fix(ai): preserve incident occurrence identity during lifecycle replay

Issue #1966 reports distinct incident IDs for one occurrence. Deterministic lifecycle replay creates eleven shells for one resolved start, including after JSON restart; open-only matching also lets an old resolution close a newer occurrence. Match exact lifecycle starts and retain closed identity on replay, without changing legacy zero-start matching or retention. Add recurrence, restart and canonical-shell regression coverage; focused incident race tests pass for twenty repetitions. This does not establish total write savings or migrate existing duplicates.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/ai-runtime.md

@@ -25,6 +25,23 @@ that same result. Successful reads retain their content and execution provenance
 
 ## 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.
+
 ### Unchanged incident JSON checkpoints
 
 Incident-memory checkpoints compare the serialized snapshot with bounded bytes

internal/ai/memory/incidents.go

@@ -361,7 +361,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)
@@ -373,6 +376,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)
 	}
@@ -434,12 +440,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
@@ -1320,8 +1334,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)

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")
+	}
+}