From 601d061dadcba23410b0b251bcb3736ab4fffe5a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 7 Sep 2026 22:41:56 +0100 Subject: [PATCH] fix(ai): backport lifecycle replay occurrence identity repair Backport of c857802df6b00e61e2229eea411d6675deff2d8a. Named regression: lifecycle replay introduced by b6babd0 revisits closed occurrences through open-only lookup, producing duplicate shells and merging genuine recurrence. Reproduced on release baseline d2f7bd0 with new tests; repaired incident-store race tests pass 20 repetitions. No unrelated unchanged-checkpoint optimisation included. Issue #1966 aggregate writes and existing duplicate migration remain unresolved. Change-source: pulse-maintainer --- .../v6/internal/subsystems/ai-runtime.md | 18 ++++ internal/ai/memory/incidents.go | 40 +++++++- internal/ai/memory/incidents_test.go | 93 +++++++++++++++++++ 3 files changed, 148 insertions(+), 3 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index daad986e1..23ed6a4c1 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -19,6 +19,24 @@ ## Purpose +### Incident lifecycle occurrence identity (7 September 2026) + +Fired, resolved and acknowledgement snapshots with a nonzero start time select +an exact (alert identifier, occurrence start) pair, not whichever incident is +open or latest. Replaying a retained closed occurrence must neither allocate a +new ID nor close a newer recurrence. A distinct start remains a distinct +occurrence even within the legacy timeline read tolerance. Zero-start legacy +callers retain best-effort matching. `TestIncidentStore_LifecycleReplayIdentity` +covers unchanged active evaluations, repeated fired/resolved transitions, +JSON checkpoint reload, canonical-backed shells and delayed historical events. +`TestIncidentStore_LifecycleRapidRecurrence` covers distinct subsecond starts. + +This does not migrate existing duplicate records, change bounded retention, +make acknowledgement event replay idempotent, or prove aggregate write-byte +reductions or recipient delivery. An evicted occurrence can still be recreated +by replay; exact identity is guaranteed only while its shell is retained. + + Incident timeline summaries must distinguish provider conditions from metric threshold evidence. For `resource-incident` alert events, numeric value and threshold placeholders must not render as a measured comparison; retain the diff --git a/internal/ai/memory/incidents.go b/internal/ai/memory/incidents.go index d8fbd4ea4..ce457221b 100644 --- a/internal/ai/memory/incidents.go +++ b/internal/ai/memory/incidents.go @@ -359,7 +359,10 @@ func (s *IncidentStore) RecordAlertFired(alert *alerts.Alert) { s.mu.Lock() defer s.mu.Unlock() - shell := s.findOpenIncidentByAlertIdentifierLocked(alert.ID) + shell := s.findLifecycleOccurrenceLocked(alert) + if alert.StartTime.IsZero() { + shell = s.findOpenIncidentByAlertIdentifierLocked(alert.ID) + } if shell == nil { shell = newIncidentShellFromAlert(alert) s.incidents = append(s.incidents, shell) @@ -371,6 +374,9 @@ func (s *IncidentStore) RecordAlertFired(alert *alerts.Alert) { "threshold": alert.Threshold, }) } + } else if incidentOccurrenceClosedAt(shell) != nil { + // A replayed fired event must not reopen a completed occurrence. + return } else { updateIncidentShellFromAlert(shell, alert) } @@ -432,12 +438,20 @@ func (s *IncidentStore) RecordAlertResolved(alert *alerts.Alert, resolvedAt time s.mu.Lock() defer s.mu.Unlock() - shell := s.findOpenIncidentByAlertIdentifierLocked(alert.ID) + shell := s.findLifecycleOccurrenceLocked(alert) + if alert.StartTime.IsZero() { + shell = s.findOpenIncidentByAlertIdentifierLocked(alert.ID) + } if shell == nil { shell = newIncidentShellFromAlert(alert) s.incidents = append(s.incidents, shell) } + // Resolution replay is idempotent, including after a checkpoint reload. + if incidentOccurrenceClosedAt(shell) != nil { + return + } + if resolvedAt.IsZero() { now := time.Now() resolvedAt = now @@ -1318,8 +1332,28 @@ func updateIncidentShellFromAlert(shell *incidentShell, alert *alerts.Alert) { shell.Message = alert.Message } +// Lifecycle snapshots carry the exact occurrence start. Unlike legacy read +// repair, they must not use a time tolerance or select an unrelated open/latest +// incident: delayed transitions and replay can arrive after a genuine recurrence. +func (s *IncidentStore) findLifecycleOccurrenceLocked(alert *alerts.Alert) *incidentShell { + if alert.ID == "" { + return nil + } + if alert.StartTime.IsZero() { + // Legacy callers without an occurrence key retain best-effort matching. + return s.findLatestIncidentByAlertIdentifierLocked(alert.ID) + } + for i := len(s.incidents) - 1; i >= 0; i-- { + shell := s.incidents[i] + if shell != nil && shell.AlertIdentifier == alert.ID && shell.OpenedAt.Equal(alert.StartTime) { + return shell + } + } + return nil +} + func (s *IncidentStore) ensureIncidentForAlertLocked(alert *alerts.Alert) *incidentShell { - shell := s.findLatestIncidentByAlertIdentifierLocked(alert.ID) + shell := s.findLifecycleOccurrenceLocked(alert) if shell == nil { shell = newIncidentShellFromAlert(alert) s.incidents = append(s.incidents, shell) diff --git a/internal/ai/memory/incidents_test.go b/internal/ai/memory/incidents_test.go index e271d2133..a842d6a3a 100644 --- a/internal/ai/memory/incidents_test.go +++ b/internal/ai/memory/incidents_test.go @@ -1059,3 +1059,96 @@ func TestIncidentStore_RecordNote_NonexistentIncident(t *testing.T) { t.Error("expected false for non-existent incident") } } + +// Lifecycle replay may revisit a closed occurrence after checkpoint restart, +// including while a newer occurrence of the same alert is already active. +func TestIncidentStore_LifecycleReplayIdentity(t *testing.T) { + for _, tc := range []struct { + name string + restart, canonical bool + }{ + {name: "memory"}, + {name: "checkpoint-restart", restart: true}, + {name: "canonical", canonical: true}, + {name: "canonical-checkpoint-restart", restart: true, canonical: true}, + } { + t.Run(tc.name, func(t *testing.T) { + store := NewIncidentStore(IncidentStoreConfig{}) + wantClosedEvents, wantOpenEvents := 2, 1 + if tc.canonical { + store.SetResourceTimelineStore(unifiedresources.NewMemoryStore()) + wantClosedEvents, wantOpenEvents = 0, 0 + } + start := time.Now().Add(-time.Hour).UTC() + old := &alerts.Alert{ID: "pbs-connectivity", ResourceID: "pbs", StartTime: start} + store.RecordAlertFired(old) + for i := 0; i < 10; i++ { + store.RecordAlertFired(old) + } + if len(store.incidents) != 1 { + t.Fatal("unchanged active evaluations duplicated incident identity") + } + oldID := store.incidents[0].ID + end := start.Add(time.Minute) + store.RecordAlertResolved(old, end) + if tc.restart { + // Synchronous JSON checkpoint/reload isolates application lifecycle + // identity from asynchronous scheduling and external event stores. + store.dataDir = t.TempDir() + store.filePath = store.dataDir + "/ai_incidents.json" + if err := store.saveToDisk(); err != nil { + t.Fatal(err) + } + reloaded := NewIncidentStore(IncidentStoreConfig{}) + reloaded.filePath = store.filePath + if err := reloaded.loadFromDisk(); err != nil { + t.Fatal(err) + } + store = reloaded + if tc.canonical { + store.SetResourceTimelineStore(unifiedresources.NewMemoryStore()) + } + } + for i := 0; i < 10; i++ { + store.RecordAlertFired(old) + store.RecordAlertResolved(old, end) + } + if len(store.incidents) != 1 { + t.Fatalf("replayed one occurrence produced %d distinct incident shells", len(store.incidents)) + } + if store.incidents[0].ID != oldID || len(store.incidents[0].Events) != wantClosedEvents { + t.Fatal("replay changed incident identity or duplicated lifecycle events") + } + newAlert := old.Clone() + newAlert.StartTime = end.Add(time.Second) + store.RecordAlertFired(newAlert) + if len(store.incidents) != 2 { + t.Fatal("genuine recurrence did not open a separate incident") + } + newID := store.incidents[1].ID + store.RecordAlertResolved(old, end) + store.RecordAlertAcknowledged(old, "historical-actor") + if len(store.incidents) != 2 || store.incidents[1].ID != newID || store.incidents[1].OccurrenceClosedAt != nil || len(store.incidents[1].Events) != wantOpenEvents { + t.Fatal("late historical lifecycle event mutated the active recurrence") + } + }) + } +} + +func TestIncidentStore_LifecycleRapidRecurrence(t *testing.T) { + store := NewIncidentStore(IncidentStoreConfig{}) + first := &alerts.Alert{ID: "same-alert", StartTime: time.Now().Add(-time.Minute)} + store.RecordAlertFired(first) + // A missed resolution must not merge distinct starts, even within the + // tolerance used by legacy timeline reads. + second := first.Clone() + second.StartTime = first.StartTime.Add(500 * time.Millisecond) + store.RecordAlertFired(second) + if len(store.incidents) != 2 { + t.Fatalf("distinct occurrence starts merged: got %d incident shells", len(store.incidents)) + } + store.RecordAlertResolved(first, first.StartTime.Add(100*time.Millisecond)) + if store.incidents[1].OccurrenceClosedAt != nil { + t.Fatal("old resolution closed rapid recurrence") + } +}