feat(localai-proxy): bridge live transcription to the upstream realtime API

AudioTranscriptionLive opens <upstream>/v1/realtime?model=<realtime_pipeline>
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 <mudler@localai.io>

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

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