feat(drive): expose authorized typed Workspace Drive API

T-722: reuse T-721 storage/CAS/receipts with clone-bound transaction authorization against attached Server authority. Add migration 84 audited Worker read_only/read_write grants and removal/deletion fencing, member-facing ordinary access, owner-controlled grant settings, Workspace-bound decimal entry references, bounded text/binary chunks, streamed HTTP receive/download with integrity and safe headers, typed errors and generated Rust/OpenAPI/TypeScript contracts.

Validation: root cargo check; server-api 83 default / 91 typescript tests; FeatureStorage 16; Drive 31 unit + 12 integration; Workspace Server 691 library + 11 binary tests (--test-threads=4), including 9 real DB/LocalFileSystem HTTP tests for two Workers/non-owner usage, revoke during reception, concurrent CAS, response-loss receipts, chunk expiry and safe binary media. cargo fmt --all -- --check and git diff --check HEAD passed. Existing OpenAPI schemas/routes semantically unchanged; dependent TS digest projections regenerated.

crates/workspace-drive/src/lib.rs

@@ -29,6 +29,8 @@ pub enum Error {
     NotFound,
     #[error("Drive revision, name, or request identity conflict")]
     Conflict,
+    #[error("Drive access denied")]
+    Denied,
     #[error("Workspace Drive is fenced for deletion")]
     Fenced,
     #[error("Drive storage failure: {0}")]
@@ -174,8 +176,11 @@ pub struct CollectionPage {
     pub next_after: Option<String>,
 }
 
+type Authorizer = dyn Fn(&Connection, bool) -> Result<()> + Send + Sync;
+
 #[derive(Clone)]
 pub struct Drive {
+    authorizer: Option<Arc<Authorizer>>,
     database: FeatureDatabase,
     blobs: Arc<blob::BlobStore>,
     workspace_id: String,
@@ -232,6 +237,7 @@ impl Drive {
             Ok(())
         })?;
         Ok(Self {
+            authorizer: None,
             database,
             blobs,
             workspace_id,
@@ -278,8 +284,36 @@ impl Drive {
         })?;
         Ok(drive)
     }
+    /// Bind immutable caller authority to this handle and its clones. The callback
+    /// runs within every client operation's IMMEDIATE transaction, never at bind
+    /// time; attached authority stays locked through blob I/O and publication.
+    /// Host maintenance must use an unbound handle.
+    pub fn with_authorizer<F>(&self, authorizer: F) -> Self
+    where
+        F: Fn(&Connection, bool) -> Result<()> + Send + Sync + 'static,
+    {
+        Self {
+            authorizer: Some(Arc::new(authorizer)),
+            ..self.clone()
+        }
+    }
+    fn authorize(&self, connection: &Connection, write: bool) -> Result<()> {
+        match &self.authorizer {
+            Some(authorizer) => authorizer(connection, write),
+            None => Ok(()),
+        }
+    }
+    /// Early admission check, not a lease: every subsequent operation still
+    /// invokes its authorizer in that operation's own transaction.
+    pub fn check_authorized(&self, write: bool) -> Result<()> {
+        self.database.try_transaction(|tx| {
+            self.authorize(tx, write)?;
+            accepting(tx)
+        })
+    }
     pub fn root(&self) -> Result<Node> {
-        self.database.try_with_connection(|c| {
+        self.database.try_transaction(|c| {
+            self.authorize(c, false)?;
             let id = c.query_row(
                 "SELECT id FROM drive_nodes WHERE parent_id IS NULL",
                 [],
@@ -289,7 +323,10 @@ impl Drive {
         })
     }
     pub fn metadata(&self, id: NodeId) -> Result<Node> {
-        self.database.try_with_connection(|c| node(c, id))
+        self.database.try_transaction(|c| {
+            self.authorize(c, false)?;
+            node(c, id)
+        })
     }
 
     /// Keyset pages are ordered by stable ID, not name. Concurrent mutations may
@@ -297,6 +334,7 @@ impl Drive {
     pub fn list(&self, parent: NodeId, after: Option<NodeId>, limit: usize) -> Result<Page> {
         validate_limit(limit, MAX_PAGE)?;
         self.database.try_transaction(|tx| {
+            self.authorize(tx, false)?;
             directory(tx, parent)?;
             let mut stmt = tx.prepare(
                 "SELECT id FROM drive_nodes WHERE parent_id=?1 AND id>?2 ORDER BY id LIMIT ?3",
@@ -320,7 +358,8 @@ impl Drive {
 
     pub fn request_status(&self, request_id: &str) -> Result<RequestStatus> {
         validate_identity(request_id, "request_id", 128)?;
-        self.database.try_with_connection(|c| {
+        self.database.try_transaction(|c| {
+            self.authorize(c, false)?;
             let result = c
                 .query_row(
                     "SELECT result_json FROM drive_receipts WHERE request_id=?1",
@@ -354,6 +393,7 @@ impl Drive {
             .map(|b| format!("{b:02x}"))
             .collect();
         self.database.try_transaction(|tx| {
+            self.authorize(tx, true)?;
             if let Some((old_fingerprint,json)) = tx.query_row("SELECT fingerprint,result_json FROM drive_receipts WHERE request_id=?1",
                 [request_id], |r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?))).optional()? {
                 if old_fingerprint != fingerprint { return Err(Error::Conflict); }
@@ -419,6 +459,7 @@ impl Drive {
     ) -> Result<ReadChunk> {
         validate_limit(length, MAX_READ_BYTES)?;
         self.database.try_transaction(|tx| {
+            self.authorize(tx, false)?;
             accepting(tx)?;
             let selected = node(tx, id)?;
             if selected.revision != revision {
@@ -488,6 +529,7 @@ impl Drive {
         validate_identity(query, "query", 256)?;
         validate_limit(budget, MAX_SEARCH_NODES)?;
         self.database.try_transaction(|tx| {
+            self.authorize(tx, false)?;
             accepting(tx)?;
             let mut stmt = tx.prepare("SELECT id FROM drive_nodes WHERE parent_id IS NOT NULL AND id>?1 ORDER BY id LIMIT ?2")?;
             let ids = stmt.query_map(params![after.map_or(0,|v|v.0),budget+1],|r|r.get::<_,i64>(0))?.collect::<std::result::Result<Vec<_>,_>>()?;

crates/workspace-drive/src/tests.rs

@@ -299,3 +299,183 @@ fn interrupted_full_permission_and_sync_failed_uploads_do_not_publish_metadata_o
         );
     }
 }
+
+fn grant_fixture(t: &tempfile::TempDir) -> (Drive, Connection) {
+    let authority = t.path().join("server.db");
+    let server = Connection::open(&authority).unwrap();
+    feature_storage::configure_connection(&server).unwrap();
+    server
+        .execute_batch(
+            "CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY,state TEXT NOT NULL);
+        INSERT INTO workspaces VALUES ('ws','active');
+        CREATE TABLE grants(write_access INTEGER NOT NULL,revoked INTEGER NOT NULL);
+        INSERT INTO grants VALUES (1,0);",
+        )
+        .unwrap();
+    let storage = feature_storage::FeatureStorage::new(t.path().join("metadata"))
+        .workspace("ws")
+        .unwrap();
+    let registration = Drive::register(&storage).unwrap();
+    (
+        Drive::open_with_workspace_authority(
+            &storage,
+            &registration,
+            &t.path().join("blobs"),
+            &authority,
+        )
+        .unwrap(),
+        server,
+    )
+}
+fn grant_authorizer(conn: &Connection, write: bool) -> Result<()> {
+    assert!(
+        !conn.is_autocommit(),
+        "authority must be checked in the operation transaction"
+    );
+    let allowed: bool = conn.query_row(
+        "SELECT revoked=0 AND (?1=0 OR write_access=1) FROM yoi_workspace_authority.grants",
+        [write],
+        |r| r.get(0),
+    )?;
+    if allowed { Ok(()) } else { Err(Error::Denied) }
+}
+
+#[test]
+fn bound_authorizer_checks_every_read_write_and_receipt_replay_without_caching() {
+    let t = tempfile::tempdir().unwrap();
+    let (raw, server) = grant_fixture(&t);
+    let root = raw.root().unwrap();
+    let bound = raw.with_authorizer(grant_authorizer);
+    let mutation = create(root.id, "file");
+    let result = bound.mutate("request", "worker", mutation.clone()).unwrap();
+    bound.check_authorized(true).unwrap();
+    server
+        .execute("UPDATE grants SET write_access=0", [])
+        .unwrap();
+    assert!(bound.check_authorized(false).is_ok());
+    assert!(matches!(bound.check_authorized(true), Err(Error::Denied)));
+    assert!(bound.root().is_ok());
+    assert!(bound.metadata(result.node.id).is_ok());
+    assert!(bound.list(root.id, None, 10).is_ok());
+    assert!(bound.search("file", true, None, 10).is_ok());
+    assert!(
+        bound
+            .read(result.node.id, result.node.revision, 0, 100)
+            .is_ok()
+    );
+    assert!(
+        bound
+            .read_text(result.node.id, result.node.revision, 0, 100)
+            .is_ok()
+    );
+    assert!(bound.request_status("request").is_ok());
+    assert!(
+        matches!(
+            bound.mutate("request", "worker", mutation.clone()),
+            Err(Error::Denied)
+        ),
+        "read-only receipt replay is a write request"
+    );
+    let clone = bound.clone();
+    server.execute("UPDATE grants SET revoked=1", []).unwrap();
+    assert!(matches!(clone.check_authorized(false), Err(Error::Denied)));
+    assert!(matches!(clone.check_authorized(true), Err(Error::Denied)));
+    assert!(matches!(clone.root(), Err(Error::Denied)));
+    assert!(matches!(clone.metadata(result.node.id), Err(Error::Denied)));
+    assert!(matches!(clone.list(root.id, None, 10), Err(Error::Denied)));
+    assert!(matches!(
+        clone.search("file", true, None, 10),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        clone.read(result.node.id, result.node.revision, 0, 100),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        clone.read_text(result.node.id, result.node.revision, 0, 100),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        clone.request_status("request"),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        clone.mutate("request", "worker", mutation),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        clone.mutate("denied", "worker", create(root.id, "denied")),
+        Err(Error::Denied)
+    ));
+    assert_eq!(
+        raw.blobs.list(None, 10).unwrap().len(),
+        1,
+        "denied write must not publish a blob"
+    );
+    assert_eq!(
+        raw.request_status("denied").unwrap(),
+        RequestStatus::Uncommitted
+    );
+    assert_eq!(raw.list(root.id, None, 10).unwrap().nodes.len(), 1);
+    raw.collect(None, 10).unwrap();
+}
+
+#[test]
+fn revoke_serializes_with_publication_and_cached_handle_denies_after_revoke_commit() {
+    use std::sync::{Mutex, mpsc};
+    use std::time::Duration;
+    let t = tempfile::tempdir().unwrap();
+    let (raw, mut server) = grant_fixture(&t);
+    let root = raw.root().unwrap();
+    let (entered_tx, entered_rx) = mpsc::channel();
+    let (release_tx, release_rx) = mpsc::channel();
+    let release_rx = Mutex::new(release_rx);
+    let publishing = raw.with_authorizer(move |conn, write| {
+        grant_authorizer(conn, write)?;
+        entered_tx.send(()).unwrap();
+        release_rx
+            .lock()
+            .unwrap()
+            .recv_timeout(Duration::from_secs(5))
+            .unwrap();
+        Ok(())
+    });
+    let cached = raw.with_authorizer(grant_authorizer);
+    let writer = std::thread::spawn(move || {
+        publishing.mutate("publishing", "worker", create(root.id, "published"))
+    });
+    entered_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+    server.busy_timeout(Duration::ZERO).unwrap();
+    assert!(
+        matches!(server.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate),
+        Err(rusqlite::Error::SqliteFailure(e,_)) if e.code==rusqlite::ErrorCode::DatabaseBusy),
+        "revoke cannot acquire authority while mutation publishes"
+    );
+    release_tx.send(()).unwrap();
+    let published = writer.join().unwrap().unwrap();
+    let revoke = server
+        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
+        .unwrap();
+    revoke.execute("UPDATE grants SET revoked=1", []).unwrap();
+    revoke.commit().unwrap();
+    assert_eq!(raw.metadata(published.node.id).unwrap(), published.node);
+    assert_eq!(
+        raw.request_status("publishing").unwrap(),
+        RequestStatus::Committed {
+            result: published.clone()
+        }
+    );
+    assert!(matches!(
+        cached.request_status("publishing"),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        cached.mutate("publishing", "worker", create(root.id, "published")),
+        Err(Error::Denied)
+    ));
+    assert!(matches!(
+        cached.mutate("after-revoke", "worker", create(root.id, "denied")),
+        Err(Error::Denied)
+    ));
+    assert_eq!(raw.blobs.list(None, 10).unwrap().len(), 1);
+}