feat: expose authorized Worker Drive tools and native WIP entries

Use T-722 identity-bound API for paged discovery, bounded text read/create/write/edit, folders, relocate/delete, revision-fixed images, Workdir artifact publication, and mutation receipts. Keep IDs as Workspace-bound decimal strings, observed CAS preconditions and Client validators hidden from model arguments, and never substitute revisions or blindly replay uncertain effects.

Activate via host-bound Feature/Profile/schema and centralized instructions. Publish only the Drive entrance as indexable; resolve selected stable IDs directly with coherent self-scoped WIP publication. Internal SubWorkers and Reviewer children deny undelegated Drive access at feature and client boundaries; no child grant or local storage authority is synthesized.

T-723 validation: root cargo check; manifest 190; tools 78 unit +39 integration +1 doctest; Worker 918 unit +110 integration; Runtime 348 unit +34 binary; Server 708 library +11 binary including 17 real DB+LocalFileSystem signed-API adapter tests. Native image capture/replay, multichunk revision bounds, Workdir import, concurrent CAS, current grants, response-loss reconciliation and dependent config WIP0.2 regressions pass. WIP crates.io 0.2.0 resolution verifier, fmt and diff checks pass.

Cargo.lock

@@ -6766,6 +6766,7 @@ dependencies = [
  "uuid",
  "webauthn-authenticator-rs",
  "webauthn-rs",
+ "wip-protocol",
  "workdir",
  "worker",
  "worker-runtime",

crates/manifest/src/config.rs

@@ -98,7 +98,9 @@ pub struct FeatureConfigPartial {
     #[serde(default)]
     pub workdir_catalog: Option<FeatureFlagConfigPartial>,
     #[serde(default)]
-    pub workspace_config: Option<WorkspaceConfigFeatureConfigPartial>,
+    pub workspace_config: Option<ActivationFeatureConfigPartial>,
+    #[serde(default)]
+    pub drive: Option<ActivationFeatureConfigPartial>,
     #[serde(default)]
     pub ticket: Option<TicketFeatureConfigPartial>,
     #[serde(default)]
@@ -149,7 +151,12 @@ impl FeatureConfigPartial {
             workspace_config: merge_option(
                 self.workspace_config,
                 other.workspace_config,
-                WorkspaceConfigFeatureConfigPartial::merge,
+                ActivationFeatureConfigPartial::merge,
+            ),
+            drive: merge_option(
+                self.drive,
+                other.drive,
+                ActivationFeatureConfigPartial::merge,
             ),
             ticket: merge_option(self.ticket, other.ticket, TicketFeatureConfigPartial::merge),
             merge_request: merge_option(
@@ -183,12 +190,12 @@ impl FeatureFlagConfigPartial {
 /// Activation only: no Workspace/grant/source/permission fields are accepted.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
-pub struct WorkspaceConfigFeatureConfigPartial {
+pub struct ActivationFeatureConfigPartial {
     #[serde(default)]
     pub enabled: Option<bool>,
 }
 
-impl WorkspaceConfigFeatureConfigPartial {
+impl ActivationFeatureConfigPartial {
     fn merge(self, other: Self) -> Self {
         Self {
             enabled: other.enabled.or(self.enabled),
@@ -196,15 +203,15 @@ impl WorkspaceConfigFeatureConfigPartial {
     }
 }
 
-impl From<WorkspaceConfigFeatureConfigPartial> for FeatureFlagConfig {
+impl From<ActivationFeatureConfigPartial> for FeatureFlagConfig {
-    fn from(value: WorkspaceConfigFeatureConfigPartial) -> Self {
+    fn from(value: ActivationFeatureConfigPartial) -> Self {
         Self {
             enabled: value.enabled.unwrap_or_default(),
         }
     }
 }
 
-impl From<FeatureFlagConfig> for WorkspaceConfigFeatureConfigPartial {
+impl From<FeatureFlagConfig> for ActivationFeatureConfigPartial {
     fn from(value: FeatureFlagConfig) -> Self {
         Self {
             enabled: Some(value.enabled),
@@ -448,6 +455,7 @@ impl From<FeatureConfigPartial> for FeatureConfig {
                 .workspace_config
                 .map(FeatureFlagConfig::from)
                 .unwrap_or_default(),
+            drive: value.drive.map(FeatureFlagConfig::from).unwrap_or_default(),
             ticket: value
                 .ticket
                 .map(TicketFeatureConfig::from)
@@ -643,6 +651,7 @@ impl From<FeatureConfig> for FeatureConfigPartial {
             manage_workdir: Some(value.manage_workdir.into()),
             workdir_catalog: Some(value.workdir_catalog.into()),
             workspace_config: Some(value.workspace_config.into()),
+            drive: Some(value.drive.into()),
             ticket: Some(value.ticket.into()),
             merge_request: Some(value.merge_request.into()),
             orchestration: Some(value.orchestration.into()),
@@ -2311,6 +2320,93 @@ worker_max_turns = 7
         assert!(!manifest.feature.merge_request.any());
     }
 
+    #[test]
+    fn drive_defaults_disabled_in_partial_resolved_and_missing_snapshot_fields() {
+        for source in ["", "[feature]\n", "[feature.drive]\n"] {
+            let partial = WorkerManifestConfig::from_toml(source).unwrap();
+            let manifest: WorkerManifest = minimal_valid().merge(partial).try_into().unwrap();
+            assert!(!manifest.feature.drive.enabled, "{source}");
+        }
+        for source in ["{}", r#"{"drive": {}}"#] {
+            let feature: FeatureConfig = serde_json::from_str(source).unwrap();
+            assert!(!feature.drive.enabled);
+        }
+        let manifest: WorkerManifest = minimal_valid().try_into().unwrap();
+        let mut snapshot = crate::write_persisted_worker_manifest_snapshot(&manifest).unwrap();
+        snapshot["manifest"]["feature"]
+            .as_object_mut()
+            .unwrap()
+            .remove("drive");
+        let restored = crate::read_persisted_worker_manifest_snapshot(snapshot).unwrap();
+        assert!(!restored.feature.drive.enabled);
+    }
+
+    #[test]
+    fn drive_activation_merges_and_roundtrips_config_and_snapshots() {
+        for enabled in [false, true] {
+            let lower =
+                WorkerManifestConfig::from_toml(&format!("[feature.drive]\nenabled = {enabled}\n"))
+                    .unwrap();
+            let empty = WorkerManifestConfig::from_toml("[feature.drive]\n").unwrap();
+            let manifest: WorkerManifest = minimal_valid()
+                .merge(lower.clone())
+                .merge(empty)
+                .try_into()
+                .unwrap();
+            assert_eq!(manifest.feature.drive.enabled, enabled);
+            assert!(!manifest.feature.workspace_config.enabled);
+            assert!(!manifest.feature.manage_workdir.enabled);
+            let partial = FeatureConfigPartial::from(manifest.feature.clone());
+            assert_eq!(partial.drive.as_ref().unwrap().enabled, Some(enabled));
+            let restored: FeatureConfigPartial =
+                toml::from_str(&toml::to_string(&partial).unwrap()).unwrap();
+            assert_eq!(FeatureConfig::from(restored), manifest.feature);
+            let json = serde_json::to_value(&manifest.feature).unwrap();
+            assert_eq!(json["drive"], serde_json::json!({"enabled": enabled}));
+            assert_eq!(
+                serde_json::from_value::<FeatureConfig>(json).unwrap(),
+                manifest.feature
+            );
+            let snapshot = crate::write_persisted_worker_manifest_snapshot(&manifest).unwrap();
+            let restored = crate::read_persisted_worker_manifest_snapshot(snapshot).unwrap();
+            assert_eq!(restored.feature.drive.enabled, enabled);
+            let upper = WorkerManifestConfig::from_toml(&format!(
+                "[feature.drive]\nenabled = {}\n",
+                !enabled
+            ))
+            .unwrap();
+            let overridden: WorkerManifest = minimal_valid()
+                .merge(lower)
+                .merge(upper)
+                .try_into()
+                .unwrap();
+            assert_eq!(overridden.feature.drive.enabled, !enabled);
+        }
+    }
+
+    #[test]
+    fn drive_activation_rejects_authority_and_unknown_fields() {
+        for (field, value) in [
+            ("workspace_id", "\"workspace-1\""),
+            ("grant_id", "\"grant-1\""),
+            ("access", "\"read_write\""),
+            ("host_path", "\"/host/drive\""),
+            ("source", "\"drive\""),
+            ("writable", "true"),
+            ("enabeld", "true"),
+        ] {
+            let source = format!("[feature.drive]\nenabled = true\n{field} = {value}\n");
+            assert!(WorkerManifestConfig::from_toml(&source).is_err(), "{field}");
+            assert!(
+                toml::from_str::<FeatureConfig>(&format!(
+                    "[drive]\nenabled = true\n{field} = {value}\n"
+                ))
+                .is_err(),
+                "{field}"
+            );
+        }
+    }
+
     #[test]
     fn workspace_config_defaults_disabled_for_partial_and_resolved_configs() {
         assert!(!FeatureConfig::default().workspace_config.enabled);

crates/manifest/src/lib.rs

@@ -143,8 +143,12 @@ pub struct FeatureConfig {
     pub workdir_catalog: FeatureFlagConfig,
     /// Opt-in WIP-only logical Workspace configuration feature. This flag is
     /// not an access grant; Backend-owned Worker grants remain authoritative.
-    #[serde(default, deserialize_with = "deserialize_workspace_config")]
+    #[serde(default, deserialize_with = "deserialize_activation_feature")]
     pub workspace_config: FeatureFlagConfig,
+    /// Opt-in Drive tools and native WIP surface. Activation grants no Drive
+    /// access: Backend-owned Worker grants are checked on every operation.
+    #[serde(default, deserialize_with = "deserialize_activation_feature")]
+    pub drive: FeatureFlagConfig,
     #[serde(default)]
     pub ticket: TicketFeatureConfig,
     #[serde(default)]
@@ -169,11 +173,11 @@ where
     })
 }
 
-fn deserialize_workspace_config<'de, D>(deserializer: D) -> Result<FeatureFlagConfig, D::Error>
+fn deserialize_activation_feature<'de, D>(deserializer: D) -> Result<FeatureFlagConfig, D::Error>
 where
     D: serde::Deserializer<'de>,
 {
-    Ok(config::WorkspaceConfigFeatureConfigPartial::deserialize(deserializer)?.into())
+    Ok(config::ActivationFeatureConfigPartial::deserialize(deserializer)?.into())
 }
 
 impl Default for FeatureConfig {
@@ -192,6 +196,7 @@ impl Default for FeatureConfig {
             manage_workdir: FeatureFlagConfig::disabled(),
             workdir_catalog: default_workdir_catalog(),
             workspace_config: FeatureFlagConfig::disabled(),
+            drive: FeatureFlagConfig::disabled(),
             ticket: TicketFeatureConfig::default(),
             merge_request: MergeRequestFeatureConfig::default(),
             orchestration: FeatureFlagConfig::disabled(),

crates/manifest/src/profile.rs

@@ -1279,6 +1279,34 @@ mod tests {
         }
     }
 
+    #[test]
+    fn builtin_drive_activation_is_workspace_role_specific_and_excludes_jobs() {
+        // Production DCDL imports are the policy boundary: changing the shared
+        // base must not accidentally activate Drive in a read-only Job/Reviewer.
+        let tmp = TempDir::new().unwrap();
+        let registry = ProfileDiscovery::with_sources(None, None)
+            .discover()
+            .unwrap();
+        for entry in registry.entries() {
+            let resolved = ProfileResolver::new()
+                .with_workspace_base(tmp.path())
+                .resolve(
+                    &ProfileSelector::source_named(ProfileRegistrySource::Builtin, &entry.name),
+                    ProfileResolveOptions::with_worker_name("drive-profile-worker"),
+                )
+                .unwrap();
+            let expected = matches!(
+                entry.name.as_str(),
+                "default" | "ticket-worker" | "coder" | "companion" | "intake" | "orchestrator"
+            );
+            assert_eq!(
+                resolved.manifest.feature.drive.enabled, expected,
+                "{}",
+                entry.name
+            );
+        }
+    }
+
     #[test]
     fn ticket_worker_profile_enables_work_reporting_without_authoring_or_integration() {
         let tmp = TempDir::new().unwrap();

crates/worker/src/controller.rs

@@ -1631,6 +1631,15 @@ where
         );
     }
     let wip_mode = worker.manifest().worker.mode == manifest::WorkerMode::Wip;
+    let drive_feature = crate::feature::builtin::drive::DriveFeature::configured(
+        worker.workspace_client_handle(),
+        feature_config.drive.enabled,
+        worker.workdir_sessions(),
+    )
+    .map(|feature| feature.with_permissions(worker.manifest().permissions.clone()));
+    if let Some(module) = &drive_feature {
+        feature_registry.add_module(module.clone());
+    }
     let workspace_config_feature =
         crate::feature::builtin::workspace_config::WorkspaceConfigFeature::configured(
             worker.workspace_client_handle(),
@@ -1731,6 +1740,15 @@ where
     let host_worker_observation_provider = worker.worker_observation_provider();
     let wip_permissions = worker.manifest().permissions.clone();
     let mut wip_mount_registry = crate::wip::WipMountRegistry::new();
+    if wip_mode && let Some(feature) = &drive_feature {
+        crate::feature::builtin::drive::wip::mount_drive_wip(&mut wip_mount_registry, feature)
+            .map_err(|error| {
+                std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    format!("mount Drive WIP projection: {error}"),
+                )
+            })?;
+    }
     if let Some(feature) = &workspace_config_feature {
         crate::feature::builtin::workspace_config::wip::mount_workspace_config_wip(
             &mut wip_mount_registry,
@@ -3969,6 +3987,149 @@ mod tests {
             }
         }
 
+        #[derive(Debug)]
+        struct DriveWorkspaceHost {
+            available: bool,
+            workspace: Option<&'static str>,
+        }
+
+        impl crate::worker::WorkspaceClient for DriveWorkspaceHost {
+            fn workspace_id(&self) -> Option<&str> {
+                self.workspace
+            }
+            fn kind(&self) -> &str {
+                "drive-test"
+            }
+            fn is_available(&self) -> bool {
+                self.available
+            }
+            fn execute(
+                &self,
+                request: crate::worker::WorkspaceRequest,
+            ) -> Result<crate::worker::WorkspaceResponse, crate::worker::WorkspaceClientError>
+            {
+                // Native discovery must not enumerate or perform any mutation.
+                assert_eq!(request.path, "/api/w/workspace-1/drive/root");
+                Ok(crate::worker::WorkspaceResponse { status: 200, body: serde_json::json!({
+                    "entry": {"workspace_id": "workspace-1", "node_id": "1"},
+                    "parent": null, "name": "", "kind": "folder", "revision": "1",
+                    "size": null, "content_type": null, "updated_by": "test",
+                    "updated_at": "2026-10-08T00:00:00Z",
+                    "latest_url": "/api/w/workspace-1/drive/download?entry_workspace_id=workspace-1&id=1"
+                }).to_string() })
+            }
+        }
+
+        #[tokio::test]
+        async fn drive_controller_wiring_requires_activation_and_workspace_host_in_both_modes() {
+            use crate::feature::FeatureModule as _;
+            for mode in [manifest::WorkerMode::Tools, manifest::WorkerMode::Wip] {
+                for (enabled, available, workspace) in [
+                    (false, true, Some("workspace-1")),
+                    (true, false, Some("workspace-1")),
+                    (true, true, None),
+                    (true, true, Some("workspace-1")),
+                ] {
+                    let root = TempDir::new().unwrap();
+                    let mut manifest = manifest::WorkerManifest::from_toml(
+                        "[worker]\nname = 'drive-wiring'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n"
+                    ).unwrap();
+                    manifest.worker.mode = mode;
+                    manifest.feature.drive.enabled = enabled;
+                    // Isolate Drive discovery from the independent Workspace catalog.
+                    manifest.feature.workdir_catalog.enabled = false;
+                    let client = ScriptedClient::new(vec![finished_response()]);
+                    let requests = client.requests.clone();
+                    let engine = Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated(client);
+                    let store = session_store::CombinedStore::new(
+                        session_store::FsStore::new(root.path().join("sessions")).unwrap(),
+                        session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(),
+                    );
+                    let host: Arc<dyn crate::worker::WorkspaceClient> =
+                        Arc::new(DriveWorkspaceHost {
+                            available,
+                            workspace,
+                        });
+                    let context = crate::worker::WorkerWorkspaceContext::with_client(
+                        workspace.map(|id| crate::worker::WorkspaceId::new(id).unwrap()),
+                        host.clone(),
+                    );
+                    let mut worker = Worker::new(
+                        manifest,
+                        engine,
+                        store,
+                        context,
+                        crate::worker::WorkerFilesystemAuthority::local(
+                            root.path().into(),
+                            root.path().into(),
+                        ),
+                        manifest::Scope::writable(root.path()).unwrap(),
+                    )
+                    .await
+                    .unwrap();
+                    worker.set_system_prompt_template(
+                        crate::SystemPromptTemplate::parse(
+                            "default",
+                            crate::PromptCatalogSource::builtins_only(),
+                        )
+                        .unwrap(),
+                    );
+                    let descriptor = crate::feature::builtin::drive::DriveFeature::new(
+                        host,
+                        worker.workdir_sessions(),
+                    )
+                    .descriptor();
+                    register_worker_tools(
+                        &mut worker,
+                        root.path().join("bash-output"),
+                        root.path().join("runtime"),
+                        SpawnedWorkerRegistry::new_for_internal_services(),
+                        None,
+                        None,
+                    )
+                    .await
+                    .unwrap();
+                    let configured = enabled && available && workspace.is_some();
+                    let tools = worker.engine().tool_server_handle();
+                    tools.flush_pending();
+                    for tool in &descriptor.tools {
+                        assert_eq!(
+                            tools.get_tool(&tool.name).is_some(),
+                            configured && mode == manifest::WorkerMode::Tools,
+                            "{mode:?}, enabled={enabled}, available={available}, workspace={workspace:?}, tool={}",
+                            tool.name
+                        );
+                    }
+                    if mode == manifest::WorkerMode::Wip {
+                        let (_, tree) = tools.get_tool("Tree").unwrap();
+                        let output = tree
+                            .execute(
+                                r#"{"path":"/drive","depth":1}"#,
+                                ToolExecutionContext::new("drive-wiring-tree", "batch", 0),
+                            )
+                            .await;
+                        assert_eq!(output.is_ok(), configured, "{output:?}");
+                        if let Ok(output) = output {
+                            let tree: serde_json::Value =
+                                serde_json::from_str(&output.content.unwrap()).unwrap();
+                            assert_eq!(tree["path"], "/drive");
+                            // A self-scoped entrance is an enumeration boundary,
+                            // not an observed empty Workspace hierarchy.
+                            assert_eq!(tree["children"], serde_json::Value::Null);
+                        }
+                    }
+                    worker.run_text("Inspect enabled features").await.unwrap();
+                    let requests = requests.lock().unwrap();
+                    let prompt = requests[0].system_prompt.as_deref().unwrap_or("");
+                    assert_eq!(
+                        prompt.contains("## Workspace Drive"),
+                        configured,
+                        "{mode:?}, enabled={enabled}, available={available}, workspace={workspace:?}"
+                    );
+                }
+            }
+        }
+
         struct LargeOutputTool;
 
         #[async_trait::async_trait]

crates/worker/src/feature/builtin.rs

@@ -6,6 +6,7 @@
 
 pub mod backend_job_result;
 pub mod chat_invocation;
+pub mod drive;
 pub mod flow_transition;
 pub mod manage_workdir;
 pub mod manage_worker;

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

@@ -0,0 +1,718 @@
+//! Worker Drive operations use only the injected Workspace API. Observations
+//! below are bounded preconditions, never a hierarchy replica or a grant cache.
+pub mod backend;
+pub(crate) mod delegation;
+#[cfg(test)]
+mod tests;
+pub mod wip;
+
+use crate::{
+    feature::{
+        FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule,
+        ToolContribution, ToolDeclaration,
+    },
+    worker::WorkspaceClient,
+};
+use agen::tool::{
+    Attachment, ImageAttachment, Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta,
+    ToolOutput,
+};
+use async_trait::async_trait;
+use backend::{DriveBackend, DriveError};
+use schemars::JsonSchema;
+use serde::{Deserialize, de::DeserializeOwned};
+use serde_json::{Value, json};
+use server_api::*;
+use sha2::{Digest, Sha256};
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::{Arc, Mutex},
+};
+use workdir::{WorkdirPath, WorkdirSessionCapability, WorkdirSessionRouter};
+
+pub const FEATURE_ID: &str = "drive";
+const MAX_OBSERVATIONS: usize = 256;
+
+#[derive(Default)]
+struct Observations {
+    entries: HashMap<String, DriveEntry>,
+    order: VecDeque<String>,
+}
+impl Observations {
+    fn insert(&mut self, entry: DriveEntry) {
+        let key = entry.entry.node_id.clone();
+        self.order.retain(|old| old != &key);
+        self.order.push_back(key.clone());
+        self.entries.insert(key, entry);
+        while self.order.len() > MAX_OBSERVATIONS {
+            if let Some(old) = self.order.pop_front() {
+                self.entries.remove(&old);
+            }
+        }
+    }
+    fn remove(&mut self, key: &str) {
+        self.entries.remove(key);
+        self.order.retain(|old| old != key);
+    }
+}
+
+#[derive(Clone)]
+pub struct DriveFeature {
+    pub(super) backend: DriveBackend,
+    pub(super) router: Arc<WorkdirSessionRouter>,
+    observations: Arc<Mutex<Observations>>,
+    nonce: Arc<str>,
+    pub(super) permissions: Option<manifest::ToolPermissionConfig>,
+}
+impl DriveFeature {
+    pub fn configured(
+        client: Arc<dyn WorkspaceClient>,
+        enabled: bool,
+        router: Arc<WorkdirSessionRouter>,
+    ) -> Option<Self> {
+        (enabled
+            && client.is_available()
+            && client
+                .workspace_id()
+                .is_some_and(|id| !id.is_empty() && !id.chars().any(char::is_control)))
+        .then(|| Self::new(client, router))
+    }
+    pub fn new(client: Arc<dyn WorkspaceClient>, router: Arc<WorkdirSessionRouter>) -> Self {
+        Self {
+            backend: DriveBackend::new(client),
+            router,
+            observations: Arc::new(Mutex::new(Observations::default())),
+            nonce: uuid::Uuid::now_v7().to_string().into(),
+            permissions: None,
+        }
+    }
+    pub fn with_permissions(mut self, permissions: Option<manifest::ToolPermissionConfig>) -> Self {
+        self.permissions = permissions;
+        self
+    }
+    pub fn tools(&self) -> Vec<(&'static str, ToolDefinition)> {
+        SPECS
+            .iter()
+            .map(|spec| {
+                let feature = self.clone();
+                let spec = *spec;
+                let definition: ToolDefinition = Arc::new(move || {
+                    (
+                        ToolMeta::new(spec.name)
+                            .description(spec.description)
+                            .input_schema((spec.schema)()),
+                        Arc::new(DriveTool {
+                            feature: feature.clone(),
+                            name: spec.name,
+                        }) as Arc<dyn Tool>,
+                    )
+                });
+                (spec.name, definition)
+            })
+            .collect()
+    }
+    fn observe(&self, entry: DriveEntry) -> Result<(), DriveError> {
+        self.backend.validate_entry(&entry)?;
+        self.observations
+            .lock()
+            .map_err(|_| DriveError::Unavailable)?
+            .insert(entry);
+        Ok(())
+    }
+    fn observed(
+        &self,
+        id: &DriveEntryRef,
+        pinned: Option<&DriveEntry>,
+    ) -> Result<DriveEntry, DriveError> {
+        self.backend.validate_ref(id)?;
+        if let Some(pinned) = pinned {
+            if pinned.entry != *id {
+                return Err(DriveError::Denied);
+            }
+            return Ok(pinned.clone());
+        }
+        self.observations.lock().map_err(|_| DriveError::Unavailable)?.entries.get(&id.node_id).cloned().ok_or_else(|| DriveError::Invalid("read or inspect metadata for this entry before mutation; observations are bounded and not persisted authority".into()))
+    }
+    fn request_id(&self, context: &ToolExecutionContext) -> String {
+        digest(format!("{}:{}", self.nonce, context.execution_id()).as_bytes())
+    }
+    pub(super) async fn execute(
+        &self,
+        name: &str,
+        input: &str,
+        context: &ToolExecutionContext,
+        pinned: Option<&DriveEntry>,
+    ) -> Result<DriveOutput, DriveError> {
+        self.backend.workspace()?;
+        let request_id = self.request_id(context);
+        let mut attachments = Vec::new();
+        let mut mutated_id = None;
+        let value = match name {
+            "DriveRoot" => {
+                let _: EmptyArgs = decode(input)?;
+                let entry = self.backend.root().await?;
+                self.observe(entry.clone())?;
+                entry_value(entry)
+            }
+            "DriveMetadata" => {
+                let args: EntryArgs = decode(input)?;
+                let entry = self.backend.metadata(&args.entry).await?;
+                self.observe(entry.clone())?;
+                entry_value(entry)
+            }
+            "DriveList" => {
+                let args: ListArgs = decode(input)?;
+                if args
+                    .limit
+                    .is_some_and(|n| n == 0 || n > DRIVE_PAGE_MAX_LIMIT)
+                {
+                    return Err(DriveError::Limit);
+                }
+                let parent = match args.parent {
+                    Some(parent) => parent,
+                    None => self.backend.root().await?.entry,
+                };
+                let response = self.backend.list(&parent, args.limit, args.after).await?;
+                page_value(response)
+            }
+            "DriveSearch" => {
+                let args: DriveSearchQuery = decode(input)?;
+                if args.query.len() > 512 || args.after.as_ref().is_some_and(|s| s.len() > 2048) {
+                    return Err(DriveError::Limit);
+                }
+                page_value(self.backend.search(args).await?)
+            }
+            "DriveRead" => {
+                let args: ReadArgs = decode(input)?;
+                if args.max_bytes == 0 || args.max_bytes > DRIVE_CHUNK_MAX_BYTES {
+                    return Err(DriveError::Limit);
+                }
+                let read = self.backend.read(&args.entry, args.max_bytes).await?;
+                if pinned.is_some_and(|entry| entry.revision != read.entry.revision) {
+                    return Err(DriveError::Conflict);
+                }
+                self.observe(read.entry.clone())?;
+                json!({"entry": entry_value(read.entry), "text": read.text, "truncated":read.truncated})
+            }
+            "DriveViewImage" => {
+                let args: EntryArgs = decode(input)?;
+                let entry = if let Some(entry) = pinned {
+                    self.observed(&args.entry, Some(entry))?
+                } else {
+                    self.backend.metadata(&args.entry).await?
+                };
+                let bytes = self
+                    .backend
+                    .bytes(&entry, tools::view_image::MAX_IMAGE_BYTES)
+                    .await?;
+                let media = tools::view_image::detect_image_mime(&bytes).ok_or_else(|| {
+                    DriveError::Invalid("supported image bytes required: PNG/JPEG/GIF/WebP".into())
+                })?;
+                let size = bytes.len();
+                attachments.push(Attachment::Image(ImageAttachment::new(
+                    media,
+                    Arc::<[u8]>::from(bytes),
+                )));
+                self.observe(entry.clone())?;
+                json!({"entry":entry_value(entry), "attached":true, "mime_type":media, "bytes":size})
+            }
+            "DriveCreateFolder" => {
+                let args: FolderArgs = decode(input)?;
+                self.backend.validate_ref(&args.parent)?;
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::CreateFolder {
+                                parent: args.parent,
+                                name: args.name,
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveCreateText" => {
+                let args: CreateArgs = decode(input)?;
+                self.backend.validate_ref(&args.parent)?;
+                let text = checked_text(args.content)?;
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::CreateText {
+                                parent: args.parent,
+                                name: args.name,
+                                text,
+                                content_type: args.content_type,
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveWrite" => {
+                let args: WriteArgs = decode(input)?;
+                let entry = self.observed(&args.entry, pinned)?;
+                let text = checked_text(args.content)?;
+                mutated_id = Some(entry.entry.node_id.clone());
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::UpdateText {
+                                id: entry.entry,
+                                expected_revision: entry.revision,
+                                text,
+                                content_type: entry.content_type.unwrap_or_else(default_media),
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveEdit" => {
+                let args: EditArgs = decode(input)?;
+                let entry = self.observed(&args.entry, pinned)?;
+                let edit = fs_operation::text::EditArgs {
+                    old_string: args.old_string,
+                    new_string: args.new_string,
+                    replace_all: args.replace_all,
+                };
+                edit.validate(text_limits()).map_err(text_error)?;
+                let read = self
+                    .backend
+                    .read(&entry.entry, DRIVE_TEXT_MAX_BYTES as u32)
+                    .await?;
+                if read.entry.revision != entry.revision {
+                    return Err(DriveError::Conflict);
+                }
+                if read.truncated {
+                    return Err(DriveError::Limit);
+                }
+                let text = fs_operation::text::edit(&read.text, &edit, text_limits())
+                    .map_err(text_error)?
+                    .content;
+                mutated_id = Some(entry.entry.node_id.clone());
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::UpdateText {
+                                id: entry.entry,
+                                expected_revision: entry.revision,
+                                text,
+                                content_type: entry.content_type.unwrap_or_else(default_media),
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveRelocate" => {
+                let args: RelocateArgs = decode(input)?;
+                self.backend.validate_ref(&args.parent)?;
+                let entry = self.observed(&args.entry, pinned)?;
+                mutated_id = Some(entry.entry.node_id.clone());
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::Relocate {
+                                id: entry.entry,
+                                expected_revision: entry.revision,
+                                parent: args.parent,
+                                name: args.name,
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveDelete" => {
+                let args: EntryArgs = decode(input)?;
+                let entry = self.observed(&args.entry, pinned)?;
+                mutated_id = Some(entry.entry.node_id.clone());
+                mutation_value(
+                    self.backend
+                        .mutate(DriveMutationRequest {
+                            request_id,
+                            mutation: DriveMutation::Delete {
+                                id: entry.entry,
+                                expected_revision: entry.revision,
+                            },
+                        })
+                        .await?,
+                )
+            }
+            "DriveSaveWorkdir" => {
+                let args: SaveArgs = decode(input)?;
+                self.backend.validate_ref(&args.parent)?;
+                // Drive API rechecks write at publication; source uses the selected
+                // Workdir read broker, with no OS-path or command fallback.
+                let selected = self
+                    .router
+                    .resolve(Some(&args.target_workdir))
+                    .map_err(|_| DriveError::Denied)?;
+                if !selected
+                    .session
+                    .capabilities()
+                    .supports(WorkdirSessionCapability::Read)
+                {
+                    return Err(DriveError::Denied);
+                }
+                if args.path.starts_with('/')
+                    || args.path.contains('\\')
+                    || args.path.split('/').any(|part| part == "..")
+                {
+                    return Err(DriveError::Invalid(
+                        "logical Workdir-relative path required".into(),
+                    ));
+                }
+                let path = WorkdirPath::new(&args.path).map_err(|_| {
+                    DriveError::Invalid("logical Workdir-relative path required".into())
+                })?;
+                let bytes = tools::view_image::read_workdir_bytes(
+                    selected.session.as_ref(),
+                    path,
+                    DRIVE_FILE_MAX_BYTES as usize,
+                )
+                .await
+                .map_err(|_| {
+                    DriveError::Invalid(
+                        "Workdir source denied, changed, oversized or unavailable".into(),
+                    )
+                })?;
+                let query = DriveUploadQuery {
+                    operation: DriveUploadOperation::Create,
+                    request_id,
+                    entry_workspace_id: args.parent.workspace_id,
+                    parent_id: Some(args.parent.node_id),
+                    name: Some(args.name),
+                    id: None,
+                    expected_revision: None,
+                    content_type: args.content_type,
+                    size: bytes.len() as u32,
+                    sha256: digest(&bytes),
+                };
+                mutation_value(self.backend.upload(query, bytes).await?)
+            }
+            "DriveRequestStatus" => {
+                let args: StatusArgs = decode(input)?;
+                serde_json::to_value(self.backend.status(&args.request_id).await?)
+                    .map_err(|_| DriveError::Unavailable)?
+            }
+            _ => return Err(DriveError::Invalid("unknown operation".into())),
+        };
+        // A successful mutation does not silently substitute its new revision
+        // into a prior observation. The caller observes explicitly next time.
+        if let Some(id) = mutated_id {
+            self.observations
+                .lock()
+                .map_err(|_| DriveError::Unavailable)?
+                .remove(&id);
+        }
+        Ok(DriveOutput { value, attachments })
+    }
+}
+
+pub(super) struct DriveOutput {
+    pub value: Value,
+    pub attachments: Vec<Attachment>,
+}
+impl DriveOutput {
+    fn tool(self, name: &str) -> Result<ToolOutput, ToolError> {
+        Ok(ToolOutput {
+            summary: format!("{name} completed through Workspace Drive API"),
+            content: Some(
+                serde_json::to_string(&self.value)
+                    .map_err(|_| ToolError::Internal("Drive result encoding failed".into()))?,
+            ),
+            attachments: self.attachments,
+        })
+    }
+}
+struct DriveTool {
+    feature: DriveFeature,
+    name: &'static str,
+}
+#[async_trait]
+impl Tool for DriveTool {
+    async fn execute(
+        &self,
+        input: &str,
+        context: ToolExecutionContext,
+    ) -> Result<ToolOutput, ToolError> {
+        self.feature
+            .execute(self.name, input, &context, None)
+            .await
+            .map_err(tool_error)?
+            .tool(self.name)
+    }
+}
+fn tool_error(error: DriveError) -> ToolError {
+    match error {
+        DriveError::Conflict => ToolError::StructuredConflict {
+            code: "drive_conflict".into(),
+            message: error.to_string(),
+        },
+        DriveError::Invalid(_) | DriveError::Limit => ToolError::InvalidArgument(error.to_string()),
+        _ => ToolError::ExecutionFailed(error.to_string()),
+    }
+}
+impl FeatureModule for DriveFeature {
+    fn descriptor(&self) -> FeatureDescriptor {
+        let mut descriptor = FeatureDescriptor::builtin(FEATURE_ID,"Workspace Drive").with_description("Explicit Backend-authorized Drive documents, images and artifact publication; no Workdir required").with_instruction(drive_instruction());
+        for spec in SPECS {
+            descriptor = descriptor.with_tool(ToolDeclaration::new(spec.name, spec.description));
+        }
+        descriptor
+    }
+    fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
+        context
+            .instructions()
+            .register(crate::feature::FeatureInstructionContribution::new(
+                drive_instruction(),
+            ))?;
+        for (name, definition) in self.tools() {
+            context
+                .tools()
+                .register(ToolContribution::new(name, definition))?;
+        }
+        Ok(())
+    }
+}
+
+fn drive_instruction() -> crate::feature::FeatureInstructionDeclaration {
+    crate::feature::FeatureInstructionDeclaration::new(
+        crate::feature::FeatureInstructionId::builtin("drive-workflow"),
+        "common.drive",
+        "Authorized Drive operation guidance",
+    )
+    .expect("static Drive instruction")
+}
+
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct EmptyArgs {}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct EntryArgs {
+    entry: DriveEntryRef,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct ListArgs {
+    #[serde(default)]
+    parent: Option<DriveEntryRef>,
+    #[serde(default)]
+    #[schemars(range(min = 1, max = 200))]
+    limit: Option<u32>,
+    #[serde(default)]
+    after: Option<String>,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct ReadArgs {
+    entry: DriveEntryRef,
+    #[serde(default = "default_read")]
+    #[schemars(range(min = 1, max = 65536))]
+    max_bytes: u32,
+}
+fn default_read() -> u32 {
+    DRIVE_CHUNK_MAX_BYTES
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct FolderArgs {
+    parent: DriveEntryRef,
+    name: String,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct CreateArgs {
+    parent: DriveEntryRef,
+    name: String,
+    content: String,
+    #[serde(default = "default_media")]
+    content_type: String,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct WriteArgs {
+    entry: DriveEntryRef,
+    content: String,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct EditArgs {
+    entry: DriveEntryRef,
+    old_string: String,
+    new_string: String,
+    #[serde(default)]
+    replace_all: bool,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct RelocateArgs {
+    entry: DriveEntryRef,
+    parent: DriveEntryRef,
+    name: String,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct SaveArgs {
+    target_workdir: String,
+    path: String,
+    parent: DriveEntryRef,
+    name: String,
+    #[serde(default = "default_binary_media")]
+    content_type: String,
+}
+fn default_binary_media() -> String {
+    "application/octet-stream".into()
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct StatusArgs {
+    request_id: String,
+}
+fn default_media() -> String {
+    "text/markdown".into()
+}
+fn decode<T: DeserializeOwned>(input: &str) -> Result<T, DriveError> {
+    if input.len() > 512 * 1024 {
+        return Err(DriveError::Limit);
+    }
+    serde_json::from_str(input)
+        .map_err(|_| DriveError::Invalid("input shape or bound rejected".into()))
+}
+fn text_limits() -> fs_operation::text::TextLimits {
+    fs_operation::text::TextLimits {
+        max_input_bytes: Some(DRIVE_TEXT_MAX_BYTES),
+        max_output_bytes: Some(DRIVE_TEXT_MAX_BYTES),
+        max_replacements: None,
+    }
+}
+fn checked_text(text: String) -> Result<String, DriveError> {
+    fs_operation::text::write(text, text_limits())
+        .map(|out| out.content)
+        .map_err(text_error)
+}
+fn text_error(error: fs_operation::text::TextError) -> DriveError {
+    DriveError::Invalid(error.to_string())
+}
+fn digest(bytes: &[u8]) -> String {
+    Sha256::digest(bytes)
+        .iter()
+        .map(|b| format!("{b:02x}"))
+        .collect()
+}
+pub(super) fn entry_path(entry: &DriveEntryRef) -> String {
+    format!(
+        "/drive/{}/{}",
+        crate::feature::builtin::manage_workdir::wip::encode_identity(&entry.workspace_id),
+        entry.node_id
+    )
+}
+fn entry_value(entry: DriveEntry) -> Value {
+    json!({"path":entry_path(&entry.entry),"metadata":entry})
+}
+fn page_value(page: DriveListResponse) -> Value {
+    json!({"entries":page.entries.into_iter().map(entry_value).collect::<Vec<_>>(),"next_after":page.next_after})
+}
+fn mutation_value(response: DriveMutationResponse) -> Value {
+    json!({"request_id":response.request_id,"entry":response.entry.map(entry_value)})
+}
+
+#[derive(Clone, Copy)]
+pub(super) struct Spec {
+    pub name: &'static str,
+    pub operation: &'static str,
+    pub description: &'static str,
+    pub schema: fn() -> Value,
+}
+fn schema<T: JsonSchema>() -> Value {
+    serde_json::to_value(schemars::schema_for!(T)).expect("static Drive argument schema")
+}
+pub(super) const SPECS: &[Spec] = &[
+    Spec {
+        name: "DriveRoot",
+        operation: "root",
+        description: "Discover the authorized Workspace Drive root. Profile activation is not a grant; no Workdir is required.",
+        schema: schema::<EmptyArgs>,
+    },
+    Spec {
+        name: "DriveMetadata",
+        operation: "metadata",
+        description: "Inspect one Workspace-bound stable entry ID and observe its revision for mutation.",
+        schema: schema::<EntryArgs>,
+    },
+    Spec {
+        name: "DriveList",
+        operation: "list",
+        description: "List a bounded folder page (default root), up to 200 entries. Results do not expand the WIP tree or observe revisions for writes.",
+        schema: schema::<ListArgs>,
+    },
+    Spec {
+        name: "DriveSearch",
+        operation: "search",
+        description: "Search names and optionally bounded text through Backend paged search; no embedding/index. Use the returned cursor to continue.",
+        schema: schema::<DriveSearchQuery>,
+    },
+    Spec {
+        name: "DriveRead",
+        operation: "read",
+        description: "Read up to 64 KiB UTF-8 text and observe entry revision. Past Session text remains an immutable observation.",
+        schema: schema::<ReadArgs>,
+    },
+    Spec {
+        name: "DriveCreateFolder",
+        operation: "create_folder",
+        description: "Create a folder through the explicit Drive write grant. Same-name conflicts are not overwritten.",
+        schema: schema::<FolderArgs>,
+    },
+    Spec {
+        name: "DriveCreateText",
+        operation: "create_text",
+        description: "Create bounded UTF-8 text (default Markdown), <=64 KiB. Success means DB publication committed.",
+        schema: schema::<CreateArgs>,
+    },
+    Spec {
+        name: "DriveWrite",
+        operation: "write",
+        description: "Replace <=64 KiB text using a previously read/metadata-observed revision. Conflicts are not retried with latest revision.",
+        schema: schema::<WriteArgs>,
+    },
+    Spec {
+        name: "DriveEdit",
+        operation: "edit",
+        description: "Partial text edit using previously observed revision and shared old_string rules. Require nonempty unique match unless replace_all. Truncated preimages cannot be edited.",
+        schema: schema::<EditArgs>,
+    },
+    Spec {
+        name: "DriveRelocate",
+        operation: "relocate",
+        description: "Rename and/or move an observed entry with revision precondition; stable ID and URL remain unchanged.",
+        schema: schema::<RelocateArgs>,
+    },
+    Spec {
+        name: "DriveDelete",
+        operation: "delete",
+        description: "Delete one observed entry with revision precondition. A recreated name has a different ID; stale observations cannot target it.",
+        schema: schema::<EntryArgs>,
+    },
+    Spec {
+        name: "DriveViewImage",
+        operation: "view_image",
+        description: "Attach revision-fixed PNG/JPEG/GIF/WebP bytes (<=10 MiB) through durable ToolOutput history/capture; a URL alone is not image viewing.",
+        schema: schema::<EntryArgs>,
+    },
+    Spec {
+        name: "DriveSaveWorkdir",
+        operation: "save_workdir",
+        description: "Publish a <=16 MiB artifact using selected Workdir alias and logical path plus Drive parent/name. Both source read and destination write are required. No Bash/raw host path/base64 input.",
+        schema: schema::<SaveArgs>,
+    },
+    Spec {
+        name: "DriveRequestStatus",
+        operation: "request_status",
+        description: "Query the same mutation request ID after OutcomeUnknown. Uncommitted snapshot is not proof of failure; never blindly recreate the request.",
+        schema: schema::<StatusArgs>,
+    },
+];

crates/worker/src/feature/builtin/drive/backend.rs

@@ -0,0 +1,455 @@
+//! Identity-bound Drive API adapter. The Backend owns grants, hierarchy, CAS and
+//! receipts. In particular, a lost mutation response is queried, never resent.
+use std::{sync::Arc, time::Duration};
+
+use crate::worker::{
+    WorkspaceBinaryRequest, WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod,
+};
+use serde::{Serialize, de::DeserializeOwned};
+use server_api::*;
+
+pub(super) const DEADLINE: Duration = Duration::from_secs(30);
+const RESPONSE_LIMIT: usize = 2 * 1024 * 1024;
+
+#[derive(Clone)]
+pub struct DriveBackend {
+    client: Arc<dyn WorkspaceClient>,
+}
+
+#[derive(Debug, Clone, thiserror::Error)]
+pub enum DriveError {
+    #[error("Drive access denied")]
+    Denied,
+    #[error("Drive resource not found")]
+    NotFound,
+    #[error("Drive revision or name conflict; observe explicitly before another operation")]
+    Conflict,
+    #[error("Invalid Drive arguments: {0}")]
+    Invalid(String),
+    #[error("Drive request exceeds a limit")]
+    Limit,
+    #[error("Drive storage or transport unavailable")]
+    Unavailable,
+    #[error(
+        "Drive mutation outcome unknown; request_id={0}; query request_status; do not blindly repeat"
+    )]
+    OutcomeUnknown(String),
+}
+
+impl DriveBackend {
+    pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
+        Self { client }
+    }
+    pub fn workspace(&self) -> Result<&str, DriveError> {
+        self.client
+            .workspace_id()
+            .filter(|id| !id.is_empty() && self.client.is_available())
+            .ok_or(DriveError::Denied)
+    }
+    fn base(&self) -> Result<String, DriveError> {
+        Ok(format!("/api/w/{}/drive", encode(self.workspace()?)))
+    }
+    pub(super) fn validate_entry(&self, entry: &DriveEntry) -> Result<(), DriveError> {
+        self.validate_ref(&entry.entry)?;
+        if let Some(parent) = &entry.parent {
+            self.validate_ref(parent)?;
+        }
+        if !decimal(&entry.revision) || entry.size.is_some_and(|n| n > DRIVE_FILE_MAX_BYTES) {
+            return Err(DriveError::Unavailable);
+        }
+        Ok(())
+    }
+    pub(super) fn validate_ref(&self, entry: &DriveEntryRef) -> Result<(), DriveError> {
+        if entry.workspace_id != self.workspace()? {
+            return Err(DriveError::Denied);
+        }
+        if !decimal(&entry.node_id) {
+            return Err(DriveError::Invalid(
+                "canonical decimal node ID required".into(),
+            ));
+        }
+        Ok(())
+    }
+    async fn json<T: DeserializeOwned + Send + 'static>(
+        &self,
+        request: WorkspaceRequest,
+        mutation: bool,
+        request_id: &str,
+    ) -> Result<T, DriveError> {
+        let client = self.client.clone();
+        let response = tokio::time::timeout(
+            DEADLINE,
+            tokio::task::spawn_blocking(move || client.execute_with_timeout(request, DEADLINE)),
+        )
+        .await;
+        let unknown = || {
+            if mutation {
+                DriveError::OutcomeUnknown(request_id.into())
+            } else {
+                DriveError::Unavailable
+            }
+        };
+        let response = response
+            .map_err(|_| unknown())?
+            .map_err(|_| unknown())?
+            .map_err(|_| unknown())?;
+        if response.body.len() > RESPONSE_LIMIT {
+            return Err(if mutation {
+                unknown()
+            } else {
+                DriveError::Limit
+            });
+        }
+        if !response.is_success() {
+            return Err(classify(
+                response.status,
+                response.body.as_bytes(),
+                mutation,
+                request_id,
+            ));
+        }
+        serde_json::from_str(&response.body).map_err(|_| unknown())
+    }
+    async fn get<T: DeserializeOwned + Send + 'static, Q: Serialize>(
+        &self,
+        suffix: &str,
+        query: &Q,
+    ) -> Result<T, DriveError> {
+        let query = query_string(query)?;
+        self.json(
+            WorkspaceRequest::get(format!("{}{suffix}{query}", self.base()?)),
+            false,
+            "",
+        )
+        .await
+    }
+    pub async fn root(&self) -> Result<DriveEntry, DriveError> {
+        let entry: DriveEntry = self
+            .json(
+                WorkspaceRequest::get(format!("{}/root", self.base()?)),
+                false,
+                "",
+            )
+            .await?;
+        self.validate_entry(&entry)?;
+        if entry.kind != DriveEntryKind::Folder || entry.parent.is_some() {
+            return Err(DriveError::Unavailable);
+        }
+        Ok(entry)
+    }
+    pub async fn metadata(&self, entry: &DriveEntryRef) -> Result<DriveEntry, DriveError> {
+        self.validate_ref(entry)?;
+        let result: DriveEntry = self
+            .get(
+                "/metadata",
+                &DriveEntryQuery {
+                    entry_workspace_id: entry.workspace_id.clone(),
+                    id: entry.node_id.clone(),
+                },
+            )
+            .await?;
+        self.validate_entry(&result)?;
+        if result.entry != *entry {
+            return Err(DriveError::Unavailable);
+        }
+        Ok(result)
+    }
+    pub async fn list(
+        &self,
+        parent: &DriveEntryRef,
+        limit: Option<u32>,
+        after: Option<String>,
+    ) -> Result<DriveListResponse, DriveError> {
+        self.validate_ref(parent)?;
+        let result: DriveListResponse = self
+            .get(
+                "/list",
+                &DriveListQuery {
+                    entry_workspace_id: parent.workspace_id.clone(),
+                    id: parent.node_id.clone(),
+                    limit,
+                    after,
+                },
+            )
+            .await?;
+        self.validate_page(&result, limit.unwrap_or(100).min(DRIVE_PAGE_MAX_LIMIT))?;
+        if result
+            .entries
+            .iter()
+            .any(|entry| entry.parent.as_ref() != Some(parent))
+        {
+            return Err(DriveError::Unavailable);
+        }
+        Ok(result)
+    }
+    pub async fn search(&self, query: DriveSearchQuery) -> Result<DriveListResponse, DriveError> {
+        let result: DriveListResponse = self.get("/search", &query).await?;
+        self.validate_page(
+            &result,
+            query.limit.unwrap_or(100).min(DRIVE_SEARCH_MAX_LIMIT),
+        )?;
+        Ok(result)
+    }
+    fn validate_page(&self, page: &DriveListResponse, limit: u32) -> Result<(), DriveError> {
+        if page.entries.len() > limit as usize {
+            return Err(DriveError::Limit);
+        }
+        for entry in &page.entries {
+            self.validate_entry(entry)?;
+        }
+        Ok(())
+    }
+    pub async fn read(
+        &self,
+        entry: &DriveEntryRef,
+        max_bytes: u32,
+    ) -> Result<DriveReadTextResponse, DriveError> {
+        self.validate_ref(entry)?;
+        let result: DriveReadTextResponse = self
+            .get(
+                "/read-text",
+                &DriveReadTextQuery {
+                    entry_workspace_id: entry.workspace_id.clone(),
+                    id: entry.node_id.clone(),
+                    max_bytes,
+                },
+            )
+            .await?;
+        self.validate_entry(&result.entry)?;
+        if result.entry.entry != *entry || result.text.len() > max_bytes as usize {
+            return Err(DriveError::Unavailable);
+        }
+        Ok(result)
+    }
+    pub async fn bytes(&self, entry: &DriveEntry, limit: usize) -> Result<Vec<u8>, DriveError> {
+        self.validate_entry(entry)?;
+        let total = entry
+            .size
+            .ok_or_else(|| DriveError::Invalid("file required".into()))?;
+        if total as usize > limit {
+            return Err(DriveError::Limit);
+        }
+        let mut bytes = Vec::with_capacity(total as usize);
+        // Even an empty file gets a revision-fixed read and current authorization.
+        loop {
+            let offset = bytes.len() as u32;
+            let length = total.saturating_sub(offset).clamp(1, DRIVE_CHUNK_MAX_BYTES);
+            let query = query_string(&DriveReadChunkQuery {
+                entry_workspace_id: entry.entry.workspace_id.clone(),
+                id: entry.entry.node_id.clone(),
+                expected_revision: entry.revision.clone(),
+                offset,
+                length,
+            })?;
+            let request = WorkspaceBinaryRequest {
+                method: WorkspaceRequestMethod::Get,
+                path: format!("{}/read-chunk{query}", self.base()?),
+                body: None,
+                max_response_bytes: DRIVE_CHUNK_MAX_BYTES as usize,
+            };
+            let client = self.client.clone();
+            let response = tokio::time::timeout(
+                DEADLINE,
+                tokio::task::spawn_blocking(move || {
+                    client.execute_binary_with_timeout(request, DEADLINE)
+                }),
+            )
+            .await
+            .map_err(|_| DriveError::Unavailable)?
+            .map_err(|_| DriveError::Unavailable)?
+            .map_err(|_| DriveError::Unavailable)?;
+            if !(200..300).contains(&response.status) {
+                return Err(classify(response.status, &response.body, false, ""));
+            }
+            let expected = total.saturating_sub(offset).min(length) as usize;
+            if response.body.len() != expected {
+                return Err(DriveError::Unavailable);
+            }
+            bytes.extend_from_slice(&response.body);
+            if bytes.len() == total as usize {
+                return Ok(bytes);
+            }
+        }
+    }
+    async fn reconcile(
+        &self,
+        id: &str,
+        result: Result<DriveMutationResponse, DriveError>,
+    ) -> Result<DriveMutationResponse, DriveError> {
+        let response = match result {
+            Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await {
+                Ok(DriveRequestStatusResponse {
+                    state: DriveRequestState::Committed,
+                    response: Some(response),
+                    ..
+                }) => response,
+                _ => return Err(DriveError::OutcomeUnknown(id.into())),
+            },
+            other => other?,
+        };
+        if response.request_id != id {
+            return Err(DriveError::OutcomeUnknown(id.into()));
+        }
+        if let Some(entry) = &response.entry {
+            self.validate_entry(entry)
+                .map_err(|_| DriveError::OutcomeUnknown(id.into()))?;
+        }
+        Ok(response)
+    }
+    pub async fn mutate(
+        &self,
+        request: DriveMutationRequest,
+    ) -> Result<DriveMutationResponse, DriveError> {
+        let body = serde_json::to_string(&request)
+            .map_err(|_| DriveError::Invalid("invalid mutation".into()))?;
+        if body.len() > 512 * 1024 {
+            return Err(DriveError::Limit);
+        }
+        let result = self
+            .json(
+                WorkspaceRequest::json(
+                    WorkspaceRequestMethod::Post,
+                    format!("{}/mutate", self.base()?),
+                    body,
+                ),
+                true,
+                &request.request_id,
+            )
+            .await;
+        self.reconcile(&request.request_id, result).await
+    }
+    pub async fn upload(
+        &self,
+        query: DriveUploadQuery,
+        bytes: Vec<u8>,
+    ) -> Result<DriveMutationResponse, DriveError> {
+        if bytes.len() > DRIVE_FILE_MAX_BYTES as usize {
+            return Err(DriveError::Limit);
+        }
+        let path = format!("{}/upload{}", self.base()?, query_string(&query)?);
+        let request = WorkspaceBinaryRequest {
+            method: WorkspaceRequestMethod::Put,
+            path,
+            body: Some(bytes),
+            max_response_bytes: RESPONSE_LIMIT,
+        };
+        let client = self.client.clone();
+        let result = tokio::time::timeout(
+            DEADLINE,
+            tokio::task::spawn_blocking(move || {
+                client.execute_binary_with_timeout(request, DEADLINE)
+            }),
+        )
+        .await;
+        let result = match result {
+            Ok(Ok(Ok(response))) if (200..300).contains(&response.status) => {
+                serde_json::from_slice(&response.body)
+                    .map_err(|_| DriveError::OutcomeUnknown(query.request_id.clone()))
+            }
+            Ok(Ok(Ok(response))) => Err(classify(
+                response.status,
+                &response.body,
+                true,
+                &query.request_id,
+            )),
+            _ => Err(DriveError::OutcomeUnknown(query.request_id.clone())),
+        };
+        self.reconcile(&query.request_id, result).await
+    }
+    pub async fn status(&self, request_id: &str) -> Result<DriveRequestStatusResponse, DriveError> {
+        if request_id.is_empty() || request_id.len() > 256 {
+            return Err(DriveError::Invalid("invalid request ID".into()));
+        }
+        let response: DriveRequestStatusResponse = self
+            .json(
+                WorkspaceRequest::get(format!("{}/requests/{}", self.base()?, encode(request_id))),
+                false,
+                "",
+            )
+            .await?;
+        if response.request_id != request_id {
+            return Err(DriveError::Unavailable);
+        }
+        if let Some(result) = &response.response {
+            if result.request_id != request_id {
+                return Err(DriveError::Unavailable);
+            }
+            if let Some(entry) = &result.entry {
+                self.validate_entry(entry)?;
+            }
+        }
+        Ok(response)
+    }
+}
+
+pub(super) fn decimal(value: &str) -> bool {
+    value
+        .as_bytes()
+        .first()
+        .is_some_and(|c| matches!(c, b'1'..=b'9'))
+        && value.bytes().all(|c| c.is_ascii_digit())
+        && value.parse::<i64>().is_ok_and(|id| id > 0)
+}
+pub(super) fn encode(value: &str) -> String {
+    value
+        .bytes()
+        .map(|b| {
+            if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_') {
+                (b as char).to_string()
+            } else {
+                format!("%{b:02X}")
+            }
+        })
+        .collect()
+}
+fn query_string<T: Serialize>(value: &T) -> Result<String, DriveError> {
+    let value =
+        serde_json::to_value(value).map_err(|_| DriveError::Invalid("invalid query".into()))?;
+    let object = value
+        .as_object()
+        .ok_or_else(|| DriveError::Invalid("invalid query".into()))?;
+    let pairs: Vec<_> = object
+        .iter()
+        .filter(|(_, v)| !v.is_null())
+        .map(|(k, v)| {
+            format!(
+                "{}={}",
+                encode(k),
+                encode(
+                    v.as_str()
+                        .map(str::to_owned)
+                        .unwrap_or_else(|| v.to_string())
+                        .as_str()
+                )
+            )
+        })
+        .collect();
+    Ok(if pairs.is_empty() {
+        String::new()
+    } else {
+        format!("?{}", pairs.join("&"))
+    })
+}
+fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError {
+    if let Ok(error) = serde_json::from_slice::<DriveApiError>(body) {
+        if error.classification == DriveFailureClassification::Unknown {
+            return DriveError::OutcomeUnknown(id.into());
+        }
+        return match error.code {
+            DriveApiErrorCode::Denied => DriveError::Denied,
+            DriveApiErrorCode::NotFound => DriveError::NotFound,
+            DriveApiErrorCode::Conflict => DriveError::Conflict,
+            DriveApiErrorCode::Invalid => DriveError::Invalid("Backend rejected arguments".into()),
+            DriveApiErrorCode::Limit => DriveError::Limit,
+            DriveApiErrorCode::StorageUnavailable => DriveError::Unavailable,
+            DriveApiErrorCode::OutcomeUnknown => DriveError::OutcomeUnknown(id.into()),
+        };
+    }
+    if matches!(status, 401 | 403) {
+        DriveError::Denied
+    } else if mutation {
+        DriveError::OutcomeUnknown(id.into())
+    } else {
+        DriveError::Unavailable
+    }
+}

crates/worker/src/feature/builtin/drive/delegation.rs

@@ -0,0 +1,477 @@
+//! Fail-closed Drive boundary for parent-owned Internal SubWorkers, including Reviewers.
+//!
+//! Internal children have no Backend Drive identity. Explicit internal Drive delegation is not
+//! supported until the Backend can approve scope bound to that child identity. A parent's current
+//! grant, enabled profile, or Reviewer attestation is not a child Drive grant. Registered Runtime
+//! Workers must use their own identity-bound, current Backend grants; this wrapper creates neither
+//! identities nor grants and keeps no local authorization cache.
+
+use std::{sync::Arc, time::Duration};
+
+use crate::worker::{
+    ReviewerContext, WorkerWorkspaceContext, WorkspaceBinaryRequest, WorkspaceBinaryResponse,
+    WorkspaceClient, WorkspaceClientError, WorkspacePromptCatalogResolution, WorkspaceRequest,
+    WorkspaceResponse, WorkspaceServerOperation, WorkspaceWorkerDiscoveryRequest,
+};
+
+/// Removes Drive authority without changing the selected client's other authority or behavior.
+/// Install outside any Reviewer client so its trusted review submission attestation is preserved.
+#[derive(Debug)]
+pub(crate) struct InternalChildWorkspaceClient {
+    inner: Arc<dyn WorkspaceClient>,
+}
+
+impl InternalChildWorkspaceClient {
+    pub(crate) fn new(inner: Arc<dyn WorkspaceClient>) -> Self {
+        Self { inner }
+    }
+
+    fn check_path(path: &str) -> Result<(), WorkspaceClientError> {
+        // Classify the route, not the method: even readonly Drive access must not inherit the
+        // parent's identity. Decode route segments and normalize dot segments before checking,
+        // so transport/router normalization cannot turn an allowed alias into a Drive request.
+        let path = path.split(['?', '#']).next().unwrap_or(path);
+        let mut decoded = Vec::with_capacity(path.len());
+        // HTTP URL parsers discard literal tabs/newlines before route normalization.
+        let bytes: Vec<_> = path
+            .bytes()
+            .filter(|byte| !matches!(byte, b'\t' | b'\r' | b'\n'))
+            .collect();
+        let mut index = 0;
+        while index < bytes.len() {
+            if bytes[index] == b'%'
+                && index + 2 < bytes.len()
+                && let (Some(high), Some(low)) =
+                    (hex_digit(bytes[index + 1]), hex_digit(bytes[index + 2]))
+            {
+                decoded.push(high * 16 + low);
+                index += 3;
+            } else {
+                decoded.push(bytes[index]);
+                index += 1;
+            }
+        }
+        let decoded = String::from_utf8_lossy(&decoded).replace('\\', "/");
+        let mut segments = Vec::new();
+        for segment in decoded.split('/') {
+            match segment {
+                "" | "." => {}
+                ".." => {
+                    segments.pop();
+                }
+                segment => segments.push(segment),
+            }
+        }
+        if matches!(segments.as_slice(), ["api", "w", _, "drive", ..]) {
+            return Err(WorkspaceClientError::Unavailable(
+                "Internal SubWorker Drive access denied: no identity-bound Backend-approved child scope; explicit internal Drive delegation is not supported".into(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+fn hex_digit(byte: u8) -> Option<u8> {
+    match byte {
+        b'0'..=b'9' => Some(byte - b'0'),
+        b'a'..=b'f' => Some(byte - b'a' + 10),
+        b'A'..=b'F' => Some(byte - b'A' + 10),
+        _ => None,
+    }
+}
+
+/// Keep Workspace identity and prompt context, but never inherit Drive authority.
+pub(crate) fn restrict_internal_child_drive(
+    context: WorkerWorkspaceContext,
+) -> WorkerWorkspaceContext {
+    WorkerWorkspaceContext::with_client(
+        context.workspace_id().cloned(),
+        Arc::new(InternalChildWorkspaceClient::new(context.client_handle())),
+    )
+}
+
+impl WorkspaceClient for InternalChildWorkspaceClient {
+    fn workspace_id(&self) -> Option<&str> {
+        self.inner.workspace_id()
+    }
+    fn kind(&self) -> &str {
+        self.inner.kind()
+    }
+    fn is_available(&self) -> bool {
+        self.inner.is_available()
+    }
+
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        Self::check_path(&request.path)?;
+        self.inner.execute(request)
+    }
+
+    fn execute_with_timeout(
+        &self,
+        request: WorkspaceRequest,
+        timeout: Duration,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        Self::check_path(&request.path)?;
+        self.inner.execute_with_timeout(request, timeout)
+    }
+
+    fn execute_binary(
+        &self,
+        request: WorkspaceBinaryRequest,
+    ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+        Self::check_path(&request.path)?;
+        self.inner.execute_binary(request)
+    }
+
+    fn execute_binary_with_timeout(
+        &self,
+        request: WorkspaceBinaryRequest,
+        timeout: Duration,
+    ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+        Self::check_path(&request.path)?;
+        self.inner.execute_binary_with_timeout(request, timeout)
+    }
+
+    fn execute_server_operation(
+        &self,
+        operation: WorkspaceServerOperation,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        self.inner.execute_server_operation(operation)
+    }
+
+    fn current_prompt_projection(
+        &self,
+        minimum_revision: Option<u64>,
+    ) -> Result<Option<WorkspacePromptCatalogResolution>, WorkspaceClientError> {
+        self.inner.current_prompt_projection(minimum_revision)
+    }
+
+    fn list_workspace_workers(
+        &self,
+        request: WorkspaceWorkerDiscoveryRequest,
+    ) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
+        self.inner.list_workspace_workers(request)
+    }
+
+    fn execute_worker_remove(
+        &self,
+        target_runtime_id: &str,
+        target_worker_id: &str,
+        reason: &str,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        self.inner
+            .execute_worker_remove(target_runtime_id, target_worker_id, reason)
+    }
+
+    fn reviewer_context(&self) -> Option<&ReviewerContext> {
+        self.inner.reviewer_context()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::worker::{ReviewerChildWorkspaceClient, WorkspaceRequestMethod};
+    use std::sync::Mutex;
+
+    #[derive(Debug, PartialEq, Eq)]
+    enum Call {
+        Json(WorkspaceRequest, Option<Duration>),
+        Binary(WorkspaceBinaryRequest, Option<Duration>),
+        Projection(Option<u64>),
+        Discovery(WorkspaceWorkerDiscoveryRequest),
+        Remove(String, String, String),
+        Server,
+    }
+
+    #[derive(Debug, Default)]
+    struct RecordingClient {
+        calls: Mutex<Vec<Call>>,
+    }
+
+    impl RecordingClient {
+        fn response(&self, call: Call) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.calls.lock().unwrap().push(call);
+            Ok(WorkspaceResponse {
+                status: 202,
+                body: "forwarded".into(),
+            })
+        }
+        fn binary_response(
+            &self,
+            call: Call,
+        ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+            self.calls.lock().unwrap().push(call);
+            Ok(WorkspaceBinaryResponse {
+                status: 206,
+                body: vec![0, 255],
+            })
+        }
+    }
+
+    impl WorkspaceClient for RecordingClient {
+        fn workspace_id(&self) -> Option<&str> {
+            Some("ws")
+        }
+        fn kind(&self) -> &str {
+            "recording"
+        }
+        fn is_available(&self) -> bool {
+            true
+        }
+        fn execute(
+            &self,
+            request: WorkspaceRequest,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.response(Call::Json(request, None))
+        }
+        fn execute_with_timeout(
+            &self,
+            request: WorkspaceRequest,
+            timeout: Duration,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.response(Call::Json(request, Some(timeout)))
+        }
+        fn execute_binary(
+            &self,
+            request: WorkspaceBinaryRequest,
+        ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+            self.binary_response(Call::Binary(request, None))
+        }
+        fn execute_binary_with_timeout(
+            &self,
+            request: WorkspaceBinaryRequest,
+            timeout: Duration,
+        ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+            self.binary_response(Call::Binary(request, Some(timeout)))
+        }
+        fn current_prompt_projection(
+            &self,
+            revision: Option<u64>,
+        ) -> Result<Option<WorkspacePromptCatalogResolution>, WorkspaceClientError> {
+            self.calls.lock().unwrap().push(Call::Projection(revision));
+            Err(WorkspaceClientError::Request("projection marker".into()))
+        }
+        fn list_workspace_workers(
+            &self,
+            request: WorkspaceWorkerDiscoveryRequest,
+        ) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
+            self.calls.lock().unwrap().push(Call::Discovery(request));
+            Err(WorkspaceClientError::Request("discovery marker".into()))
+        }
+        fn execute_worker_remove(
+            &self,
+            runtime: &str,
+            worker: &str,
+            reason: &str,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.response(Call::Remove(runtime.into(), worker.into(), reason.into()))
+        }
+        fn execute_server_operation(
+            &self,
+            _: WorkspaceServerOperation,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.response(Call::Server)
+        }
+    }
+
+    fn binary(method: WorkspaceRequestMethod, path: &str) -> WorkspaceBinaryRequest {
+        WorkspaceBinaryRequest {
+            method,
+            path: path.into(),
+            body: Some(vec![0, 128, 255]),
+            max_response_bytes: 17,
+        }
+    }
+
+    #[test]
+    fn internal_children_deny_all_drive_transports_without_parent_side_effects() {
+        let inner = Arc::new(RecordingClient::default());
+        for reviewer in [false, true] {
+            let selected: Arc<dyn WorkspaceClient> = if reviewer {
+                Arc::new(ReviewerChildWorkspaceClient::new(
+                    inner.clone(),
+                    ReviewerContext {
+                        ticket_id: "T1".into(),
+                        merge_request_id: "MR1".into(),
+                    },
+                    "attestation".into(),
+                ))
+            } else {
+                inner.clone()
+            };
+            let client = InternalChildWorkspaceClient::new(selected);
+            for path in [
+                "/api/w/ws/drive",
+                "/api/w/ws/drive?limit=1",
+                "/api/w/other/drive/root",
+                "/api/w/ws/drive/read-chunk?id=1",
+                "/api/w/ws/drive/upload",
+                "/api/w/ws/drive/requests/request-id",
+                "/api/w/ws/%64rive/root",
+                "/api/w/ws/tickets/../drive/root",
+                "/api/w/ws/tickets/%2e%2e/drive/root",
+                "/api/w/ws%2fdrive/root",
+                "/api/w/ws/dr\tive/root",
+                "/api/w/ws/dri\r\nve/root",
+            ] {
+                for method in [
+                    WorkspaceRequestMethod::Get,
+                    WorkspaceRequestMethod::Post,
+                    WorkspaceRequestMethod::Put,
+                    WorkspaceRequestMethod::Patch,
+                    WorkspaceRequestMethod::Delete,
+                ] {
+                    let request = WorkspaceRequest::json(method, path, "{}");
+                    for result in [
+                        client.execute(request.clone()),
+                        client.execute_with_timeout(request, Duration::from_secs(2)),
+                    ] {
+                        assert!(
+                            matches!(result, Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
+                            "path={path}, reviewer={reviewer}"
+                        );
+                    }
+                    for result in [
+                        client.execute_binary(binary(method, path)),
+                        client.execute_binary_with_timeout(
+                            binary(method, path),
+                            Duration::from_secs(2),
+                        ),
+                    ] {
+                        assert!(
+                            matches!(result, Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
+                            "path={path}, reviewer={reviewer}"
+                        );
+                    }
+                }
+            }
+        }
+        assert!(inner.calls.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn non_drive_requests_preserve_payloads_deadlines_and_special_operations() {
+        let inner = Arc::new(RecordingClient::default());
+        let client = InternalChildWorkspaceClient::new(inner.clone());
+        let request = WorkspaceRequest::json(
+            WorkspaceRequestMethod::Post,
+            "/api/w/ws/tickets/query",
+            "{\"query\":\"drive\"}",
+        );
+        let bytes = binary(WorkspaceRequestMethod::Put, "/api/w/ws/files/content");
+        let timeout = Duration::from_millis(37);
+        let expected = WorkspaceResponse {
+            status: 202,
+            body: "forwarded".into(),
+        };
+        assert_eq!(client.execute(request.clone()).unwrap(), expected);
+        assert_eq!(
+            client
+                .execute_with_timeout(request.clone(), timeout)
+                .unwrap(),
+            expected
+        );
+        let expected_bytes = WorkspaceBinaryResponse {
+            status: 206,
+            body: vec![0, 255],
+        };
+        assert_eq!(
+            client.execute_binary(bytes.clone()).unwrap(),
+            expected_bytes
+        );
+        assert_eq!(
+            client
+                .execute_binary_with_timeout(bytes.clone(), timeout)
+                .unwrap(),
+            expected_bytes
+        );
+        assert!(
+            matches!(client.current_prompt_projection(Some(42)), Err(WorkspaceClientError::Request(message)) if message == "projection marker")
+        );
+        let discovery = WorkspaceWorkerDiscoveryRequest {
+            cursor: Some("next".into()),
+            limit: 3,
+            query: Some("coder".into()),
+        };
+        assert!(
+            matches!(client.list_workspace_workers(discovery.clone()), Err(WorkspaceClientError::Request(message)) if message == "discovery marker")
+        );
+        assert_eq!(
+            client
+                .execute_worker_remove("runtime", "worker", "reason")
+                .unwrap(),
+            expected
+        );
+        assert_eq!(
+            client
+                .execute_server_operation(WorkspaceServerOperation::WorkerControlList)
+                .unwrap(),
+            expected
+        );
+        assert_eq!(client.workspace_id(), Some("ws"));
+        assert_eq!(client.kind(), "recording");
+        assert!(client.is_available());
+        assert_eq!(
+            *inner.calls.lock().unwrap(),
+            vec![
+                Call::Json(request.clone(), None),
+                Call::Json(request, Some(timeout)),
+                Call::Binary(bytes.clone(), None),
+                Call::Binary(bytes, Some(timeout)),
+                Call::Projection(Some(42)),
+                Call::Discovery(discovery),
+                Call::Remove("runtime".into(), "worker".into(), "reason".into()),
+                Call::Server,
+            ]
+        );
+    }
+
+    #[test]
+    fn drive_restriction_preserves_reviewer_context_and_review_attestation() {
+        let inner = Arc::new(RecordingClient::default());
+        let context = ReviewerContext {
+            ticket_id: "T1".into(),
+            merge_request_id: "MR1".into(),
+        };
+        let client =
+            InternalChildWorkspaceClient::new(Arc::new(ReviewerChildWorkspaceClient::new(
+                inner.clone(),
+                context.clone(),
+                "trusted-token".into(),
+            )));
+        assert_eq!(client.reviewer_context(), Some(&context));
+        assert_eq!(client.kind(), "runtime-reviewer-child");
+        let request = WorkspaceRequest::json(
+            WorkspaceRequestMethod::Post,
+            "/api/w/ws/merge-requests/MR1/reviews",
+            "{\"verdict\":\"approve\"}",
+        );
+        // Reviewer timeout execution uses its existing JSON attestation path, not raw parent access.
+        client
+            .execute_with_timeout(request, Duration::from_secs(1))
+            .unwrap();
+        let calls = inner.calls.lock().unwrap();
+        let [Call::Json(request, None)] = calls.as_slice() else {
+            panic!("unexpected calls: {calls:?}")
+        };
+        let body: serde_json::Value = serde_json::from_str(request.body.as_ref().unwrap()).unwrap();
+        assert_eq!(body["capability_token"], "trusted-token");
+        assert_eq!(body["verdict"], "approve");
+        drop(calls);
+        let count = inner.calls.lock().unwrap().len();
+        assert!(
+            client
+                .execute(WorkspaceRequest::json(
+                    WorkspaceRequestMethod::Delete,
+                    "/api/w/ws/tickets/T1",
+                    "{}"
+                ))
+                .is_err()
+        );
+        assert_eq!(inner.calls.lock().unwrap().len(), count);
+    }
+}

crates/worker/src/feature/builtin/drive/tests.rs

@@ -0,0 +1,514 @@
+use super::*;
+use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
+use serde::Serialize;
+use std::collections::VecDeque;
+
+#[derive(Debug)]
+struct Client {
+    responses: Mutex<VecDeque<Result<WorkspaceResponse, WorkspaceClientError>>>,
+    requests: Mutex<Vec<WorkspaceRequest>>,
+}
+impl Client {
+    fn new(responses: Vec<WorkspaceResponse>) -> Arc<Self> {
+        Arc::new(Self {
+            responses: Mutex::new(responses.into_iter().map(Ok).collect()),
+            requests: Mutex::new(Vec::new()),
+        })
+    }
+}
+impl WorkspaceClient for Client {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace")
+    }
+    fn kind(&self) -> &str {
+        "drive-test"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        self.requests.lock().unwrap().push(request);
+        self.responses
+            .lock()
+            .unwrap()
+            .pop_front()
+            .expect("unexpected Backend call")
+    }
+}
+fn response(value: impl Serialize) -> WorkspaceResponse {
+    WorkspaceResponse {
+        status: 200,
+        body: serde_json::to_string(&value).unwrap(),
+    }
+}
+fn entry(id: &str, revision: &str) -> DriveEntry {
+    DriveEntry {
+        entry: DriveEntryRef {
+            workspace_id: "workspace".into(),
+            node_id: id.into(),
+        },
+        parent: Some(DriveEntryRef {
+            workspace_id: "workspace".into(),
+            node_id: "1".into(),
+        }),
+        name: "note.md".into(),
+        kind: DriveEntryKind::File,
+        revision: revision.into(),
+        size: Some(5),
+        content_type: Some("text/markdown".into()),
+        updated_by: "worker".into(),
+        updated_at: "now".into(),
+        latest_url: format!("/api/w/workspace/drive/download?id={id}"),
+    }
+}
+fn feature(client: Arc<Client>) -> DriveFeature {
+    DriveFeature::new(client, Arc::new(WorkdirSessionRouter::new()))
+}
+async fn execute(
+    feature: &DriveFeature,
+    name: &str,
+    input: Value,
+) -> Result<DriveOutput, DriveError> {
+    feature
+        .execute(
+            name,
+            &input.to_string(),
+            &ToolExecutionContext::new(name, "batch", 0),
+            None,
+        )
+        .await
+}
+
+#[test]
+fn drive_feature_activation_never_creates_authority_and_schemas_hide_revisions() {
+    let client = Client::new(Vec::new());
+    let router = Arc::new(WorkdirSessionRouter::new());
+    assert!(DriveFeature::configured(client.clone(), false, router.clone()).is_none());
+    assert!(
+        DriveFeature::configured(
+            crate::worker::unavailable_workspace_client(None, "standalone"),
+            true,
+            router
+        )
+        .is_none()
+    );
+    let feature = feature(client);
+    for (name, definition) in feature.tools() {
+        let (meta, _) = definition();
+        assert_eq!(name, meta.name);
+        let schema = meta.input_schema.to_string();
+        assert!(!schema.contains("expected_revision"), "{name}");
+        assert!(!schema.contains("grant_id"), "{name}");
+    }
+    assert_eq!(feature.descriptor().instructions.len(), 1);
+}
+
+#[tokio::test]
+async fn foreign_or_unobserved_entry_cannot_dispatch_mutation() {
+    let client = Client::new(Vec::new());
+    let feature = feature(client.clone());
+    let own = json!({"workspace_id":"workspace","node_id":"9007199254740993"});
+    assert!(matches!(
+        execute(&feature, "DriveWrite", json!({"entry":own,"content":"new"})).await,
+        Err(DriveError::Invalid(_))
+    ));
+    assert!(matches!(
+        execute(
+            &feature,
+            "DriveDelete",
+            json!({"entry":{"workspace_id":"other","node_id":"1"}})
+        )
+        .await,
+        Err(DriveError::Denied)
+    ));
+    for id in [json!(1), json!("01"), json!("9223372036854775808")] {
+        assert!(matches!(
+            execute(
+                &feature,
+                "DriveMetadata",
+                json!({"entry":{"workspace_id":"workspace","node_id":id}})
+            )
+            .await,
+            Err(DriveError::Invalid(_))
+        ));
+    }
+    assert!(client.requests.lock().unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn stale_edit_returns_conflict_without_upgrading_revision_or_dispatching_write() {
+    let old = entry("2", "1");
+    let current = entry("2", "2");
+    let client = Client::new(vec![response(DriveReadTextResponse {
+        entry: current,
+        text: "hello".into(),
+        truncated: false,
+    })]);
+    let feature = feature(client.clone());
+    feature.observe(old.clone()).unwrap();
+    let result = execute(
+        &feature,
+        "DriveEdit",
+        json!({"entry":old.entry,"old_string":"hello","new_string":"new"}),
+    )
+    .await;
+    assert!(matches!(result, Err(DriveError::Conflict)));
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+    assert_eq!(feature.observed(&old.entry, None).unwrap().revision, "1");
+}
+
+#[tokio::test]
+async fn shared_edit_rules_reject_ambiguous_and_truncated_preimages_without_mutation() {
+    for (text, truncated) in [("hello hello", false), ("hello", true)] {
+        let old = entry("2", "1");
+        let client = Client::new(vec![response(DriveReadTextResponse {
+            entry: old.clone(),
+            text: text.into(),
+            truncated,
+        })]);
+        let feature = feature(client.clone());
+        feature.observe(old.clone()).unwrap();
+        assert!(
+            execute(
+                &feature,
+                "DriveEdit",
+                json!({"entry":old.entry,"old_string":"hello","new_string":"new"})
+            )
+            .await
+            .is_err()
+        );
+        assert!(
+            client
+                .requests
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|r| r.method == crate::worker::WorkspaceRequestMethod::Get)
+        );
+    }
+}
+
+#[tokio::test]
+async fn successful_mutation_uses_observed_decimal_revision_and_requires_next_explicit_observation()
+{
+    let old = entry("9007199254740993", "9007199254740993");
+    let updated = entry("9007199254740993", "9007199254740994");
+    let client = Client::new(Vec::new());
+    let feature = feature(client.clone());
+    feature.observe(old.clone()).unwrap();
+    let context = ToolExecutionContext::new("write", "batch", 0);
+    let id = feature.request_id(&context);
+    client
+        .responses
+        .lock()
+        .unwrap()
+        .push_back(Ok(response(DriveMutationResponse {
+            request_id: id,
+            entry: Some(updated),
+        })));
+    feature
+        .execute(
+            "DriveWrite",
+            &json!({"entry":old.entry,"content":"new"}).to_string(),
+            &context,
+            None,
+        )
+        .await
+        .unwrap();
+    let requests = client.requests.lock().unwrap();
+    let body: Value = serde_json::from_str(requests[0].body.as_ref().unwrap()).unwrap();
+    assert_eq!(body["mutation"]["expected_revision"], "9007199254740993");
+    assert_eq!(body["mutation"]["id"]["node_id"], "9007199254740993");
+    drop(requests);
+    assert!(feature.observed(&old.entry, None).is_err());
+}
+
+#[tokio::test]
+async fn lost_mutation_response_queries_same_receipt_without_resending() {
+    let old = entry("2", "1");
+    let client = Client::new(Vec::new());
+    let feature = feature(client.clone());
+    feature.observe(old.clone()).unwrap();
+    let context = ToolExecutionContext::new("write", "batch", 0);
+    let id = feature.request_id(&context);
+    client.responses.lock().unwrap().extend([
+        Err(WorkspaceClientError::Request("lost".into())),
+        Ok(response(DriveRequestStatusResponse {
+            request_id: id.clone(),
+            state: DriveRequestState::Committed,
+            response: Some(DriveMutationResponse {
+                request_id: id.clone(),
+                entry: Some(entry("2", "2")),
+            }),
+        })),
+    ]);
+    feature
+        .execute(
+            "DriveWrite",
+            &json!({"entry":old.entry,"content":"new"}).to_string(),
+            &context,
+            None,
+        )
+        .await
+        .unwrap();
+    let requests = client.requests.lock().unwrap();
+    assert_eq!(requests.len(), 2);
+    assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
+    assert_eq!(
+        requests
+            .iter()
+            .filter(|r| r.method == crate::worker::WorkspaceRequestMethod::Post)
+            .count(),
+        1
+    );
+}
+
+#[tokio::test]
+async fn uncommitted_receipt_snapshot_does_not_turn_unknown_into_failure_or_retry() {
+    let old = entry("2", "1");
+    let client = Client::new(Vec::new());
+    let feature = feature(client.clone());
+    feature.observe(old.clone()).unwrap();
+    let context = ToolExecutionContext::new("write", "batch", 0);
+    let id = feature.request_id(&context);
+    client.responses.lock().unwrap().extend([
+        Err(WorkspaceClientError::Request("lost".into())),
+        Ok(response(DriveRequestStatusResponse {
+            request_id: id.clone(),
+            state: DriveRequestState::Uncommitted,
+            response: None,
+        })),
+    ]);
+    let result = feature
+        .execute(
+            "DriveWrite",
+            &json!({"entry":old.entry,"content":"new"}).to_string(),
+            &context,
+            None,
+        )
+        .await;
+    assert!(matches!(result,Err(DriveError::OutcomeUnknown(ref observed)) if observed==&id));
+    assert_eq!(client.requests.lock().unwrap().len(), 2);
+}
+
+#[tokio::test]
+async fn observation_eviction_never_substitutes_another_entry() {
+    let feature = feature(Client::new(Vec::new()));
+    let first = entry("2", "1");
+    feature.observe(first.clone()).unwrap();
+    for id in 3..=258 {
+        feature.observe(entry(&id.to_string(), "1")).unwrap();
+    }
+    assert!(feature.observed(&first.entry, None).is_err());
+    let state = feature.observations.lock().unwrap();
+    assert_eq!(state.entries.len(), MAX_OBSERVATIONS);
+    assert_eq!(state.order.len(), MAX_OBSERVATIONS);
+}
+
+#[tokio::test]
+async fn backend_error_details_are_not_reflected_and_conflict_is_not_retried() {
+    let client = Client::new(vec![WorkspaceResponse {
+        status: 409,
+        body: serde_json::to_string(&DriveApiError {
+            code: DriveApiErrorCode::Conflict,
+            classification: DriveFailureClassification::NotCommitted,
+            message: "/secret/provider/path".into(),
+        })
+        .unwrap(),
+    }]);
+    let feature = feature(client.clone());
+    let old = entry("2", "1");
+    feature.observe(old.clone()).unwrap();
+    let result = execute(&feature, "DriveDelete", json!({"entry":old.entry}))
+        .await
+        .err()
+        .unwrap();
+    assert!(matches!(result, DriveError::Conflict));
+    assert!(!result.to_string().contains("secret"));
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn drive_native_entry_is_directly_resolvable_but_never_indexable_and_binds_identity() {
+    use crate::wip::WipSubtreeProvider;
+    let entry = entry("9007199254740993", "3");
+    let client = Client::new(vec![response(entry.clone())]);
+    let feature = feature(client.clone());
+    let mut mounts = crate::wip::WipMountRegistry::new();
+    wip::mount_drive_wip(&mut mounts, &feature).unwrap();
+    let provider = wip::DriveProvider(feature);
+    let path = entry_path(&entry.entry);
+    let publication = provider.publication(&path).await.unwrap().unwrap();
+    assert_eq!(publication.projection.route, path);
+    assert_eq!(
+        publication.scope.as_ref().unwrap().r#ref,
+        publication.projection.object.r#ref
+    );
+    assert_eq!(publication.projection.object.validator, Some(b"3".to_vec()));
+    assert!(provider.children("/drive").await.unwrap().is_empty());
+    assert!(
+        provider
+            .publication("/drive/other/9007199254740993")
+            .await
+            .unwrap()
+            .is_none()
+    );
+    let operations = &publication.projection.descriptor.operations;
+    assert!(operations.iter().any(|op| op.name == "view_image"));
+    assert!(
+        !operations
+            .iter()
+            .flat_map(|op| &op.parameters)
+            .any(|parameter| parameter.name == "expected_revision")
+    );
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+}
+
+#[derive(Debug)]
+struct ChunkClient {
+    json: Arc<Client>,
+    chunks: Mutex<VecDeque<crate::worker::WorkspaceBinaryResponse>>,
+    requests: Mutex<Vec<crate::worker::WorkspaceBinaryRequest>>,
+}
+impl WorkspaceClient for ChunkClient {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace")
+    }
+    fn kind(&self) -> &str {
+        "drive-chunk-test"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        self.json.execute(request)
+    }
+    fn execute_binary(
+        &self,
+        request: crate::worker::WorkspaceBinaryRequest,
+    ) -> Result<crate::worker::WorkspaceBinaryResponse, WorkspaceClientError> {
+        self.requests.lock().unwrap().push(request);
+        Ok(self
+            .chunks
+            .lock()
+            .unwrap()
+            .pop_front()
+            .expect("unexpected binary read"))
+    }
+}
+fn chunk_client(
+    metadata: DriveEntry,
+    chunks: Vec<crate::worker::WorkspaceBinaryResponse>,
+) -> Arc<ChunkClient> {
+    Arc::new(ChunkClient {
+        json: Client::new(vec![response(metadata)]),
+        chunks: Mutex::new(chunks.into()),
+        requests: Mutex::new(Vec::new()),
+    })
+}
+fn chunk(bytes: Vec<u8>) -> crate::worker::WorkspaceBinaryResponse {
+    crate::worker::WorkspaceBinaryResponse {
+        status: 200,
+        body: bytes,
+    }
+}
+#[tokio::test]
+async fn multi_chunk_image_uses_one_revision_and_never_exposes_binary_in_text_content() {
+    let mut metadata = entry("2", "9007199254740993");
+    let mut bytes = b"\x89PNG\r\n\x1a\n".to_vec();
+    bytes.resize(DRIVE_CHUNK_MAX_BYTES as usize + 13, 42);
+    metadata.size = Some(bytes.len() as u32);
+    let client = chunk_client(
+        metadata.clone(),
+        vec![
+            chunk(bytes[..DRIVE_CHUNK_MAX_BYTES as usize].to_vec()),
+            chunk(bytes[DRIVE_CHUNK_MAX_BYTES as usize..].to_vec()),
+        ],
+    );
+    let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
+    let output = execute(&feature, "DriveViewImage", json!({"entry":metadata.entry}))
+        .await
+        .unwrap();
+    let Attachment::Image(image) = &output.attachments[0];
+    assert_eq!(image.data(), bytes);
+    assert!(output.value.to_string().len() < 2048);
+    let requests = client.requests.lock().unwrap();
+    assert_eq!(requests.len(), 2);
+    for request in &*requests {
+        assert!(request.path.contains("expected_revision=9007199254740993"));
+        assert_eq!(request.max_response_bytes, DRIVE_CHUNK_MAX_BYTES as usize);
+    }
+    assert!(requests[0].path.split('&').any(|pair| pair == "offset=0"));
+    assert!(
+        requests[1]
+            .path
+            .split('&')
+            .any(|pair| pair == "offset=65536")
+    );
+}
+#[tokio::test]
+async fn expired_revision_or_partial_chunk_aborts_image_without_rebinding_or_retrying() {
+    for next in [
+        crate::worker::WorkspaceBinaryResponse {
+            status: 409,
+            body: serde_json::to_vec(&DriveApiError::new(DriveApiErrorCode::Conflict)).unwrap(),
+        },
+        chunk(vec![1]), // second range requires 13 bytes, not a successful prefix
+    ] {
+        let mut metadata = entry("2", "3");
+        metadata.size = Some(DRIVE_CHUNK_MAX_BYTES + 13);
+        let mut first = vec![0; DRIVE_CHUNK_MAX_BYTES as usize];
+        first[..8].copy_from_slice(b"\x89PNG\r\n\x1a\n");
+        let client = chunk_client(metadata.clone(), vec![chunk(first), next]);
+        let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
+        assert!(
+            execute(&feature, "DriveViewImage", json!({"entry":metadata.entry}))
+                .await
+                .is_err()
+        );
+        let requests = client.requests.lock().unwrap();
+        assert_eq!(
+            requests.len(),
+            2,
+            "must not retry an expired/truncated image"
+        );
+        assert!(
+            requests
+                .iter()
+                .all(|request| request.path.contains("expected_revision=3&"))
+        );
+        assert_eq!(
+            client.json.requests.lock().unwrap().len(),
+            1,
+            "must not fetch a new revision"
+        );
+    }
+}
+#[tokio::test]
+async fn oversized_images_and_unknown_arguments_fail_before_binary_transfer() {
+    let mut metadata = entry("2", "1");
+    metadata.size = Some(tools::view_image::MAX_IMAGE_BYTES as u32 + 1);
+    let client = chunk_client(metadata.clone(), vec![]);
+    let feature = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::new()));
+    assert!(matches!(
+        execute(&feature, "DriveViewImage", json!({"entry":metadata.entry})).await,
+        Err(DriveError::Limit)
+    ));
+    assert!(client.requests.lock().unwrap().is_empty());
+    assert!(
+        execute(
+            &feature,
+            "DriveRead",
+            json!({"entry":metadata.entry,"expected_revision":"1"})
+        )
+        .await
+        .is_err()
+    );
+    assert_eq!(client.json.requests.lock().unwrap().len(), 1);
+}

crates/worker/src/feature/builtin/drive/wip.rs

@@ -0,0 +1,363 @@
+//! Native non-indexable Drive entries. Publication resolves just one ID through
+//! current Backend authorization; the Client owns identity/revision validators.
+use super::{
+    DriveFeature, SPECS, Spec,
+    backend::{DriveError, decimal},
+    digest, entry_path,
+};
+use crate::wip::{
+    WipCallContext, WipMountError, WipMountRegistry, WipOperationError, WipOperationHandler,
+    WipOperationOutput, WipProjection, WipProjectionKind, WipPublication, WipSubtreeMount,
+    WipSubtreeProvider, contextual_reference, json_to_wip, wip_to_json,
+};
+use async_trait::async_trait;
+use serde_json::json;
+use server_api::{DriveEntry, DriveEntryKind, DriveEntryRef};
+use std::{collections::BTreeMap, sync::Arc};
+use wip_protocol::{
+    Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
+    ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
+};
+
+const ROOT: &str = "/drive";
+const INTERFACE: &str = "yoi.drive/entry/v1";
+
+pub fn mount_drive_wip(
+    registry: &mut WipMountRegistry,
+    feature: &DriveFeature,
+) -> Result<(), WipMountError> {
+    registry.allocate_namespace("drive", "drive")?;
+    // Registration is only a namespace entrance. The subtree provides the
+    // current target/scope/descriptor/handler in a single self-scoped snapshot.
+    let names = root_specs();
+    let descriptor = descriptor(&names, true);
+    registry.mount(WipProjection {
+        route: ROOT.into(),
+        capability: "drive:content".into(),
+        kind: WipProjectionKind::Native,
+        object: Object {
+            name: "drive".into(),
+            description: Some(
+                "Authorized Workspace Drive; discover entries via bounded list/search, not tree"
+                    .into(),
+            ),
+            interfaces: vec![contextual_reference(INTERFACE, ROOT)],
+            r#ref: None,
+            validator: None,
+        },
+        interface: contextual_reference(INTERFACE, ROOT),
+        interface_validator: Some(digest(format!("{descriptor:?}").as_bytes()).into_bytes()),
+        descriptor,
+        handler: Arc::new(Unresolved),
+    })?;
+    registry.mount_subtree(WipSubtreeMount {
+        root: ROOT.into(),
+        provider: Arc::new(DriveProvider(feature.clone())),
+    })?;
+    registry.replace_compatibility_tools(ROOT, SPECS.iter().map(|spec| spec.name))?;
+    Ok(())
+}
+struct Unresolved;
+#[async_trait]
+impl WipOperationHandler for Unresolved {
+    async fn call(
+        &self,
+        _: &str,
+        _: &BTreeMap<String, Value>,
+        _: WipCallContext,
+    ) -> Result<WipOperationOutput, WipOperationError> {
+        Err(map_error(DriveError::Denied))
+    }
+}
+pub(super) struct DriveProvider(pub(super) DriveFeature);
+#[async_trait]
+impl WipSubtreeProvider for DriveProvider {
+    // Arbitrary tree depth is safe: there are deliberately no indexable entries.
+    fn max_depth(&self) -> u32 {
+        32
+    }
+    fn max_nodes(&self) -> usize {
+        1
+    }
+    async fn publication(&self, path: &str) -> Result<Option<WipPublication>, ProtocolError> {
+        let result = if path == ROOT {
+            self.0.backend.root().await
+        } else {
+            let Some(entry) = parse_entry(&self.0, path).map_err(protocol)? else {
+                return Ok(None);
+            };
+            self.0.backend.metadata(&entry).await
+        };
+        let entry = match result {
+            Ok(entry) => entry,
+            Err(DriveError::Denied | DriveError::NotFound) => return Ok(None),
+            Err(error) => return Err(protocol(error)),
+        };
+        let root = path == ROOT;
+        let specs = if root {
+            root_specs()
+        } else {
+            entry_specs(&entry)
+        };
+        let descriptor = descriptor(&specs, root);
+        let reference = contextual_reference(INTERFACE, path);
+        let validator = entry.revision.as_bytes().to_vec();
+        let object = Object {
+            name: if root {
+                "drive".into()
+            } else {
+                entry.entry.node_id.clone()
+            },
+            description: Some(format!(
+                "Workspace Drive {:?} {:?}; stable ID {}, bodies only through bounded operations",
+                entry.kind, entry.name, entry.entry.node_id
+            )),
+            interfaces: vec![reference.clone()],
+            r#ref: Some(format!(
+                "drive:{}:{}",
+                entry.entry.workspace_id, entry.entry.node_id
+            )),
+            validator: Some(validator),
+        };
+        let projection = WipProjection {
+            route: path.into(),
+            capability: "drive:content".into(),
+            kind: WipProjectionKind::Native,
+            object,
+            interface: reference,
+            interface_validator: Some(digest(format!("{descriptor:?}").as_bytes()).into_bytes()),
+            descriptor,
+            handler: Arc::new(DriveHandler {
+                feature: self.0.clone(),
+                entry,
+                specs,
+                root,
+            }),
+        };
+        WipPublication::self_scoped(projection).map(Some)
+    }
+    async fn children(&self, _: &str) -> Result<Vec<String>, ProtocolError> {
+        Ok(Vec::new())
+    }
+}
+fn root_specs() -> Vec<Spec> {
+    SPECS
+        .iter()
+        .copied()
+        .filter(|spec| {
+            matches!(
+                spec.operation,
+                "root"
+                    | "metadata"
+                    | "list"
+                    | "search"
+                    | "create_folder"
+                    | "create_text"
+                    | "save_workdir"
+                    | "request_status"
+            )
+        })
+        .collect()
+}
+fn entry_specs(entry: &DriveEntry) -> Vec<Spec> {
+    SPECS
+        .iter()
+        .copied()
+        .filter(|spec| match spec.operation {
+            "metadata" => true,
+            "relocate" | "delete" => entry.parent.is_some(),
+            "list" | "create_folder" | "create_text" | "save_workdir" => {
+                entry.kind == DriveEntryKind::Folder
+            }
+            "read" | "write" | "edit" | "view_image" => entry.kind == DriveEntryKind::File,
+            _ => false,
+        })
+        .collect()
+}
+fn parameter(name: &str, required: bool, r#type: TypeExpr) -> ParameterDeclaration {
+    ParameterDeclaration {
+        name: name.into(),
+        required,
+        documentation: None,
+        r#type,
+    }
+}
+fn descriptor(specs: &[Spec], _root: bool) -> InterfaceDescriptor {
+    InterfaceDescriptor {format:INTERFACE_FORMAT_V1.into(),documentation:Some(Documentation {summary:"Backend-authorized Workspace Drive operations".into(),details:Some("Entry IDs are Workspace-bound. The Host/Client manage revisions; never supply validators or retry conflicts/outcome-unknown with a new revision. Read-only grants reject mutations at Backend. List/search return paths for direct Inspect, never tree inventories. Image bytes attach to durable ToolOutput; URLs are not image viewing. Workdir source read and Drive destination write are independent.".into())}),types:Vec::new(),operations:specs.iter().map(|spec|OperationDeclaration {name:spec.operation.into(),documentation:Some(Documentation {summary:spec.description.into(),details:None}),parameters:match spec.operation {
+        "list"=>vec![parameter("limit",false,TypeExpr::Integer),parameter("after",false,TypeExpr::String)],
+        "search"=>vec![parameter("query",true,TypeExpr::String),parameter("include_text",false,TypeExpr::Boolean),parameter("limit",false,TypeExpr::Integer),parameter("after",false,TypeExpr::String)],
+        "read"=>vec![parameter("max_bytes",false,TypeExpr::Integer)],
+        "write"|"edit"=>crate::file_operation::parameters(spec.operation,false).expect("shared text arguments"),
+        "create_text"=>vec![parameter("name",true,TypeExpr::String),parameter("content",true,TypeExpr::String),parameter("content_type",false,TypeExpr::String)],
+        "create_folder"=>vec![parameter("name",true,TypeExpr::String)],
+        "relocate"=>vec![parameter("parent",true,TypeExpr::Entry),parameter("name",true,TypeExpr::String)],
+        "save_workdir"=>vec![parameter("target_workdir",true,TypeExpr::String),parameter("path",true,TypeExpr::String),parameter("name",true,TypeExpr::String),parameter("content_type",false,TypeExpr::String)],
+        "request_status"=>vec![parameter("request_id",true,TypeExpr::String)],
+        _=>vec![],
+    },returns:ReturnDeclaration {documentation:None,r#type:TypeExpr::Json}}).collect()}
+}
+struct DriveHandler {
+    feature: DriveFeature,
+    entry: DriveEntry,
+    specs: Vec<Spec>,
+    root: bool,
+}
+#[async_trait]
+impl WipOperationHandler for DriveHandler {
+    async fn call(
+        &self,
+        operation: &str,
+        arguments: &BTreeMap<String, Value>,
+        context: WipCallContext,
+    ) -> Result<WipOperationOutput, WipOperationError> {
+        let spec = self
+            .specs
+            .iter()
+            .find(|spec| spec.operation == operation)
+            .ok_or_else(|| {
+                failure(
+                    ProtocolErrorCode::OperationNotFound,
+                    "Drive operation is not published",
+                )
+            })?;
+        let declaration = descriptor(&self.specs, self.root)
+            .operations
+            .into_iter()
+            .find(|op| op.name == operation)
+            .unwrap();
+        if arguments
+            .keys()
+            .any(|key| !declaration.parameters.iter().any(|p| p.name == *key))
+        {
+            return Err(map_error(DriveError::Invalid(
+                "unrecognized arguments; revision metadata is Client-managed".into(),
+            )));
+        }
+        let mut input = wip_to_json(&Value::Record(arguments.clone()))
+            .map_err(|_| map_error(DriveError::Invalid("invalid WIP arguments".into())))?;
+        let fields = input
+            .as_object_mut()
+            .ok_or_else(|| map_error(DriveError::Invalid("named arguments required".into())))?;
+        match operation {
+            "metadata" | "read" | "write" | "edit" | "delete" | "view_image" => {
+                fields.insert("entry".into(), json!(self.entry.entry));
+            }
+            "list" | "create_folder" | "create_text" | "save_workdir" => {
+                fields.insert("parent".into(), json!(self.entry.entry));
+            }
+            "relocate" => {
+                let path = match arguments.get("parent") {
+                    Some(Value::String(path)) => path.as_str(),
+                    _ => {
+                        return Err(map_error(DriveError::Invalid(
+                            "parent must be a Drive entry".into(),
+                        )));
+                    }
+                };
+                let parent = if path == ROOT {
+                    self.feature.backend.root().await.map_err(map_error)?.entry
+                } else {
+                    parse_entry(&self.feature, path)
+                        .map_err(map_error)?
+                        .ok_or_else(|| map_error(DriveError::Denied))?
+                };
+                fields.insert("parent".into(), json!(parent));
+                fields.insert("entry".into(), json!(self.entry.entry));
+            }
+            _ => {}
+        }
+        if let Some(permissions) = &self.feature.permissions {
+            if !matches!(
+                crate::permission::permission_action_for(permissions, spec.name, &input),
+                manifest::ToolPermissionAction::Allow
+            ) {
+                return Err(failure(
+                    ProtocolErrorCode::PermissionDenied,
+                    "Drive operation denied by tool permission; approval unavailable in native invocation",
+                ));
+            }
+        }
+        let output = self
+            .feature
+            .execute(
+                spec.name,
+                &input.to_string(),
+                &context.execution,
+                Some(&self.entry),
+            )
+            .await
+            .map_err(map_error)?;
+        let value = json_to_wip(&output.value).map_err(|_| {
+            WipOperationError::OutcomeUnknown(
+                "Completed Drive result unavailable; query request_status before another mutation"
+                    .into(),
+            )
+        })?;
+        // Attachments use the existing ToolOutput side channel and durable Engine
+        // capture, not transient request injection or URL-only success.
+        let result = if matches!(operation, "write" | "edit" | "relocate") {
+            let revision = output.value["entry"]["metadata"]["revision"]
+                .as_str()
+                .ok_or_else(|| {
+                    WipOperationError::OutcomeUnknown("Committed Drive revision unavailable".into())
+                })?;
+            WipOperationOutput::native_with_validator(value, revision.as_bytes().to_vec())
+        } else {
+            WipOperationOutput::native(value)
+        };
+        Ok(result.with_attachments(output.attachments))
+    }
+}
+fn parse_entry(feature: &DriveFeature, path: &str) -> Result<Option<DriveEntryRef>, DriveError> {
+    let workspace = feature.backend.workspace()?;
+    let prefix = format!(
+        "/drive/{}/",
+        crate::feature::builtin::manage_workdir::wip::encode_identity(workspace)
+    );
+    let Some(id) = path.strip_prefix(&prefix) else {
+        return Ok(None);
+    };
+    if !decimal(id) {
+        return Ok(None);
+    }
+    let entry = DriveEntryRef {
+        workspace_id: workspace.into(),
+        node_id: id.into(),
+    };
+    if entry_path(&entry) != path {
+        return Ok(None);
+    }
+    Ok(Some(entry))
+}
+fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError {
+    WipOperationError::Protocol(ProtocolError {
+        code,
+        message: message.into(),
+    })
+}
+fn map_error(error: DriveError) -> WipOperationError {
+    let code = match &error {
+        DriveError::Denied => ProtocolErrorCode::PermissionDenied,
+        DriveError::NotFound => ProtocolErrorCode::NotFound,
+        DriveError::Conflict => ProtocolErrorCode::ValidatorMismatch,
+        DriveError::Invalid(_) => ProtocolErrorCode::InvalidArguments,
+        DriveError::Limit => ProtocolErrorCode::ResourceLimitExceeded,
+        DriveError::Unavailable => ProtocolErrorCode::Internal,
+        DriveError::OutcomeUnknown(_) => {
+            return WipOperationError::OutcomeUnknown(error.to_string());
+        }
+    };
+    WipOperationError::Protocol(ProtocolError {
+        code,
+        message: error.to_string(),
+    })
+}
+fn protocol(error: DriveError) -> ProtocolError {
+    match map_error(error) {
+        WipOperationError::Protocol(error) => error,
+        _ => ProtocolError {
+            code: ProtocolErrorCode::Internal,
+            message: "Drive observation unavailable".into(),
+        },
+    }
+}

crates/worker/src/job.rs

@@ -284,6 +284,7 @@ fn validate_profile_with_grant(
         manage_workdir,
         workdir_catalog,
         workspace_config,
+        drive,
         ticket,
         merge_request,
         orchestration,
@@ -315,6 +316,7 @@ fn validate_profile_with_grant(
         ("feature.objective", objective.enabled),
         ("feature.manage_workdir", manage_workdir.enabled),
         ("feature.workspace_config", workspace_config.enabled),
+        ("feature.drive", drive.enabled),
         (
             "feature.ticket",
             ticket.enabled || ticket.authoring || ticket.thread || ticket.intake || ticket.workflow,

crates/worker/src/spawn/tool.rs

@@ -28,6 +28,7 @@ use workdir::{
 
 use crate::PromptCatalogSource;
 use crate::controller::register_worker_tools;
+use crate::feature::builtin::drive::delegation::restrict_internal_child_drive;
 use crate::internal_worker::{
     EphemeralSessionStore, InternalWorkerSessionStatus, InternalWorkerVisibility,
     prepare_internal_worker_session,
@@ -327,6 +328,9 @@ pub struct SubWorkerSpawnTool {
     /// Compact selector list shared by tool description and diagnostics.
     available_profiles: AvailableProfiles,
     internal_client_override: Option<Box<dyn agen::llm_client::LlmClient>>,
+    #[cfg(test)]
+    installed_child_workspace_client:
+        std::sync::Mutex<Option<Arc<dyn crate::worker::WorkspaceClient>>>,
 }
 
 impl SubWorkerSpawnTool {
@@ -362,6 +366,8 @@ impl SubWorkerSpawnTool {
             prompt_loader,
             available_profiles,
             internal_client_override: None,
+            #[cfg(test)]
+            installed_child_workspace_client: std::sync::Mutex::new(None),
         }
     }
 }
@@ -482,6 +488,9 @@ impl Tool for SubWorkerSpawnTool {
         // 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;
+        // Internal children (including Reviewers) have no Backend Drive identity or approved
+        // child scope. Profiles and the parent's current Drive grant cannot delegate it.
+        child_manifest.feature.drive.enabled = false;
         let reviewer_capability = input.review.as_ref().map(|review| {
             (
                 review.ticket_id.clone(),
@@ -523,6 +532,12 @@ impl Tool for SubWorkerSpawnTool {
             } else {
                 self.workspace_context.clone()
             };
+        let child_workspace_context = restrict_internal_child_drive(child_workspace_context);
+        #[cfg(test)]
+        {
+            *self.installed_child_workspace_client.lock().unwrap() =
+                Some(child_workspace_context.client_handle());
+        }
         let store = EphemeralSessionStore::default();
         let filesystem_authority = WorkerFilesystemAuthority::None;
         let mut child = Worker::<Box<dyn agen::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
@@ -1576,6 +1591,181 @@ enabled = false
         assert!(spawner_scope.snapshot().is_writable(&workspace_root));
     }
 
+    #[derive(Debug, Default)]
+    struct DriveAllowingParentClient {
+        requests: Mutex<Vec<WorkspaceRequest>>,
+    }
+
+    impl WorkspaceClient for DriveAllowingParentClient {
+        fn workspace_id(&self) -> Option<&str> {
+            Some("workspace-test")
+        }
+        fn kind(&self) -> &str {
+            "drive-allowing-parent"
+        }
+        fn is_available(&self) -> bool {
+            true
+        }
+        fn execute(
+            &self,
+            request: WorkspaceRequest,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.requests.lock().unwrap().push(request);
+            Ok(WorkspaceResponse {
+                status: 200,
+                body: "{}".into(),
+            })
+        }
+    }
+
+    #[tokio::test]
+    async fn internal_spawn_profiles_never_inherit_parent_drive_tools_or_client_authority() {
+        for (profile, review) in [
+            ("inherit", false),
+            ("default", false),
+            ("builtin:reviewer", true),
+        ] {
+            let runtime = TempDir::new().unwrap();
+            let workspace_root = runtime.path().join("project");
+            let bash_output_dir = runtime.path().join("bash-output");
+            let drive_profile =
+                format!("{INTERNAL_REVIEWER_PROFILE}\n[feature.drive]\nenabled = true\n");
+            let available_profiles = write_project_profile_registry(
+                &workspace_root,
+                Some("reviewer"),
+                &[("reviewer", "reviewer.toml", &drive_profile)],
+            );
+            let mut manifest = parent_manifest(&workspace_root, None);
+            manifest.feature.drive.enabled = true;
+            manifest
+                .scope
+                .allow
+                .push(abs_rule(&bash_output_dir, Permission::Read));
+            let spawner_scope = SharedScope::new(Scope::from_config(&manifest.scope).unwrap());
+            let registry =
+                SpawnedWorkerRegistry::new_internal("parent".into(), spawner_scope.clone());
+            let parent_client = Arc::new(DriveAllowingParentClient::default());
+            let drive_request = WorkspaceRequest::get("/api/w/workspace-test/drive/root");
+            assert!(
+                parent_client
+                    .execute(drive_request.clone())
+                    .unwrap()
+                    .is_success()
+            );
+            let workdir_tool_broker = workdir::WorkdirToolBrokerRouter::from_single(
+                workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
+                Arc::new(workdir::LocalWorkdirSession::materialized_bound(
+                    workdir::Workdir::new("test-workdir"),
+                    workspace_root.clone(),
+                    workspace_root.clone(),
+                    spawner_scope.clone(),
+                    workdir::WorkdirSessionCapabilities::ALL,
+                )),
+            )
+            .unwrap();
+            let calls = Arc::new(AtomicUsize::new(0));
+            let tool = SubWorkerSpawnTool::new(
+                "parent".into(),
+                crate::worker::WorkerWorkspaceContext::with_client(
+                    Some(WorkspaceId::new("workspace-test").unwrap()),
+                    parent_client.clone(),
+                ),
+                ParentNotificationTarget::Durable(Arc::new(|_| {})),
+                runtime.path().to_path_buf(),
+                bash_output_dir,
+                workspace_root.clone(),
+                Some(workdir_tool_broker),
+                registry.clone(),
+                manifest.clone(),
+                PromptCatalogSource::builtins_only(),
+                available_profiles,
+            )
+            .with_internal_client(Box::new(ScriptedInternalClient {
+                calls: calls.clone(),
+                parent_scope: spawner_scope,
+                delegated_path: workspace_root,
+                observed_parent_write_revoked: Arc::new(AtomicBool::new(false)),
+                observed_instruction_override: Arc::new(AtomicBool::new(false)),
+                fail_requests: Arc::new(AtomicBool::new(false)),
+            }));
+            let mut input = serde_json::json!({
+                "name": "child", "profile": profile, "instruction": "role.reviewer",
+                "task": "review", "scope": [{"target": ".", "permission": "write"}],
+                "command": true,
+            });
+            if review {
+                input["review"] = serde_json::json!({"ticket_id": "T1", "merge_request_id": "MR1"});
+            }
+            tool.execute(
+                &input.to_string(),
+                agen::tool::ToolExecutionContext::direct(),
+            )
+            .await
+            .unwrap_or_else(|error| panic!("profile={profile}: {error:?}"));
+            let record = registry.get_internal("child").unwrap();
+            assert_eq!(
+                record.session.wait_until_idle().await,
+                InternalWorkerSessionStatus::Idle
+            );
+            assert_eq!(calls.load(Ordering::SeqCst), 1);
+            assert!(
+                manifest.feature.drive.enabled,
+                "parent must keep Drive enabled"
+            );
+            assert!(
+                !record
+                    .installed_tools
+                    .iter()
+                    .any(|name| name.starts_with("Drive")),
+                "profile={profile}, installed tools={:?}",
+                record.installed_tools
+            );
+            let child_client = tool
+                .installed_child_workspace_client
+                .lock()
+                .unwrap()
+                .clone()
+                .unwrap();
+            let before = parent_client.requests.lock().unwrap().len();
+            assert!(
+                matches!(child_client.execute(drive_request), Err(WorkspaceClientError::Unavailable(message)) if message.contains("Drive access denied")),
+                "profile={profile}"
+            );
+            assert_eq!(parent_client.requests.lock().unwrap().len(), before);
+            if review {
+                assert_eq!(
+                    child_client.reviewer_context(),
+                    Some(&ReviewerContext {
+                        ticket_id: "T1".into(),
+                        merge_request_id: "MR1".into(),
+                    })
+                );
+                child_client
+                    .execute(WorkspaceRequest::json(
+                        WorkspaceRequestMethod::Post,
+                        "/api/w/workspace-test/merge-requests/MR1/reviews",
+                        "{\"verdict\":\"approve\"}",
+                    ))
+                    .unwrap();
+                let requests = parent_client.requests.lock().unwrap();
+                let review_body: serde_json::Value =
+                    serde_json::from_str(requests.last().unwrap().body.as_ref().unwrap()).unwrap();
+                assert!(
+                    review_body["capability_token"]
+                        .as_str()
+                        .is_some_and(|token| !token.is_empty())
+                );
+                assert!(
+                    requests
+                        .iter()
+                        .any(|request| request.path.ends_with("/review-capabilities"))
+                );
+            }
+            record.session.stop().await.unwrap();
+            registry.remove_internal("child").await.unwrap();
+        }
+    }
+
     #[tokio::test]
     async fn remote_subworker_spawn_keeps_worker_host_output_path_out_of_workdir_scope() {
         let runtime = TempDir::new().unwrap();

crates/workspace-server/Cargo.toml

@@ -56,6 +56,7 @@ zeroize.workspace = true
 webauthn-rs = { workspace = true }
 
 [dev-dependencies]
+wip-protocol.workspace = true
 agen.workspace = true
 tempfile.workspace = true
 tools.workspace = true

crates/workspace-server/src/server.rs

@@ -32729,6 +32729,7 @@ mod tests {
     mod ticket_worker_claim_tests;
     mod ticket_worker_launch_tests;
     mod value_profiles_tests;
+    mod worker_drive_tests;
     mod worker_operations_tests;
     include!("server_workspace_config_tests.rs");
     use super::*;

crates/workspace-server/src/server_workspace_config_wip_tests.rs

@@ -139,26 +139,33 @@ impl worker::WorkspaceClient for ConfigRouterClient {
         Ok(response)
     }
 }
-fn config_interface(path: &str) -> String {
+fn contextual_reference(base: &str, path: &str) -> wip_protocol::InterfaceReference {
-    // Use the published contextual interface identity, not a model-managed validator.
+    wip_protocol::InterfaceReference {
-    let hex = path
+        scope: path.into(),
-        .bytes()
+        name: base.into(),
-        .map(|byte| format!("{byte:02x}"))
-        .collect::<String>();
-    format!("yoi.workspace-config/node/v1/@/{hex}")
     }
-async fn config_discover(runtime: &worker::wip::WipRuntime, path: &str) {
+}
-    let observed = runtime
+fn config_interface(path: &str) -> wip_protocol::InterfaceReference {
-        .discover(path.into(), 0, true)
+    // Use the published contextual interface identity, not a model-managed validator.
-        .await
+    contextual_reference("yoi.workspace-config/node/v1", path)
-        .unwrap()
+}
-        .content
+async fn inspect_interface(runtime: &worker::wip::WipRuntime, path: &str, base: &str) -> Value {
-        .unwrap();
+    let output = runtime.inspect(path.into(), true).await.unwrap();
+    let observed: Value = serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
     assert!(
-        observed.contains(&config_interface(path)),
+        observed["interfaces"]
-        "interface must come from live discovery: {observed}"
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|interface| {
+                interface["reference"] == json!({"scope": path, "name": base})
+            }),
+        "interface must come from live inspection: {observed}"
     );
-    runtime.inspect(config_interface(path), true).await.unwrap();
+    observed
+}
+async fn config_inspect(runtime: &worker::wip::WipRuntime, path: &str) -> Value {
+    inspect_interface(runtime, path, "yoi.workspace-config/node/v1").await
 }
 async fn config_call(
     runtime: &worker::wip::WipRuntime,
@@ -167,7 +174,7 @@ async fn config_call(
     args: Value,
 ) -> std::result::Result<agen::tool::ToolOutput, agen::tool::ToolError> {
     runtime
-        .call(
+        .invoke(
             path.into(),
             config_interface(path),
             operation.into(),
@@ -183,23 +190,11 @@ async fn catalog_call(
     base: &str,
     operation: &str,
 ) -> Value {
-    let hex = path
+    inspect_interface(runtime, path, base).await;
-        .bytes()
-        .map(|byte| format!("{byte:02x}"))
-        .collect::<String>();
-    let interface = format!("{base}/@/{hex}");
-    let discovered = runtime
-        .discover(path.into(), 0, true)
-        .await
-        .unwrap()
-        .content
-        .unwrap();
-    assert!(discovered.contains(&interface), "{discovered}");
-    runtime.inspect(interface.clone(), true).await.unwrap();
     let output = runtime
-        .call(
+        .invoke(
             path.into(),
-            interface,
+            contextual_reference(base, path),
             operation.into(),
             json!({}),
             agen::tool::ToolExecutionContext::direct(),
@@ -378,7 +373,7 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
             .connection_id,
         attached.connection_id
     );
-    config_discover(&wip, CONTENT_ROOT).await;
+    config_inspect(&wip, CONTENT_ROOT).await;
     config_call(
         &wip,
         CONTENT_ROOT,
@@ -388,15 +383,10 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
     .await
     .unwrap();
     let path = "/workspace-config/notes/wip.md";
-    config_discover(&wip, path).await;
+    config_inspect(&wip, path).await;
-    assert!(
+    let read = config_call(&wip, path, "read", json!({})).await.unwrap();
-        config_call(&wip, path, "read", json!({}))
+    let read: Value = serde_json::from_str(read.content.as_deref().unwrap()).unwrap();
-            .await
+    assert_eq!(read["content"], "original note");
-            .unwrap()
-            .content
-            .unwrap()
-            .contains("original note")
-    );
     config_call(
         &wip,
         path,
@@ -422,7 +412,7 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
     config_call(&wip, path, "write", json!({"content":"replacement note"}))
         .await
         .unwrap();
-    config_discover(&wip, CONTENT_ROOT).await;
+    config_inspect(&wip, CONTENT_ROOT).await;
     config_call(&wip,CONTENT_ROOT,"apply_changes",json!({"changes":[
         {"kind":"create","path":"fragments/prompts.dcdl","content_type":"decodal","content":"{ common = { language = \"CONFIG_WIP_E2E\"; }; }"},
         {"kind":"update","path":"main.dcdl","content":"{ prompts = import \"./fragments/prompts.dcdl\"; } as WorkspaceConfigSchema"}
@@ -436,7 +426,7 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
         .unwrap();
     let prompts = crate::prompt_settings::project_prompts_from_workspace_config(&state).unwrap();
     assert_eq!(prompts.templates["common.language"], "CONFIG_WIP_E2E");
-    config_discover(&wip, path).await;
+    config_inspect(&wip, path).await;
     config_call(&wip, path, "delete", json!({})).await.unwrap();
     assert!(
         fixture
@@ -449,7 +439,7 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
             .get(&config_source::VirtualPath::parse("notes/wip.md").unwrap())
             .is_none()
     );
-    config_discover(&wip, CONTENT_ROOT).await;
+    config_inspect(&wip, CONTENT_ROOT).await;
     let before = fixture
         .api
         .config_store
@@ -492,7 +482,7 @@ async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_im
         dispatched + 1,
         "a lost response cannot trigger automatic retry"
     );
-    config_discover(&wip, CONTENT_ROOT).await;
+    config_inspect(&wip, CONTENT_ROOT).await;
     let readonly = workspace_config::attach(
         &fixture.api,
         &fixture.worker,
@@ -633,26 +623,26 @@ async fn assert_config_production_http_adapter(
     )
     .unwrap();
     assert_config_catalog(&wip, &attached).await;
-    config_discover(&wip, "/workspace-config/main.dcdl").await;
+    let main_interface = config_inspect(&wip, "/workspace-config/main.dcdl").await;
-    let main_interface = wip
-        .inspect(config_interface("/workspace-config/main.dcdl"), true)
-        .await
-        .unwrap()
-        .content
-        .unwrap();
     assert!(
-        !main_interface.contains("\"delete\""),
+        main_interface["interfaces"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .all(|interface| !interface["signature"].as_str().unwrap().contains("delete(")),
         "required entrypoint cannot advertise delete: {main_interface}"
     );
-    assert!(
+    let main = config_call(&wip, "/workspace-config/main.dcdl", "read", json!({}))
-        config_call(&wip, "/workspace-config/main.dcdl", "read", json!({}))
         .await
-            .unwrap()
+        .unwrap();
-            .content
+    let main: Value = serde_json::from_str(main.content.as_deref().unwrap()).unwrap();
+    assert!(
+        main["content"]
+            .as_str()
             .unwrap()
             .contains("WorkspaceConfigSchema")
     );
-    config_discover(&wip, "/workspace-config").await;
+    config_inspect(&wip, "/workspace-config").await;
     config_call(
         &wip,
         "/workspace-config",

docs/development/worker-drive.md

@@ -0,0 +1,167 @@
+# Worker Drive tools and WIP
+
+Drive is Workspace-owned DB hierarchy and local blob storage exposed through the
+authorized Server API. It is **not** a Workdir, mount, Git checkout, or a Session
+attachment store. Worker code never accesses the Server's physical blob root.
+
+## Enablement and authority
+
+Set `feature.drive = { enabled = true; };` in a DCDL Profile (or
+`[feature.drive] enabled = true` in TOML). Only activation is accepted; Profile
+fields cannot specify Workspace, grant, actor, access, or host paths. Default,
+Coder, Companion, Intake, Orchestrator and Ticket Worker builtin Profiles select
+the feature. Standalone, Reviewer and Job defaults do not. Host must inject an
+available Workspace-bound `WorkspaceClient`; standalone activation is inert.
+
+An authorized Workspace owner must independently grant the **registered Runtime
+Worker** whole-Drive `read_only` or `read_write` access through the T-722 grant API.
+Every read, metadata resolution, mutation, chunk and receipt query passes current
+Backend authorization. An enabled feature, Subject, Ticket assignment, path, ID,
+or observed validator is not permission. A read-only grant is never upgraded.
+The current API has no read-only capability descriptor endpoint: an authorized
+WIP inspect describes the operations for the entry kind, while Backend rejects
+writes without a write grant. Tool permission configuration also applies to
+native Invoke, including fail-closed `ask` where approval cannot be obtained.
+
+Internal SubWorkers (including Reviewer children) have no separate Backend Drive
+identity/grant contract today. Spawn suppresses Drive activation **after every
+selector resolution**, including `inherit` and explicit builtin Profiles, and
+wraps the Workspace client to deny Drive JSON/binary operations. They cannot use
+the parent's shared client to inherit Drive read or write. Perform such work in
+the authorized parent, or a separately registered Runtime Worker with its own
+current Backend grant. Any future internal delegation must introduce an explicit
+parent-bounded scope and Backend reauthorization, not remove this fence based on
+Profile activation. The wrapper preserves non-Drive Workspace operations and
+Reviewer attestation.
+
+## Normal Tools
+
+`DriveRoot {}` discovers the authorized root and returns `{path, metadata}`.
+Public entry inputs are **Workspace-bound** references:
+
+```json
+{"workspace_id":"workspace-id","node_id":"9007199254740993"}
+```
+
+Node IDs and revisions remain canonical positive decimal **strings**, including
+values above JavaScript's safe-integer range. Renaming/moving preserves IDs and
+authenticated latest URLs. Deleting/recreating the same name allocates a new ID.
+
+| Tool | Input/use |
+| --- | --- |
+| `DriveMetadata` | `entry`: one live entry and its observed revision |
+| `DriveList` | optional `parent` (default root), `limit` 1–200, opaque `after` |
+| `DriveSearch` | `query`, optional `include_text`, `limit` 1–128, opaque `after` |
+| `DriveRead` | `entry`, optional `max_bytes` 1–65,536; reports `truncated` |
+| `DriveCreateFolder` | `parent`, logical `name` |
+| `DriveCreateText` | `parent`, `name`, `content` <=64 KiB, optional `content_type` (Markdown by default) |
+| `DriveWrite` | observed `entry`, complete replacement `content` <=64 KiB |
+| `DriveEdit` | observed `entry`, `old_string`, `new_string`, optional `replace_all` |
+| `DriveRelocate` | observed `entry`, destination `parent`, new/current `name` |
+| `DriveDelete` | observed `entry`; folder deletion follows Backend rules |
+| `DriveViewImage` | `entry`; bounded revision-fixed image bytes, not a URL |
+| `DriveSaveWorkdir` | explicit `target_workdir` alias, relative `path`, destination `parent`, `name`, optional `content_type` (octet-stream by default) |
+| `DriveRequestStatus` | `request_id` from an uncertain mutation |
+
+List/search are paged discovery only, not implicit mutation observations. Read or
+metadata-inspect an entry before replacing, editing, moving or deleting it. The
+Tools adapter retains at most 256 observations, scoped to its injected Workspace.
+Eviction requires a new explicit observation; it never substitutes another entry.
+A successful mutation removes the prior Tools observation. This is a bounded
+precondition cache, not a grant cache, second hierarchy DB, or durable history of
+all Drive versions. Restored workers observe again before mutating.
+
+`DriveEdit` uses the same pure `fs_operation::text` rules as Workdir and Workspace
+config: nonempty `old_string`, different replacement, exactly one nonoverlapping
+match unless `replace_all=true`, and bounded resulting text. It fetches a full
+<=64 KiB preimage and verifies the **previously observed** revision. It cannot edit
+truncated content. The Backend's atomic CAS still decides the first winner.
+
+## Native WIP 0.2.0
+
+Start with `Inspect {"path":"/drive"}`. `/drive` offers root/metadata, list/search,
+creation, Workdir save and receipt query. Invoke selects the structured Interface
+reference shown by Inspect:
+
+```json
+{
+  "path":"/drive",
+  "interface":{"scope":"/drive","name":"yoi.drive/entry/v1"},
+  "operation":"list",
+  "arguments":{"limit":20}
+}
+```
+
+Direct entry paths returned by discovery are:
+
+```text
+/drive/<hex-encoded UTF-8 Workspace ID>/<canonical decimal node ID>
+```
+
+Use the returned path rather than hand-constructing it. A file exposes metadata,
+read, write/edit, relocate/delete and view_image; a folder exposes metadata,
+list, create_folder/create_text, save_workdir and relocate/delete where valid.
+A native target binds its entry or creation parent: no ID or revision argument is
+needed. Relocate's `parent` is a WIP typed entry path. Each entry is self-scoped;
+Host captures target, scope object, descriptor, validator and handler together.
+Client retains and supplies metadata automatically. Do not copy validators or
+`scope_ref` through model arguments.
+
+Drive has **no indexable entry children**, regardless of the size of the Drive or
+the Client's already-inspected entries. Tree shows the Drive entrance, not a file
+inventory. List/search return bounded metadata and paths; only inspect the
+selected path. No Worker index/embedding or eager body injection is added.
+
+## Conflicts, unknown outcomes and publication
+
+Mutations use a Host-generated request identity, observed node ID and exact
+revision. Conflict never causes an adapter to replace the expected revision with
+latest and retry. Observe explicitly, reconsider the edit and submit a deliberate
+new operation. Native Client stale handling likewise does not replay dispatch.
+
+Timeout, lost response, oversized/invalid mutation response or invalid completion
+may mean **OutcomeUnknown**. The adapter queries the *same* request receipt once;
+if committed it returns that confirmed DB result. Otherwise it preserves unknown
+and the `request_id`. Query `DriveRequestStatus`; an `uncommitted` snapshot is not
+proof that an in-flight request cannot later commit. Do not retry blindly with a
+fresh request ID. Backend grants also protect receipt lookup/replay. Safe typed
+Denied/NotFound/Conflict/Invalid/Limit/StorageUnavailable/OutcomeUnknown errors are
+classified without reflecting provider paths, SQL, transport URLs or secrets.
+
+Saving a Workdir artifact reads through the selected alias's scoped session
+`read_bytes`, checks path/offset/size/hash on every bounded chunk and verifies the
+assembled SHA256. Source read and destination write are independent. Unknown
+aliases, outside-scope paths, traversal/absolute host paths, changed sources,
+source read denial, oversized bodies and destination denial cannot publish a
+success. Upload transfers raw bytes internally with size/SHA256 metadata; the
+model never has to use Bash, a raw host path, or generate base64. Success is the
+Backend's **DB-publication commit**, not merely blob transfer completion.
+
+## Images and historical observations
+
+ViewImage reads <=10 MiB PNG/JPEG/GIF/WebP bytes in <=64 KiB Drive chunks at the
+metadata revision. Every chunk is authorized and revision-fixed. An intervening
+update expires the read; no mixed-generation image is attached. Media is detected
+from bytes using the existing image helper, not the filename or a download URL.
+
+Tools and native Invoke return `ImageAttachment` through `ToolOutput`. Native
+attachments accompany—not replace or bypass validation of—the WIP return value.
+Engine history, durable Session capture/log replay and normal ToolResult pruning
+own those bytes. Later Drive updates/deletion/revocation do not rewrite the text
+or image already recorded in a Session; restore reproduces that observation until
+normal pruning/correction removes details append-only. This is not full Drive
+version history and does not use unrecorded per-request message insertion.
+
+## Validation seams
+
+- `cargo test -p worker --lib feature::builtin::drive`
+- `cargo test -p worker --lib native_invoke`
+- `cargo test -p worker --lib native_attachments_do_not_bypass`
+- `cargo test -p worker-runtime --lib binary`
+- `cargo test -p yoi-workspace-server --lib worker_drive_tests`
+
+Server adapter tests use isolated DB + LocalFileSystem + production typed HTTP
+router and fresh Runtime proofs/current Worker grants. Worker tests cover
+bounded observations, exact CAS inputs, no blind resend, Tools/WIP schemas,
+conditional registration, child denial and native image capture/replay. See
+`AGENTS.md` for root check, changed-crate tests, format and diff requirements.

resources/config-schema/profile-authoring.dcdl

@@ -40,7 +40,7 @@ let
             task = Flag; web = Flag; image = Flag; sub_worker = Flag; flow = Flag;
             worker = { enabled = Bool; direct_spawn = Bool; };
             workspace_worker_discovery = Flag; objective = Flag; manage_workdir = Flag;
-            workdir_catalog = Flag; workspace_config = Flag; orchestration = Flag;
+            workdir_catalog = Flag; workspace_config = Flag; drive = Flag; orchestration = Flag;
             ticket = { enabled = Bool; authoring = Bool; thread = Bool; intake = Bool; workflow = Bool; };
             merge_request = { show = Bool; open = Bool; review = Bool; readiness_check = Bool; complete = Bool; };
             memory = {

resources/profiles/backend-job.dcdl

@@ -10,6 +10,7 @@ import "./base.dcdl" // {
     };
 
     feature = {
+        drive = { enabled = false; };
         task = { enabled = false; };
         memory = { enabled = false; };
         subjektiv = { enabled = false; extraction = { enabled = false; }; };

resources/profiles/base.dcdl

@@ -25,6 +25,8 @@ compaction = {
 };
 
 feature = {
+    # Activation only; Backend Drive grants remain independent.
+    drive = { enabled = false; };
     task = { enabled = true; };
     memory = { enabled = false; staging_tools = false; };
     subjektiv = {

resources/profiles/coder.dcdl

@@ -9,6 +9,7 @@ import "./base.dcdl" // {
     };
 
     feature = {
+        drive = { enabled = true; };
         task = { enabled = true; };
         memory = { enabled = false; };
         subjektiv = { enabled = true; };

resources/profiles/companion.dcdl

@@ -5,6 +5,7 @@ import "./base.dcdl" // {
     model = { ref = "codex-oauth/gpt-6-astra"; };
 
     feature = {
+        drive = { enabled = true; };
         task = { enabled = true; };
         memory = { enabled = false; };
         subjektiv = { enabled = true; };

resources/profiles/default.dcdl

@@ -5,6 +5,8 @@ import "./base.dcdl" // {
     delegation_scope = "workspace_write";
 
     feature = {
+        # Activation only; Backend Drive grants remain independent.
+        drive = { enabled = true; };
         task = { enabled = true; };
         memory = { enabled = false; staging_tools = false; };
         subjektiv = { enabled = false; extraction = { enabled = false; }; };

resources/profiles/intake.dcdl

@@ -5,6 +5,7 @@ import "./base.dcdl" // {
     engine = { instruction = "role.intake"; };
 
     feature = {
+        drive = { enabled = true; };
         task = { enabled = true; };
         memory = { enabled = false; };
         subjektiv = { enabled = true; };

resources/profiles/job.dcdl

@@ -9,6 +9,7 @@ engine = {
 };
 session = { record_event_trace = false; };
 feature = {
+    drive = { enabled = false; };
     task = { enabled = false; };
     memory = { enabled = false; staging_tools = false; };
     subjektiv = { enabled = false; consolidation_tools = false; extraction = { enabled = false; }; };

resources/profiles/orchestrator.dcdl

@@ -5,6 +5,7 @@ import "./base.dcdl" // {
     engine = { instruction = "role.orchestrator"; };
 
     feature = {
+        drive = { enabled = true; };
         task = { enabled = true; };
         memory = { enabled = false; };
         subjektiv = { enabled = true; };

resources/profiles/standalone.dcdl

@@ -2,6 +2,7 @@ import "./default.dcdl" // {
     slug = "standalone";
     description = "Standalone Yoi coding profile.";
     model = { ref = "codex-oauth/gpt-6.1-sol"; };
+    feature = { drive = { enabled = false; }; };
     scope = {
         intent = "workspace_write";
         symlink_policy = "logical";

resources/prompts/catalog.dcdl

@@ -6,6 +6,7 @@ let
 defaultDocument = import "./default.md";
 commonLanguage = import "./common/language.md";
 commonGit = import "./common/git.md";
+commonDrive = import "./common/drive.md";
 commonMergeRequest = import "./common/merge-request.md";
 commonTickets = import "./common/tickets.md";
 commonToolUsage = import "./common/tool-usage.md";
@@ -44,6 +45,7 @@ in
     default_prompt = defaultDocument.content;
     common = {
         git = commonGit.content;
+        drive = commonDrive.content;
         language = commonLanguage.content;
         merge_request = commonMergeRequest.content;
         tickets = commonTickets.content;

resources/prompts/common/drive.md

@@ -0,0 +1,7 @@
+## Workspace Drive
+
+Drive is Backend-owned workspace storage, not a Workdir or a host filesystem path. Feature/Profile activation is not a grant; every operation is checked by the Backend. Use the exposed Drive tools or the native WIP `/drive` entry with bounded listing and reading. Do not enumerate a Drive tree or infer authority from a visible reference.
+
+Observe an entry before replacing or deleting it; the host/client carries its revision precondition. On conflict, reread and reconcile rather than overwriting blindly. On `OutcomeUnknown`, retain the request ID and query request status before another mutation: the operation may have committed. Save a Workdir file into Drive only through a current scoped attachment using its logical relative source path; saving does not attach a Workdir or grant file access.
+
+Internal SubWorkers currently have no explicit Backend Drive grant mechanism. They must not inherit Drive access through a parent's shared WorkspaceClient, even with an enabled or inherited Profile. Ask the authorized parent to perform the operation instead.