From 008f9fdf8160cfeaa2f23ce69385dbfc3e554ef1 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 13 Oct 2026 08:44:23 +0000 Subject: [PATCH] fix(realtime): end a client response cleanly when its commit fails A client response.create waits for the in-flight commit before the LLM starts. Two exits left the client without a usable outcome: - When the wait was cancelled (response.cancel, a newer response or teardown) the response returned with no event at all. - When the commit produced no user item (transcription error, empty transcript, voice gate rejection) the LLM ran on a history that lacked the turn. Send response.created and response.done with status cancelled for the first case. For the second, do not run the LLM: send response.done with status failed and an input_not_committed error. The commit still reports its own cause. A response.create with no commit in flight is unchanged. Add specs for STT error, empty transcript, cancel and supersede during the wait. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/openai/realtime.go | 56 +++++++++ .../openai/realtime_client_commit_test.go | 115 ++++++++++++++++++ 2 files changed, 171 insertions(+) diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index 00a67519f..17592e3bd 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/core/http/endpoints/openai/realtime.go @@ -243,6 +243,10 @@ type commitSlot struct { // its own response unless that was cancelled — AND the predecessor has // finished, so the next commit is released in order. done chan struct{} + // appended records that this commit appended its user item to the + // conversation. It is written by the commit goroutine before done closes + // and must only be read after done is closed. + appended bool } // nextCommitSlot claims the next position in the commit order. It is called @@ -326,6 +330,27 @@ func (session *Session) issueClientResponse(resp types.ResponseCreateParams, con select { case <-tail.done: case <-ctx.Done(): + } + if ctx.Err() != nil { + // Cancelled, superseded or torn down while waiting: close the + // response the way a cancelled one is closed, so the client is + // never left without a terminal event. + sendUnstartedResponse(t, &resp, types.ResponseStatusCancelled, nil) + return + } + if !tail.appended { + // The commit this response waited for produced no user item + // (transcription error, empty transcript, voice gate + // rejection; the commit already reported the cause). Do not + // run the LLM on a history that lacks the turn. + sendUnstartedResponse(t, &resp, types.ResponseStatusFailed, &types.StatusDetail{ + Type: "failed", + Error: &types.Error{ + Type: "invalid_request_error", + Code: "input_not_committed", + Message: "the committed input audio produced no transcript, so no response was generated", + }, + }) return } } @@ -333,6 +358,35 @@ func (session *Session) issueClientResponse(resp types.ResponseCreateParams, con }) } +// sendUnstartedResponse emits the response.created / response.done pair for a +// response that ended before the LLM started, so the client sees a complete +// lifecycle with the given terminal status. +func sendUnstartedResponse(t Transport, overrides *types.ResponseCreateParams, status types.ResponseStatus, details *types.StatusDetail) { + r := &liveResponse{id: generateUniqueID()} + if overrides != nil { + r.metadata = overrides.Metadata + } + sendEvent(t, types.ResponseCreatedEvent{ + ServerEventBase: types.ServerEventBase{}, + Response: types.Response{ + ID: r.id, + Object: "realtime.response", + Status: types.ResponseStatusInProgress, + Metadata: r.metadata, + }, + }) + sendEvent(t, types.ResponseDoneEvent{ + ServerEventBase: types.ServerEventBase{}, + Response: types.Response{ + ID: r.id, + Object: "realtime.response", + Status: status, + StatusDetails: details, + Metadata: r.metadata, + }, + }) +} + func (s *Session) installVoiceBinding(voice string, params map[string]string, release func()) { if release == nil { release = func() {} @@ -2218,9 +2272,11 @@ func commitUtteranceWithTranscript(ctx context.Context, utt []byte, live *liveUt // when the session context was cancelled.) if ctx.Err() != nil { xlog.Debug("skipping response: turn context cancelled during transcription (barge-in); committing user item only") + slot.appended = true appendUserItem(conv, t, utt, transcript, speaker) return } + slot.appended = true generateResponse(ctx, session, utt, transcript, speaker, conv, t) } } diff --git a/core/http/endpoints/openai/realtime_client_commit_test.go b/core/http/endpoints/openai/realtime_client_commit_test.go index eb861e24d..8983ad3ad 100644 --- a/core/http/endpoints/openai/realtime_client_commit_test.go +++ b/core/http/endpoints/openai/realtime_client_commit_test.go @@ -2,6 +2,7 @@ package openai import ( "context" + "errors" "time" . "github.com/onsi/ginkgo/v2" @@ -9,6 +10,7 @@ import ( "github.com/mudler/LocalAI/core/backend" "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/http/endpoints/openai/respcoord" "github.com/mudler/LocalAI/core/http/endpoints/openai/types" ) @@ -105,6 +107,95 @@ var _ = Describe("realtime client-commit LLM/STT ordering", func() { Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1)) }) + // responseDones returns the response.done payloads sent so far. + responseDones := func(tr *fakeTransport) []types.Response { + var out []types.Response + for _, ev := range tr.events() { + if d, ok := ev.(types.ResponseDoneEvent); ok { + out = append(out, d.Response) + } + } + return out + } + + // predictCalled reports whether the response LLM ran: Predict records the + // prompt it was given. + predictCalled := func(m *heldTranscribeModel) bool { return m.lastMessages != nil } + + DescribeTable("does not run the LLM when the commit produced no user item", + func(texts []string, errs []error) { + m := newModel(texts...) + m.errs = errs + session := newSession(m) + tr := &fakeTransport{} + conv := &Conversation{} + + session.issueClientCommit(utt, conv, tr) + Eventually(m.started, "2s").Should(Receive(Equal(int32(1)))) + session.issueClientResponse(types.ResponseCreateParams{}, conv, tr) + close(m.release) + session.respSink.wait() + + Expect(predictCalled(m)).To(BeFalse()) + Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1)) + dones := responseDones(tr) + Expect(dones).To(HaveLen(1)) + Expect(dones[0].Status).To(Equal(types.ResponseStatusFailed)) + Expect(dones[0].StatusDetails).NotTo(BeNil()) + Expect(dones[0].StatusDetails.Error).NotTo(BeNil()) + Expect(dones[0].StatusDetails.Error.Code).To(Equal("input_not_committed")) + conv.Lock.Lock() + defer conv.Lock.Unlock() + Expect(conv.Items).To(BeEmpty()) + }, + Entry("STT error", []string{"x"}, []error{errors.New("stt down")}), + Entry("empty transcript", []string{" "}, nil), + ) + + It("closes the response when it is cancelled while waiting for the commit", func() { + m := newModel("hello there") + session := newSession(m) + tr := &fakeTransport{} + conv := &Conversation{} + + session.issueClientCommit(utt, conv, tr) + Eventually(m.started, "2s").Should(Receive(Equal(int32(1)))) + session.issueClientResponse(types.ResponseCreateParams{Metadata: map[string]string{"k": "v"}}, conv, tr) + Consistently(func() int { return len(responseDones(tr)) }, 100*time.Millisecond).Should(BeZero()) + + session.respSink.cancel(respcoord.SourceClient) + Eventually(func() []types.Response { return responseDones(tr) }, "2s").Should(HaveLen(1)) + dones := responseDones(tr) + Expect(dones[0].Status).To(Equal(types.ResponseStatusCancelled)) + Expect(dones[0].Metadata).To(HaveKeyWithValue("k", "v")) + Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1)) + Expect(predictCalled(m)).To(BeFalse()) + + close(m.release) + session.respSink.wait() + Expect(responseDones(tr)).To(HaveLen(1)) + }) + + It("closes the response when a newer response supersedes it during the wait", func() { + m := newModel("hello there") + session := newSession(m) + tr := &fakeTransport{} + conv := &Conversation{} + + session.issueClientCommit(utt, conv, tr) + Eventually(m.started, "2s").Should(Receive(Equal(int32(1)))) + session.issueClientResponse(types.ResponseCreateParams{}, conv, tr) + session.issueClientResponse(types.ResponseCreateParams{}, conv, tr) + Eventually(func() []types.Response { return responseDones(tr) }, "2s").Should(HaveLen(1)) + Expect(responseDones(tr)[0].Status).To(Equal(types.ResponseStatusCancelled)) + + close(m.release) + session.respSink.wait() + dones := responseDones(tr) + Expect(dones).To(HaveLen(2)) + Expect(dones[1].Status).To(Equal(types.ResponseStatusCompleted)) + }) + It("orders commit, commit, response: the LLM sees both turns in speech order", func() { m := newModel("first turn", "second turn") session := newSession(m) @@ -128,6 +219,30 @@ var _ = Describe("realtime client-commit LLM/STT ordering", func() { session.respSink.wait() Expect(userMsgTextsOf(m)).To(Equal([]string{"first turn", "second turn"})) + dones := responseDones(tr) + Expect(dones).To(HaveLen(1)) + Expect(dones[0].Status).To(Equal(types.ResponseStatusCompleted)) }) + It("does not fail a standalone response after an earlier commit failed", func() { + m := newModel("x") + m.errs = []error{errors.New("stt down")} + session := newSession(m) + tr := &fakeTransport{} + conv := &Conversation{} + + close(m.release) + session.issueClientCommit(utt, conv, tr) + session.respSink.wait() + Expect(predictCalled(m)).To(BeFalse()) + + // The failed commit is long finished: a later response.create is not + // about it and runs on the history as it is. + session.issueClientResponse(types.ResponseCreateParams{}, conv, tr) + session.respSink.wait() + Expect(predictCalled(m)).To(BeTrue()) + dones := responseDones(tr) + Expect(dones).To(HaveLen(1)) + Expect(dones[0].Status).To(Equal(types.ResponseStatusCompleted)) + }) })