Bound alert lifecycle replay with a durable projection watermark

Projection replay walked the entire alert event log on every boot, twice
(monitor start and the resource-store boundary), synchronously on the
serving path, and every replayed event queued a full incident-store JSON
rewrite. A 64k-event log made startup exceed the dev supervisor's
health-kill window, so the backend restarted forever and never served.

The event log now owns per-consumer projection watermarks in
alert_store_meta, WalkOldest takes an AfterID cursor, and the single
catch-up pass runs in the background at the canonical resource-store
boundary with periodic watermark checkpoints. The watermark only
advances when both the incident store and the canonical change recorder
are attached, so partial-surface passes repair without marking events
applied, and lowering the watermark forces a full repair replay.
Incident-store saves now coalesce: a burst of mutations queues one
whole-store serialization instead of one per event, live paths included.

Registers the alert-lifecycle-replay-startup-scalability coverage gap
and amends the alerts and monitoring subsystem contracts accordingly.

docs/release-control/v6/internal/status.json

@@ -9899,6 +9899,45 @@
           "kind": "file"
         }
       ]
+    },
+    {
+      "id": "alert-lifecycle-replay-startup-scalability",
+      "summary": "Alert lifecycle projection replay walks the entire durable event log on every boot, twice (monitor start and the canonical resource-store boundary), synchronously on the HTTP serving path, and every replayed event triggers a full incident-store JSON rewrite. With a large event log (64k events on the dev mock instance) startup exceeds the dev supervisor's health-kill window, so the backend restarts forever and never serves. Replay needs a durable projection watermark so boot only walks the un-projected tail, needs to run off the serving path, and incident-store persistence needs save coalescing so replay and live bursts stop rewriting the full store per event.",
+      "owner": "project-owner",
+      "status": "triaged",
+      "recorded_at": "2026-08-31",
+      "lane_ids": [
+        "L6",
+        "L13"
+      ],
+      "subsystem_ids": [
+        "alerts",
+        "monitoring"
+      ],
+      "proposed_resolution": "lane-expansion",
+      "coverage_impact": 6,
+      "evidence": [
+        {
+          "repo": "pulse",
+          "path": "internal/ai/memory/incidents.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "internal/alerts/event_emission.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "internal/alerts/eventlog/eventlog.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "internal/monitoring/monitor_alerts.go",
+          "kind": "file"
+        }
+      ]
     }
   ],
   "candidate_lanes": [

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

@@ -466,7 +466,12 @@ transition references recovery evidence separate from its trigger evidence.
    using normalized owned storage roots and fixed storage leaves only.
    Event-log projection changes must preserve parity with the in-memory
    JSON-history model, and migration changes must retain a retryable source or
-   backup until every legacy entry is durably imported.
+   backup until every legacy entry is durably imported. The event log also
+   owns per-consumer projection watermarks in `alert_store_meta`
+   (`projection_watermark:<consumer>`): replay consumers pass their watermark
+   as the walk's `AfterID` cursor so boot-time repair visits only the
+   un-projected tail, and lowering a watermark (including to zero) is the
+   supported way to force a full replay after a projection store rebuild.
 5. Add or change locked alert-investigation commercial handoff behavior through
    `frontend-modern/src/components/Alerts/InvestigateAlertButton.tsx` while
    preserving the shared upgrade-navigation contract; the alert surface may

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

@@ -2311,6 +2311,17 @@ The monitor-owned incident store wiring must therefore attach the canonical
 resource timeline reader whenever the unified monitor adapter is present, so
 operator alert timelines and AI incident context project those lifecycle events
 from canonical history instead of reading a second monitoring-owned timeline.
+Lifecycle projection replay is bounded by a durable projection watermark
+(`alert-lifecycle-timelines-v1`, stored beside the event log) and runs as one
+background catch-up pass scheduled at the canonical resource-store boundary —
+never synchronously on router construction or health serving, because a large
+un-projected backlog must delay projections, not startup. The watermark
+advances, with periodic mid-pass checkpoints, only when a pass runs with both
+the incident store and the canonical resource-change recorder attached; a
+partial-surface pass repairs what it can without marking events applied.
+Resetting the watermark to zero forces a full repair replay for rebuilt
+projection stores, while wiping a projection store without resetting the
+watermark leaves already-applied events to request-time read-repair only.
 
 The registry proof map now treats provider discovery and metrics history as
 their own governed runtime surfaces instead of leaving them folded into a

internal/ai/memory/incidents.go

@@ -7,6 +7,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
@@ -272,6 +273,8 @@ type IncidentStoreConfig struct {
 type IncidentStore struct {
 	mu                    sync.RWMutex
 	saveMu                sync.Mutex
+	savePending           atomic.Bool  // a queued save will capture the latest state; further requests coalesce
+	savesCompleted        atomic.Int64 // completed saveToDisk passes, observable by tests
 	incidents             []*incidentShell
 	maxIncidents          int
 	maxEvents             int
@@ -1482,6 +1485,14 @@ func (s *IncidentStore) saveAsync() {
 	if s.dataDir == "" || s.filePath == "" {
 		return
 	}
+	// Coalesce: each save marshals the whole store, so a burst of mutations
+	// (lifecycle replay, alert storms) must not queue one full serialization
+	// per call. One pending save is enough — it snapshots the store when it
+	// runs, so it covers every mutation made before it started, and any
+	// mutation after the pending flag clears queues exactly one more save.
+	if !s.savePending.CompareAndSwap(false, true) {
+		return
+	}
 	go func() {
 		if err := s.saveToDisk(); err != nil {
 			log.Warn().Err(err).Msg("failed to save incident history")
@@ -1493,6 +1504,11 @@ func (s *IncidentStore) saveToDisk() error {
 	s.saveMu.Lock()
 	defer s.saveMu.Unlock()
 
+	// Clear the coalescing flag only once this save actually starts: the
+	// snapshot below covers every mutation made before this point, and a
+	// mutation after it queues exactly one follow-up save.
+	s.savePending.Store(false)
+
 	if s.dataDir == "" || s.filePath == "" {
 		return nil
 	}
@@ -1519,6 +1535,7 @@ func (s *IncidentStore) saveToDisk() error {
 	if err := os.Rename(tmpFile, s.filePath); err != nil {
 		return err
 	}
+	s.savesCompleted.Add(1)
 	return nil
 }
 

internal/ai/memory/incidents_save_coalescing_test.go

@@ -0,0 +1,51 @@
+package memory
+
+import (
+	"testing"
+	"time"
+)
+
+func waitForCompletedSaves(t *testing.T, store *IncidentStore, want int64) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if store.savesCompleted.Load() >= want {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("saves completed = %d, want at least %d", store.savesCompleted.Load(), want)
+}
+
+func TestIncidentStoreSaveCoalescing(t *testing.T) {
+	dir := t.TempDir()
+	store := NewIncidentStore(IncidentStoreConfig{DataDir: dir})
+
+	// Hold the save lock so the queued save cannot start: every request made
+	// while one is queued must coalesce instead of queuing its own full-store
+	// serialization. This is the lifecycle-replay / alert-storm shape that
+	// previously produced one whole-store JSON write per event.
+	store.saveMu.Lock()
+	for i := 0; i < 100; i++ {
+		store.RecordAnalysis("alert-coalesce", "analysis", map[string]interface{}{"pass": i})
+	}
+	store.saveMu.Unlock()
+
+	waitForCompletedSaves(t, store, 1)
+	if saves := store.savesCompleted.Load(); saves != 1 {
+		t.Fatalf("saves completed after fully queued burst = %d, want 1", saves)
+	}
+
+	// A mutation after the coalesced save queues exactly one follow-up save.
+	store.RecordAnalysis("alert-coalesce", "follow-up", nil)
+	waitForCompletedSaves(t, store, 2)
+
+	reloaded := NewIncidentStore(IncidentStoreConfig{DataDir: dir})
+	timeline := reloaded.GetTimelineByAlertIdentifier("alert-coalesce")
+	if timeline == nil {
+		t.Fatal("coalesced saves did not persist the incident")
+	}
+	if len(timeline.Events) != 101 {
+		t.Fatalf("persisted events = %d, want all 101 mutations captured", len(timeline.Events))
+	}
+}

internal/alerts/event_emission.go

@@ -73,13 +73,16 @@ func (m *Manager) AlertEvents(filter eventlog.Filter) ([]eventlog.Event, error)
 	return store.Query(filter)
 }
 
-// ReplayLifecycleEvents visits every durable lifecycle transition and migrated
-// history snapshot oldest first. It is the projection-repair seam for
-// consumers such as incident and resource timelines: delivery callbacks are
-// deliberately not involved. Migrated snapshots carry a final occurrence
-// state rather than a transition; consumers must expand only the facts present
-// in that snapshot.
-func (m *Manager) ReplayLifecycleEvents(visit func(LifecycleEvent) error) error {
+// ReplayLifecycleEvents visits durable lifecycle transitions and migrated
+// history snapshots with ids above afterID, oldest occurrence first. It is the
+// projection-repair seam for consumers such as incident and resource
+// timelines: delivery callbacks are deliberately not involved. Migrated
+// snapshots carry a final occurrence state rather than a transition; consumers
+// must expand only the facts present in that snapshot. afterID is the
+// consumer's projection watermark (zero replays the full log); the visitor
+// receives each event's durable id so it can advance that watermark once the
+// event's projections are applied.
+func (m *Manager) ReplayLifecycleEvents(afterID int64, visit func(eventID int64, event LifecycleEvent) error) error {
 	if m == nil || visit == nil {
 		return nil
 	}
@@ -87,7 +90,7 @@ func (m *Manager) ReplayLifecycleEvents(visit func(LifecycleEvent) error) error
 	if store == nil {
 		return nil
 	}
-	return store.WalkOldest(eventlog.Filter{Types: []string{
+	return store.WalkOldest(eventlog.Filter{AfterID: afterID, Types: []string{
 		eventlog.TypeFired,
 		eventlog.TypeRefired,
 		eventlog.TypeResolved,
@@ -108,7 +111,7 @@ func (m *Manager) ReplayLifecycleEvents(visit func(LifecycleEvent) error) error
 				Msg("skipping invalid alert lifecycle snapshot during projection replay")
 			return nil
 		}
-		return visit(LifecycleEvent{
+		return visit(event.ID, LifecycleEvent{
 			Type:       event.Type,
 			OccurredAt: event.OccurredAt,
 			Alert:      &snapshot,
@@ -118,6 +121,31 @@ func (m *Manager) ReplayLifecycleEvents(visit func(LifecycleEvent) error) error
 	})
 }
 
+// LifecycleProjectionWatermark returns the durable replay watermark for a
+// named projection consumer, or zero when no event log is enabled.
+func (m *Manager) LifecycleProjectionWatermark(name string) int64 {
+	if m == nil {
+		return 0
+	}
+	return m.eventLogStore().ProjectionWatermark(name)
+}
+
+// StoreLifecycleProjectionWatermark durably advances (or resets) the replay
+// watermark for a named projection consumer. Failures are logged, not fatal:
+// a stale watermark only means the next replay revisits already-idempotent
+// projections.
+func (m *Manager) StoreLifecycleProjectionWatermark(name string, eventID int64) {
+	if m == nil {
+		return
+	}
+	if err := m.eventLogStore().SetProjectionWatermark(name, eventID); err != nil {
+		log.Warn().Err(err).
+			Str("consumer", name).
+			Int64("eventID", eventID).
+			Msg("failed to persist lifecycle projection watermark")
+	}
+}
+
 // recordAlertEvent emits one canonical lifecycle transition and appends it to
 // the durable event log. alertID is the identity fallback for callers whose
 // *Alert may be nil (some resolve paths); when the alert is present its

internal/alerts/eventlog/eventlog.go

@@ -85,6 +85,11 @@ type Filter struct {
 	Since   time.Time
 	Until   time.Time
 	Limit   int
+	// AfterID restricts matches to events with a strictly greater durable id.
+	// It is the projection-watermark cursor: replay consumers pass the highest
+	// id they have fully applied so a walk visits only the un-projected tail
+	// instead of the whole log.
+	AfterID int64
 }
 
 const (
@@ -803,6 +808,10 @@ func eventFilterWhere(filter Filter) ([]string, []any) {
 		where = append(where, "occurred_at <= ?")
 		args = append(args, filter.Until.UTC().Format(time.RFC3339Nano))
 	}
+	if filter.AfterID > 0 {
+		where = append(where, "id > ?")
+		args = append(args, filter.AfterID)
+	}
 	return where, args
 }
 

internal/alerts/eventlog/projection_watermark.go

@@ -0,0 +1,58 @@
+package eventlog
+
+// Projection watermarks record, per named consumer, the highest durable event
+// id whose lifecycle projection has been fully applied. They live in
+// alert_store_meta so the cursor travels with the log it indexes: replay walks
+// only ids above the watermark instead of the whole history on every boot.
+// The cursor does not carry the projected data — wiping a projection store
+// (incident history, unified-resource timelines) without wiping this database
+// leaves events at or below the watermark unprojected until read-repair or an
+// explicit watermark reset.
+
+import (
+	"database/sql"
+	"fmt"
+)
+
+const projectionWatermarkKeyPrefix = "projection_watermark:"
+
+// ProjectionWatermark returns the durable watermark for a named projection
+// consumer, or zero when none has been recorded. A nil store reports zero so
+// consumers without an event log replay nothing.
+func (s *Store) ProjectionWatermark(name string) int64 {
+	if s == nil || name == "" {
+		return 0
+	}
+	var value int64
+	err := s.db.QueryRow(`SELECT CAST(value AS INTEGER) FROM alert_store_meta WHERE key = ?`, projectionWatermarkKeyPrefix+name).Scan(&value)
+	if err == sql.ErrNoRows {
+		return 0
+	}
+	if err != nil || value < 0 {
+		return 0
+	}
+	return value
+}
+
+// SetProjectionWatermark durably records the watermark for a named projection
+// consumer. Callers may lower it (including to zero) to force a full replay,
+// for example after a projection store has been rebuilt.
+func (s *Store) SetProjectionWatermark(name string, eventID int64) error {
+	if s == nil {
+		return nil
+	}
+	if name == "" {
+		return fmt.Errorf("projection watermark name is required")
+	}
+	if eventID < 0 {
+		return fmt.Errorf("projection watermark id must not be negative")
+	}
+	_, err := s.db.Exec(`
+		INSERT INTO alert_store_meta (key, value) VALUES (?, ?)
+		ON CONFLICT(key) DO UPDATE SET value = excluded.value
+	`, projectionWatermarkKeyPrefix+name, eventID)
+	if err != nil {
+		return fmt.Errorf("persist projection watermark %q: %w", name, err)
+	}
+	return nil
+}

internal/alerts/eventlog/projection_watermark_test.go

@@ -0,0 +1,120 @@
+package eventlog
+
+import (
+	"testing"
+	"time"
+)
+
+func TestProjectionWatermarkRoundTripAndReset(t *testing.T) {
+	store := newTestStore(t)
+
+	if got := store.ProjectionWatermark("timelines"); got != 0 {
+		t.Fatalf("unset watermark = %d, want 0", got)
+	}
+	if err := store.SetProjectionWatermark("timelines", 42); err != nil {
+		t.Fatalf("set watermark: %v", err)
+	}
+	if got := store.ProjectionWatermark("timelines"); got != 42 {
+		t.Fatalf("watermark = %d, want 42", got)
+	}
+	if err := store.SetProjectionWatermark("timelines", 99); err != nil {
+		t.Fatalf("advance watermark: %v", err)
+	}
+	if got := store.ProjectionWatermark("timelines"); got != 99 {
+		t.Fatalf("advanced watermark = %d, want 99", got)
+	}
+	// Consumers are independent.
+	if got := store.ProjectionWatermark("other"); got != 0 {
+		t.Fatalf("unrelated consumer watermark = %d, want 0", got)
+	}
+	// Lowering (including to zero) is allowed so a rebuilt projection store
+	// can force a full replay.
+	if err := store.SetProjectionWatermark("timelines", 0); err != nil {
+		t.Fatalf("reset watermark: %v", err)
+	}
+	if got := store.ProjectionWatermark("timelines"); got != 0 {
+		t.Fatalf("reset watermark = %d, want 0", got)
+	}
+
+	if err := store.SetProjectionWatermark("", 5); err == nil {
+		t.Fatal("set watermark with empty name should fail")
+	}
+	if err := store.SetProjectionWatermark("timelines", -1); err == nil {
+		t.Fatal("set negative watermark should fail")
+	}
+
+	var nilStore *Store
+	if got := nilStore.ProjectionWatermark("timelines"); got != 0 {
+		t.Fatalf("nil store watermark = %d, want 0", got)
+	}
+	if err := nilStore.SetProjectionWatermark("timelines", 7); err != nil {
+		t.Fatalf("nil store set watermark should be a no-op, got %v", err)
+	}
+}
+
+func TestProjectionWatermarkSurvivesReopen(t *testing.T) {
+	dir := t.TempDir()
+	store, err := Open(dir)
+	if err != nil {
+		t.Fatalf("open store: %v", err)
+	}
+	if err := store.SetProjectionWatermark("timelines", 17); err != nil {
+		store.Close()
+		t.Fatalf("set watermark: %v", err)
+	}
+	store.Close()
+
+	reopened, err := Open(dir)
+	if err != nil {
+		t.Fatalf("reopen store: %v", err)
+	}
+	t.Cleanup(reopened.Close)
+	if got := reopened.ProjectionWatermark("timelines"); got != 17 {
+		t.Fatalf("watermark after reopen = %d, want 17", got)
+	}
+}
+
+func TestWalkOldestAfterIDVisitsOnlyTail(t *testing.T) {
+	store := newTestStore(t)
+	base := time.Date(2026, 8, 31, 10, 0, 0, 0, time.UTC)
+
+	for i := 0; i < 3; i++ {
+		store.Append(Event{OccurredAt: base.Add(time.Duration(i) * time.Minute), Type: TypeFired, AlertID: "a1"})
+	}
+	if err := store.Flush(); err != nil {
+		t.Fatalf("flush: %v", err)
+	}
+
+	var all []int64
+	if err := store.WalkOldest(Filter{Types: []string{TypeFired}}, func(event Event) error {
+		all = append(all, event.ID)
+		return nil
+	}); err != nil {
+		t.Fatalf("walk all: %v", err)
+	}
+	if len(all) != 3 {
+		t.Fatalf("walked %d events, want 3", len(all))
+	}
+
+	var tail []int64
+	if err := store.WalkOldest(Filter{AfterID: all[0], Types: []string{TypeFired}}, func(event Event) error {
+		tail = append(tail, event.ID)
+		return nil
+	}); err != nil {
+		t.Fatalf("walk tail: %v", err)
+	}
+	if len(tail) != 2 || tail[0] != all[1] || tail[1] != all[2] {
+		t.Fatalf("tail = %v, want %v", tail, all[1:])
+	}
+
+	var none []int64
+	if err := store.WalkOldest(Filter{AfterID: all[2], Types: []string{TypeFired}}, func(event Event) error {
+		none = append(none, event.ID)
+		return nil
+	}); err != nil {
+		t.Fatalf("walk beyond newest: %v", err)
+	}
+	if len(none) != 0 {
+		t.Fatalf("walk beyond newest visited %v, want none", none)
+	}
+}

internal/monitoring/monitor.go

@@ -1111,6 +1111,8 @@ type Monitor struct {
 	alertPushCallback          func(*alerts.Alert)
 	connectionsSnapshotLister  func() []alerts.ConnectionSnapshot // returns platform connection snapshots for the connection-degraded check
 	incidentStore              *memory.IncidentStore
+	alertProjectionReplayMu    sync.Mutex     // serializes lifecycle projection replay passes
+	alertProjectionWG          sync.WaitGroup // tracks scheduled background catch-up runs
 	notificationMgr            *notifications.NotificationManager
 	deadMan                    *deadManRuntime
 	deadManProgressUnixNano    atomic.Int64
@@ -2004,11 +2006,13 @@ func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) {
 		m.discoveryService = nil
 	}
 
-	// Set up alert callbacks
+	// Set up alert callbacks. Projection replay is deliberately absent here:
+	// the canonical resource store is not attached yet, so a replay now could
+	// not advance the durable watermark and would walk the same events again
+	// at the SetResourceStore boundary, which owns the single scheduled
+	// catch-up pass.
 	m.alertManager.SubscribeLifecycleCallback(m.handleAlertLifecycleEvent)
 	m.wireExternalAlertCallbacks(wsHub)
-	m.replayAlertLifecycleProjections()
-	m.reconcileActiveAlertTimelines()
 	m.markDeadManMonitoringProgress(time.Now().UTC())
 	if err := m.deadManConfigurationLoadError(); err != nil {
 		m.alertManager.RaiseSystemAlert(alerts.SystemAlertInput{
@@ -4572,11 +4576,12 @@ func (m *Monitor) SetResourceStore(store ResourceStoreInterface) {
 		}
 	}
 	if timelineAttached {
-		// NewMonitor replays before the API-owned durable resource store is
-		// attached. Replay again at the canonical boundary so restart repair
-		// reaches resource history as well as the incident fallback cache.
-		m.replayAlertLifecycleProjections()
-		m.reconcileActiveAlertTimelines()
+		// The canonical boundary: with the durable resource store attached,
+		// restart repair can reach resource history as well as the incident
+		// fallback cache, and the replay watermark may advance. Run it in the
+		// background — router construction and health serving must not wait
+		// behind an un-projected backlog.
+		m.scheduleAlertProjectionCatchUp()
 	}
 
 	// Immediately backfill the store from current state so ReadState

internal/monitoring/monitor_alert_handling_test.go

@@ -345,6 +345,7 @@ func TestLifecycleReplayMaterializesImportedHistoryTimeline(t *testing.T) {
 	adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(resourceStore))
 	monitor.SetResourceStore(adapter)
 	monitor.SetResourceStore(adapter)
+	monitor.alertProjectionWG.Wait()
 
 	timeline := incidentStore.GetTimelineByAlertAt(snapshot.ID, snapshot.StartTime)
 	if timeline == nil || timeline.Status != memory.IncidentStatusResolved || !timeline.Acknowledged {
@@ -362,6 +363,72 @@ func TestLifecycleReplayMaterializesImportedHistoryTimeline(t *testing.T) {
 	}
 }
 
+func TestLifecycleReplayWatermarkBoundsSubsequentPasses(t *testing.T) {
+	manager := alerts.NewManagerWithDataDir(t.TempDir(), alerts.WithoutPersistedAlertRestore())
+	t.Cleanup(manager.Stop)
+	manager.EnableEventLog()
+	config := manager.GetConfig()
+	config.Enabled = true
+	config.ActivationState = alerts.ActivationPending
+	config.TimeThresholds = map[string]int{}
+	config.SuppressionWindow = 0
+	manager.UpdateConfig(config)
+
+	vm := models.VM{ID: "watermark-vm", Name: "Watermark VM", Node: "node-1", Instance: "pve-1", Status: "stopped"}
+	manager.CheckGuest(vm, vm.Instance)
+	manager.CheckGuest(vm, vm.Instance)
+	vm.Status = "running"
+	manager.CheckGuest(vm, vm.Instance)
+
+	// A pass without the canonical resource store repairs incidents but must
+	// not advance the durable watermark, or resource-timeline projections for
+	// those events would never materialize.
+	partial := &Monitor{alertManager: manager, incidentStore: memory.NewIncidentStore(memory.IncidentStoreConfig{})}
+	partial.replayAlertLifecycleProjections()
+	if got := manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer); got != 0 {
+		t.Fatalf("watermark after partial-surface replay = %d, want 0", got)
+	}
+
+	resourceStore := unifiedresources.NewMemoryStore()
+	monitor := &Monitor{
+		alertManager:  manager,
+		incidentStore: memory.NewIncidentStore(memory.IncidentStoreConfig{}),
+		resourceStore: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(resourceStore)),
+	}
+	monitor.replayAlertLifecycleProjections()
+	watermark := manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer)
+	if watermark == 0 {
+		t.Fatal("watermark did not advance after full-surface replay")
+	}
+
+	// A later pass walks only events beyond the watermark, so fresh projection
+	// stores stay empty: nothing is left to replay.
+	freshResources := unifiedresources.NewMemoryStore()
+	rerun := &Monitor{
+		alertManager:  manager,
+		incidentStore: memory.NewIncidentStore(memory.IncidentStoreConfig{}),
+		resourceStore: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(freshResources)),
+	}
+	rerun.replayAlertLifecycleProjections()
+	if changes, err := freshResources.GetRecentChanges(vm.ID, time.Time{}, 10); err != nil || len(changes) != 0 {
+		t.Fatalf("bounded pass wrote %d changes (err %v), want none", len(changes), err)
+	}
+
+	// Resetting the watermark forces a full repair replay for rebuilt stores.
+	manager.StoreLifecycleProjectionWatermark(alertLifecycleProjectionConsumer, 0)
+	rerun.replayAlertLifecycleProjections()
+	changes, err := freshResources.GetRecentChanges(vm.ID, time.Time{}, 10)
+	if err != nil {
+		t.Fatalf("GetRecentChanges after watermark reset: %v", err)
+	}
+	if len(changes) != 2 {
+		t.Fatalf("reset replay wrote %d canonical changes, want fired and resolved", len(changes))
+	}
+	if got := manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer); got != watermark {
+		t.Fatalf("watermark after reset replay = %d, want %d", got, watermark)
+	}
+}
+
 func TestSystemAlertTimelineUsesCanonicalPulseResource(t *testing.T) {
 	resourceStore := unifiedresources.NewMemoryStore()
 	incidentStore := memory.NewIncidentStore(memory.IncidentStoreConfig{})

internal/monitoring/monitor_alerts.go

@@ -433,15 +433,84 @@ func (m *Monitor) isActiveAlertOccurrence(candidate *alerts.Alert) bool {
 	return false
 }
 
+const (
+	// alertLifecycleProjectionConsumer names the replay watermark shared by the
+	// incident-timeline and canonical resource-change projections, which one
+	// replay pass applies together.
+	alertLifecycleProjectionConsumer = "alert-lifecycle-timelines-v1"
+	// alertProjectionCheckpointEvery bounds how much replay progress a mid-pass
+	// crash can lose: the watermark is persisted after this many visited events
+	// as well as at the end of a completed pass.
+	alertProjectionCheckpointEvery = 512
+)
+
+// scheduleAlertProjectionCatchUp runs lifecycle projection replay and
+// active-alert reconciliation in the background. The walk is bounded by the
+// durable projection watermark, so it must never run synchronously on the
+// serving path: a large un-projected backlog (first boot after upgrade, a
+// reset watermark) would otherwise block router construction and health
+// serving, and dev supervisors kill an unresponsive backend long before a full
+// 200MB-log replay finishes.
+func (m *Monitor) scheduleAlertProjectionCatchUp() {
+	if m == nil {
+		return
+	}
+	m.alertProjectionWG.Add(1)
+	go func() {
+		defer m.alertProjectionWG.Done()
+		defer recoverFromPanic("alertProjectionCatchUp")
+		m.replayAlertLifecycleProjections()
+		m.reconcileActiveAlertTimelines()
+	}()
+}
+
 func (m *Monitor) replayAlertLifecycleProjections() {
 	if m == nil || m.alertManager == nil {
 		return
 	}
-	if err := m.alertManager.ReplayLifecycleEvents(func(event alerts.LifecycleEvent) error {
+	// Serialize passes instead of skipping: a second trigger waits for the
+	// in-flight pass and then walks the (now tiny) remaining tail, so callers
+	// that need replay-complete semantics can rely on a finished call.
+	m.alertProjectionReplayMu.Lock()
+	defer m.alertProjectionReplayMu.Unlock()
+
+	m.mu.RLock()
+	_, hasRecorder := m.resourceStore.(canonicalResourceChangeRecorder)
+	hasIncidents := m.incidentStore != nil
+	m.mu.RUnlock()
+	// The watermark only advances when the full projection surface is
+	// attached. A pass that runs before the canonical resource store exists
+	// repairs what it can but must not mark those events applied, or their
+	// resource-timeline projections would never materialize.
+	advance := hasRecorder && hasIncidents
+
+	afterID := m.alertManager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer)
+	maxApplied := afterID
+	visited := 0
+	err := m.alertManager.ReplayLifecycleEvents(afterID, func(eventID int64, event alerts.LifecycleEvent) error {
 		m.handleAlertLifecycleEvent(event)
+		if eventID > maxApplied {
+			maxApplied = eventID
+		}
+		visited++
+		if advance && visited%alertProjectionCheckpointEvery == 0 {
+			m.alertManager.StoreLifecycleProjectionWatermark(alertLifecycleProjectionConsumer, maxApplied)
+		}
 		return nil
-	}); err != nil {
+	})
+	if err != nil {
 		log.Error().Err(err).Msg("failed to replay canonical alert lifecycle projections")
+		return
+	}
+	if advance && maxApplied > afterID {
+		m.alertManager.StoreLifecycleProjectionWatermark(alertLifecycleProjectionConsumer, maxApplied)
+	}
+	if visited > 0 {
+		log.Info().
+			Int("events", visited).
+			Int64("watermark", maxApplied).
+			Bool("watermarkAdvanced", advance).
+			Msg("alert lifecycle projection replay completed")
 	}
 }
 
@@ -477,7 +546,11 @@ func (m *Monitor) recordAlertTimelineChange(alert *alerts.Alert, kind unifiedres
 	if alert == nil || m == nil {
 		return
 	}
+	// Background catch-up replay runs concurrently with SetResourceStore, so
+	// the store handle must be read under the monitor lock.
+	m.mu.RLock()
 	recorder, ok := m.resourceStore.(canonicalResourceChangeRecorder)
+	m.mu.RUnlock()
 	if !ok || recorder == nil {
 		return
 	}