crates/nuncio-core/src/lib.rs

@@ -93,6 +93,19 @@ pub enum CoreEvent {
         /// Action description.
         action_taken: String,
     },
+    /// Progress notification emitted during an inbound mail sync, so a
+    /// long-running sync of a large mailbox is observably progressing rather
+    /// than indistinguishable from a hang. Emitted per folder as it completes.
+    SyncProgress {
+        /// Account being synced, if the sync is scoped to one.
+        account_id: Option<String>,
+        /// Folder whose messages were just fetched and persisted.
+        folder_id: String,
+        /// Messages fetched from this folder on this pass.
+        fetched: usize,
+        /// Cumulative messages fetched across the sync so far.
+        total: usize,
+    },
     /// Progress notification during bulk filter execution over a batch of emails.
     BatchFilterProgress {
         /// Messages processed so far in the batch.

crates/nuncio-mail/Cargo.toml

@@ -31,3 +31,6 @@ reqwest.workspace = true
 [dev-dependencies]
 tempfile.workspace = true
 wiremock.workspace = true
+# `test-util` (not in tokio's `full`) enables paused-clock tests so the
+# bounded per-item FETCH timeout can be asserted without a real-time wait.
+tokio = { workspace = true, features = ["test-util"] }

crates/nuncio-mail/src/imap.rs

@@ -23,9 +23,108 @@ use tokio_stream::StreamExt;
 /// hang a connect -- and therefore a `TestAccountConnection` RPC -- forever.
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
 
+/// Upper bound on how long a single `FETCH` response item may take to arrive
+/// once the `UID FETCH` command has been acknowledged. Fetching full bodies
+/// for a large mailbox in one command can be legitimately slow overall, so the
+/// bound is per item rather than for the whole stream; but a single item that
+/// never arrives (dead socket, server-side hang) must still surface as an
+/// honest timeout rather than hanging the sync indefinitely.
+const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60);
+
 use crate::backend::MailBackend;
 use crate::parser::{MailError, MimeParserAdapter};
 
+/// Delimiter between the two components of a folder sync checkpoint. The
+/// checkpoint state string is `"{uidvalidity}{DELIM}{uidnext}"` (e.g.
+/// `"42:105"`): the mailbox's UIDVALIDITY at the time it was written, then the
+/// UID boundary to resume from. Both components are decimal `u32`.
+///
+/// UIDVALIDITY MUST be carried because IMAP UIDs are only meaningful within a
+/// stable UIDVALIDITY (RFC 3501 s2.3.1.1). If a mailbox is recreated and its
+/// UIDVALIDITY changes, UIDs restart from low numbers, so resuming from a UID
+/// boundary captured under the old UIDVALIDITY would issue `{old_uidnext}:*`
+/// against renumbered low UIDs -- which per `n:*` semantics returns only the
+/// single highest message and silently skips the rest. Comparing UIDVALIDITY
+/// first turns that into a safe full re-fetch instead.
+const CHECKPOINT_DELIM: char = ':';
+
+/// The `UID FETCH` sequence-set to issue for a folder sync, resolved from the
+/// stored checkpoint. Kept as a small typed value (rather than collapsing to a
+/// bare range string) so the distinct fall-back reasons stay observable: all
+/// of `Full`/`CorruptCheckpoint`/`UidValidityChanged` fetch `1:*`, but only the
+/// latter two are worth flagging, and tests can assert on which branch fired.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FetchRange {
+    /// No prior checkpoint (first-ever sync of this folder): fetch everything.
+    Full,
+    /// A checkpoint was stored but could not be parsed (malformed, or a legacy
+    /// bare-UID checkpoint predating UIDVALIDITY tracking). Falls back to a
+    /// full fetch, surfaced (logged) rather than silently ignored.
+    CorruptCheckpoint,
+    /// The stored checkpoint's UIDVALIDITY no longer matches the mailbox's
+    /// current UIDVALIDITY (the mailbox was renumbered) or the server did not
+    /// report a current UIDVALIDITY to compare against. Stored UIDs are no
+    /// longer meaningful, so this falls back to a safe full re-fetch.
+    UidValidityChanged {
+        /// UIDVALIDITY recorded in the stored checkpoint.
+        stored: u32,
+        /// UIDVALIDITY the server reports now, if any.
+        current: Option<u32>,
+    },
+    /// A valid checkpoint whose UIDVALIDITY still matches: fetch only UIDs at
+    /// or above the stored boundary, i.e. messages that arrived since.
+    Incremental(u32),
+}
+
+impl FetchRange {
+    /// Render the IMAP `UID FETCH` sequence-set. Every non-incremental variant
+    /// fetches all UIDs (`1:*`); an incremental checkpoint `n` fetches `n:*`,
+    /// which the server bounds to UIDs `>= n` -- a genuinely narrower fetch.
+    fn sequence_set(self) -> String {
+        match self {
+            FetchRange::Full
+            | FetchRange::CorruptCheckpoint
+            | FetchRange::UidValidityChanged { .. } => "1:*".to_string(),
+            FetchRange::Incremental(uid) => format!("{}:*", uid),
+        }
+    }
+}
+
+/// Resolve the fetch range from a stored `since_state` checkpoint and the
+/// mailbox's CURRENT UIDVALIDITY (as just reported by `SELECT`). `None` means
+/// no prior sync (full fetch). A checkpoint in the `"{uidvalidity}:{uidnext}"`
+/// format proceeds incrementally only when its UIDVALIDITY still matches the
+/// server's current one; a mismatch (mailbox renumbered) or a missing current
+/// UIDVALIDITY forces a safe full re-fetch. A malformed or legacy bare-number
+/// checkpoint is treated as corrupt (full fetch, flagged), never misread as a
+/// live UID boundary.
+fn resolve_fetch_range(since_state: Option<&str>, current_uid_validity: Option<u32>) -> FetchRange {
+    let Some(raw) = since_state else {
+        return FetchRange::Full;
+    };
+    let parsed = raw
+        .split_once(CHECKPOINT_DELIM)
+        .and_then(|(validity, uid)| {
+            Some((
+                validity.trim().parse::<u32>().ok()?,
+                uid.trim().parse::<u32>().ok()?,
+            ))
+        });
+    let Some((stored_validity, uid)) = parsed else {
+        return FetchRange::CorruptCheckpoint;
+    };
+    if uid < 1 {
+        return FetchRange::CorruptCheckpoint;
+    }
+    match current_uid_validity {
+        Some(current) if current == stored_validity => FetchRange::Incremental(uid),
+        current => FetchRange::UidValidityChanged {
+            stored: stored_validity,
+            current,
+        },
+    }
+}
+
 /// State of the dedicated IMAP IDLE socket listener.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum IdleSocketState {
@@ -449,21 +548,53 @@ impl ImapEngine {
         &self.socket_manager
     }
 
-    /// Fetch and parse all messages in a folder over an active IMAP session.
+    /// Fetch and parse the messages in a folder over an active IMAP session,
+    /// returning the fetched messages and the new sync checkpoint.
+    ///
+    /// `since_state` is the checkpoint returned by a previous sync of this
+    /// folder, encoded as `"{uidvalidity}:{uidnext}"` (see [`CHECKPOINT_DELIM`]).
+    /// The `UID FETCH` is narrowed to `{uidnext}:*` only when the checkpoint's
+    /// UIDVALIDITY still matches the mailbox's current UIDVALIDITY; a mismatch
+    /// (the mailbox was renumbered) or a malformed/legacy checkpoint forces a
+    /// safe full re-fetch instead of silently skipping renumbered mail. The
+    /// returned checkpoint is derived from real server state -- the mailbox's
+    /// UIDVALIDITY plus its `UIDNEXT` (or one past the highest UID actually
+    /// seen) after selecting the folder -- never from a synthetic counter.
     pub async fn sync_folder_messages_with_session<S>(
         &self,
         folder_id: &str,
+        since_state: Option<&str>,
         session: &mut async_imap::Session<S>,
-    ) -> Result<Vec<Email>, MailError>
+    ) -> Result<(Vec<Email>, String), MailError>
     where
         S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
     {
-        session.select(folder_id).await.map_err(|e| {
+        let mailbox = session.select(folder_id).await.map_err(|e| {
             MailError::ImapError(format!("failed to select folder '{}': {}", folder_id, e))
         })?;
+        let server_uid_next = mailbox.uid_next;
+        let server_uid_validity = mailbox.uid_validity;
+
+        let range = resolve_fetch_range(since_state, server_uid_validity);
+        match range {
+            FetchRange::CorruptCheckpoint => tracing::warn!(
+                folder_id,
+                checkpoint = since_state.unwrap_or_default(),
+                "ignoring unparseable folder sync checkpoint; falling back to a full fetch"
+            ),
+            FetchRange::UidValidityChanged { stored, current } => tracing::warn!(
+                folder_id,
+                stored_uidvalidity = stored,
+                current_uidvalidity = current,
+                "mailbox UIDVALIDITY changed (renumbered); the stored UID checkpoint is no \
+                 longer valid, so re-fetching in full to avoid silently skipping messages"
+            ),
+            FetchRange::Full | FetchRange::Incremental(_) => {}
+        }
 
         let query = build_fetch_command_query();
-        let mut fetch_stream = session.uid_fetch("1:*", query).await.map_err(|e| {
+        let sequence_set = range.sequence_set();
+        let mut fetch_stream = session.uid_fetch(&sequence_set, query).await.map_err(|e| {
             MailError::ImapError(format!(
                 "UID FETCH failed for folder '{}': {}",
                 folder_id, e
@@ -471,11 +602,29 @@ impl ImapEngine {
         })?;
 
         let mut emails = Vec::new();
-        while let Some(fetch_res) = fetch_stream.next().await {
+        let mut max_uid_seen: u32 = 0;
+        loop {
+            // Bound EVERY individual item read: a slow-but-progressing large
+            // fetch is fine, but a single item that never arrives must surface
+            // as an honest `FetchStalled` instead of hanging the whole sync.
+            let next_item = tokio::time::timeout(FETCH_ITEM_TIMEOUT, fetch_stream.next())
+                .await
+                .map_err(|_| {
+                    MailError::FetchStalled(format!(
+                        "no FETCH response for folder '{}' within {}s ({} message(s) read so far)",
+                        folder_id,
+                        FETCH_ITEM_TIMEOUT.as_secs(),
+                        emails.len()
+                    ))
+                })?;
+            let Some(fetch_res) = next_item else {
+                break;
+            };
             let fetch_data = fetch_res
                 .map_err(|e| MailError::ImapError(format!("failed reading fetch item: {}", e)))?;
 
             let uid_num = fetch_data.uid.unwrap_or(0);
+            max_uid_seen = max_uid_seen.max(uid_num);
             let email_id = format!("imap-uid-{}", uid_num);
 
             let is_read = fetch_data
@@ -559,14 +708,44 @@ impl ImapEngine {
             emails.push(email);
         }
 
-        Ok(emails)
+        // Derive the next checkpoint from real server state as
+        // `"{uidvalidity}:{uidnext}"`. The UID boundary is the mailbox's
+        // authoritative UIDNEXT, else one past the highest UID actually seen.
+        // A checkpoint is only meaningful when both a UIDVALIDITY and a UID
+        // boundary are known; if either is missing (a server that omitted
+        // UIDVALIDITY, or an empty fetch that revealed no UIDNEXT/UID), keep
+        // the prior checkpoint rather than writing a validity-less one that the
+        // next sync could misread. As a last resort (no prior checkpoint
+        // either) emit an intentionally unparseable marker so the next sync
+        // safely falls back to a full fetch.
+        let boundary_uid = server_uid_next
+            .or_else(|| (max_uid_seen > 0).then_some(max_uid_seen.saturating_add(1)));
+        let new_checkpoint = match (server_uid_validity, boundary_uid) {
+            (Some(validity), Some(uid)) => format!("{}{}{}", validity, CHECKPOINT_DELIM, uid),
+            _ => since_state
+                .map(str::to_string)
+                .unwrap_or_else(|| "full-resync-required".to_string()),
+        };
+
+        Ok((emails, new_checkpoint))
     }
 
-    /// Execute `UID FETCH 1:* (FLAGS INTERNALDATE RFC822.SIZE ENVELOPE BODY.PEEK[])` and parse returned messages into domain `Email` entities.
-    pub async fn sync_folder_messages(&self, folder_id: &str) -> Result<Vec<Email>, MailError> {
+    /// Fetch a folder's messages over a freshly authenticated session,
+    /// honoring `since_state` for an incremental fetch and returning the new
+    /// sync checkpoint alongside the messages. See
+    /// [`Self::sync_folder_messages_with_session`] for the checkpoint model.
+    pub async fn sync_folder_messages(
+        &self,
+        folder_id: &str,
+        since_state: Option<&str>,
+    ) -> Result<(Vec<Email>, String), MailError> {
         let (username, password) = match (&self.username, &self.password) {
             (Some(u), Some(p)) => (u.as_str(), p.as_str()),
             _ => {
+                // Credential-less fixture path used by unit tests that never
+                // reach a real server. The checkpoint (one past the fixture's
+                // UID) is internally consistent with the fixture rather than
+                // derived from `emails.len()`.
                 let mock_emails = vec![Email {
                     id: "imap-uid-100".to_string(),
                     account_id: self.account_id.clone(),
@@ -580,7 +759,7 @@ impl ImapEngine {
                     body_html: None,
                     attachments: Vec::new(),
                 }];
-                return Ok(mock_emails);
+                return Ok((mock_emails, "101".to_string()));
             }
         };
 
@@ -588,11 +767,11 @@ impl ImapEngine {
             .socket_manager
             .connect_session(username, password)
             .await?;
-        let emails = self
-            .sync_folder_messages_with_session(folder_id, &mut session)
-            .await?;
+        let result = self
+            .sync_folder_messages_with_session(folder_id, since_state, &mut session)
+            .await;
         let _ = session.logout().await;
-        Ok(emails)
+        result
     }
 
     /// Listen for real-time IMAP IDLE notification events on an active IMAP session.
@@ -742,11 +921,9 @@ impl MailBackend for ImapEngine {
     async fn sync_messages(
         &self,
         folder_id: &str,
-        _since_state: Option<&str>,
+        since_state: Option<&str>,
     ) -> Result<(Vec<Email>, String), MailError> {
-        let emails = self.sync_folder_messages(folder_id).await?;
-        let modseq = format!("imap-modseq-{}", emails.len());
-        Ok((emails, modseq))
+        self.sync_folder_messages(folder_id, since_state).await
     }
 
     async fn send_email(&self, _email: &Email) -> Result<(), MailError> {
@@ -781,8 +958,9 @@ mod tests {
         assert_eq!(folders.len(), 2);
         assert_eq!(folders[0].id, "INBOX");
 
-        let (emails, modseq) = engine.sync_messages("INBOX", None).await?;
-        assert_eq!(modseq, "imap-modseq-1");
+        let (emails, checkpoint) = engine.sync_messages("INBOX", None).await?;
+        // Credential-less fixture: checkpoint is one past the fixture UID.
+        assert_eq!(checkpoint, "101");
         assert_eq!(emails.len(), 1);
         assert_eq!(emails[0].id, "imap-uid-100");
 
@@ -790,6 +968,342 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn resolve_fetch_range_distinguishes_none_valid_corrupt_and_renumbered_checkpoints() {
+        // First-ever sync: full history (current UIDVALIDITY irrelevant).
+        assert_eq!(resolve_fetch_range(None, Some(1)), FetchRange::Full);
+        assert_eq!(resolve_fetch_range(None, None).sequence_set(), "1:*");
+
+        // Valid checkpoint whose UIDVALIDITY still matches: narrowed fetch.
+        assert_eq!(
+            resolve_fetch_range(Some("1:105"), Some(1)),
+            FetchRange::Incremental(105)
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("1:105"), Some(1)).sequence_set(),
+            "105:*"
+        );
+        assert_eq!(
+            resolve_fetch_range(Some(" 7 : 42 "), Some(7)),
+            FetchRange::Incremental(42)
+        );
+
+        // UIDVALIDITY changed (mailbox renumbered) or the server reported none
+        // to compare against: MUST fall back to a full fetch, distinctly, so a
+        // renumber can never silently skip the low new UIDs.
+        assert_eq!(
+            resolve_fetch_range(Some("1:105"), Some(2)),
+            FetchRange::UidValidityChanged {
+                stored: 1,
+                current: Some(2)
+            }
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("1:105"), Some(2)).sequence_set(),
+            "1:*"
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("1:105"), None),
+            FetchRange::UidValidityChanged {
+                stored: 1,
+                current: None
+            }
+        );
+
+        // Malformed or legacy bare-number (pre-UIDVALIDITY) checkpoints are
+        // corrupt -> full fetch, never misread as a live UID boundary.
+        assert_eq!(
+            resolve_fetch_range(Some(""), Some(1)),
+            FetchRange::CorruptCheckpoint
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("105"), Some(1)),
+            FetchRange::CorruptCheckpoint,
+            "a legacy bare-number checkpoint must not be read as a UID boundary"
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("1:0"), Some(1)),
+            FetchRange::CorruptCheckpoint
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("not:auid"), Some(1)),
+            FetchRange::CorruptCheckpoint
+        );
+        assert_eq!(
+            resolve_fetch_range(Some("not-a-uid"), Some(1)).sequence_set(),
+            "1:*"
+        );
+    }
+
+    /// Read a single CRLF-terminated line from a scripted-server stream,
+    /// returning `None` at EOF (client hung up). Used only by the in-memory
+    /// duplex tests below; the real fetch path never parses raw lines.
+    async fn read_scripted_line<R: AsyncRead + Unpin>(reader: &mut R) -> Option<String> {
+        let mut buf = Vec::new();
+        let mut byte = [0u8; 1];
+        loop {
+            match reader.read(&mut byte).await {
+                Ok(0) => {
+                    return if buf.is_empty() {
+                        None
+                    } else {
+                        Some(String::from_utf8_lossy(&buf).into_owned())
+                    }
+                }
+                Ok(_) => {
+                    if byte[0] == b'\n' {
+                        return Some(String::from_utf8_lossy(&buf).into_owned());
+                    }
+                    if byte[0] != b'\r' {
+                        buf.push(byte[0]);
+                    }
+                }
+                Err(_) => return None,
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn incremental_sync_narrows_the_fetch_range_on_the_second_pass() {
+        // Drives a real `async_imap::Session` over an in-memory duplex pair
+        // whose server side is a scripted IMAP responder. Proves the SECOND
+        // sync issues a GENUINELY narrower `UID FETCH` derived from the first
+        // sync's returned checkpoint -- not a full re-fetch.
+        let (client_io, server_io) = tokio::io::duplex(8192);
+        let fetch_ranges = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
+        let server_ranges = fetch_ranges.clone();
+
+        let server = tokio::spawn(async move {
+            let mut io = server_io;
+            while let Some(line) = read_scripted_line(&mut io).await {
+                let tag = line.split_whitespace().next().unwrap_or("").to_string();
+                let upper = line.to_ascii_uppercase();
+                if upper.contains("UID FETCH") {
+                    if let Ok(mut g) = server_ranges.lock() {
+                        g.push(line.clone());
+                    }
+                    let _ = io
+                        .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("LOGIN") {
+                    let _ = io
+                        .write_all(format!("{tag} OK Logged in\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("SELECT") {
+                    let resp = format!(
+                        "* FLAGS (\\Seen)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\
+                         * OK [UIDVALIDITY 1] ok\r\n* OK [UIDNEXT 105] ok\r\n\
+                         {tag} OK [READ-WRITE] SELECT done\r\n"
+                    );
+                    let _ = io.write_all(resp.as_bytes()).await;
+                } else if upper.contains("LOGOUT") {
+                    let _ = io
+                        .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes())
+                        .await;
+                    break;
+                }
+            }
+        });
+
+        let client = async_imap::Client::new(client_io);
+        let mut session = client
+            .login("user", "pass")
+            .await
+            .map_err(|(e, _)| e)
+            .expect("scripted login succeeds");
+
+        let engine = ImapEngine::new("acct-1", "example.test", 993);
+
+        // First sync: no checkpoint -> full history; returns
+        // "{uidvalidity}:{uidnext}" from real server state.
+        let (first, checkpoint) = engine
+            .sync_folder_messages_with_session("INBOX", None, &mut session)
+            .await
+            .expect("first sync succeeds");
+        assert!(first.is_empty());
+        assert_eq!(checkpoint, "1:105");
+
+        // Second sync: feed back the checkpoint; UIDVALIDITY still matches
+        // (1), so the fetch is narrowed.
+        let (_second, checkpoint2) = engine
+            .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session)
+            .await
+            .expect("second sync succeeds");
+        assert_eq!(checkpoint2, "1:105");
+
+        drop(session);
+        let _ = server.await;
+
+        let ranges = fetch_ranges.lock().expect("lock captured ranges");
+        assert_eq!(ranges.len(), 2, "expected exactly two UID FETCH commands");
+        assert!(
+            ranges[0].contains("1:*"),
+            "first sync must fetch full history, got: {}",
+            ranges[0]
+        );
+        assert!(
+            ranges[1].contains("105:*"),
+            "second sync must be narrowed to the checkpoint, got: {}",
+            ranges[1]
+        );
+        assert!(
+            !ranges[1].contains("1:*"),
+            "second sync must NOT re-fetch full history, got: {}",
+            ranges[1]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_uidvalidity_change_forces_a_full_resync_instead_of_skipping_mail() {
+        // A mailbox renumber changes UIDVALIDITY and restarts UIDs from low
+        // numbers. Resuming from the old UID boundary (`105:*`) would return
+        // only the single highest message and silently drop the rest, so a
+        // UIDVALIDITY mismatch MUST force a full `1:*` re-fetch. The scripted
+        // server reports UIDVALIDITY 1 on the first SELECT and 2 on the second.
+        let (client_io, server_io) = tokio::io::duplex(8192);
+        let fetch_ranges = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
+        let server_ranges = fetch_ranges.clone();
+
+        let server = tokio::spawn(async move {
+            let mut io = server_io;
+            let mut select_count = 0u32;
+            while let Some(line) = read_scripted_line(&mut io).await {
+                let tag = line.split_whitespace().next().unwrap_or("").to_string();
+                let upper = line.to_ascii_uppercase();
+                if upper.contains("UID FETCH") {
+                    if let Ok(mut g) = server_ranges.lock() {
+                        g.push(line.clone());
+                    }
+                    let _ = io
+                        .write_all(format!("{tag} OK FETCH completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("LOGIN") {
+                    let _ = io
+                        .write_all(format!("{tag} OK Logged in\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("SELECT") {
+                    select_count += 1;
+                    // First SELECT: UIDVALIDITY 1. Second: a DIFFERENT
+                    // UIDVALIDITY (2), i.e. the mailbox was renumbered.
+                    let uidvalidity = if select_count == 1 { 1 } else { 2 };
+                    let resp = format!(
+                        "* FLAGS (\\Seen)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\
+                         * OK [UIDVALIDITY {uidvalidity}] ok\r\n* OK [UIDNEXT 105] ok\r\n\
+                         {tag} OK [READ-WRITE] SELECT done\r\n"
+                    );
+                    let _ = io.write_all(resp.as_bytes()).await;
+                } else if upper.contains("LOGOUT") {
+                    let _ = io
+                        .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes())
+                        .await;
+                    break;
+                }
+            }
+        });
+
+        let client = async_imap::Client::new(client_io);
+        let mut session = client
+            .login("user", "pass")
+            .await
+            .map_err(|(e, _)| e)
+            .expect("scripted login succeeds");
+
+        let engine = ImapEngine::new("acct-1", "example.test", 993);
+
+        // First sync under UIDVALIDITY 1 stores "1:105".
+        let (_first, checkpoint) = engine
+            .sync_folder_messages_with_session("INBOX", None, &mut session)
+            .await
+            .expect("first sync succeeds");
+        assert_eq!(checkpoint, "1:105");
+
+        // Second sync: server now reports UIDVALIDITY 2. The stored checkpoint
+        // is stale, so the fetch MUST be a full `1:*`, and the checkpoint is
+        // rewritten under the new UIDVALIDITY.
+        let (_second, checkpoint2) = engine
+            .sync_folder_messages_with_session("INBOX", Some(&checkpoint), &mut session)
+            .await
+            .expect("second sync succeeds");
+        assert_eq!(
+            checkpoint2, "2:105",
+            "checkpoint must be rewritten under the new UIDVALIDITY"
+        );
+
+        drop(session);
+        let _ = server.await;
+
+        let ranges = fetch_ranges.lock().expect("lock captured ranges");
+        assert_eq!(ranges.len(), 2, "expected exactly two UID FETCH commands");
+        assert!(
+            ranges[0].contains("1:*"),
+            "first sync fetches full history, got: {}",
+            ranges[0]
+        );
+        assert!(
+            ranges[1].contains("1:*") && !ranges[1].contains("105:*"),
+            "a UIDVALIDITY change MUST force a full re-fetch (1:*), never 105:*, got: {}",
+            ranges[1]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_never_responding_fetch_surfaces_fetchstalled_without_hanging() {
+        // The scripted server answers LOGIN and SELECT but then goes silent on
+        // the FETCH. With the tokio clock paused, the bounded per-item timeout
+        // fast-forwards, so the stall is turned into an honest `FetchStalled`
+        // error in negligible WALL-CLOCK time -- proving it never hangs.
+        let (client_io, server_io) = tokio::io::duplex(8192);
+
+        let server = tokio::spawn(async move {
+            let mut io = server_io;
+            while let Some(line) = read_scripted_line(&mut io).await {
+                let tag = line.split_whitespace().next().unwrap_or("").to_string();
+                let upper = line.to_ascii_uppercase();
+                if upper.contains("UID FETCH") {
+                    // Deliberately never respond: a mid-stream server stall.
+                    std::future::pending::<()>().await;
+                } else if upper.contains("LOGIN") {
+                    let _ = io
+                        .write_all(format!("{tag} OK Logged in\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("SELECT") {
+                    let resp = format!(
+                        "* 1 EXISTS\r\n* OK [UIDVALIDITY 1] ok\r\n* OK [UIDNEXT 42] ok\r\n\
+                         {tag} OK [READ-WRITE] SELECT done\r\n"
+                    );
+                    let _ = io.write_all(resp.as_bytes()).await;
+                }
+            }
+        });
+
+        let client = async_imap::Client::new(client_io);
+        let mut session = client
+            .login("user", "pass")
+            .await
+            .map_err(|(e, _)| e)
+            .expect("scripted login succeeds");
+        let engine = ImapEngine::new("acct-1", "example.test", 993);
+
+        // Real wall clock, not the (paused) tokio clock: proves negligible
+        // elapsed time despite the 60s virtual timeout expiring.
+        let started = std::time::Instant::now();
+        let err = engine
+            .sync_folder_messages_with_session("INBOX", None, &mut session)
+            .await
+            .expect_err("a never-responding FETCH must surface an error");
+        assert!(
+            matches!(err, MailError::FetchStalled(_)),
+            "expected FetchStalled, got: {err:?}"
+        );
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "stalled fetch must not hang in wall-clock time, took {:?}",
+            started.elapsed()
+        );
+
+        server.abort();
+    }
+
     #[tokio::test]
     async fn with_tls_mode_records_mode_and_resolves_default_ports() {
         // A `0` port resolves to the conventional default for each mode.

crates/nuncio-mail/src/mock.rs

@@ -14,12 +14,41 @@ pub struct MockMailBackend {
     messages: Arc<Mutex<Vec<Email>>>,
     sent_messages: Arc<Mutex<Vec<Email>>>,
     should_fail: Arc<Mutex<bool>>,
+    /// Every `since_state` argument this mock's `sync_messages` was called
+    /// with, in call order, so tests can prove a caller threads the returned
+    /// checkpoint back into the next sync (real incrementality).
+    since_state_calls: Arc<Mutex<Vec<Option<String>>>>,
+    /// The checkpoint token this mock returns from `sync_messages`.
+    returned_state: Arc<Mutex<String>>,
 }
 
 impl MockMailBackend {
     /// Create a new `MockMailBackend` with empty storage.
     pub fn new() -> Self {
-        Self::default()
+        let backend = Self::default();
+        if let Ok(mut state) = backend.returned_state.lock() {
+            *state = "mock-state-token-100".to_string();
+        }
+        backend
+    }
+
+    /// Override the checkpoint token this mock returns from `sync_messages`,
+    /// so a test can control the exact state a caller must persist and thread
+    /// back on the next sync.
+    pub fn set_returned_state(&self, state: &str) {
+        if let Ok(mut guard) = self.returned_state.lock() {
+            *guard = state.to_string();
+        }
+    }
+
+    /// Every `since_state` argument passed to `sync_messages`, in call order.
+    /// A caller that genuinely resumes from a checkpoint will show `None` on
+    /// the first call and the previously-returned state on the next.
+    pub fn since_state_calls(&self) -> Vec<Option<String>> {
+        self.since_state_calls
+            .lock()
+            .map(|guard| guard.clone())
+            .unwrap_or_default()
     }
 
     /// Configure the mock to simulate network failure errors.
@@ -74,8 +103,14 @@ impl MailBackend for MockMailBackend {
     async fn sync_messages(
         &self,
         folder_id: &str,
-        _since_state: Option<&str>,
+        since_state: Option<&str>,
     ) -> Result<(Vec<Email>, String), 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() {
+            calls.push(since_state.map(str::to_string));
+        }
+
         let should_fail = self
             .should_fail
             .lock()
@@ -85,6 +120,19 @@ impl MailBackend for MockMailBackend {
                 "simulated network failure".to_string(),
             ));
         }
+        let returned_state = self
+            .returned_state
+            .lock()
+            .map_err(|e| MailError::ParseFailed(e.to_string()))?
+            .clone();
+
+        // Simulate a since_state-aware backend: with a prior checkpoint there
+        // 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));
+        }
+
         let messages = self
             .messages
             .lock()
@@ -94,7 +142,7 @@ impl MailBackend for MockMailBackend {
             .filter(|m| m.folder_id == folder_id)
             .cloned()
             .collect();
-        Ok((matches, "mock-state-token-100".to_string()))
+        Ok((matches, returned_state))
     }
 
     async fn send_email(&self, email: &Email) -> Result<(), MailError> {

crates/nuncio-mail/src/parser.rs

@@ -32,6 +32,14 @@ pub enum MailError {
     #[error("IMAP error: {0}")]
     ImapError(String),
 
+    /// A single FETCH response item exceeded its bounded read timeout without
+    /// the server responding. Distinct from a generic [`MailError::ImapError`]
+    /// so callers can tell an honest mid-stream stall (dead socket, server-side
+    /// hang) apart from a protocol/parse failure instead of the sync hanging
+    /// forever.
+    #[error("IMAP FETCH stalled: {0}")]
+    FetchStalled(String),
+
     /// Network or IO error.
     #[error("network I/O error: {0}")]
     NetworkError(String),

crates/nuncio-store/src/db.rs

@@ -649,6 +649,19 @@ impl DatabaseEngine {
                 INSERT INTO contacts_fts(id, display_name, organization, emails_json)
                 VALUES (new.id, new.display_name, COALESCE(new.organization, ''), new.emails_json);
             END;
+
+            -- Per-folder inbound-sync checkpoint. `state` is an opaque,
+            -- protocol-defined resumption token (an IMAP UID boundary such as
+            -- UIDNEXT, a JMAP state string) that a backend hands back after a
+            -- sync and expects to receive on the next sync so it can fetch only
+            -- what changed since. Keyed by (account, folder) because the token
+            -- is meaningful only within a single mailbox on a single account.
+            CREATE TABLE IF NOT EXISTS folder_sync_state (
+                account_id TEXT NOT NULL,
+                folder_id TEXT NOT NULL,
+                state TEXT NOT NULL,
+                PRIMARY KEY (account_id, folder_id)
+            );
             "#,
         )
         .execute(&self.pool)
@@ -915,6 +928,51 @@ impl DatabaseEngine {
             .collect())
     }
 
+    /// Fetch the persisted inbound-sync checkpoint for a folder, or `None` if
+    /// this account/folder pair has never completed a sync. A `None` result is
+    /// the signal to the backend that it must perform a full first-time fetch.
+    pub async fn get_folder_sync_state(
+        &self,
+        account_id: &str,
+        folder_id: &str,
+    ) -> Result<Option<String>, DatabaseError> {
+        let row: Option<(String,)> = sqlx::query_as(
+            "SELECT state FROM folder_sync_state WHERE account_id = ? AND folder_id = ?",
+        )
+        .bind(account_id)
+        .bind(folder_id)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(DatabaseError::Query)?;
+        Ok(row.map(|(state,)| state))
+    }
+
+    /// Persist the inbound-sync checkpoint a backend returned for a folder,
+    /// overwriting any prior value (a checkpoint is a running high-water mark,
+    /// not history). Written only after a folder's messages have been fetched
+    /// and persisted, so the stored checkpoint can never advance past work that
+    /// actually landed in the store.
+    pub async fn save_folder_sync_state(
+        &self,
+        account_id: &str,
+        folder_id: &str,
+        state: &str,
+    ) -> Result<(), DatabaseError> {
+        sqlx::query(
+            r#"
+            INSERT OR REPLACE INTO folder_sync_state (account_id, folder_id, state)
+            VALUES (?, ?, ?)
+            "#,
+        )
+        .bind(account_id)
+        .bind(folder_id)
+        .bind(state)
+        .execute(&self.pool)
+        .await
+        .map_err(DatabaseError::Query)?;
+        Ok(())
+    }
+
     /// Save an [`nuncio_core::model::Email`] to SQLite (INSERT OR REPLACE).
     ///
     /// The message body is encrypted (AES-256-GCM) before being written to the `messages`
@@ -2696,6 +2754,61 @@ mod tests {
             .expect("second delete is a no-op");
     }
 
+    #[tokio::test]
+    async fn folder_sync_state_round_trip_is_scoped_and_overwrites() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+
+        // Absent before any sync completes.
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-1", "INBOX")
+                .await
+                .expect("get succeeds"),
+            None
+        );
+
+        engine
+            .save_folder_sync_state("acct-1", "INBOX", "105")
+            .await
+            .expect("save succeeds");
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-1", "INBOX")
+                .await
+                .expect("get succeeds"),
+            Some("105".to_string())
+        );
+
+        // A later checkpoint overwrites (high-water mark, not history).
+        engine
+            .save_folder_sync_state("acct-1", "INBOX", "220")
+            .await
+            .expect("save succeeds");
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-1", "INBOX")
+                .await
+                .expect("get succeeds"),
+            Some("220".to_string())
+        );
+
+        // Keyed by (account, folder): a different folder/account is unaffected.
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-1", "Sent")
+                .await
+                .expect("get succeeds"),
+            None
+        );
+        assert_eq!(
+            engine
+                .get_folder_sync_state("acct-2", "INBOX")
+                .await
+                .expect("get succeeds"),
+            None
+        );
+    }
+
     /// Proves the additive `smtp_host`/`smtp_port`
     /// migration is backfill-safe. Simulates a database created BEFORE this
     /// feature existed (an `accounts` table with no `smtp_host`/`smtp_port`

crates/nunciod/src/grpc.rs

@@ -87,15 +87,17 @@ pub enum GrpcServeError {
 }
 
 /// Maps a `nuncio_core::CoreEvent` domain event onto its wire-format
-/// `nuncio.v1.Event` representation, faithfully carrying every variant's
+/// `nuncio.v1.Event` representation, carrying every wire-visible variant's
 /// fields into the matching `oneof` case (see `proto/nuncio/v1/nuncio.proto`
-/// for the wire contract).
+/// for the wire contract). Returns `None` for events that are internal-only
+/// and have no case in the versioned contract, so they are dropped from the
+/// gRPC subscription stream rather than forced onto the wire.
 ///
 /// `usize` fields (`processed`, `total`, `matched`, `salvaged_rules_count`)
 /// are narrowed to `u64` for the wire; these are in-process counters that
 /// cannot realistically approach `u64::MAX`, and protobuf has no native
 /// `usize` type.
-fn map_core_event(event: CoreEvent) -> Event {
+fn map_core_event(event: CoreEvent) -> Option<Event> {
     let kind = match event {
         CoreEvent::SyncStarted { account_id } => Kind::SyncStarted(SyncStarted { account_id }),
         CoreEvent::SyncCompleted { account_id } => {
@@ -140,8 +142,12 @@ fn map_core_event(event: CoreEvent) -> Event {
         }),
         CoreEvent::Error { message } => Kind::Error(EventError { message }),
         CoreEvent::ShuttingDown => Kind::ShuttingDown(ShuttingDown {}),
+        // Internal sync-progress signal: intentionally not forwarded to gRPC
+        // subscribers, as it has no case in the versioned wire contract. It is
+        // observed in-process (e.g. by the daemon's own subscribers) only.
+        CoreEvent::SyncProgress { .. } => return None,
     };
-    Event { kind: Some(kind) }
+    Some(Event { kind: Some(kind) })
 }
 
 /// `nuncio.v1.System` gRPC service implementation backed by the daemon's live
@@ -191,7 +197,7 @@ impl System for SystemGrpcService {
     ) -> Result<Response<Self::SubscribeStream>, Status> {
         let receiver = self.event_bus.subscribe_events();
         let mapped = BroadcastStream::new(receiver).filter_map(|item| match item {
-            Ok(core_event) => Some(Ok(map_core_event(core_event))),
+            Ok(core_event) => map_core_event(core_event).map(Ok),
             Err(BroadcastStreamRecvError::Lagged(skipped)) => {
                 tracing::warn!(
                     "gRPC Subscribe stream lagged behind the daemon event bus; skipped {} \

crates/nunciod/src/sync.rs

@@ -18,7 +18,7 @@
 //! only places that resolve a real account's credentials and construct a
 //! real backend; they then delegate the actual fetch/persist/event-emission
 //! work to [`sync_with_backend`].
-use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, EventBus};
+use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, CoreEvent, EventBus};
 use nuncio_filter::{FilterEngine, OutboxManager, RuleAction};
 use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError};
 use nuncio_store::db::DatabaseError;
@@ -235,13 +235,27 @@ async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str)
 /// [`MockMailBackend`].
 async fn fetch_and_persist(
     db: &nuncio_store::db::DatabaseEngine,
+    event_bus: &EventBus,
     backend: &dyn MailBackend,
     filter_engine: &FilterEngine,
+    account_id: Option<&str>,
 ) -> Result<usize, SyncError> {
     let folders = backend.sync_folders().await?;
     let mut synced = 0usize;
     for folder in folders {
-        let (emails, _state) = backend.sync_messages(&folder.id, None).await?;
+        // Resume from the folder's stored checkpoint so the backend can fetch
+        // only what changed since the last sync. A checkpoint is keyed by
+        // (account, folder), so a run with no account context (a mock-driven
+        // `SyncAll` with `account_id == None`) has nowhere to key it and always
+        // performs a full fetch.
+        let last_state = match account_id {
+            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())
+            .await?;
+        let fetched = emails.len();
         for email in emails {
             let is_new = is_new_message(db, &email.id).await;
             db.save_email(&email).await?;
@@ -250,6 +264,19 @@ async fn fetch_and_persist(
             }
             synced += 1;
         }
+        // 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.
+        if let Some(acct) = account_id {
+            db.save_folder_sync_state(acct, &folder.id, &new_state)
+                .await?;
+        }
+        event_bus.publish_event(CoreEvent::SyncProgress {
+            account_id: account_id.map(str::to_string),
+            folder_id: folder.id.clone(),
+            fetched,
+            total: synced,
+        });
     }
     Ok(synced)
 }
@@ -282,7 +309,8 @@ pub async fn sync_with_backend(
         None => event_bus.process_command(CoreCommand::SyncAll),
     }
 
-    let result = fetch_and_persist(db, backend, filter_engine).await;
+    let result =
+        fetch_and_persist(db, event_bus, backend, filter_engine, account_id.as_deref()).await;
 
     if let Err(e) = &result {
         event_bus.process_command(CoreCommand::ReportError {
@@ -394,7 +422,15 @@ pub async fn run_all_accounts_sync(
                 match secrets.get_secret(&config.keyring_secret_key) {
                     Ok(password) => {
                         let backend = build_mail_backend(&config, &password);
-                        match fetch_and_persist(db, backend.as_ref(), filter_engine).await {
+                        match fetch_and_persist(
+                            db,
+                            event_bus,
+                            backend.as_ref(),
+                            filter_engine,
+                            Some(&config.id),
+                        )
+                        .await
+                        {
                             Ok(count) => total_synced += count,
                             Err(e) => {
                                 event_bus.process_command(CoreCommand::ReportError {
@@ -619,6 +655,16 @@ mod tests {
                 account_id: Some("acct-mock-1".to_string())
             }
         );
+        // A per-folder progress event is published between start and complete.
+        assert_eq!(
+            events.recv().await.expect("progress event"),
+            CoreEvent::SyncProgress {
+                account_id: Some("acct-mock-1".to_string()),
+                folder_id: "inbox".to_string(),
+                fetched: 2,
+                total: 2,
+            }
+        );
         assert_eq!(
             events.recv().await.expect("complete event"),
             CoreEvent::SyncCompleted {
@@ -654,6 +700,16 @@ mod tests {
             events.recv().await.expect("start event"),
             CoreEvent::SyncStarted { account_id: None }
         );
+        // The folder is empty, but a progress event still marks it done.
+        assert_eq!(
+            events.recv().await.expect("progress event"),
+            CoreEvent::SyncProgress {
+                account_id: None,
+                folder_id: "inbox".to_string(),
+                fetched: 0,
+                total: 0,
+            }
+        );
         assert_eq!(
             events.recv().await.expect("complete event"),
             CoreEvent::SyncCompleted { account_id: None }
@@ -791,6 +847,15 @@ mod tests {
                 account_id: Some("acct-jmap-real-1".to_string())
             }
         );
+        assert_eq!(
+            events.recv().await.expect("progress event"),
+            CoreEvent::SyncProgress {
+                account_id: Some("acct-jmap-real-1".to_string()),
+                folder_id: "jmap-inbox".to_string(),
+                fetched: 1,
+                total: 1,
+            }
+        );
         assert_eq!(
             events.recv().await.expect("complete event"),
             CoreEvent::SyncCompleted {
@@ -1202,4 +1267,69 @@ mod tests {
             "no new outbox mutation should be enqueued on a repeat sync"
         );
     }
+
+    #[tokio::test]
+    async fn sync_with_backend_persists_and_threads_the_folder_checkpoint_across_syncs() {
+        // The regression this story exists to kill: the returned checkpoint
+        // must be persisted and fed back into the NEXT sync, so the second
+        // sync is genuinely incremental rather than a full re-fetch. Proven by
+        // capturing the exact `since_state` the backend received on each call.
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+
+        let mock = MockMailBackend::new();
+        mock.set_returned_state("uidnext-4242");
+        mock.add_folder(Folder {
+            id: "inbox".to_string(),
+            name: "Inbox".to_string(),
+            total_messages: 1,
+            unread_messages: 1,
+        });
+        mock.add_message(mock_email("m-inc-1", "inbox", "First"));
+
+        let filter_engine = empty_filter_engine();
+
+        // First sync: no checkpoint exists yet, so the backend is called with
+        // `None` and does a full fetch (one message).
+        let first = sync_with_backend(
+            &db,
+            &event_bus,
+            &mock,
+            &filter_engine,
+            Some("acct-inc-1".to_string()),
+        )
+        .await
+        .expect("first sync succeeds");
+        assert_eq!(first, 1);
+
+        // The returned checkpoint must now be persisted for (account, folder).
+        assert_eq!(
+            db.get_folder_sync_state("acct-inc-1", "inbox")
+                .await
+                .expect("read checkpoint"),
+            Some("uidnext-4242".to_string())
+        );
+
+        // Second sync: the daemon must feed the stored checkpoint back in. The
+        // mock returns nothing new for a non-None checkpoint -> narrower fetch.
+        let second = sync_with_backend(
+            &db,
+            &event_bus,
+            &mock,
+            &filter_engine,
+            Some("acct-inc-1".to_string()),
+        )
+        .await
+        .expect("second sync succeeds");
+        assert_eq!(second, 0, "an incremental second sync fetches nothing new");
+
+        // The proof of genuine incrementality: the backend saw `None` first,
+        // then the exact checkpoint the first sync returned.
+        assert_eq!(
+            mock.since_state_calls(),
+            vec![None, Some("uidnext-4242".to_string())],
+        );
+    }
 }