fix(speech): dictation no longer loses the recording when the GPU fails

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

@@ -763,24 +763,36 @@ it.effect(
     }),
 );
 
-it.effect("reports a failure on a GPU the person chose instead of switching to CPU", () =>
-  Effect.gen(function* () {
-    const failure = new Error("GPU failed");
-    native.transcribe.mockRejectedValueOnce(failure);
-    const speech = yield* SpeechService.SpeechService;
-    const acceleration = 'gpu:["vulkan","gpu-1"]';
-    yield* speech.updateAcceleration(acceleration);
-    const result = yield* Effect.result(
-      speech.transcribe(pcm(), DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS),
-    );
-    expect(Result.isFailure(result) && result.failure).toMatchObject({
-      _tag: "SpeechOperationError",
-      operation: "inference",
-      cause: failure,
-    });
-    expect(loadNative).toHaveBeenCalledTimes(1);
-    expect(loadNative.mock.calls[0]?.[3]).toBe(acceleration);
-  }).pipe(Effect.provide(layer)),
+it.effect.each([
+  ['gpu:["vulkan","gpu-1"]', "cpu"],
+  ["cpu", "auto"],
+] as const)(
+  "retries a recording from a chosen %s on %s only when fallback is on",
+  ([acceleration, other]) =>
+    Effect.gen(function* () {
+      const speech = yield* SpeechService.SpeechService;
+      for (const failing of ["model preparation", "inference"] as const)
+        for (const fallback of [false, true]) {
+          loadNative.mockClear();
+          const failure = new Error("device failed");
+          if (failing === "inference") native.transcribe.mockRejectedValueOnce(failure);
+          else loadNative.mockRejectedValueOnce(failure);
+          yield* speech.updateAcceleration(acceleration, fallback);
+          const result = yield* Effect.result(
+            speech.transcribe(pcm(), DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS),
+          );
+          if (fallback) expect(Result.isSuccess(result) && result.success).toBe("hello");
+          else
+            expect(Result.isFailure(result) && result.failure).toMatchObject({
+              _tag: "SpeechOperationError",
+              operation: failing,
+              cause: failure,
+            });
+          expect(loadNative.mock.calls.map((call) => call[3])).toEqual(
+            fallback ? [acceleration, other] : [acceleration],
+          );
+        }
+    }).pipe(Effect.provide(layer)),
 );
 
 it.effect("replays streamed audio on CPU after an Auto-placed feed crashes, once", () =>
@@ -895,9 +907,15 @@ it.effect.each([false, true])(
     }).pipe(Effect.provide(layer)),
 );
 
-it.effect.each(["auto", 'gpu:["vulkan","gpu-1"]', "cpu"] as const)(
-  "reports finalization failure without repeated or explicit-backend fallback: %s",
-  (acceleration) =>
+it.effect.each([
+  ["auto", false],
+  ['gpu:["vulkan","gpu-1"]', false],
+  ["cpu", false],
+  ['gpu:["vulkan","gpu-1"]', true],
+  ["cpu", true],
+] as const)(
+  "reports finalization failure after at most one fallback: %s, fallback %s",
+  ([acceleration, fallback]) =>
     Effect.gen(function* () {
       const failure = new Error("finish failed");
       const worker = {
@@ -907,15 +925,24 @@ it.effect.each(["auto", 'gpu:["vulkan","gpu-1"]', "cpu"] as const)(
           throw failure;
         }),
       };
-      const cpu = { ...worker, backend: "CPU", dispose: vi.fn() };
+      const other = {
+        ...worker,
+        backend: acceleration === "cpu" ? "Vulkan0" : "CPU",
+        dispose: vi.fn(),
+      };
+      const retries = acceleration === "auto" || fallback;
       loadNative.mockResolvedValueOnce(worker);
-      if (acceleration === "auto") loadNative.mockResolvedValueOnce(cpu);
+      if (retries) loadNative.mockResolvedValueOnce(other);
       const speech = yield* SpeechService.SpeechService;
-      yield* speech.updateAcceleration(acceleration);
+      yield* speech.updateAcceleration(acceleration, fallback);
       const stream = yield* speech.startStream(DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS);
       expect(Result.isFailure(yield* Effect.result(stream.finish))).toBe(true);
-      expect(worker.finish).toHaveBeenCalledTimes(acceleration === "auto" ? 2 : 1);
-      expect(loadNative).toHaveBeenCalledTimes(acceleration === "auto" ? 2 : 1);
+      // The failed stream frees the model before its socket closes, so a retry is not busy.
+      expect(yield* speech.status).toMatchObject({ state: "ready" });
+      expect(worker.finish).toHaveBeenCalledTimes(retries ? 2 : 1);
+      expect(loadNative.mock.calls.map((call) => call[3])).toEqual(
+        retries ? [acceleration, acceleration === "cpu" ? "auto" : "cpu"] : [acceleration],
+      );
     }).pipe(Effect.scoped, Effect.provide(layer)),
 );
 

apps/server/src/speech/SpeechService.ts

@@ -150,6 +150,7 @@ export class SpeechService extends Context.Service<
     ) => Effect.Effect<SpeechStream, SpeechError, Scope.Scope>;
     readonly updateAcceleration: (
       acceleration: SpeechAcceleration,
+      fallback?: boolean,
     ) => Effect.Effect<EnvironmentSpeechStatus, SpeechError>;
     readonly updateModelUnloadTimeout: (
       timeout: SpeechModelUnloadTimeout,
@@ -238,9 +239,16 @@ const make = Effect.gen(function* () {
   };
 
   // Auto placement may land on a GPU the native engine cannot actually run; one retry on
-  // this host's CPU recovers it. A GPU the person chose, and CPU itself, report the error.
-  const canRetryOnCpu = (acceleration: SpeechAcceleration, loaded: LoadedModel) =>
-    acceleration === "auto" && loaded.backend.toLowerCase() !== "cpu";
+  // this host's CPU recovers it. A device the person chose retries on the other only when allowed.
+  const fallbackFor = (
+    settings: { speechAcceleration: SpeechAcceleration; speechAccelerationFallback: boolean },
+    loaded: LoadedModel,
+  ) => {
+    const onCpu = loaded.backend.toLowerCase() === "cpu";
+    if (settings.speechAcceleration === "auto") return onCpu ? undefined : "cpu";
+    if (!settings.speechAccelerationFallback) return undefined;
+    return onCpu ? "auto" : "cpu";
+  };
 
   yield* Effect.addFinalizer(() =>
     Effect.promise(async () => {
@@ -397,6 +405,28 @@ const make = Effect.gen(function* () {
     }
   };
 
+  // The loader already retries "auto" on the CPU; a chosen device needs the person's fallback.
+  const loadSelected = async (
+    modelPath: string,
+    definition: SpeechModel,
+    signal: AbortSignal,
+    settings: { speechAcceleration: SpeechAcceleration; speechAccelerationFallback: boolean },
+  ) => {
+    try {
+      return await loadModel(modelPath, definition, signal, settings.speechAcceleration);
+    } catch (cause) {
+      const { speechAcceleration, speechAccelerationFallback } = settings;
+      if (speechAcceleration === "auto" || !speechAccelerationFallback) throw cause;
+      if (signal.aborted || isSpeechError(cause)) throw cause;
+      return await loadModel(
+        modelPath,
+        definition,
+        signal,
+        speechAcceleration === "cpu" ? "auto" : "cpu",
+      );
+    }
+  };
+
   const speechError = (operation: string, cause: unknown): SpeechError =>
     isSpeechError(cause) ? cause : new SpeechOperationError({ operation, cause });
 
@@ -474,6 +504,7 @@ const make = Effect.gen(function* () {
       languages: definition.languages,
       supportsLanguageDetection: definition.supportsLanguageDetection,
       acceleration: settings.speechAcceleration,
+      accelerationFallback: settings.speechAccelerationFallback,
       benchmarkTier:
         settings.speechAcceleration === "cpu" ||
         (settings.speechAcceleration === "auto" && autoUsesCpu)
@@ -560,11 +591,11 @@ const make = Effect.gen(function* () {
           Effect.andThen(installedModelPath(selectedModel(settings))),
           Effect.flatMap((modelPath) =>
             attempt("model preparation", (signal) =>
-              loadModel(
+              loadSelected(
                 modelPath,
                 selectedModel(settings),
                 AbortSignal.any([signal, lifetime.signal]),
-                settings.speechAcceleration,
+                settings,
               ),
             ).pipe(
               Effect.timeout("2 minutes"),
@@ -611,11 +642,11 @@ const make = Effect.gen(function* () {
             const modelPath = yield* installedModelPath(definition);
             const loaded = yield* Effect.tryPromise({
               try: (signal) =>
-                loadModel(
+                loadSelected(
                   modelPath,
                   definition,
                   AbortSignal.any([signal, lifetime.signal]),
-                  settings.speechAcceleration,
+                  settings,
                 ),
               catch: (cause) => speechError("model preparation", cause),
             });
@@ -653,21 +684,22 @@ const make = Effect.gen(function* () {
             let inference = loaded;
             const result = yield* transcribeWith(loaded).pipe(
               Effect.catch((error) => {
-                if (!canRetryOnCpu(settings.speechAcceleration, loaded)) return Effect.fail(error);
+                const fallback = fallbackFor(settings, loaded);
+                if (!fallback) return Effect.fail(error);
                 return Effect.tryPromise({
                   try: (signal) =>
                     loadModel(
                       modelPath,
                       definition,
                       AbortSignal.any([signal, lifetime.signal]),
-                      "cpu",
+                      fallback,
                     ),
                   catch: (cause) => speechError("model preparation", cause),
                 }).pipe(
-                  Effect.flatMap((cpu) => {
-                    autoCpuModel = definition.id;
-                    inference = cpu;
-                    return transcribeWith(cpu);
+                  Effect.flatMap((other) => {
+                    if (settings.speechAcceleration === "auto") autoCpuModel = definition.id;
+                    inference = other;
+                    return transcribeWith(other);
                   }),
                 );
               }),
@@ -752,7 +784,7 @@ const make = Effect.gen(function* () {
     let correct = makeSpeechTextCorrector(customWords, dictionary);
     const modelPath = yield* installedModelPath(definition);
     let streamModel = yield* attempt(operation, () =>
-      loadModel(modelPath, definition, signal, settings.speechAcceleration),
+      loadSelected(modelPath, definition, signal, settings),
     );
     loaded = streamModel;
     const vadPath = yield* silenceModelPath(streamModel, options);
@@ -781,19 +813,19 @@ const make = Effect.gen(function* () {
       correct = makeSpeechTextCorrector(vocabulary ? [] : customWords, dictionary);
       started = true;
     };
-    // Audio already fed is kept so a CPU restart can replay it; the limit caps it at one recording.
+    // Audio already fed is kept so a fallback restart can replay it; the limit caps it at one recording.
     const received: Float32Array[] = [];
-    let retriedOnCpu = false;
-    const recoverOnCpu = async (cause: unknown) => {
+    let recovered = false;
+    const recover = async (cause: unknown) => {
       started = false;
-      if (closed || signal.aborted || retriedOnCpu) throw cause;
-      if (!canRetryOnCpu(settings.speechAcceleration, streamModel)) throw cause;
-      retriedOnCpu = true;
+      const fallback = fallbackFor(settings, streamModel);
+      if (closed || signal.aborted || recovered || !fallback) throw cause;
+      recovered = true;
       forgetModel(streamModel);
       await streamModel.dispose();
       if (closed || signal.aborted) throw cause;
-      streamModel = await loadModel(modelPath, definition, signal, "cpu");
-      autoCpuModel = definition.id;
+      streamModel = await loadModel(modelPath, definition, signal, fallback);
+      if (settings.speechAcceleration === "auto") autoCpuModel = definition.id;
       loaded = streamModel;
       await beginStream();
       let update: Awaited<ReturnType<LoadedModel["feed"]>> | undefined;
@@ -807,7 +839,7 @@ const make = Effect.gen(function* () {
       try {
         await beginStream();
       } catch (cause) {
-        await recoverOnCpu(cause);
+        await recover(cause);
       }
     });
     let byteLength = 0;
@@ -844,7 +876,7 @@ const make = Effect.gen(function* () {
           try {
             update = await streamModel.feed(pcm);
           } catch (cause) {
-            update = await recoverOnCpu(cause);
+            update = await recover(cause);
             revisionBase += lastNativeRevision;
             lastNativeRevision = 0;
           }
@@ -863,7 +895,7 @@ const make = Effect.gen(function* () {
         let text: string;
         try {
           text = await streamModel.finish().catch(async (cause: unknown) => {
-            await recoverOnCpu(cause);
+            await recover(cause);
             if (closed || signal.aborted) throw cause;
             return streamModel.finish();
           });
@@ -888,10 +920,14 @@ const make = Effect.gen(function* () {
     transcribe,
     startStream: Effect.fn("SpeechService.startStream")(function* (options) {
       const session = yield* Scope.fork(yield* Effect.scope);
-      return yield* openStream(options).pipe(
-        Scope.provide(session),
-        Effect.onError((cause) => Scope.close(session, Exit.failCause(cause))),
-      );
+      const closeOnError = <A, E>(effect: Effect.Effect<A, E>) =>
+        effect.pipe(Effect.onError((cause) => Scope.close(session, Exit.failCause(cause))));
+      const stream = yield* closeOnError(openStream(options).pipe(Scope.provide(session)));
+      // A failed stream frees the model before the client hears of it, so its retry is not busy.
+      return {
+        feed: (pcm: Uint8Array) => closeOnError(stream.feed(pcm)),
+        finish: closeOnError(stream.finish),
+      };
     }),
     downloadModel: Effect.fn("SpeechService.downloadModel")((modelId) =>
       requireSupported.pipe(
@@ -970,10 +1006,11 @@ const make = Effect.gen(function* () {
         yield* writeSettings("model linking", { speechModelId: first });
       return yield* modelsEffect;
     }),
-    updateAcceleration: Effect.fn("SpeechService.updateAcceleration")((acceleration) =>
-      writeSettings("acceleration update", { speechAcceleration: acceleration }).pipe(
-        Effect.andThen(freshStatus("acceleration update")),
-      ),
+    updateAcceleration: Effect.fn("SpeechService.updateAcceleration")((acceleration, fallback) =>
+      writeSettings("acceleration update", {
+        speechAcceleration: acceleration,
+        ...(fallback === undefined ? {} : { speechAccelerationFallback: fallback }),
+      }).pipe(Effect.andThen(freshStatus("acceleration update"))),
     ),
     updateModelUnloadTimeout: Effect.fn("SpeechService.updateModelUnloadTimeout")((timeout) =>
       writeSettings("model unload timeout update", { speechModelUnloadTimeout: timeout }).pipe(

apps/server/src/speech/http.ts

@@ -198,7 +198,9 @@ export const speechHttpApiLayer = HttpApiBuilder.group(
         "updateAcceleration",
         Effect.fn("environment.voice.updateAcceleration")(function* (args) {
           yield* guard(args.endpoint.name, AuthSettingsWriteScope);
-          return yield* speech.updateAcceleration(args.payload.acceleration).pipe(mapSpeechError);
+          return yield* speech
+            .updateAcceleration(args.payload.acceleration, args.payload.fallback)
+            .pipe(mapSpeechError);
         }),
       )
       .handle(

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

@@ -55,6 +55,36 @@ it("returns transcription from an isolated process and disposes it", async () =>
   }
 });
 
+it("splits a piece the engine could not finish instead of failing the recording", async () => {
+  for (const error of [
+    "OutputTruncated",
+    "OutputRepetition",
+    "InputTooLong",
+    "OutOfMemory",
+    "BackendError",
+  ]) {
+    const model = await loadNativeSpeechModel(
+      "unused.gguf",
+      new AbortController().signal,
+      fixture(`async (pcm) => {
+        class OutputTruncated extends Error {}
+        const errors = { OutputTruncated, OutputRepetition: class OutputRepetition extends OutputTruncated {},
+          InputTooLong: class InputTooLong extends Error {}, OutOfMemory: class OutOfMemory extends Error {},
+          BackendError: class BackendError extends Error {} };
+        if (pcm.length > 5000) throw new errors.${error}("too much");
+        return { text: String(pcm.length) };
+      }`),
+    );
+    try {
+      expect(
+        await model.transcribe(new Float32Array(16000), { timestamps: "none" }, false),
+      ).toEqual({ text: "4000 4000 4000 4000" });
+    } finally {
+      await model.dispose();
+    }
+  }
+});
+
 it("probes generic vocabulary support and delivers ordered terms to offline inference", async () => {
   for (const supported of [false, true]) {
     const model = await loadNativeSpeechModel(

apps/server/src/speech/native.ts

@@ -31,6 +31,13 @@ const newAudio = (filterSilence) => createSpeechAudio({
   frameSamples: 512, filterSilence, nativeStreaming: model.capabilities.supportsStreaming,
   longForm: model.supports("long_form"), maxAudioMs: model.capabilities.maxAudioMs,
 });
+// Matched by class name: OutputRepetition extends OutputTruncated, and smaller input needs less memory.
+const splittable = new Set(["OutputTruncated", "InputTooLong", "OutOfMemory", "BackendError"]);
+const canSplit = (error) => {
+  for (let proto = error && Object.getPrototypeOf(error); proto; proto = Object.getPrototypeOf(proto))
+    if (splittable.has(proto.constructor?.name)) return true;
+  return false;
+};
 const openAsr = async () => {
   if (nativeStreaming) {
     session = model.createSession();
@@ -71,8 +78,7 @@ const transcribe = async (pcm) => {
     const result = await model.transcribe(pcm, options);
     texts.push(result.text);
   } catch (error) {
-    if ((error?.constructor?.name !== "OutputTruncated" &&
-         error?.constructor?.name !== "InputTooLong") || pcm.length <= 4000) throw error;
+    if (!canSplit(error) || pcm.length <= 4000) throw error;
     const middle = Math.floor(pcm.length / 2);
     await transcribe(pcm.subarray(0, middle));
     await transcribe(pcm.subarray(middle));