fix(realtime): wait for the client commit's STT before the response LLM starts

With client-driven turns (turn_detection null, e.g. a client VAD / speaker
gate) the client sends input_audio_buffer.commit and response.create
back-to-back. The response.create triggered the LLM immediately while the
commit's transcription was still in flight, so the LLM (fast) won the race
and answered an empty transcript; the response on the real transcript was
then discarded ("skipping response: turn context cancelled during
transcription").

Record the in-flight client commit's slot done channel (closed once the
commit has transcribed and appended its user item) and make the client
response.wait for it before the LLM starts, so the LLM only runs after the
STT is complete.

- issueCommit gains a slotSink callback (slot claim + issue stay under one
  lock); the VAD commit path passes nil.
- The client commit / response.create handler cores are extracted to
  issueClientCommit / issueClientResponse so the real path is unit-testable.
- Regression specs drive the real issue path with a held transcription: the
  LLM must not start until the STT completes, and it must see the user's
  transcript.

Signed-off-by: nexxtmobile.de <kai@nexxtmobile.de>
(cherry picked from commit 10555084bf)
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/http/endpoints/openai/realtime.go

@@ -220,6 +220,15 @@ type Session struct {
 	// user-item appends in speech order, not transcription-completion order
 	// (see commitSlot).
 	commitTail *commitSlot
+
+	// clientCommitMu guards clientCommitDone.
+	clientCommitMu sync.Mutex
+	// clientCommitDone is the done channel of the in-flight client commit
+	// (input_audio_buffer.commit), closed once that commit has transcribed and
+	// appended its user item. A response.create issued while it is open waits
+	// on it so the LLM starts only after the STT is complete (client-commit
+	// LLM/STT race, see clientCommitDone).
+	clientCommitDone chan struct{}
 }
 
 // commitSlot orders committed user turns in the conversation. The
@@ -281,10 +290,13 @@ func (session *Session) nextCommitSlotLocked() *commitSlot {
 // registers the body and applies the coordinator start; the commit work runs
 // in the spawned goroutine — so holding the lock across it does not stall
 // VAD/barge-in handling.
-func (session *Session) issueCommit(parent context.Context, source respcoord.Source, run func(ctx context.Context, slot *commitSlot)) {
+func (session *Session) issueCommit(parent context.Context, source respcoord.Source, slotSink func(*commitSlot), run func(ctx context.Context, slot *commitSlot)) {
 	session.commitOrderMu.Lock()
 	defer session.commitOrderMu.Unlock()
 	slot := session.nextCommitSlotLocked()
+	if slotSink != nil {
+		slotSink(slot)
+	}
 	// A non-interrupting VAD commit must not supersede a response at speech end.
 	ifIdle := source == respcoord.SourceVAD && !session.interruptResponseEnabled()
 	session.respSink.issueWithPolicy(parent, source, ifIdle, func(ctx context.Context) {
@@ -292,6 +304,65 @@ func (session *Session) issueCommit(parent context.Context, source respcoord.Sou
 	})
 }
 
+// setClientCommitDone records the in-flight client commit's slot done channel
+// (closed once that commit has transcribed and appended its user item) so a
+// following response.create can wait for the STT to complete. See
+// getClientCommitDone.
+func (session *Session) setClientCommitDone(done chan struct{}) {
+	session.clientCommitMu.Lock()
+	session.clientCommitDone = done
+	session.clientCommitMu.Unlock()
+}
+
+// getClientCommitDone returns the in-flight client commit's slot done channel
+// (nil if none is in flight). With client-driven turns (turn_detection null,
+// e.g. a client VAD / speaker gate) the client sends input_audio_buffer.commit
+// and response.create back-to-back; without waiting on this channel the LLM
+// (fast) races the commit's STT (slow) and answers an empty transcript while
+// the response on the real transcript is discarded (client-commit LLM/STT
+// race).
+func (session *Session) getClientCommitDone() chan struct{} {
+	session.clientCommitMu.Lock()
+	defer session.clientCommitMu.Unlock()
+	return session.clientCommitDone
+}
+
+// issueClientCommit issues a client commit (input_audio_buffer.commit) through
+// the shared commit-ordering boundary (slot claim + issue under one lock, shared
+// with the VAD commit path) and records the in-flight commit's slot done channel
+// so the back-to-back response.create can wait for the STT before the LLM starts
+// (client-commit LLM/STT race).
+func (session *Session) issueClientCommit(allAudio []byte, conv *Conversation, t Transport) {
+	var cs *commitSlot
+	session.issueCommit(context.Background(), respcoord.SourceClient, func(slot *commitSlot) {
+		cs = slot
+	}, func(ctx context.Context, slot *commitSlot) {
+		commitUtterance(ctx, allAudio, session, conv, t, slot)
+	})
+	if cs != nil {
+		session.setClientCommitDone(cs.done)
+	}
+}
+
+// issueClientResponse issues a client response (response.create). With
+// client-driven turns the commit and response.create are sent back-to-back; the
+// response waits for the in-flight commit's STT (and user-item append) to finish
+// so the LLM answers the real transcript, not an empty one (client-commit
+// LLM/STT race).
+func (session *Session) issueClientResponse(resp types.ResponseCreateParams, conv *Conversation, t Transport) {
+	commitDone := session.getClientCommitDone()
+	session.respSink.issue(context.Background(), respcoord.SourceClient, func(ctx context.Context) {
+		if commitDone != nil {
+			select {
+			case <-commitDone:
+			case <-ctx.Done():
+				return
+			}
+		}
+		triggerResponse(ctx, session, conv, t, &resp)
+	})
+}
+
 func (s *Session) installVoiceBinding(voice string, params map[string]string, release func()) {
 	if release == nil {
 		release = func() {}
@@ -1038,12 +1109,10 @@ func runRealtimeSession(application *application.Application, t Transport, model
 				ItemID:          generateItemID(),
 			})
 
-			// Issue through the shared commit-ordering boundary (slot claim +
-			// issue under one lock, shared with the VAD commit path) so slot
-			// order == issue order across both producers (issue #12445).
-			session.issueCommit(context.Background(), respcoord.SourceClient, func(ctx context.Context, slot *commitSlot) {
-				commitUtterance(ctx, allAudio, session, conversation, t, slot)
-			})
+			// Issue through the shared commit-ordering boundary and record the
+			// in-flight commit so the back-to-back response.create waits for the
+			// STT before the LLM starts (client-commit LLM/STT race).
+			session.issueClientCommit(allAudio, conversation, t)
 
 		case types.InputAudioBufferClearEvent:
 			xlog.Debug("recv", "message", string(msg))
@@ -1180,10 +1249,11 @@ func runRealtimeSession(application *application.Application, t Transport, model
 				conversation.Lock.Unlock()
 			}
 
-			resp := e.Response
-			session.respSink.issue(context.Background(), respcoord.SourceClient, func(ctx context.Context) {
-				triggerResponse(ctx, session, conversation, t, &resp)
-			})
+			// With client-driven turns the commit and this response.create are
+			// sent back-to-back; the response waits for the in-flight commit's
+			// STT (and user-item append) to finish so the LLM answers the real
+			// transcript, not an empty one (client-commit LLM/STT race).
+			session.issueClientResponse(e.Response, conversation, t)
 
 		case types.ResponseCancelEvent:
 			xlog.Debug("recv", "message", string(msg))

core/http/endpoints/openai/realtime_client_commit_test.go

@@ -0,0 +1,106 @@
+package openai
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/mudler/LocalAI/core/backend"
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/core/http/endpoints/openai/types"
+)
+
+// Regression tests for the client-commit LLM/STT race. With client-driven
+// turns (turn_detection null, e.g. a client VAD / speaker gate) the client
+// sends input_audio_buffer.commit and response.create back-to-back. Without the
+// fix, response.create triggers the LLM immediately while the commit's STT is
+// still in flight, so the LLM (fast) wins the race and answers an EMPTY
+// transcript; the response on the real transcript is then discarded ("skipping
+// response: turn context cancelled during transcription"). The fix makes the
+// response wait for the in-flight commit's STT (and user-item append) before
+// the LLM starts.
+var _ = Describe("realtime client-commit LLM/STT ordering", func() {
+	utt := []byte{1, 2, 3, 4}
+
+	newModel := func(texts ...string) *heldTranscribeModel {
+		return &heldTranscribeModel{
+			fakeModel: &fakeModel{
+				cfg:             &config.ModelConfig{},
+				predictTokens:   []string{"ok"},
+				predictResp:     backend.LLMResponse{Response: "ok"},
+				ttsStreamChunks: [][]byte{{1}},
+				ttsStreamRate:   24000,
+			},
+			texts:    texts,
+			release:  make(chan struct{}),
+			started:  make(chan int32, 8),
+			finished: make(chan int32, 8),
+		}
+	}
+
+	newSession := func(m *heldTranscribeModel) *Session {
+		on := true
+		session := &Session{
+			OutputSampleRate:        24000,
+			InputAudioTranscription: &types.AudioTranscription{},
+			ModelInterface:          m,
+			ModelConfig: &config.ModelConfig{
+				Pipeline: config.Pipeline{Streaming: config.PipelineStreaming{LLM: &on, TTS: &on}},
+			},
+			respSink: newResponseSink(),
+		}
+		session.sessionCtx, session.sessionCancel = context.WithCancel(context.Background())
+		return session
+	}
+
+	userMsgTextsOf := func(m *heldTranscribeModel) []string {
+		var out []string
+		for _, msg := range m.lastMessages {
+			if msg.Role == string(types.MessageRoleUser) {
+				out = append(out, msg.StringContent)
+			}
+		}
+		return out
+	}
+
+	It("does not start the LLM until the client commit's STT is complete", func() {
+		m := newModel("hello there")
+		session := newSession(m)
+		tr := &fakeTransport{}
+		conv := &Conversation{}
+
+		// Client commit — its transcription is held by the double.
+		session.issueClientCommit(utt, conv, tr)
+		Eventually(m.started, "2s").Should(Receive(Equal(int32(1))))
+
+		// Back-to-back response.create (supersedes the commit's own response).
+		// With the fix it WAITS for the held STT: no response is created while
+		// the transcription is in flight.
+		session.issueClientResponse(types.ResponseCreateParams{}, conv, tr)
+		Consistently(tr.countEvents(types.ServerEventTypeResponseCreated), 200*time.Millisecond).Should(BeZero())
+
+		// Release the STT; the commit appends its user item and the waiting
+		// response now starts on the real transcript.
+		close(m.release)
+		session.respSink.wait()
+
+		Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1))
+		// The LLM saw the user's transcript, not an empty one.
+		Expect(userMsgTextsOf(m)).To(Equal([]string{"hello there"}))
+	})
+
+	It("still responds to a standalone response.create with no commit in flight", func() {
+		m := newModel()
+		session := newSession(m)
+		tr := &fakeTransport{}
+		conv := &Conversation{}
+
+		// No commit in flight: the response must not wait (no deadlock).
+		session.issueClientResponse(types.ResponseCreateParams{}, conv, tr)
+		session.respSink.wait()
+
+		Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1))
+	})
+})

core/http/endpoints/openai/realtime_commit_order_test.go

@@ -121,7 +121,7 @@ var _ = Describe("realtime commit order and transcription lifetime (issue #12445
 	// real respcoord supersession — NOT a direct
 	// commitUtteranceWithTranscript call.
 	issue := func(session *Session, parent context.Context, source respcoord.Source, conv *Conversation, tr *fakeTransport) {
-		session.issueCommit(parent, source, func(ctx context.Context, slot *commitSlot) {
+		session.issueCommit(parent, source, nil, func(ctx context.Context, slot *commitSlot) {
 			commitUtteranceWithTranscript(ctx, utt, nil, nil, "", session, conv, tr, slot)
 		})
 	}

core/http/endpoints/openai/realtime_interrupt_test.go

@@ -61,7 +61,7 @@ var _ = Describe("realtime interrupt_response", func() {
 			ctx := holdResponse(s)
 			conv, tr := &Conversation{}, &fakeTransport{}
 			committed := make(chan struct{})
-			s.issueCommit(context.Background(), respcoord.SourceVAD, func(turnCtx context.Context, slot *commitSlot) {
+			s.issueCommit(context.Background(), respcoord.SourceVAD, nil, func(turnCtx context.Context, slot *commitSlot) {
 				commitUtteranceWithTranscript(turnCtx, []byte{1, 2, 3, 4}, nil, nil, "", s, conv, tr, slot)
 				close(committed)
 			})
@@ -86,7 +86,7 @@ var _ = Describe("realtime interrupt_response", func() {
 		holdResponse(s)
 		conv, tr := &Conversation{}, &fakeTransport{}
 		committed := make(chan struct{})
-		s.issueCommit(context.Background(), respcoord.SourceVAD, func(turnCtx context.Context, slot *commitSlot) {
+		s.issueCommit(context.Background(), respcoord.SourceVAD, nil, func(turnCtx context.Context, slot *commitSlot) {
 			commitUtteranceWithTranscript(turnCtx, []byte{1, 2, 3, 4}, nil, nil, "", s, conv, tr, slot)
 			close(committed)
 		})
@@ -115,7 +115,7 @@ var _ = Describe("realtime interrupt_response", func() {
 			s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()}
 			DeferCleanup(s.respSink.shutdown)
 			started := make(chan context.Context, 1)
-			s.issueCommit(context.Background(), source, func(ctx context.Context, slot *commitSlot) {
+			s.issueCommit(context.Background(), source, nil, func(ctx context.Context, slot *commitSlot) {
 				defer close(slot.done)
 				started <- ctx
 				<-ctx.Done()
@@ -132,7 +132,7 @@ var _ = Describe("realtime interrupt_response", func() {
 		s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()}
 		ctx := holdResponse(s)
 		done := make(chan struct{})
-		s.issueCommit(context.Background(), respcoord.SourceClient, func(_ context.Context, slot *commitSlot) {
+		s.issueCommit(context.Background(), respcoord.SourceClient, nil, func(_ context.Context, slot *commitSlot) {
 			close(slot.done)
 			close(done)
 		})

core/http/endpoints/openai/realtime_turncoord.go

@@ -130,7 +130,7 @@ func (s *turnSink) Perform(e turncoord.Effect) {
 		// Issue through the shared commit-ordering boundary (slot claim +
 		// issue under one lock, shared with the client commit path) so slot
 		// order == issue order across both producers (issue #12445).
-		s.session.issueCommit(s.vadContext, respcoord.SourceVAD, func(ctx context.Context, slot *commitSlot) {
+		s.session.issueCommit(s.vadContext, respcoord.SourceVAD, nil, func(ctx context.Context, slot *commitSlot) {
 			commitUtteranceWithTranscript(ctx, audio, live, gated, itemID, s.session, conv, s.transport, slot)
 		})
 	case turncoord.DiscardTurn: