From 2c8fb1949aaeca4699c6f476becb30436e57fa80 Mon Sep 17 00:00:00 2001
From: Loakesh Indiran <rustyloaki@gmail.com>
Date: Mon, 2 Nov 2026 23:34:39 +0530
Subject: [PATCH] feat(alerts): shared notification send path, stage 1a core

Add NotificationKind and the platform rule table, the kind, event,
object id and cluster name on the notification context, template
precedence with ordered level 1 candidates (D2 reserved rule and
transport rule), the default_templates org key, compiled anomaly and
incident cards, the replay record, and resolve, send and notify.
Preview and test send accept a kind. Example fixtures under
render/testdata/examples have golden tests. Nothing calls the path yet.

Design assumptions checked against main: 1, 7, 8 and 9 hold. 2 holds
(the table is alert_incidents). 3 holds in part (stage 3). 4 fails for
HTTP, which has no timeout, so the shared path sets 120 s. 5 holds in
effect. 6 fails: alert_composites.template also names a template.
---
 .../src/request/alerts/destinations.rs        |   5 +-
 src/config/src/meta/alerts/mod.rs             |   1 +
 .../src/meta/alerts/notification_kind.rs      | 184 +++++
 src/core/src/alerts/alert.rs                  | 128 ++-
 src/core/src/alerts/notifications/context.rs  |  30 +-
 src/core/src/alerts/notifications/custom.rs   |   4 +
 .../alerts/notifications/default_template.rs  | 134 ++-
 src/core/src/alerts/notifications/examples.rs | 353 ++++++++
 src/core/src/alerts/notifications/format.rs   |  16 +
 src/core/src/alerts/notifications/mod.rs      |   5 +-
 src/core/src/alerts/notifications/notify.rs   | 763 ++++++++++++++++++
 .../src/alerts/notifications/org_default.rs   | 611 ++++++++++++--
 src/core/src/alerts/notifications/platform.rs | 180 ++++-
 src/core/src/alerts/notifications/preview.rs  |  29 +-
 src/core/src/alerts/notifications/record.rs   | 364 +++++++++
 .../src/alerts/notifications/render/mod.rs    |   1 +
 .../render/testdata/examples/anomaly.json     | 191 +++++
 .../render/testdata/examples/incident.json    | 216 +++++
 .../alerts/notifications/render/webhook.rs    |   8 +
 src/core/src/alerts/notifications/resolve.rs  |  57 +-
 .../src/alerts/notifications/test_send.rs     |  19 +-
 src/core/src/organization.rs                  |   9 +
 src/db/src/system_settings.rs                 |   3 +-
 23 files changed, 3153 insertions(+), 158 deletions(-)
 create mode 100644 src/config/src/meta/alerts/notification_kind.rs
 create mode 100644 src/core/src/alerts/notifications/examples.rs
 create mode 100644 src/core/src/alerts/notifications/notify.rs
 create mode 100644 src/core/src/alerts/notifications/record.rs
 create mode 100644 src/core/src/alerts/notifications/render/testdata/examples/anomaly.json
 create mode 100644 src/core/src/alerts/notifications/render/testdata/examples/incident.json

diff --git a/src/api/management/src/request/alerts/destinations.rs b/src/api/management/src/request/alerts/destinations.rs
index 4771a2b98c..80e5593193 100644
--- a/src/api/management/src/request/alerts/destinations.rs
+++ b/src/api/management/src/request/alerts/destinations.rs
@@ -764,6 +764,9 @@ pub struct TestSendRequest {
     /// The name of an already-saved template to test-send. Exactly one of
     /// `definition` / `template_name` must be set.
     pub template_name: Option<String>,
+    /// Which producer's sample to render with; absent means alert.
+    #[serde(default)]
+    pub kind: config::meta::alerts::notification_kind::NotificationKind,
 }
 
 #[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
@@ -885,7 +888,7 @@ pub async fn test_send(
         return MetaHttpResponse::bad_request("Test-send is only supported for alert destinations");
     };
 
-    match core_test_send(&org_id, destination_type, &spec).await {
+    match core_test_send(&org_id, destination_type, &spec, req.kind).await {
         Ok(message) => MetaHttpResponse::json(TestSendResponse {
             success: true,
             message,
diff --git a/src/config/src/meta/alerts/mod.rs b/src/config/src/meta/alerts/mod.rs
index d03ec6682d..30fb3a8811 100644
--- a/src/config/src/meta/alerts/mod.rs
+++ b/src/config/src/meta/alerts/mod.rs
@@ -40,6 +40,7 @@ pub mod dispatch;
 pub mod grouping;
 pub mod incidents;
 pub mod level;
+pub mod notification_kind;
 pub mod priority;
 pub mod prom_rule;
 pub mod rca_runs;
diff --git a/src/config/src/meta/alerts/notification_kind.rs b/src/config/src/meta/alerts/notification_kind.rs
new file mode 100644
index 0000000000..77d44091c6
--- /dev/null
+++ b/src/config/src/meta/alerts/notification_kind.rs
@@ -0,0 +1,184 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Which producer a notification comes from, and what each of its events does on a platform.
+
+use serde::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+pub const EVENT_RESOLVED: &str = "resolved";
+pub const EVENT_ACKNOWLEDGED: &str = "acknowledged";
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum NotificationKind {
+    #[default]
+    Alert,
+    Incident,
+    Anomaly,
+    Synthetics,
+    Downtime,
+    OnCall,
+}
+
+impl NotificationKind {
+    pub const ALL: [NotificationKind; 6] = [
+        NotificationKind::Alert,
+        NotificationKind::Incident,
+        NotificationKind::Anomaly,
+        NotificationKind::Synthetics,
+        NotificationKind::Downtime,
+        NotificationKind::OnCall,
+    ];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            NotificationKind::Alert => "alert",
+            NotificationKind::Incident => "incident",
+            NotificationKind::Anomaly => "anomaly",
+            NotificationKind::Synthetics => "synthetics",
+            NotificationKind::Downtime => "downtime",
+            NotificationKind::OnCall => "on_call",
+        }
+    }
+
+    pub fn parse(s: &str) -> Option<NotificationKind> {
+        Self::ALL.into_iter().find(|k| k.as_str() == s)
+    }
+
+    /// The reserved name of the kind's compiled default; alerts keep `o2_default_content`.
+    pub fn compiled_default_name(self) -> &'static str {
+        match self {
+            NotificationKind::Alert => "o2_default_content",
+            NotificationKind::Incident => "o2_default_incident",
+            NotificationKind::Anomaly => "o2_default_anomaly",
+            NotificationKind::Synthetics => "o2_default_synthetics",
+            NotificationKind::Downtime => "o2_default_downtime",
+            NotificationKind::OnCall => "o2_default_on_call",
+        }
+    }
+
+    /// What `event` does to a PagerDuty, Opsgenie or ServiceNow record of this kind.
+    pub fn platform_event(self, event: &str) -> PlatformEvent {
+        let head = event.split(':').next().unwrap_or_default();
+        match (self, head) {
+            (NotificationKind::Alert, EVENT_RESOLVED) => PlatformEvent::Resolve,
+            (NotificationKind::Alert, _) => PlatformEvent::Trigger,
+            (NotificationKind::Downtime, _) => PlatformEvent::NotForKind,
+            (_, EVENT_RESOLVED) => PlatformEvent::Resolve,
+            (NotificationKind::OnCall, EVENT_ACKNOWLEDGED) => PlatformEvent::NoAck,
+            (NotificationKind::Anomaly, "firing")
+            | (NotificationKind::Incident, "opened" | "updated")
+            | (NotificationKind::Synthetics, "firing" | "down")
+            | (NotificationKind::OnCall, "paged" | "handoff") => PlatformEvent::Trigger,
+            _ => PlatformEvent::NotForKind,
+        }
+    }
+
+    /// True for the events that open or extend a platform record, the only ones closure refuses.
+    pub fn opens_episode(self, event: &str) -> bool {
+        self != NotificationKind::Alert && self.platform_event(event) == PlatformEvent::Trigger
+    }
+
+    /// True for events whose episode has no lifecycle, so it closes at insert.
+    pub fn closed_at_insert(self, event: &str) -> bool {
+        let head = event.split(':').next().unwrap_or_default();
+        match self {
+            NotificationKind::Downtime => true,
+            NotificationKind::Synthetics => {
+                matches!(head, "degraded" | "flaky" | "cert_expiring")
+            }
+            NotificationKind::OnCall => head == "test",
+            _ => false,
+        }
+    }
+}
+
+impl std::fmt::Display for NotificationKind {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PlatformEvent {
+    Trigger,
+    Resolve,
+    NoAck,
+    NotForKind,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn wire_spelling_round_trips() {
+        for kind in NotificationKind::ALL {
+            assert_eq!(NotificationKind::parse(kind.as_str()), Some(kind));
+            let json = serde_json::to_string(&kind).unwrap();
+            assert_eq!(json, format!("\"{}\"", kind.as_str()));
+        }
+        assert_eq!(NotificationKind::OnCall.as_str(), "on_call");
+        assert_eq!(NotificationKind::parse("pages"), None);
+    }
+
+    #[test]
+    fn platform_rule_table() {
+        use NotificationKind::*;
+        use PlatformEvent::*;
+        let cases = [
+            (Anomaly, "firing:1700", Trigger),
+            (Anomaly, "resolved", Resolve),
+            (Incident, "opened", Trigger),
+            (Incident, "updated:severity_changed:1", Trigger),
+            (Incident, "resolved", Resolve),
+            (Synthetics, "firing:r1", Trigger),
+            (Synthetics, "down", Trigger),
+            (Synthetics, "degraded:r1", NotForKind),
+            (Synthetics, "flaky:r1", NotForKind),
+            (Synthetics, "cert_expiring:r1", NotForKind),
+            (Synthetics, "resolved", Resolve),
+            (Downtime, "started", NotForKind),
+            (Downtime, "resolved", NotForKind),
+            (OnCall, "paged:1", Trigger),
+            (OnCall, "handoff:2:9", Trigger),
+            (OnCall, "acknowledged", NoAck),
+            (OnCall, "coverage_gap:1", NotForKind),
+            (OnCall, "verdict:5", NotForKind),
+            (OnCall, "test", NotForKind),
+            (OnCall, "resolved", Resolve),
+        ];
+        for (kind, event, want) in cases {
+            assert_eq!(kind.platform_event(event), want, "{kind} {event}");
+        }
+    }
+
+    #[test]
+    fn opening_and_closed_at_insert_events() {
+        use NotificationKind::*;
+        assert!(Anomaly.opens_episode("firing:1"));
+        assert!(!Anomaly.opens_episode("resolved"));
+        assert!(Incident.opens_episode("updated:x:1"));
+        assert!(!OnCall.opens_episode("acknowledged"));
+        assert!(!Downtime.opens_episode("started"));
+        assert!(!Alert.opens_episode("firing"));
+        assert!(Downtime.closed_at_insert("started"));
+        assert!(Synthetics.closed_at_insert("degraded:r1"));
+        assert!(!Synthetics.closed_at_insert("firing:r1"));
+        assert!(OnCall.closed_at_insert("test"));
+        assert!(!Anomaly.closed_at_insert("firing:1"));
+    }
+}
diff --git a/src/core/src/alerts/alert.rs b/src/core/src/alerts/alert.rs
index 5e74990e1e..28aa83522c 100644
--- a/src/core/src/alerts/alert.rs
+++ b/src/core/src/alerts/alert.rs
@@ -3255,7 +3255,7 @@ async fn build_chart_asset(
 /// A body we cannot parse is left alone and logged: saving the alert already
 /// refused this combination, so reaching here means the template changed after
 /// the fact, and dropping the firing would be the worse failure.
-fn stamp_platform_key(
+pub(crate) fn stamp_platform_key(
     dest_type: &DestinationType,
     ctx: &NotificationContext,
     msg: String,
@@ -3571,18 +3571,52 @@ fn members_to_send(kept: Vec<String>, skipped: &[String]) -> Result<Email, anyho
     Ok(Email { recipients: kept })
 }
 
-async fn send_http_notification(endpoint: &Endpoint, msg: String) -> Result<String, anyhow::Error> {
+pub(crate) async fn send_http_notification(
+    endpoint: &Endpoint,
+    msg: String,
+) -> Result<String, anyhow::Error> {
+    match send_http_attempt(endpoint, msg, None).await? {
+        HttpAttempt::Answered {
+            note: Some(note), ..
+        } => Ok(note),
+        HttpAttempt::Answered {
+            status,
+            body,
+            hide_body,
+            note: None,
+        } => send_outcome(status, &body, hide_body),
+        HttpAttempt::Blocked(e) | HttpAttempt::NoAnswer(e) => Err(anyhow::anyhow!(e)),
+    }
+}
+
+/// One HTTP send as the shared path classifies it; `note` marks a Slack resend without images.
+pub(crate) enum HttpAttempt {
+    Answered {
+        status: reqwest::StatusCode,
+        body: String,
+        hide_body: bool,
+        note: Option<String>,
+    },
+    Blocked(String),
+    NoAnswer(String),
+}
+
+pub(crate) async fn send_http_attempt(
+    endpoint: &Endpoint,
+    msg: String,
+    timeout: Option<std::time::Duration>,
+) -> Result<HttpAttempt, anyhow::Error> {
     // A plain body posted at a Jira site root does nothing, so senders without a Jira branch fail.
     if jira::is_jira(endpoint) {
-        return Err(anyhow::anyhow!(jira::UNSUPPORTED_SENDER));
+        return Ok(HttpAttempt::Blocked(jira::UNSUPPORTED_SENDER.to_string()));
     }
     // Block SSRF: validate the destination URL (including DNS resolution) before
     // making any outbound request. The client is built through `build_safe_destination_client`
     // so that redirect targets and per-connect DNS resolution are re-validated.
     if let Err(e) = SsrfGuard::validate_destination_url_with_config_async(&endpoint.url).await {
-        return Err(anyhow::anyhow!(
+        return Ok(HttpAttempt::Blocked(format!(
             "Destination URL blocked by SSRF guard: {e}"
-        ));
+        )));
     }
 
     let builder = if endpoint.skip_tls_verify {
@@ -3590,6 +3624,10 @@ async fn send_http_notification(endpoint: &Endpoint, msg: String) -> Result<Stri
     } else {
         reqwest::Client::builder()
     };
+    let builder = match timeout {
+        Some(t) => builder.timeout(t),
+        None => builder,
+    };
     let client = common::utils::ssrf_guard::build_safe_destination_client(builder)?;
     let url = url::Url::parse(&endpoint.url)?;
     let build_req = |body: String| {
@@ -3640,15 +3678,18 @@ async fn send_http_notification(endpoint: &Endpoint, msg: String) -> Result<Stri
         Ok(v) => v,
         Err(e) => {
             log::error!("error sending request to {} with error {e:?}", endpoint.url);
-            return Err(anyhow::anyhow!(
+            return Ok(HttpAttempt::NoAnswer(format!(
                 "error sending request : {:?}",
                 e.without_url()
-            ));
+            )));
         }
     };
     let resp_status = resp.status();
     let hide_body = config::utils::ssrf_guard::admitted_only_by_allowlist(&resp);
-    let resp_body = resp.text().await.map_err(reqwest::Error::without_url)?;
+    let resp_body = match resp.text().await {
+        Ok(body) => body,
+        Err(e) => return Ok(HttpAttempt::NoAnswer(e.without_url().to_string())),
+    };
 
     log::debug!(
         "Alert sent to destination {} with status: {}, body:\n{}",
@@ -3693,17 +3734,27 @@ async fn send_http_notification(endpoint: &Endpoint, msg: String) -> Result<Stri
                      — check the image URLs in the destination's template."
                 );
             }
-            return Ok(format!(
-                "sent status: {} (image stripped: Slack rejected it)",
-                reqwest::StatusCode::OK,
-            ));
+            return Ok(HttpAttempt::Answered {
+                status: reqwest::StatusCode::OK,
+                body: String::new(),
+                hide_body,
+                note: Some(format!(
+                    "sent status: {} (image stripped: Slack rejected it)",
+                    reqwest::StatusCode::OK,
+                )),
+            });
         }
         log::error!(
             "Alert http notification failed with status: {resp_status}, body: {resp_body}, payload: {msg}"
         );
     }
 
-    send_outcome(resp_status, &resp_body, hide_body)
+    Ok(HttpAttempt::Answered {
+        status: resp_status,
+        body: resp_body,
+        hide_body,
+        note: None,
+    })
 }
 
 /// Send an email; with `text_body: None` it carries only the HTML part.
@@ -3803,16 +3854,38 @@ async fn send_discord_with_attachment(
     msg: String,
     png: std::sync::Arc<Vec<u8>>,
 ) -> Result<String, anyhow::Error> {
+    match send_discord_attempt(endpoint, msg, png, None).await? {
+        HttpAttempt::Answered {
+            status,
+            body,
+            hide_body,
+            ..
+        } => send_outcome(status, &body, hide_body),
+        HttpAttempt::Blocked(e) => Err(anyhow::anyhow!(e)),
+        HttpAttempt::NoAnswer(e) => Err(anyhow::anyhow!(e)),
+    }
+}
+
+pub(crate) async fn send_discord_attempt(
+    endpoint: &Endpoint,
+    msg: String,
+    png: std::sync::Arc<Vec<u8>>,
+    timeout: Option<std::time::Duration>,
+) -> Result<HttpAttempt, anyhow::Error> {
     if let Err(e) = SsrfGuard::validate_destination_url_with_config_async(&endpoint.url).await {
-        return Err(anyhow::anyhow!(
+        return Ok(HttpAttempt::Blocked(format!(
             "Destination URL blocked by SSRF guard: {e}"
-        ));
+        )));
     }
     let builder = if endpoint.skip_tls_verify {
         reqwest::Client::builder().danger_accept_invalid_certs(true)
     } else {
         reqwest::Client::builder()
     };
+    let builder = match timeout {
+        Some(t) => builder.timeout(t),
+        None => builder,
+    };
     let client = common::utils::ssrf_guard::build_safe_destination_client(builder)?;
     let url = url::Url::parse(&endpoint.url)?;
 
@@ -3825,21 +3898,27 @@ async fn send_discord_with_attachment(
                 .mime_str("image/png")?,
         );
 
-    let resp = client
-        .post(url)
-        .multipart(form)
-        .send()
-        .await
-        .map_err(reqwest::Error::without_url)?;
+    let resp = match client.post(url).multipart(form).send().await {
+        Ok(resp) => resp,
+        Err(e) => return Ok(HttpAttempt::NoAnswer(e.without_url().to_string())),
+    };
     let resp_status = resp.status();
     let hide_body = config::utils::ssrf_guard::admitted_only_by_allowlist(&resp);
-    let resp_body = resp.text().await.map_err(reqwest::Error::without_url)?;
+    let resp_body = match resp.text().await {
+        Ok(body) => body,
+        Err(e) => return Ok(HttpAttempt::NoAnswer(e.without_url().to_string())),
+    };
     if !resp_status.is_success() {
         log::error!(
             "Alert discord notification failed with status: {resp_status}, body: {resp_body}, payload: {msg}"
         );
     }
-    send_outcome(resp_status, &resp_body, hide_body)
+    Ok(HttpAttempt::Answered {
+        status: resp_status,
+        body: resp_body,
+        hide_body,
+        note: None,
+    })
 }
 
 /// What the caller of a send sees; a peer only the operator allowlist admits shows its status only.
@@ -3856,7 +3935,7 @@ fn send_outcome(
     }
 }
 
-async fn send_sns_notification(
+pub(crate) async fn send_sns_notification(
     alert_name: &str,
     aws_sns: &AwsSns,
     msg: String,
@@ -4510,6 +4589,7 @@ async fn build_notification_context(
         level,
         chart_url: None,
         chart_png: None,
+        ..Default::default()
     }
 }
 pub fn get_row_column_map(rows: &[Map<String, Value>]) -> HashMap<String, HashSet<String>> {
diff --git a/src/core/src/alerts/notifications/context.rs b/src/core/src/alerts/notifications/context.rs
index 6b4135d1fd..e97f7d38de 100644
--- a/src/core/src/alerts/notifications/context.rs
+++ b/src/core/src/alerts/notifications/context.rs
@@ -23,7 +23,7 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
 
 use config::{
-    meta::alerts::level::AlertLevel,
+    meta::alerts::{level::AlertLevel, notification_kind::NotificationKind},
     utils::json::{Map, Value},
 };
 
@@ -99,6 +99,34 @@ pub struct NotificationContext {
     pub metadata: Vec<(String, String)>,
     pub group_labels: Option<BTreeMap<String, String>>,
     pub level: Option<AlertLevel>,
+    pub kind: NotificationKind,
+    /// The owning object a delivery row is listed under, such as the anomaly or incident id.
+    pub object_id: String,
+    /// The ledger event, e.g. `firing:{window_end_us}` or `resolved`; empty on the alert path.
+    pub event: String,
+    pub cluster_name: String,
+    /// Today's body for a generic webhook while no user template applies (D1).
+    pub legacy_payload: Option<LegacyPayload>,
+}
+
+/// Builds a producer's pre-template JSON body only when the legacy rule selects it.
+#[derive(Clone)]
+pub struct LegacyPayload(pub std::sync::Arc<dyn Fn() -> Value + Send + Sync>);
+
+impl LegacyPayload {
+    pub fn new(f: impl Fn() -> Value + Send + Sync + 'static) -> Self {
+        Self(std::sync::Arc::new(f))
+    }
+
+    pub fn build(&self) -> Value {
+        (self.0)()
+    }
+}
+
+impl std::fmt::Debug for LegacyPayload {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("LegacyPayload")
+    }
 }
 
 /// Collect row columns in deterministic order.
diff --git a/src/core/src/alerts/notifications/custom.rs b/src/core/src/alerts/notifications/custom.rs
index 81a728d6c1..0ea38a70a9 100644
--- a/src/core/src/alerts/notifications/custom.rs
+++ b/src/core/src/alerts/notifications/custom.rs
@@ -88,6 +88,9 @@ pub(crate) fn apply_custom_template_plain(
             ctx.episode_id.as_deref().unwrap_or_default(),
         )
         .replace("{recovery_reason}", &ctx.recovery_reason)
+        .replace("{notification_kind}", ctx.kind.as_str())
+        .replace("{notification_event}", &ctx.event)
+        .replace("{cluster_name}", &ctx.cluster_name)
         // Feature 2 (PT-4 / PT-9). Scope is DESTINATION TEMPLATES ONLY (D25):
         // incident notifications build custom JSON and workflows carry
         // hard-coded metadata; neither is wired here in v1. Unset priority and
@@ -564,6 +567,7 @@ mod golden {
             chart_png: None,
             alert_id: String::new(),
             incident_id: String::new(),
+            ..Default::default()
         }
     }
 
diff --git a/src/core/src/alerts/notifications/default_template.rs b/src/core/src/alerts/notifications/default_template.rs
index 76d22afe94..ef51a89334 100644
--- a/src/core/src/alerts/notifications/default_template.rs
+++ b/src/core/src/alerts/notifications/default_template.rs
@@ -26,7 +26,10 @@
 use config::{
     DEFAULT_ORG,
     meta::{
-        alerts::content_spec::{ChartSpec, ContentField, ContentSpec, RowsSpec},
+        alerts::{
+            content_spec::{ChartSpec, ContentField, ContentSpec, RowsSpec},
+            notification_kind::NotificationKind,
+        },
         destinations::{Template, TemplateKind, TemplateType},
     },
 };
@@ -73,6 +76,76 @@ pub fn compiled_default_content() -> ContentSpec {
     }
 }
 
+/// The compiled default for `kind`; kinds whose stage has not shipped use the alert card.
+pub fn compiled_content_for(kind: NotificationKind) -> ContentSpec {
+    match kind {
+        NotificationKind::Incident => compiled_incident_content(),
+        NotificationKind::Anomaly => compiled_anomaly_content(),
+        _ => compiled_default_content(),
+    }
+}
+
+/// The reserved name that `compiled_content_for(kind)` is seeded under.
+pub fn compiled_name_for(kind: NotificationKind) -> &'static str {
+    match kind {
+        NotificationKind::Incident | NotificationKind::Anomaly => kind.compiled_default_name(),
+        _ => DEFAULT_CONTENT_TEMPLATE_NAME,
+    }
+}
+
+pub fn compiled_anomaly_content() -> ContentSpec {
+    ContentSpec {
+        title: "Anomaly {alert_status}: {anomaly_name} on {stream_name}".into(),
+        title_overrides: Default::default(),
+        body: "{alert_description}".into(),
+        fields: vec![
+            field("Worst value", "{worst_actual_value}"),
+            field("Expected", "{worst_expected}"),
+            field("Window", "{window_start} to {window_end}"),
+            field("Score", "{score} vs {threshold}"),
+        ],
+        rows: RowsSpec {
+            enabled: false,
+            max: 5,
+            columns: None,
+            format: None,
+        },
+        links: vec![],
+        chart: ChartSpec { enabled: false },
+    }
+}
+
+pub fn compiled_incident_content() -> ContentSpec {
+    ContentSpec {
+        title: "Incident {incident_severity}: {incident_title}".into(),
+        title_overrides: Default::default(),
+        body: "Incident **{incident_event}** for service **{incident_service}**.\n\n{alert_description}"
+            .into(),
+        fields: vec![
+            field("Event", "{incident_event}"),
+            field("Severity", "{incident_severity}"),
+            field("Service", "{incident_service}"),
+            field("Alert", "{alert_name}"),
+        ],
+        rows: RowsSpec {
+            enabled: true,
+            max: 5,
+            columns: None,
+            format: None,
+        },
+        links: vec![],
+        chart: ChartSpec { enabled: false },
+    }
+}
+
+fn field(label: &str, value: &str) -> ContentField {
+    ContentField {
+        label: label.into(),
+        value: value.into(),
+        show_when: None,
+    }
+}
+
 /// True for template names the system reserves and manages: the
 /// `prebuilt_*` family, plus `o2_default_content`. Users can never
 /// create/update/delete these (root-only, same enforcement path as
@@ -80,31 +153,50 @@ pub fn compiled_default_content() -> ContentSpec {
 pub fn is_reserved_template_name(name: &str) -> bool {
     config::prebuilt_loader::is_prebuilt_template_name(name)
         || name == DEFAULT_CONTENT_TEMPLATE_NAME
+        || name == NotificationKind::Incident.compiled_default_name()
+        || name == NotificationKind::Anomaly.compiled_default_name()
         || crate::alerts::templates::is_downtime_template_name(name)
 }
 
-/// Create-if-missing (and overwrite under the same `should_apply_prebuilt`
-/// rule as the prebuilt loop) the seeded DB copy of the compiled default in
-/// DEFAULT_ORG. Called from `ensure_system_templates`, inside the same
-/// dist_lock + revision gate, after the prebuilt loop.
+/// Seed the read only DEFAULT_ORG copies of the compiled defaults under `should_apply_prebuilt`.
 pub(crate) async fn ensure_default_content_template(
     shipped_rev: u32,
     applied_rev: u32,
 ) -> Result<(), anyhow::Error> {
-    let spec = compiled_default_content();
-    let body = serde_json::to_string(&spec)?;
+    for kind in [
+        NotificationKind::Alert,
+        NotificationKind::Incident,
+        NotificationKind::Anomaly,
+    ] {
+        ensure_compiled_row(
+            compiled_name_for(kind),
+            &compiled_content_for(kind),
+            shipped_rev,
+            applied_rev,
+        )
+        .await?;
+    }
+    Ok(())
+}
 
+async fn ensure_compiled_row(
+    name: &str,
+    spec: &ContentSpec,
+    shipped_rev: u32,
+    applied_rev: u32,
+) -> Result<(), anyhow::Error> {
+    let body = serde_json::to_string(spec)?;
     let mut template = Template {
         id: None,
         org_id: DEFAULT_ORG.to_string(),
-        name: DEFAULT_CONTENT_TEMPLATE_NAME.to_string(),
+        name: name.to_string(),
         is_default: false,
         template_type: TemplateType::Http,
         body: body.clone(),
         kind: TemplateKind::Content,
     };
 
-    match db::alerts::templates::get(DEFAULT_ORG, DEFAULT_CONTENT_TEMPLATE_NAME).await {
+    match db::alerts::templates::get(DEFAULT_ORG, name).await {
         Ok(existing) => {
             let drifted = existing.body != body || existing.template_type != TemplateType::Http;
             if super::super::templates::should_apply_prebuilt(shipped_rev, applied_rev, drifted)
@@ -112,20 +204,16 @@ pub(crate) async fn ensure_default_content_template(
             {
                 template.id = existing.id;
                 db::alerts::templates::set(template).await?;
-                log::info!(
-                    "[TEMPLATES] Updated system template '{DEFAULT_CONTENT_TEMPLATE_NAME}' in {DEFAULT_ORG}"
-                );
+                log::info!("[TEMPLATES] Updated system template '{name}' in {DEFAULT_ORG}");
             }
         }
         Err(TemplateError::NotFound) => {
             db::alerts::templates::set(template).await?;
-            log::info!(
-                "[TEMPLATES] Created system template '{DEFAULT_CONTENT_TEMPLATE_NAME}' in {DEFAULT_ORG}"
-            );
+            log::info!("[TEMPLATES] Created system template '{name}' in {DEFAULT_ORG}");
         }
         Err(e) => {
             log::error!(
-                "[TEMPLATES] Error checking system template '{DEFAULT_CONTENT_TEMPLATE_NAME}' in {DEFAULT_ORG}: {e}"
+                "[TEMPLATES] Error checking system template '{name}' in {DEFAULT_ORG}: {e}"
             );
         }
     }
@@ -144,8 +232,14 @@ mod tests {
 
     #[test]
     fn compiled_default_parses_and_renders_everywhere() {
-        let spec = compiled_default_content();
-        assert!(!spec.title.is_empty());
+        for kind in NotificationKind::ALL {
+            let spec = compiled_content_for(kind);
+            assert!(!spec.title.is_empty());
+            renders_everywhere(&spec);
+        }
+    }
+
+    fn renders_everywhere(spec: &ContentSpec) {
         let ctx = synthetic_context(None);
         // Every channel format must render the compiled-in default without
         // error — this is the terminal fallback (see module doc comment),
@@ -162,7 +256,7 @@ mod tests {
             ChannelFormat::ServiceNow,
             ChannelFormat::Sns,
         ] {
-            let content = resolve_content(&spec, &ctx, f.channel_family());
+            let content = resolve_content(spec, &ctx, f.channel_family());
             assert!(render(f, &content, &ctx).is_ok(), "{f:?}");
         }
     }
@@ -189,6 +283,8 @@ mod tests {
     #[test]
     fn o2_default_content_is_reserved() {
         assert!(is_reserved_template_name("o2_default_content"));
+        assert!(is_reserved_template_name("o2_default_anomaly"));
+        assert!(is_reserved_template_name("o2_default_incident"));
         assert!(is_reserved_template_name("prebuilt_slack"));
         assert!(!is_reserved_template_name("my_template"));
     }
diff --git a/src/core/src/alerts/notifications/examples.rs b/src/core/src/alerts/notifications/examples.rs
new file mode 100644
index 0000000000..b8faea88b2
--- /dev/null
+++ b/src/core/src/alerts/notifications/examples.rs
@@ -0,0 +1,353 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Per kind sample contexts and variable lists, shared by preview and the example goldens.
+
+use config::meta::alerts::{level::AlertLevel, notification_kind::NotificationKind};
+
+use super::{
+    STATUS_FIRING, STATUS_RESOLVED,
+    context::{NotificationContext, build_row_columns},
+};
+
+pub const ANOMALY_VARIABLES: [&str; 13] = [
+    "anomaly_id",
+    "anomaly_name",
+    "anomaly_kind",
+    "anomaly_count",
+    "max_deviation_percent",
+    "deviation_kind",
+    "worst_actual_value",
+    "worst_expected",
+    "score",
+    "threshold",
+    "window_start",
+    "window_end",
+    "anomaly_url",
+];
+
+pub const INCIDENT_VARIABLES: [&str; 6] = [
+    "incident_id",
+    "incident_title",
+    "incident_event",
+    "incident_severity",
+    "incident_service",
+    "incident_url",
+];
+
+const COMMON_VARIABLES: [&str; 27] = [
+    "org_name",
+    "stream_type",
+    "stream_name",
+    "alert_name",
+    "alert_type",
+    "alert_period",
+    "alert_operator",
+    "alert_threshold",
+    "alert_count",
+    "alert_agg_value",
+    "alert_level",
+    "alert_status",
+    "episode_id",
+    "recovery_reason",
+    "alert_priority",
+    "alert_tags",
+    "alert_threshold_crit",
+    "alert_threshold_warn",
+    "alert_warning_threshold",
+    "alert_start_time",
+    "alert_end_time",
+    "alert_url",
+    "alert_trigger_time_str",
+    "alert_time",
+    "alert_description",
+    "rows",
+    "group.<label>",
+];
+
+/// The variables a template of `kind` can use: common, shared send path, then the kind's own.
+pub fn variables_for(kind: NotificationKind) -> Vec<String> {
+    let kind_vars: &[&str] = match kind {
+        NotificationKind::Anomaly => &ANOMALY_VARIABLES,
+        NotificationKind::Incident => &INCIDENT_VARIABLES,
+        _ => &[],
+    };
+    COMMON_VARIABLES
+        .iter()
+        .chain(super::resolve::EXTRA_SCALAR_NAMES.iter())
+        .chain(kind_vars)
+        .map(|v| v.to_string())
+        .collect()
+}
+
+/// The sample of the Examples tab for `kind`; `resolved` gives the recovery of the same episode.
+pub fn example_context(kind: NotificationKind, resolved: bool) -> NotificationContext {
+    match kind {
+        NotificationKind::Anomaly => anomaly_example(resolved),
+        NotificationKind::Incident => incident_example(resolved),
+        _ => {
+            let level = if resolved {
+                AlertLevel::Ok
+            } else {
+                AlertLevel::Critical
+            };
+            let mut ctx = super::preview::synthetic_context(Some(level));
+            ctx.kind = kind;
+            ctx
+        }
+    }
+}
+
+fn base(kind: NotificationKind, resolved: bool, level: AlertLevel) -> NotificationContext {
+    NotificationContext {
+        org_name: "acme".into(),
+        stream_type: "logs".into(),
+        stream_name: "web-logs".into(),
+        alert_status: if resolved {
+            STATUS_RESOLVED
+        } else {
+            STATUS_FIRING
+        }
+        .into(),
+        alert_level: level_str(level).into(),
+        level: Some(level),
+        kind,
+        cluster_name: "o2-acme".into(),
+        alert_start_time: "N/A".into(),
+        alert_end_time: "N/A".into(),
+        alert_start_time_millis: "N/A".into(),
+        alert_end_time_millis: "N/A".into(),
+        ..Default::default()
+    }
+}
+
+fn anomaly_example(resolved: bool) -> NotificationContext {
+    let link = "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme";
+    let level = if resolved {
+        AlertLevel::Ok
+    } else {
+        AlertLevel::Critical
+    };
+    let mut ctx = base(NotificationKind::Anomaly, resolved, level);
+    ctx.alert_name = "error rate on web-logs".into();
+    ctx.alert_url = link.into();
+    ctx.event = if resolved {
+        STATUS_RESOLVED.into()
+    } else {
+        "firing:1762077900000000".into()
+    };
+    ctx.episode_id = Some("a1b2c3:1762077900000000".into());
+    let (start, end) = if resolved {
+        ("2026-11-02 10:25 UTC", "2026-11-02 10:40 UTC")
+    } else {
+        ("2026-11-02 09:50 UTC", "2026-11-02 10:05 UTC")
+    };
+    ctx.alert_trigger_time_str = if resolved {
+        "2026-11-02 10:40:00"
+    } else {
+        "2026-11-02 10:05:00"
+    }
+    .into();
+    ctx.alert_description = if resolved {
+        format!("a full detection run found no anomalies in window {start}\u{2013}{end}")
+    } else {
+        format!(
+            "3 anomalies in window {start}\u{2013}{end} | worst value: 342.00 (expected ~98.00 \
+             for this hour) | score 0.91 vs threshold 0.80 (score 14.0% over its bar)"
+        )
+    };
+    let (count, worst, expected, score, deviation, kind) = if resolved {
+        ("0", "", "", "0.00", "", "recovery")
+    } else {
+        ("3", "342.00", "98.00", "0.91", "14.0", "value")
+    };
+    ctx.context_attributes = [
+        ("anomaly_id", "a1b2c3"),
+        ("anomaly_name", "error rate on web-logs"),
+        ("anomaly_kind", kind),
+        ("anomaly_count", count),
+        ("max_deviation_percent", deviation),
+        (
+            "deviation_kind",
+            if resolved { "" } else { "score_over_threshold" },
+        ),
+        ("worst_actual_value", worst),
+        ("worst_expected", expected),
+        ("score", score),
+        ("threshold", "0.80"),
+        ("window_start", start),
+        ("window_end", end),
+        ("anomaly_url", link),
+    ]
+    .map(|(k, v)| (k.to_string(), v.to_string()))
+    .to_vec();
+    ctx
+}
+
+fn incident_example(resolved: bool) -> NotificationContext {
+    let url = "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme";
+    let level = if resolved {
+        AlertLevel::Ok
+    } else {
+        AlertLevel::Critical
+    };
+    let mut ctx = base(NotificationKind::Incident, resolved, level);
+    ctx.alert_name = "High error rate".into();
+    ctx.alert_url = url.into();
+    ctx.incident_id = "inc_7f3a".into();
+    ctx.episode_id = Some("inc_7f3a:1762077900000000".into());
+    let event = if resolved {
+        STATUS_RESOLVED
+    } else {
+        "new_incident_created"
+    };
+    ctx.event = if resolved { "resolved" } else { "opened" }.into();
+    ctx.alert_description = if resolved {
+        "All linked alerts recovered"
+    } else {
+        "Error rate above 5% for 5 minutes"
+    }
+    .into();
+    ctx.alert_trigger_time_str = if resolved {
+        "2026-11-02 10:40:00"
+    } else {
+        "2026-11-02 10:05:00"
+    }
+    .into();
+    if !resolved {
+        ctx.rows = vec![
+            serde_json::from_str(r#"{"host":"web-1","errors":9}"#).unwrap(),
+            serde_json::from_str(r#"{"host":"web-2","errors":3}"#).unwrap(),
+        ];
+        ctx.row_columns = build_row_columns(&ctx.rows);
+    }
+    ctx.context_attributes = [
+        ("incident_id", "inc_7f3a"),
+        ("incident_title", "High error rate on web-logs"),
+        ("incident_event", event),
+        ("incident_severity", "P2"),
+        ("incident_service", "checkout"),
+        ("incident_url", url),
+    ]
+    .map(|(k, v)| (k.to_string(), v.to_string()))
+    .to_vec();
+    ctx
+}
+
+fn level_str(level: AlertLevel) -> &'static str {
+    match level {
+        AlertLevel::Critical => "critical",
+        AlertLevel::Warning => "warning",
+        AlertLevel::Ok => "ok",
+        AlertLevel::NoData => "no_data",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::alerts::notifications::{
+        default_template::compiled_content_for,
+        format::ChannelFormat,
+        render::{RenderedMessage, render},
+        resolve::resolve_content,
+    };
+
+    const EXAMPLE_ANOMALY: &str = include_str!("render/testdata/examples/anomaly.json");
+    const EXAMPLE_INCIDENT: &str = include_str!("render/testdata/examples/incident.json");
+
+    fn rendered(
+        kind: NotificationKind,
+        resolved: bool,
+        format: ChannelFormat,
+    ) -> serde_json::Value {
+        let ctx = example_context(kind, resolved);
+        let content = resolve_content(&compiled_content_for(kind), &ctx, format.channel_family());
+        match render(format, &content, &ctx).unwrap() {
+            RenderedMessage::Http { body } => serde_json::from_str(&body).unwrap(),
+            other => panic!("expected Http, got {other:?}"),
+        }
+    }
+
+    fn example(kind: NotificationKind) -> serde_json::Value {
+        serde_json::json!({
+            "slack_firing": rendered(kind, false, ChannelFormat::Slack),
+            "slack_resolved": rendered(kind, true, ChannelFormat::Slack),
+            "webhook_firing": rendered(kind, false, ChannelFormat::Webhook),
+        })
+    }
+
+    #[test]
+    fn golden_example_anomaly() {
+        let want: serde_json::Value = serde_json::from_str(EXAMPLE_ANOMALY).unwrap();
+        assert_eq!(example(NotificationKind::Anomaly), want);
+        let title = &want["slack_firing"]["attachments"][0]["blocks"][0]["text"]["text"];
+        assert!(
+            title
+                .as_str()
+                .unwrap()
+                .contains("Anomaly firing: error rate on web-logs on web-logs")
+        );
+    }
+
+    #[test]
+    fn golden_example_incident() {
+        let want: serde_json::Value = serde_json::from_str(EXAMPLE_INCIDENT).unwrap();
+        assert_eq!(example(NotificationKind::Incident), want);
+        assert_eq!(
+            want["webhook_firing"]["context"]["kind"], "incident",
+            "the envelope names the kind"
+        );
+    }
+
+    #[test]
+    fn anomaly_recovery_hides_empty_fields() {
+        let ctx = example_context(NotificationKind::Anomaly, true);
+        let content = resolve_content(
+            &compiled_content_for(NotificationKind::Anomaly),
+            &ctx,
+            "slack",
+        );
+        let labels: Vec<&str> = content.fields.iter().map(|(l, _)| l.as_str()).collect();
+        assert_eq!(labels, ["Window", "Score"]);
+    }
+
+    #[test]
+    fn variables_for_lists_the_kind_group_after_the_common_one() {
+        let vars = variables_for(NotificationKind::Anomaly);
+        assert!(vars.contains(&"alert_name".to_string()));
+        assert!(vars.contains(&"notification_kind".to_string()));
+        assert!(vars.contains(&"anomaly_count".to_string()));
+        assert!(!variables_for(NotificationKind::Alert).contains(&"anomaly_count".to_string()));
+    }
+
+    /// Regenerate `render/testdata/examples/`; hand-check every file before committing it.
+    #[test]
+    #[ignore]
+    fn capture_examples() {
+        let dir = concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/src/alerts/notifications/render/testdata/examples"
+        );
+        std::fs::create_dir_all(dir).unwrap();
+        for (kind, file) in [
+            (NotificationKind::Anomaly, "anomaly.json"),
+            (NotificationKind::Incident, "incident.json"),
+        ] {
+            let body = serde_json::to_string_pretty(&example(kind)).unwrap() + "\n";
+            std::fs::write(format!("{dir}/{file}"), body).unwrap();
+        }
+    }
+}
diff --git a/src/core/src/alerts/notifications/format.rs b/src/core/src/alerts/notifications/format.rs
index 85bb9836f0..c6ce0513b8 100644
--- a/src/core/src/alerts/notifications/format.rs
+++ b/src/core/src/alerts/notifications/format.rs
@@ -36,6 +36,22 @@ pub enum ChannelFormat {
 }
 
 impl ChannelFormat {
+    /// The channel name a delivery row stores and the preview API accepts.
+    pub fn wire_name(&self) -> &'static str {
+        match self {
+            Self::Slack => "slack",
+            Self::TeamsAdaptiveCard => "teams_adaptivecard",
+            Self::TeamsMessageCard => "teams_messagecard",
+            Self::Discord => "discord",
+            Self::PagerDuty => "pagerduty",
+            Self::Opsgenie => "opsgenie",
+            Self::ServiceNow => "servicenow",
+            Self::Email => "email",
+            Self::Sns => "sns",
+            Self::Webhook => "webhook",
+        }
+    }
+
     /// The `*_overrides` key this format resolves against
     /// ([`resolve_content`](super::resolve::resolve_content)).
     ///
diff --git a/src/core/src/alerts/notifications/mod.rs b/src/core/src/alerts/notifications/mod.rs
index 46570b480e..55c91da27a 100644
--- a/src/core/src/alerts/notifications/mod.rs
+++ b/src/core/src/alerts/notifications/mod.rs
@@ -23,17 +23,20 @@ pub mod chart;
 pub mod context;
 pub mod custom;
 pub mod default_template;
+pub mod examples;
 pub mod format;
 pub mod jira;
 pub(crate) mod modifiers;
+pub mod notify;
 pub mod org_default;
 pub mod platform;
 pub mod preview;
+pub mod record;
 pub mod render;
 pub mod resolve;
 pub mod test_send;
 
-pub use context::{NotificationContext, build_row_columns};
+pub use context::{LegacyPayload, NotificationContext, build_row_columns};
 pub use custom::apply_custom_template;
 pub use format::{ChannelFormat, derive_channel_format, teams_format_for_url};
 pub use preview::{PreviewError, PreviewRequest, PreviewResponse, preview};
diff --git a/src/core/src/alerts/notifications/notify.rs b/src/core/src/alerts/notifications/notify.rs
new file mode 100644
index 0000000000..49d6859966
--- /dev/null
+++ b/src/core/src/alerts/notifications/notify.rs
@@ -0,0 +1,763 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! The shared send path every non-alert producer calls: resolve, render, send (Backend section 5).
+
+use config::meta::{
+    alerts::notification_kind::{NotificationKind, PlatformEvent},
+    destinations::{DestinationType, Module, Template},
+};
+use db::alerts::destinations::DestinationError;
+
+use super::{
+    NotificationContext,
+    format::{ChannelFormat, derive_channel_format},
+    org_default::{EffectiveTemplate, TemplateSource, resolve_effective_template_for},
+    platform::{Platform, ResolveAnswer, platform_of},
+    record::{ReplayRecord, render_record},
+};
+use crate::alerts::{
+    alert::{HttpAttempt, send_discord_attempt, send_http_attempt, send_sns_notification},
+    destinations::{self, note_skipped, org_member_split},
+};
+
+/// Must stay under the ledger lease so a hung transport never outlives its claim.
+pub const TRANSPORT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
+
+/// The destination metadata key that forces the legacy body or the rendered envelope (D1).
+pub const PAYLOAD_FORMAT_KEY: &str = "payload_format";
+
+/// The `template_source` of a row whose body is the producer's legacy JSON.
+pub const SOURCE_LEGACY: &str = "legacy";
+pub const SOURCE_PLATFORM_RESOLVE: &str = "platform_resolve";
+
+#[derive(Debug, thiserror::Error)]
+pub enum NotifyError {
+    #[error("destination {0} not found")]
+    DestinationNotFound(String),
+    #[error("destination {0} is not an alert destination")]
+    NotAlertDestination(String),
+    #[error("destination {destination} names template {template}, which does not exist")]
+    TemplateNotFound {
+        destination: String,
+        template: String,
+    },
+    #[error("render failed: {0}")]
+    Render(String),
+    #[error("{0}")]
+    Blocked(String),
+    #[error("HTTP {status}: {body}")]
+    Rejected { status: u16, body: String },
+    #[error("{0}")]
+    Transport(String),
+    #[error("skipped: {0:?}")]
+    Skipped(SkipReason),
+}
+
+impl NotifyError {
+    /// Worth another attempt: no answer, or an answer that says try later.
+    pub fn is_retryable(&self) -> bool {
+        match self {
+            NotifyError::Rejected { status, .. } => matches!(status, 408 | 429) || *status >= 500,
+            NotifyError::Transport(_) => true,
+            _ => false,
+        }
+    }
+
+    pub fn status_code(&self) -> Option<u16> {
+        match self {
+            NotifyError::Rejected { status, .. } => Some(*status),
+            _ => None,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum SkipReason {
+    AlreadyDelivered,
+    InFlight,
+    RetryScheduled,
+    Exhausted,
+    NoOrgMemberRecipient,
+    PlatformNoAck,
+    PlatformNotForKind,
+    NoTriggerRecorded,
+    Superseded,
+    DestinationChanged,
+}
+
+impl SkipReason {
+    pub fn outcome(self) -> SkipOutcome {
+        match self {
+            SkipReason::AlreadyDelivered => SkipOutcome::Delivered,
+            SkipReason::InFlight | SkipReason::RetryScheduled => SkipOutcome::Pending,
+            SkipReason::PlatformNoAck
+            | SkipReason::PlatformNotForKind
+            | SkipReason::NoTriggerRecorded
+            | SkipReason::Superseded => SkipOutcome::Handled,
+            SkipReason::Exhausted
+            | SkipReason::NoOrgMemberRecipient
+            | SkipReason::DestinationChanged => SkipOutcome::Unsuccessful,
+        }
+    }
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            SkipReason::AlreadyDelivered => "AlreadyDelivered",
+            SkipReason::InFlight => "InFlight",
+            SkipReason::RetryScheduled => "RetryScheduled",
+            SkipReason::Exhausted => "Exhausted",
+            SkipReason::NoOrgMemberRecipient => "NoOrgMemberRecipient",
+            SkipReason::PlatformNoAck => "PlatformNoAck",
+            SkipReason::PlatformNotForKind => "PlatformNotForKind",
+            SkipReason::NoTriggerRecorded => "NoTriggerRecorded",
+            SkipReason::Superseded => "Superseded",
+            SkipReason::DestinationChanged => "DestinationChanged",
+        }
+    }
+
+    pub fn parse(s: &str) -> Option<SkipReason> {
+        [
+            SkipReason::AlreadyDelivered,
+            SkipReason::InFlight,
+            SkipReason::RetryScheduled,
+            SkipReason::Exhausted,
+            SkipReason::NoOrgMemberRecipient,
+            SkipReason::PlatformNoAck,
+            SkipReason::PlatformNotForKind,
+            SkipReason::NoTriggerRecorded,
+            SkipReason::Superseded,
+            SkipReason::DestinationChanged,
+        ]
+        .into_iter()
+        .find(|r| r.as_str() == s)
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum SkipOutcome {
+    Delivered,
+    Pending,
+    Handled,
+    Unsuccessful,
+}
+
+#[derive(Clone, Debug)]
+pub struct Resolved {
+    pub destination: config::meta::destinations::Destination,
+    pub destination_type: DestinationType,
+    pub effective: EffectiveTemplate,
+    pub channel: ChannelFormat,
+}
+
+/// The finished delivery row as the producer sees it.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DeliveryRow {
+    pub id: String,
+    pub status: String,
+    pub status_code: Option<u16>,
+    pub attempt: i32,
+}
+
+/// What `send` decided to put on the wire before the ledger sees it.
+pub struct Prepared {
+    pub record: ReplayRecord,
+    pub template_source: String,
+    pub template_name: String,
+    pub template_skipped: String,
+    pub platform: Option<Platform>,
+}
+
+/// Find the destination and the template that renders this send.
+pub async fn resolve(
+    org_id: &str,
+    dest_name: &str,
+    explicit: &[Template],
+    kind: NotificationKind,
+) -> Result<Resolved, NotifyError> {
+    let (destination, dest_template) = destinations::get_with_template(org_id, dest_name)
+        .await
+        .map_err(|e| match e {
+            DestinationError::NotFound => NotifyError::DestinationNotFound(dest_name.to_string()),
+            DestinationError::TemplateNotFound => NotifyError::TemplateNotFound {
+                destination: dest_name.to_string(),
+                template: String::new(),
+            },
+            other => NotifyError::Transport(other.to_string()),
+        })?;
+    let Module::Alert {
+        destination_type, ..
+    } = &destination.module
+    else {
+        return Err(NotifyError::NotAlertDestination(dest_name.to_string()));
+    };
+    let destination_type = destination_type.clone();
+    let effective = resolve_effective_template_for(
+        org_id,
+        kind,
+        explicit,
+        dest_template,
+        Some(&destination_type),
+    )
+    .await;
+    let channel = derive_channel_format(&destination_type);
+    Ok(Resolved {
+        destination,
+        destination_type,
+        effective,
+        channel,
+    })
+}
+
+/// Render and send one notification to one resolved destination.
+pub async fn send(
+    org_id: &str,
+    resolved: &Resolved,
+    subject: Option<&str>,
+    ctx: &mut NotificationContext,
+) -> Result<DeliveryRow, NotifyError> {
+    apply_destination_metadata(&resolved.destination_type, ctx);
+    let prepared = prepare(resolved, subject, ctx)?;
+    transport(org_id, &resolved.destination_type, &prepared.record).await?;
+    Ok(DeliveryRow {
+        status: "sent".into(),
+        attempt: 1,
+        ..Default::default()
+    })
+}
+
+/// The Slack app outbox and Jira issues have no replayable body, so both bypass the ledger.
+async fn send_outside_ledger(
+    org_id: &str,
+    resolved: &Resolved,
+    ctx: &mut NotificationContext,
+) -> Option<Result<DeliveryRow, NotifyError>> {
+    let sent = match &resolved.destination_type {
+        DestinationType::SlackApp(target) => {
+            crate::alerts::slack_publish::send_templated_post(
+                &crate::alerts::slack_publish::QueueSink,
+                org_id,
+                target,
+                resolved.effective.template(),
+                ctx,
+            )
+            .await
+        }
+        DestinationType::Http(endpoint)
+            if super::jira::is_jira(endpoint) && ctx.kind == NotificationKind::Incident =>
+        {
+            super::jira::send(endpoint, ctx).await
+        }
+        _ => return None,
+    };
+    Some(
+        sent.map(|_| DeliveryRow {
+            status: "sent".into(),
+            attempt: 1,
+            ..Default::default()
+        })
+        .map_err(|e| NotifyError::Transport(e.to_string())),
+    )
+}
+
+/// Resolve, then send.
+pub async fn notify(
+    org_id: &str,
+    dest_name: &str,
+    explicit: &[Template],
+    subject: Option<&str>,
+    ctx: &mut NotificationContext,
+) -> Result<DeliveryRow, NotifyError> {
+    let resolved = resolve(org_id, dest_name, explicit, ctx.kind).await?;
+    send(org_id, &resolved, subject, ctx).await
+}
+
+/// The platform rule, the legacy rule (D1), then the render; nothing here touches the network.
+pub fn prepare(
+    resolved: &Resolved,
+    subject: Option<&str>,
+    ctx: &NotificationContext,
+) -> Result<Prepared, NotifyError> {
+    let effective = &resolved.effective;
+    let platform = platform_of(&resolved.destination_type);
+    if let Some(platform) = platform {
+        let episode_id = ctx.episode_id.as_deref().unwrap_or_default();
+        match ctx.kind.platform_event(&ctx.event) {
+            PlatformEvent::NoAck => return Err(NotifyError::Skipped(SkipReason::PlatformNoAck)),
+            PlatformEvent::NotForKind => {
+                return Err(NotifyError::Skipped(SkipReason::PlatformNotForKind));
+            }
+            PlatformEvent::Resolve | PlatformEvent::Trigger if episode_id.is_empty() => {
+                return Err(NotifyError::Render(
+                    "episode_id empty on a platform send".to_string(),
+                ));
+            }
+            PlatformEvent::Resolve => {
+                return Ok(Prepared {
+                    record: ReplayRecord::platform_resolve(platform, episode_id),
+                    template_source: SOURCE_PLATFORM_RESOLVE.into(),
+                    template_name: String::new(),
+                    template_skipped: "[]".into(),
+                    platform: Some(platform),
+                });
+            }
+            PlatformEvent::Trigger => {}
+        }
+    }
+    if let Some(legacy) = legacy_body(resolved, ctx) {
+        return Ok(Prepared {
+            record: ReplayRecord::http(
+                &method_name(&resolved.destination_type),
+                legacy.to_string(),
+                None,
+            ),
+            template_source: SOURCE_LEGACY.into(),
+            template_name: String::new(),
+            template_skipped: effective.skipped_json(),
+            platform,
+        });
+    }
+    let record = render_record(
+        &resolved.destination_type,
+        effective.template(),
+        ctx,
+        subject,
+    )
+    .map_err(NotifyError::Render)?;
+    Ok(Prepared {
+        record,
+        template_source: effective.source.as_str().into(),
+        template_name: effective.template().name.clone(),
+        template_skipped: effective.skipped_json(),
+        platform,
+    })
+}
+
+/// D1: a generic webhook with no user template at any level keeps the producer's JSON.
+fn legacy_body(resolved: &Resolved, ctx: &NotificationContext) -> Option<serde_json::Value> {
+    let legacy = ctx.legacy_payload.as_ref()?;
+    if !matches!(
+        ctx.kind,
+        NotificationKind::Anomaly | NotificationKind::Synthetics | NotificationKind::OnCall
+    ) || resolved.channel != ChannelFormat::Webhook
+    {
+        return None;
+    }
+    let forced = match &resolved.destination_type {
+        DestinationType::Http(endpoint) => endpoint.metadata.get(PAYLOAD_FORMAT_KEY).cloned(),
+        _ => None,
+    };
+    let use_legacy = match forced.as_deref() {
+        Some("legacy") => true,
+        Some("rendered") => false,
+        _ => resolved.effective.source == TemplateSource::Compiled,
+    };
+    use_legacy.then(|| legacy.build())
+}
+
+fn method_name(dest_type: &DestinationType) -> String {
+    match dest_type {
+        DestinationType::Http(endpoint) => format!("{:?}", endpoint.method),
+        _ => "POST".into(),
+    }
+}
+
+pub(crate) fn apply_destination_metadata(
+    dest_type: &DestinationType,
+    ctx: &mut NotificationContext,
+) {
+    ctx.metadata = match dest_type {
+        DestinationType::Http(endpoint) => endpoint
+            .metadata
+            .iter()
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect(),
+        _ => Vec::new(),
+    };
+}
+
+/// Put `record` on the wire with the destination's current URL, headers, recipients or topic.
+pub async fn transport(
+    org_id: &str,
+    dest_type: &DestinationType,
+    record: &ReplayRecord,
+) -> Result<String, NotifyError> {
+    if record.truncated {
+        return Err(NotifyError::Render(
+            "replay record was truncated; nothing to resend".into(),
+        ));
+    }
+    match (record.transport(), dest_type) {
+        ("platform_resolve", DestinationType::Http(endpoint)) => {
+            let platform = record
+                .platform
+                .as_ref()
+                .and_then(|p| super::record::parse_platform(&p.kind))
+                .ok_or_else(|| NotifyError::Render("platform record has no kind".into()))?;
+            let episode_id = record
+                .platform
+                .as_ref()
+                .map(|p| p.episode_id.as_str())
+                .unwrap_or_default();
+            platform_resolve(platform, endpoint, episode_id).await
+        }
+        ("http", DestinationType::Http(endpoint)) => {
+            let http = record.http.as_ref().ok_or_else(missing)?;
+            let attempt = match record.png() {
+                Some(png) if !http.parts.is_empty() => {
+                    send_discord_attempt(
+                        endpoint,
+                        http.body.clone(),
+                        std::sync::Arc::new(png),
+                        Some(TRANSPORT_TIMEOUT),
+                    )
+                    .await
+                }
+                _ => send_http_attempt(endpoint, http.body.clone(), Some(TRANSPORT_TIMEOUT)).await,
+            }
+            .map_err(|e| NotifyError::Render(e.to_string()))?;
+            http_outcome(attempt)
+        }
+        ("http", DestinationType::SlackApp(target)) => {
+            let http = record.http.as_ref().ok_or_else(missing)?;
+            crate::alerts::slack_publish::send_post(
+                &crate::alerts::slack_publish::QueueSink,
+                org_id,
+                target,
+                "",
+                &http.body,
+            )
+            .await
+            .map_err(|e| NotifyError::Transport(e.to_string()))
+        }
+        ("email", DestinationType::Email(email)) => {
+            let rec = record.email.as_ref().ok_or_else(missing)?;
+            let (kept, skipped) = org_member_split(org_id, &email.recipients)
+                .await
+                .map_err(|e| NotifyError::Transport(e.to_string()))?;
+            if kept.is_empty() {
+                return Err(NotifyError::Skipped(SkipReason::NoOrgMemberRecipient));
+            }
+            let to = config::meta::destinations::Email { recipients: kept };
+            crate::alerts::alert::send_email_notification_with_inline_png(
+                &rec.subject,
+                &to,
+                rec.text.clone(),
+                rec.html.clone(),
+                record.png().map(std::sync::Arc::new),
+            )
+            .await
+            .map(|resp| note_skipped(resp, &skipped))
+            .map_err(|e| NotifyError::Transport(e.to_string()))
+        }
+        ("sns", DestinationType::Sns(aws_sns)) => {
+            let rec = record.sns.as_ref().ok_or_else(missing)?;
+            send_sns_notification(&rec.subject, aws_sns, rec.message.clone())
+                .await
+                .map_err(|e| NotifyError::Transport(e.to_string()))
+        }
+        _ => Err(NotifyError::Skipped(SkipReason::DestinationChanged)),
+    }
+}
+
+fn missing() -> NotifyError {
+    NotifyError::Render("replay record has no body for its transport".into())
+}
+
+fn http_outcome(attempt: HttpAttempt) -> Result<String, NotifyError> {
+    match attempt {
+        HttpAttempt::Answered {
+            note: Some(note), ..
+        } => Ok(note),
+        HttpAttempt::Answered { status, body, .. } if status.is_success() => {
+            Ok(format!("sent status: {status}, body: {body}"))
+        }
+        HttpAttempt::Answered {
+            status,
+            body,
+            hide_body,
+            ..
+        } => Err(NotifyError::Rejected {
+            status: status.as_u16(),
+            body: if hide_body { String::new() } else { cap(&body) },
+        }),
+        HttpAttempt::Blocked(e) => Err(NotifyError::Blocked(e)),
+        HttpAttempt::NoAnswer(e) => Err(NotifyError::Transport(e)),
+    }
+}
+
+async fn platform_resolve(
+    platform: Platform,
+    endpoint: &config::meta::destinations::Endpoint,
+    episode_id: &str,
+) -> Result<String, NotifyError> {
+    let answer = super::platform::resolve_for_ledger(
+        platform,
+        endpoint,
+        episode_id,
+        Some(TRANSPORT_TIMEOUT),
+    )
+    .await
+    .map_err(|e| NotifyError::Render(e.to_string()))?;
+    match answer {
+        ResolveAnswer::Closed(text) | ResolveAnswer::NothingToClose(text) => Ok(text),
+        ResolveAnswer::Rejected { status, body } => Err(NotifyError::Rejected {
+            status,
+            body: cap(&body),
+        }),
+        ResolveAnswer::Blocked(e) => Err(NotifyError::Blocked(e)),
+        ResolveAnswer::NoAnswer(e) => Err(NotifyError::Transport(e)),
+    }
+}
+
+/// Error text stored on a row is capped at 2 KB.
+pub fn cap(s: &str) -> String {
+    const MAX: usize = 2048;
+    if s.len() <= MAX {
+        return s.to_string();
+    }
+    let mut end = MAX;
+    while !s.is_char_boundary(end) {
+        end -= 1;
+    }
+    s[..end].to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use config::meta::destinations::{Destination, Endpoint, TemplateKind, TemplateType};
+
+    use super::*;
+    use crate::alerts::notifications::{LegacyPayload, org_default::compiled_fallback_for};
+
+    fn resolved_for(dest_type: DestinationType, source: TemplateSource) -> Resolved {
+        let channel = derive_channel_format(&dest_type);
+        Resolved {
+            destination: Destination {
+                name: "d1".into(),
+                ..Default::default()
+            },
+            destination_type: dest_type,
+            effective: EffectiveTemplate {
+                template: compiled_fallback_for("acme", NotificationKind::Anomaly),
+                source,
+                skipped: vec![],
+            },
+            channel,
+        }
+    }
+
+    fn http(destination_type: &str, payload_format: Option<&str>) -> DestinationType {
+        let mut endpoint = Endpoint {
+            destination_type: (!destination_type.is_empty()).then(|| destination_type.to_string()),
+            url: "https://hooks.example/x".into(),
+            ..Default::default()
+        };
+        if let Some(f) = payload_format {
+            endpoint
+                .metadata
+                .insert(PAYLOAD_FORMAT_KEY.into(), f.into());
+        }
+        DestinationType::Http(endpoint)
+    }
+
+    fn anomaly_ctx(event: &str) -> NotificationContext {
+        let mut ctx = crate::alerts::notifications::examples::example_context(
+            NotificationKind::Anomaly,
+            event == "resolved",
+        );
+        ctx.event = event.into();
+        ctx.legacy_payload = Some(LegacyPayload::new(|| serde_json::json!({"text": "legacy"})));
+        ctx
+    }
+
+    fn body(p: &Prepared) -> serde_json::Value {
+        serde_json::from_str(&p.record.http.as_ref().unwrap().body).unwrap()
+    }
+
+    #[test]
+    fn a_generic_webhook_with_no_user_template_keeps_the_legacy_body() {
+        let r = resolved_for(http("", None), TemplateSource::Compiled);
+        let p = prepare(&r, None, &anomaly_ctx("firing:1")).unwrap();
+        assert_eq!(p.template_source, SOURCE_LEGACY);
+        assert_eq!(body(&p), serde_json::json!({"text": "legacy"}));
+    }
+
+    #[test]
+    fn payload_format_rendered_sends_the_envelope_with_the_kind() {
+        let r = resolved_for(http("", Some("rendered")), TemplateSource::Compiled);
+        let p = prepare(&r, None, &anomaly_ctx("firing:1")).unwrap();
+        assert_eq!(p.template_source, "compiled");
+        assert_eq!(body(&p)["context"]["kind"], "anomaly");
+    }
+
+    #[test]
+    fn a_user_template_on_a_generic_webhook_renders() {
+        let r = resolved_for(http("", None), TemplateSource::Destination);
+        let p = prepare(&r, None, &anomaly_ctx("firing:1")).unwrap();
+        assert_eq!(p.template_source, "destination");
+        assert_eq!(body(&p)["version"], 1);
+    }
+
+    #[test]
+    fn slack_never_gets_the_legacy_body() {
+        let r = resolved_for(http("slack", Some("legacy")), TemplateSource::Compiled);
+        let p = prepare(&r, None, &anomaly_ctx("firing:1")).unwrap();
+        assert_eq!(p.template_source, "compiled");
+        assert!(body(&p)["attachments"].is_array());
+    }
+
+    #[test]
+    fn a_platform_resolve_is_a_protocol_record_with_the_episode_key() {
+        let r = resolved_for(http("pagerduty", None), TemplateSource::Compiled);
+        let p = prepare(&r, None, &anomaly_ctx("resolved")).unwrap();
+        assert_eq!(p.template_source, SOURCE_PLATFORM_RESOLVE);
+        let platform = p.record.platform.unwrap();
+        assert_eq!(platform.kind, "pagerduty");
+        assert_eq!(platform.episode_id, "a1b2c3:1762077900000000");
+    }
+
+    #[test]
+    fn a_platform_trigger_carries_the_dedup_key() {
+        let r = resolved_for(http("pagerduty", None), TemplateSource::Compiled);
+        let p = prepare(&r, None, &anomaly_ctx("firing:1")).unwrap();
+        assert_eq!(body(&p)["dedup_key"], "a1b2c3:1762077900000000");
+        assert_eq!(body(&p)["event_action"], "trigger");
+    }
+
+    #[test]
+    fn an_empty_episode_on_a_platform_send_is_a_render_error() {
+        let r = resolved_for(http("pagerduty", None), TemplateSource::Compiled);
+        let mut ctx = anomaly_ctx("resolved");
+        ctx.episode_id = None;
+        assert!(matches!(
+            prepare(&r, None, &ctx),
+            Err(NotifyError::Render(_))
+        ));
+    }
+
+    #[test]
+    fn platform_skips_follow_the_rule_table() {
+        let r = resolved_for(http("opsgenie", None), TemplateSource::Compiled);
+        let mut ctx = anomaly_ctx("firing:1");
+        ctx.kind = NotificationKind::OnCall;
+        ctx.event = "acknowledged".into();
+        assert!(matches!(
+            prepare(&r, None, &ctx),
+            Err(NotifyError::Skipped(SkipReason::PlatformNoAck))
+        ));
+        ctx.kind = NotificationKind::Downtime;
+        ctx.event = "started".into();
+        assert!(matches!(
+            prepare(&r, None, &ctx),
+            Err(NotifyError::Skipped(SkipReason::PlatformNotForKind))
+        ));
+    }
+
+    #[test]
+    fn email_and_sns_render_from_the_compiled_card() {
+        let email = DestinationType::Email(config::meta::destinations::Email {
+            recipients: vec!["a@acme.com".into()],
+        });
+        let p = prepare(
+            &resolved_for(email, TemplateSource::Compiled),
+            None,
+            &anomaly_ctx("firing:1"),
+        )
+        .unwrap();
+        let rec = p.record.email.unwrap();
+        assert!(rec.subject.contains("Anomaly firing"));
+        assert!(rec.html.contains("342.00"));
+        assert!(rec.text.unwrap().contains("342.00"));
+
+        let sns = DestinationType::Sns(config::meta::destinations::AwsSns {
+            sns_topic_arn: "arn:x".into(),
+            aws_region: "us-east-1".into(),
+        });
+        let p = prepare(
+            &resolved_for(sns, TemplateSource::Compiled),
+            Some("subject"),
+            &anomaly_ctx("firing:1"),
+        )
+        .unwrap();
+        assert_eq!(p.record.sns.unwrap().subject, "subject");
+    }
+
+    #[test]
+    fn retryable_errors() {
+        assert!(
+            NotifyError::Rejected {
+                status: 503,
+                body: String::new()
+            }
+            .is_retryable()
+        );
+        assert!(
+            NotifyError::Rejected {
+                status: 429,
+                body: String::new()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !NotifyError::Rejected {
+                status: 400,
+                body: String::new()
+            }
+            .is_retryable()
+        );
+        assert!(NotifyError::Transport("t".into()).is_retryable());
+        assert!(!NotifyError::Blocked("b".into()).is_retryable());
+    }
+
+    #[test]
+    fn skip_outcomes() {
+        assert_eq!(
+            SkipReason::AlreadyDelivered.outcome(),
+            SkipOutcome::Delivered
+        );
+        assert_eq!(SkipReason::RetryScheduled.outcome(), SkipOutcome::Pending);
+        assert_eq!(
+            SkipReason::NoTriggerRecorded.outcome(),
+            SkipOutcome::Handled
+        );
+        assert_eq!(
+            SkipReason::DestinationChanged.outcome(),
+            SkipOutcome::Unsuccessful
+        );
+        assert_eq!(
+            SkipReason::parse("Superseded"),
+            Some(SkipReason::Superseded)
+        );
+    }
+
+    #[test]
+    fn custom_templates_fall_through_on_the_wrong_transport() {
+        let custom_http = Template {
+            name: "hook".into(),
+            kind: TemplateKind::Custom,
+            template_type: TemplateType::Http,
+            ..Default::default()
+        };
+        assert!(super::super::org_default::fits_transport(
+            &custom_http,
+            &http("slack", None)
+        ));
+        assert!(!super::super::org_default::fits_transport(
+            &custom_http,
+            &DestinationType::Email(Default::default())
+        ));
+    }
+}
diff --git a/src/core/src/alerts/notifications/org_default.rs b/src/core/src/alerts/notifications/org_default.rs
index 7112ed3f06..8d0e388b8f 100644
--- a/src/core/src/alerts/notifications/org_default.rs
+++ b/src/core/src/alerts/notifications/org_default.rs
@@ -24,14 +24,19 @@
 //! module only owns reading/writing the pointer value itself.
 
 use config::meta::{
-    destinations::{Template, TemplateKind, TemplateType},
+    alerts::notification_kind::NotificationKind,
+    destinations::{DestinationType, Template, TemplateKind, TemplateType},
     system_settings::{SettingScope, SystemSetting},
 };
 use db::system_settings;
 
-use super::default_template::{DEFAULT_CONTENT_TEMPLATE_NAME, compiled_default_content};
+use super::default_template::{
+    DEFAULT_CONTENT_TEMPLATE_NAME, compiled_content_for, compiled_name_for,
+    is_reserved_template_name,
+};
 
 pub const DEFAULT_ALERT_TEMPLATE_KEY: &str = "default_alert_template";
+pub const DEFAULT_TEMPLATES_KEY: &str = "default_templates";
 
 /// Extract the template name from a stored setting value, treating an empty
 /// string the same as absent — pure so it's testable without a DB.
@@ -87,73 +92,311 @@ pub async fn set_org_default_template(
     Ok(())
 }
 
-/// Which layer of the resolution chain actually supplied the template used
-/// for a notification (design §6.1).
-pub enum EffectiveTemplate {
-    /// Alert-level or destination-level — the caller already had a `Template`
-    /// in hand (from a DB lookup elsewhere) and just wants precedence applied.
-    Explicit(Template),
-    /// Resolved via the org's `default_alert_template` pointer.
-    OrgDefault(Template),
-    /// Terminal — pointer missing/dangling/unseeded. Never an error.
-    CompiledFallback(Template),
+/// Which precedence level supplied the template (Backend section 3).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum TemplateSource {
+    Explicit,
+    Destination,
+    OrgDefault,
+    LegacyAlertDefault,
+    Compiled,
 }
 
-impl EffectiveTemplate {
-    pub fn template(&self) -> &Template {
+impl TemplateSource {
+    /// The `template_source` value a delivery row stores.
+    pub fn as_str(self) -> &'static str {
         match self {
-            EffectiveTemplate::Explicit(t)
-            | EffectiveTemplate::OrgDefault(t)
-            | EffectiveTemplate::CompiledFallback(t) => t,
+            TemplateSource::Explicit => "explicit",
+            TemplateSource::Destination => "destination",
+            TemplateSource::OrgDefault | TemplateSource::LegacyAlertDefault => "org_default",
+            TemplateSource::Compiled => "compiled",
         }
     }
 }
 
-/// A transient (never persisted) `Template` wrapping the compiled-in
-/// fallback content, for callers that need the same `Template` shape as a DB
-/// row without a DB read.
-fn compiled_fallback_template(org_id: &str) -> Template {
-    Template {
-        id: None,
-        org_id: org_id.to_string(),
-        name: DEFAULT_CONTENT_TEMPLATE_NAME.to_string(),
-        is_default: false,
-        template_type: TemplateType::Http,
-        body: serde_json::to_string(&compiled_default_content())
-            .unwrap_or_else(|_| "{}".to_string()),
-        kind: TemplateKind::Content,
+#[derive(Clone, Debug)]
+pub struct EffectiveTemplate {
+    pub template: Template,
+    pub source: TemplateSource,
+    /// Levels passed over, with the reason, in the order they were tried.
+    pub skipped: Vec<(TemplateSource, String)>,
+}
+
+impl EffectiveTemplate {
+    pub fn template(&self) -> &Template {
+        &self.template
+    }
+
+    /// `template_skipped` as stored on a delivery row.
+    pub fn skipped_json(&self) -> String {
+        let list: Vec<serde_json::Value> = self
+            .skipped
+            .iter()
+            .map(|(source, reason)| {
+                serde_json::json!({ "level": source.as_str(), "reason": reason })
+            })
+            .collect();
+        serde_json::Value::Array(list).to_string()
     }
 }
 
-/// Resolve the template that should actually render this notification:
-/// alert-level → destination-level → org `default_alert_template` → compiled-in
-/// fallback. Never returns an error — the compiled fallback is infallible by
-/// construction (asserted by `default_template`'s own tests).
-///
-/// `alert_tpl`/`dest_tpl` are `None` when nothing explicit is bound; this
-/// function does NOT decide alert-vs-destination precedence between two
-/// explicit values — see `alert::choose_template` for that pure rule. Callers
-/// pass in whichever of the two `choose_template` already picked (or `None`
-/// if neither was set) as `explicit`.
+/// Not a reserved row (`prebuilt_*`, `o2_default_*`, `prebuilt_downtime*`).
+pub fn is_user_template(t: &Template) -> bool {
+    !is_reserved_template_name(&t.name)
+}
+
+/// Level 2: alerts keep any destination template; other kinds take only user templates (D2).
+pub fn destination_template_applies(t: &Template, kind: NotificationKind) -> bool {
+    kind == NotificationKind::Alert || is_user_template(t)
+}
+
+/// A Content template fits every transport; a Custom one only its own (Http, Email, Sns).
+pub fn fits_transport(t: &Template, dt: &DestinationType) -> bool {
+    match t.kind {
+        TemplateKind::Content => true,
+        TemplateKind::Custom => matches!(
+            (&t.template_type, dt),
+            (
+                TemplateType::Http,
+                DestinationType::Http(_) | DestinationType::SlackApp(_)
+            ) | (TemplateType::Email { .. }, DestinationType::Email(_))
+                | (TemplateType::Sns, DestinationType::Sns(_))
+        ),
+    }
+}
+
+/// The alert path's resolution: `explicit` is what `choose_template` already picked.
 pub async fn resolve_effective_template(
     org_id: &str,
     explicit: Option<Template>,
 ) -> EffectiveTemplate {
-    if let Some(t) = explicit {
-        return EffectiveTemplate::Explicit(t);
-    }
+    let explicit: Vec<Template> = explicit.into_iter().collect();
+    resolve_effective_template_for(org_id, NotificationKind::Alert, &explicit, None, None).await
+}
 
-    if let Some(name) = get_org_default_template(org_id).await
-        && let Ok(t) = db::alerts::templates::get(org_id, &name).await
+/// Walk levels 1 to 4 for one send; the alert path passes no `dt` and skips the transport rule.
+pub async fn resolve_effective_template_for(
+    org_id: &str,
+    kind: NotificationKind,
+    explicit: &[Template],
+    dest_template: Option<Template>,
+    dt: Option<&DestinationType>,
+) -> EffectiveTemplate {
+    let mut skipped = Vec::new();
+    if let Some((template, source)) =
+        choose_explicit_or_destination(kind, explicit, dest_template, dt, &mut skipped)
     {
-        // `db::alerts::templates::get` already falls back to the DEFAULT_ORG
-        // row by name when the org-scoped row is absent — the pointer to
-        // `o2_default_content` resolves through exactly that fallback.
-        return EffectiveTemplate::OrgDefault(t);
+        return EffectiveTemplate {
+            template,
+            source,
+            skipped,
+        };
+    }
+    let defaults = get_default_templates(org_id).await;
+    let org_default_name = defaults.get(kind).map(str::to_string);
+    let org_default = match &org_default_name {
+        Some(name) => db::alerts::templates::get(org_id, name).await.ok(),
+        None => None,
+    };
+    let legacy_default = if kind == NotificationKind::Alert && org_default.is_none() {
+        match get_org_default_template(org_id).await {
+            Some(name) => db::alerts::templates::get(org_id, &name).await.ok(),
+            None => None,
+        }
+    } else {
+        None
+    };
+    let fallback = OrgFallback {
+        explicit,
+        org_default,
+        org_default_name,
+        legacy_default,
+    };
+    choose_org_or_compiled(org_id, kind, fallback, dt, skipped)
+}
+
+/// Levels 1 and 2; `None` sends the caller on to the org levels with `skipped` filled.
+pub fn choose_explicit_or_destination(
+    kind: NotificationKind,
+    explicit: &[Template],
+    dest_template: Option<Template>,
+    dt: Option<&DestinationType>,
+    skipped: &mut Vec<(TemplateSource, String)>,
+) -> Option<(Template, TemplateSource)> {
+    let fits = |t: &Template| dt.is_none_or(|dt| fits_transport(t, dt));
+    for t in explicit {
+        if fits(t) {
+            return Some((t.clone(), TemplateSource::Explicit));
+        }
+        skipped.push((
+            TemplateSource::Explicit,
+            format!("{}: transport_mismatch", t.name),
+        ));
+    }
+    let t = dest_template?;
+    if !destination_template_applies(&t, kind) {
+        skipped.push((TemplateSource::Destination, format!("{}: reserved", t.name)));
+        return None;
+    }
+    if !fits(&t) {
+        skipped.push((
+            TemplateSource::Destination,
+            format!("{}: transport_mismatch", t.name),
+        ));
+        return None;
+    }
+    Some((t, TemplateSource::Destination))
+}
+
+/// What levels 3 and 4 have to offer, loaded by the caller so the walk itself is pure.
+pub struct OrgFallback<'a> {
+    pub explicit: &'a [Template],
+    pub org_default: Option<Template>,
+    pub org_default_name: Option<String>,
+    pub legacy_default: Option<Template>,
+}
+
+/// Levels 3 and 4; a name already tried at level 1 is not tried again.
+pub fn choose_org_or_compiled(
+    org_id: &str,
+    kind: NotificationKind,
+    fallback: OrgFallback<'_>,
+    dt: Option<&DestinationType>,
+    mut skipped: Vec<(TemplateSource, String)>,
+) -> EffectiveTemplate {
+    let fits = |t: &Template| dt.is_none_or(|dt| fits_transport(t, dt));
+    let tried_at_level_1 = |name: &str| fallback.explicit.iter().any(|t| t.name == name);
+    match (fallback.org_default, fallback.org_default_name) {
+        (Some(t), _) if tried_at_level_1(&t.name) => {}
+        (Some(t), _) if fits(&t) => {
+            return EffectiveTemplate {
+                template: t,
+                source: TemplateSource::OrgDefault,
+                skipped,
+            };
+        }
+        (Some(t), _) => skipped.push((
+            TemplateSource::OrgDefault,
+            format!("{}: transport_mismatch", t.name),
+        )),
+        (None, Some(name)) if !tried_at_level_1(&name) => {
+            skipped.push((TemplateSource::OrgDefault, format!("{name}: missing")))
+        }
+        (None, _) => {}
+    }
+    if let Some(t) = fallback.legacy_default
+        && fits(&t)
+    {
+        return EffectiveTemplate {
+            template: t,
+            source: TemplateSource::LegacyAlertDefault,
+            skipped,
+        };
+    }
+    EffectiveTemplate {
+        template: compiled_fallback_for(org_id, kind),
+        source: TemplateSource::Compiled,
+        skipped,
+    }
+}
+
+/// One pointer per kind, from the `default_templates` org setting; never seeded.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DefaultTemplates(pub std::collections::BTreeMap<String, String>);
+
+impl DefaultTemplates {
+    pub fn get(&self, kind: NotificationKind) -> Option<&str> {
+        self.0
+            .get(kind.as_str())
+            .map(String::as_str)
+            .filter(|n| !n.is_empty())
     }
 
-    // Pointer missing, or set but dangling — terminal fallback, never an error.
-    EffectiveTemplate::CompiledFallback(compiled_fallback_template(org_id))
+    pub fn from_value(v: &serde_json::Value) -> Self {
+        let map = v
+            .as_object()
+            .map(|o| {
+                o.iter()
+                    .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
+                    .collect()
+            })
+            .unwrap_or_default();
+        Self(map)
+    }
+}
+
+/// Read the org's `default_templates` map; a missing row is cached so later sends skip the read.
+pub async fn get_default_templates(org_id: &str) -> DefaultTemplates {
+    match system_settings::get(
+        &SettingScope::Org,
+        Some(org_id),
+        None,
+        DEFAULT_TEMPLATES_KEY,
+    )
+    .await
+    {
+        Ok(Some(setting)) => DefaultTemplates::from_value(&setting.setting_value),
+        Ok(None) => {
+            system_settings::set_only_cached(&default_templates_setting(
+                org_id,
+                serde_json::json!({}),
+            ))
+            .await;
+            DefaultTemplates::default()
+        }
+        Err(e) => {
+            log::error!("[TEMPLATES] read default_templates failed for org {org_id}: {e}");
+            DefaultTemplates::default()
+        }
+    }
+}
+
+/// Write `default_templates = {}` for an org that has no row yet.
+pub async fn init_default_templates(org_id: &str) -> Result<(), anyhow::Error> {
+    let existing = system_settings::get(
+        &SettingScope::Org,
+        Some(org_id),
+        None,
+        DEFAULT_TEMPLATES_KEY,
+    )
+    .await?;
+    if existing.is_some_and(|s| s.id.is_some()) {
+        return Ok(());
+    }
+    system_settings::set(&default_templates_setting(org_id, serde_json::json!({}))).await?;
+    Ok(())
+}
+
+fn default_templates_setting(org_id: &str, value: serde_json::Value) -> SystemSetting {
+    let now = chrono::Utc::now().timestamp_micros();
+    SystemSetting {
+        id: None,
+        scope: SettingScope::Org,
+        org_id: Some(org_id.to_string()),
+        user_id: None,
+        setting_key: DEFAULT_TEMPLATES_KEY.to_string(),
+        setting_category: Some("alerts".to_string()),
+        setting_value: value,
+        description: Some("Default notification template per kind for this org".into()),
+        created_at: now,
+        updated_at: now,
+        created_by: None,
+        updated_by: None,
+    }
+}
+
+/// A transient `Template` holding the kind's compiled default; never read from the database.
+pub fn compiled_fallback_for(org_id: &str, kind: NotificationKind) -> Template {
+    Template {
+        id: None,
+        org_id: org_id.to_string(),
+        name: compiled_name_for(kind).to_string(),
+        is_default: false,
+        template_type: TemplateType::Http,
+        body: serde_json::to_string(&compiled_content_for(kind))
+            .unwrap_or_else(|_| "{}".to_string()),
+        kind: TemplateKind::Content,
+    }
 }
 
 /// Idempotent: set the pointer (to the compiled default) for every org that
@@ -173,6 +416,12 @@ pub async fn backfill_org_default_templates() -> Result<(), anyhow::Error> {
         .unwrap_or_default();
     let mut backfilled = 0;
     for org in &orgs {
+        if let Err(e) = init_default_templates(&org.identifier).await {
+            log::error!(
+                "[TEMPLATES] backfill default_templates failed for org {}: {e}",
+                org.identifier
+            );
+        }
         if get_org_default_template(&org.identifier).await.is_some() {
             continue; // already has a pointer — never overwrite
         }
@@ -212,31 +461,29 @@ mod tests {
         assert_eq!(pointer_from_value(&serde_json::json!({})), None);
     }
 
-    /// `Explicit` short-circuits before any DB read — the only branch of
-    /// `resolve_effective_template` testable without a DB connection. The
-    /// `OrgDefault`/`CompiledFallback` branches are covered by
-    /// `resolve_resolution_chain_e2e` (Task 11's E2E, DB-backed).
-    #[tokio::test]
-    async fn resolve_effective_template_explicit_short_circuits() {
-        let explicit = Template {
+    fn tpl(name: &str, kind: TemplateKind, template_type: TemplateType) -> Template {
+        Template {
             id: None,
             org_id: "org1".into(),
-            name: "my_template".into(),
+            name: name.into(),
             is_default: false,
-            template_type: TemplateType::Http,
+            template_type,
             body: "{}".into(),
-            kind: TemplateKind::Content,
-        };
-        let resolved = resolve_effective_template("org1", Some(explicit.clone())).await;
-        match resolved {
-            EffectiveTemplate::Explicit(t) => assert_eq!(t.name, explicit.name),
-            _ => panic!("expected Explicit"),
+            kind,
         }
     }
 
+    #[tokio::test]
+    async fn resolve_effective_template_explicit_short_circuits() {
+        let explicit = tpl("my_template", TemplateKind::Content, TemplateType::Http);
+        let resolved = resolve_effective_template("org1", Some(explicit.clone())).await;
+        assert_eq!(resolved.source, TemplateSource::Explicit);
+        assert_eq!(resolved.template.name, explicit.name);
+    }
+
     #[test]
     fn compiled_fallback_template_shape() {
-        let t = compiled_fallback_template("org1");
+        let t = compiled_fallback_for("org1", NotificationKind::Alert);
         assert_eq!(t.org_id, "org1");
         assert_eq!(t.name, DEFAULT_CONTENT_TEMPLATE_NAME);
         assert_eq!(t.kind, TemplateKind::Content);
@@ -246,4 +493,234 @@ mod tests {
             serde_json::from_str(&t.body).expect("body must parse as ContentSpec");
         assert!(!spec.title.is_empty());
     }
+
+    fn content(name: &str) -> Template {
+        tpl(name, TemplateKind::Content, TemplateType::Http)
+    }
+
+    fn custom_http(name: &str) -> Template {
+        tpl(name, TemplateKind::Custom, TemplateType::Http)
+    }
+
+    fn email_dest() -> DestinationType {
+        DestinationType::Email(Default::default())
+    }
+
+    fn slack_dest() -> DestinationType {
+        DestinationType::Http(config::meta::destinations::Endpoint {
+            destination_type: Some("slack".into()),
+            ..Default::default()
+        })
+    }
+
+    /// Kind, explicit, destination, org default, destination type, expected source, skip count.
+    type Case = (
+        NotificationKind,
+        Vec<Template>,
+        Option<Template>,
+        Option<Template>,
+        DestinationType,
+        TemplateSource,
+        usize,
+    );
+
+    /// One walk over loaded candidates; `org` is level 3.
+    fn walk(
+        kind: NotificationKind,
+        explicit: &[Template],
+        dest: Option<Template>,
+        org: Option<Template>,
+        dt: &DestinationType,
+    ) -> EffectiveTemplate {
+        let mut skipped = Vec::new();
+        if let Some((template, source)) =
+            choose_explicit_or_destination(kind, explicit, dest, Some(dt), &mut skipped)
+        {
+            return EffectiveTemplate {
+                template,
+                source,
+                skipped,
+            };
+        }
+        let org_default_name = org.as_ref().map(|t| t.name.clone());
+        choose_org_or_compiled(
+            "org1",
+            kind,
+            OrgFallback {
+                explicit,
+                org_default: org,
+                org_default_name,
+                legacy_default: None,
+            },
+            Some(dt),
+            skipped,
+        )
+    }
+
+    #[test]
+    fn resolution_table() {
+        use NotificationKind::*;
+        let prebuilt = content("prebuilt_slack");
+        let user = content("ops-card");
+        let hook = custom_http("hook");
+        let cases: Vec<Case> = vec![
+            // Level 1 wins when it fits.
+            (
+                Anomaly,
+                vec![user.clone()],
+                Some(user.clone()),
+                None,
+                slack_dest(),
+                TemplateSource::Explicit,
+                0,
+            ),
+            // A Custom http override on email falls through, recording transport_mismatch.
+            (
+                Anomaly,
+                vec![hook.clone()],
+                None,
+                None,
+                email_dest(),
+                TemplateSource::Compiled,
+                1,
+            ),
+            (
+                Anomaly,
+                vec![hook.clone()],
+                Some(user.clone()),
+                None,
+                email_dest(),
+                TemplateSource::Destination,
+                1,
+            ),
+            // D2: a reserved destination template never shadows a kind default.
+            (
+                Anomaly,
+                vec![],
+                Some(prebuilt.clone()),
+                None,
+                slack_dest(),
+                TemplateSource::Compiled,
+                1,
+            ),
+            (
+                Incident,
+                vec![],
+                Some(prebuilt.clone()),
+                None,
+                slack_dest(),
+                TemplateSource::Compiled,
+                1,
+            ),
+            // Alerts keep any destination template.
+            (
+                Alert,
+                vec![],
+                Some(prebuilt.clone()),
+                None,
+                slack_dest(),
+                TemplateSource::Destination,
+                0,
+            ),
+            // A user destination template applies to every kind.
+            (
+                Anomaly,
+                vec![],
+                Some(user.clone()),
+                None,
+                slack_dest(),
+                TemplateSource::Destination,
+                0,
+            ),
+            // Level 3 is used when levels 1 and 2 give nothing.
+            (
+                Anomaly,
+                vec![],
+                Some(prebuilt.clone()),
+                Some(user.clone()),
+                slack_dest(),
+                TemplateSource::OrgDefault,
+                1,
+            ),
+            // A Custom http org default on email is skipped too.
+            (
+                Anomaly,
+                vec![],
+                None,
+                Some(hook.clone()),
+                email_dest(),
+                TemplateSource::Compiled,
+                1,
+            ),
+            // Incident 1a: the org incident override first, then 1b the alert's template.
+            (
+                Incident,
+                vec![hook.clone(), user.clone()],
+                None,
+                None,
+                email_dest(),
+                TemplateSource::Explicit,
+                1,
+            ),
+            (
+                Incident,
+                vec![user.clone(), hook.clone()],
+                None,
+                None,
+                email_dest(),
+                TemplateSource::Explicit,
+                0,
+            ),
+        ];
+        for (i, (kind, explicit, dest, org, dt, want, skips)) in cases.into_iter().enumerate() {
+            let got = walk(kind, &explicit, dest, org, &dt);
+            assert_eq!(got.source, want, "case {i}");
+            assert_eq!(got.skipped.len(), skips, "case {i}: {:?}", got.skipped);
+        }
+    }
+
+    #[test]
+    fn an_incident_override_that_does_not_fit_falls_to_the_alert_template() {
+        let got = walk(
+            NotificationKind::Incident,
+            &[custom_http("incident-hook"), content("alert-card")],
+            None,
+            None,
+            &email_dest(),
+        );
+        assert_eq!(got.template.name, "alert-card");
+        assert_eq!(got.skipped[0].1, "incident-hook: transport_mismatch");
+        assert!(got.skipped_json().contains("transport_mismatch"));
+    }
+
+    #[test]
+    fn a_name_tried_at_level_1_is_not_tried_again_at_level_3() {
+        let hook = custom_http("hook");
+        let got = walk(
+            NotificationKind::Incident,
+            std::slice::from_ref(&hook),
+            None,
+            Some(hook.clone()),
+            &email_dest(),
+        );
+        assert_eq!(got.source, TemplateSource::Compiled);
+        assert_eq!(got.skipped.len(), 1);
+    }
+
+    #[test]
+    fn the_compiled_default_follows_the_kind() {
+        let got = walk(NotificationKind::Anomaly, &[], None, None, &slack_dest());
+        assert_eq!(got.template.name, "o2_default_anomaly");
+        let got = walk(NotificationKind::Incident, &[], None, None, &slack_dest());
+        assert_eq!(got.template.name, "o2_default_incident");
+    }
+
+    #[test]
+    fn default_templates_codec() {
+        let d =
+            DefaultTemplates::from_value(&serde_json::json!({"incident": "ops", "anomaly": ""}));
+        assert_eq!(d.get(NotificationKind::Incident), Some("ops"));
+        assert_eq!(d.get(NotificationKind::Anomaly), None);
+        assert_eq!(d.get(NotificationKind::OnCall), None);
+    }
 }
diff --git a/src/core/src/alerts/notifications/platform.rs b/src/core/src/alerts/notifications/platform.rs
index 8c7db54624..531a0b65a8 100644
--- a/src/core/src/alerts/notifications/platform.rs
+++ b/src/core/src/alerts/notifications/platform.rs
@@ -85,12 +85,63 @@ pub fn stamp_key(
     Ok(value.to_string())
 }
 
+/// The prefix of a resolve answer that found no open record for the key.
+pub const NOTHING_TO_CLOSE: &str = "nothing to close";
+
+/// One resolve as the delivery ledger classifies it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ResolveAnswer {
+    Closed(String),
+    NothingToClose(String),
+    Rejected { status: u16, body: String },
+    Blocked(String),
+    NoAnswer(String),
+}
+
 /// Close whatever the firing opened; ServiceNow needs two, updating only by `sys_id`.
 pub async fn send_resolve(
     platform: Platform,
     endpoint: &Endpoint,
     episode_id: &str,
 ) -> Result<String> {
+    match resolve_once(platform, endpoint, episode_id, None).await? {
+        ResolveAnswer::Closed(text) | ResolveAnswer::NothingToClose(text) => Ok(text),
+        ResolveAnswer::Rejected { status, body } => Err(answer_error(status, &body)),
+        ResolveAnswer::Blocked(e) | ResolveAnswer::NoAnswer(e) => Err(anyhow::anyhow!(e)),
+    }
+}
+
+/// Like `send_resolve`, but Opsgenie's 404 on close also reads as nothing to close.
+pub async fn resolve_for_ledger(
+    platform: Platform,
+    endpoint: &Endpoint,
+    episode_id: &str,
+    timeout: Option<std::time::Duration>,
+) -> Result<ResolveAnswer> {
+    let answer = resolve_once(platform, endpoint, episode_id, timeout).await?;
+    Ok(match answer {
+        ResolveAnswer::Rejected { status: 404, body } if platform == Platform::Opsgenie => {
+            ResolveAnswer::NothingToClose(format!("{NOTHING_TO_CLOSE}: {body}"))
+        }
+        other => other,
+    })
+}
+
+fn answer_error(status: u16, body: &str) -> anyhow::Error {
+    let status = reqwest::StatusCode::from_u16(status).unwrap_or(reqwest::StatusCode::BAD_GATEWAY);
+    if body.is_empty() {
+        anyhow::anyhow!("status {status}")
+    } else {
+        anyhow::anyhow!("status {status}: {body}")
+    }
+}
+
+async fn resolve_once(
+    platform: Platform,
+    endpoint: &Endpoint,
+    episode_id: &str,
+    timeout: Option<std::time::Duration>,
+) -> Result<ResolveAnswer> {
     match platform {
         Platform::PagerDuty => {
             let body = json!({
@@ -100,13 +151,15 @@ pub async fn send_resolve(
                 "event_action": "resolve",
                 "dedup_key": episode_id,
             });
-            request(
+            let answer = request_raw(
                 endpoint,
                 reqwest::Method::POST,
                 &endpoint.url,
                 Some(body.to_string()),
+                timeout,
             )
-            .await
+            .await?;
+            Ok(answer.into_resolve())
         }
         Platform::Opsgenie => {
             let url = format!(
@@ -114,26 +167,35 @@ pub async fn send_resolve(
                 endpoint.url.trim_end_matches('/')
             );
             let body = json!({ "source": "OpenObserve" });
-            request(
+            let answer = request_raw(
                 endpoint,
                 reqwest::Method::POST,
                 &url,
                 Some(body.to_string()),
+                timeout,
             )
-            .await
+            .await?;
+            Ok(answer.into_resolve())
         }
-        Platform::ServiceNow => resolve_servicenow(endpoint, episode_id).await,
+        Platform::ServiceNow => resolve_servicenow(endpoint, episode_id, timeout).await,
     }
 }
 
-async fn resolve_servicenow(endpoint: &Endpoint, episode_id: &str) -> Result<String> {
+async fn resolve_servicenow(
+    endpoint: &Endpoint,
+    episode_id: &str,
+    timeout: Option<std::time::Duration>,
+) -> Result<ResolveAnswer> {
     let base = endpoint.url.trim_end_matches('/');
     // No dedup, so one episode leaves several records; closed ones reject a PATCH.
     let lookup = format!(
         "{base}?sysparm_query=correlation_id={episode_id}^state!={STATE_RESOLVED}^state!={STATE_CLOSED}\
          &sysparm_fields=sys_id&sysparm_limit={LOOKUP_LIMIT}"
     );
-    let found = request(endpoint, reqwest::Method::GET, &lookup, None).await?;
+    let found = match request_raw(endpoint, reqwest::Method::GET, &lookup, None, timeout).await? {
+        RawAnswer::Answered { status, text, .. } if status.is_success() => text,
+        other => return Ok(other.into_resolve()),
+    };
     let sys_ids: Vec<String> = serde_json::from_str::<Value>(&found)
         .ok()
         .and_then(|v| v["result"].as_array().cloned())
@@ -142,9 +204,10 @@ async fn resolve_servicenow(endpoint: &Endpoint, episode_id: &str) -> Result<Str
         .filter_map(|row| row["sys_id"].as_str().map(str::to_string))
         .collect();
     if sys_ids.is_empty() {
-        return Err(anyhow::anyhow!(
-            "no open incident carries correlation_id {episode_id}"
-        ));
+        return Ok(ResolveAnswer::Rejected {
+            status: 404,
+            body: format!("no open incident carries correlation_id {episode_id}"),
+        });
     }
 
     let close_code = endpoint
@@ -161,46 +224,94 @@ async fn resolve_servicenow(endpoint: &Endpoint, episode_id: &str) -> Result<Str
     .to_string();
 
     // Runs to the end: one failure must not strand the records after it.
-    let (mut closed, mut failures) = (0usize, String::new());
+    let (mut closed, mut failures, mut last) = (0usize, String::new(), None);
     for sys_id in &sys_ids {
-        match request(
+        let answer = request_raw(
             endpoint,
             reqwest::Method::PATCH,
             &format!("{base}/{sys_id}"),
             Some(body.clone()),
+            timeout,
         )
-        .await
-        {
-            Ok(_) => closed += 1,
-            Err(e) => failures = format!("{failures} {sys_id}: {e};"),
+        .await?
+        .into_resolve();
+        match answer {
+            ResolveAnswer::Closed(_) => closed += 1,
+            other => {
+                failures = format!("{failures} {sys_id}: {other:?};");
+                last = Some(other);
+            }
         }
     }
-    if failures.is_empty() {
-        Ok(format!("closed {closed} incident(s)"))
-    } else {
-        Err(anyhow::anyhow!(
+    match last {
+        None => Ok(ResolveAnswer::Closed(format!(
+            "closed {closed} incident(s)"
+        ))),
+        Some(ResolveAnswer::Rejected { status, .. }) => Ok(ResolveAnswer::Rejected {
+            status,
+            body: format!("closed {closed} of {}, failed:{failures}", sys_ids.len()),
+        }),
+        Some(ResolveAnswer::NoAnswer(_)) => Ok(ResolveAnswer::NoAnswer(format!(
             "closed {closed} of {}, failed:{failures}",
             sys_ids.len()
-        ))
+        ))),
+        Some(other) => Ok(other),
+    }
+}
+
+enum RawAnswer {
+    Answered {
+        status: reqwest::StatusCode,
+        text: String,
+        hide_body: bool,
+    },
+    Blocked(String),
+    NoAnswer(String),
+}
+
+impl RawAnswer {
+    fn into_resolve(self) -> ResolveAnswer {
+        match self {
+            RawAnswer::Answered { status, text, .. } if status.is_success() => {
+                ResolveAnswer::Closed(text)
+            }
+            RawAnswer::Answered {
+                status,
+                text,
+                hide_body,
+            } => ResolveAnswer::Rejected {
+                status: status.as_u16(),
+                body: if hide_body { String::new() } else { text },
+            },
+            RawAnswer::Blocked(e) => ResolveAnswer::Blocked(e),
+            RawAnswer::NoAnswer(e) => ResolveAnswer::NoAnswer(e),
+        }
     }
 }
 
 /// Guards the URL this module built rather than the one the operator typed.
-async fn request(
+async fn request_raw(
     endpoint: &Endpoint,
     method: reqwest::Method,
     url: &str,
     body: Option<String>,
-) -> Result<String> {
-    SsrfGuard::validate_destination_url_with_config_async(url)
-        .await
-        .map_err(|e| anyhow::anyhow!("URL blocked by SSRF guard: {e}"))?;
+    timeout: Option<std::time::Duration>,
+) -> Result<RawAnswer> {
+    if let Err(e) = SsrfGuard::validate_destination_url_with_config_async(url).await {
+        return Ok(RawAnswer::Blocked(format!(
+            "URL blocked by SSRF guard: {e}"
+        )));
+    }
 
     let builder = if endpoint.skip_tls_verify {
         reqwest::Client::builder().danger_accept_invalid_certs(true)
     } else {
         reqwest::Client::builder()
     };
+    let builder = match timeout {
+        Some(t) => builder.timeout(t),
+        None => builder,
+    };
     let client = common::utils::ssrf_guard::build_safe_destination_client(builder)?;
     let mut req = client.request(method, url::Url::parse(url)?);
 
@@ -223,17 +334,18 @@ async fn request(
         req = req.body(body);
     }
 
-    let resp = req.send().await.map_err(reqwest::Error::without_url)?;
+    let resp = match req.send().await {
+        Ok(resp) => resp,
+        Err(e) => return Ok(RawAnswer::NoAnswer(e.without_url().to_string())),
+    };
     let status = resp.status();
     let hide_body = config::utils::ssrf_guard::admitted_only_by_allowlist(&resp);
     let text = resp.text().await.unwrap_or_default();
-    if status.is_success() {
-        Ok(text)
-    } else if hide_body {
-        Err(anyhow::anyhow!("status {status}"))
-    } else {
-        Err(anyhow::anyhow!("status {status}: {text}"))
-    }
+    Ok(RawAnswer::Answered {
+        status,
+        text,
+        hide_body,
+    })
 }
 
 #[cfg(test)]
diff --git a/src/core/src/alerts/notifications/preview.rs b/src/core/src/alerts/notifications/preview.rs
index e17b271413..38b0f84500 100644
--- a/src/core/src/alerts/notifications/preview.rs
+++ b/src/core/src/alerts/notifications/preview.rs
@@ -21,7 +21,9 @@
 //! [`render`] functions Task 9's `send_notification` calls, never a
 //! reimplementation. See `preview_uses_production_renderer` below.
 
-use config::meta::alerts::{content_spec::ContentSpec, level::AlertLevel};
+use config::meta::alerts::{
+    content_spec::ContentSpec, level::AlertLevel, notification_kind::NotificationKind,
+};
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
@@ -44,6 +46,9 @@ pub struct PreviewRequest {
     /// Reserved for the Phase-1b alert-driven sample source; ignored in 1a.
     #[serde(default)]
     pub sample: Option<serde_json::Value>,
+    /// Which producer's sample to render with; absent means alert.
+    #[serde(default)]
+    pub kind: NotificationKind,
 }
 
 #[derive(Debug, Clone, Serialize, ToSchema)]
@@ -53,6 +58,8 @@ pub struct PreviewResponse {
     /// FE card rendering schema.
     pub preview_model: PreviewModel,
     pub unknown_variables: Vec<String>,
+    /// Every variable a template of the requested kind can use.
+    pub variables: Vec<String>,
 }
 
 #[derive(Debug, Clone, Serialize, ToSchema)]
@@ -203,6 +210,7 @@ pub fn synthetic_context(level: Option<AlertLevel>) -> NotificationContext {
         chart_png: None,
         alert_id: String::new(),
         incident_id: String::new(),
+        ..Default::default()
     }
 }
 
@@ -220,7 +228,16 @@ pub fn preview(req: &PreviewRequest) -> Result<PreviewResponse, PreviewError> {
         .validate()
         .map_err(PreviewError::InvalidContent)?;
 
-    let ctx = synthetic_context(level);
+    let ctx = match req.kind {
+        NotificationKind::Alert => synthetic_context(level),
+        kind => {
+            let mut ctx = super::examples::example_context(kind, level == Some(AlertLevel::Ok));
+            if level.is_some() {
+                ctx.level = level;
+            }
+            ctx
+        }
+    };
 
     let content = resolve_content(&req.definition, &ctx, format.channel_family());
     let rendered =
@@ -277,6 +294,7 @@ pub fn preview(req: &PreviewRequest) -> Result<PreviewResponse, PreviewError> {
         payload,
         preview_model,
         unknown_variables: content.unknown_variables,
+        variables: super::examples::variables_for(req.kind),
     })
 }
 
@@ -311,6 +329,7 @@ mod tests {
             channel: "slack".into(),
             severity: Some("critical".into()),
             sample: None,
+            kind: Default::default(),
         };
         let err = preview(&req).expect_err("hostile link previewed cleanly");
         assert!(
@@ -326,6 +345,7 @@ mod tests {
             channel: "slack".into(),
             severity: Some("critical".into()),
             sample: None,
+            kind: Default::default(),
         };
         let resp = preview(&req).unwrap();
         // Same payload the send path would produce for the same spec +
@@ -357,6 +377,7 @@ mod tests {
             channel: "slack".into(),
             severity: Some("critical".into()),
             sample: None,
+            kind: Default::default(),
         })
         .unwrap();
         assert!(
@@ -372,6 +393,7 @@ mod tests {
             channel: "slack".into(),
             severity: None,
             sample: None,
+            kind: Default::default(),
         })
         .unwrap();
         assert!(
@@ -392,6 +414,7 @@ mod tests {
             channel: "webhook".into(),
             severity: None,
             sample: None,
+            kind: Default::default(),
         })
         .unwrap();
         assert_eq!(resp.unknown_variables, vec!["alert_time_zzz"]);
@@ -406,6 +429,7 @@ mod tests {
             channel: "slack".into(),
             severity: Some("criticall".into()),
             sample: None,
+            kind: Default::default(),
         })
         .unwrap_err();
         assert!(matches!(err, PreviewError::UnknownSeverity(ref s) if s == "criticall"));
@@ -421,6 +445,7 @@ mod tests {
                 channel: "slack".into(),
                 severity: ok,
                 sample: None,
+                kind: Default::default(),
             })
             .expect("documented severity values must be accepted");
             payloads.push(resp.payload);
diff --git a/src/core/src/alerts/notifications/record.rs b/src/core/src/alerts/notifications/record.rs
new file mode 100644
index 0000000000..ccd7932720
--- /dev/null
+++ b/src/core/src/alerts/notifications/record.rs
@@ -0,0 +1,364 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! The replay record: a rendered message any node can resend with the destination's current URL.
+
+use base64::Engine as _;
+use config::meta::destinations::{DestinationType, Template, TemplateKind, TemplateType};
+use serde::{Deserialize, Serialize};
+
+use super::{
+    NotificationContext, RenderedMessage, apply_custom_template,
+    format::{ChannelFormat, derive_channel_format},
+    platform::Platform,
+    render::{discord::CHART_ATTACHMENT_NAME, email::CHART_CONTENT_ID, render},
+    resolve::resolve_content,
+};
+
+/// A record larger than this drops its images, then is truncated if it is still over.
+pub const RECORD_CAP_BYTES: usize = 256 * 1024;
+pub const IMAGE_DROPPED: &str = "image dropped from replay";
+
+const B64: base64::engine::GeneralPurpose = base64::engine::general_purpose::STANDARD;
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ReplayRecord {
+    pub v: u8,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub transport: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub http: Option<HttpRecord>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub email: Option<EmailRecord>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sns: Option<SnsRecord>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub platform: Option<PlatformRecord>,
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub truncated: bool,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct HttpRecord {
+    pub method: String,
+    pub body: String,
+    pub content_type: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub parts: Vec<Attachment>,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EmailRecord {
+    pub subject: String,
+    pub html: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub text: Option<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub inline: Vec<Attachment>,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Attachment {
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub name: String,
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub filename: String,
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub cid: String,
+    pub content_type: String,
+    pub bytes_b64: String,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SnsRecord {
+    pub subject: String,
+    pub message: String,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PlatformRecord {
+    pub kind: String,
+    pub episode_id: String,
+}
+
+impl ReplayRecord {
+    pub fn http(method: &str, body: String, png: Option<&[u8]>) -> Self {
+        Self {
+            v: 1,
+            transport: Some("http".into()),
+            http: Some(HttpRecord {
+                method: method.to_string(),
+                body,
+                content_type: "application/json".into(),
+                parts: png
+                    .map(|png| {
+                        vec![Attachment {
+                            name: "files[0]".into(),
+                            filename: CHART_ATTACHMENT_NAME.into(),
+                            content_type: "image/png".into(),
+                            bytes_b64: B64.encode(png),
+                            ..Default::default()
+                        }]
+                    })
+                    .unwrap_or_default(),
+            }),
+            ..Default::default()
+        }
+    }
+
+    pub fn email(subject: String, html: String, text: Option<String>, png: Option<&[u8]>) -> Self {
+        Self {
+            v: 1,
+            transport: Some("email".into()),
+            email: Some(EmailRecord {
+                subject,
+                html,
+                text,
+                inline: png
+                    .map(|png| {
+                        vec![Attachment {
+                            cid: CHART_CONTENT_ID.into(),
+                            content_type: "image/png".into(),
+                            bytes_b64: B64.encode(png),
+                            ..Default::default()
+                        }]
+                    })
+                    .unwrap_or_default(),
+            }),
+            ..Default::default()
+        }
+    }
+
+    pub fn sns(subject: String, message: String) -> Self {
+        Self {
+            v: 1,
+            transport: Some("sns".into()),
+            sns: Some(SnsRecord { subject, message }),
+            ..Default::default()
+        }
+    }
+
+    pub fn platform_resolve(platform: Platform, episode_id: &str) -> Self {
+        Self {
+            v: 1,
+            transport: Some("platform_resolve".into()),
+            platform: Some(PlatformRecord {
+                kind: platform_name(platform).into(),
+                episode_id: episode_id.to_string(),
+            }),
+            ..Default::default()
+        }
+    }
+
+    pub fn truncated() -> Self {
+        Self {
+            v: 1,
+            truncated: true,
+            ..Default::default()
+        }
+    }
+
+    /// The transport family this record was rendered for: http, email, sns or platform_resolve.
+    pub fn transport(&self) -> &str {
+        self.transport.as_deref().unwrap_or_default()
+    }
+
+    /// The chart PNG of a Discord multipart or email inline record, if it still carries one.
+    pub fn png(&self) -> Option<Vec<u8>> {
+        let part = self
+            .http
+            .as_ref()
+            .and_then(|h| h.parts.first())
+            .or_else(|| self.email.as_ref().and_then(|e| e.inline.first()))?;
+        B64.decode(&part.bytes_b64).ok()
+    }
+
+    /// The stored form under the cap, and whether the images had to be dropped to get there.
+    pub fn to_stored(&self) -> (String, bool) {
+        let full = serde_json::to_string(self).unwrap_or_default();
+        if full.len() <= RECORD_CAP_BYTES {
+            return (full, false);
+        }
+        let mut lean = self.clone();
+        if let Some(http) = lean.http.as_mut() {
+            http.parts.clear();
+        }
+        if let Some(email) = lean.email.as_mut() {
+            email.inline.clear();
+        }
+        let lean_json = serde_json::to_string(&lean).unwrap_or_default();
+        if lean_json.len() <= RECORD_CAP_BYTES {
+            return (lean_json, true);
+        }
+        (
+            serde_json::to_string(&ReplayRecord::truncated()).unwrap_or_default(),
+            true,
+        )
+    }
+
+    pub fn parse(stored: &str) -> Option<Self> {
+        serde_json::from_str(stored).ok()
+    }
+}
+
+pub fn platform_name(platform: Platform) -> &'static str {
+    match platform {
+        Platform::PagerDuty => "pagerduty",
+        Platform::Opsgenie => "opsgenie",
+        Platform::ServiceNow => "servicenow",
+    }
+}
+
+pub fn parse_platform(name: &str) -> Option<Platform> {
+    match name {
+        "pagerduty" => Some(Platform::PagerDuty),
+        "opsgenie" => Some(Platform::Opsgenie),
+        "servicenow" => Some(Platform::ServiceNow),
+        _ => None,
+    }
+}
+
+/// The transport family a destination type sends through.
+pub fn transport_of(dest_type: &DestinationType) -> &'static str {
+    match dest_type {
+        DestinationType::Http(_) | DestinationType::SlackApp(_) => "http",
+        DestinationType::Email(_) => "email",
+        DestinationType::Sns(_) => "sns",
+    }
+}
+
+/// Render `template` for `dest_type` into a record, the same way the alert send path renders.
+pub fn render_record(
+    dest_type: &DestinationType,
+    template: &Template,
+    ctx: &NotificationContext,
+    subject: Option<&str>,
+) -> Result<ReplayRecord, String> {
+    let is_email = matches!(dest_type, DestinationType::Email(_));
+    match template.kind {
+        TemplateKind::Custom => {
+            let msg = apply_custom_template(&template.body, ctx, is_email);
+            let msg = crate::alerts::alert::stamp_platform_key(dest_type, ctx, msg);
+            let title = match &template.template_type {
+                TemplateType::Email { title } => apply_custom_template(title, ctx, is_email),
+                _ => template.name.clone(),
+            };
+            let subject = subject.map(str::to_string).unwrap_or(title);
+            Ok(match dest_type {
+                DestinationType::Email(_) => ReplayRecord::email(subject, msg, None, None),
+                DestinationType::Sns(_) => ReplayRecord::sns(subject, msg),
+                _ => ReplayRecord::http(&method_of(dest_type), msg, None),
+            })
+        }
+        TemplateKind::Content => render_content_record(dest_type, template, ctx, subject),
+    }
+}
+
+fn render_content_record(
+    dest_type: &DestinationType,
+    template: &Template,
+    ctx: &NotificationContext,
+    subject: Option<&str>,
+) -> Result<ReplayRecord, String> {
+    let spec = db::alerts::templates::get_parsed_content(template)
+        .map_err(|e| format!("Invalid content template {}: {e}", template.name))?;
+    let format = derive_channel_format(dest_type);
+    let content = resolve_content(&spec, ctx, format.channel_family());
+    let title = content.title.clone();
+    let rendered = render(format, &content, ctx).map_err(|e| format!("Renderer failed: {e}"))?;
+    let png = ctx.chart_png.as_deref().map(|p| p.as_slice());
+    match (rendered, dest_type) {
+        (
+            RenderedMessage::Http { body },
+            DestinationType::Http(_) | DestinationType::SlackApp(_),
+        ) => {
+            if matches!(format, ChannelFormat::Discord) && png.is_some() {
+                return Ok(ReplayRecord::http(&method_of(dest_type), body, png));
+            }
+            let body = crate::alerts::alert::stamp_platform_key(dest_type, ctx, body);
+            Ok(ReplayRecord::http(&method_of(dest_type), body, None))
+        }
+        (
+            RenderedMessage::Email {
+                subject: rendered_subject,
+                html,
+                text,
+            },
+            DestinationType::Email(_),
+        ) => Ok(ReplayRecord::email(
+            subject.map(str::to_string).unwrap_or(rendered_subject),
+            html,
+            Some(text),
+            png,
+        )),
+        (RenderedMessage::Sns { message, .. }, DestinationType::Sns(_)) => Ok(ReplayRecord::sns(
+            subject.map(str::to_string).unwrap_or(title),
+            message,
+        )),
+        (rendered, dest_type) => Err(format!(
+            "Rendered message {rendered:?} does not match destination type {dest_type:?}"
+        )),
+    }
+}
+
+fn method_of(dest_type: &DestinationType) -> String {
+    match dest_type {
+        DestinationType::Http(endpoint) => format!("{:?}", endpoint.method),
+        _ => "POST".to_string(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn small_records_store_whole() {
+        let rec = ReplayRecord::http("POST", "{}".into(), Some(&[1, 2, 3]));
+        let (stored, dropped) = rec.to_stored();
+        assert!(!dropped);
+        let back = ReplayRecord::parse(&stored).unwrap();
+        assert_eq!(back, rec);
+        assert_eq!(back.png(), Some(vec![1, 2, 3]));
+    }
+
+    #[test]
+    fn an_oversized_record_drops_its_image_first() {
+        let png = vec![7u8; RECORD_CAP_BYTES];
+        let rec = ReplayRecord::http("POST", "{\"x\":1}".into(), Some(&png));
+        let (stored, dropped) = rec.to_stored();
+        assert!(dropped);
+        let back = ReplayRecord::parse(&stored).unwrap();
+        assert!(!back.truncated);
+        assert_eq!(back.http.unwrap().body, "{\"x\":1}");
+        assert_eq!(ReplayRecord::parse(&stored).unwrap().png(), None);
+    }
+
+    #[test]
+    fn a_body_still_over_the_cap_is_truncated() {
+        let rec = ReplayRecord::email("s".into(), "x".repeat(RECORD_CAP_BYTES + 1), None, None);
+        let (stored, dropped) = rec.to_stored();
+        assert!(dropped);
+        assert_eq!(stored, r#"{"v":1,"truncated":true}"#);
+    }
+
+    #[test]
+    fn email_inline_png_round_trips() {
+        let rec = ReplayRecord::email("s".into(), "<p>h</p>".into(), Some("t".into()), Some(&[9]));
+        let back = ReplayRecord::parse(&rec.to_stored().0).unwrap();
+        assert_eq!(back.email.as_ref().unwrap().inline[0].cid, CHART_CONTENT_ID);
+        assert_eq!(back.png(), Some(vec![9]));
+    }
+}
diff --git a/src/core/src/alerts/notifications/render/mod.rs b/src/core/src/alerts/notifications/render/mod.rs
index 31f860f7b3..ba5bb937b2 100644
--- a/src/core/src/alerts/notifications/render/mod.rs
+++ b/src/core/src/alerts/notifications/render/mod.rs
@@ -395,6 +395,7 @@ mod tests {
             chart_png: None,
             alert_id: String::new(),
             incident_id: String::new(),
+            ..Default::default()
         }
     }
 
diff --git a/src/core/src/alerts/notifications/render/testdata/examples/anomaly.json b/src/core/src/alerts/notifications/render/testdata/examples/anomaly.json
new file mode 100644
index 0000000000..d3bcf8b8a2
--- /dev/null
+++ b/src/core/src/alerts/notifications/render/testdata/examples/anomaly.json
@@ -0,0 +1,191 @@
+{
+  "slack_firing": {
+    "attachments": [
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme|Anomaly firing: error rate on web-logs on web-logs>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "3 anomalies in window 2026-11-02 09:50 UTC–2026-11-02 10:05 UTC | worst value: 342.00 (expected ~98.00 for this hour) | score 0.91 vs threshold 0.80 (score 14.0% over its bar)"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Worst value*\n342.00"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Expected*\n98.00"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Window*\n2026-11-02 09:50 UTC to 2026-11-02 10:05 UTC"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Score*\n0.91 vs 0.80"
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "slack_resolved": {
+    "attachments": [
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme|Anomaly resolved: error rate on web-logs on web-logs>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "a full detection run found no anomalies in window 2026-11-02 10:25 UTC–2026-11-02 10:40 UTC"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Window*\n2026-11-02 10:25 UTC to 2026-11-02 10:40 UTC"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Score*\n0.00 vs 0.80"
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "webhook_firing": {
+    "version": 1,
+    "title": "Anomaly firing: error rate on web-logs on web-logs",
+    "body_markdown": "3 anomalies in window 2026\\-11\\-02 09\\:50 UTC–2026\\-11\\-02 10\\:05 UTC \\| worst value\\: 342\\.00 \\(expected \\~98\\.00 for this hour\\) \\| score 0\\.91 vs threshold 0\\.80 \\(score 14\\.0\\% over its bar\\)",
+    "fields": {
+      "Worst value": "342.00",
+      "Expected": "98.00",
+      "Window": "2026-11-02 09:50 UTC to 2026-11-02 10:05 UTC",
+      "Score": "0.91 vs 0.80"
+    },
+    "rows": [],
+    "links": [
+      {
+        "label": "",
+        "url": "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme"
+      }
+    ],
+    "severity": "critical",
+    "chart_url": null,
+    "context": {
+      "org_name": "acme",
+      "stream_type": "logs",
+      "stream_name": "web-logs",
+      "alert_name": "error rate on web-logs",
+      "alert_type": null,
+      "alert_period": null,
+      "alert_operator": null,
+      "alert_threshold": null,
+      "alert_count": null,
+      "alert_agg_value": null,
+      "alert_level": "critical",
+      "alert_status": "firing",
+      "episode_id": "a1b2c3:1762077900000000",
+      "recovery_reason": null,
+      "alert_priority": null,
+      "alert_tags": null,
+      "alert_threshold_crit": null,
+      "alert_threshold_warn": null,
+      "alert_warning_threshold": null,
+      "alert_start_time": "N/A",
+      "alert_end_time": "N/A",
+      "alert_url": "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme",
+      "alert_trigger_time_str": "2026-11-02 10:05:00",
+      "alert_time": "2026-11-02 10:05:00",
+      "alert_description": "3 anomalies in window 2026-11-02 09:50 UTC–2026-11-02 10:05 UTC | worst value: 342.00 (expected ~98.00 for this hour) | score 0.91 vs threshold 0.80 (score 14.0% over its bar)",
+      "kind": "anomaly",
+      "notification_kind": "anomaly",
+      "notification_event": "firing:1762077900000000",
+      "cluster_name": "o2-acme",
+      "alert_start_time_millis": "N/A",
+      "alert_end_time_millis": "N/A",
+      "alert_trigger_time": "0",
+      "alert_trigger_time_millis": "0",
+      "alert_trigger_time_seconds": "0",
+      "alert_promql_operator": null,
+      "alert_promql_value": null,
+      "anomaly_id": "a1b2c3",
+      "anomaly_name": "error rate on web-logs",
+      "anomaly_kind": "value",
+      "anomaly_count": "3",
+      "max_deviation_percent": "14.0",
+      "deviation_kind": "score_over_threshold",
+      "worst_actual_value": "342.00",
+      "worst_expected": "98.00",
+      "score": "0.91",
+      "threshold": "0.80",
+      "window_start": "2026-11-02 09:50 UTC",
+      "window_end": "2026-11-02 10:05 UTC",
+      "anomaly_url": "https://o2.acme.com/web/alerts/anomaly/edit/a1b2c3?org_identifier=acme"
+    }
+  }
+}
diff --git a/src/core/src/alerts/notifications/render/testdata/examples/incident.json b/src/core/src/alerts/notifications/render/testdata/examples/incident.json
new file mode 100644
index 0000000000..df077d2514
--- /dev/null
+++ b/src/core/src/alerts/notifications/render/testdata/examples/incident.json
@@ -0,0 +1,216 @@
+{
+  "slack_firing": {
+    "attachments": [
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme|Incident P2: High error rate on web-logs>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "Incident *new_incident_created* for service *checkout*.\n\nError rate above 5% for 5 minutes"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Event*\nnew_incident_created"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Severity*\nP2"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Service*\ncheckout"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Alert*\nHigh error rate"
+              }
+            ]
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "`host` web-1 · `errors` 9\n`host` web-2 · `errors` 3"
+            }
+          }
+        ]
+      },
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "slack_resolved": {
+    "attachments": [
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme|Incident P2: High error rate on web-logs>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "Incident *resolved* for service *checkout*.\n\nAll linked alerts recovered"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Event*\nresolved"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Severity*\nP2"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Service*\ncheckout"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Alert*\nHigh error rate"
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "webhook_firing": {
+    "version": 1,
+    "title": "Incident P2: High error rate on web-logs",
+    "body_markdown": "Incident **new\\_incident\\_created** for service **checkout**.\n\nError rate above 5\\% for 5 minutes",
+    "fields": {
+      "Event": "new_incident_created",
+      "Severity": "P2",
+      "Service": "checkout",
+      "Alert": "High error rate"
+    },
+    "rows": [
+      {
+        "host": "web-1",
+        "errors": 9
+      },
+      {
+        "host": "web-2",
+        "errors": 3
+      }
+    ],
+    "links": [
+      {
+        "label": "",
+        "url": "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme"
+      }
+    ],
+    "severity": "critical",
+    "chart_url": null,
+    "context": {
+      "org_name": "acme",
+      "stream_type": "logs",
+      "stream_name": "web-logs",
+      "alert_name": "High error rate",
+      "alert_type": null,
+      "alert_period": null,
+      "alert_operator": null,
+      "alert_threshold": null,
+      "alert_count": null,
+      "alert_agg_value": null,
+      "alert_level": "critical",
+      "alert_status": "firing",
+      "episode_id": "inc_7f3a:1762077900000000",
+      "recovery_reason": null,
+      "alert_priority": null,
+      "alert_tags": null,
+      "alert_threshold_crit": null,
+      "alert_threshold_warn": null,
+      "alert_warning_threshold": null,
+      "alert_start_time": "N/A",
+      "alert_end_time": "N/A",
+      "alert_url": "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme",
+      "alert_trigger_time_str": "2026-11-02 10:05:00",
+      "alert_time": "2026-11-02 10:05:00",
+      "alert_description": "Error rate above 5% for 5 minutes",
+      "kind": "incident",
+      "notification_kind": "incident",
+      "notification_event": "opened",
+      "cluster_name": "o2-acme",
+      "alert_start_time_millis": "N/A",
+      "alert_end_time_millis": "N/A",
+      "alert_trigger_time": "0",
+      "alert_trigger_time_millis": "0",
+      "alert_trigger_time_seconds": "0",
+      "alert_promql_operator": null,
+      "alert_promql_value": null,
+      "host": [
+        "web-1",
+        "web-2"
+      ],
+      "errors": [
+        "9",
+        "3"
+      ],
+      "incident_id": "inc_7f3a",
+      "incident_title": "High error rate on web-logs",
+      "incident_event": "new_incident_created",
+      "incident_severity": "P2",
+      "incident_service": "checkout",
+      "incident_url": "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme"
+    }
+  }
+}
diff --git a/src/core/src/alerts/notifications/render/webhook.rs b/src/core/src/alerts/notifications/render/webhook.rs
index 00278c8004..3abbd0cd7e 100644
--- a/src/core/src/alerts/notifications/render/webhook.rs
+++ b/src/core/src/alerts/notifications/render/webhook.rs
@@ -106,6 +106,14 @@ pub fn context_object(ctx: &NotificationContext) -> Value {
         out.insert(name.to_string(), str_or_null(value));
     }
 
+    // Alerts keep their envelope byte identical; every other kind names itself.
+    if ctx.kind != config::meta::alerts::notification_kind::NotificationKind::Alert {
+        out.insert("kind".to_string(), json!(ctx.kind.as_str()));
+        for (name, value) in crate::alerts::notifications::resolve::extra_scalar_vars(ctx) {
+            out.insert(name.to_string(), str_or_null(&value));
+        }
+    }
+
     // Row columns: the multi-value form, as arrays (the scalar `", "`-joined
     // form is a template convenience; a machine consumer wants the list).
     for (key, values) in &ctx.row_columns {
diff --git a/src/core/src/alerts/notifications/resolve.rs b/src/core/src/alerts/notifications/resolve.rs
index 38a7085299..33f3d53c23 100644
--- a/src/core/src/alerts/notifications/resolve.rs
+++ b/src/core/src/alerts/notifications/resolve.rs
@@ -32,7 +32,9 @@
 //! structure the author wrote. Title, fields, links and rows are not
 //! markdown-parsed and stay raw.
 
-use config::meta::alerts::{content_spec::ContentSpec, level::AlertLevel};
+use config::meta::alerts::{
+    content_spec::ContentSpec, level::AlertLevel, notification_kind::NotificationKind,
+};
 
 use super::{context::NotificationContext, custom::scan_and_replace_var};
 
@@ -96,6 +98,10 @@ pub fn resolve_content(
         unknown_variables.append(&mut unk);
         let (value, mut unk) = substitute_raw(&field.value, ctx);
         unknown_variables.append(&mut unk);
+        // Alerts keep an empty field so their output stays byte identical; other kinds hide it.
+        if ctx.kind != NotificationKind::Alert && value.trim().is_empty() {
+            continue;
+        }
         fields.push((label, value));
     }
 
@@ -271,6 +277,10 @@ fn substitute_with(
         scan_and_replace_var(&mut out, name, |len| esc(&truncate(value, len)));
     }
 
+    for (name, value) in extra_scalar_vars(ctx) {
+        scan_and_replace_var(&mut out, name, |len| esc(&truncate(&value, len)));
+    }
+
     for (key, values) in ctx.row_columns.iter() {
         let joined = values.join(", ");
         scan_and_replace_var(&mut out, key, |len| esc(&truncate(&joined, len)));
@@ -301,7 +311,7 @@ fn substitute_with(
 /// from the pre-substitution input, which `{name}` tokens are genuinely
 /// unknown rather than merely a substring of the raw template.
 fn is_known_var(name: &str, ctx: &NotificationContext) -> bool {
-    if scalar_vars(ctx).iter().any(|(n, _)| *n == name) {
+    if scalar_vars(ctx).iter().any(|(n, _)| *n == name) || EXTRA_SCALAR_NAMES.contains(&name) {
         return true;
     }
     if ctx.row_columns.iter().any(|(k, _)| k == name) {
@@ -446,6 +456,48 @@ pub(crate) fn scalar_vars(ctx: &NotificationContext) -> Vec<(&'static str, &str)
     ]
 }
 
+/// Scalars added for the shared send path; kept apart so the alert webhook envelope is unchanged.
+pub(crate) fn extra_scalar_vars(ctx: &NotificationContext) -> Vec<(&'static str, String)> {
+    let trigger_millis = ctx.alert_trigger_time / 1000;
+    vec![
+        ("notification_kind", ctx.kind.as_str().to_string()),
+        ("notification_event", ctx.event.clone()),
+        ("cluster_name", ctx.cluster_name.clone()),
+        (
+            "alert_start_time_millis",
+            ctx.alert_start_time_millis.clone(),
+        ),
+        ("alert_end_time_millis", ctx.alert_end_time_millis.clone()),
+        ("alert_trigger_time", ctx.alert_trigger_time.to_string()),
+        ("alert_trigger_time_millis", trigger_millis.to_string()),
+        (
+            "alert_trigger_time_seconds",
+            (trigger_millis / 1000).to_string(),
+        ),
+        (
+            "alert_promql_operator",
+            ctx.promql_operator.clone().unwrap_or_default(),
+        ),
+        (
+            "alert_promql_value",
+            ctx.promql_value.clone().unwrap_or_default(),
+        ),
+    ]
+}
+
+pub(crate) const EXTRA_SCALAR_NAMES: [&str; 10] = [
+    "notification_kind",
+    "notification_event",
+    "cluster_name",
+    "alert_start_time_millis",
+    "alert_end_time_millis",
+    "alert_trigger_time",
+    "alert_trigger_time_millis",
+    "alert_trigger_time_seconds",
+    "alert_promql_operator",
+    "alert_promql_value",
+];
+
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;
@@ -509,6 +561,7 @@ mod tests {
             chart_png: None,
             alert_id: String::new(),
             incident_id: String::new(),
+            ..Default::default()
         }
     }
 
diff --git a/src/core/src/alerts/notifications/test_send.rs b/src/core/src/alerts/notifications/test_send.rs
index d930cba04d..a1822e53ae 100644
--- a/src/core/src/alerts/notifications/test_send.rs
+++ b/src/core/src/alerts/notifications/test_send.rs
@@ -35,7 +35,10 @@ use std::{
 
 use config::{
     get_config,
-    meta::{alerts::content_spec::ContentSpec, destinations::DestinationType},
+    meta::{
+        alerts::{content_spec::ContentSpec, notification_kind::NotificationKind},
+        destinations::DestinationType,
+    },
 };
 
 use super::{
@@ -153,7 +156,7 @@ pub fn build_test_message(
     dest_type: &DestinationType,
     spec: &ContentSpec,
 ) -> Result<RenderedMessage, TestSendError> {
-    Ok(build_test_message_with_title(dest_type, spec)?.1)
+    Ok(build_test_message_with_title(dest_type, spec, NotificationKind::Alert)?.1)
 }
 
 /// Same as [`build_test_message`], but also returns the marked title used to
@@ -163,13 +166,17 @@ pub fn build_test_message(
 fn build_test_message_with_title(
     dest_type: &DestinationType,
     spec: &ContentSpec,
+    kind: NotificationKind,
 ) -> Result<(String, RenderedMessage), TestSendError> {
     // Hold the draft to the same bar `templates::save` applies, so a
     // test-send cannot "succeed" on a template that cannot be saved.
     spec.validate().map_err(TestSendError::InvalidContent)?;
 
     let format = derive_channel_format(dest_type);
-    let ctx: NotificationContext = synthetic_context(None);
+    let ctx: NotificationContext = match kind {
+        NotificationKind::Alert => synthetic_context(None),
+        kind => super::examples::example_context(kind, false),
+    };
     let mut content = resolve_content(spec, &ctx, format.channel_family());
     content.title = format!("{TEST_MARKER}{}", content.title);
     let title = content.title.clone();
@@ -188,8 +195,9 @@ pub async fn test_send(
     org_id: &str,
     dest_type: &DestinationType,
     spec: &ContentSpec,
+    kind: NotificationKind,
 ) -> Result<String, TestSendError> {
-    let (title, rendered) = build_test_message_with_title(dest_type, spec)?;
+    let (title, rendered) = build_test_message_with_title(dest_type, spec, kind)?;
     dispatch_test_message(org_id, dest_type, &title, rendered)
         .await
         .map_err(|e| TestSendError::DispatchFailed(e.to_string()))
@@ -339,7 +347,8 @@ mod tests {
             sns_topic_arn: "arn:aws:sns:us-east-1:123456789012:topic".into(),
             aws_region: "us-east-1".into(),
         });
-        let (title, rendered) = build_test_message_with_title(&dest, &sample_spec()).unwrap();
+        let (title, rendered) =
+            build_test_message_with_title(&dest, &sample_spec(), NotificationKind::Alert).unwrap();
         assert!(title.starts_with(TEST_MARKER));
         assert_eq!(title, format!("{TEST_MARKER}Sample CPU alert fired"));
         match rendered {
diff --git a/src/core/src/organization.rs b/src/core/src/organization.rs
index 831b06acb8..f681f13bd2 100644
--- a/src/core/src/organization.rs
+++ b/src/core/src/organization.rs
@@ -705,6 +705,15 @@ pub async fn create_org(
                     org.identifier
                 );
             }
+            if let Err(e) =
+                crate::alerts::notifications::org_default::init_default_templates(&org.identifier)
+                    .await
+            {
+                log::warn!(
+                    "Failed to set default_templates for org '{}': {e}",
+                    org.identifier
+                );
+            }
             #[cfg(feature = "cloud")]
             enqueue_cloud_event(CloudEvent {
                 org_id: org.identifier.clone(),
diff --git a/src/db/src/system_settings.rs b/src/db/src/system_settings.rs
index b7415d7cce..dc10007124 100644
--- a/src/db/src/system_settings.rs
+++ b/src/db/src/system_settings.rs
@@ -212,8 +212,7 @@ pub async fn set(setting: &SystemSetting) -> Result<SystemSetting> {
 /// For a setting whose absence has a defined meaning: caching that meaning stops every subsequent
 /// [`get`] falling through to a database read that can only return nothing. A real row, written
 /// here or on any other node, replaces it through [`set`] or [`watch`].
-#[cfg(feature = "enterprise")]
-pub(crate) async fn set_only_cached(setting: &SystemSetting) {
+pub async fn set_only_cached(setting: &SystemSetting) {
     let cache_k = cache_key(
         &setting.scope,
         setting.org_id.as_deref(),