Integrate reviewed connection alert confirmation repair

Preserve exact Core candidate c46e01c92b. Retain its complete source-y5x_es_d alerts race, vet and package-build proof; the integrated alerts subtree and module manifests match that tested candidate. Batch-start upstream is already preserved in the launch head. The unrelated Machines browser-receipt mismatch remains with Web request 112189e5-998f-4412-a3ab-b7896773f67c, not cleared by this backend-only integration.

Change-source: pulse-maintainer

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

@@ -15,6 +15,31 @@
 
 ## Purpose
 
+### Connection confirmations cannot span unknown or disabled observations
+
+The existing PVE/PBS/PMG/vSphere/TrueNAS connection-degraded detector requires
+three consecutive degraded observations to activate and three consecutive
+healthy observations to recover. Pending or unknown connection state is neither
+degradation nor recovery: it drops pending activation and intent grace, restarts
+recovery confirmation, and retains an active occurrence's identity, severity,
+acknowledgement, trusted metadata and observation time without fire/resolve
+callbacks. Paused connections, disabled connections and owning-resource offline
+policy disablement clear active alerts immediately and reset pending activation
+even when no active alert exists. Healthy evidence also ends pending intent grace.
+
+Resets address exactly the connection's canonical discrete-state key, not another
+connection, condition or metric. Core and optional shadow reducer observe the same
+reset. Intent bookkeeping and its persisted checkpoint lose the interrupted run;
+re-enabled detection must establish fresh confirmations and any explicit grace.
+No new alert family, threshold, grace policy, route or external probe is added.
+
+`connection_confirmation_gap_test.go` covers all five platforms, six interruption
+paths, exact-connection isolation, fresh activation/grace, acknowledged recovery
+and exactly-once callbacks with shadow parity. `TestInterruptDiscreteRun` in
+`reducer/discrete_test.go` pins exact-key isolation, retained firing state,
+acknowledgement, no fabricated resolution and idempotent interruption. These
+synthetic source checks do not establish native outage or destination delivery.
+
 ### Retained guest observations are not alert evidence
 
 Guest memory marked `last-known`, `unavailable` or an unknown observation state

internal/alerts/connection.go

@@ -155,11 +155,31 @@ func (m *Manager) suppressConnectionDegradedAlert(snap ConnectionSnapshot) {
 	alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey)
 
 	m.mu.Lock()
+	m.interruptConnectionDegradedRunNoLock(snap.ID)
 	m.mu.Unlock()
 
 	m.clearAlert(alertID)
 }
 
+// interruptConnectionDegradedRunNoLock ends confirmation and intent runs for
+// this connection only. Unknown evidence is neither degradation nor recovery:
+// keep any firing occurrence, but require fresh consecutive healthy polls.
+// Callers hold m.mu.
+func (m *Manager) interruptConnectionDegradedRunNoLock(resourceID string) {
+	specKey := canonicalDiscreteStateSpecID(resourceID, connectionDegradedStateKey)
+	for _, state := range m.mirrorStatesNoLock() {
+		state.InterruptDiscreteRun(resourceID, specKey)
+	}
+	m.clearConnectionDegradedIntentNoLock(resourceID)
+}
+
+func (m *Manager) clearConnectionDegradedIntentNoLock(resourceID string) {
+	trackingKey := canonicalDiscreteStateStateID(resourceID, connectionDegradedStateKey)
+	if m.clearIntentPendingNoLock(trackingKey) {
+		m.saveActiveAlertsAsync("connection confirmation interrupted")
+	}
+}
+
 // CheckConnection raises or clears the connection-degraded alert for one
 // platform connection. Severity scales with observed state: stale → warning,
 // unreachable / unauthorized → critical. State=active runs through the
@@ -187,10 +207,10 @@ func (m *Manager) CheckConnection(snap ConnectionSnapshot) {
 	case ConnectionStateStale, ConnectionStateUnreachable, ConnectionStateUnauthorized:
 		// fall through and fire
 	default:
-		// pending/unknown — no alert, but reset the consecutive count so a
-		// later degraded run starts from zero instead of accumulating across
-		// a transient pending blip.
+		// Pending/unknown breaks both confirmation runs without resolving an
+		// existing outage or carrying activation grace across the gap.
 		m.mu.Lock()
+		m.interruptConnectionDegradedRunNoLock(snap.ID)
 		m.mu.Unlock()
 		return
 	}
@@ -200,11 +220,8 @@ func (m *Manager) CheckConnection(snap ConnectionSnapshot) {
 		severity = AlertLevelCritical
 	}
 
-	// Another degraded observation invalidates any in-flight recovery
-	// confirmations — recovery must build back up from zero.
+	// The reducer resets recovery confirmations on degraded observations.
 	alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey)
-	m.mu.Lock()
-	m.mu.Unlock()
 
 	spec, err := buildCanonicalDiscreteStateSpec(
 		snap.ID,
@@ -310,8 +327,8 @@ func (m *Manager) clearConnectionDegradedAlert(snap ConnectionSnapshot) {
 	defer m.mu.Unlock()
 	defer m.shadowObserveRecoveryNoLock(snap.ID, specKey, alertID, offlineRecoveryConfirmationsDefault)
 
-	// Reset the legacy degraded counter; the reducer core owns the
-	// confirmation run itself.
-
+	// Healthy evidence ends pending activation grace as well as the reducer's
+	// activation count. A firing incident still uses the recovery gate below.
+	m.clearConnectionDegradedIntentNoLock(snap.ID)
 	m.resolveDiscreteRecoveryNoLock(snap.ID, specKey, alertID, offlineRecoveryConfirmationsDefault, "Connection", snap.Name, snap.ID)
 }

internal/alerts/connection_confirmation_gap_test.go

@@ -0,0 +1,201 @@
+package alerts
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func interruptConnectionSnapshot(m *Manager, snap ConnectionSnapshot, gap string) {
+	switch gap {
+	case "pending":
+		snap.State = ConnectionStatePending
+	case "unknown":
+		snap.State = ConnectionState("unknown")
+	case "healthy":
+		snap.State = ConnectionStateActive
+	case "paused":
+		snap.State = ConnectionStatePaused
+	case "disabled":
+		snap.Enabled = false
+	case "offline-policy":
+		m.mu.Lock()
+		m.config.Overrides[snap.ID] = ThresholdConfig{DisableConnectivity: true}
+		m.mu.Unlock()
+	}
+	m.CheckConnection(snap)
+	if gap == "offline-policy" {
+		m.mu.Lock()
+		delete(m.config.Overrides, snap.ID)
+		m.mu.Unlock()
+	}
+}
+
+func TestConnectionConfirmationInterrupted(t *testing.T) {
+	for _, platform := range []ConnectionType{ConnectionTypePVE, ConnectionTypePBS, ConnectionTypePMG, ConnectionTypeVMware, ConnectionTypeTrueNAS} {
+		for _, gap := range []string{"pending", "unknown", "healthy", "paused", "disabled", "offline-policy"} {
+			t.Run(string(platform)+"/"+gap, func(t *testing.T) {
+				m := newShadowFeedManager(t)
+				snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale)
+				snap.Type = platform
+				other := snap
+				other.ID = "connection-ab"
+				key := canonicalDiscreteStateSpecID(snap.ID, connectionDegradedStateKey)
+				alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey)
+				for range 2 {
+					m.CheckConnection(snap)
+					m.CheckConnection(other)
+				}
+				interruptConnectionSnapshot(m, snap, gap)
+				m.mu.RLock()
+				pending := testCoreHasIncident(m, snap.ID, key)
+				otherCount := testCoreConfirmations(m, other.ID, canonicalDiscreteStateSpecID(other.ID, connectionDegradedStateKey))
+				m.mu.RUnlock()
+				if pending {
+					t.Error("interrupted observation retained activation confirmations")
+				}
+				if otherCount != 2 {
+					t.Errorf("interruption changed another connection's count to %d", otherCount)
+				}
+				for range 2 {
+					m.CheckConnection(snap)
+					if testHasActiveAlert(t, m, alertID) {
+						t.Fatal("non-consecutive degraded observations fired an alert")
+					}
+				}
+				m.CheckConnection(snap)
+				testRequireActiveAlert(t, m, alertID)
+				m.CheckConnection(other)
+				testRequireActiveAlert(t, m, canonicalDiscreteStateStateID(other.ID, connectionDegradedStateKey))
+				if n := m.ShadowDivergences(); n != 0 {
+					t.Fatalf("shadow diverged %d times", n)
+				}
+			})
+		}
+	}
+}
+
+func TestConnectionUnknownInterruptsRecoveryWithoutResolving(t *testing.T) {
+	for _, gap := range []string{"pending", "unknown"} {
+		t.Run(gap, func(t *testing.T) {
+			m := newShadowFeedManager(t)
+			var fires, resolutions atomic.Int32
+			resolved := make(chan struct{}, 2)
+			m.mu.Lock()
+			m.config.ActivationState = ActivationActive
+			m.mu.Unlock()
+			m.SetAlertCallback(func(*Alert) { fires.Add(1) })
+			m.SetResolvedCallback(func(string) {
+				resolutions.Add(1)
+				resolved <- struct{}{}
+			})
+			snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale)
+			alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey)
+			for range 3 {
+				m.CheckConnection(snap)
+			}
+			if err := m.AcknowledgeAlert(alertID, "test-operator"); err != nil {
+				t.Fatal(err)
+			}
+			snap.State = ConnectionStateActive
+			for range 2 {
+				m.CheckConnection(snap)
+			}
+			before := testRequireActiveAlert(t, m, alertID).Clone()
+			interruptConnectionSnapshot(m, snap, gap)
+			after := testRequireActiveAlert(t, m, alertID).Clone()
+			if !reflect.DeepEqual(before, after) {
+				t.Fatal("unknown observation changed the acknowledged active occurrence")
+			}
+			for range 2 {
+				m.CheckConnection(snap)
+				if !testHasActiveAlert(t, m, alertID) {
+					t.Fatal("non-consecutive healthy observations resolved the outage")
+				}
+			}
+			if fires.Load() != 1 || resolutions.Load() != 0 {
+				t.Fatalf("unexpected callbacks: fires=%d resolutions=%d", fires.Load(), resolutions.Load())
+			}
+			m.CheckConnection(snap)
+			if testHasActiveAlert(t, m, alertID) {
+				t.Fatal("three fresh healthy observations did not resolve")
+			}
+			select {
+			case <-resolved:
+			case <-time.After(2 * time.Second):
+				t.Fatal("confirmed recovery did not dispatch its resolved callback")
+			}
+			if fires.Load() != 1 || resolutions.Load() != 1 {
+				t.Fatalf("unexpected final callbacks: fires=%d resolutions=%d", fires.Load(), resolutions.Load())
+			}
+			if n := m.ShadowDivergences(); n != 0 {
+				t.Fatalf("shadow diverged %d times", n)
+			}
+		})
+	}
+}
+
+func TestConnectionInterruptionRestartsIntentGrace(t *testing.T) {
+	for _, gap := range []string{"pending", "unknown", "healthy", "paused", "disabled", "offline-policy"} {
+		t.Run(gap, func(t *testing.T) {
+			m := newShadowFeedManager(t)
+			var tick time.Duration
+			m.mu.Lock()
+			m.intentClock = func() time.Duration { return tick }
+			m.mu.Unlock()
+			grace := 60
+			document := NewAlertIntentPolicyDocument()
+			document.Defaults[string(AlertIntentSignalOffline)] = AlertIntentRule{GraceSeconds: &grace}
+			if err := m.LoadIntentPolicies(document); err != nil {
+				t.Fatal(err)
+			}
+			snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale)
+			alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey)
+			m.CheckConnection(snap)
+			tick = 40 * time.Second
+			m.CheckConnection(snap)
+			if err := m.SaveActiveAlerts(); err != nil {
+				t.Fatal(err)
+			}
+			interruptConnectionSnapshot(m, snap, gap)
+			m.mu.RLock()
+			_, pending := m.intentPending[alertID]
+			_, ticking := m.intentRuntimeTicks[alertID]
+			m.mu.RUnlock()
+			if pending || ticking {
+				t.Error("interrupted observation retained intent grace bookkeeping")
+			}
+			if err := m.SaveActiveAlerts(); err != nil {
+				t.Fatal(err)
+			}
+			data, err := os.ReadFile(filepath.Join(m.getAlertsDir(), intentPendingFileName))
+			if err != nil {
+				t.Fatal(err)
+			}
+			var saved []IntentPendingState
+			if err := json.Unmarshal(data, &saved); err != nil {
+				t.Fatal(err)
+			}
+			if len(saved) != 0 {
+				t.Fatal("checkpoint retained interrupted intent grace for restart")
+			}
+			for _, next := range []time.Duration{70, 90, 129} {
+				tick = next * time.Second
+				m.CheckConnection(snap)
+				if testHasActiveAlert(t, m, alertID) {
+					t.Fatal("intent grace accrued across an observation gap")
+				}
+			}
+			tick = 130 * time.Second
+			m.CheckConnection(snap)
+			testRequireActiveAlert(t, m, alertID)
+			if n := m.ShadowDivergences(); n != 0 {
+				t.Fatalf("shadow diverged %d times", n)
+			}
+		})
+	}
+}

internal/alerts/reducer/discrete.go

@@ -12,6 +12,26 @@ package reducer
 
 import "time"
 
+// InterruptDiscreteRun ends the consecutive-observation evidence for one key
+// without treating missing evidence as healthy. Pending activation (including
+// intent grace) is dropped. Firing incidents keep their occurrence and
+// acknowledgement, but recovery must restart from zero. No event is emitted.
+// It reports whether a confirmation run was interrupted.
+func (s *State) InterruptDiscreteRun(resourceID, subKey string) bool {
+	key := incidentKey(resourceID, subKey)
+	incident := s.incidents[key]
+	if incident == nil {
+		return false
+	}
+	if incident.State == StatePending {
+		delete(s.incidents, key)
+		return true
+	}
+	interrupted := incident.RecoveryCount != 0
+	incident.RecoveryCount = 0
+	return interrupted
+}
+
 // DiscreteSignal is one observation of a discrete condition on a resource:
 // Matched reports whether the trigger condition was observed (offline,
 // powered-state mismatch, state in the trigger set).

internal/alerts/reducer/discrete_test.go

@@ -5,6 +5,76 @@ import (
 	"time"
 )
 
+func TestInterruptDiscreteRun(t *testing.T) {
+	t.Run("pending confirmation and grace restart for the exact key", func(t *testing.T) {
+		state := NewState()
+		rule := DiscreteRule{Confirmations: 3, Intent: &DiscreteIntent{Explicit: true, GraceSeconds: 120}}
+		signal := discreteSignalAt(true, SeverityWarning, 0)
+		state.ApplyDiscrete(signal, rule)
+		signal.ObservedAt = signal.ObservedAt.Add(time.Minute)
+		state.ApplyDiscrete(signal, rule)
+		other := signal
+		other.Key = "other-condition"
+		state.ApplyDiscrete(other, rule)
+		other.ResourceID = "node-10"
+		state.ApplyDiscrete(other, rule)
+		if !state.InterruptDiscreteRun("node-1", "connectivity") {
+			t.Fatal("pending run was not interrupted")
+		}
+		if _, ok := state.Incident("node-1", "connectivity"); ok {
+			t.Fatal("pending run survived interruption")
+		}
+		for _, id := range []string{"node-1", "node-10"} {
+			if _, ok := state.Incident(id, "other-condition"); !ok {
+				t.Fatalf("interruption removed unrelated key for %s", id)
+			}
+		}
+		if state.InterruptDiscreteRun("node-1", "connectivity") {
+			t.Fatal("absent run reported an interruption")
+		}
+		signal.ObservedAt = signal.ObservedAt.Add(time.Hour)
+		events := state.ApplyDiscrete(signal, rule)
+		incident, ok := state.Incident(signal.ResourceID, signal.Key)
+		if len(events) != 1 || events[0].Type != EventPending || !ok || incident.Confirmations != 1 || !incident.PendingSince.Equal(signal.ObservedAt) {
+			t.Fatalf("activation did not restart: events=%+v incident=%+v", events, incident)
+		}
+	})
+
+	t.Run("firing occurrence and acknowledgement survive a recovery gap", func(t *testing.T) {
+		state := NewState()
+		rule := DiscreteRule{Confirmations: 1, RecoveryConfirmations: 3}
+		signal := discreteSignalAt(true, SeverityCritical, 0)
+		state.ApplyDiscrete(signal, rule)
+		if !state.Acknowledge(signal.ResourceID, signal.Key, "test-operator", signal.ObservedAt) {
+			t.Fatal("acknowledgement failed")
+		}
+		signal.Matched = false
+		state.ApplyDiscrete(signal, rule)
+		state.ApplyDiscrete(signal, rule)
+		before, _ := state.Incident(signal.ResourceID, signal.Key)
+		if before.RecoveryCount != 2 || !state.InterruptDiscreteRun(signal.ResourceID, signal.Key) {
+			t.Fatal("recovery run was not interrupted")
+		}
+		want := before
+		want.RecoveryCount = 0
+		got, ok := state.Incident(signal.ResourceID, signal.Key)
+		if !ok || got != want {
+			t.Fatalf("interruption changed the firing occurrence: got=%+v want=%+v", got, want)
+		}
+		if len(state.resolved) != 0 || len(state.acks) != 1 || state.InterruptDiscreteRun(signal.ResourceID, signal.Key) {
+			t.Fatal("interruption manufactured resolution, lost acknowledgement or was not idempotent")
+		}
+		for range 2 {
+			if events := state.ApplyDiscrete(signal, rule); len(events) != 0 {
+				t.Fatalf("non-consecutive recovery emitted %+v", events)
+			}
+		}
+		if events := state.ApplyDiscrete(signal, rule); len(events) != 1 || events[0].Type != EventResolved {
+			t.Fatalf("fresh recovery did not resolve: %+v", events)
+		}
+	})
+}
+
 func discreteSignalAt(matched bool, severity Severity, offset time.Duration) DiscreteSignal {
 	return DiscreteSignal{
 		ResourceID: "node-1",