core: extract the store and sync engine into a shared crate (M12 step 1)

Android becomes a native Kotlin client over this same code (Scribe note 2730), so
the local store and sync engine stop being modules of the desktop app and become
`thoughtsync-core`, a crate with no UI framework in it at all.

This is a move, not a rewrite, and the measurement is why: every file in local/
and sync/ already carried ZERO Tauri references — 4,980 of 6,372 lines. The
coupling was 473 lines of command shim, which stays behind in the desktop crate
as src/commands/. Kept as git renames so history follows the files.

The desktop imports them under their old names (`use thoughtsync_core::{local,
sync}`) so every call site reads exactly as before. What moved is where they
live, not what they are.

Two things a workspace changes that are easy to miss, both caught before pushing:

[profile.release] now lives at the workspace ROOT. Cargo silently ignores
profiles declared by a non-root member — leaving it in the desktop crate would
have dropped lto/strip/opt-level from every release build with only a warning.

And a workspace shares ONE target dir, so the bundles moved from
desktop/src-tauri/target to target/. Thirteen references across publish-release,
debundle-graphics, verify.sh, package-prebuilt and the workflow now point there.
Pinning target-dir back would have been the smaller diff, but the Android lane
also produces Rust artifacts and they do not belong under desktop/.

Also retires the Tauri Android lane in the same push rather than leaving a path
that is being replaced: gen/android, android.yml and docs/android-dev.md are
gone, the mobile_entry_point attribute with them, and the lib drops to rlib —
staticlib/cdylib existed for Tauri mobile, and the .so Android loads will be
built from the core crate instead. Rule 22, no parallel path.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

core/src/sync/pull.rs

@@ -0,0 +1,840 @@
+//! Pull: bring a server's changes into the local store (M10.7b).
+//!
+//! The feed is a single monotonic sequence shared by notes and labels, so one
+//! integer cursor is a total-order watermark over both (docs/sync.md). We loop pages
+//! until the server says there are no more, persisting the cursor **in the same
+//! transaction** as the page it describes — a cursor committed ahead of its data
+//! would silently skip those rows forever, which reads as a clean sync.
+
+use chrono::{SecondsFormat, Utc};
+use rusqlite::{params, Connection, OptionalExtension};
+use serde::Serialize;
+
+use super::blobs::BlobStore;
+use super::client;
+use super::state;
+use super::wire;
+use crate::local::Db;
+
+/// Backstop against a server that never stops saying `has_more`. At the server's
+/// 1000-row page cap this is 10M rows — far past any real store, so hitting it means
+/// something is wrong, not that someone has a lot of notes.
+const MAX_PAGES: usize = 10_000;
+
+/// What a pull did — for the UI, and for the log when something looks off.
+#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
+pub struct PullSummary {
+    pub pages: usize,
+    pub notes_applied: usize,
+    pub notes_deleted: usize,
+    pub labels_applied: usize,
+    pub labels_deleted: usize,
+    pub cursor: i64,
+    /// Rows that still held unpushed local edits when the server's version landed on
+    /// top. Should be 0 in the normal cycle, because push runs first; anything higher
+    /// means local work was overwritten, which is worth saying out loud.
+    pub clobbered_dirty: usize,
+    pub blobs_downloaded: usize,
+    /// Attachments whose bytes couldn't be fetched or failed verification. Counted
+    /// rather than fatal — see `download_missing_blobs`.
+    pub blobs_failed: usize,
+}
+
+impl PullSummary {
+    fn absorb(&mut self, other: PullSummary) {
+        self.pages += other.pages;
+        self.notes_applied += other.notes_applied;
+        self.notes_deleted += other.notes_deleted;
+        self.labels_applied += other.labels_applied;
+        self.labels_deleted += other.labels_deleted;
+        self.clobbered_dirty += other.clobbered_dirty;
+        self.blobs_downloaded += other.blobs_downloaded;
+        self.blobs_failed += other.blobs_failed;
+        self.cursor = other.cursor;
+    }
+}
+
+/// `(note_id, attachment_id, sha256)` for every attachment that advertises a hash.
+/// The caller filters against the blob store — which blobs we hold isn't a SQL
+/// question.
+pub fn hashed_attachments(conn: &Connection) -> rusqlite::Result<Vec<(String, String, String)>> {
+    let mut stmt = conn.prepare(
+        "SELECT note_id, id, sha256 FROM attachments
+          WHERE sha256 IS NOT NULL AND sha256 <> ''",
+    )?;
+    let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
+    rows.collect()
+}
+
+/// Fetch the bytes for any attachment we have metadata for but no blob.
+///
+/// A failed attachment NEVER fails the sync. Notes are the primary data and they've
+/// already landed; an image that didn't arrive is retried on the next cycle simply
+/// because its blob still counts as missing. Aborting here would mean one unreachable
+/// file could block every future sync.
+async fn download_missing_blobs(
+    db: &Db,
+    blobs: &BlobStore,
+    base_url: &str,
+    token: &str,
+) -> Result<(usize, usize), String> {
+    let wanted = {
+        let conn = db.0.lock().map_err(|e| e.to_string())?;
+        hashed_attachments(&conn).map_err(|e| e.to_string())?
+    };
+
+    let mut downloaded = 0;
+    let mut failed = 0;
+    for (note_id, attachment_id, sha256) in wanted {
+        // Content-addressed, so this skips blobs we already hold — including the same
+        // image attached to a different note.
+        if blobs.has(&sha256) {
+            continue;
+        }
+        match client::fetch_attachment(base_url, token, &note_id, &attachment_id).await {
+            Ok(bytes) => match blobs.store(&sha256, &bytes) {
+                Ok(_) => downloaded += 1,
+                Err(e) => {
+                    log::warn!("attachment {attachment_id}: {e}");
+                    failed += 1;
+                }
+            },
+            Err(e) => {
+                log::warn!("attachment {attachment_id}: {e}");
+                failed += 1;
+            }
+        }
+    }
+    Ok((downloaded, failed))
+}
+
+fn now() -> String {
+    Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
+}
+
+/// Apply one page and advance the cursor, atomically.
+///
+/// Labels are applied before notes so a membership never references a label row that
+/// doesn't exist yet.
+pub fn apply_page(conn: &Connection, page: &wire::ChangesPage) -> rusqlite::Result<PullSummary> {
+    let tx = conn.unchecked_transaction()?;
+    let mut summary = PullSummary {
+        pages: 1,
+        cursor: page.cursor,
+        ..Default::default()
+    };
+
+    for label in &page.labels {
+        if label.is_tombstone() {
+            tx.execute("DELETE FROM labels WHERE id = ?1", params![label.id])?;
+            summary.labels_deleted += 1;
+        } else {
+            upsert_label(&tx, label)?;
+            summary.labels_applied += 1;
+        }
+    }
+
+    for note in &page.notes {
+        if note.is_tombstone() {
+            // A purge tombstone carries no content — its only job is to say "delete
+            // your copy". Children go with it via ON DELETE CASCADE.
+            tx.execute("DELETE FROM notes WHERE id = ?1", params![note.id])?;
+            summary.notes_deleted += 1;
+            continue;
+        }
+        if is_dirty(&tx, &note.id)? {
+            summary.clobbered_dirty += 1;
+        }
+        upsert_note(&tx, note)?;
+        summary.notes_applied += 1;
+    }
+
+    state::set_cursor(&tx, page.cursor)?;
+    tx.commit()?;
+    Ok(summary)
+}
+
+fn is_dirty(conn: &Connection, note_id: &str) -> rusqlite::Result<bool> {
+    let dirty: Option<i64> = conn
+        .query_row(
+            "SELECT dirty FROM notes WHERE id = ?1",
+            params![note_id],
+            |r| r.get(0),
+        )
+        .optional()?;
+    Ok(dirty == Some(1))
+}
+
+fn upsert_label(conn: &Connection, label: &wire::Label) -> rusqlite::Result<()> {
+    // One label per name is enforced on both sides (locally a UNIQUE index on
+    // lower(name); on the server, per owner). A label created offline can therefore
+    // collide with one the server already had under a different id — "work" typed on
+    // this machine and "work" that already existed.
+    //
+    // The server's row wins, but its MEMBERSHIPS have to survive the swap. Just
+    // deleting the local duplicate would cascade its note_labels away, stripping the
+    // label off notes that this pull never even mentions — silent loss that no later
+    // page would repair. So: free the name, insert the server's row, re-point the
+    // memberships onto it, then drop the husk.
+    let duplicates: Vec<String> = {
+        let mut stmt =
+            conn.prepare("SELECT id FROM labels WHERE lower(name) = lower(?1) AND id <> ?2")?;
+        let rows = stmt.query_map(params![label.name, label.id], |r| r.get::<_, String>(0))?;
+        rows.collect::<rusqlite::Result<Vec<String>>>()?
+    };
+    // Renaming first is what makes the insert possible at all — the unique index
+    // would otherwise reject the server's row before anything could be merged.
+    for old in &duplicates {
+        conn.execute(
+            "UPDATE labels SET name = name || ' (superseded ' || id || ')' WHERE id = ?1",
+            params![old],
+        )?;
+    }
+
+    let created = label.created_at.clone().unwrap_or_else(now);
+    conn.execute(
+        "INSERT INTO labels (id, name, color, created_at, updated_at, sync_revision, dirty)
+              VALUES (?1, ?2, ?3, ?4, ?4, ?5, 0)
+         ON CONFLICT(id) DO UPDATE SET
+              name          = excluded.name,
+              color         = excluded.color,
+              sync_revision = excluded.sync_revision,
+              dirty         = 0",
+        params![
+            label.id,
+            label.name,
+            label.color,
+            created,
+            label.sync_revision
+        ],
+    )?;
+
+    for old in &duplicates {
+        // OR IGNORE guards a (note_id, label_id) collision. Today the unique index on
+        // lower(name) makes that unreachable — two same-name labels can't coexist
+        // locally — so this is belt-and-braces against that index changing, not a
+        // case we've seen. Anything it skips cascades away with the husk below, which
+        // is correct: those are duplicates of a membership that now exists.
+        conn.execute(
+            "UPDATE OR IGNORE note_labels SET label_id = ?1 WHERE label_id = ?2",
+            params![label.id, old],
+        )?;
+        conn.execute("DELETE FROM labels WHERE id = ?1", params![old])?;
+    }
+    Ok(())
+}
+
+fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    let created = note.created_at.clone().unwrap_or_else(now);
+    let updated = note.updated_at.clone().unwrap_or_else(|| created.clone());
+    // The server's `deleted_at` is the authority on trash AGE. Taking it from the feed
+    // rather than stamping "now" locally is what keeps a note trashed three weeks ago
+    // from looking brand-new to a device that only just heard about it — otherwise
+    // every fresh install would silently reset the whole retention clock. Falls back
+    // to the note's updated_at only if an older server omits the field.
+    let trashed_at = if note.trashed {
+        note.deleted_at.clone().or_else(|| Some(updated.clone()))
+    } else {
+        None
+    };
+    // `created_at` is deliberately absent from the UPDATE clause: a note's birth time
+    // never changes, and the server's copy is the same value anyway.
+    conn.execute(
+        "INSERT INTO notes (id, title, body, color, kind, position, pinned, archived,
+                            trashed, remind_at, recurrence, created_at, updated_at,
+                            sync_revision, trashed_at, dirty)
+              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, 0)
+         ON CONFLICT(id) DO UPDATE SET
+              title         = excluded.title,
+              body          = excluded.body,
+              color         = excluded.color,
+              kind          = excluded.kind,
+              position      = excluded.position,
+              pinned        = excluded.pinned,
+              archived      = excluded.archived,
+              trashed       = excluded.trashed,
+              remind_at     = excluded.remind_at,
+              recurrence    = excluded.recurrence,
+              updated_at    = excluded.updated_at,
+              sync_revision = excluded.sync_revision,
+              trashed_at    = excluded.trashed_at,
+              dirty         = 0",
+        params![
+            note.id,
+            note.title,
+            note.body,
+            note.color,
+            note.kind,
+            note.position,
+            note.pinned,
+            note.archived,
+            note.trashed,
+            note.remind_at,
+            note.recurrence,
+            created,
+            updated,
+            note.sync_revision,
+            trashed_at,
+        ],
+    )?;
+
+    // Children are replaced wholesale: a delta carries the note's FULL current state,
+    // so "what the server sent" IS the complete set. Diffing would be more code and
+    // could leave behind a row the server no longer has.
+    replace_items(conn, note)?;
+    replace_attachments(conn, note)?;
+    replace_previews(conn, note)?;
+    replace_labels(conn, note)?;
+    Ok(())
+}
+
+fn replace_items(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    conn.execute(
+        "DELETE FROM checklist_items WHERE note_id = ?1",
+        params![note.id],
+    )?;
+    for (index, item) in note.items.iter().enumerate() {
+        conn.execute(
+            "INSERT INTO checklist_items (id, note_id, text, checked, position)
+                  VALUES (?1, ?2, ?3, ?4, ?5)",
+            params![
+                item.id,
+                note.id,
+                item.text,
+                item.checked,
+                position_of(item.position, index)
+            ],
+        )?;
+    }
+    Ok(())
+}
+
+fn replace_attachments(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    conn.execute(
+        "DELETE FROM attachments WHERE note_id = ?1",
+        params![note.id],
+    )?;
+    for (index, att) in note.attachments.iter().enumerate() {
+        // The feed carries no explicit position for attachments — they arrive in
+        // creation order, so the index preserves it.
+        conn.execute(
+            "INSERT INTO attachments (id, note_id, url, filename, mime, size, sha256, position)
+                  VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+            params![
+                att.id,
+                note.id,
+                att.url,
+                att.filename,
+                att.mime,
+                att.size,
+                att.sha256,
+                index as i64
+            ],
+        )?;
+    }
+    Ok(())
+}
+
+fn replace_previews(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    conn.execute(
+        "DELETE FROM link_previews WHERE note_id = ?1",
+        params![note.id],
+    )?;
+    for (index, preview) in note.previews.iter().enumerate() {
+        conn.execute(
+            "INSERT INTO link_previews (id, note_id, url, title, description, image_url,
+                                        site_name, position)
+                  VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+            params![
+                preview.id,
+                note.id,
+                preview.url,
+                preview.title,
+                preview.description,
+                preview.image_url,
+                preview.site_name,
+                index as i64
+            ],
+        )?;
+    }
+    Ok(())
+}
+
+fn replace_labels(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
+    conn.execute(
+        "DELETE FROM note_labels WHERE note_id = ?1",
+        params![note.id],
+    )?;
+    for label in &note.labels {
+        ensure_label_stub(conn, label)?;
+        // `via_tag` is applied verbatim rather than re-derived from the body. The
+        // server already reconciled tags when it saved the note, and re-deriving here
+        // would call the local find-or-create path, which marks new labels dirty and
+        // would push them straight back — sync churn out of nothing.
+        conn.execute(
+            "INSERT OR IGNORE INTO note_labels (note_id, label_id, via_tag)
+                  VALUES (?1, ?2, ?3)",
+            params![note.id, label.id, label.via_tag],
+        )?;
+    }
+    Ok(())
+}
+
+/// Materialize a label referenced by a note, if we don't have it yet.
+///
+/// Notes and labels page from one shared sequence, so a note can reference a label
+/// whose own delta landed in an earlier page — or, right at a page boundary, hasn't
+/// landed. The note carries enough of the label to create it, so a membership never
+/// fails on a missing row. `OR IGNORE` because the label's real delta (later in this
+/// page or a future one) is the authority on its name and color.
+fn ensure_label_stub(conn: &Connection, label: &wire::NoteLabel) -> rusqlite::Result<()> {
+    let ts = now();
+    conn.execute(
+        "INSERT OR IGNORE INTO labels (id, name, color, created_at, updated_at, dirty)
+              VALUES (?1, ?2, ?3, ?4, ?4, 0)",
+        params![label.id, label.name, label.color, ts],
+    )?;
+    Ok(())
+}
+
+/// Trust an explicit position; fall back to arrival order when the server sent 0 for
+/// everything (which is what an unordered list looks like on the wire).
+fn position_of(explicit: i64, index: usize) -> i64 {
+    if explicit > 0 {
+        explicit
+    } else {
+        index as i64
+    }
+}
+
+/// Loop the feed to exhaustion, starting from the persisted cursor.
+///
+/// NOTE ON ORDERING: the full cycle is push-then-pull (docs/sync.md). Running this
+/// against a store with unpushed edits lets the server's version land on top of them
+/// — counted as `clobbered_dirty` and logged, rather than hidden.
+pub async fn run(
+    db: &Db,
+    blobs: &BlobStore,
+    base_url: &str,
+    token: &str,
+) -> Result<PullSummary, String> {
+    let mut total = PullSummary::default();
+
+    loop {
+        let since = {
+            let conn = db.0.lock().map_err(|e| e.to_string())?;
+            state::read(&conn).map_err(|e| e.to_string())?.last_cursor
+        };
+
+        let page = client::fetch_changes(base_url, token, since).await?;
+
+        // Trust the data over the flag: a server that claims more pages without
+        // advancing the cursor would spin this loop forever.
+        if page.has_more && page.cursor <= since {
+            return Err(format!(
+                "The server reported more changes but its cursor didn't advance past \
+                 {since}. Stopping rather than looping forever."
+            ));
+        }
+
+        let has_more = page.has_more;
+        let applied = {
+            let conn = db.0.lock().map_err(|e| e.to_string())?;
+            apply_page(&conn, &page).map_err(|e| e.to_string())?
+        };
+        total.absorb(applied);
+
+        if !has_more {
+            break;
+        }
+        if total.pages >= MAX_PAGES {
+            return Err(format!(
+                "Stopped after {MAX_PAGES} pages without reaching the end of the \
+                 server's changes. Something is wrong with the feed."
+            ));
+        }
+    }
+
+    // Notes first, bytes after: the metadata is what makes the attachments knowable,
+    // and knowing one is missing is what lets the next cycle retry it.
+    let (downloaded, failed) = download_missing_blobs(db, blobs, base_url, token).await?;
+    total.blobs_downloaded = downloaded;
+    total.blobs_failed = failed;
+
+    if total.clobbered_dirty > 0 {
+        log::warn!(
+            "pull overwrote {} note(s) that still had unpushed local edits",
+            total.clobbered_dirty
+        );
+    }
+    if total.blobs_failed > 0 {
+        log::warn!(
+            "pull: {} attachment(s) couldn't be downloaded; will retry next sync",
+            total.blobs_failed
+        );
+    }
+    log::info!(
+        "pull complete: {} page(s), {} note(s) applied, {} deleted, {} label(s) applied, cursor {}",
+        total.pages,
+        total.notes_applied,
+        total.notes_deleted,
+        total.labels_applied,
+        total.cursor
+    );
+    Ok(total)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::local::schema;
+
+    fn db() -> Connection {
+        let conn = Connection::open_in_memory().expect("in-memory db");
+        schema::migrate(&conn).expect("migrate");
+        conn
+    }
+
+    fn note(id: &str, revision: i64) -> wire::Note {
+        wire::Note {
+            id: id.to_string(),
+            title: Some("Title".into()),
+            body: "Body".into(),
+            color: "default".into(),
+            kind: "text".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![],
+            items: vec![],
+            attachments: vec![],
+            previews: vec![],
+        }
+    }
+
+    fn page(notes: Vec<wire::Note>, labels: Vec<wire::Label>, cursor: i64) -> wire::ChangesPage {
+        wire::ChangesPage {
+            notes,
+            labels,
+            cursor,
+            has_more: false,
+        }
+    }
+
+    fn count(conn: &Connection, sql: &str) -> i64 {
+        conn.query_row(sql, [], |r| r.get(0)).expect("count")
+    }
+
+    fn trash_stamp(conn: &Connection, id: &str) -> Option<String> {
+        let sql = "SELECT trashed_at FROM notes WHERE id = ?1";
+        conn.query_row(sql, [id], |r| r.get(0)).expect("stamp")
+    }
+
+    #[test]
+    fn applies_a_note_and_advances_the_cursor() {
+        let conn = db();
+        let summary = apply_page(&conn, &page(vec![note("n1", 7)], vec![], 7)).expect("apply");
+        assert_eq!(summary.notes_applied, 1);
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 1);
+        assert_eq!(state::read(&conn).expect("state").last_cursor, 7);
+    }
+
+    #[test]
+    fn pulled_rows_are_not_dirty() {
+        // They came FROM the server, so pushing them back would be pure churn.
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        assert_eq!(count(&conn, "SELECT dirty FROM notes WHERE id = 'n1'"), 0);
+    }
+
+    #[test]
+    fn tombstone_deletes_the_local_note() {
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        let mut dead = note("n1", 2);
+        dead.purged_at = Some("2026-07-26T01:00:00.000Z".into());
+        let summary = apply_page(&conn, &page(vec![dead], vec![], 2)).expect("apply");
+        assert_eq!(summary.notes_deleted, 1);
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 0);
+    }
+
+    #[test]
+    fn trashed_is_not_a_tombstone() {
+        // `trashed` is ordinary state that keeps syncing; only `purged_at` deletes.
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 1);
+        assert_eq!(count(&conn, "SELECT trashed FROM notes WHERE id = 'n1'"), 1);
+    }
+
+    #[test]
+    fn trash_age_comes_from_the_server_not_from_now() {
+        // The retention countdown runs off this timestamp. Stamping it locally would
+        // hand every note a fresh 30 days on any device that syncs it for the first
+        // time — a note trashed last month would never expire anywhere.
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = Some("2026-06-01T09:30:00+00:00".into());
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        let stamped = trash_stamp(&conn, "n1");
+        assert_eq!(stamped.as_deref(), Some("2026-06-01T09:30:00+00:00"));
+    }
+
+    #[test]
+    fn restoring_a_note_server_side_clears_its_trash_stamp() {
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = Some("2026-06-01T09:30:00+00:00".into());
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        apply_page(&conn, &page(vec![note("n1", 2)], vec![], 2)).expect("apply");
+        let stamped = trash_stamp(&conn, "n1");
+        assert_eq!(stamped, None, "an untrashed note keeps no trash stamp");
+    }
+
+    #[test]
+    fn an_older_server_without_deleted_at_still_ages_the_trash() {
+        // Falls back to updated_at rather than leaving the stamp null, which would
+        // make the note un-expirable and its countdown blank.
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = None;
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        let stamped = trash_stamp(&conn, "n1");
+        assert_eq!(stamped.as_deref(), Some("2026-07-26T00:00:00.000Z"));
+    }
+
+    #[test]
+    fn children_are_replaced_not_merged() {
+        let conn = db();
+        let mut first = note("n1", 1);
+        first.items = vec![
+            wire::Item {
+                id: "i1".into(),
+                text: "one".into(),
+                checked: false,
+                position: 0,
+            },
+            wire::Item {
+                id: "i2".into(),
+                text: "two".into(),
+                checked: false,
+                position: 1,
+            },
+        ];
+        apply_page(&conn, &page(vec![first], vec![], 1)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM checklist_items"), 2);
+
+        // The server dropped an item; the local copy must drop it too.
+        let mut second = note("n1", 2);
+        second.items = vec![wire::Item {
+            id: "i1".into(),
+            text: "one".into(),
+            checked: true,
+            position: 0,
+        }];
+        apply_page(&conn, &page(vec![second], vec![], 2)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM checklist_items"), 1);
+    }
+
+    #[test]
+    fn note_label_membership_materializes_a_missing_label() {
+        // The label's own delta may have landed in an earlier page, or not yet.
+        let conn = db();
+        let mut n = note("n1", 1);
+        n.labels = vec![wire::NoteLabel {
+            id: "l1".into(),
+            name: "work".into(),
+            color: "blue".into(),
+            via_tag: true,
+        }];
+        apply_page(&conn, &page(vec![n], vec![], 1)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1);
+        assert_eq!(
+            count(
+                &conn,
+                "SELECT via_tag FROM note_labels WHERE note_id = 'n1'"
+            ),
+            1,
+            "via_tag is applied verbatim, not re-derived"
+        );
+    }
+
+    #[test]
+    fn server_label_replaces_a_local_duplicate_by_name() {
+        let conn = db();
+        conn.execute(
+            "INSERT INTO labels (id, name, color, created_at, updated_at, dirty)
+                  VALUES ('local-id', 'Work', 'default', '2026-01-01', '2026-01-01', 1)",
+            [],
+        )
+        .expect("seed local label");
+
+        let server = wire::Label {
+            id: "server-id".into(),
+            name: "work".into(),
+            color: "blue".into(),
+            sync_revision: 5,
+            purged_at: None,
+            created_at: Some("2026-07-26T00:00:00.000Z".into()),
+        };
+        apply_page(&conn, &page(vec![], vec![server], 5)).expect("apply");
+
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1);
+        let id: String = conn
+            .query_row("SELECT id FROM labels", [], |r| r.get(0))
+            .expect("label");
+        assert_eq!(id, "server-id", "the server's row wins on pull");
+    }
+
+    #[test]
+    fn merging_a_duplicate_label_keeps_its_note_memberships() {
+        // The notes carrying the local label may not be in this page at all, so a
+        // plain delete would strip the label off them with nothing to repair it.
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("seed note");
+        conn.execute(
+            "INSERT INTO labels (id, name, color, created_at, updated_at, dirty)
+                  VALUES ('local-id', 'Work', 'default', '2026-01-01', '2026-01-01', 1)",
+            [],
+        )
+        .expect("seed local label");
+        conn.execute(
+            "INSERT INTO note_labels (note_id, label_id, via_tag)
+                  VALUES ('n1', 'local-id', 0)",
+            [],
+        )
+        .expect("seed membership");
+
+        let server = wire::Label {
+            id: "server-id".into(),
+            name: "work".into(),
+            color: "blue".into(),
+            sync_revision: 5,
+            purged_at: None,
+            created_at: None,
+        };
+        apply_page(&conn, &page(vec![], vec![server], 5)).expect("apply");
+
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1);
+        let label_id: String = conn
+            .query_row(
+                "SELECT label_id FROM note_labels WHERE note_id = 'n1'",
+                [],
+                |r| r.get(0),
+            )
+            .expect("membership survived");
+        assert_eq!(label_id, "server-id", "membership re-pointed, not dropped");
+    }
+
+    #[test]
+    fn label_tombstone_deletes_and_cascades_memberships() {
+        let conn = db();
+        let mut n = note("n1", 1);
+        n.labels = vec![wire::NoteLabel {
+            id: "l1".into(),
+            name: "work".into(),
+            color: "blue".into(),
+            via_tag: false,
+        }];
+        apply_page(&conn, &page(vec![n], vec![], 1)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM note_labels"), 1);
+
+        let dead = wire::Label {
+            id: "l1".into(),
+            name: "work".into(),
+            color: "blue".into(),
+            sync_revision: 2,
+            purged_at: Some("2026-07-26T01:00:00.000Z".into()),
+            created_at: None,
+        };
+        apply_page(&conn, &page(vec![], vec![dead], 2)).expect("apply");
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 0);
+        assert_eq!(
+            count(&conn, "SELECT COUNT(*) FROM note_labels"),
+            0,
+            "membership should cascade with the label"
+        );
+    }
+
+    #[test]
+    fn overwriting_a_dirty_note_is_counted() {
+        let conn = db();
+        conn.execute(
+            "INSERT INTO notes (id, body, created_at, updated_at, dirty)
+                  VALUES ('n1', 'local edit', '2026-01-01', '2026-01-01', 1)",
+            [],
+        )
+        .expect("seed dirty note");
+        let summary = apply_page(&conn, &page(vec![note("n1", 9)], vec![], 9)).expect("apply");
+        assert_eq!(summary.clobbered_dirty, 1);
+    }
+
+    #[test]
+    fn applying_a_fresh_note_reports_no_clobber() {
+        let conn = db();
+        let summary = apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        assert_eq!(summary.clobbered_dirty, 0);
+    }
+
+    #[test]
+    fn empty_page_still_advances_the_cursor() {
+        // The server can page past rows that were trimmed to the shared watermark.
+        let conn = db();
+        apply_page(&conn, &page(vec![], vec![], 42)).expect("apply");
+        assert_eq!(state::read(&conn).expect("state").last_cursor, 42);
+    }
+
+    #[test]
+    fn note_upsert_preserves_the_original_created_at() {
+        let conn = db();
+        apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply");
+        let mut later = note("n1", 2);
+        later.created_at = Some("2099-01-01T00:00:00.000Z".into());
+        apply_page(&conn, &page(vec![later], vec![], 2)).expect("apply");
+        let created: String = conn
+            .query_row("SELECT created_at FROM notes WHERE id = 'n1'", [], |r| {
+                r.get(0)
+            })
+            .expect("created_at");
+        assert_eq!(created, "2026-07-26T00:00:00.000Z");
+    }
+
+    #[test]
+    fn a_page_that_fails_leaves_the_cursor_untouched() {
+        // Atomicity is the whole resumability story: a cursor committed ahead of its
+        // data would skip those rows forever. Force a failure with a duplicate
+        // checklist-item id inside one page.
+        let conn = db();
+        let mut n = note("n1", 3);
+        n.items = vec![
+            wire::Item {
+                id: "dup".into(),
+                text: "one".into(),
+                checked: false,
+                position: 0,
+            },
+            wire::Item {
+                id: "dup".into(),
+                text: "two".into(),
+                checked: false,
+                position: 1,
+            },
+        ];
+        assert!(apply_page(&conn, &page(vec![n], vec![], 3)).is_err());
+        assert_eq!(state::read(&conn).expect("state").last_cursor, 0);
+        assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 0);
+    }
+}