Add live transcription via remote server.

2026-07-14 20:21:48 +02:00
parent 695b97be76
commit 42fe27d2dd
9 changed files with 227 additions and 21 deletions
@@ -266,3 +266,9 @@ a few seconds.
 by aborting the `/api/transcribe` fetch instead (`AbortController`; the server
 cancels the sidecar pass between segments, D-011/D-013) — the inference worker
 and its loaded pipeline are left untouched in that path.
+
+## D-019 — Server-based live microphone transcription
+**Status:** Accepted
+**Why:** M7 future want. Enables logged-in users on weak devices to transcribe live microphone audio using the server engine.
+**Shape:** The client opens a WebSocket connection to `/api/transcribe/live`. The client streams 16kHz Float32 PCM chunks (from `MicCapture`) over the WebSocket. The server runs `LiveTranscriber` and `LiveSegmenter` (from `src/lib/live.ts`, which is pure and runs perfectly in Node/Bun). For each segment/pass, the server's `LiveTranscriber` writes the audio to a temporary file and invokes the Python sidecar. The server sends `LiveUpdate` objects back to the client over the WebSocket.
+**Consequence:** The client does zero ML and minimal buffering, acting only as a mic-streamer and UI-renderer for live passes on the server.
@@ -58,9 +58,10 @@ export function App() {
  // Live-mode session (M7, D-016). The mic and transcriber live outside React
  // state; only display text goes through setState. Segment passes are
  // correlated with worker responses via requestId.
-  const liveRef = useRef<{ mic: MicCapture; transcriber: LiveTranscriber } | null>(
+  const liveRef = useRef<{ mic: MicCapture; transcriber: LiveTranscriber | null } | null>(
    null,
  );
+  const liveWsRef = useRef<WebSocket | null>(null);
  const pendingLive = useRef(
    new Map<
      string,
@@ -143,6 +144,8 @@ export function App() {
    return () => {
      void liveRef.current?.mic.stop().catch(() => {});
      liveRef.current = null;
+      liveWsRef.current?.close();
+      liveWsRef.current = null;
      workerRef.current?.terminate();
      workerRef.current = null;
      browserEngineRef.current = null;
@@ -465,6 +468,74 @@ export function App() {
    setLiveTentative("");
    setLiveStatus("starting");

+    const useServer =
+      engineChoice === "server" &&
+      auth.user !== null &&
+      serverStatus?.available === true;
+
+    const options = (getModel(modelId)?.multilingual ?? false) ? { language, task } : {};
+
+    if (useServer) {
+      const wsUrl = new URL(`/api/transcribe/live?model=${modelId}`, window.location.href);
+      wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:";
+      if (options.language) wsUrl.searchParams.set("language", options.language);
+      if (options.task) wsUrl.searchParams.set("task", options.task);
+
+      const ws = new WebSocket(wsUrl);
+      
+      ws.onmessage = (event) => {
+        const data = JSON.parse(event.data);
+        if (data.type === "update") {
+          setLiveCommitted(data.update.committedText);
+          setLiveTentative(data.update.tentativeText);
+          if (data.update.error) setError(data.update.error);
+        } else if (data.type === "result") {
+          setResult(data.result);
+          setLiveCommitted("");
+          setLiveTentative("");
+          setLiveStatus("idle");
+          liveWsRef.current = null;
+          liveRef.current = null;
+        } else if (data.type === "error") {
+          setError(data.error);
+          setLiveStatus("idle");
+          liveWsRef.current = null;
+          liveRef.current = null;
+        }
+      };
+
+      ws.onerror = () => {
+        setError("WebSocket connection error.");
+        setLiveStatus("idle");
+        liveWsRef.current = null;
+        liveRef.current = null;
+      };
+
+      ws.onopen = async () => {
+        try {
+          const mic = await startMicCapture((chunk) => {
+            if (ws.readyState === WebSocket.OPEN) {
+              ws.send(chunk.buffer);
+            }
+          });
+          liveWsRef.current = ws;
+          liveRef.current = { mic, transcriber: null };
+          setLiveStatus("listening");
+        } catch (micError) {
+          ws.close();
+          setLiveStatus("idle");
+          setError(
+            micError instanceof DOMException && micError.name === "NotAllowedError"
+              ? "Microphone access was denied. Allow it in your browser's site settings and try again."
+              : micError instanceof Error
+                ? micError.message
+                : String(micError),
+          );
+        }
+      };
+      return;
+    }
+
    // Idempotent model load, like the file flow — a pending download runs
    // while the mic opens; segment passes queue in the worker until ready.
    const worker = ensureWorker();
@@ -474,9 +545,6 @@ export function App() {
      postToWorker(worker, { type: "load", modelId, device: devicePreference });
    }

-    // Language/task are fixed for the session; the options are disabled while
-    // live. `.en` models take neither (same rule as the file flow).
-    const options = (getModel(modelId)?.multilingual ?? false) ? { language, task } : {};
    const transcriber = new LiveTranscriber({
      ...LIVE_TUNING,
      transcribe: (audio) => transcribeLiveSegment(worker, audio, options),
@@ -509,14 +577,19 @@ export function App() {
    setLiveStatus("stopping");
    // Stop feeding audio first, then drain the remaining segment passes.
    await session.mic.stop();
-    try {
-      const final = await session.transcriber.stop();
-      setResult(final);
-      setLiveCommitted("");
-      setLiveTentative("");
-    } finally {
-      liveRef.current = null;
-      setLiveStatus("idle");
+
+    if (liveWsRef.current) {
+      liveWsRef.current.send(JSON.stringify({ type: "stop" }));
+    } else {
+      try {
+        const final = await session.transcriber!.stop();
+        setResult(final);
+        setLiveCommitted("");
+        setLiveTentative("");
+      } finally {
+        liveRef.current = null;
+        setLiveStatus("idle");
+      }
    }
  }

@@ -667,7 +740,7 @@ export function App() {
        <p>
          Model: <code>{modelId}</code> · Engine:{" "}
          {serverSelected
-            ? "server (faster-whisper) — files upload to this deployment"
+            ? "server (faster-whisper) — audio streams to this deployment"
            : activeDevice === "webgpu"
              ? "WebGPU (on-device)"
              : activeDevice === "webnn"
@@ -16,7 +16,7 @@ interface EngineToggleProps {
 * Architecture.md §5). The server option stays visible but disabled when the
 * deployment reports the engine unavailable (e.g. no faster-whisper sidecar,
 * D-013), showing the probe's reason — hiding it entirely would read as a
- * broken account feature. Live mode always stays on-device (D-016).
+ * broken account feature. Live mode streams to the server if selected (D-019).
 */
 export function EngineToggle({
  choice,
@@ -53,7 +53,7 @@ export function EngineToggle({
          ? "Checking server availability…"
          : serverAvailable
            ? effective === "server"
-              ? "Uploaded files are transcribed on the server (faster-whisper). Live mode always stays on this device."
+              ? "Audio files and live microphone stream are transcribed on the server (faster-whisper)."
              : "Your audio stays on this device."
            : (status.reason ?? "The server engine is unavailable.")}
      </span>
@@ -5,6 +5,7 @@ import { injectPrecacheManifest } from "./pwa/precache.ts";
 import { auth } from "./server/auth.ts";
 import { AUTH_ENABLED } from "./server/flags.ts";
 import { transcribe } from "./server/transcribe.ts";
+import { liveWebSocket } from "./server/live.ts";

 const ROOT = join(import.meta.dir, "..");
 const IS_PRODUCTION = process.env.NODE_ENV === "production";
@@ -56,7 +57,7 @@ async function serveOrtFile(dir: string, name: string): Promise<Response> {
 // Auth endpoints are M4 (src/server/auth.ts); /api/transcribe joins in M5.
 // ---------------------------------------------------------------------------

-async function handleApiRequest(request: Request): Promise<Response> {
+async function handleApiRequest(request: Request, server: any): Promise<Response | undefined> {
  const { pathname } = new URL(request.url);
  const apiIndex = pathname.indexOf("/api/");
  const apiPath = apiIndex !== -1 ? pathname.slice(apiIndex) : pathname;
@@ -79,11 +80,31 @@ async function handleApiRequest(request: Request): Promise<Response> {
      return auth.logout(request);
    case "GET /api/auth/me":
      return auth.me(request);
-    // Server engine (M5, D-011/D-013): availability probe + streamed run.
    case "GET /api/transcribe/status":
      return transcribe.status(request);
    case "POST /api/transcribe":
      return transcribe.post(request);
+    case "GET /api/transcribe/live": {
+      const user = auth.getSessionUser(request);
+      if (!user) return new Response("Unauthorized", { status: 401 });
+      const url = new URL(request.url);
+      const modelId = url.searchParams.get("model");
+      if (!modelId) return new Response("Missing model", { status: 400 });
+      const fasterWhisperModel = transcribe.getFasterWhisperModel?.(modelId) || (await import("./lib/models.ts")).getFasterWhisperModel(modelId);
+      if (!fasterWhisperModel) return new Response("Unknown model", { status: 400 });
+      
+      const success = server.upgrade(request, {
+        data: {
+          user,
+          model: fasterWhisperModel,
+          language: url.searchParams.get("language") || undefined,
+          task: url.searchParams.get("task") || undefined,
+          tempDir: require("node:os").tmpdir()
+        }
+      });
+      if (success) return undefined;
+      return new Response("WebSocket upgrade failed", { status: 500 });
+    }
    default:
      return NOT_FOUND();
  }
@@ -103,10 +124,14 @@ async function productionServeOptions(): Promise<ServeOptions> {
  }
  return {
    development: false,
-    async fetch(request) {
+    async fetch(request, server) {
      const url = new URL(request.url);
      // API routes are dynamic (M4/M5) and network-only in the SW (D-007).
-      if (url.pathname.includes("/api/")) return handleApiRequest(request);
+      if (url.pathname.includes("/api/")) {
+        const response = await handleApiRequest(request, server);
+        if (response) return response;
+        return new Response("Upgrading...", { status: 101 }); // actually Bun handles the real response, but TypeScript requires a return
+      }

      let relative = url.pathname === "/" ? "index.html" : decodeURIComponent(url.pathname.slice(1));
      if (relative.endsWith("/")) relative += "index.html";
@@ -190,7 +215,11 @@ async function developmentServeOptions(): Promise<ServeOptions> {
      const url = new URL(request.url);
      const pathname = url.pathname;

-      if (pathname.includes("/api/")) return handleApiRequest(request);
+      if (pathname.includes("/api/")) {
+        const response = await handleApiRequest(request, server);
+        if (response) return response;
+        return new Response("Upgrading...", { status: 101 });
+      }

      if (pathname.endsWith("/inference.worker.js")) {
        return new Response(workerScript, { headers: scriptHeaders });
@@ -232,6 +261,7 @@ async function developmentServeOptions(): Promise<ServeOptions> {

 const options = IS_PRODUCTION ? await productionServeOptions() : await developmentServeOptions();
 options.port = process.env.PORT || 3000;
+options.websocket = liveWebSocket as any;

 const server = serve(options);

@@ -0,0 +1,95 @@
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import type { ServerWebSocket } from "bun";
+import { LIVE_TUNING, LiveTranscriber, type LiveUpdate } from "../lib/live";
+import { sidecar } from "./transcribe";
+import { getSessionUser } from "./auth";
+import { getFasterWhisperModel } from "../lib/models";
+import type { WhisperTask } from "../worker/protocol";
+
+interface LiveSessionData {
+  user: any;
+  model: string;
+  language?: string;
+  task?: WhisperTask;
+  tempDir: string;
+}
+
+export const liveWebSocket = {
+  open(ws: ServerWebSocket<LiveSessionData>) {
+    // Initialize the transcriber for this connection.
+    const transcriber = new LiveTranscriber({
+      ...LIVE_TUNING,
+      transcribe: async (audio: Float32Array) => {
+        const audioPath = join(
+          ws.data.tempDir,
+          `whisper-live-${crypto.randomUUID()}.f32`
+        );
+        const sink = Bun.file(audioPath).writer();
+        // audio.buffer is an ArrayBuffer, we need a Uint8Array view
+        sink.write(new Uint8Array(audio.buffer as ArrayBuffer, audio.byteOffset, audio.byteLength));
+        await sink.end();
+
+        return new Promise((resolve, reject) => {
+          let text = "";
+          let chunks: any[] = [];
+          
+          sidecar.transcribe(
+            {
+              model: ws.data.model,
+              audioPath,
+              language: ws.data.language,
+              task: ws.data.task,
+            },
+            (segment) => {
+              text += segment.text;
+              chunks.push({
+                text: segment.text,
+                timestamp: [segment.start, segment.end],
+              });
+            }
+          )
+          .then(() => resolve({ text: text.trim(), chunks }))
+          .catch(reject)
+          .finally(() => Bun.file(audioPath).unlink().catch(() => {}));
+        });
+      },
+      onUpdate: (update: LiveUpdate) => {
+        ws.send(JSON.stringify({ type: "update", update }));
+      },
+    });
+    
+    // Store it on the websocket
+    (ws as any).transcriber = transcriber;
+  },
+
+  message(ws: ServerWebSocket<LiveSessionData>, message: string | Buffer | Uint8Array) {
+    if (typeof message === "string") {
+      const parsed = JSON.parse(message);
+      if (parsed.type === "stop") {
+        const transcriber = (ws as any).transcriber as LiveTranscriber;
+        transcriber.stop().then((result) => {
+          ws.send(JSON.stringify({ type: "result", result }));
+          ws.close();
+        }).catch((err) => {
+          ws.send(JSON.stringify({ type: "error", error: err instanceof Error ? err.message : String(err) }));
+          ws.close();
+        });
+      }
+    } else {
+      // Audio chunk arrived
+      const transcriber = (ws as any).transcriber as LiveTranscriber;
+      // Convert raw bytes to Float32Array (16kHz mono)
+      const buffer = message instanceof Buffer ? message.buffer : message.buffer;
+      const byteOffset = message instanceof Buffer ? message.byteOffset : (message as Uint8Array).byteOffset;
+      const byteLength = message.byteLength;
+      const audio = new Float32Array(buffer, byteOffset, byteLength / 4);
+      transcriber.push(audio);
+    }
+  },
+
+  close(ws: ServerWebSocket<LiveSessionData>) {
+    const transcriber = (ws as any).transcriber as LiveTranscriber;
+    transcriber?.stop().catch(() => {});
+  },
+};
@@ -470,8 +470,10 @@ export function createTranscribeHandlers(deps: TranscribeDeps): TranscribeHandle
 // Default instance (real sidecar + app sessions), wired in src/index.ts
 // ---------------------------------------------------------------------------

+export const sidecar = createSidecar();
+
 export const transcribe: TranscribeHandlers = createTranscribeHandlers({
-  sidecar: createSidecar(),
+  sidecar,
  getSessionUser: appGetSessionUser,
  maxUploadBytes: Number(process.env.TRANSCRIBE_MAX_UPLOAD_BYTES) || undefined,
 });