M10.7b: pull the change feed into the local store (task 2105)

Server -> local. sync/wire.rs mirrors the delta-feed JSON exactly as
notes/serialize.py sends it; sync/pull.rs applies it.

ATOMICITY IS THE POINT. The cursor is written in the SAME transaction as the
page it describes. A cursor committed ahead of its data would skip those rows
forever while reporting a clean sync — the worst kind of failure, because
nothing looks wrong. A test forces a mid-page failure and asserts the cursor
stayed put.

Every degradation leans toward re-downloading rather than skipping: an
unparseable cursor means full sync, wire fields are all defaulted so a newer
server adding a field (or an older one omitting one) yields a partial note
instead of a rejected page, and a page that fails rolls back whole.

Labels are applied before notes so a membership never references a row that
doesn't exist. A note also carries enough of its labels to materialize them,
because notes and labels page from ONE shared sequence and a note can arrive
referencing a label whose own delta landed in an earlier page.

via_tag is applied verbatim rather than re-deriving #tags from the body. The
server already reconciled them on save, and re-deriving would go through the
local find-or-create path, which marks new labels dirty — pushing them
straight back. Sync churn manufactured out of nothing.

Duplicate-label merge, the subtle one: a label created offline can collide by
name with one the server already had under a different id. Both sides enforce
one label per name, so the server's row has to win — but simply deleting the
local duplicate would CASCADE its note_labels away, stripping the label off
notes this pull never mentions, with no later page to repair it. So we free
the name, insert the server's row, re-point the memberships, then drop the
husk. Tested.

Children (items/attachments/previews/labels) are replaced wholesale rather
than diffed: a delta carries the note's FULL state, so what arrived IS the
complete set, and diffing could strand a row the server no longer has.

The loop trusts the data over the flag — a server claiming has_more without
advancing its cursor stops with an error instead of spinning forever.

Pull can overwrite a row with unpushed local edits. The documented cycle is
push-then-pull (M10.7c), so that should never happen; when it does it's
counted as clobbered_dirty and logged rather than hidden.

17 tests, all against an in-memory database.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SreJkbxB4gx8pPsu8QbLPi

desktop/src-tauri/src/lib.rs

@@ -93,6 +93,7 @@ pub fn run() {
             sync::commands::sync_link,
             sync::commands::sync_unlink,
             sync::commands::sync_status,
+            sync::commands::sync_pull,
         ])
         .run(tauri::generate_context!())
         .expect("error while running the ThoughtSync desktop app");

desktop/src-tauri/src/sync/client.rs

@@ -14,6 +14,7 @@ use reqwest::{RequestBuilder, StatusCode};
 use serde::{Deserialize, Serialize};
 
 use super::compat::{self, Compatibility, ServerInfo};
+use super::wire;
 
 /// Timeout for the short request/response calls in this module. Kept tight because a
 /// user is watching a button while they run, and the most common mistake — a wrong
@@ -21,6 +22,15 @@ use super::compat::{self, Compatibility, ServerInfo};
 /// just look frozen. The sync engine's bulk transfers will need their own, longer one.
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// Bulk transfers get much longer: a first full sync can be thousands of notes, and
+/// failing one at ten seconds would make a large store impossible to ever pull.
+const SYNC_TIMEOUT: Duration = Duration::from_secs(120);
+
+/// 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 \
+                              revoked. Unlink and link again to issue a new one.";
+
 /// What the link UI needs after a handshake: where we ended up (the normalized URL,
 /// which may differ from what was typed), who answered, and whether we can work
 /// with them.
@@ -48,13 +58,17 @@ struct DeviceLoginResponse {
     user: Identity,
 }
 
-fn http() -> Result<reqwest::Client, String> {
+fn http_with(timeout: Duration) -> Result<reqwest::Client, String> {
     reqwest::Client::builder()
-        .timeout(REQUEST_TIMEOUT)
+        .timeout(timeout)
         .build()
         .map_err(|e| format!("Could not start the network client: {e}"))
 }
 
+fn http() -> Result<reqwest::Client, String> {
+    http_with(REQUEST_TIMEOUT)
+}
+
 /// Attach the client-identity headers every request carries, plus a bearer token
 /// when we hold one.
 fn prepare(builder: RequestBuilder, token: Option<&str>) -> RequestBuilder {
@@ -179,6 +193,36 @@ pub async fn fetch_identity(base_url: &str, token: &str) -> Result<Identity, Str
         .map_err(|_| format!("{base_url} accepted the token but sent an unexpected reply."))
 }
 
+/// Fetch one page of the change feed, starting after `since`.
+///
+/// The caller loops until `has_more` is false (see `pull::run`); paging lives there
+/// rather than here so the transport stays a single request/response.
+pub async fn fetch_changes(
+    base_url: &str,
+    token: &str,
+    since: i64,
+) -> Result<wire::ChangesPage, String> {
+    let url = format!("{base_url}/api/sync/changes?since={since}");
+    let request = prepare(http_with(SYNC_TIMEOUT)?.get(url), Some(token));
+    let response = request
+        .send()
+        .await
+        .map_err(|e| describe_transport_error(base_url, &e))?;
+
+    let status = response.status();
+    if status == StatusCode::UNAUTHORIZED {
+        return Err(TOKEN_REJECTED.to_string());
+    }
+    if !status.is_success() {
+        return Err(unexpected_status(base_url, status));
+    }
+
+    response
+        .json()
+        .await
+        .map_err(|e| format!("Couldn't read the change feed from {base_url}: {e}"))
+}
+
 /// The public, unauthenticated endpoint carrying the handshake.
 fn config_url(base_url: &str) -> String {
     format!("{base_url}/api/config")
@@ -196,10 +240,12 @@ fn me_url(base_url: &str) -> String {
 /// Display is accurate but reads like a stack trace.
 fn describe_transport_error(base_url: &str, err: &reqwest::Error) -> String {
     if err.is_timeout() {
+        // No specific duration here: these calls run under two different budgets
+        // (interactive vs bulk sync), and naming the wrong one is worse than naming
+        // none.
         format!(
-            "{base_url} didn't respond within {} seconds. It may be offline, or \
-             unreachable from this network.",
-            REQUEST_TIMEOUT.as_secs()
+            "{base_url} didn't respond in time. It may be offline, or unreachable \
+             from this network."
         )
     } else if err.is_connect() {
         format!(

desktop/src-tauri/src/sync/commands.rs

@@ -9,6 +9,7 @@ use tauri::State;
 use crate::local::Db;
 use crate::sync::client::{self, Identity, ProbeResult};
 use crate::sync::compat::Compatibility;
+use crate::sync::pull;
 use crate::sync::state;
 
 /// Ask a server who it is, without committing to anything. The UI calls this as the
@@ -122,3 +123,26 @@ pub fn sync_status(db: State<'_, Db>) -> Result<state::Status, String> {
     let conn = db.0.lock().map_err(|e| e.to_string())?;
     state::status(&conn).map_err(|e| e.to_string())
 }
+
+/// The server URL + token, or a plain "not linked" error. Every networked sync
+/// command needs exactly this, and none of them may hold the lock past it.
+fn credentials(db: &State<'_, Db>) -> Result<(String, String), String> {
+    let conn = db.0.lock().map_err(|e| e.to_string())?;
+    let current = state::read(&conn).map_err(|e| e.to_string())?;
+    match (current.server_url, current.device_token) {
+        (Some(url), Some(token)) => Ok((url, token)),
+        _ => Err("This app isn't linked to a server yet.".to_string()),
+    }
+}
+
+/// Pull the server's changes into the local store.
+///
+/// Standalone for now; M10.7c wraps push-then-pull into a single `sync_now`, which
+/// is the ordering the protocol assumes. Run alone against unpushed local edits, the
+/// server's version lands on top of them — the returned `clobbered_dirty` count
+/// reports that rather than hiding it.
+#[tauri::command]
+pub async fn sync_pull(db: State<'_, Db>) -> Result<pull::PullSummary, String> {
+    let (base_url, token) = credentials(&db)?;
+    pull::run(db.inner(), &base_url, &token).await
+}

desktop/src-tauri/src/sync/mod.rs

@@ -15,4 +15,6 @@
 pub mod client;
 pub mod commands;
 pub mod compat;
+pub mod pull;
 pub mod state;
+pub mod wire;

desktop/src-tauri/src/sync/pull.rs

@@ -0,0 +1,699 @@
+//! 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::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,
+}
+
+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.cursor = other.cursor;
+    }
+}
+
+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());
+    // `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, dirty)
+              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, 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,
+              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,
+        ],
+    )?;
+
+    // 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, 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."
+            ));
+        }
+    }
+
+    if total.clobbered_dirty > 0 {
+        log::warn!(
+            "pull overwrote {} note(s) that still had unpushed local edits",
+            total.clobbered_dirty
+        );
+    }
+    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,
+            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")
+    }
+
+    #[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 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);
+    }
+}

desktop/src-tauri/src/sync/wire.rs

@@ -0,0 +1,161 @@
+//! The delta-feed JSON shapes, exactly as `GET /api/sync/changes` sends them.
+//!
+//! Mirrors the server's serializers (`notes/serialize.py` + `serialize.py`) — see
+//! `docs/sync.md` for the contract. Every field is `#[serde(default)]` or `Option`
+//! so a NEWER server adding fields, or an older one omitting one, degrades to a
+//! partial note rather than failing the whole page. Losing one attribute is
+//! recoverable; refusing a page stalls sync permanently at that cursor.
+
+use serde::Deserialize;
+
+#[derive(Debug, Clone, Deserialize, Default)]
+pub struct ChangesPage {
+    #[serde(default)]
+    pub notes: Vec<Note>,
+    #[serde(default)]
+    pub labels: Vec<Label>,
+    #[serde(default)]
+    pub cursor: i64,
+    #[serde(default)]
+    pub has_more: bool,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Note {
+    pub id: String,
+    #[serde(default)]
+    pub title: Option<String>,
+    #[serde(default)]
+    pub body: String,
+    #[serde(default = "default_color")]
+    pub color: String,
+    #[serde(default = "default_kind")]
+    pub kind: String,
+    #[serde(default)]
+    pub position: i64,
+    #[serde(default)]
+    pub pinned: bool,
+    #[serde(default)]
+    pub archived: bool,
+    /// The server derives this from `deleted_at` — trash, NOT a tombstone.
+    #[serde(default)]
+    pub trashed: bool,
+    #[serde(default)]
+    pub remind_at: Option<String>,
+    #[serde(default)]
+    pub recurrence: Option<String>,
+    #[serde(default)]
+    pub created_at: Option<String>,
+    #[serde(default)]
+    pub updated_at: Option<String>,
+    #[serde(default)]
+    pub sync_revision: i64,
+    /// Set means the row was permanently purged: a content-less tombstone whose only
+    /// job is to tell clients to delete their copy.
+    #[serde(default)]
+    pub purged_at: Option<String>,
+    #[serde(default)]
+    pub labels: Vec<NoteLabel>,
+    #[serde(default)]
+    pub items: Vec<Item>,
+    #[serde(default)]
+    pub attachments: Vec<Attachment>,
+    #[serde(default)]
+    pub previews: Vec<Preview>,
+}
+
+impl Note {
+    pub fn is_tombstone(&self) -> bool {
+        self.purged_at.is_some()
+    }
+}
+
+/// A label as it appears attached to a note. Carries enough to materialize the label
+/// row itself, which is what lets a membership be applied even if the label's own
+/// delta hasn't arrived (see `pull::apply_page`).
+#[derive(Debug, Clone, Deserialize)]
+pub struct NoteLabel {
+    pub id: String,
+    #[serde(default)]
+    pub name: String,
+    #[serde(default = "default_color")]
+    pub color: String,
+    /// True when the membership came from a `#tag` in the body rather than a manual
+    /// assignment. Applied verbatim rather than re-derived — see `pull::apply_page`.
+    #[serde(default)]
+    pub via_tag: bool,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Item {
+    pub id: String,
+    #[serde(default)]
+    pub text: String,
+    #[serde(default)]
+    pub checked: bool,
+    #[serde(default)]
+    pub position: i64,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Attachment {
+    pub id: String,
+    #[serde(default)]
+    pub url: String,
+    #[serde(default)]
+    pub filename: Option<String>,
+    #[serde(default = "default_mime")]
+    pub mime: String,
+    #[serde(default)]
+    pub size: Option<i64>,
+    #[serde(default)]
+    pub sha256: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Preview {
+    pub id: String,
+    #[serde(default)]
+    pub url: String,
+    #[serde(default)]
+    pub title: Option<String>,
+    #[serde(default)]
+    pub description: Option<String>,
+    #[serde(default)]
+    pub image_url: Option<String>,
+    #[serde(default)]
+    pub site_name: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Label {
+    pub id: String,
+    #[serde(default)]
+    pub name: String,
+    #[serde(default = "default_color")]
+    pub color: String,
+    #[serde(default)]
+    pub sync_revision: i64,
+    #[serde(default)]
+    pub purged_at: Option<String>,
+    #[serde(default)]
+    pub created_at: Option<String>,
+}
+
+impl Label {
+    pub fn is_tombstone(&self) -> bool {
+        self.purged_at.is_some()
+    }
+}
+
+fn default_color() -> String {
+    "default".to_string()
+}
+
+fn default_kind() -> String {
+    "text".to_string()
+}
+
+fn default_mime() -> String {
+    "application/octet-stream".to_string()
+}