realtime: announce speakers in transcription-only sessions (#12871)

feat(realtime): announce speakers in transcription-only sessions

With voice_recognition.identity.announce on, a realtime session sends
a conversation.item.speaker event for each user turn. Only
generateResponse sent this event. A transcription-only session never
calls generateResponse, so it resolved the speaker of each turn and
then dropped the result.

In a transcription-only session with announce on, send the event
before the completed transcription event of the turn, with the same
item_id. The rules are the same as on the response path: no event for
an unknown speaker unless announce_unknown is set, and no event for a
speaker that an enforcing gate rejects.

Sessions with announce off and response sessions do not change.

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

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

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,

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

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)
 

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
 {