crates/nuncio-store/src/db.rs

@@ -67,6 +67,17 @@ impl DatabaseError {
             _ => false,
         }
     }
+
+    /// Returns true if this error means "no row matched the lookup" (e.g. a
+    /// `get_message`/`get_contact`/`get_calendar_event` call for an id that
+    /// was never saved), as opposed to a genuine I/O, corruption, or key
+    /// provisioning failure.
+    ///
+    /// Exposed so callers outside this crate can distinguish "not found" from
+    /// other failures without depending on `sqlx` directly.
+    pub fn is_not_found(&self) -> bool {
+        matches!(self, DatabaseError::Query(sqlx::Error::RowNotFound))
+    }
 }
 
 /// SQLite database storage engine managing WAL connection pools and migrations.

crates/nunciod/src/grpc.rs

@@ -865,6 +865,7 @@ struct MailGrpcService {
     db: Arc<DatabaseEngine>,
     event_bus: Arc<EventBus>,
     secrets: Arc<SecretManager>,
+    filter_engine: Arc<FilterEngine>,
     overrides: MailEngineOverrides,
 }
 
@@ -1053,15 +1054,33 @@ impl Mail for MailGrpcService {
         let account_id = request.into_inner().account_id;
 
         let synced = if let Some(backend) = &self.overrides.mail_backend {
-            crate::sync::sync_with_backend(&self.db, &self.event_bus, backend.as_ref(), account_id)
-                .await
-                .map_err(|e| Status::internal(format!("sync failed: {e}")))?
+            crate::sync::sync_with_backend(
+                &self.db,
+                &self.event_bus,
+                backend.as_ref(),
+                &self.filter_engine,
+                account_id,
+            )
+            .await
+            .map_err(|e| Status::internal(format!("sync failed: {e}")))?
         } else if let Some(account_id) = account_id {
-            crate::sync::run_account_sync(&self.db, &self.secrets, &self.event_bus, &account_id)
-                .await
-                .map_err(|e| Status::internal(format!("sync failed: {e}")))?
+            crate::sync::run_account_sync(
+                &self.db,
+                &self.secrets,
+                &self.event_bus,
+                &self.filter_engine,
+                &account_id,
+            )
+            .await
+            .map_err(|e| Status::internal(format!("sync failed: {e}")))?
         } else {
-            crate::sync::run_all_accounts_sync(&self.db, &self.secrets, &self.event_bus).await
+            crate::sync::run_all_accounts_sync(
+                &self.db,
+                &self.secrets,
+                &self.event_bus,
+                &self.filter_engine,
+            )
+            .await
         };
 
         Ok(Response::new(SyncResponse {
@@ -1607,6 +1626,7 @@ pub async fn serve_on_listener_with_overrides(
         db: db.clone(),
         event_bus,
         secrets: secrets.clone(),
+        filter_engine: filter_engine.clone(),
         overrides,
     };
     let mail_interceptor = BearerAuthInterceptor::new(token.clone());

crates/nunciod/src/main.rs

@@ -54,6 +54,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         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 {
                 match cmd {
@@ -62,6 +63,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                             &db_sync,
                             &secrets_sync,
                             &event_bus_sync,
+                            &filter_engine_sync,
                         )
                         .await;
                         tracing::info!("SyncAll completed: {} message(s) synced", synced);
@@ -71,6 +73,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                             &db_sync,
                             &secrets_sync,
                             &event_bus_sync,
+                            &filter_engine_sync,
                             &account_id,
                         )
                         .await

crates/nunciod/src/sync.rs

@@ -19,6 +19,7 @@
 //! real backend; they then delegate the actual fetch/persist/event-emission
 //! work to [`sync_with_backend`].
 use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, EventBus};
+use nuncio_filter::{FilterEngine, OutboxManager, RuleAction};
 use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError};
 use nuncio_store::db::DatabaseError;
 use nuncio_store::vault::{SecretManager, VaultError};
@@ -41,12 +42,193 @@ pub enum SyncError {
     Store(#[from] DatabaseError),
 }
 
+/// Map a [`RuleAction`] to the short mutation tag persisted on
+/// [`nuncio_filter::PendingRemoteMutation::mutation_type`] and the outbox
+/// target parameter, for every action that is not applied immediately.
+///
+/// Nothing downstream currently parses `mutation_type` back into a
+/// `RuleAction`, so these tags are a fixed vocabulary owned by this module;
+/// keep them stable once the outbox worker starts consuming them.
+fn remote_action_tag(action: &RuleAction) -> Option<(&'static str, Option<String>)> {
+    match action {
+        RuleAction::MoveTo(folder) => Some(("MOVE", Some(folder.clone()))),
+        RuleAction::CopyTo(folder) => Some(("COPY", Some(folder.clone()))),
+        RuleAction::Flag => Some(("FLAG", None)),
+        RuleAction::Unflag => Some(("UNFLAG", None)),
+        RuleAction::Delete => Some(("DELETE", None)),
+        RuleAction::ForwardTo(address) => Some(("FORWARD", Some(address.clone()))),
+        RuleAction::CallWebhook(url) => Some(("WEBHOOK", Some(url.clone()))),
+        RuleAction::MarkRead | RuleAction::MarkUnread => None,
+    }
+}
+
+/// Evaluate `email` against `filter_engine` and route every matched rule's
+/// actions to their real, persisted effect.
+///
+/// Filters fire EXACTLY ONCE per newly-arrived message: callers (see
+/// [`fetch_and_persist`]) must only invoke this for a message that was
+/// genuinely new to the store on this sync pass, never for one that already
+/// existed. Sync is not yet incremental -- every sync re-fetches and
+/// re-persists messages the backend still reports -- so calling this
+/// unconditionally on every persisted message would re-fire a rule's actions
+/// on every single re-sync of the same mailbox: a fresh
+/// `PendingRemoteMutation` and a fresh `FilterExecutionLog` entry per cycle
+/// for a message that arrived once, growing the outbox and the audit ledger
+/// without bound and (once a real mutation transport lands) duplicating the
+/// real remote operation. This function itself has no way to tell "new" from
+/// "already seen" -- that determination is the caller's responsibility.
+///
+/// `MarkRead`/`MarkUnread` apply immediately to the stored message via
+/// [`DatabaseEngine::set_message_read`]. Every other action enqueues a real
+/// [`nuncio_filter::PendingRemoteMutation`] through the existing outbox
+/// (`OutboxManager::create_mutation` + `DatabaseEngine::save_pending_mutation`)
+/// for the background outbox worker to pick up -- this function never
+/// performs the real remote IMAP/JMAP mutation or webhook call itself.
+/// Every match also appends a [`nuncio_filter::FilterExecutionLog`] entry via
+/// `DatabaseEngine::save_filter_execution_log`, so there is an audit trail of
+/// what fired independent of whether the outbox item is ever drained.
+///
+/// A bookkeeping failure on one action (a DB write error) is logged via
+/// `tracing::warn!` and does not stop the remaining actions/rules from being
+/// applied -- one broken write must never silently swallow the rest of a
+/// sync's filtering. The return value counts only actions that were fully
+/// applied AND logged; a partially-failed match (e.g. the mutation persists
+/// but the execution log write fails) is not counted as success, keeping the
+/// count honest for callers that log or assert on it.
+///
+/// Standalone and reusable by design: this is the exact evaluate-and-route
+/// logic other sync paths (e.g. a bulk retroactive triage over the whole
+/// store) need, so it takes only a `DatabaseEngine`/`FilterEngine`/`Email`
+/// and has no dependency on the live inbound-sync call chain.
+pub async fn apply_filter_actions(
+    db: &nuncio_store::db::DatabaseEngine,
+    filter_engine: &FilterEngine,
+    email: &nuncio_core::model::Email,
+) -> usize {
+    let mut applied = 0usize;
+
+    for (rule, actions) in filter_engine.evaluate(email) {
+        for action in actions {
+            let immediate_ok = match &action {
+                RuleAction::MarkRead => match db.set_message_read(&email.id, true).await {
+                    Ok(()) => true,
+                    Err(e) => {
+                        tracing::warn!(
+                            "filter rule '{}' MARK READ failed for message '{}': {e}",
+                            rule.id,
+                            email.id
+                        );
+                        false
+                    }
+                },
+                RuleAction::MarkUnread => match db.set_message_read(&email.id, false).await {
+                    Ok(()) => true,
+                    Err(e) => {
+                        tracing::warn!(
+                            "filter rule '{}' MARK UNREAD failed for message '{}': {e}",
+                            rule.id,
+                            email.id
+                        );
+                        false
+                    }
+                },
+                RuleAction::MoveTo(_)
+                | RuleAction::CopyTo(_)
+                | RuleAction::Flag
+                | RuleAction::Unflag
+                | RuleAction::Delete
+                | RuleAction::ForwardTo(_)
+                | RuleAction::CallWebhook(_) => {
+                    // Remote actions are handled below via the outbox; this
+                    // arm only tracks whether the immediate-action write
+                    // itself succeeded (there is none for a remote action,
+                    // so it is unconditionally "not applicable" here). Listed
+                    // explicitly rather than a wildcard so a future
+                    // immediate-style action can't silently fall through to
+                    // remote routing by accident.
+                    true
+                }
+            };
+
+            if !immediate_ok {
+                continue;
+            }
+
+            let mut action_ok = true;
+            if let Some((tag, target)) = remote_action_tag(&action) {
+                let mutation = OutboxManager::create_mutation(&rule.id, &email.id, tag, target);
+                if let Err(e) = db.save_pending_mutation(&mutation).await {
+                    tracing::warn!(
+                        "filter rule '{}' failed to enqueue outbox mutation for message '{}': {e}",
+                        rule.id,
+                        email.id
+                    );
+                    action_ok = false;
+                }
+            }
+
+            if !action_ok {
+                continue;
+            }
+
+            let action_desc = action.to_nsql();
+            if let Err(e) = db
+                .save_filter_execution_log(&rule.id, &email.id, &action_desc)
+                .await
+            {
+                tracing::warn!(
+                    "filter rule '{}' matched message '{}' but failed to write execution log: {e}",
+                    rule.id,
+                    email.id
+                );
+                continue;
+            }
+
+            applied += 1;
+        }
+    }
+
+    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`]. 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 `INSERT OR
-/// REPLACE`, so storage stays deduplicated by message id regardless).
+/// 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).
+/// 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
+/// `INSERT OR REPLACE`, so storage stays deduplicated by message id
+/// regardless).
 ///
 /// Contains no event-bus or credential logic -- pure fetch-and-persist, so
 /// it is the smallest unit tests can exercise directly with a
@@ -54,13 +236,18 @@ pub enum SyncError {
 async fn fetch_and_persist(
     db: &nuncio_store::db::DatabaseEngine,
     backend: &dyn MailBackend,
+    filter_engine: &FilterEngine,
 ) -> Result<usize, SyncError> {
     let folders = backend.sync_folders().await?;
     let mut synced = 0usize;
     for folder in folders {
         let (emails, _state) = backend.sync_messages(&folder.id, None).await?;
         for email in emails {
+            let is_new = is_new_message(db, &email.id).await;
             db.save_email(&email).await?;
+            if is_new {
+                apply_filter_actions(db, filter_engine, &email).await;
+            }
             synced += 1;
         }
     }
@@ -85,6 +272,7 @@ pub async fn sync_with_backend(
     db: &nuncio_store::db::DatabaseEngine,
     event_bus: &EventBus,
     backend: &dyn MailBackend,
+    filter_engine: &FilterEngine,
     account_id: Option<String>,
 ) -> Result<usize, SyncError> {
     match &account_id {
@@ -94,7 +282,7 @@ pub async fn sync_with_backend(
         None => event_bus.process_command(CoreCommand::SyncAll),
     }
 
-    let result = fetch_and_persist(db, backend).await;
+    let result = fetch_and_persist(db, backend, filter_engine).await;
 
     if let Err(e) = &result {
         event_bus.process_command(CoreCommand::ReportError {
@@ -153,6 +341,7 @@ pub async fn run_account_sync(
     db: &nuncio_store::db::DatabaseEngine,
     secrets: &SecretManager,
     event_bus: &EventBus,
+    filter_engine: &FilterEngine,
     account_id: &str,
 ) -> Result<usize, SyncError> {
     let setup = async {
@@ -168,6 +357,7 @@ pub async fn run_account_sync(
                 db,
                 event_bus,
                 backend.as_ref(),
+                filter_engine,
                 Some(account_id.to_string()),
             )
             .await
@@ -192,6 +382,7 @@ pub async fn run_all_accounts_sync(
     db: &nuncio_store::db::DatabaseEngine,
     secrets: &SecretManager,
     event_bus: &EventBus,
+    filter_engine: &FilterEngine,
 ) -> usize {
     event_bus.process_command(CoreCommand::SyncAll);
 
@@ -202,7 +393,7 @@ pub async fn run_all_accounts_sync(
                 match secrets.get_secret(&config.keyring_secret_key) {
                     Ok(password) => {
                         let backend = build_mail_backend(&config, &password);
-                        match fetch_and_persist(db, backend.as_ref()).await {
+                        match fetch_and_persist(db, backend.as_ref(), filter_engine).await {
                             Ok(count) => total_synced += count,
                             Err(e) => {
                                 event_bus.process_command(CoreCommand::ReportError {
@@ -247,6 +438,12 @@ mod tests {
     use wiremock::matchers::{body_partial_json, method, path};
     use wiremock::{Mock, MockServer, ResponseTemplate};
 
+    /// A `FilterEngine` with no rules, for tests that exercise sync
+    /// mechanics unrelated to filtering.
+    fn empty_filter_engine() -> FilterEngine {
+        FilterEngine::new(Vec::new()).expect("empty rule set")
+    }
+
     fn sample_jmap_account(id: &str) -> AccountConfig {
         sample_jmap_account_at(id, "jmap.nuncio.mx")
     }
@@ -394,9 +591,16 @@ mod tests {
         mock.add_message(mock_email("m1", "inbox", "Hello"));
         mock.add_message(mock_email("m2", "inbox", "World"));
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, Some("acct-mock-1".to_string()))
-            .await
-            .expect("sync succeeds");
+        let filter_engine = empty_filter_engine();
+        let synced = sync_with_backend(
+            &db,
+            &event_bus,
+            &mock,
+            &filter_engine,
+            Some("acct-mock-1".to_string()),
+        )
+        .await
+        .expect("sync succeeds");
         assert_eq!(synced, 2);
 
         let persisted = db
@@ -439,7 +643,8 @@ mod tests {
             unread_messages: 0,
         });
 
-        let synced = sync_with_backend(&db, &event_bus, &mock, None)
+        let filter_engine = empty_filter_engine();
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
             .await
             .expect("sync succeeds");
         assert_eq!(synced, 0);
@@ -465,9 +670,16 @@ mod tests {
         let mock = MockMailBackend::new();
         mock.set_should_fail(true);
 
-        let err = sync_with_backend(&db, &event_bus, &mock, Some("acct-fail".to_string()))
-            .await
-            .expect_err("sync fails");
+        let filter_engine = empty_filter_engine();
+        let err = sync_with_backend(
+            &db,
+            &event_bus,
+            &mock,
+            &filter_engine,
+            Some("acct-fail".to_string()),
+        )
+        .await
+        .expect_err("sync fails");
         assert!(err.to_string().contains("mail backend error"));
 
         assert_eq!(
@@ -553,9 +765,16 @@ mod tests {
             .set_secret(&config.keyring_secret_key, "irrelevant-jmap-token")
             .expect("store credential in mock vault");
 
-        let synced = run_account_sync(&db, &secrets, &event_bus, "acct-jmap-real-1")
-            .await
-            .expect("sync succeeds");
+        let filter_engine = empty_filter_engine();
+        let synced = run_account_sync(
+            &db,
+            &secrets,
+            &event_bus,
+            &filter_engine,
+            "acct-jmap-real-1",
+        )
+        .await
+        .expect("sync succeeds");
         // The wiremock stubs advertise 1 folder containing 1 message.
         assert_eq!(synced, 1);
 
@@ -589,9 +808,16 @@ mod tests {
         let mut events = event_bus.subscribe_events();
         let secrets = SecretManager::mock();
 
-        let err = run_account_sync(&db, &secrets, &event_bus, "acct-does-not-exist")
-            .await
-            .expect_err("unknown account fails");
+        let filter_engine = empty_filter_engine();
+        let err = run_account_sync(
+            &db,
+            &secrets,
+            &event_bus,
+            &filter_engine,
+            "acct-does-not-exist",
+        )
+        .await
+        .expect_err("unknown account fails");
         assert!(matches!(err, SyncError::AccountNotFound(_)));
 
         match events.recv().await.expect("error event") {
@@ -617,7 +843,8 @@ mod tests {
         db.save_account(&config).await.expect("save account");
         // Deliberately never calls `secrets.set_secret(...)`.
 
-        let err = run_account_sync(&db, &secrets, &event_bus, "acct-no-cred-1")
+        let filter_engine = empty_filter_engine();
+        let err = run_account_sync(&db, &secrets, &event_bus, &filter_engine, "acct-no-cred-1")
             .await
             .expect_err("missing credential fails");
         assert!(matches!(err, SyncError::Vault(_)));
@@ -640,7 +867,8 @@ mod tests {
         let mut events = event_bus.subscribe_events();
         let secrets = SecretManager::mock();
 
-        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        let filter_engine = empty_filter_engine();
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await;
         assert_eq!(total, 0);
 
         assert_eq!(
@@ -677,7 +905,8 @@ mod tests {
             .set_secret(&config_b.keyring_secret_key, "token-b")
             .expect("store credential b");
 
-        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        let filter_engine = empty_filter_engine();
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await;
         // 2 accounts * 1 folder * 1 message each = 2 processed messages.
         assert_eq!(total, 2);
         assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
@@ -696,7 +925,8 @@ mod tests {
         db.save_account(&config).await.expect("save account");
         // No credential stored in the vault for this account.
 
-        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        let filter_engine = empty_filter_engine();
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await;
         assert_eq!(total, 0);
 
         assert_eq!(
@@ -714,4 +944,261 @@ mod tests {
             CoreEvent::SyncCompleted { account_id: None }
         );
     }
+
+    #[tokio::test]
+    async fn sync_with_backend_fires_immediate_mark_read_action_on_match() {
+        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_email("m-urgent", "inbox", "Urgent: server down"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 1);
+
+        let stored = db.get_message("m-urgent").await.expect("message persisted");
+        assert!(
+            stored.read,
+            "MARK READ action must genuinely flip the stored read flag"
+        );
+
+        let logs = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs");
+        assert_eq!(logs.len(), 1);
+        assert_eq!(logs[0].message_id, "m-urgent");
+
+        let pending = db
+            .list_pending_mutations(10)
+            .await
+            .expect("list pending mutations");
+        assert!(
+            pending.is_empty(),
+            "an immediate action must not enqueue an outbox mutation"
+        );
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_enqueues_outbox_mutation_for_remote_action_on_match() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+
+        let rule = nuncio_filter::NsqlParser::parse_rule(
+            "Archive Rule",
+            1,
+            "WHERE subject CONTAINS 'Archive Me' ACTION MOVE TO 'Archive'",
+        )
+        .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_email("m-archive", "inbox", "Please Archive Me"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 1);
+
+        let pending = db
+            .list_pending_mutations(10)
+            .await
+            .expect("list pending mutations");
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].message_id, "m-archive");
+        assert_eq!(pending[0].mutation_type, "MOVE");
+        assert_eq!(pending[0].status, "pending");
+        assert!(pending[0].payload.contains("Archive"));
+
+        let logs = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs");
+        assert_eq!(logs.len(), 1);
+        assert_eq!(logs[0].message_id, "m-archive");
+
+        // The remote action must not have applied a local read-flag change.
+        let stored = db
+            .get_message("m-archive")
+            .await
+            .expect("message persisted");
+        assert!(!stored.read);
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_non_matching_message_produces_no_filter_side_effects() {
+        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_email("m-plain", "inbox", "Just a normal update"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None)
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 1);
+
+        let stored = db.get_message("m-plain").await.expect("message persisted");
+        assert!(!stored.read, "non-matching message must stay untouched");
+
+        let logs = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs");
+        assert!(logs.is_empty(), "no rule matched, so no log should exist");
+
+        let pending = db
+            .list_pending_mutations(10)
+            .await
+            .expect("list pending mutations");
+        assert!(
+            pending.is_empty(),
+            "no rule matched, so no mutation should be enqueued"
+        );
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_does_not_refire_filters_on_a_repeat_sync_of_the_same_message() {
+        // Sync is not yet incremental: a real backend re-reports messages it
+        // already handed over on every sync pass. This proves a message's
+        // filter actions fire exactly once across repeated syncs rather than
+        // once per sync cycle -- the regression this guard exists for.
+        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 move_rule = nuncio_filter::NsqlParser::parse_rule(
+            "Archive Rule",
+            2,
+            "WHERE subject CONTAINS 'Urgent' ACTION MOVE TO 'Archive'",
+        )
+        .expect("parse rule");
+        let filter_engine =
+            FilterEngine::new(vec![mark_read_rule, move_rule]).expect("compile rules");
+
+        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_email("m-repeat", "inbox", "Urgent: server down"));
+
+        // 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)
+            .await
+            .expect("first sync succeeds");
+        assert_eq!(synced_first, 1);
+
+        let stored_after_first = db
+            .get_message("m-repeat")
+            .await
+            .expect("message persisted after first sync");
+        assert!(stored_after_first.read);
+
+        let logs_after_first = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs after first sync");
+        assert_eq!(logs_after_first.len(), 2, "both rules should fire once");
+
+        let pending_after_first = db
+            .list_pending_mutations(10)
+            .await
+            .expect("list pending mutations after first sync");
+        assert_eq!(pending_after_first.len(), 1);
+
+        // Manually flip the message back to unread, exactly as a user
+        // reading and then un-reading it would -- proves the second sync
+        // doesn't re-flip it back to read via a repeat MARK READ fire.
+        db.set_message_read("m-repeat", false)
+            .await
+            .expect("manually mark unread");
+
+        // 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)
+            .await
+            .expect("second sync succeeds");
+        assert_eq!(synced_second, 1);
+
+        let stored_after_second = db
+            .get_message("m-repeat")
+            .await
+            .expect("message persisted after second sync");
+        assert!(
+            !stored_after_second.read,
+            "a re-sync of an already-seen message must not re-fire MARK READ"
+        );
+
+        let logs_after_second = db
+            .list_filter_execution_logs(10)
+            .await
+            .expect("list execution logs after second sync");
+        assert_eq!(
+            logs_after_second.len(),
+            2,
+            "no new execution log entries should be written on a repeat sync"
+        );
+
+        let pending_after_second = db
+            .list_pending_mutations(10)
+            .await
+            .expect("list pending mutations after second sync");
+        assert_eq!(
+            pending_after_second.len(),
+            1,
+            "no new outbox mutation should be enqueued on a repeat sync"
+        );
+    }
 }