diff --git a/alembic/versions/0031_previews_bump_note_revision.py b/alembic/versions/0031_previews_bump_note_revision.py
new file mode 100644
index 0000000..62710f4
--- /dev/null
+++ b/alembic/versions/0031_previews_bump_note_revision.py
@@ -0,0 +1,38 @@
+"""a link preview's insert, update or delete bumps its note's sync revision
+
+Revision ID: 0031
+Revises: 0030
+Create Date: 2026-10-07
+
+0015 made every child table bump its parent note's `sync_revision`, so a note syncs
+as a whole. `note_link_previews` arrived later (0020) and was never added, and two
+things have been missing on every linked device since:
+
+- A preview fetched in the background after a save (`unfurl_queue.py`) never reached
+  a device that had already pulled the note. The note's revision was assigned when
+  the TEXT was saved, before the preview existed, and nothing moved it afterwards.
+- A preview dismissed on the web stayed on every other device, for the same reason.
+
+The trigger is the same function 0015 installs on the other child tables.
+
+## Downgrade
+
+Drops the trigger. Previews go back to not propagating; nothing is lost.
+"""
+from alembic import op
+
+revision = "0031"
+down_revision = "0030"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE TRIGGER trg_note_link_previews_bump_note AFTER INSERT OR UPDATE OR DELETE "
+        "ON note_link_previews FOR EACH ROW EXECUTE PROCEDURE ts_bump_parent_note_revision()"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TRIGGER IF EXISTS trg_note_link_previews_bump_note ON note_link_previews")
diff --git a/android/ffi/src/models.rs b/android/ffi/src/models.rs
index 6e07ff5..9967788 100644
--- a/android/ffi/src/models.rs
+++ b/android/ffi/src/models.rs
@@ -167,6 +167,8 @@ pub struct Attachment {
     pub mime: String,
     pub size: Option<i64>,
     pub sha256: Option<String>,
+    /// Why the server refused a file attached on this device. None otherwise.
+    pub upload_error: Option<String>,
 }
 
 #[derive(Debug, Clone, uniffi::Record)]
@@ -264,6 +266,7 @@ impl From<core_models::Attachment> for Attachment {
             mime,
             size,
             sha256,
+            upload_error,
         } = value;
         Attachment {
             id,
@@ -272,6 +275,7 @@ impl From<core_models::Attachment> for Attachment {
             mime,
             size,
             sha256,
+            upload_error,
         }
     }
 }
@@ -637,6 +641,10 @@ pub struct PushSummary {
     /// realistic case). Silently retrying forever would be the wrong shape.
     pub rejected: u64,
     pub errors: Vec<String>,
+    /// Files attached on this device that reached the server this cycle.
+    pub uploaded: u64,
+    /// Files that didn't; their reasons are in `errors`.
+    pub upload_failed: u64,
 }
 
 #[derive(Debug, Clone, uniffi::Record)]
@@ -668,6 +676,8 @@ impl From<core_push::PushSummary> for PushSummary {
             noop,
             rejected,
             errors,
+            uploaded,
+            upload_failed,
         } = value;
         PushSummary {
             batches: batches as u64,
@@ -678,6 +688,8 @@ impl From<core_push::PushSummary> for PushSummary {
             noop: noop as u64,
             rejected: rejected as u64,
             errors,
+            uploaded: uploaded as u64,
+            upload_failed: upload_failed as u64,
         }
     }
 }
diff --git a/core/src/local/models.rs b/core/src/local/models.rs
index 97234b6..29920c2 100644
--- a/core/src/local/models.rs
+++ b/core/src/local/models.rs
@@ -55,6 +55,10 @@ pub struct Attachment {
     pub mime: String,
     pub size: Option<i64>,
     pub sha256: Option<String>,
+    /// Why the server refused a file attached on this device, in words to show on it.
+    /// Absent for everything else, including a file still waiting to upload.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub upload_error: Option<String>,
 }
 
 #[derive(Serialize)]
diff --git a/core/src/local/schema.rs b/core/src/local/schema.rs
index 7cd9867..221fc40 100644
--- a/core/src/local/schema.rs
+++ b/core/src/local/schema.rs
@@ -274,6 +274,23 @@ UPDATE saved_filters
    AND params LIKE '%"color"%';
 "#;
 
+// v10 (#5168): an attachment can be created on THIS device.
+//
+// Until now every attachment row arrived on the delta feed, so every one was already on
+// the server. A file attached here, offline, is not, and `uploaded = 0` is the queue
+// push drains once the note has landed. The rows already here all came from the
+// server, which is why the default is 1.
+//
+// `upload_error` holds a refusal that retrying won't fix — over the size limit, an id
+// already in use, bytes that don't match their hash. A row carrying one leaves the
+// queue: a file refused for its size would otherwise be re-sent in full on every
+// cycle, every five minutes, for as long as the app was open. The message is what the
+// editor shows on the file instead.
+const SCHEMA_V10: &str = r#"
+ALTER TABLE attachments ADD COLUMN uploaded INTEGER NOT NULL DEFAULT 1;
+ALTER TABLE attachments ADD COLUMN upload_error TEXT;
+"#;
+
 pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
     conn.execute_batch("PRAGMA foreign_keys = ON;")?;
     let version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
@@ -313,6 +330,10 @@ pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
         conn.execute_batch(SCHEMA_V9)?;
         conn.execute_batch("PRAGMA user_version = 9;")?;
     }
+    if version < 10 {
+        conn.execute_batch(SCHEMA_V10)?;
+        conn.execute_batch("PRAGMA user_version = 10;")?;
+    }
     Ok(())
 }
 
@@ -427,7 +448,34 @@ mod tests {
         let version: i64 = conn
             .query_row("PRAGMA user_version", [], |r| r.get(0))
             .expect("version");
-        assert_eq!(version, 9);
+        assert_eq!(version, 10);
+    }
+
+    /// Every attachment that predates v10 came down the feed, so it is already on the
+    /// server. Defaulting it to "waiting to upload" would re-send every file once.
+    #[test]
+    fn v10_counts_existing_attachments_as_already_on_the_server() {
+        let conn = v7_db();
+        migrate_v8(&conn).expect("v8");
+        conn.execute_batch(SCHEMA_V9).expect("v9");
+        conn.execute_batch("PRAGMA user_version = 9;").expect("v9");
+        add_note(&conn, "n", "a note");
+        conn.execute(
+            "INSERT INTO attachments (id, note_id, url) VALUES ('a', 'n', '/x')",
+            [],
+        )
+        .expect("seed");
+
+        migrate(&conn).expect("migrate");
+        let (uploaded, error): (bool, Option<String>) = conn
+            .query_row(
+                "SELECT uploaded, upload_error FROM attachments WHERE id = 'a'",
+                [],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .expect("row");
+        assert!(uploaded);
+        assert_eq!(error, None);
     }
 
     /// The column is gone, not merely unread. Asserted by asking SQLite rather than by
diff --git a/core/src/local/store.rs b/core/src/local/store.rs
index 075db4d..55891a9 100644
--- a/core/src/local/store.rs
+++ b/core/src/local/store.rs
@@ -92,7 +92,7 @@ fn items_of(body: &str) -> Vec<ChecklistItem> {
 
 fn load_attachments(conn: &Connection, note_id: &str) -> rusqlite::Result<Vec<Attachment>> {
     let mut stmt = conn.prepare(
-        "SELECT id, url, filename, mime, size, sha256 FROM attachments WHERE note_id = ?1 ORDER BY position ASC",
+        "SELECT id, url, filename, mime, size, sha256, upload_error FROM attachments WHERE note_id = ?1 ORDER BY position ASC",
     )?;
     let rows = stmt.query_map([note_id], |r| {
         let server_url: String = r.get(1)?;
@@ -117,6 +117,7 @@ fn load_attachments(conn: &Connection, note_id: &str) -> rusqlite::Result<Vec<At
             mime,
             size: r.get(4)?,
             sha256,
+            upload_error: r.get(6)?,
         })
     })?;
     rows.collect()
@@ -660,20 +661,110 @@ pub fn delete_item(conn: &Connection, id: &str, item_id: &str) -> rusqlite::Resu
     set_body(conn, id, derive::remove_item(&body, index))
 }
 
+/// The stored form of a declared content type: the bare media type, lowercase.
+/// Matches the server's `normalize_mime`, so both sides file a type the same way.
+fn normalize_mime(raw: &str) -> String {
+    let bare = raw
+        .split(';')
+        .next()
+        .unwrap_or("")
+        .trim()
+        .to_ascii_lowercase();
+    if bare.is_empty() {
+        "application/octet-stream".to_string()
+    } else {
+        bare
+    }
+}
+
+/// A file's name reduced to its last path component, as the server's
+/// `_safe_filename` does — the name is for display and download, never a path.
+fn safe_filename(raw: &str) -> String {
+    let base = raw.trim().replace('\\', "/");
+    let base = base.rsplit('/').next().unwrap_or("").trim();
+    let capped: String = base.chars().take(255).collect();
+    if capped.is_empty() {
+        "file".to_string()
+    } else {
+        capped
+    }
+}
+
+/// Attach a file on this device, linked or not.
+///
+/// The bytes go into the blob store under their hash, and the row waits with
+/// `uploaded = 0` until push sends it — after the note itself has landed, since the
+/// server files an attachment under its note. The note is touched, so it is dirty
+/// too: that is what a background sync keys on to send it promptly.
+pub fn add_attachment(
+    conn: &Connection,
+    blobs: &crate::sync::blobs::BlobStore,
+    note_id: &str,
+    filename: &str,
+    mime: &str,
+    bytes: &[u8],
+) -> Result<Note, String> {
+    let exists: Option<i64> = conn
+        .query_row("SELECT 1 FROM notes WHERE id = ?1", [note_id], |r| r.get(0))
+        .optional()
+        .map_err(|e| e.to_string())?;
+    if exists.is_none() {
+        return Err("That note no longer exists.".to_string());
+    }
+    let sha256 = blobs.put(bytes)?;
+    let id = new_id();
+    conn.execute(
+        "INSERT INTO attachments (id, note_id, url, filename, mime, size, sha256, position, uploaded)
+              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7,
+                      (SELECT COALESCE(MAX(position) + 1, 0) FROM attachments WHERE note_id = ?2),
+                      0)",
+        params![
+            id,
+            note_id,
+            // The server's route for it, which is what the row holds once it has
+            // synced too. Nothing renders it: `load_attachments` serves the local bytes.
+            format!("/api/notes/{note_id}/attachments/{id}"),
+            safe_filename(filename),
+            normalize_mime(mime),
+            bytes.len() as i64,
+            sha256,
+        ],
+    )
+    .map_err(|e| e.to_string())?;
+    touch(conn, note_id).map_err(|e| e.to_string())?;
+    load_note(conn, note_id).map_err(|e| e.to_string())
+}
+
 pub fn delete_attachment(conn: &Connection, id: &str, att_id: &str) -> rusqlite::Result<Note> {
+    let uploaded: Option<bool> = conn
+        .query_row(
+            "SELECT uploaded FROM attachments WHERE id = ?1 AND note_id = ?2",
+            params![att_id, id],
+            |r| r.get(0),
+        )
+        .optional()?;
     conn.execute(
         "DELETE FROM attachments WHERE id = ?1 AND note_id = ?2",
         params![att_id, id],
     )?;
+    // Only a file the server holds needs a tombstone; without one the next pull would
+    // bring it straight back. One still waiting to upload leaves the queue with its row.
+    if uploaded == Some(true) {
+        record_pending_delete(conn, "attachment", att_id)?;
+    }
     touch(conn, id)?;
     load_note(conn, id)
 }
 
 pub fn delete_preview(conn: &Connection, id: &str, preview_id: &str) -> rusqlite::Result<Note> {
-    conn.execute(
+    let removed = conn.execute(
         "DELETE FROM link_previews WHERE id = ?1 AND note_id = ?2",
         params![preview_id, id],
     )?;
+    // Previews are made by the server, so every one it has is one it would send back.
+    if removed > 0 {
+        record_pending_delete(conn, "preview", preview_id)?;
+    }
     touch(conn, id)?;
     load_note(conn, id)
 }
@@ -1460,4 +1551,126 @@ mod tests {
         trash(&conn, &binned.id).expect("trash");
         assert_eq!(list_labels(&conn).expect("labels")[0].count, Some(1));
     }
+
+    fn blobs(tag: &str) -> crate::sync::blobs::BlobStore {
+        let dir = std::env::temp_dir().join(format!("ts-store-blobs-{}-{tag}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        crate::sync::blobs::BlobStore::new(dir).expect("blobs")
+    }
+
+    fn tombstones(conn: &Connection) -> Vec<(String, String)> {
+        let mut stmt = conn
+            .prepare("SELECT entity, id FROM pending_deletes ORDER BY entity, id")
+            .expect("prepare");
+        let rows = stmt
+            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
+            .expect("query");
+        rows.collect::<rusqlite::Result<_>>().expect("rows")
+    }
+
+    #[test]
+    fn an_attached_file_is_kept_here_and_queued_to_upload() {
+        let conn = db();
+        let blobs = blobs("attach");
+        let n = note(&conn, "with a receipt");
+        conn.execute("UPDATE notes SET dirty = 0", [])
+            .expect("clean");
+
+        let got = add_attachment(
+            &conn,
+            &blobs,
+            &n.id,
+            "C:\\scans\\receipt.PDF",
+            "Application/PDF; name=x",
+            b"%PDF",
+        )
+        .expect("attach");
+
+        let att = &got.attachments[0];
+        assert_eq!(
+            att.filename.as_deref(),
+            Some("receipt.PDF"),
+            "a name, not a path"
+        );
+        assert_eq!(att.mime, "application/pdf");
+        assert_eq!(att.size, Some(4));
+        let hash = att.sha256.clone().expect("hashed");
+        assert!(blobs.has(&hash), "the bytes are on this device");
+        assert!(att.url.contains(&hash), "and served from here: {}", att.url);
+        assert_eq!(att.upload_error, None);
+        let (uploaded, dirty): (bool, bool) = conn
+            .query_row(
+                "SELECT a.uploaded, n.dirty FROM attachments a JOIN notes n ON n.id = a.note_id",
+                [],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .expect("row");
+        assert!(!uploaded, "it waits for push");
+        assert!(
+            dirty,
+            "the note is touched, which is what starts a background sync"
+        );
+    }
+
+    #[test]
+    fn attaching_to_a_note_that_is_gone_writes_nothing() {
+        let conn = db();
+        let blobs = blobs("gone");
+        assert!(add_attachment(&conn, &blobs, "missing", "a.txt", "text/plain", b"hi").is_err());
+        let rows: i64 = conn
+            .query_row("SELECT COUNT(*) FROM attachments", [], |r| r.get(0))
+            .expect("count");
+        assert_eq!(rows, 0);
+        let files = std::fs::read_dir(blobs.root()).expect("dir").count();
+        assert_eq!(files, 0, "and no bytes were filed for it");
+    }
+
+    #[test]
+    fn only_a_file_the_server_holds_leaves_a_tombstone_when_removed() {
+        let conn = db();
+        let blobs = blobs("remove");
+        let n = note(&conn, "two files");
+        conn.execute(
+            "INSERT INTO attachments (id, note_id, url) VALUES ('synced', ?1, '/x')",
+            [&n.id],
+        )
+        .expect("synced row");
+        let with_local =
+            add_attachment(&conn, &blobs, &n.id, "new.txt", "text/plain", b"hi").expect("attach");
+        let local = with_local
+            .attachments
+            .iter()
+            .find(|a| a.id != "synced")
+            .expect("local")
+            .id
+            .clone();
+
+        delete_attachment(&conn, &n.id, "synced").expect("remove synced");
+        delete_attachment(&conn, &n.id, &local).expect("remove local");
+
+        // Without the tombstone the next pull would put the synced file straight back.
+        // The local one never reached the server, so there is nothing to tell it.
+        assert_eq!(
+            tombstones(&conn),
+            vec![("attachment".to_string(), "synced".to_string())]
+        );
+    }
+
+    #[test]
+    fn dismissing_a_preview_leaves_a_tombstone() {
+        let conn = db();
+        let n = note(&conn, "https://example.com");
+        conn.execute(
+            "INSERT INTO link_previews (id, note_id, url) VALUES ('p1', ?1, 'https://example.com')",
+            [&n.id],
+        )
+        .expect("preview");
+
+        delete_preview(&conn, &n.id, "p1").expect("dismiss");
+        delete_preview(&conn, &n.id, "not-there").expect("a miss is fine");
+        assert_eq!(
+            tombstones(&conn),
+            vec![("preview".to_string(), "p1".to_string())]
+        );
+    }
 }
diff --git a/core/src/sync/blobs.rs b/core/src/sync/blobs.rs
index 8a0c8f4..0f46167 100644
--- a/core/src/sync/blobs.rs
+++ b/core/src/sync/blobs.rs
@@ -78,6 +78,15 @@ impl BlobStore {
         Ok(path)
     }
 
+    /// File bytes this device produced (a file attached here) and return their hash.
+    /// The hash is computed from the bytes, so unlike [`store`](Self::store) there is
+    /// nothing to verify against.
+    pub fn put(&self, bytes: &[u8]) -> Result<String, String> {
+        let hash = digest(bytes);
+        self.store(&hash, bytes)?;
+        Ok(hash)
+    }
+
     pub fn read(&self, sha256: &str) -> Option<Vec<u8>> {
         fs::read(self.path(sha256)?).ok()
     }
@@ -140,7 +149,9 @@ fn urlencode(value: &str) -> String {
     out
 }
 
-fn urldecode(value: &str) -> String {
+/// Undo percent-encoding. Also used for the filename the desktop's attach command
+/// receives in a header, which can only carry ASCII.
+pub fn urldecode(value: &str) -> String {
     let bytes = value.as_bytes();
     let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
     let mut i = 0;
@@ -163,12 +174,14 @@ fn urldecode(value: &str) -> String {
 ///
 /// The mime rides in the URL and this scheme is an origin of its own, so echoing an
 /// arbitrary type would let an attachment claiming `text/html` run as a document
-/// there. Echoing is safe only because of the FAMILY check: nothing starting with
-/// `image/` can name a scriptable type. Everything else is served as an opaque
-/// download — the right treatment for an arbitrary file regardless.
+/// there. Echoing is safe only for the families that can't carry script, and
+/// `image/` is not quite one of them: `image/svg+xml` is a document that runs its own
+/// `<script>`, so it is served as an opaque download like everything else unfamiliar.
+/// The web made the same exclusion for the same reason (#1981).
 fn content_type_for(mime: &str) -> String {
     const RENDERABLE: &[&str] = &["image/", "audio/", "video/"];
-    let familiar = RENDERABLE.iter().any(|p| mime.starts_with(p)) || mime == "application/pdf";
+    let familiar = (RENDERABLE.iter().any(|p| mime.starts_with(p)) && mime != "image/svg+xml")
+        || mime == "application/pdf";
     // A header value can't carry control characters, and a mime type has no business
     // being long — both would only arrive from a malformed or hostile feed.
     let printable = mime.len() <= 100 && mime.bytes().all(|b| b.is_ascii_graphic());
@@ -295,6 +308,8 @@ mod tests {
         let opaque = "application/octet-stream";
         assert_eq!(content_type_for("text/html"), opaque);
         assert_eq!(content_type_for("application/javascript"), opaque);
+        // An image family member that is really a document with script in it.
+        assert_eq!(content_type_for("image/svg+xml"), opaque);
         assert_eq!(content_type_for(""), opaque);
         // A control character can't reach a header value even under a safe family.
         assert_eq!(content_type_for("image/png\r\nX-Evil: 1"), opaque);
@@ -309,6 +324,13 @@ mod tests {
         assert!(body.is_empty());
     }
 
+    #[test]
+    fn put_files_bytes_under_their_own_hash() {
+        let store = store("put");
+        assert_eq!(store.put(b"hello").expect("put"), HELLO);
+        assert_eq!(store.read(HELLO).as_deref(), Some(&b"hello"[..]));
+    }
+
     #[test]
     fn missing_blob_reads_as_none() {
         let store = store("missing");
diff --git a/core/src/sync/client.rs b/core/src/sync/client.rs
index 4a286dd..81fe10a 100644
--- a/core/src/sync/client.rs
+++ b/core/src/sync/client.rs
@@ -27,6 +27,11 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
 /// failing one at ten seconds would make a large store impossible to ever pull.
 const SYNC_TIMEOUT: Duration = Duration::from_secs(120);
 
+/// One file going up can be far larger than a page of notes, and a timeout shorter
+/// than the transfer is not a failure that retrying ever fixes — the same upload
+/// would time out again every cycle.
+const UPLOAD_TIMEOUT: Duration = Duration::from_secs(600);
+
 /// Shared by every call that presents a token, so a revoked one reads the same way
 /// wherever it surfaces.
 const TOKEN_REJECTED: &str = "This server rejected the device token — it may have been \
@@ -314,6 +319,71 @@ pub async fn fetch_attachment(
         .map_err(|e| format!("Couldn't download an attachment from {base_url}: {e}"))
 }
 
+/// Why an upload didn't land, split by whether trying again could help.
+#[derive(Debug, PartialEq, Eq)]
+pub enum UploadError {
+    /// Worth another attempt next cycle: the network, a server error, or a note the
+    /// server hasn't got yet.
+    Retry(String),
+    /// The server refused this file and will refuse it the same way every time —
+    /// too large, an id in use, bytes that don't match their hash.
+    Refused(String),
+}
+
+/// Upload one file attached on this device, under the id it was given here.
+///
+/// `PUT /api/sync/attachments/<id>` takes the raw bytes; idempotent, so a retry of
+/// an upload whose reply was lost answers 200 and stores nothing twice.
+#[allow(clippy::too_many_arguments)]
+pub async fn upload_attachment(
+    base_url: &str,
+    token: &str,
+    note_id: &str,
+    attachment_id: &str,
+    filename: &str,
+    mime: &str,
+    sha256: &str,
+    bytes: Vec<u8>,
+) -> Result<(), UploadError> {
+    let url = format!("{base_url}/api/sync/attachments/{attachment_id}");
+    let client = http_with(UPLOAD_TIMEOUT).map_err(UploadError::Retry)?;
+    let request = prepare(client.put(url), Some(token))
+        .query(&[
+            ("note_id", note_id),
+            ("filename", filename),
+            ("sha256", sha256),
+        ])
+        .header(reqwest::header::CONTENT_TYPE, mime)
+        .body(bytes);
+    let response = request
+        .send()
+        .await
+        .map_err(|e| UploadError::Retry(describe_transport_error(base_url, &e)))?;
+
+    let status = response.status();
+    if status.is_success() {
+        return Ok(());
+    }
+    if status == StatusCode::UNAUTHORIZED {
+        return Err(UploadError::Retry(TOKEN_REJECTED.to_string()));
+    }
+    // The server's own words, when it gave some: "file is too large (max 25 MB)" is
+    // what a person can act on, and a bare status code is not.
+    let reason = response
+        .json::<serde_json::Value>()
+        .await
+        .ok()
+        .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(String::from))
+        .unwrap_or_else(|| format!("HTTP {}", status.as_u16()));
+    if status.is_client_error() && status != StatusCode::NOT_FOUND {
+        Err(UploadError::Refused(reason))
+    } else {
+        // 404: the note isn't on the server yet (its push was rejected, say). 5xx:
+        // the server's problem, and likely a passing one.
+        Err(UploadError::Retry(reason))
+    }
+}
+
 /// Send a batch of changes and hand back the raw reply.
 ///
 /// Returns text rather than parsed results so this module stays pure transport —
diff --git a/core/src/sync/compat.rs b/core/src/sync/compat.rs
index 3ef597e..de6ae9c 100644
--- a/core/src/sync/compat.rs
+++ b/core/src/sync/compat.rs
@@ -23,7 +23,9 @@ use std::sync::OnceLock;
 ///
 /// v4 (M315): `color` left the note. NOT a floor raise on either side — see the note
 /// on [`MIN_SERVER_PROTOCOL_VERSION`].
-pub const CLIENT_PROTOCOL_VERSION: u32 = 4;
+/// v5 (#5168): files attached here upload, and removed attachments and previews are
+/// pushed. Additive: both go only to a server advertising `attachment_sync`.
+pub const CLIENT_PROTOCOL_VERSION: u32 = 5;
 
 /// The oldest server protocol this client can drive — the symmetric half of the
 /// server's `min_client_protocol_version`.
@@ -47,7 +49,8 @@ pub const REQUIRED_FEATURES: &[&str] = &["notes", "labels"];
 /// Capabilities whose absence costs a feature but not the link. Listing these
 /// explicitly (rather than diffing against whatever the server happens to send) is
 /// what lets the UI name exactly what the user will be missing.
-pub const OPTIONAL_FEATURES: &[&str] = &["attachments", "tombstones", "revisions"];
+pub const OPTIONAL_FEATURES: &[&str] =
+    &["attachments", "tombstones", "revisions", "attachment_sync"];
 
 /// The handshake fields of `GET /api/config`.
 ///
@@ -75,7 +78,7 @@ pub struct ServerInfo {
 }
 
 impl ServerInfo {
-    fn has_feature(&self, name: &str) -> bool {
+    pub fn has_feature(&self, name: &str) -> bool {
         self.sync_features.iter().any(|f| f.as_str() == name)
     }
 
diff --git a/core/src/sync/engine.rs b/core/src/sync/engine.rs
index 5204a2b..483928c 100644
--- a/core/src/sync/engine.rs
+++ b/core/src/sync/engine.rs
@@ -38,7 +38,17 @@ pub async fn run_cycle(
     base_url: &str,
     token: &str,
 ) -> Result<SyncOutcome, String> {
-    let push = push::run(db, base_url, token).await?;
+    // What this server can do, asked before anything is sent: files attached here,
+    // and removed attachments and previews, go only to a server advertising
+    // `attachment_sync`. An older one keeps them queued on this device until it is
+    // updated, rather than refusing each one on every cycle. Best-effort — an
+    // unanswered probe reads as "not advertised", and the cycle carries on.
+    let server = super::client::probe(base_url).await.ok().map(|p| p.server);
+    let attachment_sync = server
+        .as_ref()
+        .is_some_and(|s| s.has_feature("attachment_sync"));
+
+    let push = push::run(db, blobs, base_url, token, attachment_sync).await?;
     let pull = pull::run(db, blobs, base_url, token).await?;
 
     if pull.clobbered_dirty > 0 {
@@ -51,15 +61,11 @@ pub async fn run_cycle(
         );
     }
 
-    // While we're already talking to this server, re-read what it says about itself.
-    // Today that's the trash-retention window the Trash view counts down against, and
-    // it can change under us whenever an admin edits the setting. Best-effort on
-    // purpose: a config blip must not fail a cycle whose actual work already
-    // succeeded, and the stored value simply stays as it was.
-    let retention = super::client::probe(base_url)
-        .await
-        .ok()
-        .and_then(|p| p.server.trash_retention_days);
+    // The same answer carries the trash-retention window the Trash view counts down
+    // against, which can change whenever an admin edits the setting. Best-effort on
+    // purpose: a config blip must not fail a cycle whose actual work succeeded, and
+    // the stored value simply stays as it was.
+    let retention = server.and_then(|s| s.trash_retention_days);
 
     let status = {
         let conn = db.0.lock().map_err(|e| e.to_string())?;
diff --git a/core/src/sync/pull.rs b/core/src/sync/pull.rs
index abdace2..ea45168 100644
--- a/core/src/sync/pull.rs
+++ b/core/src/sync/pull.rs
@@ -281,14 +281,41 @@ fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
     Ok(())
 }
 
+/// Whether this device removed the row and the server hasn't acknowledged it yet.
+/// Such a row is still on the server, so it is still in the feed — and putting it
+/// back would undo a removal that is only waiting for the next push.
+fn removed_here(conn: &Connection, entity: &str, id: &str) -> rusqlite::Result<bool> {
+    let found: Option<i64> = conn
+        .query_row(
+            "SELECT 1 FROM pending_deletes WHERE entity = ?1 AND id = ?2",
+            params![entity, id],
+            |r| r.get(0),
+        )
+        .optional()?;
+    Ok(found.is_some())
+}
+
 fn replace_attachments(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    // Only rows the server already had are replaced. A file attached here and still
+    // waiting to upload exists nowhere else yet, and deleting it would lose it; once
+    // it has gone up, the server's copy arrives under the same id and takes its place.
     conn.execute(
-        "DELETE FROM attachments WHERE note_id = ?1",
+        "DELETE FROM attachments WHERE note_id = ?1 AND uploaded = 1",
         params![note.id],
     )?;
     for (index, att) in note.attachments.iter().enumerate() {
+        if removed_here(conn, "attachment", &att.id)? {
+            continue;
+        }
+        // The server listing an id this device is still waiting to upload means the
+        // upload landed and only its reply was lost. The server's row replaces it.
+        conn.execute(
+            "DELETE FROM attachments WHERE id = ?1 AND uploaded = 0",
+            params![att.id],
+        )?;
         // The feed carries no explicit position for attachments — they arrive in
-        // creation order, so the index preserves it.
+        // creation order, so the index preserves it. A plain INSERT, so an id listed
+        // twice fails the page rather than being quietly merged.
         conn.execute(
             "INSERT INTO attachments (id, note_id, url, filename, mime, size, sha256, position)
                   VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
@@ -313,6 +340,9 @@ fn replace_previews(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()
         params![note.id],
     )?;
     for (index, preview) in note.previews.iter().enumerate() {
+        if removed_here(conn, "preview", &preview.id)? {
+            continue;
+        }
         conn.execute(
             "INSERT INTO link_previews (id, note_id, url, title, description, image_url,
                                         site_name, position)
@@ -778,4 +808,88 @@ mod tests {
         assert_eq!(state::read(&conn).expect("state").last_cursor, 0);
         assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 0);
     }
+
+    fn preview(id: &str) -> wire::Preview {
+        wire::Preview {
+            id: id.to_string(),
+            url: "https://example.com/".into(),
+            title: None,
+            description: None,
+            image_url: None,
+            site_name: None,
+        }
+    }
+
+    fn queued_upload(conn: &Connection, id: &str, note_id: &str) {
+        conn.execute(
+            "INSERT INTO attachments (id, note_id, url, uploaded) VALUES (?1, ?2, '/x', 0)",
+            params![id, note_id],
+        )
+        .expect("queued upload");
+    }
+
+    fn attachment_ids(conn: &Connection) -> Vec<String> {
+        let mut stmt = conn
+            .prepare("SELECT id FROM attachments ORDER BY id")
+            .expect("prepare");
+        let rows = stmt.query_map([], |r| r.get(0)).expect("query");
+        rows.collect::<rusqlite::Result<_>>().expect("rows")
+    }
+
+    #[test]
+    fn a_pull_keeps_a_file_still_waiting_to_upload() {
+        // It exists only on this device until push sends it; a pull that replaced the
+        // note's attachments wholesale would delete the only copy.
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        queued_upload(&conn, "mine", "n1");
+
+        let mut changed = note("n1", 2);
+        changed.attachments = vec![attachment("theirs")];
+        apply_page(&conn, &page(vec![changed], vec![], 2)).expect("apply");
+
+        assert_eq!(attachment_ids(&conn), vec!["mine", "theirs"]);
+    }
+
+    #[test]
+    fn the_servers_copy_takes_over_from_an_upload_whose_reply_was_lost() {
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        queued_upload(&conn, "a1", "n1");
+
+        let mut listed = note("n1", 2);
+        listed.attachments = vec![attachment("a1")];
+        apply_page(&conn, &page(vec![listed], vec![], 2)).expect("apply");
+
+        assert_eq!(attachment_ids(&conn), vec!["a1"]);
+        let uploaded: bool = conn
+            .query_row(
+                "SELECT uploaded FROM attachments WHERE id = 'a1'",
+                [],
+                |r| r.get(0),
+            )
+            .expect("row");
+        assert!(uploaded, "not sent a second time");
+    }
+
+    #[test]
+    fn a_removal_waiting_to_be_pushed_is_not_undone_by_a_pull() {
+        let conn = db();
+        let mut first = note("n1", 1);
+        first.attachments = vec![attachment("a1")];
+        first.previews = vec![preview("p1")];
+        apply_page(&conn, &page(vec![first], vec![], 1)).expect("apply");
+
+        crate::local::store::delete_attachment(&conn, "n1", "a1").expect("remove");
+        crate::local::store::delete_preview(&conn, "n1", "p1").expect("dismiss");
+
+        // The server hasn't heard yet, so its copy of the note still lists both.
+        let mut stale = note("n1", 2);
+        stale.attachments = vec![attachment("a1")];
+        stale.previews = vec![preview("p1")];
+        apply_page(&conn, &page(vec![stale], vec![], 2)).expect("apply");
+
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM attachments"), 0);
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM link_previews"), 0);
+    }
 }
diff --git a/core/src/sync/push.rs b/core/src/sync/push.rs
index 1fc96cf..314d8e5 100644
--- a/core/src/sync/push.rs
+++ b/core/src/sync/push.rs
@@ -1,8 +1,9 @@
 //! Push: send local changes to the server and apply what it says (M10.7c).
 //!
-//! Two sources feed a push: rows flagged `dirty` (created or edited locally) and rows
+//! Three sources feed a push: rows flagged `dirty` (created or edited locally), rows
 //! in `pending_deletes` (permanently deleted locally — see `local::schema` v2 for why
-//! a delete needs its own record).
+//! a delete needs its own record), and files attached on this device that are still
+//! waiting to go up (`attachments.uploaded = 0`, schema v10).
 //!
 //! Sync is **whole-note**: an upsert carries the client's full current state, not a
 //! patch (docs/sync.md). The server resolves conflicts last-write-wins by the client's
@@ -11,7 +12,8 @@
 use rusqlite::{params, Connection, OptionalExtension};
 use serde::{Deserialize, Serialize};
 
-use super::client;
+use super::blobs::BlobStore;
+use super::client::{self, UploadError};
 use super::state;
 use crate::local::Db;
 
@@ -35,6 +37,11 @@ pub struct PushSummary {
     /// realistic case). Silently retrying forever would be the wrong shape.
     pub rejected: usize,
     pub errors: Vec<String>,
+    /// Files attached on this device that reached the server this cycle.
+    pub uploaded: usize,
+    /// Files that didn't. Their reasons are in `errors`; one the server refused for
+    /// good also carries its reason on the attachment and isn't tried again.
+    pub upload_failed: usize,
 }
 
 impl PushSummary {
@@ -47,6 +54,8 @@ impl PushSummary {
         self.noop += other.noop;
         self.rejected += other.rejected;
         self.errors.extend(other.errors);
+        self.uploaded += other.uploaded;
+        self.upload_failed += other.upload_failed;
     }
 }
 
@@ -136,9 +145,17 @@ pub struct PushResult {
 
 /// Everything waiting to go up, oldest edit first so a truncated batch still makes
 /// forward progress in a sensible order.
-pub fn collect(conn: &Connection, limit: usize) -> rusqlite::Result<Vec<Change>> {
+///
+/// `attachment_sync` is whether the server takes attachment and preview removals.
+/// Without it they stay queued here, untouched, until a server that does — an older
+/// one would answer "unknown entity" and the removal would read as a failure.
+pub fn collect(
+    conn: &Connection,
+    limit: usize,
+    attachment_sync: bool,
+) -> rusqlite::Result<Vec<Change>> {
     let mut out = Vec::new();
-    collect_deletes(conn, &mut out, limit)?;
+    collect_deletes(conn, &mut out, limit, attachment_sync)?;
     if out.len() < limit {
         collect_labels(conn, &mut out, limit)?;
     }
@@ -148,11 +165,21 @@ pub fn collect(conn: &Connection, limit: usize) -> rusqlite::Result<Vec<Change>>
     Ok(out)
 }
 
-fn collect_deletes(conn: &Connection, out: &mut Vec<Change>, limit: usize) -> rusqlite::Result<()> {
+fn collect_deletes(
+    conn: &Connection,
+    out: &mut Vec<Change>,
+    limit: usize,
+    attachment_sync: bool,
+) -> rusqlite::Result<()> {
+    // Filtered in SQL rather than skipped below, so held-back removals can't use up
+    // the LIMIT and starve the deletes that can go.
     let mut stmt = conn.prepare(
-        "SELECT entity, id, deleted_at FROM pending_deletes ORDER BY deleted_at LIMIT ?1",
+        "SELECT entity, id, deleted_at FROM pending_deletes
+          WHERE entity IN ('note', 'label')
+             OR (?2 AND entity IN ('attachment', 'preview'))
+          ORDER BY deleted_at LIMIT ?1",
     )?;
-    let rows = stmt.query_map(params![limit as i64], |r| {
+    let rows = stmt.query_map(params![limit as i64, attachment_sync], |r| {
         Ok((
             r.get::<_, String>(0)?,
             r.get::<_, String>(1)?,
@@ -161,11 +188,13 @@ fn collect_deletes(conn: &Connection, out: &mut Vec<Change>, limit: usize) -> ru
     })?;
     for row in rows {
         let (entity, id, deleted_at) = row?;
-        // Only 'note' and 'label' exist on the wire; anything else is a bug in a
-        // writer, and shipping it would earn a blanket rejection for the batch.
+        // Only these exist on the wire; anything else is a bug in a writer, and
+        // shipping it would earn a rejection that no retry could clear.
         let entity: &'static str = match entity.as_str() {
             "note" => "note",
             "label" => "label",
+            "attachment" => "attachment",
+            "preview" => "preview",
             _ => continue,
         };
         out.push(Change::delete(entity, id, deleted_at));
@@ -414,7 +443,9 @@ pub fn has_pending(conn: &Connection) -> rusqlite::Result<bool> {
         .query_row(
             "SELECT 1 FROM notes WHERE dirty = 1
               UNION ALL SELECT 1 FROM labels WHERE dirty = 1
-              UNION ALL SELECT 1 FROM pending_deletes LIMIT 1",
+              UNION ALL SELECT 1 FROM pending_deletes
+              UNION ALL SELECT 1 FROM attachments WHERE uploaded = 0 AND upload_error IS NULL
+              LIMIT 1",
             [],
             |r| r.get(0),
         )
@@ -434,22 +465,141 @@ pub fn pending_fingerprint(conn: &Connection) -> rusqlite::Result<Option<String>
     let fingerprint: String = conn.query_row(
         "SELECT (SELECT COUNT(*) || ':' || IFNULL(MAX(updated_at), '') FROM notes WHERE dirty = 1)
              || '|' || (SELECT COUNT(*) || ':' || IFNULL(MAX(updated_at), '') FROM labels WHERE dirty = 1)
-             || '|' || (SELECT COUNT(*) || ':' || IFNULL(MAX(deleted_at), '') FROM pending_deletes)",
+             || '|' || (SELECT COUNT(*) || ':' || IFNULL(MAX(deleted_at), '') FROM pending_deletes)
+             || '|' || (SELECT COUNT(*) FROM attachments WHERE uploaded = 0 AND upload_error IS NULL)",
         [],
         |r| r.get(0),
     )?;
-    Ok((fingerprint != "0:|0:|0:").then_some(fingerprint))
+    Ok((fingerprint != "0:|0:|0:|0").then_some(fingerprint))
+}
+
+/// A file attached on this device, ready to go up.
+#[derive(Debug, PartialEq, Eq)]
+pub struct PendingUpload {
+    pub note_id: String,
+    pub id: String,
+    pub filename: String,
+    pub mime: String,
+    pub sha256: String,
+}
+
+/// Files waiting to upload whose note the server already holds. A note still dirty
+/// after the push (its change was rejected) keeps its files back too: the server files
+/// an attachment under its note, and would answer 404 for one it doesn't have.
+pub fn pending_uploads(conn: &Connection) -> rusqlite::Result<Vec<PendingUpload>> {
+    let mut stmt = conn.prepare(
+        "SELECT a.note_id, a.id, IFNULL(a.filename, 'file'), a.mime, a.sha256
+           FROM attachments a JOIN notes n ON n.id = a.note_id
+          WHERE a.uploaded = 0 AND a.upload_error IS NULL AND a.sha256 IS NOT NULL
+            AND n.dirty = 0
+          ORDER BY a.note_id, a.position",
+    )?;
+    let rows = stmt.query_map([], |r| {
+        Ok(PendingUpload {
+            note_id: r.get(0)?,
+            id: r.get(1)?,
+            filename: r.get(2)?,
+            mime: r.get(3)?,
+            sha256: r.get(4)?,
+        })
+    })?;
+    rows.collect()
+}
+
+/// Record what became of one upload.
+fn settle_upload(
+    conn: &Connection,
+    id: &str,
+    result: &Result<(), UploadError>,
+) -> rusqlite::Result<()> {
+    match result {
+        Ok(()) => conn.execute(
+            "UPDATE attachments SET uploaded = 1, upload_error = NULL WHERE id = ?1",
+            params![id],
+        )?,
+        Err(UploadError::Refused(reason)) => conn.execute(
+            "UPDATE attachments SET upload_error = ?2 WHERE id = ?1",
+            params![id, reason],
+        )?,
+        Err(UploadError::Retry(_)) => 0,
+    };
+    Ok(())
+}
+
+/// Send the bytes of every file attached here whose note has landed.
+///
+/// One file failing never fails the cycle, the same as a download: the notes have
+/// already gone, and one unreachable or oversized file must not hold up every sync
+/// after it. A refusal is recorded on the attachment instead, and a passing failure
+/// is simply tried again next cycle.
+async fn upload_pending(
+    db: &Db,
+    blobs: &BlobStore,
+    base_url: &str,
+    token: &str,
+) -> Result<PushSummary, String> {
+    let wanted = {
+        let conn = db.0.lock().map_err(|e| e.to_string())?;
+        pending_uploads(&conn).map_err(|e| e.to_string())?
+    };
+    let mut summary = PushSummary::default();
+    for upload in wanted {
+        let result = match blobs.read(&upload.sha256) {
+            Some(bytes) => {
+                client::upload_attachment(
+                    base_url,
+                    token,
+                    &upload.note_id,
+                    &upload.id,
+                    &upload.filename,
+                    &upload.mime,
+                    &upload.sha256,
+                    bytes,
+                )
+                .await
+            }
+            // Nothing to send and nothing that could bring it back: the file was
+            // only ever on this device.
+            None => Err(UploadError::Refused(
+                "the file's bytes are missing from this device".to_string(),
+            )),
+        };
+        {
+            let conn = db.0.lock().map_err(|e| e.to_string())?;
+            settle_upload(&conn, &upload.id, &result).map_err(|e| e.to_string())?;
+        }
+        match result {
+            Ok(()) => summary.uploaded += 1,
+            Err(UploadError::Refused(reason)) | Err(UploadError::Retry(reason)) => {
+                log::warn!("attachment {}: {reason}", upload.id);
+                summary.upload_failed += 1;
+                summary
+                    .errors
+                    .push(format!("{} didn't upload: {reason}", upload.filename));
+            }
+        }
+    }
+    Ok(summary)
 }
 
 /// Send everything pending, in batches, applying each batch's results before the
-/// next is collected.
-pub async fn run(db: &Db, base_url: &str, token: &str) -> Result<PushSummary, String> {
+/// next is collected — then the files, once their notes are there to hold them.
+///
+/// `attachment_sync`: whether the server advertises it (see [`collect`]). Without
+/// it, uploads wait too.
+pub async fn run(
+    db: &Db,
+    blobs: &BlobStore,
+    base_url: &str,
+    token: &str,
+    attachment_sync: bool,
+) -> Result<PushSummary, String> {
     let mut total = PushSummary::default();
 
     loop {
         let batch = {
             let conn = db.0.lock().map_err(|e| e.to_string())?;
-            collect(&conn, BATCH).map_err(|e| e.to_string())?
+            collect(&conn, BATCH, attachment_sync).map_err(|e| e.to_string())?
         };
         if batch.is_empty() {
             break;
@@ -477,6 +627,10 @@ pub async fn run(db: &Db, base_url: &str, token: &str) -> Result<PushSummary, St
         }
     }
 
+    if attachment_sync {
+        total.absorb(upload_pending(db, blobs, base_url, token).await?);
+    }
+
     if total.rejected > 0 {
         log::warn!(
             "push: {} change(s) rejected by the server: {}",
@@ -485,13 +639,16 @@ pub async fn run(db: &Db, base_url: &str, token: &str) -> Result<PushSummary, St
         );
     }
     log::info!(
-        "push complete: {} sent ({} created, {} applied, {} kept, {} noop, {} rejected)",
+        "push complete: {} sent ({} created, {} applied, {} kept, {} noop, {} rejected), \
+         {} file(s) uploaded, {} not",
         total.sent,
         total.created,
         total.applied,
         total.kept,
         total.noop,
-        total.rejected
+        total.rejected,
+        total.uploaded,
+        total.upload_failed
     );
     Ok(total)
 }
@@ -548,7 +705,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "clean", 0);
         seed_note(&conn, "dirty", 1);
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         assert_eq!(batch.len(), 1);
         assert_eq!(batch[0].id, "dirty");
         assert_eq!(batch[0].op, "upsert");
@@ -573,7 +730,7 @@ mod tests {
             )
             .expect("seed membership");
         }
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         let note = batch.iter().find(|c| c.entity == "note").expect("note");
         assert_eq!(note.label_ids.as_deref(), Some(&["manual".to_string()][..]));
     }
@@ -583,7 +740,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "n1", 0);
         store::delete_forever(&conn, "n1").expect("delete");
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         assert_eq!(batch.len(), 1);
         assert_eq!(batch[0].op, "delete");
         assert_eq!(batch[0].entity, "note");
@@ -594,7 +751,7 @@ mod tests {
     fn applied_clears_dirty_and_records_the_revision() {
         let conn = db();
         seed_note(&conn, "n1", 1);
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         apply_results(&conn, &batch, &[ok("applied", Some(42))]).expect("apply");
         assert_eq!(dirty_count(&conn), 0);
         let rev: i64 = conn
@@ -611,7 +768,7 @@ mod tests {
         // every time; the following pull adopts the server's version instead.
         let conn = db();
         seed_note(&conn, "n1", 1);
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         let summary = apply_results(&conn, &batch, &[ok("kept", Some(99))]).expect("apply");
         assert_eq!(summary.kept, 1);
         assert_eq!(dirty_count(&conn), 0);
@@ -625,7 +782,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "n1", 1);
         state::set_cursor(&conn, 100).expect("cursor");
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply");
         assert_eq!(state::read(&conn).expect("state").last_cursor, 39);
     }
@@ -635,7 +792,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "n1", 1);
         state::set_cursor(&conn, 10).expect("cursor");
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply");
         assert_eq!(
             state::read(&conn).expect("state").last_cursor,
@@ -648,7 +805,7 @@ mod tests {
     fn rejected_stays_dirty_and_is_reported() {
         let conn = db();
         seed_note(&conn, "n1", 1);
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         let mut bad = ok("rejected", None);
         bad.error = Some("name in use".into());
         let summary = apply_results(&conn, &batch, &[bad]).expect("apply");
@@ -662,7 +819,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "n1", 0);
         store::delete_forever(&conn, "n1").expect("delete");
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         apply_results(&conn, &batch, &[ok("applied", Some(7))]).expect("apply");
         assert!(!has_pending(&conn).expect("pending"));
     }
@@ -673,7 +830,7 @@ mod tests {
         let conn = db();
         seed_note(&conn, "n1", 1);
         store::delete_forever(&conn, "n1").expect("delete");
-        let batch = collect(&conn, 100).expect("collect");
+        let batch = collect(&conn, 100, true).expect("collect");
         apply_results(&conn, &batch, &[ok("noop", None)]).expect("apply");
         assert!(!has_pending(&conn).expect("pending"));
     }
@@ -761,4 +918,162 @@ mod tests {
         conn.execute("UPDATE notes SET dirty = 0", []).unwrap();
         assert_eq!(pending_fingerprint(&conn).unwrap(), None);
     }
+
+    fn queued_upload(conn: &Connection, id: &str, note_id: &str, error: Option<&str>) {
+        conn.execute(
+            "INSERT INTO attachments (id, note_id, url, filename, mime, sha256, uploaded, upload_error)
+                  VALUES (?1, ?2, '/x', 'f.txt', 'text/plain', 'h', 0, ?3)",
+            params![id, note_id, error],
+        )
+        .expect("queued upload");
+    }
+
+    fn upload_ids(conn: &Connection) -> Vec<String> {
+        pending_uploads(conn)
+            .expect("uploads")
+            .into_iter()
+            .map(|u| u.id)
+            .collect()
+    }
+
+    #[test]
+    fn removals_of_files_and_previews_wait_for_a_server_that_takes_them() {
+        let conn = db();
+        store::record_pending_delete(&conn, "attachment", "a1").expect("tombstone");
+        store::record_pending_delete(&conn, "preview", "p1").expect("tombstone");
+        store::record_pending_delete(&conn, "note", "n9").expect("tombstone");
+
+        // An older server would answer "unknown entity" to each of them.
+        let older: Vec<_> = collect(&conn, 100, false)
+            .expect("collect")
+            .iter()
+            .map(|c| c.entity)
+            .collect();
+        assert_eq!(older, vec!["note"]);
+
+        let mut newer: Vec<_> = collect(&conn, 100, true)
+            .expect("collect")
+            .iter()
+            .map(|c| (c.entity, c.op))
+            .collect();
+        newer.sort();
+        assert_eq!(
+            newer,
+            vec![
+                ("attachment", "delete"),
+                ("note", "delete"),
+                ("preview", "delete")
+            ]
+        );
+    }
+
+    #[test]
+    fn a_removed_attachment_stays_removed_through_a_whole_cycle() {
+        use crate::sync::{pull, wire};
+        let conn = db();
+        let server_note = |revision: i64, attachments: Vec<wire::Attachment>| wire::Note {
+            id: "n1".into(),
+            body: "a note".into(),
+            position: 0,
+            pinned: false,
+            archived: false,
+            trashed: false,
+            deleted_at: None,
+            remind_at: None,
+            recurrence: None,
+            created_at: Some("2026-07-26T00:00:00.000Z".into()),
+            updated_at: Some("2026-07-26T00:00:00.000Z".into()),
+            sync_revision: revision,
+            purged_at: None,
+            labels: vec![],
+            attachments,
+            previews: vec![],
+        };
+        let page = |note: wire::Note, cursor: i64| wire::ChangesPage {
+            notes: vec![note],
+            labels: vec![],
+            cursor,
+            has_more: false,
+        };
+        let a1 = wire::Attachment {
+            id: "a1".into(),
+            url: "/x".into(),
+            filename: None,
+            mime: "image/png".into(),
+            size: None,
+            sha256: None,
+        };
+        pull::apply_page(&conn, &page(server_note(1, vec![a1]), 1)).expect("first pull");
+
+        store::delete_attachment(&conn, "n1", "a1").expect("remove");
+
+        // Push: the removal and the touched note go up, and the server applies both.
+        let batch = collect(&conn, 100, true).expect("collect");
+        let results: Vec<PushResult> = batch
+            .iter()
+            .map(|c| ok("applied", (c.entity == "note").then_some(5)))
+            .collect();
+        apply_results(&conn, &batch, &results).expect("results");
+        assert!(
+            !has_pending(&conn).expect("pending"),
+            "the tombstone is settled"
+        );
+
+        // Pull: the server's note now comes without it.
+        pull::apply_page(&conn, &page(server_note(6, vec![]), 6)).expect("second pull");
+        let left: i64 = conn
+            .query_row("SELECT COUNT(*) FROM attachments", [], |r| r.get(0))
+            .expect("count");
+        assert_eq!(left, 0);
+    }
+
+    #[test]
+    fn a_file_uploads_only_once_its_note_is_on_the_server() {
+        let conn = db();
+        seed_note(&conn, "landed", 0);
+        seed_note(&conn, "unsent", 1);
+        queued_upload(&conn, "a", "landed", None);
+        queued_upload(&conn, "b", "unsent", None);
+        queued_upload(&conn, "c", "landed", Some("file is too large (max 25 MB)"));
+        assert_eq!(upload_ids(&conn), vec!["a"]);
+    }
+
+    #[test]
+    fn a_refused_upload_leaves_the_queue_and_a_passing_failure_stays_in_it() {
+        let conn = db();
+        seed_note(&conn, "n", 0);
+        queued_upload(&conn, "big", "n", None);
+        queued_upload(&conn, "flaky", "n", None);
+        assert!(
+            pending_fingerprint(&conn).expect("fp").is_some(),
+            "uploads are pending work"
+        );
+
+        let refused = Err(UploadError::Refused("file is too large (max 25 MB)".into()));
+        settle_upload(&conn, "big", &refused).expect("settle");
+        settle_upload(&conn, "flaky", &Err(UploadError::Retry("offline".into()))).expect("settle");
+        assert_eq!(
+            upload_ids(&conn),
+            vec!["flaky"],
+            "the refusal is not resent every cycle"
+        );
+        assert!(has_pending(&conn).expect("pending"));
+
+        settle_upload(&conn, "flaky", &Ok(())).expect("settle");
+        assert!(upload_ids(&conn).is_empty());
+        assert!(!has_pending(&conn).expect("pending"));
+        assert_eq!(pending_fingerprint(&conn).expect("fp"), None);
+        let reason: Option<String> = conn
+            .query_row(
+                "SELECT upload_error FROM attachments WHERE id = 'big'",
+                [],
+                |r| r.get(0),
+            )
+            .expect("row");
+        assert_eq!(
+            reason.as_deref(),
+            Some("file is too large (max 25 MB)"),
+            "shown on the file"
+        );
+    }
 }
diff --git a/desktop/src-tauri/src/autosync.rs b/desktop/src-tauri/src/autosync.rs
index 61ca635..8539da7 100644
--- a/desktop/src-tauri/src/autosync.rs
+++ b/desktop/src-tauri/src/autosync.rs
@@ -270,18 +270,17 @@ fn worker(app: AppHandle, rx: Receiver<Request>) {
 
         let blobs = app.state::<BlobStore>();
         let started = Instant::now();
-        // A panic inside one cycle must not end automatic sync for the rest of the
-        // session: this thread is the only thing that runs it, and a dead thread
-        // looks exactly like a quiet one. It becomes a failed cycle instead.
-        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-            tauri::async_runtime::block_on(engine::run_cycle(
-                db.inner(),
-                blobs.inner(),
-                &base_url,
-                &token,
-            ))
-        }))
-        .unwrap_or_else(|_| Err("The sync cycle crashed; it will be retried.".to_string()));
+        // No catch_unwind here, and that is deliberate. The workspace's release profile
+        // sets `panic = "abort"`, so a panic in a cycle ends the whole app rather than
+        // this thread, and a catch would only ever work in a debug build. What
+        // matters is that the worker can't die silently and leave the app looking
+        // synced, and an abort is anything but silent.
+        let result = tauri::async_runtime::block_on(engine::run_cycle(
+            db.inner(),
+            blobs.inner(),
+            &base_url,
+            &token,
+        ));
         pace.last_attempt = Some(started);
         // Whatever is still pending after this cycle is not new on the next tick.
         pace.sent = pending(&db);
diff --git a/desktop/src-tauri/src/commands/local.rs b/desktop/src-tauri/src/commands/local.rs
index 2388069..a950131 100644
--- a/desktop/src-tauri/src/commands/local.rs
+++ b/desktop/src-tauri/src/commands/local.rs
@@ -10,6 +10,7 @@ use inkwell_core::local::models::*;
 use inkwell_core::local::retention;
 use inkwell_core::local::store;
 use inkwell_core::local::Db;
+use inkwell_core::sync::blobs::{self, BlobStore};
 use inkwell_core::sync::state;
 
 // A macro would hide the (very regular) locking; kept explicit so each command reads
@@ -120,6 +121,35 @@ pub fn notes_delete_item(id: String, item_id: String, db: State<'_, Db>) -> Resu
     store::delete_item(&conn, &id, &item_id).map_err(|e| e.to_string())
 }
 
+/// Attach a file to a note, online or not (#5168).
+///
+/// The file's bytes are the raw IPC body — a JSON number array would be several
+/// times the file's size — and its note, name and type ride in headers. Header values
+/// are ASCII-only, so the frontend percent-encodes the name. Async so hashing and
+/// writing a large file happens off the main thread, not while the window waits.
+#[tauri::command]
+pub async fn notes_add_attachment(
+    request: tauri::ipc::Request<'_>,
+    db: State<'_, Db>,
+    blobs: State<'_, BlobStore>,
+) -> Result<Note, String> {
+    let tauri::ipc::InvokeBody::Raw(bytes) = request.body() else {
+        return Err("The file's contents didn't arrive.".to_string());
+    };
+    let header = |name: &str| {
+        request
+            .headers()
+            .get(name)
+            .and_then(|v| v.to_str().ok())
+            .unwrap_or_default()
+    };
+    let note_id = header("x-note-id");
+    let filename = blobs::urldecode(header("x-filename"));
+    let mime = header("x-mime");
+    let conn = db.0.lock().map_err(|e| e.to_string())?;
+    store::add_attachment(&conn, &blobs, note_id, &filename, mime, bytes)
+}
+
 #[tauri::command]
 pub fn notes_delete_attachment(
     id: String,
diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs
index 8f7e1aa..5209468 100644
--- a/desktop/src-tauri/src/lib.rs
+++ b/desktop/src-tauri/src/lib.rs
@@ -177,6 +177,7 @@ pub fn run() {
             commands::local::notes_add_item,
             commands::local::notes_update_item,
             commands::local::notes_delete_item,
+            commands::local::notes_add_attachment,
             commands::local::notes_delete_attachment,
             commands::local::notes_delete_preview,
             commands::local::notes_reorder,
diff --git a/docs/sync.md b/docs/sync.md
index 9f6fc98..6c249fb 100644
--- a/docs/sync.md
+++ b/docs/sync.md
@@ -61,6 +61,9 @@ syncs everything else.
     falls back to the colour it derives locally, and one pushing `color` has the key
     ignored. The test is not "did a field leave" but "does either side end up
     showing something wrong".
+  - v5 (#5168): attachments became a sync entity — an upload route keyed by the
+    client's id, and `attachment`/`preview` deletes in push. Additive, so it is
+    the `attachment_sync` feature and the floor stays.
 - **Additive change** (a new field, a new capability) → add a `sync_features`
   name. Do **not** raise a minimum. Old clients keep working.
 - **Breaking change only** → raise `MIN_CLIENT_PROTOCOL_VERSION` (or the client's
@@ -222,6 +225,14 @@ Body: `{ "changes": [ ... ] }` (max 1000 per batch). Each change:
 - **Labels:** `{entity: "label", op: "upsert"|"delete", id, edited_at, name,
   color}`. A per-owner name clash on a *different* id is rejected (fix locally
   and retry).
+- **Attachments and link previews** (`attachment_sync`): `{entity:
+  "attachment"|"preview", op: "delete", id, edited_at}`. Delete is the only op —
+  attachments are created by upload (below) and previews by the server. A removal
+  is **not** weighed under last-write-wins: there is no rival version of a
+  removed file, so it always applies. A row the caller can't see answers `noop`,
+  the same as one that never existed. Removals are explicit rather than "the
+  note's current attachment set" on purpose: push runs before pull, so a set
+  would delete an attachment another device added that this one has not seen.
 
 ### Conflict resolution — last-write-wins + history
 
@@ -251,18 +262,28 @@ had a newer edit and the client should adopt the server version on its next pull
 Metadata rides the delta feed (`id, url, mime, size, sha256`); the bytes move
 over the existing routes:
 
-- **Upload:** `POST /api/notes/<note_id>/attachments` (multipart, field `file`;
-  optional field `id` to keep a client-minted attachment id). Re-uploading an id
-  the note already has is an idempotent no-op. Server stores + hashes the bytes.
+- **Sync upload** (`attachment_sync`): `PUT /api/sync/attachments/<id>?note_id=
+  &filename=&sha256=`, body = the raw bytes, type in `Content-Type`. For a file
+  attached offline, sent after the note itself has landed. `201` stores it under
+  the client's id; `200 {"status": "exists"}` if the note already holds that id
+  (a retry); `400` if the bytes don't hash to `sha256`; `409` if the id belongs to
+  another note; `413` over `max_attachment_mb`; `404` for a note the caller
+  doesn't own. A `4xx` other than `404` is permanent — the client records it on
+  the attachment and stops retrying.
+- **Web upload:** `POST /api/notes/<note_id>/attachments` (multipart, field
+  `file`; optional field `id`, same idempotency).
 - **Download:** `GET /api/notes/<note_id>/attachments/<id>` (owner/shared scoped).
 - The client uses `sha256` to skip blobs it already holds and to verify
-  integrity after download. (Currently image mimes only; broadening to any file
-  is tracked separately.)
+  integrity after download. Any file type syncs.
+- Every insert or delete of an attachment or a link preview bumps its note's
+  `sync_revision` (triggers from 0015 and 0031), so a change to either reaches
+  other devices as the note.
 
 ## A sync cycle
 
 1. **Push** local changes since the last sync (batched). Apply the per-item
-   results (mark synced, adopt server version where `kept`).
+   results (mark synced, adopt server version where `kept`). Then upload the bytes
+   of attachments added offline whose notes have now landed.
 2. **Pull** from the stored `since` cursor until `has_more` is false. Upsert
    notes/labels into the local store; delete rows whose `purged_at` is set;
    download any attachment blobs referenced by a new/changed `sha256`.
diff --git a/frontend/src/adapters/local.ts b/frontend/src/adapters/local.ts
index 64142a9..1bbb80c 100644
--- a/frontend/src/adapters/local.ts
+++ b/frontend/src/adapters/local.ts
@@ -4,8 +4,8 @@
 //
 // Argument keys are camelCase; Tauri converts them to the Rust commands' snake_case
 // parameters (e.g. labelIds -> label_ids). A few operations have no offline meaning
-// yet (account auth, device linking, attachment upload, URL unfurl, file import) —
-// those reject with a clear message rather than silently failing; the board, editor,
+// yet (account auth, device linking, URL unfurl, file import) — those reject with a
+// clear message rather than silently failing; the board, editor, attachments,
 // capture, filters, labels, checklists and reminders all work fully offline.
 
 import { invoke } from "../desktop/bridge";
@@ -58,7 +58,17 @@ export const local: Repo = {
     addItem: (id, text) => invoke<Note>("notes_add_item", { id, text }),
     updateItem: (id, itemId, changes) => invoke<Note>("notes_update_item", { id, itemId, changes }),
     deleteItem: (id, itemId) => invoke<Note>("notes_delete_item", { id, itemId }),
-    uploadAttachment: () => Promise.reject<Note>(new Error(NEEDS_SERVER)),
+    // Kept on this device and uploaded by the next sync once linked (#5168). The bytes
+    // go as the raw IPC body; the name is percent-encoded because a header carries
+    // only ASCII.
+    uploadAttachment: async (id, file) =>
+      invoke<Note>("notes_add_attachment", new Uint8Array(await file.arrayBuffer()), {
+        headers: {
+          "x-note-id": id,
+          "x-filename": encodeURIComponent(file.name),
+          "x-mime": file.type || "application/octet-stream",
+        },
+      }),
     deleteAttachment: (id, attId) => invoke<Note>("notes_delete_attachment", { id, attId }),
     unfurl: () => Promise.reject<Note>(new Error(NEEDS_SERVER)),
     deletePreview: (id, previewId) => invoke<Note>("notes_delete_preview", { id, previewId }),
diff --git a/frontend/src/api/errors.test.ts b/frontend/src/api/errors.test.ts
new file mode 100644
index 0000000..cc74358
--- /dev/null
+++ b/frontend/src/api/errors.test.ts
@@ -0,0 +1,28 @@
+import { describe, expect, it } from "vitest";
+
+import { errorMessage } from "./errors";
+
+describe("errorMessage", () => {
+  it("reads the REST client's error", () => {
+    expect(errorMessage({ error: "file is too large (max 25 MB)", status: 413 }, "x")).toBe(
+      "file is too large (max 25 MB)",
+    );
+  });
+
+  // A Tauri command rejects with its Rust error as a bare string.
+  it("reads a desktop command's string", () => {
+    expect(errorMessage("That note no longer exists.", "x")).toBe("That note no longer exists.");
+  });
+
+  it("reads a thrown Error", () => {
+    expect(errorMessage(new Error("Not running in the desktop app."), "x")).toBe(
+      "Not running in the desktop app.",
+    );
+  });
+
+  it("falls back when there is nothing to say", () => {
+    expect(errorMessage(undefined, "Could not upload file.")).toBe("Could not upload file.");
+    expect(errorMessage("  ", "fallback")).toBe("fallback");
+    expect(errorMessage({ status: 500 }, "fallback")).toBe("fallback");
+  });
+});
diff --git a/frontend/src/api/errors.ts b/frontend/src/api/errors.ts
new file mode 100644
index 0000000..aafcd1b
--- /dev/null
+++ b/frontend/src/api/errors.ts
@@ -0,0 +1,16 @@
+// What went wrong, in words to show a person, from whichever layer failed.
+//
+// The two backends reject differently: the REST client with an `ApiError`
+// (`{ error, status }`), a Tauri command with the plain string its Rust `Err` held.
+// A catch that reads only `.error` turns every desktop failure into its generic
+// fallback — "Could not upload file." when the store said exactly why.
+
+export function errorMessage(e: unknown, fallback: string): string {
+  if (typeof e === "string" && e.trim()) return e;
+  if (e && typeof e === "object") {
+    const { error, message } = e as { error?: unknown; message?: unknown };
+    if (typeof error === "string" && error.trim()) return error;
+    if (typeof message === "string" && message.trim()) return message;
+  }
+  return fallback;
+}
diff --git a/frontend/src/components/NoteCard.vue b/frontend/src/components/NoteCard.vue
index f906d9f..5836313 100644
--- a/frontend/src/components/NoteCard.vue
+++ b/frontend/src/components/NoteCard.vue
@@ -2,6 +2,7 @@
 import { computed, ref, watch } from "vue";
 import { useNotesStore } from "../stores/notes";
 import { NOTE_CARD_SURFACE, labelChipClasses } from "../notes/colors";
+import { rendersInline } from "../notes/attachments";
 import type { Note } from "../stores/notes";
 import Icon from "./Icon.vue";
 import LinkPreview from "./LinkPreview.vue";
@@ -43,8 +44,8 @@ const trashCountdown = computed(() => formatTrashCountdown(trashDays.value));
 const trashUrgent = computed(() => trashDays.value !== null && trashDays.value <= 3);
 
 // The card previews the first image inline; non-image files show as compact chips.
-const firstImage = computed(() => props.note.attachments.find((a) => a.mime.startsWith("image/")));
-const otherAttachments = computed(() => props.note.attachments.filter((a) => !a.mime.startsWith("image/")));
+const firstImage = computed(() => props.note.attachments.find((a) => rendersInline(a.mime)));
+const otherAttachments = computed(() => props.note.attachments.filter((a) => !rendersInline(a.mime)));
 
 // How much of a note the CARD shows. Android has always clamped to 8
 // (`MAX_PREVIEW_LINES`); the web rendered the whole body, so one long note could
diff --git a/frontend/src/components/NoteEditor.vue b/frontend/src/components/NoteEditor.vue
index 835ab93..7341fc0 100644
--- a/frontend/src/components/NoteEditor.vue
+++ b/frontend/src/components/NoteEditor.vue
@@ -1,6 +1,8 @@
 <script setup lang="ts">
 import { computed, nextTick, onBeforeUnmount, onMounted, ref, watch } from "vue";
 import { useNotesStore } from "../stores/notes";
+import { errorMessage } from "../api/errors";
+import { rendersInline } from "../notes/attachments";
 import Icon from "./Icon.vue";
 import LabelPicker from "./LabelPicker.vue";
 import LinkPreview from "./LinkPreview.vue";
@@ -446,11 +448,12 @@ function addChecklist(): void {
 }
 
 // ---- attachments ----
+const refusedUploads = computed(() => liveNote.value.attachments.filter((a) => a.upload_error));
 function pickFile() {
   fileInput.value?.click();
 }
 function attKind(mime: string): "image" | "audio" | "file" {
-  if (mime.startsWith("image/")) return "image";
+  if (rendersInline(mime)) return "image";
   if (mime.startsWith("audio/")) return "audio";
   return "file";
 }
@@ -467,7 +470,7 @@ async function uploadFile(file: File) {
   try {
     await notes.uploadAttachment(id, file);
   } catch (e) {
-    uploadError.value = (e as { error?: string }).error ?? "Could not upload file.";
+    uploadError.value = errorMessage(e, "Could not upload file.");
   }
 }
 
@@ -611,6 +614,16 @@ function revPreview(rev: NoteRevision): string {
           </template>
         </div>
         <p v-if="uploadError" class="text-xs text-red-600 dark:text-red-400">{{ uploadError }}</p>
+        <!-- Desktop: a file attached here that the server refused. It isn't retried,
+             so without this line it would sit on this device unsynced and unexplained. -->
+        <p
+          v-for="att in refusedUploads"
+          :key="`refused-${att.id}`"
+          class="text-xs text-amber-600 dark:text-amber-400"
+        >
+          {{ att.filename || "A file" }} won't sync: {{ att.upload_error }}. It stays on this
+          device until you remove it.
+        </p>
 
         <!-- Fetched automatically after each save; removable here and nowhere else. -->
         <div v-if="liveNote.previews.length" class="flex flex-col gap-2">
diff --git a/frontend/src/desktop/bridge.ts b/frontend/src/desktop/bridge.ts
index 1255c9f..cd8e2e9 100644
--- a/frontend/src/desktop/bridge.ts
+++ b/frontend/src/desktop/bridge.ts
@@ -4,7 +4,13 @@
 // @tauri-apps/api dependency and the web bundle is unaffected.
 
 interface TauriGlobal {
-  core: { invoke: <T>(cmd: string, args?: Record<string, unknown>) => Promise<T> };
+  core: {
+    invoke: <T>(
+      cmd: string,
+      args?: Record<string, unknown> | Uint8Array,
+      options?: { headers?: Record<string, string> },
+    ) => Promise<T>;
+  };
   // Also from `withGlobalTauri`. Needed because quick capture puts the app in TWO
   // windows, each with its own Pinia stores — a note saved in one is invisible to
   // the other until something says so, and an event is the only channel between
@@ -31,10 +37,16 @@ export function isDesktop(): boolean {
   return typeof window !== "undefined" && !!window.__TAURI__;
 }
 
-export function invoke<T>(cmd: string, args?: Record<string, unknown>): Promise<T> {
+/** Call a desktop command. `args` is either named arguments or, for a command that
+ *  takes a file, its raw bytes — with anything else it needs in `options.headers`. */
+export function invoke<T>(
+  cmd: string,
+  args?: Record<string, unknown> | Uint8Array,
+  options?: { headers?: Record<string, string> },
+): Promise<T> {
   const tauri = window.__TAURI__;
   if (!tauri) return Promise.reject(new Error("Not running in the desktop app."));
-  return tauri.core.invoke<T>(cmd, args).catch((err: unknown) => {
+  return tauri.core.invoke<T>(cmd, args, options).catch((err: unknown) => {
     // Every failed command names itself in the log — so a broken "basic function"
     // in some environment is diagnosable. Skip log_event to avoid recursion.
     if (cmd !== "log_event") logEvent("error", `invoke '${cmd}' failed: ${String(err)}`);
@@ -111,6 +123,10 @@ export interface PushSummary {
   noop: number;
   rejected: number;
   errors: string[];
+  /** Files attached on this device that reached the server this cycle. */
+  uploaded: number;
+  /** Files that didn't; their reasons are in `errors`. */
+  upload_failed: number;
 }
 
 export interface PullSummary {
diff --git a/frontend/src/notes/attachments.ts b/frontend/src/notes/attachments.ts
new file mode 100644
index 0000000..ccb0942
--- /dev/null
+++ b/frontend/src/notes/attachments.ts
@@ -0,0 +1,11 @@
+// How an attachment is drawn, decided in one place for the card and the editor.
+
+/** Whether a file is shown as a picture rather than as a chip to download.
+ *
+ *  Every `image/` type except SVG. An SVG is a document that can carry its own
+ *  script, so neither surface serves one to render: the server sends it as a download
+ *  (#1981) and the desktop's blob scheme as opaque bytes, where an `<img>` of it would
+ *  only ever show broken. */
+export function rendersInline(mime: string): boolean {
+  return mime.startsWith("image/") && mime !== "image/svg+xml";
+}
diff --git a/frontend/src/stores/notes.ts b/frontend/src/stores/notes.ts
index b847115..67daa4b 100644
--- a/frontend/src/stores/notes.ts
+++ b/frontend/src/stores/notes.ts
@@ -43,6 +43,9 @@ export interface Attachment {
   mime: string;
   size?: number;
   sha256?: string | null;
+  // Desktop only: why the server refused a file attached on this device. The file
+  // stays here, unsynced, until it is removed.
+  upload_error?: string;
 }
 
 // A cached OpenGraph/meta preview for a URL in the note (server-fetched, SSRF-guarded).
diff --git a/frontend/src/views/SyncView.vue b/frontend/src/views/SyncView.vue
index 1bca2e5..7896332 100644
--- a/frontend/src/views/SyncView.vue
+++ b/frontend/src/views/SyncView.vue
@@ -233,6 +233,8 @@ function showOutcome(outcome: SyncOutcome) {
   const blobs = outcome.pull.blobs_downloaded;
   const parts: string[] = [];
   if (sent > 0) parts.push(`sent ${sent}`);
+  const up = outcome.push.uploaded;
+  if (up > 0) parts.push(`uploaded ${up} file${up === 1 ? "" : "s"}`);
   if (received > 0) parts.push(`received ${received}`);
   if (blobs > 0) parts.push(`${blobs} attachment${blobs === 1 ? "" : "s"}`);
   lastResult.value = parts.length ? `Synced — ${parts.join(", ")}.` : "Already up to date.";
@@ -244,10 +246,12 @@ function showOutcome(outcome: SyncOutcome) {
   // Rejections are the server refusing a specific change — surfaced, never
   // swallowed, because only the person can resolve them. The background cycle does
   // not resend them on its own (autosync.rs), so this stays until they're fixed.
-  syncError.value =
-    outcome.push.rejected > 0
-      ? `${outcome.push.rejected} change(s) the server wouldn't accept: ${outcome.push.errors.join("; ")}`
-      : "";
+  // A file refused for good is listed once, here, and not retried; the editor says so
+  // on the file itself from then on.
+  const problems: string[] = [];
+  if (outcome.push.rejected > 0) problems.push(`${outcome.push.rejected} change(s) the server wouldn't accept`);
+  if (outcome.push.upload_failed > 0) problems.push(`${outcome.push.upload_failed} file(s) didn't upload`);
+  syncError.value = problems.length ? `${problems.join(" and ")}: ${outcome.push.errors.join("; ")}` : "";
 }
 
 async function syncNow() {
diff --git a/src/inkwell/notes/__init__.py b/src/inkwell/notes/__init__.py
index 87d0b2a..13e04be 100644
--- a/src/inkwell/notes/__init__.py
+++ b/src/inkwell/notes/__init__.py
@@ -10,11 +10,9 @@ External callers import from `inkwell.notes` (see `__all__`); those names are
 re-exported here so the package is a drop-in replacement for the old module."""
 from __future__ import annotations
 
-import hashlib
 import io
 import json
 import os
-import uuid
 import zipfile
 from datetime import datetime, timedelta, timezone
 
@@ -58,6 +56,8 @@ from .helpers import (
     apply_filter,
     derive_display_title,
     is_empty_note,
+    store_attachment,
+    unlink_media,
     parse_list_items,
 )
 from .import_export import (
@@ -624,44 +624,25 @@ async def upload_attachment(note_id: str):
         upload = files.get("file")
         if upload is None:
             return json_error("no file provided", 400)
-        # Any file type is allowed — images render inline, everything else downloads.
-        mime = (upload.content_type or "application/octet-stream").split(";")[0].strip().lower()
-        filename = _safe_filename(upload.filename)
-        ext = _attachment_ext(filename, mime)
-        # A native client may supply the attachment's id so an offline-attached file
-        # keeps its identity across sync. Re-uploading an id it already has is a no-op.
+        # A client may supply the attachment's id so a file keeps one identity on
+        # every device. Re-sending an id the note already has is a no-op.
         form = await request.form
-        raw_id = (form.get("id") or "").strip()
-        att_id = uuid.uuid4()
-        if raw_id:
-            parsed_att = parse_uuid(raw_id)
-            if parsed_att is None:
+        att_id = None
+        if raw_id := (form.get("id") or "").strip():
+            att_id = parse_uuid(raw_id)
+            if att_id is None:
                 return json_error("invalid attachment id", 400)
-            att_id = parsed_att
-            existing = await db.scalar(
-                select(NoteAttachment).where(NoteAttachment.id == att_id, NoteAttachment.note_id == note.id)
-            )
-            if existing is not None:
+            existing = await db.scalar(select(NoteAttachment.note_id).where(NoteAttachment.id == att_id))
+            if existing == note.id:
                 return jsonify(await _serialize_note(db, note))  # already have this blob
+            if existing is not None:
+                return json_error("attachment id already in use", 409)
         raw = upload.stream.read()
         max_mb = int(await get_setting(db, "max_attachment_mb"))
         if len(raw) > max_mb * 1024 * 1024:
             return json_error(f"file is too large (max {max_mb} MB)", 413)
-        rel = os.path.join(str(note.id), f"{att_id}{ext}")
-        dest = Config.media_root() / rel
-        dest.parent.mkdir(parents=True, exist_ok=True)
-        dest.write_bytes(raw)
-        db.add(
-            NoteAttachment(
-                id=att_id,
-                note_id=note.id,
-                path=rel,
-                filename=filename,
-                mime=mime,
-                size=len(raw),
-                sha256=hashlib.sha256(raw).hexdigest(),
-            )
-        )
+        # Any file type is allowed — images render inline, everything else downloads.
+        store_attachment(db, note, raw, upload.filename, upload.content_type, att_id)
         await db.commit()
         return jsonify(await _serialize_note(db, note)), 201
 
@@ -712,14 +693,11 @@ async def delete_attachment(note_id: str, att_id: str):
         att = await db.scalar(select(NoteAttachment).where(NoteAttachment.id == aid, NoteAttachment.note_id == note.id))
         if att is None:
             return not_found()
-        file_path = Config.media_root() / att.path
+        rel = att.path
         await db.delete(att)
         await db.commit()
         result = await _serialize_note(db, note)
-    try:
-        file_path.unlink(missing_ok=True)
-    except OSError:
-        pass
+    unlink_media(rel)
     return jsonify(result)
 
 
diff --git a/src/inkwell/notes/helpers.py b/src/inkwell/notes/helpers.py
index c197a19..9254de4 100644
--- a/src/inkwell/notes/helpers.py
+++ b/src/inkwell/notes/helpers.py
@@ -1,19 +1,27 @@
 """Small shared helpers + constants for the notes package: display-name derivation,
-board-filter narrowing, the owner-scoped fetch, and filename/slug sanitizers used by
-both the attachment routes and the importer."""
+board-filter narrowing, the owner-scoped fetch, filename/slug sanitizers, and the one
+place an attachment's bytes are written — shared by the upload route, the importer
+and sync."""
 from __future__ import annotations
 
 from .checklist import strip_marker
 
+import hashlib
+import logging
 import os
 import re
+import uuid
 
 from quart import g
 from sqlalchemy import select
 
+from ..config import Config
 from ..models.note import Note
+from ..models.note_attachment import NoteAttachment
 from ..responses import parse_uuid
 
+logger = logging.getLogger(__name__)
+
 ALLOWED_IMAGE_MIMES = {"image/png": ".png", "image/jpeg": ".jpg", "image/gif": ".gif", "image/webp": ".webp"}
 
 VALID_FILTERS = {"active", "archived", "trash"}
@@ -112,6 +120,50 @@ def _attachment_ext(filename: str, mime: str) -> str:
     return ALLOWED_IMAGE_MIMES.get(mime, "")
 
 
+def normalize_mime(raw: str | None) -> str:
+    """A declared content type reduced to its bare, lowercase media type."""
+    return (raw or "application/octet-stream").split(";")[0].strip().lower() or "application/octet-stream"
+
+
+def store_attachment(
+    db, note: Note, raw: bytes, filename: str | None, mime: str | None, att_id: uuid.UUID | None = None
+) -> NoteAttachment:
+    """Write one attachment's bytes under the media root and add its row (not committed).
+
+    The single writer for the upload route, the importer and sync, so all three
+    sanitize the name, pick the extension and record the hash the same way. A sync
+    client passes `att_id` so a file attached offline keeps the id it was given there.
+    """
+    filename = _safe_filename(filename)
+    mime = normalize_mime(mime)
+    att_id = att_id or uuid.uuid4()
+    rel = os.path.join(str(note.id), f"{att_id}{_attachment_ext(filename, mime)}")
+    dest = Config.media_root() / rel
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    dest.write_bytes(raw)
+    row = NoteAttachment(
+        id=att_id,
+        note_id=note.id,
+        path=rel,
+        filename=filename,
+        mime=mime,
+        size=len(raw),
+        sha256=hashlib.sha256(raw).hexdigest(),
+    )
+    db.add(row)
+    return row
+
+
+def unlink_media(rel: str) -> None:
+    """Remove an attachment's file, after its row is gone. A file that is already
+    missing or can't be removed is logged, never raised: the row was what the person
+    asked to delete, and failing here would undo a deletion that has committed."""
+    try:
+        (Config.media_root() / rel).unlink(missing_ok=True)
+    except OSError:
+        logger.warning("couldn't remove attachment file %s", rel, exc_info=True)
+
+
 def _header_filename(name: str) -> str:
     """Sanitize a filename for a Content-Disposition header (drop quotes/newlines)."""
     return re.sub(r'[\r\n"]', "", name or "")[:255] or "file"
diff --git a/src/inkwell/notes/import_export.py b/src/inkwell/notes/import_export.py
index 5b0c3e1..a36a99d 100644
--- a/src/inkwell/notes/import_export.py
+++ b/src/inkwell/notes/import_export.py
@@ -4,27 +4,21 @@ and materializes notes — with a decompression-size budget so an import zip bom
 exhaust memory/disk."""
 from __future__ import annotations
 
-import hashlib
 import json
-import os
 import posixpath
-import uuid
 import zipfile
 from datetime import datetime, timezone
 
 from sqlalchemy import select
 
 from ..common import parse_dt
-from ..config import Config
 from ..models.label import NoteLabel
 from ..models.note import Note
-from ..models.note_attachment import NoteAttachment
 from .helpers import (
     ALLOWED_IMAGE_MIMES,
-    _attachment_ext,
-    _safe_filename,
     derive_display_title,
     is_empty_note,
+    store_attachment,
 )
 from .checklist import append_item
 from .tags import _find_or_create_label, _lift_and_reconcile_tags
@@ -231,25 +225,7 @@ def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict, budget: _
         raw = budget.read(zf, zpath)
     except KeyError:
         return False
-    filename = _safe_filename(posixpath.basename(zpath))
-    mime = (att.get("mime") or "application/octet-stream").split(";")[0].strip().lower()
-    ext = _attachment_ext(filename, mime)
-    att_id = uuid.uuid4()
-    rel = os.path.join(str(note.id), f"{att_id}{ext}")
-    dest = Config.media_root() / rel
-    dest.parent.mkdir(parents=True, exist_ok=True)
-    dest.write_bytes(raw)
-    db.add(
-        NoteAttachment(
-            id=att_id,
-            note_id=note.id,
-            path=rel,
-            filename=filename,
-            mime=mime,
-            size=len(raw),
-            sha256=hashlib.sha256(raw).hexdigest(),
-        )
-    )
+    store_attachment(db, note, raw, posixpath.basename(zpath), att.get("mime"))
     return True
 
 
diff --git a/src/inkwell/sync.py b/src/inkwell/sync.py
index 67f1d14..c5b0568 100644
--- a/src/inkwell/sync.py
+++ b/src/inkwell/sync.py
@@ -11,6 +11,7 @@ from where it left off (since=0 = full initial sync).
 """
 from __future__ import annotations
 
+import hashlib
 import uuid
 from datetime import datetime, timezone
 
@@ -24,12 +25,17 @@ from .db import session_scope
 from .labeling import reconcile_manual_labels, resolve_owned_label_ids
 from .models.label import Label, NoteLabel
 from .models.note import Note
+from .models.note_attachment import NoteAttachment
+from .models.note_link_preview import NoteLinkPreview
 from .notes import (
     _serialize_notes,
     normalize_color,
     normalize_recurrence,
 )
 from .notes.body import write_body
+from .notes.helpers import store_attachment, unlink_media
+from .responses import json_error, not_found, parse_uuid
+from .settings import get_setting
 from .retention import purge_note
 from .serialize import serialize_label_sync
 from .unfurl_queue import schedule as schedule_unfurls
@@ -68,7 +74,10 @@ MAX_PUSH = 1000  # per-batch change cap
 # `_assign_note_fields` reads its payload key by key and never validates the shape.
 # Neither direction errors and neither loses anything visible. `title` was the note's
 # NAME; this is a field that no longer renders anywhere.
-SYNC_PROTOCOL_VERSION = 4
+# v5 (#5168): attachments became a sync entity — `PUT /attachments/<id>` uploads a file
+# attached offline, and push takes `attachment` and `preview` deletes. Additive, so the
+# floor stays: a v4 client never sends either, and gets the same notes it always did.
+SYNC_PROTOCOL_VERSION = 5
 MIN_CLIENT_PROTOCOL_VERSION = 3
 
 # Named capabilities beyond the base protocol. An ADDITIVE change earns a name
@@ -82,6 +91,7 @@ SYNC_FEATURES: tuple[str, ...] = (
     "attachments",  # blob upload/download, deduped by sha256
     "tombstones",  # purge propagates as a content-less row
     "revisions",  # an overwritten version snapshots into note history
+    "attachment_sync",  # upload by client id + attachment/preview deletes in push (v5)
 )
 
 
@@ -373,6 +383,43 @@ async def _apply_label(db, ch: dict) -> dict:
     }
 
 
+# Entities a push may only DELETE. Each is a note's child, so deleting one bumps its
+# note's revision (the parent-bump trigger), and the next pull carries the note
+# without it to every other device.
+_CHILD_ENTITIES = {"attachment": NoteAttachment, "preview": NoteLinkPreview}
+
+
+async def _apply_child_delete(db, ch: dict, removed_files: list[str]) -> dict:
+    """Delete one attachment or link preview the client removed.
+
+    Not subject to last-write-wins. LWW settles which of two versions of the SAME
+    thing survives; a removal has no rival version — nothing can re-create an id once
+    it is gone — so it always applies. That is what makes a deletion stick: a removal
+    that lost to a newer edit of the note would come straight back on the next pull.
+
+    A row this caller cannot see answers `noop`, exactly like one that never existed,
+    so the reply cannot be used to learn whether someone else's id is real.
+    """
+    entity = ch.get("entity")
+    raw_id = ch.get("id")
+    child_id = parse_uuid(str(raw_id)) if raw_id is not None else None
+    if child_id is None:
+        return {"id": raw_id, "entity": entity, "status": "rejected", "error": "invalid id"}
+    if ch.get("op") != "delete":
+        return {"id": str(child_id), "entity": entity, "status": "rejected", "error": "only delete is supported"}
+    model = _CHILD_ENTITIES[entity]
+    row = await db.scalar(
+        select(model).join(Note, Note.id == model.note_id).where(model.id == child_id, Note.owner_id == g.user_id)
+    )
+    if row is None:
+        return {"id": str(child_id), "entity": entity, "status": "noop"}
+    if isinstance(row, NoteAttachment):
+        removed_files.append(row.path)
+    await db.delete(row)
+    await db.flush()
+    return {"id": str(child_id), "entity": entity, "status": "applied"}
+
+
 @bp.post("/push")
 @login_required
 async def push():
@@ -387,6 +434,7 @@ async def push():
 
     results = []
     previews: list[tuple[uuid.UUID, str]] = []
+    removed_files: list[str] = []
     async with session_scope() as db:
         for ch in changes:
             if not isinstance(ch, dict):
@@ -397,6 +445,8 @@ async def push():
                 results.append(await _apply_note(db, ch, previews))
             elif entity == "label":
                 results.append(await _apply_label(db, ch))
+            elif entity in _CHILD_ENTITIES:
+                results.append(await _apply_child_delete(db, ch, removed_files))
             else:
                 results.append({"id": ch.get("id"), "status": "rejected", "error": "unknown entity"})
         await db.commit()
@@ -406,4 +456,47 @@ async def push():
         # than awaited — a push must not wait on somebody else's website.
         for note_id, text in previews:
             schedule_unfurls(note_id, text)
+    # Files go once their rows have committed, so a failed batch never leaves a row
+    # pointing at bytes that were already removed.
+    for rel in removed_files:
+        unlink_media(rel)
     return jsonify({"results": results})
+
+
+@bp.put("/attachments/<att_id>")
+@login_required
+async def put_attachment(att_id: str):
+    """Upload a file a client attached while offline, under the id it chose there.
+
+    The body is the file's raw bytes; `note_id`, `filename` and `sha256` ride in the
+    query and the type in Content-Type. Raw rather than multipart so a native client
+    needs no form encoder for what is one blob and three strings.
+
+    Idempotent: re-sending an id the note already holds answers 200 and changes
+    nothing, which is what lets a client retry an upload whose reply it never got.
+    The bytes must hash to `sha256`, so a truncated transfer is refused instead of
+    stored and served to every other device as the real file.
+    """
+    aid = parse_uuid(att_id)
+    nid = parse_uuid(request.args.get("note_id") or "")
+    expected = (request.args.get("sha256") or "").strip().lower()
+    if aid is None or nid is None:
+        return not_found()
+    async with session_scope() as db:
+        note = await db.scalar(select(Note).where(Note.id == nid, Note.owner_id == g.user_id))
+        if note is None:
+            return not_found()
+        existing = await db.scalar(select(NoteAttachment.note_id).where(NoteAttachment.id == aid))
+        if existing == note.id:
+            return jsonify({"id": str(aid), "status": "exists"})
+        if existing is not None:
+            return json_error("attachment id already in use", 409)
+        max_mb = int(await get_setting(db, "max_attachment_mb"))
+        raw = await request.get_data()
+        if len(raw) > max_mb * 1024 * 1024:
+            return json_error(f"file is too large (max {max_mb} MB)", 413)
+        if hashlib.sha256(raw).hexdigest() != expected:
+            return json_error("the file's bytes don't match its sha256", 400)
+        store_attachment(db, note, raw, request.args.get("filename"), request.content_type, aid)
+        await db.commit()
+    return jsonify({"id": str(aid), "status": "created"}), 201
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 8b3435b..a9d8921 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -15,18 +15,21 @@ deployment has ever seen.
 """
 from __future__ import annotations
 
+import hashlib
 import uuid
 from datetime import datetime, timedelta, timezone
 
 import pytest
 import pytest_asyncio
-from sqlalchemy import select, text
+from sqlalchemy import func, select, text
 
 from inkwell import ratelimit
 from inkwell.app import create_app
+from inkwell.config import Config
 from inkwell.db import dispose_engine, session_scope
 from inkwell.models.label import NoteLabel
 from inkwell.models.note import Note
+from inkwell.models.note_attachment import NoteAttachment
 from inkwell.models.user import User
 from inkwell.notes.tags import _lift_and_reconcile_tags
 from inkwell.settings import get_setting, live, refresh_live, reset_live, set_settings
@@ -737,3 +740,137 @@ async def test_a_pushed_edit_keeps_the_clients_edit_time_when_a_tag_is_lifted(ap
     )
     note = await (await app_client.get(f"/api/notes/{nid}")).get_json()
     assert datetime.fromisoformat(note["updated_at"]) == datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+# --- attachments as a sync entity (#5168) ---------------------------------------
+
+
+async def _note_revision(app_client, nid: str) -> int:
+    feed = await (await app_client.get("/api/sync/changes?since=0")).get_json()
+    return next(n["sync_revision"] for n in feed["notes"] if n["id"] == nid)
+
+
+async def _pushed_note(app_client, body: str = "with a file") -> str:
+    nid = str(uuid.uuid4())
+    resp = await app_client.post(
+        "/api/sync/push",
+        json={"changes": [{"entity": "note", "id": nid, "op": "upsert", "body": body,
+                           "edited_at": "2026-01-01T00:00:00Z", "created_at": "2026-01-01T00:00:00Z"}]},
+    )
+    assert (await resp.get_json())["results"][0]["status"] == "created"
+    return nid
+
+
+async def _put(app_client, aid: str, nid: str, raw: bytes, sha: str | None = None, name: str = "scan.pdf"):
+    return await app_client.put(
+        f"/api/sync/attachments/{aid}",
+        data=raw,
+        headers={"Content-Type": "application/pdf"},
+        query_string={"note_id": nid, "filename": name, "sha256": sha or hashlib.sha256(raw).hexdigest()},
+    )
+
+
+async def test_an_offline_attachment_uploads_once_under_its_own_id(app_client, db):
+    await _signed_in(app_client, "upload")
+    nid = await _pushed_note(app_client)
+    aid = str(uuid.uuid4())
+
+    assert (await _put(app_client, aid, nid, b"%PDF-1", sha="0" * 64)).status_code == 400, "hash mismatch refused"
+    assert (await app_client.get(f"/api/notes/{nid}")).status_code == 200
+    assert (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"] == []
+
+    before = await _note_revision(app_client, nid)
+    first = await _put(app_client, aid, nid, b"%PDF-1")
+    assert first.status_code == 201
+    assert await _note_revision(app_client, nid) > before, "other devices hear about it"
+
+    again = await _put(app_client, aid, nid, b"%PDF-1")
+    assert again.status_code == 200 and (await again.get_json())["status"] == "exists", "a retried upload is a no-op"
+
+    atts = (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"]
+    assert [(a["id"], a["filename"], a["mime"], a["sha256"]) for a in atts] == [
+        (aid, "scan.pdf", "application/pdf", hashlib.sha256(b"%PDF-1").hexdigest())
+    ]
+    other = await _pushed_note(app_client, "another note")
+    assert (await _put(app_client, aid, other, b"%PDF-1")).status_code == 409, "one id, one note"
+
+
+async def test_an_upload_to_a_note_the_caller_does_not_own_is_not_found(app_client, db):
+    # Signed in first: registration is open only to the first account.
+    await _signed_in(app_client, "intruder")
+    stranger = User(email="stranger@example.test", display_name="Stranger")
+    db.add(stranger)
+    await db.flush()
+    theirs = Note(owner_id=stranger.id, body="theirs", display_title="theirs")
+    db.add(theirs)
+    await db.commit()
+
+    resp = await _put(app_client, str(uuid.uuid4()), str(theirs.id), b"x")
+    assert resp.status_code == 404
+
+
+async def test_a_pushed_attachment_delete_removes_the_row_the_file_and_bumps_the_note(app_client, db):
+    await _signed_in(app_client, "remove")
+    nid = await _pushed_note(app_client)
+    aid = str(uuid.uuid4())
+    await _put(app_client, aid, nid, b"bytes")
+    stored = (await db.scalar(select(NoteAttachment).where(NoteAttachment.id == uuid.UUID(aid)))).path
+    assert (Config.media_root() / stored).is_file()
+
+    before = await _note_revision(app_client, nid)
+    resp = await app_client.post(
+        "/api/sync/push",
+        json={"changes": [{"entity": "attachment", "id": aid, "op": "delete", "edited_at": "2000-01-01T00:00:00Z"}]},
+    )
+    assert (await resp.get_json())["results"] == [{"id": aid, "entity": "attachment", "status": "applied"}]
+    assert (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"] == []
+    assert not (Config.media_root() / stored).exists()
+    # The edit time above is older than the note's: a removal is not a version
+    # competing under last-write-wins, so it applies anyway.
+    assert await _note_revision(app_client, nid) > before, "the note re-syncs without it"
+
+
+async def test_a_child_delete_cannot_reach_another_owners_rows(app_client, db):
+    await _signed_in(app_client, "prober")
+    stranger = User(email="other@example.test", display_name="Other")
+    db.add(stranger)
+    await db.flush()
+    theirs = Note(owner_id=stranger.id, body="theirs", display_title="theirs")
+    db.add(theirs)
+    await db.flush()
+    att = NoteAttachment(note_id=theirs.id, path="x/y.bin", mime="application/octet-stream", size=1)
+    preview = NoteLinkPreview(note_id=theirs.id, url="https://example.com/")
+    db.add_all([att, preview])
+    await db.commit()
+
+    resp = await app_client.post(
+        "/api/sync/push",
+        json={"changes": [
+            {"entity": "attachment", "id": str(att.id), "op": "delete", "edited_at": "2030-01-01T00:00:00Z"},
+            {"entity": "preview", "id": str(preview.id), "op": "delete", "edited_at": "2030-01-01T00:00:00Z"},
+            {"entity": "preview", "id": str(uuid.uuid4()), "op": "upsert", "edited_at": "2030-01-01T00:00:00Z"},
+        ]},
+    )
+    statuses = [r["status"] for r in (await resp.get_json())["results"]]
+    # Someone else's row answers exactly like a missing one: nothing to learn here.
+    assert statuses == ["noop", "noop", "rejected"]
+    assert await db.scalar(select(func.count()).select_from(NoteAttachment)) == 1
+    assert await db.scalar(select(func.count()).select_from(NoteLinkPreview)) == 1
+
+
+async def test_a_preview_arriving_or_leaving_moves_its_note_in_the_feed(app_client, db):
+    """0031: before it, a preview fetched after the save, or dismissed on the web,
+    never reached a device that had already pulled the note."""
+    await _signed_in(app_client, "previews")
+    nid = await _pushed_note(app_client, "read https://example.com/a")
+
+    before = await _note_revision(app_client, nid)
+    async with session_scope() as other:  # as the background unfurl does
+        other.add(NoteLinkPreview(note_id=uuid.UUID(nid), url="https://example.com/a", title="A"))
+        await other.commit()
+    arrived = await _note_revision(app_client, nid)
+    assert arrived > before
+
+    preview_id = (await (await app_client.get(f"/api/notes/{nid}")).get_json())["previews"][0]["id"]
+    await app_client.delete(f"/api/notes/{nid}/previews/{preview_id}")
+    assert await _note_revision(app_client, nid) > arrived
diff --git a/tests/test_sync.py b/tests/test_sync.py
index aed688f..9ca0c37 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -34,6 +34,12 @@ async def test_push_requires_auth(app):
     assert resp.status_code == 401
 
 
+async def test_attachment_upload_requires_auth(app):
+    client = app.test_client()
+    resp = await client.put("/api/sync/attachments/00000000-0000-0000-0000-000000000000", data=b"x")
+    assert resp.status_code == 401
+
+
 def test_client_wins():
     older = datetime(2026, 7, 20, tzinfo=timezone.utc)
     newer = datetime(2026, 7, 22, tzinfo=timezone.utc)