DLP phase 3: hold for review

The hold action now holds (dlp-and-mail-flow-rules spec, §2.6), where
until now it blocked.

- At DATA a hold decision queues the message with its release a century
  off (the queue's future-release mechanism, so the stored format is
  unchanged and an older node just never sends it), transport rules
  still applied, and replies 250 Held for review. A review record under
  R/h + queue id keeps the sender, recipients, subject, size, rules and
  detector counts. The sender is told when the rule asks.
- smtp/queue/held.rs: release (each recipient due now, its next notice
  as far off as it was, its lifetime counted from the release), reject
  (removed from the queue, the sender told, with the reviewer's note),
  and expiry: the daily clean-up rejects what nobody reviewed in 7 days,
  recorded as the server's doing.
- inbuxa:HeldMessage get/set: the review queue, sysDlpReviewGet to list
  and read (preview, 64 KB of text, only when asked for and recorded as
  blobAccess), sysDlpReviewUpdate to release or reject, a reason
  required and audited by the request layer; no create or destroy;
  server-level only.
- Guards: Emails > Queue refuses to change or delete held mail; the
  sender can't unsend it.
- Privacy catalog entry for inbuxa:HeldMessage; spec §2.6 as built.

Tests: mail_rules_tests gains the whole flow (held and listed with
counts, sender notified and nothing delivered, queue and unsend
refused, preview recorded, reject needs a reason and tells the sender
the note, release delivers, expiry returns it, decisions audited with
reasons). smtp inbound, system_tests (after one BlobNotFound in
antispam, the known flake, then clean), features and common unit tests.

crates/smtp/src/inbound/data.rs

@@ -740,35 +740,46 @@ impl<T: SessionStream> Session<T> {
 
         // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.1): after the system
         // script, before headers and signing
-        match self
+        let mut held_draft = None;
+        let (message, envelope) = match self
             .check_mail_rules(edited_message.as_deref().unwrap_or(raw_message.as_slice()))
             .await
         {
-            super::mailflow::Checked::Accept => {}
-            super::mailflow::Checked::Changed { message, envelope } => {
-                if let Some(message) = message {
-                    edited_message = Some(message);
-                }
-                for change in envelope {
-                    match change {
-                        super::mailflow::EnvelopeChange::AddRecipient(address) => {
-                            if !self.data.rcpt_to.iter().any(|r| r.address_lcase.eq_ignore_ascii_case(&address)) {
-                                self.data.rcpt_to.push(SessionAddress::new(address));
-                            }
-                        }
-                        super::mailflow::EnvelopeChange::Redirect(addresses) => {
-                            self.data.rcpt_to = addresses.into_iter().map(SessionAddress::new).collect();
-                        }
-                        super::mailflow::EnvelopeChange::Route(queue) => {
-                            self.data.mailflow_queue = Some(queue);
-                        }
-                    }
-                }
+            super::mailflow::Checked::Accept => (None, Vec::new()),
+            super::mailflow::Checked::Changed { message, envelope } => (message, envelope),
+            // §2.6: queued, but not due for a century; a reviewer releases it
+            super::mailflow::Checked::Hold { draft, message, envelope } => {
+                self.data.future_release = inbuxa_features::mailflow::held::HOLD_SECONDS;
+                held_draft = Some(draft);
+                (message, envelope)
             }
             super::mailflow::Checked::Refuse(reply, refusal) => {
                 self.data.dlp_refusal = refusal;
                 return reply.into();
             }
+        };
+        if let Some(message) = message {
+            edited_message = Some(message);
+        }
+        for change in envelope {
+            match change {
+                super::mailflow::EnvelopeChange::AddRecipient(address) => {
+                    if !self
+                        .data
+                        .rcpt_to
+                        .iter()
+                        .any(|r| r.address_lcase.eq_ignore_ascii_case(&address))
+                    {
+                        self.data.rcpt_to.push(SessionAddress::new(address));
+                    }
+                }
+                super::mailflow::EnvelopeChange::Redirect(addresses) => {
+                    self.data.rcpt_to = addresses.into_iter().map(SessionAddress::new).collect();
+                }
+                super::mailflow::EnvelopeChange::Route(queue) => {
+                    self.data.mailflow_queue = Some(queue);
+                }
+            }
         }
 
         // Build message
@@ -850,6 +861,19 @@ impl<T: SessionStream> Session<T> {
                 .server
                 .eval_signers(&ac.dkim.sign, self, self.data.session_id)
                 .await;
+            // inbuxa: §2.6, who the held message is from and to
+            let held_envelope = held_draft.as_ref().map(|_| {
+                (
+                    message.message.return_path.to_string(),
+                    message
+                        .message
+                        .recipients
+                        .iter()
+                        .map(|r| r.address.to_string())
+                        .collect::<Vec<_>>(),
+                    message.message.size,
+                )
+            });
             if message
                 .queue(
                     QueueParams::new(raw_message, self.data.session_id, &self.server)
@@ -864,9 +888,17 @@ impl<T: SessionStream> Session<T> {
             {
                 self.state = State::Accepted(queue_id);
                 self.data.messages_sent += 1;
-                format!("250 2.0.0 Message queued with id {queue_id:x}.\r\n")
-                    .into_bytes()
-                    .into()
+                if let (Some(draft), Some((sender, recipients, size))) = (held_draft, held_envelope)
+                {
+                    self.record_held(queue_id, draft, sender, recipients, size).await;
+                    format!("250 2.0.0 Held for review, id {queue_id:x}.\r\n")
+                        .into_bytes()
+                        .into()
+                } else {
+                    format!("250 2.0.0 Message queued with id {queue_id:x}.\r\n")
+                        .into_bytes()
+                        .into()
+                }
             } else {
                 (b"451 4.3.5 Unable to accept message at this time.\r\n"[..]).into()
             }

crates/smtp/src/inbound/mailflow.rs

@@ -21,6 +21,7 @@ use inbuxa_features::{
             Attachment, Content, Decision, Envelope, Outcome as RulesOutcome, Recipient, RuleRef,
         },
         extract::{self, Extracted, Limits},
+        held::{self, Held, HeldRule, KEEP_DAYS},
         rewrite,
         rules::{Action as RuleAction, Kind},
     },
@@ -44,6 +45,20 @@ pub enum Checked {
     },
     /// Refuse, with this SMTP reply, and for a JMAP submission, why.
     Refuse(Vec<u8>, Option<DlpRefusal>),
+    /// Queue it held for review (§2.6), with any transport changes.
+    Hold {
+        draft: HeldDraft,
+        message: Option<Vec<u8>>,
+        envelope: Vec<EnvelopeChange>,
+    },
+}
+
+/// What the review record will say, once the message has a queue id.
+pub struct HeldDraft {
+    pub subject: String,
+    pub rules: Vec<HeldRule>,
+    pub counts: Vec<(String, usize)>,
+    pub notify_sender: bool,
 }
 
 /// What a transport rule changes about where a message goes.
@@ -219,6 +234,7 @@ impl<T: SessionStream> Session<T> {
             None
         };
         let checked_subject = tag.as_ref().map_or(subject, |(_, rest)| rest.as_str());
+        let held_subject = checked_subject.to_string();
 
         let mut content = Content {
             subject: checked_subject,
@@ -319,22 +335,51 @@ impl<T: SessionStream> Session<T> {
             .await;
         }
 
-        match decision {
-            Decision::Block(rules) | Decision::Hold { rules, .. } => {
-                // Hold for review is phase 3: until then a hold rule blocks,
-                // rather than let the message through unreviewed
+        let hold = match decision {
+            Decision::Block(rules) => {
                 let refusal = refusal(true, &rules);
-                Checked::Refuse(format!("550 5.7.1 {}\r\n", notices(&rules)).into_bytes(), Some(refusal))
+                return Checked::Refuse(
+                    format!("550 5.7.1 {}\r\n", notices(&rules)).into_bytes(),
+                    Some(refusal),
+                );
             }
-            Decision::Warn(rules) => Checked::Refuse(
-                format!(
-                    "550 5.7.1 {} To send anyway, start the subject with [override: your reason]\r\n",
-                    notices(&rules)
-                )
-                .into_bytes(),
-                Some(refusal(false, &rules)),
-            ),
-            Decision::Pass => {
+            Decision::Warn(rules) => {
+                return Checked::Refuse(
+                    format!(
+                        "550 5.7.1 {} To send anyway, start the subject with [override: your reason]\r\n",
+                        notices(&rules)
+                    )
+                    .into_bytes(),
+                    Some(refusal(false, &rules)),
+                );
+            }
+            // Accepted and queued, but not sent until a reviewer says so
+            // (§2.6); the transport rules still apply, so what's released
+            // is what would have gone out
+            Decision::Hold {
+                rules,
+                notify_sender,
+            } => Some(HeldDraft {
+                subject: held_subject,
+                rules: rules
+                    .iter()
+                    .map(|r| HeldRule {
+                        name: r.name.clone(),
+                        notice: r.notice.clone(),
+                    })
+                    .collect(),
+                counts: outcome
+                    .matched
+                    .iter()
+                    .filter(|m| m.kind == Kind::Dlp)
+                    .flat_map(|m| m.counts.iter().cloned())
+                    .collect(),
+                notify_sender,
+            }),
+            Decision::Pass => None,
+        };
+        {
+            {
                 // The tag was an instruction to the server, not part of the
                 // subject: it doesn't go out
                 let mut current: Option<Vec<u8>> =
@@ -344,12 +389,18 @@ impl<T: SessionStream> Session<T> {
                     for action in &matched.actions {
                         let now = current.as_deref().unwrap_or(message);
                         let next = match action {
-                            RuleAction::AddDisclaimer { text, html, position } => {
-                                rewrite::add_disclaimer(now, text, html.as_deref(), *position)
+                            RuleAction::AddDisclaimer {
+                                text,
+                                html,
+                                position,
+                            } => rewrite::add_disclaimer(now, text, html.as_deref(), *position),
+                            RuleAction::AddHeader { name, value } => {
+                                Some(rewrite::add_header(now, name, value))
                             }
-                            RuleAction::AddHeader { name, value } => Some(rewrite::add_header(now, name, value)),
                             RuleAction::RemoveHeader { name } => rewrite::remove_header(now, name),
-                            RuleAction::PrefixSubject { text } => rewrite::prefix_subject(now, text),
+                            RuleAction::PrefixSubject { text } => {
+                                rewrite::prefix_subject(now, text)
+                            }
                             RuleAction::AddRecipient { address } => {
                                 changes.push(EnvelopeChange::AddRecipient(address.clone()));
                                 None
@@ -363,13 +414,16 @@ impl<T: SessionStream> Session<T> {
                                 None
                             }
                             RuleAction::Refuse { text } => {
-                                self.record_transport(&sender, &matched.name, "refused", &domains).await;
+                                self.record_transport(&sender, &matched.name, "refused", &domains)
+                                    .await;
                                 return Checked::Refuse(
                                     format!("550 5.7.1 {}\r\n", reply_text(text)).into_bytes(),
                                     None,
                                 );
                             }
-                            RuleAction::Block { .. } | RuleAction::Warn { .. } | RuleAction::Hold { .. } => None,
+                            RuleAction::Block { .. }
+                            | RuleAction::Warn { .. }
+                            | RuleAction::Hold { .. } => None,
                         };
                         if next.is_some() {
                             current = next;
@@ -381,25 +435,76 @@ impl<T: SessionStream> Session<T> {
                         .actions
                         .iter()
                         .filter_map(|a| match a {
-                            RuleAction::AddRecipient { address } => Some(format!("copied to {address}")),
-                            RuleAction::Redirect { addresses } => Some(format!("redirected to {}", addresses.join(", "))),
+                            RuleAction::AddRecipient { address } => {
+                                Some(format!("copied to {address}"))
+                            }
+                            RuleAction::Redirect { addresses } => {
+                                Some(format!("redirected to {}", addresses.join(", ")))
+                            }
                             RuleAction::Route { queue } => Some(format!("routed through {queue}")),
                             _ => None,
                         })
                         .collect();
                     if !routed.is_empty() {
-                        self.record_transport(&sender, &matched.name, &routed.join(", "), &domains).await;
+                        self.record_transport(&sender, &matched.name, &routed.join(", "), &domains)
+                            .await;
                     }
                 }
-                if current.is_none() && changes.is_empty() {
-                    Checked::Accept
-                } else {
-                    Checked::Changed { message: current, envelope: changes }
+                match hold {
+                    Some(draft) => Checked::Hold {
+                        draft,
+                        message: current,
+                        envelope: changes,
+                    },
+                    None if current.is_none() && changes.is_empty() => Checked::Accept,
+                    None => Checked::Changed {
+                        message: current,
+                        envelope: changes,
+                    },
                 }
             }
         }
     }
 
+    /// Writes the review record for a message just queued held (§2.6),
+    /// and tells the sender when the rule asks. A failure to write it is
+    /// logged: the message stays held, never sent unreviewed.
+    pub async fn record_held(
+        &self,
+        queue_id: u64,
+        draft: HeldDraft,
+        sender: String,
+        recipients: Vec<String>,
+        size: u64,
+    ) {
+        let at = store::write::now();
+        let account = self.data.authenticated_as.as_ref();
+        let record = Held {
+            queue_id,
+            sender,
+            account_id: account.map(|a| a.account_id),
+            tenant_id: account.and_then(|a| a.account.id_tenant),
+            recipients,
+            subject: draft.subject,
+            size,
+            rules: draft.rules,
+            counts: draft.counts,
+            held_at: at,
+            expires_at: at + KEEP_DAYS * 86_400,
+        };
+        if let Err(err) = held::create(self.server.store(), &record).await {
+            trc::error!(
+                err.span_id(self.data.session_id)
+                    .caused_by(trc::location!())
+                    .details("Failed to write the review record of a held message")
+            );
+            return;
+        }
+        if draft.notify_sender {
+            crate::queue::held::notify_held(&self.server, &record).await;
+        }
+    }
+
     /// A transport rule that refused a message or changed where it goes
     /// (§2.7): who sent it (or the server, for incoming mail), the rule,
     /// what it did.
@@ -485,9 +590,8 @@ impl<T: SessionStream> Session<T> {
             .collect::<Vec<_>>()
             .join("; ");
         let (what, outcome, reason) = match decision {
-            Decision::Block(_) | Decision::Hold { .. } => {
-                ("blocked", Outcome::refused("inbuxa:dlpBlocked", None), None)
-            }
+            Decision::Hold { .. } => ("held for review", Outcome::success(), None),
+            Decision::Block(_) => ("blocked", Outcome::refused("inbuxa:dlpBlocked", None), None),
             Decision::Warn(_) => ("warned", Outcome::refused("inbuxa:dlpWarning", None), None),
             Decision::Pass => (
                 "sent after a warning",

crates/smtp/src/queue/held.rs

@@ -0,0 +1,177 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! inbuxa: mail held for review (dlp-and-mail-flow-rules spec, §2.6):
+//! releasing it, rejecting it, rejecting what nobody reviewed in time, and
+//! the notices the sender gets.
+//!
+//! A held message sits in the queue with its release [`HOLD_SECONDS`] off.
+//! Releasing it undoes exactly that: each recipient due now, its next
+//! notice as far from now as it was from its retry, its lifetime counted
+//! from the release.
+
+use crate::{
+    queue::{Message, MessageWrapper, Status, spool::SmtpSpool},
+    reporting::send::MtaReportSend,
+};
+use common::{
+    Server,
+    config::smtp::queue::{QueueExpiry, QueueName},
+    ipc::QueueEvent,
+};
+use inbuxa_features::{
+    audit::{Action, Actor, Outcome, Record, Target},
+    mailflow::held::{self, HOLD_SECONDS, Held, KEEP_DAYS},
+};
+use mail_builder::{
+    MessageBuilder,
+    headers::{HeaderType, address::Address},
+};
+use store::{ahash::AHashSet, write::now};
+
+/// Puts a held message back on its way. False when it's no longer queued.
+pub async fn release(server: &Server, queue_id: u64) -> trc::Result<bool> {
+    let Some(archive) = server.read_message_archive(queue_id).await? else {
+        held::delete(server.store(), queue_id).await?;
+        return Ok(false);
+    };
+    let mut message: Message = archive.to_unarchived::<Message>()?.deserialize()?;
+    let prev_events = message.next_events();
+    let at = now();
+    let mut modified = AHashSet::new();
+    for (idx, rcpt) in message.recipients.iter_mut().enumerate() {
+        if !matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_)) {
+            continue;
+        }
+        let notify_gap = rcpt.notify.due.saturating_sub(rcpt.retry.due);
+        rcpt.retry.due = at;
+        rcpt.notify.due = at + notify_gap;
+        if let QueueExpiry::Ttl(ttl) = rcpt.expires {
+            rcpt.expires = QueueExpiry::Ttl(
+                ttl.saturating_sub(HOLD_SECONDS) + at.saturating_sub(message.created),
+            );
+        }
+        modified.insert(idx);
+    }
+    let saved = MessageWrapper::new(message, queue_id, QueueName::default())
+        .save_registry_changes(server, prev_events, modified)
+        .await;
+    held::delete(server.store(), queue_id).await?;
+    let _ = server.inner.ipc.queue_tx.send(QueueEvent::Refresh).await;
+    Ok(saved)
+}
+
+/// Takes a held message out of the queue and tells its sender, with the
+/// reviewer's note if there is one. False when it's no longer queued.
+pub async fn reject(server: &Server, record: &Held, note: Option<&str>) -> trc::Result<bool> {
+    let removed = match server
+        .read_message(record.queue_id, QueueName::default())
+        .await
+    {
+        Some(message) => message.remove(server, None).await,
+        None => false,
+    };
+    held::delete(server.store(), record.queue_id).await?;
+    let mut text = format!(
+        "Your message \"{}\" to {} was held for review under this server's rules, and wasn't sent.\r\n",
+        record.subject,
+        record.recipients.join(", ")
+    );
+    match note {
+        Some(note) => text.push_str(&format!("\r\nThe reviewer's note: {note}\r\n")),
+        None => text.push_str(&format!(
+            "\r\nNobody reviewed it within {KEEP_DAYS} days, so it was returned.\r\n"
+        )),
+    }
+    notify(
+        server,
+        record,
+        &format!("Not sent: {}", record.subject),
+        text,
+    )
+    .await;
+    let _ = server.inner.ipc.queue_tx.send(QueueEvent::Refresh).await;
+    Ok(removed)
+}
+
+/// Tells the sender their message is held (when the rule asks).
+pub async fn notify_held(server: &Server, record: &Held) {
+    let notices = record
+        .rules
+        .iter()
+        .map(|r| r.notice.as_str())
+        .collect::<Vec<_>>()
+        .join(" ");
+    let text = format!(
+        "Your message \"{}\" to {} is held for review under this server's rules: {notices}\r\n\r\n\
+         It will be sent if a reviewer releases it, and returned otherwise within {KEEP_DAYS} days.\r\n",
+        record.subject,
+        record.recipients.join(", "),
+    );
+    notify(
+        server,
+        record,
+        &format!("Held for review: {}", record.subject),
+        text,
+    )
+    .await;
+}
+
+async fn notify(server: &Server, record: &Held, subject: &str, text: String) {
+    let domain = record
+        .sender
+        .rsplit_once('@')
+        .map_or("localhost", |(_, d)| d);
+    let from = format!("postmaster@{domain}");
+    let message = MessageBuilder::new()
+        .from(Address::new_address(Some("Mail review"), from.clone()))
+        .to(Address::new_address(None::<String>, record.sender.clone()))
+        .subject(subject)
+        .header("Auto-Submitted", HeaderType::Text("auto-replied".into()))
+        .text_body(text)
+        .write_to_vec()
+        .unwrap_or_default();
+    server
+        .send_autogenerated(from, [record.sender.as_str()].into_iter(), message, None, 0)
+        .await;
+}
+
+/// Rejects every held message nobody reviewed in time (§2.6), each
+/// recorded as the server's doing. Returns how many.
+pub async fn expire(server: &Server) -> trc::Result<usize> {
+    let at = now();
+    let mut count = 0;
+    for record in held::all(server.store()).await? {
+        if !record.is_expired(at) {
+            continue;
+        }
+        reject(server, &record, None).await?;
+        count += 1;
+        server
+            .audit_note(Record {
+                at: at * 1000,
+                actor: Actor::system("DLP"),
+                via: None,
+                remote_ip: None,
+                action: Action::Destroy,
+                target: Target {
+                    kind: "inbuxa:HeldMessage".into(),
+                    id: Some(record.queue_id.to_string()),
+                    name: Some(record.subject.clone()),
+                    account_id: record.account_id,
+                    tenant_id: record.tenant_id,
+                },
+                changes: vec![],
+                details: Some(format!(
+                    "Rejected: nobody reviewed it within {KEEP_DAYS} days; the sender was told"
+                )),
+                reason: None,
+                outcome: Outcome::success(),
+            })
+            .await;
+    }
+    Ok(count)
+}

crates/smtp/src/queue/mod.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use common::{
@@ -21,6 +23,7 @@ use types::blob_hash::BlobHash;
 use utils::DomainPart;
 
 pub mod dsn;
+pub mod held; // inbuxa: mail held for review
 pub mod manager;
 pub mod quota;
 pub mod spool;