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: