fix: preserve restore provenance and cached catalog identity

Bind completed restore staging to its exact source and request, retain adopted legacy-live loss recovery, and lease the managed catalog through cached Drive transactions. Add interrupted restore, legacy recovery, and production consumer regressions.

crates/feature-storage/README.md

@@ -35,6 +35,17 @@ root is not a way to forget that fence. Cross-root locks are acquired in canonic
 root order, source shared and target exclusive; same-root recovery takes one
 exclusive lock. A restore retires, but does not remove, the previous owner
 receipt. Only the new physical identity is current, including after exact retry.
+Before creating staging, the owner reserves the complete source-root, snapshot,
+manifest and per-DB physical/request provenance. An unfinished restore cannot
+be replaced by a different source or snapshot. Copying uses a separate file;
+only a verified completed backup is renamed into reusable staging. Unowned
+staging is diagnosed, not accepted as a completed copy. Identity/completion
+response loss reconciles that exact request, including after directory publication.
+Adopted legacy-live scopes can recover after stopped, complete scope-tree loss:
+the old external record remains unchanged and its marker is reconstructed only
+after completed owner/snapshot provenance checks. Missing markers in an existing
+scope, legacy pending/deletion records and new-owner erasure fences still refuse
+normal admission; this recovery path does not clear erasure.
 A snapshot without matching source-owner evidence requires explicit, authorized
 legacy adoption with domain attribution/shape validation instead of blind restore.
 These checks cover owner-managed operations, not deliberate destruction of the

crates/feature-storage/src/identity_tests.rs

@@ -670,6 +670,223 @@ fn same_owner_loss_recovery_retires_old_receipts_and_erasure_blocks_every_restor
     assert!(!fresh_root.join(SCOPE).exists());
 }
 
+#[test]
+fn review_r1_unowned_incomplete_or_foreign_staging_never_becomes_a_successful_restore() {
+    for partial in [true, false] {
+        let temp = tempfile::tempdir().unwrap();
+        let source = temp.path().join("source");
+        let destination = temp.path().join("destination");
+        let snapshot = temp.path().join("snapshot");
+        let (manager, registration) = registered(&source);
+        let scoped = manager.scope(SCOPE).unwrap();
+        let database = scoped.open(&registration).unwrap();
+        insert(&database, "requested-source");
+        scoped.backup(&snapshot).unwrap();
+        let staged = destination.join(format!(".restore-{SCOPE}/features/{FEATURE}.sqlite"));
+        fs::create_dir_all(staged.parent().unwrap()).unwrap();
+        if partial {
+            drop(Connection::open(&staged).unwrap());
+        } else {
+            let foreign = temp.path().join("foreign");
+            let (other, registration) = registered(&foreign);
+            let other_scope = other.scope(SCOPE).unwrap();
+            let other_db = other_scope.open(&registration).unwrap();
+            insert(&other_db, "wrong-source");
+            let foreign_snapshot = temp.path().join("foreign-snapshot");
+            other_scope.backup(&foreign_snapshot).unwrap();
+            fs::copy(
+                foreign_snapshot.join(format!("features/{FEATURE}.sqlite")),
+                &staged,
+            )
+            .unwrap();
+            other_scope.delete().unwrap();
+        }
+        let (target, _) = registered(&destination);
+        assert!(
+            target
+                .scope(SCOPE)
+                .unwrap()
+                .restore_from(&snapshot, &scoped)
+                .is_err(),
+            "unowned staging must not be interpreted as a completed copy"
+        );
+        assert!(!destination.join(SCOPE).exists());
+    }
+}
+
+#[test]
+fn review_r3_adopted_legacy_live_scope_recovers_from_stopped_loss_without_erasure_undo() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("storage");
+    legacy(&root, SCOPE);
+    let incarnation = Uuid::now_v7();
+    fs::create_dir_all(root.join(".scope-locks")).unwrap();
+    let record = format!("v1 live {incarnation}\n");
+    fs::write(
+        root.join(".scope-locks").join(format!("{SCOPE}.lock")),
+        &record,
+    )
+    .unwrap();
+    fs::write(
+        root.join(SCOPE).join(".incarnation"),
+        format!("{incarnation}\n"),
+    )
+    .unwrap();
+    let (manager, registration) = registered(&root);
+    let scoped = manager.scope(SCOPE).unwrap();
+    let database = scoped
+        .adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
+        .unwrap();
+    let old = database.identity().clone();
+    let snapshot = temp.path().join("snapshot");
+    scoped.backup(&snapshot).unwrap();
+    drop(database);
+    manager.shutdown().unwrap();
+    fs::remove_dir_all(root.join(SCOPE)).unwrap();
+    let (manager, registration) = registered(&root);
+    let scoped = manager.scope(SCOPE).unwrap();
+    scoped.restore(&snapshot).unwrap();
+    let database = scoped.open(&registration).unwrap();
+    assert_eq!(values(&database), ["preserved"]);
+    assert_ne!(database.identity(), &old);
+    assert_eq!(
+        fs::read_to_string(root.join(".scope-locks").join(format!("{SCOPE}.lock"))).unwrap(),
+        record
+    );
+    assert_eq!(
+        fs::read_to_string(root.join(SCOPE).join(".incarnation")).unwrap(),
+        format!("{incarnation}\n")
+    );
+    drop(database);
+    scoped.delete().unwrap();
+    assert!(scoped.restore(&snapshot).is_err());
+}
+
+#[test]
+fn review_r1_interrupted_copy_keeps_exact_source_and_blocks_competing_or_erased_source_reuse() {
+    for (copied, erase_source) in [(false, false), (true, false), (true, true)] {
+        let temp = tempfile::tempdir().unwrap();
+        let source_root = temp.path().join("source");
+        let snapshot = temp.path().join("snapshot");
+        let (source, registration) = registered(&source_root);
+        let source_scope = source.scope(SCOPE).unwrap();
+        let database = source_scope.open(&registration).unwrap();
+        insert(&database, "source-A");
+        source_scope.backup(&snapshot).unwrap();
+        let target_root = temp.path().join("target");
+        let (target, registration) = registered(&target_root);
+        let error = target
+            .restore_scope_with_copy(SCOPE, &snapshot, &source_scope, |from, to| {
+                drop(Connection::open(to)?);
+                if copied {
+                    backup_database(from, to)?;
+                }
+                Err(FeatureStorageError::operation("injected copy interruption"))
+            })
+            .unwrap_err();
+        assert!(error.to_string().contains("injected copy interruption"));
+        assert!(!target_root.join(SCOPE).exists());
+        let stage = target_root.join(format!(".restore-{SCOPE}/features"));
+        assert!(stage.join(format!("{FEATURE}.copying")).is_file());
+        assert!(!stage.join(format!("{FEATURE}.sqlite")).exists());
+        let owner = Connection::open(target_root.join(".owner.sqlite")).unwrap();
+        let saved: (String, bool) = owner
+            .query_row(
+                "SELECT evidence,published FROM storage_restores WHERE scope=?1",
+                [SCOPE],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .unwrap();
+        assert!(!saved.1);
+        if erase_source {
+            source_scope.delete().unwrap();
+        }
+        let (other, other_registration) = registered(&temp.path().join("other"));
+        let other_scope = other.scope(SCOPE).unwrap();
+        let other_database = other_scope.open(&other_registration).unwrap();
+        insert(&other_database, "source-B");
+        let other_snapshot = temp.path().join("other-snapshot");
+        other_scope.backup(&other_snapshot).unwrap();
+        assert_identity_error(
+            target
+                .scope(SCOPE)
+                .unwrap()
+                .restore_from(&other_snapshot, &other_scope),
+        );
+        let retained: (String, bool) = owner
+            .query_row(
+                "SELECT evidence,published FROM storage_restores WHERE scope=?1",
+                [SCOPE],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .unwrap();
+        assert_eq!(retained, saved);
+        assert!(!target_root.join(SCOPE).exists());
+        if erase_source {
+            assert!(matches!(
+                target
+                    .scope(SCOPE)
+                    .unwrap()
+                    .restore_from(&snapshot, &source_scope),
+                Err(FeatureStorageError::ScopeMaintenance(_))
+            ));
+        } else {
+            // Restart, recopy incomplete bytes, and reconcile only the original source/request.
+            target.shutdown().unwrap();
+            let (target, registration) = registered(&target_root);
+            let scoped = target.scope(SCOPE).unwrap();
+            scoped.restore_from(&snapshot, &source_scope).unwrap();
+            scoped.restore_from(&snapshot, &source_scope).unwrap();
+            assert_eq!(values(&scoped.open(&registration).unwrap()), ["source-A"]);
+        }
+        let _ = registration;
+    }
+}
+
+#[test]
+fn review_r1_response_loss_after_publication_reconciles_the_saved_physical_identity() {
+    let temp = tempfile::tempdir().unwrap();
+    let (source, registration) = registered(&temp.path().join("source"));
+    let source_scope = source.scope(SCOPE).unwrap();
+    let database = source_scope.open(&registration).unwrap();
+    insert(&database, "committed-copy");
+    let snapshot = temp.path().join("snapshot");
+    source_scope.backup(&snapshot).unwrap();
+    let root = temp.path().join("target");
+    let (target, registration) = registered(&root);
+    drop(
+        target
+            .scope("unrelated")
+            .unwrap()
+            .open(&registration)
+            .unwrap(),
+    );
+    let owner = Connection::open(root.join(".owner.sqlite")).unwrap();
+    owner.execute_batch("CREATE TRIGGER injected_restore_completion BEFORE UPDATE OF completed ON storage_requests WHEN NEW.kind='restore' BEGIN SELECT RAISE(ABORT,'injected restore completion'); END;").unwrap();
+    let scoped = target.scope(SCOPE).unwrap();
+    assert!(matches!(
+        scoped.restore_from(&snapshot, &source_scope),
+        Err(FeatureStorageError::OutcomeUnknown { .. })
+    ));
+    let copied = Connection::open(database_path(&root, SCOPE)).unwrap();
+    let physical: String = copied
+        .query_row("SELECT physical FROM __yoi_storage_identity", [], |r| {
+            r.get(0)
+        })
+        .unwrap();
+    assert!(
+        scoped.open(&registration).is_err(),
+        "unfinished completion is not normal admission"
+    );
+    owner
+        .execute_batch("DROP TRIGGER injected_restore_completion;")
+        .unwrap();
+    scoped.restore_from(&snapshot, &source_scope).unwrap();
+    let database = scoped.open(&registration).unwrap();
+    assert_eq!(database.identity().physical, physical);
+    assert_eq!(values(&database), ["committed-copy"]);
+}
+
 struct ChildGuard(Child);
 
 impl ChildGuard {

crates/feature-storage/src/legacy_owner.rs

@@ -70,6 +70,16 @@ pub(super) fn identity(path: &Path, scope: &str, feature: &str) -> Result<Option
 }
 
 pub(super) fn fenced(root: &Path, scope: &str) -> Result<bool> {
+    fenced_inner(root, scope, false)
+}
+
+// Only restore may inspect a lost live scope; it must still match snapshot
+// provenance against completed new-owner receipts before admitting any bytes.
+pub(super) fn fenced_for_restore(root: &Path, scope: &str) -> Result<bool> {
+    fenced_inner(root, scope, true)
+}
+
+fn fenced_inner(root: &Path, scope: &str, recovering_loss: bool) -> Result<bool> {
     if root
         .join(".lifecycle")
         .join(format!("{scope}.deleted"))
@@ -100,7 +110,16 @@ pub(super) fn fenced(root: &Path, scope: &str) -> Result<bool> {
     match fields[1] {
         "deleted" | "deleting" => Ok(true),
         "live" => {
-            if fs::read_to_string(root.join(scope).join(".incarnation"))? != format!("{id}\n") {
+            let marker = fs::read_to_string(root.join(scope).join(".incarnation"));
+            if recovering_loss
+                && !root.join(scope).try_exists()?
+                && marker
+                    .as_ref()
+                    .is_err_and(|e| e.kind() == std::io::ErrorKind::NotFound)
+            {
+                return Ok(false);
+            }
+            if marker? != format!("{id}\n") {
                 return Err(FeatureStorageError::Identity(
                     "legacy scope incarnation differs".into(),
                 ));
@@ -113,6 +132,28 @@ pub(super) fn fenced(root: &Path, scope: &str) -> Result<bool> {
     }
 }
 
+/// Reconstruct only a previously retained live scope marker, never a deletion
+/// record or a new identity. Called after matching completed snapshot provenance.
+pub(super) fn restore_marker(root: &Path, scope: &str, temporary: &Path) -> Result<()> {
+    if fenced_for_restore(root, scope)? {
+        return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
+    }
+    let record = match fs::read_to_string(root.join(".scope-locks").join(format!("{scope}.lock"))) {
+        Ok(record) => record,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+        Err(e) => return Err(e.into()),
+    };
+    if let Some(id) = record
+        .strip_prefix("v1 live ")
+        .and_then(|s| s.strip_suffix('\n'))
+    {
+        let marker = temporary.join(".incarnation");
+        fs::write(&marker, format!("{id}\n"))?;
+        std::fs::File::open(marker)?.sync_all()?;
+    }
+    Ok(())
+}
+
 fn verify_shape(connection: &Connection, table: &str, expected: &str) -> Result<()> {
     let sql: String = connection.query_row(
         "SELECT sql FROM sqlite_schema WHERE type='table' AND name=?1",

crates/feature-storage/src/lib.rs

@@ -638,6 +638,16 @@ impl FeatureStorage {
         scope_id: &str,
         snapshot: &Path,
         source: &ScopedFeatureStorage,
+    ) -> Result<()> {
+        self.restore_scope_with_copy(scope_id, snapshot, source, backup_database)
+    }
+
+    fn restore_scope_with_copy(
+        &self,
+        scope_id: &str,
+        snapshot: &Path,
+        source: &ScopedFeatureStorage,
+        mut copy: impl FnMut(&Path, &Path) -> Result<()>,
     ) -> Result<()> {
         if source.scope_id != scope_id {
             return Err(FeatureStorageError::Identity(
@@ -659,27 +669,45 @@ impl FeatureStorage {
             let target = lifecycle.lock(true)?;
             (Some(source_lifecycle.lock(false)?), target)
         };
-        if lifecycle.fenced(scope_id)? {
+        if lifecycle.fenced_for_restore(scope_id)? {
             return Err(FeatureStorageError::ScopeMaintenance(scope_id.into()));
         }
         let feature_ids = read_snapshot_manifest(snapshot, scope_id)?;
         // Hold the source erasure fence through target publication. A fresh root
         // or an old snapshot cannot erase this source-owner dependency.
-        if source_lifecycle.fenced(scope_id)? {
+        if source_lifecycle.fenced_for_restore(scope_id)? {
             return Err(FeatureStorageError::ScopeMaintenance(scope_id.into()));
         }
+        // Length-prefixed provenance avoids ambiguous request/manifest concatenation.
+        let mut evidence = String::new();
+        let mut field = |value: &str| {
+            evidence.push_str(&format!("{}:{value}", value.len()));
+        };
+        let path_key = |path: &Path| {
+            path.as_os_str()
+                .as_encoded_bytes()
+                .iter()
+                .map(|b| format!("{b:02x}"))
+                .collect::<String>()
+        };
+        field(&path_key(&source_root));
+        field(&path_key(&fs::canonicalize(snapshot)?));
+        field(&fs::read_to_string(snapshot.join(SNAPSHOT_MANIFEST))?);
         for feature in &feature_ids {
-            source_lifecycle.verify_snapshot(
+            let (physical, request) = source_lifecycle.verify_snapshot(
                 scope_id,
                 feature,
                 &snapshot.join("features").join(format!("{feature}.sqlite")),
             )?;
+            field(feature);
+            field(&physical);
+            field(&request);
         }
         self.begin_maintenance(scope_id, true)?;
         let scope_dir = self.scope_dir(scope_id);
         let temporary = self.inner.root.join(format!(".restore-{scope_id}"));
-        let evidence = fs::canonicalize(snapshot)?.display().to_string();
         let result = (|| {
+            lifecycle.reserve_restore(scope_id, &evidence, &temporary, &scope_dir)?;
             if scope_dir.exists() {
                 // Rename/receipt response loss: only this exact saved restore
                 // may reconcile the published inode. An unrelated existing
@@ -701,6 +729,7 @@ impl FeatureStorage {
                     binding.check_database(&connection)?;
                     lifecycle.complete(&binding)?;
                 }
+                lifecycle.publish_restore(scope_id, &evidence)?;
                 return Ok(());
             }
             let temporary_features = temporary.join("features");
@@ -709,12 +738,18 @@ impl FeatureStorage {
             for feature_id in &feature_ids {
                 let target = temporary_features.join(format!("{feature_id}.sqlite"));
                 if !target.exists() {
-                    backup_database(
+                    // SQLite may create an empty destination before copying. Only
+                    // the verified-copy rename promotes it to reusable staging.
+                    let copying = temporary_features.join(format!("{feature_id}.copying"));
+                    copy(
                         &snapshot
                             .join("features")
                             .join(format!("{feature_id}.sqlite")),
-                        &target,
+                        &copying,
                     )?;
+                    source_lifecycle.verify_snapshot(scope_id, feature_id, &copying)?;
+                    fs::rename(&copying, &target)?;
+                    sync_directory(&temporary_features)?;
                 }
                 let request = format!("restore:{scope_id}:{feature_id}:{evidence}");
                 let binding = lifecycle.reserve(
@@ -723,9 +758,14 @@ impl FeatureStorage {
                 let mut connection =
                     Connection::open_with_flags(&target, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
                 configure_connection(&connection)?;
+                verify_database(&connection)?;
+                if binding.check_database(&connection).is_err() {
+                    source_lifecycle.verify_snapshot(scope_id, feature_id, &target)?;
+                }
                 binding.restore_identity(&mut connection)?;
                 bindings.push(binding);
             }
+            legacy_owner::restore_marker(&self.inner.root, scope_id, &temporary)?;
             sync_directory(&temporary_features)?;
             sync_directory(&temporary)?;
             fs::rename(&temporary, &scope_dir)?;
@@ -735,6 +775,7 @@ impl FeatureStorage {
                     .complete(binding)
                     .map_err(|error| binding.unknown(error))?;
             }
+            lifecycle.publish_restore(scope_id, &evidence)?;
             Ok(())
         })();
         // Failed/pending restore is retained for exact reconciliation, not

crates/feature-storage/src/owner.rs

@@ -78,6 +78,7 @@ CREATE TABLE IF NOT EXISTS storage_requests (
  current INTEGER NOT NULL DEFAULT 1);
 CREATE UNIQUE INDEX IF NOT EXISTS storage_current_target ON storage_requests(scope,feature) WHERE current=1;
 CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
+CREATE TABLE IF NOT EXISTS storage_restores (scope TEXT PRIMARY KEY, evidence TEXT NOT NULL, published INTEGER NOT NULL DEFAULT 0);
 ",
             )?;
         }
@@ -98,6 +99,81 @@ CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
         )?)
     }
 
+    /// Reserve the entire restore before creating staging. An unfinished copy
+    /// cannot be adopted by another source/request, even if its files look valid.
+    pub fn reserve_restore(
+        &self,
+        scope: &str,
+        evidence: &str,
+        temporary: &Path,
+        destination: &Path,
+    ) -> Result<()> {
+        let mut catalog = self.catalog(true)?;
+        let tx = catalog.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let previous: Option<(String, bool)> = tx
+            .query_row(
+                "SELECT evidence,published FROM storage_restores WHERE scope=?1",
+                [scope],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .optional()?;
+        match previous {
+            Some((saved, _)) if saved == evidence => {}
+            Some((_, false)) => {
+                return Err(FeatureStorageError::Identity(
+                    "another restore is unfinished; reconcile its exact source/request".into(),
+                ));
+            }
+            Some((_, true)) if destination.exists() || temporary.exists() => {
+                return Err(FeatureStorageError::Identity(
+                    "another restore owns the existing destination/staging".into(),
+                ));
+            }
+            None if destination.exists() => {
+                return Err(FeatureStorageError::DestinationExists(
+                    destination.display().to_string(),
+                ));
+            }
+            None if temporary.exists() => {
+                return Err(FeatureStorageError::Identity(
+                    "unowned restore staging requires explicit reconciliation".into(),
+                ));
+            }
+            _ => {
+                tx.execute("INSERT INTO storage_restores(scope,evidence,published) VALUES(?1,?2,0) ON CONFLICT(scope) DO UPDATE SET evidence=excluded.evidence,published=0", params![scope,evidence])?;
+            }
+        }
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub fn publish_restore(&self, scope: &str, evidence: &str) -> Result<()> {
+        let changed = self.catalog(false)?.execute(
+            "UPDATE storage_restores SET published=1 WHERE scope=?1 AND evidence=?2",
+            params![scope, evidence],
+        )?;
+        if changed != 1 {
+            return Err(FeatureStorageError::Identity(
+                "restore request changed".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    pub fn fenced_for_restore(&self, scope: &str) -> Result<bool> {
+        if legacy_owner::fenced_for_restore(&self.root, scope)? {
+            return Ok(true);
+        }
+        if !self.root.join(".owner.sqlite").try_exists()? {
+            return Ok(false);
+        }
+        Ok(self.catalog(false)?.query_row(
+            "SELECT EXISTS(SELECT 1 FROM storage_deletions WHERE scope=?1)",
+            [scope],
+            |r| r.get(0),
+        )?)
+    }
+
     pub fn delete(&self, scope: &str) -> Result<()> {
         let directory = self.root.join(".lifecycle");
         durable_create_directory(&directory, &mut sync_directory)?;
@@ -146,7 +222,12 @@ CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
                 "request and Host authority evidence are required".into(),
             ));
         }
-        if self.fenced(scope)? {
+        let fenced = if kind == "restore" {
+            self.fenced_for_restore(scope)?
+        } else {
+            self.fenced(scope)?
+        };
+        if fenced {
             return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
         }
         if kind != "provision" && !path.is_file() {
@@ -235,8 +316,13 @@ CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
     /// Snapshot bytes are not authority. Match their saved identity to this
     /// explicitly supplied source owner's completed provenance, including an old
     /// lifecycle superseded by a prior restore. The caller holds its lifecycle lock.
-    pub fn verify_snapshot(&self, scope: &str, feature: &str, path: &Path) -> Result<()> {
+    pub fn verify_snapshot(
-        if self.fenced(scope)? {
+        &self,
+        scope: &str,
+        feature: &str,
+        path: &Path,
+    ) -> Result<(String, String)> {
+        if self.fenced_for_restore(scope)? {
             return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
         }
         if !self.root.join(".owner.sqlite").is_file() {
@@ -261,7 +347,7 @@ CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
                 "snapshot does not belong to the selected source owner".into(),
             ));
         }
-        Ok(())
+        Ok((physical, request))
     }
 
     pub fn request_kind(&self, request: &str) -> Result<Option<String>> {

crates/workspace-drive/Cargo.toml

@@ -21,4 +21,5 @@ futures.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
 
 [dev-dependencies]
+rusqlite = { workspace = true, features = ["hooks"] }
 tempfile.workspace = true

crates/workspace-drive/src/lib.rs

@@ -197,6 +197,7 @@ type Authorizer = dyn Fn(&Connection, bool) -> Result<()> + Send + Sync;
 pub struct Drive {
     authorizer: Option<Arc<Authorizer>>,
     database: FeatureDatabase,
+    authority_database: Option<FeatureDatabase>,
     blobs: Arc<blob::BlobStore>,
     workspace_id: String,
     _storage: ScopedFeatureStorage,
@@ -224,13 +225,42 @@ impl Drive {
         })?;
         Ok(Self {
             authorizer: None,
+            authority_database: None,
             database,
             blobs,
             workspace_id,
             _storage: storage.clone(),
         })
     }
-    /// Attaches a trusted Host's Workspace authority DB to the connection.
+    /// Attaches a managed Host catalog using its existing physical identity owner.
+    /// Each operation reacquires its lease, including operations on cached clones.
+    /// Lock order is catalog lifecycle, Drive lifecycle/connection, then SQLite
+    /// IMMEDIATE locks (including the attached catalog). The catalog connection
+    /// mutex is released before acquiring Drive locks; only its lease spans the
+    /// operation, blob I/O, and commit/rollback. Domain authorization is unchanged.
+    pub fn open_with_managed_workspace_authority(
+        storage: &ScopedFeatureStorage,
+        registration: &RegisteredFeature,
+        root: &Path,
+        authority_database: &FeatureDatabase,
+    ) -> Result<Self> {
+        let _catalog_lease = authority_database.guard(authority_database.identity())?;
+        let path: String = authority_database.try_with_connection::<_, Error>(|connection| {
+            Ok(connection.query_row(
+                "SELECT file FROM pragma_database_list WHERE name='main'",
+                [],
+                |row| row.get(0),
+            )?)
+        })?;
+        let mut drive = Self::open(storage, registration, root)?;
+        drive.attach_workspace_authority(Path::new(&path), true)?;
+        drive.authority_database = Some(authority_database.clone());
+        Ok(drive)
+    }
+
+    /// Attaches an unmanaged Host authority path. This does not verify or lease
+    /// its physical identity; managed Hosts must use
+    /// [`Self::open_with_managed_workspace_authority`] instead.
     /// IMMEDIATE transactions lock authority too, rejecting mutations through cached handles after deletion is reserved.
     /// Drive does not modify attached authority or guarantee atomic writes across DBs.
     /// Blob I/O also serializes with Host DB writes, so long operations delay other Host writes.
@@ -242,6 +272,11 @@ impl Drive {
         authority_database: &Path,
     ) -> Result<Self> {
         let drive = Self::open(storage, registration, root)?;
+        drive.attach_workspace_authority(authority_database, false)?;
+        Ok(drive)
+    }
+
+    fn attach_workspace_authority(&self, authority_database: &Path, refresh: bool) -> Result<()> {
         let path = authority_database
             .to_str()
             .ok_or_else(|| Error::Invalid("authority database path must be UTF-8".into()))?;
@@ -250,7 +285,7 @@ impl Drive {
                 "Workspace authority database is missing".into(),
             ));
         }
-        drive.database.try_with_connection::<_, Error>(|c| {
+        self.database.try_with_connection::<_, Error>(|c| {
             let existing: Option<String> = c
                 .query_row(
                     "SELECT file FROM pragma_database_list WHERE name='yoi_workspace_authority'",
@@ -262,14 +297,19 @@ impl Drive {
                 Some(old) if Path::new(&old) != authority_database => Err(Error::Invalid(
                     "Drive already bound to a different Workspace authority".into(),
                 )),
-                Some(_) => Ok(()),
+                Some(_) if !refresh => Ok(()),
-                None => {
+                existing => {
+                    // A cached primary connection may still have an attachment
+                    // from an earlier catalog handle. Rebind under the new
+                    // owner's lease rather than accepting path equality alone.
+                    if existing.is_some() {
+                        c.execute_batch("DETACH DATABASE yoi_workspace_authority")?;
+                    }
                     c.execute("ATTACH DATABASE ?1 AS yoi_workspace_authority", [path])?;
                     Ok(())
                 }
             }
-        })?;
+        })
-        Ok(drive)
     }
     /// Binds the caller's authorization callback to the handle and its clones.
     /// Runs inside every client operation transaction, not at binding, holding the authority lock through blob I/O and publication.
@@ -283,6 +323,17 @@ impl Drive {
             ..self.clone()
         }
     }
+    fn transaction<T>(&self, operation: impl FnOnce(&Transaction<'_>) -> Result<T>) -> Result<T> {
+        // This is outside FeatureDatabase::try_transaction: releasing the lease
+        // inside its callback would expose the commit/rollback window.
+        let _catalog_lease = self
+            .authority_database
+            .as_ref()
+            .map(|database| database.guard(database.identity()))
+            .transpose()?;
+        self.database.try_transaction(operation)
+    }
+
     fn authorize(&self, connection: &Connection, write: bool) -> Result<()> {
         match &self.authorizer {
             Some(authorizer) => authorizer(connection, write),
@@ -291,13 +342,13 @@ impl Drive {
     }
     /// Early admission check, not a lease. Subsequent operations recheck authorization in their own transactions.
     pub fn check_authorized(&self, write: bool) -> Result<()> {
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             self.authorize(tx, write)?;
             accepting(tx)
         })
     }
     pub fn root(&self) -> Result<Node> {
-        self.database.try_transaction(|c| {
+        self.transaction(|c| {
             self.authorize(c, false)?;
             let id = c.query_row(
                 "SELECT id FROM drive_nodes WHERE parent_id IS NULL",
@@ -308,7 +359,7 @@ impl Drive {
         })
     }
     pub fn metadata(&self, id: NodeId) -> Result<Node> {
-        self.database.try_transaction(|c| {
+        self.transaction(|c| {
             self.authorize(c, false)?;
             node(c, id)
         })
@@ -318,7 +369,7 @@ impl Drive {
     /// Concurrent changes between pages may add, delete, or move entries.
     pub fn list(&self, parent: NodeId, after: Option<NodeId>, limit: usize) -> Result<Page> {
         validate_limit(limit, MAX_PAGE)?;
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             self.authorize(tx, false)?;
             directory(tx, parent)?;
             let mut stmt = tx.prepare(
@@ -346,7 +397,7 @@ impl Drive {
     /// but their existence never authorizes an operation with a different actor/payload.
     pub fn request_status(&self, request_id: &str) -> Result<RequestStatus> {
         validate_identity(request_id, "request_id", 128)?;
-        self.database.try_transaction(|c| {
+        self.transaction(|c| {
             self.authorize(c, false)?;
             let result = c
                 .query_row(
@@ -382,7 +433,7 @@ impl Drive {
             .iter()
             .map(|b| format!("{b:02x}"))
             .collect();
-        self.database.try_transaction(|tx| {
+        self.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()? {
@@ -450,7 +501,7 @@ impl Drive {
         length: usize,
     ) -> Result<ReadChunk> {
         validate_limit(length, MAX_READ_BYTES)?;
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             self.authorize(tx, false)?;
             accepting(tx)?;
             let selected = node(tx, id)?;
@@ -520,7 +571,7 @@ impl Drive {
     ) -> Result<SearchPage> {
         validate_identity(query, "query", 256)?;
         validate_limit(budget, MAX_SEARCH_NODES)?;
-        self.database.try_transaction(|tx| {
+        self.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")?;
@@ -552,7 +603,7 @@ impl Drive {
     /// New UUIDs may sort before the cursor, so future sweeps start at the beginning.
     pub fn collect(&self, after: Option<&str>, limit: usize) -> Result<CollectionPage> {
         validate_limit(limit, MAX_PAGE)?;
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             accepting(tx)?;
             let keys = self.blobs.list(after, limit + 1)?;
             let more = keys.len() > limit;
@@ -578,7 +629,7 @@ impl Drive {
     }
     /// Commits a persistent tombstone rejecting new operations. Receipt reads/replays never authorize new mutations.
     pub fn fence(&self) -> Result<()> {
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             tx.execute("UPDATE drive_state SET deleting=1 WHERE singleton=1", [])?;
             Ok(())
         })
@@ -588,7 +639,7 @@ impl Drive {
     /// A Host lifecycle operation distinct from ordinary folder deletion; never processes other Workspaces' storage roots.
     pub fn purge(&self) -> Result<()> {
         self.fence()?;
-        self.database.try_transaction(|tx| {
+        self.transaction(|tx| {
             while self.blobs.cleanup_staging(MAX_PAGE)? > 0 {}
             loop {
                 let keys = self.blobs.list(None, MAX_PAGE)?;

crates/workspace-drive/tests/catalog_authority.rs

@@ -0,0 +1,347 @@
+//! Storage-owner/Drive integration through production `open_host_catalog` and
+//! `open_with_managed_workspace_authority` on temporary SQLite files. Catalog
+//! domain preparation and the authorizer are reduced to persisted grant tables
+//! and queries. This does not execute Server factories, schema migrations, HTTP
+//! authentication, or Worker liveness checks. Requests reuse the cached Drive
+//! connection without reopening the catalog to hide stale state.
+//! Run with `cargo test -p workspace-drive --test catalog_authority`.
+
+use feature_storage::{FeatureDatabase, FeatureStorage};
+use rusqlite::Connection;
+use std::fs::{File, OpenOptions, TryLockError};
+use std::path::{Path, PathBuf};
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, AtomicUsize, Ordering},
+};
+use workspace_drive::{Drive, Error, Mutation, RequestStatus};
+
+struct CatalogDrive {
+    catalog: FeatureDatabase,
+    path: PathBuf,
+    lease_path: PathBuf,
+    drive: Drive,
+    _temp: tempfile::TempDir,
+}
+
+impl CatalogDrive {
+    fn new() -> Self {
+        let temp = tempfile::tempdir().unwrap();
+        let path = temp.path().join("server.db");
+        let owner = FeatureStorage::for_server_database(&path).unwrap();
+        let catalog = owner.open_host_catalog::<Error>(
+            &path, "catalog-test", "Host configured isolated test catalog", |conn| {
+                conn.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY, state TEXT NOT NULL);
+                    INSERT INTO workspaces VALUES('drive-ws', 'active');
+                    CREATE TABLE workspace_drive_grants(workspace_id TEXT, runtime_id TEXT, worker_id TEXT, access TEXT, revoked INTEGER);
+                    INSERT INTO workspace_drive_grants VALUES('drive-ws','runtime','worker','read_write',0);")?;
+                Ok(())
+            }
+        ).unwrap();
+        let storage = owner.workspace("drive-ws").unwrap();
+        let registration = Drive::register(&storage).unwrap();
+        let root = temp.path().join("blobs");
+        let drive =
+            Drive::open_with_managed_workspace_authority(&storage, &registration, &root, &catalog)
+                .unwrap();
+        let drive = drive.clone();
+        let lease_path = temp
+            .path()
+            .join("feature-storage/.lifecycle/_host_catalog.lock");
+        Self {
+            catalog,
+            path,
+            lease_path,
+            drive,
+            _temp: temp,
+        }
+    }
+
+    fn grant_bound(&self, callbacks: Arc<AtomicUsize>) -> Drive {
+        self.drive.with_authorizer(move |conn, write| {
+            callbacks.fetch_add(1, Ordering::SeqCst);
+            let granted: bool = conn.query_row(
+                "SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspace_drive_grants
+                 WHERE workspace_id='drive-ws' AND runtime_id='runtime' AND worker_id='worker'
+                 AND revoked=0 AND (?1=0 OR access='read_write'))",
+                [write],
+                |row| row.get(0),
+            )?;
+            if !granted {
+                return Err(Error::Denied);
+            }
+            Ok(())
+        })
+    }
+}
+
+fn catalog_lock(path: &Path) -> File {
+    OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(path)
+        .unwrap()
+}
+
+#[test]
+fn cached_drive_valid_catalog_allows_grants_and_rechecks_revocation() {
+    let fixture = CatalogDrive::new();
+    let callbacks = Arc::new(AtomicUsize::new(0));
+    let drive = fixture.grant_bound(callbacks.clone());
+    let parent = drive.root().unwrap().id;
+    let committed = drive
+        .mutate(
+            "valid",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "allowed".into(),
+            },
+        )
+        .unwrap();
+    assert_eq!(drive.metadata(committed.node.id).unwrap(), committed.node);
+    assert!(matches!(
+        drive.request_status("valid").unwrap(),
+        RequestStatus::Committed { .. }
+    ));
+    fixture
+        .catalog
+        .try_with_connection::<_, Error>(|conn| {
+            conn.execute("UPDATE workspace_drive_grants SET revoked=1", [])?;
+            Ok(())
+        })
+        .unwrap();
+    assert!(matches!(drive.root(), Err(Error::Denied)));
+    assert_eq!(callbacks.load(Ordering::SeqCst), 5);
+}
+
+#[test]
+fn cached_drive_missing_catalog_refuses_before_stale_grant_callback() {
+    assert_cached_catalog_refuses_stale_grants(Replacement::Missing);
+}
+
+#[test]
+fn cached_drive_different_database_at_same_path_refuses_before_stale_grant_callback() {
+    assert_cached_catalog_refuses_stale_grants(Replacement::OtherDatabase);
+}
+
+#[test]
+fn cached_drive_same_path_copied_catalog_refuses_before_stale_grant_callback() {
+    assert_cached_catalog_refuses_stale_grants(Replacement::CopiedIdentity);
+}
+
+enum Replacement {
+    Missing,
+    OtherDatabase,
+    CopiedIdentity,
+}
+
+fn assert_cached_catalog_refuses_stale_grants(replacement: Replacement) {
+    let fixture = CatalogDrive::new();
+    let callbacks = Arc::new(AtomicUsize::new(0));
+    let drive = fixture.grant_bound(callbacks.clone());
+    let parent = drive.root().unwrap().id;
+    drive
+        .mutate(
+            "before",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "before".into(),
+            },
+        )
+        .unwrap();
+    let before = callbacks.load(Ordering::SeqCst);
+    let copy = fixture.path.with_extension("copy");
+    if matches!(replacement, Replacement::CopiedIdentity) {
+        fixture
+            .catalog
+            .try_with_connection::<_, Error>(|conn| {
+                let mut target = Connection::open(&copy)?;
+                rusqlite::backup::Backup::new(conn, &mut target)?.run_to_completion(
+                    128,
+                    std::time::Duration::from_millis(1),
+                    None,
+                )?;
+                // Preserve UUID and grants to expose physical replacement even
+                // when a path/identity-row-only check would accept the copy.
+                let original: String = conn.query_row(
+                    "SELECT physical FROM __yoi_storage_identity WHERE singleton=1",
+                    [],
+                    |row| row.get(0),
+                )?;
+                let copied: String = target.query_row(
+                    "SELECT physical FROM __yoi_storage_identity WHERE singleton=1",
+                    [],
+                    |row| row.get(0),
+                )?;
+                assert_eq!(original, copied);
+                let grants: i64 = target.query_row(
+                    "SELECT count(*) FROM workspace_drive_grants WHERE revoked=0",
+                    [],
+                    |row| row.get(0),
+                )?;
+                assert_eq!(grants, 1);
+                Ok(())
+            })
+            .unwrap();
+    }
+    std::fs::rename(&fixture.path, fixture.path.with_extension("retired")).unwrap();
+    match replacement {
+        Replacement::Missing => {}
+        Replacement::CopiedIdentity => std::fs::rename(&copy, &fixture.path).unwrap(),
+        Replacement::OtherDatabase => {
+            let other = Connection::open(&fixture.path).unwrap();
+            other
+                .execute_batch(
+                    "CREATE TABLE unrelated(value TEXT); INSERT INTO unrelated VALUES('retain');",
+                )
+                .unwrap();
+        }
+    }
+    // Contrast the owner's managed rejection with the cached consumer.
+    // No fresh catalog open occurs before these Drive requests.
+    let entered = std::cell::Cell::new(false);
+    assert!(
+        fixture
+            .catalog
+            .try_with_connection::<_, Error>(|_| {
+                entered.set(true);
+                Ok(())
+            })
+            .is_err()
+    );
+    assert!(!entered.get());
+    assert!(matches!(drive.root(), Err(Error::Storage(_))));
+    assert!(matches!(
+        drive.mutate(
+            "after",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "forbidden".into(),
+            }
+        ),
+        Err(Error::Storage(_))
+    ));
+    assert!(matches!(
+        drive.request_status("before"),
+        Err(Error::Storage(_))
+    ));
+    assert_eq!(
+        callbacks.load(Ordering::SeqCst),
+        before,
+        "retired grants must not be consulted"
+    );
+    assert_eq!(
+        fixture.path.exists(),
+        !matches!(replacement, Replacement::Missing),
+        "a missing catalog must not be recreated"
+    );
+    if matches!(replacement, Replacement::OtherDatabase) {
+        let other = Connection::open(&fixture.path).unwrap();
+        assert_eq!(
+            other
+                .query_row("SELECT value FROM unrelated", [], |row| row
+                    .get::<_, String>(0))
+                .unwrap(),
+            "retain"
+        );
+        let identity_added: bool = other
+            .query_row(
+                "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='__yoi_storage_identity')",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert!(!identity_added, "rejected replacement must not be adopted");
+    }
+}
+
+#[test]
+fn cached_drive_catalog_lease_is_held_in_commit_and_rollback_but_not_while_idle() {
+    for rollback in [false, true] {
+        let fixture = CatalogDrive::new();
+        let parent = fixture.drive.root().unwrap().id;
+        let idle_lock = catalog_lock(&fixture.lease_path);
+        idle_lock.try_lock().unwrap();
+        idle_lock.unlock().unwrap();
+        let after_authorizer = Arc::new(AtomicBool::new(false));
+        let probe = after_authorizer.clone();
+        let lease_path = fixture.lease_path.clone();
+        let catalog = fixture.catalog.clone();
+        let drive = fixture.drive.with_authorizer(move |conn, _| {
+            let lock = catalog_lock(&lease_path);
+            assert!(
+                matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                "cached transaction must hold catalog lease"
+            );
+            // Only the lifecycle lease is retained, not the catalog mutex.
+            catalog
+                .try_with_connection::<_, Error>(|catalog_conn| {
+                    let count: i64 = catalog_conn.query_row(
+                        "SELECT count(*) FROM workspace_drive_grants",
+                        [],
+                        |row| row.get(0),
+                    )?;
+                    assert_eq!(count, 1);
+                    Ok(())
+                })
+                .unwrap();
+            // Test-only hooks observe SQLite transaction exit rather than a
+            // callback/SQL effect that merely precedes COMMIT or ROLLBACK.
+            let lease_path = lease_path.clone();
+            let probe = probe.clone();
+            if rollback {
+                conn.rollback_hook(Some(move || {
+                    let lock = catalog_lock(&lease_path);
+                    assert!(
+                        matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                        "lease must span SQLite rollback"
+                    );
+                    probe.store(true, Ordering::SeqCst);
+                }));
+                conn.execute("UPDATE drive_state SET deleting=1 WHERE singleton=1", [])?;
+                return Err(Error::Denied);
+            }
+            conn.commit_hook(Some(move || {
+                let lock = catalog_lock(&lease_path);
+                assert!(
+                    matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                    "lease must span SQLite commit"
+                );
+                probe.store(true, Ordering::SeqCst);
+                false
+            }));
+            Ok(())
+        });
+        let result = drive.mutate(
+            "lease-effect",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "transaction-effect".into(),
+            },
+        );
+        if rollback {
+            assert!(matches!(result, Err(Error::Denied)));
+        } else {
+            result.unwrap();
+        }
+        assert!(after_authorizer.load(Ordering::SeqCst));
+        idle_lock.try_lock().unwrap();
+        idle_lock.unlock().unwrap();
+        fixture.drive.check_authorized(true).unwrap();
+        assert_eq!(
+            fixture.drive.list(parent, None, 10).unwrap().nodes.len(),
+            usize::from(!rollback)
+        );
+        assert!(matches!(
+            (
+                fixture.drive.request_status("lease-effect").unwrap(),
+                rollback
+            ),
+            (RequestStatus::Uncommitted, true) | (RequestStatus::Committed { .. }, false)
+        ));
+    }
+}

crates/workspace-server/Cargo.toml

@@ -66,6 +66,7 @@ webauthn-rs = { workspace = true, features = ["conditional-ui"] }
 webauthn-rs-proto = "0.5.5"
 
 [dev-dependencies]
+rusqlite = { workspace = true, features = ["hooks"] }
 feature-storage = { workspace = true, features = ["test-support"] }
 native-tls = "0.2"
 tokio-native-tls = "0.3"

crates/workspace-server/src/server.rs

@@ -4243,7 +4243,8 @@ impl WorkspaceApi {
             .open(
                 &self.feature_storage,
                 self.config.drive_blob_root()?,
-                self.config.database_path.clone(),
+                self.config_store
+                    .drive_authority(&self.config.database_path)?,
             )
             .await
     }
@@ -4253,7 +4254,8 @@ impl WorkspaceApi {
             .fence(
                 &self.feature_storage,
                 self.config.drive_blob_root()?,
-                self.config.database_path.clone(),
+                self.config_store
+                    .drive_authority(&self.config.database_path)?,
             )
             .await
     }
@@ -4263,7 +4265,8 @@ impl WorkspaceApi {
             .purge(
                 &self.feature_storage,
                 self.config.drive_blob_root()?,
-                self.config.database_path.clone(),
+                self.config_store
+                    .drive_authority(&self.config.database_path)?,
             )
             .await
     }

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

@@ -175,7 +175,9 @@ async fn deleting_workspace_reopens_fenced_drive_and_purges_blobs_before_feature
         .open(
             &api.feature_storage,
             root.clone(),
-            config.database_path.clone(),
+            api.config_store
+                .drive_authority(&config.database_path)
+                .unwrap(),
         )
         .await
         .unwrap();

crates/workspace-server/src/store.rs

@@ -2906,6 +2906,28 @@ impl SqliteWorkspaceStore {
         self.db_executor.clone()
     }
 
+    pub(crate) fn drive_authority(
+        &self,
+        fixture_path: &Path,
+    ) -> Result<crate::workspace_drive_host::DriveAuthority> {
+        match &self.conn {
+            StoreConnection::Managed(database) => Ok(
+                crate::workspace_drive_host::DriveAuthority::Managed(database.clone()),
+            ),
+            #[cfg(test)]
+            StoreConnection::Fixture(_) => Ok(
+                crate::workspace_drive_host::DriveAuthority::Fixture(fixture_path.to_owned()),
+            ),
+            #[cfg(not(test))]
+            StoreConnection::Fixture(_) => {
+                let _ = fixture_path;
+                Err(Error::Store(
+                    "Drive requires a managed Server catalog".into(),
+                ))
+            }
+        }
+    }
+
     pub(crate) fn with_conn<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
         match &self.conn {
             StoreConnection::Managed(database) => database.try_with_connection(f),

crates/workspace-server/src/workspace_drive_catalog_tests.rs

@@ -0,0 +1,321 @@
+//! Integration of the production catalog owner, WorkspaceDriveHost, and cached
+//! Drive transactions. Only the request callback is reduced to a real persisted
+//! grant query; this does not certify HTTP authentication or Worker liveness.
+//! The minimal path-only fixtures in workspace_drive_host are deliberately not
+//! used here. All storage and lock probes are confined to each temporary root.
+
+use super::*;
+use crate::SqliteWorkspaceStore;
+use rusqlite::Connection;
+use std::fs::{File, OpenOptions, TryLockError};
+use std::sync::{Arc, atomic::AtomicUsize};
+use workspace_drive::{Mutation, RequestStatus};
+
+struct CatalogDrive {
+    catalog: SqliteWorkspaceStore,
+    path: PathBuf,
+    lease_path: PathBuf,
+    drive: Drive,
+    _temp: tempfile::TempDir,
+}
+
+impl CatalogDrive {
+    async fn new() -> Self {
+        let temp = tempfile::tempdir().unwrap();
+        let path = temp.path().join("server.db");
+        // SqliteWorkspaceStore::open, not from_connection/connect_prepared_fixture:
+        // the expected identity and _host_catalog lease are production owners.
+        let catalog = SqliteWorkspaceStore::open(&path).unwrap();
+        catalog.with_conn(|conn| {
+            conn.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at)
+                VALUES('drive-owner','user','drive-owner','Owner','1','1');
+                INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at)
+                VALUES('drive-ws','drive-owner','Workspace','active','1','1');
+                INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at)
+                VALUES('drive-ws','runtime','worker','Worker','normal','1','1');
+                INSERT INTO workspace_drive_grants(workspace_id,runtime_id,worker_id,access,revoked,created_by,created_at)
+                VALUES('drive-ws','runtime','worker','read_write',0,'drive-owner','1');")?;
+            Ok(())
+        }).unwrap();
+        let storage = crate::FeatureStorage::for_server_database(&path)
+            .unwrap()
+            .workspace("drive-ws")
+            .unwrap();
+        let root = drive_blob_root(temp.path(), "drive-ws").unwrap();
+        let host = WorkspaceDriveHost::new(&storage).await.unwrap();
+        let authority = catalog.drive_authority(&path).unwrap();
+        assert!(matches!(&authority, DriveAuthority::Managed(_)));
+        let drive = host
+            .open(&storage, root.clone(), authority.clone())
+            .await
+            .unwrap();
+        // Return the cached handle, not merely the initially opened connection.
+        drop(drive);
+        let drive = host.open(&storage, root, authority).await.unwrap();
+        let lease_path = temp
+            .path()
+            .join("feature-storage/.lifecycle/_host_catalog.lock");
+        Self {
+            _temp: temp,
+            catalog,
+            path,
+            lease_path,
+            drive,
+        }
+    }
+
+    fn grant_bound(&self, callbacks: Arc<AtomicUsize>) -> Drive {
+        self.drive.with_authorizer(move |conn, write| {
+            callbacks.fetch_add(1, Ordering::SeqCst);
+            let granted: bool = conn.query_row(
+                "SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspace_drive_grants
+                 WHERE workspace_id='drive-ws' AND runtime_id='runtime' AND worker_id='worker'
+                 AND revoked=0 AND (?1=0 OR access='read_write'))",
+                [write],
+                |row| row.get(0),
+            )?;
+            if !granted {
+                return Err(workspace_drive::Error::Denied);
+            }
+            Ok(())
+        })
+    }
+}
+
+fn catalog_lock(path: &Path) -> File {
+    OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(path)
+        .unwrap()
+}
+
+#[tokio::test]
+async fn cached_drive_valid_catalog_allows_grants_and_rechecks_revocation() {
+    let fixture = CatalogDrive::new().await;
+    let callbacks = Arc::new(AtomicUsize::new(0));
+    let drive = fixture.grant_bound(callbacks.clone());
+    let parent = drive.root().unwrap().id;
+    let committed = drive
+        .mutate(
+            "valid",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "allowed".into(),
+            },
+        )
+        .unwrap();
+    assert_eq!(drive.metadata(committed.node.id).unwrap(), committed.node);
+    assert!(matches!(
+        drive.request_status("valid").unwrap(),
+        RequestStatus::Committed { .. }
+    ));
+    fixture.catalog.with_conn(|conn| {
+        conn.execute("UPDATE workspace_drive_grants SET revoked=1, revoked_by='drive-owner', revoked_at='2'", [])?;
+        Ok(())
+    }).unwrap();
+    assert!(matches!(drive.root(), Err(workspace_drive::Error::Denied)));
+    assert_eq!(callbacks.load(Ordering::SeqCst), 5);
+}
+
+#[tokio::test]
+async fn cached_drive_missing_catalog_refuses_before_stale_grant_callback() {
+    assert_cached_catalog_refuses_stale_grants(false).await;
+}
+
+#[tokio::test]
+async fn cached_drive_same_path_copied_catalog_refuses_before_stale_grant_callback() {
+    assert_cached_catalog_refuses_stale_grants(true).await;
+}
+
+async fn assert_cached_catalog_refuses_stale_grants(replacement: bool) {
+    let fixture = CatalogDrive::new().await;
+    let callbacks = Arc::new(AtomicUsize::new(0));
+    let drive = fixture.grant_bound(callbacks.clone());
+    let parent = drive.root().unwrap().id;
+    drive
+        .mutate(
+            "before",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "before".into(),
+            },
+        )
+        .unwrap();
+    let before = callbacks.load(Ordering::SeqCst);
+    let copy = fixture.path.with_extension("copy");
+    if replacement {
+        fixture
+            .catalog
+            .with_conn(|conn| {
+                let mut target = Connection::open(&copy)?;
+                rusqlite::backup::Backup::new(conn, &mut target)?.run_to_completion(
+                    128,
+                    std::time::Duration::from_millis(1),
+                    None,
+                )?;
+                // Copy retains the UUID and grants: path/identity-row equality
+                // alone cannot distinguish this from the expected physical DB.
+                let original: String = conn.query_row(
+                    "SELECT physical FROM __yoi_storage_identity WHERE singleton=1",
+                    [],
+                    |row| row.get(0),
+                )?;
+                let copied: String = target.query_row(
+                    "SELECT physical FROM __yoi_storage_identity WHERE singleton=1",
+                    [],
+                    |row| row.get(0),
+                )?;
+                assert_eq!(original, copied);
+                let grants: i64 = target.query_row(
+                    "SELECT count(*) FROM workspace_drive_grants WHERE revoked=0",
+                    [],
+                    |row| row.get(0),
+                )?;
+                assert_eq!(grants, 1);
+                Ok(())
+            })
+            .unwrap();
+    }
+    std::fs::rename(&fixture.path, fixture.path.with_extension("retired")).unwrap();
+    if replacement {
+        std::fs::rename(&copy, &fixture.path).unwrap();
+    }
+    // Contrast the catalog repository's managed rejection with its cached
+    // consumer. No fresh WorkspaceApi lookup/owner open occurs here.
+    let entered = std::cell::Cell::new(false);
+    assert!(
+        fixture
+            .catalog
+            .with_conn(|_| {
+                entered.set(true);
+                Ok(())
+            })
+            .is_err()
+    );
+    assert!(!entered.get());
+    assert!(matches!(
+        drive.root(),
+        Err(workspace_drive::Error::Storage(_))
+    ));
+    assert!(matches!(
+        drive.mutate(
+            "after",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "forbidden".into(),
+            }
+        ),
+        Err(workspace_drive::Error::Storage(_))
+    ));
+    assert!(matches!(
+        drive.request_status("before"),
+        Err(workspace_drive::Error::Storage(_))
+    ));
+    assert_eq!(
+        callbacks.load(Ordering::SeqCst),
+        before,
+        "retired grants must not be consulted"
+    );
+    assert_eq!(
+        fixture.path.exists(),
+        replacement,
+        "a missing catalog must not be recreated"
+    );
+}
+
+#[tokio::test]
+async fn cached_drive_catalog_lease_is_held_in_commit_and_rollback_but_not_while_idle() {
+    for rollback in [false, true] {
+        let fixture = CatalogDrive::new().await;
+        let parent = fixture.drive.root().unwrap().id;
+        let idle_lock = catalog_lock(&fixture.lease_path);
+        idle_lock.try_lock().unwrap();
+        idle_lock.unlock().unwrap();
+        let after_authorizer = Arc::new(AtomicBool::new(false));
+        let probe = after_authorizer.clone();
+        let lease_path = fixture.lease_path.clone();
+        let catalog = fixture.catalog.clone();
+        let drive = fixture.drive.with_authorizer(move |conn, _| {
+            let lock = catalog_lock(&lease_path);
+            assert!(
+                matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                "cached transaction must hold catalog lease"
+            );
+            // Only the lease, not the catalog connection mutex, is retained.
+            // This also exercises the documented order against a catalog read.
+            catalog
+                .with_conn(|catalog_conn| {
+                    let count: i64 = catalog_conn.query_row(
+                        "SELECT count(*) FROM workspace_drive_grants",
+                        [],
+                        |row| row.get(0),
+                    )?;
+                    assert_eq!(count, 1);
+                    Ok(())
+                })
+                .unwrap();
+            // Observe actual SQLite transaction exit, not only the callback or
+            // a SQL effect preceding COMMIT. Hooks are test-only dependency features.
+            let lease_path = lease_path.clone();
+            let probe = probe.clone();
+            if rollback {
+                conn.rollback_hook(Some(move || {
+                    let lock = catalog_lock(&lease_path);
+                    assert!(
+                        matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                        "lease must span SQLite rollback"
+                    );
+                    probe.store(true, Ordering::SeqCst);
+                }));
+                // Make a real metadata change first, then deny it. Transaction
+                // drop must roll it back while still retaining the catalog lease.
+                conn.execute("UPDATE drive_state SET deleting=1 WHERE singleton=1", [])?;
+                return Err(workspace_drive::Error::Denied);
+            }
+            conn.commit_hook(Some(move || {
+                let lock = catalog_lock(&lease_path);
+                assert!(
+                    matches!(lock.try_lock(), Err(TryLockError::WouldBlock)),
+                    "lease must span SQLite commit"
+                );
+                probe.store(true, Ordering::SeqCst);
+                false
+            }));
+            Ok(())
+        });
+        let result = drive.mutate(
+            "lease-effect",
+            "owner",
+            Mutation::CreateFolder {
+                parent,
+                name: "transaction-effect".into(),
+            },
+        );
+        if rollback {
+            assert!(matches!(result, Err(workspace_drive::Error::Denied)));
+        } else {
+            result.unwrap();
+        }
+        assert!(after_authorizer.load(Ordering::SeqCst));
+        // The transaction has committed or rolled back, not just returned from
+        // its authorizer. No lease is retained across idle cached-handle time.
+        idle_lock.try_lock().unwrap();
+        idle_lock.unlock().unwrap();
+        fixture.drive.check_authorized(true).unwrap();
+        assert_eq!(
+            fixture.drive.list(parent, None, 10).unwrap().nodes.len(),
+            usize::from(!rollback)
+        );
+        assert!(matches!(
+            (
+                fixture.drive.request_status("lease-effect").unwrap(),
+                rollback
+            ),
+            (RequestStatus::Uncommitted, true) | (RequestStatus::Committed { .. }, false)
+        ));
+    }
+}

crates/workspace-server/src/workspace_drive_host.rs

@@ -54,6 +54,14 @@ fn reject_symlink_ancestors(path: &Path) -> Result<()> {
     Ok(())
 }
 
+#[derive(Clone)]
+pub(crate) enum DriveAuthority {
+    Managed(feature_storage::FeatureDatabase),
+    // Minimal legacy fixtures are not evidence of production catalog ownership.
+    #[cfg(test)]
+    Fixture(PathBuf),
+}
+
 pub(crate) struct WorkspaceDriveHost {
     registration: crate::RegisteredFeature,
     purged: AtomicBool,
@@ -88,7 +96,7 @@ impl WorkspaceDriveHost {
         &self,
         storage: &WorkspaceFeatureStorage,
         root: PathBuf,
-        authority_database: PathBuf,
+        authority_database: DriveAuthority,
     ) -> Result<Drive> {
         let mut cached = self.drive.lock().await;
         if let Some(drive) = cached.as_ref() {
@@ -102,12 +110,20 @@ impl WorkspaceDriveHost {
                 // 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(
+                match authority_database {
-                    &storage,
+                    DriveAuthority::Managed(catalog) => {
-                    &registration,
+                        Drive::open_with_managed_workspace_authority(
-                    &root,
+                            &storage,
-                    &authority_database,
+                            &registration,
-                )
+                            &root,
+                            &catalog,
+                        )
+                    }
+                    #[cfg(test)]
+                    DriveAuthority::Fixture(path) => {
+                        Drive::open_with_workspace_authority(&storage, &registration, &root, &path)
+                    }
+                }
                 .map_err(Error::from)
             })
             .await?;
@@ -119,7 +135,7 @@ impl WorkspaceDriveHost {
         &self,
         storage: &WorkspaceFeatureStorage,
         root: PathBuf,
-        authority_database: PathBuf,
+        authority_database: DriveAuthority,
     ) -> Result<()> {
         if self.purged.load(Ordering::Acquire) {
             return Ok(());
@@ -134,7 +150,7 @@ impl WorkspaceDriveHost {
         &self,
         storage: &WorkspaceFeatureStorage,
         root: PathBuf,
-        authority_database: PathBuf,
+        authority_database: DriveAuthority,
     ) -> Result<()> {
         if self.purged.load(Ordering::Acquire) {
             return Ok(());
@@ -161,12 +177,16 @@ impl WorkspaceDriveHost {
     }
 }
 
+#[cfg(test)]
+#[path = "workspace_drive_catalog_tests.rs"]
+mod catalog_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use workspace_drive::Mutation;
 
-    fn seed_authority(data_root: &Path) -> PathBuf {
+    fn seed_authority(data_root: &Path) -> DriveAuthority {
         let path = data_root.join("server.db");
         let conn = rusqlite::Connection::open(&path).unwrap();
         conn.execute_batch(
@@ -174,7 +194,7 @@ mod tests {
             INSERT INTO workspaces VALUES ('workspace-a', 'active');",
         )
         .unwrap();
-        path
+        DriveAuthority::Fixture(path)
     }
 
     #[test]