diff --git a/mcp/Dockerfile b/mcp/Dockerfile index b7d964095..b76bbbd88 100644 --- a/mcp/Dockerfile +++ b/mcp/Dockerfile @@ -13,7 +13,7 @@ FROM node:22-bookworm-slim WORKDIR /usr/src/app # Install dependencies -RUN apt update && apt install -y sed git git-lfs vim tree ripgrep wget curl jq python3 pandoc python3-pip python3-venv zip unzip poppler-utils ffmpeg && \ +RUN apt update && apt install -y sed git git-lfs vim tree ripgrep wget curl jq python3 pandoc python3-pip python3-venv zip unzip poppler-utils ffmpeg bzip2 && \ git lfs install --system && \ ln -s /usr/bin/python3 /usr/bin/python diff --git a/mcp/src/index.ts b/mcp/src/index.ts index b500e5bf0..4f7508f69 100644 --- a/mcp/src/index.ts +++ b/mcp/src/index.ts @@ -24,7 +24,7 @@ import * as l from "./graph/learnings.js"; import * as uploads from "./graph/uploads.js"; import * as gitree from "./gitree/routes.js"; import * as gitreeProposals from "./gitree/proposals.js"; -import { mountLab } from "./lab/mount.js"; +import { mountLab, attachLabAudio } from "./lab/mount.js"; import { loadModelPricing } from "./aieo/src/index.js"; import path from "path"; import { fileURLToPath } from "url"; @@ -447,6 +447,9 @@ const server = app.listen(port, host, () => { ); }); +// Dictation WebSocket for the lab UI (upgrades bypass Express). +attachLabAudio(server); + process.on("SIGTERM", () => void gracefulShutdown("SIGTERM")); process.on("SIGINT", () => void gracefulShutdown("SIGINT")); diff --git a/mcp/src/lab/mount.ts b/mcp/src/lab/mount.ts index ef6b15da6..082180043 100644 --- a/mcp/src/lab/mount.ts +++ b/mcp/src/lab/mount.ts @@ -1,7 +1,16 @@ import type { Express, Request, Response, NextFunction } from "express"; +import type { IncomingMessage, Server } from "node:http"; +import type { Duplex } from "node:stream"; import { getRequestListener } from "@hono/node-server"; +import { createAudioUpgradeHandler, type AudioUpgradeHandler } from "vein"; import { createLabVein } from "./createLabVein.js"; +/** The one lab vein, built on first use (HTTP request or dictation upgrade). */ +let labVeinP: ReturnType | null = null; +function labVein() { + return (labVeinP ??= createLabVein({ serveUi: true })); +} + /** * Bridge a (lazily-built) vein Hono app into Express. The instance is * created on the first request to its mount path, so mcp boot is never @@ -22,6 +31,23 @@ function bridge(factory: () => Promise<{ app: { fetch: any } }>) { }; } +/** Does this request carry the lab credential? (`labAuth` without the + * response side, so the WebSocket upgrade can apply the same rule.) */ +function labAuthorized(req: { header(name: string): string | undefined }): boolean { + const apiToken = process.env.API_TOKEN; + if (!apiToken) return true; + if (req.header("x-api-token") === apiToken) return true; + const header = req.header("authorization") ?? ""; + if (header.startsWith("Basic ")) { + const decoded = Buffer.from(header.slice(6), "base64").toString(); + const sep = decoded.indexOf(":"); + const user = decoded.slice(0, sep); + const pass = decoded.slice(sep + 1); + if (sep > 0 && user === "admin" && pass === apiToken) return true; + } + return false; +} + /** * Mount the single lab vein under `/lab` (API + run-streaming SSE). All * experiments share this one instance — they're groups of workflows @@ -44,17 +70,7 @@ function bridge(factory: () => Promise<{ app: { fetch: any } }>) { * rest of mcp for server-to-server callers. */ function labAuth(req: Request, res: Response, next: NextFunction): void { - const apiToken = process.env.API_TOKEN; - if (!apiToken) return next(); - if (req.header("x-api-token") === apiToken) return next(); - const header = req.header("authorization") ?? ""; - if (header.startsWith("Basic ")) { - const decoded = Buffer.from(header.slice(6), "base64").toString(); - const sep = decoded.indexOf(":"); - const user = decoded.slice(0, sep); - const pass = decoded.slice(sep + 1); - if (sep > 0 && user === "admin" && pass === apiToken) return next(); - } + if (labAuthorized(req)) return next(); res.set("WWW-Authenticate", 'Basic realm="stakgraph-lab"'); res.status(401).json({ error: "Unauthorized" }); } @@ -69,5 +85,46 @@ export function mountLab(app: Express): void { if (req.path === "/lab/") return next(); res.redirect(308, "/lab/"); }); - app.use("/lab", labAuth, bridge(() => createLabVein({ serveUi: true }))); + app.use("/lab", labAuth, bridge(labVein)); +} + +const LAB_AUDIO_STREAM = "/lab/audio/stream"; + +/** + * Dictation over `/lab/audio/stream` (vein `src/audio/ws.ts`). A WebSocket + * upgrade never enters Express, so the bridge above can't carry it: hook the + * Node server's `upgrade` event, apply the lab credential (browsers resend + * cached Basic auth on same-origin handshakes, so the UI's one-time prompt + * covers it), then hand the socket to vein. Built lazily like the bridge — + * the first dictation boots the lab vein if a request hasn't already. + * Other upgrade paths get a 404 rather than a socket left hanging. + */ +export function attachLabAudio(server: Server): void { + let handlerP: Promise | null = null; + const handler = () => + (handlerP ??= labVein().then((vein) => + vein.stt + ? createAudioUpgradeHandler(vein.stt, { basePath: "/lab", authorize: () => true }) + : null, + )); + const reject = (socket: Duplex, status: string) => { + socket.write(`HTTP/1.1 ${status}\r\nConnection: close\r\n\r\n`); + socket.destroy(); + }; + server.on("upgrade", (req: IncomingMessage, socket: Duplex, head: Buffer) => { + const url = new URL(req.url ?? "/", "http://localhost"); + if (url.pathname !== LAB_AUDIO_STREAM) return reject(socket, "404 Not Found"); + const header = (name: string) => req.headers[name.toLowerCase()] as string | undefined; + if (!labAuthorized({ header })) return reject(socket, "401 Unauthorized"); + handler() + .then((h) => { + if (!h) return reject(socket, "501 Not Implemented"); + if (socket.destroyed) return; + h.handle(req, socket, head); + }) + .catch((e) => { + console.error("[lab] dictation upgrade failed:", e); + reject(socket, "500 Internal Server Error"); + }); + }); } diff --git a/mcp/yarn.lock b/mcp/yarn.lock index 6b09cbd76..7ab7b4262 100644 --- a/mcp/yarn.lock +++ b/mcp/yarn.lock @@ -5996,6 +5996,48 @@ shebang-regex@^3.0.0: resolved "https://registry.npmjs.org/shebang-regex/-/shebang-regex-3.0.0.tgz" integrity sha512-7++dFhtcx3353uBaq8DDR4NuxBetBzC7ZQOhmTQInHEd6bSrXdiEyzCvG07Z44UYdLShWUyXt5M/yhz8ekcb1A== +sherpa-onnx-darwin-arm64@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-darwin-arm64/-/sherpa-onnx-darwin-arm64-1.13.7.tgz#2b46e237bf28e8d15a9de49c6dcc5a3d5f22da09" + integrity sha512-5NCE50hAvr3n2pdett0SgfPBJXaFZE0bqHwbHyiq+IKZ8Ids0l4M0VrG+ImGYIafCwie+oC3uAJ+pKj9xg/k+w== + +sherpa-onnx-darwin-x64@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-darwin-x64/-/sherpa-onnx-darwin-x64-1.13.7.tgz#8d0ac16d47973a8b25eaf74fcad246005aaa8250" + integrity sha512-N3o+T+wn9WaQmsKV5DD8bTHdo+WN2+sXwmZcGJZiDjtOMR2zFz7uVCZnYCmEAMgvChC+oHcF5RvEEKcRCAu6Pw== + +sherpa-onnx-linux-arm64@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-linux-arm64/-/sherpa-onnx-linux-arm64-1.13.7.tgz#e1726c9748252d07923d00f81c0c3fd4767583c0" + integrity sha512-TFCVpXyTh69buhOtTS8KIfkRXOVKY4Y1qjAktSItrKS4A0chnnrlXO5bKWoNAPeI6fMxTF/uvMYbYgcvjEMfNg== + +sherpa-onnx-linux-x64@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-linux-x64/-/sherpa-onnx-linux-x64-1.13.7.tgz#6766b8dc29b5f46cce44f59196b3252f71aa38c5" + integrity sha512-npmxn5WwmAmlthgBhmbZ33t3i2j4mJwQt46dMEb3j7d41y1/uJrjrVAfa/DkvV+vn49ZWfcQ2UEWDipaZBVhuw== + +sherpa-onnx-node@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-node/-/sherpa-onnx-node-1.13.7.tgz#825bd6e386a9c86ebb6e5c70083b85ef7f14185d" + integrity sha512-0XGV7arGngBCnol0m8OLyqlnaUm19Q1KmetVj1DDBdymXa1upmAHZDwNdN47gjsEhqE5hXUEyc1vRQoXrNhNVg== + optionalDependencies: + sherpa-onnx-darwin-arm64 "^1.13.7" + sherpa-onnx-darwin-x64 "^1.13.7" + sherpa-onnx-linux-arm64 "^1.13.7" + sherpa-onnx-linux-x64 "^1.13.7" + sherpa-onnx-win-ia32 "^1.13.7" + sherpa-onnx-win-x64 "^1.13.7" + +sherpa-onnx-win-ia32@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-win-ia32/-/sherpa-onnx-win-ia32-1.13.7.tgz#d9c9ed4f459e33667be20e88ed00ee51c04f61b1" + integrity sha512-sTwtpxPQ76XLn0giAbvknIDEDKD3XXi2mo2AVROEucf1pIK1DjQl+LjLkalTeFoQqbC4J3xGx/g+xgcHQD1dsw== + +sherpa-onnx-win-x64@^1.13.7: + version "1.13.7" + resolved "https://registry.yarnpkg.com/sherpa-onnx-win-x64/-/sherpa-onnx-win-x64-1.13.7.tgz#51a5baf11aefef39f37727f5e50c26082d450610" + integrity sha512-wBV1o+/zgsMrOjfCFIgGrH6S28xq6CqRCLSavCOjTZ6cqr80yGc07DUHxqsHFPZvfoJU+2JF5L2l3gyWFWoWdQ== + side-channel-list@^1.0.0: version "1.0.0" resolved "https://registry.npmjs.org/side-channel-list/-/side-channel-list-1.0.0.tgz" @@ -6594,7 +6636,10 @@ vary@^1, vary@^1.1.2: system-canvas "^0.2.22" system-canvas-react "^0.2.22" uuid "^11.1.0" + ws "^8.21.3" zod "^4.3.6" + optionalDependencies: + sherpa-onnx-node "^1.13.7" web-streams-polyfill@4.0.0-beta.3: version "4.0.0-beta.3" @@ -6676,6 +6721,11 @@ ws@^8.18.0: resolved "https://registry.npmjs.org/ws/-/ws-8.19.0.tgz" integrity sha512-blAT2mjOEIi0ZzruJfIhb3nps74PRWTCz1IjglWEEpQl5XS/UNama6u2/rjFkDDouqr4L67ry+1aGIALViWjDg== +ws@^8.21.3: + version "8.21.3" + resolved "https://registry.yarnpkg.com/ws/-/ws-8.21.3.tgz#660b4faddb6a3e575c86e078126919961f4de4fc" + integrity sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw== + y18n@^5.0.5: version "5.0.8" resolved "https://registry.npmjs.org/y18n/-/y18n-5.0.8.tgz" diff --git a/vein/AGENTS.md b/vein/AGENTS.md index 9faafc994..b020a1278 100644 --- a/vein/AGENTS.md +++ b/vein/AGENTS.md @@ -61,6 +61,13 @@ vein/ │ │ │ # validate_workflow (static YAML check, no publish — src/validate.ts) │ │ ├── stepHelpers.ts # lsSteps / searchSteps / readStepSource (filesystem-style browser) │ │ └── schemaHelpers.ts # Zod → FieldDesc[] (for get_step schema rendering) +│ ├── audio/ # speech-to-text over sherpa-onnx (plans/local-desktop-and-stt.md §4). Streaming dictation is the product surface; workflows learn AROUND it (hotword lists, "dream cycles" §4.8), no STT step in v1 +│ │ ├── stt.ts # createStt(): model download+verify, recognizer cache, streams (PCM in → partial/final out), two-recognizer mode (fast greedy partials + hotword-capable finals), batch transcribe; sherpa is an optionalDependency, lazy-imported, fakeable via `engine` +│ │ ├── models.ts # catalog: id → release URL + sha256 + chunk latency + hotwords?; VEIN_MODEL_DIR/stt/ +│ │ ├── hotwords.ts # contextual biasing: list format, synthesized bpe.vocab from tokens.txt (REQUIRED with modelingUnit "bpe" — unset = cjkchar = silent no-op), named lists under /audio/hotwords +│ │ ├── sessions.ts # /audio/sessions/.jsonl: finals + user corrections (the dream cycle's training data) +│ │ ├── ws.ts # GET /audio/stream WebSocket (raw `ws` on the Node server — @hono/node-ws doesn't support node-server 2.x); Bearer or ?key= +│ │ └── routes.ts # /audio/models (+ SSE download), /audio/transcribe (WAV body), /audio/hotwords/:name, /audio/sessions/:id (+ corrections) │ ├── graph/ # jarvis-compatible Neo4j graph backend over bolt, no jarvis in the loop (plans/jarvis-graph-compat.md). Opt-in via openGraphBackend │ │ ├── bolt.ts # neo4j-driver wrapper; int() for Integer writes (plain JS numbers write as FLOAT) │ │ ├── vein-schemas.ts# the 9 Vein node types + 14-row edge registry (label registry in plans/generic-storage.md); author-time checks @@ -121,6 +128,11 @@ npm run dev # starts Hono server on :3000 docker run -d --name vein-neo4j-test -p 7688:7687 -e NEO4J_AUTH=neo4j/veintest neo4j:5 VEIN_TEST_NEO4J_URI=bolt://localhost:7688 VEIN_TEST_NEO4J_PASSWORD=veintest npm run test:graph +# Speech-to-text live test — needs the sherpa addon (optionalDependency, +# installed by `npm install` on supported platforms) and downloads the 57 MB +# kroko model into VEIN_TEST_STT_MODEL_DIR (temp dir when unset). +VEIN_TEST_STT=1 npm run test:stt + # Web UI (dev mode with HMR) cd vein/web npm install @@ -151,7 +163,9 @@ cd vein && npm run dev # serves API + UI on :3000 | `VEIN_GRAPH_NAMESPACE` | `default` | jarvis namespace every Vein node is written into | | `VEIN_GRAPH_EMBEDDINGS` | (on) | `off` disables the local MiniLM embedder (vectors stay NULL; search is fulltext-only) | | `VEIN_GRAPH_SEED_ONTOLOGY` | (off) | `1` seeds the bundled jarvis ontology (151 schemas + edge schemas + indexes, add-only) on first open, so a standalone Neo4j can host jarvis-typed data (Document, EvalSet, Concept, …) with no jarvis process. No-op on a jarvis-seeded DB. | -| `VEIN_MODEL_CACHE` | `~/.cache/vein-models` | Where the embedding model's ONNX files are cached | +| `VEIN_MODEL_DIR` | `~/.cache/vein-models` | Local model files: MiniLM's ONNX cache and STT models under `stt//`. `VEIN_MODEL_CACHE` is the older alias. | +| `VEIN_STT_MODEL` | `zipformer-en-kroko` | Finals recognizer for `/audio/stream` + `/audio/transcribe` (hotword-capable) | +| `VEIN_STT_PARTIAL_MODEL` | `nemo-fast-conformer-en-80ms` | Fast greedy recognizer whose output is shown as live partials; `off` for single-recognizer streams | ## Auth diff --git a/vein/package.json b/vein/package.json index ab2942633..74ba51833 100644 --- a/vein/package.json +++ b/vein/package.json @@ -22,7 +22,8 @@ "build:web": "npm --prefix web run build", "dev": "npm run build:web && tsx --env-file=.env src/server.ts", "start": "node build/server.js", - "test": "tsx --test src/expr.test.ts src/core.test.ts src/runner.test.ts src/run-control.test.ts src/control-flow.test.ts src/store.test.ts src/workspace.test.ts src/storage-conformance.test.ts src/integration.test.ts src/services.test.ts src/cassette.test.ts src/run-step.test.ts src/createVein.test.ts src/authoring.test.ts src/ai-integration.test.ts src/chat-store.test.ts src/chat-endpoints.test.ts src/steps/registry.test.ts src/steps/core/agent.test.ts src/steps/core/pack.test.ts src/auth.test.ts src/secret-store.test.ts src/artifacts.test.ts src/slack.test.ts src/gdrive.test.ts src/html-extract.test.ts src/shell.test.ts src/validate.test.ts", + "test": "tsx --test src/expr.test.ts src/core.test.ts src/runner.test.ts src/run-control.test.ts src/control-flow.test.ts src/store.test.ts src/workspace.test.ts src/storage-conformance.test.ts src/integration.test.ts src/services.test.ts src/cassette.test.ts src/run-step.test.ts src/createVein.test.ts src/authoring.test.ts src/ai-integration.test.ts src/chat-store.test.ts src/chat-endpoints.test.ts src/steps/registry.test.ts src/steps/core/agent.test.ts src/steps/core/pack.test.ts src/auth.test.ts src/secret-store.test.ts src/artifacts.test.ts src/slack.test.ts src/gdrive.test.ts src/html-extract.test.ts src/shell.test.ts src/validate.test.ts src/model-dir.test.ts src/audio/hotwords.test.ts src/audio/stt.test.ts src/audio/ws.test.ts", + "test:stt": "VEIN_TEST_STT=1 tsx --test src/audio/stt.live.test.ts", "test:graph": "tsx --test --test-concurrency=1 \"src/graph/*.test.ts\" \"src/steps/lib/graph/*.test.ts\"" }, "dependencies": { @@ -41,6 +42,7 @@ "system-canvas": "^0.2.22", "system-canvas-react": "^0.2.22", "uuid": "^11.1.0", + "ws": "^8.21.3", "zod": "^4.3.6" }, "devDependencies": { @@ -48,7 +50,11 @@ "@types/js-yaml": "^4.0.9", "@types/node": "^22.19.0", "@types/uuid": "^10.0.0", + "@types/ws": "^8.18.1", "tsx": "^4.21.0", "typescript": "^5.9.0" + }, + "optionalDependencies": { + "sherpa-onnx-node": "^1.13.7" } } diff --git a/vein/plans/local-desktop-and-stt.md b/vein/plans/local-desktop-and-stt.md index d9f5d3881..d250be236 100644 --- a/vein/plans/local-desktop-and-stt.md +++ b/vein/plans/local-desktop-and-stt.md @@ -14,11 +14,17 @@ Status: nothing below is built. Findings are from reading the tree as of ## 0. Decisions up front -- **STT runs inside vein, not in the host app.** One implementation serves - every topology (desktop-local, mobile→server, server-side workflows). Native - clients only capture audio and send it. The Swift/Kotlin sherpa bindings are - reserved for a future offline-only mobile mode, and even then the client must - present the same API shape so callers can't tell local from remote. +- **STT runs inside vein, not in the host app.** Vein is the process we + package on every platform, so the recognizer lives there and native clients + only capture audio and send it. The Swift/Kotlin sherpa bindings are + reserved for a future offline-only mobile mode. +- **Streaming dictation is the product surface; there is no STT workflow + step in v1.** Workflows sit *around* the recognizer as the learning loop + ("dream cycles", §4.8), not in front of it. A batch `audio/transcribe` + step can come later if a server workflow needs one. +- **Model family: Zipformer/NeMo transducers.** They are the only sherpa + models that accept hotwords (contextual biasing), which is the mechanism + the dream cycle drives. Whisper/Moonshine/SenseVoice are out (§4.2). - **Desktop ships a Node runtime + a bundled `server.cjs` + `web/dist`**, not a single-file executable, for the first cut. Single-file (SEA / Bun / Deno compile) is a later optimization once the step loader no longer scans @@ -37,7 +43,7 @@ Status: nothing below is built. Findings are from reading the tree as of |---|---|---|---| | Desktop app | child process, `127.0.0.1:` | host captures mic → local HTTP/WS | webview loads `web/dist` from the same origin | | Mobile app | remote server | app captures mic → HTTPS/WSS | same routes, different base URL | -| Server workflow | server | files already on disk / artifacts | `audio/transcribe` step | +| Server workflow (later) | server | files already on disk / artifacts | batch `audio/transcribe` step, not in v1 | | Offline mobile (later) | none | on-device sherpa binding | out of scope; keep API shape identical | The only client-side difference between rows 1 and 2 is the base URL and how @@ -222,161 +228,240 @@ Idea, not decided. Needs team discussion. ## 4. Speech-to-text via sherpa-onnx -### 4.1 Facts - -- Apache-2.0. CPU-only on desktop (GPU is Jetson/CUDA-Linux only). -- npm: `sherpa-onnx-node` (currently 1.13.7) with one optional platform - package each: `darwin-arm64`, `darwin-x64`, `linux-x64`, `linux-arm64`, - `win-x64`, `win-ia32`; 23–34 MB unpacked each. Native addon. -- Both **offline** (whole-file) and **streaming** (partial results) - recognizers. Model families: Whisper, SenseVoice, Zipformer (streaming + - offline), Moonshine, Parakeet (NeMo TDT), Paraformer. +### 4.1 Facts (measured 2026-09-04 on an M-series Mac, Node 24, sherpa-onnx-node 1.13.7) + +- Apache-2.0. CPU-only on desktop. npm `sherpa-onnx-node` + one optional + platform package (`darwin-arm64`, `darwin-x64`, `linux-x64`, + `linux-arm64`, `win-x64`, `win-ia32`; 23–34 MB each). Native addon; on + macOS it loads with **no** `DYLD_LIBRARY_PATH` when installed normally. +- Streaming Zipformer (kroko, 57 MB): recognizer builds in ~0.9 s; decodes + 7 s of audio in ~100 ms (RTF ≈ 0.015). Output is cased and punctuated. + Its exported chunk is 128 frames = **1.28 s**, so partials arrive every + ~1.3 s — accurate but not "live". Chunk size is baked into the export. +- Streaming Zipformer 20M (2023-02, 128 MB): partials every **320 ms**, but + uppercase, no punctuation, and noticeably worse accuracy on the same + clip. Not worth shipping. +- NeMo streaming fast-conformer transducers (int8 ≈ 105 MB) measured on + the same clip: **480 ms** variant → partials every ~570 ms, decode + 220 ms / 7 s; **80 ms** variant → partials every ~150–200 ms, decode + 560 ms / 7 s (RTF 0.08). Lowercase, no punctuation, accuracy on par with + kroko ("stack work", "fink swarm", "vain work flow"). Load ≈ 0.4 s. +- **sherpa's NeMo online transducer is greedy-only** ("Unsupported decoding + method: modified_beam_search"), so the NeMo streaming models cannot take + hotwords. Fast partials and biased decoding come from different model + families — hence the two-recognizer stream in §4.4. +- Nemotron-speech streaming 0.6B, 80 ms variant (int8, 463 MB): the + accuracy ceiling — cased, punctuated, got "Sphinx" right unbiased — but + decode is 3.9 s / 7 s on two threads (RTF ≈ 0.55), load ≈ 0.9 s, and it + is the same NeMo implementation: **greedy-only, no hotwords**. An option + for fast desktops that want a single recognizer without biasing, not the + default. - Bundles its own `libonnxruntime`. Loading it alongside `onnxruntime-node` - in one process risks symbol clashes on Linux. Another reason to run MiniLM - on WASM (§3) and drop `onnxruntime-node` from desktop builds. -- Models are separate downloads (tens to hundreds of MB). - -### 4.2 Dependency layout in vein - -- Add `sherpa-onnx-node` as an **optionalDependency**. Import lazily - (`await import("sherpa-onnx-node")`) inside the step's `run()` and the - route handlers, matching the existing lazy-SDK pattern. If the import - fails, routes return `501 { error: "stt not available" }` and the step - fails with a clear message. Server installs without it still boot. -- New module: `src/audio/stt.ts` — owns recognizer construction, model - resolution, and a tiny session registry for streaming. Steps and routes - both go through it (AGENTS.md "step vs service": this is a service). -- Expose it as `ctx.services.stt?` on `VeinCapabilities` (optional, like - `artifacts`) so custom steps can transcribe without importing sherpa. + in one process risks symbol clashes on Linux. Another reason to run + MiniLM on WASM (§3) and drop `onnxruntime-node` from desktop builds. +- Models are separate `.tar.bz2` downloads. Node has no bzip2; extract by + spawning `tar xjf` (bsdtar on macOS/Windows 10+, GNU tar on Linux all + handle bz2). +- **Tail flush:** after the last audio the stream must be fed ~2 s of + silence before `inputFinished()`, or the final chunk's words are lost + (observed: "post the re" vs "post the results to hive"). +- **A stream's sample rate must never change.** sherpa `exit(-1)`s the whole + process on "You changed the input sampling rate" — no exception to catch. + The service pads the flush at the stream's own rate and rejects a + mid-stream rate change before it reaches the addon. +- **End to end over the WebSocket** (two-recognizer default, 100 ms frames + sent at real-time pace): partials arrive 50–100 ms behind the audio + position; the final lands ~300 ms after `end`; wall time ≈ audio length. + First connection pays ~1.3 s for recognizer construction, later ones ~3 ms. + +### 4.2 Hotwords — the mechanism the dream cycle drives + +- Transducer-only, `modified_beam_search` only (greedy ignores them). + Cost: decode went 80 → ~100 ms on the 7 s clip. Negligible. +- **`modelingUnit` must be `"bpe"`.** Unset, sherpa defaults to `cjkchar` + and silently splits each hotword into single characters, which the model + never emits — hotwords then have no effect at any score. This cost an + hour; do not repeat it. +- `bpe` mode needs a `bpe.vocab` (sentencepiece vocab, `piece\tscore`). + The released models don't ship one. **Synthesize it from `tokens.txt`**: + every non-special token with a constant score of `-1`; sherpa's unigram + encoder then picks the fewest-pieces segmentation. Verified: with this + vocab, hotwords `sphinx`/`Sphinx` turned "on this FHIC swarm" into + "on the Sphinx swarm" at score 1.5–3. +- Score is per matched token. 1.5–3 works; 5 over-biases ("Open the + Stak | Graph, then …"). Per-phrase `:score` suffix is supported; the + dream cycle should set it per term rather than one global number. +- Casing must match what the model would emit (kroko is cased: "Jarvis" + came out capitalized unprompted, unknown names came out lowercase). + Emit both forms for proper nouns. +- Limits: a hotword only helps when the acoustic path is already in the + beam. "Stakwork" stayed "stackwork" at every score — the model hears + "stack work" and the boosted token path is a different spelling. That is + a post-correction case (§4.8 step 4), not a hotwords case. ### 4.3 Audio contract -- Input to the engine is **16 kHz mono float32 PCM**. Accept: - - WAV files (any rate; sherpa's `readWave` handles 8/16/24-bit PCM; we - resample to 16 kHz if needed via sherpa's built-in resampler). - - Raw PCM16LE at 16 kHz for streaming frames (what native mic capture - produces cheaply on every platform). -- Not accepted in v1: compressed formats (m4a/opus/mp3). Clients decode - natively before sending. Revisit if mobile upload sizes hurt. +- Engine input is 16 kHz mono float32. sherpa resamples other rates itself + (verified with a 24 kHz WAV), so accept: + - WAV files (any rate, 8/16/24-bit PCM) for the batch route; + - raw PCM16LE frames with a declared `sampleRate` for streaming (what + native mic capture produces cheaply on every platform). +- Not accepted in v1: compressed formats. Clients decode natively. ### 4.4 Surface -**Step** `audio/transcribe` (`src/steps/lib/audio/transcribe.ts`): - -```yaml -- id: words - type: audio/transcribe - config: - path: "{{ steps.download.output.path }}" # WAV on disk or an artifact ref - model: parakeet-tdt-0.6b-v3 # optional; default from settings - language: auto # for multilingual models -``` - -Output: `{ text, segments: [{start, end, text}], language?, model, durationMs }`. -Segments come from the recognizer's timestamps where the model provides them -(Whisper/Parakeet/SenseVoice do; Zipformer streaming gives word times). - -**Route** `POST /audio/transcribe` — multipart or raw `audio/wav` body, -optional `model`/`language` query. Same output as the step. Gated by -`VEIN_API_KEY` like everything else. This is what the desktop and mobile apps -call for push-to-talk and voice memos. - -**Route** `GET /audio/stream` (WebSocket) — live dictation: - -- Client sends `{"type":"start","model":"zipformer-streaming-en","sampleRate":16000}` - then binary frames of PCM16LE, then `{"type":"end"}`. -- Server sends `{"type":"partial","text":...}` as the streaming recognizer - updates and `{"type":"final","text":...,"segments":[...]}` on endpoint - detection or `end`. -- One recognizer stream per socket; recognizer instances are shared per - model and reused (they're expensive to construct; streams are cheap). -- Needs `@hono/node-ws`. This is vein's first client-to-server streaming - route; keep it isolated in `src/audio/ws.ts` so the SSE-based rest of the - server is untouched. - -**Route** `GET /audio/models` — installed vs. available models, sizes, and -download state, so clients can show status and trigger downloads. -`POST /audio/models/:id/download` starts a download; progress via the -existing SSE event pattern. - -### 4.5 Model management - -- `VEIN_MODEL_DIR` (default `~/.cache/vein-models`, same dir MiniLM uses - today via `VEIN_MODEL_CACHE` — rename/alias so there's one setting). -- A small **catalog** in `src/audio/models.ts`: id → download URL (GitHub - releases of sherpa-onnx), archive layout, recognizer type, language list, - streaming yes/no, approximate size. Start with four entries: - -| id | Use | Streaming | -|---|---|---| -| `parakeet-tdt-0.6b-v3-int8` | English, best offline accuracy | no | -| `moonshine-base-en-int8` | English, small and fast | no | -| `sense-voice-multilingual-int8` | zh/en/ja/ko/yue offline | no | -| `zipformer-streaming-en-int8` | live dictation | yes | - -- Default model: `parakeet-tdt-0.6b-v3-int8` for `transcribe`, - `zipformer-streaming-en-int8` for `stream`. Overridable by env - (`VEIN_STT_MODEL`, `VEIN_STT_STREAM_MODEL`) and per call. -- Downloads happen on first use or via the route; never at boot. Server - images pre-bake models into `VEIN_MODEL_DIR`. -- Verify a checksum from the catalog after download; extract with Node's - `zlib` + a minimal tar reader (no new dep) or `tar` package if we already - pull one transitively. +Everything lives in `src/audio/` and is exposed as `ctx.services.stt?` +(optional, like `artifacts`) so future steps can transcribe without +importing sherpa. All `sherpa-onnx-node` imports are lazy; without the +addon the routes return `501 { error: "stt not available" }` and vein boots +and passes every other test. + +**`GET /audio/stream`** (WebSocket, `@hono/node-ws`) — live dictation: + +- Client: `{"type":"start","model":"","sampleRate":16000,"hotwords":""|["phrase", …],"session":""}` + then binary PCM16LE frames, then `{"type":"end"}`. +- Server: `{"type":"ready"}`, `{"type":"partial","text"}` on every change, + `{"type":"final","text","words":[{w,start}]}` on endpoint detection and on + `end`, `{"type":"error","error"}`. +- **Two recognizers per stream when `partialModel` is set**: the fast + greedy NeMo model produces the partials, the hotword-capable Zipformer + produces the finals and owns endpoint detection (both are reset on its + endpoint). Partials feel live (~150 ms); finals are biased, cased and + punctuated. Combined RTF ≈ 0.1 on M-series. With `partialModel` unset + one recognizer does both. +- One recognizer per (model, hotwords hash, score, endpoint rules), cached; + streams are cheap. + Decoding is synchronous native work on the event loop, ~1–2 ms per + 100 ms frame on M-series; move to a worker thread only if a slow target + shows it. +- Auth: `Authorization: Bearer` **or** `?key=` (a webview's WebSocket + cannot set headers). HTTP routes below use the normal bearer middleware. +- The upgrade happens on the Node server, not inside Hono. Vein's own + `listen()` attaches it; a host that mounts `vein.app` itself (mcp's + Express bridge under `/lab`) hooks the server's `upgrade` event and hands + matching requests to `createAudioUpgradeHandler(stt, { basePath, + authorize })` — `authorize` because an upgrade bypasses the host's HTTP + auth middleware (mcp applies its Basic `admin:API_TOKEN` rule there; + browsers resend cached Basic credentials on same-origin handshakes). + See `mcp/src/lab/mount.ts` `attachLabAudio`. + +**Web UI client** (`web/src/dictation.ts`, `SettingsDialog`, the mic in +`ChatFlyout`): mic → AudioWorklet → PCM16 → `/audio/stream`, dictating into +the AI chat input. Settings (gear icon) has "Enable dictation", which +downloads the default pair, plus an inline hotwords list for experimenting; +both are browser-local (`localStorage`), since installation is already a +server fact. Streams log under `chat-` sessions. The desktop/mobile +hosts capture natively instead (§4.6). + +**`POST /audio/transcribe`** — raw `audio/wav` body, `?model=`, +`?hotwords=`. Same output as a `final`. Push-to-talk and voice memos. + +**`GET /audio/models`** — catalog entries with `installed`, size, chunk +latency, language. **`POST /audio/models/:id/download`** — SSE progress +(bytes, then extracting, then done); sha256 verified from the catalog. + +**Sessions and hotword lists** (the dream-cycle seam, §4.8): +- `GET /audio/sessions`, `GET /audio/sessions/:id` — finals per session as + JSON, written as `/audio/sessions/.jsonl` by the stream. +- `POST /audio/sessions/:id/corrections { index, text }` — the user's edit + of a final. Highest-signal training data. +- `GET/PUT/DELETE /audio/hotwords/:name` — a named phrase list (one per + line, optional `:score`), stored under `/audio/hotwords/`. The + stream's `start.hotwords` names one. + +### 4.5 Model catalog (`src/audio/models.ts`) + +id → GitHub release URL, sha256, size, file layout, chunk latency, casing. + +| id | Size | Partials every | Hotwords | Notes | +|---|---|---|---|---| +| `zipformer-en-kroko` | 57 MB | 1.3 s | yes | cased + punctuated; finals model | +| `nemo-fast-conformer-en-80ms` | 103 MB | ~0.15 s | no (greedy-only) | partials model | +| `nemo-fast-conformer-en-480ms` | 106 MB | ~0.57 s | no (greedy-only) | middle ground if 80 ms costs too much CPU on a weak target | +| `nemotron-speech-en-80ms` | 463 MB | ~0.15 s | no (greedy-only) | accuracy ceiling; RTF ≈ 0.55 on 2 threads, cased + punctuated | + +Defaults: `model` = `zipformer-en-kroko`, `partialModel` = +`nemo-fast-conformer-en-80ms`. Env `VEIN_STT_MODEL` / `VEIN_STT_PARTIAL_MODEL` +and per-call overrides. Dropped: the 2023-02 20M Zipformer (fast but weak). +`VEIN_MODEL_DIR` (alias of the existing `VEIN_MODEL_CACHE`; default +`/vein/models` where the root is `VEIN_CACHE_DIR`, else +`XDG_CACHE_HOME`, else `~/.cache` — the same root mcp's GAIA checkout uses, +so a server's `~/.cache/vein` volume persists both) holds `stt//`. Downloads happen on first use or via +the route, never at boot; server images pre-bake. ### 4.6 Client responsibilities (Swift / Kotlin) -- Capture the microphone natively (AVAudioEngine / AudioRecord). Convert to - 16 kHz mono PCM16LE. Do **not** try to use `getUserMedia` inside the - webview; permission and device handling are unreliable there. -- Push-to-talk: buffer, wrap as WAV, `POST /audio/transcribe`. -- Live: open `/audio/stream`, send frames every ~100 ms, render partials. -- On desktop the base URL is `http://127.0.0.1:`; on mobile it's the - server. Nothing else differs. +- Capture the microphone natively (AVAudioEngine / AudioRecord), 16 kHz + mono PCM16LE. Do **not** use `getUserMedia` inside the webview. +- Live: open `/audio/stream`, send ~100 ms frames, render partials, replace + with finals. Push-to-talk: buffer, wrap as WAV, `POST /audio/transcribe`. +- Show the user's finals editable; send edits as corrections. ### 4.7 Testing -- Unit: `stt.ts` model resolution and catalog parsing without the addon - (mock the import). -- Integration (opt-in, `VEIN_TEST_STT=1`, like `VEIN_TEST_NEO`): download the - Moonshine model once into a temp dir, transcribe a checked-in 3-second WAV - fixture, assert on normalized text. Skip when the addon is missing. -- Streaming: feed the same fixture as PCM frames over the WS route and assert - the final matches the offline result within edit distance. - ---- +- Unit: catalog resolution, hotwords compile (bpe.vocab synthesis + + file layout), the WebSocket protocol over a fake engine. No addon needed. +- Live (opt-in, `VEIN_TEST_STT=1`, like `VEIN_TEST_NEO4J_URI`): download + kroko once into a temp model dir, stream its bundled `test_wavs/0.wav` + as PCM frames, assert the final; then the same with a hotwords list. + Skipped when the addon is missing. + +### 4.8 Dream cycles — how recognition evolves with a user or company + +The recognizer never retrains. What evolves is the **hotwords list** (an +environment artifact in `EVOLVE_SPEC.md` terms — layer 2: versioned, +reviewable, changes what every later run sees) and, optionally, a +correction glossary. Workflows produce both; nothing here is a new step type. + +1. **Capture.** The stream logs every final to the session file; the UI + sends the user's edits as corrections. Corrections say exactly which + words the model gets wrong. +2. **Dream.** A scheduled workflow reads recent sessions and corrections, + optionally company sources via the existing `gdrive/*`, `slack/*`, + `github/*` steps, and an `llm` step extracts the lingo: names, product + terms, acronyms, each with a suggested boost and both casings. It writes + the list with an `http` step to `PUT /audio/hotwords/`. +3. **Apply.** The next stream that names the list gets a recognizer built + with it. Partials are biased directly; no latency cost. +4. **Correct.** For misses hotwords can't fix (the "stackwork" case), a + glossary of `wrong → right` pairs applied to finals only, either as a + string replacement or a small fast LLM pass. Finals can afford 100 ms. +5. **Measure.** Corrections per hundred words per list version. The eval + substrate already knows how to score a versioned artifact; a list that + makes things worse rolls back like a workflow version. + +Open: whether the glossary lives in the same file as the hotwords (one +artifact per user/company) or beside it. Lean: same artifact, two sections. ## 5. Order of work -1. **Server prerequisites** (small, ship together): `VEIN_HOST`, - `VEIN_WEB_DIST`, structured `ready` line, `GET /health`, `?key=` handoff in - the UI, `VEIN_MODEL_DIR` alias. -2. **STT core**: `src/audio/stt.ts`, catalog, `audio/transcribe` step, - `POST /audio/transcribe`, `GET /audio/models`. Testable on a server with - no desktop work at all. -3. **Streaming**: `@hono/node-ws`, `/audio/stream`. -4. **Phase A packaging**: esbuild bundle, Node binary, native dir, a macOS - host proof-of-concept that spawns vein and shows the UI. -5. **Kotlin host**, Windows shell override. -6. Later: single-binary (phase B), local vector store or LadybugDB backend - (§3, §3.1), offline-mobile bindings. - -Steps 1–3 are pure vein work and are useful for the hosted product on their -own. - ---- +1. **STT core** (in progress on `vein-stt`): `src/audio/` service, catalog, + hotwords compiler, `/audio/stream` WebSocket, `POST /audio/transcribe`, + `/audio/models` + download, sessions + hotword lists. Testable on a + server with no desktop work at all. +2. **Model bake-off**: measure the NeMo / Nemotron streaming variants for + partial latency and accuracy on the same clips; pick the default. +3. **Server prerequisites for desktop**: `VEIN_HOST`, `VEIN_WEB_DIST`, + structured `ready` line, `?key=` handoff in the UI, `VEIN_MODEL_DIR` + alias (the last one lands with step 1). +4. **First dream cycle**: a sessions → llm → `PUT /audio/hotwords` workflow + plus the corrections UI. Proves the loop before packaging. +5. **Phase A packaging**: esbuild bundle, Node binary, native dir, a macOS + host proof-of-concept that spawns vein and streams the mic. +6. **Kotlin host**, Windows shell override. +7. Later: single-binary (phase B), local vector store or LadybugDB backend + (§3, §3.1), batch `audio/transcribe` step, offline-mobile bindings. ## 6. Open questions -- WebSocket auth: header (fine for native clients) vs. `?key=` (needed if the - webview ever opens the socket). Lean: accept both, same as HTTP. -- Should `audio/transcribe` also accept an artifact ref from a previous step, - or only a path? Lean: both; artifacts are the natural way a `gdrive`/`http` - step hands a file forward. -- Whisper models in the catalog: they're popular but slow on CPU and lack - Parakeet's accuracy in English. Include `whisper-small-int8` only if - multilingual demand shows up beyond what SenseVoice covers. +- Hotwords per user vs per company: lists are named, so both exist; the + open part is whether a stream can name several and how they merge. +- Non-English dictation: kroko ships de/es/fr streaming variants and + Nemotron 3.5 is multilingual. Catalog entries only; nothing else changes. - Speaker diarization and VAD: sherpa ships silero-VAD and speaker embedding models. Not in v1; the streaming route's endpointing is enough. +- Decoding on the event loop vs a worker thread: fine on M-series; decide + after measuring an older Intel laptop and Windows. - LadybugDB local graph (§3.1): do it at all, and if so, jarvis-shaped logical model vs. Ladybug-native schema. Team discussion pending. @@ -384,8 +469,9 @@ own. ## 7. Decision rules -- Transcription logic lives in `src/audio/`, exposed as a service; steps and - routes are thin. +- Transcription logic lives in `src/audio/`, exposed as a service; routes + are thin. No STT step in v1; workflows learn around the recognizer. +- Hotwords need `modelingUnit: "bpe"` + a synthesized `bpe.vocab`. Always. - Everything sherpa is lazy-imported and optional. A vein without the addon must boot and run every non-audio test. - Native addons are never embedded; ship one platform dir beside the binary. diff --git a/vein/src/audio/hotwords.test.ts b/vein/src/audio/hotwords.test.ts new file mode 100644 index 000000000..f5e5795fa --- /dev/null +++ b/vein/src/audio/hotwords.test.ts @@ -0,0 +1,91 @@ +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + HotwordsStore, + compileHotwords, + formatHotwords, + hotwordsHash, + parseHotwords, + synthesizeBpeVocab, +} from "./hotwords.js"; + +describe("hotwords list format", () => { + it("parses phrases, per-phrase scores, comments and blanks", () => { + const list = parseHotwords("# team lingo\nStakwork\nSphinx :3.5\n\n vein workflow : 2\n"); + assert.deepEqual(list, [ + { phrase: "Stakwork" }, + { phrase: "Sphinx", score: 3.5 }, + { phrase: "vein workflow", score: 2 }, + ]); + }); + + it("round-trips through format", () => { + const text = "Stakwork\nSphinx :3.5\n"; + assert.equal(formatHotwords(parseHotwords(text)), text); + }); + + it("hashes the canonical text, not the input spelling", () => { + assert.equal(hotwordsHash(parseHotwords("a\nb :2")), hotwordsHash(parseHotwords("# x\n\na\n b : 2 \n"))); + assert.notEqual(hotwordsHash(parseHotwords("a")), hotwordsHash(parseHotwords("a :2"))); + }); +}); + +describe("synthesizeBpeVocab", () => { + it("turns tokens.txt into a sentencepiece vocab with constant scores", () => { + const vocab = synthesizeBpeVocab(" 0\n 1\n 2\ns 3\n▁the 4\n"); + assert.equal(vocab, "\t0\n\t0\n\t0\ns\t-1\n▁the\t-1\n"); + }); +}); + +describe("compileHotwords", () => { + let dir: string; + beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), "vein-hotwords-")); + await writeFile(join(dir, "tokens.txt"), " 0\n▁S 1\np 2\nhi 3\nn 4\nx 5\n"); + }); + afterEach(() => rm(dir, { recursive: true, force: true })); + + it("writes the list beside the model and synthesizes bpe.vocab once", async () => { + const a = await compileHotwords(dir, [{ phrase: "Sphinx" }, { phrase: "Hive", score: 3 }]); + assert.equal(a.vocab, join(dir, "bpe.vocab")); + assert.equal(await readFile(a.file, "utf-8"), "Sphinx\nHive :3\n"); + assert.match(await readFile(a.vocab, "utf-8"), /^\t0\n\t0\n<\/s>\t0\n▁S\t-1\n/); + // Idempotent: same hash, same file, vocab not rewritten. + await writeFile(a.vocab, "sentinel"); + const b = await compileHotwords(dir, [{ phrase: "Sphinx" }, { phrase: "Hive", score: 3 }]); + assert.equal(b.file, a.file); + assert.equal(await readFile(a.vocab, "utf-8"), "sentinel"); + }); +}); + +describe("HotwordsStore", () => { + let dir: string; + beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), "vein-hotwords-store-")); + }); + afterEach(() => rm(dir, { recursive: true, force: true })); + + it("put/get/list/delete a named list", async () => { + const store = new HotwordsStore(dir); + assert.deepEqual(await store.list(), []); + assert.equal(await store.get("team"), null); + const list = await store.put("team", "# lingo\nStakwork\nSphinx :3\n"); + assert.equal(list.length, 2); + assert.equal(await store.get("team"), "Stakwork\nSphinx :3\n"); + const infos = await store.list(); + assert.equal(infos.length, 1); + assert.equal(infos[0]!.name, "team"); + assert.equal(infos[0]!.count, 2); + assert.equal(await store.delete("team"), true); + assert.equal(await store.delete("team"), false); + }); + + it("rejects names that could escape the directory", async () => { + const store = new HotwordsStore(dir); + await assert.rejects(() => store.put("../x", "a"), /invalid hotwords list name/); + await assert.rejects(() => store.get("a/b"), /invalid hotwords list name/); + }); +}); diff --git a/vein/src/audio/hotwords.ts b/vein/src/audio/hotwords.ts new file mode 100644 index 000000000..053617a0f --- /dev/null +++ b/vein/src/audio/hotwords.ts @@ -0,0 +1,162 @@ +/** + * Hotwords (contextual biasing) for sherpa-onnx transducer models — the + * mechanism the dream cycle drives (plans/local-desktop-and-stt.md §4.2, §4.8). + * + * Two hard-won facts, both verified live: + * 1. sherpa needs `modelingUnit: "bpe"`; left unset it defaults to `cjkchar` + * and splits every hotword into single characters the model never emits, + * so the list silently does nothing at any score. + * 2. `bpe` mode needs a sentencepiece `bpe.vocab`, which the released models + * don't ship. We synthesize one from `tokens.txt`: every non-special + * token with the same score, so sherpa's unigram encoder picks the + * fewest-pieces segmentation (longest match). Good enough to turn + * "this FHIC swarm" into "the Sphinx swarm". + * + * This module has no sherpa dependency: it only writes the files the + * recognizer config points at. + */ +import { createHash } from "node:crypto"; +import { mkdir, readFile, readdir, rm, stat, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +export interface Hotword { + phrase: string; + /** Per-phrase boost; falls back to the recognizer's global score. */ + score?: number; +} + +/** One phrase per line, optional trailing ` :score`, `#` comments, blanks + * ignored. The same text format sherpa reads, so a stored list is passed + * through verbatim. */ +export function parseHotwords(text: string): Hotword[] { + const out: Hotword[] = []; + for (const raw of text.split(/\r?\n/)) { + const line = raw.trim(); + if (!line || line.startsWith("#")) continue; + const m = line.match(/^(.*?)\s*:\s*(-?\d+(?:\.\d+)?)$/); + if (m && m[1]) out.push({ phrase: m[1].trim(), score: Number(m[2]) }); + else out.push({ phrase: line }); + } + return out; +} + +export function formatHotwords(list: readonly Hotword[]): string { + return list.map((h) => (h.score == null ? h.phrase : `${h.phrase} :${h.score}`)).join("\n") + "\n"; +} + +/** Stable id for a compiled list: sha256 of its canonical text. */ +export function hotwordsHash(list: readonly Hotword[]): string { + return createHash("sha256").update(formatHotwords(list)).digest("hex").slice(0, 16); +} + +/** Sentencepiece-style vocab (`piece\tscore`) from a sherpa `tokens.txt` + * (`piece id` per line). Special tokens (``, ``, ``) are + * replaced by the three sentencepiece specials. */ +export function synthesizeBpeVocab(tokensTxt: string): string { + const pieces = tokensTxt + .split(/\r?\n/) + .map((l) => l.trim().split(/\s+/)[0] ?? "") + .filter((p) => p && !/^<.*>$/.test(p)); + return ["\t0", "\t0", "\t0", ...pieces.map((p) => `${p}\t-1`)].join("\n") + "\n"; +} + +export interface CompiledHotwords { + hash: string; + /** The list, one phrase per line (sherpa `hotwordsFile`). */ + file: string; + /** The synthesized vocab (sherpa `bpeVocab`). */ + vocab: string; +} + +/** Materialize a list beside a model: `/hotwords/.txt` plus + * `/bpe.vocab` (synthesized once). Idempotent. */ +export async function compileHotwords(modelDir: string, list: readonly Hotword[]): Promise { + const vocab = join(modelDir, "bpe.vocab"); + if (!(await exists(vocab))) { + const tokens = await readFile(join(modelDir, "tokens.txt"), "utf-8"); + await writeFile(vocab, synthesizeBpeVocab(tokens)); + } + const hash = hotwordsHash(list); + const dir = join(modelDir, "hotwords"); + await mkdir(dir, { recursive: true }); + const file = join(dir, `${hash}.txt`); + if (!(await exists(file))) await writeFile(file, formatHotwords(list)); + return { hash, file, vocab }; +} + +// ── Named lists (the dream cycle's promotion artifact) ───────────────────── + +const NAME_RE = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; + +export interface HotwordsListInfo { + name: string; + count: number; + updatedAt: string; +} + +/** `/audio/hotwords/.txt`. A workflow's `http` step PUTs a + * list here; a stream names it in `start.hotwords`. */ +export class HotwordsStore { + readonly dir: string; + constructor(dataDir: string) { + this.dir = join(dataDir, "audio", "hotwords"); + } + + private pathOf(name: string): string { + if (!NAME_RE.test(name)) throw new Error(`invalid hotwords list name: ${JSON.stringify(name)}`); + return join(this.dir, `${name}.txt`); + } + + async list(): Promise { + let names: string[]; + try { + names = (await readdir(this.dir)).filter((f) => f.endsWith(".txt")); + } catch { + return []; + } + const out: HotwordsListInfo[] = []; + for (const f of names.sort()) { + const p = join(this.dir, f); + const [text, st] = await Promise.all([readFile(p, "utf-8"), stat(p)]); + out.push({ name: f.slice(0, -4), count: parseHotwords(text).length, updatedAt: st.mtime.toISOString() }); + } + return out; + } + + /** The raw text, or null when the list doesn't exist. */ + async get(name: string): Promise { + try { + return await readFile(this.pathOf(name), "utf-8"); + } catch (e) { + if ((e as NodeJS.ErrnoException).code === "ENOENT") return null; + throw e; + } + } + + async put(name: string, text: string): Promise { + const p = this.pathOf(name); + await mkdir(this.dir, { recursive: true }); + const list = parseHotwords(text); + await writeFile(p, formatHotwords(list)); + return list; + } + + async delete(name: string): Promise { + try { + await rm(this.pathOf(name)); + return true; + } catch (e) { + if ((e as NodeJS.ErrnoException).code === "ENOENT") return false; + throw e; + } + } +} + +async function exists(p: string): Promise { + try { + await stat(p); + return true; + } catch { + return false; + } +} diff --git a/vein/src/audio/models.ts b/vein/src/audio/models.ts new file mode 100644 index 000000000..629756313 --- /dev/null +++ b/vein/src/audio/models.ts @@ -0,0 +1,128 @@ +/** + * STT model catalog — the one list desktop, server, and the dream cycle + * agree on (plans/local-desktop-and-stt.md §4.5). Every entry is a sherpa-onnx + * GitHub release asset with its sha256, so a download is verifiable and a + * server image can pre-bake the same files. + * + * Only streaming transducers are listed: hotwords (§4.2) need a transducer, + * and the NeMo online transducer in sherpa is greedy-only, so `hotwords` + * records which entries can be biased. + */ +import { join } from "node:path"; + +export interface SttModel { + id: string; + url: string; + /** sha256 of the `.tar.bz2`. */ + sha256: string; + bytes: number; + /** Top-level directory inside the archive. */ + archiveDir: string; + language: string; + /** How often partials change, measured (ms). Baked into the export. */ + chunkMs: number; + /** Accepts a hotwords list (`modified_beam_search` transducer). */ + hotwords: boolean; + /** Emits casing and punctuation. */ + cased: boolean; + /** Silence to feed after the last audio so the final chunk flushes (ms). */ + tailPadMs: number; + description: string; +} + +const RELEASE = "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models"; + +export const STT_MODELS: readonly SttModel[] = [ + { + id: "zipformer-en-kroko", + url: `${RELEASE}/sherpa-onnx-streaming-zipformer-en-kroko-2025-08-06.tar.bz2`, + sha256: "c8676e5ff9ac2a85296e53ee0fd4d5fb1db6770e7a7647166eeafe349ade6834", + bytes: 57267600, + archiveDir: "sherpa-onnx-streaming-zipformer-en-kroko-2025-08-06", + language: "en", + chunkMs: 1280, + hotwords: true, + cased: true, + tailPadMs: 2000, + description: "Streaming Zipformer (Banafo Kroko). Cased, punctuated, hotword-capable; partials every ~1.3 s. The finals model.", + }, + { + id: "nemo-fast-conformer-en-80ms", + url: `${RELEASE}/sherpa-onnx-nemo-streaming-fast-conformer-transducer-en-80ms-int8.tar.bz2`, + sha256: "7bd33a914e93370a1ba9c2066d9e841bdcad8613fa2a00537c1ae15d851a14d8", + bytes: 102813625, + archiveDir: "sherpa-onnx-nemo-streaming-fast-conformer-transducer-en-80ms-int8", + language: "en", + chunkMs: 80, + hotwords: false, + cased: false, + tailPadMs: 1000, + description: "NeMo streaming FastConformer transducer, 80 ms chunks (int8). Lowercase, greedy-only; partials every ~150 ms. The partials model.", + }, + { + id: "nemo-fast-conformer-en-480ms", + url: `${RELEASE}/sherpa-onnx-nemo-streaming-fast-conformer-transducer-en-480ms-int8.tar.bz2`, + sha256: "da93061cbf7b708b6b65976f70b29f519be29df750d8cdcabf98c65645930f13", + bytes: 105913204, + archiveDir: "sherpa-onnx-nemo-streaming-fast-conformer-transducer-en-480ms-int8", + language: "en", + chunkMs: 480, + hotwords: false, + cased: false, + tailPadMs: 1000, + description: "NeMo streaming FastConformer transducer, 480 ms chunks (int8). Lowercase, greedy-only; partials every ~570 ms at less CPU than the 80 ms variant.", + }, + { + id: "nemotron-speech-en-80ms", + url: `${RELEASE}/sherpa-onnx-nemotron-speech-streaming-en-0.6b-80ms-int8-2026-04-25.tar.bz2`, + sha256: "caaf92069dbd1ca054f8e17cab179813bc28b4585f5c392540357ece4722333d", + bytes: 463945379, + archiveDir: "sherpa-onnx-nemotron-speech-streaming-en-0.6b-80ms-int8-2026-04-25", + language: "en", + chunkMs: 80, + hotwords: false, + cased: true, + tailPadMs: 1000, + description: "NVIDIA Nemotron-speech streaming 0.6B, 80 ms chunks (int8). Cased, punctuated, the most accurate; greedy-only and ~0.5× real-time CPU on two threads.", + }, +]; + +export const DEFAULT_MODEL = "zipformer-en-kroko"; +export const DEFAULT_PARTIAL_MODEL = "nemo-fast-conformer-en-80ms"; + +export function findModel(id: string): SttModel | undefined { + return STT_MODELS.find((m) => m.id === id); +} + +export function requireModel(id: string): SttModel { + const m = findModel(id); + if (!m) throw new Error(`unknown stt model "${id}" (known: ${STT_MODELS.map((x) => x.id).join(", ")})`); + return m; +} + +/** Shared with MiniLM (src/model-dir.ts). STT models live under `/stt//`. */ +export { modelDirFromEnv } from "../model-dir.js"; + +export function sttModelPath(modelDir: string, id: string): string { + return join(modelDir, "stt", id); +} + +/** Pick the model files inside an extracted dir. Prefers int8 exports. */ +export function pickModelFiles(entries: readonly string[]): { + encoder: string; + decoder: string; + joiner: string; + tokens: string; +} { + const pick = (re: RegExp) => { + const hits = entries.filter((f) => re.test(f)).sort((a, b) => Number(b.includes("int8")) - Number(a.includes("int8"))); + if (!hits[0]) throw new Error(`model dir is missing a file matching ${re}`); + return hits[0]; + }; + return { + encoder: pick(/^encoder.*\.onnx$/), + decoder: pick(/^decoder.*\.onnx$/), + joiner: pick(/^joiner.*\.onnx$/), + tokens: pick(/^tokens\.txt$/), + }; +} diff --git a/vein/src/audio/routes.ts b/vein/src/audio/routes.ts new file mode 100644 index 000000000..7e3d30970 --- /dev/null +++ b/vein/src/audio/routes.ts @@ -0,0 +1,114 @@ +/** + * HTTP surface over the STT service (plans/local-desktop-and-stt.md §4.4). + * The WebSocket lives in ws.ts. Everything here is gated by `requireApiKey` + * (permissive in dev, like the rest of vein). + */ +import type { Hono } from "hono"; +import { streamSSE } from "hono/streaming"; +import { requireApiKey } from "../auth.js"; +import { findModel } from "./models.js"; +import { SttUnavailableError, type SttService, type SttStreamOptions } from "./stt.js"; + +export function audioRoutes(app: Hono, stt: SttService): void { + app.use("/audio/*", requireApiKey); + + app.get("/audio/models", async (c) => { + return c.json({ available: await stt.available(), modelDir: stt.modelDir, models: await stt.models() }); + }); + + // SSE: {"phase":"download","received","total"} … {"phase":"extract"} … {"phase":"done"} + app.post("/audio/models/:id/download", async (c) => { + const id = c.req.param("id"); + if (!findModel(id)) return c.json({ error: `unknown model ${JSON.stringify(id)}` }, 404); + return streamSSE(c, async (s) => { + let last = 0; + try { + await stt.ensureModel(id, (p) => { + // Throttle byte progress to ~every 1 MB so the stream stays light. + if (p.phase === "download" && p.received - last < 1_000_000 && p.received !== p.total) return; + if (p.phase === "download") last = p.received; + void s.writeSSE({ event: "progress", data: JSON.stringify(p) }); + }); + await s.writeSSE({ event: "progress", data: JSON.stringify({ phase: "done" }) }); + } catch (e) { + await s.writeSSE({ event: "error", data: JSON.stringify({ error: (e as Error).message }) }); + } + }); + }); + + // Raw audio/wav body. Query: model, partialModel (ignored: batch is + // single-recognizer), hotwords (list name), hotwordsScore, session. + app.post("/audio/transcribe", async (c) => { + const q = c.req.query(); + const opts: Omit = {}; + if (q["model"]) opts.model = q["model"]; + if (q["hotwords"]) opts.hotwords = q["hotwords"]; + if (q["hotwordsScore"]) opts.hotwordsScore = Number(q["hotwordsScore"]); + if (q["session"]) opts.session = q["session"]; + const body = new Uint8Array(await c.req.arrayBuffer()); + if (body.byteLength < 44) return c.json({ error: "expected a WAV body" }, 400); + try { + return c.json(await stt.transcribe(body, opts)); + } catch (e) { + return errorResponse(c, e); + } + }); + + // ── Hotword lists (the dream cycle's promotion artifact) ──────────────── + + app.get("/audio/hotwords", async (c) => c.json({ lists: await stt.hotwords.list() })); + + app.get("/audio/hotwords/:name", async (c) => { + const text = await stt.hotwords.get(c.req.param("name")); + if (text == null) return c.json({ error: "not found" }, 404); + return c.text(text); + }); + + // Body: text/plain (one phrase per line, optional ` :score`) or JSON + // { "phrases": ["…", …] } / { "text": "…" }. + app.put("/audio/hotwords/:name", async (c) => { + const name = c.req.param("name"); + let text: string; + if ((c.req.header("content-type") ?? "").includes("application/json")) { + const body = (await c.req.json()) as { phrases?: string[]; text?: string }; + text = body.text ?? (body.phrases ?? []).join("\n"); + } else { + text = await c.req.text(); + } + try { + const list = await stt.hotwords.put(name, text); + return c.json({ name, count: list.length }); + } catch (e) { + return c.json({ error: (e as Error).message }, 400); + } + }); + + app.delete("/audio/hotwords/:name", async (c) => { + const ok = await stt.hotwords.delete(c.req.param("name")); + return ok ? c.json({ ok: true }) : c.json({ error: "not found" }, 404); + }); + + // ── Sessions (finals + corrections) ───────────────────────────────────── + + app.get("/audio/sessions", async (c) => c.json({ sessions: await stt.sessions.list() })); + + app.get("/audio/sessions/:id", async (c) => { + const entries = await stt.sessions.get(c.req.param("id")); + if (!entries) return c.json({ error: "not found" }, 404); + return c.json({ id: c.req.param("id"), entries }); + }); + + app.post("/audio/sessions/:id/corrections", async (c) => { + const body = (await c.req.json().catch(() => null)) as { index?: number; text?: string } | null; + if (!body || typeof body.index !== "number" || typeof body.text !== "string") { + return c.json({ error: "expected { index: number, text: string }" }, 400); + } + await stt.correct(c.req.param("id"), body.index, body.text); + return c.json({ ok: true }); + }); +} + +function errorResponse(c: { json: (b: unknown, s: 400 | 501) => Response }, e: unknown): Response { + if (e instanceof SttUnavailableError) return c.json({ error: e.message }, 501); + return c.json({ error: (e as Error).message }, 400); +} diff --git a/vein/src/audio/sessions.ts b/vein/src/audio/sessions.ts new file mode 100644 index 000000000..d2a57d276 --- /dev/null +++ b/vein/src/audio/sessions.ts @@ -0,0 +1,83 @@ +/** + * Dictation sessions — the capture half of the dream cycle + * (plans/local-desktop-and-stt.md §4.8). One JSONL file per session under + * `/audio/sessions/`: every final the stream produced, plus the + * user's corrections to those finals. A dream-cycle workflow reads these + * through `GET /audio/sessions/:id`. + */ +import { appendFile, mkdir, readdir, readFile, stat } from "node:fs/promises"; +import { join } from "node:path"; +import type { SttWord } from "./stt.js"; + +const ID_RE = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; + +export type SessionEntry = + | { + type: "final"; + t: string; + index: number; + text: string; + words: SttWord[]; + model: string; + hotwords: string | null; + } + | { type: "correction"; t: string; index: number; text: string }; + +export interface SessionInfo { + id: string; + updatedAt: string; + bytes: number; +} + +export class SessionStore { + readonly dir: string; + constructor(dataDir: string) { + this.dir = join(dataDir, "audio", "sessions"); + } + + private pathOf(id: string): string { + if (!ID_RE.test(id)) throw new Error(`invalid session id: ${JSON.stringify(id)}`); + return join(this.dir, `${id}.jsonl`); + } + + async append(id: string, entry: SessionEntry): Promise { + await mkdir(this.dir, { recursive: true }); + await appendFile(this.pathOf(id), JSON.stringify(entry) + "\n"); + } + + async list(): Promise { + let names: string[]; + try { + names = (await readdir(this.dir)).filter((f) => f.endsWith(".jsonl")); + } catch { + return []; + } + const out: SessionInfo[] = []; + for (const f of names.sort()) { + const st = await stat(join(this.dir, f)); + out.push({ id: f.slice(0, -6), updatedAt: st.mtime.toISOString(), bytes: st.size }); + } + return out; + } + + /** The index the next final in this session should carry. */ + async nextIndex(id: string): Promise { + const entries = (await this.get(id)) ?? []; + return entries.reduce((n, e) => (e.type === "final" ? Math.max(n, e.index + 1) : n), 0); + } + + /** All entries in order, or null when the session doesn't exist. */ + async get(id: string): Promise { + let text: string; + try { + text = await readFile(this.pathOf(id), "utf-8"); + } catch (e) { + if ((e as NodeJS.ErrnoException).code === "ENOENT") return null; + throw e; + } + return text + .split("\n") + .filter(Boolean) + .map((l) => JSON.parse(l) as SessionEntry); + } +} diff --git a/vein/src/audio/sherpa.d.ts b/vein/src/audio/sherpa.d.ts new file mode 100644 index 000000000..abdec5814 --- /dev/null +++ b/vein/src/audio/sherpa.d.ts @@ -0,0 +1,7 @@ +// sherpa-onnx-node ships JSDoc typedefs, not .d.ts. The engine surface vein +// uses is typed locally in stt.ts (`SttEngine`); this only makes the lazy +// `import("sherpa-onnx-node")` compile. +declare module "sherpa-onnx-node" { + const sherpa: any; + export = sherpa; +} diff --git a/vein/src/audio/stt.live.test.ts b/vein/src/audio/stt.live.test.ts new file mode 100644 index 000000000..3b51da0bb --- /dev/null +++ b/vein/src/audio/stt.live.test.ts @@ -0,0 +1,88 @@ +/** + * LIVE speech-to-text test — needs the sherpa addon and downloads the small + * kroko model (57 MB) into VEIN_TEST_STT_MODEL_DIR (default: a temp dir, so + * point it at a persistent dir to avoid re-downloading). Opt in with + * VEIN_TEST_STT=1. Skipped otherwise, like the graph tests. + */ +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createStt, loadSherpaEngine, pcm16ToFloat32, type SttEvent, type SttService } from "./stt.js"; +import { sttModelPath } from "./models.js"; + +const enabled = process.env["VEIN_TEST_STT"] === "1"; + +describe("stt live (VEIN_TEST_STT=1)", { skip: !enabled }, () => { + let dataDir: string; + let stt: SttService; + let wavPath: string; + const modelDir = process.env["VEIN_TEST_STT_MODEL_DIR"]; + let tmpModelDir: string | undefined; + + before(async () => { + assert.ok(await loadSherpaEngine(), "sherpa-onnx-node must be installed for the live test"); + dataDir = await mkdtemp(join(tmpdir(), "vein-stt-live-")); + if (!modelDir) tmpModelDir = await mkdtemp(join(tmpdir(), "vein-stt-models-")); + stt = createStt({ dataDir, modelDir: modelDir ?? tmpModelDir!, env: {} }); + const dir = await stt.ensureModel("zipformer-en-kroko"); + assert.equal(dir, sttModelPath(modelDir ?? tmpModelDir!, "zipformer-en-kroko")); + wavPath = join(dir, "test_wavs", "0.wav"); + }); + after(async () => { + await rm(dataDir, { recursive: true, force: true }); + if (tmpModelDir) await rm(tmpModelDir, { recursive: true, force: true }); + }); + + async function streamWav(opts: Parameters[0]): Promise { + const engine = (await loadSherpaEngine())!; + const wave = engine.readWave(wavPath); + const s = await stt.openStream({ ...opts, sampleRate: wave.sampleRate }); + const frame = Math.round(wave.sampleRate * 0.1); + const events: SttEvent[] = []; + for (let i = 0; i < wave.samples.length; i += frame) { + // Round-trip through PCM16 so the bytes path is what's exercised. + const chunk = wave.samples.subarray(i, i + frame); + const pcm = new Uint8Array(chunk.length * 2); + const view = new DataView(pcm.buffer); + chunk.forEach((v, j) => view.setInt16(j * 2, Math.max(-32768, Math.min(32767, Math.round(v * 32768))), true)); + assert.equal(pcm16ToFloat32(pcm).length, chunk.length); + events.push(...s.push(pcm)); + } + events.push(...s.end()); + await s.flush(); + return events; + } + + it("streams the bundled test clip to partials and a final", async () => { + const events = await streamWav({ partialModel: null, session: "live" }); + const finals = events.filter((e) => e.type === "final"); + const partials = events.filter((e) => e.type === "partial"); + assert.ok(partials.length >= 1, "expected partials"); + assert.ok(finals.length >= 1, "expected a final"); + const text = finals.map((e) => (e.type === "final" ? e.text : "")).join(" "); + assert.match(text.toLowerCase(), /ask not what your country can do for you/); + const first = finals[0]!; + assert.ok(first.type === "final" && first.words.length >= 5 && first.words[0]!.start >= 0); + const logged = await stt.sessions.get("live"); + assert.ok(logged && logged.length >= 1); + }); + + it("accepts a hotwords list (modified_beam_search path) and still transcribes", async () => { + await stt.hotwords.put("live", "country :2\nAsk\n"); + const events = await streamWav({ partialModel: null, hotwords: "live" }); + const text = events + .filter((e) => e.type === "final") + .map((e) => (e.type === "final" ? e.text : "")) + .join(" "); + assert.match(text.toLowerCase(), /ask not what your country/); + }); + + it("transcribe() over the WAV bytes matches the stream", async () => { + const { readFile } = await import("node:fs/promises"); + const r = await stt.transcribe(await readFile(wavPath)); + assert.match(r.text.toLowerCase(), /ask not what your country/); + assert.equal(r.model, "zipformer-en-kroko"); + }); +}); diff --git a/vein/src/audio/stt.test.ts b/vein/src/audio/stt.test.ts new file mode 100644 index 000000000..2cee1f9dd --- /dev/null +++ b/vein/src/audio/stt.test.ts @@ -0,0 +1,283 @@ +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + createStt, + pcm16ToFloat32, + wordsOf, + type EngineRecognizer, + type EngineResult, + type EngineStream, + type SttEngine, +} from "./stt.js"; +import { STT_MODELS, sttModelPath } from "./models.js"; + +/** + * A scripted fake engine: each recognizer is fed a queue of "what to say + * after N seconds of audio", so the stream logic (partials, endpoints, + * two-recognizer mode, flush, session logging) is exercised with no addon. + */ +interface Script { + /** text to report once at least `at` seconds of audio have been accepted */ + at: number; + text: string; + tokens?: string[]; + timestamps?: number[]; + /** report an endpoint once this text is reached */ + endpoint?: boolean; +} + +class FakeStream implements EngineStream { + seconds = 0; + finished = false; + acceptWaveform(w: { samples: Float32Array; sampleRate: number }): void { + this.seconds += w.samples.length / w.sampleRate; + } + inputFinished(): void { + this.finished = true; + } +} + +class FakeRecognizer implements EngineRecognizer { + readonly config: Record; + readonly script: Script[]; + private offsets = new WeakMap(); + constructor(config: Record, script: Script[]) { + this.config = config; + this.script = script; + } + createStream(): EngineStream { + return new FakeStream(); + } + isReady(): boolean { + return false; + } + decode(): void {} + private current(s: EngineStream): Script | undefined { + const fs = s as FakeStream; + const local = fs.seconds - (this.offsets.get(fs) ?? 0); + return [...this.script].reverse().find((x) => local >= x.at); + } + isEndpoint(s: EngineStream): boolean { + return this.current(s)?.endpoint === true; + } + reset(s: EngineStream): void { + const fs = s as FakeStream; + this.offsets.set(fs, fs.seconds); + } + getResult(s: EngineStream): EngineResult { + const cur = this.current(s); + if (!cur) return { text: "" }; + return { text: cur.text, tokens: cur.tokens, timestamps: cur.timestamps, start_time: this.offsets.get(s as FakeStream) ?? 0 }; + } +} + +function fakeEngine(scripts: Record, built: Record[] = []): SttEngine { + return { + OnlineRecognizer: class { + constructor(config: Record) { + built.push(config); + const tokens = String((config["modelConfig"] as { tokens: string }).tokens); + const id = STT_MODELS.find((m) => tokens.includes(`/stt/${m.id}/`))?.id ?? "?"; + return new FakeRecognizer(config, scripts[id] ?? []); + } + } as unknown as SttEngine["OnlineRecognizer"], + readWave: () => ({ samples: new Float32Array(16000 * 3), sampleRate: 16000 }), + }; +} + +/** Pretend `id` is installed under modelDir. */ +async function installFake(modelDir: string, id: string): Promise { + const dir = sttModelPath(modelDir, id); + await mkdir(dir, { recursive: true }); + for (const f of ["encoder.int8.onnx", "decoder.int8.onnx", "joiner.int8.onnx"]) await writeFile(join(dir, f), ""); + await writeFile(join(dir, "tokens.txt"), " 0\n▁S 1\nphinx 2\n"); + await writeFile(join(dir, ".ok"), "x"); +} + +const silence = (seconds: number) => new Uint8Array(16000 * 2 * seconds); + +describe("stt service", () => { + let root: string; + let dataDir: string; + let modelDir: string; + const env = { VEIN_STT_MODEL: undefined, VEIN_STT_PARTIAL_MODEL: undefined } as Record; + + beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), "vein-stt-")); + dataDir = join(root, "data"); + modelDir = join(root, "models"); + await installFake(modelDir, "zipformer-en-kroko"); + await installFake(modelDir, "nemo-fast-conformer-en-80ms"); + }); + afterEach(() => rm(root, { recursive: true, force: true })); + + it("reports availability from the engine loader and lists install state", async () => { + const off = createStt({ dataDir, modelDir, env, engine: async () => null, log: () => {} }); + assert.equal(await off.available(), false); + const on = createStt({ dataDir, modelDir, env, engine: async () => fakeEngine({}), log: () => {} }); + assert.equal(await on.available(), true); + const models = await on.models(); + assert.equal(models.find((m) => m.id === "zipformer-en-kroko")?.installed, true); + assert.equal(models.find((m) => m.id === "zipformer-en-kroko")?.default, "model"); + assert.equal(models.find((m) => m.id === "nemo-fast-conformer-en-80ms")?.default, "partialModel"); + assert.equal(models.find((m) => m.id === "nemo-fast-conformer-en-480ms")?.installed, false); + }); + + it("openStream throws a clear error without the addon", async () => { + const stt = createStt({ dataDir, modelDir, env, engine: async () => null, log: () => {} }); + await assert.rejects(() => stt.openStream({ partialModel: null }), /stt not available/); + }); + + it("single recognizer: partials on change, final + reset on endpoint, flush on end", async () => { + const engine = fakeEngine({ + "zipformer-en-kroko": [ + { at: 0.5, text: "hello", tokens: ["▁hello"], timestamps: [0.3] }, + { at: 1.0, text: "hello world", tokens: ["▁hello", "▁wor", "ld"], timestamps: [0.3, 0.8, 0.9], endpoint: true }, + { at: 1.5, text: "again", tokens: ["▁again"], timestamps: [0.1] }, + ], + }); + const stt = createStt({ dataDir, modelDir, env, engine: async () => engine, log: () => {} }); + const s = await stt.openStream({ partialModel: null, session: "s1" }); + assert.equal(s.model, "zipformer-en-kroko"); + assert.equal(s.partialModel, null); + const events = []; + for (let i = 0; i < 12; i++) events.push(...s.push(silence(0.1))); + events.push(...s.end()); + await s.flush(); + assert.deepEqual(events, [ + { type: "partial", text: "hello" }, + { type: "partial", text: "hello world" }, + { type: "final", index: 0, text: "hello world", words: [{ text: "hello", start: 0.3 }, { text: "world", start: 0.8 }] }, + // after the reset at ~1.1 s, the second script line is reached 1.5 s later — that's + // during end()'s 2 s silence pad, so it surfaces as the trailing final only. + { type: "final", index: 1, text: "again", words: [{ text: "again", start: 1.2 }] }, + ]); + // Session log captured both finals. + const entries = (await stt.sessions.get("s1")) ?? []; + assert.deepEqual(entries.map((e) => (e.type === "final" ? e.text : e.type)), ["hello world", "again"]); + await stt.correct("s1", 1, "again!"); + assert.equal(((await stt.sessions.get("s1")) ?? []).at(-1)?.type, "correction"); + // A later connection on the same session keeps counting. + const s2 = await stt.openStream({ partialModel: null, session: "s1" }); + const later = []; + for (let i = 0; i < 11; i++) later.push(...s2.push(silence(0.1))); + assert.equal(later.find((e) => e.type === "final")?.index, 2); + }); + + it("two recognizers: partials from the fast model, finals + endpoint from the main one", async () => { + const built: Record[] = []; + const engine = fakeEngine( + { + "zipformer-en-kroko": [{ at: 1.0, text: "Hello, world.", tokens: ["▁Hello", ","], timestamps: [0.2, 0.5], endpoint: true }], + "nemo-fast-conformer-en-80ms": [ + { at: 0.2, text: "hel" }, + { at: 0.4, text: "hello" }, + { at: 0.8, text: "hello world" }, + ], + }, + built, + ); + const stt = createStt({ dataDir, modelDir, env, engine: async () => engine, log: () => {} }); + const s = await stt.openStream({ hotwords: ["Sphinx"], hotwordsScore: 3 }); + assert.equal(s.partialModel, "nemo-fast-conformer-en-80ms"); + assert.equal(s.hotwords, "inline"); + const events = []; + for (let i = 0; i < 11; i++) events.push(...s.push(silence(0.1))); + assert.deepEqual(events, [ + { type: "partial", text: "hel" }, + { type: "partial", text: "hello" }, + { type: "partial", text: "hello world" }, + { type: "final", index: 0, text: "Hello, world.", words: [{ text: "Hello,", start: 0.2 }] }, + ]); + // Hotwords went to the main recognizer only, with bpe modeling unit + vocab. + const main = built.find((c) => String((c["modelConfig"] as { tokens: string }).tokens).includes("zipformer-en-kroko"))!; + const fast = built.find((c) => String((c["modelConfig"] as { tokens: string }).tokens).includes("nemo-fast"))!; + assert.equal(main["decodingMethod"], "modified_beam_search"); + assert.equal(main["hotwordsScore"], 3); + assert.equal((main["modelConfig"] as Record)["modelingUnit"], "bpe"); + assert.match(String((main["modelConfig"] as Record)["bpeVocab"]), /bpe\.vocab$/); + assert.equal(fast["decodingMethod"], "greedy_search"); + assert.equal(fast["hotwordsFile"], undefined); + }); + + it("caches recognizers per (model, hotwords, score, endpoint) and resolves named lists", async () => { + const built: Record[] = []; + const engine = fakeEngine({}, built); + const stt = createStt({ dataDir, modelDir, env, engine: async () => engine, log: () => {} }); + await stt.hotwords.put("team", "Stakwork\nSphinx :3\n"); + const a = await stt.openStream({ partialModel: null, hotwords: "team" }); + const b = await stt.openStream({ partialModel: null, hotwords: "team" }); + const c = await stt.openStream({ partialModel: null, hotwords: "team", hotwordsScore: 4 }); + const d = await stt.openStream({ partialModel: null }); + assert.equal(a.hotwords, "team"); + assert.equal(b.hotwords, "team"); + assert.equal(c.hotwords, "team"); + assert.equal(d.hotwords, null); + assert.equal(built.length, 3); + await assert.rejects(() => stt.openStream({ partialModel: null, hotwords: "nope" }), /unknown hotwords list/); + }); + + it("ignores hotwords on models that cannot take them", async () => { + const built: Record[] = []; + const stt = createStt({ dataDir, modelDir, env, engine: async () => fakeEngine({}, built), log: () => {} }); + const s = await stt.openStream({ model: "nemo-fast-conformer-en-80ms", partialModel: null, hotwords: ["x"] }); + assert.equal(s.model, "nemo-fast-conformer-en-80ms"); + assert.equal(built[0]!["decodingMethod"], "greedy_search"); + }); + + it("transcribe runs the single-recognizer path over a WAV and joins the finals", async () => { + const engine = fakeEngine({ + "zipformer-en-kroko": [ + { at: 1.0, text: "one", endpoint: true }, + { at: 1.0, text: "two", endpoint: true }, + ], + }); + const stt = createStt({ dataDir, modelDir, env, engine: async () => engine, log: () => {} }); + const r = await stt.transcribe(new Uint8Array(64)); + assert.equal(r.model, "zipformer-en-kroko"); + assert.ok(r.segments.length >= 1); + assert.equal(r.text, r.segments.map((s) => s.text).join(" ")); + }); + + it("ensureModel rejects unknown ids and dedupes in-flight downloads", async () => { + let fetches = 0; + const fetchImpl = (async () => { + fetches++; + return new Response("not a tarball", { status: 200 }); + }) as unknown as typeof fetch; + const stt = createStt({ dataDir, modelDir, env, engine: async () => fakeEngine({}), fetchImpl, log: () => {} }); + await assert.rejects(() => stt.ensureModel("nope"), /unknown stt model/); + const p1 = stt.ensureModel("nemo-fast-conformer-en-480ms").catch((e: Error) => e.message); + const p2 = stt.ensureModel("nemo-fast-conformer-en-480ms").catch((e: Error) => e.message); + const [m1, m2] = await Promise.all([p1, p2]); + assert.match(String(m1), /sha256 mismatch/); + assert.equal(m1, m2); + assert.equal(fetches, 1); + }); +}); + +describe("helpers", () => { + it("wordsOf groups subword tokens on the ▁/space marker with segment offsets", () => { + assert.deepEqual( + wordsOf({ text: "", tokens: [" A", "s", "k", " not", "▁now"], timestamps: [0.3, 0.4, 0.5, 0.7, 1.0], start_time: 10 }), + [ + { text: "Ask", start: 10.3 }, + { text: "not", start: 10.7 }, + { text: "now", start: 11 }, + ], + ); + }); + + it("pcm16ToFloat32 scales and honours the byte offset", () => { + const buf = Buffer.alloc(8); + buf.writeInt16LE(0, 0); + buf.writeInt16LE(-32768, 2); + buf.writeInt16LE(16384, 4); + buf.writeInt16LE(32767, 6); + const view = new Uint8Array(buf.buffer, buf.byteOffset + 2, 6); + assert.deepEqual(Array.from(pcm16ToFloat32(view)), [-1, 0.5, 32767 / 32768]); + }); +}); diff --git a/vein/src/audio/stt.ts b/vein/src/audio/stt.ts new file mode 100644 index 000000000..3d6e786c9 --- /dev/null +++ b/vein/src/audio/stt.ts @@ -0,0 +1,619 @@ +/** + * Speech-to-text service over sherpa-onnx (plans/local-desktop-and-stt.md §4). + * + * Owns: model download + verification, recognizer construction and caching, + * the streaming session (PCM in → partial/final events out), and the batch + * `transcribe` that runs the same streaming path over a WAV. Routes + * (`routes.ts`) and the WebSocket (`ws.ts`) are thin over this. + * + * `sherpa-onnx-node` is an optionalDependency, imported lazily: a vein without + * the addon boots, and `available()` says so. Tests inject a fake `engine`. + * + * Two-recognizer streams: when `partialModel` is set, a fast greedy model + * produces the partials and the hotword-capable `model` produces finals and + * owns endpoint detection (see §4.4 for why — sherpa's NeMo online + * transducer is greedy-only, so speed and biasing come from different + * families). + */ +import { createHash } from "node:crypto"; +import { createWriteStream } from "node:fs"; +import { mkdir, readdir, rename, rm, stat, writeFile } from "node:fs/promises"; +import { spawn } from "node:child_process"; +import { join } from "node:path"; +import { pipeline } from "node:stream/promises"; +import { Transform } from "node:stream"; +import { randomUUID } from "node:crypto"; +import { compileHotwords, parseHotwords, HotwordsStore, type Hotword } from "./hotwords.js"; +import { + DEFAULT_MODEL, + DEFAULT_PARTIAL_MODEL, + STT_MODELS, + modelDirFromEnv, + pickModelFiles, + requireModel, + sttModelPath, + type SttModel, +} from "./models.js"; +import { SessionStore } from "./sessions.js"; + +// ── Engine surface (what we use of sherpa-onnx-node; fakeable) ──────────── + +export interface EngineWaveform { + samples: Float32Array; + sampleRate: number; +} +export interface EngineResult { + text: string; + tokens?: string[]; + timestamps?: number[]; + start_time?: number; +} +export interface EngineStream { + acceptWaveform(w: EngineWaveform): void; + inputFinished(): void; +} +export interface EngineRecognizer { + createStream(): EngineStream; + isReady(s: EngineStream): boolean; + decode(s: EngineStream): void; + isEndpoint(s: EngineStream): boolean; + reset(s: EngineStream): void; + getResult(s: EngineStream): EngineResult; +} +export interface SttEngine { + OnlineRecognizer: new (config: Record) => EngineRecognizer; + readWave(path: string): EngineWaveform; +} + +/** Lazy sherpa import; null when the optional addon isn't installed. */ +export async function loadSherpaEngine(): Promise { + try { + const mod = (await import("sherpa-onnx-node")) as { default?: SttEngine } & SttEngine; + return mod.default ?? mod; + } catch { + return null; + } +} + +// ── Public types ─────────────────────────────────────────────────────────── + +export interface SttWord { + text: string; + /** Seconds from the start of the stream. */ + start: number; +} + +export type SttEvent = + | { type: "partial"; text: string } + | { type: "final"; index: number; text: string; words: SttWord[] }; + +export interface EndpointRules { + /** Trailing silence (s) that ends a segment even with no speech yet. */ + rule1?: number; + /** Trailing silence (s) after speech that ends a segment. */ + rule2?: number; + /** Utterance length (s) after which a segment is cut regardless. */ + rule3?: number; +} + +export interface SttStreamOptions { + /** Finals model (hotword-capable). Default `VEIN_STT_MODEL` / catalog default. */ + model?: string; + /** Fast model for partials; `null` for single-recognizer mode. Default + * `VEIN_STT_PARTIAL_MODEL` / catalog default. */ + partialModel?: string | null; + /** A stored list name, or phrases inline. */ + hotwords?: string | readonly string[] | readonly Hotword[]; + /** Global per-token boost for phrases without their own `:score`. */ + hotwordsScore?: number; + sampleRate?: number; + /** Log finals + accept corrections under this id (`SessionStore`). */ + session?: string; + endpoint?: EndpointRules; +} + +export interface SttStream { + readonly model: string; + readonly partialModel: string | null; + readonly hotwords: string | null; + /** Feed PCM16LE bytes at the stream's sample rate. */ + push(pcm16le: Uint8Array): SttEvent[]; + /** Feed float32 samples (any rate; sherpa resamples). */ + pushSamples(samples: Float32Array, sampleRate: number): SttEvent[]; + /** Flush: pads silence so the last chunk decodes, emits the trailing final. */ + end(): SttEvent[]; + /** Resolves once every session-log write so far has landed — await it + * before telling a client it may correct the finals. */ + flush(): Promise; + close(): void; +} + +export interface TranscribeResult { + text: string; + segments: { text: string; words: SttWord[] }[]; + model: string; + hotwords: string | null; + durationMs: number; +} + +export type DownloadProgress = + | { phase: "download"; received: number; total: number } + | { phase: "extract" } + | { phase: "done" }; + +export interface SttModelStatus extends SttModel { + installed: boolean; + /** Which entry env/default resolution picks. */ + default: "model" | "partialModel" | null; +} + +export interface SttService { + readonly modelDir: string; + readonly hotwords: HotwordsStore; + readonly sessions: SessionStore; + /** Whether the sherpa addon loads in this process. */ + available(): Promise; + models(): Promise; + /** Download + verify + extract (idempotent, deduped). Resolves to the dir. */ + ensureModel(id: string, onProgress?: (p: DownloadProgress) => void): Promise; + openStream(opts?: SttStreamOptions): Promise; + transcribe(wav: Uint8Array, opts?: Omit): Promise; + /** Append a user's correction to a session's final. */ + correct(session: string, index: number, text: string): Promise; +} + +export interface SttServiceOptions { + /** Local data dir (sessions, hotword lists, temp WAVs). */ + dataDir: string; + modelDir?: string; + engine?: () => Promise; + fetchImpl?: typeof fetch; + env?: Record; + log?: (msg: string) => void; +} + +// ── Implementation ───────────────────────────────────────────────────────── + +const FRAME_SECONDS = 0.1; +const DEFAULT_HOTWORDS_SCORE = 2; +const DEFAULT_ENDPOINT: Required = { rule1: 2.4, rule2: 1.0, rule3: 20 }; + +interface Loaded { + model: SttModel; + dir: string; + files: ReturnType; +} + +export function createStt(opts: SttServiceOptions): SttService { + const env = opts.env ?? process.env; + const modelDir = opts.modelDir ?? modelDirFromEnv(env); + const loadEngine = opts.engine ?? loadSherpaEngine; + const fetchImpl = opts.fetchImpl ?? fetch; + const log = opts.log ?? ((m: string) => console.log(`[stt] ${m}`)); + const hotwords = new HotwordsStore(opts.dataDir); + const sessions = new SessionStore(opts.dataDir); + + let engineP: Promise | undefined; + const engine = () => (engineP ??= loadEngine()); + const requireEngine = async (): Promise => { + const e = await engine(); + if (!e) throw new SttUnavailableError(); + return e; + }; + + const downloads = new Map>(); + const recognizers = new Map>(); + + const defaultModelId = () => env["VEIN_STT_MODEL"] ?? DEFAULT_MODEL; + const defaultPartialId = (): string | null => { + const v = env["VEIN_STT_PARTIAL_MODEL"]; + if (v === "" || v === "none" || v === "off") return null; + return v ?? DEFAULT_PARTIAL_MODEL; + }; + + async function installed(id: string): Promise { + try { + await stat(join(sttModelPath(modelDir, id), ".ok")); + return true; + } catch { + return false; + } + } + + async function ensureModel(id: string, onProgress?: (p: DownloadProgress) => void): Promise { + const model = requireModel(id); + const dir = sttModelPath(modelDir, id); + if (await installed(id)) return dir; + let p = downloads.get(id); + if (!p) { + p = download(model, dir, onProgress).finally(() => downloads.delete(id)); + downloads.set(id, p); + } + return p; + } + + async function download(model: SttModel, dir: string, onProgress?: (p: DownloadProgress) => void): Promise { + await mkdir(join(modelDir, "stt"), { recursive: true }); + const archive = `${dir}.tar.bz2.part`; + log(`downloading ${model.id} (${Math.round(model.bytes / 1e6)} MB)`); + const res = await fetchImpl(model.url); + if (!res.ok || !res.body) throw new Error(`download failed for ${model.id}: HTTP ${res.status}`); + const total = Number(res.headers.get("content-length") ?? model.bytes); + const hash = createHash("sha256"); + let received = 0; + const meter = new Transform({ + transform(chunk: Buffer, _enc, cb) { + hash.update(chunk); + received += chunk.length; + onProgress?.({ phase: "download", received, total }); + cb(null, chunk); + }, + }); + await pipeline(res.body as unknown as NodeJS.ReadableStream, meter, createWriteStream(archive)); + const digest = hash.digest("hex"); + if (digest !== model.sha256) { + await rm(archive, { force: true }); + throw new Error(`sha256 mismatch for ${model.id}: expected ${model.sha256}, got ${digest}`); + } + onProgress?.({ phase: "extract" }); + const scratch = `${dir}.extract`; + await rm(scratch, { recursive: true, force: true }); + await mkdir(scratch, { recursive: true }); + await untar(archive, scratch); + await rm(dir, { recursive: true, force: true }); + await rename(join(scratch, model.archiveDir), dir); + await rm(scratch, { recursive: true, force: true }); + await rm(archive, { force: true }); + await writeFile(join(dir, ".ok"), new Date().toISOString()); + onProgress?.({ phase: "done" }); + log(`installed ${model.id} → ${dir}`); + return dir; + } + + async function load(id: string): Promise { + const model = requireModel(id); + const dir = await ensureModel(id); + const files = pickModelFiles(await readdir(dir)); + return { model, dir, files }; + } + + async function resolveHotwords( + spec: SttStreamOptions["hotwords"], + ): Promise<{ name: string | null; list: Hotword[] }> { + if (spec == null) return { name: null, list: [] }; + if (typeof spec === "string") { + const text = await hotwords.get(spec); + if (text == null) throw new Error(`unknown hotwords list "${spec}"`); + return { name: spec, list: parseHotwords(text) }; + } + const list = spec.map((h) => (typeof h === "string" ? { phrase: h } : h)).filter((h) => h.phrase.trim()); + return { name: list.length ? "inline" : null, list }; + } + + async function recognizer( + loaded: Loaded, + list: readonly Hotword[], + score: number, + ep: Required, + ): Promise { + const e = await requireEngine(); + const biased = list.length > 0 && loaded.model.hotwords; + const compiled = biased ? await compileHotwords(loaded.dir, list) : null; + const key = [loaded.model.id, compiled?.hash ?? "-", biased ? score : "-", ep.rule1, ep.rule2, ep.rule3].join("|"); + let p = recognizers.get(key); + if (!p) { + const { dir, files } = loaded; + const modelConfig: Record = { + transducer: { encoder: join(dir, files.encoder), decoder: join(dir, files.decoder), joiner: join(dir, files.joiner) }, + tokens: join(dir, files.tokens), + numThreads: 2, + provider: "cpu", + debug: 0, + }; + if (compiled) { + // §4.2: without modelingUnit sherpa defaults to cjkchar and the list + // silently does nothing. + modelConfig["modelingUnit"] = "bpe"; + modelConfig["bpeVocab"] = compiled.vocab; + } + const config: Record = { + featConfig: { sampleRate: 16000, featureDim: 80 }, + modelConfig, + decodingMethod: compiled ? "modified_beam_search" : "greedy_search", + maxActivePaths: 4, + enableEndpoint: true, + rule1MinTrailingSilence: ep.rule1, + rule2MinTrailingSilence: ep.rule2, + rule3MinUtteranceLength: ep.rule3, + }; + if (compiled) { + config["hotwordsFile"] = compiled.file; + config["hotwordsScore"] = score; + } + p = Promise.resolve().then(() => { + const t0 = Date.now(); + const r = new e.OnlineRecognizer(config); + log(`recognizer ${loaded.model.id}${compiled ? ` +hotwords(${compiled.hash})` : ""} ready in ${Date.now() - t0} ms`); + return r; + }); + p.catch(() => recognizers.delete(key)); + recognizers.set(key, p); + } + return p; + } + + async function openStream(o: SttStreamOptions = {}): Promise { + const modelId = o.model ?? defaultModelId(); + const partialId = o.partialModel === undefined ? defaultPartialId() : o.partialModel; + const ep = { ...DEFAULT_ENDPOINT, ...stripUndefined(o.endpoint ?? {}) }; + const { name: hotwordsName, list } = await resolveHotwords(o.hotwords); + const score = o.hotwordsScore ?? DEFAULT_HOTWORDS_SCORE; + + const main = await load(modelId); + const mainRec = await recognizer(main, list, score, ep); + let partialRec: EngineRecognizer | null = null; + if (partialId && partialId !== modelId) { + partialRec = await recognizer(await load(partialId), [], score, ep); + } + // Final indexes continue across connections that share a session, so a + // correction's `index` is unambiguous within the session file. + const firstIndex = o.session ? await sessions.nextIndex(o.session) : 0; + return new Stream({ + model: main.model, + partialModel: partialId && partialRec ? partialId : null, + hotwords: hotwordsName, + mainRec, + partialRec, + sampleRate: o.sampleRate ?? 16000, + session: o.session ?? null, + sessions, + firstIndex, + }); + } + + async function transcribe(wav: Uint8Array, o: Omit = {}): Promise { + const e = await requireEngine(); + const tmp = join(opts.dataDir, "audio", "tmp"); + await mkdir(tmp, { recursive: true }); + const path = join(tmp, `${randomUUID()}.wav`); + await writeFile(path, wav); + let wave: EngineWaveform; + try { + wave = e.readWave(path); + } finally { + await rm(path, { force: true }); + } + // Batch runs single-recognizer: partials are irrelevant here. + const stream = await openStream({ ...o, partialModel: null, sampleRate: wave.sampleRate }); + const t0 = Date.now(); + const finals: SttEvent[] = []; + const frame = Math.max(1, Math.round(FRAME_SECONDS * wave.sampleRate)); + try { + for (let i = 0; i < wave.samples.length; i += frame) { + finals.push(...stream.pushSamples(wave.samples.subarray(i, i + frame), wave.sampleRate).filter((ev) => ev.type === "final")); + } + finals.push(...stream.end().filter((ev) => ev.type === "final")); + } finally { + stream.close(); + } + const segments = finals.flatMap((ev) => (ev.type === "final" ? [{ text: ev.text, words: ev.words }] : [])); + return { + text: segments.map((s) => s.text).join(" "), + segments, + model: stream.model, + hotwords: stream.hotwords, + durationMs: Date.now() - t0, + }; + } + + return { + modelDir, + hotwords, + sessions, + available: async () => (await engine()) != null, + async models() { + const d = defaultModelId(); + const pd = defaultPartialId(); + return Promise.all( + STT_MODELS.map(async (m) => ({ + ...m, + installed: await installed(m.id), + default: m.id === d ? ("model" as const) : m.id === pd ? ("partialModel" as const) : null, + })), + ); + }, + ensureModel, + openStream, + transcribe, + correct: (session, index, text) => sessions.append(session, { type: "correction", t: new Date().toISOString(), index, text }), + }; +} + +export class SttUnavailableError extends Error { + constructor() { + super("stt not available: sherpa-onnx-node is not installed for this platform"); + this.name = "SttUnavailableError"; + } +} + +// ── The streaming session ────────────────────────────────────────────────── + +interface StreamDeps { + model: SttModel; + partialModel: string | null; + hotwords: string | null; + mainRec: EngineRecognizer; + partialRec: EngineRecognizer | null; + sampleRate: number; + session: string | null; + sessions: SessionStore; + firstIndex: number; +} + +class Stream implements SttStream { + readonly model: string; + readonly partialModel: string | null; + readonly hotwords: string | null; + private readonly d: StreamDeps; + private readonly main: EngineStream; + private readonly partial: EngineStream | null; + private lastPartial = ""; + private index = 0; + private closed = false; + private writes: Promise = Promise.resolve(); + /** Rate of the last audio fed — sherpa aborts the process if a stream's + * rate changes, so the flush pad must match it. */ + private rate: number; + private fed = false; + + constructor(d: StreamDeps) { + this.d = d; + this.model = d.model.id; + this.partialModel = d.partialModel; + this.hotwords = d.hotwords; + this.main = d.mainRec.createStream(); + this.partial = d.partialRec ? d.partialRec.createStream() : null; + this.index = d.firstIndex; + this.rate = d.sampleRate; + } + + push(pcm16le: Uint8Array): SttEvent[] { + return this.pushSamples(pcm16ToFloat32(pcm16le), this.d.sampleRate); + } + + pushSamples(samples: Float32Array, sampleRate: number): SttEvent[] { + if (this.closed) throw new Error("stream is closed"); + this.feed(samples, sampleRate); + const events: SttEvent[] = []; + const text = this.currentPartialText(); + if (text !== this.lastPartial) { + this.lastPartial = text; + events.push({ type: "partial", text }); + } + if (this.d.mainRec.isEndpoint(this.main)) { + const fin = this.takeFinal(); + if (fin) events.push(fin); + this.resetAll(); + } + return events; + } + + end(): SttEvent[] { + if (this.closed) return []; + const pad = Math.round((this.d.model.tailPadMs / 1000) * this.rate); + this.feed(new Float32Array(pad), this.rate); + this.main.inputFinished(); + while (this.d.mainRec.isReady(this.main)) this.d.mainRec.decode(this.main); + const events: SttEvent[] = []; + const fin = this.takeFinal(); + if (fin) events.push(fin); + this.close(); + return events; + } + + flush(): Promise { + return this.writes; + } + + close(): void { + this.closed = true; + } + + private feed(samples: Float32Array, sampleRate: number): void { + if (sampleRate !== this.rate) { + if (this.rate !== this.d.sampleRate || samples.length) { + // Changing rate mid-stream is fatal in sherpa (it exits the process). + if (this.fed) throw new Error(`sample rate changed mid-stream (${this.rate} → ${sampleRate})`); + } + this.rate = sampleRate; + } + this.fed = true; + const w = { samples, sampleRate }; + this.main.acceptWaveform(w); + while (this.d.mainRec.isReady(this.main)) this.d.mainRec.decode(this.main); + if (this.partial && this.d.partialRec) { + this.partial.acceptWaveform(w); + while (this.d.partialRec.isReady(this.partial)) this.d.partialRec.decode(this.partial); + } + } + + private currentPartialText(): string { + const r = this.partial && this.d.partialRec ? this.d.partialRec.getResult(this.partial) : this.d.mainRec.getResult(this.main); + return r.text.trim(); + } + + private takeFinal(): SttEvent | null { + const r = this.d.mainRec.getResult(this.main); + const text = r.text.trim(); + if (!text) return null; + const ev: SttEvent = { type: "final", index: this.index++, text, words: wordsOf(r) }; + if (this.d.session) { + const entry = { + type: "final" as const, + t: new Date().toISOString(), + index: ev.index, + text, + words: ev.words, + model: this.model, + hotwords: this.hotwords, + }; + const session = this.d.session; + this.writes = this.writes + .then(() => this.d.sessions.append(session, entry)) + .catch((e) => console.warn(`[stt] session log failed:`, e)); + } + return ev; + } + + private resetAll(): void { + this.d.mainRec.reset(this.main); + if (this.partial && this.d.partialRec) this.d.partialRec.reset(this.partial); + this.lastPartial = ""; + } +} + +/** Group sherpa's subword tokens into words. A token starting with `▁` (or + * a space, as the JSON renders it) begins a word. */ +export function wordsOf(r: EngineResult): SttWord[] { + const tokens = r.tokens ?? []; + const ts = r.timestamps ?? []; + const base = r.start_time ?? 0; + const words: SttWord[] = []; + tokens.forEach((tok, i) => { + const starts = tok.startsWith("▁") || tok.startsWith(" "); + const piece = starts ? tok.slice(1) : tok; + const last = words[words.length - 1]; + if (starts || !last) words.push({ text: piece, start: round(base + (ts[i] ?? 0)) }); + else last.text += piece; + }); + return words.filter((w) => w.text.length > 0); +} + +export function pcm16ToFloat32(bytes: Uint8Array): Float32Array { + const n = bytes.byteLength >> 1; + const out = new Float32Array(n); + const view = new DataView(bytes.buffer, bytes.byteOffset, n * 2); + for (let i = 0; i < n; i++) out[i] = view.getInt16(i * 2, true) / 32768; + return out; +} + +function round(n: number): number { + return Math.round(n * 1000) / 1000; +} + +function stripUndefined(o: T): Partial { + return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== undefined)) as Partial; +} + +/** `tar xjf` — bsdtar (macOS, Windows 10+) and GNU tar both do bz2; Node's + * zlib has no bzip2. */ +function untar(archive: string, into: string): Promise { + return new Promise((resolve, reject) => { + const p = spawn("tar", ["xjf", archive, "-C", into], { stdio: ["ignore", "ignore", "pipe"] }); + let err = ""; + p.stderr.on("data", (d) => (err += d)); + p.on("error", reject); + p.on("close", (code) => (code === 0 ? resolve() : reject(new Error(`tar exited ${code}: ${err.trim()}`)))); + }); +} + diff --git a/vein/src/audio/ws.test.ts b/vein/src/audio/ws.test.ts new file mode 100644 index 000000000..1739ccd0e --- /dev/null +++ b/vein/src/audio/ws.test.ts @@ -0,0 +1,161 @@ +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import { createServer, type Server } from "node:http"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import WebSocket from "ws"; +import { attachAudioWebSocket } from "./ws.js"; +import { _resetAuthState } from "../auth.js"; +import type { SttEvent, SttService, SttStream, SttStreamOptions } from "./stt.js"; + +/** A service whose streams echo frame counts as partials and finalize on end. */ +function fakeService(overrides: Partial = {}): SttService & { opened: SttStreamOptions[] } { + const opened: SttStreamOptions[] = []; + const svc = { + opened, + modelDir: "/nowhere", + hotwords: {} as SttService["hotwords"], + sessions: {} as SttService["sessions"], + available: async () => true, + models: async () => [], + ensureModel: async () => "/nowhere", + transcribe: async () => { + throw new Error("unused"); + }, + correct: async () => {}, + async openStream(o: SttStreamOptions = {}): Promise { + opened.push(o); + let frames = 0; + let bytes = 0; + const stream: SttStream = { + model: o.model ?? "fake", + partialModel: o.partialModel ?? null, + hotwords: typeof o.hotwords === "string" ? o.hotwords : null, + push(b) { + frames++; + bytes += b.byteLength; + const ev: SttEvent[] = [{ type: "partial", text: `frame ${frames}` }]; + if (frames === 3) ev.push({ type: "final", index: 0, text: "three frames", words: [] }); + return ev; + }, + pushSamples: () => [], + end: () => [{ type: "final", index: 1, text: `done ${bytes} bytes`, words: [] }], + flush: async () => {}, + close() {}, + }; + return stream; + }, + ...overrides, + }; + return svc as SttService & { opened: SttStreamOptions[] }; +} + +async function listen(server: Server): Promise { + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + return (server.address() as { port: number }).port; +} + +function connect(url: string, headers?: Record): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url, { headers }); + ws.once("open", () => resolve(ws)); + ws.once("error", reject); + ws.once("unexpected-response", (_req, res) => reject(new Error(`HTTP ${res.statusCode}`))); + }); +} + +/** Collect every JSON message until the socket closes. */ +function collect(ws: WebSocket): Promise[]> { + const out: Record[] = []; + return new Promise((resolve) => { + ws.on("message", (d) => out.push(JSON.parse(d.toString()))); + ws.on("close", () => resolve(out)); + }); +} + +describe("/audio/stream websocket", () => { + let server: Server; + let port: number; + let detach: () => void; + let dir: string; + const originalKey = process.env["VEIN_API_KEY"]; + + beforeEach(async () => { + _resetAuthState(); + delete process.env["VEIN_API_KEY"]; + dir = await mkdtemp(join(tmpdir(), "vein-ws-")); + server = createServer((_req, res) => res.writeHead(404).end()); + }); + afterEach(async () => { + detach?.(); + await new Promise((r) => server.close(() => r())); + await rm(dir, { recursive: true, force: true }); + if (originalKey === undefined) delete process.env["VEIN_API_KEY"]; + else process.env["VEIN_API_KEY"] = originalKey; + _resetAuthState(); + }); + + it("start → binary frames → end yields ready, partials, finals, then a clean close", async () => { + const svc = fakeService(); + detach = attachAudioWebSocket(server, svc); + port = await listen(server); + const ws = await connect(`ws://127.0.0.1:${port}/audio/stream`); + const done = collect(ws); + ws.send(JSON.stringify({ type: "start", model: "m", hotwords: "team", sampleRate: 16000, session: "s" })); + for (let i = 0; i < 3; i++) ws.send(Buffer.alloc(320)); + ws.send(JSON.stringify({ type: "end" })); + const msgs = await done; + assert.deepEqual(msgs, [ + { type: "ready", model: "m", partialModel: null, hotwords: "team" }, + { type: "partial", text: "frame 1" }, + { type: "partial", text: "frame 2" }, + { type: "partial", text: "frame 3" }, + { type: "final", index: 0, text: "three frames", words: [] }, + { type: "final", index: 1, text: "done 960 bytes", words: [] }, + ]); + assert.deepEqual(svc.opened, [{ model: "m", hotwords: "team", sampleRate: 16000, session: "s" }]); + }); + + it("audio before start is an error; a bad list name surfaces as an error message", async () => { + detach = attachAudioWebSocket(server, fakeService()); + port = await listen(server); + const ws = await connect(`ws://127.0.0.1:${port}/audio/stream`); + const done = collect(ws); + ws.send(Buffer.alloc(10)); + const msgs = await done; + assert.equal(msgs[0]?.type, "error"); + assert.match(String(msgs[0]?.error), /start/); + + const failing = fakeService({ + openStream: async () => { + throw new Error("unknown hotwords list \"nope\""); + }, + }); + detach(); + detach = attachAudioWebSocket(server, failing); + const ws2 = await connect(`ws://127.0.0.1:${port}/audio/stream`); + const done2 = collect(ws2); + ws2.send(JSON.stringify({ type: "start", hotwords: "nope" })); + const msgs2 = await done2; + assert.deepEqual(msgs2, [{ type: "error", error: 'unknown hotwords list "nope"' }]); + }); + + it("leaves other upgrade paths alone", async () => { + detach = attachAudioWebSocket(server, fakeService()); + port = await listen(server); + await assert.rejects(() => connect(`ws://127.0.0.1:${port}/other`)); + }); + + it("with VEIN_API_KEY set: rejects without a key, accepts Bearer or ?key=", async () => { + process.env["VEIN_API_KEY"] = "sekret"; + detach = attachAudioWebSocket(server, fakeService()); + port = await listen(server); + await assert.rejects(() => connect(`ws://127.0.0.1:${port}/audio/stream`), /HTTP 401/); + await assert.rejects(() => connect(`ws://127.0.0.1:${port}/audio/stream?key=wrong`), /HTTP 401/); + const a = await connect(`ws://127.0.0.1:${port}/audio/stream?key=sekret`); + a.close(); + const b = await connect(`ws://127.0.0.1:${port}/audio/stream`, { authorization: "Bearer sekret" }); + b.close(); + }); +}); diff --git a/vein/src/audio/ws.ts b/vein/src/audio/ws.ts new file mode 100644 index 000000000..8f3378585 --- /dev/null +++ b/vein/src/audio/ws.ts @@ -0,0 +1,184 @@ +/** + * `GET /audio/stream` — live dictation over a WebSocket + * (plans/local-desktop-and-stt.md §4.4). + * + * Protocol (client → server): + * {"type":"start", model?, partialModel?, hotwords?, hotwordsScore?, + * sampleRate?, session?, endpoint?} — SttStreamOptions, verbatim + * — PCM16LE at sampleRate + * {"type":"end"} — flush; server closes after the final + * Server → client: + * {"type":"ready", model, partialModel, hotwords} + * {"type":"partial", text} + * {"type":"final", index, text, words} + * {"type":"error", error} — then close + * + * Hono's node-ws adapter doesn't support @hono/node-server 2.x yet, and the + * upgrade has to happen on the Node server anyway, so this hooks `ws` + * straight onto the http.Server that `listen()` creates. Vein's first + * client-to-server streaming route; the SSE-based rest is untouched. + * + * Auth: `Authorization: Bearer ` or `?key=` — a browser's + * WebSocket cannot set headers. + */ +import type { IncomingMessage, Server } from "node:http"; +import type { Duplex } from "node:stream"; +import { WebSocketServer, type WebSocket } from "ws"; +import { apiKeyMatches } from "../auth.js"; +import type { SttService, SttStream, SttStreamOptions } from "./stt.js"; + +export const AUDIO_STREAM_PATH = "/audio/stream"; + +export interface AttachOptions { + /** Mount prefix when vein sits under a parent router (e.g. `/lab`). */ + basePath?: string; + path?: string; + /** Replace the default `VEIN_API_KEY` check (Bearer or `?key=`) — a host + * that gates vein behind its own credential applies it here, since an + * upgrade bypasses its HTTP middleware. */ + authorize?: (req: IncomingMessage, url: URL) => boolean; +} + +export interface AudioUpgradeHandler { + /** Route path this handler owns (`basePath + path`). */ + readonly path: string; + /** Handle one `upgrade` event. Returns false (and touches nothing) when + * the request isn't for this path, so the caller can fall through. */ + handle(req: IncomingMessage, socket: Duplex, head: Buffer): boolean; + close(): void; +} + +/** The dictation socket's upgrade handler, for hosts that own the Node + * server themselves (e.g. an Express app that bridges `vein.app`). */ +export function createAudioUpgradeHandler(stt: SttService, opts: AttachOptions = {}): AudioUpgradeHandler { + const path = (opts.basePath ?? "") + (opts.path ?? AUDIO_STREAM_PATH); + const wss = new WebSocketServer({ noServer: true }); + const authorize = + opts.authorize ?? ((req: IncomingMessage, url: URL) => apiKeyMatches(req.headers.authorization, url.searchParams.get("key"))); + return { + path, + handle(req, socket, head) { + const url = new URL(req.url ?? "/", "http://localhost"); + if (url.pathname !== path) return false; + if (!authorize(req, url)) { + socket.write("HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return true; + } + wss.handleUpgrade(req, socket, head, (ws) => handleSocket(ws, stt)); + return true; + }, + close: () => wss.close(), + }; +} + +/** Attach the dictation socket to a Node http server. Returns a detach fn. */ +export function attachAudioWebSocket(server: Server, stt: SttService, opts: AttachOptions = {}): () => void { + const handler = createAudioUpgradeHandler(stt, opts); + const onUpgrade = (req: IncomingMessage, socket: Duplex, head: Buffer) => { + if (handler.handle(req, socket, head)) return; + // Not ours. If nobody else handles upgrades the socket would hang open + // forever, so answer 404 when we're the only listener. + if (server.listenerCount("upgrade") === 1) { + socket.write("HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n"); + socket.destroy(); + } + }; + server.on("upgrade", onUpgrade); + return () => { + server.off("upgrade", onUpgrade); + handler.close(); + }; +} + +/** Drive one socket. Exported for tests (any `ws`-shaped socket works). */ +export function handleSocket(ws: WebSocket, stt: SttService): void { + let stream: SttStream | null = null; + let opening: Promise | null = null; + let finished = false; + + const send = (msg: Record) => { + if (ws.readyState === ws.OPEN) ws.send(JSON.stringify(msg)); + }; + const fail = (err: unknown) => { + send({ type: "error", error: err instanceof Error ? err.message : String(err) }); + ws.close(1011, "error"); + }; + const emit = (events: ReturnType) => { + for (const ev of events) send(ev); + }; + + ws.on("message", (data, isBinary) => { + if (finished) return; + if (isBinary) { + if (!stream) { + if (opening) { + // Audio arriving before the recognizer is ready: wait, then feed. + const chunk = toBytes(data); + opening.then(() => stream && emit(stream.push(chunk))).catch(fail); + return; + } + return fail(new Error('send {"type":"start"} before audio')); + } + try { + emit(stream.push(toBytes(data))); + } catch (e) { + fail(e); + } + return; + } + + let msg: { type?: string } & SttStreamOptions; + try { + msg = JSON.parse(toBytes(data).toString()); + } catch { + return fail(new Error("expected JSON control message or binary PCM")); + } + if (msg.type === "start") { + if (stream || opening) return fail(new Error("stream already started")); + const { type: _t, ...o } = msg; + opening = stt + .openStream(o) + .then((s) => { + stream = s; + send({ type: "ready", model: s.model, partialModel: s.partialModel, hotwords: s.hotwords }); + }) + .catch((e) => { + opening = null; + fail(e); + }); + return; + } + if (msg.type === "end") { + finished = true; + const finish = async () => { + if (stream) { + try { + emit(stream.end()); + } catch (e) { + return fail(e); + } + // Session finals are on disk before the client can correct them. + await stream.flush(); + } + ws.close(1000, "done"); + }; + if (opening) opening.then(finish).catch(fail); + else finish().catch(fail); + return; + } + fail(new Error(`unknown message type ${JSON.stringify(msg.type)}`)); + }); + + ws.on("close", () => { + stream?.close(); + stream = null; + }); +} + +function toBytes(data: unknown): Buffer { + if (Buffer.isBuffer(data)) return data; + if (Array.isArray(data)) return Buffer.concat(data as Buffer[]); + if (data instanceof ArrayBuffer) return Buffer.from(data); + return Buffer.from(String(data)); +} diff --git a/vein/src/auth.ts b/vein/src/auth.ts index b120ee2f9..9138134f2 100644 --- a/vein/src/auth.ts +++ b/vein/src/auth.ts @@ -39,14 +39,7 @@ export function warnIfUnconfigured(): void { * token matching `VEIN_API_KEY`. Permissive when the env var is unset. */ export async function requireApiKey(c: Context, next: Next) { - const expected = configuredKey(); - if (!expected) return next(); // permissive (dev mode) - - const header = c.req.header("authorization") ?? ""; - const match = header.match(/^Bearer\s+(.+)$/i); - const got = match?.[1]?.trim(); - - if (!got || got !== expected) { + if (!apiKeyMatches(c.req.header("authorization"))) { return c.json( { error: "unauthorized: valid Authorization: Bearer required" }, 401, @@ -56,6 +49,20 @@ export async function requireApiKey(c: Context, next: Next) { return next(); } +/** + * Does a request carry the deployment key? Accepts `Authorization: Bearer` + * and, when the caller passes it, a `?key=` query value — the WebSocket + * dictation route needs the latter because a browser's WebSocket cannot set + * headers. Permissive (true) when `VEIN_API_KEY` is unset. + */ +export function apiKeyMatches(authorization: string | undefined, queryKey?: string | null): boolean { + const expected = configuredKey(); + if (!expected) return true; + const match = (authorization ?? "").match(/^Bearer\s+(.+)$/i); + const got = match?.[1]?.trim() || queryKey?.trim(); + return !!got && got === expected; +} + /** Test-only: reset the one-time-warning state so tests stay deterministic. */ export function _resetAuthState(): void { warned = false; diff --git a/vein/src/capabilities.ts b/vein/src/capabilities.ts index a09570669..eaaa8c453 100644 --- a/vein/src/capabilities.ts +++ b/vein/src/capabilities.ts @@ -1,3 +1,4 @@ +import type { SttService } from "./audio/stt.js"; /** * Standard "capabilities" — the small, generic, host-owned services that * LLM-authored adapter STEPS build on (see AGENTS.md "step vs service"). @@ -281,6 +282,10 @@ export interface VeinCapabilities { /** Per-run artifact files. Present on the standard server (rooted in the * workspace); optional because a bare in-code bag may not carry one. */ artifacts?: ArtifactsCapability; + /** Speech-to-text (src/audio). Present on the standard server; a step can + * transcribe through it without importing sherpa. Optional because a + * bare in-code bag may not carry one. */ + stt?: SttService; } /** The default standard services bag: global-fetch http + secrets. Secrets are diff --git a/vein/src/createVein.ts b/vein/src/createVein.ts index e502e9064..804c86827 100644 --- a/vein/src/createVein.ts +++ b/vein/src/createVein.ts @@ -39,6 +39,9 @@ import { buildAuthoringCapability } from "./authoring.js"; import type { CassetteMode } from "./cassette.js"; // Type-only: the graph backend stays a lazy, opt-in dependency. import type { GraphBackend } from "./graph/backend.js"; +import { createStt, type SttService } from "./audio/stt.js"; +import { audioRoutes } from "./audio/routes.js"; +import { attachAudioWebSocket } from "./audio/ws.js"; // Static import is safe: notifier depends only on chat-store, never the AI // SDK (which stays lazy-loaded inside launchChatTurn). import { createChatNotifier, formatRunNotification } from "./ai/notifier.js"; @@ -143,6 +146,14 @@ export interface VeinOptions { * relative asset paths, so it can be mounted at any sub-path (e.g. * `/lab`) as long as the host serves it with a trailing slash. */ webDist?: string; + + /** Speech-to-text (src/audio): the `/audio/*` routes, the `/audio/stream` + * WebSocket (attached by `listen()`), and `ctx.services.stt`. Defaults + * to a service rooted at `dataDir` with models under `VEIN_MODEL_DIR`; + * pass your own, or `false` to mount nothing. The sherpa addon is an + * optionalDependency loaded on first use, so the default costs nothing + * at boot and the routes answer 501 without it. */ + stt?: SttService | false; } export interface AutoResumeOptions { @@ -207,6 +218,11 @@ export interface Vein { opts?: VeinRunOptions, ) => Promise; + /** The speech-to-text service behind `/audio/*` (null when disabled). A + * host that mounts `app` itself must call `attachAudioWebSocket(server, + * vein.stt)` to get the dictation socket; `listen()` does it. */ + stt: SttService | null; + /** Boot the Hono server with `@hono/node-server`. Resolves to the * bound port. Convenience wrapper — feel free to mount `app` yourself. */ listen: (port?: number) => Promise; @@ -418,11 +434,15 @@ export async function createVein( // store (UI-managed) with `process.env` as fallback. This is what lets an // LLM-authored adapter rely on `ctx.services.http` / `ctx.services.secrets` // existing out of the box. + // Speech-to-text: sessions + hotword lists live under dataDir; models under + // VEIN_MODEL_DIR. Nothing loads until a stream or transcribe call. + const stt: SttService | null = opts.stt === false ? null : (opts.stt ?? createStt({ dataDir })); const services = { ...(standardServices({ secretStore }) as unknown as Record), // Per-run artifact files, rooted in the local data dir. A consumer bag // can override with its own ArtifactsCapability (spread below wins). artifacts: fileArtifactsCapability(join(dataDir, "artifacts")), + ...(stt ? { stt } : {}), ...((opts.services ?? {}) as Record), } as TServices; const artifacts = (services as Record)["artifacts"] as @@ -1820,6 +1840,10 @@ export async function createVein( }); }); + // ── Speech-to-text (src/audio) ───────────────────────────────────────── + + if (stt) audioRoutes(app, stt); + // ── Static files (web UI) ──────────────────────────────────────────────── if (serveUi) { @@ -1831,7 +1855,8 @@ export async function createVein( path.startsWith("/workflows") || path.startsWith("/steps") || path.startsWith("/chat") || - path.startsWith("/health") + path.startsWith("/health") || + path.startsWith("/audio") ) { return c.notFound(); } @@ -1899,7 +1924,9 @@ export async function createVein( ); console.log(`vein steps: ${Object.keys(registry).length} registered`); console.log(`vein server: http://localhost:${p}`); - serve({ fetch: app.fetch, port: p }); + const server = serve({ fetch: app.fetch, port: p }); + // The dictation socket upgrades on the Node server, not inside Hono. + if (stt) attachAudioWebSocket(server as unknown as import("node:http").Server, stt); return p; } @@ -1930,6 +1957,7 @@ export async function createVein( rebuildRegistry, autoResumeStaleRuns, run, + stt, listen, }; } diff --git a/vein/src/graph/embeddings.ts b/vein/src/graph/embeddings.ts index 3ae547424..bc499b4bd 100644 --- a/vein/src/graph/embeddings.ts +++ b/vein/src/graph/embeddings.ts @@ -15,7 +15,7 @@ * between MERGE and vector write is healed by `backfillEmbeddings` at boot — * the same NULL-scan idiom jarvis's own migration uses (`migration.py:857`). */ -import { homedir } from "node:os"; +import { modelDirFromEnv } from "../model-dir.js"; import { join } from "node:path"; import { int, type Bolt } from "./bolt.js"; import type { Embedder } from "./node-writer.js"; @@ -29,8 +29,8 @@ export const EMBEDDING_MAX_TOKENS = 256; export interface MiniLMOptions { /** HF repo id of an ONNX export of all-MiniLM-L6-v2. */ model?: string; - /** Where model files are cached. Default `~/.cache/vein-models` - * (override with `VEIN_MODEL_CACHE`). */ + /** Where model files are cached. See src/model-dir.ts for the default + * (`VEIN_MODEL_DIR`, `VEIN_MODEL_CACHE`, or `/vein/models`). */ cacheDir?: string; /** Texts per forward pass. */ batchSize?: number; @@ -52,7 +52,7 @@ export class MiniLMEmbedder implements Embedder { /** Load (downloading on first use) and self-check the output dimension. */ static async load(opts: MiniLMOptions = {}): Promise { const tf: Transformers = await import("@huggingface/transformers"); - tf.env.cacheDir = opts.cacheDir ?? process.env["VEIN_MODEL_CACHE"] ?? join(homedir(), ".cache", "vein-models"); + tf.env.cacheDir = opts.cacheDir ?? modelDirFromEnv(); tf.env.allowLocalModels = false; const model = opts.model ?? EMBEDDING_MODEL; const [tokenizer, net] = await Promise.all([ diff --git a/vein/src/index.ts b/vein/src/index.ts index 2ce0baf4a..a5d376899 100644 --- a/vein/src/index.ts +++ b/vein/src/index.ts @@ -146,6 +146,30 @@ export { type FetchLike, } from "./capabilities.js"; +// Speech-to-text (src/audio): the service behind /audio/*, the dictation +// WebSocket attach for hosts that mount `app` themselves, and the catalog. +export { + createStt, + loadSherpaEngine, + SttUnavailableError, + type SttService, + type SttStream, + type SttStreamOptions, + type SttEvent, + type SttWord, + type TranscribeResult, + type SttEngine, +} from "./audio/stt.js"; +export { + attachAudioWebSocket, + createAudioUpgradeHandler, + AUDIO_STREAM_PATH, + type AttachOptions as AudioAttachOptions, + type AudioUpgradeHandler, +} from "./audio/ws.js"; +export { STT_MODELS, DEFAULT_MODEL as DEFAULT_STT_MODEL, type SttModel } from "./audio/models.js"; +export { parseHotwords, formatHotwords, HotwordsStore, type Hotword } from "./audio/hotwords.js"; + // Secret store — deployment-scoped, encrypted credential persistence behind // the `secrets` capability + the `/secrets` admin endpoints. export { diff --git a/vein/src/model-dir.test.ts b/vein/src/model-dir.test.ts new file mode 100644 index 000000000..04da95c5b --- /dev/null +++ b/vein/src/model-dir.test.ts @@ -0,0 +1,21 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { join } from "node:path"; +import { homedir } from "node:os"; +import { modelDirFromEnv, LEGACY_MODEL_DIR } from "./model-dir.js"; + +describe("modelDirFromEnv", () => { + it("prefers VEIN_MODEL_DIR, then the VEIN_MODEL_CACHE alias", () => { + assert.equal(modelDirFromEnv({ VEIN_MODEL_DIR: "/m", VEIN_MODEL_CACHE: "/c", VEIN_CACHE_DIR: "/r" }), "/m"); + assert.equal(modelDirFromEnv({ VEIN_MODEL_CACHE: "/c", VEIN_CACHE_DIR: "/r" }), "/c"); + }); + it("falls back to /vein/models, then XDG_CACHE_HOME", () => { + assert.equal(modelDirFromEnv({ VEIN_CACHE_DIR: "/srv/cache" }), join("/srv/cache", "vein", "models")); + assert.equal(modelDirFromEnv({ XDG_CACHE_HOME: "/xdg" }), join("/xdg", "vein", "models")); + assert.equal(modelDirFromEnv({ VEIN_CACHE_DIR: "/r", XDG_CACHE_HOME: "/xdg" }), join("/r", "vein", "models")); + }); + it("with nothing set uses ~/.cache/vein/models (or the legacy dir if only that exists)", () => { + const got = modelDirFromEnv({}); + assert.ok(got === join(homedir(), ".cache", "vein", "models") || got === LEGACY_MODEL_DIR, got); + }); +}); diff --git a/vein/src/model-dir.ts b/vein/src/model-dir.ts new file mode 100644 index 000000000..294a59328 --- /dev/null +++ b/vein/src/model-dir.ts @@ -0,0 +1,29 @@ +/** + * Where vein caches downloaded model files (MiniLM embeddings, sherpa STT). + * + * Resolution order: + * 1. `VEIN_MODEL_DIR` — explicit. + * 2. `VEIN_MODEL_CACHE` — the older alias MiniLM shipped with. + * 3. `/vein/models`, where the cache root is `VEIN_CACHE_DIR`, + * else `XDG_CACHE_HOME`, else `~/.cache`. Same convention as the GAIA + * checkout in mcp (`/vein/gaia`), so on a server that + * mounts a volume at `~/.cache/vein` — or sets `VEIN_CACHE_DIR` — models + * persist across restarts with no extra configuration. + * 4. Back-compat: with nothing configured, a pre-existing `~/.cache/vein-models` + * (the old default) keeps being used so dev machines don't re-download. + */ +import { existsSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; + +export const LEGACY_MODEL_DIR = join(homedir(), ".cache", "vein-models"); + +export function modelDirFromEnv(env: Record = process.env): string { + const explicit = env["VEIN_MODEL_DIR"] ?? env["VEIN_MODEL_CACHE"]; + if (explicit) return explicit; + const root = env["VEIN_CACHE_DIR"] ?? env["XDG_CACHE_HOME"]; + if (root) return join(root, "vein", "models"); + const modern = join(homedir(), ".cache", "vein", "models"); + if (!existsSync(modern) && existsSync(LEGACY_MODEL_DIR)) return LEGACY_MODEL_DIR; + return modern; +} diff --git a/vein/web/src/api.ts b/vein/web/src/api.ts index 59fc80b4b..f8b298c6f 100644 --- a/vein/web/src/api.ts +++ b/vein/web/src/api.ts @@ -32,6 +32,71 @@ export async function fetchJSON(path: string, opts?: RequestInit): Promise return res.json() as Promise; } +// ── Speech-to-text (src/audio) ───────────────────────────────────────────── + +export interface SttModelStatus { + id: string; + bytes: number; + language: string; + /** How often partials change (ms). */ + chunkMs: number; + /** Accepts a hotwords list. */ + hotwords: boolean; + cased: boolean; + description: string; + installed: boolean; + /** Which entry the server's env/default resolution picks. */ + default: "model" | "partialModel" | null; +} + +export interface SttModelsResponse { + /** Whether the sherpa addon loads on the server. */ + available: boolean; + modelDir: string; + models: SttModelStatus[]; +} + +export const listSttModels = () => fetchJSON("/audio/models"); + +export type SttDownloadProgress = + | { phase: "download"; received: number; total: number } + | { phase: "extract" } + | { phase: "done" }; + +/** Download + verify + extract a catalog model on the server; SSE progress. */ +export async function downloadSttModel( + id: string, + onProgress: (p: SttDownloadProgress) => void, +): Promise { + const res = await fetch(`${BASE}/audio/models/${encodeURIComponent(id)}/download`, { method: "POST" }); + if (!res.ok || !res.body) throw new Error(`download ${id}: ${res.status} ${res.statusText}`); + const reader = res.body.getReader(); + const dec = new TextDecoder(); + let buf = ""; + let event = ""; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buf += dec.decode(value, { stream: true }); + const lines = buf.split("\n"); + buf = lines.pop() ?? ""; + for (const line of lines) { + if (line.startsWith("event:")) event = line.slice(6).trim(); + else if (line.startsWith("data:")) { + const data = JSON.parse(line.slice(5)); + if (event === "error") throw new Error(data.error ?? "download failed"); + onProgress(data as SttDownloadProgress); + } + } + } +} + +/** WebSocket URL for `/audio/stream` on the same origin + mount path. */ +export function sttStreamUrl(): string { + const proto = location.protocol === "https:" ? "wss:" : "ws:"; + return `${proto}//${location.host}${BASE}/audio/stream`; +} + // ── Workflows ────────────────────────────────────────────────────────────── export interface WorkflowEntry { diff --git a/vein/web/src/app.tsx b/vein/web/src/app.tsx index 603478127..e32ec5d7f 100644 --- a/vein/web/src/app.tsx +++ b/vein/web/src/app.tsx @@ -14,6 +14,9 @@ import { ChatFlyout } from "./components/ChatFlyout"; import { CategoryEditor } from "./components/CategoryEditor"; import { CreateDialog } from "./components/CreateDialog"; import { SecretsDialog } from "./components/SecretsDialog"; +import { SettingsDialog } from "./components/SettingsDialog"; +import { GearIcon } from "./icons"; +import { sttSettings as sttStore, type SttSettings } from "./storage"; import { AddStepDialog, StepTypeEntry } from "./components/AddStepDialog"; import { StepEditFlyout } from "./components/StepEditFlyout"; import { StepInfoFlyout } from "./components/StepInfoFlyout"; @@ -81,6 +84,9 @@ export function App() { const [runDrill, setRunDrill] = useState([]); const [showCreate, setShowCreate] = useState(false); const [showSecrets, setShowSecrets] = useState(false); + const [showSettings, setShowSettings] = useState(false); + const [sttSettings, setSttSettings] = useState(() => sttStore.get()); + const updateStt = (next: SttSettings) => { sttStore.set(next); setSttSettings(next); }; const [showAddStep, setShowAddStep] = useState(false); const [stepTypes, setStepTypes] = useState([]); // Sidebar Steps catalog: whether the section is expanded (persisted), and @@ -885,6 +891,9 @@ export function App() { )} + @@ -953,6 +962,9 @@ export function App() { {/* Create dialog */} {showCreate && setShowCreate(false)} onCreate={handleCreate} categories={categories} />} {showSecrets && setShowSecrets(false)} />} + {showSettings && ( + setShowSettings(false)} /> + )} {/* Add step dialog */} {showAddStep && setShowAddStep(false)} />} @@ -960,6 +972,7 @@ export function App() { {/* Chat flyout */} {showChat && ( setShowChat(false)} onWorkflowCreated={async (name) => { await refreshWorkflows(); diff --git a/vein/web/src/components/ChatFlyout.tsx b/vein/web/src/components/ChatFlyout.tsx index 6af294b19..27c4db257 100644 --- a/vein/web/src/components/ChatFlyout.tsx +++ b/vein/web/src/components/ChatFlyout.tsx @@ -2,7 +2,8 @@ import { useState, useCallback, useEffect, useRef } from "preact/hooks"; import * as api from "../api"; import * as storage from "../storage"; import { formatJson } from "../helpers"; -import { CloseIcon, HistoryIcon, CopyIcon, CheckIcon } from "../icons"; +import { CloseIcon, HistoryIcon, CopyIcon, CheckIcon, MicIcon } from "../icons"; +import { startDictation, dictationSupported, type Dictation } from "../dictation"; import { ToolResultView } from "./ToolResultView"; import { FlyoutResizer } from "./FlyoutResizer"; import { Markdown } from "./Markdown"; @@ -214,10 +215,77 @@ export function ChatFlyout(props: { onClose: () => void; onWorkflowCreated: (name: string) => void; onWorkflowRan: (name: string, runId: string) => void; + /** Dictation preferences (SettingsDialog). The mic shows once the chosen + * model is confirmed installed on the server. */ + stt: storage.SttSettings; }) { const [entries, setEntries] = useState([]); const [input, setInput] = useState(""); const [loading, setLoading] = useState(false); + + // ── Dictation ────────────────────────────────────────────────────────── + // Finals accumulate into `dictBase`; the live partial is appended after it + // and replaced by the next final. Typing while listening takes over. + const [micReady, setMicReady] = useState(false); + const [listening, setListening] = useState<"starting" | "on" | "stopping" | false>(false); + const [micError, setMicError] = useState(""); + const dictRef = useRef(null); + const dictBase = useRef(""); + const { stt } = props; + useEffect(() => { + let cancelled = false; + setMicReady(false); + if (!stt.enabled || !stt.model || !dictationSupported()) return; + api.listSttModels().then((r) => { + if (cancelled) return; + const ok = (id: string | null) => !id || !!r.models.find((m) => m.id === id)?.installed; + setMicReady(r.available && ok(stt.model) && ok(stt.partialModel)); + }).catch(() => {}); + return () => { cancelled = true; }; + }, [stt.enabled, stt.model, stt.partialModel]); + useEffect(() => () => dictRef.current?.stop(), []); + + const stopDictation = useCallback(() => { + if (!dictRef.current) return; + setListening("stopping"); + dictRef.current.stop(); + }, []); + + const toggleDictation = useCallback(async () => { + if (listening) return stopDictation(); + setMicError(""); + setListening("starting"); + dictBase.current = input.trim() ? input.replace(/\s*$/, " ") : ""; + try { + dictRef.current = await startDictation({ + model: stt.model!, + partialModel: stt.partialModel, + hotwords: stt.hotwords.split("\n").map((l) => l.trim()).filter(Boolean), + session: chatId ? `chat-${chatId}` : undefined, + onReady: () => setListening("on"), + onPartial: (text) => setInput(dictBase.current + text), + onFinal: (text) => { + // The flush on stop can emit a punctuation-only final ("."): attach + // it to the previous word instead of leaving "word . ". + if (text) { + const base = /^[.,!?;:]/.test(text) ? dictBase.current.trimEnd() : dictBase.current; + dictBase.current = base + text + " "; + } + setInput(dictBase.current); + }, + onError: (msg) => setMicError(msg), + onClose: () => { + dictRef.current = null; + setListening(false); + inputRef.current?.focus(); + }, + }); + } catch (e) { + dictRef.current = null; + setListening(false); + setMicError(e instanceof Error ? e.message : String(e)); + } + }, [listening, stopDictation, input, stt]); const [expanded, setExpanded] = useState>({}); const [chatId, setChatId] = useState(() => new URLSearchParams(location.search).get(CHAT_URL_PARAM) ?? @@ -449,6 +517,7 @@ export function ChatFlyout(props: { const send = useCallback(async () => { const text = input.trim(); if (!text || loading) return; + if (dictRef.current) stopDictation(); setEntries((prev) => [...prev, { kind: "user", content: text }]); setInput(""); @@ -473,7 +542,7 @@ export function ChatFlyout(props: { setEntries((prev) => [...prev, { kind: "text", content: "Error connecting to AI." }]); setLoading(false); } - }, [input, loading, chatId, attach, loadChat]); + }, [input, loading, chatId, attach, loadChat, stopDictation]); const handleKeyDown = (e: KeyboardEvent) => { if (e.key === "Enter" && !e.shiftKey) { @@ -618,16 +687,32 @@ export function ChatFlyout(props: { )} )} + {micError &&
{micError}
}