Integrate Core alert escalation delivery repair

Change-source: pulse-maintainer

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

@@ -2830,11 +2830,14 @@ mute or maintenance state is allowed.
 
 Only a supporting member of a `supported-cause` group suppresses duplicate
 notification delivery, with the typed `correlated_primary` reason and primary
-alert ID in the event log. The primary remains the one delivery owner.
+alert ID in the event log. The primary remains the one delivery owner. Supporting
+members also skip escalation levels and critical repeats while that supported
+cause is current; an already queued escalation is revalidated before delivery.
 `observation-set` members continue to notify independently because Pulse has
 not established causality. `incident_synthesis_test.go` pins classification,
 contradiction downgrade, bounded evidence, duplicate-delivery suppression, and
-partial-recovery behavior.
+partial-recovery behavior. `escalation_policy_change_test.go` pins escalation
+suppression and queued-callback revalidation without muting observation sets.
 
 ### Alert hydration is not resource admission
 
@@ -2996,7 +2999,8 @@ backend deletion or destination delivery.
 
 `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
+snooze, supported-cause symptom correlation, 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

internal/alerts/escalation.go

@@ -44,9 +44,9 @@ func (m *Manager) checkEscalations() {
 
 	now := m.policyNow()
 	for _, alert := range m.activeAlerts {
-		// Monitor-only alerts remain visible but must never schedule a
-		// notification, including escalation or critical repeats.
-		if alert == nil || alert.Acknowledged || isMonitorOnlyAlert(alert) {
+		// A supported symptom has a primary delivery owner. Like monitor-only
+		// alerts, it stays visible without its own escalation or critical repeat.
+		if alert == nil || alert.Acknowledged || isMonitorOnlyAlert(alert) || isSupportedInfrastructureSymptom(alert) {
 			continue
 		}
 		if _, snoozed := alertSnoozeUntil(alert, now); snoozed {
@@ -130,7 +130,7 @@ func (m *Manager) PrepareEscalationNotification(snapshot *Alert, level int) (*Al
 		return nil, EscalationLevel{}, false
 	}
 	active, ok := m.getActiveAlertNoLock(snapshot.ID)
-	if !ok || active == nil || !active.StartTime.Equal(snapshot.StartTime) || active.Acknowledged || isMonitorOnlyAlert(active) {
+	if !ok || active == nil || !active.StartTime.Equal(snapshot.StartTime) || active.Acknowledged || isMonitorOnlyAlert(active) || isSupportedInfrastructureSymptom(active) {
 		return nil, EscalationLevel{}, false
 	}
 	if _, snoozed := alertSnoozeUntil(active, m.policyNow()); snoozed {

internal/alerts/incident_synthesis_test.go

@@ -282,3 +282,86 @@ func TestSupportedInfrastructureSymptomUsesPrimaryNotification(t *testing.T) {
 		t.Fatalf("diagnosis = %+v, want correlated-primary suppression", diagnosis)
 	}
 }
+
+func TestSupportedInfrastructureSymptomDoesNotEscalateOrRepeat(t *testing.T) {
+	m := newTestManager(t)
+	now := time.Date(2026, 9, 29, 19, 0, 0, 0, time.UTC)
+	m.now = func() time.Time { return now }
+	m.mu.Lock()
+	m.config.Enabled = true
+	m.config.ActivationState = ActivationActive
+	m.config.Schedule.Escalation = EscalationConfig{
+		Enabled: true, RepeatCritical: true, RepeatEvery: 5,
+		Levels: []EscalationLevel{{After: 5, Notify: "email"}},
+	}
+	correlation := &AlertCorrelation{
+		Key: "infrastructure:host-offline", Kind: AlertCorrelationKindInfrastructureIncident,
+		Role: AlertCorrelationRoleSupporting, Reason: "Verified host dependency.",
+		Inference: AlertCorrelationInferenceSupportedCause, PrimaryAlertID: "host-offline",
+	}
+	newSymptom := &Alert{
+		ID: "new-symptom", Level: AlertLevelCritical, StartTime: now.Add(-10 * time.Minute),
+		Correlation: cloneAlertCorrelation(correlation),
+	}
+	repeatingSymptom := &Alert{
+		ID: "repeating-symptom", Level: AlertLevelCritical, StartTime: now.Add(-30 * time.Minute),
+		LastEscalation: 1, EscalationTimes: []time.Time{now.Add(-10 * time.Minute)},
+		Correlation: cloneAlertCorrelation(correlation),
+	}
+	observationOnly := &Alert{
+		ID: "observation-only", Level: AlertLevelCritical, StartTime: now.Add(-10 * time.Minute),
+		Correlation: &AlertCorrelation{
+			Key: "infrastructure:unconfirmed", Kind: AlertCorrelationKindInfrastructureIncident,
+			Role: AlertCorrelationRoleSupporting, Reason: "Coincident observations only.",
+			Inference: AlertCorrelationInferenceObservationSet, PrimaryAlertID: "host-offline",
+		},
+	}
+	m.setActiveAlertNoLock(newSymptom.ID, newSymptom)
+	m.setActiveAlertNoLock(repeatingSymptom.ID, repeatingSymptom)
+	m.setActiveAlertNoLock(observationOnly.ID, observationOnly)
+	m.mu.Unlock()
+
+	m.checkEscalations()
+	if newSymptom.LastEscalation != 0 || len(newSymptom.EscalationTimes) != 0 {
+		t.Fatalf("supported symptom scheduled an escalation: %+v", newSymptom)
+	}
+	if repeatingSymptom.LastEscalation != 1 || len(repeatingSymptom.EscalationTimes) != 1 {
+		t.Fatalf("supported symptom repeated an escalation: %+v", repeatingSymptom)
+	}
+	if observationOnly.LastEscalation != 1 || len(observationOnly.EscalationTimes) != 1 {
+		t.Fatalf("uncorroborated observation lost independent escalation: %+v", observationOnly)
+	}
+}
+
+func TestQueuedEscalationRejectsNewlySupportedInfrastructureSymptom(t *testing.T) {
+	m := newTestManager(t)
+	m.mu.Lock()
+	m.config.Enabled = true
+	m.config.ActivationState = ActivationActive
+	m.config.Schedule.Escalation = EscalationConfig{
+		Enabled: true, Levels: []EscalationLevel{{After: 5, Notify: "email"}},
+	}
+	alert := &Alert{ID: "newly-correlated", Level: AlertLevelCritical, StartTime: time.Now().Add(-10 * time.Minute)}
+	m.setActiveAlertNoLock(alert.ID, alert)
+	snapshot := cloneAlertForOutput(alert)
+	alert.Correlation = &AlertCorrelation{
+		Key: "infrastructure:host-offline", Kind: AlertCorrelationKindInfrastructureIncident,
+		Role: AlertCorrelationRoleSupporting, Reason: "Verified host dependency.",
+		Inference: AlertCorrelationInferenceSupportedCause, PrimaryAlertID: "host-offline",
+	}
+	alert.Metadata = map[string]interface{}{
+		MetadataQuietHoursSuppressed: true,
+		MetadataQuietHoursReplayAt:   time.Now().Add(time.Hour).UTC().Format(time.RFC3339),
+	}
+	m.mu.Unlock()
+
+	if _, _, eligible := m.PrepareEscalationNotification(snapshot, 1); eligible {
+		t.Fatal("queued escalation remained eligible after the symptom gained a supported primary")
+	}
+	if !m.ShouldSuppressNotification(alert) {
+		t.Fatal("supported symptom remained eligible through the public delivery helper")
+	}
+	if hasQuietHoursNotificationReplay(alert) {
+		t.Fatal("supported symptom retained a queued quiet-hours replay")
+	}
+}

internal/alerts/notification_policy.go

@@ -600,6 +600,10 @@ func (m *Manager) ShouldSuppressNotification(alert *Alert) bool {
 		clearQuietHoursNotificationReplay(alert)
 		return true
 	}
+	if isSupportedInfrastructureSymptom(alert) {
+		clearQuietHoursNotificationReplay(alert)
+		return true
+	}
 
 	if suppressed, reason := m.operatorSuppressionForAlertNoLock(alert, time.Now().UTC()); suppressed {
 		clearQuietHoursNotificationReplay(alert)