Merge pull request #2028 from rcourtman/maintainer/20260910T090007Z

Prevent stale alert escalations after alert state changes

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

@@ -2908,3 +2908,18 @@ real history hook and administration card: load a row, start a pending range
 read, confirm clear, then release the obsolete response at desktop and phone
 widths. Scripted API responses establish component behaviour, not installed
 backend deletion or destination delivery.
+
+### Escalation callback admission uses current policy and occurrence
+
+`PrepareEscalationNotification` rejects asynchronous snapshots after escalation
+or global alert disablement, inactive activation, recovery, acknowledgement,
+snooze, or replacement by another occurrence of the same alert ID. It returns
+the current alert and detached current per-level routing, not mutable manager
+state. The monitoring callback must use this admission before enqueueing.
+This check does not recall provider-accepted messages or make the subsequent
+notification enqueue atomic with alert resolution.
+
+`escalation_policy_change_test.go` pins a short-lived alert and disabled policy
+before the former 180-minute deadline, plus current payload/routing selection,
+copy isolation, snooze and invalid-level rejection. These deterministic component tests
+establish neither installed delivery timing nor the cause of a delayed email.

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

@@ -4008,3 +4008,13 @@ without a request consumer, and after acceptance, plus late completion and
 success/error propagation using channel-only fixtures without starting monitors
 or storage. This is shutdown handshake proof, not installed notification receipt
 or full server reload acceptance.
+
+### Stale escalation callback snapshots are not delivery authority
+
+`handleAlertEscalated` obtains current occurrence/policy admission from the
+alert manager before dispatch. `monitor_escalation_stale_test.go` invokes the
+callback after disablement, inactive activation, recovery, recurrence,
+and acknowledgement, requiring no queued notification. Existing routing,
+quiet-hours and cooldown tests now use real active manager occurrences rather
+than fabricated inactive alerts. This is callback-boundary proof, not a claim
+that already admitted or provider-accepted deliveries can be recalled.

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

@@ -899,3 +899,14 @@ wraps its returned error and checks both transport retry policy and the shared
 class predicate used by the queue. `TestClassFromHTTPStatus` pins the reason
 classes. This proves the classification boundary without starting queue/storage
 workers; it does not establish installed receipt or queue scheduling execution.
+
+### Initial routing edits and previously queued firing work
+
+Changing the initial target affects new sends; it does not rewrite a previously
+queued email or revoke independent per-level escalation routing. Queue delivery
+uses the admitted destination configuration while checking current global and
+destination enablement. Resolution independently cancels obsolete firing work.
+`queue_policy_change_test.go` verifies retained email routing after a webhook
+initial-target edit with an in-memory SMTP acceptance, and prevents a previously
+fetched pending item from sending after cancellation. This does not establish
+recipient inbox timing or recall mail already accepted by a provider.

internal/alerts/escalation.go

@@ -114,3 +114,28 @@ func (m *Manager) checkEscalations() {
 		m.safeCallEscalationRepeatCallback(alert, len(escalation.Levels))
 	}
 }
+
+// PrepareEscalationNotification revalidates an asynchronously dispatched
+// escalation against the current policy and occurrence. The callback snapshot
+// is not authority to send after recovery, acknowledgement or a policy edit.
+// This does not recall a notification already handed to a destination.
+func (m *Manager) PrepareEscalationNotification(snapshot *Alert, level int) (*Alert, EscalationLevel, bool) {
+	if m == nil || snapshot == nil {
+		return nil, EscalationLevel{}, false
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if !m.config.Enabled || m.config.ActivationState != ActivationActive || !m.config.Schedule.Escalation.Enabled || level <= 0 || level > len(m.config.Schedule.Escalation.Levels) {
+		return nil, EscalationLevel{}, false
+	}
+	active, ok := m.getActiveAlertNoLock(snapshot.ID)
+	if !ok || active == nil || !active.StartTime.Equal(snapshot.StartTime) || active.Acknowledged {
+		return nil, EscalationLevel{}, false
+	}
+	if _, snoozed := alertSnoozeUntil(active, m.policyNow()); snoozed {
+		return nil, EscalationLevel{}, false
+	}
+	target := m.config.Schedule.Escalation.Levels[level-1]
+	target.DestinationIDs = append([]string(nil), target.DestinationIDs...)
+	return cloneAlertForOutput(active), target, true
+}

internal/alerts/escalation_policy_change_test.go

@@ -0,0 +1,80 @@
+package alerts
+
+import (
+	"testing"
+	"time"
+)
+
+func TestEscalationPolicyChangeBeforeFormerDeadline(t *testing.T) {
+	for _, action := range []string{"disable", "resolve"} {
+		t.Run(action, func(t *testing.T) {
+			m := newTestManager(t)
+			now := time.Date(2026, 9, 10, 1, 0, 0, 0, time.UTC)
+			m.now = func() time.Time { return now }
+			cfg := m.GetConfig()
+			cfg.Enabled = true
+			cfg.ActivationState = ActivationActive
+			cfg.Schedule.Escalation = EscalationConfig{Enabled: true, Levels: []EscalationLevel{{After: 180, Notify: "email"}}}
+			m.UpdateConfig(cfg)
+			a := &Alert{ID: "short-lived", StartTime: now, Level: AlertLevelWarning}
+			m.mu.Lock()
+			m.setActiveAlertNoLock(a.ID, a)
+			m.mu.Unlock()
+			now = now.Add(time.Minute)
+			m.checkEscalations()
+			if action == "disable" {
+				cfg.Schedule.Escalation.Enabled = false
+				m.UpdateConfig(cfg)
+			} else if !m.ClearAlert(a.ID) {
+				t.Fatal("clear failed")
+			}
+			now = now.Add(180 * time.Minute)
+			m.checkEscalations()
+			// checkEscalations records admission synchronously before spawning delivery.
+			if a.LastEscalation != 0 || len(a.EscalationTimes) != 0 {
+				t.Fatalf("obsolete escalation admitted: %+v", a)
+			}
+		})
+	}
+}
+
+func TestPrepareEscalationNotificationUsesCurrentOccurrenceAndDetachedTarget(t *testing.T) {
+	m := newTestManager(t)
+	now := time.Now()
+	m.mu.Lock()
+	m.config.Enabled = true
+	m.config.ActivationState = ActivationActive
+	m.config.Schedule.Escalation = EscalationConfig{Enabled: true, Levels: []EscalationLevel{{After: 180, Notify: "email", DestinationIDs: []string{"email"}}}}
+	a := &Alert{ID: "current-escalation", Type: "cpu", ResourceID: "node/test", StartTime: now, Level: AlertLevelWarning, Value: 90}
+	m.setActiveAlertNoLock(a.ID, a)
+	snapshot := cloneAlertForOutput(a)
+	a.Value = 95
+	m.config.Schedule.Escalation.Levels[0].Notify = "webhook"
+	m.config.Schedule.Escalation.Levels[0].DestinationIDs = []string{"webhook:ops"}
+	m.mu.Unlock()
+	current, target, ok := m.PrepareEscalationNotification(snapshot, 1)
+	if !ok || current.Value != 95 || target.Notify != "webhook" || target.DestinationIDs[0] != "webhook:ops" {
+		t.Fatalf("current=%+v target=%+v ok=%v", current, target, ok)
+	}
+	current.Value = 0
+	target.DestinationIDs[0] = "mutated"
+	current, target, ok = m.PrepareEscalationNotification(snapshot, 1)
+	if !ok || current.Value != 95 || target.DestinationIDs[0] != "webhook:ops" {
+		t.Fatal("returned data aliases manager state")
+	}
+	for _, level := range []int{0, -1, 2} {
+		if _, _, ok := m.PrepareEscalationNotification(snapshot, level); ok {
+			t.Fatalf("invalid level %d admitted", level)
+		}
+	}
+	if _, _, ok := m.PrepareEscalationNotification(nil, 1); ok {
+		t.Fatal("nil snapshot admitted")
+	}
+	if err := m.SnoozeAlert(snapshot.ID, "test", time.Now().Add(time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, ok := m.PrepareEscalationNotification(snapshot, 1); ok {
+		t.Fatal("snoozed occurrence admitted")
+	}
+
+}

internal/monitoring/monitor_alert_handling_test.go

@@ -957,17 +957,7 @@ func TestMonitor_HandleAlertEscalated_QuietHoursSuppressesNotification(t *testin
 		alertManager:    mgr,
 	}
 
-	alert := &alerts.Alert{
-		ID:           "escalated-offline",
-		Type:         "connectivity",
-		Level:        alerts.AlertLevelCritical,
-		ResourceID:   "node/pve-1",
-		ResourceName: "pve-1",
-		Node:         "pve-1",
-		Instance:     "pve",
-		Message:      "Node offline",
-		StartTime:    time.Now(),
-	}
+	alert := activeEscalationFixture(t, mgr, "connectivity", alerts.AlertLevelCritical)
 
 	m.handleAlertEscalated(nil, alert, 1)
 
@@ -1013,17 +1003,7 @@ func TestMonitor_HandleAlertEscalated_SendsNotificationWhenNotSuppressed(t *test
 		alertManager:    mgr,
 	}
 
-	alert := &alerts.Alert{
-		ID:           "escalated-normal",
-		Type:         "connectivity",
-		Level:        alerts.AlertLevelCritical,
-		ResourceID:   "node/pve-1",
-		ResourceName: "pve-1",
-		Node:         "pve-1",
-		Instance:     "pve",
-		Message:      "Node offline",
-		StartTime:    time.Now(),
-	}
+	alert := activeEscalationFixture(t, mgr, "connectivity", alerts.AlertLevelCritical)
 
 	m.handleAlertEscalated(nil, alert, 1)
 
@@ -1069,17 +1049,7 @@ func TestMonitor_HandleAlertEscalated_BypassesDeliveryCooldown(t *testing.T) {
 		alertManager:    mgr,
 	}
 
-	alert := &alerts.Alert{
-		ID:           "escalated-with-cooldown",
-		Type:         "memory",
-		Level:        alerts.AlertLevelWarning,
-		ResourceID:   "vm/100",
-		ResourceName: "vm-100",
-		Node:         "pve-1",
-		Instance:     "pve",
-		Message:      "Memory threshold crossed",
-		StartTime:    time.Now().Add(-10 * time.Minute),
-	}
+	alert := activeEscalationFixture(t, mgr, "memory", alerts.AlertLevelWarning)
 
 	notifMgr.SendAlert(alert)
 	select {

internal/monitoring/monitor_alert_override_migration_test.go

@@ -62,10 +62,7 @@ func TestHandleAlertEscalatedPreservesLegacyAppriseAndExactWebhookRouting(t *tes
 		cfg.Schedule.QuietHours.Enabled = false
 		manager.UpdateConfig(cfg)
 
-		(&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, &alerts.Alert{
-			ID: "apprise-escalation", Type: "connectivity", Level: alerts.AlertLevelCritical,
-			ResourceID: "node/pve-1", ResourceName: "pve-1", StartTime: time.Now(),
-		}, 1)
+		(&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, activeEscalationFixture(t, manager, "connectivity", alerts.AlertLevelCritical), 1)
 
 		select {
 		case <-requests:
@@ -106,10 +103,7 @@ func TestHandleAlertEscalatedPreservesLegacyAppriseAndExactWebhookRouting(t *tes
 		cfg.Schedule.QuietHours.Enabled = false
 		manager.UpdateConfig(cfg)
 
-		(&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, &alerts.Alert{
-			ID: "exact-webhook-escalation", Type: "connectivity", Level: alerts.AlertLevelCritical,
-			ResourceID: "node/pve-1", ResourceName: "pve-1", StartTime: time.Now(),
-		}, 1)
+		(&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, activeEscalationFixture(t, manager, "connectivity", alerts.AlertLevelCritical), 1)
 
 		select {
 		case <-selectedRequests:

internal/monitoring/monitor_alerts.go

@@ -280,11 +280,13 @@ func (m *Monitor) handleAlertEscalated(hub *websocket.Hub, alert *alerts.Alert,
 		Int("level", level).
 		Msg("Alert escalated")
 
-	config := m.alertManager.GetConfig()
-	if level <= 0 || level > len(config.Schedule.Escalation.Levels) {
+	current, escalationLevel, eligible := m.alertManager.PrepareEscalationNotification(alert, level)
+	if !eligible {
 		return
 	}
 
+	alert = current
+
 	if m.alertManager.ShouldSuppressNotification(alert) {
 		log.Info().
 			Str("alertID", alert.ID).
@@ -295,7 +297,6 @@ func (m *Monitor) handleAlertEscalated(hub *websocket.Hub, alert *alerts.Alert,
 	}
 
 	if m.notificationMgr != nil {
-		escalationLevel := config.Schedule.Escalation.Levels[level-1]
 		if len(escalationLevel.DestinationIDs) > 0 {
 			m.notificationMgr.SendEscalatedAlertToDestinations(alert, escalationLevel.Notify, escalationLevel.DestinationIDs)
 			m.broadcastEscalatedAlert(hub, alert)

internal/monitoring/monitor_escalation_stale_test.go

@@ -0,0 +1,83 @@
+package monitoring
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
+)
+
+func activeEscalationFixture(t *testing.T, m *alerts.Manager, kind string, level alerts.AlertLevel) *alerts.Alert {
+	t.Helper()
+	t.Cleanup(m.Stop)
+	cfg := m.GetConfig()
+	cfg.Enabled = true
+	cfg.ActivationState = alerts.ActivationActive
+	cfg.Schedule.Escalation.Enabled = true
+	m.UpdateConfig(cfg)
+	m.RaiseSystemAlert(alerts.SystemAlertInput{Type: kind, Level: level, Message: "escalation fixture"})
+	active := m.GetActiveAlerts()
+	if len(active) != 1 {
+		t.Fatalf("active fixtures = %d", len(active))
+	}
+	return active[0].Clone()
+}
+
+// The callback carries a snapshot and can run after a config edit or recovery.
+// Invoke it after that boundary, rather than relying on goroutine timing.
+func TestHandleAlertEscalatedRejectsStaleAdmission(t *testing.T) {
+	for _, action := range []string{"disabled", "inactive", "alerts-disabled", "resolved", "new-occurrence", "acknowledged"} {
+		t.Run(action, func(t *testing.T) {
+			t.Setenv("PULSE_DATA_DIR", t.TempDir())
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }))
+			defer server.Close()
+			n := notifications.NewNotificationManager("https://pulse.example.test")
+			defer n.Stop()
+			if err := n.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+				t.Fatal(err)
+			}
+			n.AddWebhook(notifications.WebhookConfig{ID: "test", URL: server.URL, Enabled: true})
+			manager := alerts.NewManager()
+			cfg := manager.GetConfig()
+			cfg.Schedule.Escalation.Levels = []alerts.EscalationLevel{{After: 180, Notify: "webhook"}}
+			cfg.Schedule.QuietHours.Enabled = false
+			manager.UpdateConfig(cfg)
+			snapshot := activeEscalationFixture(t, manager, "stale-escalation", alerts.AlertLevelCritical)
+			cfg = manager.GetConfig()
+			switch action {
+			case "disabled":
+				cfg.Schedule.Escalation.Enabled = false
+				manager.UpdateConfig(cfg)
+			case "inactive":
+				cfg.ActivationState = alerts.ActivationPending
+				manager.UpdateConfig(cfg)
+			case "alerts-disabled":
+				cfg.Enabled = false
+				manager.UpdateConfig(cfg)
+			case "resolved", "new-occurrence":
+				if !manager.ClearAlert(snapshot.ID) {
+					t.Fatal("clear failed")
+				}
+				if action == "new-occurrence" {
+					activeEscalationFixture(t, manager, "stale-escalation", alerts.AlertLevelCritical)
+				}
+			case "acknowledged":
+				if err := manager.AcknowledgeAlert(snapshot.ID, "test"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			(&Monitor{alertManager: manager, notificationMgr: n}).handleAlertEscalated(nil, snapshot, 1)
+			stats, err := n.GetQueue().GetQueueStats()
+			if err != nil {
+				t.Fatal(err)
+			}
+			for status, count := range stats {
+				if count != 0 {
+					t.Errorf("stale callback queued %s=%d", status, count)
+				}
+			}
+		})
+	}
+}

internal/notifications/queue_policy_change_test.go

@@ -0,0 +1,54 @@
+package notifications
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+)
+
+// Initial routing changes apply to new sends, not already admitted work.
+// Resolution, independently, must retire that old firing work before retry.
+func TestQueuedEmailRoutingChangeAndResolution(t *testing.T) {
+	for _, resolve := range []bool{false, true} {
+		name := "still-active"
+		if resolve {
+			name = "resolved"
+		}
+		t.Run(name, func(t *testing.T) {
+			var deliveries int32
+			stubSMTPDialSuccess(t, &deliveries)
+			q, err := NewNotificationQueue(t.TempDir())
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer q.Stop()
+			n := &NotificationManager{enabled: true, queue: q, initialTarget: notificationDeliveryTargetEmail,
+				emailConfig:  EmailConfig{Enabled: true, SMTPHost: "smtp.example.com", SMTPPort: 25, From: "pulse@example.test", To: []string{"ops@example.test"}},
+				lastNotified: make(map[string]notificationRecord), deliveryReceipts: make(map[string]struct{}),
+			}
+			alert := &alerts.Alert{ID: "routing-change", ResourceName: "test-node", Level: alerts.AlertLevelWarning, StartTime: time.Now().Add(-time.Minute)}
+			n.SendAlert(alert)
+			pending, err := q.GetPending(10)
+			if err != nil || len(pending) != 1 || pending[0].Type != "email" {
+				t.Fatalf("queued email = %v, %v", pending, err)
+			}
+			old := pending[0]
+			n.SetInitialNotifyTarget("webhook")
+			// A fetched worker item must not defeat resolution cancellation.
+			if resolve {
+				n.CancelAlert(alert.ID)
+			}
+			q.SetProcessor(n.ProcessQueuedNotification)
+			q.processNotification(old)
+			want := int32(1)
+			if resolve {
+				want = 0
+			}
+			if got := atomic.LoadInt32(&deliveries); got != want {
+				t.Fatalf("SMTP acceptances = %d, want %d", got, want)
+			}
+		})
+	}
+}