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 ( +