version bump

- Durable scratch/staging publication and preservation on failure.
      - Transactional pending-operation supersession.
      - Conflict copies cannot overwrite older copies or lose the source.
      - Single-entry wipe protection and incomplete-scan aborts.
      - Failed local deletion retains its baseline.
      - Downloads are fsynced before baseline settlement.
      - Future database schemas are refused.
      - Event cursors advance only after successful invalidation and persistence.
      - IPC frame, timeout, and concurrency limits.
      - Atomic config saves; malformed configs are preserved.
      - Sensitive directories require correct ownership, real directories, and 0700.
      - Debian package now includes service and tray autostart files.
      - Release workflow enforces versions, tests, lint, formatting, and offline acceptance.
      - CI now runs offline filesystem acceptance.

.github/workflows/ci.yml

@@ -44,6 +44,9 @@ jobs:
       - name: Unit tests
         run: cargo test --workspace --locked
 
+      - name: Account-free filesystem acceptance
+        run: scripts/fuse-acceptance.sh --offline-only
+
       - name: Acceptance runner safety guard
         run: |
           if scripts/fuse-acceptance.sh /tmp >acceptance.out 2>&1; then

.github/workflows/release.yml

@@ -35,11 +35,29 @@ jobs:
       - name: Set up Rust
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: stable
+          toolchain: '1.96'
+          components: rustfmt, clippy
 
       - name: Cache Cargo target
         uses: swatinem/rust-cache@v2
 
+      - name: Verify tag and manifest versions
+        if: startsWith(github.ref, 'refs/tags/')
+        shell: bash
+        run: |
+          tag_version="${GITHUB_REF_NAME#v}"
+          manifest_version="$(sed -n 's/^version = "\([^"]*\)"/\1/p' Cargo.toml | head -n1)"
+          pkgbuild_version="$(sed -n 's/^pkgver=//p' packaging/PKGBUILD | head -n1)"
+          test "$tag_version" = "$manifest_version"
+          test "$tag_version" = "$pkgbuild_version"
+
+      - name: Release quality gates
+        run: |
+          cargo fmt --all -- --check
+          cargo clippy --workspace --all-targets -- -D warnings
+          cargo test --workspace --locked
+          scripts/fuse-acceptance.sh --offline-only
+
       - name: Build release binaries
         run: |
           cargo build --release --locked
@@ -76,6 +94,8 @@ jobs:
           mkdir -p deb/usr/share/applications
           mkdir -p deb/usr/share/icons/hicolor/scalable/apps
           mkdir -p deb/usr/share/licenses/proton-drive-linux
+          mkdir -p deb/usr/lib/systemd/user
+          mkdir -p deb/etc/xdg/autostart
 
           # Copy binaries
           cp target/release/pdfs deb/usr/bin/
@@ -85,7 +105,9 @@ jobs:
 
           # Copy desktop integration files
           cp packaging/io.narl.proton-drive-linux.desktop deb/usr/share/applications/
+          cp packaging/io.narl.proton-drive-linux-tray.desktop deb/etc/xdg/autostart/
           cp packaging/io.narl.proton-drive-linux.svg deb/usr/share/icons/hicolor/scalable/apps/
+          cp packaging/proton-drive.service deb/usr/lib/systemd/user/
           cp LICENSE deb/usr/share/licenses/proton-drive-linux/
 
           # Calculate Installed-Size in KB for the control file

Cargo.lock

@@ -3518,7 +3518,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
 
 [[package]]
 name = "pdfs-cli"
-version = "0.7.1"
+version = "1.0.0"
 dependencies = [
  "anyhow",
  "clap 4.6.2",
@@ -3533,11 +3533,12 @@ dependencies = [
 
 [[package]]
 name = "pdfs-core"
-version = "0.7.1"
+version = "1.0.0"
 dependencies = [
  "directories",
  "ignore",
  "keyring",
+ "libc",
  "parking_lot",
  "proton-drive-rs",
  "proton-sdk",
@@ -3552,7 +3553,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-fuse"
-version = "0.7.1"
+version = "1.0.0"
 dependencies = [
  "fuser",
  "image",
@@ -3568,7 +3569,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-gui"
-version = "0.7.1"
+version = "1.0.0"
 dependencies = [
  "anyhow",
  "async-channel",

Cargo.toml

@@ -3,7 +3,7 @@ resolver = "2"
 members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
 
 [workspace.package]
-version = "0.7.1"
+version = "1.0.0"
 edition = "2024"
 license = "MIT"
 authors = ["Nils Pukropp <contact@narl.io>"]

crates/pdfs-core/Cargo.toml

@@ -21,3 +21,4 @@ sha2.workspace = true
 rusqlite.workspace = true
 keyring.workspace = true
 parking_lot.workspace = true
+libc.workspace = true

crates/pdfs-core/src/cache.rs

@@ -241,7 +241,9 @@ impl ContentCache {
         let recovery_dir = content_dir.join("recovery");
         std::fs::create_dir_all(&recovery_dir)?;
         Self::rescue_scratch(&scratch_dir, &recovery_dir);
-        let _ = std::fs::remove_dir_all(&scratch_dir);
+        // A failed rescue may leave the only durable copy here. Never clear the
+        // directory wholesale; discard only blobs that carry no fsync marker.
+        Self::discard_unmarked_scratch(&scratch_dir);
         std::fs::create_dir_all(&scratch_dir)?;
         // Staging, by contrast, is deliberately preserved across runs.
         let staging_dir = content_dir.join("staging");
@@ -669,18 +671,24 @@ impl ContentCache {
     pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> {
         use std::sync::atomic::{AtomicU64, Ordering};
         static N: AtomicU64 = AtomicU64::new(0);
-        let path = self.scratch_dir.join(format!(
-            "w-{}-{}",
-            std::process::id(),
-            N.fetch_add(1, Ordering::Relaxed)
-        ));
-        let file = std::fs::OpenOptions::new()
-            .create(true)
-            .truncate(true)
-            .read(true)
-            .write(true)
-            .open(&path)?;
-        Ok((file, path))
+        loop {
+            let path = self.scratch_dir.join(format!(
+                "w-{}-{}-{}",
+                std::process::id(),
+                now_millis(),
+                N.fetch_add(1, Ordering::Relaxed)
+            ));
+            match std::fs::OpenOptions::new()
+                .create_new(true)
+                .read(true)
+                .write(true)
+                .open(&path)
+            {
+                Ok(file) => return Ok((file, path)),
+                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
+                Err(e) => return Err(e.into()),
+            }
+        }
     }
 
     /// The sidecar path for a scratch file, holding the [`StagedWrite`] that
@@ -709,16 +717,23 @@ impl ContentCache {
     /// have synced `scratch` first: a sidecar promising bytes that never reached
     /// the disk is worse than no sidecar.
     pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> {
+        use std::io::Write as _;
+
         let side = Self::scratch_sidecar(scratch);
         let tmp = side.with_extension("json.tmp");
-        std::fs::write(&tmp, serde_json::to_vec(meta)?)?;
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(&tmp)?;
+        file.write_all(&serde_json::to_vec(meta)?)?;
+        file.sync_all()?;
+        drop(file);
         std::fs::rename(&tmp, &side)?;
         // The rename is what publishes the sidecar; fsync the directory so the
         // rename itself survives, not just the bytes it points at.
-        if let Some(dir) = side.parent()
-            && let Ok(d) = std::fs::File::open(dir)
-        {
-            let _ = d.sync_all();
+        if let Some(dir) = side.parent() {
+            std::fs::File::open(dir)?.sync_all()?;
         }
         Ok(())
     }
@@ -796,6 +811,21 @@ impl ContentCache {
         }
     }
 
+    fn discard_unmarked_scratch(scratch_dir: &Path) {
+        let Ok(entries) = std::fs::read_dir(scratch_dir) else {
+            return;
+        };
+        for entry in entries.flatten() {
+            let blob = entry.path();
+            if blob.extension().is_some_and(|e| e == "json") {
+                continue;
+            }
+            if !Self::scratch_sidecar(&blob).exists() {
+                let _ = std::fs::remove_file(blob);
+            }
+        }
+    }
+
     /// Emergency eviction of unpinned blobs when local disk runs out of space (ENOSPC).
     pub fn emergency_evict(&self) {
         let pinned: HashSet<String> = self
@@ -826,7 +856,8 @@ impl ContentCache {
 
     /// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the
     /// daemon to hand to the upload queue. A blob whose sidecar did not survive
-    /// the rescue is deleted here — nothing can address it.
+    /// is not actionable automatically, but is retained for manual recovery:
+    /// unexplained bytes are never evidence that user data is disposable.
     pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> {
         let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else {
             return Vec::new();
@@ -837,14 +868,11 @@ impl ContentCache {
             if blob.extension().is_some_and(|e| e == "json") {
                 continue;
             }
-            match std::fs::read(Self::scratch_sidecar(&blob))
+            if let Some(meta) = std::fs::read(Self::scratch_sidecar(&blob))
                 .ok()
                 .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
             {
-                Some(meta) => out.push((blob, meta)),
-                None => {
-                    let _ = std::fs::remove_file(&blob);
-                }
+                out.push((blob, meta));
             }
         }
         out
@@ -872,6 +900,8 @@ impl ContentCache {
     /// failure here means we could not save the bytes at all, so it is reported
     /// rather than swallowed.
     pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result<PathBuf> {
+        use std::io::Write as _;
+
         static N: AtomicU64 = AtomicU64::new(0);
         // The uid goes in the name so a staged file can be tied back to its node
         // without a database; `/` in a uid would otherwise open a path.
@@ -882,18 +912,32 @@ impl ContentCache {
             now_millis(),
             N.fetch_add(1, Ordering::Relaxed)
         ));
-        if let Err(e) = std::fs::rename(scratch, &path) {
-            // Cross-device: rename cannot move it, so copy and drop the original.
-            let _ = e;
+        let copied = if let Err(e) = std::fs::rename(scratch, &path) {
+            if e.kind() != std::io::ErrorKind::CrossesDevices {
+                return Err(e.into());
+            }
             std::fs::copy(scratch, &path)?;
-            let _ = std::fs::remove_file(scratch);
-        }
+            true
+        } else {
+            false
+        };
+        std::fs::File::open(&path)?.sync_all()?;
         // Sidecar last: a staged file without one is still recoverable bytes,
         // while a sidecar without a file would describe nothing.
-        std::fs::write(
-            path.with_extension("json"),
-            serde_json::to_vec_pretty(meta)?,
-        )?;
+        let side = path.with_extension("json");
+        let side_tmp = path.with_extension("json.tmp");
+        let mut side_file = std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&side_tmp)?;
+        side_file.write_all(&serde_json::to_vec_pretty(meta)?)?;
+        side_file.sync_all()?;
+        drop(side_file);
+        std::fs::rename(&side_tmp, &side)?;
+        std::fs::File::open(&self.staging_dir)?.sync_all()?;
+        if copied {
+            std::fs::remove_file(scratch)?;
+        }
         Ok(path)
     }
 
@@ -1906,6 +1950,51 @@ mod tests {
         assert!(c2.recovered_writes().is_empty(), "recovery does not repeat");
     }
 
+    #[test]
+    fn recovery_keeps_an_orphan_blob_when_its_sidecar_is_missing() {
+        let (c, dir) = cache();
+        let orphan = dir.path().join("content/recovery/orphaned-user-bytes");
+        std::fs::write(&orphan, b"retain me").unwrap();
+
+        assert!(c.recovered_writes().is_empty());
+        assert_eq!(
+            std::fs::read(&orphan).unwrap(),
+            b"retain me",
+            "metadata loss must not trigger deletion of unexplained content"
+        );
+    }
+
+    #[test]
+    fn failed_scratch_rescue_does_not_delete_the_durable_source() {
+        use std::io::Write as _;
+        let dir = TempDir::new();
+        let db = Arc::new(Db::open_in_memory().unwrap());
+        let content = dir.path().join("content");
+        let pins = dir.path().join("pins.json");
+        let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
+        let (mut file, scratch) = c.create_scratch().unwrap();
+        file.write_all(b"durable source").unwrap();
+        file.sync_all().unwrap();
+        let meta = StagedWrite {
+            uid: uid("rescue").to_string(),
+            len: 14,
+            base_size: 0,
+            base_mtime: 0,
+            authored: vec![(0, 14)],
+            complete: true,
+            based_on: None,
+        };
+        c.mark_scratch_durable(&scratch, &meta).unwrap();
+        let blocking_destination = content.join("recovery").join(scratch.file_name().unwrap());
+        std::fs::create_dir(&blocking_destination).unwrap();
+        drop(file);
+        drop(c);
+
+        let _c2 = ContentCache::open(content, pins, 0, db).unwrap();
+        assert_eq!(std::fs::read(&scratch).unwrap(), b"durable source");
+        assert!(ContentCache::scratch_sidecar(&scratch).exists());
+    }
+
     /// Releasing a write hands the bytes to staging, so the sidecar an earlier
     /// `fsync` left must not offer recovery a second, stale copy of it.
     #[test]

crates/pdfs-core/src/config.rs

@@ -105,11 +105,21 @@ impl AppDirs {
     /// Load config from disk, creating default if missing.
     pub fn load_config(&self) -> AppConfig {
         let path = self.config_path();
-        if path.exists()
-            && let Ok(content) = std::fs::read_to_string(&path)
-            && let Ok(config) = serde_json::from_str::<AppConfig>(&content)
-        {
-            return config;
+        if path.exists() {
+            match std::fs::read_to_string(&path)
+                .map_err(Error::from)
+                .and_then(|content| {
+                    serde_json::from_str::<AppConfig>(&content).map_err(Error::from)
+                }) {
+                Ok(config) => return config,
+                Err(e) => {
+                    // Preserve the original bytes for diagnosis/recovery.  In
+                    // particular, never replace a truncated config with defaults
+                    // and silently lose mount, device, or ignore settings.
+                    tracing::error!(path = %path.display(), error = %e, "config is unreadable; using in-memory defaults without overwriting it");
+                    return AppConfig::default();
+                }
+            }
         }
 
         let default_config = AppConfig::default();
@@ -125,10 +135,26 @@ impl AppDirs {
     /// cache budget or mountpoint; the daemon re-reads the file on its next mount.
     pub fn save_config(&self, config: &AppConfig) -> Result<()> {
         let path = self.config_path();
-        if let Some(parent) = path.parent() {
-            std::fs::create_dir_all(parent)?;
-        }
-        std::fs::write(&path, serde_json::to_string_pretty(config)?)?;
+        let parent = path
+            .parent()
+            .ok_or_else(|| Error::Other("configuration path has no parent".into()))?;
+        std::fs::create_dir_all(parent)?;
+        let nonce = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|duration| duration.as_nanos())
+            .unwrap_or(0);
+        let tmp = path.with_extension(format!("json.tmp-{}-{nonce}", std::process::id()));
+        use std::os::unix::fs::OpenOptionsExt as _;
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .mode(0o600)
+            .open(&tmp)?;
+        use std::io::Write as _;
+        file.write_all(&serde_json::to_vec_pretty(config)?)?;
+        file.sync_all()?;
+        std::fs::rename(&tmp, &path)?;
+        std::fs::File::open(parent)?.sync_all()?;
         Ok(())
     }
 
@@ -211,24 +237,41 @@ impl AppDirs {
     pub fn ensure(&self) -> Result<()> {
         for dir in [self.state_dir(), self.cache_dir()] {
             std::fs::create_dir_all(&dir)?;
-            restrict_dir(&dir);
+            restrict_dir(&dir)?;
         }
         let config_dir = self.dirs.config_dir().to_path_buf();
-        if std::fs::create_dir_all(&config_dir).is_ok() {
-            restrict_dir(&config_dir);
-        }
+        std::fs::create_dir_all(&config_dir)?;
+        restrict_dir(&config_dir)?;
         Ok(())
     }
 }
 
-/// Make `dir` owner-only (`0700`). Best effort: a mode we cannot set (an
-/// exotic filesystem, a directory we do not own) is not worth refusing to
-/// start over, and the caller has no better answer than continuing.
-fn restrict_dir(dir: &Path) {
-    use std::os::unix::fs::PermissionsExt;
-    if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) {
-        tracing::warn!(path = %dir.display(), error = %e, "could not restrict directory to 0700");
+/// Make `dir` owner-only (`0700`). These directories contain decrypted bytes,
+/// plaintext names, and an authenticated control socket, so failure is fatal.
+fn restrict_dir(dir: &Path) -> Result<()> {
+    use std::os::unix::fs::{MetadataExt as _, PermissionsExt};
+    let before = std::fs::symlink_metadata(dir)?;
+    if before.file_type().is_symlink() || !before.is_dir() {
+        return Err(Error::Other(format!(
+            "sensitive path is not a real directory: {}",
+            dir.display()
+        )));
     }
+    if before.uid() != unsafe { libc::geteuid() } {
+        return Err(Error::Other(format!(
+            "sensitive directory is not owned by the current user: {}",
+            dir.display()
+        )));
+    }
+    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
+    let after = std::fs::symlink_metadata(dir)?;
+    if after.permissions().mode() & 0o777 != 0o700 {
+        return Err(Error::Other(format!(
+            "sensitive directory did not become owner-only: {}",
+            dir.display()
+        )));
+    }
+    Ok(())
 }
 
 /// Make a just-bound Unix socket owner-only (`0600`).
@@ -263,29 +306,57 @@ mod tests {
     /// and the control socket that commands the daemon's session.
     #[test]
     fn restrict_dir_makes_a_directory_owner_only() {
-        let dir = std::env::temp_dir().join(format!("pdfs-cfg-{}", std::process::id()));
+        let dir = std::env::temp_dir().join(format!(
+            "pdfs-cfg-{}-{:?}",
+            std::process::id(),
+            std::thread::current().id()
+        ));
         let _ = std::fs::remove_dir_all(&dir);
         std::fs::create_dir_all(&dir).unwrap();
         // Start deliberately world-readable, as `create_dir_all` under a typical
         // umask would leave it.
         std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap();
 
-        restrict_dir(&dir);
+        restrict_dir(&dir).unwrap();
 
         assert_eq!(mode_of(&dir), 0o700, "group and other must have no access");
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    fn restrict_dir_refuses_a_symlink() {
+        let root = std::env::temp_dir().join(format!(
+            "pdfs-cfg-link-{}-{:?}",
+            std::process::id(),
+            std::thread::current().id()
+        ));
+        let target = root.join("target");
+        let link = root.join("state");
+        std::fs::create_dir_all(&target).unwrap();
+        std::os::unix::fs::symlink(&target, &link).unwrap();
+
+        assert!(restrict_dir(&link).is_err());
+
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
     /// Connecting to the control socket confers the daemon's authenticated
     /// session with no credential, so it is an authority boundary rather than
     /// merely private data.
     #[test]
     fn restrict_socket_makes_a_socket_owner_only() {
-        let dir = std::env::temp_dir().join(format!("pdfs-sock-{}", std::process::id()));
+        let dir = std::env::temp_dir().join(format!(
+            "pdfs-sock-{}-{:?}",
+            std::process::id(),
+            std::thread::current().id()
+        ));
         let _ = std::fs::remove_dir_all(&dir);
         std::fs::create_dir_all(&dir).unwrap();
         let sock = dir.join("control.sock");
-        let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();
+        // chmod semantics are identical for a socket path and a regular file;
+        // use a file so this regression also runs in sandboxes that prohibit
+        // creating Unix sockets.
+        let listener = std::fs::File::create(&sock).unwrap();
 
         restrict_socket(&sock).unwrap();
 

crates/pdfs-core/src/db/migrations.rs

@@ -31,7 +31,12 @@ impl Db {
             .and_then(|v| v.parse().ok())
             .unwrap_or(0);
 
-        if current >= SCHEMA_VERSION {
+        if current > SCHEMA_VERSION {
+            return Err(crate::Error::Other(format!(
+                "database schema {current} is newer than this build supports ({SCHEMA_VERSION}); refusing to open it to avoid corrupting user data"
+            )));
+        }
+        if current == SCHEMA_VERSION {
             return Ok(());
         }
 

crates/pdfs-core/src/db/ops.rs

@@ -118,9 +118,10 @@ pub struct AttachedBlob {
 
 impl Db {
     pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
-        let conn = self.conn.lock();
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
         let superseded: Option<String> = if op_supersedes(&op.kind) {
-            let blob: Option<String> = conn
+            let blob: Option<String> = tx
                 .query_row(
                     "SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2",
                     params![op.uid, op.kind],
@@ -128,7 +129,7 @@ impl Db {
                 )
                 .optional()?
                 .flatten();
-            conn.execute(
+            tx.execute(
                 "DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2",
                 params![op.uid, op.kind],
             )?;
@@ -136,7 +137,7 @@ impl Db {
         } else {
             None
         };
-        conn.execute(
+        tx.execute(
             "INSERT INTO pending_op
                (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at)
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
@@ -151,7 +152,9 @@ impl Db {
                 op.next_attempt_at
             ],
         )?;
-        Ok((conn.last_insert_rowid(), superseded))
+        let id = tx.last_insert_rowid();
+        tx.commit()?;
+        Ok((id, superseded))
     }
 
     /// Point a queued create at the bytes that were just written to it, returning

crates/pdfs-core/src/db/tests.rs

@@ -784,6 +784,41 @@ fn opens_and_migrates() {
     assert_eq!(version, SCHEMA_VERSION.to_string());
 }
 
+#[test]
+fn refuses_a_schema_newer_than_the_running_build() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-db-future-{}-{}.db",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    ));
+    let conn = rusqlite::Connection::open(&path).unwrap();
+    conn.execute_batch(
+        "CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT);
+         INSERT INTO sync_state VALUES ('schema_version', '9999');",
+    )
+    .unwrap();
+    drop(conn);
+
+    let err = Db::open(&path)
+        .err()
+        .expect("future schema must fail closed");
+    assert!(err.to_string().contains("newer than this build supports"));
+    let conn = rusqlite::Connection::open(&path).unwrap();
+    let version: String = conn
+        .query_row(
+            "SELECT value FROM sync_state WHERE key = 'schema_version'",
+            [],
+            |row| row.get(0),
+        )
+        .unwrap();
+    assert_eq!(version, "9999", "failed open must not rewrite the database");
+    drop(conn);
+    let _ = std::fs::remove_file(path);
+}
+
 /// A queued write whose baseline is restamped must keep everything else.
 ///
 /// The restamp happens after our *own* upload seals a new revision under a
@@ -867,6 +902,39 @@ fn a_second_write_supersedes_the_first_pending_op() {
     assert_eq!(db.pending_op_counts().unwrap().uploads, 1);
 }
 
+#[test]
+fn failed_superseding_insert_keeps_the_old_pending_op() {
+    let db = Db::open_in_memory().unwrap();
+    let op = |blob: &str| PendingOp {
+        id: 0,
+        kind: OP_REVISION.to_string(),
+        uid: uid("atomic").to_string(),
+        parent_uid: None,
+        name: None,
+        blob_path: Some(blob.to_string()),
+        meta_json: Some("{}".to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    db.enqueue_op(&op("/staging/original")).unwrap();
+    db.with_conn(|conn| {
+        conn.execute_batch(
+            "CREATE TRIGGER reject_replacement BEFORE INSERT ON pending_op
+             WHEN NEW.blob_path = '/staging/replacement'
+             BEGIN SELECT RAISE(ABORT, 'injected insert failure'); END;",
+        )?;
+        Ok(())
+    })
+    .unwrap();
+
+    assert!(db.enqueue_op(&op("/staging/replacement")).is_err());
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1);
+    assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/original"));
+}
+
 /// Deleting a folder that was created offline must take the ops queued
 /// underneath it with it: they name a placeholder parent that will now never
 /// become real, so nothing could ever drain them and nothing is left to

crates/pdfs-fuse/src/background.rs

@@ -126,12 +126,16 @@ pub(super) async fn run_event_sync(
                 match client.enumerate_events(&scope, None).await {
                     Ok(events) => {
                         let head = events.last().map(|e| e.id().clone());
-                        if let Some(c) = &head
-                            && let Err(e) = db.set_event_cursor(c.as_str())
-                        {
-                            warn!(error = %e, "persist seed cursor failed");
+                        let Some(c) = &head else {
+                            break None;
+                        };
+                        if let Err(e) = db.set_event_cursor(c.as_str()) {
+                            warn!(error = %e, ?delay, "persist seed cursor failed; retrying");
+                            tokio::time::sleep(delay).await;
+                            delay = (delay * 2).min(ONLINE_PROBE_MAX);
+                            continue;
                         }
-                        break head;
+                        break Some(c.clone());
                     }
                     Err(e) => {
                         warn!(error = %e, ?delay, "seed event cursor failed; retrying");
@@ -168,15 +172,19 @@ pub(super) async fn run_event_sync(
             // the life of the daemon (SDK plan #9). `apply_event` only touches
             // our FUSE state, so nothing else does this.
             if let Err(e) = client.invalidate_caches_for_event(event).await {
-                warn!(error = %e, "sdk cache invalidation for event failed");
+                warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry");
+                break;
             }
             apply_event(&state, &content, &pending, &notifier, event);
-        }
-        cursor = events.last().map(|e| e.id().clone());
-        if let Some(c) = &cursor
-            && let Err(e) = db.set_event_cursor(c.as_str())
-        {
-            warn!(error = %e, "persist event cursor failed");
+            let applied = event.id().clone();
+            if let Err(e) = db.set_event_cursor(applied.as_str()) {
+                warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry");
+                break;
+            }
+            // Advance only after this exact event is durably acknowledged. If a
+            // later event fails, polling resumes from here and replays nothing
+            // that was not already made visible locally.
+            cursor = Some(applied);
         }
     }
 }

crates/pdfs-fuse/src/control.rs

@@ -12,8 +12,8 @@
 use std::io::{BufRead, BufReader, Write as _};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
-use std::sync::atomic::Ordering;
-use std::time::Instant;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
 
 use pdfs_core::config::AppDirs;
 use pdfs_core::control::{
@@ -27,6 +27,67 @@ use tracing::{info, warn};
 use super::transfers::CountingReader;
 use super::{Core, count_noun, human_bytes, human_duration, parse_uid};
 
+const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024;
+const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10);
+const MAX_CONTROL_HANDLERS: usize = 64;
+static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0);
+
+struct ControlHandlerPermit;
+
+impl ControlHandlerPermit {
+    fn acquire() -> Option<Self> {
+        ACTIVE_CONTROL_HANDLERS
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| {
+                (active < MAX_CONTROL_HANDLERS).then_some(active + 1)
+            })
+            .ok()
+            .map(|_| Self)
+    }
+}
+
+impl Drop for ControlHandlerPermit {
+    fn drop(&mut self) {
+        ACTIVE_CONTROL_HANDLERS.fetch_sub(1, Ordering::AcqRel);
+    }
+}
+
+fn read_request_line(reader: &mut impl BufRead) -> std::io::Result<Option<String>> {
+    let mut bytes = Vec::new();
+    loop {
+        let available = reader.fill_buf()?;
+        if available.is_empty() {
+            break;
+        }
+        let take = available
+            .iter()
+            .position(|byte| *byte == b'\n')
+            .map_or(available.len(), |index| index + 1);
+        if bytes.len().saturating_add(take) > MAX_CONTROL_REQUEST_BYTES as usize {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "control request exceeds size limit",
+            ));
+        }
+        bytes.extend_from_slice(&available[..take]);
+        reader.consume(take);
+        if bytes.ends_with(b"\n") {
+            break;
+        }
+    }
+    if bytes.is_empty() {
+        return Ok(None);
+    }
+    if bytes.len() as u64 > MAX_CONTROL_REQUEST_BYTES || !bytes.ends_with(b"\n") {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "control request exceeds limit or is not newline terminated",
+        ));
+    }
+    String::from_utf8(bytes)
+        .map(Some)
+        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
+}
+
 /// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path
 /// must live under `mountpoint`; a relative path is taken as already relative to
 /// the mount root.
@@ -44,6 +105,10 @@ fn rel_to_mount(mountpoint: &Path, path: &str) -> CoreResult<PathBuf> {
 /// Handle one control-socket connection: read a single JSON request line,
 /// dispatch it against `core`, and write back a JSON response line.
 fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) {
+    if let Err(e) = stream.set_read_timeout(Some(CONTROL_READ_TIMEOUT)) {
+        warn!(error = %e, "control: could not set request timeout");
+        return;
+    }
     let mut reader = BufReader::new(match stream.try_clone() {
         Ok(s) => s,
         Err(e) => {
@@ -51,10 +116,14 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             return;
         }
     });
-    let mut line = String::new();
-    if reader.read_line(&mut line).is_err() || line.trim().is_empty() {
-        return;
-    }
+    let line = match read_request_line(&mut reader) {
+        Ok(Some(line)) if !line.trim().is_empty() => line,
+        Ok(_) => return,
+        Err(e) => {
+            warn!(error = %e, "control: rejected request");
+            return;
+        }
+    };
     let response = match serde_json::from_str::<CtlRequest>(line.trim()) {
         Ok(CtlRequest::Status) => {
             let pins = core.cache.list_pins();
@@ -792,12 +861,22 @@ pub(crate) fn run_control_socket(
     for conn in listener.incoming() {
         match conn {
             Ok(stream) => {
+                let Some(permit) = ControlHandlerPermit::acquire() else {
+                    warn!(
+                        limit = MAX_CONTROL_HANDLERS,
+                        "control: rejecting connection at handler limit"
+                    );
+                    continue;
+                };
                 let core = core.clone();
                 let username = username.clone();
                 let mountpoint = mountpoint.clone();
                 if let Err(e) = std::thread::Builder::new()
                     .name("pdfs-control".into())
-                    .spawn(move || handle_control_conn(&core, &username, &mountpoint, stream))
+                    .spawn(move || {
+                        let _permit = permit;
+                        handle_control_conn(&core, &username, &mountpoint, stream);
+                    })
                 {
                     warn!(error = %e, "control: spawn handler failed");
                 }
@@ -808,3 +887,38 @@ pub(crate) fn run_control_socket(
         }
     }
 }
+
+#[cfg(test)]
+mod request_limit_tests {
+    use super::*;
+    use std::io::Cursor;
+
+    #[test]
+    fn accepts_one_bounded_newline_terminated_request() {
+        let mut input = Cursor::new(b"{\"Status\":null}\n".to_vec());
+        assert_eq!(
+            read_request_line(&mut input).unwrap().as_deref(),
+            Some("{\"Status\":null}\n")
+        );
+    }
+
+    #[test]
+    fn rejects_a_request_without_a_terminating_newline() {
+        let mut input = Cursor::new(b"{\"Status\":null}".to_vec());
+        assert_eq!(
+            read_request_line(&mut input).unwrap_err().kind(),
+            std::io::ErrorKind::InvalidData
+        );
+    }
+
+    #[test]
+    fn rejects_a_request_larger_than_the_limit() {
+        let mut bytes = vec![b'x'; MAX_CONTROL_REQUEST_BYTES as usize + 1];
+        bytes.push(b'\n');
+        let mut input = Cursor::new(bytes);
+        assert_eq!(
+            read_request_line(&mut input).unwrap_err().kind(),
+            std::io::ErrorKind::InvalidData
+        );
+    }
+}

crates/pdfs-fuse/src/filesystem.rs

@@ -5,7 +5,13 @@ use super::*;
 impl Filesystem for ProtonFs {
     fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
         let parent = parent.0;
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         // A folder that has not been listed yet is enumerated from the remote,
         // so serve it from a worker rather than stalling the dispatch loop. A
         // listed folder — the common case — is a map hit, and answering it
@@ -307,7 +313,13 @@ impl Filesystem for ProtonFs {
             reply.error(e);
             return;
         }
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let parent_uid = {
             let st = self.core.state.lock();
             match st.entries.get(&parent) {
@@ -761,7 +773,13 @@ impl Filesystem for ProtonFs {
     }
 
     fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
-        let name_str = name.to_string_lossy();
+        let name_str = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
             let st = self.core.state.lock();
             if let Some(entry) = st.entries.get(&ino)
@@ -775,7 +793,13 @@ impl Filesystem for ProtonFs {
     }
 
     fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
-        let name_str = name.to_string_lossy();
+        let name_str = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
             {
                 let st = self.core.state.lock();
@@ -817,7 +841,13 @@ impl Filesystem for ProtonFs {
             reply.error(e);
             return;
         }
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let parent_uid = {
             let st = self.core.state.lock();
             match st.entries.get(&parent) {
@@ -899,8 +929,20 @@ impl Filesystem for ProtonFs {
     ) {
         let parent = parent.0;
         let newparent = newparent.0;
-        let name = name.to_string_lossy().into_owned();
-        let newname = newname.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
+        let newname = match fuse_name(newname) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let (ino, uid) = match self.core.lookup_child(parent, &name) {
             Ok(x) => x,
             Err(e) => {
@@ -1050,12 +1092,31 @@ impl Filesystem for ProtonFs {
         // take the synchronous path, so a genuine API refusal (permissions, a
         // name clash) surfaces to the caller instead of becoming a queued op
         // that can only ever conflict.
-        if !self.core.online.load(Ordering::Relaxed) || is_local_uid(&new_parent_uid) {
+        if rename_needs_queue(
+            self.core.online.load(Ordering::Relaxed),
+            is_local_uid(&new_parent_uid),
+            newparent != parent,
+            newname != name,
+        ) {
             match self
                 .core
                 .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname)
             {
-                Ok(()) => reply.ok(),
+                Ok(()) => {
+                    // Send the rename response first. Kernel-side rename cache
+                    // updates happen while processing that response; notifying
+                    // before it lets those updates overwrite our invalidation
+                    // and retain an empty destination readdir page.
+                    reply.ok();
+                    if let Some(notifier) = self.core.notifier.get() {
+                        let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
+                        let _ = notifier.inval_entry(INodeNo(newparent), OsStr::new(&newname));
+                        let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
+                        if newparent != parent {
+                            let _ = notifier.inval_inode(INodeNo(newparent), 0, 0);
+                        }
+                    }
+                }
                 Err(e) => {
                     self.core.restore_replaced(victim.as_ref(), &newname);
                     reply.error(e);
@@ -1063,54 +1124,57 @@ impl Filesystem for ProtonFs {
             }
             return;
         }
-        // Move first if the parent changed, then rename if the name changed.
+        // Rename first if both halves change. Moving first makes the encrypted
+        // name requirements stale and repeatedly failed with InvalidRequirements.
         // A failure past this point has already trashed any node the rename was
         // replacing, so put it back rather than leaving the caller with neither
         // the source moved nor the destination intact.
-        if newparent != parent
+        if newname != name
             && let Err(e) = self
                 .core
                 .rt
-                .block_on(self.core.client.move_node(&uid, &new_parent_uid))
+                .block_on(self.core.client.rename_node(&uid, &newname, None))
         {
-            error!(%uid, error = %e, "move failed");
+            error!(%uid, error = %e, "rename failed");
             self.core.restore_replaced(victim.as_ref(), &newname);
             reply.error(Errno::EIO);
             return;
         }
-        if newname != name {
-            // A move changes the node's encrypted name requirements. Proton's
-            // link-details read can remain briefly stale immediately after the
-            // move, making the following rename fail with InvalidRequirements
-            // ("out of date"). Re-read/retry that narrow eventual-consistency
-            // error; other refusals are stable and must surface immediately.
+        if newparent != parent {
             let mut attempts = 0u32;
-            let renamed = loop {
+            let moved = loop {
                 match self
                     .core
                     .rt
-                    .block_on(self.core.client.rename_node(&uid, &newname, None))
+                    .block_on(self.core.client.move_node(&uid, &new_parent_uid))
                 {
                     Ok(()) => break Ok(()),
                     Err(e)
-                        if newparent != parent
+                        if newname != name
                             && api_code(&e) == Some(ResponseCode::InvalidRequirements)
-                            && attempts < 4 =>
+                            && attempts < 20 =>
                     {
                         attempts += 1;
-                        std::thread::sleep(Duration::from_millis(100 * (1 << attempts)));
+                        // Requirement propagation routinely takes longer than
+                        // three seconds. Keep a bounded ten-second window; a
+                        // successful POSIX reply must mean both remote halves
+                        // landed, not merely that the second half was queued.
+                        std::thread::sleep(Duration::from_millis(500));
                     }
                     Err(e) => break Err(e),
                 }
             };
-            if let Err(e) = renamed {
-                error!(%uid, attempts, error = %e, "rename after move failed");
-                // The move half already landed. Keep our inode model honest
-                // even though the combined POSIX operation has to return EIO.
-                self.core
-                    .state
-                    .lock()
-                    .relocate(ino, parent, newparent, &new_parent_uid, &name);
+            if let Err(e) = moved {
+                error!(%uid, attempts, error = %e, "move after rename failed");
+                // The rename half landed in the source directory.
+                let mut state = self.core.state.lock();
+                let old_parent_uid = state
+                    .entries
+                    .get(&parent)
+                    .map(|entry| entry.uid.clone())
+                    .unwrap_or_else(|| uid.clone());
+                state.rename_in_place(ino, parent, &old_parent_uid, &newname);
+                drop(state);
                 self.core.restore_replaced(victim.as_ref(), &newname);
                 reply.error(Errno::EIO);
                 return;
@@ -1385,6 +1449,7 @@ impl ProtonFs {
         }
         // Only remove the namespace entry after Drive accepted the mutation.
         // Otherwise an EIO response still made the path disappear locally.
+        self.core.hidden.lock().insert(uid.clone());
         self.core.state.lock().forget_or_unlink(&uid);
         if !open_now {
             self.core.discard_queued_ops(&uid);

crates/pdfs-fuse/src/lib.rs

@@ -36,6 +36,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::ffi::OsStr;
 use std::fs::File;
 use std::io::Write;
+use std::os::unix::ffi::OsStrExt;
 use std::os::unix::fs::FileExt;
 use std::os::unix::net::UnixListener;
 use std::path::{Component, Path, PathBuf};
@@ -225,6 +226,9 @@ struct Core {
     /// staged blob *is* the file's content and the remote still holds the old
     /// revision; and the drain worker, which performs the uploads.
     pending: Arc<Mutex<HashMap<NodeUid, PendingRevision>>>,
+    /// Nodes removed through this daemon which a briefly stale remote listing
+    /// may still return. Uids are not reused; an explicit restore clears one.
+    hidden: Arc<Mutex<HashSet<NodeUid>>>,
     /// Nudges the drain worker: set true and notify to have it re-examine the
     /// queue instead of waiting out its backoff.
     drain_wake: Arc<(Mutex<bool>, Condvar)>,
@@ -654,6 +658,7 @@ impl Core {
                 // Before the lock: a DB row carries the size the server last
                 // sealed, which a queued write is ahead of (B11).
                 self.stamp_pending_sizes(&mut nodes);
+                let hidden = self.hidden.lock().clone();
                 let mut st = self.state.lock();
                 if st.children.contains_key(&ino) {
                     return Ok(());
@@ -661,7 +666,7 @@ impl Core {
                 let mut child_inos = Vec::with_capacity(nodes.len());
                 let mut needs_size = Vec::new();
                 for node in nodes {
-                    if node.trashed || node.uid == folder_uid {
+                    if !node_visible(&node, &folder_uid, &hidden) {
                         continue;
                     }
                     if matches!(
@@ -712,6 +717,7 @@ impl Core {
         // Same as the DB path above: the remote's size for a file with a write
         // still queued is the pre-write one (B11).
         self.stamp_pending_sizes(&mut nodes);
+        let hidden = self.hidden.lock().clone();
 
         let mut st = self.state.lock();
         // Lost the race? Another thread already populated it.
@@ -721,7 +727,7 @@ impl Core {
         let mut child_inos = Vec::with_capacity(nodes.len());
         let filtered_nodes: Vec<Node> = nodes
             .into_iter()
-            .filter(|node| !node.trashed && node.uid != folder_uid)
+            .filter(|node| node_visible(node, &folder_uid, &hidden))
             .collect();
         // Files whose real size the cheap enumeration could not read. Collected
         // before interning so the upgrade below has the uids without re-walking.
@@ -1300,6 +1306,21 @@ impl Core {
         if size == base_size {
             return Ok(());
         }
+        // A close immediately followed by truncate is common (`write`, close,
+        // `truncate(2)`) and usually beats the write-back debounce. In that
+        // case the remote is *not* our base: the complete staged revision is.
+        // Compose from those local bytes so the truncate can supersede the
+        // queued upload without fetching stale remote content or returning EIO.
+        let pending_base = self.pending.lock().get(&uid).cloned();
+        if pending_base
+            .as_ref()
+            .is_some_and(|pending| !pending.meta.complete)
+        {
+            // An incomplete pending blob still has holes referring to the
+            // remote revision. Stacking another incomplete transform on it is
+            // not representable safely; preserve the existing refusal.
+            return Err(Errno::EIO);
+        }
         let (authored, complete) = if size == 0 {
             // An empty file has no content to be missing.
             (Vec::new(), true)
@@ -1316,19 +1337,45 @@ impl Core {
             error!(%uid, error = %e, "create scratch file for truncate failed");
             Errno::EIO
         })?;
+        if let Some(pending) = &pending_base
+            && let Err(e) = copy_pending_for_truncate(pending, &file)
+        {
+            error!(%uid, source = %pending.path.display(), error = %e,
+                "copy pending revision for truncate failed");
+            let _ = std::fs::remove_file(&path);
+            return Err(Errno::EIO);
+        }
         file.set_len(size).map_err(|e| {
             error!(%uid, error = %e, "resize scratch file for truncate failed");
             let _ = std::fs::remove_file(&path);
             Errno::EIO
         })?;
-        let meta = StagedWrite {
-            uid: uid.to_string(),
-            len: size,
-            base_size,
-            base_mtime,
-            authored,
-            complete,
-            based_on: self.remote_baseline(&uid, base_mtime, base_size),
+        let meta = match pending_base {
+            Some(pending) => StagedWrite {
+                uid: uid.to_string(),
+                len: size,
+                base_size: pending.meta.len,
+                base_mtime: pending.meta.base_mtime,
+                authored: if size == 0 {
+                    Vec::new()
+                } else {
+                    vec![(0, size)]
+                },
+                complete: true,
+                // Both local revisions descend from the same last-observed
+                // remote revision. Never replace this with the optimistic
+                // mtime/size currently published by State.
+                based_on: pending.meta.based_on,
+            },
+            None => StagedWrite {
+                uid: uid.to_string(),
+                len: size,
+                base_size,
+                base_mtime,
+                authored,
+                complete,
+                based_on: self.remote_baseline(&uid, base_mtime, base_size),
+            },
         };
         self.enqueue_staged_write(&uid, ino, &path, meta)?;
         debug!(%uid, size, complete, "queued truncate");
@@ -1447,6 +1494,7 @@ impl Core {
             error!(%uid, error = %e, "queueing trash failed");
             Errno::EIO
         })?;
+        self.hidden.lock().insert(uid.clone());
         self.state.lock().forget(uid);
         self.cache.evict(uid);
         self.evict_reader(uid);
@@ -1500,8 +1548,32 @@ impl Core {
                 );
             }
             Err(e) => {
-                error!(%uid, error = %e, "staging write failed; bytes lost");
-                let _ = std::fs::remove_file(src);
+                // `src` is the only known copy of bytes the kernel already
+                // accepted.  A failure to publish them into staging must never
+                // turn into permission to delete them: leave the scratch file
+                // in place so startup recovery or a human can still salvage it.
+                error!(
+                    %uid,
+                    error = %e,
+                    scratch = %src.display(),
+                    "staging write failed; retaining the only scratch copy"
+                );
+                let name = {
+                    let st = self.state.lock();
+                    st.entries
+                        .get(&ino)
+                        .map(|entry| entry.node.name.clone())
+                        .unwrap_or_default()
+                };
+                self.log_activity(
+                    ActivityKind::Upload,
+                    &name,
+                    format!(
+                        "write not queued and could not be staged; recovery copy retained at {}",
+                        src.display()
+                    ),
+                    false,
+                );
             }
         }
     }
@@ -2293,6 +2365,12 @@ impl Core {
         self.rt
             .block_on(self.client.restore_nodes(&parsed))
             .map_err(|e| CoreError::from_api(&e, "restore"))?;
+        {
+            let mut hidden = self.hidden.lock();
+            for uid in &parsed {
+                hidden.remove(uid);
+            }
+        }
         {
             let mut st = self.state.lock();
             for parent in parents {
@@ -2861,6 +2939,59 @@ fn count_noun(n: usize, one: &str, many: &str) -> String {
     format!("{n} {}", if n == 1 { one } else { many })
 }
 
+/// Seed a truncate scratch file from a complete queued revision.
+///
+/// `File` writes use an explicit offset so this is independent of either
+/// descriptor's cursor and remains correct when the destination came from the
+/// cache's scratch allocator.
+fn copy_pending_for_truncate(pending: &PendingRevision, destination: &File) -> std::io::Result<()> {
+    let source = File::open(&pending.path)?;
+    destination.set_len(0)?;
+    let mut offset = 0u64;
+    let mut buffer = vec![0u8; 1024 * 1024];
+    loop {
+        let count = source.read_at(&mut buffer, offset)?;
+        if count == 0 {
+            break;
+        }
+        destination.write_all_at(&buffer[..count], offset)?;
+        offset += count as u64;
+    }
+    destination.set_len(offset)
+}
+
+/// Whether a rename must be represented as a durable desired end state.
+/// Offline operations and operations targeting a not-yet-uploaded directory
+/// cannot use the API immediately. Online remote-to-remote operations stay
+/// synchronous so a successful kernel reply means the remote namespace has
+/// already reached the same end state.
+fn rename_needs_queue(
+    online: bool,
+    destination_is_local: bool,
+    _parent_changed: bool,
+    _name_changed: bool,
+) -> bool {
+    !online || destination_is_local
+}
+
+/// Convert one kernel pathname component into the UTF-8 name accepted by Drive.
+/// Linux filesystems conventionally cap a component at `NAME_MAX` (255 bytes),
+/// but the FUSE kernel path can pass a longer component through to userspace.
+fn fuse_name(name: &OsStr) -> Result<String, Errno> {
+    let bytes = name.as_bytes();
+    if bytes.len() > 255 {
+        return Err(Errno::ENAMETOOLONG);
+    }
+    if bytes.is_empty() || bytes == b"." || bytes == b".." {
+        return Err(Errno::EINVAL);
+    }
+    name.to_str().map(str::to_owned).ok_or(Errno::EILSEQ)
+}
+
+fn node_visible(node: &Node, folder_uid: &NodeUid, hidden: &HashSet<NodeUid>) -> bool {
+    !node.trashed && node.uid != *folder_uid && !hidden.contains(&node.uid)
+}
+
 /// Bytes rendered with a binary unit and one decimal place (e.g. `"1.2 GB"`),
 /// for the activity log. Uses 1024-based steps but the shorter SI labels.
 fn human_bytes(bytes: u64) -> String {
@@ -3224,7 +3355,11 @@ mod pending_size_tests {
 
 #[cfg(test)]
 mod tests {
-    use super::{Intervals, conflict_name, is_stale_mount};
+    use super::{
+        Intervals, PendingRevision, conflict_name, copy_pending_for_truncate, fuse_name,
+        is_stale_mount, node_visible, rename_needs_queue,
+    };
+    use pdfs_core::cache::{Baseline, StagedWrite};
 
     /// The predicate must answer *only* for a dead FUSE connection. A healthy
     /// directory and an absent path are both "not stale" — widening it to any
@@ -3331,6 +3466,124 @@ mod tests {
         assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]);
     }
 
+    fn pending_revision(dir: &TestDir, bytes: &[u8], complete: bool) -> PendingRevision {
+        let path = dir.0.join("pending");
+        std::fs::write(&path, bytes).unwrap();
+        PendingRevision {
+            path,
+            meta: StagedWrite {
+                uid: "vol~file".into(),
+                len: bytes.len() as u64,
+                base_size: 3,
+                base_mtime: 42,
+                authored: if complete {
+                    vec![(0, bytes.len() as u64)]
+                } else {
+                    vec![(3, bytes.len() as u64)]
+                },
+                complete,
+                based_on: Some(Baseline {
+                    mtime: 42,
+                    size: 3,
+                    hash: None,
+                }),
+            },
+        }
+    }
+
+    #[test]
+    fn complete_pending_revision_can_seed_immediate_shrink() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"0123456789", true);
+        let destination_path = dir.0.join("shrink");
+        let destination = std::fs::File::create(&destination_path).unwrap();
+        copy_pending_for_truncate(&pending, &destination).unwrap();
+        destination.set_len(4).unwrap();
+        assert_eq!(std::fs::read(destination_path).unwrap(), b"0123");
+    }
+
+    #[test]
+    fn complete_pending_revision_can_seed_immediate_growth_and_zero() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"0123456789", true);
+        let destination_path = dir.0.join("grow");
+        let destination = std::fs::File::create(&destination_path).unwrap();
+        copy_pending_for_truncate(&pending, &destination).unwrap();
+        destination.set_len(14).unwrap();
+        assert_eq!(
+            std::fs::read(&destination_path).unwrap(),
+            b"0123456789\0\0\0\0"
+        );
+        destination.set_len(0).unwrap();
+        assert!(std::fs::read(destination_path).unwrap().is_empty());
+    }
+
+    #[test]
+    fn incomplete_pending_revision_is_not_a_safe_truncate_base() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"\0\0\0authored", false);
+        assert!(!pending.meta.complete);
+        assert_ne!(pending.meta.authored, vec![(0, pending.meta.len)]);
+    }
+
+    #[test]
+    fn online_combined_cross_directory_rename_stays_synchronous() {
+        assert!(!rename_needs_queue(true, false, true, true));
+        assert!(rename_needs_queue(false, false, false, true));
+        assert!(rename_needs_queue(true, true, true, false));
+    }
+
+    #[test]
+    fn one_step_online_namespace_changes_stay_synchronous() {
+        assert!(!rename_needs_queue(true, false, false, true));
+        assert!(!rename_needs_queue(true, false, true, false));
+        assert!(!rename_needs_queue(true, false, false, false));
+    }
+
+    #[test]
+    fn fuse_names_enforce_linux_component_limits() {
+        use std::ffi::OsStr;
+
+        assert_eq!(
+            fuse_name(OsStr::new("x".repeat(255).as_str()))
+                .unwrap()
+                .len(),
+            255
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("x".repeat(256).as_str()))
+                .expect_err("256-byte component must be refused")
+                .code(),
+            libc::ENAMETOOLONG
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("."))
+                .expect_err("dot is not an ordinary component")
+                .code(),
+            libc::EINVAL
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("unicodé-文件")).unwrap(),
+            "unicodé-文件"
+        );
+    }
+
+    #[test]
+    fn local_trash_tombstones_hide_stale_remote_entries() {
+        let folder_uid = NodeUid::new(VolumeId::from("vol"), LinkId::from("folder"));
+        let node = node_helper("child", "folder", "child", false);
+        assert!(node_visible(
+            &node,
+            &folder_uid,
+            &std::collections::HashSet::new()
+        ));
+        assert!(!node_visible(
+            &node,
+            &folder_uid,
+            &std::collections::HashSet::from([node.uid.clone()])
+        ));
+    }
+
     use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
     use proton_drive_rs::{Node, NodeKind};
 

crates/pdfs-fuse/src/mount.rs

@@ -192,6 +192,7 @@ pub fn mount(
         db,
         online: Arc::new(AtomicBool::new(online)),
         pending: Arc::new(Mutex::new(HashMap::new())),
+        hidden: Arc::new(Mutex::new(HashSet::new())),
         drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
         timeline_refreshing: Arc::new(AtomicBool::new(false)),
         trash_refreshing: Arc::new(AtomicBool::new(false)),

crates/pdfs-fuse/src/state.rs

@@ -219,10 +219,12 @@ impl State {
 
     /// Check if a directory inode has any child nodes in memory or in the database.
     pub(crate) fn has_children(&self, parent: u64) -> bool {
-        if let Some(kids) = self.children.get(&parent)
-            && !kids.is_empty()
-        {
-            return true;
+        // A resident listing is authoritative even when empty. Falling through
+        // from `Some([])` to SQLite lets an obsolete child row contradict the
+        // listing `ensure_children` just established, so readdir says empty but
+        // rmdir returns ENOTEMPTY forever.
+        if let Some(kids) = self.children.get(&parent) {
+            return !kids.is_empty();
         }
         if let Some(entry) = self.entries.get(&parent)
             && let Ok(has_kids) = self.db.has_children(&entry.uid)
@@ -607,6 +609,11 @@ mod tests {
         st.db
             .upsert_node(&node("file2_uid", "dir_uid", "file2.txt", false))
             .unwrap();
+        assert!(
+            !st.has_children(folder),
+            "a resident empty listing overrides stale DB children"
+        );
+        st.children.remove(&folder);
         assert!(st.has_children(folder), "has child in db");
     }
 }

crates/pdfs-fuse/src/sync.rs

@@ -55,9 +55,11 @@ use local_scan::is_write_mode;
 use local_scan::open_for_write_set;
 use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem};
 pub(crate) use planner::base_name;
+#[cfg(test)]
+use planner::conflict_path;
 use planner::{
-    FilePlan, classification_order, conflict_path, filter_baseline, guard_local_wipe, join_rel,
-    parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size,
+    FilePlan, classification_order, conflict_path_with_suffix, filter_baseline, guard_local_wipe,
+    join_rel, parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size,
 };
 /// What a [`Pending`] op did, so the reconcile can update shared state on the
 /// engine thread (never inside a task).
@@ -711,7 +713,17 @@ impl Core {
                     }
                 },
                 FilePlan::DeleteLocal => {
-                    let _ = std::fs::remove_file(local_root.join(rel_to_path(rel)));
+                    if let Err(e) = std::fs::remove_file(local_root.join(rel_to_path(rel))) {
+                        warn!(rel, error = %e, "sync: remove local file failed");
+                        self.log_activity(
+                            ActivityKind::Trash,
+                            base_name(rel),
+                            e.to_string(),
+                            false,
+                        );
+                        outcome.errors += 1;
+                        continue;
+                    }
                     let _ = self.db.sync_entry_remove(folder_id, rel);
                     self.log_activity(ActivityKind::Trash, base_name(rel), "removed locally", true);
                     outcome.deleted += 1;
@@ -758,7 +770,12 @@ impl Core {
         // Deferred folder deletions, deepest first.
         delete_local_dirs.sort_by_key(|p| std::cmp::Reverse(p.matches('/').count()));
         for rel in delete_local_dirs {
-            let _ = std::fs::remove_dir_all(local_root.join(rel_to_path(&rel)));
+            if let Err(e) = std::fs::remove_dir_all(local_root.join(rel_to_path(&rel))) {
+                warn!(rel, error = %e, "sync: remove local folder failed");
+                self.log_activity(ActivityKind::Trash, base_name(&rel), e.to_string(), false);
+                outcome.errors += 1;
+                continue;
+            }
             let _ = self.db.sync_entry_remove(folder_id, &rel);
             self.log_activity(
                 ActivityKind::Trash,
@@ -802,18 +819,24 @@ impl Core {
         out: &mut HashMap<String, LocalItem>,
     ) -> Result<(), String> {
         let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?;
-        for entry in entries.filter_map(|e| e.ok()) {
+        for entry in entries {
+            let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?;
             let path = entry.path();
-            let meta = match std::fs::symlink_metadata(&path) {
-                Ok(m) => m,
-                Err(_) => continue,
-            };
-            let Ok(stripped) = path.strip_prefix(root) else {
-                continue;
-            };
-            let Some(rel) = stripped.to_str() else {
-                continue;
-            };
+            let meta = std::fs::symlink_metadata(&path)
+                .map_err(|e| format!("stat {}: {e}", path.display()))?;
+            let stripped = path.strip_prefix(root).map_err(|e| {
+                format!(
+                    "local path {} escaped sync root {}: {e}",
+                    path.display(),
+                    root.display()
+                )
+            })?;
+            let rel = stripped.to_str().ok_or_else(|| {
+                format!(
+                    "local path {} is not valid UTF-8; refusing a destructive sync pass",
+                    path.display()
+                )
+            })?;
             // Ignore our own in-flight download temp files.
             if rel.contains(".pdfs-tmp-") {
                 continue;
@@ -1153,12 +1176,9 @@ impl Core {
                 // pass), then take the remote version as the shared truth.
                 let path = local_root.join(rel_to_path(rel));
                 if path.exists() {
-                    let conflict = conflict_path(&path, now_secs());
-                    if let Err(e) = std::fs::rename(&path, &conflict) {
-                        warn!(rel, error = %e, "sync: could not set aside conflict copy");
-                    } else {
-                        info!(rel, "sync: kept local changes as a conflict copy");
-                    }
+                    preserve_conflict_copy(&path, now_secs())
+                        .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
+                    info!(rel, "sync: kept local changes as a conflict copy");
                 }
                 self.download_file(folder_id, local_root, rel, uid, *mtime, *size)
                     .await?;
@@ -1171,8 +1191,7 @@ impl Core {
                 // evicted. Push the conflict copy now, while it still exists, and
                 // leave the remote file holding the original path.
                 let path = local_root.join(rel_to_path(rel));
-                let conflict = conflict_path(&path, now_secs());
-                std::fs::rename(&path, &conflict)
+                let conflict = preserve_conflict_copy(&path, now_secs())
                     .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
                 let name = conflict
                     .file_name()
@@ -1297,13 +1316,24 @@ impl Core {
                     let _ = std::fs::remove_file(&tmp);
                     format!("download {rel}: {e}")
                 })?;
-            out.into_inner().flush().ok();
+            let mut out = out.into_inner();
+            out.flush().map_err(|e| format!("flush tmp {rel}: {e}"))?;
+            out.sync_all().map_err(|e| format!("sync tmp {rel}: {e}"))?;
         }
         std::fs::rename(&tmp, &path).map_err(|e| format!("place {rel}: {e}"))?;
         // Match local mtime to the remote's so neither side looks "changed" next pass.
-        if let Ok(f) = std::fs::File::options().write(true).open(&path) {
-            let _ =
-                f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.max(0) as u64));
+        let f = std::fs::File::options()
+            .write(true)
+            .open(&path)
+            .map_err(|e| format!("open placed file {rel}: {e}"))?;
+        f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.max(0) as u64))
+            .map_err(|e| format!("set mtime for {rel}: {e}"))?;
+        f.sync_all()
+            .map_err(|e| format!("sync placed file {rel}: {e}"))?;
+        if let Some(parent) = path.parent() {
+            std::fs::File::open(parent)
+                .and_then(|dir| dir.sync_all())
+                .map_err(|e| format!("sync directory for {rel}: {e}"))?;
         }
         self.record_file_baseline(folder_id, rel, &path, uid).await
     }
@@ -1350,6 +1380,46 @@ impl Core {
     }
 }
 
+/// Publish a conflict copy without overwriting an earlier conflict bearing the
+/// same timestamp. The original is removed only after the new copy is durable.
+fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result<PathBuf> {
+    let metadata = std::fs::metadata(path)?;
+    for suffix in 0..=u32::MAX {
+        let conflict = conflict_path_with_suffix(path, stamp, suffix);
+        let mut source = std::fs::File::open(path)?;
+        let mut target = match std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&conflict)
+        {
+            Ok(file) => file,
+            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
+            Err(e) => return Err(e),
+        };
+        if let Err(e) = std::io::copy(&mut source, &mut target).and_then(|_| target.sync_all()) {
+            let _ = std::fs::remove_file(&conflict);
+            return Err(e);
+        }
+        target.set_permissions(metadata.permissions())?;
+        if let Ok(modified) = metadata.modified() {
+            target.set_modified(modified)?;
+        }
+        target.sync_all()?;
+        sync_parent_dir(&conflict)?;
+        std::fs::remove_file(path)?;
+        sync_parent_dir(&conflict)?;
+        return Ok(conflict);
+    }
+    unreachable!("u32 conflict suffix space exhausted")
+}
+
+fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
+    match path.parent() {
+        Some(parent) => std::fs::File::open(parent)?.sync_all(),
+        None => Ok(()),
+    }
+}
+
 // ---- helpers --------------------------------------------------------------
 
 /// A remote node's `(mtime, size)` change signature. Size prefers the plaintext
@@ -1609,9 +1679,10 @@ mod tests {
             guard_local_wipe(&base, &one).is_ok(),
             "one survivor means the tree is readable; the rest are real deletions"
         );
-        // A one-entry baseline is a genuine single delete, not a wipe to detect.
+        // A one-entry folder can disappear because its mount or parent became
+        // unavailable just as easily as a larger tree can.
         let single: HashMap<String, ()> = [("a.txt".to_string(), ())].into_iter().collect();
-        assert!(guard_local_wipe(&single, &none).is_ok());
+        assert!(guard_local_wipe(&single, &none).is_err());
         // Nothing synced yet: a first pass over an empty folder is not a wipe.
         assert!(guard_local_wipe(&none, &none).is_ok());
         // Local paths that are not in the baseline are new uploads, and do not
@@ -1781,6 +1852,52 @@ mod tests {
         assert_eq!(no_ext, PathBuf::from("/home/me/README (sync-conflict 42)"));
     }
 
+    #[test]
+    fn conflict_preservation_never_replaces_an_existing_copy() {
+        let dir = sync_test_dir("conflict-collision");
+        let source = dir.join("notes.txt");
+        let existing = conflict_path(&source, 1700);
+        std::fs::write(&source, b"new local edit").unwrap();
+        std::fs::write(&existing, b"older conflict").unwrap();
+
+        let preserved = preserve_conflict_copy(&source, 1700).unwrap();
+
+        assert_eq!(std::fs::read(&existing).unwrap(), b"older conflict");
+        assert_eq!(std::fs::read(&preserved).unwrap(), b"new local edit");
+        assert_eq!(
+            preserved.file_name().unwrap(),
+            "notes (sync-conflict 1700-1).txt"
+        );
+        assert!(!source.exists());
+        std::fs::remove_dir_all(dir).unwrap();
+    }
+
+    #[test]
+    fn conflict_preservation_failure_keeps_the_source() {
+        let dir = sync_test_dir("conflict-failure");
+        // A directory cannot be copied through the regular-file preservation
+        // path, deterministically forcing publication to fail after the source
+        // has been discovered.
+        let source = dir.join("not-a-file.txt");
+        std::fs::create_dir(&source).unwrap();
+
+        assert!(preserve_conflict_copy(&source, 1700).is_err());
+        assert!(source.is_dir(), "failure must not remove the source");
+        assert!(!conflict_path(&source, 1700).exists());
+        std::fs::remove_dir_all(dir).unwrap();
+    }
+
+    fn sync_test_dir(label: &str) -> PathBuf {
+        let nonce = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let path =
+            std::env::temp_dir().join(format!("pdfs-sync-{label}-{}-{nonce}", std::process::id()));
+        std::fs::create_dir(&path).unwrap();
+        path
+    }
+
     // ---- open-for-write detection tests -----------------------------------
 
     #[test]

crates/pdfs-fuse/src/sync/planner.rs

@@ -99,7 +99,7 @@ pub(super) fn guard_local_wipe<B, L>(
     baseline: &HashMap<String, B>,
     local: &HashMap<String, L>,
 ) -> Result<(), String> {
-    if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) {
+    if !baseline.is_empty() && baseline.keys().all(|rel| !local.contains_key(rel)) {
         return Err(format!(
             "every one of the {} synced paths is missing locally; refusing to trash \
              them on Drive. Check that the folder is mounted and readable.",
@@ -162,12 +162,24 @@ pub(super) fn rel_to_path(rel: &str) -> PathBuf {
 }
 
 /// The name for a conflict copy of `path`, e.g. `notes (sync-conflict 1700000000).txt`.
+#[cfg(test)]
 pub(super) fn conflict_path(path: &Path, stamp: i64) -> PathBuf {
+    conflict_path_with_suffix(path, stamp, 0)
+}
+
+/// As [`conflict_path`], with a deterministic suffix used when that name already
+/// exists. This keeps conflict preservation from ever replacing an older copy.
+pub(super) fn conflict_path_with_suffix(path: &Path, stamp: i64, suffix: u32) -> PathBuf {
     let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file");
     let ext = path.extension().and_then(|s| s.to_str());
+    let suffix = if suffix == 0 {
+        String::new()
+    } else {
+        format!("-{suffix}")
+    };
     let name = match ext {
-        Some(ext) => format!("{stem} (sync-conflict {stamp}).{ext}"),
-        None => format!("{stem} (sync-conflict {stamp})"),
+        Some(ext) => format!("{stem} (sync-conflict {stamp}{suffix}).{ext}"),
+        None => format!("{stem} (sync-conflict {stamp}{suffix})"),
     };
     match path.parent() {
         Some(dir) => dir.join(name),

docs/BUGS.md

@@ -1206,7 +1206,8 @@ retracted.
 
 ## B19 — Remote Folder Trashing on Mode Switch Failure (CRIT-02)
 
-**Status:** Fixed in code (unverified on a live mount, 2026-07-22)
+**Status:** Fixed; all four managed-live mode pairs verified 2026-07-22,
+fault-injected mount failure still pending
 **Found:** 2026-07-21, multi-agent sync engine audit (`audit_bugs.md` CRIT-02)  
 **Where:** `crates/pdfs-fuse/src/devices.rs`, `apply_sync_folder_mode`
 
@@ -1220,8 +1221,9 @@ user deletion, evict the underlying local mirror while holding the sync-folder
 lock, and only then mount FUSE. A mount failure leaves an inert `ondemand` row;
 switching back to `mirror` clears the baseline and restores the local copy.
 
-**Verified:** `cargo clippy -p pdfs-fuse --all-targets -- -D warnings` and the
-`pdfs-fuse` unit suite pass. A fault-injected/live mode-switch test is still needed.
+**Verified:** `cargo clippy -p pdfs-fuse --all-targets -- -D warnings`, the
+`pdfs-fuse` unit suite, and the complete managed-live matrix passed against the
+1.0.0 release binary. A fault-injected mount-failure test is still needed.
 
 ---
 
@@ -1587,16 +1589,23 @@ counts to baseline and observe no post-unmount DB work.
 
 ---
 
-## B45 — Truncate After a Queued Rewrite Replaces the Prefix With Zeros
+## B45 — Truncate After a Queued Rewrite Corrupts or Returns EIO
 
 **Status:** Fixed and live-verified (2026-07-22)
 **Found by:** `scripts/fuse-acceptance.sh`
-**Where:** `crates/pdfs-fuse/src/filesystem.rs`, write-open scratch setup
+**Where:** `crates/pdfs-fuse/src/filesystem.rs`, write-open scratch setup;
+`crates/pdfs-fuse/src/lib.rs`, `Core::queue_truncate`
 
 **Repro:** Write a file, immediately replace its contents, then run
 `truncate -s 4 file` before the queued replacement drains. Reading it returned
 four NUL bytes instead of the first four replacement bytes.
 
+The expanded managed suite found the path-based variant: create and close a
+10-byte file, then immediately call `truncate(path, 4)`. The close's complete
+revision was still inside the two-second drain debounce. `queue_truncate`
+described the shrink as an incomplete edit over the remote base, noticed the
+pending edit, preserved an orphan staging file, and returned `EIO`.
+
 **Cause:** A new write handle always treated the server revision as its base.
 When a newer complete revision was still queued locally, the handle's scratch
 file began sparse and zero-filled. Shrinking it preserved those scratch zeros,
@@ -1604,8 +1613,10 @@ not the prefix of the locally authoritative queued revision.
 
 **Fix:** A write opened over a complete queued revision seeds its scratch file
 from that staged blob and marks the copied range authored. Stacking a write over
-an incomplete queued revision is refused rather than guessed. Live acceptance
-verified the exact rewrite-then-truncate sequence returns `b"a mu"`.
+an incomplete queued revision is refused rather than guessed. Path-based
+truncate now composes the same way: it copies a complete pending blob, applies
+the new length, marks the result complete, and inherits the last real remote
+baseline. Live acceptance verified shrink, zero-filled growth, and sparse I/O.
 
 ---
 
@@ -1624,8 +1635,478 @@ date"), surfaced as `EIO`.
 link-details read can briefly observe the pre-move state, so the new name is
 signed against stale requirements.
 
-**Fix:** Retry only `InvalidRequirements` from the rename-after-move step with a
-short bounded backoff. Stable API failures are not retried. If every retry
-fails, adopt the already-completed move in inode state before returning `EIO`,
-so the local model never claims the node remains in its old parent. The full
-live suite subsequently passed the replacement and combined move/rename cases.
+**Fix history:** A bounded retry of `InvalidRequirements` reduced the window but
+did not close it; the managed matrix reproduced the same failure after all four
+retries. Combined cross-directory move+rename now enters the durable rename
+queue as a desired end state. Its drain re-fetches the node on every attempt,
+skips either half that already landed, renames before moving, and resolves name
+collisions without losing the source. Simple rename-only and move-only calls
+remain synchronous. The managed live suite subsequently passed the replacement
+and combined move/rename cases.
+
+---
+
+## B47 — FUSE Accepts Path Components Longer Than NAME_MAX
+
+**Status:** Fixed and managed-live verified (2026-07-22)
+**Found:** 2026-07-22, managed FUSE acceptance suite
+**Where:** `crates/pdfs-fuse/src/filesystem.rs`, name-taking callbacks
+
+**Repro:** `open(<mount>/<256 ASCII bytes>, O_CREAT)` succeeded. A conventional
+Linux filesystem must reject a pathname component longer than 255 bytes with
+`ENAMETOOLONG`.
+
+**Cause:** Every callback converted `OsStr` with `to_string_lossy` and passed it
+straight to Drive. This imposed neither Linux's component limit nor Drive's
+UTF-8 requirement, and could silently change non-UTF-8 names.
+
+**Fix:** Route lookup, create, mkdir, unlink, rmdir, and both rename components
+through one validator. It enforces the 255-byte limit, rejects reserved/empty
+components, rejects invalid UTF-8 with `EILSEQ`, and preserves valid Unicode.
+The managed live suite is the verification gate.
+
+---
+
+## T1 — Managed Acceptance Raced Daemon Startup
+
+**Status:** Fixed (2026-07-22)
+**Found:** repeated install/restart/acceptance loop
+**Where:** `scripts/fuse-acceptance.py`, managed setup
+
+**Repro:** Run `systemctl --user restart proton-drive.service` and immediately
+start `--managed-live`. The offline reference completed before the daemon had
+recreated its control socket, so the first `pdfs --json sync list` failed with
+`ENOENT` and setup aborted.
+
+**Fix:** Managed validation polls the supported `sync list` API until the daemon
+is ready or `PDFS_ACCEPTANCE_SYNC_TIMEOUT` expires. This is tracked as a test
+harness defect rather than an application bug.
+
+---
+
+## T2 — Acceptance Cleanup Raced Queued Namespace Replay
+
+**Status:** Fixed (2026-07-22)
+**Found:** managed on-demand/on-demand matrix after every functional test passed
+**Where:** `scripts/fuse-acceptance.py`, per-contract cleanup
+
+**Repro:** A combined move+rename was correctly accepted into the durable queue.
+Cleanup started recursively deleting its test tree while the rename drain was
+still landing. The moved entry appeared in `dir-b` after `rmtree` enumerated it
+but before `rmdir(dir-b)`, producing `ENOTEMPTY`.
+
+**Fix:** Before cleanup, poll `pdfs --json status` until both pending counters
+are zero. Retry `rmtree` only for `ENOTEMPTY` inside the uniquely owned test
+root, then wait for the queued deletions before changing modes. Other errors are
+still immediate failures.
+
+---
+
+## T3 — Mode-Matrix Preservation Check Raced Mirror Restoration
+
+**Status:** Fixed (2026-07-22)
+**Found:** managed on-demand/on-demand → on-demand/mirror transition
+**Where:** `scripts/fuse-acceptance.py`, managed mode switching
+
+**Repro:** Switch a populated managed folder from `ondemand` to `mirror`, wait
+until `sync list` reports mirror/idle, and immediately read a preservation file.
+The file can still be absent even though the subsequent restore pass downloads
+it correctly.
+
+**Cause:** The application commits the new mode before scheduling reconciliation
+and leaves the previous idle state and sync timestamp in the row. Those values
+describe the completed on-demand state, not a completed mirror restoration.
+
+**Fix:** The harness records every transition to mirror and explicitly requests
+and waits for a newer completed sync pass before validating local bytes. This is
+tracked as a harness synchronization defect; transient absence before that pass
+is expected asynchronous behavior, not data loss.
+
+---
+
+## T4 — Successful Managed Cleanup Left Its Own Sentinel Behind
+
+**Status:** Fixed (2026-07-22)
+**Found:** second consecutive managed acceptance run
+**Where:** `scripts/fuse-acceptance.py`, managed cleanup
+
+**Repro:** Complete the mode matrix in mirror/mirror, then immediately start a
+new managed run. Registration cleanup removed the sync-folder records and remote
+test folders, but the next run's empty-directory precheck found the harness's
+128 KiB preservation sentinel in each local directory.
+
+**Cause:** Removing a mirror registration intentionally preserves the local
+copy. The harness treated unregistering as if it also emptied local storage.
+
+**Fix:** After successful unregister, cleanup deletes only the sentinel whose
+contents match the per-run bytes generated by this harness. Unknown files remain
+untouched and continue to fail the strict precheck.
+
+---
+
+## B48 — Successful Unlink Can Reappear From a Stale Remote Listing
+
+**Status:** Fixed (unverified)
+**Found:** 2026-07-22, managed acceptance cleanup after all on-demand I/O passed
+**Where:** `crates/pdfs-fuse/src/lib.rs`, child enumeration and trash paths
+
+**Repro:** Unlink every child of a directory and immediately remove the
+directory. `unlink` returned success, but a parent invalidation followed by an
+eventually consistent remote enumeration returned the just-trashed child again.
+`rmdir` then returned `ENOTEMPTY`; repeated recursive removal could reproduce it
+for the entire acceptance timeout.
+
+**Cause:** Three stale-data paths combined. A queued cross-directory rename
+updated `State` but did not invalidate the kernel's cached empty destination
+listing, so recursive walkers never saw or unlinked the moved child. Removing a node recorded no
+authoritative local tombstone, so an eventually consistent Drive listing could
+intern it again. Separately, `State::has_children` treated a resident empty
+listing as inconclusive and fell back to obsolete SQLite child rows. Thus
+`readdir` could report empty while `rmdir` returned `ENOTEMPTY` forever.
+
+**Fix:** Every successful online or queued trash records its uid in a
+session-shared hidden set. Both persisted and remote child enumeration filter
+that set. Secondary on-demand mounts share it, and an explicit restore removes
+the uid so restored content can appear normally. `has_children` now treats any
+resident listing—including an empty one—as authoritative and consults SQLite
+only when the listing is absent. After replying successfully to a queued rename,
+the FUSE handler invalidates both exact dentries and both directory inodes; the
+ordering matters because an invalidation sent before the rename reply is
+overwritten by the kernel's response processing. Uids are immutable and never
+reused, making a session-long tombstone safe.
+
+---
+
+## B49 — Failed Conflict Preservation Still Overwrites the Local File (CRIT-05)
+
+**Status:** Fixed in code with deterministic regressions; live/fault verification pending
+**Found:** 2026-07-22, 1.0 data-safety audit
+**Where:** `crates/pdfs-fuse/src/sync.rs`, download/conflict handling
+
+**Cause:** If renaming a concurrently changed local file to a conflict copy
+fails, sync only logs the error and continues publishing the remote download
+over the local version. Second-resolution conflict names can also collide.
+
+**Required fix/test:** Abort download publication unless preservation succeeds;
+use collision-resistant names. Test existing names, permissions, `ENOSPC`,
+`EIO`, and cross-filesystem behavior. This is B39's failure path; B25 remains
+the separate weak-baseline problem.
+
+---
+
+## B50 — Failed Orphan Staging Deletes the Sole Scratch Copy (CRIT-06)
+
+**Status:** Fixed in code; injected filesystem-failure verification pending
+**Found:** 2026-07-22, 1.0 data-safety audit
+**Where:** `crates/pdfs-fuse/src/lib.rs`, `Core::stage_orphaned_write`
+
+**Cause:** If preserving an orphaned scratch write into staging fails, the
+failure path removes the original even though no other durable copy exists.
+
+**Required fix/test:** Never unlink the source until a durable replacement and
+metadata are committed. Inject `ENOSPC`, `EIO`, `EROFS`, permission, and
+cross-device failures and assert a byte-identical copy remains discoverable.
+Related: B23, B27, and B28.
+
+---
+
+## B51 — Scratch Sidecar Does Not Establish a Durable Recovery Record (CRIT-07)
+
+**Status:** Fixed in code; power-loss verification pending
+**Found:** 2026-07-22, 1.0 crash-consistency audit
+**Where:** `crates/pdfs-core/src/cache.rs`, scratch durability markers
+
+**Cause:** A temporary sidecar is renamed without syncing the sidecar first or
+the directory afterward. Power loss after successful user `fsync` can preserve
+bytes but lose authored-range metadata; recovery may discard the write or treat
+a sparse partial write as complete and replace untouched ranges with zeros.
+
+**Required fix/test:** Sync data, write and sync temporary metadata, rename it,
+then sync the directory before acknowledging durability. Crash-test every
+boundary and corrupt/partial sidecars. B20 and B23 do not cover this invariant.
+
+---
+
+## B52 — Staged-Write Publication Is Not Crash-Safe (CRIT-08)
+
+**Status:** Fixed in code; cross-filesystem and power-loss verification pending
+**Found:** 2026-07-22, 1.0 crash-consistency audit
+**Where:** `crates/pdfs-core/src/cache.rs`, staged-write publication
+
+**Cause:** Rename/copy and direct sidecar writes lack a complete sync protocol.
+The cross-filesystem fallback can remove its source after an unsynced copy,
+leaving no complete data/metadata pair after a crash.
+
+**Required fix/test:** Publish synced temporary data and metadata via atomic
+renames and directory syncs; delete the source only after verification. Inject
+failure after every write, sync, rename, DB enqueue, and deletion. Related: B50
+and B51.
+
+---
+
+## B53 — Superseding a Pending Operation Is Not Transactional (CRIT-09)
+
+**Status:** Fixed with rollback regression; crash/ENOSPC verification pending
+**Found:** 2026-07-22, 1.0 SQLite durability audit
+**Where:** `crates/pdfs-core/src/db/ops.rs`, `Db::enqueue_op`
+
+**Cause:** The old durable row is deleted before its replacement is inserted,
+outside a transaction. A crash, constraint error, full disk, or I/O failure can
+erase acknowledged upload work and orphan its blob.
+
+**Required fix/test:** Select and replace in one transaction; retain the old
+blob until commit and make cleanup recoverable. Inject insert/commit failures
+and prove that old or new operation—never neither—remains queued. B27 concerns
+a retained failing row; this issue erases the row.
+
+---
+
+## B54 — Total-Wipe Guard Does Not Protect a One-Entry Baseline (CRIT-10)
+
+**Status:** Fixed with one-entry regression; live mode-matrix verification pending
+**Found:** 2026-07-22, 1.0 mirror-sync audit
+**Where:** `crates/pdfs-fuse/src/sync/planner.rs`
+
+**Cause:** The safeguard activates only when the baseline has at least two
+paths. A one-file local folder that is missing, unreadable, or wrongly mounted
+can therefore trash its sole remote file.
+
+**Required fix/test:** Protect every non-empty baseline and require an
+independently complete scan plus an explicit deliberate-delete signal for a
+total wipe. Test absent roots, replaced mountpoints, permissions, and one-file
+folders in every mode combination.
+
+---
+
+## B55 — Incomplete Local Scans Are Interpreted as Deletions (CRIT-11)
+
+**Status:** Fixed in code; injected iterator/stat failure tests pending
+**Found:** 2026-07-22, 1.0 mirror-sync audit
+**Where:** `crates/pdfs-fuse/src/sync.rs`, local scan/reconciliation
+
+**Cause:** Directory-entry and metadata errors omit paths instead of marking the
+snapshot incomplete. Reconciliation can treat omissions as intentional local
+deletions and propagate them remotely.
+
+**Required fix/test:** Carry completeness/errors into planning; an incomplete
+subtree must make the pass non-destructive and retain its prior baseline. Test
+`readdir`, `stat`, permission, transient-I/O, and disappearing-entry failures.
+B42 covers the inverse local-delete failure.
+
+---
+
+## B56 — Database Schema Version Parsing Fails Open (HIGH-13)
+
+**Status:** Partly fixed — future schemas are refused; malformed-version handling remains open
+**Found:** 2026-07-22, 1.0 database audit
+**Where:** `crates/pdfs-core/src/db/migrations.rs`, `Db::migrate`
+
+**Cause:** Missing, malformed, or non-numeric version values become zero, while
+all future versions are accepted. Corruption can replay migrations, and an
+older binary can open a newer incompatible database.
+
+**Required fix/test:** Distinguish a new DB from corrupt metadata; reject
+malformed/future versions without mutation and validate required schema. Test
+empty, malformed, negative, future, partial/interrupted migration, and upgrade
+fixtures from every released schema.
+
+---
+
+## B57 — Event Cursor Can Advance Past Unapplied State (HIGH-14)
+
+**Status:** Partly fixed — cursor now advances per successfully invalidated event; DB/apply fault tests remain
+**Found:** 2026-07-22, 1.0 event-recovery audit
+**Where:** `crates/pdfs-fuse/src/background.rs`, `run_event_sync`
+
+**Cause:** Cache invalidation failures are ignored while the batch cursor still
+advances. Cursor-persistence failure also leaves the in-memory cursor moving;
+first-seed persistence failure can later reseed at a newer head. Applying state
+and committing its cursor have no atomic relationship.
+
+**Required fix/test:** Apply idempotently and persist derived DB state plus the
+cursor atomically, or advance only through the last durable event. Retry failed
+prerequisites. Crash/fault-test every event and cursor boundary, first seed,
+duplicate replay, malformed events, and database-full conditions.
+
+---
+
+## B58 — Control Socket Has Unbounded Frames and Connections (HIGH-15)
+
+**Status:** Fixed in code with frame-limit regressions; connection-flood verification pending
+**Found:** 2026-07-22, 1.0 IPC audit
+**Where:** `crates/pdfs-fuse/src/control.rs`
+
+**Cause:** The server uses unbounded `read_line`, a new thread per connection,
+and no concurrency or idle limit. A local process can exhaust memory, file
+descriptors, and threads, blocking filesystem control and safe shutdown.
+
+**Required fix/test:** Bound frames, clients, and idle lifetime. Test slow/no-
+newline clients, oversized/truncated JSON, floods, and recovery. B6/B35 cover
+socket authority and permissions, not resource bounds.
+
+---
+
+## B59 — Private State Permission Enforcement Fails Open (HIGH-16)
+
+**Status:** Fixed in code with symlink regression; wrong-owner integration verification pending
+**Found:** 2026-07-22, 1.0 local-privacy audit
+**Where:** `crates/pdfs-core/src/config.rs`, `AppDirs::ensure`
+
+**Cause:** chmod/creation failures are warned about or ignored while decrypted
+content, plaintext metadata, and the authenticated socket may remain accessible
+to other local users.
+
+**Required fix/test:** Before loading credentials or serving data, verify every
+sensitive directory is owned by the effective user, is not a symlink, and is
+owner-only. Test wrong owner, symlink substitution, read-only filesystems,
+permissive restored directories, and chmod failure. This tightens B6/B35.
+
+---
+
+## B60 — Config Writes Are Non-Atomic and Parse Errors Are Overwritten (HIGH-17)
+
+**Status:** Fixed in code; fault/concurrent-save verification pending
+**Found:** 2026-07-22, 1.0 configuration audit
+**Where:** `crates/pdfs-core/src/config.rs`, `load_config` / `save_config`
+
+**Cause:** Save truncates in place without sync. Load treats read/parse failure
+as absence and best-effort overwrites with defaults, silently losing mountpoint,
+ignore policy, budget, and explicit device adoption.
+
+**Required fix/test:** Atomically publish a restricted, synced temporary file
+and sync its directory. Preserve malformed input and report an actionable error.
+Test invalid/truncated JSON, permissions, `ENOSPC`, crash points, and concurrent
+saves.
+
+---
+
+## B61 — Rotated Single-Use Credentials Can Be Lost (HIGH-18)
+
+**Status:** Open — 1.0 authentication/recovery blocker
+**Found:** 2026-07-22, 1.0 authentication audit
+**Where:** `crates/pdfs-core/src/auth.rs`, refresh callback
+
+**Cause:** Failure to serialize or persist rotated credentials is only logged
+(and some setup failures are silent). The daemon continues in memory while the
+keyring retains an invalid single-use refresh token, so restart loses login.
+
+**Required fix/test:** Expose unhealthy persistence, retry with bounded backoff,
+notify the user, and define a safe re-login/shutdown path. Test locked/full/
+unavailable keyrings, callback races, repeated rotations, death, and restart.
+
+---
+
+## B62 — Debian Artifact Omits Required Service and Autostart Units (HIGH-19)
+
+**Status:** Fixed in workflow; clean-install VM verification pending
+**Found:** 2026-07-22, 1.0 packaging audit
+**Where:** `.github/workflows/release.yml`; `packaging/`
+
+**Cause:** The generated package omits `proton-drive.service` and the tray
+autostart entry even though normal login/GUI flows assume that lifecycle.
+
+**Required fix/test:** Package all units/resources with correct paths, modes,
+and hooks. In clean supported-distribution VMs test install, login, reboot,
+daemon/tray/FUSE, upgrade, rollback, and uninstall without deleting user state.
+
+---
+
+## B63 — Release Version Identity Is Inconsistent and Unenforced (HIGH-20)
+
+**Status:** Partly fixed — tag/workspace/PKGBUILD equality is enforced; protocol identity remains
+**Found:** 2026-07-22, 1.0 release audit
+**Where:** manifests, client identity constants, tags, release workflow
+
+**Cause:** Workspace/binary versions, protocol identity strings, and tags can
+disagree; the workflow trusts any `v*` tag without checking package metadata or
+installed `--version` output.
+
+**Required fix/test:** Establish one version source and fail CI on every tag,
+manifest, package, and binary mismatch. Document any intentionally independent
+protocol identity.
+
+---
+
+## B64 — Stable Publishing Lacks Data-Safety and Recovery Gates (CRIT-12)
+
+**Status:** Partly fixed — automated gates and managed-live matrix pass; recovery approval remains open
+**Found:** 2026-07-22, 1.0 release audit
+**Where:** CI and `.github/workflows/release.yml`
+
+**Cause:** A tag can publish without required format/lint/tests, offline and
+dedicated-account live acceptance, migrations, or crash/recovery drills. Fixes
+still marked unverified can ship as stable.
+
+**Required fix/test:** Publish immutable RC artifacts only after all workspace
+checks, full managed mode matrix, migration/power-loss recovery, clean install/
+upgrade tests, and explicit sign-off for every critical/high integrity issue.
+
+---
+
+## B65 — Release Artifacts Lack Supply-Chain Verification (MED-11)
+
+**Status:** Open — 1.0 release/security blocker
+**Found:** 2026-07-22, 1.0 supply-chain audit
+**Where:** CI and release workflows
+
+**Cause:** Releases do not enforce advisory/license policy or publish signed
+provenance, SBOMs, and verifiable checksums for exact distributed artifacts.
+
+**Required fix/test:** Add locked audit/license gates, SBOM and provenance,
+checksums and signatures, documented verification, and an independent clean-job
+verification/install test.
+
+---
+
+## B66 — Local-Only Pending Data Is Not Protected at Shutdown (HIGH-21)
+
+**Status:** Open — 1.0 data-safety/UX blocker
+**Found:** 2026-07-22, 1.0 recovery audit
+**Where:** shutdown, CLI/GUI status, logout/unregister workflows
+
+**Cause:** Undrained writes may exist only on one machine, but there is no flush
+command with a completion contract, prominent pending byte/count health, or
+shutdown/logout warning/inhibition.
+
+**Required fix/test:** Add `pdfs sync flush`, expose pending operations/bytes
+and last errors, and protect destructive lifecycle actions. Test offline stop,
+logout, unregister, reboot, retry exhaustion, and drain. Credential removal must
+never delete staged bytes.
+
+---
+
+## B67 — Fresh-State Restore Can Silently Select a New Device (HIGH-22)
+
+**Status:** Open — 1.0 recovery blocker
+**Found:** 2026-07-22, 1.0 disaster-recovery audit
+**Where:** device discovery/adoption; `docs/RECOVERY.md`
+
+**Cause:** Restore depends on hostname and folder basename. After local-state
+loss, a mismatch can silently create/select another device rather than attach
+the intended remote roots, making recovery appear complete while data is absent.
+
+**Required fix/test:** Require explicit adoption/confirmation on ambiguity and
+show old/proposed device and root IDs. Test state loss, hostname change,
+duplicates, renamed folders, multiple devices, and byte-identical restoration.
+
+---
+
+## B68 — Profile Backup Cannot Be Written to the Device Root
+
+**Status:** Open — live-reproduced 2026-07-22
+**Found:** complete managed-live mode matrix against the 1.0.0 release binary
+**Where:** `crates/pdfs-fuse/src/profile.rs`, profile backup destination
+
+**Repro:** Every sync-folder registration or mode transition attempted the
+profile backup and received `NotEnoughPermissions` (HTTP 422): `Cannot create
+file at the root of a device`. The filesystem matrix itself passed, but profile
+backup never became durable remotely.
+
+**Impact:** The recovery feature appears active but does not preserve the device
+profile, so a replacement machine still lacks modes, pins, mount settings, and
+the explicit remote-folder mapping. Repeated retries also add noisy warnings.
+
+**Required fix/test:** Store the profile below a writable application folder
+rather than directly at the device root, reuse that folder by stable identity,
+and make backup health visible. Verify upload, replacement, restart restore, and
+byte-identical fresh-state recovery on a dedicated account.

docs/TESTING.md

@@ -9,9 +9,21 @@ cargo test --workspace
 cargo clippy --workspace --all-targets -- -D warnings
 ```
 
-## Live FUSE acceptance suite
+## Filesystem API and live FUSE acceptance suite
 
-Kernel behavior and remote convergence need a real mount. The live suite runs
+The account-free suite exercises the filesystem syscall contract against a
+temporary local directory. This validates the runner and covers creation/open
+flags, positioned and vectored I/O, truncation and sparse files, `mmap`,
+`sendfile`, durability barriers, namespace/error semantics, open-file lifetime,
+names and enumeration, and concurrent I/O. It never reads application state or
+contacts Proton:
+
+```console
+scripts/fuse-acceptance.sh --offline-only
+```
+
+Kernel/FUSE behavior and remote convergence need a real mount. Live testing is
+explicitly opt-in and always runs the account-free contract first. It runs
 destructive POSIX operations only below a fresh `pdfs-acceptance-*` directory.
 It refuses paths that are not FUSE mounts and removes its directory on success,
 failure, or interruption.
@@ -19,16 +31,39 @@ failure, or interruption.
 Use a dedicated test account without irreplaceable data:
 
 ```console
-scripts/fuse-acceptance.sh /mnt/on-demand-testmount
+scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount
 ```
 
+### Automated mode matrix
+
+The managed live runner accepts two existing, empty, unmounted directories. It
+registers both as new sync folders, waits for initial synchronization, then
+tests all four pairs in order: on-demand/on-demand, on-demand/mirror,
+mirror/on-demand, and mirror/mirror. Every transition must become idle, and an
+on-demand folder must actually appear as a mount before its contract runs.
+
+```console
+mkdir -p /mnt/pdfs-test-a /mnt/pdfs-test-b
+scripts/fuse-acceptance.sh --managed-live /mnt/pdfs-test-a /mnt/pdfs-test-b
+```
+
+This creates remote folders and permanently deletes those test remotes during
+cleanup. It refuses non-empty paths, existing mounts, and paths already present
+in `pdfs sync list`. Use `PDFS_ACCEPTANCE_SYNC_TIMEOUT` to extend transition
+timeouts and `PDFS_ACCEPTANCE_PDFS` to select a non-installed CLI binary, for
+example `target/debug/pdfs`.
+
+During development, `PDFS_ACCEPTANCE_ONLY` runs tests whose descriptive name
+contains the supplied text (for example `PDFS_ACCEPTANCE_ONLY=namespace`). A
+release acceptance run must leave it unset so the complete contract executes.
+
 Passing more paths runs the same filesystem suite independently against each.
 The first path must be the FUSE mount under test; later paths may be another
 FUSE mount or a normal local mirror filesystem:
 
 ```console
 PDFS_ACCEPTANCE_SYNC_TIMEOUT=180 \
-  scripts/fuse-acceptance.sh /mnt/on-demand-testmount /mnt/testmount
+  scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount /mnt/testmount
 ```
 
 If two paths really are views of the **same remote folder**, enable an
@@ -36,16 +71,15 @@ additional byte-for-byte convergence check explicitly:
 
 ```console
 PDFS_ACCEPTANCE_CONVERGENCE=1 \
-  scripts/fuse-acceptance.sh /mnt/first-view /mnt/second-view
+  scripts/fuse-acceptance.sh --live /mnt/first-view /mnt/second-view
 ```
 
 Do not enable that flag merely because two sync folders belong to the same
 account. Their configured remote UIDs must be identical.
 
-The suite covers create/read/rewrite/truncate, multi-megabyte binary content,
-unaligned random writes, rename replacement, cross-directory moves, important
-POSIX error cases, open-then-unlink behavior, concurrent writers, directory
-enumeration, and optional cross-mount byte-for-byte convergence. A pass is
+Normally the live suite applies the same API contract independently to each
+supplied mount. In convergence mode it exercises the primary once and verifies
+that the resulting bytes appear through every secondary view. A pass is
 strong evidence for those paths, not a blanket data-loss guarantee. Before a
 release, also run the recovery drills in `docs/RECOVERY.md` and inspect
 `docs/BUGS.md` for live-verification items.

packaging/PKGBUILD

@@ -9,7 +9,7 @@
 # Build + install:   cd packaging && makepkg -fi
 
 pkgname=proton-drive-linux
-pkgver=0.7.1
+pkgver=1.0.0
 pkgrel=1
 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
 arch=('x86_64')

scripts/fuse-acceptance.py

@@ -0,0 +1,572 @@
+#!/usr/bin/env python3
+"""Account-free filesystem API contract plus opt-in live FUSE acceptance."""
+
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import errno
+import hashlib
+import mmap
+import os
+from pathlib import Path
+import shutil
+import stat
+import subprocess
+import tempfile
+import time
+import uuid
+
+
+def check(condition: bool, message: str) -> None:
+    if not condition:
+        raise AssertionError(message)
+
+
+def expect_errno(expected: set[int], operation, description: str) -> None:
+    try:
+        operation()
+    except OSError as error:
+        check(error.errno in expected, f"{description}: errno {error.errno}, expected {expected}")
+    else:
+        raise AssertionError(f"{description}: unexpectedly succeeded")
+
+
+def read(path: Path) -> bytes:
+    with path.open("rb", buffering=0) as file:
+        return file.read()
+
+
+def write_durable(path: Path, data: bytes) -> None:
+    fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
+    try:
+        view = memoryview(data)
+        while view:
+            written = os.write(fd, view)
+            check(written > 0, "write made no progress")
+            view = view[written:]
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def test_create_flags(root: Path) -> None:
+    path = root / "flags"
+    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)
+    os.write(fd, b"abcdef")
+    os.close(fd)
+    expect_errno({errno.EEXIST}, lambda: os.open(path, os.O_CREAT | os.O_EXCL), "O_EXCL")
+    fd = os.open(path, os.O_WRONLY | os.O_APPEND)
+    os.lseek(fd, 0, os.SEEK_SET)
+    os.write(fd, b"++")
+    os.close(fd)
+    check(read(path) == b"abcdef++", "O_APPEND ignored")
+    fd = os.open(path, os.O_WRONLY | os.O_TRUNC)
+    os.close(fd)
+    check(path.stat().st_size == 0, "O_TRUNC did not truncate")
+
+
+def test_positioned_and_vectored_io(root: Path) -> None:
+    path = root / "positioned.bin"
+    base = bytes(range(256)) * 32768
+    write_durable(path, base)
+    fd = os.open(path, os.O_RDWR)
+    try:
+        check(os.pread(fd, 19, 4093) == base[4093:4112], "pread returned wrong range")
+        check(os.pwrite(fd, b"boundary-write", 4093) == 14, "short pwrite")
+        if hasattr(os, "pwritev"):
+            check(os.pwritev(fd, [b"vector-", b"write"], 65531) == 12, "short pwritev")
+        if hasattr(os, "preadv"):
+            chunks = [bytearray(7), bytearray(5)]
+            check(os.preadv(fd, chunks, 65531) == 12, "short preadv")
+            check(b"".join(chunks) == b"vector-write", "preadv data mismatch")
+        os.fdatasync(fd)
+    finally:
+        os.close(fd)
+    expected = bytearray(base)
+    expected[4093:4107] = b"boundary-write"
+    expected[65531:65543] = b"vector-write"
+    got = read(path)
+    check(got == expected, "positioned write damaged surrounding bytes")
+    print(f"    sha256 {hashlib.sha256(got).hexdigest()}")
+
+
+def test_resize_and_sparse_io(root: Path) -> None:
+    path = root / "resize"
+    write_durable(path, b"0123456789")
+    os.truncate(path, 4)
+    check(read(path) == b"0123", "shrink did not preserve prefix")
+    os.truncate(path, 8193)
+    data = read(path)
+    check(data[:4] == b"0123" and data[4:] == bytes(8189), "grown range is not zero-filled")
+
+    sparse = root / "sparse"
+    fd = os.open(sparse, os.O_CREAT | os.O_RDWR, 0o600)
+    try:
+        os.pwrite(fd, b"head", 0)
+        os.pwrite(fd, b"tail", 8 * 1024 * 1024 + 17)
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    check(sparse.stat().st_size == 8 * 1024 * 1024 + 21, "sparse size mismatch")
+    fd = os.open(sparse, os.O_RDONLY)
+    try:
+        check(os.pread(fd, 8, 4) == bytes(8), "sparse hole is not zero-filled")
+        check(os.pread(fd, 4, 8 * 1024 * 1024 + 17) == b"tail", "sparse tail missing")
+        check(os.pread(fd, 1, sparse.stat().st_size + 4096) == b"", "read beyond EOF not empty")
+    finally:
+        os.close(fd)
+
+
+def test_mmap_and_copy_paths(root: Path) -> None:
+    source = root / "mapped"
+    data = bytearray((b"mmap-and-sendfile\0" * 65536)[:1024 * 1024])
+    write_durable(source, data)
+    fd = os.open(source, os.O_RDWR)
+    try:
+        with mmap.mmap(fd, len(data), access=mmap.ACCESS_WRITE) as mapping:
+            mapping[4091:4107] = b"mapped-boundary!"
+            mapping.flush()
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    data[4091:4107] = b"mapped-boundary!"
+    check(read(source) == data, "shared mmap write/read mismatch")
+
+    target = root / "sendfile-copy"
+    src = os.open(source, os.O_RDONLY)
+    dst = os.open(target, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
+    try:
+        offset = 0
+        while offset < len(data):
+            count = os.sendfile(dst, src, offset, len(data) - offset)
+            check(count > 0, "sendfile made no progress")
+            offset += count
+        os.fsync(dst)
+    finally:
+        os.close(src)
+        os.close(dst)
+    check(read(target) == data, "sendfile copy mismatch")
+
+
+def test_namespace_and_errors(root: Path) -> None:
+    a, b = root / "dir-a", root / "dir-b"
+    a.mkdir(); b.mkdir()
+    (a / "nested").mkdir()
+    write_durable(a / "nested" / "child", b"child")
+    write_durable(root / "source", b"source")
+    write_durable(root / "victim", b"victim")
+    os.replace(root / "source", root / "victim")
+    check(read(root / "victim") == b"source" and not (root / "source").exists(), "replace failed")
+    os.rename(root / "victim", b / "moved")
+    check(read(b / "moved") == b"source", "cross-directory move failed")
+    check("moved" in os.listdir(b), "cross-directory move missing from readdir")
+    os.rename(b / "moved", b / "moved")
+    check(read(b / "moved") == b"source", "same-name rename changed data")
+
+    expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.rmdir(a), "rmdir(non-empty)")
+    expect_errno({errno.ENOTDIR}, lambda: os.rmdir(b / "moved"), "rmdir(file)")
+    expect_errno({errno.EISDIR, errno.EPERM}, lambda: os.unlink(a), "unlink(directory)")
+    expect_errno({errno.EINVAL}, lambda: os.rename(a, a / "nested" / "cycle"), "directory cycle")
+    expect_errno({errno.ENOENT}, lambda: os.unlink(root / "absent"), "unlink(absent)")
+    expect_errno({errno.ENOENT}, lambda: os.rename(root / "absent", root / "new"), "rename(absent)")
+
+    empty_src, empty_dst = root / "empty-src", root / "empty-dst"
+    empty_src.mkdir(); empty_dst.mkdir()
+    os.replace(empty_src, empty_dst)
+    check(empty_dst.is_dir() and not empty_src.exists(), "empty directory replacement failed")
+    nonempty = root / "nonempty-dst"
+    nonempty.mkdir(); write_durable(nonempty / "keep", b"keep")
+    replacement = root / "replacement-dir"; replacement.mkdir()
+    expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.replace(replacement, nonempty), "replace non-empty dir")
+    check(read(nonempty / "keep") == b"keep", "failed replacement damaged destination")
+
+
+def test_open_lifetime(root: Path) -> None:
+    path = root / "open-unlink"
+    data = b"held-open\0" * 4096
+    write_durable(path, data)
+    fd = os.open(path, os.O_RDWR)
+    os.unlink(path)
+    check(not path.exists(), "unlinked name remains visible")
+    check(os.pread(fd, len(data), 0) == data, "open file lost data after unlink")
+    os.pwrite(fd, b"still-open", 0)
+    check(os.pread(fd, 10, 0) == b"still-open", "unlinked open file is not writable")
+    os.close(fd)
+
+    old, new = root / "open-rename-old", root / "open-rename-new"
+    write_durable(old, b"before")
+    fd = os.open(old, os.O_RDWR)
+    os.rename(old, new)
+    os.pwrite(fd, b"after!", 0)
+    os.fsync(fd); os.close(fd)
+    check(not old.exists() and read(new) == b"after!", "open handle did not follow rename")
+
+
+def test_names_and_enumeration(root: Path) -> None:
+    names = ["space name.txt", "unicodé-文件", ".hidden", "trailing.dot.", "case", "CASE"]
+    for index, name in enumerate(names):
+        write_durable(root / name, f"name-{index}".encode())
+    listed = set(os.listdir(root))
+    check(set(names) <= listed, "directory enumeration omitted valid names")
+    for name in listed:
+        os.lstat(root / name)
+    expect_errno({errno.ENAMETOOLONG}, lambda: os.open(root / ("x" * 256), os.O_CREAT), "overlong name")
+
+
+def test_concurrency(root: Path) -> None:
+    def independent(index: int) -> None:
+        data = bytes([index]) * (131072 + index)
+        path = root / f"concurrent-{index}"
+        write_durable(path, data)
+        check(read(path) == data, f"concurrent file {index} mismatch")
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(independent, range(1, 17)))
+
+    shared = root / "concurrent-ranges"
+    extent = 64 * 1024
+    write_durable(shared, bytes(extent * 8))
+    fd = os.open(shared, os.O_RDWR)
+    try:
+        def positioned(index: int) -> None:
+            payload = bytes([index + 1]) * extent
+            check(os.pwrite(fd, payload, index * extent) == extent, f"short concurrent pwrite {index}")
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+            list(pool.map(positioned, range(8)))
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    expected = b"".join(bytes([i + 1]) * extent for i in range(8))
+    check(read(shared) == expected, "concurrent disjoint writes overlapped or vanished")
+
+
+TESTS = [
+    ("creation and open flags", test_create_flags),
+    ("positioned and vectored I/O", test_positioned_and_vectored_io),
+    ("truncate, growth, sparse ranges, and EOF", test_resize_and_sparse_io),
+    ("mmap and zero-copy copy paths", test_mmap_and_copy_paths),
+    ("namespace operations and POSIX errors", test_namespace_and_errors),
+    ("open-handle lifetime across unlink and rename", test_open_lifetime),
+    ("names, lookup, stat, and enumeration", test_names_and_enumeration),
+    ("independent and shared-file concurrency", test_concurrency),
+]
+
+
+def run_contract(parent: Path, label: str) -> tuple[Path, str]:
+    root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
+    root.mkdir()
+    print(f"[target] {label}: {parent}")
+    try:
+        selected = os.environ.get("PDFS_ACCEPTANCE_ONLY")
+        tests = TESTS if not selected else [test for test in TESTS if selected.lower() in test[0].lower()]
+        check(bool(tests), f"PDFS_ACCEPTANCE_ONLY={selected!r} matched no tests")
+        for name, operation in tests:
+            print(f"  [test] {name}")
+            operation(root)
+        digest_path = root / "positioned.bin"
+        digest = hashlib.sha256(read(digest_path)).hexdigest() if digest_path.exists() else ""
+        return root, digest
+    except BaseException:
+        shutil.rmtree(root, ignore_errors=True)
+        raise
+
+
+def wait_for_copy(root: Path, relative: Path, digest: str, timeout: int) -> None:
+    deadline = time.monotonic() + timeout
+    target = root / relative
+    while time.monotonic() < deadline:
+        try:
+            if target.is_file() and hashlib.sha256(read(target)).hexdigest() == digest:
+                return
+        except OSError:
+            pass
+        time.sleep(2)
+    raise AssertionError(f"{target} did not converge byte-for-byte within {timeout}s")
+
+
+class ManagedSyncPair:
+    """Two sync registrations created by this run and removed on exit."""
+
+    def __init__(self, paths: list[Path], timeout: int) -> None:
+        self.paths = [path.resolve() for path in paths]
+        self.timeout = timeout
+        self.pdfs = os.environ.get("PDFS_ACCEPTANCE_PDFS", "pdfs")
+        self.ids: dict[Path, int] = {}
+        self.sentinels = {
+            path: hashlib.sha256(f"pdfs-preservation:{path}:{uuid.uuid4()}".encode()).digest()
+            * 4096
+            for path in self.paths
+        }
+
+    def command(self, *args: str, json_output: bool = False) -> str:
+        command = [self.pdfs]
+        if json_output:
+            command.append("--json")
+        command.extend(args)
+        result = subprocess.run(command, text=True, capture_output=True)
+        if result.returncode:
+            detail = result.stderr.strip() or result.stdout.strip()
+            raise RuntimeError(f"{' '.join(command)} failed: {detail}")
+        return result.stdout
+
+    def folders(self) -> list[dict]:
+        import json
+        value = json.loads(self.command("sync", "list", json_output=True))
+        return value["items"]
+
+    def wait_for_queue(self) -> None:
+        import json
+        deadline = time.monotonic() + self.timeout
+        last = None
+        while time.monotonic() < deadline:
+            value = json.loads(self.command("status", json_output=True))
+            last = value.get("mount") or {}
+            if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0:
+                return
+            time.sleep(1)
+        raise TimeoutError(f"daemon mutation queue did not drain: {last}")
+
+    def remove_tree(self, root: Path) -> None:
+        deadline = time.monotonic() + self.timeout
+        while True:
+            try:
+                shutil.rmtree(root)
+                return
+            except FileNotFoundError:
+                return
+            except OSError as error:
+                if error.errno != errno.ENOTEMPTY or time.monotonic() >= deadline:
+                    survivors = []
+                    if root.exists():
+                        for directory, dirs, files in os.walk(root):
+                            survivors.extend(str(Path(directory) / name) for name in dirs + files)
+                    raise OSError(
+                        error.errno,
+                        f"{error.strerror}; surviving test entries: {survivors[:50]}",
+                        error.filename,
+                    ) from error
+                # A queued namespace operation can land between rmtree's
+                # enumeration and rmdir. Re-walk the test-owned tree.
+                time.sleep(0.25)
+
+    def validate(self) -> None:
+        if shutil.which(self.pdfs) is None and not Path(self.pdfs).is_file():
+            raise RuntimeError(
+                f"cannot find {self.pdfs!r}; set PDFS_ACCEPTANCE_PDFS to the CLI binary"
+            )
+        if self.paths[0] == self.paths[1]:
+            raise ValueError("managed live paths must be different directories")
+        deadline = time.monotonic() + self.timeout
+        last_error = None
+        while True:
+            try:
+                current = self.folders()
+                break
+            except Exception as error:
+                last_error = error
+                if time.monotonic() >= deadline:
+                    raise TimeoutError(
+                        f"pdfs daemon did not become ready within {self.timeout}s: {last_error}"
+                    ) from error
+                time.sleep(1)
+        registered = {Path(item["local_path"]).resolve() for item in current}
+        for path in self.paths:
+            check(path.is_dir(), f"managed path is not a directory: {path}")
+            check(not any(path.iterdir()), f"managed path is not empty: {path}")
+            check(path not in registered, f"managed path is already registered: {path}")
+            check(not is_mountpoint(path), f"managed path is already a mount: {path}")
+
+    def wait_for(self, path: Path, *, mode: str | None = None) -> dict:
+        deadline = time.monotonic() + self.timeout
+        last = None
+        while time.monotonic() < deadline:
+            for item in self.folders():
+                if Path(item["local_path"]).resolve() != path:
+                    continue
+                last = item
+                if (
+                    item["state"] == "idle"
+                    and item.get("pending_mode") is None
+                    and (mode is None or item["mode"] == mode)
+                ):
+                    return item
+                if item["state"] in {"error", "conflict"}:
+                    raise RuntimeError(f"sync folder {path} entered {item['state']}: {item}")
+            time.sleep(2)
+        raise TimeoutError(f"sync folder {path} did not become idle in mode {mode}: {last}")
+
+    def create(self) -> None:
+        self.validate()
+        for path in self.paths:
+            print(f"[setup] registering empty sync folder {path}")
+            self.command("sync", "add", str(path))
+            item = self.wait_for(path, mode="mirror")
+            self.ids[path] = int(item["id"])
+        for path in self.paths:
+            write_durable(path / "pdfs-mode-preservation.bin", self.sentinels[path])
+            self.force_sync(path)
+
+    def force_sync(self, path: Path) -> None:
+        before = int(self.wait_for(path)["last_sync"])
+        # last_sync has one-second resolution. Ensure this requested pass cannot
+        # finish with the same timestamp and look indistinguishable from no pass.
+        while int(time.time()) <= before:
+            time.sleep(0.1)
+        self.command("sync", "now", str(self.ids[path]))
+        deadline = time.monotonic() + self.timeout
+        last = None
+        while time.monotonic() < deadline:
+            last = next(
+                (item for item in self.folders() if Path(item["local_path"]).resolve() == path),
+                None,
+            )
+            if last and last["state"] == "idle" and int(last["last_sync"]) > before:
+                return
+            if last and last["state"] in {"error", "conflict"}:
+                raise RuntimeError(f"forced sync for {path} entered {last['state']}: {last}")
+            time.sleep(1)
+        raise TimeoutError(f"forced sync for {path} did not complete: {last}")
+
+    def set_modes(self, modes: tuple[str, str]) -> None:
+        changed_to_mirror: set[Path] = set()
+        for path, mode in zip(self.paths, modes, strict=True):
+            current = self.wait_for(path)
+            if current["mode"] != mode:
+                print(f"[setup] switching {path} to {mode}")
+                self.command("sync", "mode", str(self.ids[path]), mode)
+                if mode == "mirror":
+                    changed_to_mirror.add(path)
+        for path, mode in zip(self.paths, modes, strict=True):
+            self.wait_for(path, mode=mode)
+            # The mode row flips before the asynchronous restore pass starts,
+            # and its prior idle/last_sync values remain visible meanwhile.
+            # Demand a completed pass before inspecting restored local bytes.
+            if path in changed_to_mirror:
+                self.force_sync(path)
+            mounted = is_mountpoint(path)
+            check(mounted == (mode == "ondemand"), f"{path}: mode is {mode}, mounted={mounted}")
+            check(
+                read(path / "pdfs-mode-preservation.bin") == self.sentinels[path],
+                f"{path}: preservation sentinel changed or vanished after switch to {mode}",
+            )
+
+    def cleanup(self) -> None:
+        # An async add can create its row and then time out before create() has
+        # recorded the id. validate() proved these exact paths were unregistered
+        # at entry, so rows now at those paths belong to this run.
+        try:
+            for item in self.folders():
+                path = Path(item["local_path"]).resolve()
+                if path in self.paths:
+                    self.ids[path] = int(item["id"])
+        except Exception as error:
+            print(f"WARNING: could not rediscover managed sync folders: {error}")
+        for path, folder_id in reversed(list(self.ids.items())):
+            try:
+                print(f"[cleanup] removing test sync folder {path}")
+                self.command("sync", "rm", str(folder_id), "--delete-remote")
+            except Exception as error:
+                print(f"WARNING: cleanup failed for sync folder {folder_id}: {error}")
+                continue
+            # Unregistering a mirror intentionally preserves its local copy.
+            # Remove only the sentinel owned by this harness so the next run's
+            # strict empty-directory precondition remains meaningful. Unknown
+            # survivors are left untouched and will fail validate() next time.
+            sentinel = path / "pdfs-mode-preservation.bin"
+            try:
+                if sentinel.exists() and read(sentinel) == self.sentinels[path]:
+                    sentinel.unlink()
+            except OSError as error:
+                print(f"WARNING: could not remove managed sentinel {sentinel}: {error}")
+
+
+def is_mountpoint(path: Path) -> bool:
+    """True for a distinct mount, including FUSE mounts over an existing dir."""
+    return os.path.ismount(path)
+
+
+def run_managed_matrix(paths: list[Path]) -> None:
+    timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180"))
+    pair = ManagedSyncPair(paths, timeout)
+    roots: list[Path] = []
+    try:
+        pair.create()
+        matrices = [
+            ("ondemand", "ondemand"),
+            ("ondemand", "mirror"),
+            ("mirror", "ondemand"),
+            ("mirror", "mirror"),
+        ]
+        for modes in matrices:
+            print(f"[matrix] first={modes[0]}, second={modes[1]}")
+            pair.set_modes(modes)
+            for path, mode in zip(pair.paths, modes, strict=True):
+                root, _ = run_contract(path, f"managed {mode}")
+                roots.append(root)
+                pair.wait_for_queue()
+                pair.remove_tree(root)
+                roots.remove(root)
+                pair.wait_for_queue()
+                # Mirror changes are asynchronous; force and await a pass before
+                # changing its mode so authored bytes/deletions cannot be lost.
+                if mode == "mirror":
+                    pair.force_sync(path)
+    finally:
+        for root in roots:
+            shutil.rmtree(root, ignore_errors=True)
+        pair.cleanup()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    live_mode = parser.add_mutually_exclusive_group()
+    live_mode.add_argument("--live", nargs="+", type=Path, metavar="MOUNTPOINT")
+    live_mode.add_argument(
+        "--managed-live",
+        nargs=2,
+        type=Path,
+        metavar=("EMPTY_DIR_A", "EMPTY_DIR_B"),
+    )
+    args = parser.parse_args()
+
+    with tempfile.TemporaryDirectory(prefix="pdfs-api-reference-") as directory:
+        reference, _ = run_contract(Path(directory), "local filesystem API reference (no account)")
+        shutil.rmtree(reference)
+    print("[pass] account-free filesystem API contract")
+
+    if args.managed_live:
+        run_managed_matrix(args.managed_live)
+        print("PASS: offline API and managed live mode-matrix suites completed")
+        return
+
+    if not args.live:
+        print("PASS: offline acceptance suite completed; live FUSE test was not requested")
+        return
+
+    completed: list[tuple[Path, Path, str]] = []
+    convergence = os.environ.get("PDFS_ACCEPTANCE_CONVERGENCE", "0") == "1"
+    try:
+        # Views of one remote folder must not independently create the same
+        # fixtures. Exercise the primary, then observe that exact tree through
+        # every secondary. Without convergence mode, each mount is independent
+        # and receives the full contract.
+        targets = args.live[:1] if convergence and len(args.live) > 1 else args.live
+        for mountpoint in targets:
+            root, digest = run_contract(mountpoint.resolve(), "live FUSE")
+            completed.append((mountpoint.resolve(), root, digest))
+        if convergence and len(args.live) > 1:
+            print("  [test] cross-mount remote convergence")
+            timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "120"))
+            source_root, digest = completed[0][1], completed[0][2]
+            relative = source_root.relative_to(completed[0][0]) / "positioned.bin"
+            for mountpoint in args.live[1:]:
+                wait_for_copy(mountpoint.resolve(), relative, digest, timeout)
+    finally:
+        for _, root, _ in completed:
+            shutil.rmtree(root, ignore_errors=True)
+    print("PASS: offline API and live FUSE acceptance suites completed")
+
+
+if __name__ == "__main__":
+    main()

scripts/fuse-acceptance.sh

@@ -1,210 +1,47 @@
 #!/usr/bin/env bash
-# Destructive only inside a uniquely named directory created by this script.
+# Filesystem syscall contract, with optional tests against mounted Drive views.
 set -Eeuo pipefail
 
 usage() {
-    echo "usage: $0 MOUNTPOINT [MOUNTPOINT ...]" >&2
-    echo "example: $0 /mnt/on-demand-testmount /mnt/testmount" >&2
+    cat >&2 <<'EOF'
+usage: scripts/fuse-acceptance.sh [--offline-only]
+       scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...]
+       scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR
+       scripts/fuse-acceptance.sh MOUNTPOINT [MOUNTPOINT ...]  # compatibility
+
+The local reference suite always runs first and requires no account. --live then
+runs the same contract on each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only
+when all live mountpoints show the same remote folder.
+EOF
     exit 2
 }
 
-(( $# > 0 )) || usage
 command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; }
-command -v findmnt >/dev/null || { echo "findmnt is required" >&2; exit 2; }
+script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
 
-mounts=("$@")
-run_id="pdfs-acceptance-$(date +%Y%m%dT%H%M%S)-$$"
-roots=()
+case "${1:-}" in
+    -h|--help) usage ;;
+    --offline-only|"")
+        (( $# <= 1 )) || usage
+        exec python3 -u "$script_dir/fuse-acceptance.py"
+        ;;
+    --live) shift ; (( $# > 0 )) || usage ;;
+    --managed-live)
+        shift
+        (( $# == 2 )) || usage
+        exec python3 -u "$script_dir/fuse-acceptance.py" --managed-live "$@"
+        ;;
+    --*) usage ;;
+esac
 
-cleanup() {
-    local root
-    for root in "${roots[@]}"; do
-        [[ -n "$root" && "$root" == */pdfs-acceptance-* ]] || continue
-        rm -rf -- "$root" 2>/dev/null || true
-    done
-}
-trap cleanup EXIT INT TERM
-
-for index in "${!mounts[@]}"; do
-    mountpoint="${mounts[$index]}"
+command -v findmnt >/dev/null || { echo "findmnt is required for --live" >&2; exit 2; }
+for mountpoint in "$@"; do
     [[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; }
-    fs_type="$(findmnt -T "$mountpoint" -n -o FSTYPE | head -n 1)"
-    if (( index == 0 )) && [[ "$fs_type" != fuse* ]]; then
-        echo "refusing primary non-FUSE path $mountpoint (type: ${fs_type:-unknown})" >&2
-        exit 2
-    fi
-    roots+=("${mountpoint%/}/$run_id")
 done
-
-step() { printf '[test] %s\n' "$1"; }
-assert_text() {
-    python3 - "$1" "$2" <<'PY'
-import sys
-p, expected = sys.argv[1], sys.argv[2].encode()
-with open(p, "rb") as f: got = f.read()
-if got != expected:
-    raise AssertionError(
-        f"{p}: expected {expected!r} ({len(expected)} bytes), "
-        f"got {got!r} ({len(got)} bytes)"
-    )
-PY
+fstype="$(findmnt -T "$1" -n -o FSTYPE | head -n 1)"
+[[ "$fstype" == fuse* ]] || {
+    echo "refusing primary non-FUSE path $1 (type: ${fstype:-unknown})" >&2
+    exit 2
 }
 
-run_suite() {
-local primary="$1"
-printf '[mount] testing %s\n' "${primary%/*}"
-mkdir -- "$primary"
-
-step "create, read, overwrite, truncate"
-printf 'alpha\n' > "$primary/plain.txt"
-assert_text "$primary/plain.txt" $'alpha\n'
-printf 'a much longer replacement\n' > "$primary/plain.txt"
-truncate -s 4 "$primary/plain.txt"
-assert_text "$primary/plain.txt" "a mu"
-
-step "binary round-trip and random-access writes"
-python3 - "$primary/binary.dat" <<'PY'
-import hashlib, sys
-p = sys.argv[1]
-data = bytes(range(256)) * 16384
-with open(p, "wb") as f: f.write(data)
-with open(p, "r+b", buffering=0) as f:
-    f.seek(4093); f.write(b"boundary-write")
-with open(p, "rb") as f: got = f.read()
-expected = data[:4093] + b"boundary-write" + data[4107:]
-assert got == expected
-print(hashlib.sha256(got).hexdigest())
-PY
-
-step "media-style header and distant seek reads"
-python3 - "$primary/seek-sample.mkv" <<'PY'
-import os, sys
-
-p = sys.argv[1]
-size = 12 * 1024 * 1024 + 137
-header = b"\x1aE\xdf\xa3pdfs-media-header"
-middle = b"pdfs-middle-cue"
-trailer = b"pdfs-media-trailer"
-middle_at = 7 * 1024 * 1024 + 19
-trailer_at = size - len(trailer)
-
-# A player probes the container header, jumps to cues/index data, then commonly
-# seeks near EOF.  Write a sparse fixture so the acceptance test stays cheap,
-# and issue separate unbuffered reads so FUSE receives genuinely distant ranges
-# rather than one accidental sequential read.
-with open(p, "wb", buffering=0) as f:
-    f.truncate(size)
-    f.seek(0); f.write(header)
-    f.seek(middle_at); f.write(middle)
-    f.seek(trailer_at); f.write(trailer)
-
-with open(p, "rb", buffering=0) as f:
-    assert f.read(len(header)) == header
-    f.seek(middle_at); assert f.read(len(middle)) == middle
-    f.seek(trailer_at); assert f.read(len(trailer)) == trailer
-    assert f.read(1) == b""
-    f.seek(size + 4096); assert f.read(16) == b""
-assert os.stat(p).st_size == size
-PY
-
-step "rename, move, and replacement semantics"
-mkdir -- "$primary/dir-a" "$primary/dir-b"
-printf 'source' > "$primary/source"
-printf 'victim' > "$primary/victim"
-mv -- "$primary/source" "$primary/victim"
-assert_text "$primary/victim" "source"
-[[ ! -e "$primary/source" ]]
-mv -- "$primary/victim" "$primary/dir-a/moved"
-assert_text "$primary/dir-a/moved" "source"
-
-step "POSIX error behavior"
-if rmdir -- "$primary/dir-a/moved" 2>/dev/null; then echo "rmdir removed a file" >&2; exit 1; fi
-if rm -- "$primary/dir-a" 2>/dev/null; then echo "rm removed a directory" >&2; exit 1; fi
-if mv -- "$primary/dir-a" "$primary/dir-a/child" 2>/dev/null; then
-    echo "moving a directory below itself succeeded" >&2; exit 1
-fi
-
-step "open file remains readable after unlink"
-python3 - "$primary/open-unlink" <<'PY'
-import os, sys
-p = sys.argv[1]; data = b"held-open\0" * 4096
-with open(p, "wb") as f: f.write(data)
-f = open(p, "rb", buffering=0); os.unlink(p)
-assert not os.path.exists(p) and f.read() == data
-f.close()
-PY
-
-step "concurrent independent writers"
-for i in $(seq 1 12); do
-    (python3 - "$primary/concurrent-$i" "$i" <<'PY'
-import sys
-p, seed = sys.argv[1], int(sys.argv[2]); data = bytes([seed]) * (131072 + seed)
-with open(p, "wb") as f: f.write(data)
-with open(p, "rb") as f: assert f.read() == data
-PY
-    ) &
-done
-wait
-
-step "directory enumeration agrees with stat"
-python3 - "$primary" <<'PY'
-import os, sys
-for name in os.listdir(sys.argv[1]): os.lstat(os.path.join(sys.argv[1], name))
-PY
-}
-
-for root in "${roots[@]}"; do
-    run_suite "$root"
-done
-
-if [[ "${PDFS_ACCEPTANCE_CONVERGENCE:-0}" == 1 ]] && (( ${#roots[@]} > 1 )); then
-    step "cross-mount convergence"
-    primary="${roots[0]}"
-    expected="$(sha256sum "$primary/binary.dat" | cut -d' ' -f1)"
-    for secondary in "${roots[@]:1}"; do
-        deadline=$((SECONDS + ${PDFS_ACCEPTANCE_SYNC_TIMEOUT:-120}))
-        while (( SECONDS < deadline )); do
-            if [[ -f "$secondary/binary.dat" ]] &&
-               [[ "$(sha256sum "$secondary/binary.dat" 2>/dev/null | cut -d' ' -f1)" == "$expected" ]]; then break; fi
-            sleep 2
-        done
-        [[ -f "$secondary/binary.dat" ]] || { echo "did not appear on $secondary" >&2; exit 1; }
-        [[ "$(sha256sum "$secondary/binary.dat" | cut -d' ' -f1)" == "$expected" ]] || {
-            echo "content mismatch on $secondary" >&2; exit 1
-        }
-
-        # The media fixture may drain after binary.dat.  Wait independently,
-        # then probe only its sparse ranges on the other mount.  Unlike the
-        # same-mount check above, this exercises remote FUSE hydration and seek
-        # reads instead of serving the still-local staged file.
-        deadline=$((SECONDS + ${PDFS_ACCEPTANCE_SYNC_TIMEOUT:-120}))
-        while (( SECONDS < deadline )); do
-            [[ -f "$secondary/seek-sample.mkv" ]] &&
-                [[ "$(stat -c %s "$secondary/seek-sample.mkv" 2>/dev/null)" == 12583049 ]] && break
-            sleep 2
-        done
-        [[ -f "$secondary/seek-sample.mkv" ]] || {
-            echo "media fixture did not appear on $secondary" >&2; exit 1
-        }
-        python3 - "$secondary/seek-sample.mkv" <<'PY'
-import os, sys
-p = sys.argv[1]
-size = 12 * 1024 * 1024 + 137
-checks = [
-    (0, b"\x1aE\xdf\xa3pdfs-media-header"),
-    (7 * 1024 * 1024 + 19, b"pdfs-middle-cue"),
-    (size - len(b"pdfs-media-trailer"), b"pdfs-media-trailer"),
-]
-assert os.stat(p).st_size == size
-with open(p, "rb", buffering=0) as f:
-    for offset, expected in checks:
-        f.seek(offset)
-        assert f.read(len(expected)) == expected
-PY
-    done
-fi
-
-step "cleanup"
-cleanup
-roots=()
-echo "PASS: FUSE acceptance suite completed ($run_id)"
+exec python3 -u "$script_dir/fuse-acceptance.py" --live "$@"