From fe347e4ea522759199eaac9d2187777d04353a12 Mon Sep 17 00:00:00 2001 From: Nick Nyanjui Date: Sat, 26 Sep 2026 11:48:48 +0700 Subject: [PATCH] fix(stt): dispatch live-API-only Gemini models over the Live WebSocket transport (#4006) Addresses #4006 by letting Gemini STT models that only exist on the Live API transcribe instead of failing. transcribeGemini sends audio to :generateContent and that is the only Gemini path. A model that is realtime-only (exposed by the Live API's bidiGenerateContent WebSocket) therefore fails outright, even though the account can transcribe it. open-sse/handlers/geminiLiveStt.js owns the WebSocket lifecycle: opens :bidiGenerateContent, sends setup frame, waits for setupComplete, streams audio as realtimeInput media chunks, and settles on turnComplete. Dispatch is driven by transport marker 'gemini-live'. Adds custom model transport persistence and selection on the dashboard. --- open-sse/handlers/geminiLiveStt.js | 266 +++++++++ open-sse/handlers/sttCore.js | 46 +- open-sse/providers/registry/gemini.js | 1 + .../providers/[id]/AddCustomModelModal.js | 38 +- .../dashboard/providers/[id]/page.js | 12 +- src/app/api/models/custom/route.js | 17 +- src/lib/db/repos/aliasRepo.js | 8 +- src/shared/constants/models.js | 20 + src/sse/handlers/stt.js | 25 +- tests/unit/gemini-live-stt.test.js | 534 ++++++++++++++++++ 10 files changed, 945 insertions(+), 22 deletions(-) create mode 100644 open-sse/handlers/geminiLiveStt.js create mode 100644 tests/unit/gemini-live-stt.test.js diff --git a/open-sse/handlers/geminiLiveStt.js b/open-sse/handlers/geminiLiveStt.js new file mode 100644 index 00000000..51837732 --- /dev/null +++ b/open-sse/handlers/geminiLiveStt.js @@ -0,0 +1,266 @@ +import { Buffer } from "node:buffer"; + +// Gemini Live API realtime STT transport. +// +// The REST generateContent path (sttCore.transcribeGemini) only transcribes +// whole files inline. The Live API's `:bidiGenerateContent` WebSocket is the +// streaming counterpart: audio goes up as realtimeInput mediaChunks and the +// server pushes incremental `serverContent.inputTranscription` events back. +// This module owns the socket lifecycle only — envelope/response shaping +// stays in sttCore so the engine's single STT exit shape is preserved. +// +// Marker contract: dispatched from sttCore's format-switch when the model +// entry carries `transport: "gemini-live"` (registry) or the caller passes a +// transport string (custom models). Never keyed on a hardcoded model id here. +// +// Transport behavior: +// - Node >= 22 global WebSocket (undici). No new dependency. +// - Live API expects low-latency PCM; other containers are forwarded with +// their declared MIME unchanged (provider-side rejection is surfaced). +// - Text accumulation is append-only over inputTranscription segments and +// ends on serverContent.turnComplete (or graceful close with partial text). +// - Transcription deltas are kept per-frame (chunks[]) so sttCore can shape +// verbose_json segments without fabricating timestamps. goAway advisements +// rotate the socket once per call: setup replay + byte-offset resume. + +const SETUP_TIMEOUT_MS = 10_000; // open → setupComplete +const TURN_TIMEOUT_MS = 60_000; // audio streamed → turnComplete +const MAX_TIMEOUT_MS = 300_000; // clamp ceiling for client-supplied lifecycle knobs +const CHUNK_BYTES = 16_384; // ~0.5s of 16-bit 16kHz mono PCM +const GOAWAY_RECONNECTS = 1; // socket rotations honoured per call + +class GeminiLiveError extends Error { + constructor(message, status) { + super(message); + this.name = "GeminiLiveError"; + this.status = status || 502; + } +} + +// REST base (https://host/v1beta/models) → Live WS base +// (wss://host/ws/api/v1beta/models), then the bidiGenerateContent endpoint. +function toLiveWsUrl(baseUrl, model, token) { + const url = new URL(baseUrl); + url.protocol = "wss:"; + if (!url.pathname.startsWith("/ws/")) url.pathname = `/ws/api${url.pathname}`; + const base = url.toString().replace(/\/+$/, ""); + return `${base}/${encodeURIComponent(model)}:bidiGenerateContent?key=${encodeURIComponent(token || "")}`; +} + +// Bind socket events supporting BOTH handler styles: addEventListener +// (browser WebSocket, undici) and onopen/onmessage property assignment +// (minimal polyfills). Whichever the implementation exposes, it works. +function bindSocket(ws, { onOpen, onMessage, onError, onClose }) { + if (typeof ws.addEventListener === "function") { + ws.addEventListener("open", onOpen); + ws.addEventListener("message", onMessage); + ws.addEventListener("error", onError); + ws.addEventListener("close", onClose); + return; + } + ws.onopen = onOpen; + ws.onmessage = onMessage; + ws.onerror = onError; + ws.onclose = onClose; +} + +function parseFrame(data) { + try { + return JSON.parse(typeof data === "string" ? data : String(data)); + } catch { + return null; // non-JSON frames carry no Live API semantics + } +} + +function firstStringField(formData, key) { + const v = typeof formData?.get === "function" ? formData.get(key) : null; + return typeof v === "string" && v.trim() ? v.trim() : ""; +} + +// Lifecycle knobs the live registry entry advertises in params[] +// (setup/turn timeouts). They ride the same formData pass-through sttCore +// gives every transport — no sttCore change needed to reach this leaf. +function firstNumberField(formData, key, fallback) { + const n = Number(firstStringField(formData, key)); + return Number.isFinite(n) && n > 0 ? Math.min(n, MAX_TIMEOUT_MS) : fallback; +} + +/** + * Transcribe an audio File via the Gemini Live bidirectional stream. + * @returns {Promise<{text: string, chunks: string[]}>} transcript plus the raw + * incremental inputTranscription deltas (sttCore shapes verbose_json from them). + * @throws {GeminiLiveError} with .status for the sttCore error envelope. + */ +export async function transcribeGeminiLive({ cfg, file, model, token, formData, mimeType }) { + const WS = globalThis.WebSocket; + if (!WS) throw new GeminiLiveError("Gemini Live transport needs global WebSocket (Node >= 22)", 502); + + const buf = Buffer.from(await file.arrayBuffer()); + if (!buf.length) throw new GeminiLiveError("Empty audio file", 400); + + const instruction = firstStringField(formData, "prompt") || "Transcribe the spoken audio verbatim."; + const language = firstStringField(formData, "language"); + const setupTimeoutMs = firstNumberField(formData, "setup_timeout_ms", SETUP_TIMEOUT_MS); + const turnTimeoutMs = firstNumberField(formData, "turn_timeout_ms", TURN_TIMEOUT_MS); + // system_instruction (registry param) overrides the built-in transcription + // directive wholesale; prompt/language only shape the default. + const instructionOverride = firstStringField(formData, "system_instruction"); + const systemText = instructionOverride + || (language ? `${instruction} Language: ${language}.` : instruction); + const wsUrl = toLiveWsUrl(cfg.baseUrl, model, token); + + return await new Promise((resolve, reject) => { + let text = ""; + const chunks = []; // raw inputTranscription deltas, shaped by sttCore + let settled = false; + let timer = null; + let goAwayTimer = null; + let ws = null; + let generation = 0; // socket identity: superseded closes never settle + let sentBytes = 0; // audio prefix already handed to the live socket + let goAwayReconnects = GOAWAY_RECONNECTS; + + const arm = (ms, message) => { + if (timer) clearTimeout(timer); + timer = setTimeout(() => fail(new GeminiLiveError(message, 504)), ms); + }; + const shutdown = () => { + if (timer) { clearTimeout(timer); timer = null; } + if (goAwayTimer) { clearTimeout(goAwayTimer); goAwayTimer = null; } + // ws is null until the first open() dials (and stays null when the + // constructor throws) — fail() runs shutdown() on that path. + if (!ws) return; + try { + if (ws.readyState === WS.OPEN || ws.readyState === WS.CONNECTING) ws.close(1000); + } catch { /* socket already dead — outcome is already settled */ } + }; + const succeed = () => { + if (settled) return; + settled = true; + shutdown(); + resolve({ text, chunks }); + }; + const fail = (err) => { + if (settled) return; + settled = true; + shutdown(); + reject(err); + }; + const send = (frame) => { + if (ws.readyState !== WS.OPEN) return false; + try { + ws.send(JSON.stringify(frame)); + } catch { + return false; // socket died mid-send — streamAudioAndPrompt maps this to a 502 + } + return true; + }; + + // Streams every byte not yet sent, then the flushing text turn. After a + // goAway rotation this resumes from sentBytes — no audio re-upload. + const streamAudioAndPrompt = () => { + for (let off = sentBytes; off < buf.length; off += CHUNK_BYTES) { + const mediaChunk = buf.subarray(off, off + CHUNK_BYTES).toString("base64"); + if (!send({ realtimeInput: { mediaChunks: [{ mimeType, data: mediaChunk }] } })) { + fail(new GeminiLiveError("Gemini Live socket closed while streaming audio", 502)); + return; + } + sentBytes = Math.min(off + CHUNK_BYTES, buf.length); + } + // Final user turn: flushes the recognizer and yields turnComplete. + send({ clientContent: { turns: [{ parts: [{ text: systemText }] }], turnComplete: true } }); + }; + + // goAway: the server names the instant it will force-close this socket. + // Graceful play = rotate BEFORE the deadline: retire the live socket, + // dial a fresh one, replay setup, resume audio from sentBytes — text and + // chunks survive the hop. Once the advisory budget is spent a later + // goAway is left to the close path, which settles on partial transcript. + const scheduleGoAwayReconnect = (goAway) => { + if (settled || goAwayTimer || goAwayReconnects <= 0) return; + const deadline = Date.parse(typeof goAway?.time === "string" ? goAway.time : ""); + const delay = Number.isFinite(deadline) + ? Math.max(0, Math.min(deadline - Date.now(), setupTimeoutMs)) + : 0; + goAwayTimer = setTimeout(() => { + goAwayTimer = null; + if (settled) return; + goAwayReconnects--; + generation++; + try { ws?.close(1000); } catch { /* deadline crossed mid-flight — re-dial anyway */ } + open(); + }, delay); + }; + + const open = () => { + const gen = ++generation; + try { + ws = new WS(wsUrl); + } catch { + fail(new GeminiLiveError("Gemini Live websocket connection failed", 502)); + return; + } + bindSocket(ws, { + onOpen: () => { + if (settled || gen !== generation) return; + arm(setupTimeoutMs, "Gemini Live timed out waiting for setupComplete"); + send({ + setup: { + model: `models/${model}`, + generationConfig: { + responseModalities: ["TEXT"], + inputAudioTranscription: {}, + }, + systemInstruction: { parts: [{ text: systemText }] }, + }, + }); + }, + onMessage: (ev) => { + if (settled || gen !== generation) return; + const frame = parseFrame(ev?.data); + if (!frame) return; + + if (frame.error) { + const e = frame.error; + fail(new GeminiLiveError(`Gemini Live error${e.status ? ` (${e.status})` : ""}: ${e.message || "unknown"}`, 502)); + return; + } + if (frame.goAway) { + scheduleGoAwayReconnect(frame.goAway); + return; + } + + const sc = frame.serverContent; + if (!sc) return; + + const delta = typeof sc.inputTranscription?.text === "string" ? sc.inputTranscription.text : ""; + // Trim before testing: a padding-only frame carries no transcript and + // must not make an empty run look like a partial success on close. + if (delta.trim()) { + text += delta; + chunks.push(delta); + } + + if (sc.setupComplete) { + arm(turnTimeoutMs, "Gemini Live transcription timed out"); + streamAudioAndPrompt(); + return; + } + if (sc.turnComplete) succeed(); + }, + onError: () => { + if (settled || gen !== generation) return; + fail(new GeminiLiveError("Gemini Live websocket connection failed", 502)); + }, + onClose: (ev) => { + if (settled || gen !== generation) return; + // Partial transcript beats a hard error on graceful close; silence is one. + if (text.trim()) succeed(); + else fail(new GeminiLiveError(`Gemini Live socket closed before completion${ev?.code ? ` (code ${ev.code})` : ""}`, 502)); + }, + }); + }; + + open(); + }); +} diff --git a/open-sse/handlers/sttCore.js b/open-sse/handlers/sttCore.js index 8127782a..21582c4d 100644 --- a/open-sse/handlers/sttCore.js +++ b/open-sse/handlers/sttCore.js @@ -1,5 +1,7 @@ import { Buffer } from "node:buffer"; import { createErrorResult } from "../utils/error.js"; +import { transcribeGeminiLive } from "./geminiLiveStt.js"; +import { PROVIDER_MODELS, PROVIDER_ID_TO_ALIAS } from "../config/providerModels.js"; import { HTTP_STATUS } from "../config/runtimeConfig.js"; // Build auth headers from sttConfig + token @@ -162,11 +164,26 @@ function jsonResponse(obj) { }; } +// Model-level transport marker (registry models[].transport, e.g. the Gemini +// live STT entry's "gemini-live", or a custom model's stored transport). +// Dispatch reads the marker — never a hardcoded model id — so new realtime +// providers extend sttCore through data, not code. +function resolveModelTransport(provider, model) { + const key = PROVIDER_ID_TO_ALIAS[provider] || provider; + const models = PROVIDER_MODELS[key] || PROVIDER_MODELS[provider]; + if (!Array.isArray(models)) return null; + const entry = models.find((m) => m && m.id === model && (m.kind || "llm") === "stt"); + const marker = typeof entry?.transport === "string" ? entry.transport.trim() : ""; + return marker || null; +} + /** - * STT core handler — dispatch by sttConfig.format. + * STT core handler — dispatch by model transport marker, else sttConfig.format. + * `transport` is the caller-supplied marker override (custom models resolve + * it in the app layer; built-ins fall back to the registry entry marker). * @returns {Promise<{success, response, status?, error?}>} */ -export async function handleSttCore({ provider, model, formData, credentials, sttConfig }) { +export async function handleSttCore({ provider, model, formData, credentials, sttConfig, transport }) { const file = formData.get("file"); if (!file) return createErrorResult(HTTP_STATUS.BAD_REQUEST, "Missing required field: file"); @@ -186,8 +203,29 @@ export async function handleSttCore({ provider, model, formData, credentials, st return createErrorResult(HTTP_STATUS.UNAUTHORIZED, `No credentials for STT provider: ${provider}`); } + // Format-switch extension: an explicit caller marker wins over the registry + // marker; with neither, the provider-default sttConfig.format applies. + const marker = (typeof transport === "string" && transport.trim()) ? transport.trim() : resolveModelTransport(provider, model); + try { - switch (cfg.format) { + switch (marker || cfg.format) { + case "gemini-live": { + const live = await transcribeGeminiLive({ cfg, file, model, token, formData, mimeType: resolveAudioContentType(file) }); + // response_format parity with the OpenAI-compatible transport: default + // envelope stays {text}; verbose_json adds segments mapped from the + // Live API's incremental inputTranscription deltas. Those frames carry + // NO timestamps, so segments expose {id,text} only (id = delta order, + // Whisper-compatible 0-based) — start/end/duration are deliberately + // absent rather than fabricated as zeros, which would misrepresent + // provider data to callers diffing transports. + const fmt = typeof formData?.get === "function" + ? String(formData.get("response_format") ?? "").trim().toLowerCase() + : ""; + if (fmt === "verbose_json") { + return jsonResponse({ text: live.text, segments: live.chunks.map((segText, id) => ({ id, text: segText })) }); + } + return jsonResponse({ text: live.text }); + } case "deepgram": return await transcribeDeepgram(cfg, file, model, token, formData); case "assemblyai": return await transcribeAssemblyAI(cfg, file, model, token); case "nvidia-asr": return await transcribeNvidia(cfg, file, model, token); @@ -196,6 +234,6 @@ export async function handleSttCore({ provider, model, formData, credentials, st default: return await transcribeOpenAICompatible(cfg, file, model, token, formData); } } catch (err) { - return createErrorResult(HTTP_STATUS.BAD_GATEWAY, err.message || "STT request failed"); + return createErrorResult(err.status || HTTP_STATUS.BAD_GATEWAY, err.message || "STT request failed"); } } diff --git a/open-sse/providers/registry/gemini.js b/open-sse/providers/registry/gemini.js index 590bf48e..e0fb109d 100644 --- a/open-sse/providers/registry/gemini.js +++ b/open-sse/providers/registry/gemini.js @@ -58,6 +58,7 @@ export default { { id: "gemini-2.5-flash", name: "Gemini 2.5 Flash", params: ["language","prompt"], kind: "stt" }, { id: "gemini-2.5-flash-lite", name: "Gemini 2.5 Flash Lite (Cheapest)", params: ["language","prompt"], kind: "stt" }, { id: "gemini-2.0-flash", name: "Gemini 2.0 Flash", params: ["language","prompt"], kind: "stt" }, + { id: "gemini-2.5-flash-native-audio-preview-09-17", name: "Gemini Live Transcription (Realtime)", params: ["language","prompt","system_instruction","setup_timeout_ms","turn_timeout_ms"], kind: "stt", transport: "gemini-live" }, { id: "gemini-3.1-flash-tts-preview", name: "Gemini 3.1 Flash TTS", kind: "tts" }, { id: "gemini-2.5-flash-preview-tts", name: "Gemini 2.5 Flash TTS", kind: "tts" }, { id: "gemini-2.5-pro-preview-tts", name: "Gemini 2.5 Pro TTS", kind: "tts" }, diff --git a/src/app/(dashboard)/dashboard/providers/[id]/AddCustomModelModal.js b/src/app/(dashboard)/dashboard/providers/[id]/AddCustomModelModal.js index 3c9b21d0..2636be3d 100644 --- a/src/app/(dashboard)/dashboard/providers/[id]/AddCustomModelModal.js +++ b/src/app/(dashboard)/dashboard/providers/[id]/AddCustomModelModal.js @@ -2,8 +2,8 @@ import { useState, useEffect } from "react"; import PropTypes from "prop-types"; -import { Button, Modal, Toggle } from "@/shared/components"; -import { CAPACITY_META } from "@/shared/constants/models"; +import { Button, Modal, Select, Toggle } from "@/shared/components"; +import { CAPACITY_META, STT_TRANSPORT_META, STT_TRANSPORTS } from "@/shared/constants/models"; const defaultCaps = () => Object.fromEntries(Object.keys(CAPACITY_META).map((key) => [key, false])); @@ -13,10 +13,12 @@ export default function AddCustomModelModal({ isOpen, providerAlias, providerDis const [testStatus, setTestStatus] = useState(null); // null | "testing" | "ok" | "error" const [testError, setTestError] = useState(""); const [saving, setSaving] = useState(false); + // Realtime dispatch marker for the transport select; "" = provider default REST. + const [transport, setTransport] = useState(""); // Reset state when modal opens useEffect(() => { - if (isOpen) { setModelId(""); setCaps(defaultCaps()); setTestStatus(null); setTestError(""); } + if (isOpen) { setModelId(""); setCaps(defaultCaps()); setTransport(""); setTestStatus(null); setTestError(""); } }, [isOpen]); // Strip provider's own alias prefix (e.g. "cc/model" -> "model" for cc provider) @@ -50,7 +52,9 @@ export default function AddCustomModelModal({ isOpen, providerAlias, providerDis if (!cleanId || saving) return; setSaving(true); try { - await onSave(cleanId, caps); + // caps.stt is UI-only; the parent save flow derives the model type from + // it and forwards the pinned transport (null unless the caller picked one). + await onSave(cleanId, caps, caps.stt ? transport : null); } finally { setSaving(false); } @@ -106,6 +110,32 @@ export default function AddCustomModelModal({ isOpen, providerAlias, providerDis + {/* STT is a model TYPE, not a chat capability: the save flow turns this + flag into type "stt" (the API honours a transport only on stt + records). The select pins the realtime dispatch marker persisted + with the model; the whitelist is the shared STT_TRANSPORT_META. */} +
+ { setCaps((prev) => ({ ...prev, stt: v })); if (!v) setTransport(""); }} + label="Speech to text" + description="Transcribes audio via /v1/audio/transcriptions" + size="sm" + /> + {caps.stt && ( +
+