From c2375a7db8f48426695e7fc02c63dfc5d15724f3 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 27 Sep 2026 00:14:32 +0000 Subject: [PATCH] feat(localai-proxy): bridge live transcription to the upstream realtime API AudioTranscriptionLive opens /v1/realtime?model= as a transcription session with server VAD, forwards PCM as base64 PCM16 appends, and maps transcription deltas and completions to Delta/Eou. Closing the send side waits briefly for an in-flight utterance, then sends the final text. An upstream error, failed transcription or disconnect ends the stream with Unavailable so failover reopens on the next target. Assisted-by: Claude:claude-opus-5-5 Signed-off-by: Ettore Di Giacinto --- backend/go/localai-proxy/live.go | 425 ++++++++++++++++++++++++++ backend/go/localai-proxy/live_test.go | 380 +++++++++++++++++++++++ 2 files changed, 805 insertions(+) create mode 100644 backend/go/localai-proxy/live.go create mode 100644 backend/go/localai-proxy/live_test.go diff --git a/backend/go/localai-proxy/live.go b/backend/go/localai-proxy/live.go new file mode 100644 index 000000000..622358729 --- /dev/null +++ b/backend/go/localai-proxy/live.go @@ -0,0 +1,425 @@ +package main + +import ( + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "math" + "net/http" + "net/url" + "strings" + "time" + + "github.com/gorilla/websocket" + "github.com/mudler/xlog" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/mudler/LocalAI/pkg/grpc/grpcerrors" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" +) + +const ( + realtimePath = "/v1/realtime" + + // defaultLiveSampleRate is what TranscriptLiveConfig.sample_rate 0 means. + defaultLiveSampleRate = 16000 + + // finalWait bounds how long a closing session waits for an utterance the + // upstream VAD already started. The upstream transcribes only after its + // VAD sees the speech end, so a slow model can finish after the client + // stops sending; waiting forever would pin the gRPC call on a hung + // upstream. + finalWait = 5 * time.Second + + // liveWriteTimeout turns an upstream that stops reading into an error + // instead of a call blocked on a full socket buffer. + liveWriteTimeout = 10 * time.Second + + // liveHandshakeTimeout bounds the WebSocket upgrade only. The upstream + // warms the pipeline models before it sends session.created, which can + // take minutes on a cold box, so the setup phase after the upgrade is + // bounded by request_timeout_seconds instead. + liveHandshakeTimeout = 30 * time.Second +) + +// realtimeEvent holds the fields the bridge reads from any upstream server +// event; unrelated events decode into it harmlessly. +type realtimeEvent struct { + Type string `json:"type"` + ItemID string `json:"item_id"` + Delta string `json:"delta"` + Transcript string `json:"transcript"` + Error *struct { + Message string `json:"message"` + Code string `json:"code"` + } `json:"error"` +} + +// AudioTranscriptionLive bridges a live transcription session to the +// upstream's /v1/realtime transcription session, so a realtime pipeline on +// this box can use a remote LocalAI's streaming ASR. The upstream runs its own +// server VAD and transcribes per utterance: each completed utterance becomes +// Delta (whatever the streamed deltas did not already carry) plus Eou. Word +// timings and Eob have no upstream counterpart and stay empty. +// +// Contract (pkg/grpc/server.go): this method closes out, in closes when the +// client half-closes, and errors are returned at once because the caller +// blocks on the first Recv for the ready ack. +func (p *LocalAIProxy) AudioTranscriptionLive(in <-chan *pb.TranscriptLiveRequest, out chan<- *pb.TranscriptLiveResponse) error { + defer close(out) + + cfg, err := p.config() + if err != nil { + return err + } + if cfg.realtimePipeline == "" { + return grpcerrors.LiveTranscriptionUnsupported(backendName, "set the realtime_pipeline backend option") + } + + first, ok := <-in + if !ok { + return nil // the caller closed without sending anything + } + lc := first.GetConfig() + if lc == nil { + return status.Error(codes.InvalidArgument, "localai-proxy: the first live transcription message must carry a config") + } + rate := int(lc.GetSampleRate()) + if rate == 0 { + rate = defaultLiveSampleRate + } + + conn, err := p.dialRealtime(cfg) + if err != nil { + return err + } + defer func() { _ = conn.Close() }() + + s := &liveSession{ + conn: conn, + out: out, + pipeline: cfg.realtimePipeline, + language: lc.GetLanguage(), + rate: rate, + sent: map[string]string{}, + events: make(chan realtimeEvent), + done: make(chan struct{}), + } + // Closing done releases the reader if it is blocked handing over an + // event; closing conn (deferred above, runs after this) unblocks its read. + defer close(s.done) + go s.readLoop() + + return s.run(in, cfg.timeout) +} + +// dialRealtime opens the upstream WebSocket. A refused upgrade is mapped like +// any other upstream HTTP reply, so a 5xx trips failover and a 4xx does not. +func (p *LocalAIProxy) dialRealtime(cfg *proxyConfig) (*websocket.Conn, error) { + u, err := url.Parse(cfg.base + realtimePath) + if err != nil { + return nil, status.Errorf(codes.Internal, "localai-proxy: build %s URL: %v", realtimePath, err) + } + // Load only accepts http(s) bases. + if u.Scheme == "https" { + u.Scheme = "wss" + } else { + u.Scheme = "ws" + } + u.RawQuery = url.Values{"model": {cfg.realtimePipeline}}.Encode() + + header := http.Header{} + if cfg.apiKey != "" { + header.Set("Authorization", "Bearer "+cfg.apiKey) + } + dialer := websocket.Dialer{ + Proxy: http.ProxyFromEnvironment, + HandshakeTimeout: liveHandshakeTimeout, + } + conn, resp, err := dialer.Dial(u.String(), header) + if err != nil { + if resp != nil && resp.StatusCode != http.StatusSwitchingProtocols { + defer func() { _ = resp.Body.Close() }() + return nil, statusError(realtimePath, resp) + } + return nil, transportError(realtimePath, err) + } + return conn, nil +} + +// liveSession is the state of one bridged session. Only run touches it; the +// reader goroutine only hands events over. +type liveSession struct { + conn *websocket.Conn + out chan<- *pb.TranscriptLiveResponse + pipeline string + language string + rate int + + sent map[string]string // text already sent as deltas, per upstream item + final []string // completed transcripts, in order + pending int // utterances started upstream and not yet transcribed + + events chan realtimeEvent + readErr error // set before events is closed + done chan struct{} +} + +// readLoop decodes upstream events until the socket fails. It is the only +// reader, as gorilla/websocket requires. +func (s *liveSession) readLoop() { + defer close(s.events) + for { + _, msg, err := s.conn.ReadMessage() + if err != nil { + s.readErr = err + return + } + var ev realtimeEvent + if err := json.Unmarshal(msg, &ev); err != nil { + xlog.Debug("localai-proxy: skipping undecodable realtime event", "error", err) + continue + } + select { + case s.events <- ev: + case <-s.done: + return + } + } +} + +// run drives the session: setup, then audio forwarding and event mapping, +// then the drain once the client closes its side. It is also the only writer +// on the socket, so frames never interleave. +func (s *liveSession) run(in <-chan *pb.TranscriptLiveRequest, setupTimeout time.Duration) error { + var ( + created, ready bool + inClosed bool + backlog []*pb.TranscriptLiveRequest + setupTimer <-chan time.Time + drainTimer <-chan time.Time + drain *time.Timer + ) + defer func() { + if drain != nil { + drain.Stop() + } + }() + if setupTimeout > 0 { + t := time.NewTimer(setupTimeout) + defer t.Stop() + setupTimer = t.C + } + + // finishOrDrain finalizes once nothing is in flight, else waits (bounded) + // for the upstream to finish the utterance it already started. + finishOrDrain := func() (bool, error) { + if s.pending <= 0 { + return true, s.finish() + } + if drain == nil { + drain = time.NewTimer(finalWait) + drainTimer = drain.C + } + return false, nil + } + + for { + select { + case req, ok := <-in: + if !ok { + in, inClosed = nil, true + if ready { + if done, err := finishOrDrain(); done { + return err + } + } + continue + } + if !ready { + // Callers wait for the ready ack before streaming, but hold + // anything sent early rather than drop it. + backlog = append(backlog, req) + continue + } + if err := s.forward(req); err != nil { + return err + } + + case ev, ok := <-s.events: + if !ok { + return s.upstreamGone() + } + switch ev.Type { + case "session.created": + if created { + continue + } + created = true + if err := s.write(s.sessionUpdate()); err != nil { + return err + } + case "session.updated": + if ready || !created { + continue + } + ready = true + setupTimer = nil + s.out <- &pb.TranscriptLiveResponse{Ready: true} + for _, req := range backlog { + if err := s.forward(req); err != nil { + return err + } + } + backlog = nil + if inClosed { + if done, err := finishOrDrain(); done { + return err + } + } + case "error": + return s.upstreamError("error", ev) + case "conversation.item.input_audio_transcription.failed": + return s.upstreamError("transcription failed", ev) + case "input_audio_buffer.speech_started": + s.pending++ + case "conversation.item.input_audio_transcription.delta": + s.delta(ev) + case "conversation.item.input_audio_transcription.completed": + s.completed(ev) + if inClosed && s.pending <= 0 { + return s.finish() + } + } + + case <-setupTimer: + return status.Errorf(codes.Unavailable, "localai-proxy: upstream %s did not set up the transcription session in %s", realtimePath, setupTimeout) + + case <-drainTimer: + xlog.Warn("localai-proxy: upstream did not finish the last utterance in time; finalizing without it", + "pipeline", s.pipeline, "wait", finalWait) + return s.finish() + } + } +} + +func (s *liveSession) sessionUpdate() map[string]any { + return map[string]any{ + "type": "session.update", + "session": map[string]any{ + "type": "transcription", + "audio": map[string]any{ + "input": map[string]any{ + "format": map[string]any{"type": "audio/pcm", "rate": s.rate}, + "transcription": map[string]any{"model": s.pipeline, "language": s.language}, + "turn_detection": map[string]any{"type": "server_vad"}, + }, + }, + }, + } +} + +// forward sends one client message upstream. The upstream session rate is +// fixed at setup, so a second config cannot be honoured. +func (s *liveSession) forward(req *pb.TranscriptLiveRequest) error { + if req.GetConfig() != nil { + return status.Error(codes.InvalidArgument, "localai-proxy: a live transcription config is only accepted as the first message") + } + pcm := req.GetAudio().GetPcm() + if len(pcm) == 0 { + return nil + } + return s.write(map[string]any{ + "type": "input_audio_buffer.append", + "audio": base64.StdEncoding.EncodeToString(pcm16LE(pcm)), + }) +} + +// pcm16LE converts float samples in [-1, 1] to the little-endian PCM16 the +// realtime API takes. Out-of-range samples are clipped rather than wrapped. +func pcm16LE(pcm []float32) []byte { + buf := make([]byte, len(pcm)*2) + for i, f := range pcm { + v := math.Max(-1, math.Min(1, float64(f))) + binary.LittleEndian.PutUint16(buf[i*2:], uint16(int16(v*math.MaxInt16))) + } + return buf +} + +func (s *liveSession) delta(ev realtimeEvent) { + if ev.Delta == "" { + return + } + s.sent[ev.ItemID] += ev.Delta + s.out <- &pb.TranscriptLiveResponse{Delta: ev.Delta} +} + +// completed ends one utterance. Deltas only arrive when the upstream pipeline +// streams its transcription, so the completion carries whatever text the +// deltas did not; if the final transcript diverged from them there is no way +// to retract, and FinalResult carries the authoritative text. +func (s *liveSession) completed(ev realtimeEvent) { + sent := s.sent[ev.ItemID] + delete(s.sent, ev.ItemID) + rest, ok := strings.CutPrefix(ev.Transcript, sent) + if !ok { + rest = "" + } + if s.pending > 0 { + s.pending-- + } + if t := strings.TrimSpace(ev.Transcript); t != "" { + s.final = append(s.final, t) + } + s.out <- &pb.TranscriptLiveResponse{Delta: rest, Eou: true} +} + +// finish sends the final transcript and closes the upstream session cleanly. +func (s *liveSession) finish() error { + s.out <- &pb.TranscriptLiveResponse{FinalResult: &pb.TranscriptResult{Text: strings.Join(s.final, " ")}} + _ = s.conn.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(time.Second)) + return nil +} + +func (s *liveSession) write(v any) error { + if err := s.conn.SetWriteDeadline(time.Now().Add(liveWriteTimeout)); err != nil { + return s.writeFailed(err) + } + if err := s.conn.WriteJSON(v); err != nil { + return s.writeFailed(err) + } + return nil +} + +func (s *liveSession) writeFailed(err error) error { + xlog.Warn("localai-proxy: realtime write failed", "pipeline", s.pipeline, "error", err) + return status.Errorf(codes.Unavailable, "localai-proxy: upstream %s: %v", realtimePath, err) +} + +// upstreamGone maps a socket that stopped delivering events. It is +// Unavailable even for a clean close: the client did not end the session, so +// the upstream dropped it, and failover should reopen on the next target. +func (s *liveSession) upstreamGone() error { + err := s.readErr + if err == nil { + err = errors.New("connection closed") + } + xlog.Warn("localai-proxy: realtime upstream disconnected", "pipeline", s.pipeline, "error", err) + return status.Errorf(codes.Unavailable, "localai-proxy: upstream %s disconnected: %v", realtimePath, err) +} + +func (s *liveSession) upstreamError(what string, ev realtimeEvent) error { + msg := "no details" + if ev.Error != nil && ev.Error.Message != "" { + msg = ev.Error.Message + if ev.Error.Code != "" { + msg = fmt.Sprintf("%s (%s)", msg, ev.Error.Code) + } + } + xlog.Warn("localai-proxy: realtime upstream reported an error", "pipeline", s.pipeline, "event", what, "error", msg) + return status.Errorf(codes.Unavailable, "localai-proxy: upstream %s %s: %s", realtimePath, what, msg) +} diff --git a/backend/go/localai-proxy/live_test.go b/backend/go/localai-proxy/live_test.go new file mode 100644 index 000000000..22e30e6d1 --- /dev/null +++ b/backend/go/localai-proxy/live_test.go @@ -0,0 +1,380 @@ +package main + +import ( + "encoding/base64" + "encoding/binary" + "encoding/json" + "net/http" + "runtime" + "strings" + "time" + + "github.com/gorilla/websocket" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/mudler/LocalAI/pkg/grpc/grpcerrors" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" +) + +// wsUpstream is a fake upstream /v1/realtime endpoint. Each spec scripts the +// server side of the session in script, which runs on the upgraded socket. +func wsUpstream(script func(c *websocket.Conn, r *http.Request)) *fakeUpstream { + upgrader := websocket.Upgrader{} + return newFakeUpstreamWithHandler(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/realtime" { + http.NotFound(w, r) + return + } + c, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = c.Close() }() + script(c, r) + }) +} + +func wsSend(c *websocket.Conn, v any) { + defer GinkgoRecover() + Expect(c.WriteJSON(v)).To(Succeed()) +} + +// wsRecv reads the next client event; ok is false once the client is gone. +func wsRecv(c *websocket.Conn) (map[string]any, bool) { + var ev map[string]any + if err := c.ReadJSON(&ev); err != nil { + return nil, false + } + return ev, true +} + +// wsHandshake plays the upstream's session setup and returns the client's +// session.update event. +func wsHandshake(c *websocket.Conn) map[string]any { + defer GinkgoRecover() + wsSend(c, map[string]any{"type": "session.created", "session": map[string]any{}}) + upd, ok := wsRecv(c) + Expect(ok).To(BeTrue()) + Expect(upd["type"]).To(Equal("session.update")) + wsSend(c, map[string]any{"type": "session.updated", "session": map[string]any{}}) + return upd +} + +// wsDrain reads client events until the client closes, returning the +// decoded PCM16 samples of every input_audio_buffer.append. +func wsDrain(c *websocket.Conn) [][]int16 { + var frames [][]int16 + for { + ev, ok := wsRecv(c) + if !ok { + return frames + } + if ev["type"] != "input_audio_buffer.append" { + continue + } + raw, err := base64.StdEncoding.DecodeString(ev["audio"].(string)) + if err != nil { + return frames + } + samples := make([]int16, len(raw)/2) + for i := range samples { + samples[i] = int16(binary.LittleEndian.Uint16(raw[i*2:])) + } + frames = append(frames, samples) + } +} + +type liveCall struct { + in chan *pb.TranscriptLiveRequest + out chan *pb.TranscriptLiveResponse + errc chan error +} + +// startLive runs AudioTranscriptionLive the way pkg/grpc/server.go does: +// buffered channels, the caller owning in and the backend owning out. +func startLive(p *LocalAIProxy) *liveCall { + lc := &liveCall{ + in: make(chan *pb.TranscriptLiveRequest, 4), + out: make(chan *pb.TranscriptLiveResponse, 4), + errc: make(chan error, 1), + } + go func() { lc.errc <- p.AudioTranscriptionLive(lc.in, lc.out) }() + return lc +} + +func (lc *liveCall) config(lang string, rate int32) { + lc.in <- &pb.TranscriptLiveRequest{Payload: &pb.TranscriptLiveRequest_Config{ + Config: &pb.TranscriptLiveConfig{Language: lang, SampleRate: rate}, + }} +} + +func (lc *liveCall) audio(pcm ...float32) { + lc.in <- &pb.TranscriptLiveRequest{Payload: &pb.TranscriptLiveRequest_Audio{ + Audio: &pb.TranscriptLiveAudio{Pcm: pcm}, + }} +} + +func (lc *liveCall) next() *pb.TranscriptLiveResponse { + var r *pb.TranscriptLiveResponse + EventuallyWithOffset(1, lc.out, 2*time.Second).Should(Receive(&r)) + return r +} + +// finish asserts the call returned within 2 s and out is closed, and +// returns the call's error. +func (lc *liveCall) finish() error { + var err error + EventuallyWithOffset(1, lc.errc, 2*time.Second).Should(Receive(&err)) + for range lc.out { + } + return err +} + +// liveGoroutines counts goroutines still running code from live.go, so a +// spec can prove the bridge left nothing blocked behind. +func liveGoroutines() int { + buf := make([]byte, 1<<20) + buf = buf[:runtime.Stack(buf, true)] + n := 0 + for _, g := range strings.Split(string(buf), "\n\n") { + if strings.Contains(g, "localai-proxy/live.go:") { + n++ + } + } + return n +} + +func withPipeline(o *pb.ModelOptions) { + o.Options = append(o.Options, "realtime_pipeline:remote-pipe") +} + +var _ = Describe("AudioTranscriptionLive", func() { + It("reports live transcription unsupported without realtime_pipeline", func() { + up := newFakeUpstream() + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, nil)) + + err := lc.finish() + Expect(grpcerrors.IsLiveTranscriptionUnsupported(err)).To(BeTrue()) + Expect(err.Error()).To(ContainSubstring("realtime_pipeline")) + Expect(up.recorded()).To(BeEmpty()) + }) + + It("rejects a first message that is not a config", func() { + up := wsUpstream(func(*websocket.Conn, *http.Request) {}) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.audio(0.1) + + Expect(status.Code(lc.finish())).To(Equal(codes.InvalidArgument)) + }) + + It("opens the pipeline session and acks ready only after session.updated", func() { + gotURL := make(chan string, 1) + gotAuth := make(chan string, 1) + gotUpdate := make(chan map[string]any, 1) + release := make(chan struct{}) + up := wsUpstream(func(c *websocket.Conn, r *http.Request) { + gotURL <- r.URL.RequestURI() + gotAuth <- r.Header.Get("Authorization") + wsSend(c, map[string]any{"type": "session.created", "session": map[string]any{}}) + upd, _ := wsRecv(c) + gotUpdate <- upd + <-release + wsSend(c, map[string]any{"type": "session.updated", "session": map[string]any{}}) + wsDrain(c) + }) + DeferCleanup(up.Close) + keyFile := writeInput("key", "sekret\n") + p := loadProxy(up, func(o *pb.ModelOptions) { + withPipeline(o) + o.Proxy.ApiKeyFile = keyFile + }) + + lc := startLive(p) + lc.config("it", 24000) + + Eventually(gotURL, 2*time.Second).Should(Receive(Equal("/v1/realtime?model=remote-pipe"))) + Expect(gotAuth).To(Receive(Equal("Bearer sekret"))) + var upd map[string]any + Eventually(gotUpdate, 2*time.Second).Should(Receive(&upd)) + raw, err := json.Marshal(upd) + Expect(err).NotTo(HaveOccurred()) + Expect(raw).To(MatchJSON(`{"type":"session.update","session":{"type":"transcription","audio":{"input":{ + "format":{"type":"audio/pcm","rate":24000}, + "transcription":{"model":"remote-pipe","language":"it"}, + "turn_detection":{"type":"server_vad"}}}}}`)) + + Consistently(lc.out, 200*time.Millisecond).ShouldNot(Receive()) + close(release) + Expect(lc.next().GetReady()).To(BeTrue()) + + close(lc.in) + Expect(lc.finish()).To(Succeed()) + }) + + It("defaults the session rate to 16000", func() { + gotUpdate := make(chan map[string]any, 1) + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + gotUpdate <- wsHandshake(c) + wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("", 0) + Expect(lc.next().GetReady()).To(BeTrue()) + + var upd map[string]any + Expect(gotUpdate).To(Receive(&upd)) + format := upd["session"].(map[string]any)["audio"].(map[string]any)["input"].(map[string]any)["format"] + Expect(format).To(HaveKeyWithValue("rate", BeNumerically("==", 16000))) + + close(lc.in) + Expect(lc.finish()).To(Succeed()) + }) + + It("forwards audio as base64 PCM16 appends", func() { + frames := make(chan [][]int16, 1) + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsHandshake(c) + frames <- wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + Expect(lc.next().GetReady()).To(BeTrue()) + + lc.audio(0, 0.5, -1, 1, 2) + lc.audio(-0.25) + close(lc.in) + Expect(lc.finish()).To(Succeed()) + + Eventually(frames, 2*time.Second).Should(Receive(Equal([][]int16{ + {0, 16383, -32767, 32767, 32767}, + {-8191}, + }))) + }) + + It("maps deltas and completions to Delta and Eou, and finishes with the full text", func() { + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsHandshake(c) + wsSend(c, map[string]any{"type": "input_audio_buffer.speech_started"}) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.delta", "item_id": "a", "delta": "hel"}) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.delta", "item_id": "a", "delta": "lo"}) + wsSend(c, map[string]any{"type": "input_audio_buffer.speech_stopped"}) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.completed", "item_id": "a", "transcript": "hello world"}) + wsSend(c, map[string]any{"type": "input_audio_buffer.speech_started"}) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.completed", "item_id": "b", "transcript": "again"}) + wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + Expect(lc.next().GetReady()).To(BeTrue()) + + Expect(lc.next()).To(SatisfyAll( + WithTransform((*pb.TranscriptLiveResponse).GetDelta, Equal("hel")), + WithTransform((*pb.TranscriptLiveResponse).GetEou, BeFalse()))) + Expect(lc.next().GetDelta()).To(Equal("lo")) + r := lc.next() + Expect(r.GetDelta()).To(Equal(" world")) + Expect(r.GetEou()).To(BeTrue()) + r = lc.next() + Expect(r.GetDelta()).To(Equal("again")) + Expect(r.GetEou()).To(BeTrue()) + + close(lc.in) + Expect(lc.next().GetFinalResult().GetText()).To(Equal("hello world again")) + Expect(lc.finish()).To(Succeed()) + }) + + It("waits for an in-flight utterance before the final result", func() { + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsHandshake(c) + wsSend(c, map[string]any{"type": "input_audio_buffer.speech_started"}) + // The client closes its side now; the transcription lands later. + time.Sleep(300 * time.Millisecond) + wsSend(c, map[string]any{"type": "input_audio_buffer.speech_stopped"}) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.completed", "item_id": "a", "transcript": "late words"}) + wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + Expect(lc.next().GetReady()).To(BeTrue()) + close(lc.in) + + r := lc.next() + Expect(r.GetDelta()).To(Equal("late words")) + Expect(r.GetEou()).To(BeTrue()) + Expect(lc.next().GetFinalResult().GetText()).To(Equal("late words")) + Expect(lc.finish()).To(Succeed()) + }) + + It("ends with Unavailable on an upstream error event during setup", func() { + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsSend(c, map[string]any{"type": "session.created", "session": map[string]any{}}) + wsRecv(c) + wsSend(c, map[string]any{"type": "error", "error": map[string]any{ + "type": "invalid_request_error", "code": "session_update_error", + "message": "model is not a valid pipeline model: remote-pipe", + }}) + wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + + err := lc.finish() + Expect(status.Code(err)).To(Equal(codes.Unavailable)) + Expect(err.Error()).To(ContainSubstring("not a valid pipeline model")) + }) + + It("ends with Unavailable when a transcription fails", func() { + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsHandshake(c) + wsSend(c, map[string]any{"type": "conversation.item.input_audio_transcription.failed", "item_id": "a", + "error": map[string]any{"message": "backend crashed"}}) + wsDrain(c) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + Expect(lc.next().GetReady()).To(BeTrue()) + + err := lc.finish() + Expect(status.Code(err)).To(Equal(codes.Unavailable)) + Expect(err.Error()).To(ContainSubstring("backend crashed")) + }) + + It("maps a refused upgrade to the upstream status", func() { + up := newFakeUpstreamWithHandler(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "loading", http.StatusServiceUnavailable) + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + + Expect(status.Code(lc.finish())).To(Equal(codes.Unavailable)) + }) + + It("upstream disconnect ends the stream", func() { + before := liveGoroutines() + up := wsUpstream(func(c *websocket.Conn, _ *http.Request) { + wsHandshake(c) + // Drop the socket mid-session, as a crashed upstream would. + }) + DeferCleanup(up.Close) + lc := startLive(loadProxy(up, withPipeline)) + lc.config("en", 16000) + Expect(lc.next().GetReady()).To(BeTrue()) + // The caller keeps its send side open; the bridge must not wait on it. + + Expect(status.Code(lc.finish())).To(Equal(codes.Unavailable)) + Eventually(liveGoroutines, time.Second).Should(Equal(before)) + close(lc.in) + }) +})