init, scaffhold + fuse

.gitignore

@@ -0,0 +1,4 @@
+/target
+**/*.rs.bk
+*.pdb
+.env

Cargo.toml

@@ -0,0 +1,33 @@
+[workspace]
+resolver = "2"
+members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli"]
+
+[workspace.package]
+version = "0.1.0"
+edition = "2024"
+license = "MIT"
+authors = ["Nils Pukropp <contact@narl.io>"]
+rust-version = "1.96"
+
+[workspace.dependencies]
+# Local Proton SDK crates (path deps until published; switch to versioned later).
+proton-sdk = { path = "../proton-sdk-rs/crates/proton-sdk" }
+proton-drive-rs = { path = "../proton-sdk-rs/crates/proton-drive-rs" }
+
+# Internal crates.
+pdfs-core = { path = "crates/pdfs-core" }
+pdfs-fuse = { path = "crates/pdfs-fuse" }
+
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "fs", "io-util", "time", "signal"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+thiserror = "2"
+anyhow = "1"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+directories = "5"
+keyring = { version = "3", features = ["sync-secret-service", "crypto-rust"] }
+clap = { version = "4", features = ["derive"] }
+rpassword = "7"
+fuser = "0.17"
+libc = "0.2"

crates/pdfs-cli/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "pdfs-cli"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+rust-version.workspace = true
+description = "CLI for the Proton Drive Linux client"
+
+[[bin]]
+name = "pdfs"
+path = "src/main.rs"
+
+[dependencies]
+pdfs-core.workspace = true
+pdfs-fuse.workspace = true
+tokio.workspace = true
+anyhow.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+clap.workspace = true
+rpassword.workspace = true

crates/pdfs-cli/src/main.rs

@@ -0,0 +1,119 @@
+//! `pdfs` — command-line front-end for the Proton Drive Linux client.
+
+use std::io::Write;
+use std::path::PathBuf;
+
+use anyhow::{Context, Result};
+use clap::{Parser, Subcommand};
+use pdfs_core::auth;
+use pdfs_core::config::AppDirs;
+
+#[derive(Parser)]
+#[command(name = "pdfs", version, about = "Proton Drive for Linux (Files On-Demand)")]
+struct Cli {
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Log in to a Proton account (SRP + optional 2FA) and store the session.
+    Login {
+        /// Account email; prompted if omitted.
+        #[arg(short, long)]
+        username: Option<String>,
+    },
+    /// Forget the stored session.
+    Logout,
+    /// Show the currently logged-in account.
+    Status,
+    /// Mount Proton Drive at the given (or default) path. Blocks until unmounted.
+    Mount {
+        /// Mountpoint; defaults to ~/ProtonDrive.
+        mountpoint: Option<PathBuf>,
+    },
+}
+
+fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "info".into()),
+        )
+        .init();
+
+    let cli = Cli::parse();
+    match cli.command {
+        Command::Login { username } => cmd_login(username),
+        Command::Logout => cmd_logout(),
+        Command::Status => cmd_status(),
+        Command::Mount { mountpoint } => cmd_mount(mountpoint),
+    }
+}
+
+/// Read a line from stdin with a prompt (echoed).
+fn prompt_line(label: &str) -> Result<String> {
+    print!("{label}: ");
+    std::io::stdout().flush()?;
+    let mut line = String::new();
+    std::io::stdin().read_line(&mut line)?;
+    Ok(line.trim().to_owned())
+}
+
+fn cmd_login(username: Option<String>) -> Result<()> {
+    let username = match username {
+        Some(u) => u,
+        None => prompt_line("Proton email")?,
+    };
+    let password = rpassword::prompt_password("Password: ").context("read password")?;
+
+    // The 2FA prompt fires lazily — only if the account actually requires it.
+    let get_totp = || -> pdfs_core::Result<String> {
+        prompt_line("2FA code")
+            .map_err(|e| pdfs_core::Error::Other(format!("read 2FA code: {e}")))
+    };
+
+    let rt = tokio::runtime::Runtime::new()?;
+    rt.block_on(auth::login(&username, &password, get_totp))
+        .context("login failed")?;
+
+    println!("Logged in as {username}. Session stored in the system keyring.");
+    Ok(())
+}
+
+fn cmd_logout() -> Result<()> {
+    auth::logout()?;
+    println!("Stored session cleared.");
+    Ok(())
+}
+
+fn cmd_status() -> Result<()> {
+    match auth::load() {
+        Ok(s) => {
+            println!("Logged in as {} (user {})", s.username, s.user_id);
+            println!("Scopes: {}", s.scopes.join(", "));
+        }
+        Err(pdfs_core::Error::NotLoggedIn) => println!("Not logged in. Run `pdfs login`."),
+        Err(e) => return Err(e.into()),
+    }
+    Ok(())
+}
+
+fn cmd_mount(mountpoint: Option<PathBuf>) -> Result<()> {
+    let dirs = AppDirs::new()?;
+    dirs.ensure()?;
+    let mountpoint = mountpoint.unwrap_or_else(|| dirs.default_mountpoint());
+    std::fs::create_dir_all(&mountpoint)
+        .with_context(|| format!("create mountpoint {}", mountpoint.display()))?;
+
+    // Multi-threaded runtime: its worker threads keep servicing async SDK calls
+    // while the main thread is parked inside the blocking FUSE session loop.
+    let rt = tokio::runtime::Runtime::new()?;
+    let (client, _session) = rt
+        .block_on(auth::resume_client())
+        .context("resume session (run `pdfs login` first)")?;
+
+    let handle = rt.handle().clone();
+    pdfs_fuse::mount(client, handle, &mountpoint).context("mount failed")?;
+    Ok(())
+}

crates/pdfs-core/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "pdfs-core"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+rust-version.workspace = true
+description = "Core auth/session/config for the Proton Drive Linux client"
+
+[dependencies]
+proton-sdk.workspace = true
+proton-drive-rs.workspace = true
+tokio.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
+directories.workspace = true
+keyring.workspace = true

crates/pdfs-core/src/auth.rs

@@ -0,0 +1,132 @@
+//! Login, session persistence, and Drive-client construction.
+//!
+//! A successful login is persisted to the OS keyring as a single JSON blob
+//! holding the resumable session tokens plus the mailbox password (needed to
+//! rebuild the key chain on resume). The daemon resumes from this blob with no
+//! interactive step; the refresh token auto-renews via the HTTP client's 401
+//! path, so no fresh 2FA is required until the refresh token itself expires.
+
+use keyring::Entry;
+use proton_drive_rs::ProtonDriveClient;
+use proton_sdk::config::ProtonClientConfiguration;
+use proton_sdk::session::{PasswordMode, ProtonApiSession, ResumeParameters};
+use serde::{Deserialize, Serialize};
+
+use crate::config::{APP_VERSION, KEYRING_SERVICE, USER_AGENT};
+use crate::error::{Error, Result};
+
+/// Fixed keyring account name for the single stored session blob.
+const KEYRING_USER: &str = "session";
+
+/// Everything needed to resume a session unattended, persisted to the keyring.
+#[derive(Serialize, Deserialize, Clone)]
+pub struct StoredSession {
+    pub session_id: String,
+    pub username: String,
+    pub user_id: String,
+    pub access_token: String,
+    pub refresh_token: String,
+    pub scopes: Vec<String>,
+    /// `1` = single password, `2` = dual (Proton wire value).
+    pub password_mode: u8,
+    /// Mailbox (data) password — required by `ProtonDriveClient::new` to derive
+    /// the key chain. Lives only in the OS keyring, never on disk in cleartext.
+    pub mailbox_password: String,
+}
+
+impl StoredSession {
+    fn into_params(&self) -> ResumeParameters {
+        ResumeParameters {
+            session_id: self.session_id.clone().into(),
+            username: self.username.clone(),
+            user_id: self.user_id.clone().into(),
+            access_token: self.access_token.clone(),
+            refresh_token: self.refresh_token.clone(),
+            scopes: self.scopes.clone(),
+            is_waiting_for_second_factor_code: false,
+            password_mode: match self.password_mode {
+                1 => PasswordMode::Single,
+                _ => PasswordMode::Dual,
+            },
+        }
+    }
+}
+
+fn client_config() -> ProtonClientConfiguration {
+    ProtonClientConfiguration::new(APP_VERSION).with_user_agent(USER_AGENT)
+}
+
+fn keyring_entry() -> Result<Entry> {
+    Ok(Entry::new(KEYRING_SERVICE, KEYRING_USER)?)
+}
+
+/// Run an interactive SRP + (optional) 2FA login and persist the session.
+///
+/// `get_totp` is only invoked when the account requires a second factor, so
+/// callers can defer prompting until it is actually needed.
+pub async fn login(
+    username: &str,
+    password: &str,
+    get_totp: impl FnOnce() -> Result<String>,
+) -> Result<()> {
+    let mut session =
+        ProtonApiSession::begin(client_config(), username, password.as_bytes()).await?;
+
+    if session.is_waiting_for_second_factor() {
+        let code = get_totp()?;
+        session.apply_second_factor_code(code.trim()).await?;
+    }
+
+    save(&session, password).await?;
+    Ok(())
+}
+
+/// Persist the session's current tokens + mailbox password to the keyring.
+pub async fn save(session: &ProtonApiSession, mailbox_password: &str) -> Result<()> {
+    let tokens = session.current_tokens().await;
+    let stored = StoredSession {
+        session_id: session.session_id().as_str().to_owned(),
+        username: session.username().to_owned(),
+        user_id: session.user_id().as_str().to_owned(),
+        access_token: tokens.access_token,
+        refresh_token: tokens.refresh_token,
+        scopes: session.scopes().to_vec(),
+        password_mode: match session.password_mode() {
+            PasswordMode::Single => 1,
+            PasswordMode::Dual => 2,
+        },
+        mailbox_password: mailbox_password.to_owned(),
+    };
+    let json = serde_json::to_string(&stored)?;
+    keyring_entry()?.set_password(&json)?;
+    Ok(())
+}
+
+/// Load the persisted session blob, or `Error::NotLoggedIn` if absent.
+pub fn load() -> Result<StoredSession> {
+    let entry = keyring_entry()?;
+    match entry.get_password() {
+        Ok(json) => Ok(serde_json::from_str(&json)?),
+        Err(keyring::Error::NoEntry) => Err(Error::NotLoggedIn),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Forget the persisted session (best-effort; absent entry is not an error).
+pub fn logout() -> Result<()> {
+    match keyring_entry()?.delete_credential() {
+        Ok(()) | Err(keyring::Error::NoEntry) => Ok(()),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Resume a persisted session and build an authenticated Drive client.
+///
+/// Re-persists tokens afterward in case the refresh path rotated them. Returns
+/// `Error::NotLoggedIn` when no session has been saved.
+pub async fn resume_client() -> Result<(ProtonDriveClient, ProtonApiSession)> {
+    let stored = load()?;
+    let session = ProtonApiSession::resume(client_config(), stored.into_params())?;
+    let client = ProtonDriveClient::new(&session, stored.mailbox_password.clone().into_bytes());
+    Ok((client, session))
+}

crates/pdfs-core/src/config.rs

@@ -0,0 +1,56 @@
+//! Static client identity and per-user filesystem paths.
+
+use std::path::PathBuf;
+
+use directories::ProjectDirs;
+
+use crate::error::{Error, Result};
+
+/// Proton requires `external-drive-{name}@{semver}-{channel}` (channel ∈
+/// stable/beta/alpha); a malformed value trips the 422 anti-abuse path.
+pub const APP_VERSION: &str = "external-drive-linux@0.1.0-alpha";
+pub const USER_AGENT: &str = "proton-drive-linux/0.1.0";
+
+/// Keyring service name; one entry per credential kind keyed by username.
+pub const KEYRING_SERVICE: &str = "proton-drive-linux";
+
+/// Resolved XDG locations for state, cache, and the default mountpoint.
+pub struct AppDirs {
+    dirs: ProjectDirs,
+}
+
+impl AppDirs {
+    pub fn new() -> Result<Self> {
+        let dirs = ProjectDirs::from("io", "narl", "proton-drive-linux")
+            .ok_or_else(|| Error::Other("cannot resolve home directory".into()))?;
+        Ok(Self { dirs })
+    }
+
+    /// Persistent state: inode map / cache index DB lives here.
+    pub fn state_dir(&self) -> PathBuf {
+        // `state_dir` is Linux-only in `directories`; fall back to data dir.
+        self.dirs
+            .state_dir()
+            .unwrap_or_else(|| self.dirs.data_dir())
+            .to_path_buf()
+    }
+
+    /// Hydrated file-content cache.
+    pub fn cache_dir(&self) -> PathBuf {
+        self.dirs.cache_dir().to_path_buf()
+    }
+
+    /// Default mountpoint when the user does not pass one explicitly.
+    pub fn default_mountpoint(&self) -> PathBuf {
+        directories::UserDirs::new()
+            .map(|u| u.home_dir().join("ProtonDrive"))
+            .unwrap_or_else(|| PathBuf::from("/tmp/ProtonDrive"))
+    }
+
+    /// Create state + cache dirs if missing.
+    pub fn ensure(&self) -> Result<()> {
+        std::fs::create_dir_all(self.state_dir())?;
+        std::fs::create_dir_all(self.cache_dir())?;
+        Ok(())
+    }
+}

crates/pdfs-core/src/error.rs

@@ -0,0 +1,27 @@
+//! Crate-wide error type.
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("proton api: {0}")]
+    Proton(#[from] proton_sdk::ProtonError),
+
+    #[error("keyring: {0}")]
+    Keyring(#[from] keyring::Error),
+
+    #[error("serialization: {0}")]
+    Serde(#[from] serde_json::Error),
+
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+
+    /// No persisted session found — caller must run an interactive login.
+    #[error("not logged in (no saved session)")]
+    NotLoggedIn,
+
+    #[error("{0}")]
+    Other(String),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

crates/pdfs-core/src/lib.rs

@@ -0,0 +1,14 @@
+//! Core building blocks for the Proton Drive Linux client: static client
+//! identity, XDG paths, and the login/session/keyring layer that every
+//! front-end (CLI today, daemon + GUI later) shares.
+
+pub mod auth;
+pub mod config;
+pub mod error;
+
+pub use error::{Error, Result};
+
+// Re-export the SDK surface downstream crates need, so they depend on
+// `pdfs-core` rather than pinning the SDK directly.
+pub use proton_drive_rs::{self, Node, NodeKind, ProtonDriveClient};
+pub use proton_sdk;

crates/pdfs-fuse/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "pdfs-fuse"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+rust-version.workspace = true
+description = "FUSE on-demand filesystem for Proton Drive"
+
+[dependencies]
+pdfs-core.workspace = true
+proton-drive-rs.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+fuser.workspace = true
+libc.workspace = true

crates/pdfs-fuse/src/lib.rs

@@ -0,0 +1,503 @@
+//! On-demand FUSE filesystem for Proton Drive.
+//!
+//! Phase 1 is a read-only placeholder mount: directories are enumerated lazily
+//! from the remote on first access and file content is hydrated on `read` via
+//! [`ProtonDriveClient::download_range`]. Nothing is mirrored to local disk;
+//! every byte is fetched on demand.
+//!
+//! Phase 2 adds live remote sync: a background task polls the volume event
+//! cursor ([`ProtonDriveClient::enumerate_events`]) and pushes invalidations
+//! into the kernel via a fuser [`Notifier`], so the cache TTL can be long while
+//! remote changes still show up promptly.
+
+use std::collections::HashMap;
+use std::ffi::OsStr;
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use fuser::{
+    Config, Errno, FileAttr, FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo,
+    LockOwner, MountOption, Notifier, OpenFlags, ReplyAttr, ReplyData, ReplyDirectory, ReplyEntry,
+    ReplyOpen, Request, Session,
+};
+use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid};
+use proton_drive_rs::{DriveEvent, DriveEventScopeId, Node, NodeKind, ProtonDriveClient};
+use tracing::{debug, error, info, warn};
+
+/// Attribute/entry cache lifetime handed back to the kernel. Long because the
+/// Phase 2 event poller actively invalidates changed inodes; without a remote
+/// change this is how long the kernel may serve stale metadata.
+const TTL: Duration = Duration::from_secs(30);
+
+/// How often the background task polls the remote event cursor.
+const POLL_INTERVAL: Duration = Duration::from_secs(10);
+
+/// The FUSE root inode is always 1.
+const ROOT_INO: u64 = 1;
+
+/// A node known to the filesystem, addressed by its kernel inode.
+struct Entry {
+    uid: NodeUid,
+    parent: u64,
+    node: Node,
+}
+
+/// Mutable inode bookkeeping, guarded by a mutex because fuser drives the
+/// `Filesystem` trait through `&self`.
+struct State {
+    /// inode -> node metadata.
+    entries: HashMap<u64, Entry>,
+    /// Dedupe inodes by node uid so a node keeps a stable inode across lookups.
+    by_uid: HashMap<NodeUid, u64>,
+    /// Cached directory listings: parent inode -> child inodes. Presence of a
+    /// key means the directory has been enumerated.
+    children: HashMap<u64, Vec<u64>>,
+    next_ino: u64,
+}
+
+impl State {
+    /// Allocate (or reuse) a stable inode for a node and store its metadata.
+    fn intern(&mut self, parent: u64, node: Node) -> u64 {
+        if let Some(&ino) = self.by_uid.get(&node.uid) {
+            if let Some(e) = self.entries.get_mut(&ino) {
+                e.node = node;
+                e.parent = parent;
+            }
+            return ino;
+        }
+        let ino = self.next_ino;
+        self.next_ino += 1;
+        self.by_uid.insert(node.uid.clone(), ino);
+        self.entries
+            .insert(ino, Entry { uid: node.uid.clone(), parent, node });
+        ino
+    }
+
+    /// Forget a node entirely: drop its inode, its uid mapping, its own cached
+    /// listing, and its slot in its parent's listing. Returns `(parent_ino,
+    /// name)` when the node was known, so the caller can notify the kernel.
+    fn forget(&mut self, uid: &NodeUid) -> Option<(u64, String)> {
+        let ino = self.by_uid.remove(uid)?;
+        let entry = self.entries.remove(&ino)?;
+        self.children.remove(&ino);
+        if let Some(kids) = self.children.get_mut(&entry.parent) {
+            kids.retain(|&k| k != ino);
+        }
+        Some((entry.parent, entry.node.name))
+    }
+}
+
+/// The Proton Drive VFS. FUSE callbacks are synchronous, so the Tokio handle
+/// bridges each one to the async SDK via [`Handle::block_on`]; the fuser
+/// session thread is not a runtime worker, so blocking on it is sound.
+pub struct ProtonFs {
+    client: ProtonDriveClient,
+    rt: tokio::runtime::Handle,
+    state: Arc<Mutex<State>>,
+    uid: u32,
+    gid: u32,
+}
+
+impl ProtonFs {
+    /// Build the filesystem rooted at `root` (the user's My Files folder).
+    pub fn new(client: ProtonDriveClient, rt: tokio::runtime::Handle, root: Node) -> Self {
+        let mut entries = HashMap::new();
+        let mut by_uid = HashMap::new();
+        by_uid.insert(root.uid.clone(), ROOT_INO);
+        entries.insert(
+            ROOT_INO,
+            Entry { uid: root.uid.clone(), parent: ROOT_INO, node: root },
+        );
+        // SAFETY: geteuid/getegid are infallible and have no preconditions.
+        let (uid, gid) = unsafe { (libc::geteuid(), libc::getegid()) };
+        Self {
+            client,
+            rt,
+            state: Arc::new(Mutex::new(State {
+                entries,
+                by_uid,
+                children: HashMap::new(),
+                next_ino: 2,
+            })),
+            uid,
+            gid,
+        }
+    }
+
+    /// A handle to the shared inode state, for the event poller.
+    fn shared_state(&self) -> Arc<Mutex<State>> {
+        Arc::clone(&self.state)
+    }
+
+    /// Enumerate `ino`'s children from the remote and cache them. No-op if the
+    /// directory has already been listed. Network I/O happens without the lock
+    /// held so concurrent metadata reads aren't blocked behind a fetch.
+    fn ensure_children(&self, ino: u64) -> Result<(), Errno> {
+        let folder_uid = {
+            let st = self.state.lock().unwrap();
+            if st.children.contains_key(&ino) {
+                return Ok(());
+            }
+            match st.entries.get(&ino) {
+                Some(e) => e.uid.clone(),
+                None => return Err(Errno::ENOENT),
+            }
+        };
+
+        let uids = self
+            .rt
+            .block_on(self.client.enumerate_folder_children_node_uids(&folder_uid))
+            .map_err(|e| {
+                error!(%folder_uid, error = %e, "enumerate folder children failed");
+                Errno::EIO
+            })?;
+        let nodes = self
+            .rt
+            .block_on(self.client.enumerate_nodes(&uids))
+            .map_err(|e| {
+                error!(%folder_uid, error = %e, "enumerate nodes failed");
+                Errno::EIO
+            })?;
+
+        let mut st = self.state.lock().unwrap();
+        // Lost the race? Another thread already populated it.
+        if st.children.contains_key(&ino) {
+            return Ok(());
+        }
+        let mut child_inos = Vec::with_capacity(nodes.len());
+        for node in nodes {
+            if node.trashed {
+                continue;
+            }
+            child_inos.push(st.intern(ino, node));
+        }
+        st.children.insert(ino, child_inos);
+        Ok(())
+    }
+
+    fn attr(&self, ino: u64, node: &Node) -> FileAttr {
+        let (kind, perm) = match node.kind {
+            NodeKind::Folder => (FileType::Directory, 0o555),
+            NodeKind::File { .. } => (FileType::RegularFile, 0o444),
+        };
+        let size = node_size(node);
+        let mtime = unix_secs(node.modification_time);
+        let crtime = unix_secs(node.creation_time);
+        FileAttr {
+            ino: INodeNo(ino),
+            size,
+            blocks: size.div_ceil(512),
+            atime: mtime,
+            mtime,
+            ctime: mtime,
+            crtime,
+            kind,
+            perm,
+            nlink: if kind == FileType::Directory { 2 } else { 1 },
+            uid: self.uid,
+            gid: self.gid,
+            rdev: 0,
+            blksize: 512,
+            flags: 0,
+        }
+    }
+}
+
+/// The plaintext size, in bytes, that a node reports.
+fn node_size(node: &Node) -> u64 {
+    match &node.kind {
+        NodeKind::Folder => 0,
+        NodeKind::File { claimed_size, total_size_on_storage, .. } => {
+            claimed_size.unwrap_or(*total_size_on_storage).max(0) as u64
+        }
+    }
+}
+
+/// Convert a Unix timestamp (seconds, possibly negative) to a `SystemTime`.
+fn unix_secs(secs: i64) -> SystemTime {
+    if secs >= 0 {
+        UNIX_EPOCH + Duration::from_secs(secs as u64)
+    } else {
+        UNIX_EPOCH - Duration::from_secs(secs.unsigned_abs())
+    }
+}
+
+impl Filesystem for ProtonFs {
+    fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
+        let parent = parent.0;
+        if let Err(e) = self.ensure_children(parent) {
+            reply.error(e);
+            return;
+        }
+        let name = name.to_string_lossy();
+        let st = self.state.lock().unwrap();
+        let hit = st.children.get(&parent).and_then(|kids| {
+            kids.iter().copied().find_map(|ino| {
+                st.entries
+                    .get(&ino)
+                    .filter(|e| e.node.name == name)
+                    .map(|e| (ino, &e.node))
+            })
+        });
+        match hit {
+            Some((ino, node)) => {
+                let attr = self.attr(ino, node);
+                reply.entry(&TTL, &attr, Generation(0));
+            }
+            None => reply.error(Errno::ENOENT),
+        }
+    }
+
+    fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
+        let st = self.state.lock().unwrap();
+        match st.entries.get(&ino.0) {
+            Some(e) => {
+                let attr = self.attr(ino.0, &e.node);
+                reply.attr(&TTL, &attr);
+            }
+            None => reply.error(Errno::ENOENT),
+        }
+    }
+
+    fn readdir(
+        &self,
+        _req: &Request,
+        ino: INodeNo,
+        _fh: FileHandle,
+        offset: u64,
+        mut reply: ReplyDirectory,
+    ) {
+        let ino = ino.0;
+        if let Err(e) = self.ensure_children(ino) {
+            reply.error(e);
+            return;
+        }
+        let st = self.state.lock().unwrap();
+        let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent);
+
+        // "." and ".." occupy offsets 0 and 1; real children follow.
+        let mut listing: Vec<(u64, FileType, String)> = vec![
+            (ino, FileType::Directory, ".".to_string()),
+            (parent, FileType::Directory, "..".to_string()),
+        ];
+        if let Some(kids) = st.children.get(&ino) {
+            for &kid in kids {
+                if let Some(e) = st.entries.get(&kid) {
+                    let ft = if e.node.is_folder() {
+                        FileType::Directory
+                    } else {
+                        FileType::RegularFile
+                    };
+                    listing.push((kid, ft, e.node.name.clone()));
+                }
+            }
+        }
+        drop(st);
+
+        for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) {
+            // The stored offset is that of the *next* entry to resume at.
+            if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) {
+                break;
+            }
+        }
+        reply.ok();
+    }
+
+    fn open(&self, _req: &Request, ino: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
+        let st = self.state.lock().unwrap();
+        match st.entries.get(&ino.0) {
+            Some(e) if e.node.is_file() => reply.opened(FileHandle(0), FopenFlags::empty()),
+            Some(_) => reply.error(Errno::EISDIR),
+            None => reply.error(Errno::ENOENT),
+        }
+    }
+
+    fn read(
+        &self,
+        _req: &Request,
+        ino: INodeNo,
+        _fh: FileHandle,
+        offset: u64,
+        size: u32,
+        _flags: OpenFlags,
+        _lock_owner: Option<LockOwner>,
+        reply: ReplyData,
+    ) {
+        let uid = {
+            let st = self.state.lock().unwrap();
+            match st.entries.get(&ino.0) {
+                Some(e) if e.node.is_file() => e.uid.clone(),
+                Some(_) => {
+                    reply.error(Errno::EISDIR);
+                    return;
+                }
+                None => {
+                    reply.error(Errno::ENOENT);
+                    return;
+                }
+            }
+        };
+        match self
+            .rt
+            .block_on(self.client.download_range(&uid, offset, size as u64))
+        {
+            Ok(bytes) => reply.data(&bytes),
+            Err(e) => {
+                warn!(%uid, offset, size, error = %e, "download_range failed");
+                reply.error(Errno::EIO);
+            }
+        }
+    }
+}
+
+/// Apply one remote event to the local cache and notify the kernel so it drops
+/// any stale cached metadata/data for the affected inodes.
+///
+/// The cache is authoritative-by-absence: dropping a directory's `children`
+/// entry forces the next `lookup`/`readdir` to re-enumerate from the remote, so
+/// most events only need to invalidate listings rather than re-fetch eagerly.
+fn apply_event(state: &Mutex<State>, notifier: &Notifier, event: &DriveEvent) {
+    match event {
+        DriveEvent::NodeUpdated { node_uid, parent_node_uid, is_trashed, .. } => {
+            let mut st = state.lock().unwrap();
+            if *is_trashed {
+                // Trashing makes a node vanish from its parent listing.
+                let child = st.by_uid.get(node_uid).copied();
+                if let Some((parent, name)) = st.forget(node_uid) {
+                    match child {
+                        Some(child) => {
+                            let _ =
+                                notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name));
+                        }
+                        None => {
+                            let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
+                        }
+                    }
+                }
+            } else if let Some(&ino) = st.by_uid.get(node_uid) {
+                // Known node changed: drop its cached attrs/data (and listing if
+                // it is a directory) so the next access re-fetches.
+                st.children.remove(&ino);
+                let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
+            }
+            // A create (or move-in) shows up as a change to the parent listing;
+            // drop it so the new child is picked up on the next readdir.
+            if let Some(parent_uid) = parent_node_uid
+                && let Some(&parent) = st.by_uid.get(parent_uid)
+                && st.children.remove(&parent).is_some()
+            {
+                let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
+            }
+        }
+        DriveEvent::NodeDeleted { node_uid, .. } => {
+            let mut st = state.lock().unwrap();
+            // Capture the inode before `forget` clears the uid mapping.
+            let child = st.by_uid.get(node_uid).copied();
+            if let Some((parent, name)) = st.forget(node_uid) {
+                match child {
+                    Some(child) => {
+                        let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name));
+                    }
+                    None => {
+                        let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
+                    }
+                }
+            }
+        }
+        // Continuity or scope was lost: our cached listings may be arbitrarily
+        // stale, so drop every listing and tell the kernel to forget all
+        // metadata. Inodes stay stable; dirs simply re-enumerate on next access.
+        DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => {
+            warn!("event continuity lost; dropping all cached listings, resyncing lazily");
+            let mut st = state.lock().unwrap();
+            let dirs: Vec<u64> = st.children.keys().copied().collect();
+            st.children.clear();
+            for ino in dirs {
+                let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
+            }
+        }
+        // No substantive local change; the cursor advance is handled by the
+        // caller persisting the event id.
+        DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {}
+    }
+}
+
+/// Poll the remote event cursor forever, applying each batch to the shared
+/// state. Seeds the cursor from the current server head so we only react to
+/// changes made after mount. Runs as a Tokio task; returns only on fatal error.
+async fn run_event_sync(
+    client: ProtonDriveClient,
+    scope: DriveEventScopeId,
+    state: Arc<Mutex<State>>,
+    notifier: Notifier,
+) {
+    // Seed: a `None` cursor yields a single `CursorAdvanced` at the server head.
+    let mut cursor: Option<DriveEventId> = match client.enumerate_events(&scope, None).await {
+        Ok(events) => events.last().map(|e| e.id().clone()),
+        Err(e) => {
+            error!(error = %e, "failed to seed event cursor; live sync disabled");
+            return;
+        }
+    };
+    info!(?cursor, "event sync started");
+
+    loop {
+        tokio::time::sleep(POLL_INTERVAL).await;
+        let events = match client.enumerate_events(&scope, cursor.as_ref()).await {
+            Ok(events) => events,
+            Err(e) => {
+                warn!(error = %e, "event poll failed; retrying after interval");
+                continue;
+            }
+        };
+        if events.is_empty() {
+            continue;
+        }
+        debug!(count = events.len(), "applying remote events");
+        for event in &events {
+            apply_event(&state, &notifier, event);
+        }
+        cursor = events.last().map(|e| e.id().clone());
+    }
+}
+
+/// Mount the filesystem at `mountpoint` and block until it is unmounted.
+///
+/// Fetches the My Files root up front (so an auth/network failure surfaces
+/// before the kernel mount), then spawns the Phase 2 event-sync task and takes
+/// over the calling thread in the FUSE session loop. `rt` must be a handle to a
+/// *running* multi-threaded runtime.
+pub fn mount(
+    client: ProtonDriveClient,
+    rt: tokio::runtime::Handle,
+    mountpoint: &Path,
+) -> std::io::Result<()> {
+    let root = rt
+        .block_on(client.get_my_files_folder())
+        .map_err(|e| std::io::Error::other(format!("fetch My Files root: {e}")))?;
+    let scope = root.tree_event_scope_id();
+
+    let mut config = Config::default();
+    config.mount_options = vec![
+        MountOption::FSName("protondrive".to_string()),
+        MountOption::Subtype("protondrive".to_string()),
+        MountOption::RO,
+        MountOption::NoExec,
+        MountOption::DefaultPermissions,
+    ];
+    info!(mountpoint = %mountpoint.display(), "mounting Proton Drive");
+
+    let fs = ProtonFs::new(client.clone(), rt.clone(), root);
+    let shared = fs.shared_state();
+
+    // Build the session explicitly (not `mount2`) so we can grab a `Notifier`
+    // for the event task. `spawn` runs the session loop on its own thread;
+    // `join` then blocks here until the filesystem is unmounted, preserving
+    // `mount`'s blocking contract.
+    let session = Session::new(fs, mountpoint, &config)?.spawn()?;
+    let notifier = session.notifier();
+    rt.spawn(run_event_sync(client, scope, shared, notifier));
+
+    session.join()
+}