From 19faa8b088e0cf292c5d3857fa342d51225740e2 Mon Sep 17 00:00:00 2001 From: Iulian Popescu Date: Wed, 9 Sep 2026 18:05:31 +0300 Subject: [PATCH] feat: speech-to-text for pi via NVIDIA Speech NIM realtime ASR - asr_transcribe / asr_health tools + the /asr command, with live partials, per-word timings, transcript export and TUI rendering - zero-dependency realtime client (https + Node 22 global WebSocket): mint transcription session -> subprotocol handshake -> event loop -> quiescence - decodes anything ffmpeg supports (and native 16k mono/stereo pcm16/pcm_f32le wav) to the 16 kHz mono PCM16 the NIM mandates - credentials via ASR_API_KEY / ASR_NIM_URL or ~/.pi/agent/asr.json; never in source - README documents the six places the published API reference diverges from the deployed protocol (commit vs done, cumulative deltas, double .completed, HTTP 403 at upgrade, ?token= not rejected, model reset on session.update) --- .gitignore | 10 + README.md | 95 ++++++++++ extensions/asr.ts | 303 ++++++++++++++++++++++++++++++ package.json | 29 +++ src/client.ts | 456 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 893 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 extensions/asr.ts create mode 100644 package.json create mode 100644 src/client.ts diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..30cce9c --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +node_modules/ +package-lock.json +yarn-error.log +*.wav +*.pcm +*.mp3 +*.mp4 +asr.json +.DS_Store +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..cec8cb7 --- /dev/null +++ b/README.md @@ -0,0 +1,95 @@ +# pi-asr + +Speech-to-text for the **pi** coding agent, backed by an NVIDIA Speech NIM realtime streaming ASR +endpoint (Riva-style realtime protocol, Parakeet/Nemotron models). + +Zero runtime dependencies: `https`, the Node 22 global `WebSocket`, and `ffmpeg` for container/codec +decoding. No Python, no virtualenvs, no local model weights. + +## Install + +```bash +# from git over ssh (uses your ~/.ssh/config) +pi install git:git@git.iaminsane.online:iulianp/pi-asr.git + +# or from a local checkout +pi install /path/to/pi-asr + +# or try it without installing +pi -e /path/to/pi-asr +``` + +## Configure + +The API key is never read from source. Priority: explicit option, then env var, then config file. + +```bash +export ASR_API_KEY=sk-... +export ASR_NIM_URL=https://your-host/openai/streaming-asr # optional, has a default +``` + +or drop a config file (mode 0600 recommended) at `~/.pi/agent/asr.json` (override with `$ASR_CONFIG`): + +```json +{ + "apiKey": "sk-...", + "baseUrl": "https://your-host/openai/streaming-asr" +} +``` + +`baseUrl` is the HTTPS base. The WebSocket URL is derived from it: `wss://` + the same host/path + +`/realtime?intent=transcription`. + +## What you get + +| surface | what it does | +| --- | --- | +| `asr_transcribe` | transcribe a local audio/video file; live partials while it streams; optional per-word timings | +| `asr_health` | `health/live` + `health/ready` + a minted session, so you see the model/language/sample-rate the server will actually use | +| `/asr [--lang xx-YY] [--words] [--speed N] [--start S] [--out f.txt]` | human-facing transcript viewer | + +```bash +pi -p "use the asr_transcribe tool on ./meeting.m4a with words=true and save to meeting.txt" +``` + +Accepted input: anything ffmpeg can decode (`wav` s16le/f32le, `mp3`, `m4a`, `aac`, `ogg`, `opus`, +`flac`, `mp4`, `mov`, `mkv`), resampled to the 16 kHz mono PCM16 the NIM mandates. Without ffmpeg, +16 kHz mono wav still works. + +## Layout + +```text +extensions/asr.ts pi registration: tools, TUI rendering, /asr command +src/client.ts RealtimeAsrClient: mint, WS subprotocol handshake, event loop, decode +``` + +## Protocol notes (things the published API reference gets wrong) + +Hard-won while building this against a real deployment. The client handles all of it. + +1. `input_audio_buffer.done` does **not** flush the decoder — you must send + `input_audio_buffer.commit` first, otherwise the server returns zero results. +2. `conversation.item.input_audio_transcription.delta` carries the **cumulative** segment text, not + an increment — rendering it incrementally duplicates every word. +3. A segment emits **two** `.completed` frames; the second has an empty `words_info`. The documented + `is_last_result` flag is never `true` on this model, so completion is detected by event + quiescence (no new event for ~700 ms after `done`) rather than by a terminal flag. +4. Auth failures surface as **HTTP 403 at the upgrade handshake**, not WS close 1008. A tokenless + handshake gets `101 Switching Protocols` with no echoed subprotocol, which Node's global + `WebSocket` treats as fatal — so the minted token is mandatory, not optional. +5. The documented `?token=` query param is **not** rejected (the docs claim it is). Do not put tokens + in URLs: they leak into access logs. The token goes in an extra `Sec-WebSocket-Protocol` value. +6. A `session.update` that omits `model` silently resets it to a model the server has not loaded + (`conformer`), which yields an empty transcript. Echo the minted session config back verbatim. + +## Latency + +mint ~120 ms, handshake ~20 ms, first partial ~360 ms, final ≈ audio duration + 0.5 s. The realtime +decoder is real-time bound, so `speed=2` (ffmpeg `atempo`) is the only way to beat wall-clock. + +## Dev + +```bash +pi install /path/to/pi-asr # registers the checkout; edit, then restart pi +pi --version # fast smoke test: prints extension load errors, if any +``` diff --git a/extensions/asr.ts b/extensions/asr.ts new file mode 100644 index 0000000..83540ad --- /dev/null +++ b/extensions/asr.ts @@ -0,0 +1,303 @@ +/** + * pi ASR extension — speech-to-text through the NVIDIA Speech NIM realtime WebSocket API. + * + * Tools: + * asr_transcribe transcribe an audio/video file (wav, mp3, m4a, aac, ogg, flac, opus, mp4, mov, mkv) + * asr_health pre-flight: health/live + health/ready + the model the NIM actually advertises + * Command: + * /asr [--lang xx-YY] [--words] [--speed N] [--start S] [--out transcript.txt] + * + * Config via env (built-in defaults otherwise): + * ASR_NIM_URL https://ai.iaminsane.online/openai/streaming-asr + * ASR_API_KEY key sent as `Authorization: Bearer ...` on every HTTPS/WS call + */ + +import { existsSync, statSync } from "node:fs"; +import { mkdir, writeFile } from "node:fs/promises"; +import { dirname, isAbsolute, relative, resolve } from "node:path"; +import type { AgentToolResult, ExtensionAPI, ExtensionContext } from "@mariozechner/pi-coding-agent"; +import { Text } from "@mariozechner/pi-tui"; +import { Type } from "@sinclair/typebox"; +import { + DEFAULT_BASE_URL, RealtimeAsrClient, hasFfmpeg, loadPcm16, + type TranscriptResult, type WordInfo, +} from "../src/client.js"; + +interface TranscribeParams { + path: string; + language?: string; + model?: string; + words?: boolean; + punctuation?: boolean; + start?: number; + speed?: number; + pace?: boolean; + out?: string; +} + +interface AsrDetails { + path: string; + text: string; + seconds: number; + wallMs: number; + model: string | null; + sessionId: string; + wordCount: number; + segments: { transcript: string; isLast: boolean }[]; + words?: WordInfo[]; + outPath?: string; + partial?: string; +} + +const wordCount = (t: string): number => (t.trim() ? t.trim().split(/\s+/).length : 0); +const rel = (p: string, cwd: string): string => { + const r = relative(cwd, p); + return r && !r.startsWith("..") && !isAbsolute(r) ? r : p; +}; + +export default function asrExtension(pi: ExtensionAPI) { + async function transcribeFile( + rawPath: string, + params: TranscribeParams, + cwd: string, + signal: AbortSignal | undefined, + onDelta?: (t: string) => void, + ): Promise<{ result: TranscriptResult; wallMs: number; absPath: string }> { + const absPath = resolve(cwd, rawPath.replace(/^@+/, "")); + if (!existsSync(absPath)) throw new Error(`no such file: ${absPath}`); + if (statSync(absPath).isDirectory()) throw new Error(`${absPath} is a directory, expected an audio file`); + + const client = new RealtimeAsrClient({ + language: params.language ?? "en-US", + model: params.model || null, + punctuation: params.punctuation ?? true, + wordOffsets: params.words ?? false, + }); + + const t0 = Date.now(); + const pcm = await loadPcm16(absPath, { + sampleRate: 16000, channels: 1, + start: params.start ?? 0, speed: params.speed ?? 1, + }); + if (!pcm.length) throw new Error(`decoded ${absPath} to zero bytes of audio (corrupt or empty track?)`); + + const result = await client.transcribe(pcm, { pace: params.pace ?? true, frameMs: 100, onDelta, signal }); + if (signal?.aborted) throw new Error("transcription cancelled"); + if (result.failed.length) { + throw new Error(`server reported a transcription failure: ${JSON.stringify(result.failed[0]).slice(0, 300)}`); + } + return { result, wallMs: Date.now() - t0, absPath }; + } + + async function runTranscribe( + params: TranscribeParams, + ctx: ExtensionContext, + onUpdate?: (r: AgentToolResult) => void, + signal?: AbortSignal, + ): Promise> { + if (!params.path?.trim()) throw new Error("`path` is required"); + if (signal?.aborted) throw new Error("transcription cancelled"); + let lastPartial = ""; + const started = Date.now(); + + const { result, wallMs, absPath } = await transcribeFile(params.path.trim(), params, ctx.cwd, signal, (t) => { + if (!t || t === lastPartial) return; + lastPartial = t; + onUpdate?.({ + content: [{ type: "text", text: t }], + details: { + path: rel(absPath, ctx.cwd), text: t, seconds: 0, wallMs: Date.now() - started, + model: null, sessionId: "", wordCount: wordCount(t), segments: [], partial: t, + }, + }); + }); + + const outPath = params.out ? resolve(ctx.cwd, params.out.replace(/^@+/, "")) : undefined; + if (outPath) { await mkdir(dirname(outPath), { recursive: true }); await writeFile(outPath, `${result.text}\n`, "utf-8"); } + + const details: AsrDetails = { + path: rel(absPath, ctx.cwd), + text: result.text, + seconds: result.seconds, + wallMs, + model: result.model, + sessionId: result.sessionId, + wordCount: wordCount(result.text), + segments: result.segments, + ...(params.words ? { words: result.words } : {}), + ...(outPath ? { outPath: rel(outPath, ctx.cwd) } : {}), + }; + + const summary = `${result.text || "(empty transcript)"}\n\n— ${details.wordCount} words, ${result.seconds}s audio, ${wallMs}ms wall, model=${result.model ?? "?"}${outPath ? `, wrote ${details.outPath}` : ""}`; + const timings = params.words && result.words.length + ? `\n\nword timings:\n${result.words.map(w => ` ${(+w.start_time).toFixed(2).padStart(7)} → ${(+w.end_time).toFixed(2).padStart(7)} ${w.word}${w.confidence != null ? ` (${(+w.confidence).toFixed(2)})` : ""}`).join("\n")}` + : ""; + return { content: [{ type: "text", text: summary + timings }], details }; + } + + /* ---------------------------------------------------------------- tools -- */ + pi.registerTool({ + name: "asr_transcribe", + label: "ASR transcribe", + description: "Speech-to-text a local audio or video file via the NVIDIA Speech NIM realtime streaming endpoint (Parakeet/Nemotron streaming ASR). Accepts wav, mp3, m4a, aac, ogg, opus, flac, mp4, mov, mkv - anything ffmpeg can decode; non-16kHz/mono input is resampled automatically. Returns the final transcript, optionally with per-word timings.", + promptSnippet: "Transcribe an audio or video file to text (speech-to-text) with the streaming ASR server", + promptGuidelines: [ + "Use asr_transcribe for any speech-to-text request (voice notes, meeting recordings, podcast clips, screen recordings) instead of hand-rolling ffmpeg/python.", + "asr_transcribe is real-time bound: budget roughly the duration of the audio. Pass speed=2 to halve wall time when slight accuracy loss is fine, or start=N to skip leading silence.", + "If asr_transcribe returns an empty transcript, run asr_health to confirm the server is up and the language/model are right.", + ], + parameters: Type.Object({ + path: Type.String({ description: "Audio/video file, relative to cwd or absolute (a leading @ is stripped)" }), + language: Type.Optional(Type.String({ description: "BCP-47 language tag, default en-US" })), + model: Type.Optional(Type.String({ description: "Override the ASR model name (default: whatever the server advertises)" })), + words: Type.Optional(Type.Boolean({ description: "Include per-word timings + confidence (default false)" })), + punctuation: Type.Optional(Type.Boolean({ description: "Server-side automatic punctuation (default true)" })), + start: Type.Optional(Type.Number({ description: "Skip the first N seconds" })), + speed: Type.Optional(Type.Number({ description: "Playback rate 0.5..4 (ffmpeg atempo); >1 is faster, slightly less accurate" })), + pace: Type.Optional(Type.Boolean({ description: "Stream at 1x realtime pacing (default true)" })), + out: Type.Optional(Type.String({ description: "Also write the plain transcript to this path" })), + }), + + async execute(_id, params: TranscribeParams, signal, onUpdate, ctx): Promise> { + return runTranscribe(params, ctx, onUpdate as any, signal); + }, + + renderCall(args: any, theme: any) { + const bits = [ + theme.fg("toolTitle", theme.bold("asr_transcribe ")), + theme.fg("muted", String(args?.path ?? "")), + ]; + if (args?.language && args.language !== "en-US") bits.push(theme.fg("dim", ` ${args.language}`)); + if (args?.speed && args.speed !== 1) bits.push(theme.fg("dim", ` ${args.speed}x`)); + if (args?.words) bits.push(theme.fg("dim", " +words")); + if (args?.out) bits.push(theme.fg("dim", ` → ${args.out}`)); + return new Text(bits.join(""), 0, 0); + }, + + renderResult(result: any, options: any, theme: any) { + const d: AsrDetails | undefined = result?.details; + if (!d) return new Text("", 0, 0); + if (options?.isPartial && d.partial) return new Text(theme.fg("warning", `… ${d.partial}`), 0, 0); + const head = theme.fg("muted", `${d.path} ${d.seconds}s audio → ${d.wallMs}ms ${d.model ?? ""}`.trimEnd()); + const body = d.text + ? theme.fg("success", d.text) + : theme.fg("warning", "(empty transcript — check --lang / model, or run asr_health)"); + let out = `${head}\n${body}`; + if (options?.expanded && d.words?.length) { + out += "\n" + d.words.slice(0, 50).map(w => theme.fg("dim", ` ${(+w.start_time).toFixed(2).padStart(7)} → ${(+w.end_time).toFixed(2).padStart(7)} ${w.word}`)).join("\n"); + } else if (d.wordCount) { + out += " " + theme.fg("dim", `(${d.wordCount} words)`); + } + if (options?.expanded && d.outPath) out += "\n" + theme.fg("dim", ` wrote ${d.outPath}`); + return new Text(out, 0, 0); + }, + }); + + pi.registerTool({ + name: "asr_health", + label: "ASR health", + description: + "Check the ASR server before/while debugging: GET health/live + health/ready, and mint a realtime session to " + + "report the model, language and sample rate the server will actually use. Also verifies ffmpeg availability " + + "and reports the API key being used (masked).", + promptSnippet: "Check ASR server health and the currently advertised model/sample-rate config", + parameters: Type.Object({}), + + async execute(_id, _params, _signal, _onUpdate, ctx): Promise> { + const client = new RealtimeAsrClient(); + const [live, ready, minted, ffmpeg] = await Promise.all([ + client.health("live", 10000).catch((e: any) => ({ error: e?.message ?? String(e) })), + client.health("ready", 10000).catch((e: any) => ({ error: e?.message ?? String(e) })), + client.mint(10000).catch((e: any) => ({ error: e?.message ?? String(e) })), + hasFfmpeg(), + ]); + const ok = live?.status === "live" && ready?.status === "ready" && !minted?.error; + const key = client.cfg.apiKey; + const secs = minted?.client_secret ? Math.max(0, (minted.client_secret.expires_at ?? 0) - Math.floor(Date.now() / 1000)) : 0; + const text = [ + `baseUrl ${process.env.ASR_NIM_URL ?? DEFAULT_BASE_URL}`, + `apiKey ${key ? `${key.slice(0, 5)}…${key.slice(-3)} (${key.length} chars)` : "NOT SET"}`, + `health/live ${JSON.stringify(live)}`, + `health/ready ${JSON.stringify(ready)}`, + minted?.error + ? `mint ERROR ${minted.error}` + : [ + `mint ${minted.id}`, + ` model ${minted.input_audio_transcription?.model}`, + ` language ${minted.input_audio_transcription?.language}`, + ` audio ${minted.input_audio_format}, ${minted.input_audio_params?.sample_rate_hz}Hz x${minted.input_audio_params?.num_channels}`, + ` ws token single-use, expires in ${secs}s (mint immediately before each connect)`, + ].join("\n"), + `ffmpeg ${ffmpeg ? "available (any container/codec accepted)" : "MISSING — only 16kHz mono wav files will work"}`, + `note the realtime decoder runs at ~1x realtime; a 10min file takes ~10min (speed=2 halves it)`, + ].join("\n"); + if (!ok) throw new Error(`ASR server is not usable:\n${text}`); + return { content: [{ type: "text", text }], details: { ok, live, ready, minted, ffmpeg } }; + }, + + renderResult(result: any, _options: any, theme: any) { + const d: any = result?.details; + if (!d) return new Text("", 0, 0); + const m = d.minted?.input_audio_transcription; + const p = d.minted?.input_audio_params; + return new Text([ + d.ok ? theme.fg("success", "✓ ASR server live") : theme.fg("error", "✗ ASR server not usable"), + theme.fg("muted", ` model ${m?.model ?? d.minted?.error ?? "?"}`), + theme.fg("dim", ` ${p?.sample_rate_hz ?? "?"}Hz x${p?.num_channels ?? "?"} ${d.minted?.input_audio_format ?? ""} · ${m?.language ?? "?"} · ffmpeg ${d.ffmpeg ? "ok" : "MISSING"}`), + ].join("\n"), 0, 0); + }, + }); + + /* --------------------------------------------------------------- /asr --- */ + pi.registerCommand("asr", { + description: "Transcribe an audio/video file: /asr [--lang xx-YY] [--words] [--speed N] [--start S] [--out file.txt]", + handler: async (args: string, ctx: ExtensionContext) => { + const toks = args.trim().split(/\s+/).filter(Boolean); + if (!toks.length) { + ctx.ui.notify("usage: /asr [--lang xx-YY] [--words] [--speed N] [--start S] [--out file.txt]", "info"); + return; + } + const p: TranscribeParams = { path: "" }; + for (let i = 0; i < toks.length; i++) { + const t = toks[i]; + if (t === "--lang") p.language = toks[++i]; + else if (t === "--words") p.words = true; + else if (t === "--speed") p.speed = Number.parseFloat(toks[++i]) || 1; + else if (t === "--start") p.start = Number.parseFloat(toks[++i]) || 0; + else if (t === "--out") p.out = toks[++i]; + else if (!p.path) p.path = t; + } + if (!p.path) { ctx.ui.notify("no file given", "error"); return; } + + let last = ""; + ctx.ui.setStatus("asr", `transcribing ${p.path} …`); + try { + const r = await runTranscribe(p, ctx); + const d = r.details; + last = d.text || "(empty transcript)"; + ctx.ui.notify(`${d.path}: ${d.wordCount} words, ${d.seconds}s audio in ${d.wallMs}ms`, "info"); + if (ctx.hasUI) { + await ctx.ui.custom((_tui, theme, _kb, done) => { + const body = new Text([ + theme.fg("toolTitle", theme.bold(` transcribed ${d.path} `)), + theme.fg("muted", `${d.seconds}s audio → ${d.wallMs}ms · ${d.model ?? "?"} · ${d.wordCount} words`), + "", + theme.fg(d.text ? "success" : "warning", d.text || "(empty transcript)"), + ...(d.words?.length ? [theme.fg("dim", `\n${d.words.slice(0, 12).map(w => `${(+w.start_time).toFixed(2).padStart(7)} → ${(+w.end_time).toFixed(2).padStart(7)} ${w.word}`).join("\n")}`)] : []), + "", + theme.fg("dim", " enter/esc to close"), + ].join("\n"), 1, 1); + body.onKey = (key: string) => { if (key === "return" || key === "escape" || key === " ") { done(undefined); return true; } return true; }; + return body; + }); + } else { + console.log(last); + } + } catch (e: any) { + ctx.ui.notify(`ASR failed: ${e?.message ?? e}`, "error"); + } finally { + ctx.ui.setStatus("asr", undefined); + } + }, + }); +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..bfd546e --- /dev/null +++ b/package.json @@ -0,0 +1,29 @@ +{ + "name": "pi-asr", + "version": "0.1.0", + "description": "Speech-to-text for the pi coding agent: asr_transcribe / asr_health tools and the /asr command, backed by an NVIDIA Speech NIM realtime streaming ASR endpoint (Parakeet/Nemotron).", + "keywords": [ + "pi-package", + "pi-extension", + "asr", + "speech-to-text", + "transcription", + "nvidia-nim", + "streaming-asr" + ], + "type": "module", + "peerDependencies": { + "@mariozechner/pi-ai": "*", + "@mariozechner/pi-coding-agent": "*", + "@mariozechner/pi-tui": "*", + "@sinclair/typebox": "*" + }, + "pi": { + "extensions": [ + "./extensions" + ] + }, + "engines": { + "node": ">=22" + } +} diff --git a/src/client.ts b/src/client.ts new file mode 100644 index 0000000..528cfe1 --- /dev/null +++ b/src/client.ts @@ -0,0 +1,456 @@ +/** + * NVIDIA Speech NIM realtime ASR client (Riva realtime WebSocket API). + * Zero dependencies, Node >= 22 (global WebSocket). + * + * Docs: https://docs.nvidia.com/nim/speech/latest/reference/api-references/asr/realtime-asr.html + * + * Verified against the live deployment (behind Traefik `replacePathRegex` + * ^/openai/streaming-asr/(.*) -> /v1/$1 plus the `qwen-audio-api-key` middleware): + * + * 1. GET {base}/health/live | /health/ready Bearer key -> {status:'live'|'ready'} + * 2. POST {base}/realtime/transcription_sessions Bearer key -> session + client_secret.value + * 3. WS {base}/realtime?intent=transcription + * Sec-WebSocket-Protocol: "realtime, realtime-token." (never a URL query param) + * 4. transcription_session.update -> transcription_session.updated + * 5. input_audio_buffer.append (base64 pcm16) * N + * 6. input_audio_buffer.commit -> input_audio_buffer.committed <-- REQUIRED, see below + * 7. server emits conversation.item.input_audio_transcription.delta|.completed|.failed + * + * Three behaviours differ from the published docs and are handled here: + * a) `input_audio_buffer.done` alone never flushes the decoder -> zero results. An explicit + * `input_audio_buffer.commit` is what triggers recognition. + * b) `delta` is CUMULATIVE (each frame repeats the whole utterance so far), not incremental. + * c) `is_last_result` never arrives as true, so completion is detected by event quiescence. + * And the server sends two `.completed` frames per segment: the first carries `words_info`, + * the second has empty `words_info` -> the word-bearing one must be kept. + * d) Auth failures arrive as an HTTP **403 during the upgrade handshake**, not as WebSocket + * close code 1008 like the docs claim. A tokenless handshake returns 101 with NO echoed + * subprotocol, which Node's global WebSocket (undici) treats as a fatal error + * ("Server did not respond with sent protocols") -> the minted token is mandatory in Node. + */ + +import { spawn } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import { extname } from "node:path"; + +export const EV = { + conversationCreated: "conversation.created", + sessionUpdated: "transcription_session.updated", + bufferCommitted: "input_audio_buffer.committed", + bufferCleared: "input_audio_buffer.cleared", + delta: "conversation.item.input_audio_transcription.delta", + completed: "conversation.item.input_audio_transcription.completed", + failed: "conversation.item.input_audio_transcription.failed", + error: "error", +} as const; + +export interface AsrConfig { + baseUrl: string; + apiKey: string; + language: string; + model: string | null; + sampleRate: number; + channels: number; + punctuation: boolean; + wordOffsets: boolean; +} + +export const DEFAULT_BASE_URL = "https://ai.iaminsane.online/openai/streaming-asr"; + +export interface AsrFileConfig { baseUrl?: string; apiKey?: string } + +/** Optional on-disk config so the key never has to live in source: ~/.pi/agent/asr.json (or $ASR_CONFIG). */ +export function configFilePath(): string { + return process.env.ASR_CONFIG ?? [homedir(), ".pi", "agent", "asr.json"].join("/"); +} + +let cachedFileConfig: AsrFileConfig | null = null; +export function readConfigFile(): AsrFileConfig { + if (cachedFileConfig) return cachedFileConfig; + try { + cachedFileConfig = JSON.parse(readFileSync(configFilePath(), "utf-8")) as AsrFileConfig; + } catch { + cachedFileConfig = {}; + } + return cachedFileConfig; +} + +/** Env wins, so the key can live outside the file: ASR_API_KEY / ASR_NIM_URL. */ +export function defaultConfig(overrides: Partial = {}): AsrConfig { + return { + baseUrl: (overrides.baseUrl ?? process.env.ASR_NIM_URL ?? readConfigFile().baseUrl ?? DEFAULT_BASE_URL).replace(/\/+$/, ""), + apiKey: overrides.apiKey ?? process.env.ASR_API_KEY ?? readConfigFile().apiKey ?? "", + language: overrides.language ?? "en-US", + model: overrides.model ?? null, // null => whatever the NIM advertises on mint + sampleRate: overrides.sampleRate ?? 16000, + channels: overrides.channels ?? 1, + punctuation: overrides.punctuation ?? true, + wordOffsets: overrides.wordOffsets ?? false, + }; +} + +const uid = (p: string = "event"): string => `${p}_${crypto.randomUUID()}`; +const sleep = (ms: number): Promise => new Promise(r => setTimeout(r, ms)); + +export interface WordInfo { + word: string; + start_time: number; + end_time: number; + confidence?: number; + speaker_tag?: number; +} + +export interface TranscriptResult { + text: string; + segments: { transcript: string; isLast: boolean }[]; + words: WordInfo[]; + failed: unknown[]; + seconds: number; + model: string | null; + sessionId: string; + protocol: string | null; +} + +type Listener = (ev: any) => void; + +export class RealtimeAsrClient { + cfg: AsrConfig; + verbose: boolean; + private listeners: Map> = new Map(); + + constructor(opts: Partial & { verbose?: boolean } = {}) { + this.cfg = defaultConfig(opts); + this.verbose = opts.verbose ?? false; + } + + log(...a: unknown[]): void { + if (this.verbose) console.error("[asr]", ...a); + } + + on(type: string, fn: Listener): () => void { + if (!this.listeners.has(type)) this.listeners.set(type, new Set()); + this.listeners.get(type)!.add(fn); + return () => this.listeners.get(type)?.delete(fn); + } + + private emit(ev: any): void { + for (const fn of this.listeners.get(ev?.type) ?? []) { try { fn(ev); } catch {} } + for (const fn of this.listeners.get("*") ?? []) { try { fn(ev); } catch {} } + } + + private headers(extra: Record = {}): Record { + const h: Record = { ...extra }; + // Gateway middleware `qwen-audio-api-key` -> 403 {"Invalid or missing API Key"} without it. + if (this.cfg.apiKey) h.Authorization = `Bearer ${this.cfg.apiKey}`; + return h; + } + + get wsUrl(): string { + return `${this.cfg.baseUrl.replace(/^http/, "ws")}/realtime?intent=transcription`; + } + + /** The gateway answers 403 {Invalid or missing API Key} without a key, which is cryptic; fail with instructions instead. */ + private requireKey(): void { + if (this.cfg.apiKey) return; + throw new Error( + "No ASR API key configured. Set ASR_API_KEY, or write { \"apiKey\": \"sk-...\" } to " + configFilePath() + + " (or point ASR_CONFIG at your file). Base URL: " + this.cfg.baseUrl, + ); + } + + /** GET {base}/health/live | /health/ready */ + async health(which: "live" | "ready" = "live", timeout: number = 15000): Promise { + this.requireKey(); + const r = await fetch(`${this.cfg.baseUrl}/health/${which}`, { + headers: this.headers(), + signal: AbortSignal.timeout?.(timeout), + }); + const t = await r.text(); + if (!r.ok) throw new Error(`health/${which} -> HTTP ${r.status}: ${t.slice(0, 200)}`); + try { return JSON.parse(t); } catch { return { raw: t }; } + } + + /** POST {base}/realtime/transcription_sessions — single-use token, 60s TTL, so mint right before connecting. */ + async mint(timeout: number = 15000): Promise { + this.requireKey(); + const r = await fetch(`${this.cfg.baseUrl}/realtime/transcription_sessions`, { + method: "POST", + headers: this.headers({ "Content-Type": "application/json" }), + signal: AbortSignal.timeout?.(timeout), + }); + const t = await r.text(); + if (!r.ok) { + const why = r.status === 403 || r.status === 401 ? "gateway rejected the API key" : `HTTP ${r.status}`; + throw new Error(`mint session failed: ${why}: ${t.slice(0, 200)}`); + } + return JSON.parse(t); + } + + /** + * Payload for transcription_session.update. The server substitutes *defaults* for anything + * omitted (model silently becomes "conformer", which is not loaded here and yields zero + * results), so echo back the minted session and apply overrides on top. + */ + sessionConfig(minted: any = {}, overrides: Record = {}): Record { + const base: Record = { + modalities: minted.modalities ?? ["text"], + input_audio_format: "pcm16", + input_audio_transcription: { + language: this.cfg.language, + model: this.cfg.model ?? minted.input_audio_transcription?.model, + prompt: minted.input_audio_transcription?.prompt ?? "", + }, + input_audio_params: { + sample_rate_hz: this.cfg.sampleRate, + num_channels: this.cfg.channels, + }, + recognition_config: { + ...(minted.recognition_config ?? {}), + max_alternatives: 1, + enable_automatic_punctuation: !!this.cfg.punctuation, + enable_word_time_offsets: !!this.cfg.wordOffsets, + enable_profanity_filter: false, + enable_verbatim_transcripts: false, + }, + speaker_diarization: minted.speaker_diarization ?? { enable_speaker_diarization: false, max_speaker_count: 8 }, + word_boosting: minted.word_boosting ?? { enable_word_boosting: false, word_boosting_list: [] }, + endpointing_config: minted.endpointing_config ?? { + start_history: 0, start_threshold: 0, stop_history: 0, stop_threshold: 0, + stop_history_eou: 0, stop_threshold_eou: 0, + }, + }; + return { ...base, ...overrides }; + } + + /** Open the socket. `onDelta` gets every cumulative partial, `onFinal` each completed segment. */ + async open(opts: { token?: string; timeout?: number; onDelta?: (text: string) => void; onFinal?: (ev: any) => void } = {}): Promise { + const token = opts.token ?? null; + const protocols = token ? ["realtime", `realtime-token.${token}`] : ["realtime"]; + const ws = new WebSocket(this.wsUrl, { protocols, headers: this.headers() }); + const st: SocketHandle = { events: [], closed: null, protocol: null }; + + await new Promise((res, rej) => { + const t = setTimeout(() => { try { ws.close(); } catch {} rej(new Error("WebSocket handshake timed out")); }, opts.timeout ?? 20000); + ws.onopen = () => { clearTimeout(t); st.protocol = ws.protocol; res(); }; + ws.onerror = (e: any) => { + clearTimeout(t); + const msg = e?.message || e?.error?.message || e?.type || "unknown"; + rej(new Error(msg.includes("did not respond with sent protocols") + ? `${msg} — the server accepted a tokenless handshake but echoed no subprotocol; always mint a client_secret first` + : `WebSocket error: ${msg}`)); + }; + ws.onclose = (e: any) => { st.closed = { code: e?.code, reason: e?.reason || "" }; this.log("closed", st.closed.code, st.closed.reason); }; + ws.onmessage = (m: any) => { + const raw = m?.data ?? m; + let ev: any; + try { ev = JSON.parse(typeof raw === "string" ? raw : Buffer.from(raw as any).toString("utf-8")); } catch { return; } + st.events.push(ev); + this.log("<-", ev.type); + if (ev.type === EV.delta && opts.onDelta) opts.onDelta(ev.delta ?? ""); + if (ev.type === EV.completed && opts.onFinal) opts.onFinal(ev); + this.emit(ev); + }; + }); + + st.send = (obj: Record) => ws.send(JSON.stringify({ event_id: uid(), ...obj })); + st.expect = (type: string, ms: number = 20000) => { + const hit = st.events.find(e => e.type === type); + if (hit) return Promise.resolve(hit); + return new Promise((res, rej) => { + const t = setTimeout(() => rej(new Error(`timeout waiting for "${type}" (saw: ${st.events.map(e => e.type).join(",") || "nothing"})`)), ms); + const off = this.on(type, ev => { clearTimeout(t); res(ev); }); + if (st.closed) { clearTimeout(t); off(); rej(new Error(`socket closed (code ${st.closed.code}) while waiting for "${type}"`)); } + }); + }; + st.close = () => { try { ws.close(1000, "bye"); } catch {} }; + return st; + } + + /** Convenience wrapper: mint -> open -> configure (caller keeps the handle for live/mic streaming). */ + async openSession(opts: { session?: Record; onDelta?: (t: string) => void; onFinal?: (ev: any) => void } = {}): Promise<{ handle: SocketHandle; minted: any }> { + const minted = await this.mint(); + const handle = await this.open({ token: minted.client_secret?.value ?? null, onDelta: opts.onDelta, onFinal: opts.onFinal }); + handle.send({ type: "transcription_session.update", session: this.sessionConfig(minted, opts.session ?? {}) }); + await handle.expect(EV.sessionUpdated, 20000); + return { handle, minted }; + } + + /** + * One-shot transcribe of raw s16le pcm16 audio. + * `onDelta` receives cumulative partials (great for a live UI), return value has final text. + */ + async transcribe(pcm16: Buffer, opts: { + frameMs?: number; pace?: boolean; paceMs?: number; session?: Record; + quietMs?: number; emptyGraceMs?: number; finalTimeout?: number; tailMs?: number; + onDelta?: (t: string) => void; signal?: AbortSignal; + } = {}): Promise { + const buf = toBuffer(pcm16); + const frameMs = opts.frameMs ?? 100; + const frameBytes = Math.max(2, Math.round((this.cfg.sampleRate * frameMs) / 1000) * 2 * this.cfg.channels); + const secs = buf.length / (this.cfg.sampleRate * 2 * this.cfg.channels); + const finalTimeout = opts.finalTimeout ?? Math.max(30000, Math.round(secs * 1000) + 30000); + + const minted = await this.mint(); + if (opts.signal?.aborted) throw new Error("cancelled"); + const handle = await this.open({ token: minted.client_secret?.value ?? null }); + + const deltas: string[] = []; + const finals: Map = new Map(); + const failures: unknown[] = []; + const offs = [ + this.on(EV.delta, e => deltas.push(e.delta ?? "")), + this.on(EV.completed, (e: any) => { + // keep the word-bearing copy: the server sends a second, words-less duplicate per segment + const key = e.item_id ?? "_"; + const prev = finals.get(key); + const nw = e.words_info?.words?.length ?? 0; + const pw = prev?.words_info?.words?.length ?? 0; + finals.set(key, prev && pw > nw && prev.transcript ? { ...e, words_info: prev.words_info } : e); + }), + this.on(EV.failed, (e: any) => failures.push(e)), + ]; + + try { + handle.send({ type: "transcription_session.update", session: this.sessionConfig(minted, opts.session ?? {}) }); + await handle.expect(EV.sessionUpdated, 20000); + + for (let i = 0; i < buf.length; i += frameBytes) { + if (opts.signal?.aborted) throw new Error("cancelled"); + handle.send({ + type: "input_audio_buffer.append", + audio: buf.subarray(i, Math.min(i + frameBytes, buf.length)).toString("base64"), + }); + if (opts.pace !== false) await sleep(opts.paceMs ?? frameMs); + } + + handle.send({ type: "input_audio_buffer.commit" }); // <-- the actual flush + const t0 = Date.now(), deadline = t0 + finalTimeout; + const quietMs = opts.quietMs ?? 1500, emptyGraceMs = opts.emptyGraceMs ?? 8000; + let lastCount = 0, lastChange = t0; + while (Date.now() < deadline) { + if (opts.signal?.aborted) break; + const n = finals.size; + if (n !== lastCount) { lastCount = n; lastChange = Date.now(); } + if (failures.length) break; + if (n > 0 && Date.now() - lastChange >= quietMs) break; + if (n === 0 && Date.now() - t0 >= emptyGraceMs) break; + await sleep(120); + } + + handle.send({ type: "input_audio_buffer.done" }); + await sleep(opts.tailMs ?? 600); + + const list = Array.from(finals.values()); + const text = list.length + ? list.map(f => f.transcript ?? "").filter(Boolean).join(" ").replace(/\s+/g, " ").trim() + : (deltas.at(-1) ?? "").replace(/\s+/g, " ").trim(); + + return { + text, + segments: list.map(f => ({ transcript: f.transcript ?? "", isLast: f.is_last_result ?? false })), + words: list.flatMap(f => f.words_info?.words ?? []), + failed: failures, + seconds: +secs.toFixed(3), + model: minted.input_audio_transcription?.model ?? null, + sessionId: minted.id ?? "", + protocol: handle.protocol, + }; + } finally { + offs.forEach(f => f()); + try { handle.send?.({ type: "input_audio_buffer.clear" }); } catch {} + handle.close(); + } + } +} + +export interface SocketHandle { + events: any[]; + closed: { code: number; reason: string } | null; + protocol: string | null; + send?: (obj: Record) => void; + expect?: (type: string, ms?: number) => Promise; + close?: () => void; +} + +function toBuffer(x: any): Buffer { + if (Buffer.isBuffer(x)) return x; + if (x instanceof Uint8Array) return Buffer.from(x); + if (x instanceof ArrayBuffer) return Buffer.from(new Uint8Array(x)); + if (x?.buffer instanceof ArrayBuffer) return Buffer.from(new Uint8Array(x.buffer)); + throw new Error("expected pcm16 audio as Buffer/Uint8Array/ArrayBuffer"); +} + +/* ------------------------------------------------------------------ audio -- */ + +/** Decode wav/mp3/m4a/aac/ogg/flac/mp4/video... to interleaved s16le at the target rate. Needs ffmpeg. */ +export async function decodeToPcm16( + path: string, + { sampleRate = 16000, channels = 1, start = 0, speed = 1 }: { sampleRate?: number; channels?: number; start?: number; speed?: number } = {}, +): Promise { + const args: string[] = ["-v", "error"]; + if (start > 0) args.push("-ss", String(start)); + args.push("-i", path); + if (speed !== 1 && speed > 0) { + // atempo only accepts 0.5..100; chain it for extremes + let rate = Math.max(0.25, Math.min(4, speed)); + const filters: string[] = []; + while (rate < 0.5) { filters.push("atempo=0.5"); rate /= 0.5; } + while (rate > 2) { filters.push("atempo=2.0"); rate /= 2.0; } + if (rate !== 1) filters.push(`atempo=${rate.toFixed(6)}`); + if (filters.length) args.push("-af", filters.join(",")); + } + args.push("-ac", String(channels), "-ar", String(sampleRate), "-f", "s16le", "-"); + const chunks: Buffer[] = []; + const { code, stderr } = await run("ffmpeg", args, b => chunks.push(b)); + const out = Buffer.concat(chunks); + if (!out.length) throw new Error(`ffmpeg produced no audio for ${path} (exit ${code}) ${stderr.slice(0, 300)}`.trim()); + return out; +} + +/** Native RIFF/EXTENSIBLE parser so plain 16k-mono wav files work without ffmpeg on PATH. */ +export async function readWavAsPcm16(path: string, sampleRate: number = 16000): Promise { + const buf = await readFile(path); + if (buf.toString("ascii", 0, 4) !== "RIFF" || buf.toString("ascii", 8, 12) !== "WAVE") throw new Error("not a wav file"); + let pos = 12, fmt: any = null, data: Buffer | null = null; + while (pos + 8 <= buf.length) { + const id = buf.toString("ascii", pos, pos + 4); + const size = buf.readUInt32LE(pos + 4); + const body = buf.subarray(pos + 8, Math.min(pos + 8 + size, buf.length)); + if (id === "fmt ") fmt = { format: body.readUInt16LE(0), channels: body.readUInt16LE(2), rate: body.readUInt32LE(4), bits: body.readUInt16LE(14) }; + else if (id === "data") data = body; + else if (id === "LIST" || id === "INFO" || id === "fact") { /* skip */ } + pos += 8 + size + (size % 2); + } + if (!fmt || !data) throw new Error("wav is missing fmt /data chunks"); + if (fmt.bits !== 16 || fmt.format !== 1) throw new Error(`wav is ${fmt.bits}bit/format${fmt.format}, need 16bit PCM (use ffmpeg) — falling back`); + if (fmt.channels !== 1) throw new Error(`wav has ${fmt.channels} channels, need mono (use ffmpeg)`); + if (fmt.rate !== sampleRate) throw new Error(`wav is ${fmt.rate}Hz, need ${sampleRate}Hz (use ffmpeg)`); + return data; +} + +/** PCM16 with ffmpeg-first strategy and a native wav fast path. */ +export async function loadPcm16(path: string, opts: { sampleRate?: number; channels?: number; start?: number; speed?: number } = {}): Promise { + const sampleRate = opts.sampleRate ?? 16000; + const channels = opts.channels ?? 1; + if (!opts.start && !opts.speed && extname(path).toLowerCase() === ".wav") { + try { return await readWavAsPcm16(path, sampleRate); } catch { /* fall through to ffmpeg */ } + } + return decodeToPcm16(path, { sampleRate, channels, start: opts.start ?? 0, speed: opts.speed ?? 1 }); +} + +export async function hasFfmpeg(): Promise { + try { await run("ffmpeg", ["-version"], () => {}); return true; } catch { return false; } +} + +function run(cmd: string, args: string[], onData: (b: Buffer) => void): Promise<{ code: number; stderr: string }> { + return new Promise((res, rej) => { + const child = spawn(cmd, args, { stdio: ["ignore", "pipe", "pipe"] }); + let stderr = ""; + child.stdout.on("data", onData); + child.stderr.on("data", b => { stderr += b.toString(); }); + child.on("error", e => rej(new Error((e as any).code === "ENOENT" ? `${cmd} is not installed / not on PATH` : e.message))); + child.on("close", code => res({ code: code ?? 0, stderr })); + }); +}