diff --git a/core/http/endpoints/openai/realtime.go b/core/http/endpoints/openai/realtime.go index 17592e3bd85c..6c609999148e 100644 --- a/core/http/endpoints/openai/realtime.go +++ b/core/http/endpoints/openai/realtime.go @@ -2129,12 +2129,53 @@ func commitUtteranceWithTranscript(ctx context.Context, utt []byte, live *liveUt itemID = generateItemID() } + // joinResolve waits for the speaker resolution once and caches the outcome, + // so it can be joined early (before the transcript is final) and again at + // the gate below. + var resolved *resolveOutcome + joinResolve := func() resolveOutcome { + if resolved == nil { + out := <-resolveCh + resolved = &out + } + return *resolved + } + + // A transcription-only session never reaches generateResponse, which is + // where a response session announces the speaker. Announce it here + // instead, before the turn's completed event, so the client knows who + // spoke when the final transcript arrives. As on the response path, a + // speaker that an enforcing gate rejects is not announced. + var beforeCompleted func() + if runResolve && session.TranscriptionOnly && session.voiceGate.cfg.AnnounceEnabled() { + beforeCompleted = func() { + out := joinResolve() + if out.err != nil { + return + } + if session.voiceGate.cfg.EnforceGate() { + alreadyVerified := false + if session.voiceGate.cfg.When == config.VoiceGateWhenFirst { + session.gateMu.Lock() + alreadyVerified = session.voiceVerified + session.gateMu.Unlock() + } + allowed, _ := session.voiceGate.authorize(out.res) + if proceed, _ := session.voiceGate.decide(alreadyVerified, allowed); !proceed { + return + } + } + s := out.res.speaker + announceSpeaker(t, session, itemID, &s) + } + } + var transcript string switch { case gated != nil: // semantic_vad retranscribe gate: the batch decode is authoritative. transcript = gated.Text - if err := emitPrecomputedTranscription(t, itemID, nil, transcript); err != nil { + if err := emitPrecomputedTranscription(t, itemID, nil, transcript, beforeCompleted); err != nil { sendError(t, "transcription_failed", err.Error(), "", "event_TODO") return } @@ -2142,7 +2183,7 @@ func commitUtteranceWithTranscript(ctx context.Context, utt []byte, live *liveUt // The caption deltas already streamed during the turn under this // itemID; the completed event replaces the partial text client-side. transcript = live.Text - if err := emitPrecomputedTranscription(t, itemID, nil, transcript); err != nil { + if err := emitPrecomputedTranscription(t, itemID, nil, transcript, beforeCompleted); err != nil { sendError(t, "transcription_failed", err.Error(), "", "event_TODO") return } @@ -2162,12 +2203,12 @@ func commitUtteranceWithTranscript(ctx context.Context, utt []byte, live *liveUt // turn is still cancelled — see the ctx check below — only the // transcript survives, so the next response is built on the full // utterance. - transcript, err = emitTranscription(sctx, t, session, itemID, f.Name()) + transcript, err = emitTranscription(sctx, t, session, itemID, f.Name(), beforeCompleted) if err != nil { // Drain the gate goroutine before returning so its in-flight read of // the temp WAV finishes before the deferred os.Remove fires. if runResolve { - <-resolveCh + joinResolve() } sendError(t, "transcription_failed", err.Error(), "", "event_TODO") return @@ -2199,7 +2240,7 @@ func commitUtteranceWithTranscript(ctx context.Context, utt []byte, live *liveUt // Join on the resolution before any side-effecting step. var speaker *types.Speaker if runResolve { - out := <-resolveCh + out := joinResolve() enforce := session.voiceGate.cfg.EnforceGate() if out.err != nil { @@ -2487,20 +2528,27 @@ func generateResponse(ctx context.Context, session *Session, utt []byte, transcr // Create user message item item := appendUserItem(conv, t, utt, transcript, speaker) - // Surface the recognized speaker to the client. Skip the event for an - // unidentified speaker unless announce_unknown is set. - if speaker != nil && session.voiceGate != nil && session.voiceGate.cfg.AnnounceEnabled() { - if speaker.Matched || session.voiceGate.cfg.Identity.AnnounceUnknown { - sendEvent(t, types.ConversationItemSpeakerEvent{ - ItemID: item.User.ID, - Speaker: *speaker, - }) - } - } + announceSpeaker(t, session, item.User.ID, speaker) triggerResponse(ctx, session, conv, t, nil) } +// announceSpeaker surfaces the recognized speaker of itemID to the client as a +// conversation.item.speaker event when the pipeline enables identity.announce. +// It skips the event for an unidentified speaker unless announce_unknown is +// set, and for a nil speaker (the resolution failed or did not run). +func announceSpeaker(t Transport, session *Session, itemID string, speaker *types.Speaker) { + if speaker == nil || session.voiceGate == nil || !session.voiceGate.cfg.AnnounceEnabled() { + return + } + if speaker.Matched || session.voiceGate.cfg.Identity.AnnounceUnknown { + sendEvent(t, types.ConversationItemSpeakerEvent{ + ItemID: itemID, + Speaker: *speaker, + }) + } +} + // maxAssistantToolTurns caps the server-side agentic loop. Mirrors the // chat-page maxToolTurns:10 from useChat.js — the model gets up to this // many consecutive tool round-trips before we return control to the user diff --git a/core/http/endpoints/openai/realtime_semantic_vad_test.go b/core/http/endpoints/openai/realtime_semantic_vad_test.go index 4621901b2873..b843e4babaa1 100644 --- a/core/http/endpoints/openai/realtime_semantic_vad_test.go +++ b/core/http/endpoints/openai/realtime_semantic_vad_test.go @@ -461,7 +461,7 @@ var _ = Describe("transcribeUtterance", func() { var _ = Describe("emitPrecomputedTranscription", func() { It("emits deltas then completed, sharing the item id", func() { tr := &fakeTransport{} - Expect(emitPrecomputedTranscription(tr, "item42", []string{"a", "", "b"}, "ab")).To(Succeed()) + Expect(emitPrecomputedTranscription(tr, "item42", []string{"a", "", "b"}, "ab", nil)).To(Succeed()) Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionDelta)).To(Equal(2), "empty deltas skipped") Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) @@ -478,7 +478,7 @@ var _ = Describe("emitPrecomputedTranscription", func() { It("emits only the completed event with no deltas", func() { tr := &fakeTransport{} - Expect(emitPrecomputedTranscription(tr, "item1", nil, "hi")).To(Succeed()) + Expect(emitPrecomputedTranscription(tr, "item1", nil, "hi", nil)).To(Succeed()) Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionDelta)).To(Equal(0)) Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) }) diff --git a/core/http/endpoints/openai/realtime_transcription.go b/core/http/endpoints/openai/realtime_transcription.go index c10535f18809..f303178a1cbd 100644 --- a/core/http/endpoints/openai/realtime_transcription.go +++ b/core/http/endpoints/openai/realtime_transcription.go @@ -12,8 +12,9 @@ import ( // whose transcript already exists (semantic_vad's live stream, or the // retranscribe gate's batch decode): optional delta replays followed by the // completed event — the same contract emitTranscription produces, sharing -// one itemID — without running the backend again. -func emitPrecomputedTranscription(t Transport, itemID string, deltas []string, transcript string) error { +// one itemID — without running the backend again. beforeCompleted, when not +// nil, runs just before the completed event is sent. +func emitPrecomputedTranscription(t Transport, itemID string, deltas []string, transcript string, beforeCompleted func()) error { for _, d := range deltas { if d == "" { continue @@ -27,6 +28,9 @@ func emitPrecomputedTranscription(t Transport, itemID string, deltas []string, t return err } } + if beforeCompleted != nil { + beforeCompleted() + } return t.SendEvent(types.ConversationItemInputAudioTranscriptionCompletedEvent{ ServerEventBase: types.ServerEventBase{EventID: "event_TODO"}, ItemID: itemID, @@ -41,7 +45,8 @@ func emitPrecomputedTranscription(t Transport, itemID string, deltas []string, t // a conversation.item.input_audio_transcription.delta as the backend produces it, // then a completed event; otherwise it transcribes the whole utterance and emits // a single completed event. delta and completed events share itemID. -func emitTranscription(ctx context.Context, t Transport, session *Session, itemID, audioPath string) (string, error) { +// beforeCompleted, when not nil, runs just before the completed event is sent. +func emitTranscription(ctx context.Context, t Transport, session *Session, itemID, audioPath string, beforeCompleted func()) (string, error) { cfg := session.InputAudioTranscription diarize := session.ModelConfig != nil && session.ModelConfig.Pipeline.Diarization @@ -66,6 +71,9 @@ func emitTranscription(ctx context.Context, t Transport, session *Session, itemI } } } + if beforeCompleted != nil { + beforeCompleted() + } if err := t.SendEvent(types.ConversationItemInputAudioTranscriptionCompletedEvent{ ServerEventBase: types.ServerEventBase{EventID: "event_TODO"}, ItemID: itemID, @@ -90,6 +98,9 @@ func emitTranscription(ctx context.Context, t Transport, session *Session, itemI return "", err } } + if beforeCompleted != nil { + beforeCompleted() + } if err := t.SendEvent(types.ConversationItemInputAudioTranscriptionCompletedEvent{ ServerEventBase: types.ServerEventBase{EventID: "event_TODO"}, ItemID: itemID, diff --git a/core/http/endpoints/openai/realtime_transcription_test.go b/core/http/endpoints/openai/realtime_transcription_test.go index e8ab399fd87b..6e33718985d0 100644 --- a/core/http/endpoints/openai/realtime_transcription_test.go +++ b/core/http/endpoints/openai/realtime_transcription_test.go @@ -29,7 +29,7 @@ var _ = Describe("emitTranscription", func() { } t := &fakeTransport{} - transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav") + transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav", nil) Expect(err).ToNot(HaveOccurred()) Expect(transcript).To(Equal("Hello world")) @@ -45,7 +45,7 @@ var _ = Describe("emitTranscription", func() { } t := &fakeTransport{} - transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav") + transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav", nil) Expect(err).ToNot(HaveOccurred()) Expect(transcript).To(Equal("Hi")) @@ -82,7 +82,7 @@ var _ = Describe("emitTranscription", func() { } t := &fakeTransport{} - transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav") + transcript, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav", nil) Expect(err).ToNot(HaveOccurred()) Expect(transcript).To(Equal("hi there. hello")) @@ -111,7 +111,7 @@ var _ = Describe("emitTranscription", func() { } t := &fakeTransport{} - _, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav") + _, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav", nil) Expect(err).ToNot(HaveOccurred()) Expect(m.lastDiarize).To(BeTrue()) @@ -127,7 +127,7 @@ var _ = Describe("emitTranscription", func() { } t := &fakeTransport{} - _, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav") + _, err := emitTranscription(context.Background(), t, session, "item1", "/tmp/x.wav", nil) Expect(err).ToNot(HaveOccurred()) Expect(m.lastDiarize).To(BeFalse()) diff --git a/core/http/endpoints/openai/realtime_voicegate_integration_test.go b/core/http/endpoints/openai/realtime_voicegate_integration_test.go index 07a5d20f963b..dfc26a9a9058 100644 --- a/core/http/endpoints/openai/realtime_voicegate_integration_test.go +++ b/core/http/endpoints/openai/realtime_voicegate_integration_test.go @@ -206,6 +206,95 @@ var _ = Describe("realtime speaker surfacing (commitUtterance)", func() { }) }) +// firstEventIndex returns the position of the first recorded event of type et, +// or -1 when there is none. +func firstEventIndex(tr *fakeTransport, et types.ServerEventType) int { + for i, e := range tr.events() { + if e.ServerEventType() == et { + return i + } + } + return -1 +} + +var _ = Describe("realtime speaker surfacing in transcription-only sessions (commitUtterance)", func() { + utt := make([]byte, 32) + + transcriptionOnly := func(identity *config.VoiceIdentityConfig) *Session { + session, _ := itSession(itGate("alice", "alice", []float32{1, 0, 0}, nil, + config.VoiceGateWhenEvery, config.VoiceGateRejectEvent)) + session.TranscriptionOnly = true + session.voiceGate.cfg.Identity = identity + return session + } + + It("emits conversation.item.speaker before the completed event when announce is on", func() { + session := transcriptionOnly(&config.VoiceIdentityConfig{Announce: true}) + tr := &fakeTransport{} + + commitUtterance(context.Background(), utt, session, &Conversation{}, tr, session.nextCommitSlot()) + + Expect(tr.countEvents(types.ServerEventTypeConversationItemSpeaker)).To(Equal(1)) + speakerAt := firstEventIndex(tr, types.ServerEventTypeConversationItemSpeaker) + completedAt := firstEventIndex(tr, types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted) + Expect(completedAt).To(BeNumerically(">=", 0)) + Expect(speakerAt).To(BeNumerically("<", completedAt)) + + events := tr.events() + spk := events[speakerAt].(types.ConversationItemSpeakerEvent) + done := events[completedAt].(types.ConversationItemInputAudioTranscriptionCompletedEvent) + Expect(spk.ItemID).To(Equal(done.ItemID)) + Expect(spk.Speaker.Name).To(Equal("alice")) + Expect(spk.Speaker.Matched).To(BeTrue()) + + // Still transcription only: no response is generated. + Expect(tr.countEvents(types.ServerEventTypeResponseDone)).To(Equal(0)) + }) + + It("emits no speaker event when announce is off", func() { + for _, identity := range []*config.VoiceIdentityConfig{nil, {Personalize: true}} { + session := transcriptionOnly(identity) + tr := &fakeTransport{} + + commitUtterance(context.Background(), utt, session, &Conversation{}, tr, session.nextCommitSlot()) + + Expect(tr.countEvents(types.ServerEventTypeConversationItemSpeaker)).To(Equal(0)) + Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) + Expect(tr.countEvents(types.ServerEventTypeResponseDone)).To(Equal(0)) + } + }) + + It("does not announce a speaker that the enforcing gate rejects", func() { + // match name "mallory" is not in the allow list, so the gate denies it. + session, _ := itSession(itGate("alice", "mallory", []float32{1, 0, 0}, nil, + config.VoiceGateWhenEvery, config.VoiceGateRejectEvent)) + session.TranscriptionOnly = true + session.voiceGate.cfg.Identity = &config.VoiceIdentityConfig{Announce: true} + tr := &fakeTransport{} + + commitUtterance(context.Background(), utt, session, &Conversation{}, tr, session.nextCommitSlot()) + + Expect(tr.countEvents(types.ServerEventTypeConversationItemSpeaker)).To(Equal(0)) + Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) + }) + + It("leaves a response session unchanged: the speaker follows the completed event", func() { + session, _ := itSession(itGate("alice", "alice", []float32{1, 0, 0}, nil, + config.VoiceGateWhenEvery, config.VoiceGateRejectEvent)) + session.voiceGate.cfg.Identity = &config.VoiceIdentityConfig{Announce: true} + tr := &fakeTransport{} + + commitUtterance(context.Background(), utt, session, &Conversation{}, tr, session.nextCommitSlot()) + + Expect(tr.countEvents(types.ServerEventTypeConversationItemSpeaker)).To(Equal(1)) + speakerAt := firstEventIndex(tr, types.ServerEventTypeConversationItemSpeaker) + completedAt := firstEventIndex(tr, types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted) + Expect(completedAt).To(BeNumerically(">=", 0)) + Expect(speakerAt).To(BeNumerically(">", completedAt)) + Expect(tr.countEvents(types.ServerEventTypeResponseDone)).To(BeNumerically(">=", 1)) + }) +}) + var _ = Describe("realtime speaker personalization (triggerResponseAtTurn)", func() { utt := make([]byte, 32) diff --git a/docs/content/features/openai-realtime.md b/docs/content/features/openai-realtime.md index 96f9c9e0b84f..662fe9aa1b97 100644 --- a/docs/content/features/openai-realtime.md +++ b/docs/content/features/openai-realtime.md @@ -730,7 +730,7 @@ pipeline: ### The `conversation.item.speaker` event -When `identity.announce` is enabled, the server emits a `conversation.item.speaker` event after the user conversation item, naming the recognized speaker: +When `identity.announce` is enabled, the server emits a `conversation.item.speaker` event after the user conversation item, naming the recognized speaker. In a transcription-only session there is no conversation item, so the server emits the event before the `conversation.item.input_audio_transcription.completed` event of the turn, with the same `item_id`: ```json {