Merge pull request #2229 from rcourtman/claude/incident-store-save-flush

Join queued incident saves before test cleanup

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

@@ -76,6 +76,17 @@ does it eliminate snapshot serialization. Proof:
 `internal/ai/memory/incidents_unchanged_test.go` covers unchanged evaluations,
 metadata, restart, resolution/recurrence, file loss and failed-write retry.
 
+Coalesced asynchronous checkpoints are tracked until they finish. A queued save
+still creates the data directory and writes its temp file after the mutation
+returns, so a caller that removes or inspects that directory joins in-flight
+saves through the store's unexported `flush` first. Tests backed by `t.TempDir`
+register it as a cleanup; unjoined saves recreated the removed directory or
+failed its cleanup with "directory not empty" on loaded CI runners. This is an
+internal durability boundary. It does not change checkpoint timing,
+coalescing or the stored format, and it is not a shutdown flush.
+`TestIncidentStoreFlushJoinsQueuedSave` proves the join covers a queued save
+that has not started.
+
 Assistant owns composer registration and focus on every open, rather than only
 on component mount. Closing clears the registered input so later keyboard
 commands cannot target a detached composer. A handoff must leave Escape and

internal/ai/memory/incident_query_test.go

@@ -23,6 +23,9 @@ func TestIncidentQueryCanonicalOccurrencesAndNotes(t *testing.T) {
 			}
 			config := IncidentStoreConfig{DataDir: t.TempDir()}
 			memory := NewIncidentStore(config)
+			// Registered after TempDir, so it runs first: queued saves must
+			// finish writing before the data directory is removed.
+			t.Cleanup(memory.flush)
 			memory.SetResourceTimelineStore(canonical)
 			start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
 			second := start.Add(2 * time.Minute)
@@ -64,8 +67,10 @@ func TestIncidentQueryCanonicalOccurrencesAndNotes(t *testing.T) {
 			require.Equal(t, IncidentStatusResolved, noted.Incidents[0].Status)
 			require.Equal(t, IncidentEventNote, noted.Incidents[0].Events[2].Type)
 			require.Equal(t, "operator_note", noted.Incidents[0].Events[2].Source)
-			require.NoError(t, memory.saveToDisk())
+			// Reload from what RecordNote's own asynchronous save wrote.
+			memory.flush()
 			restored := NewIncidentStore(config)
+			t.Cleanup(restored.flush)
 			restored.SetResourceTimelineStore(canonical)
 			persisted, err := restored.QueryIncidents(IncidentQuery{AlertIdentifier: "repeat-alert", StartedAt: start, Limit: 1})
 			require.NoError(t, err)
@@ -220,6 +225,7 @@ func TestIncidentQueryCanonicalAlertContextOwnsLegacyShell(t *testing.T) {
 func TestIncidentAlertContextRetainsOperatorNote(t *testing.T) {
 	canonical := unifiedresources.NewMemoryStore()
 	store := NewIncidentStore(IncidentStoreConfig{DataDir: t.TempDir()})
+	t.Cleanup(store.flush)
 	store.SetResourceTimelineStore(canonical)
 	started := time.Now().UTC().Add(-time.Hour)
 	change := unifiedresources.BuildAlertTimelineChange("resource-a", unifiedresources.ChangeAlertFired, started, "", unifiedresources.AlertTimelineChange{AlertIdentifier: "noted-alert"})
@@ -228,9 +234,6 @@ func TestIncidentAlertContextRetainsOperatorNote(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, page.Incidents, 1)
 	require.True(t, store.RecordNote("noted-alert", page.Incidents[0].ID, "Keep the old pool until its replacement is verified", "operator"))
-	// RecordNote persists asynchronously. Keep the temporary data directory
-	// alive until that write finishes so cleanup cannot race the save goroutine.
-	waitForCompletedSaves(t, store, 1)
 	context := store.FormatForAlert("noted-alert", 10)
 	require.Contains(t, context, "Note added by operator: Keep the old pool until its replacement is verified")
 	require.Contains(t, context, "source=operator_note")

internal/ai/memory/incidents.go

@@ -282,8 +282,9 @@ 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 JSON replacements, observable by tests
+	savePending           atomic.Bool    // a queued save will capture the latest state; further requests coalesce
+	saveWG                sync.WaitGroup // asynchronous saves still running or queued; joined by flush
+	savesCompleted        atomic.Int64   // completed JSON replacements, observable by tests
 	incidents             []*incidentShell
 	maxIncidents          int
 	maxEvents             int
@@ -1459,13 +1460,25 @@ func (s *IncidentStore) saveAsync() {
 	if !s.savePending.CompareAndSwap(false, true) {
 		return
 	}
+	s.saveWG.Add(1)
 	go func() {
+		defer s.saveWG.Done()
 		if err := s.saveToDisk(); err != nil {
 			log.Warn().Err(err).Msg("failed to save incident history")
 		}
 	}()
 }
 
+// flush blocks until every asynchronous save queued by a mutation has
+// finished. A queued save still creates the data directory and writes its
+// temp file after the mutation returns, so callers that remove or inspect the
+// data directory (including tests) join it first. Stop mutating the store
+// before calling flush: WaitGroup forbids a new save being queued while a
+// previous Wait is still returning.
+func (s *IncidentStore) flush() {
+	s.saveWG.Wait()
+}
+
 func (s *IncidentStore) saveToDisk() error {
 	s.saveMu.Lock()
 	defer s.saveMu.Unlock()

internal/ai/memory/incidents_save_coalescing_test.go

@@ -1,6 +1,7 @@
 package memory
 
 import (
+	"os"
 	"testing"
 	"time"
 )
@@ -49,3 +50,39 @@ func TestIncidentStoreSaveCoalescing(t *testing.T) {
 		t.Fatalf("persisted events = %d, want all 101 mutations captured", len(timeline.Events))
 	}
 }
+
+func TestIncidentStoreFlushJoinsQueuedSave(t *testing.T) {
+	dir := t.TempDir()
+	store := NewIncidentStore(IncidentStoreConfig{DataDir: dir})
+
+	// Hold the save lock so the save queued by the mutation cannot start. This
+	// is the window where TempDir cleanup used to race it: the mutation has
+	// returned, but the goroutine has not yet created its temp file.
+	store.saveMu.Lock()
+	store.RecordAnalysis("alert-flush", "analysis", nil)
+
+	flushed := make(chan struct{})
+	go func() {
+		store.flush()
+		close(flushed)
+	}()
+	select {
+	case <-flushed:
+		store.saveMu.Unlock()
+		t.Fatal("flush returned while a queued save had not run")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	store.saveMu.Unlock()
+	select {
+	case <-flushed:
+	case <-time.After(5 * time.Second):
+		t.Fatal("flush did not return after the queued save ran")
+	}
+	if saves := store.savesCompleted.Load(); saves != 1 {
+		t.Fatalf("saves completed when flush returned = %d, want 1", saves)
+	}
+	if _, err := os.Stat(store.filePath + ".tmp"); !os.IsNotExist(err) {
+		t.Fatalf("temp file left behind after flush: %v", err)
+	}
+}

internal/ai/memory/memory_coverage_test.go

@@ -580,16 +580,10 @@ func TestIncidentStore_SaveAsyncAndPersistence(t *testing.T) {
 	}
 
 	store.saveAsync()
+	store.flush()
 
-	deadline := time.Now().Add(500 * time.Millisecond)
-	for {
-		if _, err := os.Stat(store.filePath); err == nil {
-			break
-		}
-		if time.Now().After(deadline) {
-			t.Fatalf("expected saveAsync to create file")
-		}
-		time.Sleep(10 * time.Millisecond)
+	if _, err := os.Stat(store.filePath); err != nil {
+		t.Fatalf("expected saveAsync to create file: %v", err)
 	}
 }
 
@@ -608,7 +602,12 @@ func TestIncidentStore_SaveAsync_Error(t *testing.T) {
 		filePath: filepath.Join(badDir, incidentFileName),
 	}
 	store.saveAsync()
-	time.Sleep(20 * time.Millisecond)
+	// An unjoined save could run MkdirAll after TempDir cleanup removed the
+	// blocking file, recreating the directory mid-removal.
+	store.flush()
+	if info, err := os.Stat(badDir); err != nil || info.IsDir() {
+		t.Fatalf("failed save must leave the blocking file in place: info=%v err=%v", info, err)
+	}
 }
 
 func TestIncidentStore_SaveToDisk_Scenarios(t *testing.T) {