diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs
index 9dddf90..7becb94 100644
--- a/crates/pdfs-cli/src/main.rs
+++ b/crates/pdfs-cli/src/main.rs
@@ -30,7 +30,7 @@ struct Cli {
     /// Emit machine-readable JSON instead of formatted text.
     ///
     /// Applies to the query commands (`status`, `ls`, `pins`, `sync list`,
-    /// `devices list`, `locations`, `transfers`, `activity`, `cache inspect`). Commands that
+    /// `devices list`, `locations`, `transfers`, `activity`, `trash`, `cache inspect`). Commands that
     /// perform an action keep their human output — a script that needs to know
     /// whether one succeeded has the exit code.
     #[arg(long, global = true)]
@@ -2551,7 +2551,11 @@ fn cmd_diagnostics() -> Result<()> {
 }
 
 fn cmd_trash() -> Result<()> {
-    match control_request(CtlRequest::ListTrash)? {
+    let response = control_request(CtlRequest::ListTrash)?;
+    if emit_json(&response)? {
+        return Ok(());
+    }
+    match response {
         CtlResponse::Entries { entries } if entries.is_empty() => println!("(trash is empty)"),
         CtlResponse::Entries { entries } => {
             for e in entries {
diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs
index 6f85027..05f94f5 100644
--- a/crates/pdfs-core/src/db/ops.rs
+++ b/crates/pdfs-core/src/db/ops.rs
@@ -66,6 +66,52 @@ pub fn op_supersedes(kind: &str) -> bool {
     matches!(kind, OP_REVISION | OP_RENAME | OP_TRASH)
 }
 
+/// Delete the ops that removing `uid` takes with it, returning the staged blobs
+/// they held.
+///
+/// That is every op for `uid` or for anything queued beneath it, plus the
+/// revisions of files the node tree places beneath it. A revision op carries
+/// no `parent_uid`, so the queue alone cannot tell that its file sits inside
+/// a removed folder: the upload outlived the folder, failed forever against a
+/// trashed node, and kept its blob on disk (`docs/BUGS.md` B102).
+///
+/// Only revisions are taken from the node tree. A queued rename that moved a
+/// node *into* the folder has to land, or the node would stay where it was
+/// on the server instead of going to the trash with the folder.
+fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<String>> {
+    const DOOMED: &str = "
+        WITH RECURSIVE
+          queued(uid) AS (
+            SELECT ?1
+            UNION
+            SELECT p.uid FROM pending_op p JOIN queued q ON p.parent_uid = q.uid
+          ),
+          below(uid) AS (
+            SELECT ?1
+            UNION
+            SELECT n.uid FROM nodes n JOIN below b ON n.parent_uid = b.uid
+          ),
+          doomed(id) AS (
+            SELECT id FROM pending_op
+            WHERE uid IN (SELECT uid FROM queued)
+               OR (kind = 'revision' AND uid IN (SELECT uid FROM below))
+          )";
+    let blobs: Vec<String> = {
+        let mut stmt = tx.prepare(&format!(
+            "{DOOMED}
+             SELECT blob_path FROM pending_op
+             WHERE id IN (SELECT id FROM doomed) AND blob_path IS NOT NULL"
+        ))?;
+        let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
+        rows.collect::<std::result::Result<Vec<_>, _>>()?
+    };
+    tx.execute(
+        &format!("{DOOMED} DELETE FROM pending_op WHERE id IN (SELECT id FROM doomed)"),
+        params![uid],
+    )?;
+    Ok(blobs)
+}
+
 /// The volume id given to a node that exists only on this machine, so far. A
 /// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<uuid>` and
 /// round-trips through the same `Display`/parse path as any other uid.
@@ -252,27 +298,9 @@ impl Db {
         name: &str,
         created_at: i64,
     ) -> Result<(i64, Vec<String>)> {
-        const SUBTREE: &str = "
-            WITH RECURSIVE doomed(uid) AS (
-              SELECT ?1
-              UNION
-              SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid
-            )";
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
-        let blobs: Vec<String> = {
-            let mut stmt = tx.prepare(&format!(
-                "{SUBTREE}
-                 SELECT blob_path FROM pending_op
-                 WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL"
-            ))?;
-            let rows = stmt.query_map(params![uid], |row| row.get::<_, String>(0))?;
-            rows.collect::<std::result::Result<Vec<_>, _>>()?
-        };
-        tx.execute(
-            &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"),
-            params![uid],
-        )?;
+        let blobs = drop_doomed_ops(&tx, uid)?;
         tx.execute(
             "INSERT INTO pending_op
                (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at)
@@ -326,6 +354,67 @@ impl Db {
         Ok(Some(AttachedBlob { id, superseded }))
     }
 
+    /// Retire a queued create that has just landed as `real`, and repoint
+    /// everything that named its placeholder, in one transaction.
+    ///
+    /// The drain uploads the blob the create held when it was picked up. A
+    /// write that closes while that upload is on the wire attaches a newer blob
+    /// to the *same row* (`attach_blob_to_create`), so deleting the row by id
+    /// threw that write away: the server kept the older bytes and nothing was
+    /// left queued to replace them. When the row no longer holds `uploaded`, it
+    /// becomes a revision of `real` instead, with its sidecar passed through
+    /// `rewrite` so it names the real node. Returns that blob and sidecar, or
+    /// `None` when the create is simply done.
+    pub fn finish_create(
+        &self,
+        id: i64,
+        uploaded: Option<&str>,
+        local: &str,
+        real: &str,
+        rewrite: impl FnOnce(&str) -> Option<String>,
+    ) -> Result<Option<(String, String)>> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let row: Option<(Option<String>, Option<String>)> = tx
+            .query_row(
+                "SELECT blob_path, meta_json FROM pending_op WHERE id = ?1",
+                params![id],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .optional()?;
+        let newer = match row {
+            Some((Some(blob), Some(meta))) if uploaded != Some(blob.as_str()) => {
+                rewrite(&meta).map(|meta| (blob, meta))
+            }
+            _ => None,
+        };
+        match &newer {
+            Some((_, meta)) => {
+                tx.execute(
+                    "UPDATE pending_op
+                     SET kind = ?2, uid = ?3, parent_uid = NULL, name = NULL, meta_json = ?4,
+                         attempts = 0, last_error = NULL, next_attempt_at = 0
+                     WHERE id = ?1",
+                    params![id, OP_REVISION, real, meta],
+                )?;
+            }
+            None => {
+                tx.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?;
+            }
+        }
+        tx.execute(
+            "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
+            params![local, real],
+        )?;
+        tx.execute(
+            "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
+            params![local, real],
+        )?;
+        tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?;
+        tx.commit()?;
+        Ok(newer)
+    }
+
     /// Replace the sidecar of a queued op, for a baseline that has moved under
     /// it.
     ///
@@ -428,29 +517,12 @@ impl Db {
     /// and in the user's pending count, for the life of the database.
     ///
     /// Only `create`/`mkdir` ops carry a `parent_uid`, so for a file this
-    /// recursion finds nothing and costs one query.
+    /// recursion finds nothing and costs one query. Revisions of files the
+    /// node tree places beneath `uid` go too; see [`drop_doomed_ops`].
     pub fn delete_ops_for_uid(&self, uid: &str) -> Result<Vec<String>> {
-        const SUBTREE: &str = "
-            WITH RECURSIVE doomed(uid) AS (
-              SELECT ?1
-              UNION
-              SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid
-            )";
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
-        let blobs: Vec<String> = {
-            let mut stmt = tx.prepare(&format!(
-                "{SUBTREE}
-                 SELECT blob_path FROM pending_op
-                 WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL"
-            ))?;
-            let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
-            rows.collect::<std::result::Result<Vec<_>, _>>()?
-        };
-        tx.execute(
-            &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"),
-            params![uid],
-        )?;
+        let blobs = drop_doomed_ops(&tx, uid)?;
         tx.commit()?;
         Ok(blobs)
     }
diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs
index fa7b454..f86b995 100644
--- a/crates/pdfs-core/src/db/tests.rs
+++ b/crates/pdfs-core/src/db/tests.rs
@@ -2498,6 +2498,71 @@ fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() {
     );
 }
 
+/// A revision op names no parent, so the queue alone cannot see that its file
+/// sits inside a folder being removed. The node tree can: trashing the folder
+/// has to take the uploads of the files below it along, or they fail forever
+/// against a trashed node and keep their bytes on disk (B102). A rename that
+/// moved a node into the folder still has to land.
+#[test]
+fn trashing_a_folder_drops_the_queued_revisions_of_files_below_it() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_nodes(&[
+        folder("root", None, "My Files"),
+        folder("doomed", Some("root"), "doomed"),
+        folder("sub", Some("doomed"), "sub"),
+        file("deep", "sub", "deep.bin", 1),
+        file("moved", "doomed", "moved.txt", 1),
+        file("kept", "root", "kept.bin", 1),
+    ])
+    .unwrap();
+    let op = |kind: &str, link: &str, blob: Option<&str>| PendingOp {
+        id: 0,
+        kind: kind.to_string(),
+        uid: uid(link).to_string(),
+        parent_uid: None,
+        name: None,
+        blob_path: blob.map(str::to_string),
+        meta_json: Some("{}".to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    db.enqueue_op(&op(OP_REVISION, "deep", Some("/staging/deep")))
+        .unwrap();
+    db.enqueue_op(&op(OP_RENAME, "moved", None)).unwrap();
+    db.enqueue_op(&op(OP_REVISION, "kept", Some("/staging/kept")))
+        .unwrap();
+
+    let doomed = uid("doomed").to_string();
+    let (_, blobs) = db.replace_ops_with_trash(&doomed, "doomed", 2).unwrap();
+    assert_eq!(blobs, vec!["/staging/deep"]);
+
+    let mut left: Vec<(String, String)> = db
+        .pending_ops()
+        .unwrap()
+        .into_iter()
+        .map(|op| (op.kind, op.uid))
+        .collect();
+    left.sort();
+    assert_eq!(
+        left,
+        vec![
+            (OP_RENAME.to_string(), uid("moved").to_string()),
+            (OP_REVISION.to_string(), uid("kept").to_string()),
+            (OP_TRASH.to_string(), doomed.clone()),
+        ]
+    );
+
+    // Deleting outright follows the same tree.
+    db.enqueue_op(&op(OP_REVISION, "deep", Some("/staging/deep2")))
+        .unwrap();
+    assert_eq!(
+        db.delete_ops_for_uid(&uid("sub").to_string()).unwrap(),
+        vec!["/staging/deep2"]
+    );
+}
+
 /// Deleting a folder that was created offline must take the ops queued
 /// underneath it with it: they name a placeholder parent that will now never
 /// become real, so nothing could ever drain them and nothing is left to
@@ -2699,6 +2764,99 @@ fn attaching_to_an_already_drained_create_reports_it_is_gone() {
     assert!(out.is_none());
 }
 
+#[test]
+fn a_write_attached_while_its_create_uploads_becomes_a_revision() {
+    let db = Db::open_in_memory().unwrap();
+    let local = "local~abc";
+    let real = uid("real").to_string();
+    let create = |blob: Option<&str>| PendingOp {
+        id: 0,
+        kind: OP_CREATE.to_string(),
+        uid: local.to_string(),
+        parent_uid: Some(uid("parent").to_string()),
+        name: Some("new.txt".to_string()),
+        blob_path: blob.map(str::to_string),
+        meta_json: blob.map(|_| "{\"uid\":\"local~abc\"}".to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    let (id, _) = db.enqueue_op(&create(None)).unwrap();
+    // The drain picked the create up with no blob; the file's first write
+    // closes while the empty upload is on the wire.
+    db.attach_blob_to_create(local, "/staging/written", "{\"uid\":\"local~abc\"}")
+        .unwrap()
+        .expect("create is still queued");
+
+    let newer = db
+        .finish_create(id, None, local, &real, |json| {
+            Some(json.replace("local~abc", "vol~real"))
+        })
+        .unwrap();
+
+    assert_eq!(
+        newer,
+        Some((
+            "/staging/written".to_string(),
+            "{\"uid\":\"vol~real\"}".to_string()
+        ))
+    );
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1, "the write is still queued");
+    assert_eq!(ops[0].kind, OP_REVISION);
+    assert_eq!(ops[0].uid, real);
+    assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/written"));
+    assert_eq!(ops[0].parent_uid, None);
+}
+
+#[test]
+fn a_create_whose_blob_landed_is_retired() {
+    let db = Db::open_in_memory().unwrap();
+    let local = "local~abc";
+    let real = uid("real").to_string();
+    let (id, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_CREATE.to_string(),
+            uid: local.to_string(),
+            parent_uid: Some(uid("parent").to_string()),
+            name: Some("new.txt".to_string()),
+            blob_path: Some("/staging/sent".to_string()),
+            meta_json: Some("{}".to_string()),
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+    db.enqueue_op(&PendingOp {
+        id: 0,
+        kind: OP_CREATE.to_string(),
+        uid: "local~child".to_string(),
+        parent_uid: Some(local.to_string()),
+        name: Some("inside.txt".to_string()),
+        blob_path: None,
+        meta_json: None,
+        created_at: 2,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    })
+    .unwrap();
+
+    let newer = db
+        .finish_create(id, Some("/staging/sent"), local, &real, |_| {
+            panic!("an unchanged create has no sidecar to rewrite")
+        })
+        .unwrap();
+
+    assert_eq!(newer, None);
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1, "only the child is left");
+    assert_eq!(ops[0].parent_uid.as_deref(), Some(real.as_str()));
+}
+
 #[test]
 fn draining_a_folder_repoints_its_queued_children() {
     let db = Db::open_in_memory().unwrap();
diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs
index d06fc71..655a1db 100644
--- a/crates/pdfs-fuse/src/drain.rs
+++ b/crates/pdfs-fuse/src/drain.rs
@@ -41,7 +41,7 @@ use proton_drive_rs::proton_sdk::ids::NodeUid;
 use proton_drive_rs::{Node, NodeMoveItem};
 use tracing::{debug, error, info, warn};
 
-use super::state::Intervals;
+use super::state::{Intervals, PendingRevision};
 use super::transfers::CountingReader;
 use super::{
     Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, DRAIN_REVISION_DEBOUNCE,
@@ -886,15 +886,20 @@ impl Core {
         // Retire the op before touching anything else: if we crash here the node
         // exists remotely and the local placeholder is reconciled by the event
         // sync, whereas a surviving op would create the file a second time.
-        self.db.delete_op(op.id)?;
+        let newer = self.retire_create(op, &local, &real)?;
         self.adopt_real_uid(&local, &real)?;
         // The feed will report this create back to us; the tree already has it
         // under its real uid, so that event is ours to ignore (`Core::self_changes`).
         self.note_self_change(&real);
+        // The uploaded blob is done with either way: a write that replaced it
+        // while it was on the wire already discarded it when it attached.
         if let Some(blob) = op.blob_path.as_deref() {
             self.cache.discard_staged(Path::new(blob));
         }
-        self.pending.lock().remove(&local);
+        if newer {
+            info!(%local, %real, name, "a write landed during the create; queued it as a revision");
+            self.wake_drain();
+        }
         self.log_activity(ActivityKind::Upload, &name, "created", true);
         info!(%local, %real, name, kind = %op.kind, "pending create landed");
         Ok(())
@@ -963,6 +968,65 @@ impl Core {
         Ok(uid)
     }
 
+    /// Retire a landed create's op and move its queued state to the real uid.
+    /// True when a newer write attached to the create while it was uploading,
+    /// which is now queued as a revision of `real`.
+    ///
+    /// Runs under the `pending` lock, which `Core::enqueue_staged_write` also
+    /// holds from attaching a blob to a create until it records the pending
+    /// entry. So a write either attached before this ran and is carried over,
+    /// or finds the create gone and queues against the real uid itself. The
+    /// open write handles are repointed right after, for the same reason:
+    /// a handle released later must not queue against the placeholder.
+    fn retire_create(
+        &self,
+        op: &PendingOp,
+        local: &NodeUid,
+        real: &NodeUid,
+    ) -> Result<bool, Box<dyn std::error::Error>> {
+        let real_str = real.to_string();
+        let newer = {
+            let mut pending = self.pending.lock();
+            let newer = self.db.finish_create(
+                op.id,
+                op.blob_path.as_deref(),
+                &local.to_string(),
+                &real_str,
+                |json| {
+                    let mut meta: StagedWrite = serde_json::from_str(json).ok()?;
+                    meta.uid = real_str.clone();
+                    serde_json::to_string(&meta).ok()
+                },
+            )?;
+            pending.remove(local);
+            if let Some((blob, json)) = &newer {
+                let meta: StagedWrite = serde_json::from_str(json)?;
+                pending.insert(
+                    real.clone(),
+                    PendingRevision {
+                        path: PathBuf::from(blob),
+                        meta,
+                    },
+                );
+            }
+            newer.is_some()
+        };
+        self.for_each_state(|st| {
+            if let Some(ino) = st.by_uid.remove(local) {
+                st.by_uid.insert(real.clone(), ino);
+                if let Some(e) = st.entries.get_mut(&ino) {
+                    e.uid = real.clone();
+                }
+            }
+            for aw in st.active_writes.values_mut() {
+                if aw.uid == *local {
+                    aw.uid = real.clone();
+                }
+            }
+        });
+        Ok(newer)
+    }
+
     /// Swap a placeholder uid for the real one across everything that keyed off
     /// it: queued children, the DB, the in-memory tree, and the caches.
     ///
@@ -985,14 +1049,17 @@ impl Core {
             .fetch_node(real)
             .map_err(|e| self.errno_error(e, "fetch node"))?;
         // Repoints queued children and node rows, and drops the placeholder row.
+        // A no-op after `retire_create`, which has already done both.
         self.db
             .remap_local_uid(&local.to_string(), &real.to_string())?;
 
         self.for_each_state(|st| {
-            let Some(ino) = st.by_uid.remove(local) else {
+            if let Some(ino) = st.by_uid.remove(local) {
+                st.by_uid.insert(real.clone(), ino);
+            }
+            let Some(&ino) = st.by_uid.get(real) else {
                 return;
             };
-            st.by_uid.insert(real.clone(), ino);
             if let Some(e) = st.entries.get_mut(&ino) {
                 e.uid = real.clone();
                 e.node = node.clone();
@@ -1024,6 +1091,9 @@ impl Core {
         if let Err(e) = self.db.upsert_node(&node) {
             warn!(%real, error = %e, "db upsert_node failed after remap");
         }
+        // A write carried over from the create was made against no revision at
+        // all; the one just created is what it now replaces.
+        self.rebaseline_pending(real, &node);
         if node.is_folder() {
             // It was recorded as listed while local (it was empty and had nothing
             // to enumerate). That still holds: its queued children re-intern under
diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs
index a912f82..bbabe15 100644
--- a/crates/pdfs-fuse/src/filesystem.rs
+++ b/crates/pdfs-fuse/src/filesystem.rs
@@ -442,10 +442,7 @@ impl Filesystem for ProtonFs {
                 reply.error(Errno::EBADF);
                 return;
             };
-            let end = offset + data.len() as u64;
-            aw.written.add(offset, end);
-            aw.len = aw.len.max(end);
-            aw.dirty = true;
+            aw.record_write(offset, offset + data.len() as u64);
             let len = aw.len;
             st.set_size(ino.0, len);
             len
diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs
index 69d939a..b01a139 100644
--- a/crates/pdfs-fuse/src/lib.rs
+++ b/crates/pdfs-fuse/src/lib.rs
@@ -168,6 +168,10 @@ const DRAIN_BACKOFF_MAX: Duration = Duration::from_secs(300);
 /// directly on a new write or a reconnect, so this only bounds how late a
 /// backoff can fire.
 const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30);
+/// How long a write whose create just landed waits for the drain to repoint
+/// its inode (`Core::landed_uid`): 200 polls of 10 ms, two seconds in all.
+const LANDED_UID_POLLS: u32 = 200;
+const LANDED_UID_POLL: Duration = Duration::from_millis(10);
 /// Grace period before a queued revision becomes eligible for draining.
 ///
 /// Tools like aria2c preallocate a file (truncate to target size) and then write
@@ -261,6 +265,15 @@ const TRASH_MATERIALIZE_CHUNK: usize = 150;
 /// outlives that timeout is a hang from the user's side: the client gives up,
 /// the user asks again, and the daemon accumulates another refresh.
 const TRASH_FIRST_WAIT: Duration = Duration::from_secs(20);
+/// How long one server call of a trash refresh may take before the refresh
+/// gives up.
+///
+/// Refreshes are single-flight, so a call that never returns wedges the trash
+/// listing for the life of the daemon: every later request joins the stuck
+/// refresh, times out, and answers with the listing from before it. That hid
+/// a just-trashed folder from a permanent delete for as long as the daemon ran
+/// (`docs/BUGS.md` B103). One call is one request, so this is generous.
+const TRASH_CALL_TIMEOUT: Duration = Duration::from_secs(120);
 
 /// `sync_state` keys for the freshness stamps of the two persisted listings, and
 /// for whether the account has a photos volume at all (so an account without one
@@ -3098,35 +3111,57 @@ impl Core {
                 Errno::EIO
             }
         })?;
-        let meta_json = serde_json::to_string(&meta).unwrap_or_default();
         // Whatever this write supersedes, it supersedes now — including an
         // upload already on the wire, whose bytes describe a revision the row
         // written just below replaces. Signalled before the enqueue so the
         // in-flight reader stops at its next block rather than after however
         // much of the file is left.
         self.cancel_upload(uid);
-        let superseded = if is_local_uid(uid) {
-            // The node has no server-side identity to hang a revision on, so the
-            // bytes ride on the create that will mint it.
-            let attached = self
-                .db
-                .attach_blob_to_create(&uid.to_string(), &path.to_string_lossy(), &meta_json)
-                .map_err(|e| {
-                    error!(%uid, error = %e, "attaching write to queued create failed");
-                    Errno::EIO
-                })?;
-            match attached {
-                Some(a) => a.superseded,
-                None => {
-                    // The create drained between `release` and here, so the node
-                    // has a real uid now and this handle's is stale. The bytes are
-                    // safe in staging, but nothing here can address them.
+        let mut uid = uid.clone();
+        let len = meta.len;
+        let superseded = 'queued: {
+            if is_local_uid(&uid) {
+                // The node has no server-side identity to hang a revision on, so
+                // the bytes ride on the create that will mint it. Attached and
+                // recorded under one hold of `pending`, which is what the drain
+                // takes to retire a landed create: it either sees this blob and
+                // carries it over, or has already finished and this write
+                // queues against the real uid below (`Core::retire_create`).
+                let meta_json = serde_json::to_string(&meta).unwrap_or_default();
+                let mut pending = self.pending.lock();
+                let attached = self
+                    .db
+                    .attach_blob_to_create(&uid.to_string(), &path.to_string_lossy(), &meta_json)
+                    .map_err(|e| {
+                        error!(%uid, error = %e, "attaching write to queued create failed");
+                        Errno::EIO
+                    })?;
+                if let Some(attached) = attached {
+                    pending.insert(uid.clone(), PendingRevision { path, meta });
+                    break 'queued attached.superseded;
+                }
+                drop(pending);
+                // The create landed while this handle was open. Its uid is the
+                // one the drain has just given the inode.
+                let Some(real) = self.landed_uid(ino) else {
                     error!(%uid, staged = %path.display(),
                            "queued create vanished under a write; bytes kept in staging");
                     return Err(Errno::EIO);
+                };
+                debug!(local = %uid, %real, "create landed under an open write; queueing a revision");
+                uid = real;
+                meta.uid = uid.to_string();
+                let previous = self.pending.lock().get(&uid).cloned();
+                meta.based_on = previous.as_ref().and_then(|p| p.meta.based_on.clone());
+                if !meta.complete
+                    && let Some(previous) = previous
+                    && let Err(error) = merge_over_pending(&mut meta, &path, &previous)
+                {
+                    error!(%uid, %error, staged = %path.display(),
+                           "folding in the write the create carried failed; bytes kept in staging");
+                    return Err(Errno::EIO);
                 }
             }
-        } else {
             let op = PendingOp {
                 id: 0,
                 kind: OP_REVISION.to_string(),
@@ -3134,25 +3169,25 @@ impl Core {
                 parent_uid: None,
                 name: None,
                 blob_path: Some(path.to_string_lossy().into_owned()),
-                meta_json: Some(meta_json),
+                meta_json: Some(serde_json::to_string(&meta).unwrap_or_default()),
                 created_at: now_millis(),
                 attempts: 0,
                 last_error: None,
-                next_attempt_at: now_millis() + self.revision_debounce(uid).as_millis() as i64,
+                next_attempt_at: now_millis() + self.revision_debounce(&uid).as_millis() as i64,
             };
             let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| {
                 error!(%uid, error = %e, "queueing upload failed");
                 Errno::EIO
             })?;
+            self.pending
+                .lock()
+                .insert(uid.clone(), PendingRevision { path, meta });
             superseded
         };
         if let Some(old) = superseded {
             self.cache.discard_staged(Path::new(&old));
         }
-        let len = meta.len;
-        self.pending
-            .lock()
-            .insert(uid.clone(), PendingRevision { path, meta });
+        let uid = &uid;
         // Reflect the write in the tree straight away: `ls` must show the new
         // size and mtime even though the remote still holds the old revision.
         let now = now_secs();
@@ -3187,6 +3222,23 @@ impl Core {
         Ok(())
     }
 
+    /// The real uid the drain gave `ino` when its queued create landed.
+    ///
+    /// Called by a write that found the create already retired. The drain
+    /// repoints the inode right after it drops `pending`, with no network in
+    /// between, so a short wait covers the gap; `None` means the inode is gone
+    /// or never stopped being a placeholder.
+    fn landed_uid(&self, ino: u64) -> Option<NodeUid> {
+        for _ in 0..LANDED_UID_POLLS {
+            let uid = self.state().entries.get(&ino).map(|e| e.uid.clone())?;
+            if !is_local_uid(&uid) {
+                return Some(uid);
+            }
+            std::thread::sleep(LANDED_UID_POLL);
+        }
+        None
+    }
+
     /// Queue the new content of a path-based truncate — `> file`, or any
     /// `setattr(size=…)` arriving without a write handle.
     ///
@@ -3457,10 +3509,7 @@ impl Core {
                 error!(%uid, error = %e, "queueing trash failed");
                 Errno::EIO
             })?;
-        for blob in blobs {
-            self.cache.discard_staged(Path::new(&blob));
-        }
-        self.pending.lock().remove(uid);
+        self.release_dropped_ops(uid, blobs);
         self.hidden.lock().insert(uid.clone());
         // Withdrawn from every mount, not just the one the unlink came through:
         // a sync folder maps a remote folder that also exists under My Files, so
@@ -3486,11 +3535,36 @@ impl Core {
             error!(%uid, error = %e, "dropping queued ops failed");
             Errno::EIO
         })?;
+        self.release_dropped_ops(uid, blobs);
+        Ok(())
+    }
+
+    /// Let go of what the ops just dropped for `uid` held: their pending
+    /// entries, any upload still reading their bytes, and the bytes.
+    ///
+    /// Removing a folder also drops the queued revisions of the files below it
+    /// (`docs/BUGS.md` B102), so the entries to forget are found by blob, not
+    /// only by `uid`.
+    fn release_dropped_ops(&self, uid: &NodeUid, blobs: Vec<String>) {
+        let below: Vec<NodeUid> = {
+            let mut pending = self.pending.lock();
+            pending.remove(uid);
+            let below: Vec<NodeUid> = pending
+                .iter()
+                .filter(|(_, p)| blobs.iter().any(|b| p.path == Path::new(b)))
+                .map(|(k, _)| k.clone())
+                .collect();
+            for k in &below {
+                pending.remove(k);
+            }
+            below
+        };
+        for k in &below {
+            self.cancel_upload(k);
+        }
         for blob in blobs {
             self.cache.discard_staged(Path::new(&blob));
         }
-        self.pending.lock().remove(uid);
-        Ok(())
     }
 
     /// Which staged blob is currently queued for `uid`, if any.
@@ -4317,6 +4391,11 @@ impl Core {
             .block_on(self.client.trash_nodes(std::slice::from_ref(&uid)))
             .and_then(batch::into_unit)
             .map_err(|e| CoreError::from_api(&e, "trash"))?;
+        // Uploads queued for the node or anything below it would now only fail
+        // against the trash, forever, holding their bytes (`docs/BUGS.md` B102).
+        if let Err(error) = self.discard_queued_ops(&uid) {
+            warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed");
+        }
         let name = self
             .state
             .lock()
@@ -4483,10 +4562,8 @@ impl Core {
     /// end. Every batch is a usable listing.
     async fn refresh_trash(&self) -> CoreResult<()> {
         let started = Instant::now();
-        let uids = self
-            .client
-            .enumerate_trash_node_uids()
-            .await
+        let uids = within_trash_call("enumerate trash", self.client.enumerate_trash_node_uids())
+            .await?
             .map_err(|e| CoreError::from_api(&e, "enumerate trash"))?;
         info!(
             count = uids.len(),
@@ -4512,10 +4589,8 @@ impl Core {
         let mut items: Vec<StoredTrash> = Vec::with_capacity(uids.len());
         for chunk in uids.chunks(TRASH_MATERIALIZE_CHUNK) {
             let chunk_started = Instant::now();
-            let nodes = self
-                .client
-                .enumerate_nodes(chunk)
-                .await
+            let nodes = within_trash_call("enumerate nodes", self.client.enumerate_nodes(chunk))
+                .await?
                 .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
             items.extend(nodes.into_iter().map(|node| StoredTrash {
                 uid: node.uid.to_string(),
@@ -5556,6 +5631,19 @@ fn node_size(node: &Node) -> u64 {
     }
 }
 
+/// `call`, failed with a remote error once it has run for
+/// [`TRASH_CALL_TIMEOUT`].
+async fn within_trash_call<T>(what: &str, call: impl Future<Output = T>) -> CoreResult<T> {
+    tokio::time::timeout(TRASH_CALL_TIMEOUT, call)
+        .await
+        .map_err(|_| {
+            CoreError::new(
+                ErrorKind::Remote,
+                format!("{what}: no answer within {}s", TRASH_CALL_TIMEOUT.as_secs()),
+            )
+        })
+}
+
 /// When a node a trash refresh found went to the trash, as far as the daemon
 /// can tell. A node the stored listing already had keeps the time it had. A new
 /// one went there since the last refresh, so now is close, but only when that
@@ -7848,7 +7936,7 @@ mod tests {
             assert_before(function, "require_uid_writable", first_side_effect);
         }
         let trash = function_source(source, "fn queue_trash(");
-        assert_before(trash, "replace_ops_with_trash", "discard_staged");
+        assert_before(trash, "replace_ops_with_trash", "release_dropped_ops");
 
         // §5, parallel drain: every path that unlinks a staged blob or hands
         // the same node's work to another worker has to stop the upload that
diff --git a/crates/pdfs-fuse/src/state.rs b/crates/pdfs-fuse/src/state.rs
index 101961f..747fe71 100644
--- a/crates/pdfs-fuse/src/state.rs
+++ b/crates/pdfs-fuse/src/state.rs
@@ -128,6 +128,23 @@ pub(crate) struct WriteHandle {
     pub(crate) open_count: usize,
 }
 
+impl WriteHandle {
+    /// Account for `[offset, end)` having been written through this handle.
+    ///
+    /// A write past the end leaves a hole that reads as zeros, exactly like a
+    /// grow through `setattr`, so the hole is authored too. Left unauthored, a
+    /// hole below the base size was refilled from the base at commit: a file
+    /// shrunk and then written past its new end got its old bytes back.
+    pub(crate) fn record_write(&mut self, offset: u64, end: u64) {
+        if offset > self.len {
+            self.written.add(self.len, offset);
+        }
+        self.written.add(offset, end);
+        self.len = self.len.max(end);
+        self.dirty = true;
+    }
+}
+
 /// A released write whose upload has not happened yet (offline.md Phase 3).
 ///
 /// The bytes live in the content cache's staging dir and the intent lives in the
@@ -1863,4 +1880,50 @@ mod tests {
         let shared = st.intern(owned_root, shared);
         assert_eq!(st.entries[&shared].access, Access::Viewer);
     }
+
+    fn handle(len: u64, base_size: u64) -> WriteHandle {
+        let path = std::env::temp_dir().join(format!("pdfs-handle-test-{}", std::process::id()));
+        let file = File::create(&path).unwrap();
+        let _ = std::fs::remove_file(&path);
+        WriteHandle {
+            ino: 2,
+            uid: uid("file"),
+            file: Arc::new(file),
+            path,
+            written: Intervals::default(),
+            len,
+            base_size,
+            base_mtime: 0,
+            base_revision_id: None,
+            dirty: false,
+            open_count: 1,
+        }
+    }
+
+    #[test]
+    fn a_write_past_the_end_authors_the_hole_it_leaves() {
+        // Opened over a 100-byte base, shrunk to 10, then written at 50: bytes
+        // 10..50 are zeros now, not the base's bytes 10..50.
+        let mut h = handle(100, 100);
+        h.written.clip(10);
+        h.len = 10;
+        h.record_write(50, 60);
+        assert_eq!(h.len, 60);
+        assert_eq!(
+            h.written.segments(0, 60),
+            vec![(0, 10, false), (10, 60, true)],
+            "only the head still comes from the base"
+        );
+    }
+
+    #[test]
+    fn a_write_inside_the_file_leaves_the_rest_to_the_base() {
+        let mut h = handle(100, 100);
+        h.record_write(40, 50);
+        assert_eq!(h.len, 100);
+        assert_eq!(
+            h.written.segments(0, 100),
+            vec![(0, 40, false), (40, 50, true), (50, 100, false)]
+        );
+    }
 }
diff --git a/docs/BUGS.md b/docs/BUGS.md
index ff130e1..bd70ba9 100644
--- a/docs/BUGS.md
+++ b/docs/BUGS.md
@@ -12,6 +12,105 @@ Conventions:
 
 ---
 
+## B103 — One stuck trash refresh freezes the trash listing until the daemon restarts
+
+**Status:** Fixed (unverified). The hang itself has not been reproduced on purpose.
+**Found:** 2026-09-29, while timing the acceptance suite's wide-directory case. Its cleanup trashed
+the test folder, then waited for it to show up in the trash so it could delete it permanently. It
+never did. From 06:55 UTC on, every trash listing logged `trash refresh still running; answering
+with what has materialized` and returned the same 362 entries. No `trash refresh: enumerated` line
+followed, so the first server call of the refresh never returned.
+
+**Where:** `crates/pdfs-fuse/src/lib.rs` (`Core::refresh_trash`, `within_trash_call`).
+
+**Cause.** Trash refreshes are single-flight: a request joins the refresh already running instead
+of starting another. The refresh awaited its server calls with no deadline, so one call that
+never returned kept the flag set forever. Every later request joined it, waited out its 20 s
+budget, and answered with the stale listing.
+
+**Fix.** Each server call of a refresh (`enumerate_trash_node_uids`, and each `enumerate_nodes`
+chunk) fails after `TRASH_CALL_TIMEOUT` (120 s). The refresh then ends with an error, clears the
+flag, and the next request starts a fresh one.
+
+---
+
+## B102 — Removing a folder leaves the queued uploads of the files inside it
+
+**Status:** Fixed (unverified). No daemon carrying the change has run the acceptance suite yet.
+**Found:** 2026-09-29, after the account run that found B101. Its cleanup removed both test roots
+and reported "nothing left behind", but `pdfs sync queue` still listed two `revision` ops for
+`pdfs-acceptance-…/boundaries/reshaped`, each holding about 8 MiB of staged bytes. They retried
+against the trash forever ("the share has not been writable for 300s").
+
+**Where:** `crates/pdfs-core/src/db/ops.rs` (`drop_doomed_ops`, used by
+`Db::replace_ops_with_trash` and `Db::delete_ops_for_uid`), `crates/pdfs-fuse/src/lib.rs`
+(`Core::delete`, `Core::release_dropped_ops`).
+
+**Cause.** Two gaps.
+
+- **The queue could not see which files a folder holds.** Removing a node dropped the ops for it
+  and for anything *queued* beneath it, found through `pending_op.parent_uid`. Only `create` and
+  `mkdir` ops carry a parent, so a `revision` of a file that already existed remotely was never
+  found, however deep in the removed folder it sat.
+- **`pdfs rm` dropped nothing at all.** The control socket's delete trashed the node remotely but
+  left every op queued for it, unlike an unlink through the mount.
+
+**Fix.** Removal also follows the node tree (`nodes.parent_uid`) and drops the `revision` ops of
+every file below the node, with their blobs, pending entries and in-flight uploads. Queued renames
+below it are kept: one that moved a node into the folder still has to land for the node to go to
+the trash with it. `pdfs rm` now drops the queued ops once the remote trash succeeds. The
+acceptance janitor lists the daemon queue after cleanup and reports any op still naming one of its
+roots as a leftover.
+Regression test: `trashing_a_folder_drops_the_queued_revisions_of_files_below_it`.
+
+---
+
+## B101 — A shrunk file written past its end gets old bytes back, or fails with `EIO`; a write during a create's upload is lost
+
+**Status:** Fixed (unverified). No daemon carrying the change has run the acceptance suite yet.
+**Found:** 2026-09-29, by the new acceptance case "block boundaries and overwrites that change
+the block count". It failed with `Input/output error` in My files and in an on-demand folder, every
+time. The journal had `the revision's block table disagrees with the node size ... fsize=8388611
+reader_size=0`, then `pending upload failed; will retry ... gap-fill from base failed: Errno(5)`
+on every retry.
+
+**Where:** `crates/pdfs-fuse/src/state.rs` (`WriteHandle::record_write`),
+`crates/pdfs-fuse/src/drain.rs` (`Core::drain_local_node`, `Core::retire_create`),
+`crates/pdfs-fuse/src/lib.rs` (`Core::enqueue_staged_write`),
+`crates/pdfs-core/src/db/ops.rs` (`Db::finish_create`).
+
+**Cause.** Two defects.
+
+- **The hole of a write past the end was not authored.** A write handle tracks which ranges it
+  wrote; everything else below the base size is filled from the base when it is committed. A
+  grow through `setattr` marks its new tail as written, but a `pwrite` past the end did not mark
+  the hole it leaves. After `ftruncate(fd, 4 MiB)` and `pwrite(fd, data, 4 MiB + 10)`, bytes
+  4 MiB to 4 MiB + 10 were refilled from the old base instead of reading as zeros. That silently
+  restored truncated bytes. When the base was itself a queued revision the remote did not have
+  yet, the fill read a revision of the wrong size, and B84's check failed it: `EIO` on read, and
+  a drain that retried forever.
+- **A write that closed while its file's create uploaded was lost.** The create of a new file
+  drains on its own, often while the file is still open. A write that closes meanwhile attaches
+  its blob to the create's queue row. When the create landed, the drain deleted that row by id,
+  so the server kept the older bytes (often an empty file) and nothing was queued to replace
+  them. A handle still open after the create landed carried the placeholder uid, found no create
+  to attach to, and failed its `close` with `EIO`.
+
+**Fix.** A write past the end authors the hole (`WriteHandle::record_write`).
+`Db::finish_create` retires a landed create in one transaction: if the row now holds a newer blob
+than the one uploaded, it becomes a revision of the real uid instead of being deleted. The drain
+does this under the `pending` lock, which `enqueue_staged_write` holds from attaching a blob until
+it records the pending entry, and then repoints open write handles to the real uid. A write that
+still finds the create gone waits for its inode's real uid (`Core::landed_uid`) and queues a
+revision against it.
+Regression tests: `a_write_past_the_end_authors_the_hole_it_leaves`,
+`a_write_attached_while_its_create_uploads_becomes_a_revision`,
+`a_create_whose_blob_landed_is_retired`, and the live acceptance cases "block boundaries and
+overwrites that change the block count" (now also read back after the queue drains) and
+`regression B101`.
+
+---
+
 ## B100 — A freshly copied file reads as `EIO` or 53 bytes too big
 
 **Status:** Fixed (unverified). No daemon carrying the change has read a file with a provisional
@@ -48,7 +147,7 @@ network). A folder past the cap is now queued instead of dropped. When a running
 its thread takes over the oldest queued folder, so at most 8 threads still run.
 Regression tests: `a_folder_past_the_cap_is_queued_not_dropped`,
 `a_finished_batch_hands_its_thread_the_oldest_queued_one`. The read and open paths need a live
-mount and have no unit test.
+mount and have no unit test; the acceptance case `regression B100` covers them on a real mount.
 
 ---
 
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 3d30b17..08cd8ce 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -11,7 +11,29 @@ scratch (user data in `staging/` and `recovery/` is never touched by this).
 
 ## [Unreleased]
 
+### Added
+- **`pdfs trash --json`** prints the trash listing as JSON.
+- **The FUSE acceptance suite tests the signed-in account by default.**
+  `scripts/fuse-acceptance.sh` with no arguments creates test folders in My files and new
+  on-demand and mirror sync folders, runs every case in each, and removes all of it afterwards,
+  also on a failure or interrupt. A run that dies is cleaned up by the next one. New cases cover
+  block boundaries, unusual names, a deep tree, a wide directory, rename patterns, handle coherency,
+  throughput and B100. `--offline-only` runs the account-free suite as before.
+
 ### Fixed
+- **Truncating and then writing past the end no longer brings old bytes back.** The gap
+  between the new end and the write now reads as zeros, as it should, instead of the file's
+  earlier content. When that earlier content had not been uploaded yet, the file failed with
+  `Input/output error` and its upload retried forever (B101).
+- **A write made while a new file is first uploading is no longer lost.** Drive could keep an
+  empty or older copy of a file written right after it was created, and closing a file held open
+  across its first upload could fail (B101).
+- **Removing a folder also cancels the pending uploads of the files inside it.** Before, an
+  edit that had not reached Drive yet kept retrying against the trash forever and kept its
+  staged copy on disk. `pdfs rm` now also drops what was queued for what it removes (B102).
+- **The trash no longer freezes after a server call hangs.** Until the daemon restarted, the
+  trash kept showing the same entries, and a folder just deleted could not be found there to
+  delete permanently (B103).
 - **Freshly listed files read correctly.** A file whose real size had not been fetched yet
   reported its encrypted size, about 53 bytes too big, and reading it failed with
   `Input/output error`. This happened most often right after copying a large folder. Reads and
diff --git a/docs/CLI.md b/docs/CLI.md
index 14abd3c..72349ac 100644
--- a/docs/CLI.md
+++ b/docs/CLI.md
@@ -177,7 +177,7 @@ commands are for scripts and for cases where the mount is not available.
 ## Scripting with `--json`
 
 The global `--json` flag makes query commands print JSON: `status`, `ls`, `pins`, `sync list`,
-`devices list`, `locations`, `transfers`, `activity` and `cache inspect`. Commands that change
+`devices list`, `locations`, `transfers`, `activity`, `trash` and `cache inspect`. Commands that change
 something keep their human-readable output; check their exit code.
 
 ```bash
diff --git a/docs/TESTING.md b/docs/TESTING.md
index 31f7ba9..05db5ae 100644
--- a/docs/TESTING.md
+++ b/docs/TESTING.md
@@ -25,18 +25,80 @@ Proton:
 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.
+Kernel/FUSE behavior and remote convergence need a real mount. With no
+arguments, the script tests the account the running service is logged in to:
 
-Use a dedicated test account without irreplaceable data:
+```console
+scripts/fuse-acceptance.sh           # every case, every mount kind
+scripts/fuse-acceptance.sh --quick   # one on-demand/mirror pair instead of four
+```
+
+An account run does this, in order:
+
+1. Runs the account-free contract as the reference.
+2. Runs the contract in a fresh `pdfs-acceptance-<id>` folder in My files and
+   diffs it against the reference.
+3. Registers two new sync folders under `~/.cache/pdfs-acceptance/run-<id>/`
+   and runs the contract and the move cases in every on-demand/mirror pairing.
+4. Removes everything it created, locally and remotely, and reports a
+   `cleanup` result. The run fails if anything is left behind.
+
+Cleanup identifies remote folders by uid, never by name alone, so nothing of
+yours is touched. It removes each sync folder with `sync rm --delete-remote`,
+trashes the My files folder, and then deletes both permanently from the trash.
+It only counts a folder as gone once a fresh trash listing no longer shows it.
+It then reads `pdfs sync queue` and counts any queued op that still names one
+of its folders as left behind too, since such an op keeps its staged bytes and
+retries forever (B102).
+Cleanup also runs on a failure, Ctrl-C, `SIGTERM` or `SIGHUP`. While it runs,
+two more interrupts are ignored; a third abandons it.
+
+Each run records what it is about to create in
+`~/.cache/pdfs-acceptance/run-<id>/manifest.json` before it creates it, and
+holds a lock on that directory. If a run dies without cleaning up (`kill -9`, a
+hard timeout, a power cut), the next run finds the unlocked manifest and
+finishes that cleanup first. It then removes stray `pdfs-acceptance-<32 hex>`
+folders that no live run claims: in My files (older than
+`PDFS_ACCEPTANCE_REAP_AGE`, default one hour), under the acceptance home, and
+in the trash. A run does not start while that reap leaves anything behind.
+
+The run creates and destroys real remote folders, so a dedicated test account
+without irreplaceable data is still the safest target.
+
+`--live PATH...` tests mounts you set up yourself and leaves the sync folder
+configuration alone. It runs destructive POSIX operations only below a fresh
+`pdfs-acceptance-*` directory, refuses paths that are not FUSE mounts, and
+removes its directory on success, failure, or interruption:
 
 ```console
 scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount
 ```
 
+### Edge cases and performance
+
+Besides the syscall contract, the suite covers:
+
+- **Block boundaries.** Files of 0 bytes up to two 4 MiB blocks plus 3 bytes,
+  including every size next to a boundary, then overwrites that grow, shrink
+  and reshape them.
+- **Unusual names.** A 255-byte name, NFC and NFD forms of the same text,
+  newline, tab, emoji, characters Windows forbids, leading and trailing
+  spaces, and a case-only rename.
+- **A deep tree and a wide directory.** 24 levels of nesting renamed at the
+  top; 128 entries created concurrently, concurrent unlinks, then `rm -rf`.
+  Online, each create is a round trip to the server, and the kernel runs
+  creates in one directory one at a time. So the wide case gets twice the
+  per-case timeout and notes the time per create as `wide.create_ms`.
+- **Rename patterns.** Swaps, chains, log rotation, replacing a file under an
+  open reader, the `EISDIR`/`ENOTDIR`/`ENOENT` refusals, and a folder moved out
+  and back.
+- **Handle coherency.** Unsynced writes seen by a second handle, two `O_APPEND`
+  writers, 30 rewrite generations, and a truncate under a reader.
+- **Throughput.** 64 MiB written and read sequentially, 1 MiB in 4 KiB writes,
+  and metadata operations on 100 files. The rates are printed. The case fails
+  below `PDFS_ACCEPTANCE_MIN_MIBPS` (default 10) or `PDFS_ACCEPTANCE_MIN_OPS`
+  (default 20 operations per second).
+
 ### The mount is diffed against an ordinary filesystem
 
 The local reference run is not only a self-test of the runner. Every case
@@ -78,6 +140,16 @@ in-flight `.crdownload` must not be sealed as a revision) and B74 (a file
 renamed after close must keep its content) need a real mount and a reachable
 daemon; they are skipped, not failed, without one.
 
+**B100** (a copied tree reads back at its exact sizes) copies 48 files of
+assorted sizes into the mount at once, as `cp -r` does. On a live mount it
+waits for the queue, checks every size and byte, runs `pdfs refresh` on each
+folder to drop the cached listing, and checks again. The refresh is what
+brought back the provisional sizes the bug reported.
+
+**B101** (a write during its create's upload reaches Drive) holds new files open for
+different times before writing, so their empty creates land before, during and after the
+bytes arrive, then checks every file after the queue drains.
+
 B74 was found by exactly this mechanism: the B70 case failed, and narrowing it
 produced a smaller reproduction that got its own case. The defect is fixed and
 live-validated (2026-07-26); the case now guards against its return.
@@ -136,13 +208,17 @@ stops at the first failure.
 it ran. A suite can pass every assertion while the daemon logs failures behind
 it; that has happened, and it was caught only by reading the journal by hand.
 
-Abandoned `pdfs-acceptance-*` roots from a crashed run are removed at startup,
-but only if they match the exact generated name and are over an hour old
-(`PDFS_ACCEPTANCE_REAP_AGE`), so a concurrent run is never disturbed.
+With `--live`, abandoned `pdfs-acceptance-*` roots from a crashed run are
+removed at startup, but only if they match the exact generated name and are
+over an hour old (`PDFS_ACCEPTANCE_REAP_AGE`), so a concurrent run is never
+disturbed. An account run instead uses the manifest and reaper described
+above.
 
 ### Automated mode matrix
 
-The managed live runner accepts two existing, empty, unmounted directories. It
+An account run drives this matrix itself, in folders it creates. To use
+directories of your choice instead, 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
diff --git a/scripts/fuse-acceptance.py b/scripts/fuse-acceptance.py
index 17ba976..2de3c6d 100755
--- a/scripts/fuse-acceptance.py
+++ b/scripts/fuse-acceptance.py
@@ -24,6 +24,7 @@ import json
 import mmap
 import os
 from pathlib import Path
+import random
 import re
 import shutil
 import signal
@@ -65,6 +66,10 @@ ERROR_LEVEL = re.compile(r"\bERROR\b")
 
 FALLOC_FL_KEEP_SIZE = 0x01
 FALLOC_FL_PUNCH_HOLE = 0x02
+MIB = 1024 * 1024
+# Drive stores content in blocks of this size; the edges of one are where an
+# off-by-one in the upload or the read path shows.
+BLOCK = 4 * MIB
 RENAME_NOREPLACE = 1 << 0
 RENAME_EXCHANGE = 1 << 1
 AT_FDCWD = -100
@@ -153,6 +158,19 @@ def read(path: Path) -> bytes:
         return file.read()
 
 
+def pattern(size: int, seed: str) -> bytes:
+    """Deterministic bytes that differ at every offset, so a shift cannot hide."""
+    return random.Random(seed).randbytes(size)
+
+
+def write_all(fd: int, data: bytes) -> None:
+    view = memoryview(data)
+    while view:
+        written = os.write(fd, view)
+        check(written > 0, "write made no progress")
+        view = view[written:]
+
+
 def write_durable(path: Path, data: bytes) -> None:
     fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
     try:
@@ -276,6 +294,49 @@ class TestTimeout(Exception):
     pass
 
 
+class Interrupted(KeyboardInterrupt):
+    """SIGTERM or SIGHUP, raised like Ctrl-C so that every `finally` runs."""
+
+
+def _raise_interrupted(signum, _frame):
+    raise Interrupted(f"received {signal.Signals(signum).name}")
+
+
+def install_interrupt_handlers() -> None:
+    for signum in (signal.SIGTERM, signal.SIGHUP):
+        signal.signal(signum, _raise_interrupted)
+
+
+@contextlib.contextmanager
+def shielded():
+    """Hold interrupts off while cleanup runs.
+
+    A Ctrl-C that lands in the middle of cleanup would strand exactly the
+    remote state cleanup exists to remove. Two are acknowledged and ignored;
+    a third abandons cleanup, and the next run reaps whatever it left.
+    """
+    received = 0
+
+    def on_signal(signum, _frame):
+        nonlocal received
+        received += 1
+        if received >= 3:
+            raise Interrupted("cleanup abandoned; the next run removes what is left")
+        print(
+            f"[cleanup] {signal.Signals(signum).name} ignored while cleaning up; "
+            f"send it {3 - received} more time(s) to abandon cleanup",
+            file=sys.stderr,
+        )
+
+    watched = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP)
+    previous = {signum: signal.signal(signum, on_signal) for signum in watched}
+    try:
+        yield
+    finally:
+        for signum, handler in previous.items():
+            signal.signal(signum, handler)
+
+
 @contextlib.contextmanager
 def time_limit(seconds: int, label: str):
     if seconds <= 0:
@@ -945,6 +1006,346 @@ def test_application_workloads(ctx: Context) -> None:
         ctx.note("rsync", "not installed")
 
 
+def test_block_boundaries_and_overwrites(ctx: Context) -> None:
+    """Sizes on either side of every boundary the stack has.
+
+    The kernel moves 4 KiB pages, FUSE 128 KiB requests, and Drive 4 MiB
+    blocks. An off-by-one at any of them truncates, pads or shifts a file by one
+    byte, and only a file of exactly that size shows it.
+    """
+    root = ctx.root / "boundaries"
+    root.mkdir()
+    sizes = [0, 1, 4095, 4096, 4097, 131071, 131072, 131073, BLOCK - 1, BLOCK, BLOCK + 1, 2 * BLOCK + 3]
+    for size in sizes:
+        write_durable(root / f"size-{size}", pattern(size, f"size-{size}"))
+    for size in sizes:
+        path = root / f"size-{size}"
+        reported = os.lstat(path).st_size
+        check(reported == size, f"a {size}-byte file reports {reported} bytes")
+        check_bytes(read(path), pattern(size, f"size-{size}"), f"{size}-byte file")
+    ctx.record("sizes", [os.lstat(root / f"size-{size}").st_size for size in sizes])
+
+    # Overwrites that change how many blocks a file has, in both directions.
+    path = root / "reshaped"
+    write_durable(path, pattern(2 * BLOCK + 3, "reshaped"))
+    model = bytearray(pattern(BLOCK + 7, "shorter"))
+    write_durable(path, bytes(model))
+    check_bytes(read(path), bytes(model), "an overwrite with fewer blocks")
+    fd = os.open(path, os.O_RDWR)
+    try:
+        straddle = pattern(8192, "straddle")
+        os.pwrite(fd, straddle, BLOCK - 4096)
+        model[BLOCK - 4096 : BLOCK + 4096] = straddle
+        tail = pattern(BLOCK, "tail")
+        os.pwrite(fd, tail, len(model))
+        model.extend(tail)
+        # Back to exactly one block, then past the end, leaving a hole.
+        os.ftruncate(fd, BLOCK)
+        del model[BLOCK:]
+        os.pwrite(fd, b"after-hole", BLOCK + 10)
+        model.extend(bytes(10) + b"after-hole")
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    check(os.lstat(path).st_size == len(model), "block-crossing edits left the wrong size")
+    check_bytes(read(path), bytes(model), "block-crossing edits after reopening")
+    ctx.record("reshaped.digest", hashlib.sha256(model).hexdigest())
+
+    # Until the queue drains, reads come from the staged copy. What reached
+    # Drive is only visible after it has, so read everything again from there.
+    if ctx.is_live and ctx.daemon is not None:
+        ctx.daemon.wait_for_queue()
+        for size in sizes:
+            path = root / f"size-{size}"
+            check(os.lstat(path).st_size == size, f"a {size}-byte file changed size on upload")
+            check_bytes(read(path), pattern(size, f"size-{size}"), f"uploaded {size}-byte file")
+        check_bytes(read(root / "reshaped"), bytes(model), "uploaded block-crossing edits")
+
+
+UNUSUAL_NAMES = [
+    "a" * 255,
+    "é" * 127,
+    " leading space",
+    "trailing space ",
+    "trailing dot.",
+    "...",
+    "-starts-with-dash",
+    "#hash & ampersand",
+    "percent%20encoded",
+    "semi;colon,comma",
+    "quote's \"double\"",
+    "back\\slash",
+    "colon:star*question?",
+    "<angle>|pipe",
+    "tab\there",
+    "new\nline",
+    "emoji 📁🗂️",
+    "zero​width",
+    "rtl‮override",
+    "cjk 文件夹",
+    "latin ñ ü ß",
+    "nfc-é",
+    "nfd-é",
+]
+
+
+def test_unusual_names(ctx: Context) -> None:
+    """Every byte sequence Linux allows in a name must survive a round trip.
+
+    Drive's own clients refuse some of these (Windows forbids `:*?"<>|`, macOS
+    normalizes Unicode), but a Linux program may create any of them, and one
+    that is silently renamed or dropped loses the file for that program.
+    """
+    root = ctx.root / "names"
+    root.mkdir()
+    for name in UNUSUAL_NAMES:
+        write_durable(root / name, name.encode())
+    (root / " spaced dir ").mkdir()
+    write_durable(root / " spaced dir " / "child", b"child")
+    listed = set(os.listdir(root))
+    missing = [name for name in UNUSUAL_NAMES if name not in listed]
+    check(not missing, f"names missing from readdir: {missing!r}")
+    for name in UNUSUAL_NAMES:
+        check_bytes(read(root / name), name.encode(), f"contents of {name!r}")
+    check_bytes(read(root / " spaced dir " / "child"), b"child", "child of a spaced directory")
+    for index, name in enumerate(UNUSUAL_NAMES):
+        os.rename(root / name, root / f"renamed-{index}")
+        os.rename(root / f"renamed-{index}", root / name)
+    check(set(os.listdir(root)) == listed, "renaming names away and back changed the listing")
+
+    write_durable(root / "Case-Only.txt", b"case")
+    os.rename(root / "Case-Only.txt", root / "case-only.txt")
+    names = set(os.listdir(root))
+    check("case-only.txt" in names and "Case-Only.txt" not in names, "a case-only rename did not stick")
+    check_bytes(read(root / "case-only.txt"), b"case", "a case-only rename")
+    ctx.record("names", sorted(os.listdir(root)))
+
+
+def test_deep_tree(ctx: Context) -> None:
+    root = ctx.root / "deep-tree"
+    deep = root / "deep"
+    leaf = deep.joinpath(*(f"level-{level:02d}" for level in range(24)))
+    leaf.mkdir(parents=True)
+    write_durable(leaf / "leaf.txt", b"deep leaf")
+    os.rename(deep, root / "deep-renamed")
+    moved = root / "deep-renamed" / leaf.relative_to(deep)
+    check_bytes(read(moved / "leaf.txt"), b"deep leaf", "a leaf 24 levels down after renaming the top")
+
+    shutil.rmtree(root)
+    check(not root.exists(), "rm -rf left the tree behind")
+    check("deep-tree" not in os.listdir(ctx.root), "rm -rf left the tree in readdir")
+
+
+def test_wide_directory(ctx: Context) -> None:
+    """More entries than one readdir reply holds, created and removed in parallel.
+
+    The kernel serializes creates and unlinks within one directory, so the
+    threads only overlap the parts outside that lock. The per-create time is
+    noted: online, each create is a round trip to the server.
+    """
+    root = ctx.root / "wide"
+    root.mkdir()
+    names = [f"entry-{index:04d}.txt" for index in range(128)]
+    started = time.monotonic()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(lambda name: write_durable(root / name, name.encode()), names))
+    ctx.note("wide.create_ms", round((time.monotonic() - started) * 1000 / len(names)))
+    listed = sorted(os.listdir(root))
+    check(listed == names, f"a {len(names)}-entry directory lists {len(listed)} entries")
+    with os.scandir(root) as entries:
+        sizes = {entry.name: entry.stat().st_size for entry in entries}
+    wrong = [name for name in names if sizes.get(name) != len(name)]
+    check(not wrong, f"wrong sizes in a wide directory: {wrong[:10]}")
+    for name in names[::17]:
+        check_bytes(read(root / name), name.encode(), f"wide entry {name}")
+    half = len(names) // 2
+    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(lambda name: os.unlink(root / name), names[:half]))
+    check(sorted(os.listdir(root)) == names[half:], "concurrent unlinks left the wrong entries")
+    ctx.record("wide.count", len(listed))
+
+    shutil.rmtree(root)
+    check(not root.exists(), "rm -rf left the directory behind")
+    check("wide" not in os.listdir(ctx.root), "rm -rf left the directory in readdir")
+
+
+def test_rename_patterns(ctx: Context) -> None:
+    """The rename sequences programs actually use, not just single renames."""
+    root = ctx.root / "renames"
+    root.mkdir()
+    first, second, spare = root / "a.txt", root / "b.txt", root / "swap.tmp"
+    write_durable(first, b"alpha")
+    write_durable(second, b"beta")
+    os.rename(first, spare)
+    os.rename(second, first)
+    os.rename(spare, second)
+    check(read(first) == b"beta" and read(second) == b"alpha", "a three-way swap mixed up contents")
+
+    write_durable(root / "chain-0", b"chain")
+    for step in range(1, 6):
+        os.rename(root / f"chain-{step - 1}", root / f"chain-{step}")
+    chain = sorted(name for name in os.listdir(root) if name.startswith("chain"))
+    check(chain == ["chain-5"], f"a rename chain left {chain}")
+    check_bytes(read(root / "chain-5"), b"chain", "the end of a rename chain")
+
+    # Log rotation: rename away and recreate the name straight after.
+    log = root / "app.log"
+    write_durable(log, b"first generation\n")
+    os.rename(log, root / "app.log.1")
+    write_durable(log, b"second generation\n")
+    check_bytes(read(root / "app.log.1"), b"first generation\n", "the rotated log")
+    check_bytes(read(log), b"second generation\n", "the recreated log")
+
+    # Replacing a file must not change what an already open reader sees.
+    target = root / "replaced.txt"
+    write_durable(target, b"old contents")
+    reader = os.open(target, os.O_RDONLY)
+    try:
+        write_durable(root / "replacement.tmp", b"new contents!")
+        os.replace(root / "replacement.tmp", target)
+        check_bytes(os.pread(reader, 64, 0), b"old contents", "an open reader after its file was replaced")
+    finally:
+        os.close(reader)
+    check_bytes(read(target), b"new contents!", "the replacement")
+
+    (root / "a-dir").mkdir()
+    expect_errno({errno.EISDIR}, lambda: os.rename(first, root / "a-dir"), "rename(file over directory)")
+    expect_errno({errno.ENOTDIR}, lambda: os.rename(root / "a-dir", first), "rename(directory over file)")
+    expect_errno({errno.ENOENT}, lambda: os.rename(first, root / "missing" / "a.txt"), "rename(into a missing directory)")
+
+    (root / "left").mkdir()
+    (root / "right").mkdir()
+    write_durable(root / "left" / "payload", b"travels")
+    os.rename(root / "left", root / "right" / "left")
+    os.rename(root / "right" / "left", root / "left")
+    check_bytes(read(root / "left" / "payload"), b"travels", "a directory moved out and back")
+    ctx.record("listing", sorted(os.listdir(root)))
+
+
+def test_handle_coherency(ctx: Context) -> None:
+    """What one handle writes, every other handle and every later open sees."""
+    root = ctx.root
+    path = root / "coherent"
+    write_durable(path, b"")
+    writer = os.open(path, os.O_WRONLY)
+    reader = os.open(path, os.O_RDONLY)
+    try:
+        write_all(writer, b"visible without fsync")
+        check_bytes(os.pread(reader, 64, 0), b"visible without fsync", "a second handle before fsync")
+    finally:
+        os.close(writer)
+        os.close(reader)
+
+    log = root / "appenders"
+    write_durable(log, b"")
+    handles = [os.open(log, os.O_WRONLY | os.O_APPEND) for _ in range(2)]
+    try:
+        for index in range(50):
+            write_all(handles[index % 2], f"record {index:02d}\n".encode())
+    finally:
+        for handle in handles:
+            os.close(handle)
+    lines = read(log).decode().splitlines()
+    check(lines == [f"record {index:02d}" for index in range(50)], "two O_APPEND handles overwrote each other")
+    ctx.record("appenders.lines", len(lines))
+
+    # Rewrite and reread in a tight loop: a stale page cache or attribute
+    # cache serves the previous generation, or the previous generation's size.
+    churn = root / "churn"
+    for generation in range(30):
+        payload = pattern(1000 + generation * 37, f"generation-{generation}")
+        write_durable(churn, payload)
+        check(os.lstat(churn).st_size == len(payload), f"generation {generation}: stale size")
+        check_bytes(read(churn), payload, f"generation {generation} after close and reopen")
+
+    shrink = root / "shrink-under-reader"
+    write_durable(shrink, b"x" * 8192)
+    reader = os.open(shrink, os.O_RDONLY)
+    try:
+        os.truncate(shrink, 100)
+        check(os.fstat(reader).st_size == 100, "an open handle missed a truncate by path")
+        check_bytes(os.pread(reader, 8192, 0), b"x" * 100, "an open handle read past a truncate")
+    finally:
+        os.close(reader)
+
+
+def test_throughput(ctx: Context) -> None:
+    """Measured, printed, and held to floors a working mount clears easily.
+
+    The floors catch a pathological slowdown (a server round trip per write, a
+    re-download per read), not a slow link. Writes land in local staging and
+    reads of just-written data come from the cache, so neither should depend
+    on the network. PDFS_ACCEPTANCE_MIN_MIBPS and PDFS_ACCEPTANCE_MIN_OPS move
+    the floors; 0 turns one off.
+    """
+    min_rate = float(os.environ.get("PDFS_ACCEPTANCE_MIN_MIBPS", "10"))
+    min_ops = float(os.environ.get("PDFS_ACCEPTANCE_MIN_OPS", "20"))
+    root = ctx.root / "throughput"
+    root.mkdir()
+    chunk = pattern(MIB, "throughput")
+    total = 64
+
+    path = root / "sequential.bin"
+    started = time.monotonic()
+    fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
+    try:
+        for _ in range(total):
+            write_all(fd, chunk)
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    write_rate = total / max(time.monotonic() - started, 1e-6)
+
+    started = time.monotonic()
+    digest = hashlib.sha256()
+    with path.open("rb", buffering=0) as file:
+        while block := file.read(MIB):
+            digest.update(block)
+    read_rate = total / max(time.monotonic() - started, 1e-6)
+    check(digest.hexdigest() == hashlib.sha256(chunk * total).hexdigest(), "the 64 MiB file read back wrong")
+
+    # Many small writes: one upload per write, or a lock held across each,
+    # shows up here long before it shows up in a large copy.
+    small = root / "small-writes.bin"
+    started = time.monotonic()
+    fd = os.open(small, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
+    try:
+        for offset in range(0, MIB, 4096):
+            write_all(fd, chunk[offset : offset + 4096])
+    finally:
+        os.close(fd)
+    small_rate = 1 / max(time.monotonic() - started, 1e-6)
+    check_bytes(read(small), chunk, "1 MiB written 4 KiB at a time")
+
+    started = time.monotonic()
+    names = [root / f"meta-{index:03d}" for index in range(100)]
+    for name in names:
+        write_durable(name, b"m")
+    for name in names:
+        os.lstat(name)
+    os.listdir(root)
+    for name in names:
+        os.unlink(name)
+    ops_rate = (3 * len(names) + 1) / max(time.monotonic() - started, 1e-6)
+
+    print(
+        f"    write {write_rate:.0f} MiB/s, read {read_rate:.0f} MiB/s, "
+        f"4 KiB writes {small_rate:.1f} MiB/s, metadata {ops_rate:.0f} ops/s"
+    )
+    ctx.note("write_mibps", round(write_rate, 1))
+    ctx.note("read_mibps", round(read_rate, 1))
+    ctx.note("small_write_mibps", round(small_rate, 2))
+    ctx.note("metadata_ops", round(ops_rate, 1))
+    for label, rate, floor in (
+        ("sequential write", write_rate, min_rate),
+        ("sequential read", read_rate, min_rate),
+        ("4 KiB writes", small_rate, min_rate / 10),
+    ):
+        check(not floor or rate >= floor, f"{label} ran at {rate:.1f} MiB/s, under the {floor:g} MiB/s floor")
+    check(not min_ops or ops_rate >= min_ops, f"metadata ran at {ops_rate:.1f} ops/s, under the {min_ops:g} floor")
+    ctx.record("within_floors", True)
+
+
 def _run_tool(command: list[str], cwd: Path) -> str:
     result = subprocess.run(command, cwd=cwd, text=True, capture_output=True)
     if result.returncode:
@@ -1075,6 +1476,40 @@ def test_regression_b74_rename_after_close(ctx: Context) -> None:
     ctx.record("digest", hashlib.sha256(read(final)).hexdigest())
 
 
+def test_regression_b101_write_during_create_upload(ctx: Context) -> None:
+    """A write that closes while its file's create uploads must reach Drive (B101).
+
+    The create of a new file drains on its own, often before the first write
+    closes. The write attached its bytes to the create's queue row while the
+    upload was on the wire, and the landed create deleted that row, so Drive
+    kept an empty file. A handle that stayed open across the landing still
+    carried the placeholder uid and failed its close.
+    """
+    daemon = ctx.require_daemon()
+    root = ctx.root / "b101"
+    root.mkdir()
+    payloads = {}
+    for index, pause in enumerate((0.0, 0.5, 2.0, 5.0)):
+        path = root / f"file-{index}"
+        payloads[path] = pattern(BLOCK + 4096 * index + 1, f"b101-{index}")
+        fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
+        try:
+            # Held open with nothing written, so the empty create can drain
+            # (and land) before the bytes arrive.
+            time.sleep(pause)
+            write_all(fd, payloads[path])
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+    daemon.wait_for_queue()
+    time.sleep(3)
+    daemon.wait_for_queue()
+    for path, payload in payloads.items():
+        check(os.lstat(path).st_size == len(payload), f"{path.name} has the wrong size after upload")
+        check_bytes(read(path), payload, f"{path.name} after upload")
+    check(not _conflict_copies(root), "writes during a create produced conflict copies")
+
+
 SHARED_DIR_NAME = "Shared with me"
 
 
@@ -1277,6 +1712,49 @@ def test_regression_b79_ondemand_location_accepts_writes(ctx: Context) -> None:
     ctx.record("b79.location", str(root))
 
 
+def test_regression_b100_copied_tree_exact_sizes(ctx: Context) -> None:
+    """A tree copied in from another volume reads back at its exact sizes (B100).
+
+    A listing that has not fetched a file's real size reports the encrypted
+    size, 53 bytes over, and reading at that size failed with EIO. It was found
+    by copying a tree and reading it straight away. `pdfs refresh` drops the
+    cached listing, which brings that size-less listing back on purpose.
+    """
+    sizes = [0, 1, 53, 4095, 4096, 20_000, 131_073, 1_000_003]
+    files: dict[Path, bytes] = {}
+    copy = ctx.root / "b100-copy"
+    with tempfile.TemporaryDirectory(prefix="pdfs-b100-source-") as directory:
+        source = Path(directory) / "tree"
+        for index in range(48):
+            relative = Path(f"dir-{index % 6}") / f"sub-{index % 3}" / f"file-{index:02d}.bin"
+            files[relative] = pattern(sizes[index % len(sizes)], f"b100-{index}")
+            (source / relative).parent.mkdir(parents=True, exist_ok=True)
+            write_durable(source / relative, files[relative])
+        shutil.copytree(source, copy)
+
+    def verify(when: str) -> None:
+        for relative, data in files.items():
+            size = os.lstat(copy / relative).st_size
+            check(size == len(data), f"{relative} {when}: {size} bytes, expected {len(data)}")
+            check_bytes(read(copy / relative), data, f"{relative} {when}")
+
+    verify("right after the copy")
+    ctx.record("sizes", sorted(len(data) for data in files.values()))
+    if not ctx.is_live or ctx.daemon is None:
+        return
+    ctx.daemon.wait_for_queue()
+    verify("after the upload")
+    folders = sorted({copy / relative.parent for relative in files} | {copy})
+    try:
+        for folder in folders:
+            ctx.daemon.command("refresh", str(folder))
+    except RuntimeError as error:
+        # A mirror folder is plain local storage with no listing to drop.
+        ctx.note("refresh", str(error))
+        return
+    verify("after dropping the cached listing")
+
+
 def _conflict_copies(root: Path) -> list[str]:
     return sorted(name for name in os.listdir(root) if "(sync-conflict" in name)
 
@@ -1309,10 +1787,19 @@ def test_durability_across_restart(ctx: Context) -> None:
 
 
 class Case:
-    def __init__(self, name: str, run, kinds: tuple[str, ...] = (REFERENCE, LIVE)) -> None:
+    def __init__(
+        self,
+        name: str,
+        run,
+        kinds: tuple[str, ...] = (REFERENCE, LIVE),
+        budget_scale: int = 1,
+    ) -> None:
         self.name = name
         self.run = run
         self.kinds = kinds
+        # A case whose work is bound by per-operation server round trips gets a
+        # multiple of the per-case timeout rather than a smaller workload.
+        self.budget_scale = budget_scale
 
 
 TESTS = [
@@ -1330,6 +1817,13 @@ TESTS = [
     Case("unsupported operations refuse cleanly", test_unsupported_operations),
     Case("independent and shared-file concurrency", test_concurrency),
     Case("application workloads (editor, tar, sqlite, git, rsync)", test_application_workloads),
+    Case("block boundaries and overwrites that change the block count", test_block_boundaries_and_overwrites),
+    Case("unusual but legal names", test_unusual_names),
+    Case("a deep tree", test_deep_tree),
+    Case("a wide directory", test_wide_directory, budget_scale=2),
+    Case("rename patterns: swap, chain, rotation, replace under a reader", test_rename_patterns),
+    Case("coherency between handles and across reopen", test_handle_coherency),
+    Case("throughput floors", test_throughput),
     Case("regression B7: renamed directory stays traversable", test_regression_b7_renamed_directory),
     Case("regression B69: identical rewrite makes no conflict", test_regression_b69_identical_rewrite, (LIVE,)),
     Case("regression B70: transient download name is not sealed", test_regression_b70_transient_download_name, (LIVE,)),
@@ -1338,6 +1832,8 @@ TESTS = [
     Case("regression B34b: an editor share still writes", test_regression_b34b_editor_share_still_writes, (LIVE,)),
     Case("synthetic Shared-with-me directory contract", test_shared_directory_contract, (LIVE,)),
     Case("regression B79: an on-demand device folder accepts writes", test_regression_b79_ondemand_location_accepts_writes, (LIVE,)),
+    Case("regression B100: a copied tree reads back at its exact sizes", test_regression_b100_copied_tree_exact_sizes),
+    Case("regression B101: a write during its create's upload reaches Drive", test_regression_b101_write_during_create_upload, (LIVE,)),
     Case("durability across a daemon restart", test_durability_across_restart, (LIVE,)),
 ]
 
@@ -1434,9 +1930,18 @@ def run_contract(
     daemon=None,
     timeout: int = DEFAULT_TIMEOUT,
     fail_fast: bool = False,
+    janitor: Janitor | None = None,
 ) -> Run:
-    reap_stale_roots(parent, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
-    root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
+    """Run the contract in a fresh root under `parent`.
+
+    With a janitor the root is registered before it exists and removed by the
+    janitor, through the daemon; without one it is this function's to delete.
+    """
+    if janitor is None:
+        reap_stale_roots(parent, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
+        root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
+    else:
+        root = janitor.new_root(parent)
     root.mkdir()
     run = Run(label, kind, root)
     REPORT.append(run)
@@ -1449,7 +1954,7 @@ def run_contract(
             started = time.monotonic()
             print(f"  [test] {case.name}")
             try:
-                with time_limit(timeout, case.name):
+                with time_limit(timeout * case.budget_scale, case.name):
                     case.run(context)
             except Skip as reason:
                 elapsed = time.monotonic() - started
@@ -1463,6 +1968,8 @@ def run_contract(
                 # A timeout means the mount may be wedged; further cases would
                 # only produce noise, and cleanup already has to fight for it.
                 break
+            except KeyboardInterrupt:
+                raise
             except BaseException as error:  # noqa: BLE001 - reported, then continued
                 elapsed = time.monotonic() - started
                 detail = f"{type(error).__name__}: {error}"
@@ -1482,7 +1989,8 @@ def run_contract(
             run.digest = hashlib.sha256(read(digest_path)).hexdigest()
         return run
     except BaseException:
-        shutil.rmtree(root, ignore_errors=True)
+        if janitor is None:
+            shutil.rmtree(root, ignore_errors=True)
         raise
 
 
@@ -1660,12 +2168,19 @@ class Daemon:
             return None
         return daemon
 
-    def command(self, *args: str, json_output: bool = False) -> str:
+    def command(self, *args: str, json_output: bool = False, stdin: str | None = None) -> str:
         command = [self.pdfs]
         if json_output:
             command.append("--json")
         command.extend(args)
-        result = subprocess.run(command, text=True, capture_output=True, timeout=self.timeout)
+        result = subprocess.run(
+            command,
+            text=True,
+            capture_output=True,
+            timeout=self.timeout,
+            input=stdin,
+            stdin=None if stdin is not None else subprocess.DEVNULL,
+        )
         if result.returncode:
             detail = result.stderr.strip() or result.stdout.strip()
             raise RuntimeError(f"{' '.join(command)} failed: {detail}")
@@ -1684,6 +2199,38 @@ class Daemon:
             time.sleep(1)
         raise TimeoutError(f"daemon mutation queue did not drain: {last}")
 
+    def queue(self) -> list[dict]:
+        """The daemon's queued uploads and changes, each with its path."""
+        return json.loads(self.command("sync", "queue", json_output=True)).get("items", [])
+
+    def sync_folders(self) -> list[dict]:
+        return json.loads(self.command("sync", "list", json_output=True))["items"]
+
+    def listing(self, directory: Path) -> list[dict]:
+        """A directory's entries as the daemon knows them, without touching FUSE."""
+        return json.loads(self.command("ls", str(directory), json_output=True)).get("entries", [])
+
+    def child_uid(self, directory: Path, name: str) -> str | None:
+        return next((entry["uid"] for entry in self.listing(directory) if entry["name"] == name), None)
+
+    def trashed(self, uid: str) -> bool:
+        """Whether `uid` is in the trash, asking the server rather than the cache."""
+        self.command("refresh", "trash")
+        return any(entry["uid"] == uid for entry in self.trash())
+
+    def trash(self) -> list[dict]:
+        output = self.command("trash", json_output=True)
+        try:
+            return json.loads(output).get("entries", [])
+        except ValueError:
+            # A CLI from before `trash --json`: read its table instead.
+            entries = []
+            for line in output.splitlines():
+                match = TRASH_LINE.match(line)
+                if match:
+                    entries.append({"is_dir": match[1] == "d", "name": match[3], "uid": match[4]})
+            return entries
+
     def enclosing_mount(self, path: Path) -> Path:
         current = path.resolve()
         while current != current.parent:
@@ -1714,6 +2261,337 @@ class Daemon:
         raise TimeoutError(f"{mountpoint} did not come back within {self.timeout}s")
 
 
+TRASH_LINE = re.compile(r"^([d-])\s+(\d+)  (.*)  \[([^\]]+)\]$")
+
+
+def acceptance_home() -> Path:
+    """Where account runs keep their manifests and their local sync folders."""
+    base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache")
+    return Path(base) / "pdfs-acceptance"
+
+
+class Janitor:
+    """What one account run creates, written down before it exists.
+
+    Every remote root and every sync folder goes into a manifest *before* it
+    is created, and the manifest sits next to a lock this process holds for its
+    whole life. Cleanup never has to guess: on a pass, a failure or a signal it
+    removes exactly what the manifest lists. A run killed outright (SIGKILL,
+    the hard timeout, a power cut) leaves a manifest whose lock nobody holds,
+    and the next run finishes that cleanup before it starts its own.
+
+    Removal goes through the daemon, not `rm -rf`: one server-side trash of the
+    root, then a permanent delete by uid. Deleting a trashed folder for good
+    also takes every entry that was trashed out of it one at a time, so the
+    account's trash ends up exactly as it was.
+    """
+
+    def __init__(self, daemon: Daemon, directory: Path, lock_fd: int) -> None:
+        self.daemon = daemon
+        self.directory = directory
+        self.lock_fd = lock_fd
+        self.roots: list[str] = []
+        self.sync_folders: list[str] = []
+
+    @classmethod
+    def start(cls, daemon: Daemon) -> Janitor:
+        home = acceptance_home()
+        home.mkdir(parents=True, exist_ok=True)
+        # Locked under a hidden name and only then renamed into view, so a
+        # concurrent reaper never sees a run directory without a held lock.
+        token = uuid.uuid4().hex
+        pending = home / f".run-{token}"
+        pending.mkdir()
+        lock_fd = os.open(pending / "lock", os.O_CREAT | os.O_RDWR, 0o600)
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+        directory = home / f"run-{token}"
+        os.rename(pending, directory)
+        janitor = cls(daemon, directory, lock_fd)
+        janitor.save()
+        return janitor
+
+    @property
+    def manifest(self) -> Path:
+        return self.directory / "manifest.json"
+
+    def save(self) -> None:
+        payload = {
+            "pid": os.getpid(),
+            "started": int(time.time()),
+            "roots": self.roots,
+            "sync_folders": self.sync_folders,
+        }
+        temporary = self.directory / "manifest.json.tmp"
+        temporary.write_text(json.dumps(payload, indent=2))
+        os.replace(temporary, self.manifest)
+
+    def new_root(self, parent: Path) -> Path:
+        """A fresh root in a mount, recorded before the caller creates it."""
+        root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
+        self.roots.append(str(root))
+        self.save()
+        return root
+
+    def new_sync_folder(self) -> Path:
+        """A fresh local directory to register as a sync folder.
+
+        The daemon names the remote folder after the local basename, so the
+        basename carries the same unique marker as every other root.
+        """
+        path = self.directory / f"pdfs-acceptance-{uuid.uuid4().hex}"
+        self.sync_folders.append(str(path))
+        self.save()
+        path.mkdir()
+        return path
+
+    # -- cleanup ------------------------------------------------------------
+
+    def cleanup(self) -> list[str]:
+        """Remove everything recorded; return what could not be removed."""
+        leftovers: list[str] = []
+        with shielded():
+            print(f"[cleanup] removing what this run created ({self.directory.name})")
+            with contextlib.suppress(Exception):
+                self.daemon.wait_for_queue()
+            for local in reversed(self.sync_folders):
+                leftovers.extend(self._remove_sync_folder(Path(local)))
+            for root in reversed(self.roots):
+                leftovers.extend(self._remove_root(Path(root)))
+            leftovers.extend(self._queued_leftovers())
+            if leftovers:
+                for line in leftovers:
+                    print(f"[cleanup] LEFT OVER: {line}")
+                print(f"[cleanup] the manifest stays at {self.manifest}; the next run retries")
+            else:
+                self._forget()
+                print("[cleanup] nothing left behind")
+        return leftovers
+
+    def _queued_leftovers(self) -> list[str]:
+        """Queued ops still naming something this run created.
+
+        Removing a root has to take its queued uploads along. One that
+        survives keeps its staged bytes on disk and retries against the trash
+        forever, or lands later as a conflict copy, so it is a leftover like
+        any file (`docs/BUGS.md` B102). The daemon drops them as it trashes,
+        so this only waits out a drain worker that was mid-attempt.
+        """
+        names = {Path(path).name for path in self.roots + self.sync_folders}
+        deadline = time.monotonic() + min(self.daemon.timeout, 30)
+        while True:
+            try:
+                queued = [
+                    item
+                    for item in self.daemon.queue()
+                    if names.intersection(Path(item.get("path", "")).parts)
+                ]
+            except Exception as error:  # noqa: BLE001 - reported as a leftover
+                return [f"could not read the daemon queue: {error}"]
+            if not queued or time.monotonic() >= deadline:
+                break
+            time.sleep(1)
+        return [
+            f"queued {item['kind']} #{item['id']} for {item['path']}"
+            + (f" ({item['last_error']})" if item.get("last_error") else "")
+            for item in queued
+        ]
+
+    def _forget(self) -> None:
+        with contextlib.suppress(OSError):
+            os.close(self.lock_fd)
+        shutil.rmtree(self.directory, ignore_errors=True)
+
+    def _remove_sync_folder(self, local: Path) -> list[str]:
+        leftovers: list[str] = []
+        registered = None
+        try:
+            registered = next(
+                (
+                    item
+                    for item in self.daemon.sync_folders()
+                    if Path(item["local_path"]) == local
+                ),
+                None,
+            )
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return [f"sync folder {local}: could not list sync folders: {error}"]
+        if registered is not None:
+            print(f"[cleanup] unregistering sync folder {local}")
+            try:
+                self.daemon.command("sync", "rm", str(registered["id"]), "--delete-remote")
+            except Exception as error:  # noqa: BLE001 - reported as a leftover
+                leftovers.append(f"sync folder {local}: sync rm failed: {error}")
+            uid = registered.get("remote_uid")
+            if uid and not self.purge(uid, local.name):
+                leftovers.append(f"remote folder {local.name} ({uid}) is still in the trash")
+        else:
+            # Unregistered by an earlier attempt that died before the purge.
+            leftovers.extend(self._purge_by_name(local.name))
+        if is_mountpoint(local):
+            leftovers.append(f"{local} is still mounted")
+        elif local.exists() or local.is_symlink():
+            shutil.rmtree(local, ignore_errors=True)
+            if local.exists():
+                leftovers.append(f"local directory {local} could not be removed")
+        return leftovers
+
+    def _remove_root(self, root: Path) -> list[str]:
+        uid = None
+        try:
+            uid = self.daemon.child_uid(root.parent, root.name)
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return [f"{root}: could not list its parent through the daemon: {error}"]
+        if uid is None:
+            # Never created, or trashed by an attempt that died before the purge.
+            return self._purge_by_name(root.name)
+        print(f"[cleanup] removing {root}")
+        try:
+            self.daemon.command("rm", str(root))
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return [f"{root}: pdfs rm failed: {error}"]
+        if not self.purge(uid, root.name):
+            return [f"{root.name} ({uid}) is still in the trash"]
+        return []
+
+    def _purge_by_name(self, name: str) -> list[str]:
+        try:
+            self.daemon.command("refresh", "trash")
+            matches = [entry for entry in self.daemon.trash() if entry["name"] == name]
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return [f"{name}: could not read the trash: {error}"]
+        return [
+            f"{name} ({entry['uid']}) is still in the trash"
+            for entry in matches
+            if not self.purge(entry["uid"], name)
+        ]
+
+    def purge(self, uid: str, name: str) -> bool:
+        """Permanently delete one trashed node the run owns, by uid.
+
+        By uid and never by name: a name can match something of the user's, a
+        uid cannot. The node has to show up in the trash first, which is also
+        what proves the trash step before it worked, and it has to be gone from
+        a fresh listing afterwards.
+        """
+        deadline = time.monotonic() + self.daemon.timeout
+        attempted = False
+        last_error = ""
+        while True:
+            try:
+                if self.daemon.trashed(uid):
+                    attempted = True
+                    self.daemon.command("delete-forever", uid, stdin="yes\n")
+                # Gone from the trash after a delete was sent, even one that
+                # reported an error, means the delete landed.
+                if attempted and not self.daemon.trashed(uid):
+                    print(f"[cleanup] permanently deleted {name}")
+                    return True
+            except Exception as error:  # noqa: BLE001 - retried, then reported
+                last_error = f": {error}"
+            if time.monotonic() >= deadline:
+                stage = "it is still in the trash" if attempted else "it never reached the trash"
+                print(f"WARNING: could not permanently delete {name} ({uid}): {stage}{last_error}")
+                return False
+            time.sleep(3)
+
+    # -- reaping ------------------------------------------------------------
+
+    @classmethod
+    def reap(cls, daemon: Daemon, my_files: Path | None) -> list[str]:
+        """Finish the cleanup of every earlier run that died before its own.
+
+        A run directory whose lock can be taken belongs to a dead process. Its
+        manifest says exactly what to remove. Afterwards, sync folders under
+        the acceptance directory and trash entries named like a root that no
+        live run claims are removed too: they are what a run leaves when it
+        dies between creating something and writing it down, which the
+        manifest ordering makes rare but not impossible.
+        """
+        home = acceptance_home()
+        home.mkdir(parents=True, exist_ok=True)
+        leftovers: list[str] = []
+        live_names: set[str] = set()
+        for directory in sorted(home.iterdir()):
+            if not directory.is_dir():
+                continue
+            lock_path = directory / "lock"
+            try:
+                lock_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+            except OSError:
+                continue
+            try:
+                fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except OSError:
+                os.close(lock_fd)
+                live_names.update(cls._claimed_names(directory))
+                continue
+            print(f"[reap] finishing the cleanup of an earlier run: {directory.name}")
+            janitor = cls(daemon, directory, lock_fd)
+            with contextlib.suppress(OSError, ValueError):
+                recorded = json.loads((directory / "manifest.json").read_text())
+                janitor.roots = list(recorded.get("roots", []))
+                janitor.sync_folders = list(recorded.get("sync_folders", []))
+            # Folders that were created but not yet recorded still sit here.
+            for child in directory.iterdir():
+                if STALE_ROOT.match(child.name) and str(child) not in janitor.sync_folders:
+                    janitor.sync_folders.append(str(child))
+            leftovers.extend(janitor.cleanup())
+        leftovers.extend(cls._reap_orphans(daemon, home, live_names, my_files))
+        return leftovers
+
+    @staticmethod
+    def _claimed_names(directory: Path) -> set[str]:
+        try:
+            recorded = json.loads((directory / "manifest.json").read_text())
+        except (OSError, ValueError):
+            return set()
+        return {Path(path).name for path in recorded.get("roots", []) + recorded.get("sync_folders", [])}
+
+    @classmethod
+    def _reap_orphans(
+        cls, daemon: Daemon, home: Path, live_names: set[str], my_files: Path | None
+    ) -> list[str]:
+        leftovers: list[str] = []
+        orphan = cls(daemon, home, -1)
+        if my_files is not None:
+            # Roots from `--live` runs, or from a run on another machine, have
+            # no manifest here. Age is the only safe signal for those.
+            cutoff = time.time() - int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600"))
+            try:
+                entries = daemon.listing(my_files)
+            except Exception as error:  # noqa: BLE001 - reported as a leftover
+                entries = []
+                leftovers.append(f"could not list {my_files}: {error}")
+            for entry in entries:
+                if (
+                    STALE_ROOT.match(entry["name"])
+                    and entry["name"] not in live_names
+                    and entry.get("modified", 0) < cutoff
+                ):
+                    print(f"[reap] removing an abandoned root: {my_files / entry['name']}")
+                    leftovers.extend(orphan._remove_root(my_files / entry["name"]))
+        try:
+            folders = daemon.sync_folders()
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return [f"could not list sync folders: {error}"]
+        for item in folders:
+            local = Path(item["local_path"])
+            if home in local.parents and local.name not in live_names:
+                print(f"[reap] removing orphaned sync folder {local}")
+                leftovers.extend(orphan._remove_sync_folder(local))
+        try:
+            daemon.command("refresh", "trash")
+            trashed = daemon.trash()
+        except Exception as error:  # noqa: BLE001 - reported as a leftover
+            return leftovers + [f"could not read the trash: {error}"]
+        for entry in trashed:
+            if STALE_ROOT.match(entry["name"]) and entry["name"] not in live_names:
+                print(f"[reap] permanently deleting an abandoned root from the trash: {entry['name']}")
+                if not orphan.purge(entry["uid"], entry["name"]):
+                    leftovers.append(f"{entry['name']} ({entry['uid']}) is still in the trash")
+        return leftovers
+
+
 # --------------------------------------------------------------------------
 # Live and managed modes
 # --------------------------------------------------------------------------
@@ -2096,6 +2974,34 @@ def _unchanged(location: Location, files: dict[Path, bytes], what: str) -> None:
         check_bytes(read(location.root / relative), data, f"{location.name}: {relative} {what}")
 
 
+def _copy_between(source: Location, dest: Location) -> None:
+    name = f"copied-{uuid.uuid4().hex[:8]}"
+    files = _move_tree(source.root, name)
+    source.settle()
+    shutil.copytree(source.root / name, dest.root / name)
+    for relative, data in files.items():
+        size = os.lstat(dest.root / relative).st_size
+        check(size == len(data), f"{dest.name}: {relative} is {size} bytes right after the copy, expected {len(data)}")
+        check_bytes(read(dest.root / relative), data, f"{dest.name}: {relative} right after the copy")
+    source.settle()
+    dest.settle()
+    for location in (source, dest):
+        _unchanged(location, files, "after the copy settled")
+        check((location.root / name / "empty-dir").is_dir(), f"{location.name}: the empty folder is missing")
+        copies = location.conflict_copies()
+        check(not copies, f"{location.name}: the copy left conflict copies: {copies}")
+
+
+def test_copy_tree_between_locations(mc: MoveContext) -> None:
+    _copy_between(mc.first, mc.second)
+
+
+def test_copy_tree_out_of_my_files(mc: MoveContext) -> None:
+    if mc.myfiles is None:
+        raise Skip("the daemon reports no My files mount")
+    _copy_between(mc.myfiles, mc.first)
+
+
 def test_move_refuses_a_name_taken_at_the_destination(mc: MoveContext) -> None:
     source, dest = mc.first, mc.second
     name = f"clash-{uuid.uuid4().hex[:8]}.txt"
@@ -2177,6 +3083,8 @@ MOVE_CASES = [
     Case("move a folder from the second location to the first", test_move_folder_second_to_first, (LIVE,)),
     Case("move several sources in one command", test_move_several_sources_at_once, (LIVE,)),
     Case("move into My files and out again", test_move_through_my_files, (LIVE,)),
+    Case("copy a tree between the two locations", test_copy_tree_between_locations, (LIVE,)),
+    Case("copy a tree out of My files", test_copy_tree_out_of_my_files, (LIVE,)),
     Case("move refuses a name taken at the destination", test_move_refuses_a_name_taken_at_the_destination, (LIVE,)),
     Case("move refuses a folder into itself", test_move_refuses_a_folder_into_itself, (LIVE,)),
     Case("move refuses a mirror copy Drive lacks", test_move_refuses_a_mirror_copy_drive_lacks, (LIVE,)),
@@ -2198,7 +3106,11 @@ def _my_files(pair: ManagedSyncPair) -> Path | None:
 
 
 def run_move_contract(
-    pair: ManagedSyncPair, modes: tuple[str, str], timeout: int, fail_fast: bool
+    pair: ManagedSyncPair,
+    modes: tuple[str, str],
+    timeout: int,
+    fail_fast: bool,
+    janitor: Janitor | None = None,
 ) -> Run:
     """Move trees between the two managed folders, and through My files.
 
@@ -2210,6 +3122,8 @@ def run_move_contract(
     second = Location("second", pair.paths[1], modes[1], pair)
     my_files = _my_files(pair)
     myfiles = Location("My files", my_files, "ondemand", pair) if my_files else None
+    if myfiles is not None and janitor is not None:
+        myfiles.root = janitor.new_root(my_files)
     locations = [first, second] + ([myfiles] if myfiles else [])
     label = f"managed move {modes[0]}/{modes[1]}"
     run = Run(label, LIVE, first.root)
@@ -2217,7 +3131,8 @@ def run_move_contract(
     print(f"[target] {label}")
     try:
         for location in locations:
-            reap_stale_roots(location.folder, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
+            if janitor is None:
+                reap_stale_roots(location.folder, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600")))
             location.root.mkdir()
         for location in locations:
             location.settle()
@@ -2226,7 +3141,7 @@ def run_move_contract(
             started = time.monotonic()
             print(f"  [test] {case.name}")
             try:
-                with time_limit(timeout, case.name):
+                with time_limit(timeout * case.budget_scale, case.name):
                     case.run(context)
             except Skip as reason:
                 print(f"    SKIP  {reason}")
@@ -2240,6 +3155,8 @@ def run_move_contract(
                     Result(case.name, label, "timeout", time.monotonic() - started, str(reason))
                 )
                 break
+            except KeyboardInterrupt:
+                raise
             except BaseException as error:  # noqa: BLE001 - reported, then continued
                 detail = f"{type(error).__name__}: {error}"
                 print(f"    FAIL  {detail}")
@@ -2253,6 +3170,10 @@ def run_move_contract(
             run.ran.add(case.name)
     finally:
         for location in locations:
+            # The janitor removes its My files root in one server-side step,
+            # and each sync folder goes away whole with its registration.
+            if janitor is not None:
+                continue
             try:
                 pair.remove_tree(location.root)
                 location.settle()
@@ -2266,14 +3187,18 @@ def is_mountpoint(path: Path) -> bool:
     return os.path.ismount(path)
 
 
-def run_managed_matrix(paths: list[Path], reference: Run, args) -> None:
+def run_managed_matrix(
+    paths: list[Path], reference: Run, args, janitor: Janitor | None = None
+) -> None:
     timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180"))
     pair = ManagedSyncPair(paths, timeout)
     daemon = Daemon.discover(timeout, may_restart=args.durability)
     roots: list[Path] = []
     try:
         pair.create()
-        matrices = [
+        # --quick still puts each mode on each side once; the full matrix adds
+        # the same-mode pairs, where a move never changes kind of storage.
+        matrices = [("ondemand", "mirror")] if args.quick else [
             ("ondemand", "ondemand"),
             ("ondemand", "mirror"),
             ("mirror", "ondemand"),
@@ -2303,14 +3228,16 @@ def run_managed_matrix(paths: list[Path], reference: Run, args) -> None:
                     pair.force_sync(path)
             # A pass settles Drive several times over, so it gets the sync
             # timeout per settle rather than the per-case filesystem limit.
-            run_move_contract(pair, modes, max(args.timeout, 8 * timeout), args.fail_fast)
+            run_move_contract(pair, modes, max(args.timeout, 8 * timeout), args.fail_fast, janitor)
             for path, mode in zip(pair.paths, modes, strict=True):
                 if mode == "mirror":
                     pair.force_sync(path)
     finally:
-        for root in roots:
-            shutil.rmtree(root, ignore_errors=True)
-        pair.cleanup()
+        if janitor is None:
+            with shielded():
+                for root in roots:
+                    shutil.rmtree(root, ignore_errors=True)
+                pair.cleanup()
 
 
 def summarize_divergence(reference: Run, target: Run) -> None:
@@ -2348,6 +3275,11 @@ def summarize_divergence(reference: Run, target: Run) -> None:
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     live_mode = parser.add_mutually_exclusive_group()
+    live_mode.add_argument(
+        "--account",
+        action="store_true",
+        help="run everything on the signed-in account in folders of its own, then remove them",
+    )
     live_mode.add_argument("--live", nargs="+", type=Path, metavar="MOUNTPOINT")
     live_mode.add_argument(
         "--managed-live",
@@ -2358,6 +3290,11 @@ def main() -> int:
     parser.add_argument("--list", action="store_true", help="print case names and exit")
     parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="per-case seconds")
     parser.add_argument("--fail-fast", action="store_true", help="stop at the first failure")
+    parser.add_argument(
+        "--quick",
+        action="store_true",
+        help="with --account or --managed-live: one mode pairing instead of all four",
+    )
     parser.add_argument("--report-json", type=Path, metavar="PATH")
     parser.add_argument("--report-junit", type=Path, metavar="PATH")
     parser.add_argument(
@@ -2385,6 +3322,7 @@ def main() -> int:
             print(f"{'managed':<18} {case.name}")
         return 0
 
+    install_interrupt_handlers()
     journal = JournalWatch(os.environ.get("PDFS_ACCEPTANCE_UNIT", "proton-drive.service"))
     if args.journal_check:
         journal.start()
@@ -2404,7 +3342,9 @@ def main() -> int:
         print("[pass] account-free filesystem API contract")
 
     try:
-        if args.managed_live:
+        if args.account:
+            run_account(reference, args)
+        elif args.managed_live:
             run_managed_matrix(args.managed_live, reference, args)
         elif args.live:
             run_live(args.live, reference, args)
@@ -2413,6 +3353,62 @@ def main() -> int:
     return outcome
 
 
+def run_account(reference: Run, args) -> None:
+    """Everything, on the signed-in account, with nothing left behind.
+
+    The contract runs in My files, then in two sync folders this run creates
+    and registers itself, through every pairing of on-demand and mirror, with
+    the moves between them and My files. The janitor owns all of it.
+    """
+    timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180"))
+    daemon = Daemon.discover(timeout, may_restart=args.durability)
+    if daemon is None:
+        raise RuntimeError(
+            "no running pdfs daemon: start proton-drive.service and sign in, "
+            "or pass --offline-only for the account-free contract"
+        )
+    status = daemon.status()
+    check(bool(status.get("logged_in")), "the daemon is not signed in; run `pdfs login` first")
+    mountpoint = (status.get("mount") or {}).get("mountpoint")
+    check(
+        bool(mountpoint) and is_mountpoint(Path(mountpoint)),
+        f"My files is not mounted (daemon reports {mountpoint!r})",
+    )
+    my_files = Path(mountpoint)
+    print(f"[account] {status.get('username', '?')}, My files at {my_files}")
+
+    stale = Janitor.reap(daemon, my_files)
+    check(not stale, f"leftovers from an earlier run could not be removed: {stale}")
+    janitor = Janitor.start(daemon)
+    try:
+        run = run_contract(
+            my_files,
+            f"My files {my_files}",
+            kind=LIVE,
+            daemon=daemon,
+            timeout=args.timeout,
+            fail_fast=args.fail_fast,
+            janitor=janitor,
+        )
+        summarize_divergence(reference, run)
+        paths = [janitor.new_sync_folder(), janitor.new_sync_folder()]
+        run_managed_matrix(paths, reference, args, janitor)
+    finally:
+        started = time.monotonic()
+        leftovers = janitor.cleanup()
+        cleanup = Run("cleanup", LIVE, janitor.directory)
+        cleanup.results.append(
+            Result(
+                "the run left nothing behind",
+                "cleanup",
+                "fail" if leftovers else "pass",
+                time.monotonic() - started,
+                "; ".join(leftovers),
+            )
+        )
+        REPORT.append(cleanup)
+
+
 def run_live(mountpoints: list[Path], reference: Run, args) -> None:
     completed: list[tuple[Path, Run]] = []
     convergence = os.environ.get("PDFS_ACCEPTANCE_CONVERGENCE", "0") == "1"
diff --git a/scripts/fuse-acceptance.sh b/scripts/fuse-acceptance.sh
index de69865..68ecb25 100755
--- a/scripts/fuse-acceptance.sh
+++ b/scripts/fuse-acceptance.sh
@@ -4,20 +4,28 @@ set -Eeuo pipefail
 
 usage() {
     cat >&2 <<'EOF'
-usage: scripts/fuse-acceptance.sh [--offline-only] [OPTION ...]
+usage: scripts/fuse-acceptance.sh [OPTION ...]
+       scripts/fuse-acceptance.sh --offline-only [OPTION ...]
        scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...] [OPTION ...]
        scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR [OPTION ...]
        scripts/fuse-acceptance.sh MOUNTPOINT [MOUNTPOINT ...]  # compatibility
 
-The local reference suite always runs first and requires no account; the mount
-is then diffed against its recorded behaviour. --live runs the same contract on
-each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only when all live mountpoints
+With no mode, everything runs on the signed-in account: My files, then two
+sync folders it registers itself, through every pairing of on-demand and
+mirror, with moves and copies between them. Everything it creates is removed
+afterwards, on failure and on Ctrl-C too, and permanently deleted from the
+trash; a run that was killed is cleaned up by the next one.
+
+--offline-only runs only the local reference suite, which needs no account.
+It always runs first; every mount is diffed against its recorded behaviour.
+--live runs the same contract on each given mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only when all live mountpoints
 show the same remote folder.
 
 Options are forwarded to the Python runner. The useful ones:
   --list                  print every case and the targets it runs against
   --timeout SECONDS       per-case limit (default 180); a hung case dumps stacks
   --fail-fast             stop at the first failure instead of continuing
+  --quick                 one on-demand/mirror pairing instead of all four
   --report-json PATH      machine-readable results and recorded observations
   --report-junit PATH     JUnit XML for CI
   --budget SECONDS        report any case slower than this
@@ -45,8 +53,11 @@ collect_paths() {
 
 case "${1:-}" in
     -h|--help) usage ;;
-    --offline-only|"")
-        shift || true
+    "")
+        exec "${runner[@]}" --account
+        ;;
+    --offline-only)
+        shift
         exec "${runner[@]}" "$@"
         ;;
     --live)
@@ -60,10 +71,13 @@ case "${1:-}" in
         first="$1"; second="$2"; shift 2
         exec "${runner[@]}" --managed-live "$first" "$second" "$@"
         ;;
-    --*)
-        # Bare options with no mode: an offline run with those options.
+    --account)
         exec "${runner[@]}" "$@"
         ;;
+    --*)
+        # Bare options with no mode: an account run with those options.
+        exec "${runner[@]}" --account "$@"
+        ;;
     *)
         collect_paths "$@"
         ;;