crates/nuncio-cli/src/args.rs

@@ -250,6 +250,26 @@ pub enum AccountSubcommand {
         #[arg(short, long, help = "Unique account identifier")]
         id: String,
     },
+    /// Turn the filter engine on or off for one account.
+    Filters {
+        /// Unique account identifier.
+        #[arg(short, long, help = "Unique account identifier")]
+        id: String,
+        /// Run filters against this account's incoming mail.
+        #[arg(
+            long,
+            help = "Enable filters for this account",
+            conflicts_with = "disable"
+        )]
+        enable: bool,
+        /// Stop running filters against this account's incoming mail.
+        #[arg(
+            long,
+            help = "Disable filters for this account",
+            conflicts_with = "enable"
+        )]
+        disable: bool,
+    },
     /// Edit an existing account profile configuration.
     Edit {
         /// Unique account identifier.
@@ -343,6 +363,66 @@ pub enum MailSubcommand {
         #[arg(long, help = "Mark the message as unread", conflicts_with = "read")]
         unread: bool,
     },
+    /// Move one occupancy of a message into another folder.
+    Move {
+        /// Unique message identifier.
+        #[arg(short, long, help = "Unique message identifier")]
+        id: String,
+        /// Destination folder identifier.
+        #[arg(short, long, help = "Destination folder identifier")]
+        to: String,
+        /// Which mailbox occupancy to act on, as
+        /// `account/folder/uidvalidity/uid`. Required when the message occupies
+        /// more than one mailbox; `mail read` lists them.
+        #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")]
+        placement: Option<String>,
+    },
+    /// Delete one occupancy of a message. Other occupancies survive.
+    Delete {
+        /// Unique message identifier.
+        #[arg(short, long, help = "Unique message identifier")]
+        id: String,
+        /// Which mailbox occupancy to act on, as
+        /// `account/folder/uidvalidity/uid`.
+        #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")]
+        placement: Option<String>,
+    },
+    /// Flag one occupancy of a message.
+    Flag {
+        /// Unique message identifier.
+        #[arg(short, long, help = "Unique message identifier")]
+        id: String,
+        /// Which mailbox occupancy to act on, as
+        /// `account/folder/uidvalidity/uid`.
+        #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")]
+        placement: Option<String>,
+    },
+    /// Remove the flag from one occupancy of a message.
+    Unflag {
+        /// Unique message identifier.
+        #[arg(short, long, help = "Unique message identifier")]
+        id: String,
+        /// Which mailbox occupancy to act on, as
+        /// `account/folder/uidvalidity/uid`.
+        #[arg(long, help = "Occupancy as account/folder/uidvalidity/uid")]
+        placement: Option<String>,
+    },
+    /// List mutations the server refused because another client changed the
+    /// message first.
+    Conflicts {
+        /// Also list conflicts that have already been resolved.
+        #[arg(long, help = "Include already-resolved conflicts")]
+        all: bool,
+    },
+    /// Mark a conflict handled, recording how it was resolved.
+    Resolve {
+        /// Conflict identifier, as reported by `mail conflicts`.
+        #[arg(short, long, help = "Conflict identifier")]
+        id: String,
+        /// Note recording how the conflict was resolved.
+        #[arg(short, long, help = "How the conflict was resolved")]
+        note: String,
+    },
     /// Export mailbox messages to a portable file format: MBOX, an EML
     /// zip archive, JSON, or JSON Lines.
     Export {

crates/nuncio-cli/src/runner.rs

@@ -425,6 +425,14 @@ impl HeadlessRunner {
                     .await
                 }
                 AccountSubcommand::Show { id } => self.handle_account_show(id, json_mode).await,
+                AccountSubcommand::Filters {
+                    id,
+                    enable,
+                    disable,
+                } => {
+                    self.handle_account_filters(id, *enable, *disable, json_mode)
+                        .await
+                }
                 AccountSubcommand::Edit {
                     id,
                     email,
@@ -468,6 +476,28 @@ impl HeadlessRunner {
                 MailSubcommand::Mark { id, read, unread } => {
                     self.handle_mark_read(id, *read, *unread, json_mode).await
                 }
+                MailSubcommand::Move { id, to, placement } => {
+                    self.handle_mail_mutation("move", id, placement.as_ref(), Some(to), json_mode)
+                        .await
+                }
+                MailSubcommand::Delete { id, placement } => {
+                    self.handle_mail_mutation("delete", id, placement.as_ref(), None, json_mode)
+                        .await
+                }
+                MailSubcommand::Flag { id, placement } => {
+                    self.handle_mail_mutation("flag", id, placement.as_ref(), None, json_mode)
+                        .await
+                }
+                MailSubcommand::Unflag { id, placement } => {
+                    self.handle_mail_mutation("unflag", id, placement.as_ref(), None, json_mode)
+                        .await
+                }
+                MailSubcommand::Conflicts { all } => {
+                    self.handle_list_conflicts(*all, json_mode).await
+                }
+                MailSubcommand::Resolve { id, note } => {
+                    self.handle_resolve_conflict(id, note, json_mode).await
+                }
                 MailSubcommand::Export {
                     format,
                     out,
@@ -1079,6 +1109,245 @@ impl HeadlessRunner {
         }
     }
 
+    /// Parse an `account/folder/uidvalidity/uid` occupancy selector.
+    ///
+    /// All four coordinates are required and none may be empty: a partial
+    /// selector addresses nothing, and silently completing it would aim a
+    /// destructive mutation at a mailbox the user did not name.
+    fn parse_placement(spec: &str) -> Result<nuncio_proto::v1::PlacementRef, String> {
+        let parts: Vec<&str> = spec.split('/').collect();
+        match parts.as_slice() {
+            [account_id, folder_id, uidvalidity, uid]
+                if !account_id.is_empty()
+                    && !folder_id.is_empty()
+                    && !uidvalidity.is_empty()
+                    && !uid.is_empty() =>
+            {
+                Ok(nuncio_proto::v1::PlacementRef {
+                    account_id: (*account_id).to_string(),
+                    folder_id: (*folder_id).to_string(),
+                    uidvalidity: (*uidvalidity).to_string(),
+                    uid: (*uid).to_string(),
+                })
+            }
+            _ => Err(format!(
+                "placement '{spec}' is not account/folder/uidvalidity/uid"
+            )),
+        }
+    }
+
+    /// `mail move`/`delete`/`flag`/`unflag`: queue a placement-addressed remote
+    /// mutation via the daemon and report the outbox id it was queued as.
+    ///
+    /// The reported outcome is "queued", never "done": the daemon's outbox is
+    /// what actually talks to the server, and only it can say whether the
+    /// mutation applied, conflicted, or could not be verified.
+    async fn handle_mail_mutation(
+        &self,
+        verb: &str,
+        id: &str,
+        placement: Option<&String>,
+        destination: Option<&String>,
+        json_mode: bool,
+    ) -> String {
+        let placement = match placement.map(|p| Self::parse_placement(p)) {
+            Some(Ok(placement)) => Some(placement),
+            Some(Err(e)) => return Self::render_error(&e, json_mode),
+            None => None,
+        };
+
+        let mut client = match self.connect_mail_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        let queued = match verb {
+            "move" => {
+                let Some(to) = destination else {
+                    return Self::render_error("a destination folder is required", json_mode);
+                };
+                client
+                    .move_message(nuncio_proto::v1::MoveMessageRequest {
+                        message_id: id.to_string(),
+                        placement,
+                        destination_folder_id: to.clone(),
+                    })
+                    .await
+                    .map(|r| r.into_inner().mutation_id)
+            }
+            "delete" => client
+                .delete_message(nuncio_proto::v1::DeleteMessageRequest {
+                    message_id: id.to_string(),
+                    placement,
+                })
+                .await
+                .map(|r| r.into_inner().mutation_id),
+            _ => client
+                .flag_message(nuncio_proto::v1::FlagMessageRequest {
+                    message_id: id.to_string(),
+                    placement,
+                    flagged: verb == "flag",
+                })
+                .await
+                .map(|r| r.into_inner().mutation_id),
+        };
+
+        match queued {
+            Ok(mutation_id) => {
+                if json_mode {
+                    format_json(&json!({
+                        "status": "queued",
+                        "operation": verb,
+                        "id": id,
+                        "mutation_id": mutation_id,
+                    }))
+                } else {
+                    format!("Queued {verb} of '{id}' as mutation '{mutation_id}'")
+                }
+            }
+            Err(status) => Self::render_status_error(verb, &status, json_mode),
+        }
+    }
+
+    /// `mail conflicts`: list mutations the server refused because another
+    /// client changed the message first.
+    async fn handle_list_conflicts(&self, include_resolved: bool, json_mode: bool) -> String {
+        let mut client = match self.connect_mail_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+        match client
+            .list_conflicts(nuncio_proto::v1::ListConflictsRequest {
+                include_resolved,
+                limit: 100,
+            })
+            .await
+        {
+            Ok(response) => {
+                let conflicts = response.into_inner().conflicts;
+                if json_mode {
+                    let rows: Vec<_> = conflicts
+                        .iter()
+                        .map(|c| {
+                            json!({
+                                "id": c.id,
+                                "mutation_id": c.mutation_id,
+                                "message_id": c.message_id,
+                                "operation": c.mutation_type,
+                                "observed": c.observed,
+                                "resolved": c.resolved_at.is_some(),
+                            })
+                        })
+                        .collect();
+                    format_json(&json!({ "conflicts": rows }))
+                } else if conflicts.is_empty() {
+                    "No conflicts recorded".to_string()
+                } else {
+                    let mut out = String::new();
+                    for c in &conflicts {
+                        let state = if c.resolved_at.is_some() {
+                            "resolved"
+                        } else {
+                            "open"
+                        };
+                        out.push_str(&format!(
+                            "{} [{}] {} on '{}': {}\n",
+                            c.id, state, c.mutation_type, c.message_id, c.observed
+                        ));
+                    }
+                    out.trim_end().to_string()
+                }
+            }
+            Err(status) => Self::render_status_error("list_conflicts", &status, json_mode),
+        }
+    }
+
+    /// `mail resolve`: record a human's decision about a conflict.
+    async fn handle_resolve_conflict(&self, id: &str, note: &str, json_mode: bool) -> String {
+        let mut client = match self.connect_mail_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+        match client
+            .resolve_conflict(nuncio_proto::v1::ResolveConflictRequest {
+                conflict_id: id.to_string(),
+                resolution: note.to_string(),
+            })
+            .await
+        {
+            Ok(_) => {
+                if json_mode {
+                    format_json(&json!({ "status": "resolved", "id": id }))
+                } else {
+                    format!("Conflict '{id}' resolved")
+                }
+            }
+            Err(status) => Self::render_status_error("resolve_conflict", &status, json_mode),
+        }
+    }
+
+    /// `account filters`: flip one account's `filters_enabled` and persist it
+    /// through the daemon's `UpdateAccount` RPC.
+    ///
+    /// The account is read back first so the update carries its real current
+    /// configuration; sending a locally-invented config would silently reset
+    /// every field the CLI did not know about.
+    async fn handle_account_filters(
+        &self,
+        id: &str,
+        enable: bool,
+        disable: bool,
+        json_mode: bool,
+    ) -> String {
+        if enable == disable {
+            return Self::render_error(
+                "exactly one of --enable or --disable must be specified",
+                json_mode,
+            );
+        }
+
+        let mut client = match self.connect_accounts_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        let accounts = match client
+            .list_accounts(nuncio_proto::v1::ListAccountsRequest {})
+            .await
+        {
+            Ok(response) => response.into_inner().accounts,
+            Err(status) => return Self::render_status_error("list_accounts", &status, json_mode),
+        };
+        let Some(mut config) = accounts.into_iter().find(|a| a.id == id) else {
+            return Self::render_error(&format!("account '{id}' not found"), json_mode);
+        };
+        config.filters_enabled = enable;
+
+        match client
+            .update_account(nuncio_proto::v1::UpdateAccountRequest {
+                config: Some(config),
+                password: None,
+            })
+            .await
+        {
+            Ok(_) => {
+                if json_mode {
+                    format_json(&json!({
+                        "status": "updated",
+                        "id": id,
+                        "filters_enabled": enable,
+                    }))
+                } else {
+                    format!(
+                        "Filters {} for account '{id}'",
+                        if enable { "enabled" } else { "disabled" }
+                    )
+                }
+            }
+            Err(status) => Self::render_status_error("update_account", &status, json_mode),
+        }
+    }
+
     /// `mail export`: a real thin gRPC client of the running `nunciod`
     /// daemon's `nuncio.v1.Export` API. The
     /// daemon writes the export to `out` on ITS OWN host filesystem (the
@@ -1826,6 +2095,9 @@ impl HeadlessRunner {
             keyring_secret_key: keyring_key.clone(),
             sync_interval: Some(nuncio_proto::time::duration_from_secs(300)),
             transport: Some(transport),
+            // Filters take irreversible remote actions, so adding an account
+            // never opts it in; `account filters` does, deliberately.
+            filters_enabled: false,
         };
 
         let mut client = match self.connect_accounts_client().await {
@@ -3147,6 +3419,7 @@ mod tests {
                         email_address: "stub@nuncio.mx".to_string(),
                         keyring_secret_key: "nuncio/acct-stub-1".to_string(),
                         sync_interval: Some(nuncio_proto::time::duration_from_secs(300)),
+                        filters_enabled: false,
                         transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp(
                             nuncio_proto::v1::ImapSmtpTransport {
                                 imap_host: "imap.nuncio.mx".to_string(),
@@ -3371,6 +3644,13 @@ mod tests {
                 body_plain: Some("Stub body text".to_string()),
                 body_html: None,
                 attachments: Vec::new(),
+                placements: vec![nuncio_proto::v1::Placement {
+                    account_id: "acct-stub-1".to_string(),
+                    folder_id: "inbox".to_string(),
+                    uidvalidity: "1".to_string(),
+                    uid: "42".to_string(),
+                    read: false,
+                }],
             }
         }
 
@@ -3383,10 +3663,77 @@ mod tests {
             last_mark_read: Arc<Mutex<Option<MarkReadRequest>>>,
             last_send_message: Arc<Mutex<Option<SendMessageRequest>>>,
             last_sync: Arc<Mutex<Option<SyncRequest>>>,
+            last_move: Arc<Mutex<Option<nuncio_proto::v1::MoveMessageRequest>>>,
+            last_delete: Arc<Mutex<Option<nuncio_proto::v1::DeleteMessageRequest>>>,
+            last_flag: Arc<Mutex<Option<nuncio_proto::v1::FlagMessageRequest>>>,
         }
 
         #[tonic::async_trait]
         impl MailService for StubMail {
+            async fn move_message(
+                &self,
+                request: tonic::Request<nuncio_proto::v1::MoveMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::MoveMessageResponse>, tonic::Status>
+            {
+                *self.last_move.lock().unwrap_or_else(|e| e.into_inner()) =
+                    Some(request.into_inner());
+                Ok(tonic::Response::new(
+                    nuncio_proto::v1::MoveMessageResponse {
+                        mutation_id: "mut-stub-1".to_string(),
+                    },
+                ))
+            }
+
+            async fn delete_message(
+                &self,
+                request: tonic::Request<nuncio_proto::v1::DeleteMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::DeleteMessageResponse>, tonic::Status>
+            {
+                *self.last_delete.lock().unwrap_or_else(|e| e.into_inner()) =
+                    Some(request.into_inner());
+                Ok(tonic::Response::new(
+                    nuncio_proto::v1::DeleteMessageResponse {
+                        mutation_id: "mut-stub-2".to_string(),
+                    },
+                ))
+            }
+
+            async fn flag_message(
+                &self,
+                request: tonic::Request<nuncio_proto::v1::FlagMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::FlagMessageResponse>, tonic::Status>
+            {
+                *self.last_flag.lock().unwrap_or_else(|e| e.into_inner()) =
+                    Some(request.into_inner());
+                Ok(tonic::Response::new(
+                    nuncio_proto::v1::FlagMessageResponse {
+                        mutation_id: "mut-stub-3".to_string(),
+                    },
+                ))
+            }
+
+            async fn list_conflicts(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ListConflictsRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ListConflictsResponse>, tonic::Status>
+            {
+                Ok(tonic::Response::new(
+                    nuncio_proto::v1::ListConflictsResponse {
+                        conflicts: Vec::new(),
+                    },
+                ))
+            }
+
+            async fn resolve_conflict(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ResolveConflictRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ResolveConflictResponse>, tonic::Status>
+            {
+                Ok(tonic::Response::new(
+                    nuncio_proto::v1::ResolveConflictResponse {},
+                ))
+            }
+
             async fn list_folders(
                 &self,
                 _request: tonic::Request<ListFoldersRequest>,
@@ -3717,6 +4064,46 @@ mod tests {
 
         #[tonic::async_trait]
         impl MailService for StubMailAlwaysErrors {
+            async fn move_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::MoveMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::MoveMessageResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("mailbox index is corrupt"))
+            }
+
+            async fn delete_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::DeleteMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::DeleteMessageResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("mailbox index is corrupt"))
+            }
+
+            async fn flag_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::FlagMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::FlagMessageResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("mailbox index is corrupt"))
+            }
+
+            async fn list_conflicts(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ListConflictsRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ListConflictsResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("mailbox index is corrupt"))
+            }
+
+            async fn resolve_conflict(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ResolveConflictRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ResolveConflictResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("mailbox index is corrupt"))
+            }
+
             async fn list_folders(
                 &self,
                 _request: tonic::Request<ListFoldersRequest>,
@@ -3820,6 +4207,52 @@ mod tests {
 
         #[tonic::async_trait]
         impl MailService for StubMailTypedError {
+            /// The one typed-error path this stub exists to exercise on the
+            /// mutation surface: a genuine, server-evidenced conflict.
+            async fn move_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::MoveMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::MoveMessageResponse>, tonic::Status>
+            {
+                Err(nuncio_proto::errors::status_with_metadata(
+                    ErrorReason::Conflict,
+                    "message was moved by another client",
+                    [("message_id".to_string(), "msg-stub-1".to_string())],
+                ))
+            }
+
+            async fn delete_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::DeleteMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::DeleteMessageResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("unused in this test"))
+            }
+
+            async fn flag_message(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::FlagMessageRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::FlagMessageResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("unused in this test"))
+            }
+
+            async fn list_conflicts(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ListConflictsRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ListConflictsResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("unused in this test"))
+            }
+
+            async fn resolve_conflict(
+                &self,
+                _request: tonic::Request<nuncio_proto::v1::ResolveConflictRequest>,
+            ) -> Result<tonic::Response<nuncio_proto::v1::ResolveConflictResponse>, tonic::Status>
+            {
+                Err(tonic::Status::internal("unused in this test"))
+            }
+
             async fn list_folders(
                 &self,
                 _request: tonic::Request<ListFoldersRequest>,

crates/nuncio-core/src/model.rs

@@ -183,6 +183,38 @@ pub struct PlacementKey {
     pub remote_id: String,
 }
 
+/// A mutation the server refused because another client had already changed the
+/// message, recorded as durable state rather than as a passing notification.
+///
+/// A conflict is raised only on positive evidence from the server; an
+/// unverifiable outcome is never one. Because the intent was formed against
+/// state that no longer exists, a human has to decide what happens next, and
+/// that decision may come long after the event. The event stream is a lossy
+/// broadcast -- it drops on lag and carries no sequence number to resume from --
+/// so it can only ever be an optimisation over a queryable table, never the
+/// record itself.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MutationConflict {
+    /// Stable identifier of this conflict record.
+    pub id: String,
+    /// The outbox mutation that conflicted.
+    pub mutation_id: String,
+    /// Content-addressed key of the message the mutation targeted.
+    pub message_id: String,
+    /// The operation that was refused (`MOVE`, `DELETE`, `FLAG`, ...).
+    pub mutation_type: String,
+    /// The occupancy the mutation was aimed at, when the row recorded one.
+    pub placement: Option<PlacementKey>,
+    /// The server's positive evidence that another client changed the message.
+    pub observed: String,
+    /// Unix timestamp (seconds) when the conflict was detected.
+    pub detected_at: i64,
+    /// Unix timestamp (seconds) when a human resolved it, if they have.
+    pub resolved_at: Option<i64>,
+    /// Free-form note recording how the conflict was resolved.
+    pub resolution: Option<String>,
+}
+
 impl Placement {
     /// The addressing coordinates of this occupancy, dropping the read flag.
     pub fn key(&self) -> PlacementKey {

crates/nuncio-filter/src/outbox.rs

@@ -1,6 +1,7 @@
 //! Transactional Outbox Pattern for Remote IMAP/JMAP Mutations.
 
 use crate::ast::PendingRemoteMutation;
+use nuncio_core::model::PlacementKey;
 use serde::{Deserialize, Serialize};
 
 /// Payload details stored for pending remote mutation.
@@ -10,6 +11,17 @@ pub struct MutationPayload {
     pub action_type: String,
     /// Target parameter (e.g. folder name, target email address).
     pub target: Option<String>,
+    /// The mailbox occupancy this mutation was formed against.
+    ///
+    /// Message identity is content-addressed and folder-independent, so a single
+    /// message can occupy several mailboxes at once (Gmail labels make that
+    /// ordinary). `MOVE` and `DELETE` act on *one* occupancy, and acting on the
+    /// wrong one is destructive and irreversible, so the caller that formed the
+    /// intent records which occupancy it meant. `None` preserves the older
+    /// rows' meaning -- "unrecorded" -- which the executor may resolve only when
+    /// the message occupies exactly one mailbox.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub placement: Option<PlacementKey>,
 }
 
 /// Transactional outbox helper managing retry backoffs and mutation payloads.
@@ -31,17 +43,33 @@ impl OutboxManager {
         base + jitter
     }
 
-    /// Construct a new `PendingRemoteMutation` record.
+    /// Construct a new `PendingRemoteMutation` record whose target occupancy is
+    /// left unrecorded. The executor may still address it when the message
+    /// occupies exactly one mailbox, and refuses when it occupies several.
     pub fn create_mutation(
         rule_id: impl Into<String>,
         message_id: impl Into<String>,
         action_type: impl Into<String>,
         target: Option<String>,
+    ) -> PendingRemoteMutation {
+        Self::create_mutation_at(rule_id, message_id, action_type, target, None)
+    }
+
+    /// Construct a new `PendingRemoteMutation` record that names the mailbox
+    /// occupancy it was formed against, so a message occupying several mailboxes
+    /// is still unambiguously addressable.
+    pub fn create_mutation_at(
+        rule_id: impl Into<String>,
+        message_id: impl Into<String>,
+        action_type: impl Into<String>,
+        target: Option<String>,
+        placement: Option<PlacementKey>,
     ) -> PendingRemoteMutation {
         let action_str = action_type.into();
         let payload_struct = MutationPayload {
             action_type: action_str.clone(),
             target,
+            placement,
         };
         let payload_json = serde_json::to_string(&payload_struct).unwrap_or_default();
         let now = chrono::Utc::now().timestamp();

crates/nuncio-proto/proto/nuncio/v1/nuncio.proto

@@ -45,6 +45,14 @@ enum ErrorReason {
   ERROR_REASON_UNSUPPORTED = 11;
   // An unexpected internal error the client cannot act on specifically.
   ERROR_REASON_INTERNAL = 12;
+  // Another client changed the target first and the server said so.
+  //
+  // Raised only on positive evidence; an outcome the server left unverifiable
+  // is never reported as a conflict. Retrying is not the remedy -- the intent
+  // was formed against state that no longer exists, so re-issuing it would
+  // either do nothing or overwrite a change the caller never saw. The conflict
+  // is also recorded durably and is retrievable via `Mail.ListConflicts`.
+  ERROR_REASON_CONFLICT = 13;
 }
 
 // ErrorInfo is the structured error payload the daemon attaches to a failed
@@ -378,6 +386,11 @@ message AccountConfig {
     JmapTransport jmap = 16;
     DavTransport dav = 17;
   }
+  // Whether the filter engine runs against this account's incoming mail.
+  //
+  // Defaults to false: filters take irreversible remote actions, so an account
+  // does not get them by being added, only by being told to.
+  bool filters_enabled = 18;
 }
 
 // AddAccountRequest carries the full account configuration plus the raw
@@ -478,6 +491,24 @@ service Mail {
   rpc GetMessage(GetMessageRequest) returns (GetMessageResponse);
   // MarkRead updates a message's read/unread flag.
   rpc MarkRead(MarkReadRequest) returns (MarkReadResponse);
+  // MoveMessage enqueues a verified remote move of one occupancy of a message
+  // into another folder. Returns as soon as the intent is durably queued; the
+  // outbox executes it against the server and only records success on positive
+  // proof.
+  rpc MoveMessage(MoveMessageRequest) returns (MoveMessageResponse);
+  // DeleteMessage enqueues a verified remote delete of one occupancy of a
+  // message. Other occupancies of the same message are untouched.
+  rpc DeleteMessage(DeleteMessageRequest) returns (DeleteMessageResponse);
+  // FlagMessage enqueues a verified remote flag/unflag of one occupancy of a
+  // message.
+  rpc FlagMessage(FlagMessageRequest) returns (FlagMessageResponse);
+  // ListConflicts returns mutations the server refused because another client
+  // had already changed the message. Conflicts are durable state, not a
+  // notification: the event stream is a lossy broadcast, so this table is the
+  // source of truth a client reconciles against.
+  rpc ListConflicts(ListConflictsRequest) returns (ListConflictsResponse);
+  // ResolveConflict marks a conflict handled, recording the human's decision.
+  rpc ResolveConflict(ResolveConflictRequest) returns (ResolveConflictResponse);
   // SearchMessages runs a full-text (FTS5) search over cached message
   // subjects, senders, and bodies, returning a keyset-paginated page of hits
   // ordered by relevance rank.
@@ -542,8 +573,8 @@ message Message {
   //
   // A message can occupy several folders at once. This names the one the
   // message was reached through: the requested folder for a listing, and the
-  // lowest-sorting folder id for a lookup by id. A field carrying the complete
-  // placement set is not yet part of this contract.
+  // lowest-sorting folder id for a lookup by id. See `placements` for the
+  // complete set of mailboxes this message occupies.
   string folder_id = 3;
   // Subject line.
   string subject = 4;
@@ -561,6 +592,48 @@ message Message {
   optional string body_html = 10;
   // List of attached files.
   repeated Attachment attachments = 11;
+  // Every mailbox this message occupies, folder id ascending.
+  //
+  // Message identity is content-addressed and folder-independent, so one
+  // message can sit in several mailboxes at once -- Gmail labels make that
+  // ordinary rather than exotic. Mutations that act on a single occupancy
+  // (`MoveMessage`, `DeleteMessage`) need one of these to address, and a client
+  // that only ever saw `folder_id` cannot tell a one-mailbox message from a
+  // ten-mailbox one.
+  repeated Placement placements = 12;
+}
+
+// A single mailbox occupancy of a message: where one copy of it lives, and how
+// to address that copy on the server that holds it.
+message Placement {
+  // Account owning the mailbox.
+  string account_id = 1;
+  // Mailbox folder identifier.
+  string folder_id = 2;
+  // The UIDVALIDITY generation `uid` was captured under. Addressing is only
+  // valid within this generation; the server rejects a stale one rather than
+  // acting on whatever now bears that uid.
+  string uidvalidity = 3;
+  // Protocol-native id addressing the message in this mailbox.
+  string uid = 4;
+  // Read/unread flag of this occupancy specifically.
+  bool read = 5;
+}
+
+// Names one occupancy for a mutation to act on.
+//
+// All four coordinates are required. This is deliberately not "the folder" --
+// UIDVALIDITY scopes the uid, and dropping it would let a mutation formed
+// before a mailbox was recreated land on an unrelated message.
+message PlacementRef {
+  // Account owning the mailbox.
+  string account_id = 1;
+  // Mailbox folder identifier.
+  string folder_id = 2;
+  // The UIDVALIDITY generation `uid` was captured under.
+  string uidvalidity = 3;
+  // Protocol-native id addressing the message in this mailbox.
+  string uid = 4;
 }
 
 // Mirrors `nuncio_core::model::Folder`.
@@ -647,6 +720,106 @@ message MarkReadRequest {
 // response (as opposed to a `Status` error) is confirmation enough.
 message MarkReadResponse {}
 
+// Request to move one occupancy of a message into another folder.
+message MoveMessageRequest {
+  // Content-addressed key of the message to move.
+  string message_id = 1;
+  // Which occupancy to move. Optional only when the message occupies exactly
+  // one mailbox; with several, omitting it is refused rather than guessed,
+  // because moving the wrong copy is destructive and has no undo.
+  optional PlacementRef placement = 2;
+  // Destination folder id, within the same account as `placement`.
+  string destination_folder_id = 3;
+}
+
+// Result of enqueuing a move.
+message MoveMessageResponse {
+  // Identifier of the queued outbox mutation, for correlating with conflicts.
+  string mutation_id = 1;
+}
+
+// Request to delete one occupancy of a message.
+message DeleteMessageRequest {
+  // Content-addressed key of the message to delete.
+  string message_id = 1;
+  // Which occupancy to delete. Optional only when the message occupies exactly
+  // one mailbox. Deleting removes that copy; other occupancies survive.
+  optional PlacementRef placement = 2;
+}
+
+// Result of enqueuing a delete.
+message DeleteMessageResponse {
+  // Identifier of the queued outbox mutation.
+  string mutation_id = 1;
+}
+
+// Request to flag or unflag one occupancy of a message.
+message FlagMessageRequest {
+  // Content-addressed key of the message to flag.
+  string message_id = 1;
+  // Which occupancy to flag. Optional only when the message occupies exactly
+  // one mailbox.
+  optional PlacementRef placement = 2;
+  // Desired end state: true flags, false unflags.
+  bool flagged = 3;
+}
+
+// Result of enqueuing a flag change.
+message FlagMessageResponse {
+  // Identifier of the queued outbox mutation.
+  string mutation_id = 1;
+}
+
+// A mutation the server refused because another client changed the message
+// first, recorded durably so the decision can be made later.
+message MutationConflict {
+  // Stable identifier of this conflict record.
+  string id = 1;
+  // The outbox mutation that conflicted.
+  string mutation_id = 2;
+  // Content-addressed key of the message the mutation targeted.
+  string message_id = 3;
+  // The operation that was refused (e.g. "MOVE", "DELETE", "FLAG").
+  string mutation_type = 4;
+  // The occupancy the mutation was aimed at, when the queued intent named one.
+  optional PlacementRef placement = 5;
+  // The server's positive evidence that another client changed the message.
+  // A conflict is never inferred from an unverifiable outcome.
+  string observed = 6;
+  // When the conflict was detected.
+  google.protobuf.Timestamp detected_at = 7;
+  // When a human resolved it, if they have.
+  optional google.protobuf.Timestamp resolved_at = 8;
+  // Free-form note recording how it was resolved.
+  optional string resolution = 9;
+}
+
+// Request to list recorded mutation conflicts.
+message ListConflictsRequest {
+  // Include conflicts a human has already resolved. Resolved rows are kept as
+  // the audit trail of what was decided, so they are opt-in.
+  bool include_resolved = 1;
+  // Maximum number of conflicts to return; clamped to the server's page limit.
+  uint32 limit = 2;
+}
+
+// A page of mutation conflicts, newest first.
+message ListConflictsResponse {
+  // The conflicts.
+  repeated MutationConflict conflicts = 1;
+}
+
+// Request to mark a conflict handled.
+message ResolveConflictRequest {
+  // Identifier of the conflict to resolve.
+  string conflict_id = 1;
+  // Note recording how it was resolved.
+  string resolution = 2;
+}
+
+// Result of resolving a conflict.
+message ResolveConflictResponse {}
+
 // SearchMessagesRequest. Pages hits using the shared keyset convention (see
 // `ListMessagesRequest`); the cursor here encodes `(rank, id)` so paging
 // preserves the relevance ordering while resuming strictly after the last

crates/nuncio-proto/src/errors.rs

@@ -37,6 +37,11 @@ impl ErrorReason {
             ErrorReason::BackendUnreachable => Code::Unavailable,
             ErrorReason::AuthRequired => Code::Unauthenticated,
             ErrorReason::Unsupported => Code::Unimplemented,
+            // `Aborted` is the canonical code for a concurrency conflict, and
+            // unlike `FailedPrecondition` it tells a client the call may be
+            // retried only after re-reading the current state -- which is
+            // exactly the remedy here.
+            ErrorReason::Conflict => Code::Aborted,
         }
     }
 }

crates/nuncio-store/src/db.rs

@@ -11,6 +11,64 @@ use zeroize::Zeroizing;
 /// Raw row shape for a pending remote mutation record fetched from SQLite.
 type PendingMutationRow = (String, String, String, String, String, String, i64, i64);
 
+/// `mutation_conflicts` row: id, mutation_id, message_id, mutation_type, then the
+/// four nullable placement coordinates, observed evidence, and resolution state.
+type ConflictRow = (
+    String,
+    String,
+    String,
+    String,
+    Option<String>,
+    Option<String>,
+    Option<String>,
+    Option<String>,
+    String,
+    i64,
+    Option<i64>,
+    Option<String>,
+);
+
+/// Rebuild a conflict from its row, reassembling the placement only when all
+/// four coordinates are present -- a partial placement addresses nothing.
+fn conflict_from_row(row: ConflictRow) -> nuncio_core::model::MutationConflict {
+    let (
+        id,
+        mutation_id,
+        message_id,
+        mutation_type,
+        account_id,
+        folder_id,
+        uid_validity,
+        remote_id,
+        observed,
+        detected_at,
+        resolved_at,
+        resolution,
+    ) = row;
+    let placement = match (account_id, folder_id, uid_validity, remote_id) {
+        (Some(account_id), Some(folder_id), Some(uid_validity), Some(remote_id)) => {
+            Some(nuncio_core::model::PlacementKey {
+                account_id,
+                folder_id,
+                uid_validity,
+                remote_id,
+            })
+        }
+        _ => None,
+    };
+    nuncio_core::model::MutationConflict {
+        id,
+        mutation_id,
+        message_id,
+        mutation_type,
+        placement,
+        observed,
+        detected_at,
+        resolved_at,
+        resolution,
+    }
+}
+
 /// Database errors emitted by `nuncio-store`.
 #[derive(Error, Debug)]
 pub enum DatabaseError {
@@ -919,6 +977,21 @@ impl DatabaseEngine {
                 created_at INTEGER NOT NULL
             );
 
+            CREATE TABLE IF NOT EXISTS mutation_conflicts (
+                id TEXT PRIMARY KEY NOT NULL,
+                mutation_id TEXT NOT NULL,
+                message_id TEXT NOT NULL,
+                mutation_type TEXT NOT NULL,
+                account_id TEXT,
+                folder_id TEXT,
+                uidvalidity TEXT,
+                uid TEXT,
+                observed TEXT NOT NULL,
+                detected_at INTEGER NOT NULL,
+                resolved_at INTEGER,
+                resolution TEXT
+            );
+
             CREATE TABLE IF NOT EXISTS filter_execution_logs (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 rule_id TEXT NOT NULL,
@@ -2670,6 +2743,103 @@ impl DatabaseEngine {
         Ok(())
     }
 
+    /// Record a mutation conflict as durable, queryable state.
+    ///
+    /// `INSERT OR REPLACE` keyed on the conflict id makes a re-record of the
+    /// same conflict idempotent, so a drain pass that is retried after a crash
+    /// cannot multiply one refusal into several rows a human has to dismiss.
+    pub async fn record_conflict(
+        &self,
+        conflict: &nuncio_core::model::MutationConflict,
+    ) -> Result<(), DatabaseError> {
+        let placement = conflict.placement.as_ref();
+        sqlx::query(
+            r#"
+            INSERT OR REPLACE INTO mutation_conflicts
+            (id, mutation_id, message_id, mutation_type, account_id, folder_id,
+             uidvalidity, uid, observed, detected_at, resolved_at, resolution)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            "#,
+        )
+        .bind(&conflict.id)
+        .bind(&conflict.mutation_id)
+        .bind(&conflict.message_id)
+        .bind(&conflict.mutation_type)
+        .bind(placement.map(|p| p.account_id.clone()))
+        .bind(placement.map(|p| p.folder_id.clone()))
+        .bind(placement.map(|p| p.uid_validity.clone()))
+        .bind(placement.map(|p| p.remote_id.clone()))
+        .bind(&conflict.observed)
+        .bind(conflict.detected_at)
+        .bind(conflict.resolved_at)
+        .bind(conflict.resolution.as_deref())
+        .execute(&self.pool)
+        .await
+        .map_err(DatabaseError::Query)?;
+        Ok(())
+    }
+
+    /// List recorded mutation conflicts, newest first.
+    ///
+    /// Resolved conflicts stay in the table -- they are the audit trail of what
+    /// a human decided -- so they are only returned when explicitly asked for.
+    pub async fn list_conflicts(
+        &self,
+        include_resolved: bool,
+        limit: usize,
+    ) -> Result<Vec<nuncio_core::model::MutationConflict>, DatabaseError> {
+        let sql = if include_resolved {
+            r#"
+            SELECT id, mutation_id, message_id, mutation_type, account_id, folder_id,
+                   uidvalidity, uid, observed, detected_at, resolved_at, resolution
+            FROM mutation_conflicts
+            ORDER BY detected_at DESC, id ASC
+            LIMIT ?
+            "#
+        } else {
+            r#"
+            SELECT id, mutation_id, message_id, mutation_type, account_id, folder_id,
+                   uidvalidity, uid, observed, detected_at, resolved_at, resolution
+            FROM mutation_conflicts
+            WHERE resolved_at IS NULL
+            ORDER BY detected_at DESC, id ASC
+            LIMIT ?
+            "#
+        };
+        let rows: Vec<ConflictRow> = sqlx::query_as(sql)
+            .bind(limit as i64)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(DatabaseError::Query)?;
+
+        Ok(rows.into_iter().map(conflict_from_row).collect())
+    }
+
+    /// Mark a conflict resolved. Returns `false` when no such unresolved
+    /// conflict exists, so a caller can tell "already handled" from "applied"
+    /// instead of reporting a success that never happened.
+    pub async fn resolve_conflict(
+        &self,
+        id: &str,
+        resolution: &str,
+        resolved_at: i64,
+    ) -> Result<bool, DatabaseError> {
+        let result = sqlx::query(
+            r#"
+            UPDATE mutation_conflicts
+            SET resolved_at = ?, resolution = ?
+            WHERE id = ? AND resolved_at IS NULL
+            "#,
+        )
+        .bind(resolved_at)
+        .bind(resolution)
+        .bind(id)
+        .execute(&self.pool)
+        .await
+        .map_err(DatabaseError::Query)?;
+        Ok(result.rows_affected() > 0)
+    }
+
     /// Save a [`nuncio_filter::PendingRemoteMutation`] outbox record.
     pub async fn save_pending_mutation(
         &self,

crates/nunciod/src/grpc.rs

@@ -34,27 +34,30 @@ use nuncio_proto::v1::{
     CalendarSyncRequest, CalendarSyncResponse, Contact as ContactProto,
     ContactEmail as ContactEmailProto, ContactPhone as ContactPhoneProto, ContactsSyncRequest,
     ContactsSyncResponse, CreateContactRequest, CreateContactResponse, CreateRuleRequest,
-    CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, DeleteRuleRequest,
-    DeleteRuleResponse, Event, EventError, ExportFormat as ExportFormatProto, ExportRequest,
-    ExportResponse, ExportRulesRequest, ExportRulesResponse, FilterExecuted,
-    FilterExecutionLog as FilterExecutionLogProto, FilterRule as FilterRuleProto,
-    Folder as FolderProto, GetContactRequest, GetContactResponse, GetEventRequest,
-    GetEventResponse, GetExecutionLogsRequest, GetExecutionLogsResponse, GetMessageRequest,
-    GetMessageResponse, GetStatusRequest, GetStatusResponse,
-    ImapSmtpTransport as ImapSmtpTransportProto, ImportRulesRequest, ImportRulesResponse,
-    JmapTransport as JmapTransportProto, ListAccountsRequest, ListAccountsResponse,
+    CreateRuleResponse, DatabaseRecovered, DavTransport as DavTransportProto, DeleteMessageRequest,
+    DeleteMessageResponse, DeleteRuleRequest, DeleteRuleResponse, Event, EventError,
+    ExportFormat as ExportFormatProto, ExportRequest, ExportResponse, ExportRulesRequest,
+    ExportRulesResponse, FilterExecuted, FilterExecutionLog as FilterExecutionLogProto,
+    FilterRule as FilterRuleProto, FlagMessageRequest, FlagMessageResponse, Folder as FolderProto,
+    GetContactRequest, GetContactResponse, GetEventRequest, GetEventResponse,
+    GetExecutionLogsRequest, GetExecutionLogsResponse, GetMessageRequest, GetMessageResponse,
+    GetStatusRequest, GetStatusResponse, ImapSmtpTransport as ImapSmtpTransportProto,
+    ImportRulesRequest, ImportRulesResponse, JmapTransport as JmapTransportProto,
+    ListAccountsRequest, ListAccountsResponse, ListConflictsRequest, ListConflictsResponse,
     ListContactsRequest, ListContactsResponse, ListEventsRequest, ListEventsResponse,
     ListFoldersRequest, ListFoldersResponse, ListMessagesRequest, ListMessagesResponse,
     ListRecordsRequest, ListRecordsResponse, ListRulesRequest, ListRulesResponse, MarkReadRequest,
     MarkReadResponse, Message as MessageProto, MessageFlagsChanged, MessageSearchHit,
-    PreviewRuleRequest, PreviewRuleResponse, RemoveAccountRequest, RemoveAccountResponse,
-    RuleExportFormat as RuleExportFormatProto, SearchMessagesRequest, SearchMessagesResponse,
-    SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted,
-    SyncRequest, SyncResponse, SyncStarted, SyncState as SyncStateProto,
-    TestAccountConnectionRequest, TestAccountConnectionResponse, TlsMode as TlsModeProto,
-    TriageProgress, TriageRequest, UpdateAccountRequest, UpdateAccountResponse, UpdateAvailable,
-    UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse,
-    VerifyChainRequest, VerifyChainResponse,
+    MoveMessageRequest, MoveMessageResponse, MutationConflict as MutationConflictProto,
+    Placement as PlacementProto, PlacementRef as PlacementRefProto, PreviewRuleRequest,
+    PreviewRuleResponse, RemoveAccountRequest, RemoveAccountResponse, ResolveConflictRequest,
+    ResolveConflictResponse, RuleExportFormat as RuleExportFormatProto, SearchMessagesRequest,
+    SearchMessagesResponse, SendMessageRequest, SendMessageResponse, ShuttingDown,
+    SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse, SyncStarted,
+    SyncState as SyncStateProto, TestAccountConnectionRequest, TestAccountConnectionResponse,
+    TlsMode as TlsModeProto, TriageProgress, TriageRequest, UpdateAccountRequest,
+    UpdateAccountResponse, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse,
+    ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse,
 };
 use nuncio_store::db::DatabaseEngine;
 use nuncio_store::search::SearchEngine;
@@ -429,6 +432,7 @@ fn map_account_config_to_proto(config: nuncio_core::AccountConfig) -> AccountCon
             config.sync_interval_secs,
         )),
         transport: Some(transport),
+        filters_enabled: config.filters_enabled,
     }
 }
 
@@ -490,7 +494,7 @@ fn map_account_config_from_proto(
         email_address: config.email_address,
         keyring_secret_key: config.keyring_secret_key,
         sync_interval_secs: nuncio_proto::time::duration_to_secs(&sync_interval),
-        filters_enabled: false,
+        filters_enabled: config.filters_enabled,
         transport,
     })
 }
@@ -993,6 +997,54 @@ fn map_email_to_proto(placed: nuncio_store::db::MessageWithPlacement) -> Message
             .into_iter()
             .map(map_attachment_to_proto)
             .collect(),
+        // Filled in by `fill_placements`, which needs the store.
+        placements: Vec::new(),
+    }
+}
+
+/// Maps one stored occupancy onto its wire-format `nuncio.v1.Placement`.
+fn map_placement_to_proto(placement: nuncio_core::model::Placement) -> PlacementProto {
+    PlacementProto {
+        account_id: placement.account_id,
+        folder_id: placement.folder_id,
+        uidvalidity: placement.uid_validity,
+        uid: placement.remote_id,
+        read: placement.read,
+    }
+}
+
+/// Maps a wire-format `nuncio.v1.PlacementRef` onto the store's key type.
+fn map_placement_ref_from_proto(placement: PlacementRefProto) -> nuncio_core::model::PlacementKey {
+    nuncio_core::model::PlacementKey {
+        account_id: placement.account_id,
+        folder_id: placement.folder_id,
+        uid_validity: placement.uidvalidity,
+        remote_id: placement.uid,
+    }
+}
+
+/// Fills each message's complete placement set from the store.
+///
+/// This costs one query per message rather than one per page. That is the
+/// honest price of the field: occupancy is per-message state and a page can mix
+/// messages that sit in one mailbox with messages that sit in ten. A failure to
+/// read one message's placements leaves that message's set empty rather than
+/// failing the whole listing, and is logged -- a partial page is more useful
+/// than none, and the field is additive so a client that ignores it is
+/// unaffected.
+async fn fill_placements(db: &DatabaseEngine, messages: &mut [MessageProto]) {
+    for message in messages.iter_mut() {
+        match db.placements_of(&message.id).await {
+            Ok(placements) => {
+                message.placements = placements.into_iter().map(map_placement_to_proto).collect();
+            }
+            Err(e) => {
+                tracing::warn!(
+                    message_id = %message.id,
+                    "Mail: failed to read the placement set: {e}"
+                );
+            }
+        }
     }
 }
 
@@ -1786,6 +1838,145 @@ async fn primary_placement(
     })
 }
 
+/// Rule id recorded on outbox rows a human queued directly rather than a filter.
+///
+/// The outbox row's `rule_id` explains *why* a mutation exists; for a
+/// hand-driven one the honest answer is "a person asked", not a rule that does
+/// not exist.
+const MANUAL_ORIGIN: &str = "manual";
+
+/// Seconds since the Unix epoch, saturating at 0 before it.
+fn unix_now_secs() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or_default()
+}
+
+/// Resolve which occupancy a mutation should act on.
+///
+/// A named placement is honoured only after confirming the message still
+/// occupies it, so a client working from a stale listing cannot aim a `MOVE` or
+/// `DELETE` at a mailbox the message has left. An unnamed placement is inferred
+/// only when the message occupies exactly one mailbox; with several, this is a
+/// typed validation failure that lists the candidates, because picking one
+/// would act on mail the caller never selected and there is no undo.
+async fn resolve_target_placement(
+    db: &DatabaseEngine,
+    message_id: &str,
+    named: Option<PlacementRefProto>,
+) -> Result<nuncio_core::model::PlacementKey, Status> {
+    if message_id.is_empty() {
+        return Err(Status::invalid_argument("message_id is required"));
+    }
+    let placements = db.placements_of(message_id).await.map_err(|e| {
+        Status::internal(format!("failed to read placements for '{message_id}': {e}"))
+    })?;
+
+    if let Some(named) = named {
+        let key = map_placement_ref_from_proto(named);
+        return if placements.iter().any(|p| p.key() == key) {
+            Ok(key)
+        } else {
+            Err(errors::status_with_metadata(
+                ErrorReason::MessageNotFound,
+                format!(
+                    "message '{message_id}' does not occupy {}/{} (uidvalidity {}, uid {})",
+                    key.account_id, key.folder_id, key.uid_validity, key.remote_id
+                ),
+                [("message_id".to_string(), message_id.to_string())],
+            ))
+        };
+    }
+
+    match placements.as_slice() {
+        [only] => Ok(only.key()),
+        [] => Err(errors::status_with_metadata(
+            ErrorReason::MessageNotFound,
+            format!("message '{message_id}' occupies no mailbox"),
+            [("message_id".to_string(), message_id.to_string())],
+        )),
+        several => {
+            let folders: Vec<&str> = several.iter().map(|p| p.folder_id.as_str()).collect();
+            Err(errors::status_with_metadata(
+                ErrorReason::ValidationFailed,
+                format!(
+                    "message '{message_id}' occupies {} mailboxes ({}); name one in `placement` \
+                     rather than leaving the target to be guessed",
+                    several.len(),
+                    folders.join(", ")
+                ),
+                [
+                    ("message_id".to_string(), message_id.to_string()),
+                    ("placement_count".to_string(), several.len().to_string()),
+                ],
+            ))
+        }
+    }
+}
+
+/// Maps a stored conflict onto its wire-format `nuncio.v1.MutationConflict`.
+fn map_conflict_to_proto(conflict: nuncio_core::model::MutationConflict) -> MutationConflictProto {
+    MutationConflictProto {
+        id: conflict.id,
+        mutation_id: conflict.mutation_id,
+        message_id: conflict.message_id,
+        mutation_type: conflict.mutation_type,
+        placement: conflict.placement.map(|p| PlacementRefProto {
+            account_id: p.account_id,
+            folder_id: p.folder_id,
+            uidvalidity: p.uid_validity,
+            uid: p.remote_id,
+        }),
+        observed: conflict.observed,
+        detected_at: Some(nuncio_proto::time::timestamp_from_unix_secs(
+            conflict.detected_at,
+        )),
+        resolved_at: conflict
+            .resolved_at
+            .map(nuncio_proto::time::timestamp_from_unix_secs),
+        resolution: conflict.resolution,
+    }
+}
+
+impl MailGrpcService {
+    /// Durably queue a placement-addressed remote mutation on the outbox.
+    ///
+    /// Every mutation RPC funnels through here rather than touching a backend
+    /// directly: the outbox is what makes an intent survive a crash, and what
+    /// turns the server's answer into one of three honest outcomes instead of a
+    /// hopeful assumption. The RPC returns once the intent is durable, which is
+    /// the only thing it can truthfully promise.
+    async fn enqueue_mutation(
+        &self,
+        message_id: &str,
+        op: &str,
+        target: Option<String>,
+        placement: nuncio_core::model::PlacementKey,
+    ) -> Result<String, Status> {
+        let mutation = nuncio_filter::OutboxManager::create_mutation_at(
+            MANUAL_ORIGIN,
+            message_id,
+            op,
+            target,
+            Some(placement.clone()),
+        );
+        self.db
+            .save_pending_mutation(&mutation)
+            .await
+            .map_err(|e| Status::internal(format!("failed to queue {op} mutation: {e}")))?;
+        tracing::info!(
+            mutation_id = %mutation.id,
+            op = %op,
+            message_id = %message_id,
+            account_id = %placement.account_id,
+            folder_id = %placement.folder_id,
+            "Mail: queued placement-addressed mutation"
+        );
+        Ok(mutation.id)
+    }
+}
+
 #[tonic::async_trait]
 impl Mail for MailGrpcService {
     /// Lists folders across every **configured** account, merged by folder id.
@@ -1870,7 +2061,9 @@ impl Mail for MailGrpcService {
                 ])
             })
             .unwrap_or_default();
-        let messages = messages.into_iter().map(map_email_to_proto).collect();
+        let mut messages: Vec<MessageProto> =
+            messages.into_iter().map(map_email_to_proto).collect();
+        fill_placements(&self.db, &mut messages).await;
 
         Ok(Response::new(ListMessagesResponse {
             messages,
@@ -1897,8 +2090,23 @@ impl Mail for MailGrpcService {
 
         let placement = primary_placement(&self.db, &req.message_id).await?;
 
+        let mut message = map_email_to_proto((email, placement));
+        message.placements = self
+            .db
+            .placements_of(&req.message_id)
+            .await
+            .map_err(|e| {
+                Status::internal(format!(
+                    "failed to read placements for '{}': {e}",
+                    req.message_id
+                ))
+            })?
+            .into_iter()
+            .map(map_placement_to_proto)
+            .collect();
+
         Ok(Response::new(GetMessageResponse {
-            message: Some(map_email_to_proto((email, placement))),
+            message: Some(message),
         }))
     }
 
@@ -1949,6 +2157,97 @@ impl Mail for MailGrpcService {
         Ok(Response::new(MarkReadResponse {}))
     }
 
+    async fn move_message(
+        &self,
+        request: Request<MoveMessageRequest>,
+    ) -> Result<Response<MoveMessageResponse>, Status> {
+        let req = request.into_inner();
+        if req.destination_folder_id.is_empty() {
+            return Err(Status::invalid_argument(
+                "destination_folder_id is required",
+            ));
+        }
+        let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?;
+        let mutation_id = self
+            .enqueue_mutation(
+                &req.message_id,
+                "MOVE",
+                Some(req.destination_folder_id.clone()),
+                placement,
+            )
+            .await?;
+        Ok(Response::new(MoveMessageResponse { mutation_id }))
+    }
+
+    async fn delete_message(
+        &self,
+        request: Request<DeleteMessageRequest>,
+    ) -> Result<Response<DeleteMessageResponse>, Status> {
+        let req = request.into_inner();
+        let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?;
+        let mutation_id = self
+            .enqueue_mutation(&req.message_id, "DELETE", None, placement)
+            .await?;
+        Ok(Response::new(DeleteMessageResponse { mutation_id }))
+    }
+
+    async fn flag_message(
+        &self,
+        request: Request<FlagMessageRequest>,
+    ) -> Result<Response<FlagMessageResponse>, Status> {
+        let req = request.into_inner();
+        let placement = resolve_target_placement(&self.db, &req.message_id, req.placement).await?;
+        let op = if req.flagged { "FLAG" } else { "UNFLAG" };
+        let mutation_id = self
+            .enqueue_mutation(&req.message_id, op, None, placement)
+            .await?;
+        Ok(Response::new(FlagMessageResponse { mutation_id }))
+    }
+
+    async fn list_conflicts(
+        &self,
+        request: Request<ListConflictsRequest>,
+    ) -> Result<Response<ListConflictsResponse>, Status> {
+        let req = request.into_inner();
+        let limit = pagination::clamp_page_size(req.limit);
+        let conflicts = self
+            .db
+            .list_conflicts(req.include_resolved, limit)
+            .await
+            .map_err(|e| Status::internal(format!("failed to list conflicts: {e}")))?;
+        Ok(Response::new(ListConflictsResponse {
+            conflicts: conflicts.into_iter().map(map_conflict_to_proto).collect(),
+        }))
+    }
+
+    async fn resolve_conflict(
+        &self,
+        request: Request<ResolveConflictRequest>,
+    ) -> Result<Response<ResolveConflictResponse>, Status> {
+        let req = request.into_inner();
+        if req.conflict_id.is_empty() {
+            return Err(Status::invalid_argument("conflict_id is required"));
+        }
+        let resolved_at = unix_now_secs();
+        let applied = self
+            .db
+            .resolve_conflict(&req.conflict_id, &req.resolution, resolved_at)
+            .await
+            .map_err(|e| Status::internal(format!("failed to resolve conflict: {e}")))?;
+        if !applied {
+            // Either it never existed or someone already resolved it. Reporting
+            // success would tell the caller their decision was recorded when it
+            // was not.
+            return Err(errors::status_with_metadata(
+                ErrorReason::MessageNotFound,
+                format!("no unresolved conflict '{}' to resolve", req.conflict_id),
+                [("conflict_id".to_string(), req.conflict_id.clone())],
+            ));
+        }
+        tracing::info!(conflict_id = %req.conflict_id, "Mail: conflict resolved");
+        Ok(Response::new(ResolveConflictResponse {}))
+    }
+
     async fn search_messages(
         &self,
         request: Request<SearchMessagesRequest>,
@@ -4298,6 +4597,7 @@ mod tests {
             email_address: format!("{id}@nuncio.mx"),
             keyring_secret_key: keyring_secret_key.to_string(),
             sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+            filters_enabled: false,
             transport: Some(TransportProto::ImapSmtp(ImapSmtpTransportProto {
                 imap_host: "imap.nuncio.mx".to_string(),
                 imap_port: 993,
@@ -4317,6 +4617,7 @@ mod tests {
             email_address: "no-transport@nuncio.mx".to_string(),
             keyring_secret_key: "nuncio/acct-no-transport".to_string(),
             sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+            filters_enabled: false,
             transport: None,
         };
         let err = map_account_config_from_proto(config)

crates/nunciod/src/outbox.rs

@@ -222,6 +222,7 @@ pub async fn execute_pending_mutations(
                     message_id = %item.message_id,
                     "outbox: mutation conflicted with another client: {observed}"
                 );
+                record_conflict(db, &item, &observed).await;
                 ("conflicted", item.retry_count, |s| s.failed += 1)
             }
             Disposition::Permanent(reason) => {
@@ -260,6 +261,35 @@ pub async fn execute_pending_mutations(
 
 /// Execute a single mutation and report its honest disposition. Performs no
 /// status writes itself -- the caller persists the outcome.
+/// Persist a conflict so it outlives the log line and the event stream.
+///
+/// The conflict id is derived from the mutation id, which makes re-recording
+/// the same refusal idempotent. A failure to persist is logged and swallowed:
+/// the mutation's own status write is what stops it being re-attempted, and
+/// losing the audit row must not take the drain pass down with it.
+async fn record_conflict(db: &DatabaseEngine, item: &PendingRemoteMutation, observed: &str) {
+    let placement = serde_json::from_str::<MutationPayload>(&item.payload)
+        .ok()
+        .and_then(|p| p.placement);
+    let conflict = nuncio_core::model::MutationConflict {
+        id: format!("conflict-{}", item.id),
+        mutation_id: item.id.clone(),
+        message_id: item.message_id.clone(),
+        mutation_type: item.mutation_type.clone(),
+        placement,
+        observed: observed.to_string(),
+        detected_at: std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs() as i64)
+            .unwrap_or_default(),
+        resolved_at: None,
+        resolution: None,
+    };
+    if let Err(e) = db.record_conflict(&conflict).await {
+        tracing::warn!(mutation_id = %item.id, "outbox: failed to record conflict: {e}");
+    }
+}
+
 async fn execute_one(
     db: &DatabaseEngine,
     env: &dyn RemoteExecutionEnv,
@@ -304,9 +334,12 @@ async fn execute_one(
 ///
 /// The addressing comes from a placement rather than the message row, because
 /// those coordinates are properties of a mailbox occupancy and a message can
-/// hold several. The outbox row records only the message key, so which occupancy
-/// the rule matched is not recoverable here -- and when the message occupies
-/// more than one mailbox, this **refuses** rather than picking one.
+/// hold several. When the row records the occupancy its intent was formed
+/// against, that occupancy is used verbatim -- after confirming the message
+/// still occupies it, so a mutation cannot be aimed at a mailbox the message has
+/// since left. Otherwise the occupancy is inferred, and inference is only
+/// allowed when it is unambiguous: with more than one placement this **refuses**
+/// rather than picking one.
 ///
 /// Guessing is not the conservative option. A rule that matched the `INBOX`
 /// copy would `MOVE` or `DELETE` whichever copy happened to sort first, against
@@ -317,9 +350,7 @@ async fn execute_one(
 /// guessing is destructive and irreversible.
 ///
 /// The refusal is permanent, not retryable: nothing about a later attempt makes
-/// the outbox row remember a placement it never recorded. Carrying the
-/// originating placement on the row is the real fix, and belongs with the
-/// placement-aware wire surface.
+/// a row remember a placement it never recorded.
 async fn dispatch_mailbox_mutation(
     db: &DatabaseEngine,
     env: &dyn RemoteExecutionEnv,
@@ -359,6 +390,27 @@ async fn dispatch_mailbox_mutation(
             return Disposition::Retry(format!("failed to resolve the message's placements: {e}"))
         }
     };
+    if let Some(named) = &payload.placement {
+        return match placements.iter().find(|p| &p.key() == named) {
+            Some(placement) => {
+                let placement = placement.clone();
+                apply_mailbox_mutation(env, mutation, email, &placement, kind).await
+            }
+            // The named occupancy is gone. Re-addressing to a different one
+            // would act on mail the caller never selected, so this is terminal.
+            None => Disposition::Permanent(format!(
+                "message '{}' no longer occupies {}/{} (uidvalidity {}, uid {}); \
+                 refusing to re-aim a {} at a different mailbox",
+                email.id,
+                named.account_id,
+                named.folder_id,
+                named.uid_validity,
+                named.remote_id,
+                mutation.mutation_type
+            )),
+        };
+    }
+
     let placement = match placements.as_slice() {
         [only] => only,
         [] => {
@@ -390,6 +442,18 @@ async fn dispatch_mailbox_mutation(
         }
     };
 
+    apply_mailbox_mutation(env, mutation, email, placement, kind).await
+}
+
+/// Issue an already-addressed mailbox mutation against the account's backend and
+/// translate the backend's three honest outcomes into a disposition.
+async fn apply_mailbox_mutation(
+    env: &dyn RemoteExecutionEnv,
+    mutation: &PendingRemoteMutation,
+    email: &nuncio_core::model::Email,
+    placement: &nuncio_core::model::Placement,
+    kind: RemoteMutationKind,
+) -> Disposition {
     let backend = match env.mail_backend(&placement.account_id).await {
         Ok(backend) => backend,
         Err(e) => return Disposition::Retry(format!("failed to build mail backend: {e}")),
@@ -723,6 +787,7 @@ mod tests {
             payload: serde_json::to_string(&MutationPayload {
                 action_type: "FLAG".to_string(),
                 target: None,
+                placement: None,
             })
             .expect("serialize payload"),
             status: "pending".to_string(),
@@ -826,6 +891,15 @@ mod tests {
     /// decided here anyway, so the narrower call is also the tighter assertion.
     async fn dispatch_with_placements(
         placements: &[Placement],
+    ) -> (Disposition, Vec<RemoteMutationSpec>) {
+        dispatch_with_placements_naming(placements, None).await
+    }
+
+    /// As [`dispatch_with_placements`], but with the queued row's payload
+    /// naming `named` as the occupancy the intent was formed against.
+    async fn dispatch_with_placements_naming(
+        placements: &[Placement],
+        named: Option<nuncio_core::model::PlacementKey>,
     ) -> (Disposition, Vec<RemoteMutationSpec>) {
         let (db, _dir) = DatabaseEngine::connect_ephemeral()
             .await
@@ -840,7 +914,15 @@ mod tests {
         let env = SucceedingEnv {
             backend: backend.clone(),
         };
-        let mutation = flag_mutation();
+        let mut mutation = flag_mutation();
+        if named.is_some() {
+            mutation.payload = serde_json::to_string(&MutationPayload {
+                action_type: mutation.mutation_type.clone(),
+                target: None,
+                placement: named,
+            })
+            .expect("serialize payload");
+        }
         let payload =
             serde_json::from_str::<MutationPayload>(&mutation.payload).expect("parse payload");
         let disposition =
@@ -900,4 +982,31 @@ mod tests {
             }
         }
     }
+
+    /// The same two-mailbox message stops being ambiguous once the queued row
+    /// records which occupancy the intent was formed against: there is nothing
+    /// left to guess, so the mutation is addressed at exactly that copy --
+    /// including when it is not the one a single-placement inference would have
+    /// reached.
+    #[tokio::test]
+    async fn a_named_placement_addresses_that_occupancy_rather_than_being_refused() {
+        let (disposition, applied) = dispatch_with_placements_naming(
+            &[sample_placement(), second_placement()],
+            Some(second_placement().key()),
+        )
+        .await;
+
+        assert!(
+            matches!(disposition, Disposition::Completed),
+            "a row that names its occupancy is fully addressed, not ambiguous"
+        );
+        assert_eq!(applied.len(), 1, "the backend really was asked to act");
+        assert_eq!(
+            applied[0].folder_id, "Archive",
+            "the named occupancy is the one acted on, not the other copy"
+        );
+        assert_eq!(applied[0].remote_id, "78");
+        assert_eq!(applied[0].uid_validity, "9");
+        assert_eq!(applied[0].message_id, "msg-outbox-log");
+    }
 }

crates/nunciod/tests/accounts_lifecycle_e2e_test.rs

@@ -90,6 +90,7 @@ fn sample_account_config() -> AccountConfig {
         email_address: ACCOUNT_EMAIL.to_string(),
         keyring_secret_key: KEYRING_KEY.to_string(),
         sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+        filters_enabled: false,
         transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp(
             nuncio_proto::v1::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),

crates/nunciod/tests/caldav_production_e2e_test.rs

@@ -159,6 +159,7 @@ async fn caldav_production_sync_builds_real_client_and_persists_events() {
                 email_address: ACCOUNT_EMAIL.to_string(),
                 keyring_secret_key: KEYRING_KEY.to_string(),
                 sync_interval: Some(nuncio_proto::time::duration_from_secs(300)),
+                filters_enabled: false,
                 transport: Some(nuncio_proto::v1::account_config::Transport::Dav(
                     nuncio_proto::v1::DavTransport {
                         collection_url: collection_url.clone(),

crates/nunciod/tests/mutation_rpcs_e2e_test.rs

@@ -0,0 +1,622 @@
+//! Offline end-to-end tests for the hand-driven mail mutation surface: the
+//! placement-addressed `MoveMessage`/`DeleteMessage`/`FlagMessage` RPCs, the
+//! durable conflict ledger behind `ListConflicts`/`ResolveConflict`, the
+//! complete placement set on `Message`, and the per-account filter switch.
+//!
+//! Each test boots a real `nunciod` daemon gRPC server
+//! (`nunciod::grpc::serve_on_listener_with_overrides`) on an ephemeral loopback
+//! port, backed by a real temp file-backed [`DatabaseEngine`] and a
+//! [`SecretManager::mock`] vault (never the real OS keyring), with every engine
+//! override left at its default. The store handle is shared with the server, so
+//! the assertions can seed state a sync pass would have written and read back
+//! the durable rows the RPCs produced -- which is the only thing enqueuing can
+//! truthfully promise, since nothing here reaches a remote server.
+//!
+//! Fully offline: no real network, no real OS keyring, no live server.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use nuncio_core::model::{Email, IdentitySource, MutationConflict, Placement, PlacementKey};
+use nuncio_core::EventBus;
+use nuncio_filter::{FilterEngine, MutationPayload};
+use nuncio_proto::errors;
+use nuncio_proto::v1::{
+    AccountConfig, AddAccountRequest, DeleteMessageRequest, ErrorReason, FlagMessageRequest,
+    GetMessageRequest, ListAccountsRequest, ListConflictsRequest, ListMessagesRequest,
+    MoveMessageRequest, PlacementRef, ResolveConflictRequest, UpdateAccountRequest,
+};
+use nuncio_store::db::DatabaseEngine;
+use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
+use nunciod::grpc::{
+    serve_on_listener_with_overrides, AccountsEngineOverrides, CalendarEngineOverrides,
+    ContactsEngineOverrides, MailEngineOverrides,
+};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tonic::Code;
+
+const ACCOUNT: &str = "acct-mutation-e2e";
+/// One UIDVALIDITY generation for every seeded occupancy: these tests are about
+/// which mailbox a mutation lands in, not about generation bumps.
+const UIDVALIDITY: &str = "7";
+
+/// Boots the real daemon gRPC server on an ephemeral loopback port with every
+/// override at its default.
+///
+/// Returns the dialable address, the minted bearer token, and the SAME store
+/// handle the server is using, so a test can seed rows and inspect what an RPC
+/// durably wrote. The `TempDir` guard is returned so the caller controls its
+/// lifetime for the whole test (the spawned server keeps using the db).
+async fn boot_daemon(db_name: &str) -> (String, String, Arc<DatabaseEngine>, tempfile::TempDir) {
+    let secrets = Arc::new(SecretManager::mock());
+    let dir = tempfile::tempdir().expect("create temp dir for file-backed db");
+    let db_path = dir.path().join(db_name);
+    let db = Arc::new(
+        DatabaseEngine::connect_file(&db_path, &secrets)
+            .await
+            .expect("open temp file-backed database"),
+    );
+
+    let token = hex::encode(
+        secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .expect("mint gRPC bearer token from mock vault"),
+    );
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind ephemeral loopback port");
+    let addr = listener.local_addr().expect("listener has local addr");
+
+    let event_bus = Arc::new(EventBus::new());
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+    let server_db = db.clone();
+    let server_token = token.clone();
+    tokio::spawn(async move {
+        let _ = serve_on_listener_with_overrides(
+            listener,
+            event_bus,
+            server_db,
+            filter_engine,
+            secrets,
+            server_token,
+            MailEngineOverrides::default(),
+            CalendarEngineOverrides::default(),
+            ContactsEngineOverrides::default(),
+            AccountsEngineOverrides::default(),
+        )
+        .await;
+    });
+
+    (addr.to_string(), token, db, dir)
+}
+
+/// One seeded mailbox occupancy of the test account.
+fn placement(folder_id: &str, uid: &str) -> Placement {
+    Placement {
+        account_id: ACCOUNT.to_string(),
+        folder_id: folder_id.to_string(),
+        uid_validity: UIDVALIDITY.to_string(),
+        remote_id: uid.to_string(),
+        read: false,
+    }
+}
+
+/// The wire form a client uses to name that same occupancy.
+fn placement_ref(folder_id: &str, uid: &str) -> PlacementRef {
+    PlacementRef {
+        account_id: ACCOUNT.to_string(),
+        folder_id: folder_id.to_string(),
+        uidvalidity: UIDVALIDITY.to_string(),
+        uid: uid.to_string(),
+    }
+}
+
+/// The store-side key the queued outbox row is expected to carry.
+fn placement_key(folder_id: &str, uid: &str) -> PlacementKey {
+    placement(folder_id, uid).key()
+}
+
+/// The account row the read RPCs merge over. `Mail` reads carry no account
+/// field, so mail belonging to no configured account is unreachable through
+/// them -- seeding one keeps these tests on the production path.
+async fn seed_account(db: &DatabaseEngine) {
+    db.save_account(&nuncio_core::AccountConfig {
+        id: ACCOUNT.to_string(),
+        name: "Mutation E2E".to_string(),
+        email_address: "mutation-e2e@nuncio.mx".to_string(),
+        keyring_secret_key: format!("nuncio/{ACCOUNT}"),
+        sync_interval_secs: 60,
+        filters_enabled: false,
+        transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport {
+            endpoint_host: "jmap.nuncio.mx".to_string(),
+        }),
+    })
+    .await
+    .expect("save the owning account");
+}
+
+/// Persist one message into each of the given occupancies, the way a sync pass
+/// over several mailboxes would.
+async fn seed_message(db: &DatabaseEngine, id: &str, placements: &[Placement]) {
+    let email = Email {
+        id: id.to_string(),
+        account_id: ACCOUNT.to_string(),
+        subject: format!("subject of {id}"),
+        sender: "alice@nuncio.mx".to_string(),
+        recipient: "owner@nuncio.mx".to_string(),
+        received_at: 1_700_000_000,
+        body_plain: Some(format!("body of {id}")),
+        body_html: None,
+        attachments: Vec::new(),
+        message_id: None,
+        content_hash: None,
+    };
+    for p in placements {
+        db.save_email_at(&email, IdentitySource::Surrogate, p)
+            .await
+            .expect("seed the occupancy");
+    }
+}
+
+/// The queued outbox row with this id, decoded into its operation and payload.
+async fn queued_mutation(db: &DatabaseEngine, mutation_id: &str) -> (String, MutationPayload) {
+    let pending = db
+        .list_pending_mutations(50)
+        .await
+        .expect("read the outbox queue");
+    let row = pending
+        .into_iter()
+        .find(|m| m.id == mutation_id)
+        .unwrap_or_else(|| panic!("no queued outbox row '{mutation_id}'"));
+    let payload =
+        serde_json::from_str::<MutationPayload>(&row.payload).expect("decode the queued payload");
+    assert_eq!(
+        row.rule_id, "manual",
+        "a hand-driven mutation is attributed to a person, not to a rule"
+    );
+    (row.mutation_type, payload)
+}
+
+/// A message in two mailboxes cannot be moved by naming only the message: one
+/// of the copies would be picked for the caller, and a misaimed MOVE is
+/// destructive with no undo. Naming the occupancy makes the same call succeed,
+/// and the queued row records which copy the intent was formed against.
+#[tokio::test]
+async fn moving_a_multi_placement_message_requires_the_occupancy_to_be_named() {
+    let (addr, token, db, _dir) = boot_daemon("mutation_move.db").await;
+    seed_account(&db).await;
+    seed_message(
+        &db,
+        "msg-multi",
+        &[placement("INBOX", "101"), placement("Archive", "202")],
+    )
+    .await;
+
+    let mut mail = nuncio_proto::client::connect_mail(&addr, &token)
+        .await
+        .expect("mail client connects");
+
+    // ---- unnamed: refused, and the refusal names the candidates ----
+    let status = mail
+        .move_message(MoveMessageRequest {
+            message_id: "msg-multi".to_string(),
+            placement: None,
+            destination_folder_id: "Trash".to_string(),
+        })
+        .await
+        .expect_err("an ambiguously-addressed move must be refused");
+    assert_eq!(status.code(), Code::InvalidArgument);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::ValidationFailed)
+    );
+    assert!(
+        status.message().contains("INBOX") && status.message().contains("Archive"),
+        "the refusal must name the candidate mailboxes: {}",
+        status.message()
+    );
+    assert!(
+        db.list_pending_mutations(50)
+            .await
+            .expect("read the outbox queue")
+            .is_empty(),
+        "a refused move must not leave a queued intent behind"
+    );
+
+    // ---- named: accepted, and durable ----
+    let mutation_id = mail
+        .move_message(MoveMessageRequest {
+            message_id: "msg-multi".to_string(),
+            placement: Some(placement_ref("INBOX", "101")),
+            destination_folder_id: "Trash".to_string(),
+        })
+        .await
+        .expect("naming the occupancy makes the move well-defined")
+        .into_inner()
+        .mutation_id;
+    assert!(
+        !mutation_id.is_empty(),
+        "the caller needs an id to correlate the queued intent with a conflict"
+    );
+
+    let (mutation_type, payload) = queued_mutation(&db, &mutation_id).await;
+    assert_eq!(mutation_type, "MOVE");
+    assert_eq!(payload.action_type, "MOVE");
+    assert_eq!(payload.target.as_deref(), Some("Trash"));
+    assert_eq!(
+        payload.placement,
+        Some(placement_key("INBOX", "101")),
+        "the queued row must remember the occupancy the intent was formed against"
+    );
+}
+
+/// Delete, flag and unflag queue the operation they were asked for, addressed
+/// at a specific occupancy: named explicitly when there are several to choose
+/// between, and inferred only when the message occupies exactly one mailbox.
+#[tokio::test]
+async fn delete_and_flag_queue_placement_addressed_rows() {
+    let (addr, token, db, _dir) = boot_daemon("mutation_delete_flag.db").await;
+    seed_account(&db).await;
+    seed_message(
+        &db,
+        "msg-multi",
+        &[placement("INBOX", "101"), placement("Archive", "202")],
+    )
+    .await;
+    seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await;
+
+    let mut mail = nuncio_proto::client::connect_mail(&addr, &token)
+        .await
+        .expect("mail client connects");
+
+    let delete_id = mail
+        .delete_message(DeleteMessageRequest {
+            message_id: "msg-multi".to_string(),
+            placement: Some(placement_ref("Archive", "202")),
+        })
+        .await
+        .expect("delete of a named occupancy is well-defined")
+        .into_inner()
+        .mutation_id;
+    let (mutation_type, payload) = queued_mutation(&db, &delete_id).await;
+    assert_eq!(mutation_type, "DELETE");
+    assert_eq!(payload.action_type, "DELETE");
+    assert_eq!(payload.target, None);
+    assert_eq!(payload.placement, Some(placement_key("Archive", "202")));
+
+    let flag_id = mail
+        .flag_message(FlagMessageRequest {
+            message_id: "msg-multi".to_string(),
+            placement: Some(placement_ref("INBOX", "101")),
+            flagged: true,
+        })
+        .await
+        .expect("flag of a named occupancy is well-defined")
+        .into_inner()
+        .mutation_id;
+    let (mutation_type, payload) = queued_mutation(&db, &flag_id).await;
+    assert_eq!(mutation_type, "FLAG");
+    assert_eq!(payload.action_type, "FLAG");
+    assert_eq!(payload.placement, Some(placement_key("INBOX", "101")));
+
+    // `flagged: false` is a distinct operation, not a FLAG with a parameter --
+    // the executor maps the two onto opposite remote calls.
+    let unflag_id = mail
+        .flag_message(FlagMessageRequest {
+            message_id: "msg-solo".to_string(),
+            placement: None,
+            flagged: false,
+        })
+        .await
+        .expect("a single-occupancy message needs no placement to be unambiguous")
+        .into_inner()
+        .mutation_id;
+    let (mutation_type, payload) = queued_mutation(&db, &unflag_id).await;
+    assert_eq!(mutation_type, "UNFLAG");
+    assert_eq!(payload.action_type, "UNFLAG");
+    assert_eq!(
+        payload.placement,
+        Some(placement_key("INBOX", "303")),
+        "the inferred occupancy is recorded too, so the executor never re-infers it"
+    );
+}
+
+/// Naming an occupancy the message does not hold is a not-found error, never a
+/// silent re-aim at whichever copy does exist: the caller formed its intent
+/// against a mailbox that is no longer true, and acting on a different one
+/// would touch mail it never selected.
+#[tokio::test]
+async fn naming_an_occupancy_the_message_does_not_hold_is_refused() {
+    let (addr, token, db, _dir) = boot_daemon("mutation_stale_placement.db").await;
+    seed_account(&db).await;
+    seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await;
+
+    let mut mail = nuncio_proto::client::connect_mail(&addr, &token)
+        .await
+        .expect("mail client connects");
+
+    let status = mail
+        .move_message(MoveMessageRequest {
+            message_id: "msg-solo".to_string(),
+            placement: Some(placement_ref("Spam", "999")),
+            destination_folder_id: "Trash".to_string(),
+        })
+        .await
+        .expect_err("a stale occupancy must not be honoured");
+    assert_eq!(status.code(), Code::NotFound);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::MessageNotFound)
+    );
+    assert!(
+        status.message().contains("Spam"),
+        "the error must name the occupancy that was asked for: {}",
+        status.message()
+    );
+
+    // A delete aimed at the wrong uid within a real folder is equally refused:
+    // the uid is part of the address, not a hint.
+    let status = mail
+        .delete_message(DeleteMessageRequest {
+            message_id: "msg-solo".to_string(),
+            placement: Some(placement_ref("INBOX", "304")),
+        })
+        .await
+        .expect_err("a wrong uid must not be rounded to the right one");
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::MessageNotFound)
+    );
+
+    assert!(
+        db.list_pending_mutations(50)
+            .await
+            .expect("read the outbox queue")
+            .is_empty(),
+        "a refused mutation must leave nothing queued"
+    );
+}
+
+/// Both read RPCs report the message's complete occupancy set, folder id
+/// ascending, while `folder_id` keeps its older meaning: the occupancy the
+/// message was reached through. A client that only ever saw `folder_id` cannot
+/// tell a one-mailbox message from a two-mailbox one.
+#[tokio::test]
+async fn read_rpcs_report_every_occupancy_of_a_message() {
+    let (addr, token, db, _dir) = boot_daemon("mutation_placements_read.db").await;
+    seed_account(&db).await;
+    seed_message(
+        &db,
+        "msg-multi",
+        &[placement("INBOX", "101"), placement("Archive", "202")],
+    )
+    .await;
+
+    let mut mail = nuncio_proto::client::connect_mail(&addr, &token)
+        .await
+        .expect("mail client connects");
+
+    let fetched = mail
+        .get_message(GetMessageRequest {
+            message_id: "msg-multi".to_string(),
+        })
+        .await
+        .expect("get_message succeeds")
+        .into_inner()
+        .message
+        .expect("a found message is always populated");
+    let folders: Vec<&str> = fetched
+        .placements
+        .iter()
+        .map(|p| p.folder_id.as_str())
+        .collect();
+    assert_eq!(folders, vec!["Archive", "INBOX"]);
+    let uids: Vec<&str> = fetched.placements.iter().map(|p| p.uid.as_str()).collect();
+    assert_eq!(uids, vec!["202", "101"]);
+    assert!(fetched
+        .placements
+        .iter()
+        .all(|p| p.account_id == ACCOUNT && p.uidvalidity == UIDVALIDITY));
+    assert_eq!(
+        fetched.folder_id, "Archive",
+        "folder_id still names the single occupancy this read reached the message through"
+    );
+
+    // Reached through INBOX this time: the same complete set, a different
+    // `folder_id`.
+    let listed = mail
+        .list_messages(ListMessagesRequest {
+            folder_id: "INBOX".to_string(),
+            page_size: 10,
+            page_token: String::new(),
+        })
+        .await
+        .expect("list_messages succeeds")
+        .into_inner()
+        .messages;
+    assert_eq!(listed.len(), 1);
+    assert_eq!(listed[0].folder_id, "INBOX");
+    let folders: Vec<&str> = listed[0]
+        .placements
+        .iter()
+        .map(|p| p.folder_id.as_str())
+        .collect();
+    assert_eq!(folders, vec!["Archive", "INBOX"]);
+}
+
+/// A conflict is durable state, not a passing notification: it survives in a
+/// queryable table, resolving it is recorded once and only once, and resolved
+/// rows stay as the audit trail of what was decided.
+#[tokio::test]
+async fn conflicts_round_trip_from_record_through_resolution() {
+    let (addr, token, db, _dir) = boot_daemon("mutation_conflicts.db").await;
+    seed_account(&db).await;
+    seed_message(&db, "msg-solo", &[placement("INBOX", "303")]).await;
+
+    db.record_conflict(&MutationConflict {
+        id: "conflict-1".to_string(),
+        mutation_id: "mut-conflicted".to_string(),
+        message_id: "msg-solo".to_string(),
+        mutation_type: "MOVE".to_string(),
+        placement: Some(placement_key("INBOX", "303")),
+        observed: "another client moved the message to Archive first".to_string(),
+        detected_at: 1_700_000_500,
+        resolved_at: None,
+        resolution: None,
+    })
+    .await
+    .expect("record the conflict");
+
+    let mut mail = nuncio_proto::client::connect_mail(&addr, &token)
+        .await
+        .expect("mail client connects");
+
+    let conflicts = mail
+        .list_conflicts(ListConflictsRequest {
+            include_resolved: false,
+            limit: 10,
+        })
+        .await
+        .expect("list_conflicts succeeds")
+        .into_inner()
+        .conflicts;
+    assert_eq!(conflicts.len(), 1);
+    assert_eq!(conflicts[0].id, "conflict-1");
+    assert_eq!(conflicts[0].mutation_id, "mut-conflicted");
+    assert_eq!(conflicts[0].mutation_type, "MOVE");
+    assert_eq!(
+        conflicts[0].observed,
+        "another client moved the message to Archive first"
+    );
+    assert_eq!(
+        conflicts[0].placement,
+        Some(placement_ref("INBOX", "303")),
+        "the conflict names the occupancy the refused mutation was aimed at"
+    );
+    assert_eq!(conflicts[0].resolved_at, None);
+
+    mail.resolve_conflict(ResolveConflictRequest {
+        conflict_id: "conflict-1".to_string(),
+        resolution: "kept the other client's move".to_string(),
+    })
+    .await
+    .expect("resolve_conflict succeeds");
+
+    // Resolving it again is an honest error: reporting success would tell the
+    // caller their decision was recorded when the earlier one already stands.
+    let status = mail
+        .resolve_conflict(ResolveConflictRequest {
+            conflict_id: "conflict-1".to_string(),
+            resolution: "changed my mind".to_string(),
+        })
+        .await
+        .expect_err("a conflict can only be resolved once");
+    assert_eq!(status.code(), Code::NotFound);
+    assert_eq!(
+        errors::error_reason(&status),
+        Some(ErrorReason::MessageNotFound)
+    );
+
+    let unresolved = mail
+        .list_conflicts(ListConflictsRequest {
+            include_resolved: false,
+            limit: 10,
+        })
+        .await
+        .expect("list_conflicts succeeds")
+        .into_inner()
+        .conflicts;
+    assert!(
+        unresolved.is_empty(),
+        "a handled conflict is no longer outstanding work"
+    );
+
+    let all = mail
+        .list_conflicts(ListConflictsRequest {
+            include_resolved: true,
+            limit: 10,
+        })
+        .await
+        .expect("list_conflicts succeeds")
+        .into_inner()
+        .conflicts;
+    assert_eq!(all.len(), 1, "the audit trail survives the resolution");
+    assert_eq!(
+        all[0].resolution.as_deref(),
+        Some("kept the other client's move")
+    );
+    assert!(
+        all[0].resolved_at.is_some(),
+        "a resolved conflict records when the decision was made"
+    );
+}
+
+/// `filters_enabled` is settable over the wire rather than only by editing the
+/// accounts row by hand. It still defaults to off, because filters take
+/// irreversible remote actions and an account should not acquire them merely by
+/// being added.
+#[tokio::test]
+async fn filters_enabled_round_trips_over_the_wire() {
+    let (addr, token, _db, _dir) = boot_daemon("mutation_filters_flag.db").await;
+
+    let mut accounts = nuncio_proto::client::connect_accounts(&addr, &token)
+        .await
+        .expect("accounts client connects");
+
+    let config = AccountConfig {
+        id: ACCOUNT.to_string(),
+        name: "Mutation E2E".to_string(),
+        email_address: "mutation-e2e@nuncio.mx".to_string(),
+        keyring_secret_key: format!("nuncio/{ACCOUNT}"),
+        sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+        filters_enabled: false,
+        transport: Some(nuncio_proto::v1::account_config::Transport::Jmap(
+            nuncio_proto::v1::JmapTransport {
+                endpoint_host: "jmap.nuncio.mx".to_string(),
+            },
+        )),
+    };
+    accounts
+        .add_account(AddAccountRequest {
+            config: Some(config.clone()),
+            password: "mutation-e2e-secret".to_string(),
+        })
+        .await
+        .expect("add_account succeeds");
+
+    let listed = accounts
+        .list_accounts(ListAccountsRequest {})
+        .await
+        .expect("list_accounts succeeds")
+        .into_inner()
+        .accounts;
+    assert_eq!(listed.len(), 1);
+    assert!(
+        !listed[0].filters_enabled,
+        "an account does not get filters by being added, only by being told to"
+    );
+
+    accounts
+        .update_account(UpdateAccountRequest {
+            config: Some(AccountConfig {
+                filters_enabled: true,
+                ..config
+            }),
+            password: None,
+        })
+        .await
+        .expect("update_account succeeds");
+
+    let relisted = accounts
+        .list_accounts(ListAccountsRequest {})
+        .await
+        .expect("list_accounts succeeds")
+        .into_inner()
+        .accounts;
+    assert_eq!(relisted.len(), 1);
+    assert!(
+        relisted[0].filters_enabled,
+        "the flip must be persisted and reported back, not silently discarded"
+    );
+}

crates/nunciod/tests/no_secrets_canary_test.rs

@@ -122,6 +122,7 @@ fn sample_proto_config() -> AccountConfig {
         email_address: ACCOUNT_EMAIL.to_string(),
         keyring_secret_key: KEYRING_KEY.to_string(),
         sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+        filters_enabled: false,
         transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp(
             nuncio_proto::v1::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),

crates/nunciod/tests/spine_e2e_test.rs

@@ -60,6 +60,7 @@ fn sample_account_config() -> AccountConfig {
         email_address: ACCOUNT_EMAIL.to_string(),
         keyring_secret_key: KEYRING_KEY.to_string(),
         sync_interval: Some(nuncio_proto::time::duration_from_secs(60)),
+        filters_enabled: false,
         transport: Some(nuncio_proto::v1::account_config::Transport::ImapSmtp(
             nuncio_proto::v1::ImapSmtpTransport {
                 imap_host: "imap.nuncio.mx".to_string(),