fixed fs bugs + compleance test for fuse fs

Fix/ioctl handler

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<()> {
@@ -81,6 +81,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 +372,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])?;
-            if !node.trashed {
-                tx.execute(
-                    "INSERT INTO nodes_fts (rowid, name) VALUES (?1, ?2)",
-                    params![rowid, node.name],
+            // A folder rename/move changes every descendant's searchable path.
+            // Refresh the affected subtree, while retaining efficient rowid
+            // deletes for the overwhelmingly common one-file case.
+            let mut affected = vec![rowid];
+            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/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();
@@ -1092,6 +1182,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/control.rs

@@ -241,6 +241,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) => {

crates/pdfs-fuse/src/filesystem.rs

@@ -5,7 +5,13 @@ use super::*;
 impl Filesystem for ProtonFs {
     fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
         let parent = parent.0;
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         // A folder that has not been listed yet is enumerated from the remote,
         // so serve it from a worker rather than stalling the dispatch loop. A
         // listed folder — the common case — is a map hit, and answering it
@@ -307,7 +313,13 @@ impl Filesystem for ProtonFs {
             reply.error(e);
             return;
         }
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let parent_uid = {
             let st = self.core.state.lock();
             match st.entries.get(&parent) {
@@ -761,7 +773,13 @@ impl Filesystem for ProtonFs {
     }
 
     fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
-        let name_str = name.to_string_lossy();
+        let name_str = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
             let st = self.core.state.lock();
             if let Some(entry) = st.entries.get(&ino)
@@ -775,7 +793,13 @@ impl Filesystem for ProtonFs {
     }
 
     fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
-        let name_str = name.to_string_lossy();
+        let name_str = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
             {
                 let st = self.core.state.lock();
@@ -817,7 +841,13 @@ impl Filesystem for ProtonFs {
             reply.error(e);
             return;
         }
-        let name = name.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let parent_uid = {
             let st = self.core.state.lock();
             match st.entries.get(&parent) {
@@ -899,8 +929,20 @@ impl Filesystem for ProtonFs {
     ) {
         let parent = parent.0;
         let newparent = newparent.0;
-        let name = name.to_string_lossy().into_owned();
-        let newname = newname.to_string_lossy().into_owned();
+        let name = match fuse_name(name) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
+        let newname = match fuse_name(newname) {
+            Ok(name) => name,
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
         let (ino, uid) = match self.core.lookup_child(parent, &name) {
             Ok(x) => x,
             Err(e) => {
@@ -1050,12 +1092,31 @@ impl Filesystem for ProtonFs {
         // take the synchronous path, so a genuine API refusal (permissions, a
         // name clash) surfaces to the caller instead of becoming a queued op
         // that can only ever conflict.
-        if !self.core.online.load(Ordering::Relaxed) || is_local_uid(&new_parent_uid) {
+        if rename_needs_queue(
+            self.core.online.load(Ordering::Relaxed),
+            is_local_uid(&new_parent_uid),
+            newparent != parent,
+            newname != name,
+        ) {
             match self
                 .core
                 .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname)
             {
-                Ok(()) => reply.ok(),
+                Ok(()) => {
+                    // Send the rename response first. Kernel-side rename cache
+                    // updates happen while processing that response; notifying
+                    // before it lets those updates overwrite our invalidation
+                    // and retain an empty destination readdir page.
+                    reply.ok();
+                    if let Some(notifier) = self.core.notifier.get() {
+                        let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
+                        let _ = notifier.inval_entry(INodeNo(newparent), OsStr::new(&newname));
+                        let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
+                        if newparent != parent {
+                            let _ = notifier.inval_inode(INodeNo(newparent), 0, 0);
+                        }
+                    }
+                }
                 Err(e) => {
                     self.core.restore_replaced(victim.as_ref(), &newname);
                     reply.error(e);
@@ -1063,54 +1124,58 @@ 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
-            && let Err(e) = self
+        if newname != name {
+            if 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;
+                .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 newname != name {
-            // A move changes the node's encrypted name requirements. Proton's
-            // link-details read can remain briefly stale immediately after the
-            // move, making the following rename fail with InvalidRequirements
-            // ("out of date"). Re-read/retry that narrow eventual-consistency
-            // error; other refusals are stable and must surface immediately.
+        if newparent != parent {
             let mut attempts = 0u32;
-            let renamed = loop {
+            let moved = loop {
                 match self
                     .core
                     .rt
-                    .block_on(self.core.client.rename_node(&uid, &newname, None))
+                    .block_on(self.core.client.move_node(&uid, &new_parent_uid))
                 {
                     Ok(()) => break Ok(()),
                     Err(e)
-                        if newparent != parent
+                        if newname != name
                             && api_code(&e) == Some(ResponseCode::InvalidRequirements)
-                            && attempts < 4 =>
+                            && attempts < 20 =>
                     {
                         attempts += 1;
-                        std::thread::sleep(Duration::from_millis(100 * (1 << attempts)));
+                        // Requirement propagation routinely takes longer than
+                        // three seconds. Keep a bounded ten-second window; a
+                        // successful POSIX reply must mean both remote halves
+                        // landed, not merely that the second half was queued.
+                        std::thread::sleep(Duration::from_millis(500));
                     }
                     Err(e) => break Err(e),
                 }
             };
-            if let Err(e) = renamed {
-                error!(%uid, attempts, error = %e, "rename after move failed");
-                // The move half already landed. Keep our inode model honest
-                // even though the combined POSIX operation has to return EIO.
-                self.core
-                    .state
-                    .lock()
-                    .relocate(ino, parent, newparent, &new_parent_uid, &name);
+            if let Err(e) = moved {
+                error!(%uid, attempts, error = %e, "move after rename failed");
+                // The rename half landed in the source directory.
+                let mut state = self.core.state.lock();
+                let old_parent_uid = state
+                    .entries
+                    .get(&parent)
+                    .map(|entry| entry.uid.clone())
+                    .unwrap_or_else(|| uid.clone());
+                state.rename_in_place(ino, parent, &old_parent_uid, &newname);
+                drop(state);
                 self.core.restore_replaced(victim.as_ref(), &newname);
                 reply.error(Errno::EIO);
                 return;
@@ -1385,6 +1450,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,19 +1337,45 @@ impl Core {
             error!(%uid, error = %e, "create scratch file for truncate failed");
             Errno::EIO
         })?;
+        if let Some(pending) = &pending_base
+            && let Err(e) = copy_pending_for_truncate(pending, &file)
+        {
+            error!(%uid, source = %pending.path.display(), error = %e,
+                "copy pending revision for truncate failed");
+            let _ = std::fs::remove_file(&path);
+            return Err(Errno::EIO);
+        }
         file.set_len(size).map_err(|e| {
             error!(%uid, error = %e, "resize scratch file for truncate failed");
             let _ = std::fs::remove_file(&path);
             Errno::EIO
         })?;
-        let meta = StagedWrite {
-            uid: uid.to_string(),
-            len: size,
-            base_size,
-            base_mtime,
-            authored,
-            complete,
-            based_on: self.remote_baseline(&uid, base_mtime, base_size),
+        let meta = match pending_base {
+            Some(pending) => StagedWrite {
+                uid: uid.to_string(),
+                len: size,
+                base_size: pending.meta.len,
+                base_mtime: pending.meta.base_mtime,
+                authored: if size == 0 {
+                    Vec::new()
+                } else {
+                    vec![(0, size)]
+                },
+                complete: true,
+                // Both local revisions descend from the same last-observed
+                // remote revision. Never replace this with the optimistic
+                // mtime/size currently published by State.
+                based_on: pending.meta.based_on,
+            },
+            None => StagedWrite {
+                uid: uid.to_string(),
+                len: size,
+                base_size,
+                base_mtime,
+                authored,
+                complete,
+                based_on: self.remote_baseline(&uid, base_mtime, base_size),
+            },
         };
         self.enqueue_staged_write(&uid, ino, &path, meta)?;
         debug!(%uid, size, complete, "queued truncate");
@@ -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);
@@ -1524,6 +1575,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 +1832,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 +1868,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 +1890,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 +2025,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)
-            .map_err(|e| CoreError::from_api(&e, "download"))?;
-        self.cache
-            .store(&uid, mtime, size, &bytes)
-            .map_err(|e| CoreError::internal(format!("cache store: {e}")))?;
+
+        // Keep the temporary file beside the final cache blob. `store_file`
+        // can then adopt it with a hard link and atomic rename, while the SDK
+        // streams decrypted blocks to disk instead of building a file-sized
+        // `Vec<u8>`. A unique name also keeps concurrent opens of the same node
+        // 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 +2341,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 +2915,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 +3331,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 +3442,124 @@ mod tests {
         assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]);
     }
 
+    fn pending_revision(dir: &TestDir, bytes: &[u8], complete: bool) -> PendingRevision {
+        let path = dir.0.join("pending");
+        std::fs::write(&path, bytes).unwrap();
+        PendingRevision {
+            path,
+            meta: StagedWrite {
+                uid: "vol~file".into(),
+                len: bytes.len() as u64,
+                base_size: 3,
+                base_mtime: 42,
+                authored: if complete {
+                    vec![(0, bytes.len() as u64)]
+                } else {
+                    vec![(3, bytes.len() as u64)]
+                },
+                complete,
+                based_on: Some(Baseline {
+                    mtime: 42,
+                    size: 3,
+                    hash: None,
+                }),
+            },
+        }
+    }
+
+    #[test]
+    fn complete_pending_revision_can_seed_immediate_shrink() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"0123456789", true);
+        let destination_path = dir.0.join("shrink");
+        let destination = std::fs::File::create(&destination_path).unwrap();
+        copy_pending_for_truncate(&pending, &destination).unwrap();
+        destination.set_len(4).unwrap();
+        assert_eq!(std::fs::read(destination_path).unwrap(), b"0123");
+    }
+
+    #[test]
+    fn complete_pending_revision_can_seed_immediate_growth_and_zero() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"0123456789", true);
+        let destination_path = dir.0.join("grow");
+        let destination = std::fs::File::create(&destination_path).unwrap();
+        copy_pending_for_truncate(&pending, &destination).unwrap();
+        destination.set_len(14).unwrap();
+        assert_eq!(
+            std::fs::read(&destination_path).unwrap(),
+            b"0123456789\0\0\0\0"
+        );
+        destination.set_len(0).unwrap();
+        assert!(std::fs::read(destination_path).unwrap().is_empty());
+    }
+
+    #[test]
+    fn incomplete_pending_revision_is_not_a_safe_truncate_base() {
+        let (_, dir) = state_test_helper();
+        let pending = pending_revision(&dir, b"\0\0\0authored", false);
+        assert!(!pending.meta.complete);
+        assert_ne!(pending.meta.authored, vec![(0, pending.meta.len)]);
+    }
+
+    #[test]
+    fn online_combined_cross_directory_rename_stays_synchronous() {
+        assert!(!rename_needs_queue(true, false, true, true));
+        assert!(rename_needs_queue(false, false, false, true));
+        assert!(rename_needs_queue(true, true, true, false));
+    }
+
+    #[test]
+    fn one_step_online_namespace_changes_stay_synchronous() {
+        assert!(!rename_needs_queue(true, false, false, true));
+        assert!(!rename_needs_queue(true, false, true, false));
+        assert!(!rename_needs_queue(true, false, false, false));
+    }
+
+    #[test]
+    fn fuse_names_enforce_linux_component_limits() {
+        use std::ffi::OsStr;
+
+        assert_eq!(
+            fuse_name(OsStr::new("x".repeat(255).as_str()))
+                .unwrap()
+                .len(),
+            255
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("x".repeat(256).as_str()))
+                .expect_err("256-byte component must be refused")
+                .code(),
+            libc::ENAMETOOLONG
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("."))
+                .expect_err("dot is not an ordinary component")
+                .code(),
+            libc::EINVAL
+        );
+        assert_eq!(
+            fuse_name(OsStr::new("unicodé-文件")).unwrap(),
+            "unicodé-文件"
+        );
+    }
+
+    #[test]
+    fn local_trash_tombstones_hide_stale_remote_entries() {
+        let folder_uid = NodeUid::new(VolumeId::from("vol"), LinkId::from("folder"));
+        let node = node_helper("child", "folder", "child", false);
+        assert!(node_visible(
+            &node,
+            &folder_uid,
+            &std::collections::HashSet::new()
+        ));
+        assert!(!node_visible(
+            &node,
+            &folder_uid,
+            &std::collections::HashSet::from([node.uid.clone()])
+        ));
+    }
+
     use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
     use proton_drive_rs::{Node, NodeKind};
 

crates/pdfs-fuse/src/mount.rs

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

crates/pdfs-fuse/src/state.rs

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

crates/pdfs-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.
-        if is_video_entry(&entry_open) {
+        // media file — `activate_entry` would otherwise stream it.
+        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
-/// Photos page splits on — so the file browser and the gallery agree on what a
-/// video is.
-pub(crate) fn is_video_entry(entry: &DirEntry) -> bool {
-    !entry.is_dir && PhotoKind::classify(Some(&entry.name), None) == PhotoKind::Video
+/// Whether an entry can be opened through FUSE instead of fully materialized.
+pub(crate) fn is_streamable_media_entry(entry: &DirEntry) -> bool {
+    drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia
 }
 
-/// 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) {
-        // A video streams rather than downloads: that is exactly the "play it,
+    } else if drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia {
+        // 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
-//! a short-lived process: it must draw *now*, so nothing here blocks the main
-//! loop. Every daemon round-trip runs on a worker thread and lands back through
-//! an async channel; the two searches (Drive and local) are fired in parallel and
-//! each section renders as its own reply arrives.
+//! Bind it to a system shortcut (e.g. in Hyprland) for a quick HUD search. The
+//! application is single-instance and keeps its window alive between summons,
+//! so repeat activations only reset and present the existing widget tree. Every
+//! daemon round-trip runs on a worker thread and lands back through an async
+//! 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
-    /// once both sections have answered.
+    /// In-flight requests for the current query id.
     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.
-    window.connect_is_active_notify(|w| {
-        if !w.is_active() {
-            w.close();
-        }
+    // Keep the widget tree alive when the compositor asks the window to close.
+    // This also avoids running GTK's destruction path while async replies still
+    // hold references to prompt widgets.
+    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();
-    window.present();
+    Some(ui)
 }
 
 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
-    /// the suggestions view.
+    /// Search Drive and local metadata in one daemon round-trip. An empty query
+    /// 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 drive = spawn_request(
+        let id = self.begin_query(1);
+        let search = 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 })) => {
-                    *ui.drive_hits.borrow_mut() = hits;
+                Ok(Ok(Response::SearchResultsV2 {
+                    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);
         }
-        self.render();
+        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
-    /// restore the cursor to the top. Every state change funnels through here, so
-    /// the list, the placeholder, and the footer can never disagree.
+    /// Rebuild suggestions or the globally-ranked search section, preserving the
+    /// selected identity when a result survives the update.
     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()) {
-                self.open(index, &window);
+            if let Some(index) = self.cursor.get() {
+                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();
-            }
-            // 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();
+                self.dismiss();
             }
             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/BUGS.md

@@ -1587,16 +1587,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 +1611,10 @@ not the prefix of the locally authoritative queued revision.
 
 **Fix:** A write opened over a complete queued revision seeds its scratch file
 from that staged blob and marks the copied range authored. Stacking a write over
-an incomplete queued revision is refused rather than guessed. Live acceptance
-verified the exact rewrite-then-truncate sequence returns `b"a mu"`.
+an incomplete queued revision is refused rather than guessed. Path-based
+truncate now composes the same way: it copies a complete pending blob, applies
+the new length, marks the result complete, and inherits the last real remote
+baseline. Live acceptance verified shrink, zero-filled growth, and sparse I/O.
 
 ---
 
@@ -1624,8 +1633,142 @@ date"), surfaced as `EIO`.
 link-details read can briefly observe the pre-move state, so the new name is
 signed against stale requirements.
 
-**Fix:** Retry only `InvalidRequirements` from the rename-after-move step with a
-short bounded backoff. Stable API failures are not retried. If every retry
-fails, adopt the already-completed move in inode state before returning `EIO`,
-so the local model never claims the node remains in its old parent. The full
-live suite subsequently passed the replacement and combined move/rename cases.
+**Fix history:** A bounded retry of `InvalidRequirements` reduced the window but
+did not close it; the managed matrix reproduced the same failure after all four
+retries. Combined cross-directory move+rename now enters the durable rename
+queue as a desired end state. Its drain re-fetches the node on every attempt,
+skips either half that already landed, renames before moving, and resolves name
+collisions without losing the source. Simple rename-only and move-only calls
+remain synchronous. The managed live suite subsequently passed the replacement
+and combined move/rename cases.
+
+---
+
+## B47 — FUSE Accepts Path Components Longer Than NAME_MAX
+
+**Status:** Fixed (unverified)
+**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.

docs/TESTING.md

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

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
-    echo "example: $0 /mnt/on-demand-testmount /mnt/testmount" >&2
+    cat >&2 <<'EOF'
+usage: scripts/fuse-acceptance.sh [--offline-only]
+       scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...]
+       scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR
+       scripts/fuse-acceptance.sh MOUNTPOINT [MOUNTPOINT ...]  # compatibility
+
+The local reference suite always runs first and requires no account. --live then
+runs the same contract on each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only
+when all live mountpoints show the same remote folder.
+EOF
     exit 2
 }
 
-(( $# > 0 )) || usage
 command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; }
-command -v findmnt >/dev/null || { echo "findmnt is required" >&2; exit 2; }
+script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
 
-mounts=("$@")
-run_id="pdfs-acceptance-$(date +%Y%m%dT%H%M%S)-$$"
-roots=()
+case "${1:-}" in
+    -h|--help) usage ;;
+    --offline-only|"")
+        (( $# <= 1 )) || usage
+        exec python3 -u "$script_dir/fuse-acceptance.py"
+        ;;
+    --live) shift ; (( $# > 0 )) || usage ;;
+    --managed-live)
+        shift
+        (( $# == 2 )) || usage
+        exec python3 -u "$script_dir/fuse-acceptance.py" --managed-live "$@"
+        ;;
+    --*) usage ;;
+esac
 
-cleanup() {
-    local root
-    for root in "${roots[@]}"; do
-        [[ -n "$root" && "$root" == */pdfs-acceptance-* ]] || continue
-        rm -rf -- "$root" 2>/dev/null || true
-    done
-}
-trap cleanup EXIT INT TERM
-
-for index in "${!mounts[@]}"; do
-    mountpoint="${mounts[$index]}"
+command -v findmnt >/dev/null || { echo "findmnt is required for --live" >&2; exit 2; }
+for mountpoint in "$@"; do
     [[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; }
-    fs_type="$(findmnt -T "$mountpoint" -n -o FSTYPE | head -n 1)"
-    if (( index == 0 )) && [[ "$fs_type" != fuse* ]]; then
-        echo "refusing primary non-FUSE path $mountpoint (type: ${fs_type:-unknown})" >&2
-        exit 2
-    fi
-    roots+=("${mountpoint%/}/$run_id")
 done
-
-step() { printf '[test] %s\n' "$1"; }
-assert_text() {
-    python3 - "$1" "$2" <<'PY'
-import sys
-p, expected = sys.argv[1], sys.argv[2].encode()
-with open(p, "rb") as f: got = f.read()
-if got != expected:
-    raise AssertionError(
-        f"{p}: expected {expected!r} ({len(expected)} bytes), "
-        f"got {got!r} ({len(got)} bytes)"
-    )
-PY
+fstype="$(findmnt -T "$1" -n -o FSTYPE | head -n 1)"
+[[ "$fstype" == fuse* ]] || {
+    echo "refusing primary non-FUSE path $1 (type: ${fstype:-unknown})" >&2
+    exit 2
 }
 
-run_suite() {
-local primary="$1"
-printf '[mount] testing %s\n' "${primary%/*}"
-mkdir -- "$primary"
-
-step "create, read, overwrite, truncate"
-printf 'alpha\n' > "$primary/plain.txt"
-assert_text "$primary/plain.txt" $'alpha\n'
-printf 'a much longer replacement\n' > "$primary/plain.txt"
-truncate -s 4 "$primary/plain.txt"
-assert_text "$primary/plain.txt" "a mu"
-
-step "binary round-trip and random-access writes"
-python3 - "$primary/binary.dat" <<'PY'
-import hashlib, sys
-p = sys.argv[1]
-data = bytes(range(256)) * 16384
-with open(p, "wb") as f: f.write(data)
-with open(p, "r+b", buffering=0) as f:
-    f.seek(4093); f.write(b"boundary-write")
-with open(p, "rb") as f: got = f.read()
-expected = data[:4093] + b"boundary-write" + data[4107:]
-assert got == expected
-print(hashlib.sha256(got).hexdigest())
-PY
-
-step "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)"
+exec python3 -u "$script_dir/fuse-acceptance.py" --live "$@"