From f5d8c8a153e5003fb3fa6e592abf5f8725fee5de Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Fri, 14 Aug 2026 20:52:26 -0500
Subject: [PATCH 1/2] fix(cal): refuse phrases the scheduler cannot resolve
 exactly

NaturalLanguageScheduler ignored the requested weekday and time
entirely, always returning the reference instant plus 24 hours, and
never errored on unrecognised phrasing. Every input produced a
plausible-looking event at the wrong time, which a user only discovers
by missing the meeting.

Replace it with a small, exactly-specified grammar:

  <summary> today|tomorrow|YYYY-MM-DD at <time> for <duration>

Everything outside that grammar returns an error. Weekday phrasings
("next Tuesday") are refused as genuinely ambiguous rather than
resolved, since the two common readings differ by a week. A time of
day and a duration are both mandatory, so neither is defaulted.

Resolution now requires an explicit IANA time zone; a bare Unix
timestamp cannot anchor "3pm". Wall-clock times go through the zone's
DST rules and both anomalies are reported: times inside a
spring-forward gap and times inside a fall-back overlap are refused
rather than silently shifted or disambiguated. Durations are added as
elapsed time, so events spanning a transition keep their real length.

parse now returns ParsedSchedulingIntent instead of a CalendarEvent
with invented "default" account and calendar ids. There are no
in-tree callers, so no behaviour visible to a client changes.

Refs #258
---
 crates/nuncio-cal/src/nlp.rs | 724 ++++++++++++++++++++++++++++++++---
 1 file changed, 664 insertions(+), 60 deletions(-)

diff --git a/crates/nuncio-cal/src/nlp.rs b/crates/nuncio-cal/src/nlp.rs
index e4f6819..4be50cc 100644
--- a/crates/nuncio-cal/src/nlp.rs
+++ b/crates/nuncio-cal/src/nlp.rs
@@ -1,101 +1,705 @@
-//! Natural Language Calendar Booking NLP Parser.
+//! A deliberately small, exact natural-language scheduling parser.
+//!
+//! The grammar below is the *entire* set of phrasings this parser understands.
+//! Anything outside it returns an error. That is the point: a scheduler that
+//! guesses produces a plausible-looking event at the wrong instant, and the user
+//! only discovers the mistake by missing the meeting. Refusing is recoverable;
+//! silently booking the wrong time is not.
+//!
+//! # Grammar
+//!
+//! ```text
+//! <summary> <day> at <time> for <duration>
+//! ```
+//!
+//! * `<summary>` — free text, must be non-empty. The `at`/`for` separators are
+//!   matched from the right, so a summary may itself contain them
+//!   ("Prep for review tomorrow at 9am for 30 minutes").
+//! * `<day>` — `today`, `tomorrow`, or an ISO calendar date `YYYY-MM-DD`.
+//! * `<time>` — `noon`, `midnight`, `3pm`, `3:15pm`, `9am`, `9:05am`, or a
+//!   24-hour `HH:MM`. A bare number (`at 3`) is rejected as ambiguous.
+//! * `<duration>` — `<n> minute(s)|min(s)` or `<n> hour(s)|hr(s)`, non-zero and
+//!   no longer than 24 hours.
+//!
+//! Both `at <time>` and `for <duration>` are mandatory. Without a time there is
+//! no instant to resolve, and a defaulted duration is a guess about the user's
+//! intent, so neither is invented.
+//!
+//! # Deliberately unsupported
+//!
+//! Weekday phrasings (`next Tuesday`, `this Friday`, `on Tuesday`) are rejected
+//! as ambiguous rather than resolved. English speakers do not agree on whether
+//! "next Tuesday" means the Tuesday that is coming or the Tuesday of the
+//! following week, and the two readings differ by a whole week. There is no
+//! interpretation this parser could pick that would not be wrong for a large
+//! share of users, so it asks instead.
+//!
+//! # Time zones
+//!
+//! Resolution requires an explicit IANA time zone: "3pm" is not an instant until
+//! you know where the user is, and a Unix timestamp alone does not carry that.
+//! Wall-clock times are resolved through the zone's DST rules, and the two
+//! local-time anomalies are reported rather than papered over:
+//!
+//! * A time inside a spring-forward gap does not exist → [`NlpError::NonexistentLocalTime`].
+//! * A time inside a fall-back overlap occurs twice → [`NlpError::AmbiguousLocalTime`].
+//!
+//! Durations are added as absolute elapsed seconds, so an event that spans a DST
+//! transition keeps its real length and its wall-clock end time shifts, which is
+//! what a physical meeting actually does.
 
-use nuncio_core::model::CalendarEvent;
+use chrono::{Duration, LocalResult, NaiveDate, TimeZone};
+use chrono_tz::Tz;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
-/// Parsed natural language scheduling intent result.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+/// The longest event this parser will accept, in minutes.
+const MAX_DURATION_MINUTES: u32 = 24 * 60;
+
+/// A successfully parsed scheduling request.
+///
+/// This is a parse result, not a calendar event: it deliberately carries no
+/// account or calendar identity, because the parser has no way to know which
+/// calendar the user meant and must not invent one.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ParsedSchedulingIntent {
+    /// Event title, with the trailing scheduling phrase removed.
     pub summary: String,
+    /// Start instant, Unix seconds UTC.
     pub start_time: i64,
+    /// End instant, Unix seconds UTC (`start_time` plus the parsed duration).
     pub end_time: i64,
+    /// Parsed duration in minutes.
     pub duration_minutes: u32,
-    pub location: Option<String>,
-    pub attendees: Vec<String>,
 }
 
-/// Errors originating from natural language scheduling parsing.
-#[derive(Error, Debug, PartialEq, Eq)]
+/// Why a scheduling string could not be resolved to an exact instant.
+#[derive(Error, Debug, Clone, PartialEq, Eq)]
 pub enum NlpError {
-    #[error("failed to parse scheduling string: {0}")]
-    ParseFailed(String),
-
     #[error("empty scheduling query provided")]
     EmptyQuery,
+
+    #[error("no event title found before the scheduling phrase")]
+    MissingSummary,
+
+    #[error("no time of day given; expected `... at <time> ...`")]
+    MissingTime,
+
+    #[error("no duration given; expected `... for <n> minutes`")]
+    MissingDuration,
+
+    #[error(
+        "`{0}` is ambiguous: it can mean either the coming weekday or the one a \
+         week later. Use `today`, `tomorrow`, or an explicit YYYY-MM-DD date"
+    )]
+    AmbiguousWeekday(String),
+
+    #[error("unsupported day `{0}`; expected `today`, `tomorrow`, or YYYY-MM-DD")]
+    UnsupportedDay(String),
+
+    #[error("unsupported time `{0}`; expected `noon`, `midnight`, `3pm`, `3:15pm`, or `HH:MM`")]
+    UnsupportedTime(String),
+
+    #[error("unsupported duration `{0}`; expected `<n> minutes` or `<n> hours`, up to 24 hours")]
+    UnsupportedDuration(String),
+
+    #[error("{date} {time} does not exist in {zone}: the clocks skip forward over it")]
+    NonexistentLocalTime {
+        date: String,
+        time: String,
+        zone: String,
+    },
+
+    #[error("{date} {time} occurs twice in {zone}: the clocks fall back over it")]
+    AmbiguousLocalTime {
+        date: String,
+        time: String,
+        zone: String,
+    },
+
+    #[error("the reference timestamp {0} is not a representable instant")]
+    InvalidReference(i64),
+
+    #[error("the resolved instant overflows the representable range")]
+    InstantOutOfRange,
 }
 
-/// Natural Language Scheduling NLP Parser.
+/// Parser for the small scheduling grammar documented at the module level.
 pub struct NaturalLanguageScheduler;
 
 impl NaturalLanguageScheduler {
-    /// Parse human input string (e.g. "Coffee with Bob next Tuesday at 2pm for 45 minutes") into [`CalendarEvent`].
-    pub fn parse(input: &str, reference_timestamp: i64) -> Result<CalendarEvent, NlpError> {
-        let clean = input.trim();
-        if clean.is_empty() {
+    /// Resolve `input` to an exact instant, relative to `reference_timestamp`
+    /// (Unix seconds) interpreted in `zone`.
+    ///
+    /// `zone` is required rather than defaulted: relative days and wall-clock
+    /// times are meaningless without it, and assuming a zone is the specific
+    /// mistake that produces confidently wrong bookings.
+    ///
+    /// Returns [`NlpError`] for every phrasing outside the documented grammar.
+    /// It never falls back to an approximation.
+    pub fn parse(
+        input: &str,
+        reference_timestamp: i64,
+        zone: Tz,
+    ) -> Result<ParsedSchedulingIntent, NlpError> {
+        let original = input.trim();
+        if original.is_empty() {
             return Err(NlpError::EmptyQuery);
         }
 
-        // Extract duration if specified (e.g., "45 minutes", "1 hour")
-        let duration_secs = if clean.contains("45 min") || clean.contains("45 minutes") {
-            45 * 60
-        } else if clean.contains("30 min") || clean.contains("30 minutes") {
-            30 * 60
-        } else if clean.contains("1 hour") || clean.contains("60 minutes") {
-            60 * 60
-        } else {
-            30 * 60 // Default 30 minutes
-        };
+        // ASCII lowercasing preserves byte length, so byte offsets found in the
+        // lowered copy stay valid in `original` and the summary keeps its case.
+        let lowered = original.to_ascii_lowercase();
 
-        // Determine summary (strip timing phrases)
-        let summary = if let Some((head, _)) = clean.split_once(" next ") {
-            head.to_string()
-        } else if let Some((head, _)) = clean.split_once(" tomorrow ") {
-            head.to_string()
-        } else if let Some((head, _)) = clean.split_once(" at ") {
-            head.to_string()
-        } else {
-            clean.to_string()
-        };
+        let (head, duration_text) =
+            split_last(&lowered, original, " for ").ok_or(NlpError::MissingDuration)?;
+        let duration_minutes = parse_duration(&duration_text.to_ascii_lowercase())?;
 
-        // Standard 24h offset calculation for "next" / "tomorrow"
-        let start = reference_timestamp + 86400;
-        let end = start + duration_secs;
+        let lowered_head = head.to_ascii_lowercase();
+        let (summary_and_day, time_text) =
+            split_last(&lowered_head, head, " at ").ok_or(NlpError::MissingTime)?;
+        let (hour, minute) = parse_time_of_day(&time_text.to_ascii_lowercase())?;
 
-        Ok(CalendarEvent {
-            id: format!("nlp-evt-{}", uuid_or_timestamp(start)),
-            account_id: "default".to_string(),
-            calendar_id: "default".to_string(),
-            summary: summary.trim().to_string(),
-            start_time: start,
-            end_time: end,
-            rrule: None,
-            location: None,
+        let reference = zone
+            .timestamp_opt(reference_timestamp, 0)
+            .single()
+            .ok_or(NlpError::InvalidReference(reference_timestamp))?;
+
+        let (summary, date) = split_day(summary_and_day, reference.date_naive())?;
+        if summary.is_empty() {
+            return Err(NlpError::MissingSummary);
+        }
+
+        let start = resolve_local(zone, date, hour, minute)?;
+        let end = start
+            .checked_add_signed(Duration::minutes(i64::from(duration_minutes)))
+            .ok_or(NlpError::InstantOutOfRange)?;
+
+        Ok(ParsedSchedulingIntent {
+            summary,
+            start_time: start.timestamp(),
+            end_time: end.timestamp(),
+            duration_minutes,
         })
     }
 }
 
-fn uuid_or_timestamp(ts: i64) -> String {
-    format!("evt-{ts}")
+/// Split `original` at the last occurrence of `sep` located in `lowered`.
+///
+/// `lowered` must be the ASCII-lowercased form of `original` so their byte
+/// offsets line up.
+fn split_last<'a>(lowered: &str, original: &'a str, sep: &str) -> Option<(&'a str, &'a str)> {
+    let at = lowered.rfind(sep)?;
+    Some((&original[..at], &original[at + sep.len()..]))
+}
+
+/// Turn the trailing day word of `text` into a concrete date, returning the
+/// remaining text as the summary.
+fn split_day(text: &str, reference_date: NaiveDate) -> Result<(String, NaiveDate), NlpError> {
+    let trimmed = text.trim_end();
+    let (head, last) = match trimmed.rsplit_once(char::is_whitespace) {
+        Some((head, last)) => (head, last),
+        None => ("", trimmed),
+    };
+    let lowered = last.to_ascii_lowercase();
+
+    let date = match lowered.as_str() {
+        "today" => reference_date,
+        "tomorrow" => reference_date
+            .succ_opt()
+            .ok_or(NlpError::InstantOutOfRange)?,
+        other => {
+            if let Ok(explicit) = NaiveDate::parse_from_str(other, "%Y-%m-%d") {
+                explicit
+            } else if is_weekday(other) {
+                // Rebuild the phrase with any qualifier ("next", "this", "on")
+                // so the error quotes back what the user actually wrote.
+                let qualifier = head
+                    .rsplit_once(char::is_whitespace)
+                    .map_or(head, |(_, word)| word);
+                let phrase = if matches!(
+                    qualifier.to_ascii_lowercase().as_str(),
+                    "next" | "this" | "on" | "coming"
+                ) {
+                    format!("{qualifier} {last}")
+                } else {
+                    last.to_string()
+                };
+                return Err(NlpError::AmbiguousWeekday(phrase));
+            } else {
+                return Err(NlpError::UnsupportedDay(last.to_string()));
+            }
+        }
+    };
+
+    Ok((head.trim().to_string(), date))
+}
+
+fn is_weekday(word: &str) -> bool {
+    matches!(
+        word,
+        "monday"
+            | "mon"
+            | "tuesday"
+            | "tue"
+            | "tues"
+            | "wednesday"
+            | "wed"
+            | "thursday"
+            | "thu"
+            | "thurs"
+            | "friday"
+            | "fri"
+            | "saturday"
+            | "sat"
+            | "sunday"
+            | "sun"
+    )
+}
+
+/// Parse `noon`, `midnight`, a 12-hour `am`/`pm` time, or a 24-hour `HH:MM`.
+///
+/// A bare number is rejected: "at 3" gives no way to tell 03:00 from 15:00.
+fn parse_time_of_day(text: &str) -> Result<(u32, u32), NlpError> {
+    let raw = text.trim();
+    let compact: String = raw.chars().filter(|c| !c.is_whitespace()).collect();
+    let unsupported = || NlpError::UnsupportedTime(raw.to_string());
+
+    match compact.as_str() {
+        "noon" => return Ok((12, 0)),
+        "midnight" => return Ok((0, 0)),
+        _ => {}
+    }
+
+    if let Some(clock) = compact
+        .strip_suffix("am")
+        .map(|c| (c, false))
+        .or_else(|| compact.strip_suffix("pm").map(|c| (c, true)))
+    {
+        let (digits, is_pm) = clock;
+        let (hour, minute) = split_clock(digits, 1..=12, unsupported)?;
+        let hour = match (hour, is_pm) {
+            (12, false) => 0,
+            (12, true) => 12,
+            (h, false) => h,
+            (h, true) => h + 12,
+        };
+        return Ok((hour, minute));
+    }
+
+    // 24-hour form must carry the colon, so a bare hour never parses.
+    if !compact.contains(':') {
+        return Err(unsupported());
+    }
+    split_clock(&compact, 0..=23, unsupported)
+}
+
+/// Split `H`, `H:MM`, or `HH:MM` into an hour within `hours` and a minute.
+fn split_clock(
+    text: &str,
+    hours: std::ops::RangeInclusive<u32>,
+    unsupported: impl Fn() -> NlpError,
+) -> Result<(u32, u32), NlpError> {
+    let (hour_text, minute_text) = match text.split_once(':') {
+        Some((h, m)) => (h, m),
+        None => (text, "0"),
+    };
+    if hour_text.is_empty() || !hour_text.chars().all(|c| c.is_ascii_digit()) {
+        return Err(unsupported());
+    }
+    // A minute field is always exactly two digits; "3:5" is a typo, not 03:05.
+    if text.contains(':')
+        && (minute_text.len() != 2 || !minute_text.bytes().all(|b| b.is_ascii_digit()))
+    {
+        return Err(unsupported());
+    }
+    let hour: u32 = hour_text.parse().map_err(|_| unsupported())?;
+    let minute: u32 = minute_text.parse().map_err(|_| unsupported())?;
+    if !hours.contains(&hour) || minute > 59 {
+        return Err(unsupported());
+    }
+    Ok((hour, minute))
+}
+
+/// Parse `<n> minutes` / `<n> hours` into a minute count.
+fn parse_duration(text: &str) -> Result<u32, NlpError> {
+    let raw = text.trim();
+    let unsupported = || NlpError::UnsupportedDuration(raw.to_string());
+
+    let mut parts = raw.split_whitespace();
+    let (Some(count_text), Some(unit), None) = (parts.next(), parts.next(), parts.next()) else {
+        return Err(unsupported());
+    };
+
+    let count: u32 = count_text.parse().map_err(|_| unsupported())?;
+    let minutes = match unit {
+        "minute" | "minutes" | "min" | "mins" => count,
+        "hour" | "hours" | "hr" | "hrs" => count.checked_mul(60).ok_or_else(unsupported)?,
+        _ => return Err(unsupported()),
+    };
+
+    if minutes == 0 || minutes > MAX_DURATION_MINUTES {
+        return Err(unsupported());
+    }
+    Ok(minutes)
+}
+
+/// Resolve a wall-clock time in `zone`, reporting both DST anomalies honestly.
+fn resolve_local(
+    zone: Tz,
+    date: NaiveDate,
+    hour: u32,
+    minute: u32,
+) -> Result<chrono::DateTime<Tz>, NlpError> {
+    let local = date
+        .and_hms_opt(hour, minute, 0)
+        .ok_or(NlpError::InstantOutOfRange)?;
+
+    match zone.from_local_datetime(&local) {
+        LocalResult::Single(instant) => Ok(instant),
+        LocalResult::None => Err(NlpError::NonexistentLocalTime {
+            date: date.to_string(),
+            time: format!("{hour:02}:{minute:02}"),
+            zone: zone.name().to_string(),
+        }),
+        LocalResult::Ambiguous(_, _) => Err(NlpError::AmbiguousLocalTime {
+            date: date.to_string(),
+            time: format!("{hour:02}:{minute:02}"),
+            zone: zone.name().to_string(),
+        }),
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use chrono_tz::{America::New_York, Europe::London};
+
+    // 2026-03-05 10:00 EST — a plain winter weekday, well clear of any DST edge.
+    const REF_NY: i64 = 1_772_722_800;
+    // 2026-03-07 12:00 EST — the day before US DST begins (2026-03-08 02:00).
+    const REF_NY_BEFORE_SPRING_FORWARD: i64 = 1_772_902_800;
+    // 2026-10-31 12:00 EDT — the day before US DST ends (2026-11-01 02:00).
+    const REF_NY_BEFORE_FALL_BACK: i64 = 1_793_462_400;
+    // 2026-03-05 10:00 GMT — same wall clock as REF_NY, a different instant.
+    const REF_LONDON: i64 = 1_772_704_800;
+
+    fn parse(input: &str, reference: i64, zone: Tz) -> ParsedSchedulingIntent {
+        NaturalLanguageScheduler::parse(input, reference, zone).unwrap()
+    }
+
+    fn err(input: &str, reference: i64, zone: Tz) -> NlpError {
+        NaturalLanguageScheduler::parse(input, reference, zone).unwrap_err()
+    }
+
+    // --- supported grammar: exact instants -------------------------------
 
     #[test]
-    fn parse_valid_nlp_string_creates_event() {
-        let input = "Coffee with Bob next Tuesday at 2pm for 45 minutes";
-        let ref_ts = 1700000000;
-        let event = NaturalLanguageScheduler::parse(input, ref_ts).unwrap();
-
-        assert_eq!(event.summary, "Coffee with Bob");
-        assert_eq!(event.end_time - event.start_time, 45 * 60);
+    fn tomorrow_at_a_pm_time_resolves_to_the_exact_instant() {
+        let intent = parse(
+            "Coffee with Bob tomorrow at 2pm for 45 minutes",
+            REF_NY,
+            New_York,
+        );
+        assert_eq!(intent.summary, "Coffee with Bob");
+        // 2026-03-06 14:00 EST.
+        assert_eq!(intent.start_time, 1_772_823_600);
+        assert_eq!(intent.end_time, 1_772_823_600 + 45 * 60);
+        assert_eq!(intent.duration_minutes, 45);
     }
 
     #[test]
-    fn parse_empty_query_returns_error() {
-        let err = NaturalLanguageScheduler::parse("   ", 1700000000).unwrap_err();
-        assert_eq!(err, NlpError::EmptyQuery);
+    fn noon_and_midnight_resolve_to_the_exact_instant() {
+        let noon = parse("Lunch tomorrow at noon for 1 hour", REF_NY, New_York);
+        assert_eq!(noon.start_time, 1_772_816_400); // 2026-03-06 12:00 EST
+        assert_eq!(noon.duration_minutes, 60);
+
+        let midnight = parse("Deploy tomorrow at midnight for 30 mins", REF_NY, New_York);
+        assert_eq!(midnight.start_time, 1_772_773_200); // 2026-03-06 00:00 EST
+    }
+
+    #[test]
+    fn today_resolves_against_the_reference_date_not_the_wall_clock() {
+        let intent = parse("Retro today at 9:30am for 30 minutes", REF_NY, New_York);
+        assert_eq!(intent.summary, "Retro");
+        // 2026-03-05 09:30 EST — before the reference instant, and still exact.
+        assert_eq!(intent.start_time, 1_772_721_000);
+    }
+
+    #[test]
+    fn an_explicit_iso_date_resolves_to_the_exact_instant() {
+        let intent = parse(
+            "Standup 2026-03-08 at 01:30 for 120 minutes",
+            REF_NY,
+            New_York,
+        );
+        assert_eq!(intent.summary, "Standup");
+        assert_eq!(intent.start_time, 1_772_951_400); // 2026-03-08 01:30 EST
+    }
+
+    #[test]
+    fn separators_are_matched_from_the_right_so_summaries_may_contain_them() {
+        let intent = parse(
+            "Prep for review at HQ tomorrow at 2pm for 45 minutes",
+            REF_NY,
+            New_York,
+        );
+        assert_eq!(intent.summary, "Prep for review at HQ");
+        assert_eq!(intent.start_time, 1_772_823_600);
+    }
+
+    #[test]
+    fn every_documented_duration_unit_is_accepted() {
+        for (phrase, expected) in [
+            ("15 minutes", 15),
+            ("15 minute", 15),
+            ("15 mins", 15),
+            ("15 min", 15),
+            ("2 hours", 120),
+            ("1 hour", 60),
+            ("2 hrs", 120),
+            ("1 hr", 60),
+        ] {
+            let input = format!("Sync tomorrow at 2pm for {phrase}");
+            assert_eq!(
+                parse(&input, REF_NY, New_York).duration_minutes,
+                expected,
+                "{phrase}"
+            );
+        }
+    }
+
+    #[test]
+    fn every_documented_time_form_is_accepted() {
+        for (phrase, expected_hour, expected_minute) in [
+            ("3pm", 15, 0),
+            ("3:15pm", 15, 15),
+            ("9am", 9, 0),
+            ("9:05am", 9, 5),
+            ("12am", 0, 0),
+            ("12pm", 12, 0),
+            ("00:00", 0, 0),
+            ("23:59", 23, 59),
+            ("noon", 12, 0),
+            ("midnight", 0, 0),
+        ] {
+            let input = format!("Sync tomorrow at {phrase} for 30 minutes");
+            let intent = parse(&input, REF_NY, New_York);
+            let expected = New_York
+                .with_ymd_and_hms(2026, 3, 6, expected_hour, expected_minute, 0)
+                .unwrap()
+                .timestamp();
+            assert_eq!(intent.start_time, expected, "{phrase}");
+        }
+    }
+
+    // --- time zones -------------------------------------------------------
+
+    #[test]
+    fn the_same_phrase_in_a_different_zone_is_a_different_instant() {
+        let input = "Sync tomorrow at 2pm for 30 minutes";
+        let ny = parse(input, REF_NY, New_York);
+        let london = parse(input, REF_LONDON, London);
+
+        assert_eq!(ny.start_time, 1_772_823_600); // 2026-03-06 14:00 EST
+        assert_eq!(london.start_time, 1_772_805_600); // 2026-03-06 14:00 GMT
+                                                      // Same wall clock, five hours apart — the zone is load-bearing, so a
+                                                      // parser that ignored it would be wrong for one of these callers.
+        assert_eq!(ny.start_time - london.start_time, 5 * 3600);
+    }
+
+    #[test]
+    fn a_time_inside_the_spring_forward_gap_is_refused() {
+        let error = err(
+            "Sync tomorrow at 2:30am for 30 minutes",
+            REF_NY_BEFORE_SPRING_FORWARD,
+            New_York,
+        );
+        assert_eq!(
+            error,
+            NlpError::NonexistentLocalTime {
+                date: "2026-03-08".to_string(),
+                time: "02:30".to_string(),
+                zone: "America/New_York".to_string(),
+            }
+        );
+    }
+
+    #[test]
+    fn a_time_just_after_the_spring_forward_gap_resolves_exactly() {
+        let intent = parse(
+            "Sync tomorrow at 3am for 30 minutes",
+            REF_NY_BEFORE_SPRING_FORWARD,
+            New_York,
+        );
+        // 2026-03-08 03:00 EDT — one hour of real time after 01:00 EST, not two.
+        assert_eq!(intent.start_time, 1_772_953_200);
+    }
+
+    #[test]
+    fn a_time_inside_the_fall_back_overlap_is_refused() {
+        let error = err(
+            "Sync tomorrow at 1:30am for 30 minutes",
+            REF_NY_BEFORE_FALL_BACK,
+            New_York,
+        );
+        assert_eq!(
+            error,
+            NlpError::AmbiguousLocalTime {
+                date: "2026-11-01".to_string(),
+                time: "01:30".to_string(),
+                zone: "America/New_York".to_string(),
+            }
+        );
+    }
+
+    #[test]
+    fn a_duration_spanning_a_dst_transition_keeps_its_real_length() {
+        let intent = parse(
+            "Standup 2026-03-08 at 01:30 for 120 minutes",
+            REF_NY,
+            New_York,
+        );
+        // Elapsed time is exactly two hours...
+        assert_eq!(intent.end_time - intent.start_time, 120 * 60);
+        // ...but the clocks jumped, so the wall-clock end is 04:30, not 03:30.
+        let end = New_York.timestamp_opt(intent.end_time, 0).unwrap();
+        assert_eq!(
+            end.format("%Y-%m-%d %H:%M %Z").to_string(),
+            "2026-03-08 04:30 EDT"
+        );
+    }
+
+    // --- refusals ---------------------------------------------------------
+
+    #[test]
+    fn empty_and_blank_queries_are_refused() {
+        assert_eq!(err("", REF_NY, New_York), NlpError::EmptyQuery);
+        assert_eq!(err("   ", REF_NY, New_York), NlpError::EmptyQuery);
+    }
+
+    #[test]
+    fn weekday_phrasings_are_refused_as_ambiguous_rather_than_guessed() {
+        for (input, quoted) in [
+            ("Coffee next Tuesday at 2pm for 45 minutes", "next Tuesday"),
+            ("Coffee this Friday at 2pm for 45 minutes", "this Friday"),
+            ("Coffee on Monday at 2pm for 45 minutes", "on Monday"),
+            ("Coffee Tuesday at 2pm for 45 minutes", "Tuesday"),
+        ] {
+            assert_eq!(
+                err(input, REF_NY, New_York),
+                NlpError::AmbiguousWeekday(quoted.to_string()),
+                "{input}"
+            );
+        }
+    }
+
+    #[test]
+    fn a_missing_time_or_duration_is_refused_rather_than_defaulted() {
+        assert_eq!(
+            err("Coffee tomorrow at 2pm", REF_NY, New_York),
+            NlpError::MissingDuration
+        );
+        assert_eq!(
+            err("Coffee tomorrow for 45 minutes", REF_NY, New_York),
+            NlpError::MissingTime
+        );
+        assert_eq!(
+            err("Coffee with Bob", REF_NY, New_York),
+            NlpError::MissingDuration
+        );
+    }
+
+    #[test]
+    fn a_missing_summary_is_refused() {
+        assert_eq!(
+            err("tomorrow at 2pm for 45 minutes", REF_NY, New_York),
+            NlpError::MissingSummary
+        );
+    }
+
+    #[test]
+    fn unsupported_day_phrasings_are_refused() {
+        for phrase in [
+            "next week",
+            "in 3 days",
+            "the 5th",
+            "eod",
+            "yesterday",
+            "fortnight",
+        ] {
+            let input = format!("Sync {phrase} at 2pm for 30 minutes");
+            let error = err(&input, REF_NY, New_York);
+            assert!(
+                matches!(error, NlpError::UnsupportedDay(_)),
+                "{phrase} produced {error:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn a_bare_hour_is_refused_because_it_cannot_be_disambiguated() {
+        for phrase in [
+            "3",
+            "15",
+            "half past three",
+            "teatime",
+            "3 o'clock",
+            "25:00",
+            "13pm",
+        ] {
+            let input = format!("Sync tomorrow at {phrase} for 30 minutes");
+            let error = err(&input, REF_NY, New_York);
+            assert!(
+                matches!(error, NlpError::UnsupportedTime(_)),
+                "{phrase} produced {error:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn malformed_minute_fields_are_refused() {
+        for phrase in ["3:5pm", "3:60pm", "3:", ":30", "3:015"] {
+            let input = format!("Sync tomorrow at {phrase} for 30 minutes");
+            let error = err(&input, REF_NY, New_York);
+            assert!(
+                matches!(error, NlpError::UnsupportedTime(_)),
+                "{phrase} produced {error:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn unsupported_or_out_of_range_durations_are_refused() {
+        for phrase in [
+            "a while",
+            "0 minutes",
+            "1.5 hours",
+            "25 hours",
+            "1441 minutes",
+            "30",
+            "half an hour",
+            "2 days",
+        ] {
+            let input = format!("Sync tomorrow at 2pm for {phrase}");
+            let error = err(&input, REF_NY, New_York);
+            assert!(
+                matches!(error, NlpError::UnsupportedDuration(_)),
+                "{phrase} produced {error:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn an_invalid_calendar_date_is_refused() {
+        let error = err("Sync 2026-02-30 at 2pm for 30 minutes", REF_NY, New_York);
+        assert!(matches!(error, NlpError::UnsupportedDay(_)), "{error:?}");
+    }
+
+    #[test]
+    fn the_boundary_duration_of_exactly_24_hours_is_accepted() {
+        let intent = parse("Hackathon tomorrow at noon for 24 hours", REF_NY, New_York);
+        assert_eq!(intent.duration_minutes, MAX_DURATION_MINUTES);
+        assert_eq!(intent.end_time - intent.start_time, 24 * 3600);
     }
 }
-- 
2.54.0


From c484fb048ed0f452d70eaa787db78aeab1d60879 Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Fri, 14 Aug 2026 21:07:14 -0500
Subject: [PATCH 2/2] fix(store): survive one corrupt body on bulk read paths

A single unauthenticatable body ciphertext took down whole operations:
the FTS backfill runs inside migrate(), so one bit-rotted row stopped
the daemon booting at all, and listing, triage and export each failed
outright over one bad row out of fifty.

Single-item get_message stays fail-closed and is unchanged. It names one
message, so there is no other row left to hand back, and a message with
an empty body reads exactly like a genuinely empty one.

Bulk paths now return each row as MessageRead (Ok(email) or
Err(UnreadableBody)) instead of failing the call. The row keeps its slot,
which is what lets keyset cursors and chunk boundaries stay exact, and no
row ever comes back carrying a body the store could not authenticate.
Each failure is logged at warn! naming only the message key.

The daemon reports the omissions on the wire rather than swallowing them:
unreadable_message_ids on ListMessagesResponse and ExportResponse (an
archive outlives the store, so its gap is named), unreadable_count on
TriageProgress, and the CLI prints all three. Triage never evaluates
rules against a body that failed authentication. The FTS backfill skips
the row entirely rather than indexing an empty body, so every subsequent
open re-reports it and a repaired row heals itself.

Refs #265
---
 crates/nuncio-cli/src/runner.rs               |  48 +-
 .../proto/nuncio/v1/descriptor.bin            | Bin 18496 -> 18649 bytes
 .../nuncio-proto/proto/nuncio/v1/nuncio.proto |  28 +
 crates/nuncio-store/src/db.rs                 | 574 ++++++++++++++----
 crates/nunciod/src/grpc.rs                    | 172 +++++-
 crates/nunciod/src/sync.rs                    |  12 +-
 6 files changed, 684 insertions(+), 150 deletions(-)

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 94f9059fe4a2e80ae2dcb72296afc4dc0d87de46..ccb57609f600377fc50140eb955fdea00ea03c18 100644
GIT binary patch
delta 205
zcmX>wf$`=<#tn>2Ojmn1Gc)ZHVOqjCd7oIGnUD#WSZQ8SYGO)aQch}oZfbFHVtQ(P
zW=gRHvjVdQXOIX&))yk_nNqx2NBoH(<JZZ8ny0meG`R#2YT}dgOY=%3*c2EwSc3Qw
e;?5x9$#*o@F>Pd=+^toO&teul7PDzv$pQdZr$z|?

delta 61
zcmV-D0K)&-kpaMv0k8oA0=$f~0|L4f0dAAO7H$V3G71q#WptB17%j7x7qAfl%9BVb
T%acATt^s?K?kan;pDQ{YncWr)

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!(
-- 
2.54.0