diff --git a/.changeset/green-gnani-speak.md b/.changeset/green-gnani-speak.md new file mode 100644 index 000000000..1e9c61bf3 --- /dev/null +++ b/.changeset/green-gnani-speak.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents-plugin-gnani': patch +--- + +Add the Gnani Vachana STT/TTS provider integration. diff --git a/agents/src/tts/tts.ts b/agents/src/tts/tts.ts index 3528e9c5c..fa84fafc1 100644 --- a/agents/src/tts/tts.ts +++ b/agents/src/tts/tts.ts @@ -197,6 +197,8 @@ export abstract class SynthesizeStream abstract label: string; #tts: TTS; + readonly #metricsModelProvider: string; + readonly #metricsModelName: string; #metricsPendingTexts: string[] = []; #metricsText = ''; #monitorMetricsTask?: Promise; @@ -213,6 +215,8 @@ export abstract class SynthesizeStream constructor(tts: TTS, connOptions: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS) { this.#tts = tts; + this.#metricsModelProvider = tts.provider; + this.#metricsModelName = tts.model; this.connOptions = connOptions; this.deferredInputStream = new DeferredReadableStream(); this.pumpInput(); @@ -447,8 +451,8 @@ export abstract class SynthesizeStream outputTokens: this.#outputTokens, streamed: true, metadata: { - modelProvider: this.#tts.provider, - modelName: this.#tts.model, + modelProvider: this.#metricsModelProvider, + modelName: this.#metricsModelName, }, }; if (this.#ttsRequestSpan) { @@ -595,6 +599,8 @@ export abstract class ChunkedStream implements AsyncIterableIterator [Gnani.ai](https://gnani.ai) featuring **Prisma** (STT) and **Timbre** (TTS) models, supporting 10+ Indian languages with real-time streaming, multilingual transcription, and code-switching capabilities. + +## Installation + +```bash +pnpm add @livekit/agents-plugin-gnani +``` + +## Prerequisites + +You need a Gnani API key. Email **[speechstack@gnani.ai](mailto:speechstack@gnani.ai)** to get started; all new accounts receive free credits, no credit card required. + +### Authentication + +All APIs require a single API key: no `organizationId` or `userId` needed. + +**Option 1: Environment variable (recommended):** + +```bash +export GNANI_API_KEY="your-api-key" +``` + +**Option 2: Constructor argument:** + +```ts +const stt = new gnani.STT({ apiKey: 'your-api-key', language: 'hi-IN' }); +const tts = new gnani.TTS({ apiKey: 'your-api-key' }); +``` + +> **Migration note:** If upgrading from an earlier version, remove any `organizationId` and `userId` parameters; they are no longer accepted. + +## Quick Start + +### Speech-to-Text (REST + Streaming) + +```ts +import * as gnani from '@livekit/agents-plugin-gnani'; + +const stt = new gnani.STT({ language: 'hi-IN' }); + +// REST STT (file-based transcription) +const speechEvent = await stt.recognize(audioBuffer); + +// Streaming STT (real-time WebSocket) +const speechStream = stt.stream(); +``` + +### Text-to-Speech + +```ts +import * as gnani from '@livekit/agents-plugin-gnani'; + +// REST (default): single-request batch synthesis +const ttsRest = new gnani.TTS({ voice: 'Karan' }); + +// SSE: chunked synthesis via Server-Sent Events (lower latency) +const ttsSse = new gnani.TTS({ voice: 'Karan', synthesizeMethod: 'sse' }); + +// WebSocket: chunked synthesis over WS (lowest latency) +const ttsWs = new gnani.TTS({ voice: 'Karan', synthesizeMethod: 'websocket' }); +``` + +All three modes work with the standard LiveKit voice agent pipeline. The `synthesizeMethod` controls which transport `synthesize()` uses (REST, SSE, or WebSocket). The `stream()` method always uses WebSocket regardless of this setting. + +## Full Constructor Reference + +### STT: All Parameters + +```ts +const stt = new gnani.STT({ + language: 'en-IN', // Default: 'en-IN' + sampleRate: 16000, // Default: 16000 (also: 8000) + format: 'verbatim', // Default: 'verbatim' (also: 'transcribe') + preferredLanguage: undefined, // Default: undefined + itnNativeNumerals: false, // Default: false + apiKey: undefined, // Default: reads GNANI_API_KEY env var + baseURL: 'https://api.vachana.ai', // Default +}); +``` + +### TTS: All Parameters + +```ts +const tts = new gnani.TTS({ + voice: 'Karan', // Default: 'Karan' (also: Simran, Nara, Riya, Viraj, Raju) + model: 'vachana-voice-v3', // Default: 'vachana-voice-v3' + sampleRate: 16000, // Default: 16000 (also: 8000, 22050, 44100) + encoding: 'linear_pcm', // Only supported encoding + container: 'wav', // Default: 'wav' (also: 'raw') + numChannels: 1, // Default: 1 + bitrate: undefined, // Default: undefined (also: '96k', '128k', '192k') + synthesizeMethod: 'rest', // Default: 'rest' (also: 'sse', 'websocket') + apiKey: undefined, // Default: reads GNANI_API_KEY env var + baseURL: 'https://api.vachana.ai', // Default +}); +``` + +## Features + +### STT (Prisma) + +- **REST recognition**: REST API (`POST /stt/v3`) for file-based transcription +- **Real-time streaming**: WebSocket API (`wss://api.vachana.ai/stt/v3/stream`) for live audio transcription with VAD +- **10+ Indian languages**: see [supported language codes](https://docs.gnani.ai/api/STT/stt-websocket#supported-languages) +- **Code-switching**: supports multilingual and code-mixed audio +- **Sample rates**: 8 kHz and 16 kHz +- **ITN support**: Inverse Text Normalization via `format: 'transcribe'` + +#### Streaming PCM Specification + +All streaming audio must be sent as **raw PCM binary frames**: no container format (WAV, MP3) mid-stream. + +| Property | 16 kHz | 8 kHz | +| ------------------- | --------------------------------------- | --------------------------------------- | +| Encoding | PCM signed 16-bit little-endian | PCM signed 16-bit little-endian | +| Sample Rate | 16,000 Hz | 8,000 Hz | +| Channels | 1 (mono) | 1 (mono) | +| Samples per chunk | 512 | 512 | +| **Bytes per frame** | **1,024 bytes** (512 samples x 2 bytes) | **1,024 bytes** (512 samples x 2 bytes) | +| Frame duration | 32 ms | 64 ms | + +Frames must be sent at **real-time cadence**. See **[STT Realtime: PCM Specification](https://docs.gnani.ai/api/STT/stt-websocket#pcm-specification)** for full details. + +### TTS (Timbre) + +- **REST synthesis**: single-request batch audio generation (`synthesizeMethod: 'rest'`) +- **SSE streaming**: lower-latency chunked synthesis via Server-Sent Events (`synthesizeMethod: 'sse'`) +- **WebSocket synthesis**: lowest-latency synthesis via `synthesizeMethod: 'websocket'` or the `stream()` method +- **6 voices**: Karan, Simran, Nara, Riya, Viraj, Raju +- **Decoded output**: linear PCM in raw or WAV containers; encoded formats aren't supported +- **Configurable output**: sample rate (8000-44100), raw or WAV container, and optional bitrate +- **Runtime updates**: change voice or model via `updateOptions()` + +## Supported Languages + +### STT Languages (Prisma) + +Prisma uses BCP-47 locale codes (e.g. `hi-IN`). Supported: + +- **[STT REST: Supported Languages](https://docs.gnani.ai/api/STT/speech-to-text#supported-languages)** +- **[STT Realtime: Supported Languages](https://docs.gnani.ai/api/STT/stt-websocket#supported-languages)** + +### TTS Languages (Timbre) + +For the full list of supported languages, see **[TTS: Supported Languages](https://docs.gnani.ai/api/TTS/tts-inference#supported-languages)**. + +## Available Voices + +| Voice | ID | Gender | Description | +| ------ | -------- | ------ | ------------------------ | +| Karan | `Karan` | Male | Bold, Trustworthy | +| Simran | `Simran` | Female | Confident, Bright | +| Nara | `Nara` | Female | Gentle, Expressive | +| Riya | `Riya` | Female | Cheerful, Energetic | +| Viraj | `Viraj` | Male | Commanding, Dynamic | +| Raju | `Raju` | Male | Grounded, Conversational | + +## Architecture + +This plugin directly implements the Gnani REST, SSE, and WebSocket APIs using `fetch` and `ws`, adapting them into LiveKit's `stt.STT` and `tts.TTS` base classes. It uses the **Prisma** model for speech-to-text and the **Timbre** model for text-to-speech. No external SDK is required; all connection logic, authentication, and audio format handling is self-contained. Authentication uses a single `apiKey` passed via the `X-API-Key-ID` header. + +## Documentation + +- [Gnani API Docs](https://docs.gnani.ai/) +- [LiveKit Agents Docs](https://docs.livekit.io/agents/) +- [Gnani STT Plugin Guide](https://docs.livekit.io/agents/integrations/stt/gnani/) +- [Gnani TTS Plugin Guide](https://docs.livekit.io/agents/integrations/tts/gnani/) + +## License + +Apache-2.0 diff --git a/plugins/gnani/api-extractor.json b/plugins/gnani/api-extractor.json new file mode 100644 index 000000000..32c90f0fa --- /dev/null +++ b/plugins/gnani/api-extractor.json @@ -0,0 +1,5 @@ +{ + "$schema": "https://developer.microsoft.com/json-schemas/api-extractor/v7/api-extractor.schema.json", + "extends": "../../api-extractor-shared.json", + "mainEntryPointFilePath": "./dist/index.d.ts" +} diff --git a/plugins/gnani/etc/agents-plugin-gnani.api.md b/plugins/gnani/etc/agents-plugin-gnani.api.md new file mode 100644 index 000000000..07c4619e0 --- /dev/null +++ b/plugins/gnani/etc/agents-plugin-gnani.api.md @@ -0,0 +1,213 @@ +## API Report File for "@livekit/agents-plugin-gnani" + +> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). + +```ts + +import { APIConnectOptions } from '@livekit/agents'; +import { AudioBuffer as AudioBuffer_2 } from '@livekit/agents'; +import { stt } from '@livekit/agents'; +import { tts } from '@livekit/agents'; + +// @public (undocumented) +export type ChunkedStream = RESTChunkedStream | SSEChunkedStream | WebSocketChunkedStream; + +// @public (undocumented) +export type GnaniSTTFormat = 'verbatim' | 'transcribe'; + +// @public (undocumented) +export type GnaniSTTLanguages = 'bn-IN' | 'en-IN' | 'gu-IN' | 'hi-IN' | 'kn-IN' | 'ml-IN' | 'mr-IN' | 'pa-IN' | 'ta-IN' | 'te-IN' | 'en-IN,hi-IN'; + +// @public (undocumented) +export type GnaniTTSBitrates = '96k' | '128k' | '192k'; + +// @public (undocumented) +export type GnaniTTSContainers = 'raw' | 'wav'; + +// @public (undocumented) +export type GnaniTTSEncodings = 'linear_pcm'; + +// @public (undocumented) +export type GnaniTTSSynthesizeMethod = 'rest' | 'sse' | 'websocket'; + +// @public (undocumented) +export type GnaniTTSVoices = 'Karan' | 'Simran' | 'Nara' | 'Riya' | 'Viraj' | 'Raju'; + +// @public (undocumented) +export interface ResolvedSTTOptions { + // (undocumented) + apiKey: string; + // (undocumented) + baseURL: string; + // (undocumented) + format: GnaniSTTFormat; + // (undocumented) + itnNativeNumerals: boolean; + // (undocumented) + language: string; + // (undocumented) + preferredLanguage?: string; + // (undocumented) + sampleRate: number; +} + +// @public (undocumented) +export interface ResolvedTTSOptions { + // (undocumented) + apiKey: string; + // (undocumented) + baseURL: string; + // (undocumented) + bitrate?: string; + // (undocumented) + container: GnaniTTSContainers; + // (undocumented) + encoding: GnaniTTSEncodings; + // (undocumented) + model: string; + // (undocumented) + numChannels: number; + // (undocumented) + sampleRate: number; + // (undocumented) + sampleWidth: number; + // (undocumented) + synthesizeMethod: GnaniTTSSynthesizeMethod; + // (undocumented) + voice: string; +} + +// @public (undocumented) +export class RESTChunkedStream extends tts.ChunkedStream { + constructor(ttsInstance: TTS, text: string, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions, abortSignal?: AbortSignal); + // (undocumented) + label: string; + // (undocumented) + protected run(): Promise; +} + +// @public (undocumented) +export const SAMPLE_RATE_16K = 16000; + +// @public (undocumented) +export const SAMPLE_RATE_8K = 8000; + +// @public (undocumented) +export class SpeechStream extends stt.SpeechStream { + constructor(sttInstance: STT, opts: ResolvedSTTOptions, connOptions?: APIConnectOptions); + // (undocumented) + buildWsUrl(): string; + // (undocumented) + label: string; + // (undocumented) + _opts: ResolvedSTTOptions; + // (undocumented) + protected run(): Promise; +} + +// @public (undocumented) +export class SSEChunkedStream extends tts.ChunkedStream { + constructor(ttsInstance: TTS, text: string, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions, abortSignal?: AbortSignal); + // (undocumented) + label: string; + // (undocumented) + protected run(): Promise; +} + +// @public (undocumented) +export class STT extends stt.STT { + constructor(opts?: STTOptions & Record); + // (undocumented) + label: string; + // (undocumented) + get model(): string; + // (undocumented) + _opts: ResolvedSTTOptions; + // (undocumented) + get provider(): string; + // (undocumented) + protected _recognize(buffer: AudioBuffer_2, abortSignal?: AbortSignal): Promise; + // (undocumented) + stream(options?: { + language?: string; + connOptions?: APIConnectOptions; + }): SpeechStream; +} + +// @public (undocumented) +export interface STTOptions { + apiKey?: string; + baseURL?: string; + format?: GnaniSTTFormat; + itnNativeNumerals?: boolean; + language?: GnaniSTTLanguages | string; + preferredLanguage?: string; + sampleRate?: typeof SAMPLE_RATE_8K | typeof SAMPLE_RATE_16K | number; +} + +// @public (undocumented) +export class SynthesizeStream extends tts.SynthesizeStream { + constructor(ttsInstance: TTS, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions); + // (undocumented) + buildWsUrl(): string; + // (undocumented) + label: string; + // (undocumented) + protected run(): Promise; +} + +// @public (undocumented) +export class TTS extends tts.TTS { + constructor(opts?: TTSOptions & Record); + // (undocumented) + label: string; + // (undocumented) + get model(): string; + // (undocumented) + _opts: ResolvedTTSOptions; + // (undocumented) + get provider(): string; + // (undocumented) + stream(options?: { + connOptions?: APIConnectOptions; + }): SynthesizeStream; + // (undocumented) + synthesize(text: string, connOptions?: APIConnectOptions, abortSignal?: AbortSignal): ChunkedStream; + // (undocumented) + updateOptions(opts: TTSUpdateOptions & Record): void; +} + +// @public (undocumented) +export interface TTSOptions { + apiKey?: string; + baseURL?: string; + bitrate?: GnaniTTSBitrates | string; + container?: GnaniTTSContainers; + encoding?: GnaniTTSEncodings; + model?: string; + numChannels?: number; + sampleRate?: number; + synthesizeMethod?: GnaniTTSSynthesizeMethod; + voice?: GnaniTTSVoices | string; +} + +// @public (undocumented) +export interface TTSUpdateOptions { + model?: string; + voice?: GnaniTTSVoices | string; +} + +// @public (undocumented) +export class WebSocketChunkedStream extends tts.ChunkedStream { + constructor(ttsInstance: TTS, text: string, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions, abortSignal?: AbortSignal); + // (undocumented) + buildWsUrl(): string; + // (undocumented) + label: string; + // (undocumented) + protected run(): Promise; +} + +// (No @packageDocumentation comment for this package) + +``` diff --git a/plugins/gnani/package.json b/plugins/gnani/package.json new file mode 100644 index 000000000..68ea44a14 --- /dev/null +++ b/plugins/gnani/package.json @@ -0,0 +1,51 @@ +{ + "name": "@livekit/agents-plugin-gnani", + "version": "1.5.0", + "description": "Gnani Vachana plugin for LiveKit Node Agents", + "main": "dist/index.js", + "require": "dist/index.cjs", + "types": "dist/index.d.ts", + "exports": { + "import": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + }, + "require": { + "types": "./dist/index.d.cts", + "default": "./dist/index.cjs" + } + }, + "author": "LiveKit", + "type": "module", + "repository": "git@github.com:livekit/agents-js.git", + "license": "Apache-2.0", + "files": [ + "dist", + "src", + "README.md" + ], + "scripts": { + "build": "tsup --onSuccess \"pnpm build:types\"", + "build:types": "tsc --declaration --emitDeclarationOnly && node ../../scripts/copyDeclarationOutput.js", + "clean": "rm -rf dist", + "clean:build": "pnpm clean && pnpm build", + "lint": "eslint -f unix \"src/**/*.{ts,js}\"", + "api:check": "api-extractor run --typescript-compiler-folder ../../node_modules/typescript", + "api:update": "api-extractor run --local --typescript-compiler-folder ../../node_modules/typescript --verbose" + }, + "devDependencies": { + "@livekit/agents": "workspace:*", + "@livekit/rtc-node": "catalog:", + "@microsoft/api-extractor": "^7.35.0", + "@types/ws": "catalog:", + "tsup": "^8.3.5", + "typescript": "^5.0.0" + }, + "dependencies": { + "ws": "catalog:" + }, + "peerDependencies": { + "@livekit/agents": "workspace:*", + "@livekit/rtc-node": "catalog:" + } +} diff --git a/plugins/gnani/src/index.ts b/plugins/gnani/src/index.ts new file mode 100644 index 000000000..7e30b620f --- /dev/null +++ b/plugins/gnani/src/index.ts @@ -0,0 +1,43 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { Plugin } from '@livekit/agents'; + +export { + SAMPLE_RATE_8K, + SAMPLE_RATE_16K, + SpeechStream, + STT, + type GnaniSTTFormat, + type GnaniSTTLanguages, + type ResolvedSTTOptions, + type STTOptions, +} from './stt.js'; +export { + RESTChunkedStream, + SSEChunkedStream, + SynthesizeStream, + TTS, + WebSocketChunkedStream, + type ChunkedStream, + type GnaniTTSBitrates, + type GnaniTTSContainers, + type GnaniTTSEncodings, + type GnaniTTSSynthesizeMethod, + type GnaniTTSVoices, + type ResolvedTTSOptions, + type TTSOptions, + type TTSUpdateOptions, +} from './tts.js'; + +class GnaniPlugin extends Plugin { + constructor() { + super({ + title: 'gnani', + version: __PACKAGE_VERSION__, + package: __PACKAGE_NAME__, + }); + } +} + +Plugin.registerPlugin(new GnaniPlugin()); diff --git a/plugins/gnani/src/stt.test.ts b/plugins/gnani/src/stt.test.ts new file mode 100644 index 000000000..343d1137d --- /dev/null +++ b/plugins/gnani/src/stt.test.ts @@ -0,0 +1,568 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { STTMetrics } from '@livekit/agents'; +import { AudioFrame } from '@livekit/rtc-node'; +import { EventEmitter } from 'node:events'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { STT, SpeechStream } from './stt.js'; + +interface MockWebSocket extends EventEmitter { + options?: { handshakeTimeout?: number }; + sent: unknown[]; + closed: boolean; + terminated: boolean; +} + +const wsState = vi.hoisted(() => ({ + instances: [] as MockWebSocket[], + autoOpen: true, +})); + +vi.mock('ws', () => { + return { + WebSocket: class MockWebSocket extends EventEmitter { + static OPEN = 1; + readyState = 1; + sent: unknown[] = []; + closed = false; + terminated = false; + options?: { handshakeTimeout?: number }; + + constructor(_url: string, options?: { handshakeTimeout?: number }) { + super(); + this.options = options; + wsState.instances.push(this); + if (wsState.autoOpen) queueMicrotask(() => this.emit('open')); + } + + send(data: unknown) { + this.sent.push(data); + } + + close() { + this.closed = true; + this.emit('close', 1000); + } + + terminate() { + this.terminated = true; + this.closed = true; + this.emit('close', 1006); + } + }, + }; +}); + +describe('Gnani STT', () => { + beforeEach(() => { + wsState.instances.length = 0; + wsState.autoOpen = true; + }); + + afterEach(() => { + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); + vi.useRealTimers(); + }); + + it('requires an API key', () => { + vi.stubEnv('GNANI_API_KEY', ''); + expect(() => new STT({ apiKey: undefined })).toThrow(/API key/i); + }); + + it('accepts apiKey directly', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.apiKey).toBe('test-key'); + }); + + it('rejects unknown constructor options like Python', () => { + expect(() => new STT({ apiKey: 'test-key', unknownOption: true })).toThrow( + /unexpected option.*unknownOption/i, + ); + }); + + it('accepts apiKey from env', () => { + vi.stubEnv('GNANI_API_KEY', 'env-key'); + const stt = new STT(); + expect(stt._opts.apiKey).toBe('env-key'); + }); + + it('defaults to en-IN', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.language).toBe('en-IN'); + }); + + it('accepts custom language', () => { + const stt = new STT({ apiKey: 'test-key', language: 'hi-IN' }); + expect(stt._opts.language).toBe('hi-IN'); + }); + + it('defaults to 16000 Hz sample rate', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.sampleRate).toBe(16000); + }); + + it('accepts 8000 Hz sample rate', () => { + const stt = new STT({ apiKey: 'test-key', sampleRate: 8000 }); + expect(stt._opts.sampleRate).toBe(8000); + }); + + it('rejects invalid sample rates', () => { + expect(() => new STT({ apiKey: 'test-key', sampleRate: 44100 })).toThrow(/sampleRate/i); + }); + + it('reports streaming=true and interimResults=false', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt.capabilities.streaming).toBe(true); + expect(stt.capabilities.interimResults).toBe(false); + }); + + it('returns model and provider properties', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt.model).toBe('vachana-stt-v3'); + expect(stt.provider).toBe('Gnani'); + }); + + it('defaults to Vachana API base URL', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.baseURL).toBe('https://api.vachana.ai'); + }); + + it('accepts custom base URL', () => { + const stt = new STT({ apiKey: 'test-key', baseURL: 'https://custom.api.com' }); + expect(stt._opts.baseURL).toBe('https://custom.api.com'); + }); + + it('uses only apiKey for authentication', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect('organizationId' in stt._opts).toBe(false); + expect('userId' in stt._opts).toBe(false); + }); + + it('builds wss URL from https base', () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = new SpeechStream(stt, { + apiKey: 'test-key', + language: 'en-IN', + sampleRate: 16000, + baseURL: 'https://api.vachana.ai', + format: 'verbatim', + itnNativeNumerals: false, + }); + stream.close(); + expect(stream.buildWsUrl()).toBe('wss://api.vachana.ai/stt/v3/stream'); + }); + + it('builds ws URL from http base', () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = new SpeechStream(stt, { + apiKey: 'test-key', + language: 'en-IN', + sampleRate: 16000, + baseURL: 'http://localhost:8080', + format: 'verbatim', + itnNativeNumerals: false, + }); + stream.close(); + expect(stream.buildWsUrl()).toBe('ws://localhost:8080/stt/v3/stream'); + }); + + it('defaults format to verbatim', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.format).toBe('verbatim'); + }); + + it('accepts transcribe format for ITN', () => { + const stt = new STT({ apiKey: 'test-key', format: 'transcribe' }); + expect(stt._opts.format).toBe('transcribe'); + }); + + it('defaults preferredLanguage to undefined', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.preferredLanguage).toBeUndefined(); + }); + + it('accepts custom preferredLanguage', () => { + const stt = new STT({ apiKey: 'test-key', preferredLanguage: 'hi-IN' }); + expect(stt._opts.preferredLanguage).toBe('hi-IN'); + }); + + it('defaults itnNativeNumerals to false', () => { + const stt = new STT({ apiKey: 'test-key' }); + expect(stt._opts.itnNativeNumerals).toBe(false); + }); + + it('accepts itnNativeNumerals=true', () => { + const stt = new STT({ apiKey: 'test-key', format: 'transcribe', itnNativeNumerals: true }); + expect(stt._opts.itnNativeNumerals).toBe(true); + }); + + it('sends Python-equivalent REST form fields and reports usage metrics', async () => { + let request: RequestInit | undefined; + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + request = init; + return Response.json({ transcript: 'namaste', request_id: 'request-1' }); + }), + ); + const stt = new STT({ + apiKey: 'test-key', + language: 'hi-IN', + format: 'transcribe', + preferredLanguage: 'hi-IN', + itnNativeNumerals: true, + }); + const metricsPromise = new Promise((resolve) => { + stt.once('metrics_collected', resolve); + }); + const frame = AudioFrame.create(16000, 1, 1600); + + const event = await stt.recognize(frame); + const metrics = await metricsPromise; + const form = request?.body; + + expect(form).toBeInstanceOf(FormData); + if (!(form instanceof FormData)) throw new Error('expected FormData request body'); + expect(form.get('language_code')).toBe('hi-IN'); + expect(form.get('format')).toBe('transcribe'); + expect(form.get('preferred_language')).toBe('hi-IN'); + expect(form.get('itn_native_numerals')).toBe('true'); + expect(form.get('audio_file')).toBeInstanceOf(Blob); + expect(event.requestId).toBe('request-1'); + expect(metrics).toMatchObject({ + type: 'stt_metrics', + requestId: 'request-1', + audioDurationMs: 100, + streamed: false, + metadata: { modelProvider: 'Gnani', modelName: 'vachana-stt-v3' }, + }); + }); + + it('preserves caller cancellation while a REST request is pending', async () => { + let observeFetch = () => {}; + const fetchStarted = new Promise((resolve) => { + observeFetch = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn((_url: string | URL | Request, init?: RequestInit) => { + observeFetch(); + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + 'abort', + () => reject(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + }); + }), + ); + const stt = new STT({ apiKey: 'test-key' }); + const errors = vi.fn(); + stt.on('error', errors); + const controller = new AbortController(); + const recognition = stt.recognize(AudioFrame.create(16000, 1, 1600), controller.signal); + await fetchStarted; + + controller.abort(); + + await expect(recognition).rejects.toMatchObject({ name: 'AbortError' }); + expect(errors).not.toHaveBeenCalled(); + }); + + it('preserves caller cancellation while a non-2xx body read is pending', async () => { + let observeBodyRead = () => {}; + const bodyRead = new Promise((resolve) => { + observeBodyRead = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + let bodyController: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + pull() { + observeBodyRead(); + }, + }); + init?.signal?.addEventListener( + 'abort', + () => bodyController?.error(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + return new Response(body, { status: 429, statusText: 'Too Many Requests' }); + }), + ); + const stt = new STT({ apiKey: 'test-key' }); + const errors = vi.fn(); + stt.on('error', errors); + const controller = new AbortController(); + const recognition = stt.recognize(AudioFrame.create(16000, 1, 1600), controller.signal); + await bodyRead; + + controller.abort(); + + await expect(recognition).rejects.toMatchObject({ name: 'AbortError' }); + expect(errors).not.toHaveBeenCalled(); + }); + + it('maps timeout during a non-2xx body read to APITimeoutError', async () => { + const timeoutController = new AbortController(); + const timeoutSpy = vi.spyOn(AbortSignal, 'timeout').mockReturnValue(timeoutController.signal); + let observeBodyRead = () => {}; + const bodyRead = new Promise((resolve) => { + observeBodyRead = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + let bodyController: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + pull() { + observeBodyRead(); + }, + }); + init?.signal?.addEventListener( + 'abort', + () => bodyController?.error(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + return new Response(body, { status: 503, statusText: 'Service Unavailable' }); + }), + ); + const stt = new STT({ apiKey: 'test-key' }); + const recognition = stt.recognize(AudioFrame.create(16000, 1, 1600)); + await bodyRead; + + timeoutController.abort(); + + await expect(recognition).rejects.toMatchObject({ name: 'APITimeoutError' }); + timeoutSpy.mockRestore(); + }); + + it('preserves non-2xx provider status and body details', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => { + return new Response('invalid language configuration', { + status: 422, + statusText: 'Unprocessable Entity', + }); + }), + ); + const stt = new STT({ apiKey: 'test-key' }); + + await expect(stt.recognize(AudioFrame.create(16000, 1, 1600))).rejects.toMatchObject({ + name: 'APIStatusError', + statusCode: 422, + message: expect.stringContaining('invalid language configuration'), + }); + }); + + it('maps a non-2xx body read failure to APIConnectionError', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => { + const body = new ReadableStream({ + pull(controller) { + controller.error(new Error('body disconnected')); + }, + }); + return new Response(body, { status: 500, statusText: 'Internal Server Error' }); + }), + ); + const stt = new STT({ apiKey: 'test-key' }); + + await expect(stt.recognize(AudioFrame.create(16000, 1, 1600))).rejects.toMatchObject({ + name: 'APIConnectionError', + message: expect.stringContaining('body disconnected'), + }); + }); + + it('warns about deprecated auth kwargs without raising', () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const stt = new STT({ apiKey: 'test-key', organizationId: 'old', userId: 'old' }); + expect(stt._opts.apiKey).toBe('test-key'); + warn.mockRestore(); + }); + + it('stream() returns a SpeechStream instance', () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream(); + stream.close(); + expect(stream).toBeInstanceOf(SpeechStream); + }); + + it('stream() uses the configured language', () => { + const stt = new STT({ apiKey: 'test-key', language: 'hi-IN' }); + const stream = stt.stream(); + stream.close(); + expect(stream._opts.language).toBe('hi-IN'); + }); + + it('parses WebSocket text frames delivered as non-binary Buffers', async () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 100 }, + }); + await vi.waitFor(() => expect(wsState.instances.length).toBeGreaterThan(0)); + const ws = wsState.instances.at(-1)!; + + ws.emit( + 'message', + Buffer.from(JSON.stringify({ type: 'transcript', text: 'namaste', segment_id: 'seg-1' })), + false, + ); + + const result = await Promise.race([ + stream.next(), + new Promise((_, reject) => + setTimeout(() => reject(new Error('text Buffer was not parsed')), 50), + ), + ]); + stream.close(); + expect(result.value?.alternatives[0]?.text).toBe('namaste'); + }); + + it('drains final responses for one second after audio input ends', async () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 1500 }, + }); + await vi.waitFor(() => expect(wsState.instances.length).toBeGreaterThan(0)); + const ws = wsState.instances.at(-1)!; + + stream.pushFrame(AudioFrame.create(16000, 1, 160)); + stream.endInput(); + setTimeout(() => { + ws.emit( + 'message', + Buffer.from(JSON.stringify({ type: 'transcript', text: 'final', segment_id: 'seg-2' })), + false, + ); + }, 25); + + const result = await Promise.race([ + stream.next(), + new Promise((_, reject) => + setTimeout(() => reject(new Error('final response was not drained')), 100), + ), + ]); + stream.close(); + expect(result.value?.alternatives[0]?.text).toBe('final'); + }); + + it('uses connOptions timeout for WebSocket connection and receive', async () => { + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 200 }, + }); + await vi.waitFor(() => expect(wsState.instances.length).toBeGreaterThan(0)); + const ws = wsState.instances.at(-1)!; + await vi.waitFor(() => expect(ws.listenerCount('message')).toBeGreaterThan(0)); + + expect(ws.options?.handshakeTimeout).toBe(200); + ws.emit('message', Buffer.from(JSON.stringify({ type: 'connected' })), false); + stream.close(); + }); + + it('rejects provider close before audio input completes', async () => { + const stream = Object.create(SpeechStream.prototype); + Reflect.set(stream, 'timeoutMs', 100); + Reflect.set(stream, 'abortController', new AbortController()); + Reflect.set(stream, '_opts', { language: 'en-IN' }); + Reflect.set(stream, 'queue', { put: vi.fn() }); + const ws = new EventEmitter(); + const receive = Reflect.apply(Reflect.get(stream, 'receiveMessages'), stream, [ + ws, + { allowClose: false }, + ]); + const rejection = expect(receive).rejects.toMatchObject({ name: 'APIConnectionError' }); + + ws.emit('close', 1006); + + await rejection; + }); + + it('times out a transport-connected socket with no provider response', async () => { + vi.useFakeTimers(); + const stream = Object.create(SpeechStream.prototype); + Reflect.set(stream, 'timeoutMs', 25); + Reflect.set(stream, 'abortController', new AbortController()); + Reflect.set(stream, '_opts', { language: 'en-IN' }); + Reflect.set(stream, 'queue', { put: vi.fn() }); + const ws = new EventEmitter(); + const receive = Reflect.apply(Reflect.get(stream, 'receiveMessages'), stream, [ + ws, + { allowClose: false }, + ]); + const rejection = expect(receive).rejects.toMatchObject({ name: 'APITimeoutError' }); + + await vi.advanceTimersByTimeAsync(25); + + await rejection; + }); + + it('closes a pending WebSocket handshake when the stream aborts', async () => { + vi.useFakeTimers(); + wsState.autoOpen = false; + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + stream.close(); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.closed).toBe(true); + expect(ws.terminated).toBe(false); + }); + + it('closes a pending WebSocket receive when the stream aborts', async () => { + vi.useFakeTimers(); + const stt = new STT({ apiKey: 'test-key' }); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + stream.close(); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.closed).toBe(true); + expect(ws.terminated).toBe(true); + expect(ws.listenerCount('message')).toBe(0); + expect(ws.listenerCount('close')).toBe(0); + expect(ws.listenerCount('error')).toBe(0); + }); + + it('allows provider close during the one-second final drain', async () => { + vi.useFakeTimers(); + const stt = new STT({ apiKey: 'test-key' }); + const errors: Error[] = []; + stt.on('error', (event) => errors.push(event.error)); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + ws.emit('message', Buffer.from(JSON.stringify({ type: 'connected' })), false); + + stream.endInput(); + await vi.advanceTimersByTimeAsync(0); + ws.emit('close', 1000); + await vi.advanceTimersByTimeAsync(0); + + expect(errors).toHaveLength(0); + stream.close(); + }); +}); diff --git a/plugins/gnani/src/stt.ts b/plugins/gnani/src/stt.ts new file mode 100644 index 000000000..a90fd64e5 --- /dev/null +++ b/plugins/gnani/src/stt.ts @@ -0,0 +1,537 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { + type APIConnectOptions, + APIConnectionError, + APIStatusError, + APITimeoutError, + type AudioBuffer, + AudioByteStream, + DEFAULT_API_CONNECT_OPTIONS, + log, + mergeFrames, + normalizeLanguage, + stt, +} from '@livekit/agents'; +import type { AudioFrame } from '@livekit/rtc-node'; +import { type RawData, WebSocket } from 'ws'; + +export const GNANI_STT_BASE_URL = 'https://api.vachana.ai'; + +/** @public */ +export type GnaniSTTFormat = 'verbatim' | 'transcribe'; +/** @public */ +export type GnaniSTTLanguages = + | 'bn-IN' + | 'en-IN' + | 'gu-IN' + | 'hi-IN' + | 'kn-IN' + | 'ml-IN' + | 'mr-IN' + | 'pa-IN' + | 'ta-IN' + | 'te-IN' + | 'en-IN,hi-IN'; + +export const SUPPORTED_LANGUAGES = new Set([ + 'bn-IN', + 'en-IN', + 'gu-IN', + 'hi-IN', + 'kn-IN', + 'ml-IN', + 'mr-IN', + 'pa-IN', + 'ta-IN', + 'te-IN', + 'en-IN,hi-IN', +]); + +export const STREAM_SUPPORTED_LANGUAGES = new Set([ + 'bn-IN', + 'en-IN', + 'gu-IN', + 'hi-IN', + 'kn-IN', + 'ml-IN', + 'mr-IN', + 'pa-IN', + 'ta-IN', + 'te-IN', +]); + +/** @public */ +export const SAMPLE_RATE_16K = 16000; +/** @public */ +export const SAMPLE_RATE_8K = 8000; +export const STREAM_CHUNK_BYTES = 1024; +const NUM_CHANNELS = 1; + +const DEPRECATED_STT_OPTIONS = new Set([ + 'organizationId', + 'organization_id', + 'userId', + 'user_id', + 'httpSession', + 'http_session', +]); +const STT_OPTIONS = new Set([ + 'language', + 'apiKey', + 'sampleRate', + 'baseURL', + 'preferredLanguage', + 'format', + 'itnNativeNumerals', +]); + +/** @public */ +export interface STTOptions { + /** BCP-47 language code (for example, `hi-IN` or `en-IN`). Default: `en-IN`. */ + language?: GnaniSTTLanguages | string; + /** Gnani API key. Defaults to `$GNANI_API_KEY`. */ + apiKey?: string; + /** Streaming audio sample rate. Must be 8000 or 16000. Default: 16000. */ + sampleRate?: typeof SAMPLE_RATE_8K | typeof SAMPLE_RATE_16K | number; + /** Vachana API base URL. */ + baseURL?: string; + /** Force single-language model for this code. */ + preferredLanguage?: string; + /** `verbatim` (default) or `transcribe` to enable ITN. */ + format?: GnaniSTTFormat; + /** Render digits in native script when `format` is `transcribe`. */ + itnNativeNumerals?: boolean; +} + +/** @public */ +export interface ResolvedSTTOptions { + apiKey: string; + language: string; + sampleRate: number; + baseURL: string; + preferredLanguage?: string; + format: GnaniSTTFormat; + itnNativeNumerals: boolean; +} + +function validateOptions(opts: Record, caller: string) { + for (const name of DEPRECATED_STT_OPTIONS) { + if (name in opts) { + log().warn(`\`${name}\` is deprecated and no longer used by ${caller}`); + } + } + const unknown = Object.keys(opts).filter( + (name) => !STT_OPTIONS.has(name) && !DEPRECATED_STT_OPTIONS.has(name), + ); + if (unknown.length > 0) { + throw new TypeError(`${caller}() got unexpected option(s): ${unknown.sort().join(', ')}`); + } +} + +function resolveOptions(opts: STTOptions & Record): ResolvedSTTOptions { + validateOptions(opts, 'STT'); + const apiKey = opts.apiKey ?? process.env.GNANI_API_KEY; + if (!apiKey) { + throw new Error('Gnani API key is required. Provide it directly or set GNANI_API_KEY.'); + } + + const sampleRate = opts.sampleRate ?? SAMPLE_RATE_16K; + if (sampleRate !== SAMPLE_RATE_8K && sampleRate !== SAMPLE_RATE_16K) { + throw new Error('sampleRate must be 8000 or 16000'); + } + + return { + apiKey, + language: opts.language ?? 'en-IN', + sampleRate, + baseURL: opts.baseURL ?? GNANI_STT_BASE_URL, + preferredLanguage: opts.preferredLanguage, + format: opts.format ?? 'verbatim', + itnNativeNumerals: opts.itnNativeNumerals ?? false, + }; +} + +function websocketURL(baseURL: string, path: string): string { + if (baseURL.startsWith('https://')) return `wss://${baseURL.slice('https://'.length)}${path}`; + if (baseURL.startsWith('http://')) return `ws://${baseURL.slice('http://'.length)}${path}`; + return `wss://${baseURL}${path}`; +} + +function createWav(frame: AudioFrame): Buffer { + const bitsPerSample = 16; + const byteRate = (frame.sampleRate * frame.channels * bitsPerSample) / 8; + const blockAlign = (frame.channels * bitsPerSample) / 8; + + const header = Buffer.alloc(44); + header.write('RIFF', 0); + header.writeUInt32LE(36 + frame.data.byteLength, 4); + header.write('WAVE', 8); + header.write('fmt ', 12); + header.writeUInt32LE(16, 16); + header.writeUInt16LE(1, 20); + header.writeUInt16LE(frame.channels, 22); + header.writeUInt32LE(frame.sampleRate, 24); + header.writeUInt32LE(byteRate, 28); + header.writeUInt16LE(blockAlign, 32); + header.writeUInt16LE(bitsPerSample, 34); + header.write('data', 36); + header.writeUInt32LE(frame.data.byteLength, 40); + + const pcm = Buffer.from(frame.data.buffer, frame.data.byteOffset, frame.data.byteLength); + return Buffer.concat([header, pcm]); +} + +function buildFormData(wavBlob: Blob, opts: ResolvedSTTOptions, language?: string): FormData { + const formData = new FormData(); + formData.append('audio_file', wavBlob, 'audio.wav'); + formData.append('language_code', language ?? opts.language); + formData.append('format', opts.format); + if (opts.preferredLanguage != null) { + formData.append('preferred_language', opts.preferredLanguage); + } + if (opts.itnNativeNumerals) { + formData.append('itn_native_numerals', 'true'); + } + return formData; +} + +function mapHTTPError( + error: unknown, + callerSignal: AbortSignal | undefined, + timeoutSignal: AbortSignal, +): Error { + if (callerSignal?.aborted) { + return error instanceof Error + ? error + : new DOMException('The operation was aborted', 'AbortError'); + } + if (timeoutSignal.aborted || (error instanceof DOMException && error.name === 'TimeoutError')) { + return new APITimeoutError({ message: 'Gnani STT API request timed out' }); + } + return new APIConnectionError({ message: `Gnani STT error: ${String(error)}` }); +} + +async function readErrorText( + response: Response, + callerSignal: AbortSignal | undefined, + timeoutSignal: AbortSignal, +): Promise { + try { + return await response.text(); + } catch (error) { + throw mapHTTPError(error, callerSignal, timeoutSignal); + } +} + +function mapWebSocketError(error: Error): APIConnectionError { + if (/timed? out|timeout/i.test(error.message)) { + return new APITimeoutError({ + message: `Gnani STT WebSocket connection timed out: ${error.message}`, + }); + } + return new APIConnectionError({ message: `Gnani STT WebSocket error: ${error.message}` }); +} + +async function waitForDrain(receiveTask: Promise, timeoutMs: number): Promise { + let timeout: NodeJS.Timeout | undefined; + try { + await Promise.race([ + receiveTask, + new Promise((resolve) => { + timeout = setTimeout(resolve, timeoutMs); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +} + +/** @public */ +export class STT extends stt.STT { + _opts: ResolvedSTTOptions; + label = 'gnani.STT'; + + /** Create a new Gnani Vachana Speech-to-Text instance. */ + constructor(opts: STTOptions & Record = {}) { + const resolved = resolveOptions(opts); + super({ streaming: true, interimResults: false, alignedTranscript: false }); + this._opts = resolved; + } + + get model(): string { + return 'vachana-stt-v3'; + } + + get provider(): string { + return 'Gnani'; + } + + protected async _recognize( + buffer: AudioBuffer, + abortSignal?: AbortSignal, + ): Promise { + const frame = mergeFrames(buffer); + const wavBuffer = createWav(frame); + const wavBlob = new Blob([new Uint8Array(wavBuffer)], { type: 'audio/wav' }); + + const timeoutSignal = AbortSignal.timeout(DEFAULT_API_CONNECT_OPTIONS.timeoutMs); + const signal = abortSignal ? AbortSignal.any([abortSignal, timeoutSignal]) : timeoutSignal; + let response: Response; + try { + response = await fetch(`${this._opts.baseURL}/stt/v3`, { + method: 'POST', + headers: { 'X-API-Key-ID': this._opts.apiKey }, + body: buildFormData(wavBlob, this._opts), + signal, + }); + } catch (error) { + throw mapHTTPError(error, abortSignal, timeoutSignal); + } + + if (!response.ok) { + const errorText = await readErrorText(response, abortSignal, timeoutSignal); + throw new APIStatusError({ + message: `Gnani STT API Error (${response.status}): ${errorText}`, + options: { statusCode: response.status, body: { error: errorText } }, + }); + } + + let data: { transcript?: string; request_id?: string }; + try { + data = (await response.json()) as { transcript?: string; request_id?: string }; + } catch (error) { + throw mapHTTPError(error, abortSignal, timeoutSignal); + } + return { + type: stt.SpeechEventType.FINAL_TRANSCRIPT, + requestId: data.request_id ?? '', + alternatives: [ + { + language: normalizeLanguage(this._opts.language), + text: data.transcript ?? '', + confidence: 1, + startTime: 0, + endTime: 0, + }, + ], + }; + } + + stream(options?: { language?: string; connOptions?: APIConnectOptions }): SpeechStream { + return new SpeechStream( + this, + { ...this._opts, language: options?.language ?? this._opts.language }, + options?.connOptions, + ); + } +} + +/** @public */ +export class SpeechStream extends stt.SpeechStream { + _opts: ResolvedSTTOptions; + label = 'gnani.SpeechStream'; + private readonly timeoutMs: number; + + constructor(sttInstance: STT, opts: ResolvedSTTOptions, connOptions?: APIConnectOptions) { + super(sttInstance, opts.sampleRate, connOptions); + this._opts = opts; + this.timeoutMs = connOptions?.timeoutMs ?? DEFAULT_API_CONNECT_OPTIONS.timeoutMs; + } + + buildWsUrl(): string { + return websocketURL(this._opts.baseURL, '/stt/v3/stream'); + } + + protected async run() { + const ws = new WebSocket(this.buildWsUrl(), { + headers: this.buildHeaders(), + handshakeTimeout: this.timeoutMs, + }); + const onHandshakeAbort = () => ws.close(); + let established = false; + try { + await new Promise((resolve, reject) => { + const onOpen = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(mapWebSocketError(error)); + }; + const onClose = (code: number) => { + cleanup(); + if (this.abortSignal.aborted) resolve(); + else reject(new APIConnectionError({ message: `Gnani STT WebSocket closed: ${code}` })); + }; + const cleanup = () => { + ws.removeListener('open', onOpen); + ws.removeListener('error', onError); + ws.removeListener('close', onClose); + this.abortSignal.removeEventListener('abort', onHandshakeAbort); + }; + ws.on('open', onOpen); + ws.on('error', onError); + ws.on('close', onClose); + this.abortSignal.addEventListener('abort', onHandshakeAbort, { once: true }); + if (this.abortSignal.aborted) onHandshakeAbort(); + }); + if (this.abortSignal.aborted) return; + established = true; + + const receiveState = { allowClose: false }; + const sendTask = this.sendAudio(ws).then(() => { + receiveState.allowClose = true; + }); + const receiveTask = this.receiveMessages(ws, receiveState); + const completed = await Promise.race([ + sendTask.then(() => 'send'), + receiveTask.then(() => 'receive'), + ]); + if (completed === 'send') { + await waitForDrain(receiveTask, 1000); + } + } finally { + if (this.abortSignal.aborted && established) ws.terminate(); + else ws.close(); + } + } + + private buildHeaders(): Record { + const headers: Record = { + 'x-api-key-id': this._opts.apiKey, + lang_code: this._opts.language, + 'x-sample-rate': String(this._opts.sampleRate), + }; + if (this._opts.format !== 'verbatim') { + headers['x-format'] = this._opts.format; + } + if (this._opts.preferredLanguage != null) { + headers.preferred_language = this._opts.preferredLanguage; + } + if (this._opts.itnNativeNumerals) { + headers.itn_native_numerals = 'true'; + } + return headers; + } + + private async sendAudio(ws: WebSocket) { + const stream = new AudioByteStream(this._opts.sampleRate, NUM_CHANNELS, STREAM_CHUNK_BYTES / 2); + try { + for await (const data of this.input) { + const frames = + data === SpeechStream.FLUSH_SENTINEL + ? stream.flush() + : stream.write( + data.data.buffer.slice( + data.data.byteOffset, + data.data.byteOffset + data.data.byteLength, + ) as ArrayBuffer, + ); + + for (const frame of frames) { + const chunk = Buffer.from( + frame.data.buffer, + frame.data.byteOffset, + frame.data.byteLength, + ); + ws.send(chunk); + } + } + for (const frame of stream.flush()) { + ws.send(Buffer.from(frame.data.buffer, frame.data.byteOffset, frame.data.byteLength)); + } + } catch (error) { + if (!this.abortSignal.aborted) throw error; + } + } + + private async receiveMessages(ws: WebSocket, state: { allowClose: boolean }) { + return new Promise((resolve, reject) => { + let settled = false; + const receiveTimeout = setTimeout(() => { + settle(new APITimeoutError({ message: 'Gnani STT WebSocket receive timed out' })); + }, this.timeoutMs); + const cleanup = () => { + clearTimeout(receiveTimeout); + ws.removeListener('message', onMessage); + ws.removeListener('close', onClose); + ws.removeListener('error', onError); + this.abortSignal.removeEventListener('abort', onAbort); + }; + const settle = (error?: Error) => { + if (settled) return; + settled = true; + cleanup(); + if (error) reject(error); + else resolve(); + }; + const onMessage = (msg: RawData, isBinary: boolean) => { + if (isBinary) return; + clearTimeout(receiveTimeout); + + try { + const data = JSON.parse(msg.toString()) as Record; + const msgType = data.type; + + if (msgType === 'connected' || msgType === 'processing') return; + + if (msgType === 'transcript') { + const text = typeof data.text === 'string' ? data.text : ''; + if (!text) return; + this.queue.put({ + type: stt.SpeechEventType.FINAL_TRANSCRIPT, + requestId: typeof data.segment_id === 'string' ? data.segment_id : '', + alternatives: [ + { + language: normalizeLanguage(this._opts.language), + text, + confidence: 1, + startTime: 0, + endTime: 0, + }, + ], + }); + } else if (msgType === 'speech_start' || msgType === 'vad_start') { + this.queue.put({ type: stt.SpeechEventType.START_OF_SPEECH }); + } else if (msgType === 'speech_end' || msgType === 'vad_end') { + this.queue.put({ type: stt.SpeechEventType.END_OF_SPEECH }); + } else if (msgType === 'error') { + const message = typeof data.message === 'string' ? data.message : 'Unknown error'; + settle( + new APIStatusError({ + message: `Gnani STT stream error: ${message}`, + options: { statusCode: 500, body: { error: message } }, + }), + ); + } + } catch (error) { + settle( + new APIConnectionError({ message: `Error receiving Gnani STT messages: ${error}` }), + ); + } + }; + const onClose = (code: number) => { + if (this.abortSignal.aborted || state.allowClose) settle(); + else + settle( + new APIConnectionError({ + message: `Gnani STT WebSocket closed before input completed: ${code}`, + }), + ); + }; + const onError = (error: Error) => settle(mapWebSocketError(error)); + const onAbort = () => { + settle(); + ws.terminate(); + }; + ws.on('message', onMessage); + ws.on('close', onClose); + ws.on('error', onError); + this.abortSignal.addEventListener('abort', onAbort, { once: true }); + if (this.abortSignal.aborted) onAbort(); + }); + } +} diff --git a/plugins/gnani/src/tts.test.ts b/plugins/gnani/src/tts.test.ts new file mode 100644 index 000000000..43a0d3e0a --- /dev/null +++ b/plugins/gnani/src/tts.test.ts @@ -0,0 +1,1211 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { TTSMetrics } from '@livekit/agents'; +import { EventEmitter } from 'node:events'; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + expectTypeOf, + it, + vi, +} from 'vitest'; +import { + type GnaniTTSContainers, + type GnaniTTSEncodings, + RESTChunkedStream, + SSEChunkedStream, + SynthesizeStream, + TTS, + type TTSOptions, + WebSocketChunkedStream, +} from './tts.js'; + +interface MockWebSocket extends EventEmitter { + options?: { handshakeTimeout?: number }; + sent: unknown[]; + closed: boolean; + terminated: boolean; +} + +const wsState = vi.hoisted(() => ({ + instances: [] as MockWebSocket[], + autoOpen: true, +})); + +vi.mock('ws', () => { + return { + WebSocket: class MockWebSocket extends EventEmitter { + static OPEN = 1; + readyState = 1; + sent: unknown[] = []; + closed = false; + terminated = false; + options?: { handshakeTimeout?: number }; + + constructor(_url: string, options?: { handshakeTimeout?: number }) { + super(); + this.options = options; + wsState.instances.push(this); + if (wsState.autoOpen) queueMicrotask(() => this.emit('open')); + } + + send(data: unknown) { + this.sent.push(data); + } + + close() { + this.closed = true; + this.emit('close', 1000); + } + + terminate() { + this.terminated = true; + this.closed = true; + this.emit('close', 1006); + } + }, + }; +}); + +const swallowExpectedRejection = (reason: unknown) => { + if ( + reason instanceof Error && + ['APIConnectionError', 'APIStatusError', 'APITimeoutError'].includes(reason.name) + ) { + return; + } + throw reason; +}; +beforeAll(() => process.on('unhandledRejection', swallowExpectedRejection)); +afterAll(() => void process.off('unhandledRejection', swallowExpectedRejection)); + +describe('Gnani TTS', () => { + afterEach(() => { + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); + vi.useRealTimers(); + }); + + beforeEach(() => { + wsState.instances.length = 0; + wsState.autoOpen = true; + vi.stubGlobal( + 'fetch', + vi.fn(() => new Promise(() => {})), + ); + }); + + it('requires an API key', () => { + vi.stubEnv('GNANI_API_KEY', ''); + expect(() => new TTS({ apiKey: undefined })).toThrow(/API key/i); + }); + + it('accepts apiKey directly', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts._opts.apiKey).toBe('test-key'); + }); + + it('rejects unknown constructor options like Python', () => { + expect(() => new TTS({ apiKey: 'test-key', unknownOption: true })).toThrow( + /unexpected option.*unknownOption/i, + ); + }); + + it('accepts apiKey from env', () => { + vi.stubEnv('GNANI_API_KEY', 'env-key'); + const tts = new TTS(); + expect(tts._opts.apiKey).toBe('env-key'); + }); + + it('defaults to Karan voice', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts._opts.voice).toBe('Karan'); + }); + + it('accepts custom voice', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Raju' }); + expect(tts._opts.voice).toBe('Raju'); + }); + + it('accepts all documented voices', () => { + for (const voice of ['Karan', 'Simran', 'Nara', 'Riya', 'Viraj', 'Raju']) { + const tts = new TTS({ apiKey: 'test-key', voice }); + expect(tts._opts.voice).toBe(voice); + } + }); + + it('rejects unsupported voices', () => { + expect(() => new TTS({ apiKey: 'test-key', voice: 'nonexistent' })).toThrow(/not supported/i); + }); + + it('defaults to vachana-voice-v3', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts._opts.model).toBe('vachana-voice-v3'); + }); + + it('uses vachana-voice-v3 for v3 voices', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Simran' }); + expect(tts._opts.model).toBe('vachana-voice-v3'); + expect(tts.model).toBe('vachana-voice-v3'); + }); + + it('accepts explicit model override', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Karan', model: 'custom-model' }); + expect(tts._opts.model).toBe('custom-model'); + }); + + it('returns model and provider properties', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts.model).toBe('vachana-voice-v3'); + expect(tts.provider).toBe('Gnani'); + }); + + it('reports streaming=true', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts.capabilities.streaming).toBe(true); + }); + + it('defaults to 16000 Hz sample rate', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts.sampleRate).toBe(16000); + }); + + it('accepts custom sample rate', () => { + const tts = new TTS({ apiKey: 'test-key', sampleRate: 44100 }); + expect(tts.sampleRate).toBe(44100); + }); + + it('defaults encoding and container', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts._opts.encoding).toBe('linear_pcm'); + expect(tts._opts.container).toBe('wav'); + }); + + it('publishes only decodable TTS audio formats', () => { + expectTypeOf().toEqualTypeOf<'linear_pcm'>(); + expectTypeOf().toEqualTypeOf<'raw' | 'wav'>(); + expectTypeOf().toEqualTypeOf<'linear_pcm' | undefined>(); + expectTypeOf().toEqualTypeOf<'raw' | 'wav' | undefined>(); + }); + + it('accepts custom audio config', () => { + const tts = new TTS({ apiKey: 'test-key', encoding: 'linear_pcm', container: 'raw' }); + expect(tts._opts.encoding).toBe('linear_pcm'); + expect(tts._opts.container).toBe('raw'); + }); + + it('updateOptions can change voice', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Karan' }); + tts.updateOptions({ voice: 'Simran' }); + expect(tts._opts.voice).toBe('Simran'); + }); + + it('updateOptions can change voice and model', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Karan' }); + tts.updateOptions({ voice: 'Riya', model: 'custom-model' }); + expect(tts._opts.voice).toBe('Riya'); + expect(tts._opts.model).toBe('custom-model'); + }); + + it('updateOptions rejects unsupported voices', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(() => tts.updateOptions({ voice: 'nonexistent' })).toThrow(/not supported/i); + }); + + it('updateOptions can change model', () => { + const tts = new TTS({ apiKey: 'test-key' }); + tts.updateOptions({ model: 'custom-model' }); + expect(tts._opts.model).toBe('custom-model'); + }); + + it('updateOptions rejects fields Python does not make mutable', () => { + const tts = new TTS({ apiKey: 'test-key', sampleRate: 16000, numChannels: 1 }); + + expect(() => tts.updateOptions({ sampleRate: 44100, numChannels: 2 })).toThrow( + /unexpected option.*numChannels.*sampleRate/i, + ); + expect(tts._opts.sampleRate).toBe(16000); + expect(tts._opts.numChannels).toBe(1); + expect(tts.sampleRate).toBe(16000); + expect(tts.numChannels).toBe(1); + }); + + it('updateOptions preserves base sample-rate and channel state', () => { + const tts = new TTS({ apiKey: 'test-key', sampleRate: 44100, numChannels: 2 }); + + tts.updateOptions({ voice: 'Raju', model: 'custom-model' }); + + expect(tts._opts.sampleRate).toBe(tts.sampleRate); + expect(tts._opts.numChannels).toBe(tts.numChannels); + expect(tts.sampleRate).toBe(44100); + expect(tts.numChannels).toBe(2); + }); + + it('stores synthesizeMethod options', () => { + expect(new TTS({ apiKey: 'test-key' })._opts.synthesizeMethod).toBe('rest'); + expect(new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' })._opts.synthesizeMethod).toBe( + 'sse', + ); + expect( + new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' })._opts.synthesizeMethod, + ).toBe('websocket'); + }); + + it('synthesize() routes by synthesizeMethod', () => { + const rest = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }).synthesize('hello'); + rest.close(); + expect(rest).toBeInstanceOf(RESTChunkedStream); + + const sse = new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' }).synthesize('hello'); + sse.close(); + expect(sse).toBeInstanceOf(SSEChunkedStream); + + const websocket = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }).synthesize( + 'hello', + ); + websocket.close(); + expect(websocket).toBeInstanceOf(WebSocketChunkedStream); + }); + + it('defaults to Vachana API base URL', () => { + const tts = new TTS({ apiKey: 'test-key' }); + expect(tts._opts.baseURL).toBe('https://api.vachana.ai'); + }); + + it('builds wss URL from https base', () => { + const tts = new TTS({ apiKey: 'test-key' }); + const stream = new SynthesizeStream(tts, tts._opts); + stream.close(); + expect(stream.buildWsUrl()).toBe('wss://api.vachana.ai/api/v1/tts'); + }); + + it('builds ws URL from http base', () => { + const tts = new TTS({ apiKey: 'test-key', baseURL: 'http://localhost:9090' }); + const stream = new SynthesizeStream(tts, tts._opts); + stream.close(); + expect(stream.buildWsUrl()).toBe('ws://localhost:9090/api/v1/tts'); + }); + + it('defaults and accepts numChannels', () => { + const defaults = new TTS({ apiKey: 'test-key' }); + expect(defaults._opts.numChannels).toBe(1); + expect(defaults.numChannels).toBe(1); + + const custom = new TTS({ apiKey: 'test-key', numChannels: 2 }); + expect(custom._opts.numChannels).toBe(2); + }); + + it('defaults and accepts bitrate', () => { + const defaults = new TTS({ apiKey: 'test-key' }); + expect(defaults._opts.bitrate).toBeUndefined(); + + const custom = new TTS({ apiKey: 'test-key', bitrate: '128k' }); + expect(custom._opts.bitrate).toBe('128k'); + }); + + it('rejects unsupported sample rates', () => { + expect(() => new TTS({ apiKey: 'test-key', sampleRate: 48000 })).toThrow(/sampleRate/i); + }); + + it('accepts all documented sample rates', () => { + for (const sampleRate of [8000, 16000, 22050, 44100]) { + const tts = new TTS({ apiKey: 'test-key', sampleRate }); + expect(tts.sampleRate).toBe(sampleRate); + } + }); + + it('stream() returns a SynthesizeStream instance', () => { + const tts = new TTS({ apiKey: 'test-key' }); + const stream = tts.stream(); + stream.close(); + expect(stream).toBeInstanceOf(SynthesizeStream); + }); + + it('updateOptions preserves other fields', () => { + const tts = new TTS({ apiKey: 'test-key', voice: 'Karan' }); + tts.updateOptions({ voice: 'Raju' }); + expect(tts._opts.voice).toBe('Raju'); + expect(tts._opts.model).toBe('vachana-voice-v3'); + expect(tts._opts.encoding).toBe('linear_pcm'); + }); + + it('WebSocketChunkedStream builds correct WS URL', () => { + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }); + const stream = tts.synthesize('hello'); + stream.close(); + expect(stream).toBeInstanceOf(WebSocketChunkedStream); + expect((stream as WebSocketChunkedStream).buildWsUrl()).toBe('wss://api.vachana.ai/api/v1/tts'); + }); + + it('strips the WAV container before REST audio is decoded as PCM', async () => { + const wav = wavChunk(3200, 7); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(new Uint8Array(wav))), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + + const frame = await tts.synthesize('hello').collect(); + + expect(frame.data[0]).toBe(7); + expect(frame.samplesPerChannel).toBe(1600); + }); + + it('sends the Python-equivalent REST payload and reports usage metrics', async () => { + let request: RequestInit | undefined; + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + request = init; + return new Response(new Uint8Array(wavChunk(3200, 9))); + }), + ); + const tts = new TTS({ + apiKey: 'test-key', + voice: 'Raju', + model: 'custom-model', + sampleRate: 16000, + numChannels: 1, + encoding: 'linear_pcm', + container: 'wav', + bitrate: '128k', + synthesizeMethod: 'rest', + }); + const metricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + + await tts.synthesize('hello').collect(); + const metrics = await metricsPromise; + + expect(JSON.parse(String(request?.body))).toEqual({ + text: 'hello', + voice: 'Raju', + model: 'custom-model', + audio_config: { + sample_rate: 16000, + encoding: 'linear_pcm', + num_channels: 1, + sample_width: 2, + container: 'wav', + bitrate: '128k', + }, + }); + expect(metrics).toMatchObject({ + type: 'tts_metrics', + charactersCount: 5, + audioDurationMs: 100, + inputTokens: 0, + outputTokens: 0, + streamed: false, + metadata: { modelProvider: 'Gnani', modelName: 'custom-model' }, + }); + }); + + it('attributes in-flight metrics to the request model snapshot', async () => { + const requests: RequestInit[] = []; + let completeFirstRequest: ((response: Response) => void) | undefined; + vi.stubGlobal( + 'fetch', + vi.fn((_url: string | URL | Request, init?: RequestInit) => { + if (init) requests.push(init); + if (requests.length === 1) { + return new Promise((resolve) => { + completeFirstRequest = resolve; + }); + } + return Promise.resolve(new Response(new Uint8Array(wavChunk(3200, 17)))); + }), + ); + const tts = new TTS({ + apiKey: 'test-key', + model: 'model-A', + synthesizeMethod: 'rest', + }); + const firstMetricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + const firstAudioPromise = tts.synthesize('first').collect(); + await vi.waitFor(() => expect(requests).toHaveLength(1)); + + tts.updateOptions({ model: 'model-B' }); + completeFirstRequest?.(new Response(new Uint8Array(wavChunk(3200, 13)))); + await firstAudioPromise; + const firstMetrics = await firstMetricsPromise; + + const secondMetricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + await tts.synthesize('second').collect(); + const secondMetrics = await secondMetricsPromise; + + expect(JSON.parse(String(requests[0]?.body))).toMatchObject({ model: 'model-A' }); + expect(firstMetrics).toMatchObject({ metadata: { modelName: 'model-A' } }); + expect(JSON.parse(String(requests[1]?.body))).toMatchObject({ model: 'model-B' }); + expect(secondMetrics).toMatchObject({ metadata: { modelName: 'model-B' } }); + }); + + it('reports streaming audio and usage metrics', async () => { + const tts = new TTS({ apiKey: 'test-key', container: 'raw' }); + const metricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + const stream = tts.stream(); + const drain = (async () => { + for await (const _event of stream) { + // Drain streaming output so metrics can be finalized. + } + })(); + + stream.pushText('hello'); + stream.endInput(); + await vi.waitFor(() => expect(wsState.instances.length).toBeGreaterThan(0)); + const ws = wsState.instances.at(-1)!; + await vi.waitFor(() => expect(ws.listenerCount('message')).toBeGreaterThan(0)); + ws.emit('message', Buffer.alloc(3200, 19), true); + ws.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + + await drain; + const metrics = await metricsPromise; + expect(metrics).toMatchObject({ + type: 'tts_metrics', + charactersCount: 5, + audioDurationMs: 100, + inputTokens: 0, + outputTokens: 0, + streamed: true, + metadata: { modelProvider: 'Gnani', modelName: 'vachana-voice-v3' }, + }); + }); + + it('attributes streaming metrics to each request model snapshot', async () => { + const tts = new TTS({ apiKey: 'test-key', container: 'raw', model: 'model-A' }); + const firstMetricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + const firstStream = tts.stream(); + const firstDrain = (async () => { + for await (const _event of firstStream) { + // Drain the first request so metrics can be finalized. + } + })(); + firstStream.pushText('first'); + firstStream.endInput(); + await vi.waitFor(() => expect(wsState.instances[0]?.sent).toHaveLength(1)); + const firstSocket = wsState.instances[0]!; + const firstPayload = JSON.parse(String(firstSocket.sent[0])); + + tts.updateOptions({ model: 'model-B' }); + firstSocket.emit('message', Buffer.alloc(3200, 19), true); + firstSocket.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + await firstDrain; + const firstMetrics = await firstMetricsPromise; + + const secondMetricsPromise = new Promise((resolve) => { + tts.once('metrics_collected', resolve); + }); + const secondStream = tts.stream(); + const secondDrain = (async () => { + for await (const _event of secondStream) { + // Drain the second request so metrics can be finalized. + } + })(); + secondStream.pushText('second'); + secondStream.endInput(); + await vi.waitFor(() => expect(wsState.instances[1]?.sent).toHaveLength(1)); + const secondSocket = wsState.instances[1]!; + const secondPayload = JSON.parse(String(secondSocket.sent[0])); + secondSocket.emit('message', Buffer.alloc(3200, 23), true); + secondSocket.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + await secondDrain; + const secondMetrics = await secondMetricsPromise; + + expect(firstPayload.model).toBe('model-A'); + expect(firstMetrics).toMatchObject({ metadata: { modelName: 'model-A' } }); + expect(secondPayload.model).toBe('model-B'); + expect(secondMetrics).toMatchObject({ metadata: { modelName: 'model-B' } }); + }); + + it('rejects encoded output formats that are not decoded by the plugin', () => { + expect(() => + Reflect.construct(TTS, [{ apiKey: 'test-key', encoding: 'oggopus', container: 'ogg' }]), + ).toThrow(/unsupported audio format/i); + }); + + it('emits SSE audio before the terminal event', async () => { + let controller: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(value) { + controller = value; + }, + }); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(body)), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' }); + const stream = tts.synthesize('hello'); + const encoder = new TextEncoder(); + + controller?.enqueue( + encoder.encode( + `data: ${JSON.stringify({ audio: wavChunk(3200, 11).toString('base64') })}\r\n\r\n`, + ), + ); + + try { + const first = await Promise.race([ + stream.next(), + new Promise((_, reject) => + setTimeout(() => reject(new Error('SSE audio was buffered until completion')), 50), + ), + ]); + expect(first.value?.frame.data[0]).toBe(11); + controller?.enqueue(encoder.encode(`data: ${JSON.stringify({ is_final: true })}\r\n\r\n`)); + controller?.close(); + for await (const _audio of stream) { + // Drain the stream after observing the incremental frame. + } + } finally { + stream.close(); + } + }); + + it('parses non-binary WebSocket Buffers and emits audio incrementally', async () => { + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }); + const stream = tts.synthesize('hello'); + await vi.waitFor(() => expect(wsState.instances.length).toBeGreaterThan(0)); + const ws = wsState.instances.at(-1)!; + await vi.waitFor(() => expect(ws.listenerCount('message')).toBeGreaterThan(0)); + ws.emit( + 'message', + Buffer.from( + JSON.stringify({ + type: 'audio', + data: { audio: wavChunk(3200, 13).toString('base64') }, + }), + ), + false, + ); + + try { + const first = await Promise.race([ + stream.next(), + new Promise((_, reject) => + setTimeout(() => reject(new Error('WebSocket audio was buffered until completion')), 50), + ), + ]); + expect(first.value?.frame.data[0]).toBe(13); + ws.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + for await (const _audio of stream) { + // Drain the stream after observing the incremental frame. + } + } finally { + stream.close(); + } + }); + + it('maps REST timeout through connOptions', async () => { + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 20, + }); + + expect(Reflect.get(stream, 'timeoutMs')).toBe(20); + stream.close(); + }); + + it('settles a pending REST request quietly when the caller closes it', async () => { + let observeFetch = () => {}; + const fetchStarted = new Promise((resolve) => { + observeFetch = resolve; + }); + let observeAbort = () => {}; + const abortObserved = new Promise((resolve) => { + observeAbort = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn((_url: string | URL | Request, init?: RequestInit) => { + observeFetch(); + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + 'abort', + () => { + reject(new DOMException('The operation was aborted', 'AbortError')); + observeAbort(); + }, + { once: true }, + ); + }); + }), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + const errors = vi.fn(); + tts.on('error', errors); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + const drain = (async () => { + for await (const _event of stream) { + // Drain until caller cancellation closes the output. + } + })(); + await fetchStarted; + + stream.close(); + await abortObserved; + await drain; + await new Promise(setImmediate); + + expect(errors).not.toHaveBeenCalled(); + }); + + it('settles a pending SSE body read quietly when the caller closes it', async () => { + let observeBodyRead = () => {}; + const bodyRead = new Promise((resolve) => { + observeBodyRead = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + let bodyController: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + pull() { + observeBodyRead(); + }, + }); + init?.signal?.addEventListener( + 'abort', + () => bodyController?.error(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + return new Response(body); + }), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' }); + const errors = vi.fn(); + tts.on('error', errors); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + const drain = (async () => { + for await (const _event of stream) { + // Drain until caller cancellation closes the output. + } + })(); + await bodyRead; + + stream.close(); + await drain; + await new Promise(setImmediate); + + expect(errors).not.toHaveBeenCalled(); + }); + + it.each(['rest', 'sse'] as const)( + 'settles a pending %s non-2xx error body quietly when the caller closes it', + async (synthesizeMethod) => { + let observeBodyRead = () => {}; + const bodyRead = new Promise((resolve) => { + observeBodyRead = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + let bodyController: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + pull() { + observeBodyRead(); + }, + }); + init?.signal?.addEventListener( + 'abort', + () => + bodyController?.error(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + return new Response(body, { status: 429, statusText: 'Too Many Requests' }); + }), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod }); + const errors = vi.fn(); + tts.on('error', errors); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + const drain = (async () => { + for await (const _event of stream) { + // Drain until caller cancellation closes the output. + } + })(); + await bodyRead; + + stream.close(); + await drain; + await new Promise(setImmediate); + + expect(errors).not.toHaveBeenCalled(); + }, + ); + + it.each(['rest', 'sse'] as const)( + 'maps a timeout during a %s non-2xx error body to APITimeoutError', + async (synthesizeMethod) => { + vi.useFakeTimers(); + const timeoutController = new AbortController(); + vi.spyOn(AbortSignal, 'timeout').mockReturnValue(timeoutController.signal); + let observeBodyRead = () => {}; + const bodyRead = new Promise((resolve) => { + observeBodyRead = resolve; + }); + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + let bodyController: ReadableStreamDefaultController | undefined; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + pull() { + observeBodyRead(); + }, + }); + init?.signal?.addEventListener( + 'abort', + () => + bodyController?.error(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + return new Response(body, { status: 503, statusText: 'Service Unavailable' }); + }), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod }); + const errorPromise = new Promise((resolve) => { + tts.once('error', resolve); + }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await bodyRead; + + timeoutController.abort(); + await vi.advanceTimersByTimeAsync(0); + + await expect(errorPromise).resolves.toMatchObject({ + error: { name: 'APITimeoutError' }, + }); + stream.close(); + await vi.advanceTimersByTimeAsync(0); + }, + ); + + it.each(['rest', 'sse'] as const)( + 'preserves %s non-2xx provider status and body details', + async (synthesizeMethod) => { + vi.useFakeTimers(); + vi.stubGlobal( + 'fetch', + vi.fn(async () => { + return new Response('provider quota exceeded', { + status: 429, + statusText: 'Too Many Requests', + }); + }), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod }); + const errorPromise = new Promise((resolve) => { + tts.once('error', resolve); + }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + + await expect(errorPromise).resolves.toMatchObject({ + error: { + name: 'APIStatusError', + statusCode: 429, + message: expect.stringContaining('provider quota exceeded'), + }, + }); + stream.close(); + await vi.advanceTimersByTimeAsync(0); + }, + ); + + it('maps timeout AbortError and network failures by provenance', async () => { + vi.useFakeTimers(); + const timeoutController = new AbortController(); + const timeoutSpy = vi.spyOn(AbortSignal, 'timeout').mockReturnValue(timeoutController.signal); + vi.stubGlobal( + 'fetch', + vi.fn((_url: string | URL | Request, init?: RequestInit) => { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + 'abort', + () => reject(new DOMException('The operation was aborted', 'AbortError')), + { once: true }, + ); + }); + }), + ); + const timeoutTTS = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + const timeoutError = new Promise((resolve) => { + timeoutTTS.once('error', resolve); + }); + const timeoutStream = timeoutTTS.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + expect(vi.mocked(fetch)).toHaveBeenCalledOnce(); + + timeoutController.abort(); + await vi.advanceTimersByTimeAsync(0); + + await expect(timeoutError).resolves.toMatchObject({ + error: { name: 'APITimeoutError' }, + }); + timeoutStream.close(); + await vi.advanceTimersByTimeAsync(0); + + timeoutSpy.mockRestore(); + vi.stubGlobal( + 'fetch', + vi.fn(async () => Promise.reject(new Error('network down'))), + ); + const networkTTS = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + const networkError = new Promise((resolve) => { + networkTTS.once('error', resolve); + }); + const networkStream = networkTTS.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + + await expect(networkError).resolves.toMatchObject({ + error: { name: 'APIConnectionError' }, + }); + networkStream.close(); + await vi.advanceTimersByTimeAsync(0); + }); + + it('maps WebSocket connection and receive timeout through connOptions', async () => { + const timeoutSpy = vi.spyOn(globalThis, 'setTimeout'); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 200, + }); + await vi.waitFor(() => + expect(wsState.instances.some((instance) => instance.options?.handshakeTimeout === 200)).toBe( + true, + ), + ); + const ws = wsState.instances.find((instance) => instance.options?.handshakeTimeout === 200)!; + await vi.waitFor(() => expect(ws.listenerCount('message')).toBeGreaterThan(0)); + + expect(ws.options?.handshakeTimeout).toBe(200); + expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), 200); + ws.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + for await (const _audio of stream) { + // Drain the completed stream. + } + timeoutSpy.mockRestore(); + }); + + it('parses a REST WAV data chunk after ancillary RIFF chunks', async () => { + const wav = wavWithJunkChunk(3200, 23); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(new Uint8Array(wav))), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'rest' }); + + const frame = await tts.synthesize('hello').collect(); + + expect(frame.data[0]).toBe(23); + expect(frame.samplesPerChannel).toBe(1600); + }); + + it('parses a streaming WAV header split across SSE payloads', async () => { + const wav = wavWithJunkChunk(3200, 29); + const body = new ReadableStream({ + start(controller) { + for (const chunk of [wav.subarray(0, 7), wav.subarray(7, 31), wav.subarray(31)]) { + controller.enqueue( + new TextEncoder().encode( + `data: ${JSON.stringify({ audio: chunk.toString('base64') })}\n\n`, + ), + ); + } + controller.enqueue( + new TextEncoder().encode(`data: ${JSON.stringify({ is_final: true })}\n\n`), + ); + controller.close(); + }, + }); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(body)), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' }); + + const frame = await tts.synthesize('hello').collect(); + + expect(frame.data[0]).toBe(29); + expect(frame.samplesPerChannel).toBe(1600); + }); + + it('decodes consecutive complete WAV payloads from one SSE response', async () => { + const first = wavChunk(3200, 37); + const second = wavChunk(3200, 41); + const body = new ReadableStream({ + start(controller) { + for (const chunk of [first, second]) { + controller.enqueue( + new TextEncoder().encode( + `data: ${JSON.stringify({ audio: chunk.toString('base64') })}\n\n`, + ), + ); + } + controller.enqueue( + new TextEncoder().encode(`data: ${JSON.stringify({ is_final: true })}\n\n`), + ); + controller.close(); + }, + }); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(body)), + ); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'sse' }); + + const frame = await tts.synthesize('hello').collect(); + + expect(frame.samplesPerChannel).toBe(3200); + expect(frame.data[0]).toBe(37); + expect(frame.data[1600]).toBe(41); + }); + + it('marks the actual exact-aligned WebSocket audio frame final', async () => { + vi.useFakeTimers(); + const tts = new TTS({ + apiKey: 'test-key', + synthesizeMethod: 'websocket', + container: 'raw', + }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + const eventsPromise = (async () => { + const events = []; + for await (const event of stream) events.push(event); + return events; + })(); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + ws.emit('message', Buffer.alloc(3200, 31), true); + ws.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + await vi.advanceTimersByTimeAsync(0); + const events = await eventsPromise; + + expect(events.every((event) => event.frame.samplesPerChannel > 0)).toBe(true); + expect(events.reduce((sum, event) => sum + event.frame.samplesPerChannel, 0)).toBe(1600); + expect(events.filter((event) => event.final)).toHaveLength(1); + expect(events.at(-1)!.final).toBe(true); + }); + + it('closes a pending TTS WebSocket handshake when the stream aborts', async () => { + vi.useFakeTimers(); + wsState.autoOpen = false; + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + stream.close(); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.closed).toBe(true); + expect(ws.terminated).toBe(false); + }); + + it('closes a pending TTS WebSocket receive when the stream aborts', async () => { + vi.useFakeTimers(); + const tts = new TTS({ apiKey: 'test-key', synthesizeMethod: 'websocket' }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + stream.close(); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.closed).toBe(true); + expect(ws.terminated).toBe(true); + expect(ws.listenerCount('message')).toBe(0); + expect(ws.listenerCount('close')).toBe(0); + expect(ws.listenerCount('error')).toBe(0); + }); + + it('removes receive listeners after WebSocket terminal completion', async () => { + vi.useFakeTimers(); + const tts = new TTS({ + apiKey: 'test-key', + synthesizeMethod: 'websocket', + container: 'raw', + }); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + ws.emit('message', Buffer.from(JSON.stringify({ type: 'complete' })), false); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.listenerCount('message')).toBe(0); + expect(ws.listenerCount('close')).toBe(0); + expect(ws.listenerCount('error')).toBe(0); + expect(() => ws.emit('message', Buffer.alloc(3200), true)).not.toThrow(); + stream.close(); + }); + + it('removes receive listeners after WebSocket provider error', async () => { + vi.useFakeTimers(); + const tts = new TTS({ + apiKey: 'test-key', + synthesizeMethod: 'websocket', + container: 'raw', + }); + tts.on('error', () => {}); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + ws.emit('message', Buffer.from(JSON.stringify({ type: 'error', message: 'failed' })), false); + await vi.advanceTimersByTimeAsync(0); + + expect(ws.listenerCount('message')).toBe(0); + expect(ws.listenerCount('close')).toBe(0); + expect(ws.listenerCount('error')).toBe(0); + expect(() => ws.emit('message', Buffer.alloc(3200), true)).not.toThrow(); + stream.close(); + }); + + it('removes receive listeners after WebSocket timeout', async () => { + vi.useFakeTimers(); + const tts = new TTS({ + apiKey: 'test-key', + synthesizeMethod: 'websocket', + container: 'raw', + }); + tts.on('error', () => {}); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 100, + }); + await vi.advanceTimersByTimeAsync(0); + const ws = wsState.instances.at(-1)!; + + await vi.advanceTimersByTimeAsync(100); + + expect(ws.listenerCount('message')).toBe(0); + expect(ws.listenerCount('close')).toBe(0); + expect(ws.listenerCount('error')).toBe(0); + expect(() => ws.emit('message', Buffer.alloc(3200), true)).not.toThrow(); + stream.close(); + }); +}); + +function wavChunk(pcmBytes: number, sample: number): Buffer { + const header = Buffer.alloc(44); + header.write('RIFF', 0); + header.writeUInt32LE(36 + pcmBytes, 4); + header.write('WAVE', 8); + header.write('fmt ', 12); + header.writeUInt32LE(16, 16); + header.writeUInt16LE(1, 20); + header.writeUInt16LE(1, 22); + header.writeUInt32LE(16000, 24); + header.writeUInt32LE(32000, 28); + header.writeUInt16LE(2, 32); + header.writeUInt16LE(16, 34); + header.write('data', 36); + header.writeUInt32LE(pcmBytes, 40); + const pcm = Buffer.alloc(pcmBytes); + pcm.writeInt16LE(sample, 0); + return Buffer.concat([header, pcm]); +} + +function wavWithJunkChunk(pcmBytes: number, sample: number): Buffer { + const fmt = Buffer.alloc(24); + fmt.write('fmt ', 0); + fmt.writeUInt32LE(16, 4); + fmt.writeUInt16LE(1, 8); + fmt.writeUInt16LE(1, 10); + fmt.writeUInt32LE(16000, 12); + fmt.writeUInt32LE(32000, 16); + fmt.writeUInt16LE(2, 20); + fmt.writeUInt16LE(16, 22); + const junk = Buffer.alloc(12); + junk.write('JUNK', 0); + junk.writeUInt32LE(3, 4); + junk.fill(9, 8, 11); + const dataHeader = Buffer.alloc(8); + dataHeader.write('data', 0); + dataHeader.writeUInt32LE(pcmBytes, 4); + const body = Buffer.concat([fmt, junk, dataHeader, Buffer.alloc(pcmBytes)]); + body.writeInt16LE(sample, fmt.length + junk.length + dataHeader.length); + const riff = Buffer.alloc(12); + riff.write('RIFF', 0); + riff.writeUInt32LE(body.length + 4, 4); + riff.write('WAVE', 8); + return Buffer.concat([riff, body]); +} diff --git a/plugins/gnani/src/tts.ts b/plugins/gnani/src/tts.ts new file mode 100644 index 000000000..a835ab278 --- /dev/null +++ b/plugins/gnani/src/tts.ts @@ -0,0 +1,841 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { + type APIConnectOptions, + APIConnectionError, + APIStatusError, + APITimeoutError, + AudioByteStream, + DEFAULT_API_CONNECT_OPTIONS, + log, + shortuuid, + tts, +} from '@livekit/agents'; +import type { AudioFrame } from '@livekit/rtc-node'; +import { type RawData, WebSocket } from 'ws'; + +export const GNANI_TTS_BASE_URL = 'https://api.vachana.ai'; + +/** @public */ +export type GnaniTTSVoices = 'Karan' | 'Simran' | 'Nara' | 'Riya' | 'Viraj' | 'Raju'; +/** @public */ +export type GnaniTTSEncodings = 'linear_pcm'; +/** @public */ +export type GnaniTTSContainers = 'raw' | 'wav'; +/** @public */ +export type GnaniTTSBitrates = '96k' | '128k' | '192k'; +/** @public */ +export type GnaniTTSSynthesizeMethod = 'rest' | 'sse' | 'websocket'; + +export const SUPPORTED_VOICES = new Set([ + 'Karan', + 'Simran', + 'Nara', + 'Riya', + 'Viraj', + 'Raju', +]); +export const SUPPORTED_SAMPLE_RATES = [8000, 16000, 22050, 44100] as const; + +const DEFAULT_SAMPLE_WIDTH = 2; +const DEPRECATED_TTS_OPTIONS = new Set(['language', 'httpSession', 'http_session']); +const TTS_OPTIONS = new Set([ + 'voice', + 'model', + 'sampleRate', + 'numChannels', + 'encoding', + 'container', + 'bitrate', + 'apiKey', + 'baseURL', + 'synthesizeMethod', +]); +const TTS_UPDATE_OPTIONS = new Set(['voice', 'model']); + +/** @public */ +export interface TTSOptions { + /** Voice to use for synthesis. Default: `Karan`. */ + voice?: GnaniTTSVoices | string; + /** TTS model name. Default: `vachana-voice-v3`. */ + model?: string; + /** Audio output sample rate. Default: 16000. */ + sampleRate?: number; + /** Number of audio channels. Default: 1. */ + numChannels?: number; + /** Audio encoding. Default: `linear_pcm`. */ + encoding?: GnaniTTSEncodings; + /** Audio container. Default: `wav`. */ + container?: GnaniTTSContainers; + /** Optional audio bitrate. */ + bitrate?: GnaniTTSBitrates | string; + /** Gnani API key. Defaults to `$GNANI_API_KEY`. */ + apiKey?: string; + /** Vachana API base URL. */ + baseURL?: string; + /** Synthesis transport used by `synthesize()`. Default: `rest`. */ + synthesizeMethod?: GnaniTTSSynthesizeMethod; +} + +/** @public */ +export interface TTSUpdateOptions { + /** Voice to use for future synthesis requests. */ + voice?: GnaniTTSVoices | string; + /** Model to use for future synthesis requests. */ + model?: string; +} + +/** @public */ +export interface ResolvedTTSOptions { + apiKey: string; + voice: string; + model: string; + sampleRate: number; + encoding: GnaniTTSEncodings; + container: GnaniTTSContainers; + numChannels: number; + sampleWidth: number; + bitrate?: string; + baseURL: string; + synthesizeMethod: GnaniTTSSynthesizeMethod; +} + +function validateOptions(opts: Record, allowed: Set, caller: string) { + for (const name of DEPRECATED_TTS_OPTIONS) { + if (name in opts) { + log().warn(`\`${name}\` is deprecated and no longer used by ${caller}`); + } + } + const unknown = Object.keys(opts).filter( + (name) => !allowed.has(name) && !DEPRECATED_TTS_OPTIONS.has(name), + ); + if (unknown.length > 0) { + const sorted = unknown.sort(); + throw new TypeError(`${caller}() got unexpected option(s): ${sorted.join(', ')}`); + } +} + +function resolveOptions(opts: TTSOptions & Record): ResolvedTTSOptions { + validateOptions(opts, TTS_OPTIONS, 'TTS'); + + const apiKey = opts.apiKey ?? process.env.GNANI_API_KEY; + if (!apiKey) { + throw new Error('Gnani API key is required. Provide it directly or set GNANI_API_KEY.'); + } + + const sampleRate = opts.sampleRate ?? 16000; + if (!SUPPORTED_SAMPLE_RATES.some((supported) => supported === sampleRate)) { + throw new Error(`sampleRate must be one of ${SUPPORTED_SAMPLE_RATES.join(', ')}`); + } + + const voice = opts.voice ?? 'Karan'; + if (!SUPPORTED_VOICES.has(voice)) { + throw new Error( + `Voice '${voice}' not supported. Supported voices: ${[...SUPPORTED_VOICES].sort().join(', ')}`, + ); + } + + const encoding = opts.encoding ?? 'linear_pcm'; + const container = opts.container ?? 'wav'; + if (encoding !== 'linear_pcm' || (container !== 'raw' && container !== 'wav')) { + throw new Error( + `Unsupported audio format: encoding=${encoding}, container=${container}. ` + + 'Gnani TTS currently decodes only linear_pcm in raw or wav containers.', + ); + } + + return { + apiKey, + voice, + model: opts.model ?? 'vachana-voice-v3', + sampleRate, + encoding, + container, + numChannels: opts.numChannels ?? 1, + sampleWidth: DEFAULT_SAMPLE_WIDTH, + bitrate: opts.bitrate, + baseURL: opts.baseURL ?? GNANI_TTS_BASE_URL, + synthesizeMethod: opts.synthesizeMethod ?? 'rest', + }; +} + +function websocketURL(baseURL: string, path: string): string { + if (baseURL.startsWith('https://')) return `wss://${baseURL.slice('https://'.length)}${path}`; + if (baseURL.startsWith('http://')) return `ws://${baseURL.slice('http://'.length)}${path}`; + return `wss://${baseURL}${path}`; +} + +function buildPayload(opts: ResolvedTTSOptions, text: string): Record { + const audioConfig: Record = { + sample_rate: opts.sampleRate, + encoding: opts.encoding, + num_channels: opts.numChannels, + sample_width: opts.sampleWidth, + container: opts.container, + }; + if (opts.bitrate != null) { + audioConfig.bitrate = opts.bitrate; + } + + return { + text, + voice: opts.voice, + model: opts.model, + audio_config: audioConfig, + }; +} + +function buildHeaders(opts: ResolvedTTSOptions): Record { + return { + 'X-API-Key-ID': opts.apiKey, + 'Content-Type': 'application/json', + }; +} + +function rawDataToBuffer(data: RawData): Buffer { + if (Array.isArray(data)) return Buffer.concat(data); + if (data instanceof ArrayBuffer) return Buffer.from(data); + return Buffer.from(data.buffer, data.byteOffset, data.byteLength); +} + +class AudioContainerDecoder { + private buffer = Buffer.alloc(0); + private state: 'riff-header' | 'chunk-header' | 'data' | 'data-padding' = 'riff-header'; + private riffBytesRemaining = 0; + private dataBytesRemaining = 0; + private dataPaddingBytesRemaining = 0; + private sawDataChunk = false; + private receivedInput = false; + + constructor(private readonly container: string) {} + + push(data: Buffer): Buffer[] { + if (data.length === 0) return []; + this.receivedInput = true; + if (this.container === 'raw') return [data]; + this.buffer = Buffer.concat([this.buffer, data]); + const output: Buffer[] = []; + + while (true) { + if (this.state === 'riff-header') { + if (this.buffer.length < 12) break; + if ( + this.buffer.subarray(0, 4).toString('ascii') !== 'RIFF' || + this.buffer.subarray(8, 12).toString('ascii') !== 'WAVE' + ) { + throw new APIConnectionError({ message: 'Gnani TTS returned an invalid WAV container' }); + } + const riffSize = this.buffer.readUInt32LE(4); + if (riffSize < 4) { + throw new APIConnectionError({ message: 'Gnani TTS returned an invalid RIFF size' }); + } + this.buffer = this.buffer.subarray(12); + this.riffBytesRemaining = riffSize - 4; + this.state = 'chunk-header'; + continue; + } + + if (this.state === 'chunk-header') { + if (this.riffBytesRemaining === 0) { + this.state = 'riff-header'; + continue; + } + if (this.buffer.length < 8) break; + const chunkId = this.buffer.subarray(0, 4).toString('ascii'); + const chunkSize = this.buffer.readUInt32LE(4); + const paddedChunkSize = chunkSize + (chunkSize % 2); + if (8 + paddedChunkSize > this.riffBytesRemaining) { + throw new APIConnectionError({ message: 'Gnani TTS WAV chunk exceeds RIFF container' }); + } + if (chunkId === 'data') { + this.buffer = this.buffer.subarray(8); + this.riffBytesRemaining -= 8; + this.dataBytesRemaining = chunkSize; + this.dataPaddingBytesRemaining = chunkSize % 2; + this.sawDataChunk = true; + this.state = 'data'; + continue; + } + if (this.buffer.length < 8 + paddedChunkSize) break; + this.buffer = this.buffer.subarray(8 + paddedChunkSize); + this.riffBytesRemaining -= 8 + paddedChunkSize; + continue; + } + + if (this.state === 'data') { + if (this.dataBytesRemaining === 0) { + this.state = 'data-padding'; + continue; + } + if (this.buffer.length === 0) break; + const length = Math.min(this.buffer.length, this.dataBytesRemaining); + output.push(this.buffer.subarray(0, length)); + this.buffer = this.buffer.subarray(length); + this.dataBytesRemaining -= length; + this.riffBytesRemaining -= length; + continue; + } + + if (this.dataPaddingBytesRemaining > 0) { + if (this.buffer.length === 0) break; + this.buffer = this.buffer.subarray(1); + this.dataPaddingBytesRemaining -= 1; + this.riffBytesRemaining -= 1; + continue; + } + this.state = 'chunk-header'; + } + return output; + } + + finish() { + if (this.container !== 'wav' || !this.receivedInput) return; + if (!this.sawDataChunk) { + throw new APIConnectionError({ message: 'Gnani TTS WAV response has no data chunk' }); + } + if (this.state !== 'riff-header' || this.buffer.length > 0) { + throw new APIConnectionError({ message: 'Gnani TTS returned a truncated WAV container' }); + } + } +} + +class IncrementalAudioEmitter { + private readonly stream: AudioByteStream; + private readonly decoder: AudioContainerDecoder; + private pendingFrame: AudioFrame | undefined; + + constructor( + private readonly queue: { put: (audio: tts.SynthesizedAudio) => void }, + private readonly opts: ResolvedTTSOptions, + private readonly requestId: string, + private readonly segmentId: string, + ) { + const frameSamples = Math.max(1, Math.floor(opts.sampleRate / 100)); + this.stream = new AudioByteStream(opts.sampleRate, opts.numChannels, frameSamples); + this.decoder = new AudioContainerDecoder(opts.container); + } + + push(data: Buffer) { + for (const pcm of this.decoder.push(data)) { + this.pushFrames(this.stream.write(pcm)); + } + } + + flush() { + this.decoder.finish(); + const frames = this.stream.flush(); + this.pushFrames(frames); + if (this.pendingFrame) { + this.queue.put({ + requestId: this.requestId, + segmentId: this.segmentId, + frame: this.pendingFrame, + final: true, + }); + this.pendingFrame = undefined; + } + } + + private pushFrames(frames: AudioFrame[]) { + for (const frame of frames) { + if (frame.samplesPerChannel === 0) continue; + if (this.pendingFrame) { + this.queue.put({ + requestId: this.requestId, + segmentId: this.segmentId, + frame: this.pendingFrame, + final: false, + }); + } + this.pendingFrame = frame; + } + } +} + +function apiError(message: string, statusCode = 500): APIStatusError { + return new APIStatusError({ + message, + options: { statusCode, body: { error: message } }, + }); +} + +function mapWebSocketError(error: Error): APIConnectionError { + if (/timed? out|timeout/i.test(error.message)) { + return new APITimeoutError({ + message: `Gnani TTS WebSocket connection timed out: ${error.message}`, + }); + } + return new APIConnectionError({ message: `Gnani TTS WebSocket error: ${error.message}` }); +} + +function mapHTTPError( + error: unknown, + callerSignal: AbortSignal, + timeoutSignal: AbortSignal, + requestName: string, +): APIConnectionError | undefined { + if (callerSignal.aborted) return undefined; + if (timeoutSignal.aborted || (error instanceof DOMException && error.name === 'TimeoutError')) { + return new APITimeoutError({ message: `${requestName} timed out` }); + } + return new APIConnectionError({ message: `${requestName} error: ${String(error)}` }); +} + +async function readErrorText( + response: Response, + callerSignal: AbortSignal, + timeoutSignal: AbortSignal, + requestName: string, +): Promise { + try { + return await response.text(); + } catch (error) { + const mapped = mapHTTPError(error, callerSignal, timeoutSignal, requestName); + if (!mapped) return undefined; + throw mapped; + } +} + +/** @public */ +export class TTS extends tts.TTS { + _opts: ResolvedTTSOptions; + label = 'gnani.TTS'; + + /** Create a new Gnani Vachana Text-to-Speech instance. */ + constructor(opts: TTSOptions & Record = {}) { + const resolved = resolveOptions(opts); + super(resolved.sampleRate, resolved.numChannels, { streaming: true }); + this._opts = resolved; + } + + get model(): string { + return this._opts.model; + } + + get provider(): string { + return 'Gnani'; + } + + synthesize( + text: string, + connOptions?: APIConnectOptions, + abortSignal?: AbortSignal, + ): ChunkedStream { + if (this._opts.synthesizeMethod === 'sse') { + return new SSEChunkedStream(this, text, this._opts, connOptions, abortSignal); + } + if (this._opts.synthesizeMethod === 'websocket') { + return new WebSocketChunkedStream(this, text, this._opts, connOptions, abortSignal); + } + return new RESTChunkedStream(this, text, this._opts, connOptions, abortSignal); + } + + stream(options?: { connOptions?: APIConnectOptions }): SynthesizeStream { + return new SynthesizeStream(this, this._opts, options?.connOptions); + } + + updateOptions(opts: TTSUpdateOptions & Record) { + validateOptions(opts, TTS_UPDATE_OPTIONS, 'TTS.updateOptions'); + if (opts.voice != null) { + if (!SUPPORTED_VOICES.has(opts.voice)) { + throw new Error( + `Voice '${opts.voice}' not supported. Supported voices: ${[...SUPPORTED_VOICES].sort().join(', ')}`, + ); + } + this._opts.voice = opts.voice; + } + if (opts.model != null) this._opts.model = opts.model; + } +} + +/** @public */ +export type ChunkedStream = RESTChunkedStream | SSEChunkedStream | WebSocketChunkedStream; + +/** @public */ +export class RESTChunkedStream extends tts.ChunkedStream { + private opts: ResolvedTTSOptions; + private readonly timeoutMs: number; + label = 'gnani.RESTChunkedStream'; + + constructor( + ttsInstance: TTS, + text: string, + opts: ResolvedTTSOptions, + connOptions?: APIConnectOptions, + abortSignal?: AbortSignal, + ) { + super(text, ttsInstance, connOptions, abortSignal); + this.opts = { ...opts }; + this.timeoutMs = connOptions?.timeoutMs ?? DEFAULT_API_CONNECT_OPTIONS.timeoutMs; + } + + protected async run() { + const timeoutSignal = AbortSignal.timeout(this.timeoutMs); + const signal = AbortSignal.any([this.abortSignal, timeoutSignal]); + let response: Response; + try { + response = await fetch(`${this.opts.baseURL}/api/v1/tts/inference`, { + method: 'POST', + headers: buildHeaders(this.opts), + body: JSON.stringify(buildPayload(this.opts, this.inputText)), + signal, + }); + } catch (error) { + const mapped = mapHTTPError(error, this.abortSignal, timeoutSignal, 'Gnani TTS REST request'); + if (!mapped) return; + throw mapped; + } + + if (!response.ok) { + const errorText = await readErrorText( + response, + this.abortSignal, + timeoutSignal, + 'Gnani TTS REST error response', + ); + if (errorText === undefined) return; + throw apiError(`Gnani TTS API Error (${response.status}): ${errorText}`, response.status); + } + + let audioBuffer: ArrayBuffer; + try { + audioBuffer = await response.arrayBuffer(); + } catch (error) { + const mapped = mapHTTPError( + error, + this.abortSignal, + timeoutSignal, + 'Gnani TTS REST response', + ); + if (!mapped) return; + throw mapped; + } + const audioBytes = Buffer.from(audioBuffer); + const requestId = shortuuid(); + const emitter = new IncrementalAudioEmitter(this.queue, this.opts, requestId, requestId); + emitter.push(audioBytes); + emitter.flush(); + } +} + +/** @public */ +export class SSEChunkedStream extends tts.ChunkedStream { + private opts: ResolvedTTSOptions; + private readonly timeoutMs: number; + label = 'gnani.SSEChunkedStream'; + + constructor( + ttsInstance: TTS, + text: string, + opts: ResolvedTTSOptions, + connOptions?: APIConnectOptions, + abortSignal?: AbortSignal, + ) { + super(text, ttsInstance, connOptions, abortSignal); + this.opts = { ...opts }; + this.timeoutMs = connOptions?.timeoutMs ?? DEFAULT_API_CONNECT_OPTIONS.timeoutMs; + } + + protected async run() { + const timeoutSignal = AbortSignal.timeout(this.timeoutMs); + const signal = AbortSignal.any([this.abortSignal, timeoutSignal]); + let response: Response; + try { + response = await fetch(`${this.opts.baseURL}/api/v1/tts/sse`, { + method: 'POST', + headers: buildHeaders(this.opts), + body: JSON.stringify(buildPayload(this.opts, this.inputText)), + signal, + }); + } catch (error) { + const mapped = mapHTTPError(error, this.abortSignal, timeoutSignal, 'Gnani TTS SSE request'); + if (!mapped) return; + throw mapped; + } + + if (!response.ok) { + const errorText = await readErrorText( + response, + this.abortSignal, + timeoutSignal, + 'Gnani TTS SSE error response', + ); + if (errorText === undefined) return; + throw apiError(`Gnani TTS SSE Error (${response.status}): ${errorText}`, response.status); + } + if (!response.body) { + throw new APIConnectionError({ message: 'Gnani TTS SSE returned no response body' }); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + const requestId = shortuuid(); + const segmentId = requestId; + const emitter = new IncrementalAudioEmitter(this.queue, this.opts, requestId, segmentId); + + const handlePayload = (payload: Record) => { + if (payload.status === 'error' || 'error' in payload) { + throw apiError(String(payload.message ?? payload.error ?? 'Gnani TTS SSE error')); + } + if (payload.status === 'streaming_started') return false; + const audio = typeof payload.audio === 'string' ? payload.audio : ''; + if (audio) { + emitter.push(Buffer.from(audio, 'base64')); + } + return payload.is_final === true; + }; + + while (true) { + let readResult: ReadableStreamReadResult; + try { + readResult = await reader.read(); + } catch (error) { + const mapped = mapHTTPError( + error, + this.abortSignal, + timeoutSignal, + 'Gnani TTS SSE response', + ); + if (!mapped) return; + throw mapped; + } + const { done, value } = readResult; + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + let boundary = /\r?\n\r?\n/.exec(buffer); + while (boundary) { + const event = buffer.slice(0, boundary.index); + buffer = buffer.slice(boundary.index + boundary[0].length); + const dataLines = event + .split(/\r?\n/) + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).trim()); + if (dataLines.length > 0) { + const isFinal = handlePayload(JSON.parse(dataLines.join('')) as Record); + if (isFinal) { + emitter.flush(); + return; + } + } + boundary = /\r?\n\r?\n/.exec(buffer); + } + } + + emitter.flush(); + } +} + +/** @public */ +export class WebSocketChunkedStream extends tts.ChunkedStream { + private opts: ResolvedTTSOptions; + private readonly timeoutMs: number; + label = 'gnani.WebSocketChunkedStream'; + + constructor( + ttsInstance: TTS, + text: string, + opts: ResolvedTTSOptions, + connOptions?: APIConnectOptions, + abortSignal?: AbortSignal, + ) { + super(text, ttsInstance, connOptions, abortSignal); + this.opts = { ...opts }; + this.timeoutMs = connOptions?.timeoutMs ?? DEFAULT_API_CONNECT_OPTIONS.timeoutMs; + } + + buildWsUrl(): string { + return websocketURL(this.opts.baseURL, '/api/v1/tts'); + } + + protected async run() { + const requestId = shortuuid(); + const segmentId = requestId; + const emitter = new IncrementalAudioEmitter(this.queue, this.opts, requestId, segmentId); + await synthesizeViaWebSocket( + this.buildWsUrl(), + this.inputText, + this.opts, + this.timeoutMs, + this.abortSignal, + (chunk) => emitter.push(chunk), + ); + if (this.abortSignal.aborted) return; + emitter.flush(); + } +} + +/** @public */ +export class SynthesizeStream extends tts.SynthesizeStream { + private opts: ResolvedTTSOptions; + label = 'gnani.SynthesizeStream'; + + constructor(ttsInstance: TTS, opts: ResolvedTTSOptions, connOptions?: APIConnectOptions) { + super(ttsInstance, connOptions); + this.opts = { ...opts }; + } + + buildWsUrl(): string { + return websocketURL(this.opts.baseURL, '/api/v1/tts'); + } + + protected async run() { + const textParts: string[] = []; + for await (const data of this.input) { + if (data === SynthesizeStream.FLUSH_SENTINEL) break; + textParts.push(data); + } + + const text = textParts.join('').trim(); + if (!text) return; + + const requestId = shortuuid(); + const segmentId = shortuuid(); + const emitter = new IncrementalAudioEmitter(this.queue, this.opts, requestId, segmentId); + await synthesizeViaWebSocket( + this.buildWsUrl(), + text, + this.opts, + this.connOptions.timeoutMs, + this.abortSignal, + (chunk) => emitter.push(chunk), + () => this.markStarted(), + ); + if (this.abortSignal.aborted) return; + emitter.flush(); + this.queue.put(SynthesizeStream.END_OF_STREAM); + } +} + +async function synthesizeViaWebSocket( + url: string, + text: string, + opts: ResolvedTTSOptions, + timeoutMs: number, + abortSignal: AbortSignal, + onAudio: (chunk: Buffer) => void, + onStarted?: () => void, +): Promise { + const ws = new WebSocket(url, { + headers: buildHeaders(opts), + handshakeTimeout: timeoutMs, + }); + const onHandshakeAbort = () => ws.close(); + let established = false; + try { + await new Promise((resolve, reject) => { + let settled = false; + const settle = (error?: Error) => { + if (settled) return; + settled = true; + cleanup(); + if (error) reject(error); + else resolve(); + }; + const onOpen = () => settle(); + const onError = (error: Error) => { + if (abortSignal.aborted) settle(); + else settle(mapWebSocketError(error)); + }; + const onClose = (code: number) => { + if (abortSignal.aborted) settle(); + else settle(new APIConnectionError({ message: `Gnani TTS WebSocket closed: ${code}` })); + }; + const cleanup = () => { + ws.removeListener('open', onOpen); + ws.removeListener('error', onError); + ws.removeListener('close', onClose); + abortSignal.removeEventListener('abort', onHandshakeAbort); + }; + ws.on('open', onOpen); + ws.on('error', onError); + ws.on('close', onClose); + abortSignal.addEventListener('abort', onHandshakeAbort, { once: true }); + if (abortSignal.aborted) onHandshakeAbort(); + }); + if (abortSignal.aborted) return; + established = true; + + ws.send(JSON.stringify(buildPayload(opts, text))); + onStarted?.(); + await new Promise((resolve, reject) => { + let settled = false; + const receiveTimeout = setTimeout(() => { + settle(new APITimeoutError({ message: 'Gnani TTS WebSocket receive timed out' })); + }, timeoutMs); + const cleanup = () => { + clearTimeout(receiveTimeout); + ws.removeListener('message', onMessage); + ws.removeListener('close', onClose); + ws.removeListener('error', onError); + abortSignal.removeEventListener('abort', onAbort); + }; + const settle = (error?: Error) => { + if (settled) return; + settled = true; + cleanup(); + if (error) reject(error); + else resolve(); + }; + const onMessage = (msg: RawData, isBinary: boolean) => { + try { + if (isBinary) { + onAudio(rawDataToBuffer(msg)); + return; + } + + const payload = JSON.parse(msg.toString()) as Record; + const msgType = payload.type; + const data = (payload.data ?? {}) as Record; + const audio = typeof data.audio === 'string' ? data.audio : ''; + + if (msgType === 'audio') { + if (audio) onAudio(Buffer.from(audio, 'base64')); + } else if (msgType === 'complete') { + if (audio) onAudio(Buffer.from(audio, 'base64')); + settle(); + } else if (msgType === 'error') { + settle(apiError(String(payload.message ?? data.message ?? 'Gnani TTS stream error'))); + } + } catch (error) { + if ( + error instanceof APIStatusError || + error instanceof APIConnectionError || + error instanceof APITimeoutError + ) { + settle(error); + } else { + settle(new APIConnectionError({ message: `Gnani TTS WebSocket error: ${error}` })); + } + } + }; + const onClose = (code: number, reason: Buffer) => { + if (abortSignal.aborted) settle(); + else + settle( + new APIConnectionError({ + message: `Gnani TTS WebSocket closed before completion: ${code} ${reason?.toString() ?? ''}`, + }), + ); + }; + const onError = (error: Error) => { + if (abortSignal.aborted) settle(); + else settle(mapWebSocketError(error)); + }; + const onAbort = () => { + settle(); + ws.terminate(); + }; + ws.on('message', onMessage); + ws.on('close', onClose); + ws.on('error', onError); + abortSignal.addEventListener('abort', onAbort, { once: true }); + if (abortSignal.aborted) onAbort(); + }); + } finally { + if (abortSignal.aborted && established) ws.terminate(); + else ws.close(); + } +} diff --git a/plugins/gnani/src/tts.ws.test.ts b/plugins/gnani/src/tts.ws.test.ts new file mode 100644 index 000000000..28c0b706f --- /dev/null +++ b/plugins/gnani/src/tts.ws.test.ts @@ -0,0 +1,178 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { APIError } from '@livekit/agents'; +import { AudioFrame } from '@livekit/rtc-node'; +import { createServer } from 'node:http'; +import { Socket } from 'node:net'; +import type { Duplex } from 'node:stream'; +import { afterAll, beforeAll, expect, it } from 'vitest'; +import { type WebSocket, WebSocketServer } from 'ws'; +import { STT } from './stt.js'; +import { TTS } from './tts.js'; + +const swallowExpectedRejection = (reason: unknown) => { + if (reason instanceof APIError) return; + throw reason; +}; +beforeAll(() => process.on('unhandledRejection', swallowExpectedRejection)); +afterAll(() => void process.off('unhandledRejection', swallowExpectedRejection)); + +async function establishedServer() { + const server = createServer(); + const webSocketServer = new WebSocketServer({ server }); + const connected = new Promise<{ peer: WebSocket; socket: Socket; closed: Promise }>( + (resolve, reject) => { + webSocketServer.once('connection', (peer) => { + const socket = Reflect.get(peer, '_socket'); + if (!(socket instanceof Socket)) { + reject(new Error('real WebSocket has no network socket')); + return; + } + const closed = new Promise((resolveClose) => socket.once('close', resolveClose)); + resolve({ peer, socket, closed }); + }); + }, + ); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (address === null || typeof address === 'string') { + throw new Error('local WebSocket server has no TCP address'); + } + return { + port: address.port, + connected, + async close() { + for (const peer of webSocketServer.clients) peer.terminate(); + await new Promise((resolve) => webSocketServer.close(() => resolve())); + await new Promise((resolve, reject) => + server.close((error) => { + if (error) reject(error); + else resolve(); + }), + ); + }, + }; +} + +it('settles an error-before-close abort while a real WebSocket handshake is delayed', async () => { + const server = createServer(); + const webSocketServer = new WebSocketServer({ noServer: true }); + let resolveUpgrade: (() => void) | undefined; + let upgradeSocket: Duplex | undefined; + let socketClosed: Promise | undefined; + const upgradeStarted = new Promise((resolve) => { + resolveUpgrade = resolve; + }); + server.on('upgrade', (_request, socket) => { + upgradeSocket = socket; + socket.on('error', () => {}); + socketClosed = new Promise((resolve) => socket.once('close', resolve)); + resolveUpgrade?.(); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (address === null || typeof address === 'string') { + throw new Error('local WebSocket server has no TCP address'); + } + const { port } = address; + const errors: Error[] = []; + const tts = new TTS({ + apiKey: 'test-key', + baseURL: `http://127.0.0.1:${port}`, + synthesizeMethod: 'websocket', + container: 'raw', + }); + tts.on('error', (event) => errors.push(event.error)); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 1000, + }); + + await upgradeStarted; + stream.close(); + upgradeSocket?.destroy(); + await socketClosed; + await new Promise((resolve) => setImmediate(resolve)); + + try { + expect(errors).toHaveLength(0); + } finally { + webSocketServer.close(); + await new Promise((resolve, reject) => + server.close((error) => { + if (error) reject(error); + else resolve(); + }), + ); + } +}); + +it('forcefully settles an established TTS socket when the peer ignores close frames', async () => { + const local = await establishedServer(); + const errors: Error[] = []; + const tts = new TTS({ + apiKey: 'test-key', + baseURL: `http://127.0.0.1:${local.port}`, + synthesizeMethod: 'websocket', + container: 'raw', + }); + tts.on('error', (event) => errors.push(event.error)); + const stream = tts.synthesize('hello', { + maxRetry: 0, + retryIntervalMs: 0, + timeoutMs: 1000, + }); + const { socket, closed } = await local.connected; + socket.pause(); + + try { + stream.close(); + await closed; + + expect(socket.destroyed).toBe(true); + expect(errors).toHaveLength(0); + } finally { + await local.close(); + } +}); + +it('forcefully settles established STT after its first text response', async () => { + const local = await establishedServer(); + const errors: Error[] = []; + const stt = new STT({ + apiKey: 'test-key', + baseURL: `http://127.0.0.1:${local.port}`, + }); + stt.on('error', (event) => errors.push(event.error)); + const stream = stt.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 1000 }, + }); + const { peer, socket, closed } = await local.connected; + const receivedAudio = new Promise((resolve) => peer.once('message', () => resolve())); + stream.pushFrame(AudioFrame.create(16000, 1, 2400)); + await receivedAudio; + await new Promise((resolve, reject) => { + peer.send( + JSON.stringify({ type: 'transcript', text: 'ready', segment_id: 'segment-1' }), + (error) => { + if (error) reject(error); + else resolve(); + }, + ); + }); + const result = await stream.next(); + expect(result.value?.alternatives[0]?.text).toBe('ready'); + socket.pause(); + + try { + stream.close(); + await closed; + + expect(socket.destroyed).toBe(true); + expect(errors).toHaveLength(0); + } finally { + await local.close(); + } +}); diff --git a/plugins/gnani/tsconfig.json b/plugins/gnani/tsconfig.json new file mode 100644 index 000000000..d3d3e25b3 --- /dev/null +++ b/plugins/gnani/tsconfig.json @@ -0,0 +1,14 @@ +{ + "extends": "../../tsconfig.json", + "include": ["./src"], + "compilerOptions": { + "rootDir": "./src", + "declarationDir": "./dist", + "outDir": "./dist" + }, + "typedocOptions": { + "name": "plugins/agents-plugin-gnani", + "entryPointStrategy": "resolve", + "entryPoints": ["src/index.ts"] + } +} diff --git a/plugins/gnani/tsup.config.ts b/plugins/gnani/tsup.config.ts new file mode 100644 index 000000000..b491713a4 --- /dev/null +++ b/plugins/gnani/tsup.config.ts @@ -0,0 +1,6 @@ +import { defineConfig } from 'tsup'; +import defaults from '../../tsup.config'; + +export default defineConfig({ + ...defaults, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf76e639c..050df1c0f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -741,6 +741,31 @@ importers: specifier: ^5.0.0 version: 5.9.3 + plugins/gnani: + dependencies: + ws: + specifier: 'catalog:' + version: 8.20.0 + devDependencies: + '@livekit/agents': + specifier: workspace:* + version: link:../../agents + '@livekit/rtc-node': + specifier: 'catalog:' + version: 0.13.31 + '@microsoft/api-extractor': + specifier: ^7.35.0 + version: 7.43.7(@types/node@25.6.0) + '@types/ws': + specifier: 'catalog:' + version: 8.18.1 + tsup: + specifier: ^8.3.5 + version: 8.4.0(@microsoft/api-extractor@7.43.7(@types/node@25.6.0))(postcss@8.5.17)(tsx@4.21.0)(typescript@5.9.3) + typescript: + specifier: ^5.0.0 + version: 5.9.3 + plugins/google: dependencies: '@google/genai': diff --git a/turbo.json b/turbo.json index 5699519ad..d49f046df 100644 --- a/turbo.json +++ b/turbo.json @@ -24,6 +24,7 @@ "ELEVEN_API_KEY", "FIREWORKS_API_KEY", "FISH_API_KEY", + "GNANI_API_KEY", "GROQ_API_KEY", "HEDRA_API_KEY", "HEDRA_API_URL",