crates/nuncio-mail/src/backend.rs

@@ -5,6 +5,60 @@ use nuncio_core::model::{Attachment, Email, Folder};
 
 use crate::parser::MailError;
 
+/// The remote change a [`RemoteMutationSpec`] applies to one addressed message.
+///
+/// This is the closed vocabulary of genuinely-remote filter actions that the
+/// outbox drains against a real server. `MARK READ`/`MARK UNREAD` are applied
+/// locally to the store when a filter matches, so they are deliberately NOT
+/// represented here -- only actions that mutate server-side state are.
+/// `FORWARD` and `CALL WEBHOOK` are also excluded: they are not backend
+/// mutations (one composes an outbound message, the other calls an HTTP
+/// endpoint), so they are executed through the send/webhook seams instead.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RemoteMutationKind {
+    /// Add (`value == true`) or remove (`value == false`) the "flagged"/
+    /// starred keyword: IMAP `\Flagged`, JMAP `$flagged`. Covers `FLAG`/`UNFLAG`.
+    SetFlagged {
+        /// Whether the message should end up flagged.
+        value: bool,
+    },
+    /// Move the message from its source folder to `to_folder`.
+    Move {
+        /// Destination folder (IMAP mailbox name / JMAP mailbox id).
+        to_folder: String,
+    },
+    /// Copy the message into `to_folder`, leaving the source copy in place.
+    Copy {
+        /// Destination folder (IMAP mailbox name / JMAP mailbox id).
+        to_folder: String,
+    },
+    /// Delete the message (IMAP `\Deleted` + `EXPUNGE`; JMAP `destroy`).
+    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.
+///
+/// `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.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RemoteMutationSpec {
+    /// Protocol-native message id (IMAP `imap-uid-{uid}`, JMAP object id).
+    pub message_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 change to apply.
+    pub kind: RemoteMutationKind,
+}
+
 /// Protocol-agnostic mail backend engine trait implemented by JMAP and IMAP engines.
 #[async_trait]
 pub trait MailBackend: Send + Sync {
@@ -18,6 +72,13 @@ pub trait MailBackend: Send + Sync {
         folder_id: &str,
         since_state: Option<&str>,
     ) -> Result<(Vec<Email>, String), MailError>;
+
+    /// Apply a remote mutation (move/copy/flag/delete) to a single message
+    /// against the real server. MUST return `Ok(())` only when the server
+    /// genuinely applied the change -- implementations must never fabricate
+    /// success, and must refuse to act (returning an error) when the target
+    /// message cannot be safely identified (e.g. an IMAP UIDVALIDITY mismatch).
+    async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError>;
 }
 
 /// A composed outbound email message ready to send over SMTP. Deliberately

crates/nuncio-mail/src/imap.rs

@@ -31,9 +31,36 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
 /// honest timeout rather than hanging the sync indefinitely.
 const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60);
 
-use crate::backend::MailBackend;
+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()?;
+    (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()
+}
+
 /// Delimiter between the two components of a folder sync checkpoint. The
 /// checkpoint state string is `"{uidvalidity}{DELIM}{uidnext}"` (e.g.
 /// `"42:105"`): the mailbox's UIDVALIDITY at the time it was written, then the
@@ -759,6 +786,222 @@ impl ImapEngine {
         result
     }
 
+    /// Apply a remote mutation over an active IMAP session, selecting the
+    /// source folder and verifying its UIDVALIDITY against the stored
+    /// checkpoint BEFORE issuing any command that could touch the wrong
+    /// message.
+    ///
+    /// The UIDVALIDITY guard (RFC 3501 s2.3.1.1) is the safety-critical step: a
+    /// stored IMAP UID is only meaningful under the UIDVALIDITY it was captured
+    /// with. If the mailbox was recreated its UIDVALIDITY changes and UIDs are
+    /// reassigned, so acting on the old UID would mutate a different message. A
+    /// mismatch -- or a stored checkpoint that carries no UIDVALIDITY to
+    /// compare against, or a server that reports none -- is refused with an
+    /// honest [`MailError::ImapError`]; the mutation is never applied blind.
+    ///
+    /// `MOVE` uses `UID MOVE` (RFC 6851) when the server advertises the `MOVE`
+    /// capability, else falls back to the equivalent `UID COPY` + `UID STORE
+    /// +FLAGS (\Deleted)` + UID-scoped `UID EXPUNGE`. `DELETE` is `UID STORE
+    /// +FLAGS (\Deleted)` + `UID EXPUNGE`. The expunge is always scoped to the
+    /// target UID via RFC 4315 (never a mailbox-wide `EXPUNGE`, which would
+    /// destroy other `\Deleted` messages) and requires `UIDPLUS`, failing
+    /// closed otherwise. Each command's tagged result is checked, so only a
+    /// genuine server success returns `Ok(())`.
+    pub async fn apply_mutation_with_session<S>(
+        &self,
+        spec: &RemoteMutationSpec,
+        session: &mut async_imap::Session<S>,
+    ) -> Result<(), MailError>
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
+    {
+        let uid = parse_imap_uid(&spec.message_id).ok_or_else(|| {
+            MailError::ImapError(format!(
+                "cannot parse an IMAP UID from message id '{}'",
+                spec.message_id
+            ))
+        })?;
+
+        let mailbox = session.select(&spec.folder_id).await.map_err(|e| {
+            MailError::ImapError(format!(
+                "failed to select folder '{}': {}",
+                spec.folder_id, e
+            ))
+        })?;
+
+        // 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());
+        match (stored_validity, mailbox.uid_validity) {
+            (Some(stored), Some(current)) if stored == current => {}
+            (stored, current) => {
+                return Err(MailError::ImapError(format!(
+                    "refusing to mutate message '{}' in folder '{}': UIDVALIDITY guard failed \
+                     (stored {stored:?}, current {current:?}); the stored UID may no longer \
+                     address the intended message",
+                    spec.message_id, spec.folder_id
+                )));
+            }
+        }
+
+        let uid_set = uid.to_string();
+        match &spec.kind {
+            RemoteMutationKind::SetFlagged { value } => {
+                let op = if *value { "+FLAGS" } else { "-FLAGS" };
+                self.uid_store_flags(session, &uid_set, op, "\\Flagged")
+                    .await
+            }
+            RemoteMutationKind::Copy { to_folder } => {
+                session.uid_copy(&uid_set, to_folder).await.map_err(|e| {
+                    MailError::ImapError(format!(
+                        "UID COPY of '{}' to '{}' failed: {}",
+                        spec.message_id, to_folder, e
+                    ))
+                })
+            }
+            RemoteMutationKind::Move { to_folder } => {
+                self.uid_move(session, &uid_set, to_folder, &spec.message_id)
+                    .await
+            }
+            RemoteMutationKind::Delete => {
+                self.uid_store_flags(session, &uid_set, "+FLAGS", "\\Deleted")
+                    .await?;
+                self.uid_expunge_scoped(session, &uid_set).await
+            }
+        }
+    }
+
+    /// Issue a `UID STORE {uid} {op} ({flag})` and drain its untagged `FETCH`
+    /// responses, surfacing any protocol error. `op` is `+FLAGS`/`-FLAGS`.
+    async fn uid_store_flags<S>(
+        &self,
+        session: &mut async_imap::Session<S>,
+        uid_set: &str,
+        op: &str,
+        flag: &str,
+    ) -> Result<(), MailError>
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
+    {
+        let query = format!("{op} ({flag})");
+        let updates = session.uid_store(uid_set, &query).await.map_err(|e| {
+            MailError::ImapError(format!("UID STORE {query} for uid {uid_set} failed: {e}"))
+        })?;
+        tokio::pin!(updates);
+        while let Some(item) = updates.next().await {
+            item.map_err(|e| {
+                MailError::ImapError(format!("reading UID STORE response failed: {e}"))
+            })?;
+        }
+        Ok(())
+    }
+
+    /// Permanently remove ONLY the target `uid_set` from the selected folder
+    /// via RFC 4315 `UID EXPUNGE`, draining its untagged `EXPUNGE` responses.
+    ///
+    /// A bare `EXPUNGE` (RFC 3501) removes EVERY `\Deleted` message in the
+    /// mailbox, so it would collaterally destroy any other message a different
+    /// client had already flagged `\Deleted`. `UID EXPUNGE` is scoped to
+    /// exactly the given UIDs, so it is the only safe expunge in a shared
+    /// mailbox. It requires the `UIDPLUS` capability; if the server does not
+    /// advertise it, this fails closed with an honest error rather than
+    /// falling back to the destructive mailbox-wide `EXPUNGE`.
+    async fn uid_expunge_scoped<S>(
+        &self,
+        session: &mut async_imap::Session<S>,
+        uid_set: &str,
+    ) -> Result<(), MailError>
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
+    {
+        let caps = session
+            .capabilities()
+            .await
+            .map_err(|e| MailError::ImapError(format!("CAPABILITY query failed: {e}")))?;
+        let has_uidplus = caps.has_str("UIDPLUS");
+        drop(caps);
+        if !has_uidplus {
+            return Err(MailError::ImapError(format!(
+                "refusing to expunge uid {uid_set}: server does not advertise UIDPLUS, so a \
+                 UID-scoped EXPUNGE (RFC 4315) is unavailable; a bare EXPUNGE could destroy \
+                 other \\Deleted messages in the folder"
+            )));
+        }
+
+        let expunged = session
+            .uid_expunge(uid_set)
+            .await
+            .map_err(|e| MailError::ImapError(format!("UID EXPUNGE {uid_set} failed: {e}")))?;
+        tokio::pin!(expunged);
+        while let Some(item) = expunged.next().await {
+            item.map_err(|e| {
+                MailError::ImapError(format!("reading UID EXPUNGE response failed: {e}"))
+            })?;
+        }
+        Ok(())
+    }
+
+    /// Move a message by UID, preferring RFC 6851 `UID MOVE` and falling back
+    /// to `UID COPY` + `UID STORE +FLAGS (\Deleted)` + UID-scoped `UID EXPUNGE`
+    /// when the server does not advertise the `MOVE` capability. The fallback's
+    /// expunge is scoped to the target UID (never a mailbox-wide `EXPUNGE`) and
+    /// requires `UIDPLUS`, failing closed otherwise.
+    async fn uid_move<S>(
+        &self,
+        session: &mut async_imap::Session<S>,
+        uid_set: &str,
+        to_folder: &str,
+        message_id: &str,
+    ) -> Result<(), MailError>
+    where
+        S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static,
+    {
+        let caps = session
+            .capabilities()
+            .await
+            .map_err(|e| MailError::ImapError(format!("CAPABILITY query failed: {e}")))?;
+        let has_move = caps.has_str("MOVE");
+        drop(caps);
+
+        if has_move {
+            return session.uid_mv(uid_set, to_folder).await.map_err(|e| {
+                MailError::ImapError(format!(
+                    "UID MOVE of '{message_id}' to '{to_folder}' failed: {e}"
+                ))
+            });
+        }
+
+        session.uid_copy(uid_set, to_folder).await.map_err(|e| {
+            MailError::ImapError(format!(
+                "UID COPY (move fallback) of '{message_id}' to '{to_folder}' failed: {e}"
+            ))
+        })?;
+        self.uid_store_flags(session, uid_set, "+FLAGS", "\\Deleted")
+            .await?;
+        self.uid_expunge_scoped(session, uid_set).await
+    }
+
+    /// Apply a remote mutation over a freshly authenticated session. See
+    /// [`Self::apply_mutation_with_session`] for the UIDVALIDITY guard and the
+    /// per-action protocol commands.
+    pub async fn apply_remote_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> {
+        let (username, password) = match (&self.username, &self.password) {
+            (Some(u), Some(p)) => (u.as_str(), p.as_str()),
+            _ => {
+                return Err(MailError::AuthError(
+                    "IMAP remote mutation requires credentials".to_string(),
+                ))
+            }
+        };
+        let mut session = self
+            .socket_manager
+            .connect_session(username, password)
+            .await?;
+        let result = self.apply_mutation_with_session(spec, &mut session).await;
+        let _ = session.logout().await;
+        result
+    }
+
     /// Listen for real-time IMAP IDLE notification events on an active IMAP session.
     pub async fn listen_idle_session<S, F>(
         &self,
@@ -900,6 +1143,10 @@ impl MailBackend for ImapEngine {
     ) -> Result<(Vec<Email>, String), MailError> {
         self.sync_folder_messages(folder_id, since_state).await
     }
+
+    async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> {
+        self.apply_remote_mutation(spec).await
+    }
 }
 
 #[cfg(test)]
@@ -1327,6 +1574,346 @@ mod tests {
         server.abort();
     }
 
+    #[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);
+        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);
+    }
+
+    /// Drive a mutation over a scripted IMAP server (in-memory duplex),
+    /// capturing every command line the client issued so the test can assert
+    /// on the exact protocol commands. `advertise_move`/`advertise_uidplus`
+    /// control whether the scripted `CAPABILITY` response includes `MOVE`/
+    /// `UIDPLUS`; `select_uid_validity` is the UIDVALIDITY the scripted
+    /// `SELECT` reports.
+    async fn run_scripted_mutation(
+        spec: &RemoteMutationSpec,
+        advertise_move: bool,
+        advertise_uidplus: bool,
+        select_uid_validity: u32,
+    ) -> (Result<(), MailError>, Vec<String>) {
+        let (client_io, server_io) = tokio::io::duplex(8192);
+        let commands = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
+        let server_commands = commands.clone();
+
+        let server = tokio::spawn(async move {
+            let mut io = server_io;
+            while let Some(line) = read_scripted_line(&mut io).await {
+                if let Ok(mut g) = server_commands.lock() {
+                    g.push(line.clone());
+                }
+                let tag = line.split_whitespace().next().unwrap_or("").to_string();
+                let upper = line.to_ascii_uppercase();
+                if upper.contains("LOGIN") {
+                    let _ = io
+                        .write_all(format!("{tag} OK Logged in\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("SELECT") {
+                    let resp = format!(
+                        "* FLAGS (\\Seen \\Flagged \\Deleted)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\
+                         * OK [UIDVALIDITY {select_uid_validity}] ok\r\n* OK [UIDNEXT 105] ok\r\n\
+                         {tag} OK [READ-WRITE] SELECT done\r\n"
+                    );
+                    let _ = io.write_all(resp.as_bytes()).await;
+                } else if upper.contains("CAPABILITY") {
+                    let mut caps = String::from("IMAP4rev1");
+                    if advertise_move {
+                        caps.push_str(" MOVE");
+                    }
+                    if advertise_uidplus {
+                        caps.push_str(" UIDPLUS");
+                    }
+                    let _ = io
+                        .write_all(
+                            format!("* CAPABILITY {caps}\r\n{tag} OK CAPABILITY done\r\n")
+                                .as_bytes(),
+                        )
+                        .await;
+                } else if upper.contains("UID EXPUNGE") {
+                    let _ = io
+                        .write_all(format!("{tag} OK UID EXPUNGE completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("UID STORE") {
+                    let _ = io
+                        .write_all(format!("{tag} OK STORE completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("UID COPY") {
+                    let _ = io
+                        .write_all(format!("{tag} OK COPY completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("UID MOVE") {
+                    let _ = io
+                        .write_all(format!("{tag} OK MOVE completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("EXPUNGE") {
+                    let _ = io
+                        .write_all(format!("{tag} OK EXPUNGE completed\r\n").as_bytes())
+                        .await;
+                } else if upper.contains("LOGOUT") {
+                    let _ = io
+                        .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes())
+                        .await;
+                    break;
+                }
+            }
+        });
+
+        let client = async_imap::Client::new(client_io);
+        let mut session = client
+            .login("user", "pass")
+            .await
+            .map_err(|(e, _)| e)
+            .expect("scripted login succeeds");
+        let engine = ImapEngine::new("acct-1", "example.test", 993);
+        let result = engine.apply_mutation_with_session(spec, &mut session).await;
+        drop(session);
+        let _ = server.await;
+
+        let captured = commands.lock().expect("lock captured commands").clone();
+        (result, captured)
+    }
+
+    /// True if `line` is a bare mailbox-wide `EXPUNGE` (RFC 3501), NOT a
+    /// UID-scoped `UID EXPUNGE` (RFC 4315). Lets a test prove the destructive
+    /// mailbox-wide form is never issued (a plain `contains("EXPUNGE")` would
+    /// be satisfied by the safe `UID EXPUNGE` too).
+    fn is_bare_expunge(line: &str) -> bool {
+        let u = line.to_ascii_uppercase();
+        u.contains("EXPUNGE") && !u.contains("UID EXPUNGE")
+    }
+
+    fn imap_spec(kind: RemoteMutationKind, checkpoint: Option<&str>) -> RemoteMutationSpec {
+        RemoteMutationSpec {
+            message_id: "imap-uid-42".to_string(),
+            folder_id: "INBOX".to_string(),
+            folder_checkpoint: checkpoint.map(str::to_string),
+            kind,
+        }
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_flag_issues_uid_store_add_flags() {
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(
+                RemoteMutationKind::SetFlagged { value: true },
+                Some("1:105"),
+            ),
+            false,
+            true,
+            1,
+        )
+        .await;
+        result.expect("flag mutation succeeds");
+        assert!(
+            commands
+                .iter()
+                .any(|c| c.contains("UID STORE 42 +FLAGS (\\Flagged)")),
+            "expected a UID STORE +FLAGS (\\Flagged), got: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_move_uses_uid_move_when_server_advertises_move() {
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(
+                RemoteMutationKind::Move {
+                    to_folder: "Archive".to_string(),
+                },
+                Some("1:105"),
+            ),
+            true,
+            true,
+            1,
+        )
+        .await;
+        result.expect("move mutation succeeds");
+        assert!(
+            commands.iter().any(|c| c.contains("UID MOVE 42")),
+            "expected a UID MOVE, got: {commands:?}"
+        );
+        assert!(
+            !commands.iter().any(|c| c.contains("UID COPY")),
+            "must not fall back to COPY when MOVE is advertised: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_move_falls_back_to_copy_store_uid_expunge_without_move_capability()
+    {
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(
+                RemoteMutationKind::Move {
+                    to_folder: "Archive".to_string(),
+                },
+                Some("1:105"),
+            ),
+            false,
+            true,
+            1,
+        )
+        .await;
+        result.expect("move fallback succeeds");
+        assert!(commands.iter().any(|c| c.contains("UID COPY 42")));
+        assert!(commands
+            .iter()
+            .any(|c| c.contains("UID STORE 42 +FLAGS (\\Deleted)")));
+        // The expunge MUST be scoped to the target UID, never a mailbox-wide
+        // bare EXPUNGE that could destroy other \Deleted messages.
+        assert!(
+            commands.iter().any(|c| c.contains("UID EXPUNGE 42")),
+            "fallback must issue a UID-scoped UID EXPUNGE 42, got: {commands:?}"
+        );
+        assert!(
+            !commands.iter().any(|c| is_bare_expunge(c)),
+            "fallback must NOT issue a bare mailbox-wide EXPUNGE, got: {commands:?}"
+        );
+        assert!(
+            !commands.iter().any(|c| c.contains("UID MOVE")),
+            "must not issue UID MOVE without the capability: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_delete_stores_deleted_and_uid_expunges_the_target_only() {
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(RemoteMutationKind::Delete, Some("1:105")),
+            false,
+            true,
+            1,
+        )
+        .await;
+        result.expect("delete mutation succeeds");
+        assert!(commands
+            .iter()
+            .any(|c| c.contains("UID STORE 42 +FLAGS (\\Deleted)")));
+        // The expunge MUST target exactly the deleted UID (RFC 4315), never a
+        // mailbox-wide bare EXPUNGE that would also destroy other client's
+        // \Deleted messages in the same folder.
+        assert!(
+            commands.iter().any(|c| c.contains("UID EXPUNGE 42")),
+            "delete must issue a UID-scoped UID EXPUNGE 42, got: {commands:?}"
+        );
+        assert!(
+            !commands.iter().any(|c| is_bare_expunge(c)),
+            "delete must NOT issue a bare mailbox-wide EXPUNGE, got: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_delete_fails_closed_without_uidplus_and_issues_no_bare_expunge() {
+        // Without UIDPLUS there is no UID-scoped expunge; the op MUST fail
+        // closed rather than fall back to a destructive mailbox-wide EXPUNGE.
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(RemoteMutationKind::Delete, Some("1:105")),
+            false,
+            false,
+            1,
+        )
+        .await;
+        assert!(
+            matches!(result, Err(MailError::ImapError(_))),
+            "delete without UIDPLUS must fail closed, got: {result:?}"
+        );
+        assert!(
+            !commands
+                .iter()
+                .any(|c| c.to_ascii_uppercase().contains("EXPUNGE")),
+            "no expunge of any kind may be issued when UIDPLUS is unavailable, got: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_move_fallback_fails_closed_without_uidplus() {
+        // The no-MOVE-capability move fallback also relies on a UID-scoped
+        // expunge; without UIDPLUS it must fail closed and issue no bare
+        // EXPUNGE (the preceding COPY/STORE are non-destructive to others).
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(
+                RemoteMutationKind::Move {
+                    to_folder: "Archive".to_string(),
+                },
+                Some("1:105"),
+            ),
+            false,
+            false,
+            1,
+        )
+        .await;
+        assert!(
+            matches!(result, Err(MailError::ImapError(_))),
+            "move fallback without UIDPLUS must fail closed, got: {result:?}"
+        );
+        assert!(
+            !commands
+                .iter()
+                .any(|c| c.to_ascii_uppercase().contains("EXPUNGE")),
+            "no expunge of any kind may be issued when UIDPLUS is unavailable, got: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_refuses_to_act_on_a_uidvalidity_mismatch() {
+        // The stored checkpoint captured the UID under UIDVALIDITY 1, but the
+        // mailbox now reports 2 (renumbered). The mutation MUST be refused and
+        // NO mutating command (STORE/COPY/MOVE/EXPUNGE) may be issued.
+        let (result, commands) = run_scripted_mutation(
+            &imap_spec(RemoteMutationKind::Delete, Some("1:105")),
+            false,
+            true,
+            2,
+        )
+        .await;
+        let err = result.expect_err("a UIDVALIDITY mismatch must be refused");
+        assert!(matches!(err, MailError::ImapError(_)));
+        assert!(
+            !commands.iter().any(|c| {
+                let u = c.to_ascii_uppercase();
+                u.contains("UID STORE")
+                    || u.contains("UID COPY")
+                    || u.contains("UID MOVE")
+                    || u.contains("EXPUNGE")
+            }),
+            "no mutating command may be issued on a UIDVALIDITY mismatch, got: {commands:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn imap_apply_mutation_refuses_when_no_checkpoint_proves_uidvalidity() {
+        // With no stored checkpoint there is no UIDVALIDITY to prove the UID
+        // still addresses the intended message, so the op is refused.
+        let (result, _commands) =
+            run_scripted_mutation(&imap_spec(RemoteMutationKind::Delete, None), false, true, 1)
+                .await;
+        assert!(matches!(
+            result.expect_err("missing checkpoint must be refused"),
+            MailError::ImapError(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn imap_apply_remote_mutation_requires_credentials() {
+        let engine = ImapEngine::new("acct-1", "example.test", 993);
+        let err = engine
+            .apply_remote_mutation(&imap_spec(RemoteMutationKind::Delete, Some("1:105")))
+            .await
+            .expect_err("credential-less mutation must fail");
+        assert!(matches!(err, MailError::AuthError(_)));
+    }
+
     #[tokio::test]
     async fn with_tls_mode_records_mode_and_resolves_default_ports() {
         // A `0` port resolves to the conventional default for each mode.

crates/nuncio-mail/src/jmap.rs

@@ -5,7 +5,7 @@ use nuncio_core::model::{Email, Folder};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 
-use crate::backend::MailBackend;
+use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec};
 use crate::parser::MailError;
 
 /// JMAP Session Object (RFC 8620 Section 2).
@@ -203,6 +203,100 @@ impl JmapEngine {
         })
     }
 
+    /// Build a JSON-RPC `Email/set` request applying `spec` to a single
+    /// message (RFC 8621 s4.6). A `Move`/`Copy`/`SetFlagged` becomes an
+    /// `update` patch keyed by the JMAP object id; a `Delete` becomes a
+    /// `destroy`.
+    ///
+    /// `Move` replaces the message's mailbox set with the destination (removing
+    /// the source), whereas `Copy` adds the destination via a `mailboxIds/{id}`
+    /// patch, leaving the existing membership intact. Flag toggles patch the
+    /// `$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;
+        let (update, destroy) = match &spec.kind {
+            RemoteMutationKind::SetFlagged { value } => {
+                let keyword = if *value { json!(true) } else { Value::Null };
+                (json!({ id: { "keywords/$flagged": keyword } }), json!([]))
+            }
+            RemoteMutationKind::Move { to_folder } => (
+                json!({ id: { "mailboxIds": { to_folder: true } } }),
+                json!([]),
+            ),
+            RemoteMutationKind::Copy { to_folder } => (
+                json!({ id: { format!("mailboxIds/{to_folder}"): true } }),
+                json!([]),
+            ),
+            RemoteMutationKind::Delete => (json!({}), json!([id])),
+        };
+
+        json!({
+            "using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"],
+            "methodCalls": [
+                [
+                    "Email/set",
+                    {
+                        "accountId": account_id,
+                        "update": update,
+                        "destroy": destroy
+                    },
+                    "c1"
+                ]
+            ]
+        })
+    }
+
+    /// Confirm a JMAP `Email/set` response genuinely applied the mutation to
+    /// `message_id`: the id must appear in `updated`/`destroyed` and must NOT
+    /// appear in `notUpdated`/`notDestroyed`. Anything else -- a server-side
+    /// rejection, or an id the server silently ignored -- surfaces as an honest
+    /// error so the outbox never marks a rejected mutation completed.
+    pub fn confirm_email_set_applied(
+        raw_json: &str,
+        message_id: &str,
+        expect_destroy: bool,
+    ) -> Result<(), MailError> {
+        let val: Value = serde_json::from_str(raw_json)
+            .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?;
+        let payload = Self::extract_method_response_payload(&val)?;
+
+        // A rejection is reported in notUpdated/notDestroyed keyed by id; treat
+        // its presence as the authoritative failure signal.
+        let rejected = payload
+            .get(if expect_destroy {
+                "notDestroyed"
+            } else {
+                "notUpdated"
+            })
+            .and_then(|v| v.as_object())
+            .is_some_and(|m| m.contains_key(message_id));
+        if rejected {
+            return Err(MailError::OperationFailed(format!(
+                "JMAP Email/set rejected mutation for message '{message_id}'"
+            )));
+        }
+
+        let applied = if expect_destroy {
+            payload
+                .get("destroyed")
+                .and_then(|v| v.as_array())
+                .is_some_and(|ids| ids.iter().any(|v| v.as_str() == Some(message_id)))
+        } else {
+            payload
+                .get("updated")
+                .and_then(|v| v.as_object())
+                .is_some_and(|m| m.contains_key(message_id))
+        };
+        if applied {
+            Ok(())
+        } else {
+            Err(MailError::OperationFailed(format!(
+                "JMAP Email/set did not confirm the mutation for message '{message_id}'"
+            )))
+        }
+    }
+
     /// Build JSON-RPC request invocation for `Email/changes` (differential sync).
     pub fn build_email_changes_request(account_id: &str, since_state: &str) -> Value {
         json!({
@@ -451,6 +545,21 @@ impl MailBackend for JmapEngine {
         let get_raw = self.post_jmap(&session.api_url, &get_request).await?;
         self.parse_email_get_response(&get_raw)
     }
+
+    async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> {
+        if !self.has_credentials() {
+            return Err(MailError::AuthError(
+                "JMAP remote mutation requires credentials".to_string(),
+            ));
+        }
+        let session = self.discover_session().await?;
+        let account_id = self.resolve_account_id(&session);
+
+        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)
+    }
 }
 
 #[cfg(test)]
@@ -578,6 +687,49 @@ mod tests {
         assert_eq!(email.body_snippet, Some("hello world".to_string()));
     }
 
+    fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec {
+        RemoteMutationSpec {
+            message_id: "m-1".to_string(),
+            folder_id: "mb-inbox".to_string(),
+            folder_checkpoint: None,
+            kind,
+        }
+    }
+
+    #[test]
+    fn build_email_set_request_copy_adds_target_without_removing_source() {
+        let req = JmapEngine::build_email_set_request(
+            "acct-1",
+            &spec(RemoteMutationKind::Copy {
+                to_folder: "mb-archive".to_string(),
+            }),
+        );
+        let call = &req["methodCalls"][0];
+        assert_eq!(call[0], "Email/set");
+        // A copy patches a single mailboxIds/{id} pointer to true, leaving the
+        // rest of the message's mailbox membership untouched.
+        assert_eq!(call[1]["update"]["m-1"]["mailboxIds/mb-archive"], true);
+    }
+
+    #[test]
+    fn build_email_set_request_unflag_clears_keyword_with_null() {
+        let req = JmapEngine::build_email_set_request(
+            "acct-1",
+            &spec(RemoteMutationKind::SetFlagged { value: false }),
+        );
+        assert!(req["methodCalls"][0][1]["update"]["m-1"]["keywords/$flagged"].is_null());
+    }
+
+    #[test]
+    fn confirm_email_set_applied_requires_the_id_in_the_updated_map() {
+        let ok = r#"{"methodResponses":[["Email/set",{"updated":{"m-1":null}},"c1"]]}"#;
+        assert!(JmapEngine::confirm_email_set_applied(ok, "m-1", false).is_ok());
+
+        // An empty updated map is NOT success -- the server silently ignored it.
+        let ignored = r#"{"methodResponses":[["Email/set",{"updated":{}},"c1"]]}"#;
+        assert!(JmapEngine::confirm_email_set_applied(ignored, "m-1", false).is_err());
+    }
+
     #[tokio::test]
     async fn uncredentialed_engine_requires_credentials_to_sync() {
         // Without credentials there is no server to discover a session from;
@@ -597,5 +749,11 @@ mod tests {
             .await
             .expect_err("credential-less sync_messages must fail");
         assert!(matches!(err, MailError::AuthError(_)));
+
+        let err = engine
+            .apply_mutation(&spec(RemoteMutationKind::Delete))
+            .await
+            .expect_err("credential-less apply_mutation must fail");
+        assert!(matches!(err, MailError::AuthError(_)));
     }
 }

crates/nuncio-mail/src/lib.rs

@@ -10,7 +10,9 @@ pub mod mock;
 pub mod parser;
 pub mod smtp;
 
-pub use backend::{MailBackend, MessageSender, OutboundMessage};
+pub use backend::{
+    MailBackend, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec,
+};
 pub use imap::{IdleSocketState, ImapDualSocketManager, ImapEngine};
 pub use jmap::JmapEngine;
 pub use mock::{MockMailBackend, MockMessageSender};

crates/nuncio-mail/src/mock.rs

@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use nuncio_core::model::{Email, Folder};
 use std::sync::{Arc, Mutex};
 
-use crate::backend::{MailBackend, MessageSender, OutboundMessage};
+use crate::backend::{MailBackend, MessageSender, OutboundMessage, RemoteMutationSpec};
 use crate::parser::MailError;
 
 /// Thread-safe mock mail backend for offline testing.
@@ -19,6 +19,10 @@ pub struct MockMailBackend {
     since_state_calls: Arc<Mutex<Vec<Option<String>>>>,
     /// The checkpoint token this mock returns from `sync_messages`.
     returned_state: Arc<Mutex<String>>,
+    /// Every [`RemoteMutationSpec`] passed to `apply_mutation`, in call order,
+    /// so a daemon E2E can prove the outbox executor genuinely invoked the
+    /// backend op (not a fabricated completion).
+    applied_mutations: Arc<Mutex<Vec<RemoteMutationSpec>>>,
 }
 
 impl MockMailBackend {
@@ -70,6 +74,17 @@ impl MockMailBackend {
             guard.push(email);
         }
     }
+
+    /// Every [`RemoteMutationSpec`] this mock's `apply_mutation` was called
+    /// with, in call order. A failing configuration (`set_should_fail(true)`)
+    /// records nothing, so a test can prove a failed op is never counted as
+    /// applied.
+    pub fn applied_mutations(&self) -> Vec<RemoteMutationSpec> {
+        self.applied_mutations
+            .lock()
+            .map(|guard| guard.clone())
+            .unwrap_or_default()
+    }
 }
 
 #[async_trait]
@@ -135,6 +150,26 @@ impl MailBackend for MockMailBackend {
             .collect();
         Ok((matches, returned_state))
     }
+
+    async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> {
+        let should_fail = self
+            .should_fail
+            .lock()
+            .map_err(|e| MailError::ParseFailed(e.to_string()))?;
+        if *should_fail {
+            return Err(MailError::ImapError(
+                "simulated remote mutation failure".to_string(),
+            ));
+        }
+        // Record only genuinely-applied mutations, so a test can distinguish a
+        // real op from a fabricated completion.
+        let mut applied = self
+            .applied_mutations
+            .lock()
+            .map_err(|e| MailError::ParseFailed(e.to_string()))?;
+        applied.push(spec.clone());
+        Ok(())
+    }
 }
 
 /// Deterministic mock [`MessageSender`] for offline testing of the outbound

crates/nuncio-mail/tests/wiremock_jmap_test.rs

@@ -2,9 +2,9 @@
 
 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 
-use nuncio_mail::{JmapEngine, MailBackend};
+use nuncio_mail::{JmapEngine, MailBackend, RemoteMutationKind, RemoteMutationSpec};
 use serde_json::json;
-use wiremock::matchers::{method, path};
+use wiremock::matchers::{body_partial_json, method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
 #[tokio::test]
@@ -237,3 +237,144 @@ async fn jmap_engine_sync_folders_surfaces_session_discovery_failure() {
         .expect_err("session discovery failure must not fall back to canned folders");
     assert!(err.to_string().contains("network I/O error"));
 }
+
+/// Mount JMAP session discovery so `apply_mutation` can reach the mock `/jmap/api`.
+async fn mount_session(mock_server: &MockServer) {
+    let session_body = json!({
+        "username": "james.maes@kof22.com",
+        "primaryAccounts": { "urn:ietf:params:jmap:mail": "acct-100" },
+        "apiUrl": format!("{}/jmap/api", mock_server.uri()),
+        "state": "session-state-1"
+    });
+    Mock::given(method("GET"))
+        .and(path("/.well-known/jmap"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(&session_body))
+        .mount(mock_server)
+        .await;
+}
+
+fn jmap_engine(mock_server: &MockServer) -> JmapEngine {
+    JmapEngine::with_credentials(
+        "acct-100",
+        &mock_server.uri(),
+        "james.maes@kof22.com",
+        "wiremock-password",
+    )
+}
+
+fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec {
+    RemoteMutationSpec {
+        message_id: "jmap-msg-1".to_string(),
+        folder_id: "mb-inbox".to_string(),
+        folder_checkpoint: None,
+        kind,
+    }
+}
+
+#[tokio::test]
+async fn jmap_apply_mutation_flag_issues_keywords_patch_and_confirms_success() {
+    let mock_server = MockServer::start().await;
+    mount_session(&mock_server).await;
+
+    // Assert the exact Email/set update patch reaches the server: a $flagged
+    // keyword set to true on the target id.
+    Mock::given(method("POST"))
+        .and(path("/jmap/api"))
+        .and(body_partial_json(json!({
+            "methodCalls": [[
+                "Email/set",
+                { "update": { "jmap-msg-1": { "keywords/$flagged": true } } },
+                "c1"
+            ]]
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "methodResponses": [["Email/set", {"updated": {"jmap-msg-1": null}}, "c1"]]
+        })))
+        .mount(&mock_server)
+        .await;
+
+    let engine = jmap_engine(&mock_server);
+    engine
+        .apply_mutation(&spec(RemoteMutationKind::SetFlagged { value: true }))
+        .await
+        .expect("flag mutation applied");
+}
+
+#[tokio::test]
+async fn jmap_apply_mutation_move_replaces_mailbox_ids() {
+    let mock_server = MockServer::start().await;
+    mount_session(&mock_server).await;
+
+    Mock::given(method("POST"))
+        .and(path("/jmap/api"))
+        .and(body_partial_json(json!({
+            "methodCalls": [[
+                "Email/set",
+                { "update": { "jmap-msg-1": { "mailboxIds": { "mb-archive": true } } } },
+                "c1"
+            ]]
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "methodResponses": [["Email/set", {"updated": {"jmap-msg-1": null}}, "c1"]]
+        })))
+        .mount(&mock_server)
+        .await;
+
+    let engine = jmap_engine(&mock_server);
+    engine
+        .apply_mutation(&spec(RemoteMutationKind::Move {
+            to_folder: "mb-archive".to_string(),
+        }))
+        .await
+        .expect("move mutation applied");
+}
+
+#[tokio::test]
+async fn jmap_apply_mutation_delete_issues_destroy_and_confirms_success() {
+    let mock_server = MockServer::start().await;
+    mount_session(&mock_server).await;
+
+    Mock::given(method("POST"))
+        .and(path("/jmap/api"))
+        .and(body_partial_json(json!({
+            "methodCalls": [["Email/set", { "destroy": ["jmap-msg-1"] }, "c1"]]
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "methodResponses": [["Email/set", {"destroyed": ["jmap-msg-1"]}, "c1"]]
+        })))
+        .mount(&mock_server)
+        .await;
+
+    let engine = jmap_engine(&mock_server);
+    engine
+        .apply_mutation(&spec(RemoteMutationKind::Delete))
+        .await
+        .expect("delete mutation applied");
+}
+
+/// A server that reports the id under `notUpdated` genuinely rejected the
+/// mutation; `apply_mutation` MUST surface that as an error so the outbox never
+/// marks a rejected op completed.
+#[tokio::test]
+async fn jmap_apply_mutation_surfaces_server_rejection_as_error() {
+    let mock_server = MockServer::start().await;
+    mount_session(&mock_server).await;
+
+    Mock::given(method("POST"))
+        .and(path("/jmap/api"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "methodResponses": [["Email/set", {
+                "updated": {},
+                "notUpdated": { "jmap-msg-1": { "type": "notFound" } }
+            }, "c1"]]
+        })))
+        .mount(&mock_server)
+        .await;
+
+    let engine = jmap_engine(&mock_server);
+    let err = engine
+        .apply_mutation(&spec(RemoteMutationKind::SetFlagged { value: false }))
+        .await
+        .expect_err("a server-rejected mutation must not be reported as success");
+    assert!(err.to_string().contains("jmap-msg-1"));
+}

crates/nuncio-store/src/vault.rs

@@ -27,6 +27,14 @@ pub const LEDGER_KEY_ACCOUNT: &str = "ledger-hmac-key";
 /// and hex-encoded for use as an `authorization: Bearer <token>` header value.
 pub const GRPC_TOKEN_ACCOUNT: &str = "grpc-bearer-token";
 
+/// Keyring account name for the HMAC-SHA256 key that signs outbound filter
+/// `CALL WEBHOOK` payloads (the `X-Nuncio-Signature` header produced by
+/// `nuncio_filter::WebhookDispatcher`). Minted on first use via
+/// [`SecretManager::get_or_create_key_bytes`] and hex-encoded; a single
+/// daemon-wide signing key lets a webhook receiver verify that a delivery
+/// genuinely originated from this daemon.
+pub const WEBHOOK_SIGNING_KEY_ACCOUNT: &str = "webhook-signing-key";
+
 /// Zeroized sensitive string wrapper ensuring heap bytes are wiped on drop.
 #[derive(Debug, Clone, PartialEq, Eq, Zeroize, ZeroizeOnDrop)]
 pub struct ZeroizingSecret(String);

crates/nunciod/src/lib.rs

@@ -6,6 +6,7 @@ pub mod calendar_sync;
 pub mod contacts_sync;
 pub mod grpc;
 pub mod orchestrator;
+pub mod outbox;
 pub mod send;
 pub mod sync;
 pub use orchestrator::SelfHealingSyncOrchestrator;

crates/nunciod/src/main.rs

@@ -3,7 +3,7 @@
 //! filter automation engine, outbox retries, and the multi-client gRPC API.
 
 use nuncio_core::{CoreCommand, CoreEvent, EventBus};
-use nuncio_filter::{FilterEngine, OutboxManager};
+use nuncio_filter::FilterEngine;
 use std::sync::Arc;
 
 #[tokio::main]
@@ -98,38 +98,32 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         );
     }
 
-    // Background Outbox Worker Task
+    // Background Outbox Worker Task. Each tick drains the pending remote
+    // mutations the filter engine enqueued (move/copy/flag/unflag/delete on the
+    // real mail server, forward via SMTP, or a signed webhook call) and applies
+    // them against the resolved account's backend. A mutation is marked
+    // "completed" ONLY when the real operation genuinely succeeded; a transient
+    // failure is retried on later ticks and a permanent one (gone message,
+    // unknown action) fails honestly -- success is never fabricated.
     let db_outbox = db.clone();
+    let outbox_env = Arc::new(nunciod::outbox::ProductionExecutionEnv::new(
+        db.clone(),
+        account_secrets.clone(),
+    ));
     let _outbox_task = tokio::spawn(async move {
         let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
         loop {
             interval.tick().await;
-            if let Ok(pending) = db_outbox.list_pending_mutations(50).await {
-                for item in pending {
-                    let next_retry = item.retry_count + 1;
-                    if next_retry > OutboxManager::MAX_RETRIES {
-                        let _ = db_outbox
-                            .update_mutation_status(&item.id, "failed", next_retry)
-                            .await;
-                        continue;
-                    }
-                    let backoff_ms = OutboxManager::calculate_backoff_ms(item.retry_count);
-                    tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await;
-                    // Outbound IMAP/JMAP remote mutation execution (actually
-                    // applying a filter action like move/delete/flag on the
-                    // real mail server) is not yet wired to a real mail
-                    // server. This worker deliberately does NOT mark
-                    // mutations "completed": doing so would fabricate
-                    // success for work that never happened. It only bumps
-                    // the retry counter (so an item that keeps failing still
-                    // eventually flips to "failed" once retries are
-                    // exhausted) and leaves the item "pending" so it is
-                    // retried on the next poll tick once a real sender is
-                    // wired in.
-                    let _ = db_outbox
-                        .update_mutation_status(&item.id, "pending", next_retry)
-                        .await;
-                }
+            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
+                );
             }
         }
     });

crates/nunciod/src/outbox.rs

@@ -0,0 +1,466 @@
+//! Outbox remote-mutation executor.
+//!
+//! The filter engine enqueues a [`nuncio_filter::PendingRemoteMutation`] for
+//! every genuinely-remote action a rule matches (`MOVE`/`COPY`/`FLAG`/
+//! `UNFLAG`/`DELETE`/`FORWARD`/`CALL WEBHOOK`); `MARK READ`/`MARK UNREAD` are
+//! applied to the local store inline and are never enqueued. This module
+//! drains that outbox against the real world -- the resolved account's
+//! [`MailBackend`] for the mailbox mutations, a [`MessageSender`] for
+//! `FORWARD`, and the SSRF-hardened webhook dispatcher for `CALL WEBHOOK`.
+//!
+//! # The one inviolable rule
+//!
+//! A mutation is marked `completed` ONLY when the real remote operation
+//! genuinely succeeded. A transient failure keeps the item `pending` (bumping
+//! its retry count) until [`nuncio_filter::OutboxManager::MAX_RETRIES`] is
+//! exceeded, at which point it flips to `failed`. A permanent failure (the
+//! message no longer exists, a malformed payload, an unknown action) flips to
+//! `failed` immediately -- there is nothing a retry could fix. Success is
+//! never fabricated.
+//!
+//! # Testability
+//!
+//! [`execute_pending_mutations`] takes a `&dyn RemoteExecutionEnv`, so an
+//! offline E2E injects mock backends/senders and a loopback-allowed webhook
+//! dispatcher, proving the full sync -> filter -> outbox -> execute vertical
+//! without any live network. Production wires [`ProductionExecutionEnv`],
+//! which resolves real per-account engines from the store and the OS keyring.
+
+use async_trait::async_trait;
+use nuncio_core::AccountConfig;
+use nuncio_filter::{
+    MutationPayload, OutboxManager, PendingRemoteMutation, ValidationOptions, WebhookDispatcher,
+    WebhookError,
+};
+use nuncio_mail::{
+    MailBackend, MailError, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec,
+    SmtpTransportEngine,
+};
+use nuncio_store::db::{DatabaseEngine, DatabaseError};
+use nuncio_store::vault::{SecretManager, VaultError, WEBHOOK_SIGNING_KEY_ACCOUNT};
+use std::sync::Arc;
+use thiserror::Error;
+
+/// Errors that can occur while resolving the resources needed to execute a
+/// single outbox mutation. These are surfaced from a [`RemoteExecutionEnv`]
+/// and treated as transient (retryable) by the executor unless it has already
+/// determined the failure is permanent.
+#[derive(Debug, Error)]
+pub enum OutboxExecuteError {
+    /// No persisted account configuration matches the resolved account id.
+    #[error("account '{0}' is not configured")]
+    AccountNotFound(String),
+    /// Failed to read the account's credential from the secret vault.
+    #[error("failed to read account credential from the secret vault: {0}")]
+    Vault(#[from] VaultError),
+    /// A mail backend / transport construction or operation error.
+    #[error("mail error: {0}")]
+    Mail(#[from] MailError),
+    /// A database error occurred while resolving the account/message.
+    #[error("database error: {0}")]
+    Store(#[from] DatabaseError),
+}
+
+/// Injectable seam for the resources a mutation needs: the per-account mail
+/// backend and outbound sender, plus the webhook dispatcher. Production uses
+/// [`ProductionExecutionEnv`]; tests inject mocks.
+#[async_trait]
+pub trait RemoteExecutionEnv: Send + Sync {
+    /// Build the [`MailBackend`] for the account that owns the target message.
+    async fn mail_backend(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MailBackend>, OutboxExecuteError>;
+
+    /// Build the [`MessageSender`] used to deliver a `FORWARD`.
+    async fn message_sender(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MessageSender>, OutboxExecuteError>;
+
+    /// Dispatch a signed `CALL WEBHOOK` POST, returning the HTTP status code on
+    /// delivery. The implementation owns the SSRF egress policy -- production
+    /// keeps the secure default (private targets blocked); a test env may relax
+    /// it to reach a loopback mock server.
+    async fn dispatch_webhook(
+        &self,
+        url: &str,
+        rule_id: &str,
+        message_id: &str,
+        subject: &str,
+        sender: &str,
+    ) -> Result<u16, WebhookError>;
+}
+
+/// The disposition of a single mutation attempt.
+enum Disposition {
+    /// The real operation genuinely succeeded; mark `completed`.
+    Completed,
+    /// A transient failure; keep retrying until `MAX_RETRIES` is exceeded.
+    Retry(String),
+    /// A permanent failure nothing could retry away; mark `failed` now.
+    Permanent(String),
+}
+
+/// Tally of what a drain pass did, for logging and test assertions.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub struct ExecutionSummary {
+    /// Mutations marked `completed` on genuine success.
+    pub completed: usize,
+    /// Mutations left `pending` for a later retry.
+    pub retried: usize,
+    /// Mutations flipped to `failed` (permanent, or retries exhausted).
+    pub failed: usize,
+}
+
+/// Drain up to `limit` pending mutations, executing each against `env` and
+/// persisting the honest outcome. Never marks a mutation `completed` unless the
+/// real operation succeeded. Returns a tally of the pass.
+pub async fn execute_pending_mutations(
+    db: &DatabaseEngine,
+    env: &dyn RemoteExecutionEnv,
+    limit: usize,
+) -> ExecutionSummary {
+    let pending = match db.list_pending_mutations(limit).await {
+        Ok(items) => items,
+        Err(e) => {
+            tracing::warn!("outbox: failed to list pending mutations: {e}");
+            return ExecutionSummary::default();
+        }
+    };
+
+    let mut summary = ExecutionSummary::default();
+    for item in pending {
+        let disposition = execute_one(db, env, &item).await;
+        let next_retry = item.retry_count + 1;
+        let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition {
+            Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1),
+            Disposition::Permanent(reason) => {
+                tracing::warn!(
+                    mutation_id = %item.id,
+                    "outbox: permanently failing mutation: {reason}"
+                );
+                ("failed", next_retry, |s| s.failed += 1)
+            }
+            Disposition::Retry(reason) if next_retry > OutboxManager::MAX_RETRIES => {
+                tracing::warn!(
+                    mutation_id = %item.id,
+                    "outbox: mutation failed after exhausting retries: {reason}"
+                );
+                ("failed", next_retry, |s| s.failed += 1)
+            }
+            Disposition::Retry(reason) => {
+                tracing::debug!(
+                    mutation_id = %item.id,
+                    "outbox: mutation will be retried: {reason}"
+                );
+                ("pending", next_retry, |s| s.retried += 1)
+            }
+        };
+
+        if let Err(e) = db.update_mutation_status(&item.id, status, retry).await {
+            // A status-write failure must not crash the worker; the item stays
+            // in its prior state and is reconsidered on the next poll.
+            tracing::warn!(mutation_id = %item.id, "outbox: failed to persist status: {e}");
+        } else {
+            tally(&mut summary);
+        }
+    }
+    summary
+}
+
+/// Execute a single mutation and report its honest disposition. Performs no
+/// status writes itself -- the caller persists the outcome.
+async fn execute_one(
+    db: &DatabaseEngine,
+    env: &dyn RemoteExecutionEnv,
+    mutation: &PendingRemoteMutation,
+) -> Disposition {
+    // Resolve the message to recover its account and source folder. A message
+    // that no longer exists can never be acted on -- fail permanently.
+    let email = match db.get_message(&mutation.message_id).await {
+        Ok(email) => email,
+        Err(e) if e.is_not_found() => {
+            return Disposition::Permanent(format!(
+                "message '{}' no longer exists in the store",
+                mutation.message_id
+            ));
+        }
+        Err(e) => {
+            return Disposition::Retry(format!("failed to load message: {e}"));
+        }
+    };
+
+    let payload = match serde_json::from_str::<MutationPayload>(&mutation.payload) {
+        Ok(payload) => payload,
+        Err(e) => {
+            return Disposition::Permanent(format!("malformed mutation payload: {e}"));
+        }
+    };
+
+    match mutation.mutation_type.as_str() {
+        "MOVE" | "COPY" | "FLAG" | "UNFLAG" | "DELETE" => {
+            dispatch_mailbox_mutation(db, env, mutation, &email, &payload).await
+        }
+        "FORWARD" => dispatch_forward(db, env, &email, &payload).await,
+        "WEBHOOK" => dispatch_webhook(env, mutation, &email, &payload).await,
+        other => Disposition::Permanent(format!("unknown mutation type '{other}'")),
+    }
+}
+
+/// 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.
+async fn dispatch_mailbox_mutation(
+    db: &DatabaseEngine,
+    env: &dyn RemoteExecutionEnv,
+    mutation: &PendingRemoteMutation,
+    email: &nuncio_core::model::Email,
+    payload: &MutationPayload,
+) -> Disposition {
+    let kind = match mutation.mutation_type.as_str() {
+        "FLAG" => RemoteMutationKind::SetFlagged { value: true },
+        "UNFLAG" => RemoteMutationKind::SetFlagged { value: false },
+        "DELETE" => RemoteMutationKind::Delete,
+        "MOVE" => match &payload.target {
+            Some(folder) => RemoteMutationKind::Move {
+                to_folder: folder.clone(),
+            },
+            None => {
+                return Disposition::Permanent("MOVE mutation has no target folder".to_string())
+            }
+        },
+        "COPY" => match &payload.target {
+            Some(folder) => RemoteMutationKind::Copy {
+                to_folder: folder.clone(),
+            },
+            None => {
+                return Disposition::Permanent("COPY mutation has no target folder".to_string())
+            }
+        },
+        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}")),
+    };
+
+    let spec = RemoteMutationSpec {
+        message_id: email.id.clone(),
+        folder_id: email.folder_id.clone(),
+        folder_checkpoint,
+        kind,
+    };
+
+    match backend.apply_mutation(&spec).await {
+        Ok(()) => Disposition::Completed,
+        Err(e) => Disposition::Retry(format!("remote mutation failed: {e}")),
+    }
+}
+
+/// Compose and send a forward of `email` to the payload's target address via
+/// the account's outbound transport.
+async fn dispatch_forward(
+    db: &DatabaseEngine,
+    env: &dyn RemoteExecutionEnv,
+    email: &nuncio_core::model::Email,
+    payload: &MutationPayload,
+) -> Disposition {
+    let Some(to) = payload.target.clone() else {
+        return Disposition::Permanent("FORWARD mutation has no target address".to_string());
+    };
+
+    // The From: is the forwarding account's own address, resolved from its
+    // persisted configuration rather than trusting the stored envelope.
+    let from = match find_account_config(db, &email.account_id).await {
+        Ok(config) => config.email_address,
+        Err(OutboxExecuteError::AccountNotFound(id)) => {
+            return Disposition::Permanent(format!("forwarding account '{id}' is not configured"));
+        }
+        Err(e) => return Disposition::Retry(format!("failed to resolve forwarding account: {e}")),
+    };
+
+    let sender = match env.message_sender(&email.account_id).await {
+        Ok(sender) => sender,
+        Err(e) => return Disposition::Retry(format!("failed to build message sender: {e}")),
+    };
+
+    let outbound = OutboundMessage {
+        from,
+        to,
+        cc: None,
+        subject: format!("Fwd: {}", email.subject),
+        body_plain: email.body_plain.clone(),
+        body_html: email.body_html.clone(),
+        attachments: email.attachments.clone(),
+    };
+
+    match sender.send(&outbound).await {
+        Ok(()) => Disposition::Completed,
+        Err(e) => Disposition::Retry(format!("forward send failed: {e}")),
+    }
+}
+
+/// Deliver a signed webhook for the matched message. A 2xx/3xx response is a
+/// genuine delivery; a 4xx/5xx is treated as a retryable failure so a transient
+/// receiver outage does not silently drop the notification.
+async fn dispatch_webhook(
+    env: &dyn RemoteExecutionEnv,
+    mutation: &PendingRemoteMutation,
+    email: &nuncio_core::model::Email,
+    payload: &MutationPayload,
+) -> Disposition {
+    let Some(url) = payload.target.as_deref() else {
+        return Disposition::Permanent("WEBHOOK mutation has no target URL".to_string());
+    };
+
+    match env
+        .dispatch_webhook(
+            url,
+            &mutation.rule_id,
+            &mutation.message_id,
+            &email.subject,
+            &email.sender,
+        )
+        .await
+    {
+        Ok(status) if (200..400).contains(&status) => Disposition::Completed,
+        Ok(status) => Disposition::Retry(format!("webhook returned HTTP {status}")),
+        // A security-policy violation (blocked SSRF target) is permanent: the
+        // URL will never become allowable on retry.
+        Err(WebhookError::SecurityViolation(reason)) => {
+            Disposition::Permanent(format!("webhook blocked by security policy: {reason}"))
+        }
+        Err(e) => Disposition::Retry(format!("webhook dispatch failed: {e}")),
+    }
+}
+
+/// Look up an account's persisted configuration by id.
+async fn find_account_config(
+    db: &DatabaseEngine,
+    account_id: &str,
+) -> Result<AccountConfig, OutboxExecuteError> {
+    db.list_accounts()
+        .await?
+        .into_iter()
+        .find(|a| a.id == account_id)
+        .ok_or_else(|| OutboxExecuteError::AccountNotFound(account_id.to_string()))
+}
+
+/// Production [`RemoteExecutionEnv`]: resolves real per-account IMAP/JMAP mail
+/// backends and SMTP senders from the store plus the OS keyring, and lazily
+/// builds a single SSRF-hardened [`WebhookDispatcher`] keyed by a daemon-wide
+/// signing secret provisioned from the vault.
+///
+/// The webhook signing key is provisioned lazily -- only the first time a
+/// `CALL WEBHOOK` mutation is actually dispatched -- so a vault issue with that
+/// one key can never disable the mailbox-mutation and forward paths, which do
+/// not need it.
+pub struct ProductionExecutionEnv {
+    db: Arc<DatabaseEngine>,
+    secrets: Arc<SecretManager>,
+    webhook: std::sync::OnceLock<WebhookDispatcher>,
+}
+
+impl ProductionExecutionEnv {
+    /// Build the production environment. Infallible: no vault access happens
+    /// here, so the outbox worker can always start; the webhook signing key is
+    /// provisioned on first webhook dispatch instead (see the struct doc).
+    pub fn new(db: Arc<DatabaseEngine>, secrets: Arc<SecretManager>) -> Self {
+        Self {
+            db,
+            secrets,
+            webhook: std::sync::OnceLock::new(),
+        }
+    }
+
+    async fn account(&self, account_id: &str) -> Result<AccountConfig, OutboxExecuteError> {
+        find_account_config(&self.db, account_id).await
+    }
+
+    /// Return the lazily-provisioned webhook dispatcher, minting (or loading)
+    /// the daemon-wide signing key from the vault on first use. Fails closed if
+    /// the vault is unavailable, so a `CALL WEBHOOK` is never dispatched
+    /// unsigned -- and, being lazy, that failure is isolated to webhook
+    /// dispatch and never disables the other mutation paths.
+    fn webhook_dispatcher(&self) -> Result<&WebhookDispatcher, WebhookError> {
+        if let Some(dispatcher) = self.webhook.get() {
+            return Ok(dispatcher);
+        }
+        let key_bytes = self
+            .secrets
+            .get_or_create_key_bytes(WEBHOOK_SIGNING_KEY_ACCOUNT, 32)
+            .map_err(|e| {
+                WebhookError::SigningError(format!(
+                    "failed to provision webhook signing key from the vault: {e}"
+                ))
+            })?;
+        // A concurrent caller may win the race to set it; either way the stored
+        // dispatcher is authoritative.
+        let _ = self
+            .webhook
+            .set(WebhookDispatcher::new(hex::encode(key_bytes)));
+        self.webhook.get().ok_or_else(|| {
+            WebhookError::SigningError("webhook dispatcher unavailable after provisioning".into())
+        })
+    }
+}
+
+#[async_trait]
+impl RemoteExecutionEnv for ProductionExecutionEnv {
+    async fn mail_backend(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MailBackend>, OutboxExecuteError> {
+        let config = self.account(account_id).await?;
+        let password = self.secrets.get_secret(&config.keyring_secret_key)?;
+        Ok(crate::sync::build_mail_backend(&config, &password))
+    }
+
+    async fn message_sender(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MessageSender>, OutboxExecuteError> {
+        let config = self.account(account_id).await?;
+        let password = self.secrets.get_secret(&config.keyring_secret_key)?;
+        let engine = SmtpTransportEngine::new(
+            &config.smtp_host,
+            config.smtp_port,
+            config.smtp_tls_mode,
+            &config.email_address,
+            &password,
+        )?;
+        Ok(Box::new(engine))
+    }
+
+    async fn dispatch_webhook(
+        &self,
+        url: &str,
+        rule_id: &str,
+        message_id: &str,
+        subject: &str,
+        sender: &str,
+    ) -> Result<u16, WebhookError> {
+        // Production keeps the secure default egress policy (private/loopback
+        // targets are rejected).
+        self.webhook_dispatcher()?
+            .dispatch_with_options(
+                url,
+                rule_id,
+                message_id,
+                subject,
+                sender,
+                &ValidationOptions::default(),
+            )
+            .await
+    }
+}

crates/nunciod/src/sync.rs

@@ -339,7 +339,7 @@ async fn find_account(
 /// keyring by the caller). This is the ONLY place a concrete protocol engine
 /// is chosen; every other function in this module only ever sees `&dyn
 /// MailBackend`.
-fn build_mail_backend(config: &AccountConfig, password: &str) -> Box<dyn MailBackend> {
+pub(crate) fn build_mail_backend(config: &AccountConfig, password: &str) -> Box<dyn MailBackend> {
     match config.protocol {
         AccountProtocol::ImapSmtp => Box::new(ImapEngine::with_credentials(
             &config.id,

crates/nunciod/tests/outbox_executor_e2e_test.rs

@@ -0,0 +1,329 @@
+//! Offline end-to-end test for the outbox remote-mutation executor.
+//!
+//! Proves the full `sync -> filter -> outbox -> execute` vertical without any
+//! live network: a matched filter rule enqueues a real
+//! `PendingRemoteMutation`, and `execute_pending_mutations` drains it against
+//! injected mocks -- a `MockMailBackend` that records each applied mutation, a
+//! `MockMessageSender` for `FORWARD`, and a loopback-allowed webhook dispatcher
+//! hitting a wiremock server. The safety-critical property under test is that a
+//! mutation is marked `completed` ONLY when the real op genuinely succeeded: a
+//! failing backend never records an op and never completes the item.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use async_trait::async_trait;
+use nuncio_core::model::Email;
+use nuncio_filter::{FilterEngine, NsqlParser, ValidationOptions, WebhookDispatcher, WebhookError};
+use nuncio_mail::{
+    MailBackend, MessageSender, MockMailBackend, MockMessageSender, RemoteMutationKind,
+};
+use nuncio_store::db::DatabaseEngine;
+use nunciod::outbox::{execute_pending_mutations, RemoteExecutionEnv};
+use nunciod::sync::apply_filter_actions;
+use wiremock::matchers::{method, path};
+use wiremock::{Mock, MockServer, ResponseTemplate};
+
+const ACCOUNT_ID: &str = "acct-outbox-e2e";
+const FOLDER_ID: &str = "INBOX";
+const CHECKPOINT: &str = "1:105";
+
+/// Test [`RemoteExecutionEnv`] injecting recording mocks and a webhook
+/// dispatcher whose SSRF egress policy is relaxed so it can reach a loopback
+/// wiremock server (production keeps the secure default).
+struct MockEnv {
+    backend: MockMailBackend,
+    sender: MockMessageSender,
+    webhook: WebhookDispatcher,
+}
+
+impl MockEnv {
+    fn new() -> Self {
+        Self {
+            backend: MockMailBackend::new(),
+            sender: MockMessageSender::new(),
+            webhook: WebhookDispatcher::new("test-signing-key"),
+        }
+    }
+}
+
+#[async_trait]
+impl RemoteExecutionEnv for MockEnv {
+    async fn mail_backend(
+        &self,
+        _account_id: &str,
+    ) -> Result<Box<dyn MailBackend>, nunciod::outbox::OutboxExecuteError> {
+        // Clones share the same Arc-backed recording storage, so the returned
+        // boxed backend records into the `self.backend` the test inspects.
+        Ok(Box::new(self.backend.clone()))
+    }
+
+    async fn message_sender(
+        &self,
+        _account_id: &str,
+    ) -> Result<Box<dyn MessageSender>, nunciod::outbox::OutboxExecuteError> {
+        Ok(Box::new(self.sender.clone()))
+    }
+
+    async fn dispatch_webhook(
+        &self,
+        url: &str,
+        rule_id: &str,
+        message_id: &str,
+        subject: &str,
+        sender: &str,
+    ) -> Result<u16, WebhookError> {
+        let opts = ValidationOptions {
+            block_private_webhooks: false,
+            ..Default::default()
+        };
+        self.webhook
+            .dispatch_with_options(url, rule_id, message_id, subject, sender, &opts)
+            .await
+    }
+}
+
+fn sample_account() -> nuncio_core::AccountConfig {
+    nuncio_core::AccountConfig {
+        id: ACCOUNT_ID.to_string(),
+        name: "Outbox E2E Account".to_string(),
+        email_address: "owner@nuncio.mx".to_string(),
+        protocol: nuncio_core::AccountProtocol::ImapSmtp,
+        server_host: "imap.nuncio.mx".to_string(),
+        server_port: 993,
+        smtp_host: "smtp.nuncio.mx".to_string(),
+        smtp_port: 465,
+        use_tls: true,
+        imap_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+        smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls,
+        keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"),
+        sync_interval_secs: 60,
+    }
+}
+
+fn sample_email(subject: &str) -> Email {
+    Email {
+        id: "imap-uid-42".to_string(),
+        account_id: ACCOUNT_ID.to_string(),
+        folder_id: FOLDER_ID.to_string(),
+        subject: subject.to_string(),
+        sender: "alice@nuncio.mx".to_string(),
+        recipient: "owner@nuncio.mx".to_string(),
+        received_at: 1_700_000_000,
+        read: false,
+        body_plain: Some("original body".to_string()),
+        body_html: None,
+        attachments: Vec::new(),
+    }
+}
+
+/// Seed the account, source-folder checkpoint, and a matching message, then run
+/// the given rule's actions through `apply_filter_actions` so the outbox is
+/// populated exactly as a real sync would populate it.
+async fn seed_and_enqueue(db: &DatabaseEngine, rule_nsql: &str, subject: &str) {
+    db.save_account(&sample_account())
+        .await
+        .expect("save account");
+    db.save_folder_sync_state(ACCOUNT_ID, FOLDER_ID, CHECKPOINT)
+        .await
+        .expect("save folder checkpoint");
+
+    let email = sample_email(subject);
+    db.save_email(&email).await.expect("save email");
+
+    let rule = NsqlParser::parse_rule("Outbox Rule", 1, rule_nsql).expect("parse rule");
+    let engine = FilterEngine::new(vec![rule]).expect("compile rule");
+    apply_filter_actions(db, &engine, &email).await;
+
+    let pending = db.list_pending_mutations(10).await.expect("list pending");
+    assert_eq!(
+        pending.len(),
+        1,
+        "the rule must enqueue exactly one mutation"
+    );
+}
+
+#[tokio::test]
+async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_checkpoint() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'",
+        "Please Archive",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    assert_eq!(summary.completed, 1);
+    assert_eq!(summary.failed, 0);
+
+    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");
+    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].kind,
+        RemoteMutationKind::Move {
+            to_folder: "Archive".to_string()
+        }
+    );
+
+    // A completed mutation is no longer pending.
+    assert!(db.list_pending_mutations(10).await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn flag_rule_executes_as_set_flagged() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Star' ACTION FLAG",
+        "Star this",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    assert_eq!(summary.completed, 1);
+    let applied = env.backend.applied_mutations();
+    assert_eq!(applied.len(), 1);
+    assert_eq!(
+        applied[0].kind,
+        RemoteMutationKind::SetFlagged { value: true }
+    );
+}
+
+#[tokio::test]
+async fn delete_rule_executes_as_delete() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Trash' ACTION DELETE",
+        "Trash me",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    assert_eq!(summary.completed, 1);
+    let applied = env.backend.applied_mutations();
+    assert_eq!(applied.len(), 1);
+    assert_eq!(applied[0].kind, RemoteMutationKind::Delete);
+}
+
+#[tokio::test]
+async fn a_failing_backend_op_is_never_marked_completed() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'",
+        "Please Archive",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    env.backend.set_should_fail(true);
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    // The op failed: nothing recorded, nothing completed, and the item is kept
+    // for retry (still pending, with a bumped retry count) -- never faked.
+    assert_eq!(summary.completed, 0);
+    assert_eq!(summary.retried, 1);
+    assert!(env.backend.applied_mutations().is_empty());
+
+    let pending = db.list_pending_mutations(10).await.unwrap();
+    assert_eq!(pending.len(), 1, "a failed op must remain pending");
+    assert_eq!(pending[0].retry_count, 1, "the retry count must be bumped");
+}
+
+#[tokio::test]
+async fn forward_rule_sends_via_the_message_sender() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Forward' ACTION FORWARD TO 'boss@nuncio.mx'",
+        "Please Forward",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    assert_eq!(summary.completed, 1);
+    let sent = env.sender.sent_messages();
+    assert_eq!(sent.len(), 1, "the forward must genuinely be sent");
+    assert_eq!(sent[0].to, "boss@nuncio.mx");
+    // The From: is the forwarding account's own configured address.
+    assert_eq!(sent[0].from, "owner@nuncio.mx");
+    assert!(sent[0].subject.contains("Please Forward"));
+}
+
+#[tokio::test]
+async fn webhook_rule_dispatches_a_signed_post_to_the_receiver() {
+    let mock_server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/hooks/nuncio"))
+        .and(wiremock::matchers::header(
+            "Content-Type",
+            "application/json",
+        ))
+        .respond_with(ResponseTemplate::new(200))
+        .expect(1)
+        .mount(&mock_server)
+        .await;
+
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    let url = format!("{}/hooks/nuncio", mock_server.uri());
+    seed_and_enqueue(
+        &db,
+        &format!("WHERE subject CONTAINS 'Ping' ACTION CALL WEBHOOK '{url}'"),
+        "Ping me",
+    )
+    .await;
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    // Completion is genuine: wiremock's `.expect(1)` verifies the POST actually
+    // arrived when the server drops.
+    assert_eq!(summary.completed, 1);
+    assert!(db.list_pending_mutations(10).await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn a_mutation_for_a_vanished_message_fails_permanently() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    db.save_account(&sample_account())
+        .await
+        .expect("save account");
+
+    // Enqueue a mutation whose target message was never persisted.
+    let mutation = nuncio_filter::OutboxManager::create_mutation(
+        "rule-x",
+        "imap-uid-999",
+        "MOVE",
+        Some("Archive".to_string()),
+    );
+    db.save_pending_mutation(&mutation)
+        .await
+        .expect("save mutation");
+
+    let env = MockEnv::new();
+    let summary = execute_pending_mutations(&db, &env, 50).await;
+
+    // Cannot act on a message that no longer exists -- fail honestly, never
+    // fabricate success, and never touch the backend.
+    assert_eq!(summary.failed, 1);
+    assert_eq!(summary.completed, 0);
+    assert!(env.backend.applied_mutations().is_empty());
+    assert!(
+        db.list_pending_mutations(10).await.unwrap().is_empty(),
+        "a permanently-failed mutation is no longer pending"
+    );
+}