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 <mudler@localai.io>

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)
 	}
 }

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))
+	})
 })