chore: merge develop into hare/develop

# Conflicts:
#	crates/api-macros/src/axum.rs

crates/workspace-server/Cargo.toml

@@ -11,12 +11,14 @@ name = "yoi-server"
 path = "src/main.rs"
 
 [dependencies]
+api-macros = { workspace = true, features = ["axum"] }
 async-trait.workspace = true
 axum = { workspace = true, features = ["ws"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 futures.workspace = true
 config-source.workspace = true
 feature-storage.workspace = true
+workspace-drive.workspace = true
 subjektiv.workspace = true
 flow = { path = "../flow" }
 manifest.workspace = true

crates/workspace-server/src/authority.rs

@@ -750,9 +750,9 @@ impl SqliteWorkspaceAuthority {
                 assigned_at: assignment.assigned_at,
             })
             .collect::<Vec<_>>();
-        let current_coder = role_assignments
+        let current_worker = role_assignments
             .iter()
-            .find(|assignment| assignment.role == TicketAssignmentRole::Coder)
+            .find(|assignment| assignment.role == TicketAssignmentRole::Worker)
             .and_then(|assignment| {
                 assignment
                     .principal
@@ -779,9 +779,9 @@ impl SqliteWorkspaceAuthority {
         let has_orchestrator = active_role_assignments
             .iter()
             .any(|assignment| assignment.role == TicketAssignmentRole::Orchestrator);
-        let has_coder = active_role_assignments
+        let has_worker = active_role_assignments
             .iter()
-            .any(|assignment| assignment.role == TicketAssignmentRole::Coder);
+            .any(|assignment| assignment.role == TicketAssignmentRole::Worker);
         let targets = ticket
             .meta
             .targets
@@ -799,7 +799,6 @@ impl SqliteWorkspaceAuthority {
             })
             .collect::<Result<Vec<_>>>()?;
         let write_repository_keys = ticket_write_target_repository_keys(&targets);
-        let has_target = !write_repository_keys.is_empty();
         let has_blockers = !ticket.relations.blockers.is_empty();
         let mut assignment_diagnostics = Vec::new();
         if let Some(legacy_assignee) = ticket
@@ -818,17 +817,20 @@ impl SqliteWorkspaceAuthority {
                 ticket.meta.workflow_state,
                 TicketWorkflowState::Planning | TicketWorkflowState::Ready
             ) && !has_orchestrator
-                && !has_coder,
+                && !has_worker,
             can_unassign_orchestrator: has_orchestrator
                 && matches!(
                     ticket.meta.workflow_state,
                     TicketWorkflowState::Planning | TicketWorkflowState::Ready
                 ),
             can_queue: true,
-            can_start_manual_coder: ticket.meta.workflow_state == TicketWorkflowState::Ready
-                && !has_orchestrator
-                && !has_coder
-                && has_target
+            can_start_manual_worker: matches!(
+                ticket.meta.workflow_state,
+                TicketWorkflowState::Planning
+                    | TicketWorkflowState::Ready
+                    | TicketWorkflowState::Queued
+                    | TicketWorkflowState::InProgress
+            ) && !has_worker
                 && !has_blockers,
             queue_tickets: dependency_check.queue_tickets.clone(),
             blockers: action_blockers,
@@ -960,7 +962,7 @@ impl SqliteWorkspaceAuthority {
             linked_objectives,
             implementation_reports,
             assignments,
-            current_coder,
+            current_worker,
             assignment_diagnostics,
             action_eligibility,
             merge_requests,
@@ -2353,7 +2355,7 @@ fn ticket_query_item(
         evidence: detail.evidence.clone(),
         merge_requests: detail.merge_requests.clone(),
         merge_request: detail.merge_request.clone(),
-        current_coder: detail.current_coder.clone(),
+        current_worker: detail.current_worker.clone(),
     }
 }
 
@@ -3193,6 +3195,143 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("workspace.db");
+        let store = SqliteWorkspaceStore::open(&path).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "w".into(),
+                owner_account_id: "owner-account".into(),
+                display_name: "W".into(),
+                state: "active".into(),
+                created_at: "1".into(),
+                updated_at: "1".into(),
+            })
+            .await
+            .unwrap();
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
+        store
+            .upsert_worker_registry(&crate::store::WorkerRegistryRecord {
+                workspace_id: "w".into(),
+                worker: worker_runtime::identity::RuntimeWorkerRef::new("runtime", "worker"),
+                display_name: "Worker".into(),
+                profile: Some("builtin:default".into()),
+                retention_state: "normal".into(),
+                transcript_ref: None,
+                session_ref: None,
+                summary_ref: None,
+                diagnostics_ref: None,
+                created_at: "1".into(),
+                updated_at: "1".into(),
+            })
+            .unwrap();
+        let backend = SqliteTicketBackend::open_verified(&path, "w").unwrap();
+        let ticket = backend
+            .create(ticket::NewTicket::new("Explicit work"))
+            .unwrap();
+        let authority = SqliteWorkspaceAuthority::new(&path, "w").unwrap();
+        for (state, allowed) in [
+            ("planning", true),
+            ("ready", true),
+            ("queued", true),
+            ("inprogress", true),
+            ("done", false),
+            ("closed", false),
+        ] {
+            conn.execute(
+                "UPDATE typed_tickets SET workflow_state=?1 WHERE ticket_id=?2",
+                rusqlite::params![state, ticket.id],
+            )
+            .unwrap();
+            assert_eq!(
+                authority
+                    .ticket(&ticket.id)
+                    .unwrap()
+                    .action_eligibility
+                    .can_start_manual_worker,
+                allowed,
+                "{state}"
+            );
+        }
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        let record = crate::store::TicketRoleAssignmentRecord {
+            workspace_id: "w".into(),
+            ticket_id: ticket.id.clone(),
+            assignment_id: "assignment".into(),
+            role: TicketAssignmentRole::Worker,
+            principal: TicketAssignmentPrincipal::Worker {
+                runtime_id: "runtime".into(),
+                worker_id: "worker".into(),
+            },
+            assigned_by: "owner-account".into(),
+            assigned_at: "2".into(),
+        };
+        let orchestrator = crate::store::TicketRoleAssignmentRecord {
+            role: TicketAssignmentRole::Orchestrator,
+            assignment_id: "orchestrator".into(),
+            principal: TicketAssignmentPrincipal::WorkspaceAgent {
+                agent_key: "workspace-orchestrator".into(),
+            },
+            ..record.clone()
+        };
+        store
+            .set_current_ticket_role_assignment(
+                &orchestrator,
+                None,
+                "orchestrator-event",
+                "orchestrator-op",
+                false,
+            )
+            .unwrap();
+        assert!(
+            authority
+                .ticket(&ticket.id)
+                .unwrap()
+                .action_eligibility
+                .can_start_manual_worker
+        );
+        store
+            .start_ticket_with_worker_assignment(&record, "event", "operation", None)
+            .unwrap();
+        assert!(
+            !authority
+                .ticket(&ticket.id)
+                .unwrap()
+                .action_eligibility
+                .can_start_manual_worker
+        );
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='closed' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        let detail = authority.ticket(&ticket.id).unwrap();
+        assert!(detail.action_eligibility.can_start_manual_worker);
+        assert_eq!(detail.assignments.len(), 2);
+        let worker = detail
+            .assignments
+            .iter()
+            .find(|a| a.role == "worker")
+            .unwrap();
+        assert_eq!(worker.assignment_id, record.assignment_id);
+        assert_eq!(
+            detail.current_worker.unwrap().assignment_id,
+            record.assignment_id
+        );
+    }
+
     #[tokio::test]
     async fn sqlite_workspace_authority_reads_sqlite_records_without_filesystem_authority() {
         let dir = tempfile::tempdir().unwrap();

crates/workspace-server/src/config.rs

@@ -104,13 +104,10 @@ impl ResolvedWorkspaceBackendConfig {
         host_config: &ServerHostConfigFile,
     ) -> Result<Self> {
         let workspace_root = workspace_root.as_ref();
-        let data_root = ServerConfig::default_workspace_backend_data_root(&workspace.workspace_id);
-        let database_path = ServerConfig::default_server_database_path();
         let (browser_public_url, browser_rp_id) =
             resolve_browser_public_url(&host_config.browser.public_url)?;
         let mut server = ServerConfig::local_dev(workspace_root.to_path_buf(), workspace);
-        server.database_path = database_path.clone();
-        server.embedded_runtime_store_root = data_root.join("embedded-runtime");
+        let database_path = server.database_path.clone();
         server.max_records = DEFAULT_MAX_RECORDS;
         server.remote_runtime_sources = Vec::new();
         server.auth = AuthConfig::Passkey {

crates/workspace-server/src/hosts.rs

@@ -484,7 +484,7 @@ pub(crate) fn worker_spawn_idempotency(
     let Some(assignment) = request.ticket_assignment.as_ref() else {
         return Ok(None);
     };
-    let encoded = serde_json::to_vec(request)
+    let encoded = serde_json::to_vec(&(request, request.resolved_workdir_attachments.as_slice()))
         .map_err(|error| format!("serialize Worker spawn idempotency input: {error}"))?;
     Ok(Some((
         assignment.operation_id.clone(),
@@ -578,7 +578,7 @@ pub enum WorkerSpawnIntent {
 pub enum TicketWorkerRole {
     Intake,
     Orchestrator,
-    Coder,
+    Worker,
     Reviewer,
 }
 
@@ -6457,7 +6457,7 @@ fn worker_spawn_intent_label(intent: &WorkerSpawnIntent) -> &'static str {
         WorkerSpawnIntent::TicketRole { role, .. } => match role {
             TicketWorkerRole::Intake => "ticket_intake",
             TicketWorkerRole::Orchestrator => "ticket_orchestrator",
-            TicketWorkerRole::Coder => "ticket_coder",
+            TicketWorkerRole::Worker => "ticket_worker",
             TicketWorkerRole::Reviewer => "ticket_reviewer",
         },
     }
@@ -6524,6 +6524,27 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use std::thread;
 
+    #[test]
+    fn ticket_worker_spawn_intent_uses_worker_wire_role() {
+        let intent = WorkerSpawnIntent::TicketRole {
+            ticket_id: "ticket".into(),
+            role: TicketWorkerRole::Worker,
+        };
+        let encoded = json!({"kind": "ticket_role", "ticket_id": "ticket", "role": "worker"});
+        assert_eq!(serde_json::to_value(&intent).unwrap(), encoded);
+        assert_eq!(
+            serde_json::from_value::<WorkerSpawnIntent>(encoded).unwrap(),
+            intent
+        );
+        assert!(
+            serde_json::from_value::<WorkerSpawnIntent>(
+                json!({"kind": "ticket_role", "ticket_id": "ticket", "role": "coder"})
+            )
+            .is_err()
+        );
+        assert_eq!(worker_spawn_intent_label(&intent), "ticket_worker");
+    }
+
     #[test]
     fn remote_restore_transport_failure_requires_reconciliation() {
         let provider = RemoteWorkerRuntime::new(
@@ -7541,7 +7562,7 @@ mod tests {
         WorkerSpawnRequest {
             intent: WorkerSpawnIntent::TicketRole {
                 ticket_id: "00001KVZSGT0Q".to_string(),
-                role: TicketWorkerRole::Coder,
+                role: TicketWorkerRole::Worker,
             },
             requested_worker_name: None,
             singleton_key: None,
@@ -7830,7 +7851,7 @@ mod tests {
                 WorkerSpawnRequest {
                     intent: WorkerSpawnIntent::TicketRole {
                         ticket_id: "00001KVZSGT0Q".to_string(),
-                        role: TicketWorkerRole::Coder,
+                        role: TicketWorkerRole::Worker,
                     },
                     requested_worker_name: Some("friendly-name-is-not-authority".to_string()),
                     singleton_key: None,
@@ -7942,7 +7963,7 @@ mod tests {
                 WorkerSpawnRequest {
                     intent: WorkerSpawnIntent::TicketRole {
                         ticket_id: "00001KVZSGT0Q".to_string(),
-                        role: TicketWorkerRole::Coder,
+                        role: TicketWorkerRole::Worker,
                     },
                     requested_worker_name: None,
                     singleton_key: None,

crates/workspace-server/src/latest_schema.sql

@@ -335,8 +335,8 @@ CREATE TABLE ticket_assignment_operations (
             operation_id TEXT NOT NULL,
             action TEXT NOT NULL CHECK (action IN ('assign', 'reassign', 'unassign')),
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
             principal_id TEXT,
@@ -346,8 +346,14 @@ CREATE TABLE ticket_assignment_operations (
             expected_assignment_id TEXT,
             created_at TEXT NOT NULL,
             request_fingerprint TEXT,
+            claim_state TEXT NOT NULL DEFAULT 'committed'
+                CHECK(claim_state IN ('pending', 'failed', 'committed')),
+            failure_reason TEXT,
+            binding_recovery_json TEXT,
             PRIMARY KEY (workspace_id, operation_id),
-            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
+            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+            CHECK(claim_state = 'committed' OR assignment_id IS NULL),
+            CHECK((claim_state = 'failed') = (failure_reason IS NOT NULL))
         );
 CREATE TABLE ticket_assignment_ticket_tombstones (
     workspace_id TEXT NOT NULL,
@@ -367,8 +373,8 @@ CREATE TABLE ticket_assignment_worker_tombstones (
 CREATE TABLE ticket_current_worker_assignments (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             assignment_id TEXT NOT NULL,
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
@@ -393,8 +399,8 @@ CREATE TABLE ticket_current_worker_assignments (
 CREATE TABLE ticket_worker_assignment_events (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             event_id TEXT NOT NULL,
             action TEXT NOT NULL CHECK (action IN ('assigned', 'reassigned', 'unassigned')),
             assignment_id TEXT,
@@ -410,8 +416,8 @@ CREATE TABLE ticket_worker_assignments (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
             assignment_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
             principal_id TEXT,
@@ -1215,7 +1221,7 @@ CREATE INDEX ticket_current_principal_idx
             ON ticket_current_worker_assignments(workspace_id, principal_kind, principal_id, runtime_id, worker_id);
 CREATE UNIQUE INDEX ticket_current_singleton_role_idx
             ON ticket_current_worker_assignments(workspace_id, ticket_id, role)
-            WHERE role IN ('orchestrator', 'coder');
+            WHERE role IN ('orchestrator', 'worker');
 CREATE INDEX typed_ticket_events_workspace_kind_ticket
             ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
 CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id
@@ -1623,3 +1629,28 @@ CREATE TABLE workspace_config_grants (
 );
 CREATE UNIQUE INDEX workspace_config_grants_active_worker
     ON workspace_config_grants(workspace_id, runtime_id, worker_id) WHERE revoked = 0;
+
+CREATE TABLE workspace_drive_grants (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    access TEXT NOT NULL CHECK(access IN ('read_only', 'read_write')),
+    revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0, 1)),
+    created_by TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    revoked_by TEXT,
+    revoked_at TEXT,
+    CHECK((revoked=0 AND revoked_by IS NULL AND revoked_at IS NULL)
+       OR (revoked=1 AND revoked_by IS NOT NULL AND revoked_at IS NOT NULL))
+);
+CREATE UNIQUE INDEX workspace_drive_grants_active_worker
+    ON workspace_drive_grants(workspace_id, runtime_id, worker_id) WHERE revoked=0;
+-- Retain grant audit across Worker deletion, but never carry its authority into
+-- a later lifetime reusing the same Runtime/Worker identity.
+CREATE TRIGGER workspace_drive_grants_worker_deleted BEFORE DELETE ON worker_registry BEGIN
+    UPDATE workspace_drive_grants SET revoked=1, revoked_by='server:worker-removal',
+        revoked_at=strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+    WHERE workspace_id=OLD.workspace_id AND runtime_id=OLD.runtime_id
+      AND worker_id=OLD.worker_id AND revoked=0;
+END;

crates/workspace-server/src/lib.rs

@@ -36,6 +36,7 @@ mod worker_projection;
 pub mod worker_source;
 pub mod workspace_catalog;
 mod workspace_deletion;
+mod workspace_drive_host;
 pub mod workspace_signing_identity;
 mod workspace_subscription;
 
@@ -151,6 +152,8 @@ pub enum Error {
     WorkerSourceIdentity(String),
     #[error("Workspace signing identity error ({code}): {message}")]
     WorkspaceSigningIdentity { code: String, message: String },
+    #[error("Workspace Drive error: {0}")]
+    Drive(#[from] workspace_drive::Error),
     #[error("store error: {0}")]
     Store(String),
 }

crates/workspace-server/src/retention.rs

@@ -69,6 +69,10 @@ pub struct WorkerRemovalPlanRequest {
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum WorkerRemovalBlocker {
     Hold,
+    PendingTicketClaim {
+        operation_id: String,
+        ticket_id: String,
+    },
     UnfinishedWork {
         assignment_id: String,
         ticket_id: String,
@@ -232,6 +236,9 @@ impl SqliteWorkspaceStore {
             if let Some((assignment_id,ticket_id))=tx.query_row("SELECT a.assignment_id,a.ticket_id FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()? {
                 blockers.push(WorkerRemovalBlocker::UnfinishedWork{assignment_id,ticket_id});
             }
+            if let Some((operation_id,ticket_id))=tx.query_row("SELECT operation_id,ticket_id FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign')",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()? {
+                blockers.push(WorkerRemovalBlocker::PendingTicketClaim{operation_id,ticket_id});
+            }
             let fp=fingerprint(req,&worker.updated_at,inv,&policy,&blockers)?;
             let plan_id=stable("wrp",&fp); let operation_id=stable("wro",&fp);
             let archive_id=(policy.session_disposition==SessionDisposition::Archive).then(||stable("wra",&fp));
@@ -276,7 +283,7 @@ impl SqliteWorkspaceStore {
                 mark_stale(&tx,&plan,"hold added")?;tx.commit()?;
                 return Err(stale_error(&plan,"hold added"));
             }
-            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?;
+            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?;
             if assigned{
                 mark_stale(&tx,&plan,"unfinished work added")?;tx.commit()?;
                 return Err(stale_error(&plan,"unfinished work added"));
@@ -462,7 +469,7 @@ impl SqliteWorkspaceStore {
             let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput("Worker missing before commit".into()))?;
             if worker.updated_at!=plan.worker_revision{return Err(StoreError::InvalidInput(format!("stale:{}:Worker revision changed",plan.plan_id)));}
             if worker.retention_state=="pinned" { return Err(StoreError::InvalidInput(format!("stale:{}:hold added",plan.plan_id))); }
-            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?;
+            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?;
             if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:unfinished work added",plan.plan_id))); }
             let now=Utc::now().to_rfc3339();
             if let Some(a)=&result.archive{
@@ -1092,7 +1099,7 @@ fn parse_state(v: &str) -> rusqlite::Result<WorkerRemovalPlanState> {
 mod tests {
     use super::*;
     use crate::store::{
-        ControlPlaneStore, TicketCoderAssignmentRecord, WorkerCatalogChange, WorkerRegistryRecord,
+        ControlPlaneStore, TicketWorkerAssignmentRecord, WorkerCatalogChange, WorkerRegistryRecord,
     };
     use worker_runtime::identity::WorkerId;
     fn worker_id() -> WorkerId {
@@ -1431,7 +1438,7 @@ mod tests {
         s: &SqliteWorkspaceStore,
         assignment_id: &str,
     ) -> crate::Result<()> {
-        let assignment = TicketCoderAssignmentRecord {
+        let assignment = TicketWorkerAssignmentRecord {
             workspace_id: "w".into(),
             ticket_id: "ticket".into(),
             assignment_id: assignment_id.into(),
@@ -1439,7 +1446,7 @@ mod tests {
             assigned_by: "test".into(),
             assigned_at: "t".into(),
         };
-        s.set_current_ticket_coder_assignment(
+        s.set_current_ticket_worker_assignment(
             &assignment,
             None,
             &format!("event-{assignment_id}"),
@@ -1474,6 +1481,38 @@ mod tests {
         store
     }
 
+    #[test]
+    fn pending_ticket_claim_blocks_removal_and_is_rechecked_after_planning() {
+        let s = setup();
+        let old = s.plan_worker_removal(&req(), &inv()).unwrap();
+        s.reserve_ticket_assignment_operation(
+            "w",
+            "claim",
+            "ticket",
+            "r",
+            Some(&worker_id().to_string()),
+            "intent",
+            "t",
+        )
+        .unwrap();
+        let blocked = s.plan_worker_removal(&req(), &inv()).unwrap();
+        assert!(
+            matches!(&blocked.blockers[..], [WorkerRemovalBlocker::PendingTicketClaim{operation_id,ticket_id}] if operation_id=="claim" && ticket_id=="ticket")
+        );
+        assert!(
+            s.begin_worker_removal("w", &old.plan_id, &old.input_fingerprint)
+                .is_err()
+        );
+        s.fail_ticket_assignment_operation("w", "claim", "No Runtime effects dispatched")
+            .unwrap();
+        assert!(
+            s.plan_worker_removal(&req(), &inv())
+                .unwrap()
+                .blockers
+                .is_empty()
+        );
+    }
+
     #[test]
     fn assignment_and_orphan_are_authoritative() {
         let s = setup();
@@ -1758,7 +1797,7 @@ mod tests {
                 let replaced = tx.execute(
                     "UPDATE ticket_current_worker_assignments
                      SET assignment_id='new-assignment', updated_at='new'
-                     WHERE workspace_id='w' AND ticket_id='ticket' AND role='coder'
+                     WHERE workspace_id='w' AND ticket_id='ticket' AND role='worker'
                        AND assignment_id='assignment'
                        AND EXISTS (
                            SELECT 1 FROM ticket_assignment_work_releases
@@ -2088,7 +2127,7 @@ mod tests {
         ).map_err(StoreError::from)).unwrap();
         assert_eq!(revision, "rev1");
 
-        let assignment = TicketCoderAssignmentRecord {
+        let assignment = TicketWorkerAssignmentRecord {
             workspace_id: "w".into(),
             ticket_id: "new-ticket".into(),
             assignment_id: "new-assignment".into(),
@@ -2101,7 +2140,7 @@ mod tests {
         };
         assert!(
             store
-                .set_current_ticket_coder_assignment(
+                .set_current_ticket_worker_assignment(
                     &assignment,
                     None,
                     "event",

crates/workspace-server/src/server/tests/drive_tests.rs

@@ -0,0 +1,217 @@
+use super::*;
+use workspace_drive::{Error as DriveError, Mutation};
+
+#[test]
+fn custom_database_does_not_change_explicit_drive_placement() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut config = test_server_config(temp.path());
+    config.database_path = temp.path().join("custom-authority/authority.sqlite");
+    config.data_root = temp.path().join("explicit-yoi-data");
+    let mut workspace = test_workspace();
+    workspace.workspace_id = "other-internal-id".into();
+    workspace.display_name = "../Not a storage key".into();
+    let scoped = config
+        .for_catalog_workspace(&workspace, Vec::new())
+        .unwrap();
+    assert_eq!(
+        scoped.drive_blob_root().unwrap(),
+        config.data_root.join("drives/other-internal-id/blobs")
+    );
+}
+
+#[tokio::test]
+async fn cached_drive_handle_rejects_mutation_after_deletion_reservation() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut config = test_server_config(temp.path());
+    config.repositories.clear();
+    let store = Arc::new(test_control_store(&config));
+    seed_test_registered_workspace(store.as_ref(), &config)
+        .await
+        .unwrap();
+    let owner = format!("account-{}", config.workspace_id);
+    store
+        .upsert_workspace(&WorkspaceRecord {
+            workspace_id: "retained-workspace".into(),
+            owner_account_id: owner.clone(),
+            display_name: "Retain".into(),
+            state: "active".into(),
+            created_at: "1".into(),
+            updated_at: "1".into(),
+        })
+        .await
+        .unwrap();
+    let api = WorkspaceApi::new(config.clone(), store.clone())
+        .await
+        .unwrap();
+    let cached = api.drive().await.unwrap();
+    let for_root = cached.clone();
+    let parent = tokio::task::spawn_blocking(move || for_root.root().unwrap().id)
+        .await
+        .unwrap();
+    let preflight = store
+        .workspace_deletion_preflight(&owner, &config.workspace_id)
+        .unwrap();
+    store
+        .reserve_workspace_deletion(
+            &owner,
+            &config.workspace_id,
+            &WorkspaceDeletionRequest {
+                operation_id: "delete-cached-drive".into(),
+                expected_revision: preflight.expected_revision,
+                confirmation: preflight.display_name,
+            },
+        )
+        .unwrap();
+    // No WorkspaceApi access or host/domain fence between reservation and the
+    // cached operation: the committed server authority must reject it directly.
+    let result = tokio::task::spawn_blocking(move || {
+        cached.mutate(
+            "cached-write",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "forbidden".into(),
+            },
+        )
+    })
+    .await
+    .unwrap();
+    assert!(matches!(result, Err(DriveError::Fenced)), "{result:?}");
+}
+
+#[tokio::test]
+async fn deleting_workspace_reopens_fenced_drive_and_purges_blobs_before_feature_storage() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut config = test_server_config(temp.path());
+    config.repositories.clear();
+    let store = Arc::new(test_control_store(&config));
+    seed_test_registered_workspace(store.as_ref(), &config)
+        .await
+        .unwrap();
+    let owner = format!("account-{}", config.workspace_id);
+    store
+        .upsert_workspace(&WorkspaceRecord {
+            workspace_id: "retained-workspace".into(),
+            owner_account_id: owner.clone(),
+            display_name: "Retain".into(),
+            state: "active".into(),
+            created_at: "1".into(),
+            updated_at: "1".into(),
+        })
+        .await
+        .unwrap();
+    let root = config.drive_blob_root().unwrap();
+    // Use a separate embedded-runtime namespace for the first API: this test
+    // reconstructs Drive authority in-process rather than spawning/restarting
+    // product processes (runtime observers may retain their namespace).
+    let first_api = WorkspaceApi::new(config.clone(), store.clone())
+        .await
+        .unwrap();
+    // Merely constructing a Workspace API must not create blob storage.
+    assert!(!root.exists());
+    let drive = first_api.drive().await.unwrap();
+    let parent = tokio::task::spawn_blocking(move || {
+        let parent = drive.root().unwrap().id;
+        drive
+            .mutate(
+                "upload",
+                "owner",
+                Mutation::CreateFile {
+                    parent,
+                    name: "file".into(),
+                    content_type: "text/plain".into(),
+                    bytes: b"content".to_vec(),
+                },
+            )
+            .unwrap();
+        parent
+    })
+    .await
+    .unwrap();
+    let preflight = store
+        .workspace_deletion_preflight(&owner, &config.workspace_id)
+        .unwrap();
+    store
+        .reserve_workspace_deletion(
+            &owner,
+            &config.workspace_id,
+            &WorkspaceDeletionRequest {
+                operation_id: "delete-drive-workspace".into(),
+                expected_revision: preflight.expected_revision,
+                confirmation: preflight.display_name,
+            },
+        )
+        .unwrap();
+    // Blocked operations are not automatically resumed on restart, but their
+    // Workspace remains deleting and must still reconstruct a fenced Drive.
+    store
+        .update_workspace_deletion_operation(
+            "delete-drive-workspace",
+            WorkspaceDeletionState::Blocked,
+            &[],
+            &[],
+            None,
+        )
+        .unwrap();
+    // Simulate a crash between the catalog reservation and Drive fencing.
+    let feature_storage = first_api.feature_storage.clone();
+    tokio::task::spawn_blocking(move || feature_storage.shutdown())
+        .await
+        .unwrap()
+        .unwrap();
+    drop(first_api);
+    let recovered = WorkspaceServerApi::new(config.clone(), store.clone());
+    let api = recovered
+        .api_for_workspace(&config.workspace_id)
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(matches!(
+        api.drive().await,
+        Err(Error::WorkspaceConfigConflict(_))
+    ));
+    let reopened = api
+        .drive_host
+        .open(
+            &api.feature_storage,
+            root.clone(),
+            config.database_path.clone(),
+        )
+        .await
+        .unwrap();
+    let mutation = tokio::task::spawn_blocking(move || {
+        reopened.mutate(
+            "restart-write",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "forbidden".into(),
+            },
+        )
+    })
+    .await
+    .unwrap();
+    assert!(matches!(mutation, Err(DriveError::Fenced)), "{mutation:?}");
+    let completed = recovered
+        .execute_workspace_deletion("delete-drive-workspace")
+        .await
+        .unwrap();
+    assert_eq!(completed.state, WorkspaceDeletionState::Succeeded);
+    assert!(!root.exists());
+    assert!(
+        !config
+            .database_path
+            .parent()
+            .unwrap()
+            .join("feature-storage")
+            .join(&config.workspace_id)
+            .exists()
+    );
+    assert!(
+        store
+            .get_workspace("retained-workspace")
+            .await
+            .unwrap()
+            .is_some()
+    );
+}

crates/workspace-server/src/server/tests/ticket_evidence_tests.rs

@@ -74,8 +74,8 @@ impl EvidenceApiFixture {
             .unwrap();
         let assignment_id = "evidence-assignment".to_string();
         api.store
-            .set_current_ticket_coder_assignment(
-                &TicketCoderAssignmentRecord {
+            .set_current_ticket_worker_assignment(
+                &TicketWorkerAssignmentRecord {
                     workspace_id: TEST_WORKSPACE_ID.into(),
                     ticket_id: ticket_id.clone(),
                     assignment_id: assignment_id.clone(),
@@ -319,7 +319,7 @@ async fn done_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_
         fixture
             .api
             .store
-            .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
             .unwrap()
             .is_some()
     );
@@ -512,7 +512,7 @@ async fn open_source_without_runtime_is_typed_unavailable_not_stale_after_rescop
         fixture
             .api
             .store
-            .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
             .unwrap()
             .is_some()
     );

crates/workspace-server/src/server/tests/ticket_worker_claim_tests.rs

@@ -0,0 +1,657 @@
+use super::*;
+
+async fn claim(
+    fixture: &ManualCoderAssignmentFixture,
+    request: server_api::SetTicketRoleAssignmentRequest,
+) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
+    scoped_set_ticket_assignment(
+        State(fixture.api.clone()),
+        AxumPath((
+            TEST_WORKSPACE_ID.into(),
+            fixture.ticket_id.clone(),
+            "worker".into(),
+        )),
+        Json(request),
+    )
+    .await
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_accepts_no_resources_and_preserves_assignment_on_retry() {
+    let fixture = manual_worker_assignment_fixture().await;
+    for id in [&fixture.main_workdir_id, &fixture.docs_workdir_id] {
+        fixture
+            .api
+            .store
+            .detach_worker_workdir(
+                TEST_WORKSPACE_ID,
+                &fixture.worker,
+                Some(id),
+                TEST_CREATED_AT,
+            )
+            .unwrap();
+    }
+    sync_runtime_worker_workdir_attachments(&fixture.api, &fixture.worker).unwrap();
+    let backend = browser_ticket_backend(&fixture.api).unwrap();
+    backend
+        .edit_item(
+            fixture.ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Clear),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    backend
+        .set_workflow_state(
+            fixture.ticket_id.clone().into(),
+            TicketStateChange::new(
+                "ready",
+                "planning",
+                "Explicit investigation",
+                "No repository resources required",
+            ),
+        )
+        .unwrap();
+    let request = manual_worker_assignment_request(&fixture, "claim-without-resources");
+    assert!(request.workdir_bindings.is_empty());
+    let Json(first) = claim(&fixture, request.clone()).await.unwrap();
+    let Json(replay) = claim(&fixture, request).await.unwrap();
+    assert_eq!(
+        first.assignment.unwrap().assignment_id,
+        replay.assignment.unwrap().assignment_id
+    );
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "inprogress"
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+    assert!(
+        fixture
+            .runtime
+            .logical_attachments(&fixture.worker.worker_id)
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_narrows_read_only_targets_without_orchestrator_conflict() {
+    let fixture = manual_worker_assignment_fixture().await;
+    let backend = browser_ticket_backend(&fixture.api).unwrap();
+    backend
+        .edit_item(
+            fixture.ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Set {
+                    targets: vec![
+                        test_ticket_target(
+                            "test-repository",
+                            "develop",
+                            TicketTargetAccess::ReadOnly,
+                        ),
+                        test_ticket_target("docs", "develop", TicketTargetAccess::ReadOnly),
+                    ],
+                }),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    assign_test_orchestrator(&fixture.api, &fixture.ticket_id);
+    let request = manual_worker_assignment_request(&fixture, "claim-read-only");
+    let _ = claim(&fixture, request).await.unwrap();
+    for link in fixture
+        .api
+        .store
+        .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
+        .unwrap()
+        .into_iter()
+        .filter(|link| link.unlinked_at.is_none())
+    {
+        assert_eq!(
+            link.capabilities,
+            workdir::WorkdirSessionCapabilities::READ_ONLY
+        );
+        let workdir = fixture
+            .api
+            .store
+            .get_workdir_registry(TEST_WORKSPACE_ID, &link.workdir_id)
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            effective_worker_workdir_capabilities(
+                &fixture.api,
+                &fixture.worker,
+                &workdir,
+                link.capabilities
+            )
+            .unwrap(),
+            workdir::WorkdirSessionCapabilities::READ_ONLY
+        );
+    }
+    assert!(
+        fixture
+            .runtime
+            .logical_attachments(&fixture.worker.worker_id)
+            .iter()
+            .all(|a| a.capabilities == workdir::WorkdirSessionCapabilities::READ_ONLY)
+    );
+    assert!(
+        require_ticket_read_write_target(
+            &fixture.api,
+            &fixture.ticket_id,
+            &test_repository_id(&fixture.api),
+            "develop"
+        )
+        .is_err()
+    );
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_requires_connections_instead_of_ambient_attachments() {
+    let fixture = manual_worker_assignment_fixture().await;
+    let mut request = manual_worker_assignment_request(&fixture, "claim-ambient-denied");
+    request.workdir_bindings.clear();
+    let response = claim(&fixture, request).await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("exact live Workdir connections"));
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 0);
+    assert!(
+        fixture
+            .api
+            .store
+            .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-ambient-denied")
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn ambiguous_existing_worker_restore_keeps_claim_pending_until_accepted_recovery() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let request = manual_worker_assignment_request(&fixture, "claim-unknown");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let operation = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(operation.claim_state, "pending");
+    assert!(operation.assignment_id.is_none());
+    assert_eq!(operation.worker, Some(fixture.worker.clone()));
+    assert!(operation.request_fingerprint.is_some());
+    let recovery = fixture
+        .api
+        .config_store
+        .get_ticket_claim_recovery(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        recovery.item_revision,
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .item_revision
+    );
+    assert_eq!(recovery.selected.len(), request.workdir_bindings.len());
+    for selected in &request.workdir_bindings {
+        assert!(recovery.selected.contains(selected));
+    }
+    assert!(
+        recovery
+            .original_links
+            .iter()
+            .all(|link| link.capabilities == workdir::WorkdirSessionCapabilities::ALL)
+    );
+    assert_eq!(
+        recovery
+            .effective_links
+            .iter()
+            .find(|link| link.alias == "docs")
+            .unwrap()
+            .capabilities,
+        workdir::WorkdirSessionCapabilities::READ_ONLY
+    );
+    // This fake proves the domain transition, not execution-level restore deduplication.
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.request_fingerprint, operation.request_fingerprint);
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 2);
+    assert!(
+        fixture
+            .api
+            .store
+            .reserve_ticket_assignment_operation(
+                TEST_WORKSPACE_ID,
+                "competing-launch",
+                &fixture.ticket_id,
+                &fixture.worker.runtime_id,
+                None,
+                "other-intent",
+                TEST_CREATED_AT
+            )
+            .is_err()
+    );
+    let worker = WorkspaceWorker::resolve(
+        &fixture.api,
+        &fixture.worker.runtime_id,
+        &fixture.worker.worker_id,
+    )
+    .unwrap();
+    let denied = worker
+        .input(
+            &WorkerOperationContext::Backend,
+            WorkerInputRequest {
+                kind: WorkerInputKind::User,
+                content: "Do not run while claim is unresolved".into(),
+                segments: None,
+                submission_request_id: None,
+            },
+        )
+        .await
+        .unwrap_err();
+    assert!(denied.error.to_string().contains("pending Ticket claim"));
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "ready"
+    );
+    assert!(
+        fixture
+            .api
+            .store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .unwrap()
+            .is_none()
+    );
+
+    let Json(accepted) = claim(&fixture, request.clone()).await.unwrap();
+    let assignment = accepted.assignment.unwrap();
+    let committed = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(committed.claim_state, "committed");
+    assert_eq!(
+        committed.assignment_id,
+        Some(assignment.assignment_id.clone())
+    );
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "inprogress"
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 3);
+    assert_eq!(
+        fixture
+            .api
+            .config_store
+            .get_ticket_claim_recovery(TEST_WORKSPACE_ID, "claim-unknown")
+            .unwrap()
+            .unwrap(),
+        recovery
+    );
+    let Json(replay) = claim(&fixture, request).await.unwrap();
+    assert_eq!(
+        replay.assignment.unwrap().assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 3);
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_rejects_changed_bindings_without_restore() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let mut request = manual_worker_assignment_request(&fixture, "claim-changed-bindings");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let admitted = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    request.workdir_bindings[0].connection_id = "changed-connection".into();
+    let response = claim(&fixture, request.clone())
+        .await
+        .unwrap_err()
+        .into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("different input"));
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.request_fingerprint, admitted.request_fingerprint);
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_without_pinned_intent_rejects_replay_without_restore() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let request = manual_worker_assignment_request(&fixture, "claim-missing-intent");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    // Simulate an interrupted/legacy pending receipt with no exact restore intent.
+    let request_id = format!("manual-worker-restore:{}", request.operation_id);
+    fixture.api.config_store.with_conn(|conn| {
+        let deleted = conn.execute(
+            "DELETE FROM worker_restore_intents WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND request_id=?4",
+            rusqlite::params![TEST_WORKSPACE_ID, fixture.worker.runtime_id, fixture.worker.worker_id, request_id],
+        )?;
+        assert_eq!(deleted, 1);
+        Ok(())
+    }).unwrap();
+    let response = claim(&fixture, request.clone())
+        .await
+        .unwrap_err()
+        .into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("no pinned restore intent"));
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+    assert!(
+        fixture
+            .api
+            .config_store
+            .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &fixture.worker, &request_id)
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits_without_input() {
+    let workspace = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(workspace.path()).await;
+    execution.accept_restores();
+    let worker = spawn_ticket_check_source(&api, "manual-claim-reconciliation");
+    assert!(
+        api.store
+            .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_empty()
+    );
+    let backend = browser_ticket_backend(&api).unwrap();
+    let mut input = ticket::NewTicket::new("Manual investigation without resources");
+    input.workflow_state = Some(TicketWorkflowState::Ready);
+    let ticket_id = backend.create(input).unwrap().id;
+    backend
+        .edit_item(
+            ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Clear),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    backend
+        .set_workflow_state(
+            ticket_id.clone().into(),
+            TicketStateChange::new(
+                "ready",
+                "planning",
+                "Explicit investigation",
+                "No repository resources required",
+            ),
+        )
+        .unwrap();
+    let admitted_revision = api.authority.ticket(&ticket_id).unwrap().item_revision;
+    let request = server_api::SetTicketRoleAssignmentRequest {
+        operation_id: "claim-runtime-reconciliation".into(),
+        principal: server_api::TicketAssignmentPrincipal::Worker {
+            runtime_id: worker.runtime_id.clone(),
+            worker_id: worker.worker_id.clone(),
+        },
+        expected_assignment_id: None,
+        workdir_bindings: Vec::new(),
+    };
+    let submit_claim = || {
+        scoped_set_ticket_assignment(
+            State(api.clone()),
+            AxumPath((TEST_WORKSPACE_ID.into(), ticket_id.clone(), "worker".into())),
+            Json(request.clone()),
+        )
+    };
+    let uncertain_restore = || {
+        worker_runtime::execution::WorkerExecutionSpawnResult::Errored(
+            worker_runtime::execution::WorkerExecutionResult::rejected(
+                worker_runtime::execution::WorkerExecutionOperation::Restore,
+                "uncertain manual claim restore",
+            ),
+        )
+    };
+    *execution.restore_failure.lock().unwrap() = Some(uncertain_restore());
+    let response = submit_claim().await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("unknown"));
+    let pending = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.worker, Some(worker.clone()));
+    assert!(pending.request_fingerprint.is_some());
+    assert_eq!(api.authority.ticket(&ticket_id).unwrap().state, "planning");
+    assert!(
+        api.store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket_id)
+            .unwrap()
+            .is_none()
+    );
+    let recovery = api
+        .config_store
+        .get_ticket_claim_recovery(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(recovery.item_revision, admitted_revision);
+    assert!(recovery.selected.is_empty());
+    assert!(recovery.original_links.is_empty());
+    assert!(recovery.effective_links.is_empty());
+    let request_id = format!("manual-worker-restore:{}", request.operation_id);
+    let pinned = api
+        .config_store
+        .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &worker, &request_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pinned.request_id, request_id);
+    let original = execution.restore_operations.lock().unwrap()[0];
+    assert_eq!(
+        execution.restore_operations.lock().unwrap().as_slice(),
+        &[original]
+    );
+    assert!(execution.reconcile_operations.lock().unwrap().is_empty());
+
+    let handle = WorkspaceWorker::resolve(&api, &worker.runtime_id, &worker.worker_id).unwrap();
+    let denied = handle
+        .input(
+            &WorkerOperationContext::Backend,
+            WorkerInputRequest {
+                kind: WorkerInputKind::User,
+                content: "Do not submit while the manual claim is pending".into(),
+                segments: None,
+                submission_request_id: None,
+            },
+        )
+        .await
+        .unwrap_err();
+    assert!(denied.error.to_string().contains("pending Ticket claim"));
+
+    *execution.restore_failure.lock().unwrap() = Some(uncertain_restore());
+    let response = submit_claim().await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("pending"));
+    let still_pending = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(still_pending.claim_state, "pending");
+    assert!(still_pending.assignment_id.is_none());
+    assert_eq!(
+        still_pending.request_fingerprint,
+        pending.request_fingerprint
+    );
+    assert_eq!(
+        api.authority.ticket(&ticket_id).unwrap().item_revision,
+        admitted_revision
+    );
+    assert_eq!(
+        api.config_store
+            .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &worker, &request_id)
+            .unwrap()
+            .unwrap(),
+        pinned
+    );
+    assert_eq!(
+        execution.reconcile_operations.lock().unwrap().as_slice(),
+        &[original]
+    );
+
+    let Json(accepted) = submit_claim().await.unwrap();
+    let assignment = accepted.assignment.unwrap();
+    assert_eq!(assignment.principal, request.principal);
+    let committed = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(committed.claim_state, "committed");
+    assert_eq!(
+        committed.assignment_id,
+        Some(assignment.assignment_id.clone())
+    );
+    assert_eq!(committed.request_fingerprint, pending.request_fingerprint);
+    assert_eq!(
+        api.authority.ticket(&ticket_id).unwrap().state,
+        "inprogress"
+    );
+    assert_eq!(
+        api.store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket_id)
+            .unwrap()
+            .unwrap()
+            .assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(
+        api.config_store
+            .get_ticket_claim_recovery(TEST_WORKSPACE_ID, &request.operation_id)
+            .unwrap()
+            .unwrap(),
+        recovery
+    );
+    assert!(
+        api.store
+            .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_empty()
+    );
+    // The executor logs reconcile_restore in restore_operations as well. There is
+    // exactly one fresh dispatch; both subsequent entries reconcile its same id.
+    assert_eq!(
+        execution.restore_operations.lock().unwrap().as_slice(),
+        &[original, original, original]
+    );
+    assert_eq!(
+        execution.reconcile_operations.lock().unwrap().as_slice(),
+        &[original, original]
+    );
+    let Json(replay) = submit_claim().await.unwrap();
+    assert_eq!(
+        replay.assignment.unwrap().assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(execution.restore_operations.lock().unwrap().len(), 3);
+    assert_eq!(execution.reconcile_operations.lock().unwrap().len(), 2);
+    assert!(
+        !execution
+            .protocol_methods()
+            .iter()
+            .any(|(_, method)| matches!(
+                method,
+                protocol::Method::Submit { .. }
+                    | protocol::Method::SubmitTracked { .. }
+                    | protocol::Method::SubmitIfIdle { .. }
+            )),
+        "manual claims and denied input must not generate a Submit"
+    );
+}

crates/workspace-server/src/server/worker_operations.rs

@@ -359,6 +359,8 @@ impl WorkspaceWorker {
             "send_input"
         };
         let _guard = self.authorize_operation(context, permission).await?;
+        let session_lock = current_worker_session_lock(&self.api, &self.identity);
+        let _session_guard = session_lock.lock().await;
         self.ensure_interactive()?;
         if matches!(context, WorkerOperationContext::WorkerControl { .. })
             && request.kind == WorkerInputKind::User
@@ -373,6 +375,19 @@ impl WorkspaceWorker {
     }
 
     fn ensure_interactive(&self) -> ApiResult<()> {
+        if let Some(claim) = self
+            .api
+            .store
+            .get_pending_ticket_assignment_operation_for_worker(
+                self.api.workspace_id(),
+                &self.identity,
+            )?
+        {
+            return Err(Error::TicketAssignmentConflict(format!(
+                "Worker has pending Ticket claim {}; reconcile that exact operation before admitting input",
+                claim.operation_id,
+            )).into());
+        }
         if self
             .api
             .store
@@ -578,6 +593,8 @@ impl WorkspaceWorkerMethodSender {
             .worker
             .authorize_operation(context, "send_input")
             .await?;
+        let session_lock = current_worker_session_lock(&self.worker.api, &self.worker.identity);
+        let _session_guard = session_lock.lock().await;
         self.worker.ensure_interactive()?;
         let WorkerOperationContext::Browser { source, .. } = context else {
             return Err(Error::WorkspacePermissionDenied(

crates/workspace-server/src/server_drive.rs

@@ -0,0 +1,736 @@
+//! Authorized Workspace document API. Storage, CAS and receipts stay in workspace-drive.
+//! HTTP upload consumes frames with a hard bound; DB publication occurs only after validation.
+use super::*;
+use futures::StreamExt;
+use server_api::*;
+use sha2::{Digest, Sha256};
+use workspace_drive::{Drive, Mutation, NodeId, Revision};
+
+fn mutation_actor(context: &ServerRequestContext) -> DriveResult<String> {
+    if let Some(source) = &context.runtime_source {
+        let worker = source.worker_id.as_ref().ok_or_else(|| failure(403))?;
+        Ok(format!(
+            "worker:{}",
+            serde_json::to_string(&(&source.runtime_id, worker)).map_err(|_| failure(400))?
+        ))
+    } else {
+        Ok(format!(
+            "user:{}",
+            context.actor.as_ref().ok_or_else(|| failure(403))?.user_id
+        ))
+    }
+}
+
+type DriveResult<T> = std::result::Result<T, DriveApiError>;
+const TRANSFER_CHUNK: usize = 64 * 1024;
+
+pub(super) fn failure(status: u16) -> DriveApiError {
+    let code = match status {
+        401 | 403 => DriveApiErrorCode::Denied,
+        404 => DriveApiErrorCode::NotFound,
+        409 => DriveApiErrorCode::Conflict,
+        413 => DriveApiErrorCode::Limit,
+        500..=599 => DriveApiErrorCode::StorageUnavailable,
+        _ => DriveApiErrorCode::Invalid,
+    };
+    DriveApiError::new(code)
+}
+fn domain_error(error: workspace_drive::Error, mutation: bool) -> DriveApiError {
+    use workspace_drive::Error;
+    match error {
+        Error::Denied => failure(403),
+        Error::NotFound => failure(404),
+        Error::Conflict | Error::Fenced => failure(409),
+        Error::Invalid(_) => failure(400),
+        Error::Blob(_) => failure(503),
+        _ if mutation => DriveApiError::new(DriveApiErrorCode::OutcomeUnknown),
+        _ => failure(503),
+    }
+}
+async fn blocking<T: Send + 'static>(
+    mutation: bool,
+    action: impl FnOnce() -> workspace_drive::Result<T> + Send + 'static,
+) -> DriveResult<T> {
+    tokio::task::spawn_blocking(action)
+        .await
+        .map_err(|_| {
+            if mutation {
+                DriveApiError::new(DriveApiErrorCode::OutcomeUnknown)
+            } else {
+                failure(503)
+            }
+        })?
+        .map_err(|error| domain_error(error, mutation))
+}
+fn node_id(value: &str) -> DriveResult<NodeId> {
+    value.to_owned().try_into().map_err(|_| failure(400))
+}
+fn revision(value: &str) -> DriveResult<Revision> {
+    value.to_owned().try_into().map_err(|_| failure(400))
+}
+fn entry(workspace: &str, reference: &DriveEntryRef) -> DriveResult<NodeId> {
+    if reference.workspace_id != workspace {
+        return Err(failure(403));
+    }
+    node_id(&reference.node_id)
+}
+fn scoped_id(workspace: &str, entry_workspace: &str, id: &str) -> DriveResult<NodeId> {
+    if workspace != entry_workspace {
+        return Err(failure(403));
+    }
+    node_id(id)
+}
+fn safe_content_type(value: &str) -> DriveResult<String> {
+    // Deliberately accept a bare MIME type only, not attacker-controlled header parameters.
+    if value.len() > 128
+        || !value.contains('/')
+        || value.matches('/').count() != 1
+        || value.split('/').any(str::is_empty)
+        || !value
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b"!#$&^_.+-/".contains(&b))
+    {
+        return Err(failure(400));
+    }
+    Ok(value.to_ascii_lowercase())
+}
+fn disposition(name: &str) -> String {
+    // ASCII fallback plus RFC 5987 UTF-8 filename. Never place raw names in headers.
+    let encoded: String = name
+        .as_bytes()
+        .iter()
+        .map(|b| {
+            if b.is_ascii_alphanumeric() || b"-._".contains(b) {
+                (*b as char).to_string()
+            } else {
+                format!("%{b:02X}")
+            }
+        })
+        .collect();
+    format!("attachment; filename=\"download\"; filename*=UTF-8''{encoded}")
+}
+fn error_response(error: DriveApiError) -> Response {
+    let status = match error.code {
+        DriveApiErrorCode::Denied => 403,
+        DriveApiErrorCode::NotFound => 404,
+        DriveApiErrorCode::Conflict => 409,
+        DriveApiErrorCode::Limit => 413,
+        DriveApiErrorCode::Invalid => 400,
+        DriveApiErrorCode::StorageUnavailable | DriveApiErrorCode::OutcomeUnknown => 503,
+    };
+    (StatusCode::from_u16(status).unwrap(), Json(error)).into_response()
+}
+// No raw storage diagnostic is ever placed in a public error or stream failure.
+fn stream_error() -> std::io::Error {
+    std::io::Error::other("Drive transfer unavailable")
+}
+
+/// Stream HTTP frames while enforcing both announced and observed limits. Storage currently
+/// accepts a bounded whole-file Vec; this is binary buffering, never JSON/base64 expansion.
+async fn receive(body: axum::body::Body, declared: u64, digest: &str) -> DriveResult<Vec<u8>> {
+    if declared > workspace_drive::MAX_FILE_BYTES as u64 {
+        return Err(failure(413));
+    }
+    if digest.len() != 64
+        || !digest
+            .bytes()
+            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+    {
+        return Err(failure(400));
+    }
+    let mut stream = body.into_data_stream();
+    let mut bytes = Vec::new();
+    let mut hasher = Sha256::new();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(|_| failure(400))?;
+        let size = bytes
+            .len()
+            .checked_add(chunk.len())
+            .ok_or_else(|| failure(413))?;
+        if size > workspace_drive::MAX_FILE_BYTES {
+            return Err(failure(413));
+        }
+        if size > declared as usize {
+            return Err(failure(400));
+        }
+        hasher.update(&chunk);
+        bytes.extend_from_slice(&chunk);
+    }
+    if bytes.len() != declared as usize
+        || hasher
+            .finalize()
+            .iter()
+            .map(|b| format!("{b:02x}"))
+            .collect::<String>()
+            != digest
+    {
+        return Err(failure(400));
+    }
+    Ok(bytes)
+}
+
+fn public_node(node: workspace_drive::Node) -> DriveEntry {
+    let workspace_id = node.workspace_id.clone();
+    let time = chrono::DateTime::from_timestamp_millis(node.updated_at_ms)
+        .map(|t| t.to_rfc3339())
+        .unwrap_or_default();
+    DriveEntry {
+        entry: DriveEntryRef {
+            workspace_id: workspace_id.clone(),
+            node_id: node.id.get().to_string(),
+        },
+        parent: node.parent_id.map(|id| DriveEntryRef {
+            workspace_id,
+            node_id: id.get().to_string(),
+        }),
+        name: node.name,
+        kind: match node.kind {
+            workspace_drive::Kind::File => DriveEntryKind::File,
+            workspace_drive::Kind::Directory => DriveEntryKind::Folder,
+        },
+        revision: node.revision.get().to_string(),
+        size: (node.kind == workspace_drive::Kind::File).then_some(node.size as u32),
+        content_type: node.content_type,
+        updated_by: node.updated_by,
+        updated_at: time,
+        latest_url: format!(
+            "/api/w/{}/drive/{}?entry_workspace_id={}&id={}",
+            node.workspace_id,
+            if node.kind == workspace_drive::Kind::File {
+                "download"
+            } else {
+                "metadata"
+            },
+            node.workspace_id,
+            node.id.get()
+        ),
+    }
+}
+fn cursor(workspace: &str, id: Option<NodeId>) -> Option<String> {
+    id.map(|id| format!("{workspace}:{}", id.get()))
+}
+fn after(workspace: &str, value: Option<String>) -> DriveResult<Option<NodeId>> {
+    value
+        .map(|value| {
+            let (scope, id) = value.rsplit_once(':').ok_or_else(|| failure(400))?;
+            scoped_id(workspace, scope, id)
+        })
+        .transpose()
+}
+fn limit(value: Option<u32>, maximum: usize) -> DriveResult<usize> {
+    let limit = value.unwrap_or(maximum as u32) as usize;
+    if limit == 0 || limit > maximum {
+        return Err(failure(413));
+    }
+    Ok(limit)
+}
+pub(super) async fn root(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+) -> DriveResult<DriveEntry> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    blocking(false, move || drive.root()).await.map(public_node)
+}
+pub(super) async fn metadata(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveEntryQuery,
+) -> DriveResult<DriveEntry> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
+    blocking(false, move || drive.metadata(id))
+        .await
+        .map(public_node)
+}
+pub(super) async fn list(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveListQuery,
+) -> DriveResult<DriveListResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
+    let after = after(workspace, query.after)?;
+    let limit = limit(query.limit, workspace_drive::MAX_PAGE)?;
+    let page = blocking(false, move || drive.list(id, after, limit)).await?;
+    Ok(DriveListResponse {
+        entries: page.nodes.into_iter().map(public_node).collect(),
+        next_after: cursor(workspace, page.next_after),
+    })
+}
+pub(super) async fn search(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveSearchQuery,
+) -> DriveResult<DriveListResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let after = after(workspace, query.after)?;
+    let limit = limit(query.limit, workspace_drive::MAX_SEARCH_NODES)?;
+    let page = blocking(false, move || {
+        drive.search(&query.query, query.include_text, after, limit)
+    })
+    .await?;
+    Ok(DriveListResponse {
+        entries: page.nodes.into_iter().map(public_node).collect(),
+        next_after: cursor(workspace, page.next_after),
+    })
+}
+pub(super) async fn read_text(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveReadTextQuery,
+) -> DriveResult<DriveReadTextResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
+    let max = query.max_bytes as usize;
+    if max == 0 || max > DRIVE_TEXT_MAX_BYTES {
+        return Err(failure(413));
+    }
+    let (node, chunk) = blocking(false, move || {
+        let node = drive.metadata(id)?;
+        let chunk = drive.read(id, node.revision, 0, max)?;
+        Ok((node, chunk))
+    })
+    .await?;
+    let text = match std::str::from_utf8(&chunk.bytes) {
+        Ok(text) => text,
+        Err(error) if !chunk.eof && error.error_len().is_none() => {
+            std::str::from_utf8(&chunk.bytes[..error.valid_up_to()]).map_err(|_| failure(400))?
+        }
+        Err(_) => return Err(failure(400)),
+    };
+    Ok(DriveReadTextResponse {
+        entry: public_node(node),
+        text: text.into(),
+        truncated: !chunk.eof,
+    })
+}
+pub(super) async fn read_chunk(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveReadChunkQuery,
+) -> DriveResult<BinaryBody> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
+    let revision = revision(&query.expected_revision)?;
+    if query.length == 0 || query.length > DRIVE_CHUNK_MAX_BYTES {
+        return Err(failure(413));
+    }
+    let chunk = blocking(false, move || {
+        drive.read(id, revision, query.offset.into(), query.length as usize)
+    })
+    .await?;
+    Ok(chunk.bytes.into())
+}
+fn text_bytes(text: String) -> DriveResult<Vec<u8>> {
+    if text.len() > DRIVE_TEXT_MAX_BYTES {
+        return Err(failure(413));
+    }
+    Ok(text.into_bytes())
+}
+fn mutation(workspace: &str, mutation: DriveMutation) -> DriveResult<Mutation> {
+    Ok(match mutation {
+        DriveMutation::CreateFolder { parent, name } => Mutation::CreateFolder {
+            parent: entry(workspace, &parent)?,
+            name,
+        },
+        DriveMutation::CreateText {
+            parent,
+            name,
+            text,
+            content_type,
+        } => Mutation::CreateFile {
+            parent: entry(workspace, &parent)?,
+            name,
+            content_type: safe_content_type(&content_type)?,
+            bytes: text_bytes(text)?,
+        },
+        DriveMutation::UpdateText {
+            id,
+            expected_revision,
+            text,
+            content_type,
+        } => Mutation::Update {
+            id: entry(workspace, &id)?,
+            expected_revision: revision(&expected_revision)?,
+            content_type: safe_content_type(&content_type)?,
+            bytes: text_bytes(text)?,
+        },
+        DriveMutation::Relocate {
+            id,
+            expected_revision,
+            parent,
+            name,
+        } => Mutation::Relocate {
+            id: entry(workspace, &id)?,
+            expected_revision: revision(&expected_revision)?,
+            parent: entry(workspace, &parent)?,
+            name,
+        },
+        DriveMutation::Delete {
+            id,
+            expected_revision,
+        } => Mutation::Delete {
+            id: entry(workspace, &id)?,
+            expected_revision: revision(&expected_revision)?,
+        },
+    })
+}
+fn result(request_id: String, result: workspace_drive::MutationResult) -> DriveMutationResponse {
+    DriveMutationResponse {
+        request_id,
+        entry: (!result.deleted).then(|| public_node(result.node)),
+    }
+}
+pub(super) async fn mutate(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    request: DriveMutationRequest,
+) -> DriveResult<DriveMutationResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, true).await?;
+    let actor = mutation_actor(context)?;
+    let operation = mutation(workspace, request.mutation)?;
+    let request_id = request.request_id.clone();
+    let mutated = blocking(true, move || {
+        drive.mutate(&request.request_id, &actor, operation)
+    })
+    .await?;
+    Ok(result(request_id, mutated))
+}
+fn upload_mutation(
+    workspace: &str,
+    query: &DriveUploadQuery,
+    bytes: Vec<u8>,
+) -> DriveResult<Mutation> {
+    if query.entry_workspace_id != workspace {
+        return Err(failure(403));
+    }
+    let content_type = safe_content_type(&query.content_type)?;
+    match query.operation {
+        DriveUploadOperation::Create if query.id.is_none() && query.expected_revision.is_none() => {
+            Ok(Mutation::CreateFile {
+                parent: node_id(query.parent_id.as_deref().ok_or_else(|| failure(400))?)?,
+                name: query.name.clone().ok_or_else(|| failure(400))?,
+                content_type,
+                bytes,
+            })
+        }
+        DriveUploadOperation::Update if query.parent_id.is_none() && query.name.is_none() => {
+            Ok(Mutation::Update {
+                id: node_id(query.id.as_deref().ok_or_else(|| failure(400))?)?,
+                expected_revision: revision(
+                    query
+                        .expected_revision
+                        .as_deref()
+                        .ok_or_else(|| failure(400))?,
+                )?,
+                content_type,
+                bytes,
+            })
+        }
+        _ => Err(failure(400)),
+    }
+}
+pub(super) async fn upload(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveUploadQuery,
+    body: BinaryBody,
+) -> DriveResult<DriveMutationResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, true).await?;
+    let actor = mutation_actor(context)?;
+    let bytes = receive(
+        axum::body::Body::from(body.into_bytes()),
+        query.size.into(),
+        &query.sha256,
+    )
+    .await?;
+    publish_upload(drive, actor, workspace, query, bytes).await
+}
+async fn publish_upload(
+    drive: Drive,
+    actor: String,
+    workspace: &str,
+    query: DriveUploadQuery,
+    bytes: Vec<u8>,
+) -> DriveResult<DriveMutationResponse> {
+    let mutation = upload_mutation(workspace, &query, bytes)?;
+    let request_id = query.request_id.clone();
+    let mutated = blocking(true, move || {
+        drive.mutate(&query.request_id, &actor, mutation)
+    })
+    .await?;
+    Ok(result(request_id, mutated))
+}
+pub(super) async fn request_status(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    request_id: String,
+) -> DriveResult<DriveRequestStatusResponse> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = request_id.clone();
+    let status = blocking(false, move || drive.request_status(&id)).await?;
+    Ok(match status {
+        workspace_drive::RequestStatus::Uncommitted => DriveRequestStatusResponse {
+            request_id,
+            state: DriveRequestState::Uncommitted,
+            response: None,
+        },
+        workspace_drive::RequestStatus::Committed { result: mutation } => {
+            DriveRequestStatusResponse {
+                request_id: request_id.clone(),
+                state: DriveRequestState::Committed,
+                response: Some(result(request_id, mutation)),
+            }
+        }
+    })
+}
+
+async fn select_download(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveDownloadQuery,
+) -> DriveResult<(Drive, workspace_drive::Node)> {
+    let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
+    let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
+    let for_node = drive.clone();
+    let node = blocking(false, move || for_node.metadata(id)).await?;
+    if node.kind != workspace_drive::Kind::File {
+        return Err(failure(400));
+    }
+    if query
+        .expected_revision
+        .as_deref()
+        .is_some_and(|v| v != node.revision.get().to_string())
+    {
+        return Err(failure(409));
+    }
+    safe_content_type(node.content_type.as_deref().unwrap_or("")).map_err(|_| failure(503))?;
+    Ok((drive, node))
+}
+pub(super) async fn download(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace: &str,
+    query: DriveDownloadQuery,
+) -> DriveResult<server_api::server_api_responses::DriveDownload> {
+    let (drive, node) = select_download(api, context, workspace, query).await?;
+    let mut bytes = Vec::new();
+    let mut offset = 0;
+    loop {
+        let copy = drive.clone();
+        let chunk = blocking(false, move || {
+            copy.read(node.id, node.revision, offset, TRANSFER_CHUNK)
+        })
+        .await?;
+        offset += chunk.bytes.len() as u64;
+        bytes.extend_from_slice(&chunk.bytes);
+        if chunk.eof {
+            break;
+        }
+    }
+    Ok(server_api::server_api_responses::DriveDownload::Status200 {
+        body: bytes.into(),
+        header_content_disposition: disposition(&node.name),
+        header_cache_control: "private, no-store".into(),
+        header_etag: validator(&node),
+        header_x_content_type_options: "nosniff".into(),
+        header_content_security_policy: "sandbox; default-src 'none'".into(),
+    })
+}
+fn validator(node: &workspace_drive::Node) -> String {
+    format!(
+        "\"drive-{}-{}-{}\"",
+        node.workspace_id,
+        node.id.get(),
+        node.revision.get()
+    )
+}
+
+pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
+    Router::new()
+        .merge(streaming_router(service.clone()))
+        .merge(server_api::server_api_axum::drive_root(service.clone()))
+        .merge(server_api::server_api_axum::drive_metadata(service.clone()))
+        .merge(server_api::server_api_axum::drive_list(service.clone()))
+        .merge(server_api::server_api_axum::drive_search(service.clone()))
+        .merge(server_api::server_api_axum::drive_read_text(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::drive_read_chunk(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::drive_mutate(service.clone()))
+        .merge(server_api::server_api_axum::drive_request_status(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::drive_grant_create(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::drive_grant_revoke(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::drive_grant_list(
+            service.clone(),
+        ))
+        .layer(DefaultBodyLimit::max(512 * 1024))
+        .layer(middleware::from_fn(normalize_rejections))
+}
+async fn normalize_rejections(request: Request, next: Next) -> Response {
+    let response = next.run(request).await;
+    if matches!(response.status().as_u16(), 415 | 422) {
+        error_response(failure(400))
+    } else if matches!(response.status().as_u16(), 400 | 413)
+        && response
+            .headers()
+            .get(CONTENT_TYPE)
+            .is_none_or(|value| value != "application/json")
+    {
+        error_response(failure(response.status().as_u16()))
+    } else {
+        response
+    }
+}
+
+fn streaming_router(service: Arc<ServerApiContractService>) -> Router {
+    // The operation inventory is the route authority, not duplicated string literals.
+    use api_macros::Operation;
+    use axum::routing::put;
+    use server_api::server_api_operations::{DriveDownload, DriveUpload};
+    Router::new()
+        .route(DriveUpload::METADATA.path, put(streaming_upload))
+        .route(DriveDownload::METADATA.path, get(streaming_download))
+        .with_state(service)
+}
+async fn streaming_upload(
+    State(service): State<Arc<ServerApiContractService>>,
+    AxumPath(workspace): AxumPath<String>,
+    Extension(context): Extension<ServerRequestContext>,
+    query: std::result::Result<Query<DriveUploadQuery>, axum::extract::rejection::QueryRejection>,
+    request: Request,
+) -> Response {
+    let result = async {
+        let api = service.workspace_api().map_err(|_| failure(404))?;
+        let drive = drive_grants::authorized_drive(api, &context, &workspace, true).await?;
+        let actor = mutation_actor(&context)?;
+        let Query(query) = query.map_err(|_| failure(400))?;
+        // Reject malformed metadata, foreign refs and over-limit declarations before reading.
+        upload_mutation(&workspace, &query, Vec::new())?;
+        if query.size as usize > workspace_drive::MAX_FILE_BYTES {
+            return Err(failure(413));
+        }
+        if let Some(length) = request.headers().get(axum::http::header::CONTENT_LENGTH) {
+            let length: u64 = length
+                .to_str()
+                .ok()
+                .and_then(|s| s.parse().ok())
+                .ok_or_else(|| failure(400))?;
+            if length > workspace_drive::MAX_FILE_BYTES as u64 {
+                return Err(failure(413));
+            }
+            if length != u64::from(query.size) {
+                return Err(failure(400));
+            }
+        }
+        let bytes = receive(request.into_body(), query.size.into(), &query.sha256).await?;
+        // The bound Drive authorizer runs again under attached-authority IMMEDIATE,
+        // so a grant revoked during reception cannot publish or replay a receipt.
+        publish_upload(drive, actor, &workspace, query, bytes).await
+    }
+    .await;
+    match result {
+        Ok(result) => Json(result).into_response(),
+        Err(error) => error_response(error),
+    }
+}
+async fn streaming_download(
+    State(service): State<Arc<ServerApiContractService>>,
+    AxumPath(workspace): AxumPath<String>,
+    Extension(context): Extension<ServerRequestContext>,
+    query: std::result::Result<Query<DriveDownloadQuery>, axum::extract::rejection::QueryRejection>,
+) -> Response {
+    let result = async {
+        let api = service.workspace_api().map_err(|_| failure(404))?;
+        let Query(query) = query.map_err(|_| failure(400))?;
+        let (drive, node) = select_download(api, &context, &workspace, query).await?;
+        // Validate the referenced blob before HTTP success, including a zero-byte file.
+        let first_drive = drive.clone();
+        let first = blocking(false, move || {
+            first_drive.read(node.id, node.revision, 0, TRANSFER_CHUNK)
+        })
+        .await?;
+        let id = node.id;
+        let revision = node.revision;
+        let state = (drive, id, revision, Some(first), 0_u64, false);
+        let stream = futures::stream::try_unfold(
+            state,
+            |(drive, id, revision, first, offset, done)| async move {
+                if done {
+                    return Ok(None);
+                }
+                let chunk = match first {
+                    Some(chunk) => chunk,
+                    None => {
+                        let copy = drive.clone();
+                        blocking(false, move || {
+                            copy.read(id, revision, offset, TRANSFER_CHUNK)
+                        })
+                        .await
+                        .map_err(|_| stream_error())?
+                    }
+                };
+                let next = offset + chunk.bytes.len() as u64;
+                let eof = chunk.eof;
+                Ok::<_, std::io::Error>(Some((
+                    axum::body::Bytes::from(chunk.bytes),
+                    (drive, id, revision, None, next, eof),
+                )))
+            },
+        );
+        let mut response = axum::body::Body::from_stream(stream).into_response();
+        let headers = response.headers_mut();
+        headers.insert(
+            CONTENT_TYPE,
+            HeaderValue::from_str(&safe_content_type(
+                node.content_type.as_deref().unwrap_or(""),
+            )?)
+            .map_err(|_| failure(503))?,
+        );
+        headers.insert(
+            axum::http::header::CONTENT_LENGTH,
+            HeaderValue::from_str(&node.size.to_string()).map_err(|_| failure(503))?,
+        );
+        headers.insert(
+            axum::http::header::CONTENT_DISPOSITION,
+            HeaderValue::from_str(&disposition(&node.name)).map_err(|_| failure(503))?,
+        );
+        headers.insert(
+            axum::http::header::ETAG,
+            HeaderValue::from_str(&validator(&node)).map_err(|_| failure(503))?,
+        );
+        headers.insert(CACHE_CONTROL, HeaderValue::from_static("private, no-store"));
+        headers.insert(
+            "x-content-type-options",
+            HeaderValue::from_static("nosniff"),
+        );
+        headers.insert(
+            "content-security-policy",
+            HeaderValue::from_static("sandbox; default-src 'none'"),
+        );
+        Ok::<_, DriveApiError>(response)
+    }
+    .await;
+    match result {
+        Ok(response) => response,
+        Err(error) => error_response(error),
+    }
+}

crates/workspace-server/src/server_drive_grants.rs

@@ -0,0 +1,229 @@
+//! Drive authority is rechecked inside the metadata transaction against the
+//! attached Server DB. A bound handle is not an authorization cache or lease.
+use super::*;
+use rusqlite::params;
+use server_api::{
+    DriveApiError, DriveApiErrorCode, DriveGrantCreateRequest, DriveGrantListQuery,
+    DriveGrantListResponse, DriveGrantResponse, ServerRequestContext,
+};
+
+type DriveResult<T> = std::result::Result<T, DriveApiError>;
+fn fail(code: DriveApiErrorCode) -> DriveApiError {
+    DriveApiError::new(code)
+}
+
+pub(super) fn pre_error(error: impl Into<ApiError>) -> DriveApiError {
+    let error = error.into();
+    fail(match api_error_status(&error.error).as_u16() {
+        401 | 403 => DriveApiErrorCode::Denied,
+        404 => DriveApiErrorCode::NotFound,
+        409 => DriveApiErrorCode::Conflict,
+        413 => DriveApiErrorCode::Limit,
+        500..=599 => DriveApiErrorCode::StorageUnavailable,
+        _ => DriveApiErrorCode::Invalid,
+    })
+}
+fn mutation_error(error: Error) -> DriveApiError {
+    if matches!(error, Error::Sqlite(_) | Error::Store(_)) {
+        fail(DriveApiErrorCode::OutcomeUnknown)
+    } else {
+        pre_error(error)
+    }
+}
+fn worker_request(context: &ServerRequestContext) -> bool {
+    context.runtime_source.is_some()
+        || context.worker_source.is_some()
+        || context.transport_headers.iter().any(|(name, _)| {
+            name.eq_ignore_ascii_case("x-yoi-runtime-id")
+                || name.eq_ignore_ascii_case("x-yoi-worker-id")
+        })
+}
+fn browser_actor(context: &ServerRequestContext) -> DriveResult<&RequestActor> {
+    if worker_request(context) {
+        return Err(fail(DriveApiErrorCode::Denied));
+    }
+    context
+        .actor
+        .as_ref()
+        .ok_or_else(|| fail(DriveApiErrorCode::Denied))
+}
+
+pub(super) async fn authorized_drive(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace_id: &str,
+    write: bool,
+) -> DriveResult<workspace_drive::Drive> {
+    validate_workspace_scope(api, workspace_id).map_err(pre_error)?;
+    let workspace = workspace_id.to_owned();
+    let drive = if worker_request(context) {
+        // Never fall back to an actor when a Worker/Runtime source is present,
+        // incomplete, or inconsistent with its transport identity.
+        let headers = current_worker_contract_headers(context)
+            .map_err(|_| fail(DriveApiErrorCode::Denied))?;
+        let worker = current_worker_identity(api, workspace_id, &headers)
+            .map_err(|_| fail(DriveApiErrorCode::Denied))?;
+        if context.worker_source.as_ref().is_some_and(|source| {
+            source.runtime_id != worker.runtime_id || source.worker_id != worker.worker_id
+        }) {
+            return Err(fail(DriveApiErrorCode::Denied));
+        }
+        let drive = api.drive().await.map_err(pre_error)?;
+        drive.with_authorizer(move |conn, write| {
+            if !crate::store::drive_worker_is_live(conn, true, &workspace, &worker)? {
+                return Err(workspace_drive::Error::Denied);
+            }
+            let granted: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspace_drive_grants
+                WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND revoked=0 AND (?4=0 OR access='read_write'))",
+                params![workspace,worker.runtime_id,worker.worker_id,write],|row|row.get(0))?;
+            if !granted { return Err(workspace_drive::Error::Denied); }
+            Ok(())
+        })
+    } else {
+        let actor = browser_actor(context)?;
+        let account = actor.account_id.clone();
+        let user = actor.user_id.clone();
+        let drive = api.drive().await.map_err(pre_error)?;
+        // Ordinary Workspace use is member-facing: the current Server policy
+        // admits persisted authenticated users, not only the Workspace owner.
+        drive.with_authorizer(move |conn, _write| {
+            let member: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspaces ws
+                CROSS JOIN yoi_workspace_authority.users u
+                WHERE ws.workspace_id=?1 AND ws.state='active' AND u.account_id=?2 AND u.user_id=?3)",
+                params![workspace,account,user],|row|row.get(0))?;
+            if !member { return Err(workspace_drive::Error::Denied); }
+            Ok(())
+        })
+    };
+    // Reject missing/read-only grants before upload body reception. This check
+    // is admission only: every later operation rechecks under its own lock.
+    tokio::task::spawn_blocking(move || {
+        drive.check_authorized(write)?;
+        Ok::<_, workspace_drive::Error>(drive)
+    })
+    .await
+    .map_err(|_| fail(DriveApiErrorCode::StorageUnavailable))?
+    .map_err(|error| {
+        fail(match error {
+            workspace_drive::Error::Denied => DriveApiErrorCode::Denied,
+            workspace_drive::Error::Fenced | workspace_drive::Error::Conflict => {
+                DriveApiErrorCode::Conflict
+            }
+            workspace_drive::Error::NotFound => DriveApiErrorCode::NotFound,
+            workspace_drive::Error::Invalid(_) => DriveApiErrorCode::Invalid,
+            _ => DriveApiErrorCode::StorageUnavailable,
+        })
+    })
+}
+
+pub(super) async fn create_grant(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace_id: &str,
+    request: DriveGrantCreateRequest,
+) -> DriveResult<DriveGrantResponse> {
+    let actor = browser_actor(context)?;
+    require_workspace_owner(api, workspace_id, actor, "Drive grants")
+        .await
+        .map_err(pre_error)?;
+    for identity in [&request.runtime_id, &request.worker_id] {
+        if identity.is_empty() || identity.len() > 256 || identity.chars().any(char::is_control) {
+            return Err(fail(DriveApiErrorCode::Invalid));
+        }
+    }
+    let worker = RuntimeWorkerRef::new(&request.runtime_id, &request.worker_id);
+    let summary = api
+        .runtime
+        .worker(&worker)
+        .map_err(|_| fail(DriveApiErrorCode::NotFound))?;
+    if summary.workspace.workspace_id.as_deref() != Some(workspace_id) {
+        return Err(fail(DriveApiErrorCode::Denied));
+    }
+    let grant = DriveGrantResponse {
+        grant_id: String::new(),
+        workspace_id: workspace_id.into(),
+        runtime_id: request.runtime_id,
+        worker_id: request.worker_id,
+        access: request.access,
+        revoked: false,
+        created_by: actor.account_id.clone(),
+        created_at: String::new(),
+        revoked_by: None,
+        revoked_at: None,
+    };
+    let store = api.store.clone();
+    tokio::task::spawn_blocking(move || store.create_workspace_drive_grant(&grant))
+        .await
+        .map_err(|_| fail(DriveApiErrorCode::OutcomeUnknown))?
+        .map_err(mutation_error)
+}
+
+pub(super) async fn revoke_grant(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace_id: &str,
+    grant_id: &str,
+) -> DriveResult<DriveGrantResponse> {
+    let actor = browser_actor(context)?;
+    require_workspace_owner(api, workspace_id, actor, "Drive grants")
+        .await
+        .map_err(pre_error)?;
+    validate_grant_id(grant_id)?;
+    let (workspace, grant, actor) = (
+        workspace_id.to_owned(),
+        grant_id.to_owned(),
+        actor.account_id.clone(),
+    );
+    let store = api.store.clone();
+    tokio::task::spawn_blocking(move || {
+        store.revoke_workspace_drive_grant(&workspace, &grant, &actor)
+    })
+    .await
+    .map_err(|_| fail(DriveApiErrorCode::OutcomeUnknown))?
+    .map_err(mutation_error)?
+    .ok_or_else(|| fail(DriveApiErrorCode::NotFound))
+}
+
+fn validate_grant_id(id: &str) -> DriveResult<()> {
+    if !id
+        .parse::<i64>()
+        .is_ok_and(|n| n > 0 && n.to_string() == id)
+    {
+        return Err(fail(DriveApiErrorCode::Invalid));
+    }
+    Ok(())
+}
+pub(super) async fn list_grants(
+    api: &WorkspaceApi,
+    context: &ServerRequestContext,
+    workspace_id: &str,
+    query: DriveGrantListQuery,
+) -> DriveResult<DriveGrantListResponse> {
+    let actor = browser_actor(context)?;
+    require_workspace_owner(api, workspace_id, actor, "Drive grants")
+        .await
+        .map_err(pre_error)?;
+    let limit = query.limit.unwrap_or(100) as usize;
+    if limit == 0 || limit > 200 {
+        return Err(fail(DriveApiErrorCode::Invalid));
+    }
+    if let Some(after) = &query.after {
+        validate_grant_id(after)?;
+    }
+    let workspace = workspace_id.to_owned();
+    let store = api.store.clone();
+    let mut grants = tokio::task::spawn_blocking(move || {
+        store.list_workspace_drive_grants(&workspace, limit + 1, query.after.as_deref())
+    })
+    .await
+    .map_err(|_| fail(DriveApiErrorCode::StorageUnavailable))?
+    .map_err(pre_error)?;
+    let more = grants.len() > limit;
+    grants.truncate(limit);
+    let next_after = if more {
+        grants.last().map(|g| g.grant_id.clone())
+    } else {
+        None
+    };
+    Ok(DriveGrantListResponse { grants, next_after })
+}

crates/workspace-server/src/server_workspace_config_tests.rs

@@ -129,7 +129,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_signed_editor_routes_cannot_bypass_wip_authority() {
-        let mut fixture = manual_coder_assignment_fixture().await;
+        let mut fixture = manual_worker_assignment_fixture().await;
         let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
         configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
         let api = &fixture.api;
@@ -222,7 +222,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_entrypoint_support_is_exact_and_canonical_mutations_stay_guarded() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -299,7 +299,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_grants_owner_scope_and_signed_routes() {
-        let mut fixture = manual_coder_assignment_fixture().await;
+        let mut fixture = manual_worker_assignment_fixture().await;
         let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
         configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
         let api = &fixture.api;
@@ -615,7 +615,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_readonly_revoke_and_conditional_lifetime_fences() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let read_grant = grant(api, &fixture.worker, Access::ReadOnly).await;
         assert!(
@@ -827,7 +827,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_two_workers_ui_cas_and_identical_virtual_replacement() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let second = second_worker(&fixture).await;
         grant(api, &fixture.worker, Access::ReadWrite).await;
@@ -982,7 +982,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_invalid_bounds_and_failed_save_do_not_corrupt_tree() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1114,7 +1114,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_operation_and_revoke_share_the_detach_session_lock() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let granted = grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1168,7 +1168,7 @@ mod workspace_config_integration {
     }
     #[tokio::test]
     async fn workspace_config_grant_restart_source_immutability_and_owner_acceptance_guard() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let granted = grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1255,7 +1255,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_auxiliary_secret_effects_are_unknown_not_false_rollback() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1324,7 +1324,7 @@ mod workspace_config_integration {
     }
     #[tokio::test]
     async fn workspace_config_observe_node_limit_is_complete_or_error_not_silent_truncation() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;

crates/workspace-server/src/server_workspace_config_wip_tests.rs

@@ -259,7 +259,7 @@ async fn assert_config_catalog(runtime: &worker::wip::WipRuntime, attached: &Att
 async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_import_and_unknown() {
     use worker::feature::builtin::workspace_config::wip::mount_workspace_config_wip;
     use worker::feature::builtin::workspace_config::{CONTENT_ROOT, WorkspaceConfigFeature};
-    let mut fixture = manual_coder_assignment_fixture().await;
+    let mut fixture = manual_worker_assignment_fixture().await;
     let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
     configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
     let granted = grant(&fixture.api, &fixture.worker, Access::ReadWrite).await;
@@ -706,7 +706,7 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w
         decode_runtime_request_source_claims,
     };
     // Remote Runtime-owned HTTP adapter, including its real request signer.
-    let mut fixture = manual_coder_assignment_fixture().await;
+    let mut fixture = manual_worker_assignment_fixture().await;
     let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
     configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
     let claims = Arc::new(Mutex::new(Vec::new()));

crates/workspace-server/src/workspace_deletion.rs

@@ -87,6 +87,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "worker_workdir_links",
     "workspace_config_entries",
     "workspace_config_grants",
+    "workspace_drive_grants",
     "workspace_config_tree_revisions",
     "workspace_config_trees",
     "workspace_create_operations",

crates/workspace-server/src/workspace_drive_grants.sql

@@ -0,0 +1,24 @@
+CREATE TABLE workspace_drive_grants (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    access TEXT NOT NULL CHECK(access IN ('read_only', 'read_write')),
+    revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0, 1)),
+    created_by TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    revoked_by TEXT,
+    revoked_at TEXT,
+    CHECK((revoked=0 AND revoked_by IS NULL AND revoked_at IS NULL)
+       OR (revoked=1 AND revoked_by IS NOT NULL AND revoked_at IS NOT NULL))
+);
+CREATE UNIQUE INDEX workspace_drive_grants_active_worker
+    ON workspace_drive_grants(workspace_id, runtime_id, worker_id) WHERE revoked=0;
+-- Retain grant audit across Worker deletion, but never carry its authority into
+-- a later lifetime reusing the same Runtime/Worker identity.
+CREATE TRIGGER workspace_drive_grants_worker_deleted BEFORE DELETE ON worker_registry BEGIN
+    UPDATE workspace_drive_grants SET revoked=1, revoked_by='server:worker-removal',
+        revoked_at=strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+    WHERE workspace_id=OLD.workspace_id AND runtime_id=OLD.runtime_id
+      AND worker_id=OLD.worker_id AND revoked=0;
+END;

crates/workspace-server/src/workspace_drive_host.rs

@@ -0,0 +1,315 @@
+//! Trusted Host placement and lifecycle for Workspace Drive (no HTTP/DTO surface).
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use tokio::sync::Mutex;
+use workspace_drive::Drive;
+
+use crate::{Error, Result, WorkspaceFeatureStorage};
+
+pub(crate) fn drive_blob_root(data_root: &Path, workspace_id: &str) -> Result<PathBuf> {
+    if !data_root.is_absolute() {
+        return Err(Error::Config(
+            "Server data_root must be an absolute path".into(),
+        ));
+    }
+    if workspace_id.is_empty()
+        || workspace_id.len() > 128
+        || !workspace_id
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
+    {
+        return Err(Error::Config(
+            "Drive requires an internal Workspace storage identifier".into(),
+        ));
+    }
+    Ok(data_root.join("drives").join(workspace_id).join("blobs"))
+}
+
+// Reject redirected Host-managed ancestors before opening or deleting storage.
+// These directories are server-owned, not writable by Drive API clients.
+fn reject_symlink_ancestors(path: &Path) -> Result<()> {
+    for ancestor in path.ancestors() {
+        match std::fs::symlink_metadata(ancestor) {
+            Ok(metadata) if metadata.file_type().is_symlink() => {
+                return Err(Error::Config(
+                    "Drive storage path contains a symbolic link".into(),
+                ));
+            }
+            Ok(metadata) if !metadata.is_dir() => {
+                return Err(Error::Config(
+                    "Drive storage path contains a non-directory".into(),
+                ));
+            }
+            Ok(_) => {}
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error.into()),
+        }
+    }
+    Ok(())
+}
+
+pub(crate) struct WorkspaceDriveHost {
+    registration: crate::RegisteredFeature,
+    purged: AtomicBool,
+    drive: Mutex<Option<Drive>>,
+}
+
+impl WorkspaceDriveHost {
+    pub(crate) async fn new(storage: &WorkspaceFeatureStorage) -> Result<Self> {
+        let storage = storage.clone();
+        let registration = tokio::task::spawn_blocking(move || Drive::register(&storage))
+            .await
+            .map_err(|error| Error::Store(format!("Drive registration task failed: {error}")))?
+            .map_err(|error| Error::Store(format!("failed to register Drive storage: {error}")))?;
+        Ok(Self {
+            registration,
+            purged: AtomicBool::new(false),
+            drive: Mutex::new(None),
+        })
+    }
+    pub(crate) async fn open(
+        &self,
+        storage: &WorkspaceFeatureStorage,
+        root: PathBuf,
+        authority_database: PathBuf,
+    ) -> Result<Drive> {
+        let mut cached = self.drive.lock().await;
+        if let Some(drive) = cached.as_ref() {
+            return Ok(drive.clone());
+        }
+        let storage = storage.clone();
+        let registration = self.registration.clone();
+        let drive = tokio::task::spawn_blocking(move || {
+            reject_symlink_ancestors(&root)?;
+            // Domain IMMEDIATE transactions lock the attached server authority
+            // as well as Drive metadata. Deletion reservation and cached-handle
+            // operations therefore serialize with other server DB writes.
+            Drive::open_with_workspace_authority(
+                &storage,
+                &registration,
+                &root,
+                &authority_database,
+            )
+            .map_err(Error::from)
+        })
+        .await
+        .map_err(|error| Error::Store(format!("Drive open task failed: {error}")))??;
+        *cached = Some(drive.clone());
+        Ok(drive)
+    }
+
+    pub(crate) async fn fence(
+        &self,
+        storage: &WorkspaceFeatureStorage,
+        root: PathBuf,
+        authority_database: PathBuf,
+    ) -> Result<()> {
+        if self.purged.load(Ordering::Acquire) {
+            return Ok(());
+        }
+        let drive = self.open(storage, root, authority_database).await?;
+        tokio::task::spawn_blocking(move || drive.fence().map_err(Error::from))
+            .await
+            .map_err(|error| Error::Store(format!("Drive fence task failed: {error}")))?
+    }
+
+    pub(crate) async fn purge(
+        &self,
+        storage: &WorkspaceFeatureStorage,
+        root: PathBuf,
+        authority_database: PathBuf,
+    ) -> Result<()> {
+        if self.purged.load(Ordering::Acquire) {
+            return Ok(());
+        }
+        let drive = self.open(storage, root.clone(), authority_database).await?;
+        tokio::task::spawn_blocking(move || -> Result<()> {
+            reject_symlink_ancestors(&root)?;
+            // Domain purge persists its tombstone before removing blobs/nodes/receipts.
+            drive.purge()?;
+            // Remove the empty blob root, never the data root or another Workspace.
+            // Missing roots are successful retries; other failures retain FeatureStorage.
+            match std::fs::remove_dir_all(&root) {
+                Ok(()) => Ok(()),
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+                Err(error) => Err(error.into()),
+            }
+        })
+        .await
+        .map_err(|error| Error::Store(format!("Drive purge task failed: {error}")))??;
+        // A later signing-material/finalization failure can retry after the
+        // feature DB has been closed and removed by FeatureStorage.delete.
+        self.purged.store(true, Ordering::Release);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use workspace_drive::Mutation;
+
+    fn seed_authority(data_root: &Path) -> PathBuf {
+        let path = data_root.join("server.db");
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute_batch(
+            "CREATE TABLE workspaces (workspace_id TEXT PRIMARY KEY, state TEXT NOT NULL);
+            INSERT INTO workspaces VALUES ('workspace-a', 'active');",
+        )
+        .unwrap();
+        path
+    }
+
+    #[test]
+    fn drive_path_uses_explicit_data_root_and_internal_workspace_id() {
+        let temp = tempfile::tempdir().unwrap();
+        let data = temp.path().join("data");
+        assert_eq!(
+            drive_blob_root(&data, "workspace-a").unwrap(),
+            data.join("drives/workspace-a/blobs")
+        );
+        for id in [
+            "",
+            ".",
+            "..",
+            "../other",
+            "/absolute",
+            "a/b",
+            "a\\b",
+            "display name",
+        ] {
+            assert!(drive_blob_root(&data, id).is_err(), "accepted {id:?}");
+        }
+        assert!(drive_blob_root(Path::new("relative"), "workspace-a").is_err());
+    }
+
+    #[tokio::test]
+    async fn drive_fence_survives_host_reconstruction_before_purge() {
+        let temp = tempfile::tempdir().unwrap();
+        let feature_root = temp.path().join("feature-storage");
+        let root = drive_blob_root(temp.path(), "workspace-a").unwrap();
+        let authority = seed_authority(temp.path());
+        let storage = crate::FeatureStorage::new(&feature_root)
+            .workspace("workspace-a")
+            .unwrap();
+        let host = WorkspaceDriveHost::new(&storage).await.unwrap();
+        let drive = host
+            .open(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        let parent = tokio::task::spawn_blocking(move || drive.root().unwrap().id)
+            .await
+            .unwrap();
+        host.fence(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        drop(host);
+        drop(storage);
+        let storage = crate::FeatureStorage::new(&feature_root)
+            .workspace("workspace-a")
+            .unwrap();
+        let host = WorkspaceDriveHost::new(&storage).await.unwrap();
+        let reopened = host.open(&storage, root, authority.clone()).await.unwrap();
+        let result = tokio::task::spawn_blocking(move || {
+            reopened.mutate(
+                "after-restart",
+                "owner",
+                Mutation::CreateFolder {
+                    parent,
+                    name: "forbidden".into(),
+                },
+            )
+        })
+        .await
+        .unwrap();
+        assert!(
+            matches!(result, Err(workspace_drive::Error::Fenced)),
+            "{result:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn drive_purge_removes_only_its_blob_root_and_retries_after_feature_delete() {
+        let temp = tempfile::tempdir().unwrap();
+        let feature_root = temp.path().join("feature-storage");
+        let root = drive_blob_root(temp.path(), "workspace-a").unwrap();
+        let authority = seed_authority(temp.path());
+        let other_root = drive_blob_root(temp.path(), "workspace-b").unwrap();
+        std::fs::create_dir_all(&other_root).unwrap();
+        std::fs::write(other_root.join("keep"), b"other workspace").unwrap();
+        let storage = crate::FeatureStorage::new(&feature_root)
+            .workspace("workspace-a")
+            .unwrap();
+        let host = WorkspaceDriveHost::new(&storage).await.unwrap();
+        let drive = host
+            .open(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        tokio::task::spawn_blocking(move || {
+            let parent = drive.root().unwrap().id;
+            drive
+                .mutate(
+                    "upload",
+                    "owner",
+                    Mutation::CreateFile {
+                        parent,
+                        name: "file".into(),
+                        content_type: "text/plain".into(),
+                        bytes: b"content".to_vec(),
+                    },
+                )
+                .unwrap();
+        })
+        .await
+        .unwrap();
+        host.purge(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        assert!(!root.exists());
+        assert_eq!(
+            std::fs::read(other_root.join("keep")).unwrap(),
+            b"other workspace"
+        );
+        let feature_exists = feature_root.join("workspace-a/features/workspace-drive.sqlite");
+        assert!(
+            feature_exists.exists(),
+            "purge must retain the fenced DB: {}",
+            feature_exists.display()
+        );
+        let storage_for_delete = storage.clone();
+        tokio::task::spawn_blocking(move || storage_for_delete.delete())
+            .await
+            .unwrap()
+            .unwrap();
+        host.fence(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        host.purge(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        assert!(!root.exists());
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn redirected_drive_root_blocks_cleanup_and_preserves_external_files() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = drive_blob_root(temp.path(), "workspace-a").unwrap();
+        let authority = seed_authority(temp.path());
+        let outside = temp.path().join("outside");
+        std::fs::create_dir_all(&outside).unwrap();
+        std::fs::write(outside.join("keep"), b"external").unwrap();
+        let storage = crate::FeatureStorage::new(temp.path().join("feature-storage"))
+            .workspace("workspace-a")
+            .unwrap();
+        let host = WorkspaceDriveHost::new(&storage).await.unwrap();
+        host.fence(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        std::fs::remove_dir_all(&root).unwrap();
+        std::os::unix::fs::symlink(&outside, &root).unwrap();
+        assert!(host.purge(&storage, root, authority).await.is_err());
+        assert_eq!(std::fs::read(outside.join("keep")).unwrap(), b"external");
+    }
+}