crates/nuncio-mail/src/backend.rs

@@ -65,19 +65,63 @@ pub struct RemoteMutationSpec {
     pub kind: RemoteMutationKind,
 }
 
+/// What one folder-sync pass observed on the server.
+///
+/// The reason this is not just a list of messages: a sync that can only report
+/// what *is* there can never report what stopped being there. A message moved
+/// or deleted by any other client -- another Nuncio daemon, a phone, webmail --
+/// simply stays in the local store forever. Expressing absence is the whole
+/// point of the type.
+#[derive(Debug, Default, Clone)]
+pub struct FolderChanges {
+    /// Messages to store, whether newly arrived or changed.
+    pub upserts: Vec<Email>,
+    /// Surrogate ids the server **explicitly reported as gone**, via QRESYNC
+    /// `VANISHED`. Always safe to delete.
+    pub removals: Vec<String>,
+    /// Surrogate ids of everything the folder currently holds, when this pass
+    /// enumerated the complete UID set. The caller may delete anything it
+    /// stores for this folder that is absent from this list.
+    ///
+    /// `None` and `Some(vec![])` mean different things, and conflating them
+    /// deletes a mailbox. `None` is "this pass was incremental and cannot
+    /// speak to absence"; `Some(vec![])` is "the folder is genuinely empty".
+    pub present: Option<Vec<String>>,
+    /// The checkpoint to resume from next time.
+    pub next_state: String,
+}
+
 /// Protocol-agnostic mail backend engine trait implemented by JMAP and IMAP engines.
 #[async_trait]
 pub trait MailBackend: Send + Sync {
     /// Synchronize and list available mailbox folders.
     async fn sync_folders(&self) -> Result<Vec<Folder>, MailError>;
 
-    /// Synchronize message envelopes for a specific folder since a checkpoint state.
-    /// Returns the updated email list and the new server state checkpoint.
+    /// Enumerate what changed in a folder since `since_state`.
+    ///
+    /// Replaces a fetch-only sync: implementations must report removals when
+    /// the protocol can express them, and must say honestly (via
+    /// [`FolderChanges::present`]) whether the pass was able to observe
+    /// absence at all.
+    async fn sync_changes(
+        &self,
+        folder_id: &str,
+        since_state: Option<&str>,
+    ) -> Result<FolderChanges, MailError>;
+
+    /// Fetch-only view of [`MailBackend::sync_changes`], for callers that
+    /// genuinely only want the messages.
+    ///
+    /// Provided rather than required so no implementation can satisfy the
+    /// trait by supplying this and quietly never reporting a removal.
     async fn sync_messages(
         &self,
         folder_id: &str,
         since_state: Option<&str>,
-    ) -> Result<(Vec<Email>, String), MailError>;
+    ) -> Result<(Vec<Email>, String), MailError> {
+        let changes = self.sync_changes(folder_id, since_state).await?;
+        Ok((changes.upserts, changes.next_state))
+    }
 
     /// Apply a remote mutation (move/copy/flag/delete) to a single message
     /// against the real server. MUST return `Ok(())` only when the server

crates/nuncio-mail/src/imap_raw.rs

@@ -42,6 +42,14 @@ use tokio::io::{AsyncRead, AsyncWrite};
 
 use crate::parser::MailError;
 
+/// Bounded wait for any single response line.
+///
+/// A server that accepts a command and then stops answering must surface as a
+/// stall, not as a sync that never returns. Mirrors the per-item FETCH timeout
+/// the typed path already applies -- an unbounded read here would reintroduce
+/// exactly the hang that bounded it.
+const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
+
 /// A `COPYUID` response code (RFC 4315 section 3): the destination mailbox's
 /// UIDVALIDITY, the source UIDs, and the UIDs they became.
 ///
@@ -84,6 +92,8 @@ pub struct RawExchange {
     pub highest_modseq: Option<u64>,
     /// UIDVALIDITY, from an untagged `OK` during `SELECT`.
     pub uid_validity: Option<u32>,
+    /// UIDNEXT, from an untagged `OK` during `SELECT`.
+    pub uid_next: Option<u32>,
 }
 
 impl RawExchange {
@@ -170,9 +180,13 @@ where
     let mut exchange = RawExchange::default();
 
     loop {
-        let response = session
-            .read_response()
+        let response = tokio::time::timeout(RESPONSE_READ_TIMEOUT, session.read_response())
             .await
+            .map_err(|_| {
+                MailError::FetchStalled(format!(
+                    "no response to '{command}' within {RESPONSE_READ_TIMEOUT:?}"
+                ))
+            })?
             .map_err(|e| MailError::ImapError(format!("reading response to '{command}': {e}")))?
             .ok_or_else(|| {
                 MailError::ImapError(format!("connection closed before '{command}' completed"))
@@ -232,6 +246,7 @@ fn absorb_code(exchange: &mut RawExchange, code: Option<&ResponseCode<'_>>) {
         }
         Some(ResponseCode::HighestModSeq(modseq)) => exchange.highest_modseq = Some(*modseq),
         Some(ResponseCode::UidValidity(validity)) => exchange.uid_validity = Some(*validity),
+        Some(ResponseCode::UidNext(uid_next)) => exchange.uid_next = Some(*uid_next),
         _ => {}
     }
 }

crates/nuncio-mail/src/jmap.rs

@@ -561,11 +561,11 @@ impl MailBackend for JmapEngine {
         fields(account_id = %self.account_id, folder_id = %folder_id),
         err
     )]
-    async fn sync_messages(
+    async fn sync_changes(
         &self,
         folder_id: &str,
         _since_state: Option<&str>,
-    ) -> Result<(Vec<Email>, String), MailError> {
+    ) -> Result<crate::backend::FolderChanges, MailError> {
         if !self.has_credentials() {
             return Err(MailError::AuthError(
                 "JMAP message sync requires credentials".to_string(),
@@ -589,7 +589,16 @@ impl MailBackend for JmapEngine {
             "jmap message sync complete"
         );
 
-        Ok((emails, state))
+        Ok(crate::backend::FolderChanges {
+            upserts: emails,
+            removals: Vec::new(),
+            // `Email/query` was filtered to one mailbox, but this engine does
+            // not yet use `Email/changes`, so it cannot claim to have seen the
+            // complete set. Reporting `None` keeps the caller from deleting
+            // anything on the strength of a partial view.
+            present: None,
+            next_state: state,
+        })
     }
 
     #[tracing::instrument(skip(self), fields(account_id = %self.account_id), err)]

crates/nuncio-mail/src/mock.rs

@@ -106,11 +106,11 @@ impl MailBackend for MockMailBackend {
         Ok(folders.clone())
     }
 
-    async fn sync_messages(
+    async fn sync_changes(
         &self,
         folder_id: &str,
         since_state: Option<&str>,
-    ) -> Result<(Vec<Email>, String), MailError> {
+    ) -> Result<crate::backend::FolderChanges, MailError> {
         // Record the checkpoint arg before any early return, so even a failing
         // call is visible to a test asserting on how the caller resumes.
         if let Ok(mut calls) = self.since_state_calls.lock() {
@@ -136,19 +136,30 @@ impl MailBackend for MockMailBackend {
         // is nothing new to hand back, so the incremental fetch is genuinely
         // narrower (empty) rather than a full re-report of every message.
         if since_state.is_some() {
-            return Ok((Vec::new(), returned_state));
+            return Ok(crate::backend::FolderChanges {
+                next_state: returned_state,
+                ..Default::default()
+            });
         }
 
         let messages = self
             .messages
             .lock()
             .map_err(|e| MailError::ParseFailed(e.to_string()))?;
-        let matches = messages
+        let matches: Vec<_> = messages
             .iter()
             .filter(|m| m.folder_id == folder_id)
             .cloned()
             .collect();
-        Ok((matches, returned_state))
+        // A full pass over the mock's store did see everything in the folder,
+        // so it can honestly report what is present.
+        let present = matches.iter().map(|m| m.id.clone()).collect();
+        Ok(crate::backend::FolderChanges {
+            upserts: matches,
+            removals: Vec::new(),
+            present: Some(present),
+            next_state: returned_state,
+        })
     }
 
     async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> {

crates/nuncio-mail/tests/imap_raw_wire_test.rs

@@ -299,3 +299,182 @@ async fn a_qresync_select_without_enable_is_refused() {
         "the refusal must be surfaced, got: {err}"
     );
 }
+
+/// Drive a real folder sync against the mock and return what it observed.
+async fn sync_folder(
+    server: &MockImapServer,
+    since: Option<&str>,
+) -> nuncio_mail::backend::FolderChanges {
+    use nuncio_mail::MailBackend;
+    let engine = nuncio_mail::ImapEngine::with_credentials(
+        "acct-1",
+        server.host(),
+        server.port(),
+        nuncio_core::TlsMode::Plain,
+        "user",
+        "pass",
+    );
+    engine
+        .sync_changes("INBOX", since)
+        .await
+        .expect("folder sync succeeds")
+}
+
+#[tokio::test]
+async fn every_rung_reports_a_message_that_disappeared() {
+    // The defect this whole change exists for. A forward-only sync can only
+    // ever add, so a message removed by another client stays local forever.
+    // Each rung must be able to say it is gone -- QRESYNC because the server
+    // states it, the other two because the pass enumerated what remains.
+    for profile in [
+        ServerProfile::Qresync,
+        ServerProfile::Condstore,
+        ServerProfile::Basic,
+    ] {
+        let server = MockImapServer::start(profile).await.expect("server starts");
+        server.create_mailbox("INBOX");
+        let keep = server.append_message("INBOX", "Subject: keep\r\n\r\nkeep", &[]);
+        let gone = server.append_message("INBOX", "Subject: gone\r\n\r\ngone", &[]);
+
+        let first = sync_folder(&server, None).await;
+        assert_eq!(first.upserts.len(), 2, "{profile:?}: first pass sees both");
+
+        // Another client removes one while we are away.
+        assert!(server.expunge("INBOX", gone));
+
+        let second = sync_folder(&server, Some(&first.next_state)).await;
+
+        // However the rung learned it, the caller must end up able to identify
+        // the removed message and only that one.
+        let mut reported: Vec<String> = second.removals.clone();
+        if let Some(present) = &second.present {
+            let stored_gone = first
+                .upserts
+                .iter()
+                .map(|e| e.id.clone())
+                .filter(|id| !present.contains(id));
+            reported.extend(stored_gone);
+        }
+        reported.sort();
+        reported.dedup();
+
+        assert_eq!(
+            reported.len(),
+            1,
+            "{profile:?}: exactly one message must be reported gone, got {reported:?}"
+        );
+        let surviving = first
+            .upserts
+            .iter()
+            .find(|e| e.remote_id == keep.to_string())
+            .expect("the kept message was in the first pass");
+        assert!(
+            !reported.contains(&surviving.id),
+            "{profile:?}: the surviving message must not be reported gone"
+        );
+    }
+}
+
+#[tokio::test]
+async fn an_incremental_pass_that_cannot_see_absence_says_so() {
+    // `present: None` versus `Some(vec![])` is the difference between "I did
+    // not look" and "the folder is empty". Conflating them deletes a mailbox,
+    // so QRESYNC -- which never enumerates -- must report None.
+    let server = MockImapServer::start(ServerProfile::Qresync)
+        .await
+        .expect("server starts");
+    server.create_mailbox("INBOX");
+    server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]);
+
+    let first = sync_folder(&server, None).await;
+    let second = sync_folder(&server, Some(&first.next_state)).await;
+
+    assert!(
+        second.next_state.starts_with("v2:modseq:"),
+        "a QRESYNC-capable mailbox must checkpoint by mod-sequence, got {:?}",
+        second.next_state
+    );
+    assert_eq!(
+        second.present, None,
+        "a QRESYNC pass does not enumerate the folder, so it cannot claim to know what is present"
+    );
+}
+
+#[tokio::test]
+async fn a_condstore_only_server_still_checkpoints_by_modseq_and_enumerates() {
+    // Gmail's rung. It cannot be told what vanished, so it must enumerate --
+    // and it must still upgrade its checkpoint so the next pass is narrow.
+    let server = MockImapServer::start(ServerProfile::Condstore)
+        .await
+        .expect("server starts");
+    server.create_mailbox("INBOX");
+    server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]);
+
+    let first = sync_folder(&server, None).await;
+    assert!(
+        first.next_state.starts_with("v2:modseq:"),
+        "CONDSTORE alone is enough to checkpoint by mod-sequence, got {:?}",
+        first.next_state
+    );
+
+    let second = sync_folder(&server, Some(&first.next_state)).await;
+    assert!(
+        second.present.is_some(),
+        "without QRESYNC the only way to find removals is to enumerate"
+    );
+    assert!(
+        second.upserts.is_empty(),
+        "nothing changed, so CHANGEDSINCE must fetch no bodies"
+    );
+}
+
+#[tokio::test]
+async fn a_server_without_modseq_falls_back_to_a_uid_boundary_checkpoint() {
+    // Exchange's rung: no CONDSTORE, no QRESYNC.
+    let server = MockImapServer::start(ServerProfile::Basic)
+        .await
+        .expect("server starts");
+    server.create_mailbox("INBOX");
+    server.append_message("INBOX", "Subject: one\r\n\r\nx", &[]);
+
+    let first = sync_folder(&server, None).await;
+    assert!(
+        first.next_state.starts_with("v1:uidnext:"),
+        "a mailbox with no mod-sequences must checkpoint by UID boundary, got {:?}",
+        first.next_state
+    );
+    assert!(first.present.is_some());
+}
+
+#[tokio::test]
+async fn a_uidvalidity_change_refetches_rather_than_deleting_the_folder() {
+    // After a renumbering every stored UID names a different message. The
+    // dangerous outcome would be enumerating the new UID space, finding none
+    // of the old ids in it, and concluding the whole folder was deleted.
+    let server = MockImapServer::start(ServerProfile::Qresync)
+        .await
+        .expect("server starts");
+    server.create_mailbox("INBOX");
+    server.append_message("INBOX", "Subject: before\r\n\r\nx", &[]);
+
+    let first = sync_folder(&server, None).await;
+    assert_eq!(first.upserts.len(), 1);
+
+    server.bump_uid_validity("INBOX");
+    server.append_message("INBOX", "Subject: after\r\n\r\nx", &[]);
+
+    let second = sync_folder(&server, Some(&first.next_state)).await;
+
+    assert_eq!(
+        second.upserts.len(),
+        1,
+        "a renumbered mailbox must be re-fetched in full"
+    );
+    assert!(
+        second.removals.is_empty(),
+        "a renumbering is not a removal report"
+    );
+    // The re-fetched message has a different surrogate, because UIDVALIDITY is
+    // part of identity -- so the caller replaces rather than merges.
+    assert_ne!(second.upserts[0].id, first.upserts[0].id);
+}

crates/nuncio-store/src/db.rs

@@ -29,6 +29,22 @@ pub enum DatabaseError {
     /// Recovery salvage operation failed.
     #[error("database recovery error: {0}")]
     RecoveryFailed(String),
+    /// The database was written by a newer binary than this one.
+    ///
+    /// Fail-closed: the schema is built from `CREATE TABLE IF NOT EXISTS`, so
+    /// an older binary would otherwise open a newer database successfully and
+    /// write the old model into it, leaving two mutually invisible truths in
+    /// one file. There is no safe way to continue.
+    #[error(
+        "database identity schema v{on_disk} is newer than this build supports (v{supported}); \
+         upgrade nunciod rather than running an older binary against it"
+    )]
+    SchemaTooNew {
+        /// The generation recorded in the database's `user_version`.
+        on_disk: i64,
+        /// The newest generation this binary understands.
+        supported: i64,
+    },
     /// Audit chain verification failed.
     #[error("audit chain integrity error: {0}")]
     ChainIntegrityFailed(String),
@@ -496,6 +512,102 @@ impl DatabaseEngine {
         &self.pool
     }
 
+    /// The identity-schema generation this binary understands.
+    ///
+    /// Distinct from the additive column migrations elsewhere in this module.
+    /// Those add nullable columns and leave every stored row meaningful. This
+    /// counter tracks changes to what a stored message *is* -- how it is keyed
+    /// and what a sync checkpoint means -- where old rows cannot be reinterpreted
+    /// under the new model at all.
+    ///
+    /// Bump this only for a change of that kind, and only together with the
+    /// reset it implies (see [`DatabaseEngine::reconcile_schema_version`]).
+    pub const IDENTITY_SCHEMA_VERSION: i64 = 1;
+
+    /// Reconcile the on-disk identity-schema generation with this binary's.
+    ///
+    /// Three outcomes:
+    ///
+    /// - **Same version** (and the common case): nothing happens.
+    /// - **Database older**: the cached message store is rebuilt and the account
+    ///   re-syncs from the server. Accounts, credentials and filter rules are
+    ///   preserved; cached mail, sync checkpoints and the execution ledger are
+    ///   not. This is deliberate and pre-release only -- see ADR 0002. Message
+    ///   identity cannot be recomputed for rows already stored, so the
+    ///   alternative is a permanently two-tier store in which old messages are
+    ///   invisible to every guarantee the new model provides. The server holds
+    ///   the authoritative copy either way.
+    /// - **Database newer**: refuse to open. A `CREATE TABLE IF NOT EXISTS`
+    ///   schema means an older binary can open a newer database and quietly
+    ///   write the old model into it, leaving two mutually invisible truths in
+    ///   one file. Failing loudly is the only way that does not corrupt.
+    pub async fn reconcile_schema_version(&self) -> Result<(), DatabaseError> {
+        let (on_disk,): (i64,) = sqlx::query_as("PRAGMA user_version")
+            .fetch_one(&self.pool)
+            .await
+            .map_err(DatabaseError::Query)?;
+
+        if on_disk > Self::IDENTITY_SCHEMA_VERSION {
+            return Err(DatabaseError::SchemaTooNew {
+                on_disk,
+                supported: Self::IDENTITY_SCHEMA_VERSION,
+            });
+        }
+
+        // 0 is both "brand new database" and "predates this counter". Neither
+        // has anything worth preserving that a re-sync will not restore, and
+        // the reset is a no-op on an empty store.
+        if on_disk < Self::IDENTITY_SCHEMA_VERSION {
+            if on_disk > 0 {
+                tracing::warn!(
+                    on_disk,
+                    supported = Self::IDENTITY_SCHEMA_VERSION,
+                    "message identity schema changed: rebuilding the cached message store; \
+                     accounts and rules are preserved and mail will re-sync from the server"
+                );
+            }
+            self.reset_cached_message_store().await?;
+            sqlx::query(&format!(
+                "PRAGMA user_version = {}",
+                Self::IDENTITY_SCHEMA_VERSION
+            ))
+            .execute(&self.pool)
+            .await
+            .map_err(DatabaseError::Query)?;
+        }
+
+        Ok(())
+    }
+
+    /// Drop everything derived from the server, keeping everything the user
+    /// configured.
+    ///
+    /// Cleared: messages and their FTS index, per-folder sync checkpoints, and
+    /// the filter execution ledger. The ledger goes because its hash chain is
+    /// keyed on message ids that will no longer resolve, and a chain pointing
+    /// at absent rows has no audit value -- reinitialising it is more honest
+    /// than carrying a permanently mixed-namespace log.
+    ///
+    /// Kept: accounts, credentials (which live in the OS keyring regardless),
+    /// filter rules, and the WORM audit records, whose triggers forbid deletion
+    /// and whose entries do not reference message identity.
+    async fn reset_cached_message_store(&self) -> Result<(), DatabaseError> {
+        let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?;
+        for statement in [
+            "DELETE FROM messages",
+            "DELETE FROM messages_fts",
+            "DELETE FROM folder_sync_state",
+            "DELETE FROM filter_execution_logs",
+        ] {
+            sqlx::query(statement)
+                .execute(&mut *tx)
+                .await
+                .map_err(DatabaseError::Query)?;
+        }
+        tx.commit().await.map_err(DatabaseError::Query)?;
+        Ok(())
+    }
+
     /// Execute initial database migrations creating core envelope tables.
     pub async fn migrate(&self) -> Result<(), DatabaseError> {
         sqlx::query(
@@ -750,6 +862,12 @@ impl DatabaseEngine {
         // The statements above create the schema in full, so there is nothing to
         // upgrade: every column this engine reads is declared in its own
         // `CREATE TABLE IF NOT EXISTS`. Databases are created, never migrated.
+        //
+        // Reconciling the identity-schema generation is a different question and
+        // still applies: it asks whether a store was written under a model this
+        // binary can still interpret, and rebuilds (or refuses) accordingly.
+        // Runs after the tables exist, since the rebuild deletes from them.
+        self.reconcile_schema_version().await?;
         self.backfill_message_fts().await?;
 
         Ok(())
@@ -1374,6 +1492,51 @@ impl DatabaseEngine {
     /// Lets a caller classify a whole fetched batch as new-vs-seen with one round trip instead
     /// of one existence lookup per id. Passing an empty slice short-circuits to an empty set
     /// without querying at all, since `IN ()` is not valid SQL.
+    /// Every message id currently stored for one folder of one account.
+    ///
+    /// The local half of a UID-set diff: a sync that enumerated the folder's
+    /// complete contents can subtract what the server reported from this to
+    /// find what disappeared. Scoped by account as well as folder because
+    /// folder ids are not globally unique -- two accounts both have an
+    /// `INBOX`, and diffing across them would delete one account's mail
+    /// because the other's server did not mention it.
+    pub async fn message_ids_in_folder(
+        &self,
+        account_id: &str,
+        folder_id: &str,
+    ) -> Result<Vec<String>, DatabaseError> {
+        let rows: Vec<(String,)> =
+            sqlx::query_as("SELECT id FROM messages WHERE account_id = ? AND folder_id = ?")
+                .bind(account_id)
+                .bind(folder_id)
+                .fetch_all(&self.pool)
+                .await
+                .map_err(DatabaseError::Query)?;
+        Ok(rows.into_iter().map(|(id,)| id).collect())
+    }
+
+    /// Delete stored messages by id, returning how many rows went.
+    ///
+    /// The FTS index follows via the `messages_ad` trigger, so a deleted
+    /// message cannot survive as a searchable plaintext orphan.
+    pub async fn delete_messages(&self, ids: &[String]) -> Result<u64, DatabaseError> {
+        if ids.is_empty() {
+            return Ok(0);
+        }
+        let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?;
+        let mut deleted = 0u64;
+        for id in ids {
+            let result = sqlx::query("DELETE FROM messages WHERE id = ?")
+                .bind(id)
+                .execute(&mut *tx)
+                .await
+                .map_err(DatabaseError::Query)?;
+            deleted += result.rows_affected();
+        }
+        tx.commit().await.map_err(DatabaseError::Query)?;
+        Ok(deleted)
+    }
+
     pub async fn existing_message_ids(
         &self,
         ids: &[String],
@@ -3868,6 +4031,104 @@ mod tests {
         assert_eq!(fetched.folder_id, "INBOX");
     }
 
+    #[tokio::test]
+    async fn opening_a_newer_database_fails_closed_instead_of_writing_the_old_model() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+        sqlx::query(&format!(
+            "PRAGMA user_version = {}",
+            DatabaseEngine::IDENTITY_SCHEMA_VERSION + 1
+        ))
+        .execute(engine.pool())
+        .await
+        .unwrap();
+
+        let err = engine
+            .reconcile_schema_version()
+            .await
+            .expect_err("a database from a newer build must not be opened");
+        assert!(
+            matches!(err, DatabaseError::SchemaTooNew { .. }),
+            "got {err:?}"
+        );
+        // The schema is CREATE TABLE IF NOT EXISTS, so without this an older
+        // binary opens a newer database happily and writes the old model into
+        // it, leaving two mutually invisible truths in one file.
+        assert!(err.to_string().contains("newer than this build supports"));
+    }
+
+    #[tokio::test]
+    async fn an_identity_schema_bump_rebuilds_cached_mail_and_keeps_configuration() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+
+        let account = nuncio_core::AccountConfig {
+            id: "acct-reset".to_string(),
+            name: "Reset Account".to_string(),
+            email_address: "reset@nuncio.mx".to_string(),
+            keyring_secret_key: "nuncio/acct-reset".to_string(),
+            sync_interval_secs: 60,
+            transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
+                imap_host: "imap.nuncio.mx".to_string(),
+                imap_port: 993,
+                imap_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+                smtp_host: "smtp.nuncio.mx".to_string(),
+                smtp_port: 465,
+                smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+            }),
+        };
+        engine.save_account(&account).await.expect("save account");
+        engine
+            .save_email(&synced_email("acct-reset", "INBOX", "42", "7"))
+            .await
+            .expect("save email");
+        engine
+            .save_folder_sync_state("acct-reset", "INBOX", "v1:uidnext:42:8")
+            .await
+            .expect("save checkpoint");
+
+        // Pretend this store was written by the previous identity generation.
+        sqlx::query("PRAGMA user_version = 0")
+            .execute(engine.pool())
+            .await
+            .unwrap();
+
+        engine
+            .reconcile_schema_version()
+            .await
+            .expect("an older store is rebuilt, not rejected");
+
+        // Server-derived state goes: it cannot be reinterpreted under a new
+        // identity model, and the server holds the authoritative copy.
+        let (message_count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM messages")
+            .fetch_one(engine.pool())
+            .await
+            .unwrap();
+        assert_eq!(message_count, 0, "cached mail must be rebuilt");
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-reset", "INBOX")
+                .await
+                .unwrap(),
+            None,
+            "a checkpoint must not survive an identity change"
+        );
+
+        // Everything the user configured survives.
+        let accounts = engine.list_accounts().await.expect("list accounts");
+        assert_eq!(accounts.len(), 1, "accounts must be preserved");
+        assert_eq!(accounts[0].id, "acct-reset");
+
+        // The version is recorded, so the next open is a no-op.
+        let (version,): (i64,) = sqlx::query_as("PRAGMA user_version")
+            .fetch_one(engine.pool())
+            .await
+            .unwrap();
+        assert_eq!(version, DatabaseEngine::IDENTITY_SCHEMA_VERSION);
+        engine
+            .reconcile_schema_version()
+            .await
+            .expect("re-running at the current version is a no-op");
+    }
+
     fn export_test_email(id: &str, account_id: &str, folder_id: &str) -> nuncio_core::model::Email {
         nuncio_core::model::Email {
             id: id.to_string(),

crates/nunciod/src/sync.rs

@@ -245,10 +245,12 @@ async fn fetch_and_persist(
             Some(acct) => db.get_folder_sync_state(acct, &folder.id).await?,
             None => None,
         };
-        let (emails, new_state) = backend
-            .sync_messages(&folder.id, last_state.as_deref())
+        let changes = backend
+            .sync_changes(&folder.id, last_state.as_deref())
             .await?;
+        let emails = changes.upserts;
         let fetched = emails.len();
+        let new_state = changes.next_state;
 
         // One round trip classifies the whole chunk instead of one lookup per
         // message. A lookup failure that is not "not found" would be a
@@ -268,6 +270,37 @@ async fn fetch_and_persist(
             }
             synced += 1;
         }
+        // Reconcile what the server no longer has. Until this existed, a sync
+        // could only ever add: a message moved or deleted by any other client
+        // -- another daemon, a phone, webmail -- stayed in the local store
+        // forever, and every account accumulated ghosts.
+        if let Some(acct) = account_id {
+            let mut gone: Vec<String> = changes.removals;
+
+            // `present` is the folder's complete contents when the pass was
+            // able to enumerate them. `None` and `Some(vec![])` are NOT the
+            // same: `None` means the pass was incremental and cannot speak to
+            // absence, and treating it as "nothing is present" would delete
+            // the folder.
+            if let Some(present) = changes.present {
+                let present: std::collections::HashSet<String> = present.into_iter().collect();
+                let stored = db.message_ids_in_folder(acct, &folder.id).await?;
+                gone.extend(stored.into_iter().filter(|id| !present.contains(id)));
+            }
+
+            gone.sort_unstable();
+            gone.dedup();
+            if !gone.is_empty() {
+                let deleted = db.delete_messages(&gone).await?;
+                tracing::info!(
+                    account_id = %acct,
+                    folder_id = %folder.id,
+                    deleted,
+                    "removed messages that are no longer on the server"
+                );
+            }
+        }
+
         // Persist the returned checkpoint only AFTER this folder's messages
         // have landed, so the stored high-water mark can never advance past
         // work that actually reached the store.