fix(ai): serialize KnowledgeStore disk saves and track in-flight writes

SaveNote launched an untracked goroutine per note. Concurrent saves shared the fixed knowledge_store.json.tmp path, so the atomic temp+fsync+rename write could be defeated by an interleaved truncate/rename, and a still-running writer could recreate files in the data directory after a caller removed it (the flaky t.TempDir cleanup observed in TestKnowledgeStore_SaveLoad during release qualification).

Guard disk writes with a dedicated save mutex so only one save touches the temp path at a time, track in-flight saves with a WaitGroup, and add an unexported flush() the test uses to join them before cleanup. Add a concurrent SaveNote reload regression covering all entries, and record the persistence boundary in the ai-runtime subsystem contract.

Change-source: pulse-maintainer

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

@@ -8424,3 +8424,15 @@ passing because capability wiring is absent. Surface regressions also remove
 the planner while retaining execute authority and transport, and assert control
 is absent in both provider tools and the surface contract. These are offline
 fixture and policy proofs, not hosted-model or installed acceptance.
+
+### Knowledge-store persistence serialization
+
+`internal/ai/adapters.KnowledgeStore` persists resource notes through an atomic
+temp-file, fsync and rename. Concurrent `SaveNote` calls must not interleave on
+that temp path: disk writes are serialized behind a dedicated save mutex, and
+asynchronous saves are tracked so a caller that removes or inspects the data
+directory can join them. An observer must read either the previous store or a
+complete new snapshot, never a partially written file. This is an internal
+durability boundary, not a public API or stored-format change. The
+`TestKnowledgeStore_SaveLoad` cleanup join and the concurrent-save reload
+regression in `internal/ai/adapters/adapters_additional_test.go` enforce it.

internal/ai/adapters/adapters.go

@@ -155,6 +155,8 @@ func (a *EventCorrelatorToolAdapter) GetCorrelationsForResource(resourceID strin
 // KnowledgeStore provides persistent storage for resource notes
 type KnowledgeStore struct {
 	mu      sync.RWMutex
+	saveMu  sync.Mutex
+	saveWG  sync.WaitGroup
 	entries map[string][]KnowledgeEntry
 	dataDir string
 }
@@ -213,7 +215,11 @@ func (s *KnowledgeStore) SaveNote(resourceID, note, category string) error {
 	s.entries[resourceID] = append(s.entries[resourceID], entry)
 
 	if s.dataDir != "" {
-		go s.saveToDisk() // Async save
+		s.saveWG.Add(1)
+		go func() {
+			defer s.saveWG.Done()
+			s.saveToDisk()
+		}()
 	}
 
 	return nil
@@ -247,7 +253,17 @@ func (s *KnowledgeStore) GetKnowledge(resourceID string, category string) []Know
 	return filtered
 }
 
+// flush blocks until every asynchronous save started by SaveNote has finished.
+// Callers that remove or inspect the data directory (including tests) use it to
+// avoid racing a still-running writer.
+func (s *KnowledgeStore) flush() {
+	s.saveWG.Wait()
+}
+
 func (s *KnowledgeStore) saveToDisk() {
+	s.saveMu.Lock()
+	defer s.saveMu.Unlock()
+
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 

internal/ai/adapters/adapters_additional_test.go

@@ -1,8 +1,10 @@
 package adapters
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -73,6 +75,7 @@ func TestEventCorrelatorToolAdapter(t *testing.T) {
 func TestKnowledgeStore_SaveLoad(t *testing.T) {
 	dir := t.TempDir()
 	store := NewKnowledgeStore(dir)
+	defer store.flush()
 
 	if err := store.SaveNote("agent:res-1", "note", "general"); err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -128,3 +131,32 @@ func TestKnowledgeStore_LoadFromDisk_DoesNotCanonicalizeLegacyHostKeys(t *testin
 		t.Fatalf("expected unsupported host query alias to be rejected")
 	}
 }
+
+func TestKnowledgeStore_ConcurrentSaveNotePersistsAllEntries(t *testing.T) {
+	dir := t.TempDir()
+	store := NewKnowledgeStore(dir)
+
+	const count = 25
+	var wg sync.WaitGroup
+	for i := 0; i < count; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			if err := store.SaveNote(fmt.Sprintf("agent:res-%d", i), "note", "general"); err != nil {
+				t.Errorf("SaveNote(agent:res-%d): %v", i, err)
+			}
+		}(i)
+	}
+	wg.Wait()
+	store.flush()
+
+	loaded := NewKnowledgeStore(dir)
+	if err := loaded.loadFromDisk(); err != nil {
+		t.Fatalf("load after concurrent saves: %v", err)
+	}
+	for i := 0; i < count; i++ {
+		if got := loaded.GetKnowledge(fmt.Sprintf("agent:res-%d", i), "general"); len(got) != 1 {
+			t.Fatalf("expected exactly one entry for agent:res-%d, got %d", i, len(got))
+		}
+	}
+}