feat(speech): filter silence and transcribe live audio chunks

apps/server/src/speech/SpeechModelStore.test.ts

@@ -128,6 +128,9 @@ it.effect("resumes an interrupted transfer, replaces a corrupt same-size file, a
     ]);
     expect(yield* h.store.isReady(h.model)).toBe(true);
     expect(yield* h.store.installedPath(h.model)).toBe(installed);
+    const requestsBeforeReuse = h.requests.length;
+    expect(yield* h.store.download(h.model)).toBe(installed);
+    expect(h.requests).toHaveLength(requestsBeforeReuse);
   }).pipe(Effect.provide(dependencies)),
 );
 

apps/server/src/speech/SpeechModelStore.ts

@@ -17,6 +17,8 @@ import * as Stream from "effect/Stream";
 import * as ServerConfig from "../config.ts";
 import type { SpeechModel } from "./model.ts";
 
+type ModelFile = Pick<SpeechModel, "id" | "filename" | "revision" | "size" | "sha256">;
+
 class SpeechModelDownloadError extends Schema.TaggedError<SpeechModelDownloadError>()(
   "SpeechModelDownloadError",
   {
@@ -39,17 +41,17 @@ export class SpeechModelStore extends Context.Service<
   SpeechModelStore,
   {
     /** Cheap size check for status polls; never hashes the file. */
-    readonly isReady: (model: SpeechModel) => Effect.Effect<boolean>;
+    readonly isReady: (model: ModelFile) => Effect.Effect<boolean>;
     /** The installed file once its SHA-256 has been verified, hashing it on first use. */
-    readonly installedPath: (model: SpeechModel) => Effect.Effect<string | undefined>;
+    readonly installedPath: (model: ModelFile) => Effect.Effect<string | undefined>;
     readonly download: (
-      model: SpeechModel,
+      model: ModelFile,
       progress?: SpeechModelDownloadProgress,
     ) => Effect.Effect<
       string,
       SpeechModelDownloadError | PlatformError | HttpClientError.HttpClientError
     >;
-    readonly remove: (model: SpeechModel) => Effect.Effect<void, PlatformError>;
+    readonly remove: (model: ModelFile) => Effect.Effect<void, PlatformError>;
   }
 >()("t3/speech/SpeechModelStore") {}
 
@@ -62,11 +64,11 @@ const make = Effect.gen(function* () {
   const random = yield* SecureRandom.SecureRandom;
   const config = yield* ServerConfig.ServerConfig;
   const directory = path.join(config.stateDir, "speech", "models");
-  const modelPath = (model: SpeechModel) => path.join(directory, model.filename);
+  const modelPath = (model: ModelFile) => path.join(directory, model.filename);
   // Files whose digest failed, by identity, so status polls stop calling them ready.
   const corrupt = new Map<string, string>();
 
-  const identity = (model: SpeechModel, info: FileSystem.File.Info) =>
+  const identity = (model: ModelFile, info: FileSystem.File.Info) =>
     [
       model.sha256,
       info.size,
@@ -77,15 +79,15 @@ const make = Effect.gen(function* () {
 
   const statFile = (file: string) => fs.stat(file).pipe(Effect.orElseSucceed(() => undefined));
 
-  const expectedFile = (model: SpeechModel, file: string) =>
+  const expectedFile = (model: ModelFile, file: string) =>
     statFile(file).pipe(
       Effect.map((info) =>
         info?.type === "File" && Number(info.size) === model.size ? info : undefined,
       ),
     );
 
   const digestMatches = Effect.fn("SpeechModelStore.digest")(
-    function* (model: SpeechModel, file: string) {
+    function* (model: ModelFile, file: string) {
       const hash = NodeCrypto.createHash("sha256");
       yield* fs
         .stream(file)
@@ -97,7 +99,7 @@ const make = Effect.gen(function* () {
 
   // A verified file is trusted while its size, mtime, device and inode are unchanged.
   const verifiedPath = Effect.fn("SpeechModelStore.verifiedPath")(function* (
-    model: SpeechModel,
+    model: ModelFile,
     file: string,
     onVerificationStart?: () => void,
   ) {
@@ -119,7 +121,7 @@ const make = Effect.gen(function* () {
   });
 
   const fetchInto = Effect.fn("SpeechModelStore.fetch")(function* (
-    model: SpeechModel,
+    model: ModelFile,
     url: string,
     partialPath: string,
     offset: number,

apps/server/src/speech/SpeechService.test.ts

@@ -7,18 +7,21 @@ import * as TestClock from "effect/testing/TestClock";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Result from "effect/Result";
+import * as PlatformError from "effect/PlatformError";
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import { DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS } from "@t3tools/contracts";
 import * as ServerSettings from "../serverSettings.ts";
 import * as SpeechModelStore from "./SpeechModelStore.ts";
 import * as SpeechService from "./SpeechService.ts";
+import { SILERO_VAD } from "./silero.ts";
 
 const native = vi.hoisted(() => ({
   backend: "Vulkan0",
   supportsStreaming: true,
+  supportsLongForm: false,
   supportsTranslation: true,
   supportsInitialPrompt: false,
-  begin: vi.fn(async (_language?: string, _words?: readonly string[], _dictionary?: unknown) => {}),
+  begin: vi.fn(async (_options?: unknown, _filterSilence?: boolean, _vadPath?: string) => {}),
   feed: vi.fn(async (_pcm: Float32Array) => ({
     revision: 1,
     text: { committed: "", tentative: "hello" },
@@ -61,13 +64,13 @@ const modelDefinitions = vi.hoisted(() => [
     name: "fallback",
     description: "fallback",
     size: 8,
-    languages: ["en"],
+    languages: ["en", "fr"],
     accuracy: 2,
     speed: 2,
     recommended: false,
     supportsStreaming: false,
     supportsLanguageDetection: false,
-    supportsTranslation: false,
+    supportsTranslation: true,
   },
 ]);
 vi.mock("./native.ts", () => ({
@@ -108,6 +111,7 @@ beforeEach(() => {
   vi.clearAllMocks();
   native.backend = "Vulkan0";
   native.supportsInitialPrompt = false;
+  native.supportsStreaming = true;
   readyModels.clear();
   readyModels.add("test-model");
   readyModels.add("fallback-model");
@@ -140,6 +144,21 @@ it.effect("prepares a batch model before audio arrives and reuses it for transcr
     yield* speech.prepareModel;
     expect(loadNative).toHaveBeenCalledTimes(1);
     expect(native.transcribe).not.toHaveBeenCalled();
+    native.supportsStreaming = false;
+    yield* Effect.gen(function* () {
+      const stream = yield* speech.startStream({
+        ...DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS,
+        speechLanguage: "fr",
+        speechTranslateToEnglish: true,
+        speechSilenceFiltering: false,
+      });
+      expect(native.begin).toHaveBeenLastCalledWith(
+        { timestamps: "none", language: "fr", task: "translate", targetLanguage: "en" },
+        false,
+        "test.gguf",
+      );
+      expect(yield* stream.finish).toBe("hello");
+    }).pipe(Effect.scoped);
     expect(yield* speech.transcribe(pcm(), DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS)).toBe("hello");
     expect(loadNative).toHaveBeenCalledTimes(1);
   }).pipe(Effect.provide(layer)),
@@ -150,28 +169,103 @@ it.effect("keeps a detached download exclusive until the transfer finishes", ()
     const speech = yield* SpeechService.SpeechService;
     const started = yield* Deferred.make<void>();
     const finish = yield* Deferred.make<void>();
-    download.mockImplementationOnce(() =>
-      Deferred.succeed(started, undefined).pipe(
-        Effect.andThen(Deferred.await(finish)),
-        Effect.as("test.gguf"),
-      ),
-    );
+    const helperStarted = yield* Deferred.make<void>();
+    const helperFinish = yield* Deferred.make<void>();
+    let helperStartedDuringInstall = false;
+    let helperInterrupted = false;
+    download
+      .mockImplementationOnce(() =>
+        Deferred.succeed(started, undefined).pipe(
+          Effect.andThen(Deferred.await(finish)),
+          Effect.as("test.gguf"),
+        ),
+      )
+      .mockImplementationOnce(() =>
+        Deferred.isDone(finish).pipe(
+          Effect.tap((done) =>
+            Effect.sync(() => {
+              helperStartedDuringInstall = !done;
+            }),
+          ),
+          Effect.andThen(Deferred.succeed(helperStarted, undefined)),
+          Effect.andThen(Deferred.await(helperFinish)),
+          Effect.as("silero.gguf"),
+          Effect.onInterrupt(() =>
+            Effect.sync(() => {
+              helperInterrupted = true;
+            }),
+          ),
+        ),
+      );
     const request = yield* speech.downloadModel("test-model").pipe(Effect.forkChild);
     yield* Deferred.await(started);
+    yield* Deferred.await(helperStarted);
     yield* Fiber.interrupt(request);
+    yield* Deferred.succeed(finish, undefined);
+    yield* TestClock.adjust("16 seconds");
+    expect(helperInterrupted).toBe(false);
     const removed = yield* Effect.result(speech.removeModel("test-model"));
     expect(Result.isFailure(removed) && removed.failure).toMatchObject({ _tag: "SpeechBusyError" });
     const join = yield* speech
       .downloadModel("test-model")
       .pipe(Effect.forkChild({ startImmediately: true }));
-    yield* Deferred.succeed(finish, undefined);
+    yield* Deferred.succeed(helperFinish, undefined);
     yield* Fiber.join(join);
-    expect(download).toHaveBeenCalledTimes(1);
+    expect(download.mock.calls.map(([model]) => model.id)).toEqual(["test-model", SILERO_VAD.id]);
+    expect(helperStartedDuringInstall).toBe(true);
     yield* speech.removeModel("test-model");
     expect(readyModels.has("test-model")).toBe(false);
   }).pipe(Effect.provide(layer)),
 );
 
+it.effect("bounds missing Silero recovery during recording and continues with raw audio", () =>
+  Effect.gen(function* () {
+    const helperStarted = yield* Deferred.make<void>();
+    const helperInterrupted = yield* Deferred.make<void>();
+    download.mockImplementationOnce(() =>
+      Deferred.succeed(helperStarted, undefined).pipe(
+        Effect.andThen(Effect.never),
+        Effect.onInterrupt(() => Deferred.succeed(helperInterrupted, undefined)),
+      ),
+    );
+    const speech = yield* SpeechService.SpeechService;
+    const recording = yield* speech
+      .startStream(DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS)
+      .pipe(Effect.forkChild);
+    yield* Deferred.await(helperStarted);
+    yield* TestClock.adjust("14999 millis");
+    expect(native.begin).not.toHaveBeenCalled();
+    yield* TestClock.adjust("1 millis");
+    yield* Deferred.await(helperInterrupted);
+    const stream = yield* Fiber.join(recording);
+    expect(native.begin).toHaveBeenLastCalledWith(expect.anything(), true, undefined);
+    yield* stream.feed(pcm());
+    expect(yield* stream.finish).toBe("hello");
+  }).pipe(Effect.scoped, Effect.provide(layer)),
+);
+
+it.effect("keeps the installed speech model usable when optional Silero acquisition fails", () =>
+  Effect.gen(function* () {
+    download
+      .mockImplementationOnce(() => Effect.succeed("test.gguf"))
+      .mockImplementationOnce(() =>
+        Effect.fail(
+          new PlatformError.PlatformError(
+            new PlatformError.SystemError({
+              _tag: "NotFound",
+              module: "FileSystem",
+              method: "open",
+            }),
+          ),
+        ),
+      );
+    const speech = yield* SpeechService.SpeechService;
+    const installed = yield* speech.downloadModel("test-model");
+    expect(installed.models.find((model) => model.id === "test-model")?.state).toBe("installed");
+    expect(yield* speech.transcribe(pcm(), DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS)).toBe("hello");
+  }).pipe(Effect.provide(layer)),
+);
+
 it.effect("honors exact aliases and phrase boundaries in batch and streamed text", () =>
   Effect.gen(function* () {
     const speech = yield* SpeechService.SpeechService;