fix(notifications): deliver severity increases through cooldown

Treat higher delivered severity as new information for the same occurrence without bypassing manager admission, grouping or destination routing. Keep the highest delivered level across late lower-severity completions. Include the contract and observation-to-HTTP/history regressions in this commit.

Change-source: pulse-maintainer

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

@@ -433,10 +433,18 @@ pending backlog through the same canonical batch path instead of relying on a
 separate direct-send shortcut or waiting for an unrelated timer tick.
 Alert delivery cooldown is also owned at this boundary. Normal alert delivery
 must suppress duplicate sends for the same active alert occurrence when
-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.
+cooldown is disabled or still active. A manager-admitted increase above the
+highest delivered severity for that occurrence is new information and must pass
+this gate, while retaining normal destination selection and grouping. Late
+lower-severity delivery receipts must not lower that high-water mark and permit
+repeat critical sends. A new occurrence has its own severity history. This
+exception does not bypass manager acknowledgement, snooze, flapping or hourly
+budget admission, nor destination routing or rate limits. Scheduled escalation
+retains its dedicated send path so the alert schedule, not transport cooldown,
+controls escalation cadence and channel targeting.
+`TestMonitorSeverityDelivery` covers observation-to-HTTP delivery and the
+persisted delivery log for custom sensors, CPU metrics and ZFS health; callback
+success alone is not delivery evidence.
 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

internal/monitoring/monitor_severity_delivery_test.go

@@ -0,0 +1,195 @@
+package monitoring
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
+	"github.com/stretchr/testify/require"
+)
+
+// Issue 1801: exercise actual observation, manager admission, monitor callback,
+// persistent queue, HTTP destination and delivery history, not just callbacks.
+func TestMonitorSeverityDelivery(t *testing.T) {
+	for _, tc := range []struct {
+		cooldown, group int
+		policy          string
+	}{
+		{0, 0, "ready"}, {30, 0, "ready"}, {30, 1, "ready"},
+		{30, 0, "acknowledged"}, {30, 0, "snoozed"}, {30, 0, "rate-limited"},
+		{30, 0, "routing"}, {30, 0, "metric"}, {30, 0, "health"},
+	} {
+		t.Run(fmt.Sprintf("%s/cooldown%d/group%d", tc.policy, tc.cooldown, tc.group), func(t *testing.T) {
+			requests := make(chan string, 20)
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var payload struct {
+					Level string `json:"level"`
+				}
+				if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
+					t.Error(err)
+				}
+				requests <- payload.Level
+				w.WriteHeader(http.StatusOK)
+			}))
+			defer server.Close()
+			nm := notifications.NewNotificationManagerWithDataDir("", t.TempDir())
+			defer nm.Stop()
+			require.NotNil(t, nm.GetQueue())
+			nm.SetCooldown(tc.cooldown)
+			nm.SetGroupingWindow(tc.group)
+			require.NoError(t, nm.UpdateAllowedPrivateCIDRs("127.0.0.1/32"))
+			nm.AddWebhook(notifications.WebhookConfig{ID: "hook", Name: "hook", URL: server.URL, Enabled: true, Template: `{"level":"{{.Level}}"}`})
+			mgr := alerts.NewManagerWithDataDir(t.TempDir())
+			defer mgr.Stop()
+			cfg := mgr.GetConfig()
+			cfg.Enabled = true
+			cfg.ActivationState = alerts.ActivationActive
+			cfg.FlappingEnabled = false
+			cfg.Schedule.MaxAlertsHour = 100
+			// Disable fixture activation delay explicitly; absent entries default to five seconds.
+			cfg.TimeThresholds = map[string]int{"agent": 0, "storage": 0}
+			cfg.SuppressionWindow = 0
+			cfg.MetricEvaluationWindows = map[string]map[string]int{"agent": {"cpu": 0}}
+			if tc.policy == "metric" {
+				cfg.AgentDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75}
+			}
+			if tc.policy == "rate-limited" {
+				cfg.Schedule.MaxAlertsHour = 1
+			}
+			mgr.UpdateConfig(cfg)
+			monitor := &Monitor{alertManager: mgr, notificationMgr: nm}
+			mgr.SetAlertCallback(monitor.handleAlertFired)
+			host := models.Host{ID: "sensor-host", Hostname: "sensor-host", Sensors: models.HostSensorSummary{
+				Custom: []models.HostCustomSensorMetric{{ID: "probe", Name: "Probe", Status: "warning", ObservedAt: time.Now()}},
+			}}
+			alertType := "custom-sensor"
+			storage := models.Storage{ID: "pool-store", Name: "tank", Status: "online", ZFSPool: &models.ZFSPool{Name: "tank", State: "DEGRADED"}}
+			if tc.policy == "metric" {
+				host.Sensors.Custom = nil
+				host.CPUUsage = 85
+				alertType = "cpu"
+			}
+			if tc.policy == "health" {
+				alertType = "zfs-pool-state"
+			}
+			observe := func(critical bool) {
+				if tc.policy == "health" {
+					if critical {
+						storage.ZFSPool.State = "FAULTED"
+					} else {
+						storage.ZFSPool.State = "DEGRADED"
+					}
+					mgr.CheckStorage(storage)
+				} else {
+					if tc.policy == "metric" {
+						if critical {
+							host.CPUUsage = 95
+						} else {
+							host.CPUUsage = 85
+						}
+					} else {
+						if critical {
+							host.Sensors.Custom[0].Status = "critical"
+						} else {
+							host.Sensors.Custom[0].Status = "warning"
+						}
+					}
+					mgr.CheckHost(host)
+				}
+			}
+			ready := tc.policy == "ready" || tc.policy == "metric" || tc.policy == "health"
+			await := func(level string, sent int) {
+				t.Helper()
+				select {
+				case got := <-requests:
+					require.Equal(t, level, got)
+				case <-time.After(4 * time.Second):
+					t.Fatalf("missing %s destination delivery; history=%+v", level, deliveryHistory(t, nm))
+				}
+				// Wait for successful queue completion (which follows durable receipts and
+				// the cooldown marker), not merely arrival at the HTTP handler.
+				require.Eventually(t, func() bool {
+					stats, err := nm.GetQueueStats()
+					return err == nil && stats["sent"] == sent && len(deliveryHistory(t, nm)) == sent
+				}, 3*time.Second, 10*time.Millisecond)
+				history := deliveryHistory(t, nm)
+				require.Len(t, history, sent)
+				for _, row := range history {
+					require.True(t, row.Success)
+					require.Equal(t, notifications.DeliveryOutcomeSent, row.Outcome)
+				}
+			}
+			observe(false)
+			await("warning", 1)
+			var original alerts.Alert
+			for _, a := range mgr.GetActiveAlerts() {
+				if a.Type == alertType {
+					original = a
+				}
+			}
+			require.NotEmpty(t, original.ID)
+			switch tc.policy {
+			case "acknowledged":
+				require.NoError(t, mgr.AcknowledgeAlert(original.ID, "test"))
+			case "snoozed":
+				require.NoError(t, mgr.SnoozeAlert(original.ID, "test", time.Now().Add(time.Hour)))
+			case "routing":
+				nm.SetInitialNotifyTarget("email") // no email configured; must not widen to webhook
+			}
+			observe(false)
+			observe(true)
+			want := 1
+			if ready {
+				await("critical", 2)
+				want = 2
+			}
+			found := false
+			for _, a := range mgr.GetActiveAlerts() {
+				if a.ID == original.ID {
+					found = true
+					require.Equal(t, alerts.AlertLevelCritical, a.Level)
+					require.Equal(t, original.StartTime, a.StartTime)
+				}
+			}
+			require.True(t, found, "same occurrence must remain active")
+			observe(true)
+			observe(false)
+			// Also probe notification-layer repeat/downgrade suppression directly once
+			// the ready path has completed both sends.
+			if ready {
+				nm.SendAlert(&original)
+				critical := original.Clone()
+				critical.Level = alerts.AlertLevelCritical
+				nm.SendAlert(critical)
+			}
+			select {
+			case extra := <-requests:
+				t.Fatalf("unexpected repeat/suppressed delivery: %s", extra)
+			case <-time.After(1200 * time.Millisecond):
+			}
+			history := deliveryHistory(t, nm)
+			require.Len(t, history, want)
+			for _, row := range history {
+				require.Equal(t, []string{original.ID}, row.AlertIDs)
+				require.Equal(t, "webhook:hook", row.DestinationID)
+				require.Equal(t, 1, row.Attempts)
+			}
+			dlq, err := nm.GetQueue().GetDLQ(100)
+			require.NoError(t, err)
+			require.Empty(t, dlq)
+		})
+	}
+}
+
+func deliveryHistory(t *testing.T, nm *notifications.NotificationManager) []notifications.DeliveryLogEntry {
+	t.Helper()
+	rows, err := nm.GetDeliveryLog(time.Time{}, 100)
+	require.NoError(t, err)
+	return rows
+}

internal/notifications/notifications.go

@@ -550,6 +550,7 @@ func defaultAppriseExec(ctx context.Context, args []string) ([]byte, error) {
 type notificationRecord struct {
 	lastSent   time.Time
 	alertStart time.Time
+	level      alerts.AlertLevel // Highest delivered severity for this occurrence.
 }
 
 // Alert represents an alert (interface to avoid circular dependency)
@@ -1164,10 +1165,12 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp
 	}
 
 	// Check cooldown. A disabled cooldown means "do not send repeat
-	// notifications for the same active alert occurrence"; escalation uses an
-	// explicit bypass because its cadence is owned by the alert schedule.
+	// notifications for the same active alert occurrence". A higher severity is
+	// new information, not a repeat. Scheduled escalation retains its explicit
+	// bypass because its cadence is owned by the alert schedule.
 	record, exists := n.lastNotified[alert.ID]
-	if !options.bypassCooldown && exists && record.alertStart.Equal(alert.StartTime) && (n.cooldown <= 0 || time.Since(record.lastSent) < n.cooldown) {
+	severityIncreased := notificationSeverityRank(record.level) > 0 && notificationSeverityRank(alert.Level) > notificationSeverityRank(record.level)
+	if !options.bypassCooldown && !severityIncreased && exists && record.alertStart.Equal(alert.StartTime) && (n.cooldown <= 0 || time.Since(record.lastSent) < n.cooldown) {
 		elapsed := time.Since(record.lastSent)
 		remainingCooldown := time.Duration(0)
 		if n.cooldown > elapsed {
@@ -1221,6 +1224,20 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp
 	n.mu.Unlock()
 }
 
+// Unknown levels cannot opt into the severity cooldown exception.
+func notificationSeverityRank(level alerts.AlertLevel) int {
+	switch level {
+	case alerts.AlertLevelInfo:
+		return 1
+	case alerts.AlertLevelWarning:
+		return 2
+	case alerts.AlertLevelCritical:
+		return 3
+	default:
+		return 0
+	}
+}
+
 func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, sentAt time.Time) {
 	n.mu.Lock()
 	if n.lastNotified == nil {
@@ -1230,7 +1247,14 @@ func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, s
 		if alert == nil {
 			continue
 		}
+		level := alert.Level
+		// Destinations and retries can complete out of order. A late warning
+		// receipt must not re-enable critical delivery for this occurrence.
+		if previous, ok := n.lastNotified[alert.ID]; ok && previous.alertStart.Equal(alert.StartTime) && notificationSeverityRank(previous.level) > notificationSeverityRank(level) {
+			level = previous.level
+		}
 		n.lastNotified[alert.ID] = notificationRecord{
+			level:      level,
 			lastSent:   sentAt,
 			alertStart: alert.StartTime,
 		}

internal/notifications/severity_cooldown_test.go

@@ -0,0 +1,59 @@
+package notifications
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNotificationSeverityCooldown(t *testing.T) {
+	for _, cooldown := range []time.Duration{0, time.Hour} {
+		for _, tc := range []struct {
+			from, to alerts.AlertLevel
+			send     bool
+		}{
+			{alerts.AlertLevelInfo, alerts.AlertLevelWarning, true},
+			{alerts.AlertLevelWarning, alerts.AlertLevelCritical, true},
+			{alerts.AlertLevelCritical, alerts.AlertLevelCritical, false},
+			{alerts.AlertLevelCritical, alerts.AlertLevelWarning, false},
+			{alerts.AlertLevelWarning, alerts.AlertLevelWarning, false},
+			{"", alerts.AlertLevelCritical, false},
+			{alerts.AlertLevelWarning, "unknown", false},
+		} {
+			t.Run(cooldown.String()+"/"+string(tc.from)+"-"+string(tc.to), func(t *testing.T) {
+				nm := NewNotificationManagerWithDataDir("", t.TempDir())
+				defer nm.Stop()
+				nm.cooldown = cooldown
+				nm.SetGroupingWindow(3600)
+				a := &alerts.Alert{ID: "severity", StartTime: time.Now(), Level: tc.from}
+				nm.markAlertsNotified([]*alerts.Alert{a}, time.Now())
+				a.Level = tc.to
+				nm.SendAlert(a)
+				nm.mu.RLock()
+				count := len(nm.pendingAlerts)
+				nm.mu.RUnlock()
+				if tc.send {
+					require.Equal(t, 1, count)
+				} else {
+					require.Zero(t, count)
+				}
+			})
+		}
+	}
+}
+
+func TestNotificationSeverityReceiptOrder(t *testing.T) {
+	nm := NewNotificationManagerWithDataDir("", t.TempDir())
+	defer nm.Stop()
+	a := &alerts.Alert{ID: "severity", StartTime: time.Now(), Level: alerts.AlertLevelCritical}
+	nm.markAlertsNotified([]*alerts.Alert{a}, time.Now())
+	a.Level = alerts.AlertLevelWarning
+	nm.markAlertsNotified([]*alerts.Alert{a}, time.Now())
+	require.Equal(t, alerts.AlertLevelCritical, nm.lastNotified[a.ID].level)
+	// A new occurrence starts its own severity history.
+	a.StartTime = a.StartTime.Add(time.Minute)
+	nm.markAlertsNotified([]*alerts.Alert{a}, time.Now())
+	require.Equal(t, alerts.AlertLevelWarning, nm.lastNotified[a.ID].level)
+}