Fix alert notification delivery correctness

Fixes #1681

Fixes #1682

Fixes #1683

Contract-Neutral: Notification grouping initialization and alert-config propagation do not alter the broadly referenced agent-lifecycle or storage-recovery contracts; primary alerts, notifications, API, and monitoring contracts and regression proofs are updated.

docs/release-control/v6/internal/subsystems/alerts.md

@@ -24,6 +24,11 @@ target. The selected initial target owns firing, grouped, and recovery delivery,
 while every escalation level retains its own independently normalized target.
 Saving the alert configuration must update the live notification manager as
 well as persistence so delivery does not differ before and after restart.
+Configuration-change reevaluation may resolve metric-backed alerts against
+their updated thresholds and may apply explicit resource-disable policies, but
+it must not treat provider-owned incidents as missing thresholds. Unrelated
+configuration saves preserve those incidents and their acknowledgement state
+until their provider evaluator supplies recovery evidence.
 Docker and Podman container CPU thresholds evaluate host-capacity-normalized
 CPU percent, not Docker's runtime-native per-core percent. Alert metadata may
 carry the raw per-core value and reporting host CPU count for evidence, but the
@@ -508,6 +513,10 @@ resource type metadata before the legacy node fallback. Host-agent Ceph pool
 alerts may carry no-colon resource ids with `Instance == Node`, but when
 metadata or resource type says storage they must keep using storage threshold
 resolution and source-alias overrides instead of node defaults.
+That reevaluation boundary distinguishes metric-backed alert types from
+provider-owned incidents. A nil metric threshold may resolve only a known
+threshold-backed metric; it is not recovery evidence for resource incidents,
+and must not erase their acknowledgement state during a schedule-only save.
 
 Browser metric severity colors are also alert-backed. Workloads,
 Infrastructure, Storage, and the platform-page tables (Docker hosts and

docs/release-control/v6/internal/subsystems/api-contracts.md

@@ -29,6 +29,10 @@ The accepted values are `all`, `email`, `webhook`, and `apprise`; the backend
 normalizes aliases and invalid values before responding, persisting, and
 updating the live notification runtime. Escalation levels use the same value
 vocabulary but remain independent of the initial target.
+The same live update applies all four `schedule.grouping` fields atomically.
+When grouping is disabled, pending and subsequent alerts are delivered
+individually; the API boundary must not reduce that setting to the window or
+grouping-key fields while silently ignoring `enabled`.
 `DELETE /api/ai/patrol/suppressions/finding_{findingID}` is the canonical
 reopen path for a dismissed Patrol finding. It removes the finding-backed
 suppression row, preserves the operator note, clears dismissal state in both

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -24,6 +24,9 @@ initial-delivery target to the tenant notification manager. This is runtime
 wiring only: monitoring does not choose destinations or own notification
 policy, and live API saves must apply the same setting without requiring a
 monitor restart.
+Monitor construction also applies the persisted grouping enabled flag, window,
+and node/guest keys as one notification-manager policy, so restart behavior is
+identical to a live alert-configuration save.
 Monitoring also owns the distinction between Proxmox VM power state and QEMU
 guest-agent reachability: fresh or never-healthy VMs with an enabled but
 unavailable guest agent stay `not-running`, while only VMs with recent healthy

docs/release-control/v6/internal/subsystems/notifications.md

@@ -23,6 +23,11 @@ Escalation delivery remains independently targetable per level, allowing an
 Apprise/ntfy first notification to escalate through email, or the reverse.
 Unknown and absent persisted targets preserve historical all-destination
 behavior, and destination tag filters still apply after target selection.
+Grouping is an explicit runtime policy: `grouping.enabled=false` or a zero
+window delivers each alert independently, and disabling grouping flushes any
+pending alerts as individual deliveries. Grouped provider payloads must retain
+every alert, while live ntfy firing deliveries and webhook tests share the same
+severity-derived title, priority, and tags.
 
 ## Canonical Files
 
@@ -258,6 +263,12 @@ cooldown is disabled or still active; scheduled escalation delivery is the
 explicit exception and must route through the dedicated escalation send path so
 the alert schedule, not transport cooldown, controls escalation cadence and
 channel targeting.
+The grouping timer is also notification-owned delivery state. Live alert
+configuration must apply enabled, window, node, and guest grouping fields as
+one policy update. Turning grouping off must stop the timer and deliver every
+already-pending alert separately; service templates must be rendered only
+after the grouped summary contains every alert, so provider-specific payloads
+cannot silently collapse to the first alert.
 
 `internal/api/notifications.go` and
 `frontend-modern/src/api/notifications.ts` are shared boundaries with

internal/alerts/alerts_test.go

@@ -4568,6 +4568,64 @@ func TestReevaluateClearsDockerContainerAlertWhenOverrideDisabled(t *testing.T)
 	}
 }
 
+func TestUpdateConfigPreservesAcknowledgedProviderIncident(t *testing.T) {
+	m := newTestManager(t)
+
+	resourceID := "truenas:system-1:app-container:plex"
+	alertID := resourceID + "::resource-incident"
+	ackTime := time.Now().Add(-time.Minute).UTC()
+	alert := &Alert{
+		ID:           alertID,
+		Type:         "resource-incident",
+		Level:        AlertLevelCritical,
+		ResourceID:   resourceID,
+		ResourceName: "plex",
+		Node:         "truenas-1",
+		Instance:     "TrueNAS",
+		Message:      "Container permissions need attention",
+		StartTime:    time.Now().Add(-10 * time.Minute),
+		LastSeen:     time.Now(),
+		Acknowledged: true,
+		AckTime:      &ackTime,
+		AckUser:      "operator",
+		Metadata: map[string]interface{}{
+			"resourceType": "app-container",
+			"provider":     "truenas",
+		},
+	}
+
+	resolved := make(chan string, 1)
+	m.SetResolvedCallback(func(id string) {
+		resolved <- id
+	})
+
+	m.mu.Lock()
+	m.setActiveAlertNoLock(alertID, alert)
+	m.mu.Unlock()
+
+	config := m.GetConfig()
+	config.Schedule.Cooldown++
+	m.UpdateConfig(config)
+
+	m.mu.RLock()
+	_, retained := testLookupActiveAlert(t, m, alertID)
+	retainedAlert := m.activeAlerts[alertID]
+	m.mu.RUnlock()
+
+	if !retained {
+		t.Fatalf("provider incident was resolved by an unrelated config save")
+	}
+	if retainedAlert == nil || !retainedAlert.Acknowledged || retainedAlert.AckUser != "operator" || retainedAlert.AckTime == nil || !retainedAlert.AckTime.Equal(ackTime) {
+		t.Fatalf("acknowledgement state was not preserved: %#v", retainedAlert)
+	}
+
+	select {
+	case id := <-resolved:
+		t.Fatalf("resolved callback fired for retained provider incident %q", id)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
 func TestReevaluateClearsDockerContainerAlertWhenIgnoredPrefixAdded(t *testing.T) {
 	m := newTestManager(t)
 

internal/alerts/config_runtime.go

@@ -150,10 +150,14 @@ func (m *Manager) applyGlobalOfflineSettingsLocked() {
 	if m.config.DisableAllNodesOffline {
 		var nodeAlerts []string
 		for storageKey, alert := range m.activeAlerts {
-			if alert != nil && alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) {
-				if resourceType, _ := alert.Metadata["resourceType"].(string); resourceType == "node" {
-					nodeAlerts = append(nodeAlerts, storageKey)
-				}
+			if alert == nil {
+				continue
+			}
+			resourceType, _ := alert.Metadata["resourceType"].(string)
+			isCanonicalNodeConnectivity := alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) && resourceType == "node"
+			isLegacyNodeConnectivity := alert.Type == "connectivity" && (strings.HasPrefix(alert.ID, "node-offline-") || strings.HasPrefix(storageKey, "node-offline-"))
+			if isCanonicalNodeConnectivity || isLegacyNodeConnectivity {
+				nodeAlerts = append(nodeAlerts, storageKey)
 			}
 		}
 		for _, alertID := range nodeAlerts {
@@ -322,6 +326,15 @@ func alertResourceTypeKeysContain(keys []string, target string) bool {
 	return false
 }
 
+func isConfigReevaluatedMetricType(metricType string) bool {
+	switch metricType {
+	case "cpu", "memory", "disk", "diskRead", "diskWrite", "networkIn", "networkOut", "temperature", "usage":
+		return true
+	default:
+		return false
+	}
+}
+
 // reevaluateActiveAlertsLocked re-evaluates all active alerts against the current configuration.
 // This should only be called with m.mu already locked.
 func (m *Manager) reevaluateActiveAlertsLocked() {
@@ -524,6 +537,14 @@ func (m *Manager) reevaluateActiveAlertsLocked() {
 			threshold = getThresholdForMetric(guestThresholds, metricType)
 		}
 
+		// Provider-owned incidents and other non-metric alerts are not threshold
+		// observations. An unrelated config save must leave their lifecycle and
+		// acknowledgement state intact; only their evaluator (or an explicit
+		// resource-disable policy handled above) can resolve them.
+		if !isConfigReevaluatedMetricType(metricType) {
+			continue
+		}
+
 		if threshold == nil || threshold.Trigger <= 0 {
 			alertsToResolve = append(alertsToResolve, alertID)
 			continue

internal/api/alerts.go

@@ -211,8 +211,9 @@ func (h *AlertHandlers) UpdateAlertConfig(w http.ResponseWriter, r *http.Request
 	notificationMgr := h.getMonitor(r.Context()).GetNotificationManager()
 	notificationMgr.SetEnabled(updatedConfig.Enabled && updatedConfig.ActivationState == alerts.ActivationActive)
 	notificationMgr.SetCooldown(updatedConfig.Schedule.Cooldown)
-	notificationMgr.SetGroupingWindow(updatedConfig.Schedule.Grouping.Window)
-	notificationMgr.SetGroupingOptions(
+	notificationMgr.SetGroupingConfig(
+		updatedConfig.Schedule.Grouping.Enabled,
+		updatedConfig.Schedule.Grouping.Window,
 		updatedConfig.Schedule.Grouping.ByNode,
 		updatedConfig.Schedule.Grouping.ByGuest,
 	)

internal/monitoring/monitor.go

@@ -1739,8 +1739,9 @@ func New(cfg *config.Config) (*Monitor, error) {
 		// Apply schedule settings to notification manager
 		m.notificationMgr.SetEnabled(alertConfig.Enabled && alertConfig.ActivationState == alerts.ActivationActive)
 		m.notificationMgr.SetCooldown(alertConfig.Schedule.Cooldown)
-		m.notificationMgr.SetGroupingWindow(alertConfig.Schedule.Grouping.Window)
-		m.notificationMgr.SetGroupingOptions(
+		m.notificationMgr.SetGroupingConfig(
+			alertConfig.Schedule.Grouping.Enabled,
+			alertConfig.Schedule.Grouping.Window,
 			alertConfig.Schedule.Grouping.ByNode,
 			alertConfig.Schedule.Grouping.ByGuest,
 		)

internal/notifications/notifications.go

@@ -241,6 +241,7 @@ type NotificationManager struct {
 	lastNotified       map[string]notificationRecord
 	deliveryReceipts   map[string]struct{}
 	groupWindow        time.Duration
+	groupingEnabled    bool
 	pendingAlerts      []*alerts.Alert
 	groupTimer         *time.Timer
 	groupByNode        bool
@@ -726,6 +727,7 @@ func NewNotificationManagerWithDataDir(publicURL string, dataDir string) *Notifi
 			APIKeyHeader:   "X-API-KEY",
 		},
 		groupWindow:       30 * time.Second,
+		groupingEnabled:   true,
 		tenantID:          strings.TrimSpace(os.Getenv("PULSE_TENANT_ID")),
 		tenantName:        strings.TrimSpace(os.Getenv("PULSE_TENANT_NAME")),
 		pendingAlerts:     make([]*alerts.Alert, 0),
@@ -925,6 +927,54 @@ func (n *NotificationManager) SetGroupingWindow(seconds int) {
 	log.Info().Int("seconds", seconds).Msg("updated notification grouping window")
 }
 
+// SetGroupingConfig atomically applies the complete grouping policy. Disabling
+// grouping (or selecting a zero-second window) flushes already-pending alerts
+// as individual deliveries so a live config change cannot merge or lose them.
+func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNode, byGuest bool) {
+	if seconds < 0 {
+		seconds = 0
+	}
+
+	n.mu.Lock()
+	n.groupingEnabled = enabled
+	n.groupWindow = time.Duration(seconds) * time.Second
+	n.groupByNode = byNode
+	n.groupByGuest = byGuest
+
+	var pending []*alerts.Alert
+	var emailConfig EmailConfig
+	var webhooks []WebhookConfig
+	var appriseConfig AppriseConfig
+	var initialTarget notificationDeliveryTarget
+	var queue *NotificationQueue
+	if !enabled || seconds == 0 {
+		pending = append(pending, n.pendingAlerts...)
+		n.pendingAlerts = n.pendingAlerts[:0]
+		if n.groupTimer != nil {
+			n.groupTimer.Stop()
+			n.groupTimer = nil
+		}
+		emailConfig = copyEmailConfig(n.emailConfig)
+		webhooks = copyWebhookConfigs(n.webhooks)
+		appriseConfig = copyAppriseConfig(n.appriseConfig)
+		initialTarget = n.initialTarget
+		queue = n.queue
+	}
+	n.mu.Unlock()
+
+	for _, alert := range pending {
+		n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, initialTarget, queue)
+	}
+
+	log.Info().
+		Bool("enabled", enabled).
+		Int("seconds", seconds).
+		Bool("byNode", byNode).
+		Bool("byGuest", byGuest).
+		Int("flushedIndividually", len(pending)).
+		Msg("updated notification grouping configuration")
+}
+
 // SetGroupingOptions updates grouping options
 func (n *NotificationManager) SetGroupingOptions(byNode, byGuest bool) {
 	n.mu.Lock()
@@ -1109,32 +1159,14 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp
 		return
 	}
 
-	if options.immediate {
+	if options.immediate || !n.groupingEnabled || n.groupWindow <= 0 {
 		emailConfig := copyEmailConfig(n.emailConfig)
 		webhooks := copyWebhookConfigs(n.webhooks)
 		appriseConfig := copyAppriseConfig(n.appriseConfig)
 		queue := n.queue
 		n.mu.Unlock()
 
-		alertsToSend := []*alerts.Alert{alert}
-		jobs := buildNotificationDeliveryJobsForTarget(
-			emailConfig,
-			webhooks,
-			appriseConfig,
-			alertsToSend,
-			eventAlert,
-			time.Time{},
-			options.target,
-		)
-		if len(jobs) == 0 {
-			n.markAlertsNotified(alertsToSend, time.Now())
-			return
-		}
-		if queue != nil {
-			n.enqueueNotificationJobs(queue, jobs)
-		} else {
-			n.dispatchNotificationJobsAsync(jobs)
-		}
+		n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, options.target, queue)
 		return
 	}
 
@@ -1180,6 +1212,34 @@ func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, s
 	n.mu.Unlock()
 }
 
+func (n *NotificationManager) dispatchFiringAlerts(
+	emailConfig EmailConfig,
+	webhooks []WebhookConfig,
+	appriseConfig AppriseConfig,
+	alertsToSend []*alerts.Alert,
+	target notificationDeliveryTarget,
+	queue *NotificationQueue,
+) {
+	jobs := buildNotificationDeliveryJobsForTarget(
+		emailConfig,
+		webhooks,
+		appriseConfig,
+		alertsToSend,
+		eventAlert,
+		time.Time{},
+		target,
+	)
+	if len(jobs) == 0 {
+		n.markAlertsNotified(alertsToSend, time.Now())
+		return
+	}
+	if queue != nil {
+		n.enqueueNotificationJobs(queue, jobs)
+		return
+	}
+	n.dispatchNotificationJobsAsync(jobs)
+}
+
 func notificationDeliveryDestinationKey(job notificationDeliveryJob) string {
 	var identity string
 	switch job.Type {
@@ -1467,30 +1527,7 @@ func (n *NotificationManager) sendGroupedAlerts() {
 	queue := n.queue
 	n.mu.Unlock()
 
-	jobs := buildNotificationDeliveryJobsForTarget(
-		emailConfig,
-		webhooks,
-		appriseConfig,
-		alertsToSend,
-		eventAlert,
-		time.Time{},
-		initialTarget,
-	)
-	if len(jobs) == 0 {
-		// Preserve cooldown semantics when notifications are globally enabled
-		// but no destination is configured. Delivery receipts remain empty, so
-		// a later recovery is still correctly suppressed.
-		n.markAlertsNotified(alertsToSend, time.Now())
-		return
-	}
-
-	// Use persistent queue if available, otherwise send directly
-	if queue != nil {
-		n.enqueueNotificationJobs(queue, jobs)
-		// Note: Cooldown will be marked after successful dequeue and send
-	} else {
-		n.dispatchNotificationJobsAsync(jobs)
-	}
+	n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, alertsToSend, initialTarget, queue)
 }
 
 func buildNotificationDeliveryJobs(emailConfig EmailConfig, webhooks []WebhookConfig, appriseConfig AppriseConfig, alertsToSend []*alerts.Alert, event notificationEvent, resolvedAt time.Time) []notificationDeliveryJob {
@@ -2413,6 +2450,55 @@ func (n *NotificationManager) renderWebhookPayloadJSON(webhook WebhookConfig, da
 	return fallback()
 }
 
+func withNtfyAlertHeaders(webhook WebhookConfig, alertList []*alerts.Alert) WebhookConfig {
+	webhook = copyWebhookConfig(webhook)
+	if webhook.Headers == nil {
+		webhook.Headers = make(map[string]string)
+	}
+
+	level := alerts.AlertLevelWarning
+	var primary *alerts.Alert
+	for _, alert := range alertList {
+		if alert == nil {
+			continue
+		}
+		if primary == nil {
+			primary = alert
+		}
+		if alert.Level == alerts.AlertLevelCritical {
+			level = alerts.AlertLevelCritical
+			break
+		}
+	}
+
+	levelLabel := "WARNING"
+	priority := "high"
+	severityTag := "warning"
+	if level == alerts.AlertLevelCritical {
+		levelLabel = "CRITICAL"
+		priority = "urgent"
+		severityTag = "rotating_light"
+	}
+
+	titleSubject := fmt.Sprintf("%d alerts", len(alertList))
+	typeTag := "grouped"
+	if len(alertList) == 1 && primary != nil {
+		titleSubject = primary.ResourceName
+		if primary.Node != "" {
+			titleSubject += " on " + primary.Node
+		}
+		if primary.Type != "" {
+			typeTag = primary.Type
+		}
+	}
+
+	webhook.Headers["Content-Type"] = "text/plain"
+	webhook.Headers["Title"] = fmt.Sprintf("%s: %s", levelLabel, titleSubject)
+	webhook.Headers["Priority"] = priority
+	webhook.Headers["Tags"] = fmt.Sprintf("%s,pulse,%s", severityTag, typeTag)
+	return webhook
+}
+
 // sendGroupedWebhook sends a grouped webhook notification
 func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertList []*alerts.Alert) error {
 	if len(alertList) == 0 {
@@ -2424,57 +2510,44 @@ func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertLis
 	originalPrimary := alertList[0]
 	alertCopy := *originalPrimary
 	primaryAlert := &alertCopy
+	usesRenderedAlert := strings.TrimSpace(webhook.Template) != "" || (webhook.Service != "" && webhook.Service != "generic")
+	if usesRenderedAlert && len(alertList) > 1 {
+		otherAlerts := make([]string, 0, len(alertList)-1)
+		for i := 1; i < len(alertList); i++ {
+			if alertList[i] == nil {
+				continue
+			}
+			alertLabel := alertList[i].ResourceName
+			if alertList[i].Node != "" {
+				alertLabel += " on " + alertList[i].Node
+			}
+			if alertList[i].Message != "" {
+				otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %s", alertLabel, alertList[i].Message))
+			} else {
+				otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertLabel, alertList[i].Value))
+			}
+		}
+		if len(otherAlerts) > 0 {
+			if webhook.Service == "discord" {
+				primaryAlert.Message = fmt.Sprintf("%s | %d alerts: %s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, ", "))
+			} else if strings.TrimSpace(webhook.Template) != "" {
+				primaryAlert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, "\\n"))
+			} else {
+				primaryAlert.Message = fmt.Sprintf("%s\n\nAll %d alerts:\n%s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, "\n"))
+			}
+		}
+	}
+
+	// Prepare template data only after enriching the primary alert. Preparing it
+	// first captured the original message and silently omitted grouped alerts
+	// from built-in service payloads.
 	customFields := convertWebhookCustomFields(webhook.CustomFields)
 	data := n.prepareWebhookData(primaryAlert, customFields)
 	data.AlertCount = len(alertList)
 	data.Alerts = alertList
 	data.Mention = webhook.Mention
-
-	// Check if webhook has a custom template first
-	// Only use custom template if it's not empty
-	if webhook.Template != "" && strings.TrimSpace(webhook.Template) != "" && len(alertList) > 0 {
-		// Use custom template with enhanced message for grouped alerts
-		alert := primaryAlert
-		if len(alertList) > 1 {
-			// Build a full list of all alerts
-			summary := alert.Message
-			otherAlerts := []string{}
-			for i := 1; i < len(alertList); i++ { // Show ALL alerts
-				otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertList[i].ResourceName, alertList[i].Value))
-			}
-			if len(otherAlerts) > 0 {
-				// For custom templates, we need to escape newlines since they're likely
-				// used in shell commands or other contexts that need escaping
-				alert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", summary, len(alertList), strings.Join(otherAlerts, "\\n"))
-			}
-		}
-	}
-
-	if webhook.Service != "" && webhook.Service != "generic" && len(alertList) > 0 {
-		// For service-specific webhooks, use the first alert with a note about others
-		// For simplicity, send the first alert with a note about others
-		// Most webhook services work better with single structured payloads
-		alert := primaryAlert
-
-		// Modify message if multiple alerts - but format differently for Discord
-		if len(alertList) > 1 {
-			summary := alert.Message
-			otherAlerts := []string{}
-			for i := 1; i < len(alertList); i++ {
-				otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertList[i].ResourceName, alertList[i].Value))
-			}
-			if len(otherAlerts) > 0 {
-				// For Discord, format as a single line list to avoid newline issues
-				// Discord embeds don't render \n in description anyway
-				if webhook.Service == "discord" {
-					// Use comma-separated list for Discord
-					alert.Message = fmt.Sprintf("%s | %d alerts: %s", summary, len(alertList), strings.Join(otherAlerts, ", "))
-				} else {
-					// For other services, escape newlines properly
-					alert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", summary, len(alertList), strings.Join(otherAlerts, "\\n"))
-				}
-			}
-		}
+	if webhook.Service == "ntfy" {
+		webhook = withNtfyAlertHeaders(webhook, alertList)
 	}
 
 	var err error

internal/notifications/notifications_additional_test.go

@@ -129,6 +129,92 @@ func TestSendGroupedWebhookGeneric(t *testing.T) {
 	}
 }
 
+func TestSendGroupedWebhookNtfyIncludesLiveMetadataAndEveryAlert(t *testing.T) {
+	var gotTitle string
+	var gotPriority string
+	var gotTags string
+	var gotAuthorization string
+	var gotBody string
+
+	server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotTitle = r.Header.Get("Title")
+		gotPriority = r.Header.Get("Priority")
+		gotTags = r.Header.Get("Tags")
+		gotAuthorization = r.Header.Get("Authorization")
+		body, _ := io.ReadAll(r.Body)
+		gotBody = string(body)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	manager := NewNotificationManager("https://pulse.example")
+	defer manager.Stop()
+	manager.webhookClient = server.Client()
+	if err := manager.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+		t.Fatalf("allowlist: %v", err)
+	}
+
+	alertList := []*alerts.Alert{
+		{
+			ID:           "alert-critical",
+			Type:         "cpu",
+			Level:        alerts.AlertLevelCritical,
+			ResourceID:   "vm-101",
+			ResourceName: "database",
+			Node:         "node-a",
+			Message:      "CPU is critical",
+			Value:        99,
+			Threshold:    90,
+			StartTime:    time.Now().Add(-2 * time.Minute),
+		},
+		{
+			ID:           "alert-warning",
+			Type:         "memory",
+			Level:        alerts.AlertLevelWarning,
+			ResourceID:   "vm-102",
+			ResourceName: "cache",
+			Node:         "node-b",
+			Message:      "Memory is high",
+			Value:        88,
+			Threshold:    80,
+			StartTime:    time.Now().Add(-time.Minute),
+		},
+	}
+
+	webhook := WebhookConfig{
+		Name:    "ntfy-live",
+		URL:     server.URL + "/topic",
+		Enabled: true,
+		Service: "ntfy",
+		Headers: map[string]string{"Authorization": "Bearer secret"},
+	}
+
+	if err := manager.sendGroupedWebhook(webhook, alertList); err != nil {
+		t.Fatalf("sendGroupedWebhook error: %v", err)
+	}
+
+	if gotTitle != "CRITICAL: 2 alerts" {
+		t.Fatalf("Title = %q, want %q", gotTitle, "CRITICAL: 2 alerts")
+	}
+	if gotPriority != "urgent" {
+		t.Fatalf("Priority = %q, want urgent", gotPriority)
+	}
+	if !strings.Contains(gotTags, "rotating_light") || !strings.Contains(gotTags, "pulse") {
+		t.Fatalf("Tags = %q, want critical Pulse tags", gotTags)
+	}
+	if gotAuthorization != "Bearer secret" {
+		t.Fatalf("Authorization = %q, want configured header preserved", gotAuthorization)
+	}
+	for _, resourceName := range []string{"database", "cache"} {
+		if !strings.Contains(gotBody, resourceName) {
+			t.Fatalf("ntfy body %q does not include alert resource %q", gotBody, resourceName)
+		}
+	}
+	if !strings.Contains(gotBody, "Memory is high") {
+		t.Fatalf("ntfy body %q does not include the second alert message", gotBody)
+	}
+}
+
 func TestSendGroupedWebhookDiscordEscapesSpecialCharacters(t *testing.T) {
 	var gotBody []byte
 

internal/notifications/notifications_test.go

@@ -237,6 +237,84 @@ func TestSetGroupingWindowClampsNegativeValues(t *testing.T) {
 	nm.mu.RUnlock()
 }
 
+func TestGroupingDisabledFlushesPendingAndDeliversIndividually(t *testing.T) {
+	t.Setenv("PULSE_DATA_DIR", t.TempDir())
+
+	originalSpawn := spawnAsync
+	spawnAsync = func(f func()) { f() }
+	t.Cleanup(func() { spawnAsync = originalSpawn })
+
+	var requestMu sync.Mutex
+	requestCount := 0
+	server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestMu.Lock()
+		requestCount++
+		requestMu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	nm := NewNotificationManager("")
+	defer nm.Stop()
+	nm.webhookClient = server.Client()
+	if err := nm.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+		t.Fatalf("allowlist: %v", err)
+	}
+	nm.mu.Lock()
+	if nm.queue != nil {
+		_ = nm.queue.Stop()
+		nm.queue = nil
+	}
+	nm.mu.Unlock()
+	nm.AddWebhook(WebhookConfig{Name: "capture", URL: server.URL, Enabled: true})
+	nm.SetGroupingConfig(true, 3600, true, false)
+
+	newAlert := func(id string) *alerts.Alert {
+		return &alerts.Alert{
+			ID:           id,
+			Type:         "cpu",
+			Level:        alerts.AlertLevelWarning,
+			ResourceID:   id,
+			ResourceName: id,
+			Message:      "CPU is high",
+			Value:        90,
+			Threshold:    80,
+			StartTime:    time.Now(),
+		}
+	}
+
+	nm.SendAlert(newAlert("vm-1"))
+	nm.SendAlert(newAlert("vm-2"))
+	requestMu.Lock()
+	if requestCount != 0 {
+		requestMu.Unlock()
+		t.Fatalf("grouped alerts delivered before the grouping window elapsed: %d", requestCount)
+	}
+	requestMu.Unlock()
+
+	nm.SetGroupingConfig(false, 30, true, false)
+	requestMu.Lock()
+	if requestCount != 2 {
+		requestMu.Unlock()
+		t.Fatalf("disabling grouping delivered %d requests, want 2 individual requests", requestCount)
+	}
+	requestMu.Unlock()
+
+	nm.mu.RLock()
+	pendingCount := len(nm.pendingAlerts)
+	nm.mu.RUnlock()
+	if pendingCount != 0 {
+		t.Fatalf("pending alerts after disabling grouping = %d, want 0", pendingCount)
+	}
+
+	nm.SendAlert(newAlert("vm-3"))
+	requestMu.Lock()
+	defer requestMu.Unlock()
+	if requestCount != 3 {
+		t.Fatalf("alert sent with grouping disabled produced %d total requests, want 3", requestCount)
+	}
+}
+
 func TestSendGroupedAppriseInvokesExecutor(t *testing.T) {
 	t.Setenv("PULSE_DATA_DIR", t.TempDir())
 	nm := NewNotificationManager("")

internal/notifications/webhook_enhanced.go

@@ -568,57 +568,8 @@ func (n *NotificationManager) TestEnhancedWebhook(webhook EnhancedWebhookConfig)
 	}
 
 	if webhook.Service == "ntfy" {
-		headers := make(map[string]string, len(webhook.Headers)+4)
-		for key, value := range webhook.Headers {
-			if !strings.Contains(value, "{{") {
-				headers[key] = value
-			}
-		}
-		webhook.Headers = headers
-		webhook.Headers["Content-Type"] = "text/plain"
-
-		// Set dynamic headers based on alert level
-		title := fmt.Sprintf("%s: %s",
-			func() string {
-				switch testAlert.Level {
-				case alerts.AlertLevelCritical:
-					return "CRITICAL"
-				case alerts.AlertLevelWarning:
-					return "WARNING"
-				default:
-					return "INFO"
-				}
-			}(),
-			testAlert.ResourceName,
-		)
-		webhook.Headers["Title"] = title
-
-		priority := func() string {
-			switch testAlert.Level {
-			case alerts.AlertLevelCritical:
-				return "urgent"
-			case alerts.AlertLevelWarning:
-				return "high"
-			default:
-				return "default"
-			}
-		}()
-		webhook.Headers["Priority"] = priority
-
-		tags := fmt.Sprintf("%s,pulse,%s",
-			func() string {
-				switch testAlert.Level {
-				case alerts.AlertLevelCritical:
-					return "rotating_light"
-				case alerts.AlertLevelWarning:
-					return "warning"
-				default:
-					return "white_check_mark"
-				}
-			}(),
-			testAlert.Type,
-		)
-		webhook.Headers["Tags"] = tags
+		canonical := withNtfyAlertHeaders(canonicalWebhookConfigForEnhanced(webhook), []*alerts.Alert{testAlert})
+		webhook.Headers = canonical.Headers
 	}
 
 	result, err := n.executeEnhancedWebhookRequest(webhook, payload, WebhookTestTimeout, "Pulse-Monitoring/2.0 (Test)", webhookEventID(testAlert.ID, "alert"))