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(),