feat(synthetics): synthetics and downtimes on the shared notification path, stage 3

Synthetics checks get alerting_since, part of the alert state swap, which keys the episode {check_id}:{alerting_since}. A run that closes an episode hands its swap to a closer the server registers, so the state flip and the queued resolve rows commit together; a muted recovery closes with no rows. Firings, degraded, flaky and certificate runs send through notify() with the compiled synthetics card, today's subject line as its title, the check's template override, and today's {"text"} body on generic webhooks. Private locations send down and up on the same path. Downtimes render the seeded pair or a user template through notify(), platforms are skipped, and the D3 sniff, the private renderer and dispatch_notification are removed. The deliveries read API serves synthetics and downtime rows.
2026-11-05 00:09:35 +00:00 · 2026-11-03 10:22:36 +05:30
parent 7433927064
commit f866bb330d
30 changed files with 1846 additions and 494 deletions
@@ -173,9 +173,15 @@ fn validate(query: &DeliveriesQuery) -> Result<(NotificationKind, String), Strin
        )
    })?;
    // Only the kinds whose producers write rows today; the others land with their stages.
-    if !matches!(kind, NotificationKind::Anomaly | NotificationKind::Incident) {
+    if !matches!(
+        kind,
+        NotificationKind::Anomaly
+            | NotificationKind::Incident
+            | NotificationKind::Synthetics
+            | NotificationKind::Downtime
+    ) {
        return Err(format!(
-            "kind: '{}' has no delivery rows yet; allowed: anomaly, incident",
+            "kind: '{}' has no delivery rows yet; allowed: anomaly, incident, synthetics, downtime",
            query.kind
        ));
    }
@@ -200,6 +206,8 @@ fn owner_of_episode(episode_id: &str) -> String {
 async fn can_read(org_id: &str, user_id: &str, kind: NotificationKind, owner: &str) -> bool {
    let object_type = match kind {
        NotificationKind::Incident => "incidents",
+        NotificationKind::Synthetics => "synthetics",
+        NotificationKind::Downtime => "downtimes",
        _ => "alerts",
    };
    openobserve_core::auth::check_permissions(
@@ -250,9 +258,15 @@ mod tests {

    #[test]
    fn kinds_without_rows_yet_are_refused() {
-        for kind in ["alert", "synthetics", "downtime", "on_call"] {
+        for kind in ["alert", "on_call"] {
            let err = validate(&q(kind, Some("x1"), None)).unwrap_err();
-            assert!(err.contains("allowed: anomaly, incident"), "{kind}: {err}");
+            assert!(
+                err.contains("allowed: anomaly, incident, synthetics, downtime"),
+                "{kind}: {err}"
+            );
+        }
+        for kind in ["synthetics", "downtime"] {
+            assert!(validate(&q(kind, Some("x1"), None)).is_ok(), "{kind}");
        }
    }

@@ -713,6 +713,28 @@ async fn composition_error_response(
    })
 }

+/// A template override must name a user template that applies to synthetics; "" clears it.
+async fn checked_template(
+    org_id: &str,
+    body: &mut config::meta::synthetics::Synthetic,
+) -> Result<(), Response> {
+    body.template = body
+        .template
+        .take()
+        .map(|t| t.trim().to_string())
+        .filter(|t| !t.is_empty());
+    if let Some(name) = body.template.as_deref() {
+        openobserve_core::alerts::templates::check_binding(
+            org_id,
+            name,
+            config::meta::alerts::notification_kind::NotificationKind::Synthetics,
+        )
+        .await
+        .map_err(|e| MetaHttpResponse::bad_request(e.as_str()))?;
+    }
+    Ok(())
+}
+
 #[utoipa::path(
    post,
    path = "/{org_id}/synthetics",
@@ -775,6 +797,9 @@ pub async fn create_synthetic(
        }
    }

+    if let Err(response) = checked_template(&org_id, &mut body).await {
+        return response;
+    }
    let created_by = user_email.user_id.as_str();
    match openobserve_synthetics::service::create_synthetic(&org_id, body, created_by).await {
        Ok(check) => MetaHttpResponse::json(check),
@@ -945,6 +970,10 @@ pub async fn update_synthetic(
    #[cfg(not(feature = "enterprise"))]
    let user_id = "";

+    let mut body = body;
+    if let Err(response) = checked_template(&org_id, &mut body).await {
+        return response;
+    }
    match openobserve_synthetics::service::update_synthetic(&org_id, &id, body).await {
        Ok(check) => MetaHttpResponse::json(check),
        Err(e) => {
@@ -1718,7 +1747,7 @@ async fn process_ack(
    let should_notify =
        !matches!(resp.alert, AlertDecision::Silent) && resp.suppressed_by.is_none();
    // Enterprise-gated because alert *destinations* are: the dispatch it ends in
-    // (`alerts::alert::dispatch_notification`) is not built in OSS. An OSS build
+    // (the shared notification path) is not built in OSS. An OSS build
    // runs the check, records the run and serves the result — it just has
    // nowhere to send a notification, exactly as it has for alerts.
    #[cfg(feature = "enterprise")]
@@ -1744,6 +1773,8 @@ async fn process_ack(
            failing_locations: resp.failing_locations.clone(),
            failing_environments: resp.failing_environments.clone(),
            passing_locations: resp.passing_locations.clone(),
+            template: resp.template.clone(),
+            alerting_since: resp.alerting_since,
        };
        tokio::spawn(async move {
            openobserve_core::synthetics::notify_check_result(notification).await;
@@ -2448,6 +2479,8 @@ mod tests {
            failing_environments: Vec::new(),
            usage_events,
            suppressed_by: None,
+            alerting_since: 0,
+            template: None,
        }
    }

@@ -136,7 +136,8 @@ pub type RwBTreeMap<K, V> = tokio::sync::RwLock<BTreeMap<K, V>>;
 // 136: add applies_to to templates.
 // 137: rename user templates under the reserved o2_default_ prefix.
 // 138: add template to anomaly_detection_config.
-pub const DB_SCHEMA_VERSION: u64 = 138;
+// 139: add template and alerting_since to synthetics.
+pub const DB_SCHEMA_VERSION: u64 = 139;
 pub const DB_SCHEMA_KEY: &str = "/db_schema_version/";

 // global version variables
@@ -140,6 +140,9 @@ pub struct Synthetic {
    /// Alert destination names to notify on check failure.
    #[serde(default)]
    pub destinations: Vec<String>,
+    /// Template override for this check's messages; null or "" uses the destination's.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub template: Option<String>,
    /// Number of retries before marking a check failed (0 = no retry).
    #[serde(default)]
    pub retries: i32,
@@ -3418,42 +3418,6 @@ async fn send_to_destination_type(
    }
 }

-/// Send a pre-built message string to a single destination type.
-///
-/// Used by synthetics and downtime notifications, which build their own
-/// payload rather than going through the alert template system.
-#[cfg(feature = "enterprise")]
-pub(crate) async fn dispatch_notification(
-    org_id: &str,
-    dest_type: &DestinationType,
-    subject: &str,
-    msg: String,
-) -> Result<String, anyhow::Error> {
-    match dest_type {
-        DestinationType::Http(endpoint) => send_http_notification(endpoint, msg).await,
-        // Callers build one payload with no HTML/plaintext split, so both
-        // parts carry the same string — byte-identical to the MIME this path
-        // emitted before the signature took two bodies.
-        DestinationType::Email(email) => {
-            let (email, skipped) = org_member_email(org_id, email).await?;
-            send_email_notification(subject, &email, Some(msg.clone()), msg)
-                .await
-                .map(|resp| note_skipped(resp, &skipped))
-        }
-        DestinationType::Sns(aws_sns) => send_sns_notification(subject, aws_sns, msg).await,
-        DestinationType::SlackApp(target) => {
-            crate::alerts::slack_publish::send_post(
-                &crate::alerts::slack_publish::QueueSink,
-                org_id,
-                target,
-                subject,
-                &msg,
-            )
-            .await
-        }
-    }
-}
-
 /// Dispatch a rendered test-send message to one destination.
 ///
 /// `title` is the already-`[TEST] `-marked title `build_test_message` stamped
@@ -3464,14 +3428,6 @@ pub(crate) async fn dispatch_notification(
 /// of re-deriving it from `rendered`, or hardcoding a literal) keeps this
 /// function agnostic to the rendered payload's shape and guarantees the SNS
 /// subject is never anything other than the actual marked title.
-///
-/// Unlike [`dispatch_notification`], this is NOT enterprise-gated: test-send
-/// (Task 15) is an OSS-visible feature, and it needs the same private
-/// `send_http_notification` / `send_email_notification` / `send_sns_notification`
-/// transports this module already owns. Kept as a distinct, narrowly-scoped
-/// function rather than removing the `#[cfg]` from `dispatch_notification`
-/// itself, so the enterprise-only synthetics and downtime call sites are
-/// untouched.
 pub(crate) async fn dispatch_test_message(
    org_id: &str,
    dest_type: &DestinationType,
@@ -107,6 +107,16 @@ pub struct NotificationContext {
    pub cluster_name: String,
    /// Today's body for a generic webhook while no user template applies (D1).
    pub legacy_payload: Option<LegacyPayload>,
+    pub render_hints: RenderHints,
+}
+
+/// Producer choices the renderer honors; the defaults keep the alert output unchanged.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct RenderHints {
+    /// The email or SNS subject when a Custom template has no title of its own.
+    pub fallback_subject: Option<String>,
+    /// HTML-escape context attributes in email bodies, as the producer's own sender did.
+    pub escape_email_attributes: bool,
 }

 /// Builds a producer's pre-template JSON body only when the legacy rule selects it.
@@ -232,7 +232,12 @@ pub(crate) fn apply_custom_template_plain(
        }
    }
    for (key, value) in ctx.context_attributes.iter() {
-        process_variable_replace(&mut resp, key, &VarValue::Str(value), is_email);
+        if is_email && ctx.render_hints.escape_email_attributes {
+            let escaped = html_escape(value);
+            process_variable_replace(&mut resp, key, &VarValue::Str(&escaped), is_email);
+        } else {
+            process_variable_replace(&mut resp, key, &VarValue::Str(value), is_email);
+        }
    }

    // Substitute endpoint metadata variables (e.g., credential_assignmentGroup,
@@ -494,6 +499,15 @@ impl VarValue<'_> {
    }
 }

+fn html_escape(value: &str) -> String {
+    value
+        .replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+        .replace('"', "&quot;")
+        .replace('\'', "&#39;")
+}
+
 #[cfg(test)]
 mod golden {
    //! Byte-identity corpus for the custom-kind renderer.
@@ -81,6 +81,7 @@ pub fn compiled_content_for(kind: NotificationKind) -> ContentSpec {
    match kind {
        NotificationKind::Incident => compiled_incident_content(),
        NotificationKind::Anomaly => compiled_anomaly_content(),
+        NotificationKind::Synthetics => compiled_synthetics_content(),
        _ => compiled_default_content(),
    }
 }
@@ -88,7 +89,9 @@ pub fn compiled_content_for(kind: NotificationKind) -> ContentSpec {
 /// 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(),
+        NotificationKind::Incident | NotificationKind::Anomaly | NotificationKind::Synthetics => {
+            kind.compiled_default_name()
+        }
        _ => DEFAULT_CONTENT_TEMPLATE_NAME,
    }
 }
@@ -115,6 +118,29 @@ pub fn compiled_anomaly_content() -> ContentSpec {
    }
 }

+/// The title is today's subject line, so a mail or SNS subject reads exactly as before.
+pub fn compiled_synthetics_content() -> ContentSpec {
+    ContentSpec {
+        title: "{check_title}".into(),
+        title_overrides: Default::default(),
+        body: "{alert_description}".into(),
+        fields: vec![
+            field("Check", "{check_name} ({check_type})"),
+            field("Target", "{check_target}"),
+            field("Locations", "{check_locations_line}"),
+            field("Error", "{check_error}"),
+        ],
+        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(),
@@ -165,6 +191,7 @@ pub(crate) async fn ensure_default_content_template(
        NotificationKind::Alert,
        NotificationKind::Incident,
        NotificationKind::Anomaly,
+        NotificationKind::Synthetics,
    ] {
        ensure_compiled_row(
            compiled_name_for(kind),
@@ -915,3 +915,142 @@ fn an_email_with_an_inline_chart_replays_and_filters_recipients_again() {
        },
    );
 }
+
+#[cfg(feature = "enterprise")]
+async fn kind_rows(
+    kind: &str,
+    object: &str,
+) -> Vec<infra::table::entity::notification_deliveries::Model> {
+    let filter = ListFilter {
+        org: "acme".to_string(),
+        kind: kind.into(),
+        object_id: Some(object.to_string()),
+        size: 50,
+        ..Default::default()
+    };
+    ledger::list(infra::db::get_orm_client_rw().await, &filter)
+        .await
+        .unwrap()
+        .1
+}
+
+#[cfg(feature = "enterprise")]
+fn synthetics_close(
+    check: &str,
+    since: i64,
+    send_resolves: bool,
+    won: bool,
+) -> openobserve_synthetics::alerting::EpisodeClose {
+    openobserve_synthetics::alerting::EpisodeClose {
+        org_id: "acme".into(),
+        check_id: check.to_string(),
+        check_name: "checkout".into(),
+        check_type: "http".into(),
+        target: "https://shop.acme.com".into(),
+        episode_id: format!("{check}:{since}"),
+        run_id: "r2".into(),
+        destinations: vec!["pd-syn".into(), "slack-syn".into()],
+        template: None,
+        send_resolves,
+        swap: Box::new(move |_| Box::pin(async move { Ok(won) })),
+    }
+}
+
+#[cfg(feature = "enterprise")]
+#[test]
+fn a_synthetics_recovery_resolves_the_firing_record_and_a_muted_one_writes_no_row() {
+    run(
+        concat!(
+            module_path!(),
+            "::a_synthetics_recovery_resolves_the_firing_record_and_a_muted_one_writes_no_row"
+        ),
+        &[],
+        || async {
+            store().await;
+            let (base, seen) = mock().await;
+            register("pd-syn", http(format!("{base}/pd"), "pagerduty"));
+            register("slack-syn", http(format!("{base}/ok"), "slack"));
+            let check = svix_ksuid::Ksuid::new(None, None).to_string();
+            crate::synthetics::notify_check_result(crate::synthetics::CheckNotification {
+                org_id: "acme".into(),
+                check_name: "checkout".into(),
+                check_id: check.clone(),
+                check_type: "http".into(),
+                target: "https://shop.acme.com".into(),
+                destinations: vec!["pd-syn".into(), "slack-syn".into()],
+                run_id: "r1".into(),
+                status: "failed".into(),
+                job_count: 1,
+                error: Some("timeout".into()),
+                checked_at: 1,
+                recovery: false,
+                consecutive_failures: 1,
+                flaky: false,
+                status_reason: None,
+                error_source: String::new(),
+                degraded: false,
+                failing_locations: vec!["us-east-1".into()],
+                passing_locations: vec![],
+                failing_environments: vec![],
+                template: None,
+                alerting_since: 100,
+            })
+            .await;
+            let triggers = calls(&seen, "/pd");
+            assert_eq!(triggers.len(), 1, "the firing reached PagerDuty once");
+
+            // A lost swap closes nothing and writes no row.
+            assert!(
+                !crate::synthetics::close_alert_episode(synthetics_close(&check, 100, true, false))
+                    .await
+                    .unwrap()
+            );
+            assert!(
+                kind_rows("synthetics", &check)
+                    .await
+                    .iter()
+                    .all(|r| r.event != "resolved")
+            );
+
+            assert!(
+                crate::synthetics::close_alert_episode(synthetics_close(&check, 100, true, true))
+                    .await
+                    .unwrap()
+            );
+            let pd = calls(&seen, "/pd");
+            assert_eq!(pd.len(), 2, "one trigger, then one resolve");
+            let resolve = String::from_utf8_lossy(&pd[1].1).to_string();
+            let trigger = String::from_utf8_lossy(&pd[0].1).to_string();
+            assert!(resolve.contains("resolve"), "{resolve}");
+            assert!(
+                trigger.contains(&format!("{check}:100"))
+                    && resolve.contains(&format!("{check}:100"))
+            );
+            let resolved: Vec<_> = kind_rows("synthetics", &check)
+                .await
+                .into_iter()
+                .filter(|r| r.event == "resolved")
+                .collect();
+            assert_eq!(resolved.len(), 2);
+            assert!(resolved.iter().all(|r| r.status == "sent"), "{resolved:?}");
+
+            // A muted recovery closes its episode without a resolve row.
+            assert!(
+                crate::synthetics::close_alert_episode(synthetics_close(&check, 200, false, true))
+                    .await
+                    .unwrap()
+            );
+            assert!(
+                kind_rows("synthetics", &check)
+                    .await
+                    .iter()
+                    .all(|r| r.episode_id != format!("{check}:200"))
+            );
+            assert_eq!(
+                calls(&seen, "/pd").len(),
+                2,
+                "nothing more reached PagerDuty"
+            );
+        },
+    );
+}
@@ -47,6 +47,36 @@ pub const INCIDENT_VARIABLES: [&str; 6] = [
    "incident_url",
 ];

+pub const SYNTHETICS_VARIABLES: [&str; 16] = [
+    "check_id",
+    "check_name",
+    "check_type",
+    "check_target",
+    "check_status",
+    "check_status_label",
+    "check_status_reason",
+    "check_locations_line",
+    "check_location_count",
+    "check_consecutive_failures",
+    "check_flaky",
+    "check_degraded",
+    "check_run_id",
+    "check_url",
+    "check_error",
+    "check_title",
+];
+
+pub const DOWNTIME_VARIABLES: [&str; 8] = [
+    "downtime_name",
+    "downtime_event",
+    "downtime_reason",
+    "downtime_starts_at",
+    "downtime_ends_at",
+    "downtime_targets",
+    "downtime_counts",
+    "downtime_url",
+];
+
 const COMMON_VARIABLES: [&str; 27] = [
    "org_name",
    "stream_type",
@@ -82,6 +112,8 @@ pub fn variables_for(kind: NotificationKind) -> Vec<String> {
    let kind_vars: &[&str] = match kind {
        NotificationKind::Anomaly => &ANOMALY_VARIABLES,
        NotificationKind::Incident => &INCIDENT_VARIABLES,
+        NotificationKind::Synthetics => &SYNTHETICS_VARIABLES,
+        NotificationKind::Downtime => &DOWNTIME_VARIABLES,
        _ => &[],
    };
    COMMON_VARIABLES
@@ -100,10 +132,12 @@ pub struct VariableGroup {
 }

 /// The kinds whose producer sends through templates in this release; the UI hides the others.
-pub const SHIPPED_KINDS: [NotificationKind; 3] = [
+pub const SHIPPED_KINDS: [NotificationKind; 5] = [
    NotificationKind::Alert,
    NotificationKind::Incident,
    NotificationKind::Anomaly,
+    NotificationKind::Synthetics,
+    NotificationKind::Downtime,
 ];

 /// `common`, then one group per shipped kind with its own variables.
@@ -122,6 +156,8 @@ pub fn variable_groups() -> Vec<VariableGroup> {
        let vars: &[&str] = match kind {
            NotificationKind::Anomaly => &ANOMALY_VARIABLES,
            NotificationKind::Incident => &INCIDENT_VARIABLES,
+            NotificationKind::Synthetics => &SYNTHETICS_VARIABLES,
+            NotificationKind::Downtime => &DOWNTIME_VARIABLES,
            _ => &[],
        };
        groups.push(group(kind.as_str(), vars));
@@ -134,6 +170,7 @@ pub fn example_context(kind: NotificationKind, resolved: bool) -> NotificationCo
    match kind {
        NotificationKind::Anomaly => anomaly_example(resolved),
        NotificationKind::Incident => incident_example(resolved),
+        NotificationKind::Synthetics => synthetics_example(resolved),
        _ => {
            let level = if resolved {
                AlertLevel::Ok
@@ -233,6 +270,69 @@ fn anomaly_example(resolved: bool) -> NotificationContext {
    ctx
 }

+fn synthetics_example(resolved: bool) -> NotificationContext {
+    let url = "https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme";
+    let level = if resolved {
+        AlertLevel::Ok
+    } else {
+        AlertLevel::Critical
+    };
+    let mut ctx = base(NotificationKind::Synthetics, resolved, level);
+    ctx.stream_type = "synthetics".into();
+    ctx.stream_name = "https://shop.acme.com/api/health".into();
+    ctx.alert_name = "checkout API".into();
+    ctx.alert_url = url.into();
+    ctx.episode_id = Some("chk_9d:1762077900000000".into());
+    ctx.event = if resolved {
+        STATUS_RESOLVED.into()
+    } else {
+        "firing:run_42".into()
+    };
+    ctx.alert_trigger_time_str = "2026-11-02 10:05:00".into();
+    let (title, description, label, locations, error) = if resolved {
+        (
+            "[OpenObserve Synthetics] ✅ checkout API has RECOVERED",
+            "checkout API has recovered",
+            "recovered",
+            "us-east-1, eu-west-1 recovered",
+            "",
+        )
+    } else {
+        (
+            "[OpenObserve Synthetics] 🔴 checkout API is FAILED",
+            "checkout API is failing\nconnection timed out after 10s",
+            "failed",
+            "1 of 2 failed: us-east-1",
+            "connection timed out after 10s",
+        )
+    };
+    ctx.alert_description = description.into();
+    ctx.context_attributes = [
+        ("check_id", "chk_9d"),
+        ("check_name", "checkout API"),
+        ("check_type", "http"),
+        ("check_target", "https://shop.acme.com/api/health"),
+        ("check_status", if resolved { "passed" } else { "failed" }),
+        ("check_status_label", label),
+        ("check_status_reason", ""),
+        ("check_locations_line", locations),
+        ("check_location_count", "2"),
+        (
+            "check_consecutive_failures",
+            if resolved { "0" } else { "3" },
+        ),
+        ("check_flaky", "false"),
+        ("check_degraded", "false"),
+        ("check_run_id", "run_42"),
+        ("check_url", url),
+        ("check_error", error),
+        ("check_title", title),
+    ]
+    .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 {
@@ -304,6 +404,7 @@ mod tests {

    const EXAMPLE_ANOMALY: &str = include_str!("render/testdata/examples/anomaly.json");
    const EXAMPLE_INCIDENT: &str = include_str!("render/testdata/examples/incident.json");
+    const EXAMPLE_SYNTHETICS: &str = include_str!("render/testdata/examples/synthetics.json");

    fn rendered(
        kind: NotificationKind,
@@ -349,6 +450,19 @@ mod tests {
        );
    }

+    #[test]
+    fn golden_example_synthetics_keeps_the_subject_line_as_its_title() {
+        let want: serde_json::Value = serde_json::from_str(EXAMPLE_SYNTHETICS).unwrap();
+        assert_eq!(example(NotificationKind::Synthetics), want);
+        let title = &want["slack_firing"]["attachments"][0]["blocks"][0]["text"]["text"];
+        assert!(
+            title
+                .as_str()
+                .unwrap()
+                .contains("[OpenObserve Synthetics] 🔴 checkout API is FAILED")
+        );
+    }
+
    #[test]
    fn anomaly_recovery_hides_empty_fields() {
        let ctx = example_context(NotificationKind::Anomaly, true);
@@ -382,6 +496,7 @@ mod tests {
        for (kind, file) in [
            (NotificationKind::Anomaly, "anomaly.json"),
            (NotificationKind::Incident, "incident.json"),
+            (NotificationKind::Synthetics, "synthetics.json"),
        ] {
            let body = serde_json::to_string_pretty(&example(kind)).unwrap() + "\n";
            std::fs::write(format!("{dir}/{file}"), body).unwrap();
@@ -92,6 +92,20 @@ pub async fn set_org_default_template(
    Ok(())
 }

+/// A producer's level 1 override by name; a missing one falls through to the next level.
+pub async fn load_override(org_id: &str, name: Option<&str>) -> Vec<Template> {
+    let Some(name) = name.filter(|n| !n.is_empty()) else {
+        return Vec::new();
+    };
+    match db::alerts::templates::get(org_id, name).await {
+        Ok(t) => vec![t],
+        Err(e) => {
+            log::warn!("[TEMPLATES] template override '{name}' in {org_id} not found: {e}");
+            Vec::new()
+        }
+    }
+}
+
 /// A `default_templates` value: an object of kind to user template name that applies to it.
 pub async fn validate_default_templates(
    org_id: &str,
@@ -322,7 +336,7 @@ pub fn choose_org_or_compiled(
        };
    }
    EffectiveTemplate {
-        template: compiled_fallback_for(org_id, kind),
+        template: compiled_fallback_for_transport(org_id, kind, dt),
        source: TemplateSource::Compiled,
        skipped,
    }
@@ -428,6 +442,25 @@ pub fn compiled_fallback_for(org_id: &str, kind: NotificationKind) -> Template {
    }
 }

+/// Level 4 for one transport: downtimes keep the seeded pair (D3), other kinds their card.
+pub fn compiled_fallback_for_transport(
+    org_id: &str,
+    kind: NotificationKind,
+    dt: Option<&DestinationType>,
+) -> Template {
+    if kind != NotificationKind::Downtime {
+        return compiled_fallback_for(org_id, kind);
+    }
+    let [http, email] = crate::alerts::templates::downtime_templates();
+    let mut template = if matches!(dt, Some(DestinationType::Email(_))) {
+        email
+    } else {
+        http
+    };
+    template.org_id = org_id.to_string();
+    template
+}
+
 /// Idempotent: set the pointer (to the compiled default) for every org that
 /// doesn't already have one. Runs at startup, after
 /// `ensure_default_content_template`, under its own dist_lock — safe to run
@@ -591,6 +624,36 @@ mod tests {
        )
    }

+    #[test]
+    fn a_downtime_falls_back_to_the_seeded_pair_for_its_transport() {
+        let http = compiled_fallback_for_transport(
+            "acme",
+            NotificationKind::Downtime,
+            Some(&slack_dest()),
+        );
+        assert_eq!(http.name, "prebuilt_downtime");
+        assert_eq!(http.org_id, "acme");
+        let email = compiled_fallback_for_transport(
+            "acme",
+            NotificationKind::Downtime,
+            Some(&email_dest()),
+        );
+        assert_eq!(email.name, "prebuilt_downtime_email");
+        assert_eq!(
+            compiled_fallback_for_transport("acme", NotificationKind::Synthetics, None).name,
+            "o2_default_synthetics"
+        );
+        // D3: a user template applies to downtimes without naming a downtime variable.
+        assert!(destination_template_applies(
+            &content("ops-card"),
+            NotificationKind::Downtime
+        ));
+        assert!(!destination_template_applies(
+            &content("prebuilt_slack"),
+            NotificationKind::Downtime
+        ));
+    }
+
    #[test]
    fn resolution_table() {
        use NotificationKind::*;
@@ -432,7 +432,17 @@ mod tests {
            .iter()
            .map(|g| g.name.as_str())
            .collect();
-        assert_eq!(groups, ["common", "alert", "incident", "anomaly"]);
+        assert_eq!(
+            groups,
+            [
+                "common",
+                "alert",
+                "incident",
+                "anomaly",
+                "synthetics",
+                "downtime"
+            ]
+        );
        let incident = &resp.variable_groups[2];
        assert!(incident.variables.iter().any(|v| v == "incident_severity"));
        assert!(
@@ -251,9 +251,12 @@ pub fn render_record(
        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 fallback = ctx.render_hints.fallback_subject.as_ref();
            let title = match &template.template_type {
-                TemplateType::Email { title } => apply_custom_template(title, ctx, is_email),
-                _ => template.name.clone(),
+                TemplateType::Email { title } if !(title.is_empty() && fallback.is_some()) => {
+                    apply_custom_template(title, ctx, is_email)
+                }
+                _ => fallback.cloned().unwrap_or_else(|| template.name.clone()),
            };
            let subject = subject.map(str::to_string).unwrap_or(title);
            Ok(match dest_type {
@@ -0,0 +1,198 @@
+{
+  "slack_firing": {
+    "attachments": [
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme|[OpenObserve Synthetics] 🔴 checkout API is FAILED>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "checkout API is failing\nconnection timed out after 10s"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Check*\ncheckout API (http)"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Target*\nhttps://shop.acme.com/api/health"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Locations*\n1 of 2 failed: us-east-1"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Error*\nconnection timed out after 10s"
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "color": "#E01E5A",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "slack_resolved": {
+    "attachments": [
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "*<https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme|[OpenObserve Synthetics] ✅ checkout API has RECOVERED>*"
+            }
+          },
+          {
+            "type": "section",
+            "text": {
+              "type": "mrkdwn",
+              "text": "checkout API has recovered"
+            }
+          },
+          {
+            "type": "section",
+            "fields": [
+              {
+                "type": "mrkdwn",
+                "text": "*Check*\ncheckout API (http)"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Target*\nhttps://shop.acme.com/api/health"
+              },
+              {
+                "type": "mrkdwn",
+                "text": "*Locations*\nus-east-1, eu-west-1 recovered"
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "color": "#2EB67D",
+        "blocks": [
+          {
+            "type": "actions",
+            "elements": [
+              {
+                "type": "button",
+                "text": {
+                  "type": "plain_text",
+                  "text": "View in OpenObserve"
+                },
+                "url": "https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme",
+                "style": "primary"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "webhook_firing": {
+    "version": 1,
+    "title": "[OpenObserve Synthetics] 🔴 checkout API is FAILED",
+    "body_markdown": "checkout API is failing\nconnection timed out after 10s",
+    "fields": {
+      "Check": "checkout API (http)",
+      "Target": "https://shop.acme.com/api/health",
+      "Locations": "1 of 2 failed: us-east-1",
+      "Error": "connection timed out after 10s"
+    },
+    "rows": [],
+    "links": [
+      {
+        "label": "",
+        "url": "https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme"
+      }
+    ],
+    "severity": "critical",
+    "chart_url": null,
+    "context": {
+      "org_name": "acme",
+      "stream_type": "synthetics",
+      "stream_name": "https://shop.acme.com/api/health",
+      "alert_name": "checkout API",
+      "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": "chk_9d: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/synthetic/chk_9d/results?org_identifier=acme",
+      "alert_trigger_time_str": "2026-11-02 10:05:00",
+      "alert_time": "2026-11-02 10:05:00",
+      "alert_description": "checkout API is failing\nconnection timed out after 10s",
+      "kind": "synthetics",
+      "notification_kind": "synthetics",
+      "notification_event": "firing:run_42",
+      "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,
+      "check_id": "chk_9d",
+      "check_name": "checkout API",
+      "check_type": "http",
+      "check_target": "https://shop.acme.com/api/health",
+      "check_status": "failed",
+      "check_status_label": "failed",
+      "check_status_reason": null,
+      "check_locations_line": "1 of 2 failed: us-east-1",
+      "check_location_count": "2",
+      "check_consecutive_failures": "3",
+      "check_flaky": "false",
+      "check_degraded": "false",
+      "check_run_id": "run_42",
+      "check_url": "https://o2.acme.com/web/synthetic/chk_9d/results?org_identifier=acme",
+      "check_error": "connection timed out after 10s",
+      "check_title": "[OpenObserve Synthetics] 🔴 checkout API is FAILED"
+    }
+  }
+}
@@ -279,6 +279,18 @@ pub(crate) async fn bindings_for(
                by.push(format!("anomaly config '{config}'"));
            }
        }
+        NotificationKind::Synthetics => {
+            let db_conn = infra::db::get_orm_client_ro().await;
+            for (_, check) in infra::table::synthetics_checks::names_using_template(
+                db_conn,
+                binding_scope(org_id),
+                name,
+            )
+            .await?
+            {
+                by.push(format!("synthetics check '{check}'"));
+            }
+        }
        NotificationKind::Alert => {
            let (_, alerts) = db::alerts::templates::cached_references(org_id, name).await;
            by.extend(alerts.into_iter().map(|a| format!("alert '{a}'")));
@@ -349,13 +361,13 @@ pub struct UsedBy {
    /// Kinds whose org default pointer names the template.
    pub org_default: Vec<String>,
    #[serde(skip)]
-    anomaly_names: Vec<String>,
+    binding_names: Vec<String>,
 }

 impl UsedBy {
    /// The references the cached delete guard does not see, as one message.
    fn bindings_outside_caches(&self) -> Option<String> {
-        let mut by = self.anomaly_names.clone();
+        let mut by = self.binding_names.clone();
        if !self.org_default.is_empty() {
            by.push(format!("org default for {}", self.org_default.join(", ")));
        }
@@ -374,6 +386,15 @@ pub async fn used_by(org_id: &str, name: &str) -> Result<UsedBy, TemplateError>
    .into_iter()
    .map(|(_, config)| format!("anomaly config '{config}'"))
    .collect();
+    let synthetics_names: Vec<String> = infra::table::synthetics_checks::names_using_template(
+        infra::db::get_orm_client_ro().await,
+        binding_scope(org_id),
+        name,
+    )
+    .await?
+    .into_iter()
+    .map(|(_, check)| format!("synthetics check '{check}'"))
+    .collect();
    let mut org_default: Vec<String> = Vec::new();
    for (_, kinds) in pointer_orgs(org_id, name).await? {
        for kind in kinds {
@@ -387,10 +408,10 @@ pub async fn used_by(org_id: &str, name: &str) -> Result<UsedBy, TemplateError>
        destinations: destinations.len(),
        alerts: alerts.len(),
        anomaly: anomaly_names.len(),
-        synthetics: 0,
+        synthetics: synthetics_names.len(),
        on_call: 0,
        org_default,
-        anomaly_names,
+        binding_names: anomaly_names.into_iter().chain(synthetics_names).collect(),
    })
 }

@@ -3529,7 +3529,9 @@ pub async fn send_anomaly_alert(
    actx: o2_enterprise::enterprise::anomaly_detection::types::AnomalyAlertContext,
 ) -> anyhow::Result<bool> {
    let org_id = actx.org_id.clone();
-    let explicit = load_override(&org_id, actx.template.as_deref()).await;
+    let explicit =
+        crate::alerts::notifications::org_default::load_override(&org_id, actx.template.as_deref())
+            .await;
    let mut ctx = anomaly_notification_context(&actx);
    let sent = crate::alerts::notifications::notify::notify(
        &org_id,
@@ -3597,7 +3599,8 @@ pub async fn resolve_anomaly_episode(
    let ctx = anomaly_notification_context(&actx);
    let episode_id = ctx.episode_id.clone().unwrap_or_default();
    let explicit = if recovery.send_resolves {
-        load_override(&org_id, actx.template.as_deref()).await
+        crate::alerts::notifications::org_default::load_override(&org_id, actx.template.as_deref())
+            .await
    } else {
        Vec::new()
    };
@@ -3664,24 +3667,6 @@ pub async fn resolve_anomaly_episode(
    })
 }

-/// The config's template override, as level 1 candidates; a missing template falls through.
-#[cfg(feature = "enterprise")]
-async fn load_override(
-    org_id: &str,
-    name: Option<&str>,
-) -> Vec<config::meta::destinations::Template> {
-    let Some(name) = name.filter(|n| !n.is_empty()) else {
-        return Vec::new();
-    };
-    match db::alerts::templates::get(org_id, name).await {
-        Ok(t) => vec![t],
-        Err(e) => {
-            log::warn!("[anomaly_detection] template override '{name}' not found: {e}");
-            Vec::new()
-        }
-    }
-}
-
 /// The shared path's context for one anomaly message (Backend section 6 field mapping).
 #[cfg(feature = "enterprise")]
 fn anomaly_notification_context(
@@ -3797,7 +3782,7 @@ fn anomaly_attributes(

 /// The alert convention for the standard time fields: server local time with no suffix.
 #[cfg(feature = "enterprise")]
-fn fmt_local_us(us: i64) -> String {
+pub(crate) fn fmt_local_us(us: i64) -> String {
    use chrono::TimeZone;
    chrono::Local
        .timestamp_micros(us)
@@ -15,11 +15,8 @@

 use config::{
    get_config,
-    meta::{
-        destinations::{DestinationType, Module, Template, TemplateType},
-        downtimes::{
-            Downtime, DowntimeNotificationLogEntry, DowntimeWindow, NotificationEvent, TargetModule,
-        },
+    meta::downtimes::{
+        Downtime, DowntimeNotificationLogEntry, DowntimeWindow, NotificationEvent, TargetModule,
    },
    utils::time::now_micros,
 };
@@ -27,13 +24,12 @@ use infra::table::downtime_notifications::{self as log_table, DowntimeNotificati
 use o2_enterprise::enterprise::common::config::get_config as get_o2_config;

 use super::{DowntimeError, MatchCounts};
-use crate::alerts::templates::{DOWNTIME_EMAIL_TITLE, downtime_templates};
+use crate::alerts::templates::DOWNTIME_EMAIL_TITLE;

 pub const LOG_LIMIT: u64 = 50;
 /// How often the notifier ticks; a start one tick back is already behind every healthy checkpoint.
 pub const TICK_INTERVAL_SECS: u64 = 30;
 pub const RESULT_OK: &str = "ok";
-const DOWNTIME_VARIABLE_PREFIX: &str = "{downtime_";

 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct DueEvent {
@@ -52,31 +48,6 @@ impl DueEvent {
    }
 }

-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-enum Escape {
-    Json,
-    Html,
-    Plain,
-}
-
-impl Escape {
-    fn apply(self, value: &str) -> String {
-        match self {
-            Self::Json => {
-                let quoted = serde_json::Value::String(value.to_string()).to_string();
-                quoted[1..quoted.len() - 1].to_string()
-            }
-            Self::Html => value
-                .replace('&', "&amp;")
-                .replace('<', "&lt;")
-                .replace('>', "&gt;")
-                .replace('"', "&quot;")
-                .replace('\'', "&#39;"),
-            Self::Plain => value.to_string(),
-        }
-    }
-}
-
 pub fn this_region() -> String {
    let o2 = get_o2_config();
    if o2.super_cluster.enabled {
@@ -107,7 +78,7 @@ pub async fn deliver(row: &Downtime, due: &DueEvent) -> Result<bool, DowntimeErr
        return Ok(false);
    }
    let destinations = destinations_of(row);
-    let result = match send(row, due, &destinations).await {
+    let result = match send(row, due, record.sent_at, &destinations).await {
        Ok(()) => RESULT_OK.to_string(),
        Err(e) => {
            log::warn!(
@@ -183,7 +154,14 @@ fn log_entry(row: DowntimeNotification) -> DowntimeNotificationLogEntry {
    }
 }

-async fn send(row: &Downtime, due: &DueEvent, destinations: &[String]) -> Result<(), String> {
+async fn send(
+    row: &Downtime,
+    due: &DueEvent,
+    claimed_at: i64,
+    destinations: &[String],
+) -> Result<(), String> {
+    use crate::alerts::notifications::notify::{NotifyError, SkipOutcome, notify};
+
    let counts = super::match_counts(row).await.ok();
    let vars = variables(
        row,
@@ -192,10 +170,19 @@ async fn send(row: &Downtime, due: &DueEvent, destinations: &[String]) -> Result
        counts.as_ref(),
        &get_config().common.web_url,
    );
+    let ctx = downtime_context(row, due, claimed_at, vars);
    let mut errors = Vec::new();
    for name in destinations {
-        if let Err(e) = send_one(&row.org, name, &vars).await {
-            errors.push(format!("{name}: {e}"));
+        let mut ctx = ctx.clone();
+        match notify(&row.org, name, &[], None, &mut ctx).await {
+            Ok(_) => {}
+            // Platforms hear nothing about a maintenance window; that is not a failure.
+            Err(NotifyError::Skipped(reason))
+                if matches!(
+                    reason.outcome(),
+                    SkipOutcome::Delivered | SkipOutcome::Handled
+                ) => {}
+            Err(e) => errors.push(format!("{name}: {e}")),
        }
    }
    if errors.is_empty() {
@@ -205,48 +192,76 @@ async fn send(row: &Downtime, due: &DueEvent, destinations: &[String]) -> Result
    }
 }

-async fn send_one(org: &str, name: &str, vars: &[(&str, String)]) -> Result<(), String> {
-    let dest = crate::alerts::destinations::get(org, name)
-        .await
-        .map_err(|e| e.to_string())?;
-    let Module::Alert {
-        template,
-        destination_type,
-    } = dest.module
-    else {
-        return Err("not an alert destination".to_string());
+/// The shared path's context for one downtime event, keyed `{downtime_id}:{window_start}`.
+fn downtime_context(
+    row: &Downtime,
+    due: &DueEvent,
+    claimed_at: i64,
+    vars: Vec<(&'static str, String)>,
+) -> crate::alerts::notifications::NotificationContext {
+    use config::meta::alerts::notification_kind::NotificationKind;
+
+    use crate::alerts::notifications::{
+        NotificationContext, STATUS_FIRING, STATUS_RESOLVED, context::RenderHints,
    };
-    let own = match template {
-        Some(template) => db::alerts::templates::get(org, &template).await.ok(),
-        None => None,
+
+    let ended = matches!(
+        due.event,
+        NotificationEvent::Ended | NotificationEvent::Cancelled
+    );
+    // Each extension is its own claim, so its claim time keeps the rows apart.
+    let event = match due.event {
+        NotificationEvent::Extended => format!("extended:{claimed_at}"),
+        other => other.as_str().to_string(),
    };
-    let template = template_for(own, &destination_type);
-    let escape = if matches!(destination_type, DestinationType::Email(_)) {
-        Escape::Html
-    } else {
-        Escape::Json
-    };
-    let body = render(&template.body, vars, escape);
-    let subject = match &template.template_type {
-        TemplateType::Email { title } if !title.is_empty() => render(title, vars, Escape::Plain),
-        _ => render(DOWNTIME_EMAIL_TITLE, vars, Escape::Plain),
-    };
-    crate::alerts::alert::dispatch_notification(org, &destination_type, &subject, body)
-        .await
-        .map(|_| ())
-        .map_err(|e| e.to_string())
+    let subject = render_title(&vars);
+    let url = vars
+        .iter()
+        .find(|(name, _)| *name == "downtime_url")
+        .map(|(_, v)| v.clone())
+        .unwrap_or_default();
+    NotificationContext {
+        org_name: row.org.clone(),
+        alert_name: row.name.clone(),
+        alert_status: if ended {
+            STATUS_RESOLVED
+        } else {
+            STATUS_FIRING
+        }
+        .to_string(),
+        episode_id: Some(format!("{}:{}", row.id, due.key)),
+        alert_start_time: utc(due.window.start),
+        alert_end_time: utc(due.window.end),
+        alert_start_time_millis: (due.window.start / 1000).to_string(),
+        alert_end_time_millis: (due.window.end / 1000).to_string(),
+        alert_url: url,
+        alert_trigger_time: claimed_at,
+        alert_trigger_time_str: utc(claimed_at),
+        alert_description: row.reason.clone().unwrap_or_default(),
+        context_attributes: vars
+            .into_iter()
+            .filter(|(name, _)| *name != "org_name")
+            .map(|(k, v)| (k.to_string(), v))
+            .collect(),
+        kind: NotificationKind::Downtime,
+        object_id: row.id.clone(),
+        event,
+        cluster_name: config::get_cluster_name(),
+        render_hints: RenderHints {
+            fallback_subject: Some(subject),
+            escape_email_attributes: true,
+        },
+        ..Default::default()
+    }
 }

-fn template_for(own: Option<Template>, destination_type: &DestinationType) -> Template {
-    if let Some(own) = own.filter(|t| t.body.contains(DOWNTIME_VARIABLE_PREFIX)) {
-        return own;
-    }
-    let [http, email] = downtime_templates();
-    if matches!(destination_type, DestinationType::Email(_)) {
-        email
-    } else {
-        http
+/// Today's subject, for a destination whose template has no title of its own.
+fn render_title(vars: &[(&'static str, String)]) -> String {
+    let mut title = DOWNTIME_EMAIL_TITLE.to_string();
+    for (name, value) in vars {
+        title = title.replace(&format!("{{{name}}}"), value);
    }
+    title
 }

 fn variables(
@@ -272,34 +287,6 @@ fn variables(
    ]
 }

-/// One pass, so a value that itself reads like `{downtime_url}` is not substituted again.
-fn render(body: &str, vars: &[(&str, String)], escape: Escape) -> String {
-    let mut out = String::with_capacity(body.len());
-    let mut rest = body;
-    while let Some(open) = rest.find('{') {
-        out.push_str(&rest[..open]);
-        let tail = &rest[open..];
-        let hit = tail.find('}').and_then(|close| {
-            let name = &tail[1..close];
-            vars.iter()
-                .find(|(n, _)| *n == name)
-                .map(|(_, value)| (close, value))
-        });
-        match hit {
-            Some((close, value)) => {
-                out.push_str(&escape.apply(value));
-                rest = &tail[close + 1..];
-            }
-            None => {
-                out.push('{');
-                rest = &tail[1..];
-            }
-        }
-    }
-    out.push_str(rest);
-    out
-}
-
 fn event_phrase(event: NotificationEvent) -> &'static str {
    match event {
        NotificationEvent::Started => "started",
@@ -350,17 +337,15 @@ fn module_words(module: TargetModule) -> (&'static str, &'static str) {

 #[cfg(test)]
 mod tests {
-    use config::meta::{
-        destinations::{Email, Endpoint},
-        downtimes::{
-            DowntimeNotifications, DowntimeSchedule, DowntimeTarget, NotificationEvents, Repeat,
-            TargetFolders,
-        },
+    use config::meta::downtimes::{
+        DowntimeNotifications, DowntimeSchedule, DowntimeTarget, NotificationEvents, Repeat,
+        TargetFolders,
    };
    use infra::table::entity::downtime_notifications::Entity as LogEntity;
    use sea_orm::{ConnectionTrait, Database, DatabaseConnection, Schema};

    use super::*;
+    use crate::alerts::templates::downtime_templates;

    const HOUR: i64 = 3_600_000_000;

@@ -499,8 +484,14 @@ mod tests {
        assert_eq!(entries[0].destinations, ["slack", "pagerduty"]);
    }

+    fn due(event: NotificationEvent) -> DueEvent {
+        DueEvent::of_window(event, window())
+    }
+
    #[test]
-    fn variables_fill_both_prebuilt_templates_escaped_for_their_format() {
+    fn the_seeded_pair_renders_through_the_shared_path_escaped_for_its_format() {
+        use crate::alerts::notifications::custom::apply_custom_template;
+
        let counts = MatchCounts {
            alerts: 2,
            synthetics: 1,
@@ -513,9 +504,10 @@ mod tests {
            Some(&counts),
            "https://o2.example.com",
        );
+        let ctx = downtime_context(&row(None), &due(NotificationEvent::EndingSoon), 7, vars);
        let [http, email] = downtime_templates();

-        let body = render(&http.body, &vars, Escape::Json);
+        let body = apply_custom_template(&http.body, &ctx, false);
        let json: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(json["downtime"]["name"], "Nightly \"deploy\" <db>");
        assert_eq!(json["downtime"]["event"], "ends soon");
@@ -530,47 +522,31 @@ mod tests {
        assert_eq!(json["downtime"]["org"], "acme");
        assert_eq!(json["downtime"]["reason"], "CHG-1");

-        let html = render(&email.body, &vars, Escape::Html);
+        let html = apply_custom_template(&email.body, &ctx, true);
        assert!(html.contains("Nightly &quot;deploy&quot; &lt;db&gt;"));
        assert!(!html.contains("{downtime_"));
        assert!(!html.contains("{org_name}"));
    }

    #[test]
-    fn render_substitutes_once_and_leaves_unknown_braces_alone() {
-        let vars = vec![
-            ("downtime_name", "{downtime_url}".to_string()),
-            ("downtime_url", "https://x".to_string()),
-        ];
+    fn a_downtime_event_is_keyed_on_its_window_and_each_extension_is_its_own_row() {
+        let vars = || variables(&row(None), NotificationEvent::Started, &window(), None, "");
+        let started = downtime_context(&row(None), &due(NotificationEvent::Started), 5, vars());
+        assert_eq!(started.kind.as_str(), "downtime");
+        assert_eq!(started.event, "started");
        assert_eq!(
-            render(
-                "{downtime_name} {alert_name} {downtime_url} {",
-                &vars,
-                Escape::Plain
-            ),
-            "{downtime_url} {alert_name} https://x {"
+            started.episode_id.as_deref(),
+            Some(format!("{}:{}", row(None).id, window().start).as_str())
        );
-        assert_eq!(Escape::Json.apply("a\"b\nc"), "a\\\"b\\nc");
-    }
-
-    #[test]
-    fn a_destination_template_is_used_only_when_written_for_downtimes() {
-        let http = DestinationType::Http(Endpoint::default());
-        let email = DestinationType::Email(Email::default());
-        let own = |body: &str| Template {
-            name: "mine".to_string(),
-            body: body.to_string(),
-            ..Default::default()
-        };
+        assert_eq!(started.alert_status, "firing");
+        let ended = downtime_context(&row(None), &due(NotificationEvent::Ended), 6, vars());
+        assert_eq!(ended.alert_status, "resolved");
+        let extended = downtime_context(&row(None), &due(NotificationEvent::Extended), 9, vars());
+        assert_eq!(extended.event, "extended:9");
        assert_eq!(
-            template_for(Some(own("{downtime_name} is {downtime_event}")), &http).name,
-            "mine"
+            started.render_hints.fallback_subject.as_deref(),
+            Some("OpenObserve downtime Nightly \"deploy\" <db>: started")
        );
-        assert_eq!(
-            template_for(Some(own("[{alert_status}] {alert_name}")), &http).name,
-            "prebuilt_downtime"
-        );
-        assert_eq!(template_for(None, &email).name, "prebuilt_downtime_email");
    }

    #[test]
@@ -74,93 +74,307 @@ pub struct CheckNotification {
    pub passing_locations: Vec<String>,
    /// Environments that did not pass, worst first.
    pub failing_environments: Vec<String>,
+    /// The check's template override.
+    pub template: Option<String>,
+    /// Start of the alert episode; keys `{check_id}:{alerting_since}`.
+    pub alerting_since: i64,
 }

-/// Fires once per run (when all jobs have completed) for non-passing runs.
-/// Passing runs are suppressed — operators want alerts, not confirmations.
-///
-/// The message body is built per destination type: Slack-compatible JSON for
-/// HTTP webhooks, an HTML card for email, plain text for SNS.
+/// Fires once per run for a non-passing run; a recovery is sent by the episode close instead.
 #[cfg(feature = "enterprise")]
 pub async fn notify_check_result(n: CheckNotification) {
-    // A passing run is worth sending exactly once: when it ends an incident
-    // somebody was already told about. Otherwise it is a confirmation nobody
-    // asked for.
-    //
-    // A flaky run reports as `warning`, not `passed`, so it is not caught here —
-    // the decision to send it was already made upstream against `cooldown_mins`.
-    if !n.recovery && (n.status == "passed" || n.status == "up") {
+    use crate::alerts::notifications::notify::{NotifyError, notify};
+
+    // A passing run matters only when it ends an incident, and the close sends that.
+    if n.recovery || n.status == "passed" || n.status == "up" {
        return;
    }
-
-    use config::meta::destinations::{DestinationType, Module};
-
+    let ctx = synthetics_notification_context(&n);
+    let explicit =
+        crate::alerts::notifications::org_default::load_override(&n.org_id, n.template.as_deref())
+            .await;
    for dest_name in &n.destinations {
-        match crate::alerts::destinations::get_with_template(&n.org_id, dest_name).await {
-            Ok((dest, _tpl)) => {
-                let Module::Alert {
-                    destination_type, ..
-                } = &dest.module
-                else {
-                    continue;
-                };
-
-                let msg = match destination_type {
-                    DestinationType::Email(_) => build_email_html(&n),
-                    DestinationType::Sns(_) => build_plain_text(&n),
-                    DestinationType::Http(_) | DestinationType::SlackApp(_) => build_slack_json(&n),
-                };
-
-                let subject = if n.recovery {
-                    format!("[OpenObserve Synthetics] ✅ {} has RECOVERED", n.check_name)
-                } else if n.degraded {
-                    // Not "is WARNING": the point of the message is that this needs
-                    // action before it becomes an outage. Named where we know the
-                    // condition, because the subject is the line that decides
-                    // whether anyone opens the alert — and "CERTIFICATE EXPIRING"
-                    // gets renewed where a generic "DEGRADED" gets skimmed.
-                    match n.status_reason.as_deref() {
-                        Some("cert_expiring") => format!(
-                            "[OpenObserve Synthetics] 🟡 {} — CERTIFICATE EXPIRING SOON",
-                            n.check_name
-                        ),
-                        Some("sftp_degraded") => format!(
-                            "[OpenObserve Synthetics] 🟡 {} — SFTP DEGRADED",
-                            n.check_name
-                        ),
-                        _ => format!("[OpenObserve Synthetics] 🟡 {} is DEGRADED", n.check_name),
-                    }
-                } else if n.flaky {
-                    format!("[OpenObserve Synthetics] 🔁 {} is FLAKY", n.check_name)
-                } else {
-                    format!(
-                        "[OpenObserve Synthetics] {} {} is {}",
-                        status_emoji(&n),
-                        n.check_name,
-                        n.status.to_uppercase()
-                    )
-                };
-                if let Err(e) = crate::alerts::alert::dispatch_notification(
-                    &n.org_id,
-                    destination_type,
-                    &subject,
-                    msg,
-                )
-                .await
-                {
-                    log::error!(
-                        "[synthetics] notify dest={dest_name} check={}: {e}",
-                        n.check_id
-                    );
-                }
-            }
-            Err(e) => {
-                log::error!("[synthetics] load dest={dest_name} org={}: {e}", n.org_id);
-            }
+        let mut ctx = ctx.clone();
+        match notify(&n.org_id, dest_name, &explicit, None, &mut ctx).await {
+            Ok(_) => {}
+            Err(NotifyError::Skipped(reason)) => log::debug!(
+                "[synthetics] notify dest={dest_name} check={} skipped: {}",
+                n.check_id,
+                reason.as_str()
+            ),
+            Err(e) => log::error!(
+                "[synthetics] notify dest={dest_name} check={}: {e}",
+                n.check_id
+            ),
        }
    }
 }

+/// Closes a check's alert episode with its state swap in one transaction, then sends the recovery.
+#[cfg(feature = "enterprise")]
+pub fn close_alert_episode(
+    close: openobserve_synthetics::alerting::EpisodeClose,
+) -> openobserve_synthetics::alerting::BoxFuture<'static, anyhow::Result<bool>> {
+    Box::pin(close_alert_episode_inner(close))
+}
+
+#[cfg(feature = "enterprise")]
+async fn close_alert_episode_inner(
+    close: openobserve_synthetics::alerting::EpisodeClose,
+) -> anyhow::Result<bool> {
+    use config::meta::alerts::notification_kind::NotificationKind;
+    use infra::table::notification_deliveries::{CLOSED_BY_PRODUCER, CLOSED_BY_RESOLVE};
+
+    use crate::alerts::notifications::delivery;
+
+    let openobserve_synthetics::alerting::EpisodeClose {
+        org_id,
+        check_id,
+        check_name,
+        check_type,
+        target,
+        episode_id,
+        run_id,
+        destinations,
+        template,
+        send_resolves,
+        swap,
+    } = close;
+    let outcomes = infra::table::synthetics_jobs::run_location_outcomes(
+        infra::db::get_orm_client_ro().await,
+        &run_id,
+    )
+    .await
+    .unwrap_or_default();
+    let n = CheckNotification {
+        org_id: org_id.clone(),
+        check_name,
+        check_id: check_id.clone(),
+        check_type,
+        target,
+        destinations: destinations.clone(),
+        run_id,
+        status: "passed".to_string(),
+        job_count: (outcomes.failing.len() + outcomes.passing.len()) as i64,
+        error: None,
+        checked_at: config::utils::time::now_micros(),
+        recovery: true,
+        consecutive_failures: 0,
+        flaky: false,
+        status_reason: None,
+        error_source: String::new(),
+        degraded: false,
+        failing_locations: outcomes.failing,
+        passing_locations: outcomes.passing,
+        failing_environments: Vec::new(),
+        template: template.clone(),
+        alerting_since: episode_id
+            .rsplit_once(':')
+            .and_then(|(_, since)| since.parse().ok())
+            .unwrap_or_default(),
+    };
+    let ctx = synthetics_notification_context(&n);
+    let explicit = if send_resolves {
+        crate::alerts::notifications::org_default::load_override(&org_id, template.as_deref()).await
+    } else {
+        Vec::new()
+    };
+    let mut rows = Vec::new();
+    let mut recorded_rows = Vec::new();
+    if send_resolves {
+        for dest in &destinations {
+            rows.push(delivery::fanout_row(&org_id, dest, &explicit, None, &ctx).await);
+        }
+        for dest in
+            delivery::recorded_destinations(&org_id, NotificationKind::Synthetics, &episode_id)
+                .await
+                .into_iter()
+                .filter(|d| !destinations.contains(d))
+        {
+            recorded_rows.push(delivery::fanout_row(&org_id, &dest, &explicit, None, &ctx).await);
+        }
+    }
+    let (current, mut outside) = delivery::split_fanout(rows);
+    let (recorded, recorded_outside) = delivery::split_fanout(recorded_rows);
+    outside.extend(recorded_outside);
+    let state_change: delivery::StateChange = Box::new(move |txn| swap(txn));
+    let closed_by = if send_resolves {
+        CLOSED_BY_RESOLVE
+    } else {
+        CLOSED_BY_PRODUCER
+    };
+    let results = delivery::resolve_episode(
+        &org_id,
+        NotificationKind::Synthetics,
+        &episode_id,
+        &check_id,
+        current,
+        recorded,
+        closed_by,
+        send_resolves,
+        state_change,
+    )
+    .await?;
+    let Some(results) = results else {
+        return Ok(false);
+    };
+    delivery::send_outside_after_close(&org_id, &outside, &explicit, None, &ctx).await;
+    for result in results.iter().filter_map(|r| r.as_ref().err()) {
+        log::warn!("[synthetics] recovery of {check_id} not delivered yet: {result}");
+    }
+    Ok(true)
+}
+
+/// The subject line each run sends today; the synthetics card keeps it as its title.
+#[cfg(feature = "enterprise")]
+fn check_title(n: &CheckNotification) -> String {
+    if n.recovery {
+        format!("[OpenObserve Synthetics] ✅ {} has RECOVERED", n.check_name)
+    } else if n.degraded {
+        // Named where the condition is known: the subject decides whether anyone opens the alert.
+        match n.status_reason.as_deref() {
+            Some("cert_expiring") => format!(
+                "[OpenObserve Synthetics] 🟡 {} — CERTIFICATE EXPIRING SOON",
+                n.check_name
+            ),
+            Some("sftp_degraded") => format!(
+                "[OpenObserve Synthetics] 🟡 {} — SFTP DEGRADED",
+                n.check_name
+            ),
+            _ => format!("[OpenObserve Synthetics] 🟡 {} is DEGRADED", n.check_name),
+        }
+    } else if n.flaky {
+        format!("[OpenObserve Synthetics] 🔁 {} is FLAKY", n.check_name)
+    } else {
+        format!(
+            "[OpenObserve Synthetics] {} {} is {}",
+            status_emoji(n),
+            n.check_name,
+            n.status.to_uppercase()
+        )
+    }
+}
+
+/// `(episode_id, event)`: firing and recovery share an episode; informational runs own one.
+#[cfg(feature = "enterprise")]
+fn episode_and_event(n: &CheckNotification) -> (String, String) {
+    let alert = format!("{}:{}", n.check_id, n.alerting_since);
+    let run = format!("{}:{}", n.check_id, n.run_id);
+    if n.recovery {
+        (alert, "resolved".to_string())
+    } else if n.degraded && n.status_reason.as_deref() == Some("cert_expiring") {
+        (run, format!("cert_expiring:{}", n.run_id))
+    } else if n.degraded {
+        (run, format!("degraded:{}", n.run_id))
+    } else if n.flaky {
+        (run, format!("flaky:{}", n.run_id))
+    } else {
+        (alert, format!("firing:{}", n.run_id))
+    }
+}
+
+/// The shared path's context for one synthetics message (Backend section 6 field mapping).
+#[cfg(feature = "enterprise")]
+pub fn synthetics_notification_context(
+    n: &CheckNotification,
+) -> crate::alerts::notifications::NotificationContext {
+    use config::meta::alerts::{level::AlertLevel, notification_kind::NotificationKind};
+
+    use crate::alerts::notifications::{
+        LegacyPayload, NotificationContext, STATUS_FIRING, STATUS_RESOLVED, context::RenderHints,
+    };
+
+    let (level, level_str) = if n.recovery {
+        (AlertLevel::Ok, "ok")
+    } else if n.degraded || n.flaky || n.status == "warning" {
+        (AlertLevel::Warning, "warning")
+    } else {
+        (AlertLevel::Critical, "critical")
+    };
+    let (episode_id, event) = episode_and_event(n);
+    let mut description = status_headline(n);
+    if let Some(e) = n.error.as_deref().filter(|e| !e.is_empty()) {
+        description.push_str(&format!("\n{e}"));
+    }
+    let checked = crate::anomaly_detection::fmt_local_us(n.checked_at);
+    let legacy = slack_payload(n);
+    NotificationContext {
+        org_name: n.org_id.clone(),
+        stream_type: "synthetics".to_string(),
+        stream_name: n.target.clone(),
+        alert_name: n.check_name.clone(),
+        alert_level: level_str.to_string(),
+        alert_status: if n.recovery {
+            STATUS_RESOLVED
+        } else {
+            STATUS_FIRING
+        }
+        .to_string(),
+        episode_id: Some(episode_id),
+        alert_start_time: checked.clone(),
+        alert_end_time: checked.clone(),
+        alert_start_time_millis: (n.checked_at / 1000).to_string(),
+        alert_end_time_millis: (n.checked_at / 1000).to_string(),
+        alert_url: run_url(n),
+        alert_trigger_time: n.checked_at,
+        alert_trigger_time_str: checked,
+        alert_description: description,
+        context_attributes: synthetics_attributes(n),
+        level: Some(level),
+        kind: NotificationKind::Synthetics,
+        object_id: n.check_id.clone(),
+        event,
+        cluster_name: config::get_cluster_name(),
+        legacy_payload: Some(LegacyPayload::new(move || legacy.clone())),
+        render_hints: RenderHints {
+            fallback_subject: Some(check_title(n)),
+            escape_email_attributes: true,
+        },
+        ..Default::default()
+    }
+}
+
+/// The synthetics variables (Backend section B2) plus `check_title`, today's subject line.
+#[cfg(feature = "enterprise")]
+fn synthetics_attributes(n: &CheckNotification) -> Vec<(String, String)> {
+    let label = if n.recovery {
+        "recovered"
+    } else if n.degraded {
+        "degraded"
+    } else if n.flaky {
+        "flaky"
+    } else {
+        n.status.as_str()
+    };
+    [
+        ("check_id", n.check_id.clone()),
+        ("check_name", n.check_name.clone()),
+        ("check_type", n.check_type.clone()),
+        ("check_target", n.target.clone()),
+        ("check_status", n.status.clone()),
+        ("check_status_label", label.to_string()),
+        (
+            "check_status_reason",
+            n.status_reason.clone().unwrap_or_default(),
+        ),
+        ("check_locations_line", locations_line(n)),
+        ("check_location_count", n.job_count.to_string()),
+        (
+            "check_consecutive_failures",
+            n.consecutive_failures.to_string(),
+        ),
+        ("check_flaky", n.flaky.to_string()),
+        ("check_degraded", n.degraded.to_string()),
+        ("check_run_id", n.run_id.clone()),
+        ("check_url", run_url(n)),
+        ("check_error", n.error.clone().unwrap_or_default()),
+        ("check_title", check_title(n)),
+    ]
+    .into_iter()
+    .map(|(k, v)| (k.to_string(), v))
+    .collect()
+}
+
 #[cfg(feature = "enterprise")]
 /// Emoji for a notification, branching on the **same flags** as
 /// [`status_headline`] and in the same order.
@@ -335,7 +549,7 @@ fn locations_line(n: &CheckNotification) -> String {
 /// Slack-compatible webhook payload (also renders fine in Teams/Discord-style
 /// webhooks that accept a `text` field).
 #[cfg(feature = "enterprise")]
-fn build_slack_json(n: &CheckNotification) -> String {
+fn slack_payload(n: &CheckNotification) -> serde_json::Value {
    let checked_secs = n.checked_at / 1_000_000;
    let mut lines = vec![
        format!("{} *{}*", status_emoji(n), status_headline(n)),
@@ -355,89 +569,7 @@ fn build_slack_json(n: &CheckNotification) -> String {
        "*Time:* <!date^{checked_secs}^{{date_time_secs}}|{checked_human}>"
    ));

-    serde_json::json!({ "text": lines.join("\n") }).to_string()
-}
-
-/// Plain text — SNS fans out to SMS/lambda/etc. where markup is noise.
-#[cfg(feature = "enterprise")]
-fn build_plain_text(n: &CheckNotification) -> String {
-    let mut lines = vec![
-        status_headline(n),
-        format!("Check: {} ({})", n.check_name, n.check_type),
-        format!("Target: {}", n.target),
-        format!("Status: {}", n.status),
-        format!("Locations: {}", locations_line(n)),
-    ];
-    if let Some(envs) = environments_line(n) {
-        lines.push(format!("Environments: {envs}"));
-    }
-    if let Some(e) = n.error.as_deref().filter(|e| !e.is_empty()) {
-        lines.push(format!("Error: {e}"));
-    }
-    lines.push(format!("Time: {}", checked_at_utc(n.checked_at)));
-    lines.push(format!("Run details: {}", run_url(n)));
-    lines.join("\n")
-}
-
-/// HTML card for email destinations (lettre sends it as the HTML alternative).
-#[cfg(feature = "enterprise")]
-fn build_email_html(n: &CheckNotification) -> String {
-    // TODO: update with a better template for all the checks
-    let color = if n.recovery {
-        "#2e7d32"
-    } else {
-        match n.status.as_str() {
-            "warning" => "#b58105",
-            "error" => "#b45309",
-            _ => "#c62828",
-        }
-    };
-    let error_row = match n.error.as_deref().filter(|e| !e.is_empty()) {
-        Some(e) => format!(
-            r#"<tr><td style="padding:6px 12px;color:#666;">Error</td>
-                <td style="padding:6px 12px;"><code>{}</code></td></tr>"#,
-            html_escape(e)
-        ),
-        None => String::new(),
-    };
-    format!(
-        r#"<div style="font-family:sans-serif;max-width:560px;">
-  <h2 style="color:{color};margin-bottom:4px;">{emoji} {headline}</h2>
-  <table style="border-collapse:collapse;background:#f7f7f7;border-radius:6px;width:100%;">
-    <tr><td style="padding:6px 12px;color:#666;width:140px;">Check</td>
-        <td style="padding:6px 12px;">{name} ({mtype})</td></tr>
-    <tr><td style="padding:6px 12px;color:#666;">Target</td>
-        <td style="padding:6px 12px;">{target}</td></tr>
-    <tr><td style="padding:6px 12px;color:#666;">Status</td>
-        <td style="padding:6px 12px;font-weight:bold;color:{color};">{status}</td></tr>
-    <tr><td style="padding:6px 12px;color:#666;">Locations</td>
-        <td style="padding:6px 12px;">{jobs}</td></tr>
-    <tr><td style="padding:6px 12px;color:#666;">Time</td>
-        <td style="padding:6px 12px;">{checked_at}</td></tr>
-    {error_row}
-  </table>
-  <p style="margin-top:12px;">
-    <a href="{url}" style="background:{color};color:#fff;padding:8px 16px;border-radius:4px;text-decoration:none;">View run details</a>
-  </p>
-</div>"#,
-        emoji = status_emoji(n),
-        headline = html_escape(&status_headline(n)),
-        name = html_escape(&n.check_name),
-        mtype = html_escape(&n.check_type),
-        target = html_escape(&n.target),
-        status = n.status.to_uppercase(),
-        jobs = html_escape(&locations_line(n)),
-        checked_at = checked_at_utc(n.checked_at),
-        url = run_url(n),
-    )
-}
-
-#[cfg(feature = "enterprise")]
-fn html_escape(s: &str) -> String {
-    s.replace('&', "&amp;")
-        .replace('<', "&lt;")
-        .replace('>', "&gt;")
-        .replace('"', "&quot;")
+    serde_json::json!({ "text": lines.join("\n") })
 }

 // ── Org provisioning ──────────────────────────────────────────────────────────
@@ -613,12 +745,12 @@ pub async fn location_staleness_watcher() {
                .await
                .unwrap_or_default();
            if !loc.enabled || agents.is_empty() {
-                clear_down(&loc.id).await;
+                clear_down(&org_id, &loc).await;
                continue;
            }
            let any_live = agents.iter().any(|a| now - a.last_seen_at <= window_us);
            if any_live {
-                clear_down(&loc.id).await;
+                clear_down(&org_id, &loc).await;
                continue;
            }

@@ -663,8 +795,11 @@ pub async fn location_staleness_watcher() {
                        notify_location_down(
                            &org_id,
                            &loc,
-                            checks.len(),
-                            window_us / 1_000_000,
+                            LocationEvent::Down {
+                                affected: checks.len(),
+                                stale_secs: window_us / 1_000_000,
+                                since: now,
+                            },
                            destinations,
                        )
                        .await
@@ -703,6 +838,212 @@ async fn report_location_down(
    !destinations.is_empty()
 }

+/// A private location went down at `since`, or came back from the outage that started then.
+#[cfg(feature = "enterprise")]
+enum LocationEvent {
+    Down {
+        affected: usize,
+        stale_secs: i64,
+        since: i64,
+    },
+    Up {
+        since: i64,
+    },
+}
+
+#[cfg(feature = "enterprise")]
+impl LocationEvent {
+    fn since(&self) -> i64 {
+        match self {
+            Self::Down { since, .. } | Self::Up { since } => *since,
+        }
+    }
+}
+
+#[cfg(feature = "enterprise")]
+async fn notify_location_down(
+    org_id: &str,
+    loc: &infra::table::synthetics_locations::SyntheticsLocationRecord,
+    event: LocationEvent,
+    destinations: &[String],
+) {
+    use crate::alerts::notifications::notify::{NotifyError, notify};
+
+    let ctx = location_context(org_id, loc, &event);
+    for dest_name in destinations {
+        let mut ctx = ctx.clone();
+        match notify(org_id, dest_name, &[], None, &mut ctx).await {
+            Ok(_) | Err(NotifyError::Skipped(_)) => {}
+            Err(e) => log::error!(
+                "[synthetics] location-down notify dest={dest_name} location={}: {e}",
+                loc.id
+            ),
+        }
+    }
+}
+
+/// Clears a location's down claim and resolves the outage to every destination it reached.
+#[cfg(feature = "enterprise")]
+async fn clear_down(
+    org_id: &str,
+    loc: &infra::table::synthetics_locations::SyntheticsLocationRecord,
+) {
+    use config::meta::alerts::notification_kind::NotificationKind;
+    use infra::table::notification_deliveries::CLOSED_BY_RESOLVE;
+
+    use crate::alerts::notifications::delivery;
+
+    let since = match infra::table::synthetics_locations::down_notified_at(&loc.id).await {
+        Ok(0) => return,
+        Ok(since) => since,
+        Err(e) => {
+            log::error!(
+                "[synthetics] staleness watcher: read down flag for {}: {e}",
+                loc.id
+            );
+            return;
+        }
+    };
+    let event = LocationEvent::Up { since };
+    let ctx = location_context(org_id, loc, &event);
+    let episode_id = ctx.episode_id.clone().unwrap_or_default();
+    let mut rows = Vec::new();
+    for dest in
+        delivery::recorded_destinations(org_id, NotificationKind::Synthetics, &episode_id).await
+    {
+        rows.push(delivery::fanout_row(org_id, &dest, &[], None, &ctx).await);
+    }
+    let (recorded, outside) = delivery::split_fanout(rows);
+    let id = loc.id.clone();
+    let state_change: delivery::StateChange = Box::new(move |txn| {
+        Box::pin(async move {
+            Ok(
+                infra::table::synthetics_locations::clear_down_notification_from_with(
+                    txn, &id, since,
+                )
+                .await?,
+            )
+        })
+    });
+    match delivery::resolve_episode(
+        org_id,
+        NotificationKind::Synthetics,
+        &episode_id,
+        &loc.id,
+        Vec::new(),
+        recorded,
+        CLOSED_BY_RESOLVE,
+        true,
+        state_change,
+    )
+    .await
+    {
+        Ok(Some(_)) => {
+            delivery::send_outside_after_close(org_id, &outside, &[], None, &ctx).await;
+        }
+        Ok(None) => {}
+        Err(e) => log::error!(
+            "[synthetics] staleness watcher: clear down flag for {}: {e}",
+            loc.id
+        ),
+    }
+}
+
+/// Context for a private location down or up message, keyed `{location_id}:{since}`.
+#[cfg(feature = "enterprise")]
+fn location_context(
+    org_id: &str,
+    loc: &infra::table::synthetics_locations::SyntheticsLocationRecord,
+    event: &LocationEvent,
+) -> crate::alerts::notifications::NotificationContext {
+    use config::meta::alerts::{level::AlertLevel, notification_kind::NotificationKind};
+
+    use crate::alerts::notifications::{
+        LegacyPayload, NotificationContext, STATUS_FIRING, STATUS_RESOLVED, context::RenderHints,
+    };
+
+    let (subject, text, status, level, ledger_event) = match event {
+        LocationEvent::Down {
+            affected,
+            stale_secs,
+            ..
+        } => (
+            format!(
+                "[OpenObserve Synthetics] 🔴 Private location {} is DOWN",
+                loc.label
+            ),
+            format!(
+                "Private location {} ({}) is down — no live agent for over {}s.\nAffected checks: {}\nRestart the agent container or check its network path to OpenObserve.",
+                loc.label, loc.region, stale_secs, affected
+            ),
+            "down",
+            AlertLevel::Critical,
+            "down",
+        ),
+        LocationEvent::Up { .. } => (
+            format!(
+                "[OpenObserve Synthetics] ✅ Private location {} is UP",
+                loc.label
+            ),
+            format!(
+                "Private location {} ({}) is back: an agent is live again.",
+                loc.label, loc.region
+            ),
+            "up",
+            AlertLevel::Ok,
+            STATUS_RESOLVED,
+        ),
+    };
+    let legacy = serde_json::json!({ "text": text });
+    let now = config::utils::time::now_micros();
+    let at = crate::anomaly_detection::fmt_local_us(now);
+    NotificationContext {
+        org_name: org_id.to_string(),
+        stream_type: "synthetics".to_string(),
+        alert_name: loc.label.clone(),
+        alert_level: if status == "up" { "ok" } else { "critical" }.to_string(),
+        alert_status: if status == "up" {
+            STATUS_RESOLVED
+        } else {
+            STATUS_FIRING
+        }
+        .to_string(),
+        episode_id: Some(format!("{}:{}", loc.id, event.since())),
+        alert_url: locations_url(org_id),
+        alert_trigger_time: now,
+        alert_trigger_time_str: at,
+        alert_description: text,
+        context_attributes: vec![
+            ("check_name".to_string(), loc.label.clone()),
+            ("check_status".to_string(), status.to_string()),
+            ("check_status_label".to_string(), status.to_string()),
+            ("check_target".to_string(), loc.region.clone()),
+            ("check_url".to_string(), locations_url(org_id)),
+            ("check_title".to_string(), subject.clone()),
+        ],
+        level: Some(level),
+        kind: NotificationKind::Synthetics,
+        object_id: loc.id.clone(),
+        event: ledger_event.to_string(),
+        cluster_name: config::get_cluster_name(),
+        legacy_payload: Some(LegacyPayload::new(move || legacy.clone())),
+        render_hints: RenderHints {
+            fallback_subject: Some(subject),
+            escape_email_attributes: true,
+        },
+        ..Default::default()
+    }
+}
+
+#[cfg(feature = "enterprise")]
+fn locations_url(org_id: &str) -> String {
+    let cfg = config::get_config();
+    format!(
+        "{}{}/web/settings/synthetics_locations?org_identifier={org_id}",
+        cfg.common.web_url, cfg.common.base_uri
+    )
+}
+
 #[cfg(all(test, feature = "enterprise"))]
 mod tests {
    use super::*;
@@ -753,6 +1094,8 @@ mod tests {
            failing_locations: vec!["aws-us-east-1".into(), "aws-us-west-1".into()],
            passing_locations: vec!["aws-eu-central-1".into()],
            failing_environments: vec![],
+            template: None,
+            alerting_since: 0,
        }
    }

@@ -889,6 +1232,8 @@ mod tests {
            recovery: true,
            passing_locations: vec![],
            failing_environments: vec![],
+            template: None,
+            alerting_since: 0,
            failing_locations: vec![],
            ..recovered()
        };
@@ -1018,84 +1363,98 @@ mod tests {
        assert_eq!(down_notified_at(&db).await, 200);
        assert_eq!(sent.load(std::sync::atomic::Ordering::SeqCst), 1);
    }
-}

-/// Sends the "location down" notification to each destination, matching the
-/// per-type message formats of `notify_check_result`.
-#[cfg(feature = "enterprise")]
-async fn notify_location_down(
-    org_id: &str,
-    loc: &infra::table::synthetics_locations::SyntheticsLocationRecord,
-    affected: usize,
-    stale_secs: i64,
-    destinations: &[String],
-) {
-    use config::meta::destinations::{DestinationType, Module};
-
-    let subject = format!(
-        "[OpenObserve Synthetics] 🔴 Private location {} is DOWN",
-        loc.label
-    );
-    let text = format!(
-        "Private location {} ({}) is down — no live agent for over {}s.\nAffected checks: {}\nRestart the agent container or check its network path to OpenObserve.",
-        loc.label, loc.region, stale_secs, affected
-    );
-
-    for dest_name in destinations {
-        match crate::alerts::destinations::get_with_template(org_id, dest_name).await {
-            Ok((dest, _tpl)) => {
-                let Module::Alert {
-                    destination_type, ..
-                } = &dest.module
-                else {
-                    continue;
-                };
-                let msg = match destination_type {
-                    DestinationType::Email(_) => format!(
-                        r#"<div style="font-family:sans-serif;max-width:560px;">
-  <h2 style="color:#c62828;margin-bottom:4px;">🔴 Private location {} is down</h2>
-  <p>No live agent for over {}s ({}).</p>
-  <p>Affected checks: <b>{}</b></p>
-  <p>Restart the agent container or check its network path to OpenObserve.</p>
-</div>"#,
-                        html_escape(&loc.label),
-                        stale_secs,
-                        html_escape(&loc.region),
-                        affected
-                    ),
-                    DestinationType::Sns(_) => text.clone(),
-                    DestinationType::Http(_) | DestinationType::SlackApp(_) => {
-                        serde_json::json!({ "text": text }).to_string()
-                    }
-                };
-                if let Err(e) = crate::alerts::alert::dispatch_notification(
-                    org_id,
-                    destination_type,
-                    &subject,
-                    msg,
-                )
-                .await
-                {
-                    log::error!(
-                        "[synthetics] location-down notify dest={dest_name} location={}: {e}",
-                        loc.id
-                    );
-                }
-            }
-            Err(e) => {
-                log::error!("[synthetics] load dest={dest_name} org={org_id}: {e}");
-            }
+    #[test]
+    fn every_subject_line_stays_byte_identical_as_the_card_title() {
+        let cases = [
+            (
+                recovered(),
+                "[OpenObserve Synthetics] ✅ checkout has RECOVERED",
+            ),
+            (
+                CheckNotification {
+                    degraded: true,
+                    status: "warning".into(),
+                    status_reason: Some("cert_expiring".into()),
+                    ..firing()
+                },
+                "[OpenObserve Synthetics] 🟡 checkout — CERTIFICATE EXPIRING SOON",
+            ),
+            (
+                CheckNotification {
+                    degraded: true,
+                    status: "warning".into(),
+                    status_reason: Some("sftp_degraded".into()),
+                    ..firing()
+                },
+                "[OpenObserve Synthetics] 🟡 checkout — SFTP DEGRADED",
+            ),
+            (
+                CheckNotification {
+                    degraded: true,
+                    status: "warning".into(),
+                    ..firing()
+                },
+                "[OpenObserve Synthetics] 🟡 checkout is DEGRADED",
+            ),
+            (
+                CheckNotification {
+                    flaky: true,
+                    status: "warning".into(),
+                    ..firing()
+                },
+                "[OpenObserve Synthetics] 🔁 checkout is FLAKY",
+            ),
+            (firing(), "[OpenObserve Synthetics] 🔴 checkout is FAILED"),
+        ];
+        for (mut n, want) in cases {
+            n.check_name = "checkout".into();
+            assert_eq!(check_title(&n), want);
+            let ctx = synthetics_notification_context(&n);
+            assert_eq!(ctx.render_hints.fallback_subject.as_deref(), Some(want));
+            assert!(
+                ctx.context_attributes
+                    .iter()
+                    .any(|(k, v)| k == "check_title" && v == want)
+            );
        }
    }
-}

-/// Clears a location's down flag so a future outage notifies again.
-///
-/// Every scheduler node calls this on recovery; the underlying update is
-/// idempotent, so they cannot disagree.
-#[cfg(feature = "enterprise")]
-async fn clear_down(location_id: &str) {
-    if let Err(e) = infra::table::synthetics_locations::clear_down_notification(location_id).await {
-        log::error!("[synthetics] staleness watcher: clear down flag for {location_id}: {e}");
+    #[test]
+    fn a_firing_and_its_recovery_share_the_episode_and_informational_runs_own_one() {
+        let opened = CheckNotification {
+            alerting_since: 77,
+            ..firing()
+        };
+        let (episode, event) = episode_and_event(&opened);
+        assert_eq!(episode, format!("{}:77", opened.check_id));
+        assert_eq!(event, format!("firing:{}", opened.run_id));
+        let closed = CheckNotification {
+            alerting_since: 77,
+            ..recovered()
+        };
+        assert_eq!(
+            episode_and_event(&closed),
+            (format!("{}:77", closed.check_id), "resolved".to_string())
+        );
+        let flaky = CheckNotification {
+            flaky: true,
+            ..firing()
+        };
+        let (episode, event) = episode_and_event(&flaky);
+        assert_eq!(episode, format!("{}:{}", flaky.check_id, flaky.run_id));
+        assert!(event.starts_with("flaky:"));
+    }
+
+    #[test]
+    fn a_generic_webhook_keeps_todays_text_body_and_email_escapes_the_values() {
+        let n = firing();
+        let ctx = synthetics_notification_context(&n);
+        let body = ctx.legacy_payload.as_ref().unwrap().build().to_string();
+        assert_eq!(body, slack_payload(&n).to_string());
+        assert!(body.starts_with("{\"text\":"));
+        assert_eq!(ctx.kind.as_str(), "synthetics");
+        assert!(ctx.render_hints.escape_email_attributes);
+        assert_eq!(ctx.object_id, n.check_id);
    }
 }
@@ -48,6 +48,10 @@ pub struct Model {
    /// a certificate is `warning` on every run for weeks — so it needs
    /// transition-based suppression rather than a time window.
    pub degraded_notified_at: i64,
+    /// When the open alert episode started, in microseconds; 0 when not alerting.
+    pub alerting_since: i64,
+    /// Template override by name for this check's messages; NULL uses the destination's.
+    pub template: Option<String>,
    pub owner: Option<String>,
    pub created_at: i64,
    pub updated_at: i64,
@@ -88,6 +92,8 @@ mod tests {
            last_alert_at: 0,
            alerting: false,
            degraded_notified_at: 0,
+            alerting_since: 0,
+            template: None,
            owner: None,
            created_at: 1750000000000000,
            updated_at: 1750000000000000,
@@ -0,0 +1,122 @@
+// 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/>.
+
+//! `template` (the check's override) and `alerting_since` (episode start) on synthetics.
+
+use sea_orm_migration::prelude::*;
+
+const TABLE: &str = "synthetics";
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[derive(DeriveIden, Clone)]
+enum Synthetics {
+    Table,
+    Template,
+    AlertingSince,
+}
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // The table is created from the entity on a fresh database, so each column is guarded.
+        if !manager.has_column(TABLE, "template").await? {
+            manager.alter_table(add_template_stmt()).await?;
+        }
+        if !manager.has_column(TABLE, "alerting_since").await? {
+            manager.alter_table(add_alerting_since_stmt()).await?;
+        }
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        for (column, iden) in [
+            ("template", Synthetics::Template),
+            ("alerting_since", Synthetics::AlertingSince),
+        ] {
+            if manager.has_column(TABLE, column).await? {
+                manager.alter_table(drop_stmt(iden)).await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+fn add_template_stmt() -> TableAlterStatement {
+    Table::alter()
+        .table(Synthetics::Table)
+        .add_column(ColumnDef::new(Synthetics::Template).string_len(256).null())
+        .to_owned()
+}
+
+/// Not null with a default, so a check that is alerting at upgrade keys its episode on 0.
+fn add_alerting_since_stmt() -> TableAlterStatement {
+    Table::alter()
+        .table(Synthetics::Table)
+        .add_column(
+            ColumnDef::new(Synthetics::AlertingSince)
+                .big_integer()
+                .not_null()
+                .default(0),
+        )
+        .to_owned()
+}
+
+fn drop_stmt(column: Synthetics) -> TableAlterStatement {
+    Table::alter()
+        .table(Synthetics::Table)
+        .drop_column(column)
+        .to_owned()
+}
+
+#[cfg(test)]
+mod tests {
+    use collapse::*;
+
+    use super::*;
+
+    #[test]
+    fn postgres() {
+        collapsed_eq!(
+            &add_template_stmt().to_string(PostgresQueryBuilder),
+            r#"ALTER TABLE "synthetics" ADD COLUMN "template" varchar(256) NULL"#
+        );
+        collapsed_eq!(
+            &add_alerting_since_stmt().to_string(PostgresQueryBuilder),
+            r#"ALTER TABLE "synthetics" ADD COLUMN "alerting_since" bigint NOT NULL DEFAULT 0"#
+        );
+    }
+
+    #[test]
+    fn sqlite() {
+        collapsed_eq!(
+            &add_template_stmt().to_string(SqliteQueryBuilder),
+            r#"ALTER TABLE "synthetics" ADD COLUMN "template" varchar(256) NULL"#
+        );
+        collapsed_eq!(
+            &add_alerting_since_stmt().to_string(SqliteQueryBuilder),
+            r#"ALTER TABLE "synthetics" ADD COLUMN "alerting_since" bigint NOT NULL DEFAULT 0"#
+        );
+    }
+
+    #[test]
+    fn down_drops_each_column() {
+        collapsed_eq!(
+            &drop_stmt(Synthetics::AlertingSince).to_string(PostgresQueryBuilder),
+            r#"ALTER TABLE "synthetics" DROP COLUMN "alerting_since""#
+        );
+    }
+}
@@ -254,6 +254,7 @@ mod m20261110_000003_add_incident_episode_opened_at;
 mod m20261120_000001_add_template_applies_to;
 mod m20261120_000002_rename_reserved_template_collisions;
 mod m20261120_000003_add_anomaly_template;
+mod m20261201_000001_add_synthetics_template_and_alerting_since;
 /// Shared body of the two `folder_id` migrations above; not a migration itself.
 mod workflow_folder_id;

@@ -789,6 +790,7 @@ impl MigratorTrait for Migrator {
            Box::new(m20261120_000001_add_template_applies_to::Migration),
            Box::new(m20261120_000002_rename_reserved_template_collisions::Migration),
            Box::new(m20261120_000003_add_anomaly_template::Migration),
+            Box::new(m20261201_000001_add_synthetics_template_and_alerting_since::Migration),
        ]
    }
 }
@@ -895,6 +897,10 @@ mod tests {
        (136, "m20261120_000001_add_template_applies_to"),
        (137, "m20261120_000002_rename_reserved_template_collisions"),
        (138, "m20261120_000003_add_anomaly_template"),
+        (
+            139,
+            "m20261201_000001_add_synthetics_template_and_alerting_since",
+        ),
    ];

    // Two branches that each claim the next version merge cleanly, so a reused number must fail
@@ -143,6 +143,7 @@ impl TryFrom<synthetics_checks::Model> for Synthetic {
            locations,
            enabled: m.enabled,
            destinations,
+            template: m.template,
            retries: settings.retries,
            cooldown_mins: settings.cooldown_mins,
            wait_before_retry_secs: settings.wait_before_retry_secs,
@@ -943,6 +944,27 @@ pub async fn update_last_check_status<C: ConnectionTrait>(
    Ok(res.rows_affected > 0)
 }

+/// `(org, name)` of each check whose template override is `template`; `None` is every org.
+pub async fn names_using_template<C: ConnectionTrait>(
+    conn: &C,
+    org_id: Option<&str>,
+    template: &str,
+) -> Result<Vec<(String, String)>, errors::Error> {
+    let mut query = Entity::find()
+        .select_only()
+        .column(Column::OrgId)
+        .column(Column::Name)
+        .filter(Column::Template.eq(template));
+    if let Some(org) = org_id {
+        query = query.filter(Column::OrgId.eq(org));
+    }
+    Ok(query
+        .order_by_asc(Column::Name)
+        .into_tuple::<(String, String)>()
+        .all(conn)
+        .await?)
+}
+
 /// The alert bookkeeping a completed run needs in order to decide whether to
 /// notify. Read and written by the ack path only.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -957,6 +979,8 @@ pub struct AlertState {
    /// degraded. Degradation persists for as long as the certificate takes to
    /// expire, so it is suppressed by transition rather than by a time window.
    pub degraded_notified_at: i64,
+    /// Start of the open alert episode in microseconds; 0 when not alerting.
+    pub alerting_since: i64,
 }

 /// Reads the alert state without pulling the whole check (config, secrets and
@@ -971,15 +995,19 @@ pub async fn get_alert_state<C: ConnectionTrait>(
        .column(Column::LastAlertAt)
        .column(Column::Alerting)
        .column(Column::DegradedNotifiedAt)
-        .into_tuple::<(i32, i64, bool, i64)>()
+        .column(Column::AlertingSince)
+        .into_tuple::<(i32, i64, bool, i64, i64)>()
        .one(conn)
        .await?;
    Ok(row.map(
-        |(consecutive_failures, last_alert_at, alerting, degraded_notified_at)| AlertState {
-            consecutive_failures,
-            last_alert_at,
-            alerting,
-            degraded_notified_at,
+        |(consecutive_failures, last_alert_at, alerting, degraded_notified_at, alerting_since)| {
+            AlertState {
+                consecutive_failures,
+                last_alert_at,
+                alerting,
+                degraded_notified_at,
+                alerting_since,
+            }
        },
    ))
 }
@@ -1021,6 +1049,7 @@ pub async fn update_alert_state_if<C: ConnectionTrait>(
            Column::DegradedNotifiedAt,
            Expr::value(state.degraded_notified_at),
        )
+        .col_expr(Column::AlertingSince, Expr::value(state.alerting_since))
        // Watermark for the status-page rebuilder's delta read. Safe here
        // because the caller only writes when `next != prior` (job_api), so a
        // healthy check acking every minute never bumps it.
@@ -1033,6 +1062,7 @@ pub async fn update_alert_state_if<C: ConnectionTrait>(
        .filter(Column::LastAlertAt.eq(expected.last_alert_at))
        .filter(Column::Alerting.eq(expected.alerting))
        .filter(Column::DegradedNotifiedAt.eq(expected.degraded_notified_at))
+        .filter(Column::AlertingSince.eq(expected.alerting_since))
        .exec(conn)
        .await?;
    Ok(res.rows_affected > 0)
@@ -1187,6 +1217,7 @@ fn update_mutable_fields(am: &mut ActiveModel, check: &Synthetic) -> Result<(),
    am.locations = Set(locations);
    am.enabled = Set(check.enabled);
    am.destinations = Set(destinations);
+    am.template = Set(check.template.clone());
    am.settings = Set(settings);
    am.secrets = Set(pack_secrets(check)?);
    Ok(())
@@ -1208,6 +1239,7 @@ fn build_active_model(check: &Synthetic) -> Result<ActiveModel, errors::Error> {
        locations: Set(locations),
        enabled: Set(check.enabled),
        destinations: Set(destinations),
+        template: Set(check.template.clone()),
        settings: Set(settings),
        secrets: Set(pack_secrets(check)?),
        ..Default::default()
@@ -1281,6 +1313,8 @@ pub(crate) mod tests {
            last_alert_at: 0,
            alerting: false,
            degraded_notified_at: 0,
+            alerting_since: 0,
+            template: None,
            owner: None,
            created_at: 1750000000000000,
            updated_at: 1750000000000000,
@@ -1861,6 +1895,7 @@ pub(crate) mod tests {
        m.last_alert_at = 1_749_998_000_000_000;
        m.alerting = true;
        m.degraded_notified_at = 1_749_997_000_000_000;
+        m.alerting_since = 1_749_996_000_000_000;

        let mut check = Synthetic::try_from(m.clone()).unwrap();
        check.name = "renamed".to_string();
@@ -1888,6 +1923,7 @@ pub(crate) mod tests {
        );
        assert!(am.alerting.is_unchanged());
        assert!(am.degraded_notified_at.is_unchanged());
+        assert!(am.alerting_since.is_unchanged(), "it keys the open episode");
    }

    /// `create` mints an id unless told otherwise. The super-cluster consumer
@@ -1942,6 +1978,58 @@ pub(crate) mod tests {
        db
    }

+    #[tokio::test]
+    async fn the_alert_state_swap_guards_and_writes_the_episode_start() {
+        let db = db_with_one_check(0).await;
+        let id = make_model().id;
+        let prior = get_alert_state(&db, &id).await.unwrap().unwrap();
+        let opened = AlertState {
+            alerting: true,
+            alerting_since: 42,
+            ..prior
+        };
+        assert!(
+            update_alert_state_if(&db, &id, prior, opened)
+                .await
+                .unwrap()
+        );
+        assert_eq!(get_alert_state(&db, &id).await.unwrap().unwrap(), opened);
+
+        // A reader that saw another episode start loses the race and writes nothing.
+        let stale = AlertState {
+            alerting_since: 7,
+            ..opened
+        };
+        assert!(
+            !update_alert_state_if(&db, &id, stale, AlertState::default())
+                .await
+                .unwrap()
+        );
+        assert_eq!(get_alert_state(&db, &id).await.unwrap().unwrap(), opened);
+    }
+
+    #[tokio::test]
+    async fn names_using_template_finds_the_checks_that_name_it() {
+        use sea_orm::{ActiveModelTrait, IntoActiveModel};
+
+        let db = db_with_one_check(0).await;
+        let mut am = make_model().into_active_model();
+        am.template = Set(Some("ops-card".to_string()));
+        am.update(&db).await.unwrap();
+        assert_eq!(
+            names_using_template(&db, Some("org1"), "ops-card")
+                .await
+                .unwrap(),
+            vec![("org1".to_string(), "Login Flow".to_string())]
+        );
+        assert!(
+            names_using_template(&db, Some("other"), "ops-card")
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
+
    async fn stored_status(db: &sea_orm::DatabaseConnection, id: &str) -> i32 {
        Entity::find_by_id(id)
            .one(db)
@@ -26,7 +26,8 @@ use std::{
 };

 use sea_orm::{
-    ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, Set, sea_query::Expr,
+    ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
+    sea_query::Expr,
 };
 use tokio::sync::RwLock;

@@ -309,6 +310,35 @@ pub async fn clear_down_notification(id: &str) -> Result<(), errors::Error> {
    Ok(())
 }

+/// The time the open down claim was taken, in microseconds; 0 when the location is not down.
+pub async fn down_notified_at(id: &str) -> Result<i64, errors::Error> {
+    let client = get_orm_client_ro().await;
+    let at = Entity::find_by_id(id)
+        .select_only()
+        .column(Column::DownNotifiedAt)
+        .into_tuple::<i64>()
+        .one(client)
+        .await
+        .map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))?;
+    Ok(at.unwrap_or(0))
+}
+
+/// Clears the claim only while it still holds `since`; true when this call cleared it.
+pub async fn clear_down_notification_from_with<C: ConnectionTrait>(
+    conn: &C,
+    id: &str,
+    since: i64,
+) -> Result<bool, errors::Error> {
+    let res = Entity::update_many()
+        .col_expr(Column::DownNotifiedAt, Expr::value(0i64))
+        .filter(Column::Id.eq(id))
+        .filter(Column::DownNotifiedAt.eq(since))
+        .exec(conn)
+        .await
+        .map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))?;
+    Ok(res.rows_affected > 0)
+}
+
 pub async fn try_claim_down_notification_with<C: ConnectionTrait>(
    conn: &C,
    id: &str,
@@ -290,6 +290,8 @@ mod tests {
                last_alert_at BIGINT NOT NULL DEFAULT 0,
                alerting BOOLEAN NOT NULL DEFAULT 0,
                degraded_notified_at BIGINT NOT NULL DEFAULT 0,
+                alerting_since BIGINT NOT NULL DEFAULT 0,
+                template TEXT,
                owner TEXT NULL,
                created_at BIGINT NOT NULL,
                updated_at BIGINT NOT NULL
@@ -770,6 +770,11 @@ pub async fn init() -> Result<(), anyhow::Error> {
            })
        },
    );
+    // A check's recovery swap and its resolve rows commit together on the delivery ledger.
+    #[cfg(feature = "enterprise")]
+    openobserve_synthetics::alerting::register_episode_closer(
+        openobserve_core::synthetics::close_alert_episode,
+    );
    // Every node: the validator, which reads the token cache, runs wherever HTTP is served.
    #[cfg(feature = "enterprise")]
    if openobserve_core::fleet::enabled() {
@@ -635,6 +635,8 @@ mod tests {
            last_alert_at: 0,
            alerting: false,
            degraded_notified_at: 0,
+            alerting_since: 0,
+            template: None,
            owner: None,
            created_at: 0,
            updated_at: 0,
@@ -783,6 +785,8 @@ mod tests {
                last_alert_at BIGINT NOT NULL DEFAULT 0,
                alerting BOOLEAN NOT NULL DEFAULT 0,
                degraded_notified_at BIGINT NOT NULL DEFAULT 0,
+                alerting_since BIGINT NOT NULL DEFAULT 0,
+                template TEXT,
                owner TEXT NULL,
                created_at BIGINT NOT NULL,
                updated_at BIGINT NOT NULL
@@ -46,6 +46,32 @@ pub type MayMute = fn(org: &str) -> bool;

 static MAY_MUTE: OnceLock<MayMute> = OnceLock::new();

+/// The state swap of a run that closes an alert episode, run inside the closer's transaction.
+pub type EpisodeSwap = Box<
+    dyn for<'a> FnOnce(&'a sea_orm::DatabaseTransaction) -> BoxFuture<'a, anyhow::Result<bool>>
+        + Send,
+>;
+
+/// What a closing run hands the closer; `send_resolves` is false for a muted recovery.
+pub struct EpisodeClose {
+    pub org_id: String,
+    pub check_id: String,
+    pub check_name: String,
+    pub check_type: String,
+    pub target: String,
+    pub episode_id: String,
+    pub run_id: String,
+    pub destinations: Vec<String>,
+    pub template: Option<String>,
+    pub send_resolves: bool,
+    pub swap: EpisodeSwap,
+}
+
+/// Closes the episode and runs the swap in one transaction; true when the swap won.
+pub type EpisodeCloser = fn(EpisodeClose) -> BoxFuture<'static, anyhow::Result<bool>>;
+
+static EPISODE_CLOSER: OnceLock<EpisodeCloser> = OnceLock::new();
+
 /// What a completed run is, before any suppression is applied.
 ///
 /// The distinction that matters is between an OUTAGE and a DEGRADATION. An
@@ -220,14 +246,26 @@ pub async fn muted_by(

 /// A muted run may close `alerting` but never open it, so a muted recovery is dropped.
 pub fn suppressed_state(prior: AlertState, next: AlertState) -> AlertState {
+    let alerting = prior.alerting && next.alerting;
    AlertState {
        last_alert_at: prior.last_alert_at,
-        alerting: prior.alerting && next.alerting,
+        alerting,
        degraded_notified_at: prior.degraded_notified_at,
+        alerting_since: if alerting { prior.alerting_since } else { 0 },
        ..next
    }
 }

+/// Registers the writer that closes an alert episode; OSS registers none and keeps the plain swap.
+pub fn register_episode_closer(f: EpisodeCloser) {
+    let _ = EPISODE_CLOSER.set(f);
+}
+
+/// The registered writer, when this build has a ledger to close the episode on.
+pub fn episode_closer() -> Option<EpisodeCloser> {
+    EPISODE_CLOSER.get().copied()
+}
+
 /// Decides the outcome for one completed run and returns the state to persist.
 ///
 /// `alert_if_fails` is clamped to at least 1 — a threshold of 0 would fire before
@@ -269,6 +307,7 @@ pub fn decide(
                alerting: false,
                // The target is up, so any degradation has cleared too.
                degraded_notified_at: 0,
+                alerting_since: 0,
            };

            // Closing an incident takes precedence over reporting flakiness. The
@@ -306,6 +345,7 @@ pub fn decide(
                last_alert_at: prior.last_alert_at,
                alerting: prior.alerting,
                degraded_notified_at: prior.degraded_notified_at,
+                alerting_since: prior.alerting_since,
            };
            let reminder = cooldown_us.max(DEGRADED_REMINDER_US);
            let first_time = prior.degraded_notified_at == 0;
@@ -334,6 +374,7 @@ pub fn decide(
                // longer the headline, and re-entering the degraded state later
                // is worth saying again.
                degraded_notified_at: 0,
+                alerting_since: prior.alerting_since,
            };

            if consecutive_failures < threshold {
@@ -352,6 +393,9 @@ pub fn decide(
                return (AlertOutcome::Silent, next);
            }

+            if !prior.alerting {
+                next.alerting_since = now_us;
+            }
            next.alerting = true;
            next.last_alert_at = now_us;
            (AlertOutcome::Firing, next)
@@ -372,6 +416,7 @@ mod tests {
            last_alert_at,
            alerting,
            degraded_notified_at: 0,
+            alerting_since: 0,
        }
    }

@@ -640,6 +685,55 @@ mod tests {
        assert_eq!(outcome, AlertOutcome::Silent);
    }

+    #[test]
+    fn an_episode_starts_on_the_opening_firing_and_ends_on_the_recovery() {
+        let (outcome, opened) = decide(AlertState::default(), RunClass::Failing, 1, 30, 10 * MIN);
+        assert_eq!(outcome, AlertOutcome::Firing);
+        assert_eq!(
+            opened.alerting_since,
+            10 * MIN,
+            "the opening firing keys the episode"
+        );
+
+        // A firing after the cooldown re-pages inside the same episode.
+        let (outcome, again) = decide(opened, RunClass::Failing, 1, 30, 50 * MIN);
+        assert_eq!(outcome, AlertOutcome::Firing);
+        assert_eq!(again.alerting_since, 10 * MIN);
+        let (_, degraded) = decide(again, RunClass::Degraded, 1, 30, 51 * MIN);
+        assert_eq!(degraded.alerting_since, 10 * MIN);
+
+        let (outcome, closed) = decide(again, RunClass::Healthy, 1, 30, 60 * MIN);
+        assert_eq!(outcome, AlertOutcome::Recovered);
+        assert_eq!(
+            closed.alerting_since, 0,
+            "a recovered check holds no episode"
+        );
+        let (_, next) = decide(closed, RunClass::Failing, 1, 30, 70 * MIN);
+        assert_eq!(
+            next.alerting_since,
+            70 * MIN,
+            "the next outage is a new episode"
+        );
+    }
+
+    #[test]
+    fn a_muted_recovery_clears_the_episode_and_a_muted_firing_keeps_it() {
+        let paged = AlertState {
+            alerting_since: 5 * MIN,
+            ..state(3, MIN, true)
+        };
+        let (_, recovered) = decide(paged, RunClass::Healthy, 3, 30, 45 * MIN);
+        assert_eq!(suppressed_state(paged, recovered).alerting_since, 0);
+        let (_, firing) = decide(paged, RunClass::Failing, 3, 30, 45 * MIN);
+        assert_eq!(suppressed_state(paged, firing).alerting_since, 5 * MIN);
+        let (_, opening) = decide(AlertState::default(), RunClass::Failing, 1, 30, 45 * MIN);
+        assert_eq!(
+            suppressed_state(AlertState::default(), opening).alerting_since,
+            0,
+            "a muted run never opens an episode"
+        );
+    }
+
    #[test]
    fn a_check_that_never_alerted_does_not_announce_recovery() {
        let (outcome, _) = decide(AlertState::default(), RunClass::Healthy, 1, 0, 1000 * MIN);
@@ -1033,6 +1033,12 @@ pub struct AckResponse {
    /// The downtime that muted this run; server-side only, since the ack body goes to a probe.
    #[serde(skip)]
    pub suppressed_by: Option<String>,
+    /// Start of the alert episode a firing or a recovery belongs to; server-side only.
+    #[serde(skip)]
+    pub alerting_since: i64,
+    /// The check's template override; server-side only.
+    #[serde(skip)]
+    pub template: Option<String>,
 }

 /// The notification a completed run should send, resolved against the check's
@@ -1642,6 +1648,8 @@ fn stale_lease_response(
        failing_environments: Vec::new(),
        usage_events: Vec::new(),
        suppressed_by: None,
+        alerting_since: 0,
+        template: None,
    }
 }

@@ -1832,10 +1840,11 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
    // Decide the notification, once per RUN. Per-job would alert once per
    // location for the same outage, and would advance the failure streak by the
    // fan-out factor rather than by one.
-    let (alert, consecutive_failures, suppressed_by) = if run_complete {
+    let (alert, consecutive_failures, suppressed_by, alerting_since) = if run_complete {
        resolve_alert(
            conn,
            &check.synthetics_id,
+            &check.run_id,
            RunOutcome {
                status: run_status.as_deref(),
                attempts: req.attempts,
@@ -1847,7 +1856,7 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
        )
        .await
    } else {
-        (AlertDecision::Silent, 0, None)
+        (AlertDecision::Silent, 0, None, 0)
    };

    // Which locations broke and which came back. Only worth a query when the run
@@ -1895,6 +1904,8 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
        usage_events,
        failing_environments,
        suppressed_by,
+        alerting_since,
+        template: synthetic.template.clone(),
    })
 }

@@ -2019,10 +2030,11 @@ struct RunOutcome<'a> {
 async fn resolve_alert<C: sea_orm::ConnectionTrait>(
    conn: &C,
    synthetics_id: &str,
+    run_id: &str,
    outcome: RunOutcome<'_>,
    check: &config::meta::synthetics::Synthetic,
    now_us: i64,
-) -> (AlertDecision, i32, Option<String>) {
+) -> (AlertDecision, i32, Option<String>, i64) {
    // See `alerting::classify`. Four outcomes, not two: an outage accumulates and
    // drives the streak, a degradation does not (a certificate is not "more
    // expired" on the twentieth check), and `error` is an outage rather than a
@@ -2036,10 +2048,10 @@ async fn resolve_alert<C: sea_orm::ConnectionTrait>(

    let prior = match infra::table::synthetics_checks::get_alert_state(conn, synthetics_id).await {
        Ok(Some(state)) => state,
-        Ok(None) => return (AlertDecision::Silent, 0, None), // deleted mid-run
+        Ok(None) => return (AlertDecision::Silent, 0, None, 0), // deleted mid-run
        Err(e) => {
            tracing::warn!(%synthetics_id, "[synthetics] get_alert_state: {e}");
-            return (AlertDecision::Silent, 0, None);
+            return (AlertDecision::Silent, 0, None, 0);
        }
    };

@@ -2068,7 +2080,7 @@ async fn resolve_alert<C: sea_orm::ConnectionTrait>(
        {
            tracing::warn!(%synthetics_id, "[synthetics] clear_alert_state: {e}");
        }
-        return (AlertDecision::Silent, 0, None);
+        return (AlertDecision::Silent, 0, None, 0);
    }

    let (outcome, next) = crate::alerting::decide(
@@ -2096,25 +2108,60 @@ async fn resolve_alert<C: sea_orm::ConnectionTrait>(
    // decision was made against a stale read, so we discard it rather than
    // overwrite the winner: the streak would otherwise undercount and the
    // notification would double.
-    if next != prior {
-        match infra::table::synthetics_checks::update_alert_state_if(
-            conn,
-            synthetics_id,
-            prior,
-            next,
-        )
-        .await
-        {
-            Ok(true) => {}
-            Ok(false) => {
-                tracing::info!(
-                    %synthetics_id,
-                    "[synthetics] alert state changed under us; another run decided this one"
-                );
-                return (AlertDecision::Silent, next.consecutive_failures, None);
-            }
-            Err(e) => tracing::warn!(%synthetics_id, "[synthetics] update_alert_state: {e}"),
+    let closes_episode = prior.alerting && !next.alerting;
+    let swapped = match crate::alerting::episode_closer().filter(|_| closes_episode) {
+        Some(closer) => {
+            let id = synthetics_id.to_string();
+            let close = crate::alerting::EpisodeClose {
+                org_id: check.org_id.clone(),
+                check_id: synthetics_id.to_string(),
+                check_name: check.name.clone(),
+                check_type: check_type_name(&check.check_type),
+                target: check.target.clone(),
+                episode_id: format!("{synthetics_id}:{}", prior.alerting_since),
+                run_id: run_id.to_string(),
+                destinations: check.destinations.clone(),
+                template: check.template.clone(),
+                send_resolves: suppressed_by.is_none(),
+                swap: Box::new(move |txn| {
+                    Box::pin(async move {
+                        Ok(infra::table::synthetics_checks::update_alert_state_if(
+                            txn, &id, prior, next,
+                        )
+                        .await?)
+                    })
+                }),
+            };
+            // A failed transaction changed nothing, so the next run repeats the recovery.
+            closer(close).await.unwrap_or_else(|e| {
+                tracing::warn!(%synthetics_id, "[synthetics] close alert episode: {e}");
+                false
+            })
        }
+        None if next != prior => {
+            match infra::table::synthetics_checks::update_alert_state_if(
+                conn,
+                synthetics_id,
+                prior,
+                next,
+            )
+            .await
+            {
+                Ok(won) => won,
+                Err(e) => {
+                    tracing::warn!(%synthetics_id, "[synthetics] update_alert_state: {e}");
+                    true
+                }
+            }
+        }
+        None => true,
+    };
+    if !swapped {
+        tracing::info!(
+            %synthetics_id,
+            "[synthetics] alert state changed under us; another run decided this one"
+        );
+        return (AlertDecision::Silent, next.consecutive_failures, None, 0);
    }

    let decision = match outcome {
@@ -2124,7 +2171,25 @@ async fn resolve_alert<C: sea_orm::ConnectionTrait>(
        crate::alerting::AlertOutcome::Degraded => AlertDecision::Degraded,
        crate::alerting::AlertOutcome::Recovered => AlertDecision::Recovered,
    };
-    (decision, next.consecutive_failures, suppressed_by)
+    let alerting_since = if closes_episode {
+        prior.alerting_since
+    } else {
+        next.alerting_since
+    };
+    (
+        decision,
+        next.consecutive_failures,
+        suppressed_by,
+        alerting_since,
+    )
+}
+
+/// The wire name of a check type, as the ack response spells it.
+fn check_type_name(check_type: &config::meta::synthetics::SyntheticType) -> String {
+    serde_json::to_value(check_type)
+        .ok()
+        .and_then(|v| v.as_str().map(str::to_owned))
+        .unwrap_or_default()
 }

 #[cfg(test)]
@@ -887,6 +887,8 @@ pub(crate) mod tests {
                last_alert_at BIGINT NOT NULL DEFAULT 0,
                alerting BOOLEAN NOT NULL DEFAULT 0,
                degraded_notified_at BIGINT NOT NULL DEFAULT 0,
+                alerting_since BIGINT NOT NULL DEFAULT 0,
+                template TEXT,
                owner TEXT NULL,
                created_at BIGINT NOT NULL,
                updated_at BIGINT NOT NULL
@@ -2919,6 +2919,7 @@ mod tests {
            "last_alert_at",
            "alerting",
            "degraded_notified_at",
+            "alerting_since",
        ] {
            let start = checks.find(&format!("\"{col}\"")).unwrap();
            let at = start + checks[start..].find("NOT NULL").unwrap() + "NOT NULL".len();