From 3a9a67aeb9ac3fa104345196541a2cfd6e0da7e8 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 13 Oct 2026 08:44:22 +0000 Subject: [PATCH] refactor(realtime): gate client responses on the commit tail A client response.create waited on a separate mutex-guarded channel that only the last client commit recorded, so it ignored in-flight VAD commits. The commit chain already ends in session.commitTail, whose done channel closes after every earlier commit has finished. Read the tail under commitOrderMu when the response is issued and wait on it. This drops the extra mutex, the clientCommitDone field and the slotSink callback of issueCommit, and the response is now ordered against both commit producers. Fix the commitSlot.done comment: it closes when the commit function returns, not when the user item is appended. Add a spec for commit, commit, response ordering. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/openai/realtime.go | 84 ++++++------------- .../openai/realtime_client_commit_test.go | 25 ++++++ .../openai/realtime_commit_order_test.go | 2 +- .../openai/realtime_interrupt_test.go | 8 +- .../endpoints/openai/realtime_turncoord.go | 2 +- 5 files changed, 58 insertions(+), 63 deletions(-) diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index 17474a663..00a67519f 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/core/http/endpoints/openai/realtime.go @@ -220,15 +220,6 @@ 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 @@ -247,9 +238,10 @@ type commitSlot struct { // this commit's item append waits on it, and its slot release waits on it // too (on every exit path). prevDone chan struct{} - // done is closed when this commit has fully finished — appended its user - // item or aborted (error, empty transcript, teardown) — AND the - // predecessor has finished, so the next commit is released in order. + // done is closed when the commit function has returned — it appended its + // user item or aborted (error, empty transcript, teardown), and generated + // its own response unless that was cancelled — AND the predecessor has + // finished, so the next commit is released in order. done chan struct{} } @@ -290,13 +282,10 @@ 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, slotSink func(*commitSlot), run func(ctx context.Context, slot *commitSlot)) { +func (session *Session) issueCommit(parent context.Context, source respcoord.Source, 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) { @@ -304,57 +293,38 @@ 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). +// the shared commit-ordering boundary, so slot order == issue order across the +// client and VAD producers. 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) { + session.issueCommit(context.Background(), respcoord.SourceClient, 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). +// client-driven turns the commit and response.create are sent back-to-back, so +// the response waits for the newest commit (client or VAD) to finish before the +// LLM starts: the LLM must answer the real transcript, not an empty one +// (client-commit LLM/STT race). Reading the commit tail and issuing the +// response happen under the same lock as issueCommit, so the response is +// ordered after every commit issued before it and before any issued later. func (session *Session) issueClientResponse(resp types.ResponseCreateParams, conv *Conversation, t Transport) { - commitDone := session.getClientCommitDone() + session.commitOrderMu.Lock() + defer session.commitOrderMu.Unlock() + tail := session.commitTail + pending := false + if tail != nil { + select { + case <-tail.done: + default: + pending = true + } + } session.respSink.issue(context.Background(), respcoord.SourceClient, func(ctx context.Context) { - if commitDone != nil { + if pending { select { - case <-commitDone: + case <-tail.done: case <-ctx.Done(): return } diff --git a/core/http/endpoints/openai/realtime_client_commit_test.go b/core/http/endpoints/openai/realtime_client_commit_test.go index 85021ab3c..eb861e24d 100644 --- a/core/http/endpoints/openai/realtime_client_commit_test.go +++ b/core/http/endpoints/openai/realtime_client_commit_test.go @@ -105,4 +105,29 @@ var _ = Describe("realtime client-commit LLM/STT ordering", func() { Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(Equal(1)) }) + It("orders commit, commit, response: the LLM sees both turns in speech order", func() { + m := newModel("first turn", "second turn") + session := newSession(m) + tr := &fakeTransport{} + conv := &Conversation{} + + session.issueClientCommit(utt, conv, tr) + Eventually(m.started, "2s").Should(Receive(Equal(int32(1)))) + session.issueClientCommit(utt, conv, tr) + Eventually(m.started, "2s").Should(Receive(Equal(int32(2)))) + session.issueClientResponse(types.ResponseCreateParams{}, conv, tr) + + // The second transcription finishes first, but the response must wait + // for the whole chain. + Eventually(m.finished, "2s").Should(Receive(Equal(int32(2)))) + Consistently(func() int { + return tr.countEvents(types.ServerEventTypeResponseCreated) + }, 200*time.Millisecond).Should(BeZero()) + + close(m.release) + session.respSink.wait() + + Expect(userMsgTextsOf(m)).To(Equal([]string{"first turn", "second turn"})) + }) + }) diff --git a/core/http/endpoints/openai/realtime_commit_order_test.go b/core/http/endpoints/openai/realtime_commit_order_test.go index 31456257f..a5b024c18 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, nil, func(ctx context.Context, slot *commitSlot) { + session.issueCommit(parent, source, 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 835ef4ddb..35b875729 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, nil, func(turnCtx context.Context, slot *commitSlot) { + s.issueCommit(context.Background(), respcoord.SourceVAD, 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, nil, func(turnCtx context.Context, slot *commitSlot) { + s.issueCommit(context.Background(), respcoord.SourceVAD, 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, nil, func(ctx context.Context, slot *commitSlot) { + s.issueCommit(context.Background(), source, 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, nil, func(_ context.Context, slot *commitSlot) { + s.issueCommit(context.Background(), respcoord.SourceClient, 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 e7749b959..066614b5a 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, nil, func(ctx context.Context, slot *commitSlot) { + s.session.issueCommit(s.vadContext, respcoord.SourceVAD, func(ctx context.Context, slot *commitSlot) { commitUtteranceWithTranscript(ctx, audio, live, gated, itemID, s.session, conv, s.transport, slot) }) case turncoord.DiscardTurn: