diff --git a/.changeset/sarvam-realtime-stt.md b/.changeset/sarvam-realtime-stt.md new file mode 100644 index 000000000..23e3ae9b8 --- /dev/null +++ b/.changeset/sarvam-realtime-stt.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents-plugin-sarvam': minor +--- + +Add `STTRealtime`, a Sarvam realtime speech-to-text plugin (`saaras:v3-realtime`) with server-side VAD, partial/final transcript gating, and per-connection usage reporting. Ported from `livekit/agents` (Python) PR #6562. Note: this stream never reconnects after a socket failure — Sarvam bills per connection, so `stream()` forces `connOptions.maxRetry` to `0`. diff --git a/plugins/sarvam/README.md b/plugins/sarvam/README.md index efb73e338..c49a5b295 100644 --- a/plugins/sarvam/README.md +++ b/plugins/sarvam/README.md @@ -77,6 +77,28 @@ const stt = new sarvam.STT({ Set the `SARVAM_API_KEY` environment variable or pass `apiKey` directly. +### STT (Realtime) + +```typescript +import * as sarvam from '@livekit/agents-plugin-sarvam'; + +const stt = new sarvam.STTRealtime({ + language: 'en-IN', + streamType: 'balanced', + endpointing: 'vad', +}); +``` + +`STTRealtime` connects to Sarvam's realtime API (`saaras:v3-realtime`, not configurable), which +streams partial transcripts and uses Sarvam's own VAD for turn detection by default. Set +`endpointing: 'manual'` to delimit turns from your application instead — the plugin then emits +`START_OF_SPEECH` on the first audio frame of a turn and `END_OF_SPEECH` when you flush the +stream. + +Realtime streams don't reconnect after a socket failure, because Sarvam bills per connection — +`stream()` forces `connOptions.maxRetry` to `0`. Create a new stream (or restart the session) if +the connection drops. + ## STT Models | Model | Endpoint | Languages | Modes | Prompt | diff --git a/plugins/sarvam/etc/agents-plugin-sarvam.api.md b/plugins/sarvam/etc/agents-plugin-sarvam.api.md index e476699db..3d05f583b 100644 --- a/plugins/sarvam/etc/agents-plugin-sarvam.api.md +++ b/plugins/sarvam/etc/agents-plugin-sarvam.api.md @@ -30,7 +30,37 @@ export class ChunkedStream extends tts.ChunkedStream { protected run(): Promise; } +// Warning: (ae-missing-release-tag) "RealtimeEncoding" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export type RealtimeEncoding = 'linear16' | 'linear32' | 'mulaw' | 'alaw'; + +// Warning: (ae-missing-release-tag) "RealtimeEndpointing" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export type RealtimeEndpointing = 'vad' | 'manual'; + // Warning: (ae-forgotten-export) The symbol "stt" needs to be exported by the entry point index.d.ts +// Warning: (ae-missing-release-tag) "RealtimeSpeechStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +export class RealtimeSpeechStream extends stt.SpeechStream { + // Warning: (ae-forgotten-export) The symbol "ResolvedRealtimeOptions" needs to be exported by the entry point index.d.ts + constructor(sttInstance: STTRealtime, opts: ResolvedRealtimeOptions, connOptions: APIConnectOptions, onClose?: () => void); + // (undocumented) + close(): void; + // (undocumented) + label: string; + // (undocumented) + protected run(): Promise; + updateOptions(opts: Partial): void; +} + +// Warning: (ae-missing-release-tag) "RealtimeStreamType" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export type RealtimeStreamType = 'fast' | 'balanced' | 'simulated'; + // Warning: (ae-missing-release-tag) "SpeechStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -84,6 +114,56 @@ export type STTModes = 'transcribe' | 'translate' | 'verbatim' | 'translit' | 'c // @public export type STTOptions = STTV2Options | STTTranslateOptions | STTV3Options; +// Warning: (ae-missing-release-tag) "STTRealtime" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export class STTRealtime extends stt.STT { + constructor(opts?: Partial); + // (undocumented) + label: string; + // (undocumented) + get model(): string; + // (undocumented) + get provider(): string; + // (undocumented) + _recognize(): Promise; + // (undocumented) + stream(options?: { + language?: string; + connOptions?: APIConnectOptions; + }): RealtimeSpeechStream; + updateOptions(opts: Partial): void; +} + +// Warning: (ae-missing-release-tag) "STTRealtimeLanguages" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export type STTRealtimeLanguages = 'auto' | 'as-IN' | 'bn-IN' | 'brx-IN' | 'doi-IN' | 'en-IN' | 'gu-IN' | 'hi-IN' | 'kn-IN' | 'kok-IN' | 'ks-IN' | 'mai-IN' | 'ml-IN' | 'mni-IN' | 'mr-IN' | 'ne-IN' | 'or-IN' | 'pa-IN' | 'sa-IN' | 'sat-IN' | 'sd-IN' | 'ta-IN' | 'te-IN' | 'ur-IN'; + +// Warning: (ae-missing-release-tag) "STTRealtimeModel" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export type STTRealtimeModel = 'saaras:v3-realtime'; + +// Warning: (ae-missing-release-tag) "STTRealtimeOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +export interface STTRealtimeOptions { + apiKey?: string; + encoding?: RealtimeEncoding; + endpointing?: RealtimeEndpointing; + language?: STTRealtimeLanguages | string; + mode?: STTModes | string; + prompt?: string; + returnTimestamps?: boolean; + sampleRate?: number; + streamType?: RealtimeStreamType; + vadMinSilenceMs?: number; + vadMinSpeechMs?: number; + vadPrefixPaddingMs?: number; + vadSotThreshold?: number; +} + // Warning: (ae-forgotten-export) The symbol "STTBaseOptions" needs to be exported by the entry point index.d.ts // Warning: (ae-missing-release-tag) "STTTranslateOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // diff --git a/plugins/sarvam/src/_utils.ts b/plugins/sarvam/src/_utils.ts new file mode 100644 index 000000000..3f2ead75d --- /dev/null +++ b/plugins/sarvam/src/_utils.ts @@ -0,0 +1,39 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** Accumulates pushed numeric values and reports the total on a fixed interval. */ +export class PeriodicCollector { + private duration: number; + private callback: (value: number) => void; + private lastFlushTime: number; + private total: number | null = null; + + /** + * @param callback - function to call with the accumulated value when the duration expires + * @param options - options object + */ + constructor(callback: (value: number) => void, options: { duration: number }) { + this.duration = options.duration; + this.callback = callback; + this.lastFlushTime = performance.now() / 1000; + } + + /** Add a value to the accumulator. */ + push(value: number): void { + this.total = this.total === null ? value : this.total + value; + + if (performance.now() / 1000 - this.lastFlushTime >= this.duration) { + this.flush(); + } + } + + /** Force the callback to be called with the current total if non-zero. */ + flush(): void { + if (this.total !== null) { + this.callback(this.total); + this.total = null; + } + this.lastFlushTime = performance.now() / 1000; + } +} diff --git a/plugins/sarvam/src/index.ts b/plugins/sarvam/src/index.ts index 659aea50a..668f1e98b 100644 --- a/plugins/sarvam/src/index.ts +++ b/plugins/sarvam/src/index.ts @@ -12,6 +12,7 @@ export { type STTTranslateOptions, type STTV3Options, } from './stt.js'; +export { STTRealtime, RealtimeSpeechStream, type STTRealtimeOptions } from './stt_realtime.js'; export { ChunkedStream, SynthesizeStream, diff --git a/plugins/sarvam/src/models.ts b/plugins/sarvam/src/models.ts index 6bf70374a..357e6be67 100644 --- a/plugins/sarvam/src/models.ts +++ b/plugins/sarvam/src/models.ts @@ -136,3 +136,56 @@ export type STTV3Languages = /** All supported STT language codes */ export type STTLanguages = STTV2Languages | STTV3Languages; + +// --------------------------------------------------------------------------- +// Realtime STT model types +// --------------------------------------------------------------------------- + +/** + * Model used by {@link STTRealtime}. Pinned to Sarvam's realtime API and not configurable. + * + * @see {@link https://docs.sarvam.ai/api-reference/speech-to-text/transcribe/realtime/ws | Sarvam realtime STT WebSocket docs} + */ +export type STTRealtimeModel = 'saaras:v3-realtime'; + +/** Latency profile for the realtime stream. */ +export type RealtimeStreamType = 'fast' | 'balanced' | 'simulated'; + +/** How turn boundaries are determined on the realtime API. */ +export type RealtimeEndpointing = 'vad' | 'manual'; + +/** Wire audio encodings accepted by the realtime API. */ +export type RealtimeEncoding = 'linear16' | 'linear32' | 'mulaw' | 'alaw'; + +/** + * Languages supported by the realtime API (BCP-47), plus `'auto'` for adaptive language + * identification. + * + * @remarks + * Odia is `or-IN` on this API, unlike the legacy {@link STTLanguages} which uses `od-IN`. + */ +export type STTRealtimeLanguages = + | 'auto' + | 'as-IN' + | 'bn-IN' + | 'brx-IN' + | 'doi-IN' + | 'en-IN' + | 'gu-IN' + | 'hi-IN' + | 'kn-IN' + | 'kok-IN' + | 'ks-IN' + | 'mai-IN' + | 'ml-IN' + | 'mni-IN' + | 'mr-IN' + | 'ne-IN' + | 'or-IN' + | 'pa-IN' + | 'sa-IN' + | 'sat-IN' + | 'sd-IN' + | 'ta-IN' + | 'te-IN' + | 'ur-IN'; diff --git a/plugins/sarvam/src/stt_realtime.test.ts b/plugins/sarvam/src/stt_realtime.test.ts new file mode 100644 index 000000000..04a2706e5 --- /dev/null +++ b/plugins/sarvam/src/stt_realtime.test.ts @@ -0,0 +1,391 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { DEFAULT_API_CONNECT_OPTIONS, stt } from '@livekit/agents'; +import type { stt as sttNamespace } from '@livekit/agents'; +import { AudioFrame } from '@livekit/rtc-node'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { RealtimeSpeechStream } from './stt_realtime.js'; +import { STTRealtime, encodePcmForWire } from './stt_realtime.js'; + +interface ServerEvent { + event: string; + [key: string]: unknown; +} + +// vi.mock/vi.hoisted factories are hoisted above this file's own imports, so the fake +// WebSocket (including its own minimal event emitter) is defined entirely inside the callback. +const { sockets, FakeWebSocket } = vi.hoisted(() => { + class MiniEmitterImpl { + #listeners = new Map void>>(); + + on(event: string, listener: (...args: unknown[]) => void): this { + if (!this.#listeners.has(event)) this.#listeners.set(event, new Set()); + this.#listeners.get(event)!.add(listener); + return this; + } + + once(event: string, listener: (...args: unknown[]) => void): this { + const wrapped = (...args: unknown[]) => { + this.off(event, wrapped); + listener(...args); + }; + return this.on(event, wrapped); + } + + off(event: string, listener: (...args: unknown[]) => void): this { + this.#listeners.get(event)?.delete(listener); + return this; + } + + emit(event: string, ...args: unknown[]): boolean { + const listeners = this.#listeners.get(event); + if (!listeners || listeners.size === 0) return false; + for (const listener of [...listeners]) listener(...args); + return true; + } + } + + const sockets: InstanceType[] = []; + + class FakeWebSocketImpl extends MiniEmitterImpl { + static readonly OPEN = 1; + static readonly CLOSED = 3; + readyState = 0; + sent: (string | Buffer)[] = []; + url: string; + options?: unknown; + + constructor(url: string, options?: unknown) { + super(); + this.url = url; + this.options = options; + sockets.push(this); + queueMicrotask(() => { + this.readyState = FakeWebSocketImpl.OPEN; + this.emit('open'); + }); + } + + send(data: string | Buffer) { + this.sent.push(data); + } + + close(code = 1000, reason = '') { + if (this.readyState === FakeWebSocketImpl.CLOSED) return; + this.readyState = FakeWebSocketImpl.CLOSED; + this.emit('close', code, Buffer.from(reason)); + } + } + + return { sockets, FakeWebSocket: FakeWebSocketImpl }; +}); + +vi.mock('ws', () => ({ WebSocket: FakeWebSocket })); + +type FakeWebSocketInstance = InstanceType; + +function frame(): AudioFrame { + return new AudioFrame(new Int16Array(160), 16000, 1, 160); +} + +function toneFrame(samples: number[]): AudioFrame { + return new AudioFrame(new Int16Array(samples), 16000, 1, samples.length); +} + +// Standard ITU G.711 A-law decoder — independent of this plugin's encoder, used to round-trip +// verify linearToAlaw's output rather than asserting on encoder-internal byte values. +function alawDecode(aVal: number): number { + aVal ^= 0x55; + let t = (aVal & 0x0f) << 4; + const seg = (aVal & 0x70) >> 4; + if (seg === 0) t += 8; + else if (seg === 1) t += 0x108; + else { + t += 0x108; + t <<= seg - 1; + } + return aVal & 0x80 ? t : -t; +} + +// Standard ITU G.711 mu-law decoder — independent of this plugin's encoder. +function mulawDecode(uVal: number): number { + uVal = ~uVal & 0xff; + let t = ((uVal & 0x0f) << 3) + 0x84; + t <<= (uVal & 0x70) >> 4; + return uVal & 0x80 ? 0x84 - t : t - 0x84; +} + +async function waitForSocket(): Promise { + for (let i = 0; i < 100 && sockets.length === 0; i++) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } + if (sockets.length === 0) throw new Error('no Sarvam realtime STT socket was created'); + return sockets[sockets.length - 1]!; +} + +function onceOpen(socket: FakeWebSocketInstance): Promise { + if (socket.readyState === FakeWebSocket.OPEN) return Promise.resolve(); + return new Promise((resolve) => socket.once('open', () => resolve())); +} + +function send(socket: FakeWebSocketInstance, event: ServerEvent): void { + socket.emit('message', Buffer.from(JSON.stringify(event)), false); +} + +function texts(events: sttNamespace.SpeechEvent[], type: sttNamespace.SpeechEventType): string[] { + return events.filter((e) => e.type === type).map((e) => e.alternatives![0]!.text); +} + +async function scriptAndDrain( + stream: RealtimeSpeechStream, + script: (socket: FakeWebSocketInstance) => void | Promise, +): Promise { + const socket = await waitForSocket(); + await onceOpen(socket); + await script(socket); + + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + return events; +} + +describe('Sarvam realtime STT', () => { + beforeEach(() => { + sockets.length = 0; + }); + + it('gates the final transcript behind speech_end in vad endpointing mode', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const stream = sttRealtime.stream(); + stream.pushFrame(frame()); + stream.endInput(); + + const events = await scriptAndDrain(stream, (socket) => { + send(socket, { event: 'session.begin' }); + send(socket, { event: 'vad.speech_start' }); + send(socket, { event: 'transcript.partial', text: 'Hel' }); + send(socket, { event: 'transcript.final', text: 'Hello there', confidence: 0.9 }); + // The final must not surface yet — it's held until vad.speech_end supplies the boundary. + send(socket, { event: 'vad.speech_end' }); + send(socket, { event: 'session.end', audio_duration_s: 1 }); + }); + + expect(texts(events, stt.SpeechEventType.INTERIM_TRANSCRIPT)).toEqual(['Hel']); + expect(texts(events, stt.SpeechEventType.FINAL_TRANSCRIPT)).toEqual(['Hello there']); + + const eosIndex = events.findIndex((e) => e.type === stt.SpeechEventType.END_OF_SPEECH); + const finalIndex = events.findIndex((e) => e.type === stt.SpeechEventType.FINAL_TRANSCRIPT); + expect(eosIndex).toBeGreaterThanOrEqual(0); + expect(finalIndex).toBeLessThan(eosIndex); + + expect(events.some((e) => e.type === stt.SpeechEventType.RECOGNITION_USAGE)).toBe(true); + }); + + it('emits the final transcript immediately in manual endpointing mode', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key', endpointing: 'manual' }); + const stream = sttRealtime.stream(); + stream.pushFrame(frame()); + stream.flush(); + stream.endInput(); + + const events = await scriptAndDrain(stream, (socket) => { + // No vad.speech_end is ever sent — manual mode must not wait for one. + send(socket, { event: 'transcript.final', text: 'Manual mode transcript' }); + send(socket, { event: 'session.end', audio_duration_s: 1 }); + }); + + expect(texts(events, stt.SpeechEventType.FINAL_TRANSCRIPT)).toEqual(['Manual mode transcript']); + expect(events.some((e) => e.type === stt.SpeechEventType.START_OF_SPEECH)).toBe(true); + expect(events.some((e) => e.type === stt.SpeechEventType.END_OF_SPEECH)).toBe(true); + }); + + it('never reconnects, even if a higher maxRetry is requested by the caller', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const errors: unknown[] = []; + sttRealtime.on('error', (e) => errors.push(e)); + + const stream = sttRealtime.stream({ + connOptions: { ...DEFAULT_API_CONNECT_OPTIONS, maxRetry: 5 }, + }); + stream.pushFrame(frame()); + stream.endInput(); + + const socket = await waitForSocket(); + await onceOpen(socket); + socket.close(1008, 'session timed out'); + + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + + expect(sockets).toHaveLength(1); + expect(errors).toHaveLength(1); + }); + + it('throws when _recognize is called (streaming only)', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + await expect(sttRealtime._recognize()).rejects.toThrow(/only supports streaming/i); + }); + + it('encodes A-law samples with the correct sign and magnitude (regression for inverted sign bug)', () => { + // Round-trip through an independent reference decoder — not the encoder's own logic — + // so this actually catches a sign inversion instead of just re-asserting the bug. + // Values well above A-law's quantization floor (its lowest segment collapses everything + // under ~16 to one bucket, so sign isn't meaningful that close to zero — see below). + for (const sample of [100, 5000, 16000, 32000, -100, -5000, -16000, -32000]) { + const encoded = encodePcmForWire('alaw', toneFrame([sample])); + const decoded = alawDecode(encoded[0]!); + expect(Math.sign(decoded)).toBe(Math.sign(sample)); + // A-law is lossy/companded, not lossless — allow generous relative tolerance. + expect(Math.abs(decoded - sample)).toBeLessThan(Math.abs(sample) * 0.15 + 32); + } + + // A-law's zero code decodes to a small nonzero value by design (segment-0 offset of 8) — + // just confirm it stays near silence rather than asserting an exact sign. + const zeroEncoded = encodePcmForWire('alaw', toneFrame([0])); + expect(Math.abs(alawDecode(zeroEncoded[0]!))).toBeLessThanOrEqual(8); + }); + + it('encodes mu-law samples with the correct sign and magnitude', () => { + for (const sample of [100, 5000, 16000, 32000, -100, -5000, -16000, -32000]) { + const encoded = encodePcmForWire('mulaw', toneFrame([sample])); + const decoded = mulawDecode(encoded[0]!); + expect(Math.sign(decoded)).toBe(Math.sign(sample)); + expect(Math.abs(decoded - sample)).toBeLessThan(Math.abs(sample) * 0.15 + 32); + } + }); + + it('does not hang when the server ends the session before the caller ends input', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const stream = sttRealtime.stream(); + stream.pushFrame(frame()); + // Deliberately not calling endInput()/flush() — the caller may keep the stream open + // across turns, so a server-initiated end must not depend on the caller closing input. + + const socket = await waitForSocket(); + await onceOpen(socket); + send(socket, { event: 'vad.speech_start' }); + send(socket, { event: 'transcript.final', text: 'done' }); + send(socket, { event: 'vad.speech_end' }); + send(socket, { event: 'session.end', audio_duration_s: 1 }); + socket.close(1000, ''); + + const drain = (async () => { + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + return events; + })(); + const timeout = new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 2000)); + + const result = await Promise.race([drain, timeout]); + expect(result).not.toBe('timeout'); + expect( + texts(result as sttNamespace.SpeechEvent[], stt.SpeechEventType.FINAL_TRANSCRIPT), + ).toEqual(['done']); + }); + + it('stops forwarding updateOptions to a stream that completed naturally', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const stream = sttRealtime.stream(); + const updateOptionsSpy = vi.spyOn(stream, 'updateOptions'); + stream.pushFrame(frame()); + stream.endInput(); + + await scriptAndDrain(stream, (socket) => { + send(socket, { event: 'session.end', audio_duration_s: 1 }); + }); + + sttRealtime.updateOptions({ language: 'hi-IN' }); + expect(updateOptionsSpy).not.toHaveBeenCalled(); + }); + + it('rejects non-mono audio instead of silently corrupting it', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const errors: unknown[] = []; + sttRealtime.on('error', (e) => errors.push(e)); + + const stream = sttRealtime.stream(); + const stereoFrame = new AudioFrame(new Int16Array(320), 16000, 2, 160); + stream.pushFrame(stereoFrame); + stream.endInput(); + + // Deliberately not synchronizing with the socket's 'open' event here: the channel + // validation error fires almost synchronously once the (mocked) socket opens, which can + // close the socket before this test's own listener attaches — waiting on 'open' again + // would then hang forever. Draining the stream is enough to observe the error. + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + + expect(errors).toHaveLength(1); + expect(String((errors[0] as { error: Error }).error.message)).toMatch(/mono/i); + }); + + it('keeps a live stream on its original wire encoding after updateOptions', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key', encoding: 'linear16' }); + const stream = sttRealtime.stream(); + + const socket = await waitForSocket(); + await onceOpen(socket); + + stream.pushFrame(toneFrame(new Array(800).fill(1000))); + await new Promise((r) => setTimeout(r, 0)); + + stream.updateOptions({ encoding: 'mulaw' }); + + stream.pushFrame(toneFrame(new Array(800).fill(1000))); + stream.endInput(); + await new Promise((r) => setTimeout(r, 0)); + socket.close(1000, ''); + + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + + const binaryPayloads = socket.sent.filter((d): d is Buffer => Buffer.isBuffer(d)); + expect(binaryPayloads.length).toBeGreaterThanOrEqual(2); + for (const payload of binaryPayloads) { + expect(payload.byteLength).toBe(1600); // still linear16 (2 bytes/sample), not mulaw + } + }); + + it('sends the final buffered audio chunk when input ends without a trailing flush()', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key' }); + const stream = sttRealtime.stream(); + + const socket = await waitForSocket(); + await onceOpen(socket); + + stream.pushFrame(toneFrame(new Array(200).fill(1000))); + stream.endInput(); + await new Promise((r) => setTimeout(r, 0)); + socket.close(1000, ''); + + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + + const binaryPayloads = socket.sent.filter((d): d is Buffer => Buffer.isBuffer(d)); + expect(binaryPayloads.some((p) => p.byteLength === 400)).toBe(true); + }); + + it('ends the manual-mode turn when input ends without a trailing flush()', async () => { + const sttRealtime = new STTRealtime({ apiKey: 'test-key', endpointing: 'manual' }); + const stream = sttRealtime.stream(); + + const socket = await waitForSocket(); + await onceOpen(socket); + + stream.pushFrame(toneFrame(new Array(200).fill(1000))); + stream.endInput(); + await new Promise((r) => setTimeout(r, 0)); + socket.close(1000, ''); + + const events: sttNamespace.SpeechEvent[] = []; + for await (const event of stream) events.push(event); + + const jsonPayloads = socket.sent + .filter((d): d is string => typeof d === 'string') + .map((d) => JSON.parse(d) as { event: string }); + expect(jsonPayloads.some((p) => p.event === 'speech_start')).toBe(true); + expect(jsonPayloads.some((p) => p.event === 'speech_end')).toBe(true); + expect(events.some((e) => e.type === stt.SpeechEventType.END_OF_SPEECH)).toBe(true); + }); +}); diff --git a/plugins/sarvam/src/stt_realtime.ts b/plugins/sarvam/src/stt_realtime.ts new file mode 100644 index 000000000..1115dbc3e --- /dev/null +++ b/plugins/sarvam/src/stt_realtime.ts @@ -0,0 +1,1035 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { + type APIConnectOptions, + APIConnectionError, + APIStatusError, + AudioByteStream, + DEFAULT_API_CONNECT_OPTIONS, + log, + normalizeLanguage, + stt, + waitForAbort, + waitForWebSocketOpen, +} from '@livekit/agents'; +import type { AudioFrame } from '@livekit/rtc-node'; +import { type RawData, WebSocket } from 'ws'; +import { PeriodicCollector } from './_utils.js'; +import type { + RealtimeEncoding, + RealtimeEndpointing, + RealtimeStreamType, + STTModes, + STTRealtimeLanguages, + STTRealtimeModel, +} from './models.js'; + +const REALTIME_MODEL: STTRealtimeModel = 'saaras:v3-realtime'; +const REALTIME_WS_URL = 'wss://api.sarvam.ai/speech-to-text-realtime/ws'; +const AUDIO_CHUNK_MS = 50; +const NUM_CHANNELS = 1; +const USAGE_FLUSH_INTERVAL_S = 5; + +const SUPPORTED_SAMPLE_RATES = new Set([8000, 16000]); +const SUPPORTED_STREAM_TYPES = new Set(['fast', 'balanced', 'simulated']); +const SUPPORTED_ENDPOINTING = new Set(['vad', 'manual']); +const SUPPORTED_ENCODINGS = new Set(['linear16', 'linear32', 'mulaw', 'alaw']); +const SUPPORTED_MODES = new Set([ + 'transcribe', + 'translate', + 'verbatim', + 'translit', + 'codemix', +]); + +// --------------------------------------------------------------------------- +// Options +// --------------------------------------------------------------------------- + +/** + * Options for {@link STTRealtime}. + * + * @see {@link https://docs.sarvam.ai/api-reference/speech-to-text/transcribe/realtime/ws | Sarvam realtime STT WebSocket docs} + */ +export interface STTRealtimeOptions { + /** Sarvam API key. Defaults to $SARVAM_API_KEY */ + apiKey?: string; + /** Language code (BCP-47), or 'auto' for adaptive language identification. Default: 'en-IN'. */ + language?: STTRealtimeLanguages | string; + /** Latency profile for the stream. Default: 'balanced'. */ + streamType?: RealtimeStreamType; + /** The task applied to final transcripts. Default: 'transcribe'. */ + mode?: STTModes | string; + /** How turn boundaries are determined. Default: 'vad'. */ + endpointing?: RealtimeEndpointing; + /** Wire audio encoding. Default: 'linear16'. */ + encoding?: RealtimeEncoding; + /** Input audio sample rate. Must be 8000 or 16000. Default: 16000. */ + sampleRate?: number; + /** Terminology or context hint used to bias decoding. */ + prompt?: string; + /** Whether final transcripts include start/end timestamps. Default: false. */ + returnTimestamps?: boolean; + /** Speech activation threshold (0.0-1.0). Applies only when `endpointing: 'vad'`. */ + vadSotThreshold?: number; + /** Minimum speech duration (ms) before a turn opens. Applies only when `endpointing: 'vad'`. */ + vadMinSpeechMs?: number; + /** End-of-turn silence duration (ms). Applies only when `endpointing: 'vad'`. */ + vadMinSilenceMs?: number; + /** + * Pre-speech padding (ms) included at the start of a turn. Applies only when + * `endpointing: 'vad'`. Connection-time only — updates only affect newly created streams. + */ + vadPrefixPaddingMs?: number; +} + +interface ResolvedRealtimeOptions { + apiKey: string; + language: string; + streamType: RealtimeStreamType; + mode: string; + endpointing: RealtimeEndpointing; + encoding: RealtimeEncoding; + sampleRate: number; + prompt?: string; + returnTimestamps: boolean; + vadSotThreshold?: number; + vadMinSpeechMs?: number; + vadMinSilenceMs?: number; + vadPrefixPaddingMs?: number; +} + +function resolveRealtimeOptions(opts: Partial): ResolvedRealtimeOptions { + const apiKey = opts.apiKey ?? process.env.SARVAM_API_KEY; + if (!apiKey) { + throw new Error('Sarvam API key is required, whether as an argument or as $SARVAM_API_KEY'); + } + + const streamType = opts.streamType ?? 'balanced'; + if (!SUPPORTED_STREAM_TYPES.has(streamType)) { + throw new Error(`unsupported Sarvam realtime STT streamType: ${streamType}`); + } + const mode = opts.mode ?? 'transcribe'; + if (!SUPPORTED_MODES.has(mode)) { + throw new Error(`unsupported Sarvam realtime STT mode: ${mode}`); + } + const endpointing = opts.endpointing ?? 'vad'; + if (!SUPPORTED_ENDPOINTING.has(endpointing)) { + throw new Error(`unsupported Sarvam realtime STT endpointing: ${endpointing}`); + } + const encoding = opts.encoding ?? 'linear16'; + if (!SUPPORTED_ENCODINGS.has(encoding)) { + throw new Error(`unsupported Sarvam realtime STT encoding: ${encoding}`); + } + const sampleRate = opts.sampleRate ?? 16000; + if (!SUPPORTED_SAMPLE_RATES.has(sampleRate)) { + throw new Error( + `unsupported Sarvam realtime STT sampleRate: ${sampleRate} (must be 8000 or 16000)`, + ); + } + if ( + opts.vadSotThreshold !== undefined && + (opts.vadSotThreshold < 0 || opts.vadSotThreshold > 1) + ) { + throw new Error('vadSotThreshold must be between 0.0 and 1.0'); + } + for (const [name, value] of [ + ['vadMinSpeechMs', opts.vadMinSpeechMs], + ['vadMinSilenceMs', opts.vadMinSilenceMs], + ['vadPrefixPaddingMs', opts.vadPrefixPaddingMs], + ] as const) { + if (value !== undefined && value < 0) { + throw new Error(`${name} must be non-negative`); + } + } + + return { + apiKey, + language: normalizeLanguage(opts.language ?? 'en-IN'), + streamType, + mode, + endpointing, + encoding, + sampleRate, + prompt: opts.prompt, + returnTimestamps: opts.returnTimestamps ?? false, + vadSotThreshold: opts.vadSotThreshold, + vadMinSpeechMs: opts.vadMinSpeechMs, + vadMinSilenceMs: opts.vadMinSilenceMs, + vadPrefixPaddingMs: opts.vadPrefixPaddingMs, + }; +} + +function buildRealtimeWsUrl(opts: ResolvedRealtimeOptions): string { + const params = new URLSearchParams(); + params.set('language_code', opts.language); + params.set('stream_type', opts.streamType); + params.set('endpointing', opts.endpointing); + params.set('encoding', opts.encoding); + params.set('sample_rate', String(opts.sampleRate)); + params.set('model', REALTIME_MODEL); + params.set('mode', opts.mode); + params.set('return_timestamps', String(opts.returnTimestamps)); + if (opts.prompt != null) { + params.set('prompt', opts.prompt); + } + + if (opts.endpointing === 'vad') { + if (opts.vadSotThreshold != null) { + params.set('threshold', String(opts.vadSotThreshold)); + } + if (opts.vadMinSpeechMs != null) { + params.set('min_speech_duration_ms', String(opts.vadMinSpeechMs)); + } + if (opts.vadMinSilenceMs != null) { + params.set('silence_duration_ms', String(opts.vadMinSilenceMs)); + } + if (opts.vadPrefixPaddingMs != null) { + params.set('prefix_padding_ms', String(opts.vadPrefixPaddingMs)); + } + } + + return `${REALTIME_WS_URL}?${params.toString()}`; +} + +function buildConfigUpdatePayload( + previous: ResolvedRealtimeOptions, + current: ResolvedRealtimeOptions, +): Record | null { + const payload: Record = { event: 'config.update' }; + const entries: [string, unknown, unknown][] = [ + ['language_code', previous.language, current.language], + ['stream_type', previous.streamType, current.streamType], + ['mode', previous.mode, current.mode], + ['prompt', previous.prompt, current.prompt], + ['endpointing', previous.endpointing, current.endpointing], + ['threshold', previous.vadSotThreshold, current.vadSotThreshold], + ['min_speech_duration_ms', previous.vadMinSpeechMs, current.vadMinSpeechMs], + ['silence_duration_ms', previous.vadMinSilenceMs, current.vadMinSilenceMs], + ]; + for (const [key, oldValue, newValue] of entries) { + if (oldValue !== newValue) { + payload[key] = key === 'prompt' && newValue == null ? '' : newValue; + } + } + return Object.keys(payload).length > 1 ? payload : null; +} + +// --------------------------------------------------------------------------- +// PCM wire encoders +// --------------------------------------------------------------------------- + +// Faithful port of the standard ITU-T G.711 reference encoders (the same `linear2ulaw`/ +// `linear2alaw` algorithm — segment tables, bit-exact shifts and masks — found in Sun's +// canonical g711.c and used by virtually every G.711 implementation since). +const MULAW_BIAS = 0x84; +const MULAW_CLIP = 8159; +const SEG_UEND = [0x3f, 0x7f, 0xff, 0x1ff, 0x3ff, 0x7ff, 0xfff, 0x1fff]; +const SEG_AEND = [0x1f, 0x3f, 0x7f, 0xff, 0x1ff, 0x3ff, 0x7ff, 0xfff]; + +function search(val: number, table: number[]): number { + for (let i = 0; i < table.length; i++) { + if (val <= table[i]!) return i; + } + return table.length; +} + +function linearToMulaw(sample: number): number { + let pcm = sample >> 2; + let mask: number; + if (pcm < 0) { + pcm = -pcm; + mask = 0x7f; + } else { + mask = 0xff; + } + if (pcm > MULAW_CLIP) pcm = MULAW_CLIP; + pcm += MULAW_BIAS >> 2; + + const seg = search(pcm, SEG_UEND); + if (seg >= 8) return 0x7f ^ mask; + const uval = (seg << 4) | ((pcm >> (seg + 1)) & 0x0f); + return (uval ^ mask) & 0xff; +} + +function linearToAlaw(sample: number): number { + let pcm = sample >> 3; + let mask: number; + if (pcm >= 0) { + mask = 0xd5; + } else { + mask = 0x55; + pcm = -pcm - 1; + } + + const seg = search(pcm, SEG_AEND); + if (seg >= 8) return (0x7f ^ mask) & 0xff; + let aval = seg << 4; + aval |= seg < 2 ? (pcm >> 1) & 0x0f : (pcm >> seg) & 0x0f; + return (aval ^ mask) & 0xff; +} + +/** @internal exported only for unit tests */ +export function encodePcmForWire(encoding: RealtimeEncoding, frame: AudioFrame): Buffer { + const int16 = frame.data; + switch (encoding) { + case 'linear16': + return Buffer.from(int16.buffer, int16.byteOffset, int16.byteLength); + case 'linear32': { + const out = Buffer.alloc(int16.length * 4); + for (let i = 0; i < int16.length; i++) { + out.writeInt32LE(int16[i]! * 65536, i * 4); + } + return out; + } + case 'mulaw': { + const out = Buffer.alloc(int16.length); + for (let i = 0; i < int16.length; i++) out[i] = linearToMulaw(int16[i]!); + return out; + } + case 'alaw': { + const out = Buffer.alloc(int16.length); + for (let i = 0; i < int16.length; i++) out[i] = linearToAlaw(int16[i]!); + return out; + } + } +} + +function looksLikeErrorText(value: string): boolean { + const lowered = value.toLowerCase(); + return [ + 'error', + 'invalid', + 'failed', + 'forbidden', + 'unauthorized', + 'not found', + 'rate limit', + 'timeout', + ].some((hint) => lowered.includes(hint)); +} + +// --------------------------------------------------------------------------- +// Server event shape (partial — only fields we read) +// --------------------------------------------------------------------------- + +interface RealtimeServerEvent { + event?: string; + request_id?: string; + session_id?: string; + data?: { request_id?: string }; + metadata?: { request_id?: string }; + utterance_idx?: number; + text?: string; + language?: string; + language_confidence?: number; + confidence?: number; + start_s?: number; + end_s?: number; + audio_duration_s?: number; + applied?: unknown[]; + is_fatal?: boolean; + code?: string; + status_code?: number; + message?: string; +} + +// --------------------------------------------------------------------------- +// STTRealtime — connects to Sarvam's realtime WebSocket API (saaras:v3-realtime) +// --------------------------------------------------------------------------- + +/** + * Sarvam AI realtime speech-to-text, using the `saaras:v3-realtime` model. + * + * @remarks + * Realtime streams don't reconnect after a socket failure — Sarvam bills per connection, so + * `stream()` forces `connOptions.maxRetry = 0`. Create a new stream (or restart the session) if + * the connection drops. + * + * `apiKey` must be set via the constructor argument or the `SARVAM_API_KEY` environment variable. + * + * @see {@link https://docs.sarvam.ai/api-reference/speech-to-text/transcribe/realtime/ws | Sarvam realtime STT WebSocket docs} + */ +export class STTRealtime extends stt.STT { + label = 'sarvam.STTRealtime'; + #opts: ResolvedRealtimeOptions; + #streams = new Set(); + + constructor(opts: Partial = {}) { + const resolved = resolveRealtimeOptions(opts); + super({ streaming: true, interimResults: true, alignedTranscript: false }); + this.#opts = resolved; + } + + get model(): string { + return REALTIME_MODEL; + } + + get provider(): string { + return 'Sarvam'; + } + + /** + * Update connection options for future streams. Fields marked connection-time only + * (`sampleRate`, `returnTimestamps`, `vadPrefixPaddingMs`) apply only to newly created streams; + * other fields are also forwarded as an in-band `config.update` to every currently open stream. + */ + updateOptions(opts: Partial): void { + this.#opts = resolveRealtimeOptions({ ...this.#opts, ...opts }); + for (const stream of this.#streams) { + stream.updateOptions(opts); + } + } + + async _recognize(): Promise { + throw new Error('Sarvam realtime STT only supports streaming recognition'); + } + + stream(options?: { language?: string; connOptions?: APIConnectOptions }): RealtimeSpeechStream { + // This endpoint bills per connection, so the stream must never silently reconnect — + // forcing max_retry to 0 mirrors the Python SDK's `stream()`. + const connOptions: APIConnectOptions = { + ...(options?.connOptions ?? DEFAULT_API_CONNECT_OPTIONS), + maxRetry: 0, + }; + const opts: ResolvedRealtimeOptions = + options?.language !== undefined + ? { ...this.#opts, language: normalizeLanguage(options.language) } + : this.#opts; + + const stream = new RealtimeSpeechStream(this, opts, connOptions, () => { + this.#streams.delete(stream); + }); + this.#streams.add(stream); + return stream; + } +} + +// --------------------------------------------------------------------------- +// RealtimeSpeechStream — per-connection state machine +// --------------------------------------------------------------------------- + +export class RealtimeSpeechStream extends stt.SpeechStream { + label = 'sarvam.RealtimeSpeechStream'; + #opts: ResolvedRealtimeOptions; + #logger = log(); + #onClose?: () => void; + #closeNotified = false; + #ws?: WebSocket; + + #requestId = ''; + #sessionId = ''; + #sessionEnded = false; + + #activeEndpointing: RealtimeEndpointing; + #pendingEndpointing?: RealtimeEndpointing; + #endpointingUpdateAcknowledged = true; + #endpointingUpdateSent = false; + #pendingConfigUpdate: Record | null = null; + + #manualSpeechStarted = false; + #utteranceInProgress = false; + #pendingFinalData: RealtimeServerEvent | null = null; + #utteranceSpeechEndAudioPos: number | null = null; + #utteranceSpeechEndWall: number | null = null; + #finalReceivedForUtterance = false; + #eosEmittedForUtterance = false; + + #audioPosition = 0; + #totalReportedAudioDuration = 0; + #serverAudioDurationReported = false; + #audioDurationCollector: PeriodicCollector; + + constructor( + sttInstance: STTRealtime, + opts: ResolvedRealtimeOptions, + connOptions: APIConnectOptions, + onClose?: () => void, + ) { + super(sttInstance, opts.sampleRate, connOptions); + this.#opts = opts; + this.#activeEndpointing = opts.endpointing; + this.#onClose = onClose; + this.#audioDurationCollector = new PeriodicCollector((duration) => this.#emitUsage(duration), { + duration: USAGE_FLUSH_INTERVAL_S, + }); + } + + /** Update this stream's live options — see {@link STTRealtime.updateOptions}. */ + updateOptions(opts: Partial): void { + const previous = this.#opts; + let next = resolveRealtimeOptions({ ...previous, ...opts }); + + const connectionOnly: string[] = []; + if (next.sampleRate !== previous.sampleRate) { + connectionOnly.push('sampleRate'); + next = { ...next, sampleRate: previous.sampleRate }; + } + if (next.returnTimestamps !== previous.returnTimestamps) { + connectionOnly.push('returnTimestamps'); + next = { ...next, returnTimestamps: previous.returnTimestamps }; + } + if (next.vadPrefixPaddingMs !== previous.vadPrefixPaddingMs) { + connectionOnly.push('vadPrefixPaddingMs'); + next = { ...next, vadPrefixPaddingMs: previous.vadPrefixPaddingMs }; + } + if (next.encoding !== previous.encoding) { + // Server's decoder is fixed at connect time by the `encoding` query param. + connectionOnly.push('encoding'); + next = { ...next, encoding: previous.encoding }; + } + if (connectionOnly.length > 0) { + this.#logger.warn( + { options: connectionOnly }, + 'Sarvam realtime STT connection-only option updates only apply to new streams', + ); + } + + this.#opts = next; + + if (next.endpointing !== previous.endpointing) { + this.#pendingEndpointing = next.endpointing; + this.#endpointingUpdateAcknowledged = false; + this.#endpointingUpdateSent = false; + } + + const update = buildConfigUpdatePayload(previous, next); + if (update) { + this.#pendingConfigUpdate = { ...this.#pendingConfigUpdate, ...update }; + } + } + + close(): void { + this.#flushLocalUsageFallback(); + try { + this.#ws?.close(); + } catch { + // already closing/closed + } + super.close(); + this.#notifyClosed(); + } + + /** + * Notify {@link STTRealtime} that this stream is done, so it stops forwarding + * `updateOptions()` calls to it. Idempotent: called from both `close()` (caller-initiated) + * and `run()`'s `finally` (natural completion or error), which can race each other. + */ + #notifyClosed(): void { + if (this.#closeNotified) return; + this.#closeNotified = true; + this.#onClose?.(); + } + + protected async run(): Promise { + const url = buildRealtimeWsUrl(this.#opts); + const ws = new WebSocket(url, { + headers: { + 'API-SUBSCRIPTION-KEY': this.#opts.apiKey, + 'User-Agent': 'LiveKit-Agents-JS', + }, + }); + this.#ws = ws; + // Scoped to this connection attempt: aborted once the server->client side settles (session + // ended, or the socket closed) so the client->server audio pump — which would otherwise + // block forever on this.input.next() waiting for a frame that may never come — stops too. + const audioAbort = new AbortController(); + + try { + await waitForWebSocketOpen(ws, 'Sarvam realtime STT'); + + const audioTask = this.#processAudio(ws, audioAbort.signal); + const messagesTask = this.#processMessages(ws); + + const first = await Promise.race([ + audioTask.then(() => 'audio' as const), + messagesTask.then(() => 'messages' as const), + waitForAbort(this.abortSignal).then(() => 'abort' as const), + ]); + if (first === 'abort') return; + + // Whichever side finished first, stop the other and wait for it to settle. If audio + // finished first (endInput()/flush()), this just waits for any trailing server messages + // (e.g. a final transcript, session.end) as before. If messages finished first (server + // ended the session or closed cleanly before the caller ended input), this now unblocks + // the audio pump instead of leaving it hanging. + audioAbort.abort(); + await Promise.all([audioTask, messagesTask]); + } catch (error) { + if (this.abortSignal.aborted) return; + if (error instanceof APIStatusError || error instanceof APIConnectionError) throw error; + throw new APIConnectionError({ + message: `Sarvam realtime STT connection failed: ${error instanceof Error ? error.message : String(error)}`, + }); + } finally { + audioAbort.abort(); + try { + ws.close(); + } catch { + // already closing/closed + } + this.#ws = undefined; + this.#notifyClosed(); + } + } + + // ------------------------------------------------------------------------- + // Client -> server (audio pump) + // ------------------------------------------------------------------------- + + async #processAudio(ws: WebSocket, signal: AbortSignal): Promise { + const samplesPerChannel = Math.max( + Math.floor((this.#opts.sampleRate * AUDIO_CHUNK_MS) / 1000), + 1, + ); + const audioStream = new AudioByteStream(this.#opts.sampleRate, NUM_CHANNELS, samplesPerChannel); + + while (!this.#sessionEnded && ws.readyState === WebSocket.OPEN && !signal.aborted) { + let result: IteratorResult; + try { + result = await this.input.next({ signal }); + } catch (error) { + if (signal.aborted) break; + throw error; + } + if (result.done) break; + + const data = result.value; + this.#sendPendingConfigUpdate(ws); + + const isFlush = data === RealtimeSpeechStream.FLUSH_SENTINEL; + let frames: AudioFrame[]; + if (isFlush) { + frames = audioStream.flush(); + } else { + if (data.channels !== NUM_CHANNELS) { + throw new Error( + `Sarvam realtime STT only supports mono audio (${NUM_CHANNELS} channel), got ${data.channels} channels`, + ); + } + frames = audioStream.write( + data.data.buffer.slice( + data.data.byteOffset, + data.data.byteOffset + data.data.byteLength, + ) as ArrayBuffer, + ); + } + + this.#sendAudioFrames(ws, frames); + + if (isFlush) { + this.#audioDurationCollector.flush(); + if (this.#activeEndpointing === 'manual' && this.#manualSpeechStarted) { + this.#safeSendJson(ws, { event: 'speech_end' }); + this.#manualSpeechStarted = false; + this.#endManualUtterance(); + } + } + } + + // Drain any audio still buffered in the framer (endInput() without a trailing flush()). + if (!this.#sessionEnded && !signal.aborted && ws.readyState === WebSocket.OPEN) { + this.#sendAudioFrames(ws, audioStream.flush()); + if (this.#activeEndpointing === 'manual' && this.#manualSpeechStarted) { + this.#safeSendJson(ws, { event: 'speech_end' }); + this.#manualSpeechStarted = false; + this.#endManualUtterance(); + } + } + + this.#flushLocalUsageFallback(); + if (!this.#sessionEnded && !signal.aborted && ws.readyState === WebSocket.OPEN) { + this.#safeSendJson(ws, { event: 'end' }); + } + } + + #sendAudioFrames(ws: WebSocket, frames: AudioFrame[]): void { + for (const frame of frames) { + if (this.#activeEndpointing === 'manual' && !this.#manualSpeechStarted) { + this.#safeSendJson(ws, { event: 'speech_start' }); + this.#manualSpeechStarted = true; + this.#beginManualUtterance(); + } + + const duration = frame.samplesPerChannel / frame.sampleRate; + this.#audioDurationCollector.push(duration); + this.#audioPosition += duration; + this.#safeSendBinary(ws, encodePcmForWire(this.#opts.encoding, frame)); + } + } + + #sendPendingConfigUpdate(ws: WebSocket): void { + if (!this.#pendingConfigUpdate) return; + const payload = this.#pendingConfigUpdate; + this.#pendingConfigUpdate = null; + if ('endpointing' in payload) { + this.#endpointingUpdateSent = true; + this.#endpointingUpdateAcknowledged = false; + } + this.#safeSendJson(ws, payload); + } + + #safeSendJson(ws: WebSocket, payload: Record): void { + try { + ws.send(JSON.stringify(payload)); + } catch (error) { + this.#logger.debug({ error }, 'failed to send message to Sarvam realtime STT'); + } + } + + #safeSendBinary(ws: WebSocket, data: Buffer): void { + try { + ws.send(data); + } catch (error) { + this.#logger.debug({ error }, 'failed to send audio to Sarvam realtime STT'); + } + } + + // ------------------------------------------------------------------------- + // Server -> client (event loop) + // ------------------------------------------------------------------------- + + #processMessages(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + const cleanup = () => { + ws.off('message', onMessage); + ws.off('error', onError); + ws.off('close', onClose); + }; + + const onMessage = (raw: RawData, isBinary: boolean) => { + if (isBinary) return; // Sarvam realtime only sends JSON text frames + let parsed: RealtimeServerEvent; + try { + parsed = JSON.parse(raw.toString()) as RealtimeServerEvent; + } catch { + const text = raw.toString(); + if (looksLikeErrorText(text)) { + cleanup(); + reject(new APIStatusError({ message: `Sarvam realtime STT error: ${text}` })); + } + return; + } + + try { + this.#handleMessage(parsed); + } catch (error) { + cleanup(); + reject(error); + return; + } + + if (this.#sessionEnded) { + cleanup(); + resolve(); + } + }; + + const onError = (error: Error) => { + cleanup(); + reject( + new APIConnectionError({ + message: `Sarvam realtime STT WebSocket error: ${error.message}`, + }), + ); + }; + + const onClose = (code: number, reasonBuf: Buffer) => { + cleanup(); + const reason = reasonBuf.toString(); + if (this.#sessionEnded && (code === 1000 || code === 1001)) { + resolve(); + return; + } + if ((code === 1000 || code === 1001) && !looksLikeErrorText(reason)) { + resolve(); + return; + } + reject(this.#statusErrorFromClose(code, reason)); + }; + + ws.on('message', onMessage); + ws.on('error', onError); + ws.on('close', onClose); + }); + } + + #statusErrorFromClose(code: number, reason: string): APIStatusError { + const retryable = code === 1013; + let message = `Sarvam realtime STT WebSocket closed unexpectedly: ${reason}`; + if (code === 1003) { + message = 'Sarvam realtime STT authentication, quota, or rate limit error'; + } else if (code === 1008) { + message = 'Sarvam realtime STT session timed out or exceeded the maximum duration'; + } else if (code === 1013) { + message = 'Sarvam realtime STT backend temporarily unavailable'; + } else if (code === 4000) { + message = `Sarvam realtime STT rejected the session: ${reason}`; + } + return new APIStatusError({ + message, + options: { + statusCode: code, + requestId: this.#requestId || null, + body: { closeCode: code, closeReason: reason }, + retryable, + }, + }); + } + + // ------------------------------------------------------------------------- + // Event dispatch + // ------------------------------------------------------------------------- + + #handleMessage(data: RealtimeServerEvent): void { + this.#captureServerIds(data); + switch (data.event) { + case 'session.begin': + return; + case 'vad.speech_start': + this.#resetUtteranceState(); + this.#utteranceInProgress = true; + this.#put({ type: stt.SpeechEventType.START_OF_SPEECH, requestId: this.#requestId }); + return; + case 'vad.speech_end': + this.#handleSpeechEnd(); + return; + case 'transcript.partial': + this.#sendTranscriptEvent(stt.SpeechEventType.INTERIM_TRANSCRIPT, data); + return; + case 'transcript.final': + if (this.#activeEndpointing === 'vad') { + if (this.#isValidTranscript(data)) { + this.#pendingFinalData = data; + this.#finalReceivedForUtterance = true; + this.#tryCommitUtterance(); + } + } else if (this.#sendTranscriptEvent(stt.SpeechEventType.FINAL_TRANSCRIPT, data)) { + this.#finalReceivedForUtterance = true; + this.#completeUtterance(); + } + return; + case 'session.end': + this.#handleSessionEnd(data); + return; + case 'config.updated': + this.#handleConfigUpdated(data); + return; + case 'error': + this.#handleErrorEvent(data); + return; + case 'pong': + return; + default: + this.#logger.debug({ event: data.event }, 'unknown Sarvam realtime STT event'); + } + } + + #isValidTranscript(data: RealtimeServerEvent): boolean { + return typeof data.text === 'string' && data.text.trim().length > 0; + } + + #resetUtteranceState(): void { + this.#pendingFinalData = null; + this.#finalReceivedForUtterance = false; + this.#eosEmittedForUtterance = false; + this.#utteranceSpeechEndAudioPos = null; + this.#utteranceSpeechEndWall = null; + } + + #handleSpeechEnd(): void { + this.#utteranceSpeechEndAudioPos = this.#audioPosition; + this.#utteranceSpeechEndWall = Date.now(); + if (this.#activeEndpointing !== 'vad') { + this.#emitEndOfSpeech(); + } else if (!this.#eosEmittedForUtterance) { + // Commit a pending final first so consumers never see END_OF_SPEECH before it. + if (this.#finalReceivedForUtterance) { + this.#tryCommitUtterance(); + } else { + this.#emitEndOfSpeech(); + } + } + this.#completeUtterance(); + } + + #tryCommitUtterance(): void { + if (!this.#pendingFinalData || this.#utteranceSpeechEndAudioPos === null) return; + const committed = this.#pendingFinalData; + if (this.#sendTranscriptEvent(stt.SpeechEventType.FINAL_TRANSCRIPT, committed)) { + if (!this.#eosEmittedForUtterance) this.#emitEndOfSpeech(); + this.#pendingFinalData = null; + this.#completeUtterance(); + } + } + + #emitEndOfSpeech(): void { + if (this.#eosEmittedForUtterance) return; + this.#eosEmittedForUtterance = true; + this.#put({ + type: stt.SpeechEventType.END_OF_SPEECH, + requestId: this.#requestId, + speechEndTime: this.#utteranceSpeechEndWall ?? Date.now(), + }); + } + + #completeUtterance(): void { + this.#utteranceInProgress = false; + this.#applyPendingEndpointing(); + } + + #applyPendingEndpointing(): void { + if (this.#pendingEndpointing && this.#endpointingUpdateAcknowledged) { + this.#activeEndpointing = this.#pendingEndpointing; + this.#pendingEndpointing = undefined; + } + } + + #beginManualUtterance(): void { + this.#resetUtteranceState(); + this.#utteranceInProgress = true; + this.#put({ type: stt.SpeechEventType.START_OF_SPEECH, requestId: this.#requestId }); + } + + #endManualUtterance(): void { + this.#utteranceSpeechEndAudioPos = this.#audioPosition; + this.#utteranceSpeechEndWall = Date.now(); + this.#emitEndOfSpeech(); + this.#completeUtterance(); + } + + #sendTranscriptEvent(type: stt.SpeechEventType, data: RealtimeServerEvent): boolean { + const text = data.text; + if (typeof text !== 'string' || !text.trim()) return false; + + const language = normalizeLanguage(data.language || this.#opts.language); + let confidence = data.confidence; + if (typeof confidence !== 'number' || Number.isNaN(confidence)) confidence = 1; + + const metadata: Record = {}; + if (typeof data.utterance_idx === 'number') metadata.utteranceIdx = data.utterance_idx; + if (typeof data.language_confidence === 'number') { + metadata.languageConfidence = data.language_confidence; + } + if (type === stt.SpeechEventType.FINAL_TRANSCRIPT && this.#utteranceSpeechEndWall !== null) { + metadata.speechEndWallTime = this.#utteranceSpeechEndWall; + } + + let startTime = 0; + let endTime = 0; + if (type === stt.SpeechEventType.FINAL_TRANSCRIPT) { + if (typeof data.start_s === 'number') startTime = Math.max(data.start_s, 0); + if (typeof data.end_s === 'number') endTime = Math.max(data.end_s, 0); + if (endTime === 0) { + endTime = + this.#utteranceSpeechEndAudioPos ?? (this.#audioPosition > 0 ? this.#audioPosition : 0); + } + } + + this.#put({ + type, + requestId: this.#requestId, + alternatives: [ + { + language, + text, + startTime, + endTime, + confidence, + metadata: Object.keys(metadata).length > 0 ? metadata : undefined, + }, + ], + }); + return true; + } + + #handleSessionEnd(data: RealtimeServerEvent): void { + this.#captureServerIds(data); + this.#flushTerminalUtterance(); + + const audioDuration = data.audio_duration_s; + if (typeof audioDuration === 'number' && !this.#serverAudioDurationReported) { + this.#audioDurationCollector.flush(); + const serverAudioDuration = Math.max(audioDuration, 0); + const delta = Math.max(serverAudioDuration - this.#totalReportedAudioDuration, 0); + if (delta > 0) this.#emitUsage(delta); + this.#serverAudioDurationReported = true; + } else { + this.#flushLocalUsageFallback(); + } + this.#sessionEnded = true; + } + + #flushTerminalUtterance(): void { + if (this.#pendingFinalData && this.#utteranceSpeechEndAudioPos === null) { + this.#utteranceSpeechEndAudioPos = this.#audioPosition; + this.#utteranceSpeechEndWall = Date.now(); + } + this.#tryCommitUtterance(); + } + + #flushLocalUsageFallback(): void { + this.#audioDurationCollector.flush(); + } + + #emitUsage(duration: number): void { + this.#totalReportedAudioDuration += duration; + this.#put({ + type: stt.SpeechEventType.RECOGNITION_USAGE, + requestId: this.#requestId, + recognitionUsage: { audioDuration: duration }, + }); + } + + #handleConfigUpdated(data: RealtimeServerEvent): void { + const applied = Array.isArray(data.applied) ? data.applied.map(String) : []; + const appliedEndpointing = applied.some((entry) => entry.startsWith('endpointing')); + if (appliedEndpointing && this.#endpointingUpdateSent) { + this.#endpointingUpdateAcknowledged = true; + this.#endpointingUpdateSent = false; + if (!this.#utteranceInProgress) this.#applyPendingEndpointing(); + } + } + + #handleErrorEvent(data: RealtimeServerEvent): void { + const code = data.code ?? 'unknown'; + if (!data.is_fatal) { + this.#logger.warn( + { code, 'lk.pii.message': data.message }, + 'non-fatal Sarvam realtime STT error', + ); + return; + } + const statusCode = typeof data.status_code === 'number' ? data.status_code : -1; + this.#logger.error( + { code, statusCode, 'lk.pii.message': data.message }, + 'fatal Sarvam realtime STT error', + ); + // The raw provider message/payload may carry account or transcript-adjacent details, so it's + // only logged (tagged lk.pii.* above) — the thrown error's own message and body stay generic. + throw new APIStatusError({ + message: `Sarvam realtime STT error (${code})`, + options: { + statusCode, + requestId: this.#requestId || null, + retryable: code === 'model_unavailable', + }, + }); + } + + #captureServerIds(data: RealtimeServerEvent): void { + if (typeof data.session_id === 'string' && data.session_id) { + this.#sessionId = data.session_id; + } + if (!this.#requestId) { + const requestId = data.request_id ?? data.data?.request_id ?? data.metadata?.request_id; + if (typeof requestId === 'string' && requestId) this.#requestId = requestId; + } + void this.#sessionId; // tracked for diagnostics; not currently surfaced on SpeechEvent + } + + #put(event: stt.SpeechEvent): void { + if (!this.queue.closed) this.queue.put(event); + } +}