From 5ca62b0672a83294d75dc77112e6aac622396bd4 Mon Sep 17 00:00:00 2001 From: badcuban <108198679+badcuban@users.noreply.github.com> Date: Sun, 6 Sep 2026 12:37:57 -0400 Subject: [PATCH 1/2] feat(chat): dictate into the composer with local speech-to-text MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There was no way to speak a prompt. The old Codex realtime voice mode needed paid API access and had been switched off, leaving a dead mic slot. The composer now has a push-to-talk mic. Audio goes to the Threadlines server, which runs sherpa-onnx in a child process with a locally downloaded model (Parakeet by default, Moonshine as the small option). The native runtime and the model files are downloaded on first use, so nothing is bundled and nothing leaves the machine. The model is picked in Settings › General › Dictation; hold-to-record and the microphone are per-device client settings. The realtime voice button is no longer rendered; its code stays. --- .../resources/entitlements.mac.inherit.plist | 2 + apps/desktop/resources/entitlements.mac.plist | 2 + .../settings/DesktopClientSettings.test.ts | 2 + apps/desktop/src/window/DesktopWindow.test.ts | 3 + apps/desktop/src/window/DesktopWindow.ts | 23 +- apps/server/src/bin.ts | 9 +- apps/server/src/cli/dictationWorker.ts | 15 + apps/server/src/cli/marketingStudioSeed.ts | 1 + apps/server/src/config.ts | 4 + apps/server/src/dictation/DictationEngine.ts | 308 +++++++++++ .../src/dictation/DictationService.test.ts | 368 +++++++++++++ apps/server/src/dictation/DictationService.ts | 460 ++++++++++++++++ apps/server/src/dictation/catalog.test.ts | 73 +++ apps/server/src/dictation/catalog.ts | 230 ++++++++ apps/server/src/dictation/download.ts | 132 +++++ apps/server/src/dictation/npmTarball.test.ts | 99 ++++ apps/server/src/dictation/npmTarball.ts | 139 +++++ apps/server/src/dictation/worker.ts | 152 ++++++ apps/server/src/server.test.ts | 4 + apps/server/src/server.ts | 6 +- apps/server/src/ws.ts | 32 ++ apps/web/src/components/ChatView.browser.tsx | 9 + apps/web/src/components/ChatView.tsx | 47 -- apps/web/src/components/chat/ChatComposer.tsx | 70 ++- .../chat/ComposerDictationControl.browser.tsx | 214 ++++++++ .../chat/ComposerDictationControl.tsx | 491 ++++++++++++++++++ .../settings/DiagnosticsSettings.tsx | 13 +- .../settings/DictationSettings.browser.tsx | 131 +++++ .../components/settings/DictationSettings.tsx | 209 ++++++++ .../settings/SettingsPanels.browser.tsx | 4 +- .../components/settings/SettingsPanels.tsx | 3 + .../dictation/DictationDownloadProgress.tsx | 46 ++ apps/web/src/dictation/dictationModels.ts | 52 ++ .../web/src/dictation/dictationStatusStore.ts | 101 ++++ .../src/dictation/useDictation.browser.tsx | 206 ++++++++ apps/web/src/dictation/useDictation.ts | 230 ++++++++ .../web/src/dictation/useMicrophoneDevices.ts | 63 +++ apps/web/src/environmentApi.ts | 8 + apps/web/src/lib/formatBytes.ts | 48 ++ apps/web/src/localApi.test.ts | 12 + apps/web/src/realtimeAudio.ts | 62 ++- apps/web/src/rpc/wsRpcClient.ts | 28 + apps/web/test/wsRpcHarness.ts | 1 + packages/contracts/src/dictation.ts | 89 ++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/ipc.ts | 18 + packages/contracts/src/rpc.ts | 57 ++ packages/contracts/src/settings.ts | 18 + scripts/build-desktop-artifact.ts | 10 +- 49 files changed, 4214 insertions(+), 91 deletions(-) create mode 100644 apps/server/src/cli/dictationWorker.ts create mode 100644 apps/server/src/dictation/DictationEngine.ts create mode 100644 apps/server/src/dictation/DictationService.test.ts create mode 100644 apps/server/src/dictation/DictationService.ts create mode 100644 apps/server/src/dictation/catalog.test.ts create mode 100644 apps/server/src/dictation/catalog.ts create mode 100644 apps/server/src/dictation/download.ts create mode 100644 apps/server/src/dictation/npmTarball.test.ts create mode 100644 apps/server/src/dictation/npmTarball.ts create mode 100644 apps/server/src/dictation/worker.ts create mode 100644 apps/web/src/components/chat/ComposerDictationControl.browser.tsx create mode 100644 apps/web/src/components/chat/ComposerDictationControl.tsx create mode 100644 apps/web/src/components/settings/DictationSettings.browser.tsx create mode 100644 apps/web/src/components/settings/DictationSettings.tsx create mode 100644 apps/web/src/dictation/DictationDownloadProgress.tsx create mode 100644 apps/web/src/dictation/dictationModels.ts create mode 100644 apps/web/src/dictation/dictationStatusStore.ts create mode 100644 apps/web/src/dictation/useDictation.browser.tsx create mode 100644 apps/web/src/dictation/useDictation.ts create mode 100644 apps/web/src/dictation/useMicrophoneDevices.ts create mode 100644 apps/web/src/lib/formatBytes.ts create mode 100644 packages/contracts/src/dictation.ts diff --git a/apps/desktop/resources/entitlements.mac.inherit.plist b/apps/desktop/resources/entitlements.mac.inherit.plist index dcbc23b49..ceeb6bb05 100644 --- a/apps/desktop/resources/entitlements.mac.inherit.plist +++ b/apps/desktop/resources/entitlements.mac.inherit.plist @@ -6,5 +6,7 @@ com.apple.security.cs.disable-library-validation + com.apple.security.device.audio-input + diff --git a/apps/desktop/resources/entitlements.mac.plist b/apps/desktop/resources/entitlements.mac.plist index dcbc23b49..ceeb6bb05 100644 --- a/apps/desktop/resources/entitlements.mac.plist +++ b/apps/desktop/resources/entitlements.mac.plist @@ -6,5 +6,7 @@ com.apple.security.cs.disable-library-validation + com.apple.security.device.audio-input + diff --git a/apps/desktop/src/settings/DesktopClientSettings.test.ts b/apps/desktop/src/settings/DesktopClientSettings.test.ts index 917cf1e28..1241d72a9 100644 --- a/apps/desktop/src/settings/DesktopClientSettings.test.ts +++ b/apps/desktop/src/settings/DesktopClientSettings.test.ts @@ -19,6 +19,8 @@ const clientSettings: ClientSettings = { confirmThreadDelete: false, wrapUpThreadsOnPullRequestSettled: true, dismissedProviderUpdateNotificationKeys: [], + dictationHoldToRecord: true, + dictationMicrophoneDeviceId: null, diffChangesOnly: false, diffIgnoreWhitespace: true, diffRenderMode: "stacked", diff --git a/apps/desktop/src/window/DesktopWindow.test.ts b/apps/desktop/src/window/DesktopWindow.test.ts index 16ebbe522..00a9badda 100644 --- a/apps/desktop/src/window/DesktopWindow.test.ts +++ b/apps/desktop/src/window/DesktopWindow.test.ts @@ -62,6 +62,9 @@ function makeFakeBrowserWindow(input?: { openDevTools: vi.fn(), replaceMisspelling: vi.fn(), send: vi.fn(), + session: { + setPermissionRequestHandler: vi.fn(), + }, setWindowOpenHandler: vi.fn(), }; diff --git a/apps/desktop/src/window/DesktopWindow.ts b/apps/desktop/src/window/DesktopWindow.ts index 84dfbf031..0476b2c48 100644 --- a/apps/desktop/src/window/DesktopWindow.ts +++ b/apps/desktop/src/window/DesktopWindow.ts @@ -13,7 +13,7 @@ import * as PlatformError from "effect/PlatformError"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import type * as Electron from "electron"; +import * as Electron from "electron"; import * as DesktopAssets from "../app/DesktopAssets.ts"; import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; @@ -484,6 +484,27 @@ const make = Effect.gen(function* () { } }); + // Composer dictation is the only thing here that asks for a device. Grant + // audio-only capture (behind the OS prompt on macOS) and leave every other + // permission on the answer Electron gives today. + window.webContents.session.setPermissionRequestHandler( + (_contents, permission, callback, details) => { + const mediaTypes = "mediaTypes" in details ? (details.mediaTypes ?? []) : []; + const audioOnly = + permission === "media" && + mediaTypes.length > 0 && + mediaTypes.every((mediaType) => mediaType === "audio"); + if (!audioOnly || environment.platform !== "darwin") { + callback(true); + return; + } + void Electron.systemPreferences + .askForMediaAccess("microphone") + .then((granted) => callback(granted)) + .catch(() => callback(false)); + }, + ); + if ( !environment.marketingCaptureMode && Option.isSome(persistedWindowState) && diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index 00dd4d1fe..8cc69d5d3 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -7,6 +7,7 @@ import { Command } from "effect/unstable/cli"; import * as NetService from "@threadlines/shared/Net"; import packageJson from "../package.json" with { type: "json" }; import { authCommand } from "./cli/auth.ts"; +import { dictationWorkerCommand } from "./cli/dictationWorker.ts"; import { sharedServerCommandFlags } from "./cli/config.ts"; import { assertSingleEffectRuntime } from "./cli/effectRuntimeCheck.ts"; import { projectCommand } from "./cli/project.ts"; @@ -17,7 +18,13 @@ const CliRuntimeLayer = Layer.mergeAll(NodeServices.layer, NetService.layer); export const cli = Command.make("threadlines", { ...sharedServerCommandFlags }).pipe( Command.withDescription("Run the Threadlines server."), Command.withHandler((flags) => runServerCommand(flags)), - Command.withSubcommands([startCommand, serveCommand, authCommand, projectCommand]), + Command.withSubcommands([ + startCommand, + serveCommand, + authCommand, + projectCommand, + dictationWorkerCommand, + ]), ); if (import.meta.main) { diff --git a/apps/server/src/cli/dictationWorker.ts b/apps/server/src/cli/dictationWorker.ts new file mode 100644 index 000000000..5a1bfae3d --- /dev/null +++ b/apps/server/src/cli/dictationWorker.ts @@ -0,0 +1,15 @@ +import * as Effect from "effect/Effect"; +import { Command } from "effect/unstable/cli"; + +import { runDictationWorker } from "../dictation/worker.ts"; + +/** + * Internal entry point the server forks for speech-to-text. Routing it through + * the CLI means the same path (`process.argv[1]`) works in dev, npm and + * desktop builds. The worker exits itself when the parent disconnects, so the + * handler simply never completes. + */ +export const dictationWorkerCommand = Command.make("dictation-worker").pipe( + Command.withDescription("Internal: run the local speech-to-text worker process."), + Command.withHandler(() => Effect.sync(runDictationWorker).pipe(Effect.andThen(Effect.never))), +); diff --git a/apps/server/src/cli/marketingStudioSeed.ts b/apps/server/src/cli/marketingStudioSeed.ts index 913f007f3..17bacb2e5 100644 --- a/apps/server/src/cli/marketingStudioSeed.ts +++ b/apps/server/src/cli/marketingStudioSeed.ts @@ -174,6 +174,7 @@ const makeServerConfig = (input: MarketingStudioSeedInput): ServerConfigShape => environmentIdPath: NodePath.join(stateDir, "environment-id"), serverRuntimeStatePath: NodePath.join(stateDir, "server-runtime.json"), secretsDir: NodePath.join(stateDir, "secrets"), + speechModelsDir: NodePath.join(input.baseDir, "models", "speech"), }; }; diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index 3dfbf31f9..fd3f034fd 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -43,6 +43,8 @@ export interface ServerDerivedPaths { readonly environmentIdPath: string; readonly serverRuntimeStatePath: string; readonly secretsDir: string; + /** Shared by dev and userdata: speech models are large and platform-wide. */ + readonly speechModelsDir: string; } /** @@ -135,6 +137,7 @@ export const deriveServerPaths = Effect.fn(function* ( environmentIdPath: join(stateDir, "environment-id"), serverRuntimeStatePath: join(stateDir, "server-runtime.json"), secretsDir: join(stateDir, "secrets"), + speechModelsDir: join(baseDir, "models", "speech"), }; }); @@ -155,6 +158,7 @@ export const ensureServerDirectories = Effect.fn(function* (derivedPaths: Server fs.makeDirectory(derivedPaths.providerStatusCacheDir, { recursive: true }), fs.makeDirectory(path.dirname(derivedPaths.anonymousIdPath), { recursive: true }), fs.makeDirectory(path.dirname(derivedPaths.serverRuntimeStatePath), { recursive: true }), + fs.makeDirectory(derivedPaths.speechModelsDir, { recursive: true }), ], { concurrency: "unbounded" }, ); diff --git a/apps/server/src/dictation/DictationEngine.ts b/apps/server/src/dictation/DictationEngine.ts new file mode 100644 index 000000000..46406a57d --- /dev/null +++ b/apps/server/src/dictation/DictationEngine.ts @@ -0,0 +1,308 @@ +// @effect-diagnostics nodeBuiltinImport:off - forks the worker child process +/** + * DictationEngine - owns the transcription worker child process. + * + * Exactly one worker exists at a time. It is spawned lazily on the first + * `load`, keeps the recognizer resident between clips, and is killed on a + * crash, on a request timeout, after ten idle minutes, or when the layer's + * scope closes. Callers see the transitions through `changes`. + * + * @module dictation/DictationEngine + */ +import * as childProcess from "node:child_process"; +import * as nodePath from "node:path"; + +import { + DictationError, + type DictationEngineState, + type DictationModelId, +} from "@threadlines/contracts"; +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FiberHandle from "effect/FiberHandle"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import { DICTATION_ADDON_FILE, type DictationLibraryPathEnv } from "./catalog.ts"; +import type { DictationWorkerRequest, DictationWorkerResponse } from "./worker.ts"; + +/** Long enough for a cold model load plus a two-minute clip on slow hardware. */ +const REQUEST_TIMEOUT = Duration.seconds(60); +/** The recognizer holds hundreds of megabytes; drop it when nobody dictates. */ +const IDLE_UNLOAD_DELAY = Duration.minutes(10); +const STDERR_KEEP_CHARS = 2000; + +export interface DictationEngineSnapshot { + readonly state: DictationEngineState; + readonly loadedModel: DictationModelId | null; +} + +export interface DictationEngineLoadInput { + readonly model: DictationModelId; + readonly config: unknown; + readonly runtimeDir: string; + readonly libraryPathEnv: DictationLibraryPathEnv; +} + +export interface DictationEngineShape { + readonly snapshot: Effect.Effect; + /** Emits on every transition; the current value is read via `snapshot`. */ + readonly changes: Stream.Stream; + /** No-op when the same model is already loaded. */ + readonly load: (input: DictationEngineLoadInput) => Effect.Effect; + readonly transcribe: ( + samples: Int16Array, + sampleRate: number, + ) => Effect.Effect; + readonly unload: Effect.Effect; +} + +export class DictationEngine extends Context.Service()( + "threadlines/dictation/DictationEngine", +) {} + +const engineFailed = (message: string) => new DictationError({ code: "engine_failed", message }); + +/** + * Prepends `directory` to a library-path variable, matching the existing key's + * casing so Windows does not end up with both `Path` and `PATH`. + */ +function withLibraryPath( + env: NodeJS.ProcessEnv, + variable: DictationLibraryPathEnv, + directory: string, +): NodeJS.ProcessEnv { + const existingKey = + Object.keys(env).find((key) => key.toLowerCase() === variable.toLowerCase()) ?? variable; + const existingValue = env[existingKey]; + return { + ...env, + [existingKey]: + existingValue && existingValue.length > 0 + ? `${directory}${nodePath.delimiter}${existingValue}` + : directory, + }; +} + +const makeDictationEngine = Effect.gen(function* () { + const stateRef = yield* SubscriptionRef.make({ + state: "idle", + loadedModel: null, + }); + const inFlightRef = yield* Ref.make(false); + const workerLock = yield* Semaphore.make(1); + const idleFiber = yield* FiberHandle.make(); + + let child: childProcess.ChildProcess | undefined; + let loadedModel: DictationModelId | null = null; + let stderrTail = ""; + + // Suspended so the published `loadedModel` is read when the transition runs, + // not when the effect is built. + const setState = (state: DictationEngineState) => + Effect.suspend(() => SubscriptionRef.set(stateRef, { state, loadedModel })); + + const killChild = Effect.sync(() => { + const current = child; + child = undefined; + loadedModel = null; + if (current && current.exitCode === null && current.signalCode === null) { + current.removeAllListeners("exit"); + current.kill(); + } + }); + + const spawnWorker = (input: DictationEngineLoadInput) => + Effect.try({ + try: () => { + const env = withLibraryPath( + { + ...process.env, + // Electron's binary only behaves like Node with this set, and the + // desktop app spawns the worker from inside Electron. + ...(process.versions.electron ? { ELECTRON_RUN_AS_NODE: "1" } : {}), + }, + input.libraryPathEnv, + input.runtimeDir, + ); + const spawned = childProcess.fork(process.argv[1] ?? "", ["dictation-worker"], { + execPath: process.execPath, + // Not inherited: `node --watch src/bin.ts` in dev would otherwise + // hand the worker a `--watch` of its own. + execArgv: [], + env, + serialization: "advanced", + stdio: ["ignore", "pipe", "pipe", "ipc"], + }); + stderrTail = ""; + // Both pipes must be drained or a chatty native library blocks on a + // full buffer; stderr is kept around to explain a crash. + spawned.stdout?.resume(); + spawned.stderr?.on("data", (chunk: Buffer) => { + stderrTail = `${stderrTail}${chunk.toString("utf8")}`.slice(-STDERR_KEEP_CHARS); + }); + spawned.on("exit", () => { + child = undefined; + loadedModel = null; + Effect.runFork(SubscriptionRef.set(stateRef, { state: "idle", loadedModel: null })); + }); + return spawned; + }, + catch: (cause) => engineFailed(`Failed to start the dictation worker: ${String(cause)}`), + }); + + /** + * Sends one request and waits for the matching reply. A worker exit or an + * `error` reply fails the call; interruption detaches the listeners. + */ + const request = ( + message: DictationWorkerRequest, + match: (response: DictationWorkerResponse) => A | undefined, + ) => + Effect.callback((resume) => { + const current = child; + if (!current || !current.connected) { + resume(Effect.fail(engineFailed("The dictation worker is not running."))); + return; + } + + const cleanup = () => { + current.off("message", onMessage); + current.off("exit", onExit); + current.off("error", onError); + }; + const onMessage = (response: DictationWorkerResponse) => { + const matched = match(response); + if (matched !== undefined) { + cleanup(); + resume(Effect.succeed(matched)); + return; + } + if (response.type === "error") { + cleanup(); + resume(Effect.fail(engineFailed(response.message))); + } + }; + const onExit = () => { + cleanup(); + resume( + Effect.fail( + engineFailed( + `The dictation worker exited unexpectedly.${stderrTail ? ` ${stderrTail.trim()}` : ""}`, + ), + ), + ); + }; + const onError = (cause: Error) => { + cleanup(); + resume(Effect.fail(engineFailed(cause.message))); + }; + + current.on("message", onMessage); + current.on("exit", onExit); + current.on("error", onError); + current.send(message); + + return Effect.sync(cleanup); + }).pipe( + Effect.timeoutOrElse({ + duration: REQUEST_TIMEOUT, + orElse: () => Effect.fail(engineFailed("The dictation worker timed out.")), + }), + // A timed-out or crashed worker is not trustworthy; the next call + // respawns from scratch. + Effect.tapError(() => killChild.pipe(Effect.andThen(setState("idle")))), + ); + + const loadUnsafe = (input: DictationEngineLoadInput) => + Effect.gen(function* () { + if (child !== undefined && child.connected && loadedModel === input.model) { + return; + } + if (child !== undefined && child.connected && loadedModel !== input.model) { + yield* killChild; + } + yield* setState("loading"); + if (child === undefined || !child.connected) { + child = yield* spawnWorker(input); + } + yield* request( + { + type: "load", + model: input.model, + addonPath: nodePath.join(input.runtimeDir, DICTATION_ADDON_FILE), + config: input.config, + }, + (response) => (response.type === "loaded" ? true : undefined), + ); + loadedModel = input.model; + yield* setState("ready"); + }); + + const unloadUnsafe = Effect.gen(function* () { + if (child !== undefined && child.connected) { + yield* Effect.sync(() => child?.send({ type: "shutdown" } satisfies DictationWorkerRequest)); + } + yield* killChild; + yield* setState("idle"); + }); + + const scheduleIdleUnload = FiberHandle.run( + idleFiber, + Effect.sleep(IDLE_UNLOAD_DELAY).pipe( + Effect.andThen(workerLock.withPermits(1)(unloadUnsafe)), + Effect.ignoreCause({ log: true }), + ), + ).pipe(Effect.asVoid); + + const load: DictationEngineShape["load"] = (input) => + workerLock.withPermits(1)(loadUnsafe(input)); + + const transcribe: DictationEngineShape["transcribe"] = (samples, sampleRate) => + Effect.gen(function* () { + const acquired = yield* Ref.modify(inFlightRef, (busy) => [!busy, true] as const); + if (!acquired) { + return yield* Effect.fail( + new DictationError({ code: "busy", message: "A dictation request is already running." }), + ); + } + + return yield* workerLock + .withPermits(1)( + Effect.gen(function* () { + yield* FiberHandle.clear(idleFiber); + yield* setState("busy"); + const requestId = crypto.randomUUID(); + const text = yield* request( + { type: "transcribe", requestId, samples, sampleRate }, + (response) => + response.type === "result" && response.requestId === requestId + ? response.text + : undefined, + ); + yield* setState("ready"); + return text; + }), + ) + .pipe( + Effect.ensuring(Ref.set(inFlightRef, false)), + Effect.tap(() => scheduleIdleUnload), + ); + }); + + yield* Effect.addFinalizer(() => FiberHandle.clear(idleFiber).pipe(Effect.andThen(killChild))); + + return { + snapshot: SubscriptionRef.get(stateRef), + changes: SubscriptionRef.changes(stateRef), + load, + transcribe, + unload: workerLock.withPermits(1)(unloadUnsafe), + } satisfies DictationEngineShape; +}); + +export const DictationEngineLive = Layer.effect(DictationEngine, makeDictationEngine); diff --git a/apps/server/src/dictation/DictationService.test.ts b/apps/server/src/dictation/DictationService.test.ts new file mode 100644 index 000000000..b5101d0e8 --- /dev/null +++ b/apps/server/src/dictation/DictationService.test.ts @@ -0,0 +1,368 @@ +// @effect-diagnostics nodeBuiltinImport:off - path joins for fixture assertions +import * as nodePath from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import type { + DictationDownloadState, + DictationModelId, + DictationModelStatus, +} from "@threadlines/contracts"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Latch from "effect/Latch"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import { ServerConfig } from "../config.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { + DICTATION_ADDON_FILE, + dictationModel, + modelTotalBytes, + resolveRuntimePackage, + runtimeDirectoryName, +} from "./catalog.ts"; +import { + DictationEngine, + type DictationEngineShape, + type DictationEngineSnapshot, +} from "./DictationEngine.ts"; +import { DictationService, DictationServiceLive } from "./DictationService.ts"; +import { DictationDownloader, DictationDownloadError } from "./download.ts"; + +const MODEL: DictationModelId = "moonshine"; + +interface EngineProbe { + readonly loads: Ref.Ref>; + readonly unloads: Ref.Ref; +} + +/** Records what the service asked the worker to do and returns fixed text. */ +const makeEngineLayer = (probe: EngineProbe) => + Layer.effect( + DictationEngine, + Effect.gen(function* () { + const stateRef = yield* SubscriptionRef.make({ + state: "idle", + loadedModel: null, + }); + return { + snapshot: SubscriptionRef.get(stateRef), + changes: SubscriptionRef.changes(stateRef), + load: (input) => + Ref.update(probe.loads, (loads) => [...loads, input.model]).pipe( + Effect.andThen( + SubscriptionRef.set(stateRef, { state: "ready", loadedModel: input.model }), + ), + ), + transcribe: () => Effect.succeed(" hello there "), + unload: Ref.update(probe.unloads, (count) => count + 1).pipe( + Effect.andThen(SubscriptionRef.set(stateRef, { state: "idle", loadedModel: null })), + ), + } satisfies DictationEngineShape; + }), + ); + +/** + * Writes each requested file at exactly its expected size without moving real + * bytes. `gate`, when given, holds the first file open so a test can observe + * the in-progress status. + */ +const makeDownloaderLayer = (options: { readonly gate?: Latch.Latch; readonly fail?: boolean }) => + Layer.effect( + DictationDownloader, + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + return { + downloadFile: (input) => + Effect.gen(function* () { + const partPath = `${input.destPath}.part`; + const expectedBytes = input.expectedBytes ?? 1024; + yield* fs.makeDirectory(nodePath.dirname(input.destPath), { recursive: true }); + yield* fs.writeFile(partPath, new Uint8Array(0)); + yield* input.onProgress?.(Math.floor(expectedBytes / 2)) ?? Effect.void; + if (options.fail) { + return yield* Effect.fail(new DictationDownloadError("network is down")); + } + if (options.gate) { + yield* options.gate.await; + } + yield* fs.truncate(partPath, expectedBytes); + yield* fs.rename(partPath, input.destPath); + }).pipe( + Effect.onExit((exit) => + exit._tag === "Success" + ? Effect.void + : fs.remove(`${input.destPath}.part`, { force: true }).pipe(Effect.ignore), + ), + Effect.mapError((cause) => + cause instanceof DictationDownloadError + ? cause + : new DictationDownloadError(String(cause)), + ), + ), + }; + }), + ); + +const makeLayer = ( + probe: EngineProbe, + options: { readonly gate?: Latch.Latch; readonly fail?: boolean }, +) => + DictationServiceLive.pipe( + Layer.provide(makeEngineLayer(probe)), + Layer.provide(makeDownloaderLayer(options)), + Layer.provideMerge(ServerSettingsService.layerTest({ dictationModel: MODEL })), + Layer.provideMerge( + Layer.fresh(ServerConfig.layerTest(process.cwd(), { prefix: "threadlines-dictation-test-" })), + ), + ); + +const withService = ( + run: ( + probe: EngineProbe, + ) => Effect.Effect, + options: { readonly gate?: Latch.Latch; readonly fail?: boolean } = {}, +) => + Effect.gen(function* () { + const probe: EngineProbe = { + loads: yield* Ref.make>([]), + unloads: yield* Ref.make(0), + }; + return yield* run(probe).pipe( + Effect.provide(makeLayer(probe, options)), + Effect.provide(NodeServices.layer), + ); + }); + +/** + * Waits on the status stream (the same one clients subscribe to) until the + * model matches, with a real-clock timeout so a bug fails fast instead of + * hanging the suite. + */ +const awaitModel = (description: string, matches: (model: DictationModelStatus) => boolean) => + Effect.gen(function* () { + const dictation = yield* DictationService; + const matched = yield* dictation.streamChanges.pipe( + Stream.filter((status) => { + const model = status.models.find((entry) => entry.id === MODEL); + return model !== undefined && matches(model); + }), + Stream.runHead, + Effect.timeoutOrElse({ + duration: Duration.seconds(10), + orElse: () => Effect.die(`dictation model never became ${description}`), + }), + ); + return yield* Option.match(matched, { + onNone: () => Effect.die(`dictation status stream ended before ${description}`), + onSome: Effect.succeed, + }); + }); + +const awaitModelState = (state: DictationDownloadState) => + awaitModel(state, (model) => model.state === state); + +const leftoverPartFiles = (modelDir: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const entries = yield* fs.readDirectory(modelDir).pipe(Effect.orElseSucceed(() => [])); + return entries.filter((entry) => entry.endsWith(".part")); + }); + +/** Marks the native runtime as present so the model paths can be reached. */ +const installRuntime = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const config = yield* ServerConfig; + const runtimePackage = resolveRuntimePackage(process.platform, process.arch); + assert.isNotNull(runtimePackage, "test host has no prebuilt dictation runtime"); + const runtimeDir = nodePath.join( + config.speechModelsDir, + "runtime", + runtimeDirectoryName(runtimePackage!), + ); + yield* fs.makeDirectory(runtimeDir, { recursive: true }); + yield* fs.writeFile(nodePath.join(runtimeDir, DICTATION_ADDON_FILE), new Uint8Array(1)); +}); + +const audioClip = (durationMs: number) => { + const samples = new Int16Array(Math.round((durationMs / 1000) * 16000)); + return { + audioBase64: Buffer.from(samples.buffer).toString("base64"), + sampleRate: 16000 as const, + durationMs, + }; +}; + +// `it.live` rather than `it.effect`: the waits below need the real clock so a +// stalled download fails in ten seconds instead of at the suite timeout. +describe("DictationService", () => { + it.live("starts with the runtime and both models missing", () => + withService(() => + Effect.gen(function* () { + const dictation = yield* DictationService; + const status = yield* dictation.status; + assert.deepEqual( + status.models.map((model) => [model.id, model.state]), + [ + ["parakeet", "missing"], + ["moonshine", "missing"], + ], + ); + assert.equal(status.runtime.state, "missing"); + assert.equal(status.engine, "idle"); + assert.isTrue(status.runtime.supported); + }), + ), + ); + + it.live("moves a model through downloading with progress to ready", () => + Effect.gen(function* () { + const gate = yield* Latch.make(false); + yield* withService( + () => + Effect.gen(function* () { + const dictation = yield* DictationService; + yield* installRuntime; + yield* dictation.downloadModel(MODEL); + + // "downloading" is published the moment the job is queued; the + // progress the UI draws arrives with the first chunk after that. + const downloading = yield* awaitModel( + "downloading with progress", + (model) => model.state === "downloading" && model.bytesDownloaded > 0, + ); + const inProgress = downloading.models.find((model) => model.id === MODEL); + assert.isAbove(inProgress?.bytesDownloaded ?? 0, 0); + assert.isBelow(inProgress?.bytesDownloaded ?? 0, inProgress?.bytesTotal ?? 0); + + yield* gate.open; + const ready = yield* awaitModelState("ready"); + const done = ready.models.find((model) => model.id === MODEL); + assert.equal(done?.bytesDownloaded, modelTotalBytes(dictationModel(MODEL))); + assert.isNull(done?.error ?? null); + }), + { gate }, + ); + }), + ); + + it.live("keeps the failure on the model and removes partial files", () => + withService( + () => + Effect.gen(function* () { + const config = yield* ServerConfig; + const dictation = yield* DictationService; + yield* installRuntime; + yield* dictation.downloadModel(MODEL); + + const failed = yield* awaitModel("failed", (model) => model.error !== null); + const model = failed.models.find((entry) => entry.id === MODEL); + assert.equal(model?.state, "missing"); + assert.include(model?.error ?? "", "network is down"); + assert.deepEqual( + yield* leftoverPartFiles(nodePath.join(config.speechModelsDir, MODEL)), + [], + ); + }), + { fail: true }, + ), + ); + + it.live("cancelling a download clears the state and leaves no partial files", () => + Effect.gen(function* () { + const gate = yield* Latch.make(false); + yield* withService( + () => + Effect.gen(function* () { + const config = yield* ServerConfig; + const dictation = yield* DictationService; + yield* installRuntime; + yield* dictation.downloadModel(MODEL); + yield* awaitModelState("downloading"); + + yield* dictation.cancelDownload(MODEL); + + const cancelled = yield* awaitModelState("missing"); + const model = cancelled.models.find((entry) => entry.id === MODEL); + assert.isNull(model?.error ?? null); + assert.deepEqual( + yield* leftoverPartFiles(nodePath.join(config.speechModelsDir, MODEL)), + [], + ); + }), + { gate }, + ); + }), + ); + + it.live("rejects a clip whose declared length does not match the audio", () => + withService(() => + Effect.gen(function* () { + const dictation = yield* DictationService; + const clip = audioClip(1000); + + const mismatched = yield* dictation + .transcribe({ ...clip, durationMs: 20_000 }) + .pipe(Effect.flip); + assert.equal(mismatched.code, "audio_invalid"); + + const tooLong = yield* dictation + .transcribe({ ...clip, durationMs: 200_000 }) + .pipe(Effect.flip); + assert.equal(tooLong.code, "audio_invalid"); + }), + ), + ); + + it.live("reports the missing piece before download and transcribes after", () => + withService((probe) => + Effect.gen(function* () { + const dictation = yield* DictationService; + const clip = audioClip(1000); + + const noRuntime = yield* dictation.transcribe(clip).pipe(Effect.flip); + assert.equal(noRuntime.code, "runtime_missing"); + + yield* installRuntime; + const noModel = yield* dictation.transcribe(clip).pipe(Effect.flip); + assert.equal(noModel.code, "model_missing"); + + yield* dictation.downloadModel(MODEL); + yield* awaitModelState("ready"); + + const result = yield* dictation.transcribe(clip); + assert.equal(result.text, "hello there"); + assert.deepEqual(yield* Ref.get(probe.loads), [MODEL]); + }), + ), + ); + + it.live("unloads the engine and deletes the files when a model is removed", () => + withService((probe) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const config = yield* ServerConfig; + const dictation = yield* DictationService; + + yield* installRuntime; + yield* dictation.downloadModel(MODEL); + yield* awaitModelState("ready"); + yield* dictation.warmUp; + assert.deepEqual(yield* Ref.get(probe.loads), [MODEL]); + + yield* dictation.removeModel(MODEL); + + assert.equal(yield* Ref.get(probe.unloads), 1); + assert.isFalse(yield* fs.exists(nodePath.join(config.speechModelsDir, MODEL))); + const status = yield* dictation.status; + assert.equal(status.models.find((model) => model.id === MODEL)?.state, "missing"); + }), + ), + ); +}); diff --git a/apps/server/src/dictation/DictationService.ts b/apps/server/src/dictation/DictationService.ts new file mode 100644 index 000000000..8e4cb324a --- /dev/null +++ b/apps/server/src/dictation/DictationService.ts @@ -0,0 +1,460 @@ +// @effect-diagnostics nodeBuiltinImport:off - platform detection and path joins +/** + * DictationService - the server side of composer dictation. + * + * Owns the observable `DictationStatus` (native runtime, both models, engine), + * the on-demand downloads that produce it, and transcription itself. Audio + * arrives as one base64 PCM16 clip per request and is handed to the worker + * process; nothing leaves the machine. + * + * @module dictation/DictationService + */ +import * as nodePath from "node:path"; + +import { + DICTATION_MAX_CLIP_MS, + DictationError, + type DictationModelId, + type DictationModelStatus, + type DictationStatus, + type DictationTranscribeInput, + type DictationTranscribeResult, +} from "@threadlines/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FiberMap from "effect/FiberMap"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import { ServerConfig } from "../config.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { + DICTATION_ADDON_FILE, + DICTATION_MODELS, + dictationModel, + dictationNumThreads, + modelBytesOnDisk, + modelFileUrl, + modelIsReady, + modelTotalBytes, + resolveRuntimePackage, + runtimeDirectoryName, + type DictationRuntimePackage, +} from "./catalog.ts"; +import { DictationEngine, DictationEngineLive } from "./DictationEngine.ts"; +import { DictationDownloader, DictationDownloaderLive } from "./download.ts"; +import { extractNpmTarball } from "./npmTarball.ts"; + +/** + * The registry tarballs are all around 22 MB. The exact size is not known + * before the request, so this is only what progress is shown against. + */ +export const DICTATION_RUNTIME_APPROX_BYTES = 22 * 1024 * 1024; + +/** Progress is noisy; publishing more often than this just burns frames. */ +const PROGRESS_PUBLISH_INTERVAL_MS = 100; + +/** Sample count may drift from the declared duration by at most this share. */ +const DURATION_TOLERANCE = 0.1; + +export interface DictationServiceShape { + readonly status: Effect.Effect; + /** Current status first, then every change. */ + readonly streamChanges: Stream.Stream; + readonly downloadModel: (model: DictationModelId) => Effect.Effect; + readonly cancelDownload: (model: DictationModelId) => Effect.Effect; + readonly removeModel: (model: DictationModelId) => Effect.Effect; + /** Loads the selected model ahead of the first clip. Never fails. */ + readonly warmUp: Effect.Effect; + readonly transcribe: ( + input: DictationTranscribeInput, + ) => Effect.Effect; +} + +export class DictationService extends Context.Service()( + "threadlines/dictation/DictationService", +) {} + +const dictationError = (code: DictationError["code"], message: string) => + new DictationError({ code, message }); + +/** Decodes base64 PCM16 into the sample array the worker expects. */ +function decodePcm16(audioBase64: string): Int16Array { + const bytes = Buffer.from(audioBase64, "base64"); + const samples = new Int16Array(Math.floor(bytes.length / 2)); + for (let index = 0; index < samples.length; index += 1) { + samples[index] = bytes.readInt16LE(index * 2); + } + return samples; +} + +function validateAudio(input: DictationTranscribeInput): DictationError | undefined { + if (input.durationMs <= 0 || input.durationMs > DICTATION_MAX_CLIP_MS) { + return dictationError( + "audio_invalid", + `Recording must be between 0 and ${DICTATION_MAX_CLIP_MS / 1000} seconds.`, + ); + } + const byteLength = Buffer.from(input.audioBase64, "base64").length; + if (byteLength === 0 || byteLength % 2 !== 0) { + return dictationError("audio_invalid", "Recording is not 16-bit PCM audio."); + } + const expectedSamples = (input.durationMs / 1000) * input.sampleRate; + const actualSamples = byteLength / 2; + if (Math.abs(actualSamples - expectedSamples) > expectedSamples * DURATION_TOLERANCE) { + return dictationError("audio_invalid", "Recording length does not match the audio sent."); + } + return undefined; +} + +const initialModelStatus = (id: DictationModelId): DictationModelStatus => ({ + id, + state: "missing", + bytesDownloaded: 0, + bytesTotal: modelTotalBytes(dictationModel(id)), + error: null, +}); + +const makeDictationService = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const config = yield* ServerConfig; + const settings = yield* ServerSettingsService; + const engine = yield* DictationEngine; + const downloader = yield* DictationDownloader; + + const runtimePackage = resolveRuntimePackage(process.platform, process.arch); + const speechDir = config.speechModelsDir; + const runtimeRoot = nodePath.join(speechDir, "runtime"); + const runtimeDir = + runtimePackage === null + ? runtimeRoot + : nodePath.join(runtimeRoot, runtimeDirectoryName(runtimePackage)); + const addonPath = nodePath.join(runtimeDir, DICTATION_ADDON_FILE); + const modelDirFor = (model: DictationModelId) => nodePath.join(speechDir, model); + + const runtimeIsReady = fs.exists(addonPath).pipe(Effect.orElseSucceed(() => false)); + + const withFileSystem = (effect: Effect.Effect) => + effect.pipe(Effect.provideService(FileSystem.FileSystem, fs)); + + const statusRef = yield* SubscriptionRef.make({ + runtime: { + supported: runtimePackage !== null, + state: "missing", + bytesDownloaded: 0, + bytesTotal: DICTATION_RUNTIME_APPROX_BYTES, + error: null, + }, + models: DICTATION_MODELS.map((entry) => initialModelStatus(entry.id)), + engine: "idle", + loadedModel: null, + modelsDir: speechDir, + }); + + const updateRuntime = ( + update: (current: DictationStatus["runtime"]) => DictationStatus["runtime"], + ) => + SubscriptionRef.update(statusRef, (status) => ({ ...status, runtime: update(status.runtime) })); + + const updateModel = ( + model: DictationModelId, + update: (current: DictationModelStatus) => DictationModelStatus, + ) => + SubscriptionRef.update(statusRef, (status) => ({ + ...status, + models: status.models.map((entry) => (entry.id === model ? update(entry) : entry)), + })); + + // Initial scan: whatever is already on disk decides the starting status. + yield* Effect.gen(function* () { + const readyRuntime = yield* runtimeIsReady; + if (readyRuntime) { + yield* updateRuntime((runtime) => ({ + ...runtime, + state: "ready", + bytesDownloaded: runtime.bytesTotal, + })); + } + for (const entry of DICTATION_MODELS) { + const modelDir = modelDirFor(entry.id); + const ready = yield* modelIsReady(entry, modelDir); + const bytes = ready ? modelTotalBytes(entry) : yield* modelBytesOnDisk(entry, modelDir); + yield* updateModel(entry.id, (current) => ({ + ...current, + state: ready ? "ready" : "missing", + bytesDownloaded: bytes, + })); + } + }); + + // The engine transitions on its own (idle unload, worker crash), so status + // mirrors it instead of being written at each call site. + yield* engine.changes.pipe( + Stream.runForEach((snapshot) => + SubscriptionRef.update(statusRef, (status) => ({ + ...status, + engine: snapshot.state, + loadedModel: snapshot.loadedModel, + })), + ), + Effect.forkScoped, + ); + + const downloads = yield* FiberMap.make(); + const runtimeLock = yield* Semaphore.make(1); + + /** Publishes at most every `PROGRESS_PUBLISH_INTERVAL_MS`. */ + const makeProgressPublisher = (publish: (bytes: number) => Effect.Effect) => { + let lastPublishedAt = 0; + return (bytes: number) => + Effect.suspend(() => { + const now = Date.now(); + if (now - lastPublishedAt < PROGRESS_PUBLISH_INTERVAL_MS) { + return Effect.void; + } + lastPublishedAt = now; + return publish(bytes); + }); + }; + + const ensureRuntime = (packageInfo: DictationRuntimePackage) => + runtimeLock.withPermits(1)( + Effect.gen(function* () { + if (yield* runtimeIsReady) { + return; + } + yield* updateRuntime((runtime) => ({ + ...runtime, + state: "downloading", + bytesDownloaded: 0, + error: null, + })); + + const tarballPath = nodePath.join(runtimeRoot, ".download.tgz"); + const onProgress = makeProgressPublisher((bytes) => + updateRuntime((runtime) => ({ ...runtime, bytesDownloaded: bytes })), + ); + yield* downloader + .downloadFile({ url: packageInfo.tarballUrl, destPath: tarballPath, onProgress }) + .pipe(Effect.mapError((cause) => dictationError("download_failed", cause.message))); + yield* extractNpmTarball(tarballPath, runtimeDir).pipe( + Effect.mapError((cause) => + dictationError("download_failed", `Failed to unpack the speech runtime: ${cause}`), + ), + ); + yield* fs.remove(tarballPath, { force: true }).pipe(Effect.ignore); + + if (!(yield* runtimeIsReady)) { + return yield* Effect.fail( + dictationError( + "download_failed", + `The speech runtime package did not contain ${DICTATION_ADDON_FILE}.`, + ), + ); + } + yield* updateRuntime((runtime) => ({ + ...runtime, + state: "ready", + bytesDownloaded: runtime.bytesTotal, + error: null, + })); + }).pipe( + Effect.tapError((error) => + updateRuntime((runtime) => ({ + ...runtime, + state: "missing", + bytesDownloaded: 0, + error: error.message, + })), + ), + Effect.onInterrupt(() => + updateRuntime((runtime) => ({ ...runtime, state: "missing", bytesDownloaded: 0 })), + ), + ), + ); + + const downloadJob = (model: DictationModelId, packageInfo: DictationRuntimePackage) => + Effect.gen(function* () { + yield* ensureRuntime(packageInfo); + + const entry = dictationModel(model); + const modelDir = modelDirFor(model); + yield* fs + .makeDirectory(modelDir, { recursive: true }) + .pipe( + Effect.mapError((cause) => + dictationError("download_failed", `Failed to create ${modelDir}: ${cause}`), + ), + ); + + let completedBytes = yield* modelBytesOnDisk(entry, modelDir); + const onProgress = makeProgressPublisher((bytes) => + updateModel(model, (current) => ({ ...current, bytesDownloaded: bytes })), + ); + yield* updateModel(model, (current) => ({ ...current, bytesDownloaded: completedBytes })); + + for (const file of entry.files) { + const destPath = nodePath.join(modelDir, file.name); + const alreadyThere = yield* fs.stat(destPath).pipe( + Effect.map((info) => info.type === "File" && Number(info.size) === file.bytes), + Effect.orElseSucceed(() => false), + ); + if (alreadyThere) { + continue; + } + const bytesBefore = completedBytes; + yield* downloader + .downloadFile({ + url: modelFileUrl(entry, file), + destPath, + expectedBytes: file.bytes, + onProgress: (bytes) => onProgress(bytesBefore + bytes), + }) + .pipe(Effect.mapError((cause) => dictationError("download_failed", cause.message))); + completedBytes += file.bytes; + } + + yield* updateModel(model, (current) => ({ + ...current, + state: "ready", + bytesDownloaded: current.bytesTotal, + error: null, + })); + }).pipe( + Effect.tapError((error) => + updateModel(model, (current) => ({ + ...current, + state: "missing", + error: error.message, + })), + ), + // A cancelled download keeps whatever whole files landed and clears the + // spinner; `downloadFile` already removed the in-flight `.part`. + Effect.onInterrupt(() => + updateModel(model, (current) => ({ ...current, state: "missing", error: null })), + ), + Effect.ignoreCause({ log: true }), + ); + + const downloadModel = (model: DictationModelId) => + Effect.gen(function* () { + if (runtimePackage === null) { + return yield* Effect.fail( + dictationError( + "unsupported_platform", + "Dictation is not available for this platform yet.", + ), + ); + } + const status = yield* SubscriptionRef.get(statusRef); + const current = status.models.find((entry) => entry.id === model); + if (current?.state === "ready" || current?.state === "downloading") { + return; + } + yield* updateModel(model, (entry) => ({ ...entry, state: "downloading", error: null })); + yield* FiberMap.run(downloads, model, downloadJob(model, runtimePackage)); + }); + + const cancelDownload = (model: DictationModelId) => FiberMap.remove(downloads, model); + + const removeModel = (model: DictationModelId) => + Effect.gen(function* () { + yield* cancelDownload(model); + const snapshot = yield* engine.snapshot; + if (snapshot.loadedModel === model) { + yield* engine.unload; + } + yield* fs + .remove(modelDirFor(model), { recursive: true, force: true }) + .pipe( + Effect.mapError((cause) => + dictationError("download_failed", `Failed to remove the model files: ${cause}`), + ), + ); + yield* updateModel(model, (current) => ({ + ...current, + state: "missing", + bytesDownloaded: 0, + error: null, + })); + }); + + const selectedModel = settings.getSettings.pipe( + Effect.map((current) => current.dictationModel), + Effect.orElseSucceed(() => "parakeet" as const satisfies DictationModelId), + ); + + const loadSelectedModel = (model: DictationModelId, packageInfo: DictationRuntimePackage) => + engine.load({ + model, + config: dictationModel(model).recognizerConfig(modelDirFor(model), dictationNumThreads()), + runtimeDir, + libraryPathEnv: packageInfo.libraryPathEnv, + }); + + const warmUp = Effect.gen(function* () { + if (runtimePackage === null || !(yield* runtimeIsReady)) { + return; + } + const model = yield* selectedModel; + if (!(yield* modelIsReady(dictationModel(model), modelDirFor(model)))) { + return; + } + yield* loadSelectedModel(model, runtimePackage); + }).pipe(Effect.ignoreCause({ log: true })); + + const transcribe = (input: DictationTranscribeInput) => + Effect.gen(function* () { + const invalid = validateAudio(input); + if (invalid) { + return yield* Effect.fail(invalid); + } + if (runtimePackage === null) { + return yield* Effect.fail( + dictationError( + "unsupported_platform", + "Dictation is not available for this platform yet.", + ), + ); + } + if (!(yield* runtimeIsReady)) { + return yield* Effect.fail( + dictationError("runtime_missing", "The speech runtime has not been downloaded yet."), + ); + } + const model = yield* selectedModel; + if (!(yield* modelIsReady(dictationModel(model), modelDirFor(model)))) { + return yield* Effect.fail( + dictationError("model_missing", "The selected dictation model is not downloaded yet."), + ); + } + + yield* loadSelectedModel(model, runtimePackage); + const text = yield* engine.transcribe(decodePcm16(input.audioBase64), input.sampleRate); + return { text: text.trim() } satisfies DictationTranscribeResult; + }); + + return { + status: SubscriptionRef.get(statusRef), + // Current value first, then every change; nothing is lost in between. + streamChanges: SubscriptionRef.changes(statusRef), + // The service owns a `FileSystem` for its whole lifetime, so callers get + // plain effects with nothing left to provide. + downloadModel: (model) => withFileSystem(downloadModel(model)), + cancelDownload, + removeModel: (model) => withFileSystem(removeModel(model)), + warmUp: withFileSystem(warmUp), + transcribe: (input) => withFileSystem(transcribe(input)), + } satisfies DictationServiceShape; +}); + +export const DictationServiceLive = Layer.effect(DictationService, makeDictationService); + +/** Everything dictation needs: the worker engine, the downloader, the service. */ +export const DictationLive = DictationServiceLive.pipe( + Layer.provide(DictationEngineLive), + Layer.provide(DictationDownloaderLive), +); diff --git a/apps/server/src/dictation/catalog.test.ts b/apps/server/src/dictation/catalog.test.ts new file mode 100644 index 000000000..9c1ad4dd3 --- /dev/null +++ b/apps/server/src/dictation/catalog.test.ts @@ -0,0 +1,73 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; + +import { + DICTATION_MODELS, + DICTATION_RUNTIME_VERSION, + dictationModel, + modelIsReady, + resolveRuntimePackage, +} from "./catalog.ts"; + +const SUPPORTED_PLATFORMS: ReadonlyArray = [ + ["win32", "x64"], + ["darwin", "arm64"], + ["darwin", "x64"], + ["linux", "x64"], + ["linux", "arm64"], +]; + +it.layer(NodeServices.layer)("dictation catalog", (it) => { + it.effect("resolves a tarball url for every supported platform", () => + Effect.sync(() => { + for (const [platform, arch] of SUPPORTED_PLATFORMS) { + const runtimePackage = resolveRuntimePackage(platform, arch); + assert.isNotNull(runtimePackage, `${platform}-${arch}`); + assert.equal( + runtimePackage?.tarballUrl, + `https://registry.npmjs.org/${runtimePackage?.packageName}/-/${runtimePackage?.packageName}-${DICTATION_RUNTIME_VERSION}.tgz`, + ); + } + assert.isNull(resolveRuntimePackage("linux", "ia32")); + assert.isNull(resolveRuntimePackage("freebsd", "x64")); + }), + ); + + it.effect("names both models in catalog order", () => + Effect.sync(() => { + assert.deepEqual( + DICTATION_MODELS.map((entry) => entry.id), + ["parakeet", "moonshine"], + ); + }), + ); + + it.effect("treats a model as ready only when every file has its exact size", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fs.makeTempDirectoryScoped({ prefix: "threadlines-dictation-catalog-" }); + const entry = dictationModel("moonshine"); + const modelDir = path.join(dir, "moonshine"); + yield* fs.makeDirectory(modelDir, { recursive: true }); + + assert.isFalse(yield* modelIsReady(entry, modelDir)); + + // Sized with `truncate` so the fixture costs no real bytes. + for (const file of entry.files) { + const filePath = path.join(modelDir, file.name); + yield* fs.writeFile(filePath, new Uint8Array(0)); + yield* fs.truncate(filePath, file.bytes); + } + assert.isTrue(yield* modelIsReady(entry, modelDir)); + + // A short file is what an interrupted download leaves behind. + const short = entry.files[0]!; + yield* fs.truncate(path.join(modelDir, short.name), short.bytes - 1); + assert.isFalse(yield* modelIsReady(entry, modelDir)); + }), + ); +}); diff --git a/apps/server/src/dictation/catalog.ts b/apps/server/src/dictation/catalog.ts new file mode 100644 index 000000000..02ebeb997 --- /dev/null +++ b/apps/server/src/dictation/catalog.ts @@ -0,0 +1,230 @@ +// @effect-diagnostics nodeBuiltinImport:off - platform detection and path joins +/** + * Catalog of everything dictation downloads: the prebuilt sherpa-onnx native + * runtime for the host platform, and the speech models. Sizes are exact so a + * model directory can be declared ready without hashing, and so download + * progress is byte-accurate before the first byte arrives. + * + * @module dictation/catalog + */ +import * as os from "node:os"; +import * as nodePath from "node:path"; + +import type { DictationModelId } from "@threadlines/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; + +/** npm version of the `sherpa-onnx--` runtime packages. */ +export const DICTATION_RUNTIME_VERSION = "1.13.7"; + +/** File the native addon is loaded from inside an extracted runtime package. */ +export const DICTATION_ADDON_FILE = "sherpa-onnx.node"; + +/** + * Environment variable the child process needs pointed at the runtime + * directory so the addon's sibling shared libraries resolve. + */ +export type DictationLibraryPathEnv = "PATH" | "LD_LIBRARY_PATH" | "DYLD_LIBRARY_PATH"; + +export interface DictationRuntimePackage { + readonly packageName: string; + readonly tarballUrl: string; + readonly addonFile: string; + readonly libraryPathEnv: DictationLibraryPathEnv; +} + +const RUNTIME_PACKAGES: ReadonlyArray<{ + readonly platform: NodeJS.Platform; + readonly arch: string; + readonly packageName: string; + readonly libraryPathEnv: DictationLibraryPathEnv; +}> = [ + // The publisher renamed win32-x64 to win-x64 to dodge registry spam filters. + { platform: "win32", arch: "x64", packageName: "sherpa-onnx-win-x64", libraryPathEnv: "PATH" }, + { + platform: "darwin", + arch: "arm64", + packageName: "sherpa-onnx-darwin-arm64", + libraryPathEnv: "DYLD_LIBRARY_PATH", + }, + { + platform: "darwin", + arch: "x64", + packageName: "sherpa-onnx-darwin-x64", + libraryPathEnv: "DYLD_LIBRARY_PATH", + }, + { + platform: "linux", + arch: "x64", + packageName: "sherpa-onnx-linux-x64", + libraryPathEnv: "LD_LIBRARY_PATH", + }, + { + platform: "linux", + arch: "arm64", + packageName: "sherpa-onnx-linux-arm64", + libraryPathEnv: "LD_LIBRARY_PATH", + }, +]; + +/** `null` when no prebuilt addon is published for this platform/arch. */ +export function resolveRuntimePackage( + platform: NodeJS.Platform, + arch: string, +): DictationRuntimePackage | null { + const entry = RUNTIME_PACKAGES.find( + (candidate) => candidate.platform === platform && candidate.arch === arch, + ); + if (!entry) { + return null; + } + return { + packageName: entry.packageName, + tarballUrl: `https://registry.npmjs.org/${entry.packageName}/-/${entry.packageName}-${DICTATION_RUNTIME_VERSION}.tgz`, + addonFile: DICTATION_ADDON_FILE, + libraryPathEnv: entry.libraryPathEnv, + }; +} + +/** Directory name the extracted runtime lives under, inside `runtime/`. */ +export function runtimeDirectoryName(runtimePackage: DictationRuntimePackage): string { + return `${runtimePackage.packageName}-${DICTATION_RUNTIME_VERSION}`; +} + +export interface DictationModelFile { + readonly name: string; + readonly bytes: number; +} + +/** + * Recognizer config accepted by `createOfflineRecognizer`. Model paths are + * absolute; the shape mirrors what `sherpa-onnx-node` builds. + */ +export interface DictationRecognizerConfig { + readonly featConfig: { readonly sampleRate: number; readonly featureDim: number }; + readonly modelConfig: Record; + readonly decodingMethod: string; +} + +export interface DictationModelEntry { + readonly id: DictationModelId; + readonly label: string; + readonly hfRepo: string; + readonly files: ReadonlyArray; + readonly recognizerConfig: (modelDir: string, numThreads: number) => DictationRecognizerConfig; +} + +const FEAT_CONFIG = { sampleRate: 16000, featureDim: 80 } as const; + +const joinModelPath = (modelDir: string, file: string) => nodePath.join(modelDir, file); + +export const DICTATION_MODELS: ReadonlyArray = [ + { + id: "parakeet", + label: "Parakeet TDT 0.6B v2", + hfRepo: "csukuangfj/sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8", + files: [ + { name: "encoder.int8.onnx", bytes: 652184296 }, + { name: "decoder.int8.onnx", bytes: 7257753 }, + { name: "joiner.int8.onnx", bytes: 1739080 }, + { name: "tokens.txt", bytes: 9384 }, + ], + recognizerConfig: (modelDir, numThreads) => ({ + featConfig: FEAT_CONFIG, + modelConfig: { + transducer: { + encoder: joinModelPath(modelDir, "encoder.int8.onnx"), + decoder: joinModelPath(modelDir, "decoder.int8.onnx"), + joiner: joinModelPath(modelDir, "joiner.int8.onnx"), + }, + tokens: joinModelPath(modelDir, "tokens.txt"), + modelType: "nemo_transducer", + numThreads, + debug: 0, + }, + decodingMethod: "greedy_search", + }), + }, + { + id: "moonshine", + label: "Moonshine tiny en", + hfRepo: "csukuangfj/sherpa-onnx-moonshine-tiny-en-int8", + files: [ + { name: "preprocess.onnx", bytes: 6800738 }, + { name: "encode.int8.onnx", bytes: 18249187 }, + { name: "uncached_decode.int8.onnx", bytes: 53216096 }, + { name: "cached_decode.int8.onnx", bytes: 45264830 }, + { name: "tokens.txt", bytes: 436688 }, + ], + recognizerConfig: (modelDir, numThreads) => ({ + featConfig: FEAT_CONFIG, + modelConfig: { + moonshine: { + preprocessor: joinModelPath(modelDir, "preprocess.onnx"), + encoder: joinModelPath(modelDir, "encode.int8.onnx"), + uncachedDecoder: joinModelPath(modelDir, "uncached_decode.int8.onnx"), + cachedDecoder: joinModelPath(modelDir, "cached_decode.int8.onnx"), + }, + tokens: joinModelPath(modelDir, "tokens.txt"), + numThreads, + debug: 0, + }, + decodingMethod: "greedy_search", + }), + }, +]; + +export function dictationModel(id: DictationModelId): DictationModelEntry { + const entry = DICTATION_MODELS.find((model) => model.id === id); + if (!entry) { + throw new Error(`Unknown dictation model: ${id}`); + } + return entry; +} + +export function modelTotalBytes(entry: DictationModelEntry): number { + return entry.files.reduce((total, file) => total + file.bytes, 0); +} + +/** Hugging Face serves individual files, so no archive extraction is needed. */ +export function modelFileUrl(entry: DictationModelEntry, file: DictationModelFile): string { + return `https://huggingface.co/${entry.hfRepo}/resolve/main/${file.name}`; +} + +/** Decoding threads: enough to matter, few enough to leave the UI responsive. */ +export function dictationNumThreads(): number { + return Math.min(4, os.availableParallelism()); +} + +/** + * A model directory counts as downloaded only when every catalog file is + * present at exactly its catalog size, so an interrupted download can never + * be mistaken for a usable model. + */ +export const modelIsReady = Effect.fn(function* (entry: DictationModelEntry, modelDir: string) { + const fs = yield* FileSystem.FileSystem; + for (const file of entry.files) { + const info = yield* fs + .stat(joinModelPath(modelDir, file.name)) + .pipe(Effect.orElseSucceed(() => undefined)); + if (info === undefined || info.type !== "File" || Number(info.size) !== file.bytes) { + return false; + } + } + return true; +}); + +/** Bytes already on disk for a model, counting only files at catalog size. */ +export const modelBytesOnDisk = Effect.fn(function* (entry: DictationModelEntry, modelDir: string) { + const fs = yield* FileSystem.FileSystem; + let bytes = 0; + for (const file of entry.files) { + const info = yield* fs + .stat(joinModelPath(modelDir, file.name)) + .pipe(Effect.orElseSucceed(() => undefined)); + if (info !== undefined && info.type === "File" && Number(info.size) === file.bytes) { + bytes += file.bytes; + } + } + return bytes; +}); diff --git a/apps/server/src/dictation/download.ts b/apps/server/src/dictation/download.ts new file mode 100644 index 000000000..255263ef8 --- /dev/null +++ b/apps/server/src/dictation/download.ts @@ -0,0 +1,132 @@ +// @effect-diagnostics nodeBuiltinImport:off - path resolution for the .part file +/** + * Streaming file download used for the dictation runtime tarball and model + * files. Bytes land in a sibling `.part` file and are renamed into place only + * once the expected byte count arrives, so a cancelled or failed download can + * never leave something that looks complete. + * + * @module dictation/download + */ +import * as nodePath from "node:path"; + +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; +import { HttpClient, HttpClientRequest } from "effect/unstable/http"; + +export class DictationDownloadError extends Error { + readonly _tag = "DictationDownloadError"; +} + +export interface DownloadFileOptions { + readonly url: string; + readonly destPath: string; + /** Catalog size; a mismatch fails the download. `undefined` skips the check. */ + readonly expectedBytes?: number; + /** Called with the running byte count for this file. */ + readonly onProgress?: (bytesDownloaded: number) => Effect.Effect; +} + +export interface DictationDownloaderShape { + /** + * Downloads one file. Interruptible: on interrupt or failure the partial + * file is removed and nothing appears at `destPath`. + */ + readonly downloadFile: ( + options: DownloadFileOptions, + ) => Effect.Effect; +} + +/** + * Isolated so `DictationService` can be tested without a network: tests + * provide a layer that writes bytes straight to disk. + */ +export class DictationDownloader extends Context.Service< + DictationDownloader, + DictationDownloaderShape +>()("threadlines/dictation/DictationDownloader") {} + +const removeQuietly = (fs: FileSystem.FileSystem, path: string) => + fs.remove(path, { force: true }).pipe(Effect.ignore); + +export const DictationDownloaderLive = Layer.effect( + DictationDownloader, + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + // Redirects are followed by the underlying fetch, which Hugging Face + // relies on to hand off to its CDN. + const client = yield* HttpClient.HttpClient; + + const downloadFile: DictationDownloaderShape["downloadFile"] = Effect.fn( + "dictation.downloadFile", + )(function* (options: DownloadFileOptions) { + const partPath = `${options.destPath}.part`; + + yield* fs + .makeDirectory(nodePath.dirname(options.destPath), { recursive: true }) + .pipe( + Effect.mapError( + (cause) => new DictationDownloadError(`Failed to prepare download directory: ${cause}`), + ), + ); + yield* removeQuietly(fs, partPath); + + const download = Effect.gen(function* () { + const response = yield* client + .execute(HttpClientRequest.get(options.url)) + .pipe( + Effect.mapError( + (cause) => new DictationDownloadError(`Failed to request ${options.url}: ${cause}`), + ), + ); + + if (response.status < 200 || response.status >= 300) { + return yield* Effect.fail( + new DictationDownloadError( + `Download of ${options.url} failed with HTTP ${response.status}`, + ), + ); + } + + let bytesDownloaded = 0; + yield* response.stream.pipe( + Stream.tap((chunk) => { + bytesDownloaded += chunk.length; + return options.onProgress?.(bytesDownloaded) ?? Effect.void; + }), + Stream.run(fs.sink(partPath)), + Effect.mapError( + (cause) => new DictationDownloadError(`Download of ${options.url} failed: ${cause}`), + ), + ); + + if (options.expectedBytes !== undefined && bytesDownloaded !== options.expectedBytes) { + return yield* Effect.fail( + new DictationDownloadError( + `Download of ${options.url} returned ${bytesDownloaded} bytes, expected ${options.expectedBytes}`, + ), + ); + } + + yield* fs + .rename(partPath, options.destPath) + .pipe( + Effect.mapError( + (cause) => + new DictationDownloadError(`Failed to finish ${options.destPath}: ${cause}`), + ), + ); + }); + + yield* download.pipe( + Effect.onExit((exit) => + exit._tag === "Success" ? Effect.void : removeQuietly(fs, partPath), + ), + ); + }); + + return { downloadFile } satisfies DictationDownloaderShape; + }), +); diff --git a/apps/server/src/dictation/npmTarball.test.ts b/apps/server/src/dictation/npmTarball.test.ts new file mode 100644 index 000000000..49c04a410 --- /dev/null +++ b/apps/server/src/dictation/npmTarball.test.ts @@ -0,0 +1,99 @@ +// @effect-diagnostics nodeBuiltinImport:off - builds a tarball fixture +import * as zlib from "node:zlib"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; + +import { extractNpmTarball } from "./npmTarball.ts"; + +const BLOCK_SIZE = 512; + +/** Builds a gzipped ustar archive with one 512-byte header per file. */ +function makeTarball(entries: ReadonlyArray<{ name: string; contents: string }>): Uint8Array { + const blocks: Array = []; + for (const entry of entries) { + const data = Buffer.from(entry.contents, "utf8"); + const header = Buffer.alloc(BLOCK_SIZE); + header.write(entry.name, 0, 100, "utf8"); + header.write("000644 \0", 100, 8, "utf8"); + header.write("000000 \0", 108, 8, "utf8"); + header.write("000000 \0", 116, 8, "utf8"); + header.write(`${data.length.toString(8).padStart(11, "0")} `, 124, 12, "utf8"); + header.write("00000000000 ", 136, 12, "utf8"); + header.write("0", 156, 1, "utf8"); + header.write("ustar\0", 257, 6, "utf8"); + header.write("00", 263, 2, "utf8"); + // Checksum is computed over the header with the checksum field blanked. + header.write(" ".repeat(8), 148, 8, "utf8"); + let checksum = 0; + for (const byte of header) { + checksum += byte; + } + header.write(`${checksum.toString(8).padStart(6, "0")}\0 `, 148, 8, "utf8"); + + blocks.push(header); + const padded = Buffer.alloc(Math.ceil(data.length / BLOCK_SIZE) * BLOCK_SIZE); + data.copy(padded); + blocks.push(padded); + } + blocks.push(Buffer.alloc(BLOCK_SIZE * 2)); + return zlib.gzipSync(Buffer.concat(blocks)); +} + +const withTempDir = (run: (dir: string) => Effect.Effect) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const dir = yield* fs.makeTempDirectoryScoped({ prefix: "threadlines-npm-tarball-test-" }); + return yield* run(dir); + }); + +it.layer(NodeServices.layer)("npmTarball", (it) => { + it.effect("extracts files and strips the package prefix", () => + withTempDir((dir) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tgzPath = path.join(dir, "runtime.tgz"); + const destDir = path.join(dir, "out"); + yield* fs.writeFile( + tgzPath, + makeTarball([ + { name: "package/sherpa-onnx.node", contents: "addon" }, + { name: "package/nested/onnxruntime.dll", contents: "library" }, + ]), + ); + + yield* extractNpmTarball(tgzPath, destDir); + + assert.equal(yield* fs.readFileString(path.join(destDir, "sherpa-onnx.node")), "addon"); + assert.equal( + yield* fs.readFileString(path.join(destDir, "nested", "onnxruntime.dll")), + "library", + ); + }), + ), + ); + + it.effect("rejects an entry that escapes the destination directory", () => + withTempDir((dir) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tgzPath = path.join(dir, "evil.tgz"); + const destDir = path.join(dir, "out"); + yield* fs.writeFile( + tgzPath, + makeTarball([{ name: "package/../../escaped.txt", contents: "nope" }]), + ); + + const error = yield* extractNpmTarball(tgzPath, destDir).pipe(Effect.flip); + + assert.include(error.message, "escapes the destination directory"); + assert.isFalse(yield* fs.exists(path.join(dir, "..", "escaped.txt"))); + }), + ), + ); +}); diff --git a/apps/server/src/dictation/npmTarball.ts b/apps/server/src/dictation/npmTarball.ts new file mode 100644 index 000000000..634a64e3f --- /dev/null +++ b/apps/server/src/dictation/npmTarball.ts @@ -0,0 +1,139 @@ +// @effect-diagnostics nodeBuiltinImport:off - gunzip and path resolution +/** + * Minimal reader for npm package tarballs, used to unpack the prebuilt + * sherpa-onnx runtime. npm tarballs are gzipped ustar archives with every + * entry under a single `package/` directory; only regular files are kept and + * that prefix is stripped. Anything that would land outside the destination + * is rejected rather than skipped, because a tarball that tries is not one we + * want to half-extract. + * + * @module dictation/npmTarball + */ +import * as nodePath from "node:path"; +import * as zlib from "node:zlib"; + +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; + +const BLOCK_SIZE = 512; +const NAME_OFFSET = 0; +const NAME_SIZE = 100; +const SIZE_OFFSET = 124; +const SIZE_FIELD_SIZE = 12; +const TYPE_FLAG_OFFSET = 156; +const PREFIX_OFFSET = 345; +const PREFIX_SIZE = 155; + +export class NpmTarballError extends Error { + readonly _tag = "NpmTarballError"; +} + +const readString = (block: Uint8Array, offset: number, size: number): string => { + const slice = block.subarray(offset, offset + size); + const end = slice.indexOf(0); + return Buffer.from(end === -1 ? slice : slice.subarray(0, end)).toString("utf8"); +}; + +const readOctal = (block: Uint8Array, offset: number, size: number): number => { + const raw = readString(block, offset, size).trim(); + if (raw.length === 0) { + return 0; + } + const parsed = Number.parseInt(raw, 8); + return Number.isFinite(parsed) ? parsed : 0; +}; + +const isZeroBlock = (block: Uint8Array): boolean => block.every((byte) => byte === 0); + +interface TarEntry { + readonly name: string; + readonly typeFlag: string; + readonly data: Uint8Array; +} + +/** Walks the ustar stream, resolving `L`/`x` long-name entries as it goes. */ +function* readTarEntries(archive: Uint8Array): Generator { + let offset = 0; + let pendingLongName: string | undefined; + + while (offset + BLOCK_SIZE <= archive.length) { + const header = archive.subarray(offset, offset + BLOCK_SIZE); + offset += BLOCK_SIZE; + if (isZeroBlock(header)) { + continue; + } + + const size = readOctal(header, SIZE_OFFSET, SIZE_FIELD_SIZE); + const typeFlag = readString(header, TYPE_FLAG_OFFSET, 1) || "0"; + const data = archive.subarray(offset, offset + size); + offset += Math.ceil(size / BLOCK_SIZE) * BLOCK_SIZE; + + if (typeFlag === "L") { + pendingLongName = Buffer.from(data).toString("utf8").replace(/\0+$/, ""); + continue; + } + + const prefix = readString(header, PREFIX_OFFSET, PREFIX_SIZE); + const shortName = readString(header, NAME_OFFSET, NAME_SIZE); + const name = pendingLongName ?? (prefix.length > 0 ? `${prefix}/${shortName}` : shortName); + pendingLongName = undefined; + + yield { name, typeFlag, data }; + } +} + +/** Strips the single leading directory segment npm wraps every entry in. */ +const stripPackagePrefix = (name: string): string | undefined => { + const normalized = name.replace(/\\/g, "/").replace(/^\.\//, ""); + const separator = normalized.indexOf("/"); + if (separator === -1) { + return undefined; + } + const rest = normalized.slice(separator + 1); + return rest.length > 0 ? rest : undefined; +}; + +/** + * Extracts the regular files of an npm tarball into `destDir`, dropping the + * `package/` prefix. Fails if any entry resolves outside `destDir`. + */ +export const extractNpmTarball = Effect.fn("dictation.extractNpmTarball")(function* ( + tgzPath: string, + destDir: string, +) { + const fs = yield* FileSystem.FileSystem; + const compressed = yield* fs.readFile(tgzPath); + const archive = yield* Effect.try({ + try: () => zlib.gunzipSync(compressed), + catch: (cause) => new NpmTarballError(`Failed to gunzip ${tgzPath}: ${String(cause)}`), + }); + + const resolvedDest = nodePath.resolve(destDir); + yield* fs.makeDirectory(resolvedDest, { recursive: true }); + + const written: Array = []; + for (const entry of readTarEntries(archive)) { + // Regular files only: "0"/"\0" are files, everything else is a + // directory, link, or pax header we have no use for. + if (entry.typeFlag !== "0" && entry.typeFlag !== "\0") { + continue; + } + const relative = stripPackagePrefix(entry.name); + if (relative === undefined) { + continue; + } + + const target = nodePath.resolve(resolvedDest, relative); + if (target !== resolvedDest && !target.startsWith(resolvedDest + nodePath.sep)) { + return yield* Effect.fail( + new NpmTarballError(`Tarball entry escapes the destination directory: ${entry.name}`), + ); + } + + yield* fs.makeDirectory(nodePath.dirname(target), { recursive: true }); + yield* fs.writeFile(target, entry.data); + written.push(target); + } + + return written; +}); diff --git a/apps/server/src/dictation/worker.ts b/apps/server/src/dictation/worker.ts new file mode 100644 index 000000000..93c5f7ecd --- /dev/null +++ b/apps/server/src/dictation/worker.ts @@ -0,0 +1,152 @@ +// @effect-diagnostics nodeBuiltinImport:off - child process entry, plain Node +/** + * Dictation worker: the child process that owns the sherpa-onnx native addon. + * + * It runs as a hidden subcommand of the server binary (`threadlines + * dictation-worker`) so the entry path resolves the same in dev, npm and + * desktop builds. Keeping the addon out of the server process means a native + * crash cannot take the server down, and the multi-second model load never + * blocks the event loop. + * + * The parent sets the platform's library-path variable in the spawn env + * before this process starts, so the addon's sibling shared libraries + * resolve; the worker itself does nothing special about that. + * + * @module dictation/worker + */ +import { createRequire } from "node:module"; + +export type DictationWorkerRequest = + | { + readonly type: "load"; + readonly model: string; + readonly addonPath: string; + readonly config: unknown; + } + | { + readonly type: "transcribe"; + readonly requestId: string; + readonly samples: Int16Array; + readonly sampleRate: number; + } + | { readonly type: "unload" } + | { readonly type: "shutdown" }; + +export type DictationWorkerResponse = + | { readonly type: "loaded"; readonly model: string } + | { readonly type: "result"; readonly requestId: string; readonly text: string } + | { readonly type: "error"; readonly requestId?: string; readonly message: string }; + +/** The handful of raw addon calls offline recognition needs. */ +interface SherpaAddon { + readonly createOfflineRecognizer: (config: unknown) => unknown; + readonly createOfflineStream: (recognizer: unknown) => unknown; + readonly acceptWaveformOffline: ( + stream: unknown, + waveform: { readonly samples: Float32Array; readonly sampleRate: number }, + ) => void; + readonly decodeOfflineStream: (recognizer: unknown, stream: unknown) => void; + readonly getOfflineStreamResultAsJson: (stream: unknown) => string; +} + +const toFloat32 = (samples: Int16Array): Float32Array => { + const float = new Float32Array(samples.length); + for (let index = 0; index < samples.length; index += 1) { + float[index] = samples[index]! / 32768; + } + return float; +}; + +const readText = (json: string): string => { + const parsed: unknown = JSON.parse(json); + if (typeof parsed === "object" && parsed !== null && "text" in parsed) { + const text = (parsed as { readonly text?: unknown }).text; + return typeof text === "string" ? text : ""; + } + return ""; +}; + +/** + * Runs the worker's message loop until the parent asks it to shut down or the + * IPC channel closes. Exported so the CLI subcommand is a one-liner. + */ +export function runDictationWorker(): void { + const send = (response: DictationWorkerResponse): void => { + process.send?.(response); + }; + + let addon: SherpaAddon | undefined; + let recognizer: unknown; + let loadedModel: string | undefined; + + const unload = (): void => { + recognizer = undefined; + loadedModel = undefined; + }; + + const handle = (message: DictationWorkerRequest): void => { + switch (message.type) { + case "load": { + if (loadedModel === message.model && recognizer !== undefined) { + send({ type: "loaded", model: message.model }); + return; + } + unload(); + if (addon === undefined) { + addon = createRequire(import.meta.url)(message.addonPath) as SherpaAddon; + } + recognizer = addon.createOfflineRecognizer(message.config); + loadedModel = message.model; + send({ type: "loaded", model: message.model }); + return; + } + case "transcribe": { + if (addon === undefined || recognizer === undefined) { + send({ + type: "error", + requestId: message.requestId, + message: "No dictation model is loaded.", + }); + return; + } + const stream = addon.createOfflineStream(recognizer); + addon.acceptWaveformOffline(stream, { + samples: toFloat32(message.samples), + sampleRate: message.sampleRate, + }); + addon.decodeOfflineStream(recognizer, stream); + send({ + type: "result", + requestId: message.requestId, + text: readText(addon.getOfflineStreamResultAsJson(stream)), + }); + return; + } + case "unload": { + unload(); + return; + } + case "shutdown": { + unload(); + process.exit(0); + } + } + }; + + process.on("message", (message: DictationWorkerRequest) => { + try { + handle(message); + } catch (cause) { + send({ + type: "error", + ...(message.type === "transcribe" ? { requestId: message.requestId } : {}), + message: cause instanceof Error ? cause.message : String(cause), + }); + } + }); + + // A disconnected parent means nothing will ever read our results again. + process.on("disconnect", () => { + process.exit(0); + }); +} diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 60db81eb5..bc750cd6f 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -64,6 +64,7 @@ import { ProviderAuthSessions, type ProviderAuthSessionsShape, } from "./provider/auth/ProviderAuthSessions.ts"; +import { DictationLive } from "./dictation/DictationService.ts"; import { makeRoutesLayer } from "./server.ts"; import { resolveAttachmentRelativePath } from "./attachmentPaths.ts"; import { @@ -631,6 +632,9 @@ const buildAppUnderTest = (options?: { // Real rather than mocked: it holds no resources and its whole // behaviour is the rendezvous, so a mock would only assert wiring. PreviewAutomationBroker.layer, + // Real too: with no models on disk it only reports "missing" and + // never starts the worker, so a mock would assert nothing. + DictationLive, Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]), refresh: () => Effect.succeed([]), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5ac9c39f2..eb798c396 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -52,6 +52,7 @@ import { ThreadDiffStatBaselineReactorLive } from "./orchestration/Layers/Thread import { SleepInhibitorLive } from "./power/Layers/SleepInhibitor.ts"; import { StorageMaintenanceDaemonLive } from "./persistence/Layers/StorageMaintenance.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; +import { DictationLive } from "./dictation/DictationService.ts"; import * as PreviewAutomationBroker from "./preview/PreviewAutomationBroker.ts"; import { ProviderAuthSessionsLive } from "./provider/auth/ProviderAuthSessions.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; @@ -345,7 +346,10 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // The browser side of the agent's tools. Holds no resources of its own -- // it is a rendezvous between a provider turn and whichever client is showing // the thread -- so it merges in flat, with nothing beneath it. - Layer.provideMerge(PreviewAutomationBroker.layer), + // `DictationLive` sits alongside the broker: local speech-to-text, owning a + // worker child process and the model downloads, reading the selected model + // from the settings layer below. + Layer.provideMerge(Layer.mergeAll(PreviewAutomationBroker.layer, DictationLive)), Layer.provideMerge(PersistenceLayerLive), Layer.provideMerge(KeybindingsLive), Layer.provideMerge(ProviderRegistryLive), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 7b3274d51..42a51bcd6 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -117,6 +117,7 @@ import { redactServerSettingsForClient, ServerSettingsService } from "./serverSe import { ProviderAuthSessions } from "./provider/auth/ProviderAuthSessions.ts"; import { TerminalManager } from "./terminal/Services/Manager.ts"; import { realtimeAudioHub } from "./realtime/RealtimeAudioHub.ts"; +import { DictationService } from "./dictation/DictationService.ts"; import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries.ts"; import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem.ts"; import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths.ts"; @@ -260,6 +261,7 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const terminalManager = yield* TerminalManager; const providerAuthSessions = yield* ProviderAuthSessions; + const dictation = yield* DictationService; const providerRegistry = yield* ProviderRegistry; const providerService = yield* ProviderService; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; @@ -2302,6 +2304,36 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => realtimeAudioHub.subscribe(input.threadId), { "rpc.aggregate": "realtime" }, ), + [WS_METHODS.dictationSubscribeStatus]: (_input) => + observeRpcStream(WS_METHODS.dictationSubscribeStatus, dictation.streamChanges, { + "rpc.aggregate": "dictation", + }), + [WS_METHODS.dictationDownloadModel]: (input) => + observeRpcEffect( + WS_METHODS.dictationDownloadModel, + dictation.downloadModel(input.model), + { + "rpc.aggregate": "dictation", + }, + ), + [WS_METHODS.dictationCancelDownload]: (input) => + observeRpcEffect( + WS_METHODS.dictationCancelDownload, + dictation.cancelDownload(input.model), + { "rpc.aggregate": "dictation" }, + ), + [WS_METHODS.dictationRemoveModel]: (input) => + observeRpcEffect(WS_METHODS.dictationRemoveModel, dictation.removeModel(input.model), { + "rpc.aggregate": "dictation", + }), + [WS_METHODS.dictationWarmUp]: (_input) => + observeRpcEffect(WS_METHODS.dictationWarmUp, dictation.warmUp, { + "rpc.aggregate": "dictation", + }), + [WS_METHODS.dictationTranscribe]: (input) => + observeRpcEffect(WS_METHODS.dictationTranscribe, dictation.transcribe(input), { + "rpc.aggregate": "dictation", + }), [WS_METHODS.subscribeServerConfig]: (_input) => observeRpcStreamEffect( WS_METHODS.subscribeServerConfig, diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index 635727a05..72cde2cd8 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -268,6 +268,15 @@ function createMockEnvironmentApi(input: { git: {} as EnvironmentApi["git"], pullRequests: {} as EnvironmentApi["pullRequests"], realtime: {} as EnvironmentApi["realtime"], + dictation: { + // The composer subscribes on mount, so this one has to be callable. + subscribeStatus: () => () => undefined, + downloadModel: async () => undefined, + cancelDownload: async () => undefined, + removeModel: async () => undefined, + warmUp: async () => undefined, + transcribe: async () => ({ text: "" }), + }, orchestration: { dispatchCommand: input.dispatchCommand, getTurnDiff: (() => { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 9e8ea6c98..d0194af8c 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -129,8 +129,6 @@ import { import { useTheme } from "../hooks/useTheme"; import { useTurnDiffSummaries } from "../hooks/useTurnDiffSummaries"; import { useMediaQuery } from "../hooks/useMediaQuery"; -import { useRealtimeVoiceMode } from "../hooks/useRealtimeVoiceMode"; -import { useWsConnectionStatus } from "../rpc/wsConnectionState"; import { useCommandPaletteStore } from "../commandPaletteStore"; import { RIGHT_PANEL_INLINE_LAYOUT_MEDIA_QUERY, @@ -1491,7 +1489,6 @@ export default function ChatView(props: ChatViewProps) { // drive the environment picker in BranchToolbar. const allProjects = useStore(useShallow(selectProjectsAcrossEnvironments)); const primaryEnvironmentId = usePrimaryEnvironmentId(); - const primaryWsConnectionStatus = useWsConnectionStatus(); const savedEnvironmentRegistry = useSavedEnvironmentRegistryStore((s) => s.byId); const savedEnvironmentRuntimeById = useSavedEnvironmentRuntimeStore((s) => s.byId); const activeSavedEnvironmentRecord = @@ -2515,36 +2512,6 @@ export default function ChatView(props: ChatViewProps) { }, [activeProviderInstanceId, providerStatuses, selectedProvider]); const activeProviderDriver = activeProviderStatus?.driver ?? activeThread?.session?.provider ?? selectedProvider; - // Realtime voice remains dormant until Threadlines can support its separate - // API-key billing model as a complete product experience. Keeping this gate - // here also lets the hook clean up any projected session left by an older build. - const voiceSupported = false; - const voiceConnectionAvailable = activeThread - ? activeThread.environmentId === primaryEnvironmentId || primaryEnvironmentId === null - ? primaryWsConnectionStatus.phase === "connected" - : activeSavedEnvironmentConnectionState === "connected" - : false; - const activeSessionCanStartVoice = - activeThread?.session !== null && - activeThread?.session !== undefined && - activeThread.session.orchestrationStatus !== "starting" && - activeThread.session.orchestrationStatus !== "stopped" && - activeThread.session.orchestrationStatus !== "error"; - const voiceMode = useRealtimeVoiceMode({ - threadId: activeThread?.id ?? null, - environmentId, - supported: voiceSupported, - canStart: isServerThread && activeSessionCanStartVoice, - connectionAvailable: voiceConnectionAvailable, - projectedActive: activeThread?.voiceActive ?? false, - }); - const voiceStartDisabledReason = !isServerThread - ? "Send a message to create this thread before starting voice mode" - : !activeSessionCanStartVoice - ? "Start the Codex session before starting voice mode" - : !voiceConnectionAvailable - ? "Reconnect before starting voice mode" - : null; const activeProviderLabel = activeProviderStatus?.displayName?.trim() || formatProviderDriverKindLabel(activeProviderDriver); @@ -6881,20 +6848,6 @@ export default function ChatView(props: ChatViewProps) { scheduleStickToBottom={scheduleTimelineStickToBottom} onSend={onSend} onInterrupt={onInterrupt} - voiceControl={{ - supported: voiceSupported, - canStart: - voiceStartDisabledReason === null && - voiceMode.state.status !== "starting" && - voiceMode.state.status !== "active", - disabledReason: voiceStartDisabledReason, - projectedActive: voiceMode.projectedActive, - state: voiceMode.state, - onStart: voiceMode.start, - onToggleMute: voiceMode.toggleMute, - onStop: voiceMode.stop, - onModalityChange: voiceMode.setModality, - }} onCompactContext={contextCompactControlVisible ? onCompactContext : undefined} goalDispatching={ activeThread !== undefined && goalDispatchingThreadId === activeThread.id diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 56eb85594..b330d3ecf 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -115,7 +115,7 @@ import { ComposerPendingApprovalPanel } from "./ComposerPendingApprovalPanel"; import { ComposerPendingUserInputPanel } from "./ComposerPendingUserInputPanel"; import { ComposerGoalBar, type ComposerGoalSetInput } from "./ComposerGoalBar"; import { ComposerPlanFollowUpBanner } from "./ComposerPlanFollowUpBanner"; -import type { ComposerNotice } from "./composerNotices"; +import { type ComposerNotice, selectComposerNotices } from "./composerNotices"; import { ComposerDock, hasComposerDockContent } from "./ComposerDock"; import type { ComposerPullRequest } from "./ComposerPullRequestRow"; import { ComposerPendingDrawingContexts } from "./ComposerPendingDrawingContexts"; @@ -187,7 +187,8 @@ import { searchProviderSkills } from "../../providerSkillSearch"; import { resolveComposerSkillReferences } from "../../providerSkillReferences"; import { useHorizontalOverflow } from "../../hooks/useHorizontalOverflow"; import { useMediaQuery } from "../../hooks/useMediaQuery"; -import { ComposerVoiceControls, type ComposerVoiceControlsProps } from "./ComposerVoiceControls"; +import { ComposerDictationControl } from "./ComposerDictationControl"; +import { useDictation } from "../../dictation/useDictation"; const ATTACHMENT_SIZE_LIMIT_LABEL = `${Math.round(PROVIDER_SEND_TURN_MAX_IMAGE_BYTES / (1024 * 1024))}MB`; const ALL_ATTACHMENT_ACCEPT = `image/*,${FILE_ATTACHMENT_ACCEPT}`; @@ -593,7 +594,6 @@ export interface ChatComposerProps { // Callbacks onSend: (e?: { preventDefault: () => void }) => void; onInterrupt: () => void; - voiceControl?: ComposerVoiceControlsProps | undefined; // Goal (Codex goal mode) goalDispatching?: boolean | undefined; /** Freshly dispatched goal state not yet confirmed by the projection — @@ -683,7 +683,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) scheduleStickToBottom, onSend, onInterrupt, - voiceControl, goalDispatching, optimisticGoal, onSetThreadGoal, @@ -1515,6 +1514,17 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const attachmentsDisabledReason = attachmentsDisabled ? "Finish the pending prompt before adding attachments" : null; + // Dictation writes into the prompt, so it is blocked by everything that + // blocks typing, named the way the placeholder names it. + const dictationDisabledReason = isComposerApprovalState + ? "Resolve the approval first" + : hasBlockingQuestion + ? "Answer the question first" + : environmentUnavailable + ? `${environmentUnavailable.label} is ${ + environmentUnavailable.connectionState === "connecting" ? "connecting" : "disconnected" + }` + : null; // ------------------------------------------------------------------ // Prompt helpers @@ -2333,6 +2343,37 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) [promptRef, setPrompt], ); + // Dictation adds to what is typed instead of replacing it, so it lands at + // the caret through the same replacement path the command menu uses, with a + // space added on either side only where the neighbouring character needs one. + const insertDictatedText = useCallback( + (text: string) => { + const currentText = promptRef.current; + const cursor = Math.max(0, Math.min(currentText.length, composerCursor)); + const before = currentText.slice(0, cursor).slice(-1); + const after = currentText.slice(cursor).slice(0, 1); + const prefix = before !== "" && !/\s/.test(before) ? " " : ""; + const suffix = after !== "" && !/\s/.test(after) ? " " : ""; + applyPromptReplacement(cursor, cursor, `${prefix}${text}${suffix}`, { + focusEditorAfterReplace: true, + }); + }, + [applyPromptReplacement, composerCursor, promptRef], + ); + + const dictation = useDictation({ environmentId, onText: insertDictatedText }); + const dictationNotice: ComposerNotice | null = dictation.error + ? { + id: "dictation", + severity: "error", + lead: "Dictation failed.", + detail: dictation.error, + dismissLabel: "Dismiss", + onDismiss: dictation.clearError, + } + : null; + const dockedNotices = selectComposerNotices([...notices, dictationNotice]); + const readComposerSnapshot = useCallback((): { value: string; cursor: number; @@ -3142,7 +3183,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) about sending: those stay in front of the send button. */}
- {voiceControl?.state.error ? ( -
- {voiceControl.state.error} -
- ) : null} {composerMenuOpen && !isComposerApprovalState && (
- {voiceControl ? : null} + + undefined); + +function installEnvironmentApi(status: DictationStatus) { + __setEnvironmentApiOverrideForTests(ENVIRONMENT_ID, { + dictation: { + subscribeStatus: (callback: (next: DictationStatus) => void) => { + callback(status); + return () => undefined; + }, + downloadModel, + cancelDownload: async () => undefined, + removeModel: async () => undefined, + warmUp: async () => undefined, + transcribe: async () => ({ text: "" }), + }, + } as unknown as EnvironmentApi); +} + +/** Stands in for `useDictation`, which `ChatComposer` owns in the real app. */ +function useFakeDictation(): DictationControl { + const [status, setStatus] = useState("idle"); + return { + status, + elapsedMs: 7_000, + error: null, + start: () => setStatus("recording"), + stop: () => setStatus("transcribing"), + cancel: () => setStatus("idle"), + clearError: () => undefined, + }; +} + +function Harness({ disabled = false }: { disabled?: boolean }) { + const dictation = useFakeDictation(); + return ( + + ); +} + +function renderInApp(children: ReactNode) { + const rootRoute = createRootRoute({ component: () => children }); + const indexRoute = createRoute({ getParentRoute: () => rootRoute, path: "/" }); + const router = createRouter({ + routeTree: rootRoute.addChildren([indexRoute]), + history: createMemoryHistory({ initialEntries: ["/"] }), + }); + return render( + + + , + ); +} + +describe("ComposerDictationControl", () => { + beforeEach(async () => { + resetServerStateForTests(); + resetAppAtomRegistryForTests(); + __resetDictationStatusForTests(); + downloadModel.mockClear(); + window.nativeApi = { + persistence: { + getClientSettings: vi.fn().mockResolvedValue(null), + setClientSettings: vi.fn().mockResolvedValue(undefined), + }, + } as unknown as LocalApi; + await __resetLocalApiForTests(); + }); + + afterEach(async () => { + __resetEnvironmentApiOverridesForTests(); + __resetDictationStatusForTests(); + Reflect.deleteProperty(window, "nativeApi"); + await __resetLocalApiForTests(); + resetServerStateForTests(); + document.body.innerHTML = ""; + }); + + it("offers the mic and its options once the model is on the server", async () => { + installEnvironmentApi(makeStatus("ready")); + const mounted = await renderInApp(); + + await expect + .element(page.getByRole("button", { name: "Dictate" }), { timeout: 5_000 }) + .toBeVisible(); + await page.getByRole("button", { name: "Dictation options" }).click(); + await expect + .element(page.getByRole("menuitemradio", { name: "Default" }), { timeout: 5_000 }) + .toBeVisible(); + await expect + .element(page.getByRole("menuitemcheckbox", { name: /Hold to record/ }), { timeout: 5_000 }) + .toBeVisible(); + await expect + .element(page.getByRole("link", { name: "Dictation settings" }), { timeout: 5_000 }) + .toBeVisible(); + await mounted.unmount(); + }); + + it("records on click and offers a stop square when hold-to-record is off", async () => { + installEnvironmentApi(makeStatus("ready")); + updateSettings({ dictationHoldToRecord: false }); + const mounted = await renderInApp(); + + await page.getByRole("button", { name: "Dictate" }).click(); + + const stop = page.getByRole("button", { name: "Stop recording" }); + await expect.element(stop, { timeout: 5_000 }).toBeVisible(); + await expect.element(page.getByText("0:07"), { timeout: 5_000 }).toBeVisible(); + + await stop.click(); + await expect.element(page.getByText("Transcribing…"), { timeout: 5_000 }).toBeVisible(); + await mounted.unmount(); + updateSettings({ dictationHoldToRecord: true }); + }); + + it("asks to download the selected model instead of recording when it is missing", async () => { + installEnvironmentApi(makeStatus("missing")); + const mounted = await renderInApp(); + + await page.getByRole("button", { name: "Dictate" }).click(); + + await expect.element(page.getByText("Set up dictation"), { timeout: 5_000 }).toBeVisible(); + const download = page.getByRole("button", { name: "Download Parakeet · 631 MB" }); + await expect.element(download, { timeout: 5_000 }).toBeVisible(); + + await download.click(); + await vi.waitFor(() => expect(downloadModel).toHaveBeenCalledWith({ model: "parakeet" }), { + timeout: 5_000, + }); + await mounted.unmount(); + }); + + it("explains why the mic is off when the composer is blocked", async () => { + installEnvironmentApi(makeStatus("ready")); + const mounted = await renderInApp(); + + // The tooltip itself cannot open: a disabled button takes no pointer + // events, so the inert control and its reason are what there is to assert. + await expect + .element(page.getByRole("button", { name: "Dictate" }), { timeout: 5_000 }) + .toBeDisabled(); + await expect + .element(page.getByRole("button", { name: "Dictation options" }), { timeout: 5_000 }) + .toBeDisabled(); + await mounted.unmount(); + }); +}); diff --git a/apps/web/src/components/chat/ComposerDictationControl.tsx b/apps/web/src/components/chat/ComposerDictationControl.tsx new file mode 100644 index 000000000..b751595cd --- /dev/null +++ b/apps/web/src/components/chat/ComposerDictationControl.tsx @@ -0,0 +1,491 @@ +/** + * The composer's dictation control. + * + * A split button in the right-hand cluster: the mic records, the narrow + * chevron opens microphone and hold-to-record options. While recording, the + * control is replaced in place by a timer pill so nothing else in the row + * moves. The first use, before the speech model is on the server, opens a + * setup popover instead of recording. + * + * The recording state itself lives in `useDictation`, owned by `ChatComposer`, + * because a failed clip is reported in the composer's notice dock. + * + * @module ComposerDictationControl + */ +import type { DictationModelId, EnvironmentApi, EnvironmentId } from "@threadlines/contracts"; +import { Link } from "@tanstack/react-router"; +import { ChevronDownIcon, LoaderCircleIcon, MicIcon, Settings2Icon } from "lucide-react"; +import { memo, useEffect, useRef, useState, type PointerEvent as ReactPointerEvent } from "react"; + +import { cn } from "~/lib/utils"; +import { + DictationProgressBar, + DictationProgressLabel, +} from "../../dictation/DictationDownloadProgress"; +import { + DICTATION_MODEL_PRESENTATION, + findDictationModel, + selectedModelReady, +} from "../../dictation/dictationModels"; +import { useDictationStatus } from "../../dictation/dictationStatusStore"; +import type { DictationControl } from "../../dictation/useDictation"; +import { useMicrophoneDevices } from "../../dictation/useMicrophoneDevices"; +import { useSettings, useUpdateSettings } from "../../hooks/useSettings"; +import { readEnvironmentApi } from "../../environmentApi"; +import { formatDownloadSize } from "../../lib/formatBytes"; +import { Button } from "../ui/button"; +import { + Menu, + MenuCheckboxItem, + MenuGroup, + MenuGroupLabel, + MenuPopup, + MenuRadioGroup, + MenuRadioItem, + MenuSeparator, + MenuTrigger, +} from "../ui/menu"; +import { Popover, PopoverPopup, PopoverTrigger } from "../ui/popover"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; + +const UNSUPPORTED_PLATFORM_REASON = "Dictation isn't available on this server's platform."; +/** A press this short in hold mode is a tap, not speech: drop it and explain. */ +const HOLD_TAP_MS = 300; +const HOLD_HINT_MS = 2_500; +/** Chromium lists the system default as its own device under this id. */ +const BROWSER_DEFAULT_DEVICE_ID = "default"; + +const MIC_CLASS_NAME = + "rounded-s-full rounded-e-none text-muted-foreground/70 hover:text-foreground/80"; +const CHEVRON_CLASS_NAME = + "h-8 w-3.5 rounded-s-none rounded-e-full px-0 text-muted-foreground/70 hover:text-foreground/80 sm:h-7"; + +export interface ComposerDictationControlProps { + environmentId: EnvironmentId | null | undefined; + /** Dictation can't run right now (pending approval, blocking question, offline). */ + disabled: boolean; + disabledReason: string | null; + isMobileViewport: boolean; + dictation: DictationControl; +} + +function formatTimer(elapsedMs: number): string { + const seconds = Math.max(0, Math.floor(elapsedMs / 1_000)); + return `${Math.floor(seconds / 60)}:${String(seconds % 60).padStart(2, "0")}`; +} + +export const ComposerDictationControl = memo(function ComposerDictationControl( + props: ComposerDictationControlProps, +) { + const { environmentId, disabled, disabledReason, isMobileViewport, dictation } = props; + const status = useDictationStatus(environmentId); + const { updateSettings } = useUpdateSettings(); + const holdToRecord = useSettings((settings) => settings.dictationHoldToRecord); + const microphoneDeviceId = useSettings((settings) => settings.dictationMicrophoneDeviceId); + const selectedModel: DictationModelId = useSettings((settings) => settings.dictationModel); + + const [menuOpen, setMenuOpen] = useState(false); + const [setupOpen, setSetupOpen] = useState(false); + // A tap in hold mode records nothing. Rather than a "nothing was heard" + // error, the control says how the mic works for a moment. + const [holdHint, setHoldHint] = useState(false); + const holdHintTimerRef = useRef | null>(null); + const micRef = useRef(null); + const pressStartedAtRef = useRef(null); + const devices = useMicrophoneDevices(menuOpen); + const hasBrowserDefault = devices.some((device) => device.deviceId === BROWSER_DEFAULT_DEVICE_ID); + + useEffect( + () => () => { + if (holdHintTimerRef.current !== null) { + clearTimeout(holdHintTimerRef.current); + } + }, + [], + ); + + const isRecording = dictation.status === "recording"; + const isTranscribing = dictation.status === "transcribing"; + const platformUnsupported = status?.runtime.supported === false; + const modelReady = selectedModelReady(status, selectedModel); + const modelStatus = findDictationModel(status, selectedModel); + const modelPresentation = DICTATION_MODEL_PRESENTATION[selectedModel]; + const runtimeDownloading = status?.runtime.state === "downloading"; + const isDownloading = runtimeDownloading || modelStatus?.state === "downloading"; + // The server keeps the last failure on whichever piece failed; either one + // belongs in the popover so a retry is not a blind click. + const downloadError = status?.runtime.error ?? modelStatus?.error ?? null; + + // Until the first status lands there is no way to tell "download needed" + // from "not connected yet", so the mic waits rather than guessing. + const statusUnknown = status === undefined; + const blocked = disabled || platformUnsupported || statusUnknown; + const blockedReason = platformUnsupported + ? UNSUPPORTED_PLATFORM_REASON + : statusUnknown + ? "Checking dictation on this server…" + : disabledReason; + + // Derived rather than stored: once the model lands there is nothing left to + // set up. In hold mode the mic itself is the pill; in click mode a separate + // pill with a stop square takes its place. + const setupVisible = setupOpen && !modelReady; + const showStopSquare = isRecording && !holdToRecord; + + const showHoldHint = () => { + if (holdHintTimerRef.current !== null) { + clearTimeout(holdHintTimerRef.current); + } + setHoldHint(true); + holdHintTimerRef.current = setTimeout(() => { + holdHintTimerRef.current = null; + setHoldHint(false); + }, HOLD_HINT_MS); + }; + + const toggleRecording = () => { + if (blocked || isTranscribing) { + return; + } + if (isRecording) { + dictation.stop(); + return; + } + setHoldHint(false); + dictation.start(); + }; + + const onMicPointerDown = (event: ReactPointerEvent) => { + // A missing model is the popover trigger's business, not a recording gesture. + if (!holdToRecord || blocked || isTranscribing || isRecording || !modelReady) { + return; + } + pressStartedAtRef.current = Date.now(); + setHoldHint(false); + event.currentTarget.setPointerCapture(event.pointerId); + dictation.start(); + }; + + const onMicPointerRelease = () => { + const pressStartedAt = pressStartedAtRef.current; + if (pressStartedAt === null) { + return; + } + pressStartedAtRef.current = null; + if (Date.now() - pressStartedAt < HOLD_TAP_MS) { + dictation.cancel(); + showHoldHint(); + return; + } + dictation.stop(); + }; + + const downloadProgress = runtimeDownloading + ? { done: status?.runtime.bytesDownloaded ?? 0, total: status?.runtime.bytesTotal ?? 0 } + : { done: modelStatus?.bytesDownloaded ?? 0, total: modelStatus?.bytesTotal ?? 0 }; + + const runDictationCommand = (run: (api: EnvironmentApi) => Promise) => { + if (!environmentId) { + return; + } + const api = readEnvironmentApi(environmentId); + if (!api) { + return; + } + void run(api).catch(() => undefined); + }; + + const setupPopup = ( + + {isDownloading ? ( +
+

Downloading {modelPresentation.name}

+ {runtimeDownloading ? ( +

Preparing speech engine…

+ ) : null} + +
+ + +
+
+ ) : ( +
+
+

Set up dictation

+

+ Speech is turned into text on this computer. Nothing is sent to the internet. Download + the speech model once to start. +

+ {downloadError ? ( +

{downloadError}

+ ) : null} +
+
+ + +
+
+ )} + + ); + + if (isTranscribing) { + return ( + + + ); + } + + return ( + { + if (event.key === "Escape" && isRecording) { + event.stopPropagation(); + dictation.cancel(); + } + }} + > + {/* Anchored to the mic rather than triggered by it: the mic is already + the setup popover's trigger, and this only shows after a tap. */} + { + if (!open) { + setHoldHint(false); + } + }} + > + + Hold the mic while you talk, then let go + + + {blocked ? ( + + + } + > + + {blockedReason ?? "Dictate"} + + ) : showStopSquare ? null : ( + // The mic is the setup popover's trigger so Base UI anchors and + // dismisses it properly, but the popover only opens while the model is + // still missing: once it is there the same click records instead. + setSetupOpen(open && !modelReady)}> + + } + onPointerDown={onMicPointerDown} + onPointerUp={onMicPointerRelease} + onPointerCancel={onMicPointerRelease} + onLostPointerCapture={onMicPointerRelease} + onContextMenu={(event) => { + if (isRecording) { + event.preventDefault(); + } + }} + onClick={(event) => { + // Pointer presses are already handled above in hold mode; a + // keyboard activation reports no pointer detail and still toggles. + if (!modelReady || (holdToRecord && event.detail !== 0)) { + return; + } + toggleRecording(); + }} + > + {isRecording ? ( + <> + + {setupPopup} + + )} + + {showStopSquare ? ( + + + ) : null} + + {/* Phones have one microphone and keep the hold switch in Settings, so + the options menu would only cost row width there. */} + {isRecording || isMobileViewport ? null : ( + + + } + > + + + + {/* The gear rides in the label's empty right half instead of + spending a row and a divider on a "Dictation settings…" item. */} + + Microphone + + + {/* "Follow the system default" is stored as null. Chromium lists + that default as a device of its own, so it stands in for the + plain "Default" row wherever it exists. */} + + updateSettings({ + dictationMicrophoneDeviceId: + typeof value === "string" && + value !== "" && + value !== BROWSER_DEFAULT_DEVICE_ID + ? value + : null, + }) + } + > + {hasBrowserDefault ? null : Default} + {devices.map((device) => ( + + {device.label} + + ))} + + + + + updateSettings({ dictationHoldToRecord: Boolean(checked) }) + } + > + + Hold to record + + {isMobileViewport + ? "Off: tap to start, tap again to stop." + : "Off: click to start, click again to stop."} + + + + + + )} + + ); +}); diff --git a/apps/web/src/components/settings/DiagnosticsSettings.tsx b/apps/web/src/components/settings/DiagnosticsSettings.tsx index c5c7c62b6..6269ae0c5 100644 --- a/apps/web/src/components/settings/DiagnosticsSettings.tsx +++ b/apps/web/src/components/settings/DiagnosticsSettings.tsx @@ -18,6 +18,7 @@ import * as Option from "effect/Option"; import { ensureLocalApi } from "../../localApi"; import { useRelativeTimeTick } from "../../hooks/useRelativeTimeTick"; import { copyTextToClipboard } from "../../lib/clipboard"; +import { formatBytes } from "../../lib/formatBytes"; import { cn } from "../../lib/utils"; import { resolveAndPersistPreferredEditor } from "../../editorPreferences"; import { formatRelativeTime } from "../../timestampFormat"; @@ -51,18 +52,6 @@ function formatDuration(value: number): string { return `${(value / 1_000).toFixed(value >= 10_000 ? 1 : 2)} s`; } -function formatBytes(value: number): string { - if (value < 1024) return `${value} B`; - const units = ["KB", "MB", "GB"] as const; - let unitIndex = -1; - let next = value; - do { - next /= 1024; - unitIndex += 1; - } while (next >= 1024 && unitIndex < units.length - 1); - return `${next.toFixed(next >= 10 ? 1 : 2)} ${units[unitIndex]}`; -} - function formatRelative(value: DateTime.Utc | null): string { if (!value) return "No trace records"; const relative = formatRelativeTime(DateTime.formatIso(value)); diff --git a/apps/web/src/components/settings/DictationSettings.browser.tsx b/apps/web/src/components/settings/DictationSettings.browser.tsx new file mode 100644 index 000000000..ce9ef2dae --- /dev/null +++ b/apps/web/src/components/settings/DictationSettings.browser.tsx @@ -0,0 +1,131 @@ +import "../../index.css"; + +import { + DEFAULT_SERVER_SETTINGS, + type DictationStatus, + type EnvironmentApi, + EnvironmentId, + type LocalApi, +} from "@threadlines/contracts"; +import { page } from "vite-plus/test/browser"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { render } from "vitest-browser-react"; + +import { __resetDictationStatusForTests } from "../../dictation/dictationStatusStore"; +import { + __resetEnvironmentApiOverridesForTests, + __setEnvironmentApiOverrideForTests, +} from "../../environmentApi"; +import { writePrimaryEnvironmentDescriptor } from "../../environments/primary/context"; +import { __resetLocalApiForTests } from "../../localApi"; +import { AppAtomRegistryProvider, resetAppAtomRegistryForTests } from "../../rpc/atomRegistry"; +import { resetServerStateForTests } from "../../rpc/serverState"; +import { DictationSettings } from "./DictationSettings"; + +const ENVIRONMENT_ID = EnvironmentId.make("environment-dictation-settings"); + +const STATUS: DictationStatus = { + runtime: { supported: true, state: "ready", bytesDownloaded: 0, bytesTotal: 0, error: null }, + models: [ + { + id: "parakeet", + state: "missing", + bytesDownloaded: 0, + bytesTotal: 661_190_513, + error: null, + }, + { + id: "moonshine", + state: "missing", + bytesDownloaded: 0, + bytesTotal: 123_967_539, + error: null, + }, + ], + engine: "idle", + loadedModel: null, + modelsDir: "/home/will/.threadlines/models/speech", +}; + +const downloadModel = vi.fn(async () => undefined); +const updateServerSettings = vi.fn().mockResolvedValue(DEFAULT_SERVER_SETTINGS); +const setClientSettings = vi.fn().mockResolvedValue(undefined); + +describe("DictationSettings", () => { + beforeEach(async () => { + resetServerStateForTests(); + resetAppAtomRegistryForTests(); + __resetDictationStatusForTests(); + downloadModel.mockClear(); + updateServerSettings.mockClear(); + setClientSettings.mockClear(); + window.nativeApi = { + persistence: { + getClientSettings: vi.fn().mockResolvedValue(null), + setClientSettings, + }, + server: { updateSettings: updateServerSettings }, + } as unknown as LocalApi; + await __resetLocalApiForTests(); + writePrimaryEnvironmentDescriptor({ + environmentId: ENVIRONMENT_ID, + label: "This computer", + } as never); + __setEnvironmentApiOverrideForTests(ENVIRONMENT_ID, { + dictation: { + subscribeStatus: (callback: (next: DictationStatus) => void) => { + callback(STATUS); + return () => undefined; + }, + downloadModel, + cancelDownload: async () => undefined, + removeModel: async () => undefined, + warmUp: async () => undefined, + transcribe: async () => ({ text: "" }), + }, + } as unknown as EnvironmentApi); + }); + + afterEach(async () => { + __resetEnvironmentApiOverridesForTests(); + __resetDictationStatusForTests(); + writePrimaryEnvironmentDescriptor(null); + Reflect.deleteProperty(window, "nativeApi"); + await __resetLocalApiForTests(); + resetServerStateForTests(); + document.body.innerHTML = ""; + }); + + it("picks a model, starts its download, and flips hold to record", async () => { + const mounted = await render( + + + , + ); + + const moonshine = page.getByRole("radio", { name: "Moonshine" }); + await expect.element(moonshine, { timeout: 5_000 }).toBeVisible(); + await moonshine.click(); + await vi.waitFor( + () => expect(updateServerSettings).toHaveBeenCalledWith({ dictationModel: "moonshine" }), + { timeout: 5_000 }, + ); + + await page.getByRole("button", { name: "Download" }).first().click(); + await vi.waitFor(() => expect(downloadModel).toHaveBeenCalledWith({ model: "parakeet" }), { + timeout: 5_000, + }); + + const holdSwitch = page.getByRole("switch", { name: "Hold the mic button to record" }); + await expect.element(holdSwitch, { timeout: 5_000 }).toHaveAttribute("aria-checked", "true"); + await holdSwitch.click(); + await vi.waitFor( + () => + expect(setClientSettings).toHaveBeenCalledWith( + expect.objectContaining({ dictationHoldToRecord: false }), + ), + { timeout: 5_000 }, + ); + await mounted.unmount(); + }); +}); diff --git a/apps/web/src/components/settings/DictationSettings.tsx b/apps/web/src/components/settings/DictationSettings.tsx new file mode 100644 index 000000000..310e302ea --- /dev/null +++ b/apps/web/src/components/settings/DictationSettings.tsx @@ -0,0 +1,209 @@ +/** + * Settings › General › Dictation. + * + * Where the speech model is picked and downloaded, where the disk it uses is + * freed, and the second home of the "Hold to record" switch. The status comes + * from the primary environment's server, because that is the machine the model + * files and the transcription live on. + * + * @module DictationSettings + */ +import type { DictationModelId, EnvironmentApi } from "@threadlines/contracts"; + +import { + DictationProgressBar, + DictationProgressLabel, +} from "../../dictation/DictationDownloadProgress"; +import { DICTATION_MODEL_PRESENTATION } from "../../dictation/dictationModels"; +import { useDictationStatus } from "../../dictation/dictationStatusStore"; +import { readEnvironmentApi } from "../../environmentApi"; +import { usePrimaryEnvironmentId } from "../../environments/primary/context"; +import { useSettings, useUpdateSettings } from "../../hooks/useSettings"; +import { formatDownloadSize } from "../../lib/formatBytes"; +import { Button } from "../ui/button"; +import { Radio, RadioGroup } from "../ui/radio-group"; +import { Switch } from "../ui/switch"; +import { SettingsRow, SettingsSection } from "./settingsLayout"; + +const MODEL_ORDER: ReadonlyArray = ["parakeet", "moonshine"]; + +export function DictationSettings() { + const environmentId = usePrimaryEnvironmentId(); + const status = useDictationStatus(environmentId); + const { updateSettings } = useUpdateSettings(); + const selectedModel = useSettings((settings) => settings.dictationModel); + const holdToRecord = useSettings((settings) => settings.dictationHoldToRecord); + + const runDictationCommand = (run: (api: EnvironmentApi) => Promise) => { + if (!environmentId) { + return; + } + const api = readEnvironmentApi(environmentId); + if (!api) { + return; + } + void run(api).catch(() => undefined); + }; + + if (status?.runtime.supported === false) { + return ( + +
+ Dictation isn't available on this server's platform. +
+ + ); + } + + const statusUnknown = status === undefined; + const readyModels = status?.models.filter((model) => model.state === "ready") ?? []; + const readyBytes = readyModels.reduce((sum, model) => sum + model.bytesTotal, 0); + + return ( + + + { + if (value === "parakeet" || value === "moonshine") { + updateSettings({ dictationModel: value }); + } + }} + > + {MODEL_ORDER.map((modelId) => { + const presentation = DICTATION_MODEL_PRESENTATION[modelId]; + const modelStatus = status?.models.find((entry) => entry.id === modelId); + return ( +
+
+ +
+
+ + {presentation.name} + + + {presentation.meta} + +
+

{presentation.description}

+ {modelStatus?.error ? ( +

{modelStatus.error}

+ ) : null} +
+
+
+ {modelStatus?.state === "downloading" ? ( + <> + + + + + + + ) : modelStatus?.state === "ready" ? ( + <> + + Downloaded + + + + ) : ( + + )} +
+
+ ); + })} + + + + + updateSettings({ dictationHoldToRecord: Boolean(checked) }) + } + aria-label="Hold the mic button to record" + /> + } + /> + + {readyModels.length > 0 && status ? ( + + {formatDownloadSize(readyBytes)} · {status.modelsDir} + + } + control={ + + } + /> + ) : null} + + ); +} diff --git a/apps/web/src/components/settings/SettingsPanels.browser.tsx b/apps/web/src/components/settings/SettingsPanels.browser.tsx index 83bd5dcf1..bd30dbe4a 100644 --- a/apps/web/src/components/settings/SettingsPanels.browser.tsx +++ b/apps/web/src/components/settings/SettingsPanels.browser.tsx @@ -1294,7 +1294,9 @@ describe("GeneralSettingsPanel observability", () => { , ); - await expect.element(page.getByText("About")).toBeInTheDocument(); + await expect + .element(page.getByRole("heading", { name: "About", exact: true })) + .toBeInTheDocument(); await expect .element(page.getByRole("heading", { name: "Diagnostics", exact: true })) .toBeInTheDocument(); diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 3d86a6d85..119304eb5 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -102,6 +102,7 @@ import { type ProviderSettingsRow, } from "./SettingsPanels.logic"; import { useRelativeTimeTick } from "../../hooks/useRelativeTimeTick"; +import { DictationSettings } from "./DictationSettings"; import { SettingResetButton, SettingsPageContainer, @@ -934,6 +935,8 @@ export function GeneralSettingsPanel({ surface = "full" }: { surface?: "full" | ) : null} + + {!isPhoneSurface ? ( <> diff --git a/apps/web/src/dictation/DictationDownloadProgress.tsx b/apps/web/src/dictation/DictationDownloadProgress.tsx new file mode 100644 index 000000000..78248a090 --- /dev/null +++ b/apps/web/src/dictation/DictationDownloadProgress.tsx @@ -0,0 +1,46 @@ +/** + * The download line shared by the composer's setup popover and the Settings + * model rows: a hairline bar and the byte count under it. Kept in one place so + * a download reads identically wherever the user happens to be watching it. + * + * @module DictationDownloadProgress + */ +import { cn } from "~/lib/utils"; +import { formatDownloadProgress } from "../lib/formatBytes"; + +export function DictationProgressBar({ + bytesDownloaded, + bytesTotal, + className, +}: { + bytesDownloaded: number; + bytesTotal: number; + className?: string; +}) { + const percent = + bytesTotal > 0 ? Math.min(100, Math.max(0, (bytesDownloaded / bytesTotal) * 100)) : 0; + return ( +