fix(messaging): harden proactive broadcast parity

src/messaging/discord.rs

@@ -400,34 +400,63 @@ impl Messaging for DiscordAdapter {
     }
 
     async fn broadcast(&self, target: &str, response: OutboundResponse) -> crate::Result<()> {
-        let http = self.get_http().await?;
+        crate::messaging::traits::ensure_supported_broadcast_response(
+            "discord",
+            &response,
+            |response| {
+                matches!(
+                    response,
+                    OutboundResponse::Text(_) | OutboundResponse::RichMessage { .. }
+                )
+            },
+        )?;
+
+        let http = self
+            .get_http()
+            .await
+            .map_err(crate::messaging::traits::mark_retryable_broadcast)?;
 
         // Support "dm:{user_id}" targets for opening DM channels
         let channel_id = if let Some(user_id_str) = target.strip_prefix("dm:") {
             let user_id = UserId::new(
                 user_id_str
                     .parse::<u64>()
-                    .context("invalid discord user id for DM broadcast target")?,
+                    .context("invalid discord user id for DM broadcast target")
+                    .map_err(crate::messaging::traits::mark_permanent_broadcast)?,
             );
             user_id
                 .create_dm_channel(&*http)
                 .await
-                .context("failed to open DM channel")?
+                .context("failed to open DM channel")
+                .map_err(crate::messaging::traits::mark_classified_broadcast)?
                 .id
         } else {
             ChannelId::new(
                 target
                     .parse::<u64>()
-                    .context("invalid discord channel id for broadcast target")?,
+                    .context("invalid discord channel id for broadcast target")
+                    .map_err(crate::messaging::traits::mark_permanent_broadcast)?,
             )
         };
 
         if let OutboundResponse::Text(text) = response {
+            let mut sent_any = false;
             for chunk in split_message(&text, 2000) {
-                channel_id
+                match channel_id
                     .say(&*http, &chunk)
                     .await
-                    .context("failed to broadcast discord message")?;
+                    .context("failed to broadcast discord message")
+                {
+                    Ok(_) => sent_any = true,
+                    Err(error) => {
+                        let error = crate::messaging::traits::mark_classified_broadcast(error);
+                        return Err(
+                            crate::messaging::traits::classify_chunked_broadcast_failure(
+                                "discord", error, sent_any,
+                            ),
+                        );
+                    }
+                }
             }
         } else if let OutboundResponse::RichMessage {
             text,
@@ -446,6 +475,7 @@ impl Messaging for DiscordAdapter {
             }
 
             let chunks = split_message(&parts.text, 2000);
+            let mut sent_any = false;
             for (i, chunk) in chunks.iter().enumerate() {
                 let is_last = i == chunks.len() - 1;
                 let mut msg = CreateMessage::new();
@@ -474,10 +504,21 @@ impl Messaging for DiscordAdapter {
                     }
                 }
 
-                channel_id
+                match channel_id
                     .send_message(&*http, msg)
                     .await
-                    .context("failed to broadcast discord rich message")?;
+                    .context("failed to broadcast discord rich message")
+                {
+                    Ok(_) => sent_any = true,
+                    Err(error) => {
+                        let error = crate::messaging::traits::mark_classified_broadcast(error);
+                        return Err(
+                            crate::messaging::traits::classify_chunked_broadcast_failure(
+                                "discord", error, sent_any,
+                            ),
+                        );
+                    }
+                }
             }
         }
 
@@ -1243,6 +1284,7 @@ fn build_poll(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::messaging::traits::{BroadcastFailureKind, broadcast_failure_kind};
     use crate::{Button, ButtonStyle, Card, CardField, InteractiveElements, Poll};
 
     #[test]
@@ -1356,4 +1398,18 @@ mod tests {
         assert_eq!(parts.text, "Status\n\nAll green");
         assert!(!parts.dropped_invalid_poll);
     }
+
+    #[test]
+    fn discord_partial_delivery_failures_become_permanent() {
+        let error = crate::messaging::traits::classify_chunked_broadcast_failure(
+            "discord",
+            crate::messaging::traits::mark_classified_broadcast(anyhow::anyhow!("timeout")),
+            true,
+        );
+
+        assert_eq!(
+            broadcast_failure_kind(&error),
+            BroadcastFailureKind::Permanent
+        );
+    }
 }

src/messaging/email.rs

@@ -517,17 +517,30 @@ impl Messaging for EmailAdapter {
     }
 
     async fn broadcast(&self, target: &str, response: OutboundResponse) -> crate::Result<()> {
+        crate::messaging::traits::ensure_supported_broadcast_response(
+            "email",
+            &response,
+            supports_email_broadcast_response,
+        )?;
+
         let recipient = normalize_email_target(target)
-            .ok_or_else(|| anyhow::anyhow!("invalid email target '{target}'"))?;
+            .ok_or_else(|| anyhow::anyhow!("invalid email target '{target}'"))
+            .map_err(crate::messaging::traits::mark_permanent_broadcast)?;
 
         match response {
             OutboundResponse::Text(text) => {
                 self.send_email(&recipient, "Spacebot message", text, None, Vec::new(), None)
-                    .await?;
+                    .await
+                    .map_err(crate::messaging::traits::mark_classified_broadcast)?;
             }
-            OutboundResponse::RichMessage { text, .. } => {
-                self.send_email(&recipient, "Spacebot message", text, None, Vec::new(), None)
-                    .await?;
+            OutboundResponse::RichMessage {
+                text, cards, poll, ..
+            } => {
+                let body = rich_message_plaintext_fallback(&text, &cards, poll.as_ref())
+                    .expect("supported rich email broadcasts must produce plaintext");
+                self.send_email(&recipient, "Spacebot message", body, None, Vec::new(), None)
+                    .await
+                    .map_err(crate::messaging::traits::mark_classified_broadcast)?;
             }
             OutboundResponse::File {
                 filename,
@@ -544,12 +557,14 @@ impl Messaging for EmailAdapter {
                     Vec::new(),
                     Some((filename, data, mime_type)),
                 )
-                .await?;
+                .await
+                .map_err(crate::messaging::traits::mark_classified_broadcast)?;
             }
             OutboundResponse::ThreadReply { text, .. }
             | OutboundResponse::Ephemeral { text, .. } => {
                 self.send_email(&recipient, "Spacebot message", text, None, Vec::new(), None)
-                    .await?;
+                    .await
+                    .map_err(crate::messaging::traits::mark_classified_broadcast)?;
             }
             OutboundResponse::ScheduledMessage { text, post_at } => {
                 tracing::warn!(
@@ -558,14 +573,10 @@ impl Messaging for EmailAdapter {
                     "email adapter does not support scheduled delivery; sending immediately"
                 );
                 self.send_email(&recipient, "Spacebot message", text, None, Vec::new(), None)
-                    .await?;
+                    .await
+                    .map_err(crate::messaging::traits::mark_classified_broadcast)?;
             }
-            OutboundResponse::Reaction(_)
-            | OutboundResponse::RemoveReaction(_)
-            | OutboundResponse::Status(_)
-            | OutboundResponse::StreamStart
-            | OutboundResponse::StreamChunk(_)
-            | OutboundResponse::StreamEnd => {}
+            _ => unreachable!("unsupported broadcast responses are rejected up front"),
         }
 
         Ok(())
@@ -675,6 +686,56 @@ impl Messaging for EmailAdapter {
     }
 }
 
+fn supports_email_broadcast_response(response: &OutboundResponse) -> bool {
+    match response {
+        OutboundResponse::Text(_)
+        | OutboundResponse::File { .. }
+        | OutboundResponse::ThreadReply { .. }
+        | OutboundResponse::Ephemeral { .. }
+        | OutboundResponse::ScheduledMessage { .. } => true,
+        OutboundResponse::RichMessage {
+            text, cards, poll, ..
+        } => rich_message_plaintext_fallback(text, cards, poll.as_ref()).is_some(),
+        _ => false,
+    }
+}
+
+fn rich_message_plaintext_fallback(
+    text: &str,
+    cards: &[crate::Card],
+    poll: Option<&crate::Poll>,
+) -> Option<String> {
+    let mut sections = Vec::new();
+    let text = text.trim();
+    if !text.is_empty() {
+        sections.push(text.to_string());
+    } else {
+        let card_text = OutboundResponse::text_from_cards(cards);
+        if !card_text.trim().is_empty() {
+            sections.push(card_text);
+        }
+    }
+
+    if let Some(poll) = poll {
+        let question = poll.question.trim();
+        if !question.is_empty() {
+            let answers = poll
+                .answers
+                .iter()
+                .map(|answer| answer.trim())
+                .filter(|answer| !answer.is_empty())
+                .map(|answer| format!("- {answer}"))
+                .collect::<Vec<_>>();
+            let mut poll_lines = vec![question.to_string()];
+            poll_lines.extend(answers);
+            sections.push(poll_lines.join("\n"));
+        }
+    }
+
+    let combined = sections.join("\n\n");
+    (!combined.trim().is_empty()).then_some(combined)
+}
+
 fn build_smtp_transport(config: &EmailConfig) -> crate::Result<AsyncSmtpTransport<Tokio1Executor>> {
     let builder = if config.smtp_use_starttls {
         AsyncSmtpTransport::<Tokio1Executor>::starttls_relay(&config.smtp_host)
@@ -1758,7 +1819,10 @@ mod tests {
         EmailSearchHit, EmailSearchQuery, build_imap_search_criterion, derive_thread_key,
         extract_message_ids, is_local_mail_host, normalize_email_target, normalize_reply_subject,
         normalize_search_folders, parse_primary_mailbox, sort_and_limit_search_hits,
+        supports_email_broadcast_response,
     };
+    use crate::OutboundResponse;
+    use crate::messaging::traits::{BroadcastFailureKind, broadcast_failure_kind};
 
     #[test]
     fn parse_primary_mailbox_parses_display_name() {
@@ -1917,4 +1981,71 @@ mod tests {
         assert_eq!(results[0].subject, "newest");
         assert_eq!(results[1].subject, "middle");
     }
+
+    #[test]
+    fn email_supported_broadcast_variants_include_degraded_message_forms() {
+        assert!(supports_email_broadcast_response(&OutboundResponse::Text(
+            "hello".to_string()
+        )));
+        assert!(supports_email_broadcast_response(
+            &OutboundResponse::RichMessage {
+                text: String::new(),
+                blocks: Vec::new(),
+                cards: vec![crate::Card {
+                    title: Some("Digest".to_string()),
+                    description: Some("One item needs attention".to_string()),
+                    color: None,
+                    url: None,
+                    fields: Vec::new(),
+                    footer: None,
+                    thumbnail: None,
+                    image: None,
+                    author: None,
+                    timestamp: None,
+                }],
+                interactive_elements: Vec::new(),
+                poll: None,
+            }
+        ));
+        assert!(supports_email_broadcast_response(
+            &OutboundResponse::ScheduledMessage {
+                text: "hello".to_string(),
+                post_at: 123,
+            }
+        ));
+        assert!(supports_email_broadcast_response(&OutboundResponse::File {
+            filename: "note.txt".to_string(),
+            data: vec![1, 2, 3],
+            mime_type: "text/plain".to_string(),
+            caption: None,
+        }));
+    }
+
+    #[test]
+    fn email_unsupported_broadcast_variants_are_permanent_failures() {
+        let error = crate::messaging::traits::ensure_supported_broadcast_response(
+            "email",
+            &OutboundResponse::Reaction("thumbsup".to_string()),
+            supports_email_broadcast_response,
+        )
+        .expect_err("unsupported variants should error");
+
+        assert_eq!(
+            broadcast_failure_kind(&error),
+            BroadcastFailureKind::Permanent
+        );
+    }
+
+    #[test]
+    fn email_rejects_rich_message_without_plaintext_fallback() {
+        assert!(!supports_email_broadcast_response(
+            &OutboundResponse::RichMessage {
+                text: String::new(),
+                blocks: Vec::new(),
+                cards: Vec::new(),
+                interactive_elements: Vec::new(),
+                poll: None,
+            }
+        ));
+    }
 }