diff --git a/crates/nuncio-cli/src/runner.rs b/crates/nuncio-cli/src/runner.rs
index cb21ad3..c9d05ec 100644
--- a/crates/nuncio-cli/src/runner.rs
+++ b/crates/nuncio-cli/src/runner.rs
@@ -875,6 +875,11 @@ impl HeadlessRunner {
         // Follow `next_page_token` to the end so the command shows the whole
         // folder, transparently exercising the keyset pagination convention.
         let mut messages = Vec::new();
+        // Messages the daemon could not return because their stored bodies
+        // failed authentication. Reported rather than quietly missing: a
+        // listing that is one message short with no explanation is a lie of
+        // omission, and this is the only place the user would ever see it.
+        let mut unreadable: Vec<String> = Vec::new();
         let mut page_token = String::new();
         loop {
             match client
@@ -888,6 +893,7 @@ impl HeadlessRunner {
                 Ok(response) => {
                     let page = response.into_inner();
                     messages.extend(page.messages);
+                    unreadable.extend(page.unreadable_message_ids);
                     if page.next_page_token.is_empty() {
                         break;
                     }
@@ -904,10 +910,20 @@ impl HeadlessRunner {
                 messages.iter().map(message_proto_to_json).collect();
             format_json(&json!({
                 "folder": folder,
-                "messages": messages_json
+                "messages": messages_json,
+                "unreadable_message_ids": unreadable
             }))
-        } else {
+        } else if unreadable.is_empty() {
             format!("Folder '{}': {} message(s) found", folder, messages.len())
+        } else {
+            format!(
+                "Folder '{}': {} message(s) found; {} omitted (stored body failed \
+                 authentication): {}",
+                folder,
+                messages.len(),
+                unreadable.len(),
+                unreadable.join(", ")
+            )
         }
     }
 
@@ -1450,12 +1466,26 @@ impl HeadlessRunner {
                         "output_path": resp.output_path,
                         "message_count": resp.message_count,
                         "bytes_written": resp.bytes_written,
+                        "unreadable_message_ids": resp.unreadable_message_ids,
                     }))
-                } else {
+                } else if resp.unreadable_message_ids.is_empty() {
                     format!(
                         "Exported {} message(s) to '{}' ({} bytes)",
                         resp.message_count, resp.output_path, resp.bytes_written
                     )
+                } else {
+                    // Named, not just counted: this archive outlives the store,
+                    // so the user needs to be able to record exactly which mail
+                    // it does not contain.
+                    format!(
+                        "Exported {} message(s) to '{}' ({} bytes); {} message(s) omitted \
+                         (stored body failed authentication): {}",
+                        resp.message_count,
+                        resp.output_path,
+                        resp.bytes_written,
+                        resp.unreadable_message_ids.len(),
+                        resp.unreadable_message_ids.join(", ")
+                    )
                 }
             }
             Err(status) => Self::render_status_error("export_mailbox", &status, json_mode),
@@ -2018,6 +2048,7 @@ impl HeadlessRunner {
                         "actions_applied_count": p.actions_applied_count,
                         "last_message_id": p.last_message_id,
                         "done": p.done,
+                        "unreadable_count": p.unreadable_count,
                     })
                 })
                 .collect();
@@ -2025,8 +2056,11 @@ impl HeadlessRunner {
         } else {
             match updates.last() {
                 Some(last) => format!(
-                    "Triage complete: scanned={} matched={} actions_applied={}",
-                    last.scanned_count, last.matched_count, last.actions_applied_count
+                    "Triage complete: scanned={} matched={} actions_applied={} unreadable={}",
+                    last.scanned_count,
+                    last.matched_count,
+                    last.actions_applied_count,
+                    last.unreadable_count
                 ),
                 None => "Triage produced no progress updates.".to_string(),
             }
@@ -3844,6 +3878,7 @@ mod tests {
                 let mut message = stub_message();
                 message.folder_id = req.folder_id;
                 Ok(tonic::Response::new(ListMessagesResponse {
+                    unreadable_message_ids: Vec::new(),
                     messages: vec![message],
                     next_page_token: String::new(),
                 }))
@@ -5055,6 +5090,7 @@ mod tests {
             ) -> Result<tonic::Response<Self::TriageStream>, tonic::Status> {
                 let updates = vec![
                     Ok(TriageProgress {
+                        unreadable_count: 0,
                         scanned_count: 2,
                         matched_count: 1,
                         actions_applied_count: 1,
@@ -5062,6 +5098,7 @@ mod tests {
                         done: false,
                     }),
                     Ok(TriageProgress {
+                        unreadable_count: 0,
                         scanned_count: 3,
                         matched_count: 2,
                         actions_applied_count: 2,
@@ -5377,6 +5414,7 @@ mod tests {
                 let output_path = req.output_path.clone();
                 *self.last_request.lock().unwrap_or_else(|e| e.into_inner()) = Some(req);
                 Ok(tonic::Response::new(ExportResponse {
+                    unreadable_message_ids: Vec::new(),
                     output_path,
                     message_count: 7,
                     bytes_written: 4096,
diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin
index 94f9059..ccb5760 100644
Binary files a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin and b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin differ
diff --git a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
index bba6752..e198420 100644
--- a/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
+++ b/crates/nuncio-proto/proto/nuncio/v1/nuncio.proto
@@ -695,6 +695,16 @@ message ListMessagesResponse {
   repeated Message messages = 1;
   // Opaque cursor for the next page, or empty when this is the last page.
   string next_page_token = 2;
+  // IDs of messages that occupied this page but could not be returned: their
+  // stored body ciphertext failed authentication (tampering or bit-rot).
+  //
+  // One unreadable row does not fail the page -- the other messages are still
+  // listed -- but it is never returned with an empty or substituted body
+  // either, because that is indistinguishable from a genuinely empty message.
+  // It is named here instead, so a caller can say what it is not showing.
+  // `GetMessage` on one of these IDs still fails, which is the honest answer
+  // when the whole point of the call is that one message's content.
+  repeated string unreadable_message_ids = 3;
 }
 
 // GetMessageRequest.
@@ -1220,6 +1230,15 @@ message TriageProgress {
   // True only on the final message of the stream, once the whole store has
   // been scanned.
   bool done = 5;
+  // Total messages walked but not evaluated, because their stored body
+  // ciphertext failed authentication.
+  //
+  // A count rather than a list: this accumulates over a scan of the entire
+  // store and has no bound worth streaming. Each one is logged by name at
+  // WARN. Rules are never evaluated against a body the store could not
+  // authenticate -- a filter that acts on invented content is worse than one
+  // that reports it did not run.
+  uint64 unreadable_count = 6;
 }
 
 // Export exposes the daemon's real mailbox export engine
@@ -1287,6 +1306,15 @@ message ExportResponse {
   uint64 message_count = 2;
   // Total bytes actually written to `output_path`.
   uint64 bytes_written = 3;
+  // IDs of messages in scope that were NOT written, because their stored body
+  // ciphertext failed authentication.
+  //
+  // An export is the artifact that outlives the store, so the gap in it is
+  // named rather than merely counted: the caller can record exactly which mail
+  // its archive does not contain. Writing them with an empty or placeholder
+  // body instead would put text into an archive that the store never held, and
+  // no later reader could tell it apart from the real thing.
+  repeated string unreadable_message_ids = 4;
 }
 
 // Calendar exposes the READ path over the daemon's real, persistent calendar
diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs
index ca96d34..ff4b7ec 100644
--- a/crates/nuncio-store/src/db.rs
+++ b/crates/nuncio-store/src/db.rs
@@ -323,17 +323,87 @@ pub struct SaveOutcome {
 /// as the only place both can reach.
 pub use nuncio_core::model::PlacementKey;
 
-/// A message paired with the mailbox occupancy it was read through.
+/// A stored row whose body ciphertext would not authenticate on read.
+///
+/// Carries only columns the store keeps in the clear -- the message key and its
+/// receipt time. It never carries body content, because the body is precisely
+/// what failed to authenticate and is therefore not trustworthy enough to hand
+/// on, log, or index.
+///
+/// `received_at` travels with the key so an unreadable row can still hold its
+/// slot in an ordering or a keyset page. Dropping it from the sequence outright
+/// would move every later row forward and make a keyset cursor point at the
+/// wrong place.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UnreadableBody {
+    /// Store key of the row whose body would not authenticate.
+    pub message_key: String,
+    /// The row's receipt time, in the same units as `Email::received_at`.
+    pub received_at: i64,
+}
+
+/// One row of a **bulk** message read: `Ok` when the stored body authenticated,
+/// `Err` when it did not.
+///
+/// Single-item and bulk reads answer different questions, and this type is why
+/// they are allowed to answer them differently.
+///
+/// [`DatabaseEngine::get_message`] is asked for one named message. It fails
+/// closed -- returning `Err` for the whole call -- because a message handed back
+/// with an empty body is indistinguishable from a genuinely empty one, which
+/// would hide exactly the tampering the AEAD exists to detect.
+///
+/// A listing, a rescan or an export is asked for a *set*. Failing the whole set
+/// over one bit-rotted row hides the other forty-nine just as effectively as an
+/// empty body hides the one. So the bulk paths keep every row in its slot and
+/// refuse to invent a body for the ones that will not authenticate: the caller
+/// receives `Err(UnreadableBody)` for those rows and intact messages for the
+/// rest. Because the failure rides in the row rather than in the call, a caller
+/// cannot consume the page without meeting it.
+pub type MessageRead = Result<nuncio_core::model::Email, UnreadableBody>;
+
+/// The identity columns every bulk-read row carries, readable body or not.
+///
+/// Lets ordering, keyset cursors and progress counters treat an unreadable row
+/// as the row it still is, without first having to know whether its body
+/// decrypted.
+pub trait BulkReadRow {
+    /// Store key of this row.
+    fn message_key(&self) -> &str;
+    /// Receipt time of this row.
+    fn received_at(&self) -> i64;
+}
+
+impl BulkReadRow for MessageRead {
+    fn message_key(&self) -> &str {
+        match self {
+            Ok(email) => &email.id,
+            Err(unreadable) => &unreadable.message_key,
+        }
+    }
+
+    fn received_at(&self) -> i64 {
+        match self {
+            Ok(email) => email.received_at,
+            Err(unreadable) => unreadable.received_at,
+        }
+    }
+}
+
+/// A bulk-read message row paired with the mailbox occupancy it was read
+/// through.
 ///
 /// The two travel together out of every folder-scoped read: the caller almost
 /// always needs the per-mailbox facts (the read flag, the addressing UID) as
-/// well as the message, and re-deriving them would cost a query per row.
+/// well as the message, and re-deriving them would cost a query per row. The
+/// placement is present even for a row whose body would not authenticate --
+/// where the row sits is not encrypted, and the caller still needs it to page.
 ///
 /// Deliberately not called `PlacedMessage`: `nuncio_mail::PlacedMessage` is a
 /// different type carrying a third field (the identity tier a backend derived
 /// the key from), and the sync path imports both. One name for two shapes in
 /// one call site is a trap worth a longer name to avoid.
-pub type MessageWithPlacement = (nuncio_core::model::Email, nuncio_core::model::Placement);
+pub type MessageWithPlacement = (MessageRead, nuncio_core::model::Placement);
 
 /// Keyset cursor for [`DatabaseEngine::list_messages_page`]: the
 /// `(received_at, message_key, uidvalidity, uid)` of the last row of a page.
@@ -377,6 +447,15 @@ type MessageWithPlacementRow = (
     i64,
 );
 
+/// Drop a joined row's placement columns, leaving the identity-only
+/// [`MessageRow`] the body decoders take. Consumes the row so the message
+/// columns move rather than clone.
+fn narrow_placed_row(row: MessageWithPlacementRow) -> MessageRow {
+    (
+        row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
+    )
+}
+
 /// What a [`DatabaseEngine::delete_placements`] call removed.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct DeleteOutcome {
@@ -1235,12 +1314,31 @@ impl DatabaseEngine {
         .await
         .map_err(DatabaseError::Query)?;
 
+        let mut unreadable = 0usize;
         for (message_key, subject, sender, body_plain) in rows {
-            let dec_plain = body_plain
+            let dec_plain = match body_plain
                 .map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
                 .transpose()
-                .map_err(|e| DatabaseError::Decryption(e.to_string()))?
-                .unwrap_or_default();
+            {
+                Ok(plain) => plain.unwrap_or_default(),
+                Err(e) => {
+                    // This runs on every open, so failing the whole backfill on
+                    // one bit-rotted row would stop the daemon booting at all.
+                    // The row is left out of the index entirely rather than
+                    // indexed with an empty body: an FTS entry marks a message
+                    // permanently indexed, so an empty one would make its body
+                    // permanently unsearchable with nothing left to report it.
+                    // Leaving it out means every subsequent open re-reports it,
+                    // and a repaired row indexes itself without intervention.
+                    unreadable += 1;
+                    tracing::warn!(
+                        message_key = %message_key,
+                        error = %e,
+                        "stored message body failed authentication; leaving it out of the search index"
+                    );
+                    continue;
+                }
+            };
 
             sqlx::query(
                 "INSERT INTO messages_fts (message_key, subject, sender, body_plain) VALUES (?, ?, ?, ?)",
@@ -1254,6 +1352,14 @@ impl DatabaseEngine {
             .map_err(DatabaseError::Query)?;
         }
 
+        if unreadable > 0 {
+            tracing::warn!(
+                unreadable,
+                "search index backfill skipped messages whose stored bodies failed \
+                 authentication; their bodies are not searchable until the rows are repaired"
+            );
+        }
+
         Ok(())
     }
 
@@ -1879,6 +1985,11 @@ impl DatabaseEngine {
     /// The placement travels with the message because the caller almost always
     /// needs the per-mailbox facts -- the read flag and the addressing UID --
     /// and re-deriving them would mean a second query per row.
+    ///
+    /// A row whose stored body will not authenticate comes back as
+    /// `Err(UnreadableBody)` in its own slot rather than failing the listing --
+    /// see [`MessageRead`] for why a bulk read differs from
+    /// [`Self::get_message`] here.
     pub async fn list_messages(
         &self,
         account_id: &str,
@@ -1906,39 +2017,14 @@ impl DatabaseEngine {
 
         let mut out = Vec::with_capacity(rows.len());
         for row in rows {
-            let dec_plain = row
-                .6
-                .map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
-                .transpose()
-                .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-            let dec_html = row
-                .7
-                .map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
-                .transpose()
-                .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-
-            out.push((
-                nuncio_core::model::Email {
-                    id: row.0,
-                    account_id: row.1,
-                    subject: row.2,
-                    sender: row.3,
-                    recipient: row.4,
-                    received_at: row.5,
-                    body_plain: dec_plain,
-                    body_html: dec_html,
-                    attachments: Vec::new(),
-                    message_id: row.8,
-                    content_hash: row.9,
-                },
-                nuncio_core::model::Placement {
-                    account_id: account_id.to_string(),
-                    folder_id: folder_id.to_string(),
-                    uid_validity: row.10,
-                    remote_id: row.11,
-                    read: row.12 != 0,
-                },
-            ));
+            let placement = nuncio_core::model::Placement {
+                account_id: account_id.to_string(),
+                folder_id: folder_id.to_string(),
+                uid_validity: row.10.clone(),
+                remote_id: row.11.clone(),
+                read: row.12 != 0,
+            };
+            out.push((self.bulk_email_from_row(narrow_placed_row(row)), placement));
         }
         Ok(out)
     }
@@ -1962,6 +2048,11 @@ impl DatabaseEngine {
     /// `uidvalidity`/`uid` are compared as the text they are stored as, which is
     /// an arbitrary but total order -- and matching the `ORDER BY` exactly is
     /// all a keyset needs.
+    ///
+    /// A row whose stored body will not authenticate occupies its slot in the
+    /// page as `Err(UnreadableBody)` -- see [`MessageRead`]. Keeping the slot is
+    /// what lets the keyset stay exact: the page is still as long as the query's
+    /// window, so the cursor still names a real position in the sequence.
     pub async fn list_messages_page(
         &self,
         account_id: &str,
@@ -2018,45 +2109,27 @@ impl DatabaseEngine {
         let has_more = rows.len() > page_size;
         let mut placed = Vec::with_capacity(rows.len().min(page_size));
         for row in rows.into_iter().take(page_size) {
-            let dec_plain = row
-                .6
-                .map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
-                .transpose()
-                .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-            let dec_html = row
-                .7
-                .map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
-                .transpose()
-                .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-            placed.push((
-                nuncio_core::model::Email {
-                    id: row.0,
-                    account_id: row.1,
-                    subject: row.2,
-                    sender: row.3,
-                    recipient: row.4,
-                    received_at: row.5,
-                    body_plain: dec_plain,
-                    body_html: dec_html,
-                    attachments: Vec::new(),
-                    message_id: row.8,
-                    content_hash: row.9,
-                },
-                nuncio_core::model::Placement {
-                    account_id: account_id.to_string(),
-                    folder_id: folder_id.to_string(),
-                    uid_validity: row.10,
-                    remote_id: row.11,
-                    read: row.12 != 0,
-                },
-            ));
+            let placement = nuncio_core::model::Placement {
+                account_id: account_id.to_string(),
+                folder_id: folder_id.to_string(),
+                uid_validity: row.10.clone(),
+                remote_id: row.11.clone(),
+                read: row.12 != 0,
+            };
+            // An unreadable row stays in the page rather than being filtered
+            // out of it. The keyset is positional: dropping rows here would
+            // make the page shorter than the query's own window, so `has_more`
+            // and the cursor below would both describe a sequence the store
+            // never returned, and rows between the last surviving row and the
+            // query's real boundary would become unreachable.
+            placed.push((self.bulk_email_from_row(narrow_placed_row(row)), placement));
         }
 
         let next = if has_more {
-            placed.last().map(|(e, p)| {
+            placed.last().map(|(read, p)| {
                 (
-                    e.received_at,
-                    e.id.clone(),
+                    read.received_at(),
+                    read.message_key().to_string(),
                     p.uid_validity.clone(),
                     p.remote_id.clone(),
                 )
@@ -2089,30 +2162,11 @@ impl DatabaseEngine {
         .await
         .map_err(DatabaseError::Query)?;
 
-        let dec_plain = row
-            .6
-            .map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
-            .transpose()
-            .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-        let dec_html = row
-            .7
-            .map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
-            .transpose()
-            .map_err(|e| DatabaseError::Decryption(e.to_string()))?;
-
-        Ok(nuncio_core::model::Email {
-            id: row.0,
-            account_id: row.1,
-            subject: row.2,
-            sender: row.3,
-            recipient: row.4,
-            received_at: row.5,
-            body_plain: dec_plain,
-            body_html: dec_html,
-            attachments: Vec::new(),
-            message_id: row.8,
-            content_hash: row.9,
-        })
+        // Fail-closed on purpose, and deliberately NOT the bulk decoder: this
+        // call names one message, so there is no other row for the caller to
+        // still receive, and a message returned with an empty body reads
+        // exactly like a genuinely empty one.
+        self.email_from_row(row)
     }
 
     /// Every placement currently stored for one folder of one account.
@@ -3182,11 +3236,18 @@ impl DatabaseEngine {
     ///
     /// Walks message identities, not placements: a chunked pass over the store
     /// wants each message exactly once, however many folders it occupies.
+    ///
+    /// A row whose stored body will not authenticate is returned as
+    /// `Err(UnreadableBody)` in its own slot -- see [`MessageRead`]. Keeping the
+    /// slot matters twice over here: a scan pages on chunk length and resumes
+    /// from the last key it saw, so silently shortening a chunk would both
+    /// declare the scan finished early and rewind the cursor past rows it never
+    /// looked at.
     pub async fn get_message_chunk(
         &self,
         last_key: &str,
         limit: usize,
-    ) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
+    ) -> Result<Vec<MessageRead>, DatabaseError> {
         let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
             "SELECT message_key, account_id, subject, sender, recipient, received_at, \
              body_plain, body_html, message_id, content_hash FROM messages ",
@@ -3205,7 +3266,10 @@ impl DatabaseEngine {
             .await
             .map_err(DatabaseError::Query)?;
 
-        rows.into_iter().map(|r| self.email_from_row(r)).collect()
+        Ok(rows
+            .into_iter()
+            .map(|r| self.bulk_email_from_row(r))
+            .collect())
     }
 
     /// Query messages across the WHOLE store for export purposes,
@@ -3217,11 +3281,18 @@ impl DatabaseEngine {
     /// The folder filter is an `EXISTS` over `placements` rather than a join:
     /// a message occupying two folders must still export as one message, and a
     /// join would emit it once per matching placement.
+    ///
+    /// A row whose stored body will not authenticate is returned as
+    /// `Err(UnreadableBody)` rather than failing the export -- see
+    /// [`MessageRead`]. An archive is the one artifact a caller keeps after the
+    /// store is gone, so the caller is handed the key of every message it will
+    /// not contain and can name the gap; what it must never do is write a
+    /// message with a body the store could not authenticate.
     pub async fn list_messages_for_export(
         &self,
         account_id: Option<&str>,
         folder_id: Option<&str>,
-    ) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
+    ) -> Result<Vec<MessageRead>, DatabaseError> {
         let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
             "SELECT m.message_key, m.account_id, m.subject, m.sender, m.recipient, \
              m.received_at, m.body_plain, m.body_html, m.message_id, m.content_hash \
@@ -3257,7 +3328,35 @@ impl DatabaseEngine {
             .await
             .map_err(DatabaseError::Query)?;
 
-        rows.into_iter().map(|r| self.email_from_row(r)).collect()
+        Ok(rows
+            .into_iter()
+            .map(|r| self.bulk_email_from_row(r))
+            .collect())
+    }
+
+    /// The bulk-read counterpart to [`Self::email_from_row`]: the same
+    /// decryption, but a body that will not authenticate is reported as
+    /// [`UnreadableBody`] for that row alone instead of failing the whole read.
+    ///
+    /// The failure is logged at `warn!` naming the message key, so an operator
+    /// watching the daemon sees which row is at fault rather than a listing that
+    /// is quietly one message short. The log deliberately carries the key and
+    /// the cipher's own error only -- never body content, which is the exact
+    /// material that failed to authenticate and must not be echoed anywhere.
+    fn bulk_email_from_row(&self, row: MessageRow) -> MessageRead {
+        let message_key = row.0.clone();
+        let received_at = row.5;
+        self.email_from_row(row).map_err(|e| {
+            tracing::warn!(
+                message_key = %message_key,
+                error = %e,
+                "stored message body failed authentication; returning this row as unreadable"
+            );
+            UnreadableBody {
+                message_key,
+                received_at,
+            }
+        })
     }
 
     /// Rebuild an [`nuncio_core::model::Email`] from the identity-column row
@@ -3552,6 +3651,13 @@ fn compute_log_hash(
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    /// The message behind a bulk-read row that the fixture guarantees is
+    /// readable. Panics on an unreadable row so a test asserting on message
+    /// content can never silently pass over one that failed to authenticate.
+    fn readable(read: &MessageRead) -> &nuncio_core::model::Email {
+        read.as_ref().expect("this fixture row's body is readable")
+    }
     use sqlx::Connection;
 
     /// Proves that a transient error surfacing from the integrity probe (here, a
@@ -3831,7 +3937,7 @@ mod tests {
             .await
             .expect("list messages succeeds");
         assert_eq!(msgs.len(), 1);
-        assert_eq!(msgs[0].0.id, "msg-db-100");
+        assert_eq!(readable(&msgs[0].0).id, "msg-db-100");
         assert_eq!(msgs[0].1.remote_id, "100");
 
         let folders = engine
@@ -3844,9 +3950,13 @@ mod tests {
     }
 
     /// A `body_plain` column corrupted after a legitimate write (simulating tampering or
-    /// bit-rot) must make `get_message` and `list_messages` return
+    /// bit-rot) must make the SINGLE-ITEM `get_message` return
     /// `Err(DatabaseError::Decryption)`, never a silently-empty body -- that would defeat
     /// the AEAD's integrity guarantee exactly where it matters: detecting tampering on read.
+    ///
+    /// The bulk paths are covered by
+    /// `one_corrupt_body_does_not_destroy_a_bulk_read`, which asserts the other half of
+    /// the contract: they must not fail closed over one row, and must not hide it either.
     #[tokio::test]
     async fn corrupted_body_ciphertext_fails_closed_on_read() {
         let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
@@ -3906,20 +4016,218 @@ mod tests {
             .expect_err("get_message must fail closed on corrupted ciphertext");
         assert!(matches!(err, DatabaseError::Decryption(_)));
 
-        let err = engine
+        // The bulk paths deliberately do NOT fail the whole call here -- but the one
+        // corrupt row still never comes back carrying a body.
+        let listed = engine
             .list_messages("acct-1", "INBOX", 10)
             .await
-            .expect_err("list_messages must fail closed on corrupted ciphertext");
-        assert!(matches!(err, DatabaseError::Decryption(_)));
+            .expect("a bulk listing must survive one unreadable row");
+        assert_eq!(listed.len(), 1);
+        assert_eq!(
+            listed[0].0.as_ref().err().map(|u| u.message_key.as_str()),
+            Some("msg-db-corrupt"),
+            "the corrupt row must be reported as unreadable, not returned with a body"
+        );
+    }
 
-        // The export path reads the same ciphertext column and must fail closed too --
-        // an export that silently dropped an unauthenticated body would write a file
-        // the user could not tell apart from a genuinely empty message.
-        let err = engine
+    /// Corrupt the stored `body_plain` ciphertext of one message in place, leaving it
+    /// valid hex so the failure is an AEAD authentication failure rather than a decode
+    /// error -- the same shape a flipped bit on disk or a tampering edit would produce.
+    async fn corrupt_stored_body(engine: &DatabaseEngine, message_key: &str) {
+        let (stored,): (String,) =
+            sqlx::query_as("SELECT body_plain FROM messages WHERE message_key = ?")
+                .bind(message_key)
+                .fetch_one(&engine.pool)
+                .await
+                .expect("fetch stored ciphertext");
+        let mut bytes = hex::decode(&stored).expect("stored ciphertext is valid hex");
+        let last_idx = bytes.len() - 1;
+        bytes[last_idx] ^= 0xFF;
+
+        sqlx::query("UPDATE messages SET body_plain = ? WHERE message_key = ?")
+            .bind(hex::encode(bytes))
+            .bind(message_key)
+            .execute(&engine.pool)
+            .await
+            .expect("corrupt stored ciphertext");
+    }
+
+    /// Split a bulk read into the keys that came back whole and the keys reported
+    /// unreadable, asserting along the way that nothing arrived with a fabricated body.
+    fn split_bulk_read(rows: &[MessageRead]) -> (Vec<&str>, Vec<&str>) {
+        let mut good = Vec::new();
+        let mut bad = Vec::new();
+        for row in rows {
+            match row {
+                Ok(email) => {
+                    assert!(
+                        email.body_plain.as_deref().is_some_and(|b| !b.is_empty()),
+                        "a readable row must carry its real body, never an empty one"
+                    );
+                    good.push(email.id.as_str());
+                }
+                Err(unreadable) => bad.push(unreadable.message_key.as_str()),
+            }
+        }
+        (good, bad)
+    }
+
+    /// Store `keys` as messages in one folder, with a distinct body per position.
+    async fn save_bulk_fixture(engine: &DatabaseEngine, keys: &[&str]) {
+        for (idx, key) in keys.iter().enumerate() {
+            let email = nuncio_core::model::Email {
+                id: (*key).to_string(),
+                account_id: "acct-1".to_string(),
+                subject: format!("Message {idx}"),
+                sender: "alice@nuncio.mx".to_string(),
+                recipient: "bob@nuncio.mx".to_string(),
+                received_at: 1_700_000_000 + idx as i64,
+                body_plain: Some(format!("body number {idx}")),
+                body_html: None,
+                attachments: Vec::new(),
+                message_id: None,
+                content_hash: None,
+            };
+            let placement = nuncio_core::model::Placement {
+                account_id: "acct-1".to_string(),
+                folder_id: "INBOX".to_string(),
+                uid_validity: "1".to_string(),
+                remote_id: idx.to_string(),
+                read: false,
+            };
+            engine
+                .save_email_at(
+                    &email,
+                    nuncio_core::model::IdentitySource::Surrogate,
+                    &placement,
+                )
+                .await
+                .expect("save email succeeds");
+        }
+    }
+
+    /// The counterpart to `corrupted_body_ciphertext_fails_closed_on_read`: one corrupt
+    /// row must not destroy a whole page, a whole rescan, or a whole export -- and must
+    /// not vanish from them either.
+    ///
+    /// Three messages are stored and the middle one's ciphertext is corrupted. Every bulk
+    /// path must return the two good messages, account for the bad one by key, and never
+    /// hand back a row carrying a body it could not authenticate. The single-item read of
+    /// that same message must still fail closed, which is what makes the two behaviours a
+    /// deliberate distinction rather than an inconsistency.
+    #[tokio::test]
+    async fn one_corrupt_body_does_not_destroy_a_bulk_read() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+        save_bulk_fixture(&engine, &["msg-good-1", "msg-rotten", "msg-good-2"]).await;
+        corrupt_stored_body(&engine, "msg-rotten").await;
+
+        let listed = engine
+            .list_messages("acct-1", "INBOX", 10)
+            .await
+            .expect("listing survives one unreadable row");
+        let rows: Vec<MessageRead> = listed.into_iter().map(|(read, _)| read).collect();
+        let (good, bad) = split_bulk_read(&rows);
+        assert_eq!(good.len(), 2, "both intact messages are still listed");
+        assert_eq!(bad, vec!["msg-rotten"]);
+
+        // The keyset page must keep the unreadable row in its slot: a page one row short
+        // of its own window would make the cursor name a position the store never
+        // returned, and rows past it unreachable.
+        let (page, next) = engine
+            .list_messages_page("acct-1", "INBOX", None, 2)
+            .await
+            .expect("paged listing survives one unreadable row");
+        assert_eq!(page.len(), 2, "the unreadable row still occupies a slot");
+        assert!(next.is_some(), "a third row remains, so a cursor is owed");
+        let page_rows: Vec<MessageRead> = page.into_iter().map(|(read, _)| read).collect();
+        let (_, page_bad) = split_bulk_read(&page_rows);
+        assert_eq!(page_bad, vec!["msg-rotten"]);
+
+        let chunk = engine
+            .get_message_chunk("", 10)
+            .await
+            .expect("chunked scan survives one unreadable row");
+        assert_eq!(chunk.len(), 3, "the scan still walks every row");
+        let (good, bad) = split_bulk_read(&chunk);
+        assert_eq!(good.len(), 2);
+        assert_eq!(bad, vec!["msg-rotten"]);
+
+        let exported = engine
             .list_messages_for_export(None, None)
             .await
-            .expect_err("export must fail closed on corrupted ciphertext");
+            .expect("export survives one unreadable row");
+        let (good, bad) = split_bulk_read(&exported);
+        assert_eq!(
+            good.len(),
+            2,
+            "an export must still contain every message it can authenticate"
+        );
+        assert_eq!(
+            bad,
+            vec!["msg-rotten"],
+            "the message missing from the archive must be named, not silently dropped"
+        );
+
+        // The whole point of the distinction: asking for that one message by name still
+        // fails closed, because there is no other row left for the caller to receive.
+        let err = engine
+            .get_message("msg-rotten")
+            .await
+            .expect_err("the single-item read must still fail closed");
         assert!(matches!(err, DatabaseError::Decryption(_)));
+
+        // And the good messages are still individually readable -- the corruption is
+        // isolated to its own row, not smeared across the store.
+        assert_eq!(
+            engine
+                .get_message("msg-good-1")
+                .await
+                .expect("an intact message is still readable")
+                .body_plain
+                .as_deref(),
+            Some("body number 0")
+        );
+    }
+
+    /// `migrate()` runs on every open and backfills the search index from stored
+    /// ciphertext, so a single bit-rotted body must not stop the daemon booting. The
+    /// affected row is left out of the index (never indexed with an empty body, which
+    /// would mark it permanently indexed and permanently unsearchable), while every other
+    /// message indexes and searches normally.
+    #[tokio::test]
+    async fn fts_backfill_skips_one_corrupt_body_instead_of_failing_the_open() {
+        let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
+        save_bulk_fixture(&engine, &["msg-index-good", "msg-index-rotten"]).await;
+
+        // Drop both FTS entries so the backfill has real work, then corrupt one body.
+        sqlx::query("DELETE FROM messages_fts")
+            .execute(&engine.pool)
+            .await
+            .expect("clear the index");
+        corrupt_stored_body(&engine, "msg-index-rotten").await;
+
+        engine
+            .migrate()
+            .await
+            .expect("one unreadable body must not stop the store opening");
+
+        let search = crate::search::SearchEngine::new(&engine);
+        let hits = search
+            .search_messages("number 0")
+            .await
+            .expect("search succeeds");
+        assert_eq!(hits.len(), 1, "the intact message is indexed and findable");
+        assert_eq!(hits[0].id, "msg-index-good");
+
+        // The corrupt row is absent from the index rather than present with an empty
+        // body: it must stay a candidate for backfill so a repaired row heals itself.
+        let (indexed,): (i64,) =
+            sqlx::query_as("SELECT COUNT(*) FROM messages_fts WHERE message_key = ?")
+                .bind("msg-index-rotten")
+                .fetch_one(&engine.pool)
+                .await
+                .expect("count index rows");
+        assert_eq!(indexed, 0);
     }
 
     fn sample_calendar_event(id: &str, account_id: &str) -> nuncio_core::model::CalendarEvent {
@@ -4758,14 +5066,14 @@ mod tests {
 
         let chunk1 = engine.get_message_chunk("", 3).await.unwrap();
         assert_eq!(chunk1.len(), 3);
-        assert_eq!(chunk1[0].id, "msg-001");
-        assert_eq!(chunk1[2].id, "msg-003");
+        assert_eq!(readable(&chunk1[0]).id, "msg-001");
+        assert_eq!(readable(&chunk1[2]).id, "msg-003");
 
-        let last_id = &chunk1.last().unwrap().id;
+        let last_id = &readable(chunk1.last().unwrap()).id;
         let chunk2 = engine.get_message_chunk(last_id, 3).await.unwrap();
         assert_eq!(chunk2.len(), 2);
-        assert_eq!(chunk2[0].id, "msg-004");
-        assert_eq!(chunk2[1].id, "msg-005");
+        assert_eq!(readable(&chunk2[0]).id, "msg-004");
+        assert_eq!(readable(&chunk2[1]).id, "msg-005");
 
         let mutation = nuncio_filter::OutboxManager::create_mutation(
             "rule-1",
@@ -5920,7 +6228,7 @@ mod tests {
             assert!(rows.len() <= 2, "page must not exceed page_size");
             seen.extend(
                 rows.iter()
-                    .map(|(e, p)| (e.id.clone(), p.remote_id.clone())),
+                    .map(|(e, p)| (readable(e).id.clone(), p.remote_id.clone())),
             );
             match next {
                 Some(cursor) => after = Some(cursor),
@@ -6055,10 +6363,10 @@ mod tests {
             .expect("list");
         let found = listed
             .iter()
-            .find(|(m, _)| m.id == pair.0.id)
+            .find(|(m, _)| readable(m).id == pair.0.id)
             .expect("the saved message is listed");
-        assert_eq!(found.0.message_id, pair.0.message_id);
-        assert_eq!(found.0.content_hash, pair.0.content_hash);
+        assert_eq!(readable(&found.0).message_id, pair.0.message_id);
+        assert_eq!(readable(&found.0).content_hash, pair.0.content_hash);
 
         let exported = engine
             .list_messages_for_export(None, None)
@@ -6066,10 +6374,10 @@ mod tests {
             .expect("export");
         let found = exported
             .iter()
-            .find(|m| m.id == pair.0.id)
+            .find(|m| readable(m).id == pair.0.id)
             .expect("the saved message is exportable");
-        assert_eq!(found.message_id, pair.0.message_id);
-        assert_eq!(found.content_hash, pair.0.content_hash);
+        assert_eq!(readable(found).message_id, pair.0.message_id);
+        assert_eq!(readable(found).content_hash, pair.0.content_hash);
 
         // A message with no Message-ID stays None rather than becoming "".
         let plain = synced_email("acct-1", "INBOX", "42", "8");
@@ -6134,8 +6442,8 @@ mod tests {
 
         let all = engine.list_messages_for_export(None, None).await.unwrap();
         assert_eq!(all.len(), 2);
-        assert_eq!(all[0].id, "msg-1");
-        assert_eq!(all[1].id, "msg-2");
+        assert_eq!(readable(&all[0]).id, "msg-1");
+        assert_eq!(readable(&all[1]).id, "msg-2");
     }
 
     #[tokio::test]
@@ -6149,7 +6457,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(scoped.len(), 1);
-        assert_eq!(scoped[0].id, "msg-1");
+        assert_eq!(readable(&scoped[0]).id, "msg-1");
     }
 
     #[tokio::test]
@@ -6163,7 +6471,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(scoped.len(), 1);
-        assert_eq!(scoped[0].id, "msg-2");
+        assert_eq!(readable(&scoped[0]).id, "msg-2");
     }
 
     #[tokio::test]
@@ -6178,7 +6486,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(scoped.len(), 1);
-        assert_eq!(scoped[0].id, "msg-1");
+        assert_eq!(readable(&scoped[0]).id, "msg-1");
     }
 
     /// An export whose WORM audit record cannot be persisted must never report success --
@@ -6189,7 +6497,13 @@ mod tests {
     async fn export_messages_to_file_fails_when_worm_audit_write_fails() {
         let (engine, dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
         save_export_message(&engine, "msg-1", "acct-a", "inbox").await;
-        let messages = engine.list_messages_for_export(None, None).await.unwrap();
+        let messages: Vec<_> = engine
+            .list_messages_for_export(None, None)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|read| read.expect("fixture bodies are readable"))
+            .collect();
 
         engine.close().await;
 
diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs
index 481c2a8..dd07c5f 100644
--- a/crates/nunciod/src/grpc.rs
+++ b/crates/nunciod/src/grpc.rs
@@ -59,7 +59,7 @@ use nuncio_proto::v1::{
     UpdateAccountResponse, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse,
     ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse,
 };
-use nuncio_store::db::DatabaseEngine;
+use nuncio_store::db::{BulkReadRow, DatabaseEngine};
 use nuncio_store::search::SearchEngine;
 use nuncio_store::vault::SecretManager;
 use std::pin::Pin;
@@ -977,7 +977,9 @@ fn map_attachment_from_proto(attachment: AttachmentProto) -> nuncio_core::model:
 /// `folder_id` and `read` come from the placement: both are per-mailbox facts
 /// now, and the wire message still carries exactly one of each, so it describes
 /// one occupancy of the message rather than the message as a whole.
-fn map_email_to_proto(placed: nuncio_store::db::MessageWithPlacement) -> MessageProto {
+fn map_email_to_proto(
+    placed: (nuncio_core::model::Email, nuncio_core::model::Placement),
+) -> MessageProto {
     let (email, placement) = placed;
     MessageProto {
         id: email.id,
@@ -1778,11 +1780,14 @@ impl MailGrpcService {
         // Re-impose the store's ordering over the concatenated per-account
         // pages, so the cursor this returns is a position in the merged
         // sequence rather than in whichever account happened to be last.
+        // Ordered on the identity columns, which every row carries whether or
+        // not its body decrypted -- so a row the store could not read still
+        // sorts into its rightful place and still occupies a page slot.
         rows.sort_by(|(a_mail, a_place), (b_mail, b_place)| {
             b_mail
-                .received_at
-                .cmp(&a_mail.received_at)
-                .then_with(|| b_mail.id.cmp(&a_mail.id))
+                .received_at()
+                .cmp(&a_mail.received_at())
+                .then_with(|| b_mail.message_key().cmp(a_mail.message_key()))
                 .then_with(|| b_place.uid_validity.cmp(&a_place.uid_validity))
                 .then_with(|| b_place.remote_id.cmp(&a_place.remote_id))
         });
@@ -1791,10 +1796,10 @@ impl MailGrpcService {
         rows.truncate(page_size);
         let next = has_more
             .then(|| {
-                rows.last().map(|(email, placement)| {
+                rows.last().map(|(read, placement)| {
                     (
-                        email.received_at,
-                        email.id.clone(),
+                        read.received_at(),
+                        read.message_key().to_string(),
                         placement.uid_validity.clone(),
                         placement.remote_id.clone(),
                     )
@@ -2061,13 +2066,36 @@ impl Mail for MailGrpcService {
                 ])
             })
             .unwrap_or_default();
-        let mut messages: Vec<MessageProto> =
-            messages.into_iter().map(map_email_to_proto).collect();
+        // A row whose stored body would not authenticate is reported by ID
+        // instead of being listed. It is neither dropped in silence (the caller
+        // would see a page one message short and have no way to know) nor
+        // substituted with an empty body (indistinguishable from a message that
+        // genuinely has none). The rest of the page is unaffected: one
+        // bit-rotted row must not take a folder listing down with it.
+        let mut unreadable_message_ids = Vec::new();
+        let mut placed = Vec::with_capacity(messages.len());
+        for (read, placement) in messages {
+            match read {
+                Ok(email) => placed.push((email, placement)),
+                Err(unreadable) => unreadable_message_ids.push(unreadable.message_key),
+            }
+        }
+        if !unreadable_message_ids.is_empty() {
+            tracing::warn!(
+                folder_id = %req.folder_id,
+                unreadable = unreadable_message_ids.len(),
+                "Mail: omitted messages from a listing page whose stored bodies failed \
+                 authentication"
+            );
+        }
+
+        let mut messages: Vec<MessageProto> = placed.into_iter().map(map_email_to_proto).collect();
         fill_placements(&self.db, &mut messages).await;
 
         Ok(Response::new(ListMessagesResponse {
             messages,
             next_page_token,
+            unreadable_message_ids,
         }))
     }
 
@@ -2971,6 +2999,7 @@ impl Filters for FiltersGrpcService {
                 let mut scanned: u64 = 0;
                 let mut matched: u64 = 0;
                 let mut applied: u64 = 0;
+                let mut unreadable: u64 = 0;
 
                 loop {
                     let batch = match db.get_message_chunk(&last_id, chunk_size).await {
@@ -2986,7 +3015,26 @@ impl Filters for FiltersGrpcService {
                     };
                     let is_last_page = batch.len() < chunk_size;
 
-                    for email in &batch {
+                    for read in &batch {
+                        // A message whose stored body will not authenticate is
+                        // counted and skipped, never evaluated: rules read the
+                        // body, so acting on one the store could not
+                        // authenticate would let corrupted bytes drive real
+                        // side effects. Skipping it does not stop the scan --
+                        // one bad row must not cost the whole store its rescan.
+                        let email = match read {
+                            Ok(email) => email,
+                            Err(un) => {
+                                unreadable += 1;
+                                tracing::warn!(
+                                    message_key = %un.message_key,
+                                    "Filters: skipping a message during triage whose stored \
+                                     body failed authentication"
+                                );
+                                scanned += 1;
+                                continue;
+                            }
+                        };
                         // Messages belonging to an account this daemon does not
                         // own are counted as scanned but never acted on, so the
                         // progress total still reflects the whole store. A
@@ -3025,8 +3073,11 @@ impl Filters for FiltersGrpcService {
                         }
                     }
 
+                    // Advanced from the last row the query returned, readable or
+                    // not: resuming from the last *readable* row would re-scan
+                    // every unreadable row after it, forever.
                     if let Some(last) = batch.last() {
-                        last_id = last.id.clone();
+                        last_id = last.message_key().to_string();
                     }
 
                     let done = is_last_page;
@@ -3037,6 +3088,7 @@ impl Filters for FiltersGrpcService {
                             actions_applied_count: applied,
                             last_message_id: last_id.clone(),
                             done,
+                            unreadable_count: unreadable,
                         }))
                         .await
                         .is_err()
@@ -3051,6 +3103,7 @@ impl Filters for FiltersGrpcService {
                             scanned,
                             matched,
                             applied,
+                            unreadable,
                             "Filters: triage rescan finished"
                         );
                         return;
@@ -3115,7 +3168,7 @@ impl Export for ExportGrpcService {
         };
         tracing::info!(format = ?format, scope = scope_kind, "Export: mailbox export started");
 
-        let messages = match req.scope {
+        let read_rows = match req.scope {
             Some(export_request::Scope::AccountId(account_id)) => {
                 self.db
                     .list_messages_for_export(Some(&account_id), None)
@@ -3130,6 +3183,29 @@ impl Export for ExportGrpcService {
         }
         .map_err(|e| Status::internal(format!("failed to load messages for export: {e}")))?;
 
+        // A message whose stored body will not authenticate is left out of the
+        // archive and named in the response. The alternative -- writing it with
+        // an empty or "unavailable" body -- would put content into a file that
+        // outlives this store and that no later reader could distinguish from
+        // mail the user actually received. Reporting the gap keeps the archive
+        // honest about what it does contain, and one such row does not cost the
+        // user the export of everything else.
+        let mut unreadable_message_ids = Vec::new();
+        let mut messages = Vec::with_capacity(read_rows.len());
+        for read in read_rows {
+            match read {
+                Ok(email) => messages.push(email),
+                Err(unreadable) => unreadable_message_ids.push(unreadable.message_key),
+            }
+        }
+        if !unreadable_message_ids.is_empty() {
+            tracing::warn!(
+                unreadable = unreadable_message_ids.len(),
+                "Export: omitted messages whose stored bodies failed authentication; \
+                 the archive is incomplete and the omitted ids are reported to the caller"
+            );
+        }
+
         let output_path = std::path::PathBuf::from(&req.output_path);
         let summary = self
             .db
@@ -3147,6 +3223,7 @@ impl Export for ExportGrpcService {
             output_path: summary.output_path,
             message_count: summary.message_count as u64,
             bytes_written: summary.bytes_written,
+            unreadable_message_ids,
         }))
     }
 }
@@ -5370,6 +5447,75 @@ mod tests {
         );
     }
 
+    /// One corrupted stored body must not take a folder listing down with it,
+    /// and must not disappear from it either.
+    ///
+    /// This is the wire-level half of the store's
+    /// `one_corrupt_body_does_not_destroy_a_bulk_read`: the page still carries
+    /// every message whose body authenticated, the one that did not is named in
+    /// `unreadable_message_ids` rather than returned with an invented body, and
+    /// `GetMessage` on that same id still fails closed -- there the caller asked
+    /// for exactly that content, so an answer without it would be a lie.
+    #[tokio::test]
+    async fn list_messages_reports_a_corrupt_body_instead_of_failing_the_page() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("connect ephemeral test db");
+        seed_in_account(&db, "acct-a", "m-1", "inbox", 1_700_000_001).await;
+        seed_in_account(&db, "acct-a", "m-rotten", "inbox", 1_700_000_002).await;
+        seed_in_account(&db, "acct-a", "m-3", "inbox", 1_700_000_003).await;
+
+        // Flip a byte of the stored ciphertext, leaving it valid hex, so the
+        // read fails AEAD authentication exactly as bit-rot or tampering would.
+        let (stored,): (String,) =
+            sqlx::query_as("SELECT body_plain FROM messages WHERE message_key = ?")
+                .bind("m-rotten")
+                .fetch_one(db.pool())
+                .await
+                .expect("fetch stored ciphertext");
+        let mut bytes = hex::decode(&stored).expect("stored ciphertext is valid hex");
+        let last = bytes.len() - 1;
+        bytes[last] ^= 0xFF;
+        sqlx::query("UPDATE messages SET body_plain = ? WHERE message_key = ?")
+            .bind(hex::encode(bytes))
+            .bind("m-rotten")
+            .execute(db.pool())
+            .await
+            .expect("corrupt stored ciphertext");
+
+        let mut client = mail_client_over(db).await;
+
+        let resp = client
+            .list_messages(authed_bearer_request(ListMessagesRequest {
+                folder_id: "inbox".to_string(),
+                page_size: 10,
+                page_token: String::new(),
+            }))
+            .await
+            .expect("one unreadable row must not fail the whole listing")
+            .into_inner();
+
+        let ids: Vec<&str> = resp.messages.iter().map(|m| m.id.as_str()).collect();
+        assert_eq!(
+            ids,
+            vec!["m-3", "m-1"],
+            "every message that authenticated must still be listed"
+        );
+        assert_eq!(
+            resp.unreadable_message_ids,
+            vec!["m-rotten".to_string()],
+            "the omitted message must be named, not silently missing"
+        );
+
+        let err = client
+            .get_message(authed_bearer_request(GetMessageRequest {
+                message_id: "m-rotten".to_string(),
+            }))
+            .await
+            .expect_err("the single-item read must still fail closed");
+        assert_eq!(err.code(), tonic::Code::NotFound);
+    }
+
     /// A configured account holding nothing in this folder must not truncate
     /// the listing.
     ///
diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs
index f30e8b6..bae7d3c 100644
--- a/crates/nunciod/src/sync.rs
+++ b/crates/nunciod/src/sync.rs
@@ -747,7 +747,12 @@ mod tests {
         assert_eq!(persisted.len(), 2);
         let subjects: Vec<String> = persisted
             .iter()
-            .map(|(email, _placement)| email.subject.clone())
+            .map(|(read, _placement)| {
+                read.as_ref()
+                    .expect("fixture body is readable")
+                    .subject
+                    .clone()
+            })
             .collect();
         assert!(subjects.contains(&"Hello".to_string()));
         assert!(subjects.contains(&"World".to_string()));
@@ -948,7 +953,10 @@ mod tests {
             .expect("jmap message persisted");
         assert_eq!(persisted.len(), 1);
         let (email, placement) = &persisted[0];
-        assert_eq!(email.subject, "Welcome to JMAP Sync");
+        assert_eq!(
+            email.as_ref().expect("fixture body is readable").subject,
+            "Welcome to JMAP Sync"
+        );
         assert_eq!(placement.remote_id, "jmap-msg-1");
 
         assert_eq!(