From c27848c8c7c73fbcec256d7a3d90760cdd6f7f7b Mon Sep 17 00:00:00 2001 From: kai Date: Sat, 10 Oct 2026 22:10:17 +0000 Subject: [PATCH] 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 (cherry picked from commit 10555084bfb6e543097a04cd84d3fb8f0d492a55) Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/openai/realtime.go | 92 +++++++++++++-- .../openai/realtime_client_commit_test.go | 106 ++++++++++++++++++ .../openai/realtime_commit_order_test.go | 2 +- .../openai/realtime_interrupt_test.go | 8 +- .../endpoints/openai/realtime_turncoord.go | 2 +- 5 files changed, 193 insertions(+), 17 deletions(-) create mode 100644 core/http/endpoints/openai/realtime_client_commit_test.go diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index cd61329f2..17474a663 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/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)) diff --git a/core/http/endpoints/openai/realtime_client_commit_test.go b/core/http/endpoints/openai/realtime_client_commit_test.go new file mode 100644 index 000000000..272dea2de --- /dev/null +++ b/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)) + }) +}) diff --git a/core/http/endpoints/openai/realtime_commit_order_test.go b/core/http/endpoints/openai/realtime_commit_order_test.go index a5b024c18..31456257f 100644 --- a/core/http/endpoints/openai/realtime_commit_order_test.go +++ b/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) }) } diff --git a/core/http/endpoints/openai/realtime_interrupt_test.go b/core/http/endpoints/openai/realtime_interrupt_test.go index 35b875729..835ef4ddb 100644 --- a/core/http/endpoints/openai/realtime_interrupt_test.go +++ b/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) }) diff --git a/core/http/endpoints/openai/realtime_turncoord.go b/core/http/endpoints/openai/realtime_turncoord.go index 066614b5a..e7749b959 100644 --- a/core/http/endpoints/openai/realtime_turncoord.go +++ b/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: