From 5c89ae52ce42bfae4ff98e8707d94f8bb3c73de0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 9 Feb 2026 13:33:40 +0700 Subject: [PATCH] Update files --- .../src/server/agent/stt-manager.test.ts | 69 ++++++ .../server/src/server/agent/stt-manager.ts | 204 +++++++++++++++--- .../dictation-stream-manager.test.ts | 53 +++++ .../dictation/dictation-stream-manager.ts | 67 ++++++ packages/server/src/server/session.ts | 2 +- .../server/speech/providers/local/runtime.ts | 2 +- .../providers/local/sherpa/model-catalog.ts | 11 +- .../speech/speech-config-resolver.test.ts | 12 +- 8 files changed, 382 insertions(+), 38 deletions(-) diff --git a/packages/server/src/server/agent/stt-manager.test.ts b/packages/server/src/server/agent/stt-manager.test.ts index 8b8f91a68..9cbf1ca4f 100644 --- a/packages/server/src/server/agent/stt-manager.test.ts +++ b/packages/server/src/server/agent/stt-manager.test.ts @@ -51,6 +51,49 @@ class FakeStt implements SpeechToTextProvider { } } +class SequencedFakeStt implements SpeechToTextProvider { + public readonly id = "fake-sequenced"; + constructor(private readonly transcripts: string[]) {} + + createSession(_params: { + logger: any; + language?: string; + prompt?: string; + }): StreamingTranscriptionSession { + const emitter = new EventEmitter(); + const transcripts = this.transcripts; + let segmentId = "seg-1"; + let previousSegmentId: string | null = null; + let idx = 0; + + return { + requiredSampleRate: 24000, + async connect() {}, + appendPcm16() {}, + commit() { + const transcript = transcripts[idx] ?? ""; + idx += 1; + (emitter as any).emit("committed", { segmentId, previousSegmentId }); + (emitter as any).emit("transcript", { + segmentId, + transcript, + isFinal: true, + language: "en", + isLowConfidence: transcript.length === 0, + }); + previousSegmentId = segmentId; + segmentId = `seg-${idx + 1}`; + }, + clear() {}, + close() {}, + on(event: any, handler: any) { + emitter.on(event, handler); + return undefined; + }, + }; + } +} + describe("STTManager", () => { it("returns empty text for low-confidence transcriptions", async () => { const manager = new STTManager( @@ -79,4 +122,30 @@ describe("STTManager", () => { expect(result.language).toBe("en"); expect(result.byteLength).toBe(4); }); + + it("uses streaming segmentation for batch transcription and concatenates segment finals", async () => { + const original = process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS; + process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS = "1"; + + try { + const manager = new STTManager( + "s1", + pino({ level: "silent" }), + new SequencedFakeStt(["alpha", "beta", "gamma"]) + ); + + const threeSecondsPcm = Buffer.alloc(24000 * 2 * 3); + const result = await manager.transcribe(threeSecondsPcm, "audio/pcm;rate=24000"); + + expect(result.text).toBe("alpha beta gamma"); + expect(result.language).toBe("en"); + expect(result.byteLength).toBe(threeSecondsPcm.length); + } finally { + if (original === undefined) { + delete process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS; + } else { + process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS = original; + } + } + }); }); diff --git a/packages/server/src/server/agent/stt-manager.ts b/packages/server/src/server/agent/stt-manager.ts index 6cdd35ea8..878e05abc 100644 --- a/packages/server/src/server/agent/stt-manager.ts +++ b/packages/server/src/server/agent/stt-manager.ts @@ -10,6 +10,22 @@ interface TranscriptionMetadata { label?: string; } +const BATCH_APPEND_CHUNK_SECONDS = 1; +const DEFAULT_BATCH_COMMIT_EVERY_SECONDS = 15; +const BATCH_FINAL_TIMEOUT_MS = 120_000; + +function resolveBatchCommitEverySeconds(): number { + const fromEnv = process.env.PASEO_STT_BATCH_COMMIT_EVERY_SECONDS; + if (!fromEnv) { + return DEFAULT_BATCH_COMMIT_EVERY_SECONDS; + } + const parsed = Number.parseFloat(fromEnv); + if (!Number.isFinite(parsed) || parsed < 0) { + return DEFAULT_BATCH_COMMIT_EVERY_SECONDS; + } + return parsed; +} + export interface SessionTranscriptionResult extends TranscriptionResult { debugRecordingPath?: string; byteLength: number; @@ -97,38 +113,168 @@ export class STTManager { try { const startedAt = Date.now(); - const finalEventPromise = new Promise<{ - transcript: string; - language?: string; - logprobs?: TranscriptionResult["logprobs"]; - avgLogprob?: number; - isLowConfidence?: boolean; - }>((resolve, reject) => { - session.on("error", reject); - session.on("transcript", (payload) => { - if (!payload.isFinal) { - return; - } - resolve({ - transcript: payload.transcript, - language: payload.language, - logprobs: payload.logprobs, - avgLogprob: payload.avgLogprob, - isLowConfidence: payload.isLowConfidence, - }); - }); + await session.connect(); + + const committedSegmentIds: string[] = []; + const transcriptsBySegmentId = new Map(); + const finalTranscriptSegmentIds = new Set(); + const transcriptMetaBySegmentId = new Map< + string, + { + language?: string; + logprobs?: TranscriptionResult["logprobs"]; + avgLogprob?: number; + isLowConfidence?: boolean; + } + >(); + + let expectedFinals = 0; + let settle: (() => void) | null = null; + let fail: ((error: Error) => void) | null = null; + let settled = false; + const allFinalsReady = new Promise((resolve, reject) => { + settle = resolve; + fail = reject; }); - await session.connect(); - session.appendPcm16(pcmForModel); - session.commit(); - const finalEvent = await finalEventPromise; + const resolveIfComplete = () => { + if (settled) { + return; + } + if (expectedFinals > 0 && finalTranscriptSegmentIds.size >= expectedFinals) { + settled = true; + settle?.(); + return; + } + if (expectedFinals === 0 && finalTranscriptSegmentIds.size > 0) { + settled = true; + settle?.(); + } + }; + + const rejectWith = (error: Error) => { + if (settled) { + return; + } + settled = true; + fail?.(error); + }; + + session.on("error", (error) => { + rejectWith(error instanceof Error ? error : new Error(String(error))); + }); + + session.on("committed", ({ segmentId }) => { + committedSegmentIds.push(segmentId); + expectedFinals += 1; + resolveIfComplete(); + }); + + session.on("transcript", (payload) => { + transcriptsBySegmentId.set(payload.segmentId, payload.transcript); + if (!payload.isFinal) { + return; + } + finalTranscriptSegmentIds.add(payload.segmentId); + transcriptMetaBySegmentId.set(payload.segmentId, { + language: payload.language, + logprobs: payload.logprobs, + avgLogprob: payload.avgLogprob, + isLowConfidence: payload.isLowConfidence, + }); + resolveIfComplete(); + }); + + const appendChunkBytes = Math.max( + 1, + Math.round(session.requiredSampleRate * 2 * BATCH_APPEND_CHUNK_SECONDS) + ); + const commitEverySeconds = resolveBatchCommitEverySeconds(); + const commitEveryBytes = + commitEverySeconds > 0 + ? Math.max(1, Math.round(session.requiredSampleRate * 2 * commitEverySeconds)) + : 0; + + let bytesSinceCommit = 0; + for (let offset = 0; offset < pcmForModel.length; offset += appendChunkBytes) { + const chunk = pcmForModel.subarray( + offset, + Math.min(pcmForModel.length, offset + appendChunkBytes) + ); + if (chunk.length === 0) { + continue; + } + session.appendPcm16(chunk); + bytesSinceCommit += chunk.length; + + if (commitEveryBytes > 0 && bytesSinceCommit >= commitEveryBytes) { + session.commit(); + bytesSinceCommit = 0; + } + } + + if (bytesSinceCommit > 0 || expectedFinals === 0) { + session.commit(); + } + + const finalTimeout = setTimeout(() => { + if (settled) { + return; + } + settled = true; + this.logger.warn( + { + expectedFinals, + receivedFinals: finalTranscriptSegmentIds.size, + label: metadata?.label, + }, + "Timed out waiting for final STT segments; returning available transcripts" + ); + settle?.(); + }, BATCH_FINAL_TIMEOUT_MS); + + await allFinalsReady; + clearTimeout(finalTimeout); + + const committedSet = new Set(committedSegmentIds); + const orderedSegmentIds: string[] = [...committedSegmentIds]; + for (const segmentId of transcriptsBySegmentId.keys()) { + if (!committedSet.has(segmentId)) { + orderedSegmentIds.push(segmentId); + } + } + + const transcript = orderedSegmentIds + .map((segmentId) => transcriptsBySegmentId.get(segmentId) ?? "") + .join(" ") + .trim(); + const orderedFinalMeta = orderedSegmentIds + .filter((segmentId) => finalTranscriptSegmentIds.has(segmentId)) + .map((segmentId) => transcriptMetaBySegmentId.get(segmentId)) + .filter( + ( + meta + ): meta is { + language?: string; + logprobs?: TranscriptionResult["logprobs"]; + avgLogprob?: number; + isLowConfidence?: boolean; + } => Boolean(meta) + ); + const language = orderedFinalMeta.find((meta) => meta.language)?.language; + const singleSegmentMeta = orderedFinalMeta.length === 1 ? orderedFinalMeta[0] : null; + const allLowConfidence = + orderedFinalMeta.length > 0 && + orderedFinalMeta.every((meta) => meta.isLowConfidence === true); + const result: TranscriptionResult = { - text: finalEvent.transcript, - language: finalEvent.language, - logprobs: finalEvent.logprobs, - avgLogprob: finalEvent.avgLogprob, - isLowConfidence: finalEvent.isLowConfidence, + text: transcript, + ...(language ? { language } : {}), + ...(singleSegmentMeta?.logprobs ? { logprobs: singleSegmentMeta.logprobs } : {}), + ...(singleSegmentMeta?.avgLogprob !== undefined + ? { avgLogprob: singleSegmentMeta.avgLogprob } + : {}), + ...(allLowConfidence ? { isLowConfidence: true } : {}), duration: Date.now() - startedAt, }; diff --git a/packages/server/src/server/dictation/dictation-stream-manager.test.ts b/packages/server/src/server/dictation/dictation-stream-manager.test.ts index 2e8ae0251..f4085d0aa 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.test.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.test.ts @@ -149,4 +149,57 @@ describe("DictationStreamManager (provider-agnostic provider)", () => { } } }); + + it("auto-commits while streaming and assembles final transcript in segment order", async () => { + const originalDebug = process.env.PASEO_DICTATION_DEBUG; + process.env.PASEO_DICTATION_DEBUG = "false"; + + try { + const session = new FakeRealtimeSession(); + const emitted: Array<{ type: string; payload: any }> = []; + const manager = new DictationStreamManager({ + logger: pino({ level: "silent" }), + emit: (msg) => emitted.push(msg), + sessionId: "s1", + stt: new FakeSttProvider(session), + autoCommitSeconds: 1, + }); + + await manager.handleStart("d-segmented", "audio/pcm;rate=24000;bits=16"); + + await manager.handleChunk({ + dictationId: "d-segmented", + seq: 0, + audioBase64: buildPcmBase64(2000, 24000), + format: "audio/pcm;rate=24000;bits=16", + }); + expect(session.commitCalls).toBe(1); + + session.emitCommitted("seg-1"); + session.emitTranscript("seg-1", "hello", true); + + await manager.handleChunk({ + dictationId: "d-segmented", + seq: 1, + audioBase64: buildPcmBase64(2000, 12000), + format: "audio/pcm;rate=24000;bits=16", + }); + + await manager.handleFinish("d-segmented", 1); + expect(session.commitCalls).toBe(2); + + session.emitCommitted("seg-2"); + session.emitTranscript("seg-2", "world", true); + await tick(); + + const final = emitted.find((msg) => msg.type === "dictation_stream_final"); + expect(final?.payload.text).toBe("hello world"); + } finally { + if (originalDebug === undefined) { + delete process.env.PASEO_DICTATION_DEBUG; + } else { + process.env.PASEO_DICTATION_DEBUG = originalDebug; + } + } + }); }); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.ts b/packages/server/src/server/dictation/dictation-stream-manager.ts index 0a6f21b38..c1c0ae1cd 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.ts @@ -16,11 +16,23 @@ import { parsePcmRateFromFormat, pcm16lePeakAbs } from "../speech/audio.js"; const PCM_CHANNELS = 1; const PCM_BITS_PER_SAMPLE = 16; const DEFAULT_DICTATION_FINAL_TIMEOUT_MS = 10000; +const DEFAULT_DICTATION_AUTO_COMMIT_SECONDS = 15; const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt( process.env.PASEO_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300", 10 ); +function parseNonNegativeNumber(value: string | undefined): number | null { + if (value === undefined) { + return null; + } + const parsed = Number.parseFloat(value); + if (!Number.isFinite(parsed) || parsed < 0) { + return null; + } + return parsed; +} + function convertPCMToWavBuffer( pcmBuffer: Buffer, sampleRate: number, @@ -64,6 +76,7 @@ type DictationStreamState = { receivedChunks: Map; nextSeqToForward: number; ackSeq: number; + autoCommitBytes: number; bytesSinceCommit: number; peakSinceCommit: number; committedSegmentIds: string[]; @@ -98,6 +111,7 @@ export class DictationStreamManager { private readonly sessionId: string; private readonly stt: SpeechToTextProvider | null; private readonly finalTimeoutMs: number; + private readonly autoCommitSeconds: number; private readonly streams = new Map(); constructor(params: { @@ -106,12 +120,17 @@ export class DictationStreamManager { sessionId: string; stt: SpeechToTextProvider | null; finalTimeoutMs?: number; + autoCommitSeconds?: number; }) { this.logger = params.logger.child({ component: "dictation-stream-manager" }); this.emit = params.emit; this.sessionId = params.sessionId; this.stt = params.stt; this.finalTimeoutMs = params.finalTimeoutMs ?? DEFAULT_DICTATION_FINAL_TIMEOUT_MS; + this.autoCommitSeconds = + params.autoCommitSeconds ?? + parseNonNegativeNumber(process.env.PASEO_DICTATION_AUTO_COMMIT_SECONDS) ?? + DEFAULT_DICTATION_AUTO_COMMIT_SECONDS; } public cleanupAll(): void { @@ -213,6 +232,10 @@ export class DictationStreamManager { ); const outputRate = stt.requiredSampleRate; + const autoCommitBytes = + this.autoCommitSeconds > 0 + ? Math.max(1, Math.round(this.autoCommitSeconds * outputRate * 2)) + : 0; this.streams.set(dictationId, { dictationId, @@ -234,6 +257,7 @@ export class DictationStreamManager { receivedChunks: new Map(), nextSeqToForward: 0, ackSeq: -1, + autoCommitBytes, bytesSinceCommit: 0, peakSinceCommit: 0, committedSegmentIds: [], @@ -290,6 +314,13 @@ export class DictationStreamManager { state.debugAudioChunks.push(resampled); state.bytesSinceCommit += resampled.length; state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled)); + try { + this.maybeAutoCommitDictationSegment(params.dictationId, state); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + void this.failAndCleanupDictationStream(params.dictationId, message, true); + return; + } if (state.debugChunkWriter) { void state.debugChunkWriter.writeChunk(seq, resampled).catch((err) => { @@ -451,6 +482,42 @@ export class DictationStreamManager { this.streams.delete(dictationId); } + private maybeAutoCommitDictationSegment(dictationId: string, state: DictationStreamState): void { + if (state.finishRequested) { + return; + } + if (state.autoCommitBytes <= 0 || state.bytesSinceCommit < state.autoCommitBytes) { + return; + } + if (state.peakSinceCommit < DICTATION_SILENCE_PEAK_THRESHOLD) { + this.logger.debug( + { + dictationId, + autoCommitBytes: state.autoCommitBytes, + bytesSinceCommit: state.bytesSinceCommit, + peakSinceCommit: state.peakSinceCommit, + }, + "Dictation auto-segment: clearing silence-only segment" + ); + state.stt.clear(); + state.bytesSinceCommit = 0; + state.peakSinceCommit = 0; + return; + } + + this.logger.debug( + { + dictationId, + autoCommitBytes: state.autoCommitBytes, + bytesSinceCommit: state.bytesSinceCommit, + }, + "Dictation auto-segment: committing buffered audio" + ); + state.bytesSinceCommit = 0; + state.peakSinceCommit = 0; + state.stt.commit(); + } + private maybeSealDictationStreamFinish(dictationId: string): void { const state = this.streams.get(dictationId); if (!state) { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d361880d6..4afe08043 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -433,7 +433,7 @@ export class Session { this.defaultLocalSpeechModelIds = dictation?.localModels?.defaultModelIds && dictation.localModels.defaultModelIds.length > 0 ? [...new Set(dictation.localModels.defaultModelIds)] - : ["parakeet-tdt-0.6b-v3-int8", "pocket-tts-onnx-int8"]; + : ["parakeet-tdt-0.6b-v2-int8", "pocket-tts-onnx-int8"]; this.registerVoiceSpeakHandler = voiceBridge?.registerVoiceSpeakHandler; this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; diff --git a/packages/server/src/server/speech/providers/local/runtime.ts b/packages/server/src/server/speech/providers/local/runtime.ts index 13bd4ef9d..6b57a9f0c 100644 --- a/packages/server/src/server/speech/providers/local/runtime.ts +++ b/packages/server/src/server/speech/providers/local/runtime.ts @@ -107,7 +107,7 @@ async function createLocalSttEngine(params: { }): Promise { const { modelId, modelsDir, logger } = params; - if (modelId === "parakeet-tdt-0.6b-v3-int8") { + if (modelId === "parakeet-tdt-0.6b-v3-int8" || modelId === "parakeet-tdt-0.6b-v2-int8") { const modelDir = getLocalSpeechModelDir(modelsDir, modelId); return { kind: "offline", diff --git a/packages/server/src/server/speech/providers/local/sherpa/model-catalog.ts b/packages/server/src/server/speech/providers/local/sherpa/model-catalog.ts index 084c89b9d..0537c665d 100644 --- a/packages/server/src/server/speech/providers/local/sherpa/model-catalog.ts +++ b/packages/server/src/server/speech/providers/local/sherpa/model-catalog.ts @@ -39,6 +39,16 @@ export const SHERPA_ONNX_MODEL_CATALOG = { description: "Streaming Paraformer (often strong accuracy; heavier).", aliases: ["paraformer"], }, + "parakeet-tdt-0.6b-v2-int8": { + kind: "stt-offline", + archiveUrl: + "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8.tar.bz2", + extractedDir: "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8", + requiredFiles: ["encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx", "tokens.txt"], + description: "NVIDIA Parakeet TDT v2 (offline NeMo transducer, English).", + aliases: ["parakeet-v2", "parakeet-tdt-v2"], + defaultFor: "stt", + }, "parakeet-tdt-0.6b-v3-int8": { kind: "stt-offline", archiveUrl: @@ -47,7 +57,6 @@ export const SHERPA_ONNX_MODEL_CATALOG = { requiredFiles: ["encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx", "tokens.txt"], description: "NVIDIA Parakeet TDT v3 (offline NeMo transducer, multilingual).", aliases: ["parakeet", "parakeet-v3", "parakeet-tdt"], - defaultFor: "stt", }, "kitten-nano-en-v0_1-fp16": { kind: "tts", diff --git a/packages/server/src/server/speech/speech-config-resolver.test.ts b/packages/server/src/server/speech/speech-config-resolver.test.ts index 99c5a955f..c609be12d 100644 --- a/packages/server/src/server/speech/speech-config-resolver.test.ts +++ b/packages/server/src/server/speech/speech-config-resolver.test.ts @@ -34,13 +34,13 @@ describe("resolveSpeechConfig", () => { modelsDir: path.join(paseoHome, "models", "local-speech"), autoDownload: true, models: { - dictationStt: "parakeet-tdt-0.6b-v3-int8", - voiceStt: "parakeet-tdt-0.6b-v3-int8", + dictationStt: "parakeet-tdt-0.6b-v2-int8", + voiceStt: "parakeet-tdt-0.6b-v2-int8", voiceTts: "pocket-tts-onnx-int8", }, }); - expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v3-int8"); - expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v3-int8"); + expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v2-int8"); + expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v2-int8"); expect(result.speech.local?.models.voiceTts).toBe("pocket-tts-onnx-int8"); }); @@ -120,8 +120,8 @@ describe("resolveSpeechConfig", () => { persisted, }); - expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v3-int8"); - expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v3-int8"); + expect(result.speech.local?.models.dictationStt).toBe("parakeet-tdt-0.6b-v2-int8"); + expect(result.speech.local?.models.voiceStt).toBe("parakeet-tdt-0.6b-v2-int8"); expect(result.speech.local?.models.voiceTts).toBe("pocket-tts-onnx-int8"); }); });