Cargo.lock

@@ -2604,7 +2604,6 @@ dependencies = [
  "thiserror 1.0.69",
  "tokio",
  "tracing",
- "tracing-subscriber",
  "wiremock",
  "zeroize",
 ]

crates/nuncio-core/src/config.rs

@@ -154,6 +154,21 @@ pub struct AccountConfig {
     pub keyring_secret_key: String,
     /// Background sync polling interval in seconds (minimum 10s).
     pub sync_interval_secs: u64,
+    /// Whether this daemon executes filter rules for this account.
+    ///
+    /// **Defaults to `false`**, which is the conservative answer to a problem
+    /// the engine cannot solve on its own. Every daemon syncing an account
+    /// decides independently that a message is new, so every daemon fires the
+    /// same rules against it. `MOVE`/`COPY`/`FLAG` survive that -- ten moves
+    /// converge on one outcome -- but `FORWARD` and `CALL WEBHOOK` land on a
+    /// third party with no shared state to compare against, so N daemons send
+    /// N forwards and N webhook calls.
+    ///
+    /// Enabling this on exactly one daemon per account makes filter execution
+    /// single-owner. It is a deliberate, documented limitation rather than a
+    /// silent duplication bug; a server-side claim is the eventual fix where
+    /// the server supports one.
+    pub filters_enabled: bool,
     /// The account's single transport.
     pub transport: Transport,
 }
@@ -343,6 +358,7 @@ mod tests {
             email_address: "user@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-123".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: Transport::ImapSmtp(valid_imap_smtp_transport()),
         }
     }
@@ -354,6 +370,7 @@ mod tests {
             email_address: "user@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-jmap-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: Transport::Jmap(JmapTransport {
                 endpoint_host: "jmap.nuncio.mx".to_string(),
             }),
@@ -367,6 +384,7 @@ mod tests {
             email_address: "user@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-cal-1".to_string(),
             sync_interval_secs: 300,
+            filters_enabled: false,
             transport: Transport::Dav(DavTransport {
                 collection_url: "https://caldav.example.com/calendars/user/work/".to_string(),
             }),

crates/nuncio-filter/src/webhook.rs

@@ -189,9 +189,26 @@ impl WebhookDispatcher {
             mac.update(format!("{timestamp}.{payload_str}").as_bytes());
             let signature = hex::encode(mac.finalize().into_bytes());
 
+            // Stable across every retry of this rule/message pair, so a
+            // receiver can discard a redelivery caused by a lost response
+            // rather than acting twice. It is NOT yet stable across daemons:
+            // `rule_id` is random per daemon, so two daemons executing the
+            // same rule produce different keys. Single-owner filter execution
+            // is what prevents that case today; a rule key derived from the
+            // normalized NSQL would make it dedupable directly.
+            let idempotency_key = {
+                let mut hasher = <Sha256 as sha2::Digest>::new();
+                for field in [rule_id, message_id] {
+                    sha2::Digest::update(&mut hasher, (field.len() as u64).to_le_bytes());
+                    sha2::Digest::update(&mut hasher, field.as_bytes());
+                }
+                hex::encode(sha2::Digest::finalize(hasher))
+            };
+
             let send_result = client
                 .post(url)
                 .header("Content-Type", "application/json")
+                .header("Idempotency-Key", &idempotency_key)
                 .header(
                     "X-Nuncio-Signature",
                     format!("t={timestamp},v1={signature}"),

crates/nuncio-store/src/db.rs

@@ -858,7 +858,13 @@ impl DatabaseEngine {
                 smtp_port INTEGER,
                 imap_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls',
                 smtp_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls',
-                collection_url TEXT
+                collection_url TEXT,
+                -- Defaults to 0 (off), including for accounts written before
+                -- the column existed. Filter execution is not safe to run on
+                -- more than one daemon per account, and defaulting a shared
+                -- account to "every daemon forwards" would be the wrong way
+                -- to be wrong.
+                filters_enabled INTEGER NOT NULL DEFAULT 0
             );
 
             CREATE TABLE IF NOT EXISTS filter_rules (
@@ -1184,8 +1190,8 @@ impl DatabaseEngine {
         sqlx::query(
             r#"
             INSERT OR REPLACE INTO accounts
-            (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             "#,
         )
         .bind(&config.id)
@@ -1208,6 +1214,7 @@ impl DatabaseEngine {
         .bind(tls_mode_to_db(imap_tls_mode))
         .bind(tls_mode_to_db(smtp_tls_mode))
         .bind(&collection_url)
+        .bind(i64::from(config.filters_enabled))
         .execute(&self.pool)
         .await
         .map_err(DatabaseError::Query)?;
@@ -1253,9 +1260,10 @@ impl DatabaseEngine {
             String,
             String,
             Option<String>,
+            i64,
         )> = sqlx::query_as(
             r#"
-            SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url
+            SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled
             FROM accounts
             "#,
         )
@@ -1280,6 +1288,7 @@ impl DatabaseEngine {
                     imap_tls_mode,
                     smtp_tls_mode,
                     collection_url,
+                    filters_enabled,
                 )| {
                     let protocol = serde_json::from_str(&protocol_str)
                         .unwrap_or(nuncio_core::AccountProtocol::ImapSmtp);
@@ -1326,6 +1335,7 @@ impl DatabaseEngine {
                         email_address,
                         keyring_secret_key,
                         sync_interval_secs: sync_interval_secs as u64,
+                        filters_enabled: filters_enabled != 0,
                         transport,
                     }
                 },
@@ -3292,6 +3302,7 @@ mod tests {
                 email_address: "contended@nuncio.mx".to_string(),
                 keyring_secret_key: "nuncio/acct-contended-1".to_string(),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                     imap_host: "imap.nuncio.mx".to_string(),
                     imap_port: 993,
@@ -3921,6 +3932,7 @@ mod tests {
             email_address: "work@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-test-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -3965,6 +3977,7 @@ mod tests {
             email_address: "mixed@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-tls-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 143,
@@ -4007,6 +4020,7 @@ mod tests {
             email_address: "cal@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-caldav-store-1".to_string(),
             sync_interval_secs: 300,
+            filters_enabled: false,
             transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport {
                 collection_url: "https://dav.example.com/calendars/user/work/".to_string(),
             }),
@@ -4039,6 +4053,7 @@ mod tests {
             email_address: "imap@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-mix-imap".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 143,
@@ -4054,6 +4069,7 @@ mod tests {
             email_address: "jmap@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-mix-jmap".to_string(),
             sync_interval_secs: 120,
+            filters_enabled: false,
             transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport {
                 endpoint_host: "jmap.nuncio.mx".to_string(),
             }),
@@ -4064,6 +4080,7 @@ mod tests {
             email_address: "dav@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-mix-dav".to_string(),
             sync_interval_secs: 300,
+            filters_enabled: false,
             transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport {
                 collection_url: "https://dav.example.com/cal/".to_string(),
             }),
@@ -4103,6 +4120,7 @@ mod tests {
             email_address: "del@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-del-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -4631,6 +4649,7 @@ mod tests {
             email_address: "reset@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-reset".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: true,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -4678,6 +4697,13 @@ mod tests {
         let accounts = engine.list_accounts().await.expect("list accounts");
         assert_eq!(accounts.len(), 1, "accounts must be preserved");
         assert_eq!(accounts[0].id, "acct-reset");
+        // Including which daemon owns filter execution: a rebuild that dropped
+        // it would silently switch the owner off, or -- worse, if it defaulted
+        // the other way -- switch a bystander on.
+        assert!(
+            accounts[0].filters_enabled,
+            "filter ownership must survive an identity rebuild"
+        );
 
         // The version is recorded, so the next open is a no-op.
         let (version,): (i64,) = sqlx::query_as("PRAGMA user_version")

crates/nuncio-store/src/recovery.rs

@@ -453,6 +453,13 @@ impl SqliteRecoveryEngine {
                             email_address,
                             keyring_secret_key,
                             sync_interval_secs: sync_interval_secs as u64,
+                            // Salvage deliberately does not recover this flag.
+                            // The column may be among what the corruption took,
+                            // and re-enabling filter execution on a guess would
+                            // start forwarding mail from a daemon the user never
+                            // chose as the owner. Off is recoverable by one
+                            // setting; wrongly on is not recoverable at all.
+                            filters_enabled: false,
                             transport,
                         }
                     },
@@ -618,6 +625,7 @@ mod tests {
                 email_address: "work@nuncio.mx".to_string(),
                 keyring_secret_key: "nuncio/acct-test-1".to_string(),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                     imap_host: "imap.nuncio.mx".to_string(),
                     imap_port: 993,
@@ -679,6 +687,7 @@ mod tests {
                 email_address: "salvage@nuncio.mx".to_string(),
                 keyring_secret_key: "nuncio/acct-salvage-1".to_string(),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                     imap_host: "imap.nuncio.mx".to_string(),
                     imap_port: 993,
@@ -816,6 +825,7 @@ mod tests {
                 email_address: "starttls@nuncio.mx".to_string(),
                 keyring_secret_key: "nuncio/acct-starttls-1".to_string(),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                     imap_host: "imap.nuncio.mx".to_string(),
                     imap_port: 143,

crates/nunciod/src/grpc.rs

@@ -490,6 +490,7 @@ fn map_account_config_from_proto(
         email_address: config.email_address,
         keyring_secret_key: config.keyring_secret_key,
         sync_interval_secs: nuncio_proto::time::duration_to_secs(&sync_interval),
+        filters_enabled: false,
         transport,
     })
 }
@@ -2069,12 +2070,29 @@ impl Mail for MailGrpcService {
         }
 
         let synced = if let Some(backend) = &self.overrides.mail_backend {
+            // Filter execution is owned by whichever daemon the account opts
+            // in (see `AccountConfig::filters_enabled`). Resolve it from the
+            // stored account rather than assuming: an injected backend must
+            // take the same path production does, or the gate would only ever
+            // be exercised in production.
+            let filters_enabled = match &account_id {
+                Some(id) => self
+                    .db
+                    .list_accounts()
+                    .await
+                    .map_err(|e| Status::internal(format!("sync failed: {e}")))?
+                    .into_iter()
+                    .find(|a| &a.id == id)
+                    .is_some_and(|a| a.filters_enabled),
+                None => false,
+            };
             crate::sync::sync_with_backend(
                 &self.db,
                 &self.event_bus,
                 backend.as_ref(),
                 &self.filter_engine,
                 account_id,
+                filters_enabled,
             )
             .await
             .map_err(|e| Status::internal(format!("sync failed: {e}")))?
@@ -2612,7 +2630,34 @@ impl Filters for FiltersGrpcService {
             req.chunk_size as usize
         };
 
-        tracing::info!(chunk_size, "Filters: triage rescan started");
+        // Triage produces exactly the side effects a live sync produces, so it
+        // is bound by the same single-owner rule: only accounts this daemon
+        // owns filter execution for may be acted on. Resolved once here rather
+        // than per message.
+        let owning_accounts: std::collections::HashSet<String> = self
+            .db
+            .list_accounts()
+            .await
+            .map_err(|e| Status::internal(format!("failed to read accounts for triage: {e}")))?
+            .into_iter()
+            .filter(|a| a.filters_enabled)
+            .map(|a| a.id)
+            .collect();
+
+        if owning_accounts.is_empty() {
+            // An honest refusal beats streaming a scan that reports progress
+            // and applies nothing, which would read as "no rules matched".
+            return Err(Status::failed_precondition(
+                "no account on this daemon has filter execution enabled; \
+                 enable filters_enabled on exactly one daemon per account",
+            ));
+        }
+
+        tracing::info!(
+            chunk_size,
+            owning_accounts = owning_accounts.len(),
+            "Filters: triage rescan started"
+        );
 
         let (tx, rx) = tokio::sync::mpsc::channel::<Result<TriageProgress, Status>>(16);
         let db = self.db.clone();
@@ -2643,27 +2688,36 @@ impl Filters for FiltersGrpcService {
                     let is_last_page = batch.len() < chunk_size;
 
                     for email in &batch {
-                        // Rules can ask where a message is, so a retroactive
-                        // pass evaluates each mailbox the message occupies --
-                        // the same fan-out live sync does. The fire-once claim
-                        // inside `apply_filter_actions` keeps a message in two
-                        // folders from acting twice.
-                        let placements = match db.placements_of(&email.id).await {
-                            Ok(placements) => placements,
-                            Err(e) => {
-                                let _ = tx
-                                    .send(Err(Status::internal(format!(
-                                        "failed to read placements during triage: {e}"
-                                    ))))
-                                    .await;
-                                return;
-                            }
-                        };
+                        // Messages belonging to an account this daemon does not
+                        // own are counted as scanned but never acted on, so the
+                        // progress total still reflects the whole store. A
+                        // non-owner does not even look up the placements: it has
+                        // nothing to do with them, and leaving the fire-once
+                        // claim untouched keeps it available to the owner.
                         let mut actions_for_email = 0usize;
-                        for placement in &placements {
-                            actions_for_email +=
-                                crate::sync::apply_filter_actions(&db, &engine, email, placement)
-                                    .await;
+                        if owning_accounts.contains(&email.account_id) {
+                            // Rules can ask where a message is, so a retroactive
+                            // pass evaluates each mailbox the message occupies --
+                            // the same fan-out live sync does. The fire-once claim
+                            // inside `apply_filter_actions` keeps a message in two
+                            // folders from acting twice.
+                            let placements = match db.placements_of(&email.id).await {
+                                Ok(placements) => placements,
+                                Err(e) => {
+                                    let _ = tx
+                                        .send(Err(Status::internal(format!(
+                                            "failed to read placements during triage: {e}"
+                                        ))))
+                                        .await;
+                                    return;
+                                }
+                            };
+                            for placement in &placements {
+                                actions_for_email += crate::sync::apply_filter_actions(
+                                    &db, &engine, email, placement,
+                                )
+                                .await;
+                            }
                         }
                         scanned += 1;
                         applied += actions_for_email as u64;
@@ -3893,6 +3947,7 @@ mod tests {
                 email_address: format!("{id}@nuncio.mx"),
                 keyring_secret_key: format!("nuncio/{id}"),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport {
                     endpoint_host: "jmap.nuncio.mx".to_string(),
                 }),
@@ -4654,6 +4709,7 @@ mod tests {
             email_address: "rotate@nuncio.mx".to_string(),
             keyring_secret_key: key.to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -4697,6 +4753,7 @@ mod tests {
             email_address: "rotate2@nuncio.mx".to_string(),
             keyring_secret_key: key.to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -4779,6 +4836,9 @@ mod tests {
                 email_address: format!("{}@nuncio.mx", placement.account_id),
                 keyring_secret_key: format!("nuncio/{}", placement.account_id),
                 sync_interval_secs: 60,
+                // This daemon owns filter execution for the seeded account, so
+                // triage tests exercise the acting path rather than the refusal.
+                filters_enabled: true,
                 transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport {
                     endpoint_host: "jmap.nuncio.mx".to_string(),
                 }),
@@ -5597,6 +5657,7 @@ mod tests {
             email_address: "sender@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-send-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -5770,6 +5831,7 @@ mod tests {
             email_address: "sender@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-injected-send-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -5886,6 +5948,7 @@ mod tests {
             email_address: "sender@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-injected-send-fail-1".to_string(),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,

crates/nunciod/src/orchestrator.rs

@@ -154,6 +154,7 @@ mod tests {
                 email_address: "orch@nuncio.mx".to_string(),
                 keyring_secret_key: "nuncio/acct-orch-1".to_string(),
                 sync_interval_secs: 60,
+                filters_enabled: false,
                 transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                     imap_host: "imap.nuncio.mx".to_string(),
                     imap_port: 993,

crates/nunciod/src/send.rs

@@ -292,6 +292,7 @@ mod tests {
             email_address: format!("{id}@nuncio.mx"),
             keyring_secret_key: format!("nuncio/{id}"),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,

crates/nunciod/src/sync.rs

@@ -289,6 +289,7 @@ async fn fetch_and_persist(
     backend: &dyn MailBackend,
     filter_engine: &FilterEngine,
     account_id: Option<&str>,
+    filters_enabled: bool,
 ) -> Result<usize, SyncError> {
     let folders = backend.sync_folders().await?;
     let mut synced = 0usize;
@@ -326,7 +327,14 @@ async fn fetch_and_persist(
             let is_new = !already_present.contains(&key) && seen_this_pass.insert(key);
             db.save_email_at(&message.email, message.source, &message.placement)
                 .await?;
-            if is_new {
+            // Filter execution is single-owner per account (see
+            // `AccountConfig::filters_enabled`). A daemon that is not the owner
+            // still syncs and stores the message; it just does not act on it,
+            // because the side-effecting actions -- FORWARD, CALL WEBHOOK --
+            // are not idempotent across daemons. The gate sits outside the
+            // fire-once claim on purpose: a non-owner must not consume the
+            // claim it is declining to act on.
+            if is_new && filters_enabled {
                 apply_filter_actions(db, filter_engine, &message.email, &message.placement).await;
             }
             synced += 1;
@@ -407,6 +415,7 @@ pub async fn sync_with_backend(
     backend: &dyn MailBackend,
     filter_engine: &FilterEngine,
     account_id: Option<String>,
+    filters_enabled: bool,
 ) -> Result<usize, SyncError> {
     match &account_id {
         Some(id) => event_bus.process_command(CoreCommand::SyncAccount {
@@ -415,8 +424,15 @@ pub async fn sync_with_backend(
         None => event_bus.process_command(CoreCommand::SyncAll),
     }
 
-    let result =
-        fetch_and_persist(db, event_bus, backend, filter_engine, account_id.as_deref()).await;
+    let result = fetch_and_persist(
+        db,
+        event_bus,
+        backend,
+        filter_engine,
+        account_id.as_deref(),
+        filters_enabled,
+    )
+    .await;
 
     if let Err(e) = &result {
         event_bus.process_command(CoreCommand::ReportError {
@@ -488,18 +504,20 @@ pub async fn run_account_sync(
     let setup = async {
         let config = find_account(db, account_id).await?;
         let password = secrets.get_secret(&config.keyring_secret_key)?;
-        build_mail_backend(&config, &password)
+        let filters_enabled = config.filters_enabled;
+        build_mail_backend(&config, &password).map(|backend| (backend, filters_enabled))
     }
     .await;
 
     match setup {
-        Ok(backend) => {
+        Ok((backend, filters_enabled)) => {
             sync_with_backend(
                 db,
                 event_bus,
                 backend.as_ref(),
                 filter_engine,
                 Some(account_id.to_string()),
+                filters_enabled,
             )
             .await
         }
@@ -540,6 +558,7 @@ mod tests {
             email_address: format!("{id}@nuncio.mx"),
             keyring_secret_key: format!("nuncio/{id}"),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: Transport::Jmap(nuncio_core::JmapTransport {
                 endpoint_host: host.to_string(),
             }),
@@ -626,6 +645,7 @@ mod tests {
             email_address: format!("{id}@nuncio.mx"),
             keyring_secret_key: format!("nuncio/{id}"),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -714,6 +734,7 @@ mod tests {
             &mock,
             &filter_engine,
             Some("acct-mock-1".to_string()),
+            true,
         )
         .await
         .expect("sync succeeds");
@@ -773,7 +794,7 @@ mod tests {
         });
 
         let filter_engine = empty_filter_engine();
-        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 0);
@@ -816,6 +837,7 @@ mod tests {
             &mock,
             &filter_engine,
             Some("acct-fail".to_string()),
+            true,
         )
         .await
         .expect_err("sync fails");
@@ -1036,7 +1058,7 @@ mod tests {
         });
         mock.add_message(mock_placed("m-urgent", "inbox", "Urgent: server down"));
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 1);
@@ -1063,6 +1085,96 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn a_daemon_that_does_not_own_filters_stores_the_mail_but_fires_nothing() {
+        // The single-owner rule. A non-owning daemon must still sync and store
+        // -- it is a full mail client -- while producing none of the side
+        // effects that would be duplicated across the fleet.
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+
+        let 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![rule]).expect("compile rule");
+
+        let mock = MockMailBackend::new();
+        mock.add_folder(Folder {
+            id: "inbox".to_string(),
+            name: "Inbox".to_string(),
+            total_messages: 1,
+            unread_messages: 1,
+        });
+        mock.add_message(mock_placed("m-urgent", "inbox", "Urgent: server down"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, false)
+            .await
+            .expect("sync succeeds");
+
+        assert_eq!(synced, 1, "the message must still be synced and stored");
+        assert!(
+            !read_flag_of(&db, "m-urgent").await,
+            "a non-owning daemon must not apply MARK READ"
+        );
+        assert!(
+            db.list_filter_execution_logs(10)
+                .await
+                .expect("list logs")
+                .is_empty(),
+            "no rule ran, so nothing may be logged as executed"
+        );
+        assert!(
+            db.list_pending_mutations(10)
+                .await
+                .expect("list mutations")
+                .is_empty(),
+            "a non-owning daemon must enqueue no remote mutation"
+        );
+    }
+
+    #[tokio::test]
+    async fn run_account_sync_takes_filter_ownership_from_the_stored_account() {
+        // The flag has to come from the account row, not a caller's guess --
+        // otherwise the gate is only ever exercised by tests that pass it.
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+
+        let mut account = sample_imap_account("acct-owner");
+        assert!(
+            !account.filters_enabled,
+            "a freshly constructed account must default to not owning filters"
+        );
+        db.save_account(&account).await.expect("save account");
+        let reloaded = db
+            .list_accounts()
+            .await
+            .expect("list accounts")
+            .into_iter()
+            .find(|a| a.id == "acct-owner")
+            .expect("account round-trips");
+        assert!(!reloaded.filters_enabled);
+
+        account.filters_enabled = true;
+        db.save_account(&account).await.expect("update account");
+        let reloaded = db
+            .list_accounts()
+            .await
+            .expect("list accounts")
+            .into_iter()
+            .find(|a| a.id == "acct-owner")
+            .expect("account round-trips");
+        assert!(
+            reloaded.filters_enabled,
+            "opting one daemon in must persist"
+        );
+    }
+
     #[tokio::test]
     async fn sync_with_backend_enqueues_outbox_mutation_for_remote_action_on_match() {
         let (db, _dir) = DatabaseEngine::connect_ephemeral()
@@ -1087,7 +1199,7 @@ mod tests {
         });
         mock.add_message(mock_placed("m-archive", "inbox", "Please Archive Me"));
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 1);
@@ -1137,7 +1249,7 @@ mod tests {
         });
         mock.add_message(mock_placed("m-plain", "inbox", "Just a normal update"));
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 1);
@@ -1201,7 +1313,7 @@ mod tests {
         mock.add_message(repeat);
 
         // First sync: the message is genuinely new, so both rules must fire.
-        let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("first sync succeeds");
         assert_eq!(synced_first, 1);
@@ -1229,7 +1341,7 @@ mod tests {
 
         // Second sync: the mock backend reports the SAME message again
         // (unchanged id), simulating a non-incremental re-sync.
-        let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("second sync succeeds");
         assert_eq!(synced_second, 1);
@@ -1303,7 +1415,7 @@ mod tests {
         mock.add_message(mock_placed("m-new-1", "inbox", "Urgent: brand new"));
         mock.add_message(mock_placed("m-new-2", "inbox", "Urgent: also new"));
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 3, "every fetched message is still persisted");
@@ -1363,6 +1475,7 @@ mod tests {
             &mock,
             &filter_engine,
             Some("acct-inc-1".to_string()),
+            true,
         )
         .await
         .expect("first sync succeeds");
@@ -1384,6 +1497,7 @@ mod tests {
             &mock,
             &filter_engine,
             Some("acct-inc-1".to_string()),
+            true,
         )
         .await
         .expect("second sync succeeds");
@@ -1436,6 +1550,7 @@ mod tests {
             &backend,
             &filter_engine,
             Some(MockMailBackend::SHARED_ACCOUNT_ID),
+            true,
         )
         .await
         .expect("sync succeeds");
@@ -1489,6 +1604,7 @@ mod tests {
             &backend,
             &filter_engine,
             Some(MockMailBackend::SHARED_ACCOUNT_ID),
+            true,
         )
         .await
         .expect("sync succeeds");
@@ -1602,7 +1718,7 @@ mod tests {
 
         // `None` account: the pass has nowhere to key a checkpoint and cannot
         // run the `present` diff, but the removal is still actionable.
-        let synced = fetch_and_persist(&db, &event_bus, &mock, &filter_engine, None)
+        let synced = fetch_and_persist(&db, &event_bus, &mock, &filter_engine, None, true)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 0, "nothing new arrived");

crates/nunciod/src/sync_dispatcher.rs

@@ -806,6 +806,7 @@ mod tests {
             email_address: format!("{id}@nuncio.mx"),
             keyring_secret_key: format!("nuncio/{id}"),
             sync_interval_secs: 60,
+            filters_enabled: false,
             transport: Transport::Jmap(JmapTransport {
                 endpoint_host: "jmap.nuncio.mx".to_string(),
             }),

crates/nunciod/tests/filters_triage_e2e_test.rs

@@ -54,6 +54,27 @@ fn sample_email(id: &str, subject: &str) -> Email {
     }
 }
 
+/// The account the seeded messages belong to, opted in as this daemon's
+/// filter-execution owner.
+fn owning_account() -> nuncio_core::AccountConfig {
+    nuncio_core::AccountConfig {
+        id: ACCOUNT_ID.to_string(),
+        name: "Triage E2E".to_string(),
+        email_address: "triage-e2e@nuncio.mx".to_string(),
+        keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"),
+        sync_interval_secs: 60,
+        filters_enabled: true,
+        transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
+            imap_host: "imap.nuncio.mx".to_string(),
+            imap_port: 993,
+            imap_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+            smtp_host: "smtp.nuncio.mx".to_string(),
+            smtp_port: 465,
+            smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+        }),
+    }
+}
+
 /// The INBOX occupancy a seeded message sits in. `id` doubles as the UID, so
 /// distinct messages occupy distinct rows.
 fn sample_placement(id: &str) -> Placement {
@@ -111,6 +132,13 @@ async fn boot_daemon(
             .expect("open temp file-backed database"),
     );
 
+    // Triage only acts on accounts this daemon owns filter execution for, so
+    // the account must exist and be opted in -- otherwise the RPC honestly
+    // refuses rather than scanning and applying nothing.
+    db.save_account(&owning_account())
+        .await
+        .expect("seed the triage account");
+
     let token_bytes = secrets
         .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
         .expect("mint gRPC bearer token from mock vault");
@@ -292,3 +320,54 @@ async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() {
         prev_scanned = update.scanned_count;
     }
 }
+
+/// A daemon that owns filter execution for no account must refuse `Triage`
+/// outright. Streaming a scan that reports progress and applies nothing would
+/// be indistinguishable, from the caller's side, from "no rules matched" --
+/// and the whole point of single-owner execution is that the non-owner stays
+/// visibly inert rather than quietly so.
+#[tokio::test]
+async fn triage_refuses_when_this_daemon_owns_no_account() {
+    let secrets = Arc::new(SecretManager::mock());
+    let rule = NsqlParser::parse_rule(
+        "Urgent Auto-Read",
+        1,
+        "WHERE subject CONTAINS 'Urgent' ACTION MARK READ",
+    )
+    .expect("parse rule");
+    let filter_engine = Arc::new(FilterEngine::new(vec![rule]).expect("compile rule"));
+
+    let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await;
+
+    // Take ownership away: the account exists, but this daemon is not its
+    // filter owner.
+    let mut account = owning_account();
+    account.filters_enabled = false;
+    db.save_account(&account)
+        .await
+        .expect("opt the account out");
+
+    seed(&db, "m-1", "Urgent: server down").await;
+
+    let mut client = FiltersClient::connect(format!("http://{addr}"))
+        .await
+        .expect("connect to the daemon");
+    let status = client
+        .triage(authed_request(&token, TriageRequest::default()))
+        .await
+        .expect_err("triage must refuse without an owning account");
+
+    assert_eq!(status.code(), tonic::Code::FailedPrecondition);
+    assert!(
+        status.message().contains("filter execution"),
+        "the refusal must say why, got: {}",
+        status.message()
+    );
+
+    // And it must genuinely have applied nothing: the occupancy the rule
+    // would have marked read is still unread.
+    assert!(
+        !read_flag_of(&db, "m-1").await,
+        "a refused triage must not apply any action"
+    );
+}

crates/nunciod/tests/outbox_executor_e2e_test.rs

@@ -101,6 +101,7 @@ fn sample_account() -> nuncio_core::AccountConfig {
         email_address: "owner@nuncio.mx".to_string(),
         keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"),
         sync_interval_secs: 60,
+        filters_enabled: false,
         transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport {
             imap_host: "imap.nuncio.mx".to_string(),
             imap_port: 993,