From 6d62c09fcdd1a5886f34b8d1ab4563f4cf6528df Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Fri, 31 Jul 2026 13:55:17 -0500
Subject: [PATCH 1/5] feat(nunciod): graceful shutdown with WAL-checkpointing
 close (#340)

Install a Ctrl+C/SIGTERM signal handler and share one shutdown signal
across the real-sync command consumer, the outbox worker, the
update-check loop, and the gRPC server, so a clean exit drains every
task and forces the WAL checkpoint via DatabaseEngine::close() instead
of the OS killing the process mid-write. The signal also finally
triggers the existing CoreCommand::Shutdown / CoreEvent::ShuttingDown
event-bus path outside of tests.

Refs #289

Co-authored-by: James Maes <jmaes@dmdbrands.com>
---
 crates/nunciod/src/grpc.rs                    | 108 ++++++++++-
 crates/nunciod/src/lib.rs                     |   1 +
 crates/nunciod/src/lifecycle.rs               | 157 ++++++++++++++++
 crates/nunciod/src/main.rs                    | 168 +++++++++++++----
 .../nunciod/tests/graceful_shutdown_test.rs   | 176 ++++++++++++++++++
 5 files changed, 572 insertions(+), 38 deletions(-)
 create mode 100644 crates/nunciod/src/lifecycle.rs
 create mode 100644 crates/nunciod/tests/graceful_shutdown_test.rs

diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs
index 23f0f66..5aa2251 100644
--- a/crates/nunciod/src/grpc.rs
+++ b/crates/nunciod/src/grpc.rs
@@ -2283,6 +2283,73 @@ pub async fn serve_on_listener(
     .await
 }
 
+/// Binds a loopback TCP listener at `addr` exactly like [`serve`], but stops
+/// accepting new connections and begins draining in-flight requests as soon
+/// as `shutdown` resolves, instead of running until the transport errors.
+///
+/// `addr` MUST be a loopback address; see [`serve`].
+pub async fn serve_with_shutdown(
+    addr: &str,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
+) -> Result<(), GrpcServeError> {
+    let socket_addr = addr
+        .parse::<std::net::SocketAddr>()
+        .map_err(|_| GrpcServeError::NonLoopbackAddress(addr.to_string()))?;
+    if !socket_addr.ip().is_loopback() {
+        return Err(GrpcServeError::NonLoopbackAddress(addr.to_string()));
+    }
+
+    let listener = TcpListener::bind(addr)
+        .await
+        .map_err(|source| GrpcServeError::Bind {
+            addr: addr.to_string(),
+            source,
+        })?;
+    serve_on_listener_with_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        shutdown,
+    )
+    .await
+}
+
+/// Identical to [`serve_on_listener`], except the server stops accepting new
+/// connections and begins draining in-flight requests as soon as `shutdown`
+/// resolves.
+pub async fn serve_on_listener_with_shutdown(
+    listener: TcpListener,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
+) -> Result<(), GrpcServeError> {
+    serve_on_listener_with_overrides_and_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        MailEngineOverrides::default(),
+        CalendarEngineOverrides::default(),
+        ContactsEngineOverrides::default(),
+        AccountsEngineOverrides::default(),
+        shutdown,
+    )
+    .await
+}
+
 /// Identical to [`serve_on_listener`], except the `nuncio.v1.Mail` service's
 /// inbound sync and outbound send engines, the `nuncio.v1.Calendar`
 /// service's sync backend, and the `nuncio.v1.Contacts` service's sync
@@ -2306,6 +2373,45 @@ pub async fn serve_on_listener_with_overrides(
     calendar_overrides: CalendarEngineOverrides,
     contacts_overrides: ContactsEngineOverrides,
     accounts_overrides: AccountsEngineOverrides,
+) -> Result<(), GrpcServeError> {
+    // Delegates to the shutdown-aware variant with a shutdown future that
+    // never resolves, preserving this function's exact prior behavior (run
+    // until the transport itself errors) for every existing caller.
+    serve_on_listener_with_overrides_and_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        overrides,
+        calendar_overrides,
+        contacts_overrides,
+        accounts_overrides,
+        std::future::pending(),
+    )
+    .await
+}
+
+/// Identical to [`serve_on_listener_with_overrides`], except the server
+/// stops accepting new connections and begins draining in-flight requests
+/// as soon as `shutdown` resolves, instead of running until the transport
+/// errors. This is the function production actually calls (via
+/// [`serve_with_shutdown`]); [`serve_on_listener_with_overrides`] delegates
+/// here with a `shutdown` future that never resolves.
+#[allow(clippy::too_many_arguments)]
+pub async fn serve_on_listener_with_overrides_and_shutdown(
+    listener: TcpListener,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    overrides: MailEngineOverrides,
+    calendar_overrides: CalendarEngineOverrides,
+    contacts_overrides: ContactsEngineOverrides,
+    accounts_overrides: AccountsEngineOverrides,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
 ) -> Result<(), GrpcServeError> {
     let token: Arc<str> = token.into();
 
@@ -2392,7 +2498,7 @@ pub async fn serve_on_listener_with_overrides(
         .add_service(audit_svc)
         .add_service(calendar_svc)
         .add_service(contacts_svc)
-        .serve_with_incoming(TcpListenerStream::new(listener))
+        .serve_with_incoming_shutdown(TcpListenerStream::new(listener), shutdown)
         .await
         .map_err(GrpcServeError::Transport)
 }
diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs
index 7b3413b..9a10ad9 100644
--- a/crates/nunciod/src/lib.rs
+++ b/crates/nunciod/src/lib.rs
@@ -5,6 +5,7 @@
 pub mod calendar_sync;
 pub mod contacts_sync;
 pub mod grpc;
+pub mod lifecycle;
 pub mod orchestrator;
 pub mod outbox;
 pub mod send;
diff --git a/crates/nunciod/src/lifecycle.rs b/crates/nunciod/src/lifecycle.rs
new file mode 100644
index 0000000..dcef665
--- /dev/null
+++ b/crates/nunciod/src/lifecycle.rs
@@ -0,0 +1,157 @@
+//! Coordinates graceful shutdown across every long-running task `nunciod`
+//! spawns: the gRPC server, the outbox worker, the auto-update-check loop,
+//! and the real-sync `CoreCommand` consumer.
+//!
+//! Previously the daemon installed no signal handler at all: every real
+//! exit was an OS kill (SIGKILL/SIGTERM without a handler, a terminated
+//! console), which races [`nuncio_store::DatabaseEngine::close`]'s WAL
+//! checkpoint and can corrupt the database. A single [`ShutdownSignal`],
+//! shared by every task, lets each one stop accepting new work and finish
+//! cleanly instead.
+
+use nuncio_core::{CoreCommand, EventBus};
+use std::sync::Arc;
+use tokio::sync::watch;
+
+/// Read-only handle to the shared shutdown flag. Cloned into every
+/// long-running task; `wait` resolves once shutdown has been requested,
+/// including if it was already requested before this clone started
+/// waiting.
+#[derive(Clone)]
+pub struct ShutdownSignal {
+    rx: watch::Receiver<bool>,
+}
+
+impl ShutdownSignal {
+    /// Resolves once shutdown has been requested. Safe to call from
+    /// `tokio::select!` on every loop iteration: a task that observes
+    /// shutdown mid-iteration finishes its current unit of work (the other
+    /// `select!` branch) before this is polled again, so no in-flight
+    /// database write is abandoned half-done.
+    pub async fn wait(&mut self) {
+        // `watch::Receiver::wait_for` returns immediately if the current
+        // value already satisfies the predicate, so a clone created after
+        // shutdown was already requested does not hang waiting for a
+        // change that already happened.
+        let _ = self.rx.wait_for(|requested| *requested).await;
+    }
+
+    /// Non-blocking check for loops that need to poll rather than await.
+    pub fn is_requested(&self) -> bool {
+        *self.rx.borrow()
+    }
+}
+
+/// Owns the shutdown trigger. Production wires [`ShutdownController::trigger`]
+/// to OS signal delivery via [`install_signal_handlers`]; tests call it
+/// directly to drive the shutdown path deterministically without sending a
+/// real OS signal.
+pub struct ShutdownController {
+    tx: watch::Sender<bool>,
+    event_bus: Arc<EventBus>,
+}
+
+impl ShutdownController {
+    /// Creates a controller/signal pair. `event_bus` is the SAME bus every
+    /// other subsystem shares, so [`trigger`](Self::trigger) drives the
+    /// existing `CoreCommand::Shutdown` / `CoreEvent::ShuttingDown` path
+    /// visibly (state flips to `ShuttingDown`, subscribers are notified)
+    /// instead of that path staying reachable only from tests.
+    pub fn new(event_bus: Arc<EventBus>) -> (Self, ShutdownSignal) {
+        let (tx, rx) = watch::channel(false);
+        (Self { tx, event_bus }, ShutdownSignal { rx })
+    }
+
+    /// Initiates shutdown. Idempotent: calling it more than once (e.g. a
+    /// second Ctrl+C while already draining) is a no-op beyond the first.
+    pub fn trigger(&self) {
+        self.event_bus.process_command(CoreCommand::Shutdown);
+        let _ = self.tx.send(true);
+    }
+}
+
+/// Waits for Ctrl+C, or (on Unix) SIGTERM, then triggers shutdown. Intended
+/// to be spawned once at startup and left running for the process lifetime.
+pub async fn install_signal_handlers(controller: Arc<ShutdownController>) {
+    wait_for_os_signal().await;
+    controller.trigger();
+}
+
+#[cfg(unix)]
+async fn wait_for_os_signal() {
+    use tokio::signal::unix::{signal, SignalKind};
+
+    match signal(SignalKind::terminate()) {
+        Ok(mut sigterm) => {
+            tokio::select! {
+                _ = tokio::signal::ctrl_c() => {
+                    tracing::info!("received Ctrl+C; starting graceful shutdown");
+                }
+                _ = sigterm.recv() => {
+                    tracing::info!("received SIGTERM; starting graceful shutdown");
+                }
+            }
+        }
+        Err(err) => {
+            // Ctrl+C still works even if SIGTERM registration failed (e.g.
+            // an exotic sandboxed environment); log so the gap is visible
+            // rather than silently dropping SIGTERM handling.
+            tracing::error!("failed to install SIGTERM handler: {err}; SIGTERM will kill the process without a clean shutdown");
+            let _ = tokio::signal::ctrl_c().await;
+            tracing::info!("received Ctrl+C; starting graceful shutdown");
+        }
+    }
+}
+
+#[cfg(not(unix))]
+async fn wait_for_os_signal() {
+    let _ = tokio::signal::ctrl_c().await;
+    tracing::info!("received Ctrl+C; starting graceful shutdown");
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nuncio_core::EngineStatus;
+
+    #[tokio::test]
+    async fn trigger_flips_signal_and_event_bus_state() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, mut signal) = ShutdownController::new(event_bus.clone());
+
+        assert!(!signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+
+        controller.trigger();
+
+        assert!(signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown);
+        // Resolves immediately: the signal was already flipped above.
+        signal.wait().await;
+    }
+
+    #[tokio::test]
+    async fn signal_clone_observes_a_trigger_that_happened_before_it_was_created() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, signal) = ShutdownController::new(event_bus);
+        controller.trigger();
+
+        // A clone taken AFTER the trigger must still resolve `wait`
+        // immediately rather than hang waiting for a fresh `changed()`.
+        let mut late_clone = signal.clone();
+        late_clone.wait().await;
+        assert!(late_clone.is_requested());
+    }
+
+    #[tokio::test]
+    async fn trigger_is_idempotent() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, signal) = ShutdownController::new(event_bus.clone());
+
+        controller.trigger();
+        controller.trigger();
+
+        assert!(signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown);
+    }
+}
diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs
index c98aa72..d50f90b 100644
--- a/crates/nunciod/src/main.rs
+++ b/crates/nunciod/src/main.rs
@@ -4,7 +4,16 @@
 
 use nuncio_core::{CoreCommand, CoreEvent, EventBus};
 use nuncio_filter::FilterEngine;
+use nunciod::lifecycle::{install_signal_handlers, ShutdownController};
 use std::sync::Arc;
+use std::time::Duration;
+
+/// Bound on how long shutdown waits for in-flight work (gRPC requests still
+/// being served, an outbox drain pass in progress, a background task's
+/// current loop iteration) to finish on its own before the process moves on
+/// to closing the database and exiting anyway. Keeps a stuck task from
+/// turning a graceful shutdown into a hang.
+const SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(10);
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -17,6 +26,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let mut event_bus_owned = EventBus::new();
     let command_rx = event_bus_owned.take_command_receiver();
     let event_bus = Arc::new(event_bus_owned);
+
+    // Single shutdown signal shared by every long-running task spawned
+    // below (the real-sync command consumer, the outbox worker, the
+    // update-check loop, and the gRPC server itself), so a Ctrl+C/SIGTERM
+    // drains all of them cleanly instead of the OS killing the process
+    // mid-write. `install_signal_handlers` also drives the existing
+    // `CoreCommand::Shutdown` / `CoreEvent::ShuttingDown` event-bus path,
+    // previously wired but never triggered outside tests.
+    let (shutdown_controller, shutdown_signal) = ShutdownController::new(event_bus.clone());
+    let shutdown_controller = Arc::new(shutdown_controller);
+    let signal_task = tokio::spawn(install_signal_handlers(shutdown_controller));
     // PERSISTENT database path: defaults to `~/.nuncio/nuncio.db` -- NOT a
     // temp/ephemeral path -- so accounts (and everything else) survive a
     // daemon restart.
@@ -52,13 +72,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // trigger_background_resync`'s post-recovery `send_command` calls a
     // consumer: previously nothing read this channel, so that resync
     // trigger was inert.
-    if let Some(mut command_rx) = command_rx {
+    let sync_command_task = if let Some(mut command_rx) = command_rx {
         let db_sync = db.clone();
         let secrets_sync = account_secrets.clone();
         let event_bus_sync = event_bus.clone();
         let filter_engine_sync = filter_engine.clone();
-        let _sync_command_task = tokio::spawn(async move {
-            while let Some(cmd) = command_rx.recv().await {
+        let mut shutdown = shutdown_signal.clone();
+        Some(tokio::spawn(async move {
+            loop {
+                let cmd = tokio::select! {
+                    _ = shutdown.wait() => {
+                        tracing::info!("sync command consumer exiting on shutdown signal");
+                        break;
+                    }
+                    received = command_rx.recv() => match received {
+                        Some(cmd) => cmd,
+                        None => break,
+                    },
+                };
                 match cmd {
                     CoreCommand::SyncAll => {
                         let synced = nunciod::sync::run_all_accounts_sync(
@@ -93,12 +124,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                     _ => {}
                 }
             }
-        });
+        }))
     } else {
         tracing::error!(
             "EventBus command receiver was already taken; real sync command processing will not run"
         );
-    }
+        None
+    };
 
     // Background Outbox Worker Task. Each tick drains the pending remote
     // mutations the filter engine enqueued (move/copy/flag/unflag/delete on the
@@ -112,20 +144,34 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         db.clone(),
         account_secrets.clone(),
     ));
-    let _outbox_task = tokio::spawn(async move {
+    let mut outbox_shutdown = shutdown_signal.clone();
+    let outbox_task = tokio::spawn(async move {
         let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
         loop {
-            interval.tick().await;
-            let summary =
-                nunciod::outbox::execute_pending_mutations(&db_outbox, outbox_env.as_ref(), 50)
+            tokio::select! {
+                _ = outbox_shutdown.wait() => {
+                    tracing::info!("outbox worker exiting on shutdown signal");
+                    break;
+                }
+                _ = interval.tick() => {
+                    // Runs to completion once selected: a drain pass already
+                    // in progress is never aborted mid-write by a shutdown
+                    // signal that arrives after this branch was chosen.
+                    let summary = nunciod::outbox::execute_pending_mutations(
+                        &db_outbox,
+                        outbox_env.as_ref(),
+                        50,
+                    )
                     .await;
-            if summary.completed > 0 || summary.failed > 0 {
-                tracing::info!(
-                    "outbox drain: {} completed, {} retried, {} failed",
-                    summary.completed,
-                    summary.retried,
-                    summary.failed
-                );
+                    if summary.completed > 0 || summary.failed > 0 {
+                        tracing::info!(
+                            "outbox drain: {} completed, {} retried, {} failed",
+                            summary.completed,
+                            summary.retried,
+                            summary.failed
+                        );
+                    }
+                }
             }
         }
     });
@@ -138,21 +184,29 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // closed, nunciod must never autonomously check for or install
     // updates, so this loop only runs when an operator explicitly opts in
     // via `NUNCIO_AUTO_UPDATE_ENABLED=1`.
-    let _update_task = if nunciod::auto_update_task_enabled() {
+    let update_task = if nunciod::auto_update_task_enabled() {
         let event_bus_update = event_bus.clone();
+        let mut update_shutdown = shutdown_signal.clone();
         Some(tokio::spawn(async move {
             let mut interval = tokio::time::interval(std::time::Duration::from_secs(86400));
             interval.tick().await;
             loop {
-                interval.tick().await;
-                if let Ok(updater) = nuncio_core::UpdateEngine::new() {
-                    if let Ok(result) = updater.check_for_updates().await {
-                        if result.update_available {
-                            if let Some(info) = result.release_info {
-                                event_bus_update.publish_event(CoreEvent::UpdateAvailable {
-                                    version: info.version,
-                                    release_notes: info.release_notes,
-                                });
+                tokio::select! {
+                    _ = update_shutdown.wait() => {
+                        tracing::info!("update-check loop exiting on shutdown signal");
+                        break;
+                    }
+                    _ = interval.tick() => {
+                        if let Ok(updater) = nuncio_core::UpdateEngine::new() {
+                            if let Ok(result) = updater.check_for_updates().await {
+                                if result.update_available {
+                                    if let Some(info) = result.release_info {
+                                        event_bus_update.publish_event(CoreEvent::UpdateAvailable {
+                                            version: info.version,
+                                            release_notes: info.release_notes,
+                                        });
+                                    }
+                                }
                             }
                         }
                     }
@@ -192,17 +246,57 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 
     // The gRPC server is the long-running foreground task that keeps this
     // process alive; every other subsystem above is a background worker
-    // spawned on top of it.
-    nunciod::grpc::serve(
-        &grpc_addr,
-        event_bus,
-        db,
-        filter_engine,
-        grpc_secrets,
-        grpc_token,
-    )
-    .await
-    .map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
+    // spawned on top of it. `serve_with_shutdown` stops accepting new
+    // connections and starts draining in-flight requests as soon as
+    // `shutdown_signal` fires; the `select!` below bounds how long that
+    // drain is allowed to take before this process moves on regardless.
+    let db_for_serve = db.clone();
+    let mut grpc_shutdown_future = shutdown_signal.clone();
+    let mut grace_timer_signal = shutdown_signal.clone();
+    let grpc_result = tokio::select! {
+        result = nunciod::grpc::serve_with_shutdown(
+            &grpc_addr,
+            event_bus,
+            db_for_serve,
+            filter_engine,
+            grpc_secrets,
+            grpc_token,
+            async move { grpc_shutdown_future.wait().await },
+        ) => result,
+        _ = async move {
+            grace_timer_signal.wait().await;
+            tokio::time::sleep(SHUTDOWN_GRACE_PERIOD).await;
+        } => {
+            tracing::warn!(
+                "gRPC server did not finish draining in-flight requests within the {:?} \
+                 shutdown grace period; closing the database and exiting anyway",
+                SHUTDOWN_GRACE_PERIOD
+            );
+            Ok(())
+        }
+    };
+
+    // Bound how long the background workers get to notice the shutdown
+    // signal and exit their loops before the database is closed out from
+    // under them.
+    let _ = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async {
+        if let Some(task) = sync_command_task {
+            let _ = task.await;
+        }
+        let _ = outbox_task.await;
+        if let Some(task) = update_task {
+            let _ = task.await;
+        }
+    })
+    .await;
+    signal_task.abort();
+
+    // Force the WAL checkpoint before the process exits so no committed
+    // page is left for SQLite's own deferred "last connection closes"
+    // teardown to race (see `DatabaseEngine::close`'s doc comment).
+    db.close().await;
+
+    grpc_result.map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
 
     Ok(())
 }
diff --git a/crates/nunciod/tests/graceful_shutdown_test.rs b/crates/nunciod/tests/graceful_shutdown_test.rs
new file mode 100644
index 0000000..40d784c
--- /dev/null
+++ b/crates/nunciod/tests/graceful_shutdown_test.rs
@@ -0,0 +1,176 @@
+//! Integration test for `nunciod`'s graceful shutdown path.
+//!
+//! Before this, `nunciod` installed no signal handler and never called
+//! [`DatabaseEngine::close`], so every real exit was an OS kill racing the
+//! WAL checkpoint that `close` exists to force. This test drives the same
+//! building blocks `main.rs` wires together -- a [`ShutdownController`]
+//! shared by the gRPC server and a background worker, triggered directly
+//! (never a real OS signal, for determinism) -- and asserts:
+//!   (a) the gRPC server stops serving once the shutdown signal fires;
+//!   (b) a background worker exits its loop on the same signal instead of
+//!       being aborted mid-iteration;
+//!   (c) `DatabaseEngine::close()` runs and leaves the pool unusable for
+//!       any subsequent query.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use nuncio_core::EventBus;
+use nuncio_filter::FilterEngine;
+use nuncio_proto::v1::system_client::SystemClient;
+use nuncio_proto::v1::GetStatusRequest;
+use nuncio_store::db::DatabaseEngine;
+use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
+use nunciod::grpc::serve_on_listener_with_shutdown;
+use nunciod::lifecycle::ShutdownController;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::net::TcpListener;
+use tonic::Request;
+
+const JOIN_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[tokio::test]
+async fn shutdown_signal_stops_grpc_server_exits_background_worker_and_closes_the_database() {
+    let event_bus = Arc::new(EventBus::new());
+    let (controller, shutdown_signal) = ShutdownController::new(event_bus.clone());
+
+    let secrets = Arc::new(SecretManager::mock());
+    let token = hex::encode(
+        secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .expect("token provisioned from mock vault"),
+    );
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+    let (db, _dir) = DatabaseEngine::connect_ephemeral()
+        .await
+        .expect("connect ephemeral test db");
+    let db = Arc::new(db);
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind ephemeral loopback port");
+    let addr = listener.local_addr().expect("listener has local addr");
+
+    // (a) the gRPC server: mirrors main.rs's `serve_with_shutdown` wiring --
+    // the future it's given resolves from the SAME `ShutdownSignal` the
+    // worker below observes.
+    let mut grpc_shutdown = shutdown_signal.clone();
+    let grpc_task = tokio::spawn(serve_on_listener_with_shutdown(
+        listener,
+        event_bus.clone(),
+        db.clone(),
+        filter_engine,
+        secrets,
+        token.clone(),
+        async move { grpc_shutdown.wait().await },
+    ));
+
+    // Confirm the server actually accepts a call before shutdown, so a
+    // trivially-passing "stopped" observation later isn't just "it never
+    // started".
+    let mut client = SystemClient::connect(format!("http://{addr}"))
+        .await
+        .expect("client connects before shutdown");
+    let mut request = Request::new(GetStatusRequest {});
+    request.metadata_mut().insert(
+        "authorization",
+        format!("Bearer {token}")
+            .parse()
+            .expect("valid ascii metadata value"),
+    );
+    client
+        .get_status(request)
+        .await
+        .expect("authenticated call succeeds before shutdown");
+
+    // (b) a background worker built the same way the outbox worker and
+    // update-check loop are in `main.rs`: a `tokio::select!` between the
+    // shared shutdown signal and a repeating unit of work. `tick_count`
+    // proves the loop body actually ran, ruling out the trivial case where
+    // the select immediately picks the shutdown branch every time.
+    let tick_count = Arc::new(AtomicUsize::new(0));
+    let tick_count_worker = tick_count.clone();
+    let mut worker_shutdown = shutdown_signal.clone();
+    let worker_task = tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_millis(20));
+        loop {
+            tokio::select! {
+                _ = worker_shutdown.wait() => break,
+                _ = interval.tick() => {
+                    tick_count_worker.fetch_add(1, Ordering::SeqCst);
+                }
+            }
+        }
+    });
+
+    // Let the worker tick a few times before shutdown is requested.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert!(
+        tick_count.load(Ordering::SeqCst) > 0,
+        "worker must have run at least one iteration before shutdown"
+    );
+
+    // Trigger shutdown once, from a single controller shared by both tasks
+    // -- exactly the production wiring, minus the real OS signal.
+    controller.trigger();
+
+    let grpc_result = tokio::time::timeout(JOIN_TIMEOUT, grpc_task)
+        .await
+        .expect("gRPC server task must exit within the join timeout, not hang")
+        .expect("gRPC server task must not panic");
+    assert!(
+        grpc_result.is_ok(),
+        "serve_on_listener_with_shutdown must return Ok once it finishes draining, got {grpc_result:?}"
+    );
+
+    tokio::time::timeout(JOIN_TIMEOUT, worker_task)
+        .await
+        .expect("background worker must exit on the shutdown signal within the join timeout, not be killed")
+        .expect("background worker task must not panic");
+
+    // A dial attempt after the server task has exited must fail: the
+    // listener is gone, proving the server actually stopped accepting
+    // connections rather than merely returning while still bound.
+    assert!(
+        SystemClient::connect(format!("http://{addr}"))
+            .await
+            .is_err(),
+        "gRPC server must no longer be reachable after shutdown"
+    );
+
+    // (c) `DatabaseEngine::close()` -- forces the WAL checkpoint and closes
+    // the pool. A query issued afterward must fail, proving `close()` ran
+    // rather than merely being reachable dead code.
+    db.close().await;
+    let err = db
+        .list_filter_rules()
+        .await
+        .expect_err("a query after close() must fail because the pool is closed");
+    let message = err.to_string();
+    assert!(
+        message.to_lowercase().contains("closed") || message.to_lowercase().contains("pool"),
+        "expected a pool-closed style error after DatabaseEngine::close(), got: {message}"
+    );
+}
+
+#[tokio::test]
+async fn controller_trigger_is_observed_by_every_clone_of_the_signal() {
+    let event_bus = Arc::new(EventBus::new());
+    let (controller, shutdown_signal) = ShutdownController::new(event_bus);
+
+    let mut first = shutdown_signal.clone();
+    let mut second = shutdown_signal.clone();
+    assert!(!first.is_requested());
+    assert!(!second.is_requested());
+
+    controller.trigger();
+
+    tokio::time::timeout(JOIN_TIMEOUT, first.wait())
+        .await
+        .expect("first clone must observe the trigger");
+    tokio::time::timeout(JOIN_TIMEOUT, second.wait())
+        .await
+        .expect("second clone must observe the trigger");
+    assert!(second.is_requested());
+}
-- 
2.54.0


From f44412f1ab2def55bd2e120ed476383002231b52 Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Fri, 31 Jul 2026 13:57:22 -0500
Subject: [PATCH 2/5] fix(store): deterministic message surrogate identity (C1)
 (#337)

* fix(store): give messages a deterministic surrogate identity

IMAP UIDs are unique only per mailbox+UIDVALIDITY, yet messages.id was the
sole key and Email.id was "imap-uid-{uid}", so the same UID in two folders or
accounts collapsed to one row and silently overwrote (audit C1). The id shape
also leaked internal encoding on the wire.

Messages now carry an opaque surrogate id = SHA-256 over length-prefixed
(account, folder, uid_validity, remote_id), plus first-class remote_id and
uid_validity columns and a UNIQUE(account, folder, uid_validity, remote_id)
index (backfill-safe migration seeding legacy rows from the old id). The IMAP
and JMAP backends populate remote_id/uid_validity and hash the id; the outbox
executor recovers remote addressing from the columns instead of parsing the
id, still SELECT-verifying UIDVALIDITY before mutating. Re-sync recomputes the
same id, so it upserts one row and the fire-once filter guard still holds.

Message.id stays a proto string; only its value becomes opaque (no proto
change).

Refs #281

* fix(store): make message-identity migration atomic and self-healing

The identity migration ran ADD COLUMN, backfill, and CREATE UNIQUE INDEX as
separate autocommit statements with the backfill gated on having just added
the column. A crash between the ADD and the backfill could leave legacy rows
at remote_id='' with the column present, so the next startup skipped the
backfill and the index creation then collided on
(account, folder, uid_validity='', remote_id='') -- wedging the daemon.

Run the whole step in one transaction (a crash rolls back atomically) and make
the backfill unconditional (idempotent UPDATE ... WHERE remote_id = '') so any
committed partial state is completed before the index is built. No-op on a
fresh or already-migrated database.

Refs #281

---------

Co-authored-by: James Maes <jmaes@dmdbrands.com>
---
 crates/nuncio-core/src/export.rs              |   4 +
 crates/nuncio-core/src/model.rs               |  81 +++++-
 crates/nuncio-filter/src/engine.rs            |   4 +
 crates/nuncio-mail/src/backend.rs             |  32 ++-
 crates/nuncio-mail/src/imap.rs                |  96 ++++---
 crates/nuncio-mail/src/jmap.rs                |  36 ++-
 crates/nuncio-mail/src/mock.rs                |   2 +
 crates/nuncio-mail/src/parser.rs              |   4 +
 crates/nuncio-mail/src/smtp.rs                |   2 +
 .../nuncio-mail/tests/wiremock_jmap_test.rs   |  15 +-
 crates/nuncio-store/src/db.rs                 | 257 +++++++++++++++++-
 crates/nuncio-store/src/search.rs             |   2 +
 crates/nunciod/src/grpc.rs                    |  15 +
 crates/nunciod/src/outbox.rs                  |  23 +-
 crates/nunciod/src/sync.rs                    |  11 +-
 .../nunciod/tests/filters_triage_e2e_test.rs  |   2 +
 .../nunciod/tests/outbox_executor_e2e_test.rs |  14 +-
 crates/nunciod/tests/spine_e2e_test.rs        |   2 +
 18 files changed, 510 insertions(+), 92 deletions(-)

diff --git a/crates/nuncio-core/src/export.rs b/crates/nuncio-core/src/export.rs
index 360773f..d180291 100644
--- a/crates/nuncio-core/src/export.rs
+++ b/crates/nuncio-core/src/export.rs
@@ -185,6 +185,8 @@ mod tests {
                 id: "msg-100".to_string(),
                 account_id: "acct-1".to_string(),
                 folder_id: "INBOX".to_string(),
+                remote_id: "100".to_string(),
+                uid_validity: "1".to_string(),
                 subject: "Test Subject 1".to_string(),
                 sender: "alice@nuncio.mx".to_string(),
                 recipient: "bob@nuncio.mx".to_string(),
@@ -198,6 +200,8 @@ mod tests {
                 id: "msg-101".to_string(),
                 account_id: "acct-1".to_string(),
                 folder_id: "INBOX".to_string(),
+                remote_id: "101".to_string(),
+                uid_validity: "1".to_string(),
                 subject: "Test Subject 2".to_string(),
                 sender: "charlie@nuncio.mx".to_string(),
                 recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nuncio-core/src/model.rs b/crates/nuncio-core/src/model.rs
index 311c1b5..b6e5034 100644
--- a/crates/nuncio-core/src/model.rs
+++ b/crates/nuncio-core/src/model.rs
@@ -2,6 +2,7 @@
 
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
 
 /// Real-time daemon health, performance metrics, and telemetry payload.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -63,12 +64,30 @@ pub struct Attachment {
 /// Email message domain entity owned by Nuncio core.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Email {
-    /// Unique message identifier.
+    /// Opaque, deterministic surrogate identifier for the message.
+    ///
+    /// Derived from [`Email::surrogate_id`] over the account, folder,
+    /// UIDVALIDITY, and protocol-native id. It carries no decodable internal
+    /// encoding (callers must treat it as opaque) and is stable across
+    /// re-syncs of the same message, so persisting it upserts the same row
+    /// rather than duplicating. Distinct messages that happen to share a
+    /// protocol id across folders or accounts hash to distinct surrogates, so
+    /// one can never silently overwrite another.
     pub id: String,
     /// Account identifier owning the message.
     pub account_id: String,
     /// Mailbox folder identifier (e.g. "inbox").
     pub folder_id: String,
+    /// Protocol-native message id used to address the message on its server:
+    /// the IMAP UID as a decimal string, or the JMAP Email object id. This is
+    /// what a remote mutation must use to identify the message on the wire --
+    /// never the opaque [`Email::id`].
+    pub remote_id: String,
+    /// The addressing scope the `remote_id` is only meaningful within: the IMAP
+    /// folder UIDVALIDITY as a decimal string. Protocols without a UIDVALIDITY
+    /// (JMAP) carry a stable sentinel instead, so the surrogate id stays
+    /// deterministic.
+    pub uid_validity: String,
     /// Subject line.
     pub subject: String,
     /// Sender address (e.g. "alice@nuncio.mx").
@@ -87,6 +106,33 @@ pub struct Email {
     pub attachments: Vec<Attachment>,
 }
 
+impl Email {
+    /// Compute the deterministic, collision-resistant surrogate id for a
+    /// message from its addressing coordinates.
+    ///
+    /// The four coordinates together uniquely identify a stored message: a
+    /// protocol-native `remote_id` (IMAP UID / JMAP object id) is only unique
+    /// within one `uid_validity` scope of one `folder_id` of one `account_id`,
+    /// so all four must participate. Each field is fed length-prefixed so that
+    /// no two distinct coordinate tuples can serialize to the same byte stream
+    /// (e.g. `("a", "bc")` and `("ab", "c")` must not collide). The SHA-256
+    /// digest is hex-encoded, yielding a stable opaque token that leaks none of
+    /// its inputs' encoding on the wire.
+    pub fn surrogate_id(
+        account_id: &str,
+        folder_id: &str,
+        uid_validity: &str,
+        remote_id: &str,
+    ) -> String {
+        let mut hasher = Sha256::new();
+        for field in [account_id, folder_id, uid_validity, remote_id] {
+            hasher.update((field.len() as u64).to_le_bytes());
+            hasher.update(field.as_bytes());
+        }
+        hex::encode(hasher.finalize())
+    }
+}
+
 /// Mailbox folder entity.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Folder {
@@ -168,4 +214,37 @@ mod tests {
         assert_eq!(parsed.status, "HEALTHY");
         assert_eq!(parsed.total_emails, 15000);
     }
+
+    #[test]
+    fn surrogate_id_is_deterministic_and_opaque() {
+        let a = Email::surrogate_id("acct-1", "INBOX", "42", "5");
+        let b = Email::surrogate_id("acct-1", "INBOX", "42", "5");
+        assert_eq!(a, b, "identical coordinates must hash identically");
+        // A 32-byte SHA-256 digest hex-encodes to 64 chars, and the token must
+        // not leak the protocol-native id it was built from.
+        assert_eq!(a.len(), 64);
+        assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
+        assert!(!a.contains("imap-uid-"));
+    }
+
+    #[test]
+    fn surrogate_id_distinguishes_the_cross_folder_and_cross_account_collision() {
+        // The same protocol UID (5) in two different folders of one account, or
+        // in two different accounts, must NOT collapse to the same surrogate.
+        let inbox = Email::surrogate_id("acct-1", "INBOX", "42", "5");
+        let sent = Email::surrogate_id("acct-1", "Sent", "7", "5");
+        let other_account = Email::surrogate_id("acct-2", "INBOX", "42", "5");
+        assert_ne!(inbox, sent, "same UID in two folders must not collide");
+        assert_ne!(
+            inbox, other_account,
+            "same UID in two accounts must not collide"
+        );
+
+        // Field boundaries are unambiguous: shifting a delimiter-like split
+        // across two adjacent fields yields a different id.
+        assert_ne!(
+            Email::surrogate_id("a", "bc", "1", "1"),
+            Email::surrogate_id("ab", "c", "1", "1"),
+        );
+    }
 }
diff --git a/crates/nuncio-filter/src/engine.rs b/crates/nuncio-filter/src/engine.rs
index bce77e8..6037db1 100644
--- a/crates/nuncio-filter/src/engine.rs
+++ b/crates/nuncio-filter/src/engine.rs
@@ -373,6 +373,8 @@ mod tests {
             id: "msg-1".to_string(),
             account_id: account_id.to_string(),
             folder_id: folder_id.to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: subject.to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -394,6 +396,8 @@ mod tests {
             id: "msg-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Urgent Meeting".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nuncio-mail/src/backend.rs b/crates/nuncio-mail/src/backend.rs
index 5c9bb96..cb999bd 100644
--- a/crates/nuncio-mail/src/backend.rs
+++ b/crates/nuncio-mail/src/backend.rs
@@ -36,25 +36,31 @@ pub enum RemoteMutationKind {
     Delete,
 }
 
-/// A fully-addressed remote mutation for a single message: which message
-/// (protocol-native id), the folder it currently lives in, that folder's
-/// stored sync checkpoint (carrying IMAP UIDVALIDITY), and the change to
-/// apply.
+/// A fully-addressed remote mutation for a single message: the protocol-native
+/// id to address it on the server, the folder it currently lives in, the
+/// UIDVALIDITY scope that id was captured under, and the change to apply.
 ///
-/// `folder_checkpoint` is the value persisted by a prior sync
-/// (`"{uidvalidity}:{uidnext}"` for IMAP). The IMAP backend uses it to refuse
-/// to act when the mailbox's current UIDVALIDITY no longer matches -- a
-/// renumbered mailbox reassigns UIDs, so the stored UID would otherwise
-/// address the wrong message. Protocols that do not use UIDVALIDITY (JMAP)
-/// ignore it.
+/// Addressing is driven by `remote_id` (the IMAP UID / JMAP object id), NOT by
+/// the opaque `message_id` -- so remote-mutation targeting never depends on the
+/// shape of the surrogate id. `uid_validity` is the mailbox UIDVALIDITY the
+/// `remote_id` was captured under; the IMAP backend refuses to act unless the
+/// mailbox's current UIDVALIDITY still matches it, because a renumbered mailbox
+/// (RFC 3501 s2.3.1.1) reassigns UIDs and the stored UID would otherwise
+/// address the wrong message. Protocols without a UIDVALIDITY (JMAP) carry a
+/// stable sentinel and ignore this guard.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RemoteMutationSpec {
-    /// Protocol-native message id (IMAP `imap-uid-{uid}`, JMAP object id).
+    /// Opaque surrogate id of the target message. Diagnostic only -- it names
+    /// the message in errors/logs and is never used to address it on the wire.
     pub message_id: String,
+    /// Protocol-native id used to address the message on its server: the IMAP
+    /// UID as a decimal string, or the JMAP Email object id.
+    pub remote_id: String,
     /// Folder the message currently lives in.
     pub folder_id: String,
-    /// The source folder's stored sync checkpoint, if any.
-    pub folder_checkpoint: Option<String>,
+    /// The UIDVALIDITY the `remote_id` was captured under (IMAP; decimal
+    /// string), or a protocol sentinel for servers without UIDVALIDITY (JMAP).
+    pub uid_validity: String,
     /// The change to apply.
     pub kind: RemoteMutationKind,
 }
diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs
index adfbf63..c5288ac 100644
--- a/crates/nuncio-mail/src/imap.rs
+++ b/crates/nuncio-mail/src/imap.rs
@@ -34,31 +34,21 @@ const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60);
 use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec};
 use crate::parser::{MailError, MimeParserAdapter};
 
-/// Prefix an IMAP [`Email::id`] carries in front of its raw numeric UID (see
-/// [`ImapEngine::sync_folder_messages_with_session`], which formats every id as
-/// `"imap-uid-{uid}"`). A remote mutation must recover the UID from the id to
-/// address the message, so the prefix is defined once here.
-const IMAP_UID_ID_PREFIX: &str = "imap-uid-";
-
-/// Recover the raw IMAP UID from an [`Email::id`] of the form
-/// `"imap-uid-{uid}"`. Returns `None` for any id that does not carry the
-/// expected prefix and a valid non-zero `u32` UID, so a mutation against a
-/// malformed/foreign id fails honestly rather than addressing UID 0.
-fn parse_imap_uid(message_id: &str) -> Option<u32> {
-    let raw = message_id.strip_prefix(IMAP_UID_ID_PREFIX)?;
-    let uid = raw.trim().parse::<u32>().ok()?;
+/// Recover a raw IMAP UID from a [`RemoteMutationSpec::remote_id`] (a decimal
+/// string). Returns `None` for anything that is not a valid non-zero `u32`, so
+/// a mutation against a malformed/foreign remote id fails honestly rather than
+/// addressing UID 0.
+fn parse_imap_uid(remote_id: &str) -> Option<u32> {
+    let uid = remote_id.trim().parse::<u32>().ok()?;
     (uid >= 1).then_some(uid)
 }
 
-/// Extract the UIDVALIDITY component of a stored folder checkpoint
-/// (`"{uidvalidity}:{uidnext}"`). Returns `None` for a missing, malformed, or
-/// legacy bare-number checkpoint -- the caller treats an absent stored
-/// UIDVALIDITY as "cannot prove the mailbox has not been renumbered" and
-/// refuses to act.
-fn checkpoint_uid_validity(checkpoint: Option<&str>) -> Option<u32> {
-    let raw = checkpoint?;
-    let (validity, _uid) = raw.split_once(CHECKPOINT_DELIM)?;
-    validity.trim().parse::<u32>().ok()
+/// Parse a stored [`RemoteMutationSpec::uid_validity`] (a decimal string) into
+/// a `u32`. Returns `None` for a missing or non-numeric value -- the caller
+/// treats an unparseable UIDVALIDITY as "cannot prove the mailbox has not been
+/// renumbered" and refuses to act.
+fn parse_uid_validity(uid_validity: &str) -> Option<u32> {
+    uid_validity.trim().parse::<u32>().ok()
 }
 
 /// Delimiter between the two components of a folder sync checkpoint. The
@@ -652,7 +642,16 @@ impl ImapEngine {
 
             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);
+            // The message is addressed on the wire by its UID within the
+            // folder's UIDVALIDITY scope; the persisted id is an opaque
+            // surrogate hashed over both plus the account and folder, so the
+            // same UID in another folder/account can never collide.
+            let remote_id = uid_num.to_string();
+            let uid_validity = server_uid_validity
+                .map(|v| v.to_string())
+                .unwrap_or_default();
+            let email_id =
+                Email::surrogate_id(&self.account_id, folder_id, &uid_validity, &remote_id);
 
             let is_read = fetch_data
                 .flags()
@@ -665,6 +664,8 @@ impl ImapEngine {
                     folder_id,
                     raw_bytes,
                 )?;
+                email.remote_id = remote_id;
+                email.uid_validity = uid_validity;
                 email.read = is_read;
                 email
             } else {
@@ -721,6 +722,8 @@ impl ImapEngine {
                     id: email_id,
                     account_id: self.account_id.clone(),
                     folder_id: folder_id.to_string(),
+                    remote_id,
+                    uid_validity,
                     subject,
                     sender,
                     recipient,
@@ -815,10 +818,10 @@ impl ImapEngine {
     where
         S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
     {
-        let uid = parse_imap_uid(&spec.message_id).ok_or_else(|| {
+        let uid = parse_imap_uid(&spec.remote_id).ok_or_else(|| {
             MailError::ImapError(format!(
-                "cannot parse an IMAP UID from message id '{}'",
-                spec.message_id
+                "cannot parse an IMAP UID from remote id '{}' (message '{}')",
+                spec.remote_id, spec.message_id
             ))
         })?;
 
@@ -831,7 +834,7 @@ impl ImapEngine {
 
         // Refuse to act unless the mailbox's current UIDVALIDITY provably
         // matches the one the stored UID was captured under.
-        let stored_validity = checkpoint_uid_validity(spec.folder_checkpoint.as_deref());
+        let stored_validity = parse_uid_validity(&spec.uid_validity);
         match (stored_validity, mailbox.uid_validity) {
             (Some(stored), Some(current)) if stored == current => {}
             (stored, current) => {
@@ -1575,23 +1578,24 @@ mod tests {
     }
 
     #[test]
-    fn parse_imap_uid_recovers_uid_and_rejects_malformed_ids() {
-        assert_eq!(parse_imap_uid("imap-uid-42"), Some(42));
-        assert_eq!(parse_imap_uid("imap-uid-1"), Some(1));
-        // UID 0 is not a valid IMAP UID; a bare/foreign id yields nothing.
-        assert_eq!(parse_imap_uid("imap-uid-0"), None);
-        assert_eq!(parse_imap_uid("42"), None);
+    fn parse_imap_uid_recovers_uid_and_rejects_non_numeric_ids() {
+        assert_eq!(parse_imap_uid("42"), Some(42));
+        assert_eq!(parse_imap_uid(" 42 "), Some(42));
+        assert_eq!(parse_imap_uid("1"), Some(1));
+        // UID 0 is not a valid IMAP UID; a non-numeric/foreign id yields nothing.
+        assert_eq!(parse_imap_uid("0"), None);
+        assert_eq!(parse_imap_uid("imap-uid-42"), None);
         assert_eq!(parse_imap_uid("jmap-object-id"), None);
     }
 
     #[test]
-    fn checkpoint_uid_validity_extracts_only_a_well_formed_validity() {
-        assert_eq!(checkpoint_uid_validity(Some("7:105")), Some(7));
-        assert_eq!(checkpoint_uid_validity(Some(" 7 : 105 ")), Some(7));
-        assert_eq!(checkpoint_uid_validity(None), None);
-        // A legacy bare-number checkpoint carries no validity to compare.
-        assert_eq!(checkpoint_uid_validity(Some("105")), None);
-        assert_eq!(checkpoint_uid_validity(Some("x:105")), None);
+    fn parse_uid_validity_accepts_only_a_decimal_value() {
+        assert_eq!(parse_uid_validity("7"), Some(7));
+        assert_eq!(parse_uid_validity(" 7 "), Some(7));
+        assert_eq!(parse_uid_validity(""), None);
+        // The JMAP sentinel and a full checkpoint string are not a bare validity.
+        assert_eq!(parse_uid_validity("jmap"), None);
+        assert_eq!(parse_uid_validity("7:105"), None);
     }
 
     /// Drive a mutation over a scripted IMAP server (in-memory duplex),
@@ -1697,10 +1701,18 @@ mod tests {
     }
 
     fn imap_spec(kind: RemoteMutationKind, checkpoint: Option<&str>) -> RemoteMutationSpec {
+        // The tests express the captured scope as a "{uidvalidity}:{uidnext}"
+        // checkpoint string (mirroring how a sync stores it); the spec now
+        // carries just the UIDVALIDITY component the guard compares against.
+        let uid_validity = checkpoint
+            .and_then(|c| c.split(':').next())
+            .unwrap_or_default()
+            .to_string();
         RemoteMutationSpec {
-            message_id: "imap-uid-42".to_string(),
+            message_id: "surrogate-42".to_string(),
+            remote_id: "42".to_string(),
             folder_id: "INBOX".to_string(),
-            folder_checkpoint: checkpoint.map(str::to_string),
+            uid_validity,
             kind,
         }
     }
diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs
index b7a1e0a..cfbc0fa 100644
--- a/crates/nuncio-mail/src/jmap.rs
+++ b/crates/nuncio-mail/src/jmap.rs
@@ -8,6 +8,14 @@ use serde_json::{json, Value};
 use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec};
 use crate::parser::MailError;
 
+/// Stand-in for the UIDVALIDITY scope on JMAP, which has no such concept: a
+/// JMAP Email object id is already globally stable within its account (RFC
+/// 8621), so a fixed sentinel keeps the surrogate-id computation uniform with
+/// IMAP without implying a UIDVALIDITY guard. The IMAP UIDVALIDITY guard treats
+/// this non-numeric value as "no validity to compare" and is never engaged for
+/// a JMAP mutation, which addresses messages by object id.
+const JMAP_UID_VALIDITY_SENTINEL: &str = "jmap";
+
 /// JMAP Session Object (RFC 8620 Section 2).
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -214,7 +222,9 @@ impl JmapEngine {
     /// `$flagged` keyword (`true` to set, `null` to clear), the JMAP equivalent
     /// of IMAP `\Flagged`.
     pub fn build_email_set_request(account_id: &str, spec: &RemoteMutationSpec) -> Value {
-        let id = &spec.message_id;
+        // Address the message by its protocol-native JMAP object id, never the
+        // opaque surrogate `message_id`.
+        let id = &spec.remote_id;
         let (update, destroy) = match &spec.kind {
             RemoteMutationKind::SetFlagged { value } => {
                 let keyword = if *value { json!(true) } else { Value::Null };
@@ -383,10 +393,23 @@ impl JmapEngine {
                     .map(|a| a.email.clone())
                     .unwrap_or_else(|| "me@nuncio.mx".to_string());
 
+                let folder_id = "inbox".to_string();
+                // The JMAP object id is the protocol-native id; the persisted
+                // id is an opaque surrogate hashed over it plus the account,
+                // folder, and (sentinel) scope so it stays uniform with IMAP.
+                let remote_id = item.id;
+                let id = Email::surrogate_id(
+                    &self.account_id,
+                    &folder_id,
+                    JMAP_UID_VALIDITY_SENTINEL,
+                    &remote_id,
+                );
                 Email {
-                    id: item.id,
+                    id,
                     account_id: self.account_id.clone(),
-                    folder_id: "inbox".to_string(),
+                    folder_id,
+                    remote_id,
+                    uid_validity: JMAP_UID_VALIDITY_SENTINEL.to_string(),
                     subject: item.subject.unwrap_or_else(|| "No Subject".to_string()),
                     sender,
                     recipient,
@@ -558,7 +581,7 @@ impl MailBackend for JmapEngine {
         let request = Self::build_email_set_request(&account_id, spec);
         let raw = self.post_jmap(&session.api_url, &request).await?;
         let expect_destroy = matches!(spec.kind, RemoteMutationKind::Delete);
-        Self::confirm_email_set_applied(&raw, &spec.message_id, expect_destroy)
+        Self::confirm_email_set_applied(&raw, &spec.remote_id, expect_destroy)
     }
 }
 
@@ -689,9 +712,10 @@ mod tests {
 
     fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec {
         RemoteMutationSpec {
-            message_id: "m-1".to_string(),
+            message_id: "surrogate-m-1".to_string(),
+            remote_id: "m-1".to_string(),
             folder_id: "mb-inbox".to_string(),
-            folder_checkpoint: None,
+            uid_validity: JMAP_UID_VALIDITY_SENTINEL.to_string(),
             kind,
         }
     }
diff --git a/crates/nuncio-mail/src/mock.rs b/crates/nuncio-mail/src/mock.rs
index 4a581d5..487d09a 100644
--- a/crates/nuncio-mail/src/mock.rs
+++ b/crates/nuncio-mail/src/mock.rs
@@ -244,6 +244,8 @@ mod tests {
             id: "msg-mock-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Mock Test".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nuncio-mail/src/parser.rs b/crates/nuncio-mail/src/parser.rs
index 28976b9..95e19f7 100644
--- a/crates/nuncio-mail/src/parser.rs
+++ b/crates/nuncio-mail/src/parser.rs
@@ -129,6 +129,10 @@ impl MimeParserAdapter {
             id: id.to_string(),
             account_id: account_id.to_string(),
             folder_id: folder_id.to_string(),
+            // The MIME body carries no protocol addressing; the caller (the
+            // IMAP sync) fills these from the transport before persisting.
+            remote_id: String::new(),
+            uid_validity: String::new(),
             subject,
             sender,
             recipient,
diff --git a/crates/nuncio-mail/src/smtp.rs b/crates/nuncio-mail/src/smtp.rs
index 35e592d..9cef1f0 100644
--- a/crates/nuncio-mail/src/smtp.rs
+++ b/crates/nuncio-mail/src/smtp.rs
@@ -329,6 +329,8 @@ mod tests {
             id: "msg-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "outbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Status Update".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nuncio-mail/tests/wiremock_jmap_test.rs b/crates/nuncio-mail/tests/wiremock_jmap_test.rs
index f7773af..f809590 100644
--- a/crates/nuncio-mail/tests/wiremock_jmap_test.rs
+++ b/crates/nuncio-mail/tests/wiremock_jmap_test.rs
@@ -84,7 +84,11 @@ async fn wiremock_jmap_session_discovery_and_email_get_sync() {
         .expect("parse email get");
 
     assert_eq!(emails.len(), 1);
-    assert_eq!(emails[0].id, "msg-wm-100");
+    // The JMAP object id is preserved as the protocol-native `remote_id`, while
+    // the persisted id is an opaque 64-char surrogate hash (never the raw id).
+    assert_eq!(emails[0].remote_id, "msg-wm-100");
+    assert_ne!(emails[0].id, "msg-wm-100");
+    assert_eq!(emails[0].id.len(), 64);
     assert_eq!(emails[0].subject, "WireMock JMAP E2E Test");
     assert_eq!(emails[0].sender, "sender@nuncio.mx");
     assert_eq!(new_state, "sync-state-500");
@@ -200,7 +204,9 @@ async fn jmap_engine_sync_folders_and_messages_through_mail_backend_trait() {
         .await
         .expect("sync_messages succeeds");
     assert_eq!(emails.len(), 1);
-    assert_eq!(emails[0].id, "msg-wm-200");
+    assert_eq!(emails[0].remote_id, "msg-wm-200");
+    assert_ne!(emails[0].id, "msg-wm-200");
+    assert_eq!(emails[0].id.len(), 64);
     assert_eq!(emails[0].subject, "Trait-Level JMAP Sync");
     assert_eq!(emails[0].sender, "sender@nuncio.mx");
     assert_eq!(emails[0].received_at, 1700002000);
@@ -264,9 +270,10 @@ fn jmap_engine(mock_server: &MockServer) -> JmapEngine {
 
 fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec {
     RemoteMutationSpec {
-        message_id: "jmap-msg-1".to_string(),
+        message_id: "surrogate-jmap-msg-1".to_string(),
+        remote_id: "jmap-msg-1".to_string(),
         folder_id: "mb-inbox".to_string(),
-        folder_checkpoint: None,
+        uid_validity: "jmap".to_string(),
         kind,
     }
 }
diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs
index a40fd63..5dd7ae9 100644
--- a/crates/nuncio-store/src/db.rs
+++ b/crates/nuncio-store/src/db.rs
@@ -471,6 +471,8 @@ impl DatabaseEngine {
                 id TEXT PRIMARY KEY NOT NULL,
                 account_id TEXT NOT NULL,
                 folder_id TEXT NOT NULL,
+                remote_id TEXT NOT NULL DEFAULT '',
+                uid_validity TEXT NOT NULL DEFAULT '',
                 subject TEXT NOT NULL,
                 sender TEXT NOT NULL,
                 recipient TEXT NOT NULL,
@@ -711,11 +713,84 @@ impl DatabaseEngine {
         self.ensure_accounts_smtp_columns().await?;
         self.ensure_accounts_tls_mode_columns().await?;
         self.ensure_accounts_dav_columns().await?;
+        self.ensure_messages_identity_columns().await?;
         self.backfill_message_fts().await?;
 
         Ok(())
     }
 
+    /// Additive, backfill-safe migration that adds the `remote_id` /
+    /// `uid_validity` message-identity columns to a pre-existing `messages`
+    /// table that predated the opaque surrogate id, and installs the
+    /// `UNIQUE(account_id, folder_id, uid_validity, remote_id)` index that is
+    /// the real correctness guard against a shared protocol id (e.g. an IMAP
+    /// UID reused across folders/accounts) overwriting an unrelated message.
+    ///
+    /// A fresh database already gets both columns from `CREATE TABLE IF NOT
+    /// EXISTS messages` above, so on a fresh database the `PRAGMA table_info`
+    /// check finds them present and the backfill matches no rows -- only the
+    /// index is created. For a pre-existing database file the columns are added
+    /// with an empty-string default and every legacy row is backfilled to
+    /// `remote_id = id` (the old primary key, which was globally unique) so the
+    /// UNIQUE index can be built without collisions. SQLite has no `ADD COLUMN
+    /// IF NOT EXISTS`, so column presence is checked explicitly first, making
+    /// this safe to run on every daemon startup.
+    ///
+    /// The whole step runs in a single transaction, and the backfill is
+    /// unconditional (idempotent `WHERE remote_id = ''`) rather than gated on
+    /// having just added the column. Both properties make a partial run
+    /// self-healing: a crash mid-migration rolls back atomically, and even a
+    /// committed intermediate state (column present but a legacy row still
+    /// carrying the empty default) is completed by the next startup's backfill
+    /// before the index is built -- so the daemon can never wedge on an index
+    /// collision it could have avoided.
+    async fn ensure_messages_identity_columns(&self) -> Result<(), DatabaseError> {
+        let existing_columns: Vec<String> = sqlx::query("PRAGMA table_info(messages)")
+            .fetch_all(&self.pool)
+            .await
+            .map_err(DatabaseError::Query)?
+            .iter()
+            .map(|row| row.get::<String, _>("name"))
+            .collect();
+
+        let mut tx = self.pool.begin().await.map_err(DatabaseError::Query)?;
+
+        if !existing_columns.iter().any(|c| c == "remote_id") {
+            sqlx::query("ALTER TABLE messages ADD COLUMN remote_id TEXT NOT NULL DEFAULT ''")
+                .execute(&mut *tx)
+                .await
+                .map_err(DatabaseError::Query)?;
+        }
+        if !existing_columns.iter().any(|c| c == "uid_validity") {
+            sqlx::query("ALTER TABLE messages ADD COLUMN uid_validity TEXT NOT NULL DEFAULT ''")
+                .execute(&mut *tx)
+                .await
+                .map_err(DatabaseError::Query)?;
+        }
+
+        // Seed any legacy row still carrying the empty-string default with a
+        // unique, non-empty remote id (the old primary key) so the UNIQUE
+        // identity index is satisfiable. Unconditional so a partial prior run
+        // is always completed before the index is (re)built; a no-op on a fresh
+        // or already-migrated database.
+        sqlx::query("UPDATE messages SET remote_id = id WHERE remote_id = ''")
+            .execute(&mut *tx)
+            .await
+            .map_err(DatabaseError::Query)?;
+
+        sqlx::query(
+            "CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_identity \
+             ON messages (account_id, folder_id, uid_validity, remote_id)",
+        )
+        .execute(&mut *tx)
+        .await
+        .map_err(DatabaseError::Query)?;
+
+        tx.commit().await.map_err(DatabaseError::Query)?;
+
+        Ok(())
+    }
+
     /// Additive, backfill-safe migration that adds the `imap_tls_mode` /
     /// `smtp_tls_mode` columns to a pre-existing `accounts` table that
     /// predates persisted per-protocol TLS transport modes.
@@ -1081,13 +1156,15 @@ impl DatabaseEngine {
         sqlx::query(
             r#"
             INSERT OR REPLACE INTO messages
-            (id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            (id, account_id, folder_id, remote_id, uid_validity, subject, sender, recipient, received_at, read_flag, body_plain, body_html)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             "#,
         )
         .bind(&email.id)
         .bind(&email.account_id)
         .bind(&email.folder_id)
+        .bind(&email.remote_id)
+        .bind(&email.uid_validity)
         .bind(&email.subject)
         .bind(&email.sender)
         .bind(&email.recipient)
@@ -1143,9 +1220,11 @@ impl DatabaseEngine {
             i64,
             Option<String>,
             Option<String>,
+            String,
+            String,
         )> = sqlx::query_as(
             r#"
-            SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html
+            SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity
             FROM messages
             WHERE folder_id = ?
             ORDER BY received_at DESC
@@ -1171,6 +1250,8 @@ impl DatabaseEngine {
                     read_flag,
                     body_plain,
                     body_html,
+                    remote_id,
+                    uid_validity,
                 )| {
                     let dec_plain = body_plain
                         .map(|p| {
@@ -1194,6 +1275,8 @@ impl DatabaseEngine {
                         id,
                         account_id,
                         folder_id,
+                        remote_id,
+                        uid_validity,
                         subject,
                         sender,
                         recipient,
@@ -1225,9 +1308,11 @@ impl DatabaseEngine {
             i64,
             Option<String>,
             Option<String>,
+            String,
+            String,
         ) = sqlx::query_as(
             r#"
-            SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html
+            SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity
             FROM messages
             WHERE id = ?
             "#,
@@ -1252,6 +1337,8 @@ impl DatabaseEngine {
             id: row.0,
             account_id: row.1,
             folder_id: row.2,
+            remote_id: row.10,
+            uid_validity: row.11,
             subject: row.3,
             sender: row.4,
             recipient: row.5,
@@ -1902,7 +1989,7 @@ impl DatabaseEngine {
         limit: usize,
     ) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
         let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
-            "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html FROM messages "
+            "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity FROM messages "
         );
         if !last_id.is_empty() {
             builder.push("WHERE id > ");
@@ -1922,6 +2009,8 @@ impl DatabaseEngine {
             i64,
             Option<String>,
             Option<String>,
+            String,
+            String,
         )>();
 
         let rows = query
@@ -1947,6 +2036,8 @@ impl DatabaseEngine {
                     id: r.0,
                     account_id: r.1,
                     folder_id: r.2,
+                    remote_id: r.10,
+                    uid_validity: r.11,
                     subject: r.3,
                     sender: r.4,
                     recipient: r.5,
@@ -1972,7 +2063,7 @@ impl DatabaseEngine {
         folder_id: Option<&str>,
     ) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
         let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
-            "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html FROM messages"
+            "SELECT id, account_id, folder_id, subject, sender, recipient, received_at, read_flag, body_plain, body_html, remote_id, uid_validity FROM messages"
         );
 
         let mut has_filter = false;
@@ -2002,6 +2093,8 @@ impl DatabaseEngine {
             i64,
             Option<String>,
             Option<String>,
+            String,
+            String,
         )>();
 
         let rows = query
@@ -2027,6 +2120,8 @@ impl DatabaseEngine {
                     id: r.0,
                     account_id: r.1,
                     folder_id: r.2,
+                    remote_id: r.10,
+                    uid_validity: r.11,
                     subject: r.3,
                     sender: r.4,
                     recipient: r.5,
@@ -2483,6 +2578,8 @@ mod tests {
             id: "msg-db-100".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "INBOX".to_string(),
+            remote_id: "100".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Database Sync Test".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -2530,6 +2627,8 @@ mod tests {
             id: "msg-db-corrupt".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "INBOX".to_string(),
+            remote_id: "corrupt".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Corrupted At Rest".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -2832,6 +2931,8 @@ mod tests {
             id: "msg-mark-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "INBOX".to_string(),
+            remote_id: "mark-1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Mark Read Test".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -3362,6 +3463,8 @@ mod tests {
                 id: format!("msg-{i:03}"),
                 account_id: "acct-1".to_string(),
                 folder_id: "inbox".to_string(),
+                remote_id: format!("{i}"),
+                uid_validity: "1".to_string(),
                 subject: format!("Subject {i}"),
                 sender: "alice@nuncio.mx".to_string(),
                 recipient: "bob@nuncio.mx".to_string(),
@@ -3469,11 +3572,153 @@ mod tests {
         );
     }
 
+    /// Build a message exactly as a real sync would: an opaque surrogate id
+    /// hashed over its addressing coordinates, with the protocol-native id and
+    /// UIDVALIDITY scope carried in their own columns.
+    fn synced_email(
+        account_id: &str,
+        folder_id: &str,
+        uid_validity: &str,
+        remote_id: &str,
+    ) -> nuncio_core::model::Email {
+        nuncio_core::model::Email {
+            id: nuncio_core::model::Email::surrogate_id(
+                account_id,
+                folder_id,
+                uid_validity,
+                remote_id,
+            ),
+            account_id: account_id.to_string(),
+            folder_id: folder_id.to_string(),
+            remote_id: remote_id.to_string(),
+            uid_validity: uid_validity.to_string(),
+            subject: format!("{folder_id}/{remote_id}"),
+            sender: "alice@nuncio.mx".to_string(),
+            recipient: "bob@nuncio.mx".to_string(),
+            received_at: 1_700_000_000,
+            read: false,
+            body_plain: Some("body".to_string()),
+            body_html: None,
+            attachments: Vec::new(),
+        }
+    }
+
+    /// The C1 regression: the same protocol UID reused across folders or
+    /// accounts must persist as DISTINCT rows (never silently overwrite), and a
+    /// re-sync of the SAME message must upsert the one row it already owns.
+    #[tokio::test]
+    async fn message_identity_prevents_cross_folder_and_cross_account_collision() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+
+        // Same UID (5) in two folders of one account, and again in a second
+        // account -- the exact shapes that used to collapse to `imap-uid-5`.
+        let inbox = synced_email("acct-1", "INBOX", "42", "5");
+        let sent = synced_email("acct-1", "Sent", "7", "5");
+        let other_account = synced_email("acct-2", "INBOX", "99", "5");
+
+        // Every surrogate id is distinct, so none can overwrite another.
+        assert_ne!(inbox.id, sent.id);
+        assert_ne!(inbox.id, other_account.id);
+        assert_ne!(sent.id, other_account.id);
+
+        engine.save_email(&inbox).await.expect("save inbox");
+        engine.save_email(&sent).await.expect("save sent");
+        engine
+            .save_email(&other_account)
+            .await
+            .expect("save other account");
+
+        // All three coexist as separate rows, addressed by their own folder.
+        assert_eq!(engine.list_messages("INBOX", 10).await.unwrap().len(), 2);
+        assert_eq!(engine.list_messages("Sent", 10).await.unwrap().len(), 1);
+
+        // Re-syncing the SAME message recomputes the SAME id, so the upsert
+        // updates the one row rather than duplicating it.
+        let inbox_resynced = synced_email("acct-1", "INBOX", "42", "5");
+        assert_eq!(inbox_resynced.id, inbox.id);
+        engine
+            .save_email(&inbox_resynced)
+            .await
+            .expect("re-sync upserts");
+        assert_eq!(
+            engine.list_messages("INBOX", 10).await.unwrap().len(),
+            2,
+            "a re-sync must upsert, not duplicate"
+        );
+
+        // The recovered row round-trips its protocol addressing columns.
+        let fetched = engine.get_message(&inbox.id).await.expect("get message");
+        assert_eq!(fetched.remote_id, "5");
+        assert_eq!(fetched.uid_validity, "42");
+        assert_eq!(fetched.folder_id, "INBOX");
+    }
+
+    /// A crash between `ADD COLUMN remote_id` and its backfill can commit an
+    /// intermediate state -- the column present but a legacy row still carrying
+    /// the empty-string default -- with the UNIQUE index not yet built. The
+    /// migration must complete that backfill on the next open and build the
+    /// index without a collision, rather than wedging the daemon.
+    #[tokio::test]
+    async fn messages_identity_migration_completes_from_a_partial_run() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+
+        // Reproduce the partial state: drop the identity index, then insert two
+        // rows in the same folder that both still hold the empty default. With
+        // the index gone this is permitted, and the two would collide on
+        // (account, folder, uid_validity='', remote_id='') if it were rebuilt now.
+        sqlx::query("DROP INDEX IF EXISTS idx_messages_identity")
+            .execute(engine.pool())
+            .await
+            .unwrap();
+        for id in ["legacy-a", "legacy-b"] {
+            sqlx::query(
+                "INSERT INTO messages \
+                 (id, account_id, folder_id, remote_id, uid_validity, subject, sender, recipient, received_at, read_flag) \
+                 VALUES (?, 'acct-1', 'INBOX', '', '', 'subj', 'a@nuncio.mx', 'b@nuncio.mx', 0, 0)",
+            )
+            .bind(id)
+            .execute(engine.pool())
+            .await
+            .unwrap();
+        }
+
+        // Re-running the migration must finish the backfill and build the index.
+        engine
+            .ensure_messages_identity_columns()
+            .await
+            .expect("a partial migration must complete on the next open");
+
+        // Both legacy rows survive with a unique, non-empty backfilled remote id.
+        assert_eq!(
+            engine.get_message("legacy-a").await.unwrap().remote_id,
+            "legacy-a"
+        );
+        assert_eq!(
+            engine.get_message("legacy-b").await.unwrap().remote_id,
+            "legacy-b"
+        );
+
+        // The unique identity index is now present, and re-running is idempotent.
+        let index: Option<(String,)> = sqlx::query_as(
+            "SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_messages_identity'",
+        )
+        .fetch_optional(engine.pool())
+        .await
+        .unwrap();
+        assert!(index.is_some(), "the unique identity index must be built");
+        engine
+            .ensure_messages_identity_columns()
+            .await
+            .expect("re-running an already-complete migration 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(),
             account_id: account_id.to_string(),
             folder_id: folder_id.to_string(),
+            remote_id: id.to_string(),
+            uid_validity: "1".to_string(),
             subject: format!("Subject {id}"),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nuncio-store/src/search.rs b/crates/nuncio-store/src/search.rs
index ed6d1ad..0befdf5 100644
--- a/crates/nuncio-store/src/search.rs
+++ b/crates/nuncio-store/src/search.rs
@@ -105,6 +105,8 @@ mod tests {
             id: id.to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: id.to_string(),
+            uid_validity: "1".to_string(),
             subject: subject.to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs
index 5aa2251..43c73ef 100644
--- a/crates/nunciod/src/grpc.rs
+++ b/crates/nunciod/src/grpc.rs
@@ -1592,6 +1592,9 @@ fn synthetic_preview_email(id: &str) -> nuncio_core::model::Email {
         id: id.to_string(),
         account_id: "acct-1".to_string(),
         folder_id: "inbox".to_string(),
+        // Synthetic preview only -- never persisted or mutated remotely.
+        remote_id: id.to_string(),
+        uid_validity: String::new(),
         subject: "Test Subject".to_string(),
         sender: "test@nuncio.mx".to_string(),
         recipient: "me@nuncio.mx".to_string(),
@@ -3540,6 +3543,8 @@ mod tests {
             id: id.to_string(),
             account_id: "acct-mail-1".to_string(),
             folder_id: folder_id.to_string(),
+            remote_id: id.to_string(),
+            uid_validity: "1".to_string(),
             subject: subject.to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -4426,6 +4431,8 @@ mod tests {
             id: "msg-filters-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: subject.to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -4526,6 +4533,8 @@ mod tests {
                 id: "msg-1".to_string(),
                 account_id: "acct-1".to_string(),
                 folder_id: "inbox".to_string(),
+                remote_id: "1".to_string(),
+                uid_validity: "1".to_string(),
                 subject: "anything".to_string(),
                 sender: "a@b.com".to_string(),
                 recipient: "c@d.com".to_string(),
@@ -4668,6 +4677,8 @@ mod tests {
             id: "msg-preview-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "preview-1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Urgent Board Meeting".to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -4834,6 +4845,8 @@ mod tests {
             id: "msg-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Urgent Meeting".to_string(),
             sender: "a@b.com".to_string(),
             recipient: "c@d.com".to_string(),
@@ -4988,6 +5001,8 @@ mod tests {
             id: "msg-1".to_string(),
             account_id: "acct-1".to_string(),
             folder_id: "inbox".to_string(),
+            remote_id: "1".to_string(),
+            uid_validity: "1".to_string(),
             subject: "Urgent Meeting".to_string(),
             sender: "a@b.com".to_string(),
             recipient: "c@d.com".to_string(),
diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs
index f022ea5..062c38a 100644
--- a/crates/nunciod/src/outbox.rs
+++ b/crates/nunciod/src/outbox.rs
@@ -204,7 +204,7 @@ async fn execute_one(
 
     match mutation.mutation_type.as_str() {
         "MOVE" | "COPY" | "FLAG" | "UNFLAG" | "DELETE" => {
-            dispatch_mailbox_mutation(db, env, mutation, &email, &payload).await
+            dispatch_mailbox_mutation(env, mutation, &email, &payload).await
         }
         "FORWARD" => dispatch_forward(db, env, &email, &payload).await,
         "WEBHOOK" => dispatch_webhook(env, mutation, &email, &payload).await,
@@ -213,10 +213,10 @@ async fn execute_one(
 }
 
 /// Execute a mailbox-affecting mutation (move/copy/flag/unflag/delete) against
-/// the account's mail backend, threading the source folder's stored checkpoint
-/// so the backend can enforce its UIDVALIDITY guard.
+/// the account's mail backend, recovering the remote addressing (protocol id +
+/// UIDVALIDITY scope) from the stored message row so the backend can enforce
+/// its UIDVALIDITY guard.
 async fn dispatch_mailbox_mutation(
-    db: &DatabaseEngine,
     env: &dyn RemoteExecutionEnv,
     mutation: &PendingRemoteMutation,
     email: &nuncio_core::model::Email,
@@ -245,23 +245,20 @@ async fn dispatch_mailbox_mutation(
         other => return Disposition::Permanent(format!("unexpected mailbox mutation '{other}'")),
     };
 
-    let folder_checkpoint = match db
-        .get_folder_sync_state(&email.account_id, &email.folder_id)
-        .await
-    {
-        Ok(checkpoint) => checkpoint,
-        Err(e) => return Disposition::Retry(format!("failed to read folder checkpoint: {e}")),
-    };
-
     let backend = match env.mail_backend(&email.account_id).await {
         Ok(backend) => backend,
         Err(e) => return Disposition::Retry(format!("failed to build mail backend: {e}")),
     };
 
+    // Recover the remote addressing from the stored message row -- the
+    // protocol-native id and the UIDVALIDITY scope it was captured under --
+    // rather than parsing it out of the opaque surrogate id. The backend still
+    // SELECT-verifies the UIDVALIDITY before mutating.
     let spec = RemoteMutationSpec {
         message_id: email.id.clone(),
+        remote_id: email.remote_id.clone(),
         folder_id: email.folder_id.clone(),
-        folder_checkpoint,
+        uid_validity: email.uid_validity.clone(),
         kind,
     };
 
diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs
index bbb395c..bbab07d 100644
--- a/crates/nunciod/src/sync.rs
+++ b/crates/nunciod/src/sync.rs
@@ -628,6 +628,8 @@ mod tests {
             id: id.to_string(),
             account_id: "acct-mock-1".to_string(),
             folder_id: folder_id.to_string(),
+            remote_id: id.to_string(),
+            uid_validity: "1".to_string(),
             subject: subject.to_string(),
             sender: "alice@nuncio.mx".to_string(),
             recipient: "bob@nuncio.mx".to_string(),
@@ -864,11 +866,16 @@ mod tests {
         // The wiremock stubs advertise 1 folder containing 1 message.
         assert_eq!(synced, 1);
 
+        // The message persists under an opaque surrogate id; it is recovered by
+        // folder, and its protocol-native JMAP object id round-trips in
+        // `remote_id`.
         let persisted = db
-            .get_message("jmap-msg-1")
+            .list_messages("inbox", 10)
             .await
             .expect("jmap message persisted");
-        assert_eq!(persisted.subject, "Welcome to JMAP Sync");
+        assert_eq!(persisted.len(), 1);
+        assert_eq!(persisted[0].subject, "Welcome to JMAP Sync");
+        assert_eq!(persisted[0].remote_id, "jmap-msg-1");
 
         assert_eq!(
             events.recv().await.expect("start event"),
diff --git a/crates/nunciod/tests/filters_triage_e2e_test.rs b/crates/nunciod/tests/filters_triage_e2e_test.rs
index bb34869..60689d3 100644
--- a/crates/nunciod/tests/filters_triage_e2e_test.rs
+++ b/crates/nunciod/tests/filters_triage_e2e_test.rs
@@ -43,6 +43,8 @@ fn sample_email(id: &str, subject: &str) -> Email {
         id: id.to_string(),
         account_id: ACCOUNT_ID.to_string(),
         folder_id: "inbox".to_string(),
+        remote_id: id.to_string(),
+        uid_validity: "1".to_string(),
         subject: subject.to_string(),
         sender: "alice@nuncio.mx".to_string(),
         recipient: "triage-e2e@nuncio.mx".to_string(),
diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs
index 6f823ff..d88509a 100644
--- a/crates/nunciod/tests/outbox_executor_e2e_test.rs
+++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs
@@ -103,9 +103,11 @@ fn sample_account() -> nuncio_core::AccountConfig {
 
 fn sample_email(subject: &str) -> Email {
     Email {
-        id: "imap-uid-42".to_string(),
+        id: "surrogate-inbox-42".to_string(),
         account_id: ACCOUNT_ID.to_string(),
         folder_id: FOLDER_ID.to_string(),
+        remote_id: "42".to_string(),
+        uid_validity: "1".to_string(),
         subject: subject.to_string(),
         sender: "alice@nuncio.mx".to_string(),
         recipient: "owner@nuncio.mx".to_string(),
@@ -161,11 +163,13 @@ async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_check
 
     let applied = env.backend.applied_mutations();
     assert_eq!(applied.len(), 1, "the backend op must genuinely run");
-    assert_eq!(applied[0].message_id, "imap-uid-42");
+    // Addressing is recovered from the stored row's columns, not by parsing the
+    // opaque surrogate id: the protocol-native UID and the UIDVALIDITY scope it
+    // was captured under, so the backend can enforce its guard.
+    assert_eq!(applied[0].message_id, "surrogate-inbox-42");
+    assert_eq!(applied[0].remote_id, "42");
     assert_eq!(applied[0].folder_id, FOLDER_ID);
-    // The source folder's stored checkpoint (carrying UIDVALIDITY) is threaded
-    // through so the backend can enforce its guard.
-    assert_eq!(applied[0].folder_checkpoint.as_deref(), Some(CHECKPOINT));
+    assert_eq!(applied[0].uid_validity, "1");
     assert_eq!(
         applied[0].kind,
         RemoteMutationKind::Move {
diff --git a/crates/nunciod/tests/spine_e2e_test.rs b/crates/nunciod/tests/spine_e2e_test.rs
index 058225e..187ef0d 100644
--- a/crates/nunciod/tests/spine_e2e_test.rs
+++ b/crates/nunciod/tests/spine_e2e_test.rs
@@ -77,6 +77,8 @@ fn mock_inbound_email(id: &str, subject: &str, body: &str) -> Email {
         id: id.to_string(),
         account_id: ACCOUNT_ID.to_string(),
         folder_id: "inbox".to_string(),
+        remote_id: id.to_string(),
+        uid_validity: "1".to_string(),
         subject: subject.to_string(),
         sender: "alice@nuncio.mx".to_string(),
         recipient: ACCOUNT_EMAIL.to_string(),
-- 
2.54.0


From 8cefa903f9c81d774eb4461a70573cf1ed06eade Mon Sep 17 00:00:00 2001
From: James Maes <jmaes@dmdbrands.com>
Date: Fri, 31 Jul 2026 14:08:58 -0500
Subject: [PATCH 3/5] feat(proto): add typed error taxonomy and Status helper

Introduce a frozen, machine-readable error taxonomy for the nuncio.v1
contract: an ErrorReason enum and an ErrorInfo message carrying a reason,
optional metadata, and a human message. A nuncio-proto errors helper builds
a tonic::Status that encodes ErrorInfo into the grpc-status-details-bin
trailer with the canonical gRPC code per reason, and decodes the reason back
on the client side. Regenerate the committed descriptor golden.

Refs #273
---
 Cargo.lock                                    |   1 +
 crates/nuncio-proto/Cargo.toml                |   1 +
 .../proto/nuncio/v1/descriptor.bin            | Bin 12854 -> 13517 bytes
 .../nuncio-proto/proto/nuncio/v1/nuncio.proto |  58 +++++++
 crates/nuncio-proto/src/errors.rs             | 153 ++++++++++++++++++
 crates/nuncio-proto/src/lib.rs                |   1 +
 6 files changed, 214 insertions(+)
 create mode 100644 crates/nuncio-proto/src/errors.rs

diff --git a/Cargo.lock b/Cargo.lock
index df41e19..f94cbed 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2594,6 +2594,7 @@ dependencies = [
 name = "nuncio-proto"
 version = "0.1.0"
 dependencies = [
+ "bytes",
  "prost",
  "protoc-bin-vendored",
  "thiserror 1.0.69",
diff --git a/crates/nuncio-proto/Cargo.toml b/crates/nuncio-proto/Cargo.toml
index ccb1ec7..71359e1 100644
--- a/crates/nuncio-proto/Cargo.toml
+++ b/crates/nuncio-proto/Cargo.toml
@@ -13,6 +13,7 @@ workspace = true
 tonic.workspace = true
 tonic-prost.workspace = true
 prost.workspace = true
+bytes.workspace = true
 thiserror.workspace = true
 
 [build-dependencies]
diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin
index 1f17b041e64990ea6ac8d64c00bbd63f1cdc35ec..850edc681e2a0d676fc2fee02bcd32376c5c1d8b 100644
GIT binary patch
delta 687
zcmZvY%}T>i5QR-+|E5*KMI~yfrBo3Y0lN|f;hNmqi!rG;NzqjZHcHW^kp4hdeFh&z
z1h?)5H@<|g;Z3dDSa&nbcg~!d=hj<WgxB_iSQ)<ykR@T*4QX)GMGB-s-|KY)Sttp+
z%hfd{7zEett}@(LCKQ`Y*_1-Zkm>k+ujTbUIZ+aKbJg-+MdYiTPazWY!x8(@AlZR*
z$M5yLTVEFW#w`18H;!N_ra}^kcm8M+EX$%2tw3_<-4A@eI+A`hhlvB>apz-u{31eb
zCNnC4Oc*nn%Luki!*vX+K~!3!M3WX!8H%xK6YEsNHsxRqQ=K!y9D`NWbPU@yOxvyT
zN|O?33pT%e#?v*-(a8)U&bgcu%fhF#%pxj5A(n<TNo-GY&P6p18>=%}#!~7x%rpzu
zd7)jo<dRypC8g(p=o0Q!qo)qzCudLvO0gg(SUn?##-GQX)H*)ViIhb}*oaZsvFjTD
k^Mykr8~E?{PNQKmn@@9S9adv<YS@GsSeNoo`JKJ|CrG}(I{*Lx

delta 20
ccmX?`xh;j6Yjes(W|fV9o~dqLqM^wH0AF$lnE(I)

diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
index f120980..171b418 100644
--- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
+++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
@@ -2,6 +2,64 @@ syntax = "proto3";
 
 package nuncio.v1;
 
+// ErrorReason is the stable, machine-readable taxonomy of failure conditions
+// the daemon reports to clients. It is carried inside an `ErrorInfo` in a
+// failed RPC's gRPC status-details trailer so a client can branch on a
+// discriminant instead of string-matching human-readable (English) error
+// text. Each reason maps to a fixed canonical gRPC status code (see the
+// server-side helper that builds the `Status`), so the transport-level code
+// and this application-level reason always agree.
+//
+// Values are append-only: never renumber or repurpose an existing value.
+// New failure conditions get a new value; `ERROR_REASON_UNSPECIFIED` is the
+// zero value and is treated as a generic internal error by clients that do
+// not recognize a (newer) reason.
+enum ErrorReason {
+  ERROR_REASON_UNSPECIFIED = 0;
+  // A request was syntactically or semantically invalid (missing/blank
+  // required field, out-of-range value, unparseable NSQL, etc.).
+  ERROR_REASON_VALIDATION_FAILED = 1;
+  // No account exists for the referenced id.
+  ERROR_REASON_ACCOUNT_NOT_FOUND = 2;
+  // No filter rule exists for the referenced id.
+  ERROR_REASON_RULE_NOT_FOUND = 3;
+  // No stored message exists for the referenced id.
+  ERROR_REASON_MESSAGE_NOT_FOUND = 4;
+  // No stored calendar event exists for the referenced id.
+  ERROR_REASON_EVENT_NOT_FOUND = 5;
+  // No stored contact exists for the referenced id.
+  ERROR_REASON_CONTACT_NOT_FOUND = 6;
+  // The resource being created already exists.
+  ERROR_REASON_ALREADY_EXISTS = 7;
+  // A precondition is unmet: the operation requires configuration that is
+  // not present yet (e.g. no CalDAV/CardDAV account configured to sync).
+  ERROR_REASON_NOT_CONFIGURED = 8;
+  // A required backend (mail/calendar/contacts server) could not be reached.
+  ERROR_REASON_BACKEND_UNREACHABLE = 9;
+  // The call was not authenticated (missing/invalid bearer credentials).
+  ERROR_REASON_AUTH_REQUIRED = 10;
+  // The requested operation is not supported (no honest implementation).
+  ERROR_REASON_UNSUPPORTED = 11;
+  // An unexpected internal error the client cannot act on specifically.
+  ERROR_REASON_INTERNAL = 12;
+}
+
+// ErrorInfo is the structured error payload the daemon attaches to a failed
+// RPC's gRPC status-details (`grpc-status-details-bin`) trailer. It carries a
+// stable `reason` discriminant, optional structured `metadata` (e.g. the id
+// of the resource that was not found), and the same human-readable `message`
+// that the enclosing `Status` carries. Clients decode this to branch on
+// `reason` rather than parsing `message`.
+message ErrorInfo {
+  // Stable machine-readable failure discriminant.
+  ErrorReason reason = 1;
+  // Optional structured context (e.g. {"account_id": "acct-1"}). Keys are
+  // stable identifiers; values are the concrete offending inputs.
+  map<string, string> metadata = 2;
+  // Human-readable message, identical to the enclosing `Status` message.
+  string message = 3;
+}
+
 // System service exposes daemon-level introspection RPCs.
 //
 // This is the foundation of the versioned gRPC API (see
diff --git a/crates/nuncio-proto/src/errors.rs b/crates/nuncio-proto/src/errors.rs
new file mode 100644
index 0000000..c13f1c6
--- /dev/null
+++ b/crates/nuncio-proto/src/errors.rs
@@ -0,0 +1,153 @@
+//! Typed, machine-readable gRPC errors for the `nuncio.v1` contract.
+//!
+//! A failed RPC should carry a structured [`v1::ErrorInfo`] -- a stable
+//! [`v1::ErrorReason`] discriminant, optional metadata, and a human message
+//! -- in the gRPC `grpc-status-details-bin` status-details trailer, alongside
+//! the canonical gRPC [`Code`] that the reason maps to. Servers build such a
+//! `Status` with [`status`] / [`status_with_metadata`]; clients read the
+//! reason back with [`error_reason`] / [`error_info`] and branch on it,
+//! rather than string-matching the human-readable (English) message text.
+//!
+//! The reason-to-code mapping is centralized in [`ErrorReason::code`] so the
+//! transport-level status code and the application-level reason can never
+//! drift apart at a call site.
+
+use crate::v1::{ErrorInfo, ErrorReason};
+use bytes::Bytes;
+use prost::Message;
+use tonic::{Code, Status};
+
+impl ErrorReason {
+    /// The canonical gRPC [`Code`] this reason maps to. Both the unspecified
+    /// zero value and an explicit internal error map to [`Code::Internal`],
+    /// so an older client that receives a reason it does not recognize (and
+    /// decodes it as `Unspecified`) still sees an internal-class failure.
+    #[must_use]
+    pub fn code(self) -> Code {
+        match self {
+            ErrorReason::Unspecified | ErrorReason::Internal => Code::Internal,
+            ErrorReason::ValidationFailed => Code::InvalidArgument,
+            ErrorReason::AccountNotFound
+            | ErrorReason::RuleNotFound
+            | ErrorReason::MessageNotFound
+            | ErrorReason::EventNotFound
+            | ErrorReason::ContactNotFound => Code::NotFound,
+            ErrorReason::AlreadyExists => Code::AlreadyExists,
+            ErrorReason::NotConfigured => Code::FailedPrecondition,
+            ErrorReason::BackendUnreachable => Code::Unavailable,
+            ErrorReason::AuthRequired => Code::Unauthenticated,
+            ErrorReason::Unsupported => Code::Unimplemented,
+        }
+    }
+}
+
+/// Builds a [`Status`] for `reason` carrying `message` both as the `Status`
+/// message and inside a prost-encoded [`v1::ErrorInfo`] in the
+/// `grpc-status-details-bin` trailer, with the canonical code from
+/// [`ErrorReason::code`].
+pub fn status(reason: ErrorReason, message: impl Into<String>) -> Status {
+    build(reason, message.into(), Vec::new())
+}
+
+/// Like [`status`], but also attaches structured `metadata` (e.g. the id of
+/// the resource that was not found) to the encoded [`v1::ErrorInfo`].
+pub fn status_with_metadata(
+    reason: ErrorReason,
+    message: impl Into<String>,
+    metadata: impl IntoIterator<Item = (String, String)>,
+) -> Status {
+    build(reason, message.into(), metadata.into_iter().collect())
+}
+
+fn build(reason: ErrorReason, message: String, metadata: Vec<(String, String)>) -> Status {
+    let info = ErrorInfo {
+        reason: reason as i32,
+        metadata: metadata.into_iter().collect(),
+        message: message.clone(),
+    };
+    Status::with_details(reason.code(), message, Bytes::from(info.encode_to_vec()))
+}
+
+/// Decodes the [`v1::ErrorInfo`] carried in `status`'s details trailer, if
+/// one is present and well-formed. Returns `None` when the status carries no
+/// details (e.g. an error not built through this module) or the details are
+/// not a valid `ErrorInfo`.
+#[must_use]
+pub fn error_info(status: &Status) -> Option<ErrorInfo> {
+    let details = status.details();
+    if details.is_empty() {
+        return None;
+    }
+    ErrorInfo::decode(details).ok()
+}
+
+/// Decodes just the [`v1::ErrorReason`] from `status`'s details trailer,
+/// returning `None` when no typed [`v1::ErrorInfo`] is present.
+#[must_use]
+pub fn error_reason(status: &Status) -> Option<ErrorReason> {
+    error_info(status).map(|info| info.reason())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn each_reason_maps_to_its_canonical_code() {
+        assert_eq!(ErrorReason::Unspecified.code(), Code::Internal);
+        assert_eq!(ErrorReason::ValidationFailed.code(), Code::InvalidArgument);
+        assert_eq!(ErrorReason::AccountNotFound.code(), Code::NotFound);
+        assert_eq!(ErrorReason::RuleNotFound.code(), Code::NotFound);
+        assert_eq!(ErrorReason::MessageNotFound.code(), Code::NotFound);
+        assert_eq!(ErrorReason::EventNotFound.code(), Code::NotFound);
+        assert_eq!(ErrorReason::ContactNotFound.code(), Code::NotFound);
+        assert_eq!(ErrorReason::AlreadyExists.code(), Code::AlreadyExists);
+        assert_eq!(ErrorReason::NotConfigured.code(), Code::FailedPrecondition);
+        assert_eq!(ErrorReason::BackendUnreachable.code(), Code::Unavailable);
+        assert_eq!(ErrorReason::AuthRequired.code(), Code::Unauthenticated);
+        assert_eq!(ErrorReason::Unsupported.code(), Code::Unimplemented);
+        assert_eq!(ErrorReason::Internal.code(), Code::Internal);
+    }
+
+    #[test]
+    fn status_carries_reason_code_and_decodable_details() {
+        let status = status(ErrorReason::MessageNotFound, "message 'm-1' not found");
+        assert_eq!(status.code(), Code::NotFound);
+        assert_eq!(status.message(), "message 'm-1' not found");
+
+        let info = error_info(&status).expect("typed details present");
+        assert_eq!(info.reason(), ErrorReason::MessageNotFound);
+        assert_eq!(info.message, "message 'm-1' not found");
+        assert!(info.metadata.is_empty());
+
+        assert_eq!(
+            error_reason(&status),
+            Some(ErrorReason::MessageNotFound),
+            "the reason round-trips through the encoded details"
+        );
+    }
+
+    #[test]
+    fn status_with_metadata_round_trips_context() {
+        let status = status_with_metadata(
+            ErrorReason::AccountNotFound,
+            "account 'acct-1' not found",
+            [("account_id".to_string(), "acct-1".to_string())],
+        );
+        assert_eq!(status.code(), Code::NotFound);
+
+        let info = error_info(&status).expect("typed details present");
+        assert_eq!(info.reason(), ErrorReason::AccountNotFound);
+        assert_eq!(
+            info.metadata.get("account_id").map(String::as_str),
+            Some("acct-1")
+        );
+    }
+
+    #[test]
+    fn plain_status_without_details_decodes_to_none() {
+        let status = Status::not_found("no typed details here");
+        assert_eq!(error_info(&status), None);
+        assert_eq!(error_reason(&status), None);
+    }
+}
diff --git a/crates/nuncio-proto/src/lib.rs b/crates/nuncio-proto/src/lib.rs
index 6fe756c..6709375 100644
--- a/crates/nuncio-proto/src/lib.rs
+++ b/crates/nuncio-proto/src/lib.rs
@@ -22,6 +22,7 @@
 
 pub mod addr;
 pub mod client;
+pub mod errors;
 
 pub use addr::{grpc_addr_from_env, DEFAULT_GRPC_ADDR, GRPC_ADDR_ENV_VAR};
 
-- 
2.54.0


From b1538b2cb4b9af46ec98912ee5d40ba6c9d00a8e Mon Sep 17 00:00:00 2001
From: James Maes <jmaes@dmdbrands.com>
Date: Fri, 31 Jul 2026 14:09:51 -0500
Subject: [PATCH 4/5] feat(nunciod): return typed ErrorInfo details from error
 sites

Convert a representative set of daemon gRPC error returns to carry a typed
ErrorInfo in the status-details trailer via the nuncio-proto errors helper:
account/message/event/contact/rule not-found (with the offending id as
metadata), NSQL validation failures, and the calendar not-configured
precondition. Canonical status codes are unchanged, so existing error-path
assertions still hold. Add an offline over-the-wire E2E test that decodes the
reason and metadata from a returned Status against the real daemon server.

Refs #273
---
 crates/nunciod/src/grpc.rs                    | 119 +++++++++---
 crates/nunciod/tests/typed_errors_e2e_test.rs | 171 ++++++++++++++++++
 2 files changed, 261 insertions(+), 29 deletions(-)
 create mode 100644 crates/nunciod/tests/typed_errors_e2e_test.rs

diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs
index 23f0f66..1033f81 100644
--- a/crates/nunciod/src/grpc.rs
+++ b/crates/nunciod/src/grpc.rs
@@ -10,6 +10,7 @@ use nuncio_contacts::ContactsBackend;
 use nuncio_core::{CoreCommand, CoreEvent, EventBus};
 use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions};
 use nuncio_mail::{ImapEngine, MailBackend, MessageSender, SmtpTransportEngine};
+use nuncio_proto::errors;
 use nuncio_proto::v1::accounts_server::{Accounts, AccountsServer};
 use nuncio_proto::v1::audit_server::{Audit, AuditServer};
 use nuncio_proto::v1::calendar_server::{Calendar, CalendarServer};
@@ -19,6 +20,7 @@ use nuncio_proto::v1::export_server::{Export, ExportServer};
 use nuncio_proto::v1::filters_server::{Filters, FiltersServer};
 use nuncio_proto::v1::mail_server::{Mail, MailServer};
 use nuncio_proto::v1::system_server::{System, SystemServer};
+use nuncio_proto::v1::ErrorReason;
 use nuncio_proto::v1::{
     export_request, AccountConfig as AccountConfigProto, AccountProtocol as AccountProtocolProto,
     AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto,
@@ -549,7 +551,7 @@ impl Accounts for AccountsGrpcService {
         let config = map_account_config_from_proto(proto_config)?;
         config
             .validate()
-            .map_err(|e| Status::invalid_argument(e.to_string()))?;
+            .map_err(|e| errors::status(ErrorReason::ValidationFailed, e.to_string()))?;
 
         // Store the password credential in the OS keyring FIRST, keyed by
         // this account's `keyring_secret_key` -- it is never written to
@@ -615,7 +617,7 @@ impl Accounts for AccountsGrpcService {
         let config = map_account_config_from_proto(proto_config)?;
         config
             .validate()
-            .map_err(|e| Status::invalid_argument(e.to_string()))?;
+            .map_err(|e| errors::status(ErrorReason::ValidationFailed, e.to_string()))?;
 
         // Updating a non-existent account is a client error, not a silent
         // create: reject it so a typo'd id never conjures a new account row.
@@ -672,7 +674,13 @@ impl Accounts for AccountsGrpcService {
             .get_account(&req.id)
             .await
             .map_err(|e| Status::internal(format!("failed to look up account: {e}")))?
-            .ok_or_else(|| Status::not_found(format!("account '{}' not found", req.id)))?;
+            .ok_or_else(|| {
+                errors::status_with_metadata(
+                    ErrorReason::AccountNotFound,
+                    format!("account '{}' not found", req.id),
+                    [("account_id".to_string(), req.id.clone())],
+                )
+            })?;
 
         self.db
             .delete_account(&req.id)
@@ -708,7 +716,13 @@ impl Accounts for AccountsGrpcService {
             .get_account(&req.id)
             .await
             .map_err(|e| Status::internal(format!("failed to look up account: {e}")))?
-            .ok_or_else(|| Status::not_found(format!("account '{}' not found", req.id)))?;
+            .ok_or_else(|| {
+                errors::status_with_metadata(
+                    ErrorReason::AccountNotFound,
+                    format!("account '{}' not found", req.id),
+                    [("account_id".to_string(), req.id.clone())],
+                )
+            })?;
 
         let password = self
             .secrets
@@ -954,16 +968,19 @@ impl CalendarGrpcService {
             .collect();
 
         if caldav_accounts.is_empty() {
-            return Err(Status::failed_precondition(if req.account_id.is_empty() {
-                "no CalDAV account is configured; add one with protocol CALDAV and a \
-                 collection_url before syncing"
-                    .to_string()
-            } else {
-                format!(
-                    "no CalDAV account with id '{}' is configured",
-                    req.account_id
-                )
-            }));
+            return Err(errors::status(
+                ErrorReason::NotConfigured,
+                if req.account_id.is_empty() {
+                    "no CalDAV account is configured; add one with protocol CALDAV and a \
+                     collection_url before syncing"
+                        .to_string()
+                } else {
+                    format!(
+                        "no CalDAV account with id '{}' is configured",
+                        req.account_id
+                    )
+                },
+            ));
         }
 
         let mut total = 0usize;
@@ -1076,7 +1093,13 @@ impl Calendar for CalendarGrpcService {
             .db
             .get_calendar_event(&req.event_id)
             .await
-            .map_err(|e| Status::not_found(format!("event '{}' not found: {e}", req.event_id)))?;
+            .map_err(|e| {
+                errors::status_with_metadata(
+                    ErrorReason::EventNotFound,
+                    format!("event '{}' not found: {e}", req.event_id),
+                    [("event_id".to_string(), req.event_id.clone())],
+                )
+            })?;
 
         Ok(Response::new(GetEventResponse {
             event: Some(map_calendar_event_to_proto(event)),
@@ -1183,7 +1206,11 @@ impl Contacts for ContactsGrpcService {
         }
 
         let contact = self.db.get_contact(&req.contact_id).await.map_err(|e| {
-            Status::not_found(format!("contact '{}' not found: {e}", req.contact_id))
+            errors::status_with_metadata(
+                ErrorReason::ContactNotFound,
+                format!("contact '{}' not found: {e}", req.contact_id),
+                [("contact_id".to_string(), req.contact_id.clone())],
+            )
         })?;
 
         Ok(Response::new(GetContactResponse {
@@ -1358,7 +1385,11 @@ impl Mail for MailGrpcService {
         }
 
         let email = self.db.get_message(&req.message_id).await.map_err(|e| {
-            Status::not_found(format!("message '{}' not found: {e}", req.message_id))
+            errors::status_with_metadata(
+                ErrorReason::MessageNotFound,
+                format!("message '{}' not found: {e}", req.message_id),
+                [("message_id".to_string(), req.message_id.clone())],
+            )
         })?;
 
         Ok(Response::new(GetMessageResponse {
@@ -1387,7 +1418,11 @@ impl Mail for MailGrpcService {
             .set_message_read(&req.message_id, req.read)
             .await
             .map_err(|e| {
-                Status::not_found(format!("message '{}' not found: {e}", req.message_id))
+                errors::status_with_metadata(
+                    ErrorReason::MessageNotFound,
+                    format!("message '{}' not found: {e}", req.message_id),
+                    [("message_id".to_string(), req.message_id.clone())],
+                )
             })?;
 
         self.event_bus.process_command(CoreCommand::MarkRead {
@@ -1652,11 +1687,19 @@ impl Filters for FiltersGrpcService {
     ) -> Result<Response<CreateRuleResponse>, Status> {
         let req = request.into_inner();
 
-        let rule = NsqlParser::parse_rule(req.name, req.priority, &req.nsql)
-            .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?;
+        let rule = NsqlParser::parse_rule(req.name, req.priority, &req.nsql).map_err(|e| {
+            errors::status(
+                ErrorReason::ValidationFailed,
+                format!("NSQL syntax error: {e}"),
+            )
+        })?;
 
-        NsqlValidator::validate(&rule, &ValidationOptions::default())
-            .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?;
+        NsqlValidator::validate(&rule, &ValidationOptions::default()).map_err(|e| {
+            errors::status(
+                ErrorReason::ValidationFailed,
+                format!("NSQL validation error: {e}"),
+            )
+        })?;
 
         self.db
             .save_filter_rule(&rule)
@@ -1748,8 +1791,12 @@ impl Filters for FiltersGrpcService {
     ) -> Result<Response<PreviewRuleResponse>, Status> {
         let req = request.into_inner();
 
-        let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql)
-            .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?;
+        let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql).map_err(|e| {
+            errors::status(
+                ErrorReason::ValidationFailed,
+                format!("NSQL syntax error: {e}"),
+            )
+        })?;
 
         let preview_engine = FilterEngine::new(vec![rule])
             .map_err(|e| Status::internal(format!("failed to build preview engine: {e}")))?;
@@ -1788,17 +1835,31 @@ impl Filters for FiltersGrpcService {
             .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))?
             .into_iter()
             .find(|r| r.id == req.id)
-            .ok_or_else(|| Status::not_found(format!("filter rule '{}' not found", req.id)))?;
+            .ok_or_else(|| {
+                errors::status_with_metadata(
+                    ErrorReason::RuleNotFound,
+                    format!("filter rule '{}' not found", req.id),
+                    [("rule_id".to_string(), req.id.clone())],
+                )
+            })?;
 
         let name = req.name.unwrap_or(existing.name);
         let nsql = req.nsql.unwrap_or(existing.nsql_text);
         let priority = req.priority.unwrap_or(existing.priority);
 
-        let mut rule = NsqlParser::parse_rule(name, priority, &nsql)
-            .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?;
+        let mut rule = NsqlParser::parse_rule(name, priority, &nsql).map_err(|e| {
+            errors::status(
+                ErrorReason::ValidationFailed,
+                format!("NSQL syntax error: {e}"),
+            )
+        })?;
 
-        NsqlValidator::validate(&rule, &ValidationOptions::default())
-            .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?;
+        NsqlValidator::validate(&rule, &ValidationOptions::default()).map_err(|e| {
+            errors::status(
+                ErrorReason::ValidationFailed,
+                format!("NSQL validation error: {e}"),
+            )
+        })?;
 
         rule.id = existing.id;
 
diff --git a/crates/nunciod/tests/typed_errors_e2e_test.rs b/crates/nunciod/tests/typed_errors_e2e_test.rs
new file mode 100644
index 0000000..436f49c
--- /dev/null
+++ b/crates/nunciod/tests/typed_errors_e2e_test.rs
@@ -0,0 +1,171 @@
+//! Offline end-to-end test that typed `ErrorInfo` details survive the gRPC
+//! wire.
+//!
+//! Boots a real `nunciod` daemon gRPC server
+//! (`nunciod::grpc::serve_on_listener_with_overrides`) on an ephemeral
+//! loopback port, backed by a real temp file-backed [`DatabaseEngine`] and a
+//! [`SecretManager::mock`] vault (never the real OS keyring), with every
+//! engine override left at its default (no injected backends).
+//!
+//! Driving the authenticated `nuncio-proto` clients, it triggers several
+//! known error conditions and proves a client can machine-read each failure:
+//! the returned `tonic::Status` carries the correct canonical gRPC code AND a
+//! prost-encoded [`nuncio_proto::v1::ErrorInfo`] in the
+//! `grpc-status-details-bin` trailer whose `reason` (and metadata) decode
+//! back exactly, so a client never has to string-match the English message.
+//!
+//! Fully offline: no real network, no real OS keyring, no live server.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use nuncio_core::EventBus;
+use nuncio_filter::FilterEngine;
+use nuncio_proto::errors;
+use nuncio_proto::v1::{
+    CalendarSyncRequest, CreateRuleRequest, ErrorReason, GetMessageRequest, RemoveAccountRequest,
+};
+use nuncio_store::db::DatabaseEngine;
+use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
+use nunciod::grpc::{
+    serve_on_listener_with_overrides, AccountsEngineOverrides, CalendarEngineOverrides,
+    ContactsEngineOverrides, MailEngineOverrides,
+};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tonic::Code;
+
+/// Boots the real daemon gRPC server on an ephemeral loopback port with every
+/// override at its default. Returns the server address and minted bearer
+/// token; the `TempDir` guard is returned so the caller controls its lifetime
+/// for the whole test (the spawned server keeps using the db).
+async fn boot_daemon() -> (std::net::SocketAddr, String, tempfile::TempDir) {
+    let secrets = Arc::new(SecretManager::mock());
+    let dir = tempfile::tempdir().expect("create temp dir for file-backed db");
+    let db_path = dir.path().join("typed_errors_e2e.db");
+    let db = Arc::new(
+        DatabaseEngine::connect_file(&db_path, &secrets)
+            .await
+            .expect("open temp file-backed database"),
+    );
+
+    let token = hex::encode(
+        secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .expect("mint gRPC bearer token from mock vault"),
+    );
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind ephemeral loopback port");
+    let addr = listener.local_addr().expect("listener has local addr");
+
+    let event_bus = Arc::new(EventBus::new());
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+    let server_secrets = secrets.clone();
+    let server_token = token.clone();
+    tokio::spawn(async move {
+        let _ = serve_on_listener_with_overrides(
+            listener,
+            event_bus,
+            db,
+            filter_engine,
+            server_secrets,
+            server_token,
+            MailEngineOverrides::default(),
+            CalendarEngineOverrides::default(),
+            ContactsEngineOverrides::default(),
+            AccountsEngineOverrides::default(),
+        )
+        .await;
+    });
+
+    (addr, token, dir)
+}
+
+#[tokio::test]
+async fn typed_error_reasons_decode_from_status_details_over_the_wire() {
+    let (addr, token, _db_dir) = boot_daemon().await;
+    let addr_str = addr.to_string();
+
+    // ---- account-not-found: NotFound + ACCOUNT_NOT_FOUND + account_id ----
+    let mut accounts = nuncio_proto::client::connect_accounts(&addr_str, &token)
+        .await
+        .expect("accounts client connects");
+    let status = accounts
+        .remove_account(RemoveAccountRequest {
+            id: "no-such-account".to_string(),
+        })
+        .await
+        .expect_err("removing a nonexistent account must fail");
+    assert_eq!(status.code(), Code::NotFound);
+    let info = errors::error_info(&status).expect("typed ErrorInfo present in details trailer");
+    assert_eq!(info.reason(), ErrorReason::AccountNotFound);
+    assert_eq!(
+        info.metadata.get("account_id").map(String::as_str),
+        Some("no-such-account"),
+        "the missing id is carried as machine-readable metadata"
+    );
+    assert_eq!(info.message, status.message());
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::AccountNotFound)
+    );
+
+    // ---- message-not-found: NotFound + MESSAGE_NOT_FOUND + message_id ----
+    let mut mail = nuncio_proto::client::connect_mail(&addr_str, &token)
+        .await
+        .expect("mail client connects");
+    let status = mail
+        .get_message(GetMessageRequest {
+            message_id: "no-such-message".to_string(),
+        })
+        .await
+        .expect_err("fetching a nonexistent message must fail");
+    assert_eq!(status.code(), Code::NotFound);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::MessageNotFound)
+    );
+    let info = errors::error_info(&status).expect("typed ErrorInfo present");
+    assert_eq!(
+        info.metadata.get("message_id").map(String::as_str),
+        Some("no-such-message")
+    );
+
+    // ---- not-configured: FailedPrecondition + NOT_CONFIGURED ----
+    let mut calendar = nuncio_proto::client::connect_calendar(&addr_str, &token)
+        .await
+        .expect("calendar client connects");
+    let status = calendar
+        .sync(CalendarSyncRequest {
+            account_id: String::new(),
+            calendar_id: "personal".to_string(),
+            start_window: 0,
+            end_window: 0,
+        })
+        .await
+        .expect_err("syncing with no CalDAV account configured must fail");
+    assert_eq!(status.code(), Code::FailedPrecondition);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::NotConfigured)
+    );
+
+    // ---- validation-failed: InvalidArgument + VALIDATION_FAILED ----
+    let mut filters = nuncio_proto::client::connect_filters(&addr_str, &token)
+        .await
+        .expect("filters client connects");
+    let status = filters
+        .create_rule(CreateRuleRequest {
+            name: "bad rule".to_string(),
+            nsql: "this is not valid nsql".to_string(),
+            priority: 0,
+        })
+        .await
+        .expect_err("creating a rule from unparseable NSQL must fail");
+    assert_eq!(status.code(), Code::InvalidArgument);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::ValidationFailed)
+    );
+}
-- 
2.54.0


From dfc7cc9358fc1c4e886b99b81e87c765fd3e7699 Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Fri, 31 Jul 2026 14:11:05 -0500
Subject: [PATCH 5/5] perf(sync): batch new-message detection to remove the
 is_new_message N+1 (#341)

Replace the per-message get_message existence check in fetch_and_persist
with a single existing_message_ids(chunk_ids) lookup per fetched folder
chunk, plus a per-pass seen-set to keep duplicate ids within one chunk
firing at most once. Fire-once semantics for filter actions are
unchanged: a message is new iff it was absent from the store before this
sync pass and this is its first occurrence in the current chunk.

Refs #325

Co-authored-by: James Maes <jmaes@dmdbrands.com>
---
 crates/nuncio-store/src/db.rs |  67 +++++++++++++++++
 crates/nunciod/src/sync.rs    | 135 ++++++++++++++++++++++++++--------
 2 files changed, 171 insertions(+), 31 deletions(-)

diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs
index 5dd7ae9..7cba21a 100644
--- a/crates/nuncio-store/src/db.rs
+++ b/crates/nuncio-store/src/db.rs
@@ -1350,6 +1350,37 @@ impl DatabaseEngine {
         })
     }
 
+    /// Return the subset of `ids` already present in the `messages` table, as a single
+    /// `SELECT id FROM messages WHERE id IN (...)` query.
+    ///
+    /// 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.
+    pub async fn existing_message_ids(
+        &self,
+        ids: &[String],
+    ) -> Result<std::collections::HashSet<String>, DatabaseError> {
+        if ids.is_empty() {
+            return Ok(std::collections::HashSet::new());
+        }
+
+        let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> =
+            sqlx::QueryBuilder::new("SELECT id FROM messages WHERE id IN (");
+        let mut separated = builder.separated(", ");
+        for id in ids {
+            separated.push_bind(id);
+        }
+        builder.push(")");
+
+        let rows: Vec<(String,)> = builder
+            .build_query_as()
+            .fetch_all(&self.pool)
+            .await
+            .map_err(DatabaseError::Query)?;
+
+        Ok(rows.into_iter().map(|(id,)| id).collect())
+    }
+
     /// Save a [`nuncio_core::model::CalendarEvent`] to SQLite (INSERT OR REPLACE).
     ///
     /// Unlike [`Self::save_email`], calendar event summary/location are never encrypted at
@@ -3730,6 +3761,42 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn existing_message_ids_returns_only_the_ids_already_persisted() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+        engine
+            .save_email(&export_test_email("msg-1", "acct-a", "inbox"))
+            .await
+            .unwrap();
+        engine
+            .save_email(&export_test_email("msg-2", "acct-a", "inbox"))
+            .await
+            .unwrap();
+
+        let found = engine
+            .existing_message_ids(&[
+                "msg-1".to_string(),
+                "msg-2".to_string(),
+                "msg-3-never-saved".to_string(),
+            ])
+            .await
+            .unwrap();
+
+        assert_eq!(found.len(), 2);
+        assert!(found.contains("msg-1"));
+        assert!(found.contains("msg-2"));
+        assert!(!found.contains("msg-3-never-saved"));
+    }
+
+    #[tokio::test]
+    async fn existing_message_ids_with_empty_input_returns_empty_set_without_querying() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+
+        let found = engine.existing_message_ids(&[]).await.unwrap();
+
+        assert!(found.is_empty());
+    }
+
     #[tokio::test]
     async fn list_messages_for_export_with_no_filters_returns_every_message() {
         let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs
index bbab07d..98dc78a 100644
--- a/crates/nunciod/src/sync.rs
+++ b/crates/nunciod/src/sync.rs
@@ -195,39 +195,26 @@ pub async fn apply_filter_actions(
     applied
 }
 
-/// Reports whether `message_id` is not yet present in the store.
-///
-/// Sync is not yet incremental (a re-sync re-fetches every message the
-/// backend still reports, not just genuinely new ones), so this is the guard
-/// that keeps filter actions firing exactly once per message: it must be
-/// checked BEFORE `save_email` persists (or re-persists) the message, since
-/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the
-/// distinction between "arriving for the first time" and "seen again".
-///
-/// A lookup failure that is not "not found" (a genuine DB error) is logged
-/// and treated as "not new", the conservative choice: firing filters again on
-/// a message we can't positively identify as new risks duplicate remote
-/// mutations, which is worse than occasionally missing a fire on a message
-/// that was in fact new.
-async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str) -> bool {
-    match db.get_message(message_id).await {
-        Ok(_) => false,
-        Err(e) if e.is_not_found() => true,
-        Err(e) => {
-            tracing::warn!(
-                "could not determine whether message '{message_id}' is new before sync \
-                 (treating as not-new to avoid duplicate filter firing): {e}"
-            );
-            false
-        }
-    }
-}
-
 /// Fetch every folder and every message in every folder from `backend`,
 /// persisting each message via [`DatabaseEngine::save_email`] and, for
 /// messages genuinely new to the store, evaluating it against
-/// `filter_engine` via [`apply_filter_actions`] (see [`is_new_message`] for
-/// why this must be gated rather than run on every persisted message).
+/// `filter_engine` via [`apply_filter_actions`].
+///
+/// Sync is not yet incremental (a re-sync re-fetches every message the
+/// backend still reports, not just genuinely new ones), so new-vs-seen must
+/// be determined BEFORE `save_email` persists (or re-persists) anything --
+/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the
+/// distinction between "arriving for the first time" and "seen again". Doing
+/// that with one existence lookup per message (`DatabaseEngine::get_message`)
+/// is an N+1 query pattern over a fetched chunk; instead, this classifies an
+/// entire folder's chunk with a single batched
+/// `DatabaseEngine::existing_message_ids` lookup up front, then folds in a
+/// per-batch "already seen this pass" set while iterating so a chunk that
+/// itself contains a repeated id (e.g. a backend surfacing the same message
+/// twice in one fetch) still fires at most once -- exactly as if each
+/// message's existence had been checked one at a time immediately before its
+/// own `save_email`.
+///
 /// Returns the total number of messages processed (a message id "processed"
 /// more than once, e.g. by a backend that returns the same id from multiple
 /// folders, is counted once per occurrence -- `save_email` itself is
@@ -260,8 +247,19 @@ async fn fetch_and_persist(
             .sync_messages(&folder.id, last_state.as_deref())
             .await?;
         let fetched = emails.len();
+
+        // One round trip classifies the whole chunk instead of one lookup per
+        // message. A lookup failure that is not "not found" would be a
+        // genuine DB error; `existing_message_ids` surfaces those as `Err`
+        // rather than silently treating the batch as "not new", so callers
+        // never persist and skip-filter on an unverified assumption.
+        let chunk_ids: Vec<String> = emails.iter().map(|e| e.id.clone()).collect();
+        let already_present = db.existing_message_ids(&chunk_ids).await?;
+        let mut seen_this_pass = std::collections::HashSet::new();
+
         for email in emails {
-            let is_new = is_new_message(db, &email.id).await;
+            let is_new =
+                !already_present.contains(&email.id) && seen_this_pass.insert(email.id.clone());
             db.save_email(&email).await?;
             if is_new {
                 apply_filter_actions(db, filter_engine, &email).await;
@@ -1304,6 +1302,81 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn sync_with_backend_fires_filters_only_for_new_messages_in_a_mixed_batch() {
+        // A single fetched chunk containing a mix of an already-seen message
+        // (persisted by a prior sync) and genuinely new ones must fire filter
+        // actions ONLY for the new ones, via the batched
+        // `existing_message_ids` classification -- not a per-message lookup.
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+
+        let mark_read_rule = nuncio_filter::NsqlParser::parse_rule(
+            "Urgent Auto-Read",
+            1,
+            "WHERE subject CONTAINS 'Urgent' ACTION MARK READ",
+        )
+        .expect("parse rule");
+        let filter_engine = FilterEngine::new(vec![mark_read_rule]).expect("compile rule");
+
+        // Pre-seed one message directly, simulating it having landed in a
+        // prior sync pass. It must NOT re-fire even though this pass's
+        // backend still reports it (sync is not yet incremental).
+        db.save_email(&mock_email(
+            "m-already-seen",
+            "inbox",
+            "Urgent: known issue",
+        ))
+        .await
+        .expect("pre-seed existing message");
+
+        let mock = MockMailBackend::new();
+        mock.add_folder(Folder {
+            id: "inbox".to_string(),
+            name: "Inbox".to_string(),
+            total_messages: 3,
+            unread_messages: 3,
+        });
+        // Mixed chunk: one already-present id, two genuinely new ids.
+        mock.add_message(mock_email("m-already-seen", "inbox", "Urgent: known issue"));
+        mock.add_message(mock_email("m-new-1", "inbox", "Urgent: brand new"));
+        mock.add_message(mock_email("m-new-2", "inbox", "Urgent: also new"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 3, "every fetched message is still persisted");
+
+        let logs = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs");
+        assert_eq!(
+            logs.len(),
+            2,
+            "filters fire only for the two genuinely new messages in the mixed batch"
+        );
+        let fired_ids: std::collections::HashSet<String> =
+            logs.iter().map(|l| l.message_id.clone()).collect();
+        assert!(fired_ids.contains("m-new-1"));
+        assert!(fired_ids.contains("m-new-2"));
+        assert!(!fired_ids.contains("m-already-seen"));
+
+        // The pre-seeded message must still be persisted (INSERT OR REPLACE
+        // semantics are unaffected by the batched new/seen classification)
+        // but its read flag must be untouched by this pass's MARK READ rule.
+        let already_seen = db
+            .get_message("m-already-seen")
+            .await
+            .expect("still persisted");
+        assert!(
+            !already_seen.read,
+            "a message that already existed before this sync must not re-fire MARK READ"
+        );
+    }
+
     #[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
-- 
2.54.0