Keep observed unacknowledged alerts through retention cleanup

Age-based housekeeping deleted continuing conditions when automatic acknowledgement was disabled. Require last-observation inactivity as well as occurrence age, retain legacy timestamp fallback and inactive cleanup, and document the unchanged acknowledgement/recovery boundaries. Regression exercises cached, absent and failed Docker registry reports, callback/history identity and actual checkpoint restart.

Change-source: pulse-maintainer

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

@@ -795,6 +795,26 @@ inspectability, or convert missing/stale evidence into health.
 
 ## Current State
 
+### Continuing unacknowledged alerts survive age-based cleanup
+
+`MaxAlertAgeDays` removes an unacknowledged alert only when both its occurrence
+start and last observation predate the configured limit. A still-observed
+condition keeps its occurrence identity, age and dispatch timestamp; cleanup
+must not hide it or manufacture another firing on the next poll. Legacy alerts
+without `LastSeen` use `StartTime` for inactivity. Zero disables this retention
+rule. Acknowledged-alert cleanup, automatic acknowledgement, explicit clears,
+confirmed recovery and notification eligibility remain unchanged.
+
+`TestCleanupRetentionRequiresObservationInactivity` in `alerts_test.go` covers
+continuing/inactive, legacy, disabled, acknowledged and inconsistent-timestamp
+controls. `TestCleanupContinuingDockerUpdateKeepsOccurrence` in
+`cleanup_observation_retention_test.go` exercises the real Docker update
+detector with cached, absent and failed registry observations through four
+cleanup/next-poll cycles, JSON and durable active-state restart, callback counts,
+history and affirmative recovery. This is source-fixture lifecycle proof, not
+an installed registry/destination result or a diagnosis of a reporter's daily
+resolve/reopen cycle.
+
 ### Confirmed empty storage is recovery evidence
 
 Static storage capacity evaluation must admit a zero usage observation when

internal/alerts/active_cleanup.go

@@ -73,7 +73,11 @@ func (m *Manager) Cleanup(maxAge time.Duration) {
 	if m.config.MaxAlertAgeDays > 0 {
 		alertTTL := time.Duration(m.config.MaxAlertAgeDays) * 24 * time.Hour
 		for id, alert := range m.activeAlerts {
-			if !alert.Acknowledged && now.Sub(alert.StartTime) > alertTTL {
+			// Retention is housekeeping for inactive alerts, not recovery
+			// evidence. As with acknowledged alerts, keep a condition still
+			// observed; deleting it would re-fire on the next failing poll.
+			if !alert.Acknowledged && now.Sub(alert.StartTime) > alertTTL &&
+				lastSeenTooOld(alert, alertTTL) {
 				log.Info().
 					Str("alertID", id).
 					Dur("age", now.Sub(alert.StartTime)).

internal/alerts/alerts_test.go

@@ -21639,3 +21639,53 @@ func TestAsyncActiveCheckpointBurstIsBounded(t *testing.T) {
 		t.Fatalf("restart lost latest pending intent: %+v", pending)
 	}
 }
+
+func TestCleanupRetentionRequiresObservationInactivity(t *testing.T) {
+	now := time.Now()
+	old := now.Add(-48 * time.Hour)
+	recent := now.Add(-5 * time.Minute)
+	for _, tc := range []struct {
+		name         string
+		start, seen  time.Time
+		acknowledged bool
+		ttl          int
+		keep         bool
+	}{
+		{name: "continuing unacknowledged condition", start: old, seen: recent, ttl: 1, keep: true},
+		{name: "inactive unacknowledged condition", start: old, seen: old, ttl: 1},
+		{name: "legacy inactive without last seen", start: old, ttl: 1},
+		{name: "legacy recent without last seen", start: recent, ttl: 1, keep: true},
+		{name: "recent occurrence with inconsistent old last seen", start: recent, seen: old, ttl: 1, keep: true},
+		{name: "disabled unacknowledged retention", start: old, seen: old, keep: true},
+		{name: "continuing acknowledged condition", start: old, seen: recent, acknowledged: true, ttl: 1, keep: true},
+		{name: "inactive acknowledged condition", start: old, seen: old, acknowledged: true, ttl: 1},
+		{name: "legacy without either timestamp", ttl: 1},
+		{name: "future observation does not expire", start: old, seen: now.Add(time.Hour), ttl: 1, keep: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := newCleanupRetentionManager(t, t.TempDir(), false)
+			alert := &Alert{
+				ID: "retention-resource::metric-threshold:usage", ResourceID: "retention-resource",
+				CanonicalSpecID: "metric-threshold:usage", Type: "usage", Level: AlertLevelWarning,
+				StartTime: tc.start, LastSeen: tc.seen, Acknowledged: tc.acknowledged,
+			}
+			if tc.acknowledged {
+				ackTime := old
+				alert.AckTime = &ackTime
+				alert.AckUser = "operator"
+			}
+			m.mu.Lock()
+			m.config.MaxAlertAgeDays = tc.ttl
+			m.setActiveAlertNoLock(alert.ID, alert)
+			m.mu.Unlock()
+			m.Cleanup(time.Hour)
+			active := m.GetActiveAlerts()
+			if kept := len(active) == 1; kept != tc.keep {
+				t.Fatalf("cleanup kept = %v, want %v (start %s, last seen %s)", kept, tc.keep, tc.start, tc.seen)
+			}
+			if tc.keep && (!active[0].StartTime.Equal(tc.start) || !active[0].LastSeen.Equal(tc.seen) || active[0].Acknowledged != tc.acknowledged) {
+				t.Fatalf("cleanup changed the retained occurrence: %+v", active[0])
+			}
+		})
+	}
+}

internal/alerts/cleanup_observation_retention_test.go

@@ -0,0 +1,143 @@
+package alerts
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+)
+
+func newCleanupRetentionManager(t *testing.T, dir string, durable bool) *Manager {
+	t.Helper()
+	var options []ManagerOption
+	if durable {
+		options = append(options, WithDurableAlertStore())
+	}
+	m := NewManagerWithDataDir(dir, options...)
+	t.Cleanup(m.Stop)
+	m.mu.Lock()
+	m.config.Enabled = true
+	m.config.ActivationState = ActivationActive
+	m.config.AutoAcknowledgeAfterHours = 0
+	m.config.MaxAlertAgeDays = 1
+	m.config.MaxAcknowledgedAgeDays = 1
+	m.config.DockerDefaults.UpdateAlertDelayHours = 24
+	m.config.Schedule.Cooldown = 0 // Only the first notification per occurrence.
+	m.config.FlappingEnabled = false
+	m.mu.Unlock()
+	return m
+}
+
+// Exercise the actual detector, callbacks, history, checkpoint and restart,
+// not just a hand-seeded active map. A cached registry observation is a current
+// report; missing/failed checks must not be invented as recovery either.
+func TestCleanupContinuingDockerUpdateKeepsOccurrence(t *testing.T) {
+	for _, durable := range []bool{false, true} {
+		name := "JSON recovery"
+		if durable {
+			name = "durable active authority"
+		}
+		t.Run(name, func(t *testing.T) {
+			dir := t.TempDir()
+			m := newCleanupRetentionManager(t, dir, durable)
+			host := models.DockerHost{ID: "retention-host", Hostname: "retention-host"}
+			pending := &models.DockerContainerUpdateStatus{
+				UpdateAvailable: true, CurrentDigest: "sha256:old", LatestDigest: "sha256:new",
+				LastChecked: time.Now().Add(-6 * time.Hour),
+			}
+			container := models.DockerContainer{ID: "retention-container", Name: "web", Image: "mongo:7", UpdateStatus: pending}
+			resourceID := DockerResourceID(host.ID, container.ID)
+			trackingKey := dockerUpdateTrackingKey(host, container)
+			first := time.Now().Add(-25 * time.Hour)
+			m.mu.Lock()
+			m.dockerUpdateFirstSeen[resourceID] = first
+			m.dockerUpdateFirstSeenByIdentity[trackingKey] = first
+			m.mu.Unlock()
+			var deliveries, fired, resolved atomic.Int32
+			m.SetAlertCallback(func(*Alert) { deliveries.Add(1) })
+			m.SubscribeLifecycleCallback(func(event LifecycleEvent) {
+				switch event.Type {
+				case eventlog.TypeFired, eventlog.TypeRefired:
+					fired.Add(1)
+				case eventlog.TypeResolved:
+					resolved.Add(1)
+				}
+			})
+			evaluate := func() {
+				m.checkDockerContainerImageUpdate(host, container, resourceID, container.Name, host.Hostname, host.Hostname)
+			}
+			evaluate()
+			active := m.GetActiveAlerts()
+			if len(active) != 1 || deliveries.Load() != 1 || fired.Load() != 1 {
+				t.Fatalf("initial update did not fire once: active %d, callbacks %d, firings %d", len(active), deliveries.Load(), fired.Load())
+			}
+			initial := active[0]
+			for _, status := range []*models.DockerContainerUpdateStatus{pending, nil, {Error: "registry unavailable"}, pending} {
+				container.UpdateStatus = status
+				evaluate()
+				m.Cleanup(time.Hour)
+				if got := len(m.GetActiveAlerts()); got != 1 {
+					t.Errorf("cleanup hid the continuing condition: %d active alerts", got)
+				}
+				container.UpdateStatus = pending
+				evaluate() // The next poll must not fire a replacement occurrence.
+			}
+			if got := deliveries.Load(); got != 1 {
+				t.Errorf("continuing update dispatched %d notifications, want one", got)
+			}
+			if got := fired.Load(); got != 1 || resolved.Load() != 0 {
+				t.Errorf("continuing lifecycle has %d firings / %d resolutions, want 1 / 0", got, resolved.Load())
+			}
+			active = m.GetActiveAlerts()
+			if len(active) != 1 || active[0].ID != initial.ID || !active[0].StartTime.Equal(first) || active[0].Acknowledged || active[0].LastNotified == nil || !active[0].LastNotified.Equal(*initial.LastNotified) {
+				t.Fatalf("continuing occurrence/notification identity changed: %+v", active)
+			}
+			if got := len(m.GetAlertHistory(0)); got != 1 {
+				t.Errorf("continuing condition produced %d history rows, want one", got)
+			}
+			if durable {
+				events := queryAlertEvents(t, m, eventlog.Filter{Types: []string{eventlog.TypeFired, eventlog.TypeRefired, eventlog.TypeResolved}})
+				if len(events) != 1 || events[0].Type != eventlog.TypeFired {
+					t.Errorf("durable lifecycle invented transitions during cleanup: %+v", events)
+				}
+			}
+			t.Logf("four cleanup/next-poll cycles: %d firing callbacks, %d dispatch callbacks", fired.Load(), deliveries.Load())
+			lastSeen := active[0].LastSeen
+			m.Stop() // Includes the real final active-state checkpoint.
+
+			restarted := newCleanupRetentionManager(t, dir, durable)
+			active = restarted.GetActiveAlerts()
+			if len(active) != 1 || active[0].ID != initial.ID || !active[0].StartTime.Equal(first) || !active[0].LastSeen.Equal(lastSeen) || active[0].Acknowledged || active[0].LastNotified == nil || !active[0].LastNotified.Equal(*initial.LastNotified) {
+				t.Fatalf("restart did not preserve the continuing unacknowledged occurrence: %+v", active)
+			}
+			var afterRestartDeliveries, afterRestartFirings, afterRestartResolutions atomic.Int32
+			restarted.SetAlertCallback(func(*Alert) { afterRestartDeliveries.Add(1) })
+			restarted.SubscribeLifecycleCallback(func(event LifecycleEvent) {
+				switch event.Type {
+				case eventlog.TypeFired, eventlog.TypeRefired:
+					afterRestartFirings.Add(1)
+				case eventlog.TypeResolved:
+					afterRestartResolutions.Add(1)
+				}
+			})
+			container.UpdateStatus = nil
+			restarted.checkDockerContainerImageUpdate(host, container, resourceID, container.Name, host.Hostname, host.Hostname)
+			restarted.Cleanup(time.Hour)
+			if len(restarted.GetActiveAlerts()) != 1 || afterRestartFirings.Load() != 0 || afterRestartDeliveries.Load() != 0 {
+				t.Fatal("restart/unknown report lost or re-fired the occurrence")
+			}
+			container.UpdateStatus = &models.DockerContainerUpdateStatus{LastChecked: time.Now()}
+			restarted.checkDockerContainerImageUpdate(host, container, resourceID, container.Name, host.Hostname, host.Hostname)
+			if len(restarted.GetActiveAlerts()) != 0 || afterRestartResolutions.Load() != 1 || afterRestartDeliveries.Load() != 0 {
+				t.Fatal("affirmative clear did not resolve once without another firing notification")
+			}
+			history := restarted.GetAlertHistory(0)
+			if len(history) != 1 || history[0].OperationalRecord == nil || history[0].OperationalRecord.ResolvedAt == nil {
+				t.Fatalf("affirmative clear did not preserve one resolved history occurrence: %+v", history)
+			}
+			t.Log("persisted restart preserved identity, age, last observation and dispatch time; affirmative clear resolved once")
+		})
+	}
+}

internal/alerts/config/types.go

@@ -381,7 +381,7 @@ type AlertConfig struct {
 	TimeThresholds                 map[string]int                 `json:"timeThresholds"`               // Per-type delays: guest, node, agent, storage, pbs
 	MetricTimeThresholds           map[string]map[string]int      `json:"metricTimeThresholds"`         // Optional per-metric delays keyed by resource type
 	MetricEvaluationWindows        map[string]map[string]int      `json:"metricEvaluationWindows"`      // Rolling-average seconds keyed by resource type and metric; explicit zero uses the current value
-	MaxAlertAgeDays                int                            `json:"maxAlertAgeDays"`              // Maximum age for alerts before auto-cleanup (0 = disabled)
+	MaxAlertAgeDays                int                            `json:"maxAlertAgeDays"`              // Inactivity limit in days for unacknowledged alert cleanup (0 = disabled)
 	MaxAcknowledgedAgeDays         int                            `json:"maxAcknowledgedAgeDays"`       // Maximum age for acknowledged alerts (0 = disabled)
 	AutoAcknowledgeAfterHours      int                            `json:"autoAcknowledgeAfterHours"`    // Auto-acknowledge alerts after X hours (0 = disabled)
 	FlappingEnabled                bool                           `json:"flappingEnabled"`              // Enable flapping detection

internal/alerts/default_config.go

@@ -194,7 +194,7 @@ func defaultAlertConfig() AlertConfig {
 			},
 		},
 		// Alert TTL defaults
-		MaxAlertAgeDays:           7,  // Auto-cleanup alerts older than 7 days
+		MaxAlertAgeDays:           7,  // Cleanup unacknowledged alerts inactive for 7 days
 		MaxAcknowledgedAgeDays:    1,  // Auto-cleanup acknowledged alerts older than 1 day
 		AutoAcknowledgeAfterHours: 24, // Auto-acknowledge alerts after 24 hours
 		// Flapping detection defaults