test: fix the three data races that made -race runs noisy

None was introduced by this branch and all three are in test code, which is
what made them survive: every suite passed on every run and only the race
detector said otherwise. A known-failing -race run is worse than a noisy one,
because a real race raised by production code lands in the same report and is
read as one of these.

galleryop: gatedModelManager guarded the recorded names and not the gate
channel itself. A spec frees the parked worker by closing the gate and
installing a fresh one, on the spec goroutine, while the worker goroutine reads
the field to park on it. The channel is now read and replaced under the same
mutex, and cleanup closes idempotently.

pkg/model: two specs swapped xlog's package logger to capture output and
swapped it back on cleanup. xlog.SetLogger writes an unsynchronised global, so
the restore raced with the backend process watcher, which logs while a process
is stopping; the captured bytes.Buffer was written by that goroutine and read
by an Eventually at the same time. SetLogger is now called once for the whole
test binary, from init, before a goroutine exists to race with, and a spec
swaps the DESTINATION under a mutex through a routing slog.Handler. Per-spec
level filtering is preserved deliberately: one of these specs asserts that a
debug emission is filtered OUT and would pass vacuously against a handler that
recorded everything.

openai: fakeTransport appended to its event and audio logs from the response
and turn coordinators' goroutines while a spec ranged over them. Both are
behind a mutex and are read through snapshot accessors; the fields are renamed
so a raw read from another spec file does not compile.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/http/endpoints/openai/realtime_classifier_test.go

@@ -45,7 +45,7 @@ var classifierTestHistory = schema.Messages{
 
 func classifierResultEvents(t *fakeTransport) []types.ClassifierResultEvent {
 	var out []types.ClassifierResultEvent
-	for _, e := range t.events {
+	for _, e := range t.recordedEvents() {
 		if ev, ok := e.(types.ClassifierResultEvent); ok {
 			out = append(out, ev)
 		}
@@ -57,7 +57,7 @@ func classifierResultEvents(t *fakeTransport) []types.ClassifierResultEvent {
 // item — what a classifier response actually "spoke".
 func replyTexts(t *fakeTransport) []string {
 	var out []string
-	for _, e := range t.events {
+	for _, e := range t.recordedEvents() {
 		if ev, ok := e.(types.ResponseOutputTextDoneEvent); ok {
 			out = append(out, ev.Text)
 		}
@@ -277,7 +277,7 @@ var _ = Describe("classifierRespond", func() {
 		Expect(t.countEvents(types.ServerEventTypeResponseOutputTextDone)).To(Equal(1))
 		Expect(t.countEvents(types.ServerEventTypeResponseFunctionCallArgumentsDone)).To(Equal(1))
 		var fcArgs string
-		for _, e := range t.events {
+		for _, e := range t.recordedEvents() {
 			if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok {
 				fcArgs = done.Arguments
 			}
@@ -656,7 +656,7 @@ var _ = Describe("classifierRespond slot filling", func() {
 		Expect(results[0].Arguments).To(MatchJSON(`{"direction":"up","distance":3,"units":"meters"}`))
 
 		var fcArgs string
-		for _, e := range t.events {
+		for _, e := range t.recordedEvents() {
 			if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok {
 				fcArgs = done.Arguments
 			}
@@ -695,7 +695,7 @@ var _ = Describe("classifierRespond slot filling", func() {
 
 		Expect(handled).To(BeTrue())
 		var fcArgs string
-		for _, e := range t.events {
+		for _, e := range t.recordedEvents() {
 			if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok {
 				fcArgs = done.Arguments
 			}

core/http/endpoints/openai/realtime_doubles_test.go

@@ -16,9 +16,17 @@ import (
 // fakeTransport records the server events and audio sent to a realtime client
 // so streaming behaviour can be asserted without a real WebSocket/WebRTC peer.
 // It is not a *WebRTCTransport, so handler code takes the WebSocket path.
+//
+// Every field is behind the mutex, and the recorded slices are read only
+// through recordedEvents and recordedAudio. A real transport is written to by
+// the response and turn coordinators' goroutines while the spec goroutine
+// reads what has arrived so far, so a double that appended without a lock could
+// not be driven the way production drives it. Both fields are named with a
+// `Log` suffix so a raw read from another spec file does not compile.
 type fakeTransport struct {
-	events []types.ServerEvent
-	audio  []fakeAudioChunk
+	mu       sync.Mutex
+	eventLog []types.ServerEvent
+	audioLog []fakeAudioChunk
 }
 
 type fakeAudioChunk struct {
@@ -27,23 +35,42 @@ type fakeAudioChunk struct {
 }
 
 func (f *fakeTransport) SendEvent(e types.ServerEvent) error {
-	f.events = append(f.events, e)
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.eventLog = append(f.eventLog, e)
 	return nil
 }
 
 func (f *fakeTransport) ReadEvent() ([]byte, error) { return nil, nil }
 
 func (f *fakeTransport) SendAudio(_ context.Context, pcm []byte, sampleRate int) error {
-	f.audio = append(f.audio, fakeAudioChunk{pcm: pcm, sampleRate: sampleRate})
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.audioLog = append(f.audioLog, fakeAudioChunk{pcm: pcm, sampleRate: sampleRate})
 	return nil
 }
 
 func (f *fakeTransport) Close() error { return nil }
 
+// recordedEvents returns a snapshot of the events sent so far. A COPY, because
+// the caller ranges over it while the coordinators may still be sending.
+func (f *fakeTransport) recordedEvents() []types.ServerEvent {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]types.ServerEvent(nil), f.eventLog...)
+}
+
+// recordedAudio returns a snapshot of the audio chunks sent so far.
+func (f *fakeTransport) recordedAudio() []fakeAudioChunk {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]fakeAudioChunk(nil), f.audioLog...)
+}
+
 // countEvents returns how many recorded events have the given type.
 func (f *fakeTransport) countEvents(et types.ServerEventType) int {
 	n := 0
-	for _, e := range f.events {
+	for _, e := range f.recordedEvents() {
 		if e.ServerEventType() == et {
 			n++
 		}
@@ -55,7 +82,7 @@ func (f *fakeTransport) countEvents(et types.ServerEventType) int {
 // delta event — i.e. the text streamed to the client as it is generated.
 func (f *fakeTransport) transcriptDeltaText() string {
 	var b strings.Builder
-	for _, e := range f.events {
+	for _, e := range f.recordedEvents() {
 		if d, ok := e.(types.ResponseOutputAudioTranscriptDeltaEvent); ok {
 			b.WriteString(d.Delta)
 		}

core/http/endpoints/openai/realtime_semantic_vad_test.go

@@ -269,7 +269,7 @@ var _ = Describe("liveTurnState", func() {
 			lts.drainEvents(1.0)
 
 			var got []types.ConversationItemInputAudioTranscriptionDeltaEvent
-			for _, e := range ftr.events {
+			for _, e := range ftr.recordedEvents() {
 				if d, ok := e.(types.ConversationItemInputAudioTranscriptionDeltaEvent); ok {
 					got = append(got, d)
 				}
@@ -335,7 +335,7 @@ var _ = Describe("commitUtteranceWithTranscript", func() {
 		Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1))
 
 		var completed types.ConversationItemInputAudioTranscriptionCompletedEvent
-		for _, e := range tr.events {
+		for _, e := range tr.recordedEvents() {
 			if c, ok := e.(types.ConversationItemInputAudioTranscriptionCompletedEvent); ok {
 				completed = c
 			}
@@ -394,7 +394,7 @@ var _ = Describe("emitPrecomputedTranscription", func() {
 
 		Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionDelta)).To(Equal(2), "empty deltas skipped")
 		Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1))
-		for _, e := range tr.events {
+		for _, e := range tr.recordedEvents() {
 			switch ev := e.(type) {
 			case types.ConversationItemInputAudioTranscriptionDeltaEvent:
 				Expect(ev.ItemID).To(Equal("item42"))

core/http/endpoints/openai/realtime_sound_detection_test.go

@@ -38,7 +38,7 @@ var _ = Describe("emitSoundDetection", func() {
 		Expect(err).ToNot(HaveOccurred())
 		Expect(t.countEvents(types.ServerEventTypeConversationItemSoundDetection)).To(Equal(1))
 
-		ev, ok := t.events[0].(types.ConversationItemSoundDetectionEvent)
+		ev, ok := t.recordedEvents()[0].(types.ConversationItemSoundDetectionEvent)
 		Expect(ok).To(BeTrue())
 		Expect(ev.ItemID).To(Equal("item1"))
 		Expect(ev.ContentIndex).To(Equal(0))
@@ -62,7 +62,7 @@ var _ = Describe("emitSoundDetection", func() {
 
 		Expect(err).ToNot(HaveOccurred())
 		Expect(t.countEvents(types.ServerEventTypeConversationItemSoundDetection)).To(Equal(1))
-		ev, ok := t.events[0].(types.ConversationItemSoundDetectionEvent)
+		ev, ok := t.recordedEvents()[0].(types.ConversationItemSoundDetectionEvent)
 		Expect(ok).To(BeTrue())
 		Expect(ev.Detections).To(BeEmpty())
 	})

core/http/endpoints/openai/realtime_stream_test.go

@@ -250,8 +250,8 @@ var _ = Describe("triggerResponse", func() {
 		// The single terminal carries the produced output item and the usage —
 		// both empty in the legacy code.
 		var done *types.ResponseDoneEvent
-		for i := range t.events {
-			if d, ok := t.events[i].(types.ResponseDoneEvent); ok {
+		for _, e := range t.recordedEvents() {
+			if d, ok := e.(types.ResponseDoneEvent); ok {
 				done = &d
 			}
 		}
@@ -287,8 +287,8 @@ var _ = Describe("triggerResponse", func() {
 
 		var created *types.ResponseCreatedEvent
 		var done *types.ResponseDoneEvent
-		for i := range t.events {
-			switch e := t.events[i].(type) {
+		for _, sent := range t.recordedEvents() {
+			switch e := sent.(type) {
 			case types.ResponseCreatedEvent:
 				created = &e
 			case types.ResponseDoneEvent:
@@ -317,8 +317,8 @@ var _ = Describe("triggerResponse", func() {
 
 		triggerResponse(context.Background(), session, &Conversation{}, t, nil)
 
-		for i := range t.events {
-			if d, ok := t.events[i].(types.ResponseDoneEvent); ok {
+		for _, e := range t.recordedEvents() {
+			if d, ok := e.(types.ResponseDoneEvent); ok {
 				Expect(d.Response.Metadata).To(BeEmpty())
 			}
 		}

core/http/endpoints/openai/realtime_voicegate_integration_test.go

@@ -67,7 +67,7 @@ func itSession(gate *voiceGate) (*Session, *fakeModel) {
 // hasSpeakerNotAuthorized reports whether a speaker_not_authorized error event
 // was emitted to the client.
 func hasSpeakerNotAuthorized(tr *fakeTransport) bool {
-	for _, e := range tr.events {
+	for _, e := range tr.recordedEvents() {
 		if ev, ok := e.(types.ErrorEvent); ok && ev.Error.Code == "speaker_not_authorized" {
 			return true
 		}

core/services/galleryop/cancellable_phase_test.go

@@ -17,6 +17,13 @@ import (
 // Parking is what makes the running phase observable at all: without it the
 // handler-entry status is overwritten by the terminal write before a spec can
 // read it.
+//
+// The gate CHANNEL is under the mutex, not only the recorded names. A spec that
+// frees the worker replaces the channel so the next operation parks on a fresh
+// one, and the worker goroutine reads the field to park on it; the two happen
+// on different goroutines and raced. It was a race in this double and not in
+// anything it stands for, which is exactly why it survived: the suite passed
+// every run and only -race said so.
 type gatedModelManager struct {
 	mu      sync.Mutex
 	started []string
@@ -39,15 +46,44 @@ func (m *gatedModelManager) Started() []string {
 	return append([]string(nil), m.started...)
 }
 
+// gateCh reads the current gate, for a handler about to park on it.
+func (m *gatedModelManager) gateCh() chan struct{} {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.gate
+}
+
+// releaseGate frees every handler parked on the current gate and arms a fresh
+// one, so the next operation parks again.
+func (m *gatedModelManager) releaseGate() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	close(m.gate)
+	m.gate = make(chan struct{})
+}
+
+// closeGate frees every parked handler without arming another. It is what
+// cleanup does, and it is idempotent so a spec that already released can be
+// cleaned up after.
+func (m *gatedModelManager) closeGate() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	select {
+	case <-m.gate:
+	default:
+		close(m.gate)
+	}
+}
+
 func (m *gatedModelManager) InstallModel(_ context.Context, op *galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig], _ galleryop.ProgressCallback) error {
 	m.record(op.GalleryElementName)
-	<-m.gate
+	<-m.gateCh()
 	return nil
 }
 
 func (m *gatedModelManager) DeleteModel(name string) error {
 	m.record(name)
-	<-m.gate
+	<-m.gateCh()
 	return nil
 }
 
@@ -120,7 +156,7 @@ var _ = Describe("operation cancellability by phase", func() {
 
 		BeforeEach(func() {
 			manager = newGatedModelManager()
-			DeferCleanup(func() { close(manager.gate) })
+			DeferCleanup(manager.closeGate)
 
 			svc.SetModelManager(manager)
 			ctx, cancel := context.WithCancel(context.Background())
@@ -188,8 +224,7 @@ var _ = Describe("operation cancellability by phase", func() {
 
 			// Free the worker: it loops back to an empty channel because the
 			// delivery goroutine gave up on the send.
-			close(manager.gate)
-			manager.gate = make(chan struct{})
+			manager.releaseGate()
 
 			Consistently(manager.Started, "500ms", "20ms").ShouldNot(
 				ContainElement("localai@doomed-removal"),

pkg/model/initializers_load_logging_test.go

@@ -1,12 +1,10 @@
 package model
 
 import (
-	"bytes"
 	"log/slog"
 
 	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
 	"github.com/mudler/LocalAI/pkg/system"
-	"github.com/mudler/xlog"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -23,7 +21,7 @@ import (
 var _ = Describe("backendLoader load logging", func() {
 	var (
 		ml       *ModelLoader
-		captured *bytes.Buffer
+		captured *syncBuffer
 	)
 
 	BeforeEach(func() {
@@ -33,16 +31,10 @@ var _ = Describe("backendLoader load logging", func() {
 
 		// Capture at info level so a debug-level emission is filtered out: the
 		// assertions then fail on severity, not merely on wording.
-		captured = &bytes.Buffer{}
-		handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelInfo})
-		xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelInfo))
+		captured = captureLogs(slog.LevelInfo)
 	})
 
-	AfterEach(func() {
-		// xlog exposes no getter for the package logger, so restore the same
-		// default the suite entrypoint installs rather than the prior value.
-		xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text"))
-	})
+	AfterEach(stopCapturingLogs)
 
 	Context("when the model is already resident", func() {
 		BeforeEach(func() {

pkg/model/log_capture_test.go

@@ -0,0 +1,104 @@
+package model
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"sync"
+
+	"github.com/mudler/xlog"
+)
+
+// xlog keeps its logger in a package-level variable that SetLogger writes
+// without a lock, so every xlog.Info in the process reads that variable
+// concurrently with the write. A spec that swapped the logger to capture output
+// and swapped it back on cleanup therefore raced with any goroutine this
+// package had left logging: the backend process watcher, which logs while a
+// process is stopping, is one of those on every run.
+//
+// So SetLogger is called exactly ONCE for this whole test binary, from init,
+// before a goroutine exists to race with. What a spec swaps afterwards is the
+// destination, under a mutex, through the routing handler below. That keeps the
+// per-spec level filtering intact, which matters: at least one spec asserts
+// that a debug emission is filtered OUT, and would pass vacuously against a
+// handler that simply recorded everything.
+type routingLogHandler struct {
+	mu sync.Mutex
+	to slog.Handler
+}
+
+// sharedLogHandler is the one handler xlog is given for this binary.
+var sharedLogHandler = &routingLogHandler{}
+
+func init() {
+	xlog.SetLogger(xlog.NewLoggerWithHandler(sharedLogHandler, xlog.LogLevelInfo))
+}
+
+func (h *routingLogHandler) current() slog.Handler {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.to
+}
+
+// arm points the shared handler at inner, or discards everything when inner is
+// nil. Returns nothing: a spec restores by arming nil, because xlog exposes no
+// getter and there is no previous value to hand back.
+func (h *routingLogHandler) arm(inner slog.Handler) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.to = inner
+}
+
+func (h *routingLogHandler) Enabled(ctx context.Context, level slog.Level) bool {
+	inner := h.current()
+	return inner != nil && inner.Enabled(ctx, level)
+}
+
+func (h *routingLogHandler) Handle(ctx context.Context, r slog.Record) error {
+	inner := h.current()
+	if inner == nil {
+		return nil
+	}
+	return inner.Handle(ctx, r)
+}
+
+// WithAttrs and WithGroup hand back the router itself. xlog never calls either
+// (it has no With), and a copy would be a second handler holding the same
+// mutex by value.
+func (h *routingLogHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
+func (h *routingLogHandler) WithGroup(string) slog.Handler      { return h }
+
+var _ slog.Handler = (*routingLogHandler)(nil)
+
+// syncBuffer is a bytes.Buffer that a log handler and a poller may share.
+//
+// The diagnostic under test is written by a goroutine and read by an Eventually
+// on the spec goroutine, which is a plain concurrent use of a bytes.Buffer.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+// captureLogs sends everything logged through xlog at or above level into a
+// fresh buffer, until stopCapturingLogs is called.
+func captureLogs(level slog.Level) *syncBuffer {
+	captured := &syncBuffer{}
+	sharedLogHandler.arm(slog.NewTextHandler(captured, &slog.HandlerOptions{Level: level}))
+	return captured
+}
+
+// stopCapturingLogs sends everything logged afterwards nowhere, which is what
+// a test binary wants of a package whose goroutines outlive their spec.
+func stopCapturingLogs() { sharedLogHandler.arm(nil) }

pkg/model/process_exit_test.go

@@ -1,13 +1,11 @@
 package model
 
 import (
-	"bytes"
 	"log/slog"
 	"os"
 	"path/filepath"
 
 	"github.com/mudler/LocalAI/pkg/system"
-	"github.com/mudler/xlog"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -18,12 +16,8 @@ var _ = Describe("backend process exit diagnostics", func() {
 		backendPath := filepath.Join(tmpDir, "failing-backend")
 		Expect(os.WriteFile(backendPath, []byte("#!/bin/sh\necho 'first diagnostic' >&2\necho 'fatal metal pipeline error' >&2\nexit 42\n"), 0o700)).To(Succeed())
 
-		captured := &bytes.Buffer{}
-		handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelWarn})
-		xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelWarn))
-		DeferCleanup(func() {
-			xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text"))
-		})
+		captured := captureLogs(slog.LevelWarn)
+		DeferCleanup(stopCapturingLogs)
 
 		loader := NewModelLoader(&system.SystemState{Model: system.Model{ModelsPath: tmpDir}})
 		process, err := loader.startProcess(backendPath, "test-model", "127.0.0.1:65535")