updated documentation + readme

- 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>"]

README.md

@@ -10,11 +10,18 @@ A fast, unofficial Proton Drive client for Linux. This client features an advanc
 - **Shared Links & Invites**: Browse files shared with you by other users, and view/manage your own public shared links directly in the GUI.
 - **Local Backup (Computers)**: Sync and back up local directories (like Downloads, Documents, Pictures, etc.) directly to your Proton Drive account.
 - **System Tray Integration**: Background indicator for status monitoring, quick actions, and fast sync controls.
-- **Fuzzy Search Launcher Prompt (HUD)**: A Google Drive-style search launcher (`pdfs-prompt`) for finding and opening files or folders instantly, ideal for binding to a system-wide hotkey.
+- **Unified Search Launcher (HUD)**: A resident Google Drive-style launcher (`pdfs-prompt`) that searches Proton Drive and local files together, ranks the best matches, tolerates abbreviations and typos, and is ideal for a system-wide hotkey.
 - **Secure Credential Storage**: Integrates with the system Secret Service (GNOME Keyring, KWallet, etc.) with smart in-memory credential caching to avoid UI thread blockages.
 - **Proton Photos Support**: Access your Proton Photos timeline, view thumbnails, and download backed-up media natively (available in the GUI as a navigation tab and via the CLI).
 - **Human Verification (CAPTCHA) Recovery**: Detects sign-in gates (VPN/new IP challenges) and launches an embedded `WebKitWebView` dialog to safely complete the challenge, transparently retrying authentication with the earned token.
 - **Selective Sync (`.pdfsignore`)**: Keep build trees, dependency directories, and editor leftovers out of synced folders using gitignore-style rules.
+- **Data-Safe Offline Writes**: Durable scratch/staging files and a transactional pending queue preserve acknowledged writes across network failures and restarts.
+
+## Quick Search and Media Opening
+
+`pdfs-prompt` searches Drive metadata and the local home-directory index in one request. Results are ranked together rather than split by source. Matching is case-insensitive and supports exact names, prefixes, parent paths, multiple terms, punctuation-separated words, ordered abbreviations, common typos, and adjacent transpositions.
+
+The prompt stays resident after its first launch, so invoking the shortcut again reuses and resets the existing window. Drive folders open from the mount. Audio and video results also open through FUSE, allowing media players to seek and request ranges without downloading the entire file first; ordinary Drive files are materialized into the local cache before opening.
 
 ## Selective Sync (`.pdfsignore`)
 
@@ -131,6 +138,19 @@ The client includes several optimizations designed for high efficiency, a low me
 - **Disk-Backed Writes**: Large file writes are staged on disk in temporary scratch files (rather than fully buffered in RAM) and track modified byte intervals. Only the unedited remote segments are fetched at commit time, keeping memory usage minimal.
 - **Non-Blocking GTK4 Loop**: To prevent UI freezes, all synchronous D-Bus credential checks, control socket requests, and cache usage calculations are offloaded to background worker threads or fetched asynchronously.
 - **Flicker-Free UI Rendering**: The GTK4 application performs differential rendering of the pins list, only modifying list rows when the list content changes, preserving the user's scroll position and widget focus.
+- **Durable Staging**: Scratch metadata and staged writes are synced and atomically published before an upload is acknowledged locally. Rapid revisions supersede pending work transactionally.
+
+## Filesystem Safety
+
+Version 1.0 strengthens the boundaries around local-only data and destructive reconciliation:
+
+- Truncate composes with queued revisions, preserving the authoritative prefix and correct zero-filled growth.
+- Combined move-and-rename operations are queued durably and tolerate partially completed remote state.
+- Session tombstones prevent successfully unlinked entries from reappearing through eventually consistent listings.
+- Names are valid UTF-8 and no longer than Linux's 255-byte component limit.
+- Incomplete mirror scans are non-destructive, and total-wipe protection covers every non-empty baseline.
+- Failed conflict preservation, local deletion, or staging publication does not settle a new baseline or discard the only local copy.
+- State, cache, and config directories must be real, current-user-owned `0700` directories; control sockets are `0600`.
 
 
 ## Screenshots
@@ -258,15 +278,18 @@ This project has a GitHub Actions CI workflow configured under `.github/workflow
 1. **Triggers**: 
    - Pushing a git tag matching `v*` (e.g. `git tag v0.1.0 && git push origin v0.1.0`).
    - Manual runs via the **Actions** tab in GitHub (**workflow_dispatch**).
-2. **Build Process**:
+2. **Quality Gates and Build Process**:
+   - For tagged builds, verifies that the tag, workspace version, and `packaging/PKGBUILD` version agree.
+   - Runs `cargo fmt`, Clippy with warnings denied, the locked workspace test suite, and the account-free FUSE acceptance contract.
    - Spawns an Ubuntu runner and installs GTK4, Libadwaita, FUSE3, and Secret Service packages.
    - Sets up the Rust compiler and caches build targets to speed up runs.
    - Compiles the workspace members in release mode.
 3. **Artifact Packaging**:
    - Generates a `.tar.gz` containing the raw binaries (`pdfs`, `pdfs-app`, `pdfs-tray`, `pdfs-prompt`).
-   - Packs them into a Debian package (`.deb`).
+   - Packs them into Debian (`.deb`) and Fedora (`.rpm`) packages.
+   - Includes the systemd user service and tray autostart entry in the Debian package.
 4. **Publishing**:
-   - Creates a GitHub Release matching the pushed tag and uploads the `.deb` and `.tar.gz` packages as release assets.
+   - Creates a GitHub Release matching the pushed tag and uploads the `.deb`, `.rpm`, and `.tar.gz` packages as release assets.
    - For manual runs, compiles and exposes the packages as workflow run artifacts for testing.
 
 ---

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);
+        loop {
             let path = self.scratch_dir.join(format!(
-            "w-{}-{}",
+                "w-{}-{}-{}",
                 std::process::id(),
+                now_millis(),
                 N.fetch_add(1, Ordering::Relaxed)
             ));
-        let file = std::fs::OpenOptions::new()
+            match std::fs::OpenOptions::new()
-            .create(true)
+                .create_new(true)
-            .truncate(true)
                 .read(true)
                 .write(true)
-            .open(&path)?;
+                .open(&path)
-        Ok((file, 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()
+        if let Some(dir) = side.parent() {
-            && let Ok(d) = std::fs::File::open(dir)
+            std::fs::File::open(dir)?.sync_all()?;
-        {
-            let _ = d.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)),
+                out.push((blob, meta));
-                None => {
-                    let _ = std::fs::remove_file(&blob);
-                }
             }
         }
         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) {
+        let copied = if let Err(e) = std::fs::rename(scratch, &path) {
-            // Cross-device: rename cannot move it, so copy and drop the original.
+            if e.kind() != std::io::ErrorKind::CrossesDevices {
-            let _ = e;
+                return Err(e.into());
-            std::fs::copy(scratch, &path)?;
-            let _ = std::fs::remove_file(scratch);
             }
+            std::fs::copy(scratch, &path)?;
+            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(
+        let side = path.with_extension("json");
-            path.with_extension("json"),
+        let side_tmp = path.with_extension("json.tmp");
-            serde_json::to_vec_pretty(meta)?,
+        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()
+        if path.exists() {
-            && let Ok(content) = std::fs::read_to_string(&path)
+            match std::fs::read_to_string(&path)
-            && let Ok(config) = serde_json::from_str::<AppConfig>(&content)
+                .map_err(Error::from)
-        {
+                .and_then(|content| {
-            return config;
+                    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() {
+        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()
-        std::fs::write(&path, serde_json::to_string_pretty(config)?)?;
+            .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() {
+        std::fs::create_dir_all(&config_dir)?;
-            restrict_dir(&config_dir);
+        restrict_dir(&config_dir)?;
-        }
         Ok(())
     }
 }
 
-/// Make `dir` owner-only (`0700`). Best effort: a mode we cannot set (an
+/// Make `dir` owner-only (`0700`). These directories contain decrypted bytes,
-/// exotic filesystem, a directory we do not own) is not worth refusing to
+/// plaintext names, and an authenticated control socket, so failure is fatal.
-/// start over, and the caller has no better answer than continuing.
+fn restrict_dir(dir: &Path) -> Result<()> {
-fn restrict_dir(dir: &Path) {
+    use std::os::unix::fs::{MetadataExt as _, PermissionsExt};
-    use std::os::unix::fs::PermissionsExt;
+    let before = std::fs::symlink_metadata(dir)?;
-    if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) {
+    if before.file_type().is_symlink() || !before.is_dir() {
-        tracing::warn!(path = %dir.display(), error = %e, "could not restrict directory to 0700");
+        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/control.rs

@@ -101,6 +101,19 @@ pub enum Request {
     /// Independent of [`Request::Search`] so a front-end can fire both at once and
     /// render whichever lands first. Replies with [`Response::LocalResults`].
     SearchLocal { query: String, limit: usize },
+    /// Search both the Drive metadata index and the local-file index in one
+    /// round-trip. This is the preferred prompt API; the two legacy search
+    /// requests remain available for older clients. `limit` applies to each
+    /// source independently. `filters` is applied before limiting, so selecting
+    /// (for example) images cannot hide valid matches below an unfiltered top
+    /// `limit`. Older clients omit it and therefore search both sources and all
+    /// kinds. Replies with [`Response::SearchResultsV2`].
+    SearchV2 {
+        query: String,
+        limit: usize,
+        #[serde(default)]
+        filters: SearchFilters,
+    },
     /// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a
     /// single path component (no separators). Replies with [`Response::Ok`].
     Rename { path: String, new_name: String },
@@ -665,6 +678,15 @@ pub struct SearchHit {
     pub pinned: bool,
     /// Node uid in `volume~link` form.
     pub uid: String,
+    /// Absolute path through the most specific active on-demand mount, when the
+    /// node is covered by one. Clients should prefer this for folders and
+    /// streamable media, and fall back to `mountpoint/path` when absent.
+    #[serde(default)]
+    pub mounted_path: Option<String>,
+    /// Relevance assigned by the daemon. Higher scores sort first. Legacy
+    /// daemons omit it, in which case clients retain the daemon's result order.
+    #[serde(default)]
+    pub score: i64,
 }
 
 /// One hit in a [`Request::SearchLocal`] result: a file on this machine, outside
@@ -679,6 +701,107 @@ pub struct LocalHit {
     pub size: u64,
     /// Modification time, epoch seconds.
     pub modified: i64,
+    /// Relevance assigned by the daemon. Higher scores sort first. Legacy
+    /// daemons omit it, in which case clients retain the daemon's result order.
+    #[serde(default)]
+    pub score: i64,
+}
+
+/// A searchable corpus in [`Request::SearchV2`]. Kept explicit on the wire so
+/// future sources can be added without multiplying request variants.
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum SearchSource {
+    Drive,
+    Local,
+}
+
+/// Coarse result kind used by the prompt's facets. Classification belongs in
+/// core/the daemon and must happen before the per-source limit is applied.
+#[derive(Serialize, Deserialize, Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum SearchKind {
+    #[default]
+    All,
+    Folders,
+    Documents,
+    Images,
+    Media,
+}
+
+impl SearchKind {
+    /// Classify a result before the caller applies its final limit. This stays
+    /// beside the wire enum so every front end and both search corpora agree.
+    pub fn accepts(self, name: &str, is_dir: bool) -> bool {
+        if self == Self::All {
+            return true;
+        }
+        if self == Self::Folders {
+            return is_dir;
+        }
+        if is_dir {
+            return false;
+        }
+        let extension = std::path::Path::new(name)
+            .extension()
+            .and_then(|value| value.to_str())
+            .unwrap_or_default()
+            .to_ascii_lowercase();
+        match self {
+            Self::Documents => matches!(
+                extension.as_str(),
+                "pdf"
+                    | "doc"
+                    | "docx"
+                    | "odt"
+                    | "rtf"
+                    | "txt"
+                    | "md"
+                    | "xls"
+                    | "xlsx"
+                    | "ods"
+                    | "csv"
+                    | "ppt"
+                    | "pptx"
+                    | "odp"
+                    | "epub"
+            ),
+            Self::Images => matches!(
+                extension.as_str(),
+                "png" | "jpg" | "jpeg" | "webp" | "gif" | "bmp" | "svg" | "avif" | "heic" | "tiff"
+            ),
+            Self::Media => {
+                PhotoKind::classify(Some(name), None) == PhotoKind::Video
+                    || matches!(
+                        extension.as_str(),
+                        "mp3" | "flac" | "wav" | "ogg" | "opus" | "m4a" | "aac"
+                    )
+            }
+            Self::All | Self::Folders => unreachable!(),
+        }
+    }
+}
+
+/// Optional constraints for a unified search. The default deliberately means
+/// the original SearchV2 behaviour, preserving old JSON requests.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+pub struct SearchFilters {
+    /// Sources to query. An explicit empty list searches neither source.
+    #[serde(default = "default_search_sources")]
+    pub sources: Vec<SearchSource>,
+    #[serde(default)]
+    pub kind: SearchKind,
+}
+
+impl Default for SearchFilters {
+    fn default() -> Self {
+        Self {
+            sources: default_search_sources(),
+            kind: SearchKind::All,
+        }
+    }
+}
+
+fn default_search_sources() -> Vec<SearchSource> {
+    vec![SearchSource::Drive, SearchSource::Local]
 }
 
 /// What kind of media a timeline entry is, so the Photos page can split into
@@ -947,6 +1070,14 @@ pub enum Response {
     /// is true while a scan of the machine is still running, so a front-end can
     /// say "still indexing" instead of "no matches" on a cold first launch.
     LocalResults { hits: Vec<LocalHit>, indexing: bool },
+    /// Drive and local-file search results returned together (reply to
+    /// [`Request::SearchV2`]). `local_indexing` distinguishes a genuinely empty
+    /// local result set from one observed while the initial scan is running.
+    SearchResultsV2 {
+        drive_hits: Vec<SearchHit>,
+        local_hits: Vec<LocalHit>,
+        local_indexing: bool,
+    },
     /// A snapshot of what the daemon is working on (reply to
     /// [`Request::GetQueueStatus`]): `items` are byte-moving transfers, `jobs`
     /// the longer non-transfer work around them (scans, folder skeletons, the
@@ -1104,6 +1235,58 @@ mod tests {
         assert_eq!(PhotoKind::classify(Some("x.jpg"), Some("video/mp4")), Photo);
     }
 
+    /// Keep every extension accepted by the shared classifier covered.  Drive
+    /// activation uses this classification before it knows whether opening a
+    /// file should stream through FUSE or materialise a complete cache entry;
+    /// silently dropping one of these formats would therefore regress large
+    /// videos back to the expensive whole-file path.
+    #[test]
+    fn photo_kind_recognises_every_supported_video_extension() {
+        for ext in VIDEO_EXTS {
+            let lower = format!("movie.{ext}");
+            let upper = format!("MOVIE.{}", ext.to_ascii_uppercase());
+            assert_eq!(
+                PhotoKind::classify(Some(&lower), None),
+                PhotoKind::Video,
+                "lower-case .{ext}"
+            );
+            assert_eq!(
+                PhotoKind::classify(Some(&upper), None),
+                PhotoKind::Video,
+                "upper-case .{ext}"
+            );
+        }
+    }
+
+    #[test]
+    fn photo_kind_uses_video_mime_only_when_extension_is_not_authoritative() {
+        use PhotoKind::*;
+
+        assert_eq!(
+            PhotoKind::classify(Some("extensionless"), Some("video/mp4; codecs=avc1")),
+            Video
+        );
+        assert_eq!(
+            PhotoKind::classify(Some("clip.unknown"), Some("video/x-matroska")),
+            Video
+        );
+        // Audio is not a video and must not accidentally take the video
+        // streaming policy merely because both are time-based media.
+        assert_eq!(
+            PhotoKind::classify(Some("recording"), Some("audio/flac")),
+            Photo
+        );
+        // A known still/raw suffix remains authoritative over server metadata.
+        assert_eq!(
+            PhotoKind::classify(Some("poster.AVIF"), Some("video/mp4")),
+            Photo
+        );
+        assert_eq!(
+            PhotoKind::classify(Some("negative.NEF"), Some("video/mp4")),
+            Raw
+        );
+    }
+
     #[test]
     fn photo_kind_i64_round_trips() {
         for k in [PhotoKind::Photo, PhotoKind::Video, PhotoKind::Raw] {
@@ -1167,6 +1350,150 @@ mod tests {
         }
     }
 
+    #[test]
+    fn unified_search_request_and_response_roundtrip() {
+        let request = Request::SearchV2 {
+            query: "quarterly report".into(),
+            limit: 37,
+            filters: SearchFilters::default(),
+        };
+        let line = serde_json::to_string(&request).unwrap();
+        assert_eq!(
+            line,
+            r#"{"SearchV2":{"query":"quarterly report","limit":37,"filters":{"sources":["Drive","Local"],"kind":"All"}}}"#
+        );
+        let decoded: Request = serde_json::from_str(&line).unwrap();
+        assert_eq!(line, serde_json::to_string(&decoded).unwrap());
+
+        let response = Response::SearchResultsV2 {
+            drive_hits: vec![SearchHit {
+                name: "report.pdf".into(),
+                path: "Work/report.pdf".into(),
+                is_dir: false,
+                size: 12,
+                modified: 123,
+                pinned: true,
+                uid: "vol~link".into(),
+                mounted_path: Some("/home/me/Videos/report.pdf".into()),
+                score: 985,
+            }],
+            local_hits: vec![LocalHit {
+                name: "notes.txt".into(),
+                path: "/home/me/notes.txt".into(),
+                is_dir: false,
+                size: 7,
+                modified: 456,
+                score: 720,
+            }],
+            local_indexing: true,
+        };
+        let line = serde_json::to_string(&response).unwrap();
+        assert!(!line.contains('\n'));
+        let decoded: Response = serde_json::from_str(&line).unwrap();
+        match decoded {
+            Response::SearchResultsV2 {
+                drive_hits,
+                local_hits,
+                local_indexing,
+            } => {
+                assert_eq!(drive_hits.len(), 1);
+                assert_eq!(drive_hits[0].uid, "vol~link");
+                assert_eq!(drive_hits[0].score, 985);
+                assert_eq!(local_hits.len(), 1);
+                assert_eq!(local_hits[0].path, "/home/me/notes.txt");
+                assert_eq!(local_hits[0].score, 720);
+                assert!(local_indexing);
+            }
+            other => panic!("expected unified search results, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn search_v2_legacy_wire_form_gets_compatible_defaults() {
+        let decoded: Request =
+            serde_json::from_str(r#"{"SearchV2":{"query":"old client","limit":20}}"#).unwrap();
+        match decoded {
+            Request::SearchV2 {
+                query,
+                limit,
+                filters,
+            } => {
+                assert_eq!(query, "old client");
+                assert_eq!(limit, 20);
+                assert_eq!(filters, SearchFilters::default());
+            }
+            other => panic!("expected SearchV2, got {other:?}"),
+        }
+
+        // New daemons add scores, but old-daemon replies remain readable and
+        // signal that their existing order should be retained with score 0.
+        let decoded: Response = serde_json::from_str(
+            r#"{"SearchResultsV2":{"drive_hits":[{"name":"a","path":"a","is_dir":false,"size":1,"modified":2,"pinned":false,"uid":"v~a"}],"local_hits":[{"name":"b","path":"/b","is_dir":false,"size":3,"modified":4}],"local_indexing":false}}"#,
+        )
+        .unwrap();
+        match decoded {
+            Response::SearchResultsV2 {
+                drive_hits,
+                local_hits,
+                ..
+            } => {
+                assert_eq!(drive_hits[0].score, 0);
+                assert_eq!(local_hits[0].score, 0);
+            }
+            other => panic!("expected SearchResultsV2, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn search_v2_filters_roundtrip() {
+        let request = Request::SearchV2 {
+            query: "holiday".into(),
+            limit: 50,
+            filters: SearchFilters {
+                sources: vec![SearchSource::Drive],
+                kind: SearchKind::Images,
+            },
+        };
+        let line = serde_json::to_string(&request).unwrap();
+        let decoded: Request = serde_json::from_str(&line).unwrap();
+        assert!(matches!(
+            decoded,
+            Request::SearchV2 {
+                filters: SearchFilters { sources, kind: SearchKind::Images },
+                ..
+            } if sources == vec![SearchSource::Drive]
+        ));
+    }
+
+    #[test]
+    fn search_kind_classifies_before_results_are_limited() {
+        assert!(SearchKind::Folders.accepts("movie.mp4", true));
+        assert!(!SearchKind::Folders.accepts("movie.mp4", false));
+        assert!(SearchKind::Documents.accepts("REPORT.PDF", false));
+        assert!(SearchKind::Images.accepts("scan.HEIC", false));
+        assert!(SearchKind::Media.accepts("film.mkv", false));
+        assert!(SearchKind::Media.accepts("recording.flac", false));
+        assert!(!SearchKind::Media.accepts("poster.jpg", false));
+        assert!(SearchKind::All.accepts("anything", true));
+    }
+
+    #[test]
+    fn legacy_search_wire_forms_still_decode() {
+        let drive: Request =
+            serde_json::from_str(r#"{"Search":{"query":"old","limit":20}}"#).unwrap();
+        assert!(matches!(
+            drive,
+            Request::Search { query, limit } if query == "old" && limit == 20
+        ));
+
+        let local: Request =
+            serde_json::from_str(r#"{"SearchLocal":{"query":"old","limit":10}}"#).unwrap();
+        assert!(matches!(
+            local,
+            Request::SearchLocal { query, limit } if query == "old" && limit == 10
+        ));
+    }
+
     /// The trash requests carry uids rather than paths; they must survive the same
     /// line-delimited round-trip, since a mangled uid would restore or destroy the
     /// wrong node.

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

@@ -7,6 +7,7 @@ use super::Db;
 use crate::Result;
 use crate::localindex::LocalEntry;
 
+use super::nodes::candidate_trigrams;
 use super::utils::{TRIGRAM_MIN, like_escape};
 
 /// One hit from [`Db::search_local`]: an indexed file on the machine itself, not
@@ -147,6 +148,58 @@ impl Db {
         }
         Ok(out)
     }
+
+    /// Bounded candidates for the fuzzy scorer, including matches in the full
+    /// local path. ORed trigrams admit misspellings while the FTS index prevents
+    /// this from becoming an unbounded filesystem-table scan.
+    pub fn search_local_candidates(&self, query: &str, limit: usize) -> Result<Vec<LocalFileHit>> {
+        let query = query.trim();
+        if query.is_empty() || limit == 0 {
+            return Ok(Vec::new());
+        }
+        let conn = self.conn.lock();
+        let terms = candidate_trigrams(query);
+        let lane_limit = limit.div_ceil(2);
+        let mut rows: Vec<LocalFileHit> = if terms.is_empty() {
+            let pat = format!("%{}%", like_escape(query));
+            let mut stmt = conn.prepare(
+                "SELECT path, name, is_dir, size, mtime FROM local_files
+                 WHERE name LIKE ?1 ESCAPE '\\' OR path LIKE ?1 ESCAPE '\\'
+                 ORDER BY mtime DESC LIMIT ?2",
+            )?;
+            stmt.query_map(params![pat, limit as i64], local_hit)?
+                .collect::<std::result::Result<_, _>>()?
+        } else {
+            let mut stmt = conn.prepare(
+                "SELECT f.path, f.name, f.is_dir, f.size, f.mtime
+                   FROM local_fts x JOIN local_files f ON f.id = x.rowid
+                  WHERE local_fts MATCH ?1
+                  ORDER BY x.rank LIMIT ?2",
+            )?;
+            stmt.query_map(params![terms.join(" OR "), lane_limit as i64], local_hit)?
+                .collect::<std::result::Result<_, _>>()?
+        };
+        if !terms.is_empty()
+            && let Some(first) = query.chars().next()
+        {
+            let prefix = format!("{}%", like_escape(&first.to_string()));
+            let mut stmt = conn.prepare(
+                "SELECT path, name, is_dir, size, mtime FROM local_files
+                 WHERE name COLLATE NOCASE LIKE ?1 ESCAPE '\\'
+                 ORDER BY name COLLATE NOCASE LIMIT ?2",
+            )?;
+            let fallback = stmt
+                .query_map(params![prefix, lane_limit as i64], local_hit)?
+                .collect::<std::result::Result<Vec<_>, _>>()?;
+            for hit in fallback {
+                if !rows.iter().any(|existing| existing.path == hit.path) {
+                    rows.push(hit);
+                }
+            }
+            rows.truncate(limit);
+        }
+        Ok(rows)
+    }
 }
 
 /// Row → [`LocalFileHit`]. Every local-search query selects the same

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

@@ -9,7 +9,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 15;
+pub(super) const SCHEMA_VERSION: i64 = 16;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -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(());
         }
 
@@ -81,6 +86,9 @@ impl Db {
         if current < 15 {
             tx.execute_batch(MIGRATION_V15)?;
         }
+        if current < 16 {
+            tx.execute_batch(MIGRATION_V16)?;
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -369,3 +377,37 @@ CREATE VIRTUAL TABLE nodes_fts USING fts5(
 INSERT INTO nodes_fts (rowid, name)
   SELECT rowid, name FROM nodes WHERE trashed = 0;
 ";
+
+/// Schema v16: fuzzy search needs a bounded candidate set that includes parent
+/// paths.  Keeping the path in the trigram indexes lets SQLite narrow both name
+/// and path candidates without walking either base table for every keystroke.
+/// Node paths are initially backfilled below and subsequently maintained by the
+/// node write path (including descendants after a folder move/rename).
+const MIGRATION_V16: &str = "
+DROP TABLE IF EXISTS nodes_fts;
+CREATE VIRTUAL TABLE nodes_fts USING fts5(
+  name, path, tokenize='trigram'
+);
+CREATE INDEX IF NOT EXISTS idx_nodes_name_nocase ON nodes(name COLLATE NOCASE);
+INSERT INTO nodes_fts (rowid, name, path)
+WITH RECURSIVE paths(rowid, uid, path) AS (
+  SELECT n.rowid, n.uid, '' FROM nodes n
+   WHERE n.parent_uid IS NULL
+      OR NOT EXISTS (SELECT 1 FROM nodes p WHERE p.uid = n.parent_uid)
+  UNION ALL
+  SELECT n.rowid, n.uid,
+         CASE WHEN paths.path = '' THEN n.name ELSE paths.path || '/' || n.name END
+    FROM nodes n JOIN paths ON n.parent_uid = paths.uid
+)
+SELECT n.rowid, n.name, paths.path
+  FROM nodes n JOIN paths ON paths.rowid = n.rowid
+ WHERE n.trashed = 0;
+
+DROP TABLE IF EXISTS local_fts;
+CREATE VIRTUAL TABLE local_fts USING fts5(
+  name, path, content='local_files', content_rowid='id', tokenize='trigram'
+);
+CREATE INDEX IF NOT EXISTS idx_local_files_name_nocase
+  ON local_files(name COLLATE NOCASE);
+INSERT INTO local_fts(local_fts) VALUES('rebuild');
+";

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

@@ -81,12 +81,42 @@ impl Db {
                 params![uid],
                 |row| row.get(0),
             )?;
-            tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?;
+            // A folder rename/move changes every descendant's searchable path.
-            if !node.trashed {
+            // Refresh the affected subtree, while retaining efficient rowid
-                tx.execute(
+            // deletes for the overwhelmingly common one-file case.
-                    "INSERT INTO nodes_fts (rowid, name) VALUES (?1, ?2)",
+            let mut affected = vec![rowid];
-                    params![rowid, node.name],
+            if node.is_folder() {
+                let mut stmt = tx.prepare(
+                    "WITH RECURSIVE descendants(rowid, uid) AS (
+                       SELECT rowid, uid FROM nodes WHERE parent_uid = ?1
+                       UNION ALL
+                       SELECT n.rowid, n.uid FROM nodes n
+                         JOIN descendants d ON n.parent_uid = d.uid
+                     ) SELECT rowid FROM descendants",
                 )?;
+                for descendant in stmt.query_map(params![uid], |row| row.get(0))? {
+                    affected.push(descendant?);
+                }
+            }
+            for affected_rowid in affected {
+                tx.execute(
+                    "DELETE FROM nodes_fts WHERE rowid = ?1",
+                    params![affected_rowid],
+                )?;
+                let indexed: Option<(String, String)> = tx
+                    .query_row(
+                        "SELECT name, uid FROM nodes WHERE rowid = ?1 AND trashed = 0",
+                        params![affected_rowid],
+                        |row| Ok((row.get(0)?, row.get(1)?)),
+                    )
+                    .optional()?;
+                if let Some((name, indexed_uid)) = indexed {
+                    let path = path_of(&tx, &indexed_uid)?;
+                    tx.execute(
+                        "INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
+                        params![affected_rowid, name, path],
+                    )?;
+                }
             }
         }
         tx.commit()?;
@@ -130,6 +160,40 @@ impl Db {
         Ok(count > 0)
     }
 
+    /// Resolve `uid` relative to an ancestor node.
+    ///
+    /// This is used when a remote subtree has its own local sync mount: the
+    /// sync folder stores the ancestor UID, while search results identify the
+    /// selected descendant. Returning `None` for an unrelated or incomplete
+    /// chain keeps callers from accidentally joining a Drive-wide path onto the
+    /// wrong mountpoint. The ancestor itself resolves to the empty path.
+    pub fn path_relative_to(&self, ancestor_uid: &str, uid: &str) -> Result<Option<String>> {
+        let conn = self.conn.lock();
+        let mut stmt = conn.prepare(
+            "WITH RECURSIVE anc(uid, parent_uid, name, depth) AS (
+               SELECT uid, parent_uid, name, 0 FROM nodes WHERE uid = ?1
+               UNION ALL
+               SELECT n.uid, n.parent_uid, n.name, anc.depth + 1
+               FROM nodes n JOIN anc ON n.uid = anc.parent_uid
+               WHERE anc.depth < 1024
+             )
+             SELECT uid, name FROM anc ORDER BY depth",
+        )?;
+        let rows = stmt.query_map(params![uid], |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+        })?;
+        let mut parts = Vec::new();
+        for row in rows {
+            let (current_uid, name) = row?;
+            if current_uid == ancestor_uid {
+                parts.reverse();
+                return Ok(Some(parts.join("/")));
+            }
+            parts.push(name);
+        }
+        Ok(None)
+    }
+
     /// Full-text search over node names, newest schema's trigram index giving
     /// substring (not just prefix) matches. Returns up to `limit` non-trashed
     /// hits, each with its mountpoint-relative path resolved. Queries shorter
@@ -175,6 +239,68 @@ impl Db {
         Ok(hits)
     }
 
+    /// Return a bounded, deliberately broad candidate pool for fuzzy ranking.
+    /// Unlike [`search`](Self::search), trigram terms are ORed across both the
+    /// basename and parent path, so a typo can still share enough trigrams to
+    /// enter the pool. Final relevance ordering belongs to the caller.
+    pub fn search_candidates(&self, query: &str, limit: usize) -> Result<Vec<SearchHit>> {
+        let query = query.trim();
+        if query.is_empty() || limit == 0 {
+            return Ok(Vec::new());
+        }
+        let conn = self.conn.lock();
+        let terms = candidate_trigrams(query);
+        let lane_limit = limit.div_ceil(2);
+        let mut rows: Vec<(String, String)> = if terms.is_empty() {
+            let pat = format!("%{}%", like_escape(query));
+            let mut stmt = conn.prepare(
+                "SELECT node_json, uid FROM nodes
+                 WHERE name LIKE ?1 ESCAPE '\\' AND trashed = 0 AND node_json IS NOT NULL
+                 ORDER BY name LIMIT ?2",
+            )?;
+            collect_pairs(stmt.query_map(params![pat, limit as i64], pair)?)?
+        } else {
+            let expression = terms.join(" OR ");
+            let mut stmt = conn.prepare(
+                "SELECT n.node_json, n.uid
+                   FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid
+                  WHERE nodes_fts MATCH ?1 AND n.trashed = 0 AND n.node_json IS NOT NULL
+                  ORDER BY f.rank LIMIT ?2",
+            )?;
+            collect_pairs(stmt.query_map(params![expression, lane_limit as i64], pair)?)?
+        };
+        // A short substitution can destroy every trigram (`vedio` vs `video`).
+        // Feed the scorer an additional indexed same-initial lane so it can
+        // recover those candidates without scanning the full node table.
+        if !terms.is_empty()
+            && let Some(first) = query.chars().next()
+        {
+            let prefix = format!("{}%", like_escape(&first.to_string()));
+            let mut stmt = conn.prepare(
+                "SELECT node_json, uid FROM nodes
+                 WHERE name COLLATE NOCASE LIKE ?1 ESCAPE '\\'
+                   AND trashed = 0 AND node_json IS NOT NULL
+                 ORDER BY name COLLATE NOCASE LIMIT ?2",
+            )?;
+            let fallback =
+                collect_pairs(stmt.query_map(params![prefix, lane_limit as i64], pair)?)?;
+            for row in fallback {
+                if !rows.iter().any(|(_, uid)| uid == &row.1) {
+                    rows.push(row);
+                }
+            }
+            rows.truncate(limit);
+        }
+        rows.into_iter()
+            .map(|(json, uid)| {
+                Ok(SearchHit {
+                    node: serde_json::from_str(&json)?,
+                    path: path_of(&conn, &uid)?,
+                })
+            })
+            .collect()
+    }
+
     /// Mark (or unmark) a folder's child listing as complete. A listed folder
     /// rehydrates its `children` map on mount even when empty; an unlisted one
     /// re-enumerates from the remote on next access.
@@ -261,6 +387,23 @@ impl Db {
     // in sync on every store/read/evict, so it is authoritative for eviction.
 }
 
+/// Unique quoted character trigrams suitable for an FTS5 OR expression.
+pub(super) fn candidate_trigrams(query: &str) -> Vec<String> {
+    let chars: Vec<char> = query.to_lowercase().chars().collect();
+    let mut terms = Vec::new();
+    for window in chars.windows(TRIGRAM_MIN) {
+        if window.iter().all(|c| c.is_whitespace()) {
+            continue;
+        }
+        let raw: String = window.iter().collect();
+        let quoted = format!("\"{}\"", raw.replace('"', "\"\""));
+        if !terms.contains(&quoted) {
+            terms.push(quoted);
+        }
+    }
+    terms
+}
+
 /// Effective plaintext size of a node for the indexed `size` column: the
 /// claimed size when known, else the on-storage size; folders are 0.
 pub(super) fn node_size(node: &Node) -> i64 {

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

@@ -5,6 +5,7 @@ use crate::control::{ActivityEntry, ActivityKind};
 use crate::localindex::LocalEntry;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
 use proton_drive_rs::{Node, NodeKind};
+use serde_json::json;
 
 /// `Db::open` applies the pragmas that `open_in_memory` skips, so the WAL
 /// settings can only be checked against a real file.
@@ -542,6 +543,95 @@ fn search_short_query_like_fallback() {
     assert_eq!(hits[0].node.name, "ab.txt");
 }
 
+#[test]
+fn fuzzy_candidates_include_typo_and_parent_path_matches() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("projects", Some("root"), "Projects"))
+        .unwrap();
+    db.upsert_node(&file("report", "projects", "quarterly-report.pdf", 1))
+        .unwrap();
+    db.upsert_node(&file("video", "projects", "video.mp4", 1))
+        .unwrap();
+
+    let typo = db.search_candidates("quartrly", 20).unwrap();
+    assert!(
+        typo.iter()
+            .any(|hit| hit.node.name == "quarterly-report.pdf")
+    );
+    assert!(
+        db.search_candidates("vedio", 20)
+            .unwrap()
+            .iter()
+            .any(|hit| hit.node.name == "video.mp4")
+    );
+
+    let path = db.search_candidates("projects", 20).unwrap();
+    assert!(
+        path.iter()
+            .any(|hit| hit.node.name == "quarterly-report.pdf")
+    );
+}
+
+#[test]
+fn renaming_parent_refreshes_descendant_search_paths() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("folder", Some("root"), "zzbeforezz"))
+        .unwrap();
+    db.upsert_node(&file("child", "folder", "plain.txt", 1))
+        .unwrap();
+    assert!(
+        db.search_candidates("zzbeforezz", 20)
+            .unwrap()
+            .iter()
+            .any(|h| h.node.name == "plain.txt")
+    );
+
+    db.upsert_node(&folder("folder", Some("root"), "yyafteryy"))
+        .unwrap();
+    assert!(
+        !db.search_candidates("zzbeforezz", 20)
+            .unwrap()
+            .iter()
+            .any(|h| h.node.name == "plain.txt")
+    );
+    assert!(
+        db.search_candidates("yyafteryy", 20)
+            .unwrap()
+            .iter()
+            .any(|h| h.node.name == "plain.txt")
+    );
+}
+
+#[test]
+fn local_fuzzy_candidates_include_typo_and_parent_path() {
+    let db = Db::open_in_memory().unwrap();
+    let generation = db.local_begin_scan().unwrap();
+    db.local_upsert_batch(
+        generation,
+        &[
+            local(
+                "/home/u/Projects/quarterly-report.pdf",
+                "quarterly-report.pdf",
+                false,
+            ),
+            local("/home/u/Videos/video.mp4", "video.mp4", false),
+        ],
+    )
+    .unwrap();
+    db.local_finish_scan(generation, 1).unwrap();
+
+    assert_eq!(db.search_local_candidates("quartrly", 20).unwrap().len(), 1);
+    assert_eq!(db.search_local_candidates("projects", 20).unwrap().len(), 1);
+    assert!(
+        db.search_local_candidates("vedio", 20)
+            .unwrap()
+            .iter()
+            .any(|hit| hit.name == "video.mp4")
+    );
+}
+
 #[test]
 fn search_drops_fts_row_on_delete_and_trash() {
     let db = Db::open_in_memory().unwrap();
@@ -694,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
@@ -777,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
@@ -1092,6 +1250,96 @@ fn pending_mode_is_queued_until_the_mode_is_reached() {
     assert_eq!(folder.pending_mode, None);
 }
 
+#[test]
+fn node_path_resolves_relative_to_a_sync_root_uid() {
+    let db = Db::open_in_memory().unwrap();
+    let root = node_from(serde_json::Value::Null, "root", "My Files", json!("Folder"));
+    let videos = node_from(json!(uid("root")), "videos", "Videos", json!("Folder"));
+    let anime = node_from(json!(uid("videos")), "anime", "anime", json!("Folder"));
+    let show = node_from(json!(uid("anime")), "show", "Oshi no Ko", json!("Folder"));
+    let episode = node_from(
+        json!(uid("show")),
+        "episode",
+        "episode 01.mkv",
+        json!({"File": {
+            "media_type": "video/x-matroska",
+            "total_size_on_storage": 42,
+            "claimed_size": 42,
+            "claimed_modification_time": null
+        }}),
+    );
+    db.upsert_nodes(&[root, videos, anime, show, episode])
+        .unwrap();
+
+    assert_eq!(
+        db.path_relative_to(&uid("videos").to_string(), &uid("episode").to_string())
+            .unwrap()
+            .as_deref(),
+        Some("anime/Oshi no Ko/episode 01.mkv")
+    );
+    assert_eq!(
+        db.path_relative_to(&uid("videos").to_string(), &uid("videos").to_string())
+            .unwrap()
+            .as_deref(),
+        Some("")
+    );
+    assert_eq!(
+        db.path_relative_to(&uid("anime").to_string(), &uid("episode").to_string())
+            .unwrap()
+            .as_deref(),
+        Some("Oshi no Ko/episode 01.mkv")
+    );
+}
+
+#[test]
+fn node_path_rejects_unrelated_missing_and_broken_ancestor_chains() {
+    let db = Db::open_in_memory().unwrap();
+    let root = node_from(serde_json::Value::Null, "root", "My Files", json!("Folder"));
+    let videos = node_from(json!(uid("root")), "videos", "Videos", json!("Folder"));
+    let episode = node_from(
+        json!(uid("videos")),
+        "episode",
+        "episode.mkv",
+        json!({"File": {
+            "media_type": "video/x-matroska",
+            "total_size_on_storage": 42,
+            "claimed_size": 42,
+            "claimed_modification_time": null
+        }}),
+    );
+    let unrelated = node_from(
+        json!(uid("root")),
+        "documents",
+        "Documents",
+        json!("Folder"),
+    );
+    let orphan = node_from(
+        json!(uid("missing-parent")),
+        "orphan",
+        "orphan.mkv",
+        json!({"File": {
+            "media_type": "video/x-matroska",
+            "total_size_on_storage": 42,
+            "claimed_size": 42,
+            "claimed_modification_time": null
+        }}),
+    );
+    db.upsert_nodes(&[root, videos, episode, unrelated, orphan])
+        .unwrap();
+
+    for (ancestor, descendant) in [
+        ("documents", "episode"),
+        ("videos", "missing-node"),
+        ("videos", "orphan"),
+    ] {
+        assert_eq!(
+            db.path_relative_to(&uid(ancestor).to_string(), &uid(descendant).to_string())
+                .unwrap(),
+            None
+        );
+    }
+}
+
 #[test]
 fn cache_index_touch_access_and_lru_order() {
     let db = Db::open_in_memory().unwrap();

crates/pdfs-core/src/lib.rs

@@ -10,6 +10,7 @@ pub mod db;
 pub mod error;
 pub mod localindex;
 pub mod profile;
+pub mod search;
 pub mod service;
 pub mod shell;
 pub mod syncignore;

crates/pdfs-core/src/search.rs

@@ -0,0 +1,214 @@
+//! Shared, deterministic search-result relevance scoring.
+//!
+//! The scorer deliberately operates on a small candidate set rather than
+//! replacing an index. Callers should use SQLite (or another index) to find
+//! candidates, then use this module to put local and remote results on the
+//! same relevance scale.
+
+/// Scores a candidate against a user query.
+///
+/// Every query term must match either `basename` or `parent_path`. Basename
+/// matches are weighted twice as strongly as path matches. The returned score
+/// is stable and only meaningful when compared with scores produced by this
+/// function.
+#[must_use]
+pub fn relevance_score(query: &str, basename: &str, parent_path: &str) -> Option<i64> {
+    let query = normalize(query);
+    if query.is_empty() {
+        return Some(0);
+    }
+
+    let name = normalize(basename);
+    let path = normalize(parent_path);
+    let query_terms: Vec<_> = query.split_whitespace().collect();
+    let name_words: Vec<_> = name.split_whitespace().collect();
+    let path_words: Vec<_> = path.split_whitespace().collect();
+
+    let mut total = 0_i64;
+    for term in query_terms {
+        let name_score = field_term_score(term, &name, &name_words).map(|score| score * 2);
+        let path_score = field_term_score(term, &path, &path_words);
+        total += name_score.into_iter().chain(path_score).max()?;
+    }
+
+    // Whole-field bonuses make ordering intuitive without changing whether a
+    // candidate matches. They dominate small differences between term scores.
+    if name == query {
+        total += 20_000;
+    } else if name.starts_with(&query) {
+        total += 10_000;
+    } else if name.contains(&query) {
+        total += 4_000;
+    }
+
+    Some(total)
+}
+
+fn normalize(value: &str) -> String {
+    let mut normalized = String::with_capacity(value.len());
+    let mut last_was_space = true;
+
+    for character in value.chars().flat_map(char::to_lowercase) {
+        if character.is_alphanumeric() {
+            normalized.push(character);
+            last_was_space = false;
+        } else if !last_was_space {
+            normalized.push(' ');
+            last_was_space = true;
+        }
+    }
+
+    if last_was_space {
+        normalized.pop();
+    }
+    normalized
+}
+
+fn field_term_score(term: &str, field: &str, words: &[&str]) -> Option<i64> {
+    if field == term {
+        return Some(10_000);
+    }
+    if field.starts_with(term) {
+        return Some(8_500);
+    }
+    if words.contains(&term) {
+        return Some(8_000);
+    }
+    if words.iter().any(|word| word.starts_with(term)) {
+        return Some(7_000);
+    }
+    if field.contains(term) {
+        return Some(6_000);
+    }
+
+    // Subsequence matching supports useful abbreviations such as "qtrrpt".
+    let subsequence = std::iter::once(field)
+        .chain(words.iter().copied())
+        .filter_map(|word| subsequence_gap(term, word))
+        .min();
+    if let Some(gap) = subsequence {
+        return Some(4_500 - i64::try_from(gap.min(500)).unwrap_or(500));
+    }
+
+    // Typo matching is deliberately restricted to substantial terms to avoid
+    // noisy matches for one- and two-letter searches.
+    if term.chars().count() >= 4
+        && words
+            .iter()
+            .any(|word| bounded_damerau_levenshtein(term, word, typo_limit(term)))
+    {
+        return Some(3_500);
+    }
+
+    None
+}
+
+fn subsequence_gap(needle: &str, haystack: &str) -> Option<usize> {
+    let needle_length = needle.chars().count();
+    let mut needle = needle.chars();
+    let mut wanted = needle.next()?;
+    let mut first = None;
+
+    for (index, character) in haystack.chars().enumerate() {
+        if character == wanted {
+            first.get_or_insert(index);
+            match needle.next() {
+                Some(next) => wanted = next,
+                None => return Some(index + 1 - first.unwrap_or(index) - needle_length),
+            }
+        }
+    }
+    None
+}
+
+fn typo_limit(term: &str) -> usize {
+    usize::from(term.chars().count() >= 5) + 1
+}
+
+/// A bounded optimal-string-alignment distance check. This treats a single
+/// adjacent transposition (for example, `vidoe`) as one edit.
+fn bounded_damerau_levenshtein(left: &str, right: &str, limit: usize) -> bool {
+    let left: Vec<_> = left.chars().collect();
+    let right: Vec<_> = right.chars().collect();
+    if left.len().abs_diff(right.len()) > limit || left.len().max(right.len()) > 64 {
+        return false;
+    }
+
+    let mut distances = vec![vec![0_usize; right.len() + 1]; left.len() + 1];
+    for (index, row) in distances.iter_mut().enumerate() {
+        row[0] = index;
+    }
+    for (index, distance) in distances[0].iter_mut().enumerate() {
+        *distance = index;
+    }
+
+    for i in 1..=left.len() {
+        for j in 1..=right.len() {
+            let cost = usize::from(left[i - 1] != right[j - 1]);
+            distances[i][j] = (distances[i - 1][j] + 1)
+                .min(distances[i][j - 1] + 1)
+                .min(distances[i - 1][j - 1] + cost);
+            if i > 1 && j > 1 && left[i - 1] == right[j - 2] && left[i - 2] == right[j - 1] {
+                distances[i][j] = distances[i][j].min(distances[i - 2][j - 2] + 1);
+            }
+        }
+    }
+
+    distances[left.len()][right.len()] <= limit
+}
+
+#[cfg(test)]
+mod tests {
+    use super::relevance_score;
+
+    #[test]
+    fn exact_name_beats_prefix_and_substring() {
+        let exact = relevance_score("report", "report", "work").unwrap();
+        let prefix = relevance_score("report", "report final", "work").unwrap();
+        let substring = relevance_score("report", "annual report", "work").unwrap();
+        assert!(exact > prefix);
+        assert!(prefix > substring);
+    }
+
+    #[test]
+    fn basename_is_weighted_above_parent_path() {
+        let name = relevance_score("budget", "budget notes", "archive").unwrap();
+        let path = relevance_score("budget", "notes", "work/budget").unwrap();
+        assert!(name > path);
+    }
+
+    #[test]
+    fn matches_typos_and_adjacent_transpositions() {
+        assert!(relevance_score("vedio", "video.mp4", "media").is_some());
+        assert!(relevance_score("vidoe", "video.mp4", "media").is_some());
+        assert!(relevance_score("presentaton", "presentation.pdf", "work").is_some());
+    }
+
+    #[test]
+    fn separators_form_searchable_word_boundaries() {
+        assert!(relevance_score("final draft", "Final_Draft-v2.pdf", "work").is_some());
+        assert!(relevance_score("quarter report", "quarterly-report.pdf", "work").is_some());
+    }
+
+    #[test]
+    fn unicode_case_folding_preserves_accented_matches() {
+        assert!(relevance_score("CAFÉ", "Café Photos", "Trips").is_some());
+        assert!(relevance_score("ångström", "ÅNGSTRÖM notes", "Research").is_some());
+    }
+
+    #[test]
+    fn matches_terms_across_name_and_path() {
+        assert!(relevance_score("tax invoice", "invoice-1042.pdf", "archive/tax/2025").is_some());
+    }
+
+    #[test]
+    fn ordered_subsequence_supports_abbreviations() {
+        assert!(relevance_score("qtrrpt", "quarter-report.pdf", "finance").is_some());
+    }
+
+    #[test]
+    fn rejects_unrelated_candidates_and_short_typos() {
+        assert_eq!(relevance_score("video", "holiday.jpg", "photos"), None);
+        assert_eq!(relevance_score("xy", "xz.txt", "misc"), None);
+    }
+}

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 Some(c) = &head else {
-                            && let Err(e) = db.set_event_cursor(c.as_str())
+                            break None;
-                        {
+                        };
-                            warn!(error = %e, "persist seed cursor failed");
+                        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);
+            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;
             }
-        cursor = events.last().map(|e| e.id().clone());
+            // Advance only after this exact event is durably acknowledged. If a
-        if let Some(c) = &cursor
+            // later event fails, polling resumes from here and replays nothing
-            && let Err(e) = db.set_event_cursor(c.as_str())
+            // that was not already made visible locally.
-        {
+            cursor = Some(applied);
-            warn!(error = %e, "persist event cursor failed");
         }
     }
 }

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::sync::atomic::{AtomicUsize, Ordering};
-use std::time::Instant;
+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();
+    let line = match read_request_line(&mut reader) {
-    if reader.read_line(&mut line).is_err() || line.trim().is_empty() {
+        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();
@@ -241,6 +310,18 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             },
             Err(e) => CtlResponse::error(e),
         },
+        Ok(CtlRequest::SearchV2 {
+            query,
+            limit,
+            filters,
+        }) => match core.search_v2(&query, limit, &filters) {
+            Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 {
+                drive_hits,
+                local_hits,
+                local_indexing: core.indexing.load(Ordering::Relaxed),
+            },
+            Err(e) => CtlResponse::error(e),
+        },
         Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) {
             Ok(rel) => match core.rename(&rel, &new_name) {
                 Ok(name) => {
@@ -780,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");
                 }
@@ -796,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 name = match fuse_name(name) {
-        let newname = newname.to_string_lossy().into_owned();
+            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))
-        {
-            error!(%uid, error = %e, "move 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.
-            let mut attempts = 0u32;
-            let renamed = loop {
-                match self
                 .core
                 .rt
                 .block_on(self.core.client.rename_node(&uid, &newname, None))
+        {
+            error!(%uid, error = %e, "rename failed");
+            self.core.restore_replaced(victim.as_ref(), &newname);
+            reply.error(Errno::EIO);
+            return;
+        }
+        if newparent != parent {
+            let mut attempts = 0u32;
+            let moved = loop {
+                match self
+                    .core
+                    .rt
+                    .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 {
+            if let Err(e) = moved {
-                error!(%uid, attempts, error = %e, "rename after move failed");
+                error!(%uid, attempts, error = %e, "move after rename failed");
-                // The move half already landed. Keep our inode model honest
+                // The rename half landed in the source directory.
-                // even though the combined POSIX operation has to return EIO.
+                let mut state = self.core.state.lock();
-                self.core
+                let old_parent_uid = state
-                    .state
+                    .entries
-                    .lock()
+                    .get(&parent)
-                    .relocate(ino, parent, newparent, &new_parent_uid, &name);
+                    .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

@@ -35,6 +35,8 @@ use parking_lot::{Condvar, Mutex};
 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};
@@ -54,13 +56,15 @@ use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite};
 use pdfs_core::config::AppDirs;
 use pdfs_core::control::{
     ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
-    SearchHit, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection,
+    SearchFilters, SearchHit, SearchSource, SyncFolderInfo, SyncPhase, SyncProgress,
+    TransferDirection,
 };
 use pdfs_core::db::{
     Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, StoredNode,
     StoredSyncFolder, StoredTrash,
 };
 use pdfs_core::localindex;
+use pdfs_core::search::relevance_score;
 use pdfs_core::{CoreError, CoreResult};
 use proton_drive_rs::proton_sdk::api::ResponseCode;
 use proton_drive_rs::proton_sdk::error::ProtonError;
@@ -222,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)>,
@@ -651,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(());
@@ -658,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!(
@@ -709,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.
@@ -718,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.
@@ -1297,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)
@@ -1313,12 +1337,37 @@ 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 {
+        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,
@@ -1326,6 +1375,7 @@ impl Core {
                 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");
@@ -1444,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);
@@ -1497,8 +1548,32 @@ impl Core {
                 );
             }
             Err(e) => {
-                error!(%uid, error = %e, "staging write failed; bytes lost");
+                // `src` is the only known copy of bytes the kernel already
-                let _ = std::fs::remove_file(src);
+                // 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,
+                );
             }
         }
     }
@@ -1524,6 +1599,27 @@ impl Core {
         Ok(out.into_inner())
     }
 
+    /// Download a whole file directly into `out`, without retaining its
+    /// plaintext in memory. This is the disk-backed twin of
+    /// [`Core::download_file_tracked`], and preserves the same transfer
+    /// accounting while allowing callers that only need an on-disk file to
+    /// keep memory use independent of file size.
+    fn download_file_tracked_to<W: Write>(
+        &self,
+        uid: &NodeUid,
+        name: &str,
+        total: u64,
+        out: W,
+    ) -> std::result::Result<W, ProtonError> {
+        let guard = self
+            .transfers
+            .begin(name, uid.to_string(), TransferDirection::Download, total);
+        let mut out = CountingWriter::new(out, &guard);
+        self.rt
+            .block_on(self.client.download_file_to(uid, &mut out))?;
+        Ok(out.into_inner())
+    }
+
     /// Like [`download_file_tracked`] for a photo, streaming through the photos
     /// client's [`download_photo_to`].
     ///
@@ -1760,6 +1856,24 @@ impl Core {
             .collect())
     }
 
+    /// Return an absolute path through the most specific live on-demand mount
+    /// covering `uid`. A node may be below several configured roots, so prefer
+    /// the root leaving the shortest descendant path.
+    fn mounted_search_path(&self, uid: &str) -> Option<String> {
+        let live_mounts = self.mounts.lock();
+        let folders = self.db.sync_folder_list().ok()?;
+        folders
+            .into_iter()
+            .filter(|folder| folder.mode == "ondemand" && live_mounts.contains_key(&folder.id))
+            .filter_map(|folder| {
+                let relative = self.db.path_relative_to(&folder.remote_uid, uid).ok()??;
+                let depth = relative.split('/').count();
+                Some((depth, PathBuf::from(folder.local_path).join(relative)))
+            })
+            .min_by_key(|(depth, _)| *depth)
+            .map(|(_, path)| path.to_string_lossy().into_owned())
+    }
+
     /// Full-text search node names against the local SQLite index, mapping each
     /// DB hit to the wire [`SearchHit`] (resolving live pin state from the cache,
     /// which the DB doesn't track). Pure local lookup — never hits the network.
@@ -1778,6 +1892,8 @@ impl Core {
                 modified: h.node.modification_time,
                 pinned: self.cache.is_pinned(&h.node.uid),
                 uid: h.node.uid.to_string(),
+                mounted_path: self.mounted_search_path(&h.node.uid.to_string()),
+                score: 0,
             })
             .collect())
     }
@@ -1798,10 +1914,96 @@ impl Core {
                 is_dir: h.is_dir,
                 size: h.size.max(0) as u64,
                 modified: h.mtime,
+                score: 0,
             })
             .collect())
     }
 
+    /// Search both prompt sources with identical query and per-source limits.
+    /// Keeping the composition in the daemon core gives every control surface
+    /// the same semantics while preserving the legacy source-specific methods.
+    fn search_v2(
+        &self,
+        query: &str,
+        limit: usize,
+        filters: &SearchFilters,
+    ) -> CoreResult<(Vec<SearchHit>, Vec<LocalHit>)> {
+        if query.trim().is_empty() || limit == 0 {
+            return Ok((Vec::new(), Vec::new()));
+        }
+        // Candidate generation is intentionally broader than the final result
+        // cap: filtering and fuzzy rejection happen below, so limiting at the
+        // SQLite boundary to `limit` would recreate the old empty-filter bug.
+        let candidate_limit = limit.saturating_mul(10).clamp(100, 1_000);
+        let mut drive_hits = if filters.sources.contains(&SearchSource::Drive) {
+            self.db
+                .search_candidates(query, candidate_limit)
+                .map_err(|e| CoreError::from_api(&e, "search candidates"))?
+                .into_iter()
+                .filter(|hit| filters.kind.accepts(&hit.node.name, hit.node.is_folder()))
+                .filter_map(|hit| {
+                    let parent = Path::new(&hit.path)
+                        .parent()
+                        .map_or_else(String::new, |path| path.display().to_string());
+                    let pinned = self.cache.is_pinned(&hit.node.uid);
+                    let score = relevance_score(query, &hit.node.name, &parent)?
+                        + if pinned { 250 } else { 0 };
+                    Some(SearchHit {
+                        name: hit.node.name.clone(),
+                        path: hit.path,
+                        is_dir: hit.node.is_folder(),
+                        size: node_size(&hit.node),
+                        modified: hit.node.modification_time,
+                        pinned,
+                        uid: hit.node.uid.to_string(),
+                        mounted_path: self.mounted_search_path(&hit.node.uid.to_string()),
+                        score,
+                    })
+                })
+                .collect()
+        } else {
+            Vec::new()
+        };
+        let mut local_hits = if filters.sources.contains(&SearchSource::Local) {
+            self.db
+                .search_local_candidates(query, candidate_limit)
+                .map_err(|e| CoreError::from_api(&e, "local search candidates"))?
+                .into_iter()
+                .filter(|hit| filters.kind.accepts(&hit.name, hit.is_dir))
+                .filter_map(|hit| {
+                    let parent = Path::new(&hit.path)
+                        .parent()
+                        .map_or_else(String::new, |path| path.display().to_string());
+                    Some(LocalHit {
+                        score: relevance_score(query, &hit.name, &parent)?,
+                        name: hit.name,
+                        path: hit.path,
+                        is_dir: hit.is_dir,
+                        size: hit.size.max(0) as u64,
+                        modified: hit.mtime,
+                    })
+                })
+                .collect()
+        } else {
+            Vec::new()
+        };
+        drive_hits.sort_by(|a, b| {
+            b.score
+                .cmp(&a.score)
+                .then_with(|| a.name.cmp(&b.name))
+                .then_with(|| a.path.cmp(&b.path))
+        });
+        local_hits.sort_by(|a, b| {
+            b.score
+                .cmp(&a.score)
+                .then_with(|| a.name.cmp(&b.name))
+                .then_with(|| a.path.cmp(&b.path))
+        });
+        drive_hits.truncate(limit);
+        local_hits.truncate(limit);
+        Ok((drive_hits, local_hits))
+    }
+
     /// A page of the photos timeline (newest first), sliced out of the DB.
     /// `Ok(None)` when the account has no photos volume.
     ///
@@ -1847,12 +2049,43 @@ impl Core {
         if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
             return Ok(p);
         }
-        let bytes = self
+
-            .download_file_tracked(&uid, &name, size)
+        // Keep the temporary file beside the final cache blob. `store_file`
-            .map_err(|e| CoreError::from_api(&e, "download"))?;
+        // can then adopt it with a hard link and atomic rename, while the SDK
-        self.cache
+        // streams decrypted blocks to disk instead of building a file-sized
-            .store(&uid, mtime, size, &bytes)
+        // `Vec<u8>`. A unique name also keeps concurrent opens of the same node
-            .map_err(|e| CoreError::internal(format!("cache store: {e}")))?;
+        // from sharing a partially written staging file.
+        static OPEN_TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
+        let seq = OPEN_TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
+        let tmp = self.cache.content_path(&uid).with_extension(format!(
+            "open-{}-{}-{seq}.tmp",
+            std::process::id(),
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .map(|d| d.as_nanos())
+                .unwrap_or(0)
+        ));
+        let file = File::options()
+            .write(true)
+            .create_new(true)
+            .open(&tmp)
+            .map_err(|e| CoreError::internal(format!("create download temp: {e}")))?;
+        let file = match self.download_file_tracked_to(&uid, &name, size, file) {
+            Ok(file) => file,
+            Err(e) => {
+                let _ = std::fs::remove_file(&tmp);
+                return Err(CoreError::from_api(&e, "download"));
+            }
+        };
+        if let Err(e) = file.sync_all() {
+            let _ = std::fs::remove_file(&tmp);
+            return Err(CoreError::internal(format!("sync download temp: {e}")));
+        }
+        if let Err(e) = self.cache.store_file(&uid, mtime, size, &tmp) {
+            let _ = std::fs::remove_file(&tmp);
+            return Err(CoreError::internal(format!("cache store: {e}")));
+        }
+        let _ = std::fs::remove_file(&tmp);
         Ok(self.cache.content_path(&uid))
     }
 
@@ -2132,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 {
@@ -2700,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 {
@@ -3063,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
@@ -3170,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)
+        // A resident listing is authoritative even when empty. Falling through
-            && !kids.is_empty()
+        // from `Some([])` to SQLite lets an obsolete child row contradict the
-        {
+        // listing `ensure_children` just established, so readdir says empty but
-            return true;
+        // 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,
+    FilePlan, classification_order, conflict_path_with_suffix, filter_baseline, guard_local_wipe,
-    parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size,
+    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) {
+            let meta = std::fs::symlink_metadata(&path)
-                Ok(m) => m,
+                .map_err(|e| format!("stat {}: {e}", path.display()))?;
-                Err(_) => continue,
+            let stripped = path.strip_prefix(root).map_err(|e| {
-            };
+                format!(
-            let Ok(stripped) = path.strip_prefix(root) else {
+                    "local path {} escaped sync root {}: {e}",
-                continue;
+                    path.display(),
-            };
+                    root.display()
-            let Some(rel) = stripped.to_str() else {
+                )
-                continue;
+            })?;
-            };
+            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,13 +1176,10 @@ 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());
+                    preserve_conflict_copy(&path, now_secs())
-                    if let Err(e) = std::fs::rename(&path, &conflict) {
+                        .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
-                        warn!(rel, error = %e, "sync: could not set aside conflict copy");
-                    } else {
                     info!(rel, "sync: kept local changes as a conflict copy");
                 }
-                }
                 self.download_file(folder_id, local_root, rel, uid, *mtime, *size)
                     .await?;
                 Ok(Applied::Conflict)
@@ -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());
+                let conflict = preserve_conflict_copy(&path, now_secs())
-                std::fs::rename(&path, &conflict)
                     .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 = std::fs::File::options()
-            let _ =
+            .write(true)
-                f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.max(0) as u64));
+            .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}"),
+        Some(ext) => format!("{stem} (sync-conflict {stamp}{suffix}).{ext}"),
-        None => format!("{stem} (sync-conflict {stamp})"),
+        None => format!("{stem} (sync-conflict {stamp}{suffix})"),
     };
     match path.parent() {
         Some(dir) => dir.join(name),

crates/pdfs-gui/src/activation.rs

@@ -0,0 +1,89 @@
+//! Shared Drive-result activation policy for the GTK browser and quick prompt.
+//!
+//! Media should be handed to its mounted path so players can issue range reads
+//! through FUSE. Other files retain the materialize-then-open behaviour because
+//! many desktop applications expect a stable ordinary file.
+
+use std::path::{Path, PathBuf};
+
+use pdfs_core::control::PhotoKind;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum DriveActivation {
+    Folder,
+    MountedMedia,
+    Materialize,
+}
+
+pub(crate) fn drive_activation(name: &str, is_dir: bool) -> DriveActivation {
+    if is_dir {
+        DriveActivation::Folder
+    } else if is_streamable_media(name) {
+        DriveActivation::MountedMedia
+    } else {
+        DriveActivation::Materialize
+    }
+}
+
+pub(crate) fn mounted_path(mountpoint: &Path, relative_path: &str) -> PathBuf {
+    mountpoint.join(relative_path)
+}
+
+fn is_streamable_media(name: &str) -> bool {
+    if PhotoKind::classify(Some(name), None) == PhotoKind::Video {
+        return true;
+    }
+
+    let extension = Path::new(name)
+        .extension()
+        .and_then(|value| value.to_str())
+        .unwrap_or_default();
+
+    matches!(
+        extension.to_ascii_lowercase().as_str(),
+        "mp3" | "flac" | "wav" | "ogg" | "opus" | "m4a" | "aac"
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn folders_use_the_mount_regardless_of_extension() {
+        assert_eq!(
+            drive_activation("recordings.mp4", true),
+            DriveActivation::Folder
+        );
+    }
+
+    #[test]
+    fn media_extensions_use_fuse_path_case_insensitively() {
+        for name in ["movie.MKV", "voice.opus", "song.FLAC", "clip.m4v"] {
+            assert_eq!(
+                drive_activation(name, false),
+                DriveActivation::MountedMedia,
+                "{name}"
+            );
+        }
+    }
+
+    #[test]
+    fn ordinary_files_are_materialized() {
+        for name in ["report.pdf", "notes.txt", "archive.zip", "no-extension"] {
+            assert_eq!(
+                drive_activation(name, false),
+                DriveActivation::Materialize,
+                "{name}"
+            );
+        }
+    }
+
+    #[test]
+    fn mounted_paths_preserve_relative_components() {
+        assert_eq!(
+            mounted_path(Path::new("/mnt/drive"), "Videos/a movie.mkv"),
+            Path::new("/mnt/drive/Videos/a movie.mkv")
+        );
+    }
+}

crates/pdfs-gui/src/app/main.rs

@@ -1,3 +1,5 @@
+#[path = "../activation.rs"]
+pub(crate) mod activation;
 pub(crate) mod pages;
 pub(crate) mod widgets;
 

crates/pdfs-gui/src/app/pages/browser.rs

@@ -1,3 +1,4 @@
+use crate::activation::{DriveActivation, drive_activation, mounted_path};
 use crate::*;
 
 pub(crate) struct BrowserState {
@@ -493,9 +494,9 @@ pub(crate) fn show_context_menu(ui: &Rc<Ui>, entry: &DirEntry, anchor: &gtk4::Bo
     popover.set_parent(anchor);
     popover.connect_closed(|p| p.unparent());
 
-    // Videos lead with Play (stream from the mount, no download); the plain Open
+    // Media leads with Play (stream from the mount, no download); the plain Open
     // below still downloads a local copy for anyone who wants one.
-    if is_video_entry(entry) {
+    if is_streamable_media_entry(entry) {
         let play = menu_item("Play (stream)", "media-playback-start-symbolic");
         let ui_play = ui.clone();
         let entry_play = entry.clone();
@@ -514,8 +515,8 @@ pub(crate) fn show_context_menu(ui: &Rc<Ui>, entry: &DirEntry, anchor: &gtk4::Bo
     open.connect_clicked(move |_| {
         pop.popdown();
         // Open always means "download a local copy and hand off", even for a
-        // video — `activate_entry` would otherwise stream it.
+        // media file — `activate_entry` would otherwise stream it.
-        if is_video_entry(&entry_open) {
+        if is_streamable_media_entry(&entry_open) {
             download_and_open(&ui_open, &entry_open);
         } else {
             activate_entry(&ui_open, &entry_open);
@@ -610,14 +611,12 @@ pub(crate) fn entry_at(model: Option<&impl IsA<gio::ListModel>>, pos: u32) -> Op
     Some(entry)
 }
 
-/// Whether an entry's name marks it as a video, reusing the same classifier the
+/// Whether an entry can be opened through FUSE instead of fully materialized.
-/// Photos page splits on — so the file browser and the gallery agree on what a
+pub(crate) fn is_streamable_media_entry(entry: &DirEntry) -> bool {
-/// video is.
+    drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia
-pub(crate) fn is_video_entry(entry: &DirEntry) -> bool {
-    !entry.is_dir && PhotoKind::classify(Some(&entry.name), None) == PhotoKind::Video
 }
 
-/// Stream a video straight from the mount, no download. Drive folders *are* part
+/// Stream media straight from the mount, no download. Drive folders *are* part
 /// of the FUSE mount, so a player pointed at `<mountpoint>/<rel>` reads the file
 /// through [`Core::read_range_remote`] — 4 MB blocks fetched on demand as it
 /// seeks and buffers — instead of waiting for the whole file to land like
@@ -626,11 +625,11 @@ pub(crate) fn is_video_entry(entry: &DirEntry) -> bool {
 pub(crate) fn stream_entry(ui: &Rc<Ui>, entry: &DirEntry) {
     let rel = entry_rel(ui, entry);
     let mountpoint = ui.dirs.resolved_mountpoint(&ui.dirs.load_config());
-    let abs = mountpoint.join(&rel);
+    let abs = mounted_path(&mountpoint, &rel);
     let Some(path) = abs.to_str() else {
         toast_error(
             ui,
-            "Couldn't play video",
+            "Couldn't play media",
             "The file path isn't valid UTF-8.",
         );
         return;
@@ -639,7 +638,7 @@ pub(crate) fn stream_entry(ui: &Rc<Ui>, entry: &DirEntry) {
     play_external(path);
 }
 
-/// Open an entry the Nautilus way: folders descend, videos stream from the mount,
+/// Open an entry the Nautilus way: folders descend, media streams from the mount,
 /// other files download-and-open.
 pub(crate) fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) {
     let rel = entry_rel(ui, entry);
@@ -651,8 +650,8 @@ pub(crate) fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) {
         }
         *ui.browser.path.borrow_mut() = rel;
         load_browser(ui);
-    } else if is_video_entry(entry) {
+    } else if drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia {
-        // A video streams rather than downloads: that is exactly the "play it,
+        // Media streams rather than downloads: that is exactly the "play it,
         // don't fetch the whole thing" behaviour this is for.
         stream_entry(ui, entry);
     } else {

crates/pdfs-gui/src/prompt.rs

@@ -1,11 +1,11 @@
 //! `pdfs-prompt` — the launcher: one search field over both Proton Drive and the
 //! files on this machine, styled after Google Drive's search overlay.
 //!
-//! Bind it to a system shortcut (e.g. in Hyprland) for a quick HUD search. It is
+//! Bind it to a system shortcut (e.g. in Hyprland) for a quick HUD search. The
-//! a short-lived process: it must draw *now*, so nothing here blocks the main
+//! application is single-instance and keeps its window alive between summons,
-//! loop. Every daemon round-trip runs on a worker thread and lands back through
+//! so repeat activations only reset and present the existing widget tree. Every
-//! an async channel; the two searches (Drive and local) are fired in parallel and
+//! daemon round-trip runs on a worker thread and lands back through an async
-//! each section renders as its own reply arrives.
+//! channel. Drive and local lookup share one request so a result set paints once.
 
 use std::cell::{Cell, RefCell};
 use std::path::{Path, PathBuf};
@@ -17,7 +17,12 @@ use adw::prelude::*;
 use gtk4::{gio, glib};
 
 use pdfs_core::config::AppDirs;
-use pdfs_core::control::{LocalHit, Request, Response, SearchHit, send};
+use pdfs_core::control::{
+    LocalHit, Request, Response, SearchFilters, SearchHit, SearchKind, SearchSource, send,
+};
+
+mod activation;
+use activation::{DriveActivation, drive_activation, mounted_path};
 
 const APP_ID: &str = "io.narl.proton-drive-linux-prompt";
 
@@ -60,6 +65,16 @@ impl Filter {
             Filter::Media => !is_dir && is_media(name),
         }
     }
+
+    fn search_kind(self) -> SearchKind {
+        match self {
+            Self::All => SearchKind::All,
+            Self::Folders => SearchKind::Folders,
+            Self::Documents => SearchKind::Documents,
+            Self::Images => SearchKind::Images,
+            Self::Media => SearchKind::Media,
+        }
+    }
 }
 
 fn extension(name: &str) -> String {
@@ -174,6 +189,13 @@ impl Hit {
         matches!(self, Hit::Drive(h) if h.pinned)
     }
 
+    fn score(&self) -> i64 {
+        match self {
+            Self::Drive(hit) => hit.score,
+            Self::Local(hit) => hit.score,
+        }
+    }
+
     /// Stable identity used to preserve the keyboard cursor across re-renders.
     /// The bool keeps a Drive and a local hit at the same path distinct.
     fn key(&self) -> (bool, String) {
@@ -184,6 +206,15 @@ impl Hit {
     }
 }
 
+fn rank_hits(hits: &mut [Hit]) {
+    hits.sort_by(|a, b| {
+        b.score()
+            .cmp(&a.score())
+            .then_with(|| a.name().cmp(b.name()))
+            .then_with(|| a.key().cmp(&b.key()))
+    });
+}
+
 /// Everything the window needs to render and act, in one place so the many
 /// callbacks below capture a single `Rc` instead of a dozen clones each.
 struct Ui {
@@ -195,6 +226,7 @@ struct Ui {
     stack: gtk4::Stack,
     scroller: gtk4::ScrolledWindow,
     results: gtk4::Box,
+    search_section: Section,
     drive_section: Section,
     local_section: Section,
     placeholder: adw::StatusPage,
@@ -217,8 +249,7 @@ struct Ui {
     /// Monotonic query id. A reply whose id is stale (the user typed again while
     /// it was in flight) is dropped instead of overwriting fresher results.
     query_id: Cell<u64>,
-    /// In-flight searches for the current query id, so the spinner stops only
+    /// In-flight requests for the current query id.
-    /// once both sections have answered.
     pending: Cell<u8>,
     /// True once a Drive open is running: the window is about to close, so
     /// further input is ignored.
@@ -229,6 +260,10 @@ struct Ui {
     /// against the selected row once the fresh results settle, rather than being
     /// dropped — so a plain "type, Enter" always opens something.
     open_pending: Cell<bool>,
+    /// Prevent lifecycle-driven entry resets from also scheduling the normal
+    /// debounced empty-query request; activation performs one explicit daemon
+    /// bootstrap instead.
+    suppress_entry_change: Cell<bool>,
     /// The window, so a deferred open (above) can reach it from `render`.
     window: RefCell<Option<adw::ApplicationWindow>>,
 }
@@ -301,17 +336,33 @@ fn main() -> glib::ExitCode {
         .init();
 
     let app = adw::Application::builder().application_id(APP_ID).build();
+    // GtkApplication normally remains active while it owns a window, including
+    // when that window is hidden. Hold it explicitly as well: residency is a
+    // product requirement here, not an incidental window-lifetime side effect.
+    let _resident = app.hold();
     app.connect_startup(|_| load_css());
-    app.connect_activate(build_window);
+    let prompt: Rc<RefCell<Option<Rc<Ui>>>> = Rc::new(RefCell::new(None));
+    app.connect_activate(move |app| {
+        let ui = if let Some(ui) = prompt.borrow().clone() {
+            ui
+        } else {
+            let Some(ui) = build_window(app) else {
+                return;
+            };
+            *prompt.borrow_mut() = Some(ui.clone());
+            ui
+        };
+        ui.activate();
+    });
     app.run()
 }
 
-fn build_window(app: &adw::Application) {
+fn build_window(app: &adw::Application) -> Option<Rc<Ui>> {
     let dirs = match AppDirs::new() {
         Ok(d) => d,
         Err(e) => {
             tracing::error!("cannot resolve app dirs: {e}");
-            return;
+            return None;
         }
     };
 
@@ -362,8 +413,10 @@ fn build_window(app: &adw::Application) {
     let results = gtk4::Box::new(gtk4::Orientation::Vertical, 0);
     results.add_css_class("results");
 
+    let search_section = Section::new("Best matches", "system-search-symbolic");
     let drive_section = Section::new("Proton Drive", "folder-remote-symbolic");
     let local_section = Section::new("This computer", "drive-harddisk-symbolic");
+    search_section.attach(&results);
     drive_section.attach(&results);
     local_section.attach(&results);
 
@@ -430,6 +483,7 @@ fn build_window(app: &adw::Application) {
         stack,
         scroller,
         results,
+        search_section,
         drive_section,
         local_section,
         placeholder,
@@ -446,6 +500,7 @@ fn build_window(app: &adw::Application) {
         opening: Cell::new(false),
         indexing: Cell::new(false),
         open_pending: Cell::new(false),
+        suppress_entry_change: Cell::new(false),
         window: RefCell::new(None),
     });
     *ui.window.borrow_mut() = Some(window.clone());
@@ -477,15 +532,17 @@ fn build_window(app: &adw::Application) {
         (&ui.local_section.list, true),
     ] {
         let ui_row = ui.clone();
-        let window_row = window.clone();
         list.connect_row_activated(move |_, row| {
             let offset = if is_local { ui_row.drive_count() } else { 0 };
-            ui_row.open(offset + row.index() as usize, &window_row);
+            ui_row.open(offset + row.index() as usize);
         });
     }
+    let ui_row = ui.clone();
+    ui.search_section
+        .list
+        .connect_row_activated(move |_, row| ui_row.open(row.index() as usize));
 
     let ui_key = ui.clone();
-    let window_key = window.clone();
     let keys = gtk4::EventControllerKey::new();
     keys.connect_key_pressed(move |_, key, _, state| {
         if ui_key.opening.get() {
@@ -496,7 +553,7 @@ fn build_window(app: &adw::Application) {
                 // Spotlight-style: first Escape clears a non-empty query, a
                 // second (or an already-empty box) dismisses the launcher.
                 if ui_key.entry.text().is_empty() {
-                    window_key.close();
+                    ui_key.dismiss();
                 } else {
                     ui_key.entry.set_text("");
                 }
@@ -532,19 +589,41 @@ fn build_window(app: &adw::Application) {
     // focused entry is never reached. Enter with focus in the list is covered by
     // `row_activated` above.
     let ui_activate = ui.clone();
-    let window_activate = window.clone();
     entry.connect_activate(move |_| {
         if ui_activate.opening.get() {
             return;
         }
         // Fall back to the top row when nothing is explicitly selected (e.g.
         // results haven't rendered yet); `open` guards the rest.
-        ui_activate.open(ui_activate.cursor.get().unwrap_or(0), &window_activate);
+        ui_activate.open(ui_activate.cursor.get().unwrap_or(0));
     });
 
+    // Some GTK/input-method combinations consume the physical Return binding in
+    // GtkText without emitting Entry::activate. Capture only Return ahead of the
+    // text widget and leave every other key to normal GTK/IME handling. Stopping
+    // propagation also prevents the activate signal above from opening twice;
+    // that signal remains the programmatic/accessibility activation path.
+    let ui_return = ui.clone();
+    let return_key = gtk4::EventControllerKey::new();
+    return_key.set_propagation_phase(gtk4::PropagationPhase::Capture);
+    return_key.connect_key_pressed(move |_, key, _, _| {
+        if matches!(key, gtk4::gdk::Key::Return | gtk4::gdk::Key::KP_Enter) {
+            if !ui_return.opening.get() {
+                ui_return.open(ui_return.cursor.get().unwrap_or(0));
+            }
+            glib::Propagation::Stop
+        } else {
+            glib::Propagation::Proceed
+        }
+    });
+    entry.add_controller(return_key);
+
     let debounce: Rc<RefCell<Option<glib::SourceId>>> = Rc::new(RefCell::new(None));
     let ui_changed = ui.clone();
     entry.connect_changed(move |entry| {
+        if ui_changed.suppress_entry_change.get() {
+            return;
+        }
         // A fresh keystroke supersedes any Enter that was waiting on results:
         // the user is still refining the query, not asking to open yet.
         ui_changed.open_pending.set(false);
@@ -564,22 +643,59 @@ fn build_window(app: &adw::Application) {
     let ui_retry = ui.clone();
     retry.connect_clicked(move |_| ui_retry.connect());
 
-    // A launcher is modal by convention: losing focus dismisses it.
+    // Keep the widget tree alive when the compositor asks the window to close.
-    window.connect_is_active_notify(|w| {
+    // This also avoids running GTK's destruction path while async replies still
-        if !w.is_active() {
+    // hold references to prompt widgets.
-            w.close();
+    let ui_close = ui.clone();
-        }
+    window.connect_close_request(move |_| {
+        ui_close.dismiss();
+        glib::Propagation::Stop
     });
     let entry_map = entry.clone();
     window.connect_map(move |_| {
         entry_map.grab_focus();
     });
 
-    ui.connect();
+    Some(ui)
-    window.present();
 }
 
 impl Ui {
+    /// Reset and present the resident prompt in response to GApplication
+    /// activation. A second invocation while a file is materialising merely
+    /// raises the progress window; it must not make the in-flight open mutable.
+    fn activate(self: &Rc<Self>) {
+        let Some(window) = self.window.borrow().clone() else {
+            return;
+        };
+        if !self.opening.get() {
+            // Supersede replies from the previous showing before clearing the
+            // query. The daemon work may finish, but it can no longer repaint
+            // the newly activated prompt with stale rows.
+            self.query_id.set(self.query_id.get() + 1);
+            self.pending.set(0);
+            self.open_pending.set(false);
+            self.spinner.stop();
+            self.spinner.set_visible(false);
+            self.suppress_entry_change.set(true);
+            self.entry.set_text("");
+            self.suppress_entry_change.set(false);
+            self.set_filter(Filter::All);
+            self.connect();
+        }
+        window.present();
+        self.entry.grab_focus();
+    }
+
+    /// Dismiss without destroying the application window. GTK continues to
+    /// associate the hidden window with the single application instance, so a
+    /// later desktop shortcut invocation is a cheap `activate` round-trip.
+    fn dismiss(&self) {
+        self.open_pending.set(false);
+        if let Some(window) = self.window.borrow().as_ref() {
+            window.set_visible(false);
+        }
+    }
+
     /// Ask the daemon for status and, if it answers, load the suggested (pinned)
     /// files. Runs off-thread: a dead daemon must not freeze the first frame.
     fn connect(self: &Rc<Self>) {
@@ -640,6 +756,8 @@ impl Ui {
                         modified: 0,
                         pinned: true,
                         uid: pin.uid,
+                        mounted_path: None,
+                        score: 0,
                     })
                     .collect(),
                 _ => Vec::new(),
@@ -650,32 +768,41 @@ impl Ui {
         });
     }
 
-    /// Fire both searches for `query` in parallel. An empty query falls back to
+    /// Search Drive and local metadata in one daemon round-trip. An empty query
-    /// the suggestions view.
+    /// falls back to the suggestions view.
     fn search(self: &Rc<Self>, query: &str) {
         if query.is_empty() {
             self.load_suggestions();
             return;
         }
 
-        let id = self.begin_query(2);
+        let id = self.begin_query(1);
-
+        let search = spawn_request(
-        let drive = spawn_request(
             self.socket.clone(),
-            Request::Search {
+            Request::SearchV2 {
                 query: query.to_string(),
                 limit: SEARCH_LIMIT,
+                filters: SearchFilters {
+                    sources: vec![SearchSource::Drive, SearchSource::Local],
+                    kind: self.filter.get().search_kind(),
+                },
             },
         );
         let ui = self.clone();
         glib::spawn_future_local(async move {
-            let reply = drive.recv().await;
+            let reply = search.recv().await;
             if !ui.finish_query(id) {
                 return;
             }
             match reply {
-                Ok(Ok(Response::SearchResults { hits })) => {
+                Ok(Ok(Response::SearchResultsV2 {
-                    *ui.drive_hits.borrow_mut() = hits;
+                    drive_hits,
+                    local_hits,
+                    local_indexing,
+                })) => {
+                    *ui.drive_hits.borrow_mut() = drive_hits;
+                    *ui.local_hits.borrow_mut() = local_hits;
+                    ui.indexing.set(local_indexing);
                     ui.render();
                 }
                 // Transport failure: the daemon is gone. Don't paint "No
@@ -683,32 +810,6 @@ impl Ui {
                 Ok(Err(_)) | Err(_) => ui.go_offline(),
                 _ => {
                     ui.drive_hits.borrow_mut().clear();
-                    ui.render();
-                }
-            }
-        });
-
-        let local = spawn_request(
-            self.socket.clone(),
-            Request::SearchLocal {
-                query: query.to_string(),
-                limit: SEARCH_LIMIT,
-            },
-        );
-        let ui = self.clone();
-        glib::spawn_future_local(async move {
-            let reply = local.recv().await;
-            if !ui.finish_query(id) {
-                return;
-            }
-            match reply {
-                Ok(Ok(Response::LocalResults { hits, indexing })) => {
-                    ui.indexing.set(indexing);
-                    *ui.local_hits.borrow_mut() = hits;
-                    ui.render();
-                }
-                Ok(Err(_)) | Err(_) => ui.go_offline(),
-                _ => {
                     ui.local_hits.borrow_mut().clear();
                     ui.render();
                 }
@@ -751,7 +852,12 @@ impl Ui {
         for (chip_filter, chip) in self.chips.borrow().iter() {
             chip.set_active(*chip_filter == filter);
         }
+        let query = self.entry.text().trim().to_string();
+        if query.is_empty() {
             self.render();
+        } else {
+            self.search(&query);
+        }
     }
 
     fn cycle_filter(self: &Rc<Self>, backwards: bool) {
@@ -776,9 +882,8 @@ impl Ui {
             .count()
     }
 
-    /// Rebuild both sections from the last replies under the active chip, then
+    /// Rebuild suggestions or the globally-ranked search section, preserving the
-    /// restore the cursor to the top. Every state change funnels through here, so
+    /// selected identity when a result survives the update.
-    /// the list, the placeholder, and the footer can never disagree.
     fn render(self: &Rc<Self>) {
         let filter = self.filter.get();
 
@@ -808,17 +913,24 @@ impl Ui {
 
         let query = self.entry.text().trim().to_string();
         let searching = !query.is_empty();
-        let drive_title = if searching {
-            "Proton Drive"
-        } else {
-            "Pinned in Proton Drive"
-        };
-        self.drive_section.set_rows(&drive, Some(drive_title));
-        self.local_section.set_rows(&local, None);
-
         let total = drive.len() + local.len();
         let mut visible = drive;
         visible.extend(local);
+        if searching {
+            rank_hits(&mut visible);
+            self.search_section.set_rows(&visible, Some("Best matches"));
+            self.drive_section.set_rows(&[], None);
+            self.local_section.set_rows(&[], None);
+        } else {
+            self.search_section.set_rows(&[], None);
+            let drive_count = visible
+                .iter()
+                .take_while(|hit| matches!(hit, Hit::Drive(_)))
+                .count();
+            self.drive_section
+                .set_rows(&visible[..drive_count], Some("Pinned in Proton Drive"));
+            self.local_section.set_rows(&visible[drive_count..], None);
+        }
 
         // Find where the previously-selected row landed in the rebuilt list.
         let restored = selected_key.and_then(|key| visible.iter().position(|hit| hit.key() == key));
@@ -854,8 +966,8 @@ impl Ui {
         // so it lands on the final list, not a half-populated one.
         if self.open_pending.get() && self.pending.get() == 0 {
             self.open_pending.set(false);
-            if let (Some(index), Some(window)) = (self.cursor.get(), self.window.borrow().clone()) {
+            if let Some(index) = self.cursor.get() {
-                self.open(index, &window);
+                self.open(index);
             }
         }
     }
@@ -892,6 +1004,19 @@ impl Ui {
     /// section owns it, clear the other section's selection, and scroll it into
     /// view. Focus stays in the entry so typing never breaks.
     fn select(self: &Rc<Self>, index: usize) {
+        if !self.rendered_query.borrow().is_empty() {
+            self.drive_section.list.unselect_all();
+            self.local_section.list.unselect_all();
+            let Some(row) = self.search_section.list.row_at_index(index as i32) else {
+                return;
+            };
+            self.search_section.list.select_row(Some(&row));
+            self.cursor.set(Some(index));
+            self.scroll_into_view(&row);
+            return;
+        }
+
+        self.search_section.list.unselect_all();
         let drive_count = self.drive_count();
         let (list, other, local_index) = if index < drive_count {
             (
@@ -938,7 +1063,7 @@ impl Ui {
     /// straight to the desktop; a Drive hit must first be hydrated into the cache
     /// by the daemon (`OpenFile`), which can take a moment — the window stays up,
     /// with the spinner running, until the path comes back.
-    fn open(self: &Rc<Self>, index: usize, window: &adw::ApplicationWindow) {
+    fn open(self: &Rc<Self>, index: usize) {
         if self.opening.get() {
             return;
         }
@@ -957,15 +1082,36 @@ impl Ui {
         match hit {
             Hit::Local(local) => {
                 xdg_open(Path::new(&local.path));
-                window.close();
+                self.dismiss();
-            }
-            // A Drive folder exists in the mount — open it in the file manager
-            // rather than downloading anything.
-            Hit::Drive(drive) if drive.is_dir => {
-                xdg_open(&self.mountpoint.borrow().join(&drive.path));
-                window.close();
             }
             Hit::Drive(drive) => {
+                // Empty-query rows come from ListPins, whose `recursive` flag is
+                // pin policy rather than reliable node kind metadata. Every pin
+                // already has a mounted path, so activate that path directly;
+                // this handles non-recursively pinned folders without mistaking
+                // them for files and sending an invalid OpenFile request.
+                if self.rendered_query.borrow().is_empty() {
+                    let path = drive.mounted_path.as_deref().map_or_else(
+                        || mounted_path(&self.mountpoint.borrow(), &drive.path),
+                        PathBuf::from,
+                    );
+                    xdg_open(&path);
+                    self.dismiss();
+                    return;
+                }
+                match drive_activation(&drive.name, drive.is_dir) {
+                    DriveActivation::Folder | DriveActivation::MountedMedia => {
+                        let path = drive.mounted_path.as_deref().map_or_else(
+                            || mounted_path(&self.mountpoint.borrow(), &drive.path),
+                            PathBuf::from,
+                        );
+                        xdg_open(&path);
+                        self.dismiss();
+                        return;
+                    }
+                    DriveActivation::Materialize => {}
+                }
+
                 self.opening.set(true);
                 self.spinner.set_visible(true);
                 self.spinner.start();
@@ -978,7 +1124,6 @@ impl Ui {
                     },
                 );
                 let ui = self.clone();
-                let window = window.clone();
                 glib::spawn_future_local(async move {
                     let reply = rx.recv().await;
                     ui.spinner.stop();
@@ -987,7 +1132,7 @@ impl Ui {
                     match reply {
                         Ok(Ok(Response::FilePath { path })) => {
                             xdg_open(Path::new(&path));
-                            window.close();
+                            ui.dismiss();
                         }
                         Ok(Ok(Response::Error { message, .. })) => {
                             ui.hint.set_label(&format!("Could not open: {message}"));
@@ -1288,3 +1433,37 @@ fn load_css() {
         );
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ranked_hits_merge_sources_on_one_score_scale() {
+        let mut hits = vec![
+            Hit::Drive(SearchHit {
+                name: "alpha.pdf".into(),
+                path: "Drive/alpha.pdf".into(),
+                is_dir: false,
+                size: 1,
+                modified: 1,
+                pinned: false,
+                uid: "v~a".into(),
+                mounted_path: None,
+                score: 100,
+            }),
+            Hit::Local(LocalHit {
+                name: "better.pdf".into(),
+                path: "/home/me/better.pdf".into(),
+                is_dir: false,
+                size: 1,
+                modified: 1,
+                score: 300,
+            }),
+        ];
+
+        rank_hits(&mut hits);
+        assert!(matches!(&hits[0], Hit::Local(hit) if hit.name == "better.pdf"));
+        assert!(matches!(&hits[1], Hit::Drive(hit) if hit.name == "alpha.pdf"));
+    }
+}

docs/ARCHITECTURE.md

@@ -30,10 +30,10 @@ graph TD
 
 | Crate | Primary Role | Key Components | State Management |
 |---|---|---|---|
-| [`pdfs-core`](file:///home/narl/dev/private/proton-drive-linux/crates/pdfs-core) | Core Infrastructure & Services | Cache Bookkeeping, Database migrations/schemas, IPC protocol payloads. | Holds the unified SQLite DB (`Db`) connection and the on-disk cache metadata (`ContentCache`). |
+| [`pdfs-core`](../crates/pdfs-core) | Core Infrastructure & Services | Cache bookkeeping, database migrations/schemas, IPC protocol payloads, and shared search relevance scoring. | Holds the unified SQLite DB (`Db`) connection and the on-disk cache metadata (`ContentCache`). |
-| [`pdfs-fuse`](file:///home/narl/dev/private/proton-drive-linux/crates/pdfs-fuse) | VFS Layer & Reconciliation | FUSE callbacks, background upload queue (`drain`), two-way sync runner. | Manages in-memory inode maps (`State`), active descriptors (`WriteHandle`), and background task threads. |
+| [`pdfs-fuse`](../crates/pdfs-fuse) | VFS Layer & Reconciliation | FUSE callbacks, background upload queue (`drain`), two-way sync runner. | Manages in-memory inode maps (`State`), active descriptors (`WriteHandle`), and background task threads. |
-| [`pdfs-cli`](file:///home/narl/dev/private/proton-drive-linux/crates/pdfs-cli) | Command Line Interface | Command routing, daemon launcher, IPC client wrapper. | Stateless; communicates with daemon over IPC control socket. |
+| [`pdfs-cli`](../crates/pdfs-cli) | Command Line Interface | Command routing, daemon launcher, IPC client wrapper. | Stateless; communicates with daemon over IPC control socket. |
-| [`pdfs-gui`](file:///home/narl/dev/private/proton-drive-linux/crates/pdfs-gui) | Graphical Interface | GTK Page timelines (My files, Shared, Shared with me, Computers, Photos, Activity, Trash, Settings). | Stateless; polls daemon for status and lists timelines via IPC socket. |
+| [`pdfs-gui`](../crates/pdfs-gui) | Graphical Interface | GTK pages, tray, and the resident quick-search prompt. | Keeps UI state only; all durable state and Drive access remain behind the IPC socket. |
 
 ---
 
@@ -127,8 +127,8 @@ Because Proton Drive does not support partial byte writes, modified files must b
 1. **Staging writes (`WriteHandle`):** Writes are stored locally in a `scratch` file. The daemon tracks modified regions using `Intervals` (which holds ranges of edited bytes).
 2. **Close/Release (`queue_revision`):** When the application closes the file descriptor, the daemon:
    - Fetches any untouched gaps from the remote base file to compile the full file.
-   - Moves the scratch file to `staging` under a `{uid}-{millis}-{counter}` name, so a staged blob can be tied back to its node without consulting the database.
+   - Durably publishes the scratch data and authored-range sidecar into `staging` under a `{uid}-{millis}-{counter}` name. Temporary data and metadata are synced before atomic rename, and their directory is synced before the source is removed.
-   - Queues a pending database operation (`PendingOp`). For `OP_REVISION` ops, execution is debounced by 2 seconds (`DRAIN_REVISION_DEBOUNCE = 2s`) to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission.
+   - Transactionally queues or supersedes a pending database operation (`PendingOp`), so insertion failure cannot erase the previously acknowledged upload. For `OP_REVISION` ops, execution is debounced by 2 seconds (`DRAIN_REVISION_DEBOUNCE = 2s`) to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission.
 3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revision uploads, resolves conflicts, and cleans up staging files. Upon landing a revision upload (`refresh_after_upload`), it rebaselines both still-queued ops (`rebaseline_pending`) and open write handles targeting the same node to prevent false self-conflict copies on subsequent writes.
 
 ```mermaid
@@ -170,7 +170,7 @@ sequenceDiagram
 The sync engine handles offline-capable, bidirectional synchronization between the local disk and Proton Drive for directories marked in `mirror` mode.
 
 ### Lifecycle of a Sync Pass
-1. **Walk Local:** Walks the local directory tree recursively, scanning sizes and modification times.
+1. **Walk Local:** Walks the local directory tree recursively, scanning sizes and modification times while carrying a completeness result. A `readdir`, metadata, permission, or transient I/O failure makes the pass non-destructive instead of turning omitted paths into deletions.
 2. **Walk Remote:** Walks the remote database representation. If remote file modification times are updated, it calls the API to decrypt their sizes.
 3. **Load Baseline:** Loads the `sync_entry` database table, which contains the snapshot of both sides during the *last successful sync*.
 4. **Permutation Diffing:** The loop compares the three states (`local`, `remote`, `baseline`) to classify items:
@@ -192,7 +192,7 @@ graph TD
 ```
 
 5. **Depth-Ascending Batching:** Folders are processed first to ensure hierarchies exist before files are placed. Work is executed concurrently up to a set limit.
-6. **Post-Sync Settle:** On success, baseline entries are upserted, timestamps updated, and any pending mode switches (e.g. going on-demand) are evaluated.
+6. **Safety Gate and Settle:** Destructive plans are rejected when the scan is incomplete. Total-wipe protection applies to every non-empty baseline, including a single-entry folder. On success, baseline entries are upserted, timestamps updated, and any pending mode switches (e.g. going on-demand) are evaluated. Failed local deletion or conflict preservation retains the previous baseline and prevents remote content from overwriting the local source.
 
 ---
 
@@ -202,11 +202,18 @@ The CLI and GUI front-ends do not access database files or make network calls di
 
 * **Transport:** IPC over Unix Stream Socket.
 * **Framing:** Line-delimited JSON payloads.
+* **Resource limits:** Requests are limited to 1 MiB, must end with a newline, and must arrive within 10 seconds. At most 64 connection handlers are active at once.
 * **Control Protocol:**
   * Client sends a single JSON line (`Request`).
   * Daemon parses, handles the request, and replies with a single JSON line (`Response`).
   * Timeout durations are separated: **2 seconds** for writes (avoids hangs on defunct sockets) and **120 seconds** for reads (accommodates heavy transfers).
 
+### Unified Search
+
+`SearchV2` is the shared search boundary for the resident prompt. One request carries the query, result limit, requested sources (`Drive`, `Local`), and content kind. The daemon queries Drive metadata and the local home-directory index, then applies [`pdfs_core::search::relevance_score`](../crates/pdfs-core/src/search.rs) to both result sets. Exact and prefix matches rank above substring, abbreviation, and bounded typo matches; all query terms must match either the basename or parent path. Returning scores on one scale lets the GUI merge both sources into a deterministic **Best matches** list.
+
+The prompt is a single-instance GTK application. It retains and hides its window between activations, resets its query state when summoned, and ignores stale asynchronous replies. Folder and streamable audio/video results open through the FUSE mount so applications can issue range reads; ordinary Drive files use the daemon's materialize-then-open path.
+
 ---
 
 ## 7. Subsystem Interaction & Thread Map
@@ -215,7 +222,7 @@ The background daemon relies on the following thread topology:
 
 1. **Main Thread / Dispatch Loop:** Blocks on `fuser::Session` loop. Reads kernel FUSE events and hands off network-bound VFS work to the FUSE workers pool.
 2. **FUSE Workers Pool (11 threads, two lanes):** Bounded thread pool handling network operations. Split into 3 threads reserved for metadata (`lookup`, `readdir`) and 8 general threads that serve transfers (block reads) and fall back to metadata when no transfer is waiting. Reserved threads never accept a transfer — that is what keeps a directory listing from queuing behind saturated downloads (audit A6).
-3. **IPC listener Thread:** Listens on Unix socket connections, spawning a lightweight task per connection to serve front-end status/configuration requests.
+3. **IPC listener Thread:** Listens on the Unix socket and admits at most 64 concurrent, timeout-bound handlers for front-end status/configuration requests.
 4. **Sync Engine Loop Thread:** Serializes sync runs. Wakes on debounced local inotify filesystem changes, remote polling intervals, or manual user requests.
 5. **Drain Queue Worker Thread:** Processes staged writes (`PendingOp`) sequentially, uploading revisions and retrying with exponential backoff on failures.
 
@@ -261,13 +268,13 @@ Two exposures this client does **not** currently mitigate, stated plainly rather
 - **Swap.** Content keys, session keys, and decrypted buffers live in ordinary heap memory. Nothing calls `mlock(2)`, so under memory pressure they may be paged out. Raising `LimitMEMLOCK` in the systemd unit would *not* change this — there is no locking to permit. The effective mitigation is encrypted swap (dm-crypt / `systemd-cryptsetup`), which is standard on a LUKS install.
 - **Kernel page cache.** Plaintext returned through FUSE is cached by the kernel like any other file data, and is likewise swappable. Defeating this would mean `direct_io` on every read, forfeiting the readahead and caching that make the mount usable. The trade is taken deliberately in favour of performance.
 
-### 8.5 File modes: currently inherited, not enforced
+### 8.5 Enforced ownership and file modes
 
-**The daemon does not set restrictive permissions on anything it creates.** Cache and state directories come from `create_dir_all`, and `control.sock` from a plain `UnixListener::bind` — all subject to the process umask, which on a typical desktop yields `0755` directories and a `0755` socket.
+On every start, `AppDirs::ensure` verifies that the state, cache, and configuration paths are real directories owned by the effective user, rejects symlink substitution or the wrong owner, and enforces mode `0700`. Both Unix sockets are changed to `0600` immediately after binding; failure is fatal rather than falling back to an unguarded daemon.
 
-What protects them today is the mode of the XDG parents (`~/.cache`, `~/.local/state`), which are conventionally `0700` — but that is a property of the user's system, not something this client establishes or verifies. On a machine where the home directory is group- or world-traversable, the consequences differ by artifact:
+These checks protect the artifacts even when the surrounding home or XDG parent is traversable:
 
-| Artifact | Exposure if the parent is traversable |
+| Artifact | Why access is restricted |
 |---|---|
 | `content/` | Another local user can read cached plaintext file content |
 | `cache.db` | Another local user can read the full decrypted name/structure index |
@@ -275,14 +282,14 @@ What protects them today is the mode of the XDG parents (`~/.cache`, `~/.local/s
 
 The socket is the sharpest of the three, because it is an authority boundary rather than a data one — connecting to it confers the daemon's authenticated session without any credential.
 
-**Fixed (bugs.md B6):** `AppDirs::ensure` now sets `0700` on the state, cache, and config directories on every start — not only at creation, so a directory that already exists with a permissive mode is tightened — and `config::restrict_socket` sets `0600` on both sockets immediately after `bind`. The directories are restricted *before* the socket is bound, which closes the window between `bind` and `chmod`. A control socket whose mode cannot be set takes the daemon down rather than serving unguarded.
+Configuration publication follows the same fail-closed model: a restricted temporary file is written and synced, atomically renamed, and followed by a directory sync. A malformed existing configuration is reported and preserved rather than overwritten with defaults.
 
 See [RECOVERY.md](RECOVERY.md) for what a lost machine means for the plaintext described in this section, and what to revoke.
 
 ### 8.6 Implications for deployment
 
 - Treat the cache and state directories as being as sensitive as the Drive contents themselves.
-- Multi-user machines warrant checking `~/.cache` and `~/.local/state` are `0700` until §8.5 is addressed.
+- Restored or manually copied profiles must remain owned by the user. The daemon refuses to start against wrong-owner or symlinked sensitive directories.
 - Purging the cache (`pdfs` settings, or `PurgeCache` over IPC) removes content but **not** `cache.db`, and deliberately never removes undrained `staging/` or `recovery/` blobs.
 
 ---
@@ -343,4 +350,3 @@ sequenceDiagram
    This ensures compatibility with all versions of Proton's client verification scripts.
 3. **SRP Handshake Reset:** Because a gated login burns the SRP handshake on the API side, the client cannot simply resume the previous request. Instead, `auth::login_interactive` restarts the SRP process from scratch with the verification credentials attached, keeping the complex handshake details isolated from the front-end.
 4. **CLI Fallback:** Since the CLI has no native web browser engine, hitting the CAPTCHA gate fails immediately with a user-friendly message directing the user to sign in once via the GUI (`pdfs-app`) to persist the authenticated session keys to the system keyring.
-

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
+**Verified:** `cargo clippy -p pdfs-fuse --all-targets -- -D warnings`, the
-`pdfs-fuse` unit suite pass. A fault-injected/live mode-switch test is still needed.
+`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
+an incomplete queued revision is refused rather than guessed. Path-based
-verified the exact rewrite-then-truncate sequence returns `b"a mu"`.
+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
+**Fix history:** A bounded retry of `InvalidRequirements` reduced the window but
-short bounded backoff. Stable API failures are not retried. If every retry
+did not close it; the managed matrix reproduced the same failure after all four
-fails, adopt the already-completed move in inode state before returning `EIO`,
+retries. Combined cross-directory move+rename now enters the durable rename
-so the local model never claims the node remains in its old parent. The full
+queue as a desired end state. Its drain re-fetches the node on every attempt,
-live suite subsequently passed the replacement and combined move/rename cases.
+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/DEVELOPMENT.md

@@ -1,11 +1,11 @@
-# Pending Development & Verification Tasks
+# Development and Verification Status for 1.0
 
-This document compiles all pending, deferred, and unstarted tasks from the various roadmap, audit, and bug-tracking markdown files in the repository. Completed items have been filtered out.
+This document summarizes the work still pending after the 1.0.0 release branch and records the most important recently completed changes. [`BUGS.md`](BUGS.md) is the authoritative issue ledger; [`TESTING.md`](TESTING.md) defines the acceptance commands.
 
 ---
 
 ## 1. Core Feature Roadmap (Horizon Tasks)
-Derived from: [features.md](file:///home/narl/dev/private/proton-drive-linux/features.md) & [features-plan.md](file:///home/narl/dev/private/proton-drive-linux/features-plan.md)
+Consolidated from the historical feature roadmap; current bug status lives in [`BUGS.md`](BUGS.md).
 
 * **Local Cache & Metadata Encryption at Rest (Priority 5 / Horizon 1)**: Encrypt the SQLite database `cache.db` on disk using SQLCipher (`rusqlite` feature `bundled-sqlcipher`), and encrypt raw content cache blocks (`content/blocks/` and `content/scratch/`) using a fast symmetric scheme like AES-GCM or ChaCha20-Poly1305. The encryption key should be derived from the OS Keyring.
 * **Active Bandwidth Throttling & Traffic Shaping (Horizon 1)**: Add speed governors for daemon uploads and downloads. Support configuration variables `max_upload_rate_kbps` and `max_download_rate_kbps` in `config.json` that can be dynamically adjusted over the Unix control socket.
@@ -26,7 +26,7 @@ Derived from: [features.md](file:///home/narl/dev/private/proton-drive-linux/fea
 ---
 
 ## 2. Performance, Correctness & Robustness Items
-Derived from: [plan.md](file:///home/narl/dev/private/proton-drive-linux/plan.md), [improvements.md](file:///home/narl/dev/private/proton-drive-linux/improvements.md), & [photos-video-streaming.md](file:///home/narl/dev/private/proton-drive-linux/photos-video-streaming.md)
+Consolidated from the performance and media-streaming audit notes.
 
 * **B6. Debounce the LRU access touch**: Keep last-touch times in memory and flush them in batches (debounce window of 30-60 seconds) to avoid performing a SQLite `UPDATE` to the `cache_entries` table on every cache hit.
 * **E12. Client in-flight cap shape**: Address the lock discrepancy between `MAX_CONCURRENT_BLOCK_DOWNLOADS = 10` (per-file) and `DEFAULT_MAX_INFLIGHT_BLOCKS = 12` (global client-wide). Either raise the global cap, lower the per-file window, or dynamically calculate the window to prevent single files from starving concurrent transfers.
@@ -40,13 +40,12 @@ Derived from: [plan.md](file:///home/narl/dev/private/proton-drive-linux/plan.md
   * **F3. Short block yields a short read / EOF**: Address the kernel EOF interpretation on short block reads. Either pad the block or fail loudly rather than silently serving short data.
   * **F4. `stream_readahead` check-then-spawn race**: Prevent duplicate concurrent readahead tasks for the same block. Check-then-spawn is currently racey.
   * **F5. `refresh_blocks` permit lock (SDK)**: Ensure URL refreshing (`refresh_blocks`) does not hold an in-flight block permit, preventing expired URLs from pinning client permits.
-  * **F6. Uncapped Unix control socket thread spawns**: Replace the unbounded `std::thread::spawn` per accepted socket connection in `run_control_socket` with a bounded pool or handler limits to prevent thread exhaustion.
   * **F7. Document temporary cache file safety**: Add explanatory comment to `cache.rs:362` explaining why `with_extension("tmp")` is safe there but not in `store_thumbnail`.
 
 ---
 
 ## 3. Open & Unverified Bugs
-Derived from: [docs/bugs.md](file:///home/narl/dev/private/proton-drive-linux/docs/bugs.md)
+Derived from: [`BUGS.md`](BUGS.md)
 
 * **B2. Unattributed Trash Origin**: Investigate why deleted/moved files are occasionally sent to trash instead of vanishing on older rename operations.
 * **B10. GLib Critical Warning on HUD Close**: Assertion failure `g_list_store_remove: assertion '!g_sequence_iter_is_end (it)' failed` when prompt window closes during active launch (`xdg-open`). Needs investigation under `G_DEBUG=fatal-criticals`.
@@ -54,11 +53,17 @@ Derived from: [docs/bugs.md](file:///home/narl/dev/private/proton-drive-linux/do
 * **B8 (CAPTCHA sign-in bridge)**: Verify a real gated sign-in using Webkit CAPTCHA completion.
 * **Draft Revision Upload loop**: Verify SDK-side deletion and retry logic when a conflicting draft revision exists under the same client UID.
 * **Optimistic Size Loss on Restart**: Verify that pending uploads retain their optimistic sizes across a daemon restart.
+* **B48 — stale listing after unlink**: Session tombstones and authoritative empty listings are implemented; run focused live verification against eventually consistent remote enumeration.
+* **B49–B55 — data-safety fixes**: Conflict preservation, orphan/staging durability, transactional supersession, one-entry wipe protection, and incomplete-scan handling have regression coverage but still require the fault, power-loss, and live mode-matrix cases recorded in `BUGS.md`.
+* **B56–B60 — fail-closed state and IPC fixes**: Future schemas, event cursor advancement, socket limits, private-directory validation, and atomic config saves are implemented. Malformed-schema, database-full, connection-flood, wrong-owner, and concurrent/faulted-save verification remains.
+* **B61 — credential rotation persistence**: Surface and retry keyring persistence failures so a rotated single-use refresh token cannot be lost on restart.
+* **B62–B65 — release assurance**: Clean-install package verification, complete version/protocol identity, recovery sign-off, advisory/license policy, SBOM, provenance, checksums, and signatures remain.
+* **B66–B68 — lifecycle and recovery**: Add queue flush/shutdown protection, explicit device adoption during fresh-state restore, and move profile backup below a writable remote folder.
 
 ---
 
 ## 4. Testing, Verification & Disaster Recovery Drills
-Derived from: [testing.md](file:///home/narl/dev/private/proton-drive-linux/testing.md) & [docs/RECOVERY.md](file:///home/narl/dev/private/proton-drive-linux/docs/RECOVERY.md)
+Derived from: [`TESTING.md`](TESTING.md) and [`RECOVERY.md`](RECOVERY.md)
 
 * **IPC Memory and Performance Stress Tests (Phase 3)**: Verify memory allocations do not balloon when sending large `UploadPhoto` JSON payloads. Test control socket timeout behaviors under long-running block operations.
 * **Fault Injection and Resiliency Tests (Phase 4)**:
@@ -71,11 +76,13 @@ Derived from: [testing.md](file:///home/narl/dev/private/proton-drive-linux/test
   3. Wipe the local state directory and SQLite cache.
   4. Restart the daemon.
   5. Run `pdfs sync restore` and assert the recovered filesystem matches the source byte-for-byte.
+* **Managed live mode matrix**: Before release, run `scripts/fuse-acceptance.sh --managed-live <empty-a> <empty-b>` with `PDFS_ACCEPTANCE_ONLY` unset. It validates on-demand/on-demand, on-demand/mirror, mirror/on-demand, and mirror/mirror transitions.
+* **Crash-consistency matrix**: Inject failure after scratch/sidecar sync, each staging rename, directory sync, pending-op insert/commit, and source deletion; prove that a complete discoverable copy and either the old or new queue record always survives.
 
 ---
 
 ## 5. Fixed & Resolved Bugs
-Derived from: [docs/bugs.md](file:///home/narl/dev/private/proton-drive-linux/docs/bugs.md)
+Derived from: [`BUGS.md`](BUGS.md)
 
 ### Completely Fixed & Verified
 * **B1 — FUSE rename loses the file (data loss)**: In `rename`, the node's database row was deleted while only updating the in-memory state, causing the file to vanish on next listing sync. Fixed by correctly calling `st.invalidate_listing(newparent)` to refresh SQLite cache states.
@@ -88,6 +95,19 @@ Derived from: [docs/bugs.md](file:///home/narl/dev/private/proton-drive-linux/do
 * **B12 — cold enumeration is slow per entry, and goes superlinear past ~500**: S2K decryption and thumbnail resolution loops were serial. Fixed by parallelizing key derivations and caching metadata.
 * **B13 — `rename` over an existing destination fails instead of replacing it**: Fixed POSIX rename target overwrite behavior to replace existing destinations.
 * **B14 — provisional (ciphertext) sizes make rsync read short and abort the file**: Adjusted size reporting to align logical and ciphertext sizing, preventing premature EOF.
+* **B45 — truncate over a queued rewrite**: Path-based and handle-based truncate now compose over the complete pending blob; shrink, growth, and sparse I/O are live-verified.
+* **B46 — combined cross-directory move and rename**: The desired end state is durably queued and reconciles partially completed remote operations; replacement and combined cases are live-verified.
+* **B47 — overlong/invalid names**: All name-taking callbacks share a validator for `NAME_MAX`, reserved components, and UTF-8; managed-live verification passed.
+
+### Implemented in 1.0 (Further Fault Verification Pending)
+
+* **Unified search and resident prompt**: `SearchV2` combines Drive and local lookup with shared relevance scoring and content/source filters. The prompt reuses its window and opens streamable media through FUSE.
+* **Durable write publication**: Scratch sidecars and staged blobs use synced temporary files, atomic renames, and directory synchronization; orphan rescue retains its source on failure.
+* **Transactional pending operations**: Superseding a queued operation cannot delete the old durable record unless the replacement commits.
+* **Safer mirror reconciliation**: Conflict-copy failure aborts publication, incomplete scans are non-destructive, and total-wipe protection includes one-entry baselines.
+* **Bounded IPC**: Control requests are limited to 1 MiB and 10 seconds, with at most 64 active handlers.
+* **Private local state**: Sensitive directories require correct ownership, real-directory identity, and `0700`; sockets use `0600`; config saves are atomic and malformed input is preserved.
+* **Release gates**: Tag/workspace/package versions are checked and formatting, lint, tests, and offline FUSE acceptance run before packaging.
 
 ### Code Fixed / Implemented (Pending Verification)
 * **B8 — no way to complete a CAPTCHA, so a gated sign-in is unrecoverable**: Implemented a Webkit-based bridge to prompt the user to resolve interactive CAPTCHAs during gated logins. (Needs verification on a real gate request).
@@ -100,7 +120,7 @@ Derived from: [docs/bugs.md](file:///home/narl/dev/private/proton-drive-linux/do
 ## 6. Recently Implemented Features
 
 ### Open-for-Write Deferral for Mirror Sync (Implemented)
-**File**: [`sync.rs`](file:///home/narl/dev/private/proton-drive-linux/crates/pdfs-fuse/src/sync.rs)
+**File**: [`sync.rs`](../crates/pdfs-fuse/src/sync.rs)
 
 The mirror folder sync engine now defers uploading any file that is currently held open for writing by another process, matching the guarantee the FUSE mount path already provides (where uploads are deferred until `close(fd)`).
 
@@ -127,4 +147,3 @@ The mirror folder sync engine now defers uploading any file that is currently he
 |---|---|---|
 | FUSE mount | ✅ Perfect | Upload deferred until `close(fd)` |
 | Mirror folders | ✅ Perfect | `/proc/*/fd` scan per reconcile pass |
-

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,
+Normally the live suite applies the same API contract independently to each
-unaligned random writes, rename replacement, cross-directory moves, important
+supplied mount. In convergence mode it exercises the primary once and verifies
-POSIX error cases, open-then-unlink behavior, concurrent writers, directory
+that the resulting bytes appear through every secondary view. A pass is
-enumeration, and optional cross-mount byte-for-byte convergence. 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,150 +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
+    cat >&2 <<'EOF'
-    echo "example: $0 /mnt/on-demand-testmount /mnt/testmount" >&2
+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=("$@")
+case "${1:-}" in
-run_id="pdfs-acceptance-$(date +%Y%m%dT%H%M%S)-$$"
+    -h|--help) usage ;;
-roots=()
+    --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() {
+command -v findmnt >/dev/null || { echo "findmnt is required for --live" >&2; exit 2; }
-    local root
+for mountpoint in "$@"; do
-    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]}"
     [[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; }
-    fs_type="$(findmnt -T "$mountpoint" -n -o FSTYPE | head -n 1)"
+done
-    if (( index == 0 )) && [[ "$fs_type" != fuse* ]]; then
+fstype="$(findmnt -T "$1" -n -o FSTYPE | head -n 1)"
-        echo "refusing primary non-FUSE path $mountpoint (type: ${fs_type:-unknown})" >&2
+[[ "$fstype" == fuse* ]] || {
+    echo "refusing primary non-FUSE path $1 (type: ${fstype:-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
 }
 
-run_suite() {
+exec python3 -u "$script_dir/fuse-acceptance.py" --live "$@"
-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 "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
-        }
-    done
-fi
-
-step "cleanup"
-cleanup
-roots=()
-echo "PASS: FUSE acceptance suite completed ($run_id)"