Run server-engine file transcriptions through faster-whisper's batched pipeline with VAD

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server-py/sidecar.py

@@ -9,7 +9,8 @@ stdio — one JSON object per line, no framing beyond that:
 
   stdin  -> {"type": "transcribe", "id": "1", "model": "tiny.en",
              "audio_path": "/tmp/...", "language": "en" | null,
-             "task": "transcribe" | "translate"}
+             "task": "transcribe" | "translate",
+             "mode": "file" | "live"}                   # optional, default "file"
   stdout <- {"type": "segment", "id": "1", "start": 0.0, "end": 4.2,
              "text": " Hello world."}                       # 0..n of these
   stdout <- {"type": "done", "id": "1", "cancelled": false}
@@ -24,6 +25,17 @@ container decoding happens here. Requests are processed strictly one at a
 time; a reader thread keeps consuming stdin meanwhile so `cancel` can
 interrupt the segment loop of the request in flight.
 
+Decoding strategy by mode (Decisions.md D-WS5a):
+  "file"  faster-whisper's BatchedInferencePipeline with its Silero VAD
+          (vad_filter=True): silence is dropped, speech is cut into <= 30 s
+          chunks and BATCH_SIZE chunks decode per forward pass. Segments come
+          out in bursts (one burst per batch), so a cancel takes effect at the
+          next burst boundary; the Bun side's heartbeat (D-021) covers the
+          longer quiet stretch before the first burst.
+  "live"  the plain sequential WhisperModel.transcribe: server-live passes
+          are short (<= 29 s, often a few seconds), so batching has nothing to
+          batch and VAD only adds latency and can clip a short utterance.
+
 The last-used model stays loaded between requests (WorkBreakdown M5 warm-up:
 never reload weights per request) and is replaced — not accumulated — when a
 request names a different model, bounding memory to one model.
@@ -47,6 +59,7 @@ import sys
 import threading
 
 SAMPLE_RATE = 16000
+BATCH_SIZE = 8  # D-WS5a: VAD chunks decoded per batched forward pass
 
 
 def emit(message):
@@ -101,6 +114,7 @@ def main():
 
     cached_name = None
     cached_model = None
+    cached_pipeline = None  # BatchedInferencePipeline wrapping cached_model
 
     emit({"type": "ready", "version": getattr(faster_whisper, "__version__", "?")})
 
@@ -115,26 +129,41 @@ def main():
             audio_path = request["audio_path"]
             language = request.get("language") or None
             task = request.get("task") or "transcribe"
+            mode = request.get("mode") or "file"
 
             if model_name != cached_name:
-                cached_model = None  # release before loading the replacement
+                # Release both before loading the replacement: one model resident.
+                cached_model = cached_pipeline = None
+                cached_name = None
                 cached_model = faster_whisper.WhisperModel(
                     model_name,
                     device=device,
                     compute_type=compute_type,
                     download_root=download_root,
                 )
+                # A thin wrapper (no weights of its own) — shares cached_model.
+                cached_pipeline = faster_whisper.BatchedInferencePipeline(cached_model)
                 cached_name = model_name
 
             audio = np.fromfile(audio_path, dtype=np.float32)
-            segments, _info = cached_model.transcribe(
-                audio, language=language, task=task
-            )
+            if mode == "live":
+                segments, _info = cached_model.transcribe(
+                    audio, language=language, task=task
+                )
+            else:
+                segments, _info = cached_pipeline.transcribe(
+                    audio,
+                    language=language,
+                    task=task,
+                    vad_filter=True,
+                    batch_size=BATCH_SIZE,
+                )
 
             was_cancelled = False
             # `segments` is a generator: inference happens during iteration,
-            # so each segment can stream out (and cancel can cut in) while the
-            # next one is still being decoded.
+            # so segments stream out (and cancel can cut in) while later ones
+            # are still being decoded — per segment in live mode, per batch of
+            # BATCH_SIZE speech chunks in file mode (D-WS5a).
             for segment in segments:
                 if request_id in cancelled:
                     was_cancelled = True

src/server/transcribe.ts

@@ -47,8 +47,15 @@ export interface SidecarTranscribeRequest {
   audioPath: string;
   language?: string;
   task?: WhisperTask;
+  /**
+   * "file" (default): whole uploads — batched inference over VAD speech
+   * chunks. "live": short server-live passes — sequential, no VAD (D-WS5a).
+   */
+  mode?: SidecarMode;
 }
 
+export type SidecarMode = "file" | "live";
+
 export interface SidecarStatus {
   available: boolean;
   reason?: string;
@@ -265,6 +272,7 @@ export function createSidecar(command?: readonly string[]): Sidecar {
             audio_path: request.audioPath,
             language: request.language ?? null,
             task: request.task ?? "transcribe",
+            mode: request.mode ?? "file",
           });
         } catch (error) {
           pending.delete(id);

tests/fixtures/stub_fw/faster_whisper/__init__.py

@@ -0,0 +1,61 @@
+"""faster_whisper stand-in that reports how server-py/sidecar.py called it.
+
+Lets tests/transcribe.test.ts check the real sidecar's request routing
+(D-WS5a: file → BatchedInferencePipeline + VAD, live → sequential) without
+CTranslate2 or model weights. Every segment's text describes the call:
+
+  " seq model=<name> loads=<n> <sorted kwargs>"      WhisperModel.transcribe
+  " batched model=<name> loads=<n> <sorted kwargs>"  BatchedInferencePipeline
+
+Model names with special behaviour:
+  "slow"     20 segments, 0.1 s apart (cancel lands between them)
+  "silence"  no segments at all (VAD found no speech)
+"""
+
+import time
+
+__version__ = "stub"
+
+_loads = {"count": 0}
+
+
+class _Segment:
+    def __init__(self, start, end, text):
+        self.start = start
+        self.end = end
+        self.text = text
+
+
+def _describe(kind, model, kwargs):
+    options = " ".join(
+        "%s=%s" % (key, kwargs[key]) for key in sorted(kwargs) if kwargs[key] is not None
+    )
+    return " %s model=%s loads=%d %s" % (kind, model.name, _loads["count"], options)
+
+
+def _segments(kind, model, kwargs):
+    if model.name == "silence":
+        return
+    if model.name == "slow":
+        for index in range(20):
+            yield _Segment(float(index), float(index + 1), " slow %d" % index)
+            time.sleep(0.1)
+        return
+    yield _Segment(0.0, 1.0, _describe(kind, model, kwargs))
+
+
+class WhisperModel:
+    def __init__(self, name, device=None, compute_type=None, download_root=None):
+        _loads["count"] += 1
+        self.name = name
+
+    def transcribe(self, audio, **kwargs):
+        return _segments("seq", self, kwargs), None
+
+
+class BatchedInferencePipeline:
+    def __init__(self, model):
+        self.model = model
+
+    def transcribe(self, audio, **kwargs):
+        return _segments("batched", self.model, kwargs), None

tests/fixtures/stub_fw/numpy/__init__.py

@@ -0,0 +1,13 @@
+"""Minimal numpy stand-in for running server-py/sidecar.py without numpy.
+
+Put this directory on PYTHONPATH (with `python3 -S` so the real site-packages
+stay out). The sidecar only calls `np.fromfile(path, dtype=np.float32)`; the
+stub returns the raw bytes so the faster_whisper stub can report their size.
+"""
+
+float32 = "float32"
+
+
+def fromfile(path, dtype=None):
+    with open(path, "rb") as handle:
+        return handle.read()

tests/transcribe.test.ts

@@ -455,6 +455,88 @@ describe("createSidecar (degradation, D-013)", () => {
   });
 });
 
+// ---------------------------------------------------------------------------
+// The real server-py/sidecar.py against a stubbed faster_whisper/numpy
+// (tests/fixtures/stub_fw): request routing, model cache, cancel (D-WS5a).
+// `-S` keeps real site-packages out, so this runs the same everywhere.
+// ---------------------------------------------------------------------------
+
+describe("server-py/sidecar.py (stubbed faster-whisper, D-WS5a)", () => {
+  const STUB_DIR = join(import.meta.dir, "fixtures/stub_fw");
+  const stubbedSidecar = () =>
+    createSidecar(["env", `PYTHONPATH=${STUB_DIR}`, "python3", "-S", REAL_SIDECAR]);
+
+  async function run(
+    sidecar: Sidecar,
+    request: Omit<SidecarTranscribeRequest, "audioPath">,
+    signal?: AbortSignal,
+    onSegment: (segment: SidecarSegment) => void = () => {},
+  ): Promise<{ texts: string[]; cancelled: boolean }> {
+    const audioPath = join(await makeTempDir(), "audio.f32");
+    await Bun.write(audioPath, pcmBody([0, 0.5, -0.5, 0]));
+    const texts: string[] = [];
+    const outcome = await sidecar.transcribe(
+      { ...request, audioPath },
+      (segment) => {
+        texts.push(segment.text);
+        onSegment(segment);
+      },
+      signal,
+    );
+    return { texts, cancelled: outcome.cancelled };
+  }
+
+  test("file mode (and no mode) runs the batched pipeline with VAD, batch_size=8", async () => {
+    const sidecar = stubbedSidecar();
+    const explicit = await run(sidecar, { model: "m1", mode: "file", language: "da" });
+    expect(explicit.texts).toHaveLength(1);
+    expect(explicit.texts[0]).toContain("batched model=m1");
+    expect(explicit.texts[0]).toContain("vad_filter=True");
+    expect(explicit.texts[0]).toContain("batch_size=8");
+    expect(explicit.texts[0]).toContain("language=da");
+    expect(explicit.texts[0]).toContain("task=transcribe");
+
+    const implicit = await run(sidecar, { model: "m1" });
+    expect(implicit.texts[0]).toContain("batched model=m1");
+  });
+
+  test("live mode stays sequential without VAD (short passes, D-WS5a)", async () => {
+    const sidecar = stubbedSidecar();
+    const { texts } = await run(sidecar, { model: "m1", mode: "live", task: "translate" });
+    expect(texts[0]).toContain("seq model=m1");
+    expect(texts[0]).not.toContain("vad_filter");
+    expect(texts[0]).not.toContain("batch_size");
+    expect(texts[0]).toContain("task=translate");
+  });
+
+  test("one model stays loaded across modes; a switch reloads and rewraps", async () => {
+    const sidecar = stubbedSidecar();
+    expect((await run(sidecar, { model: "m1" })).texts[0]).toContain("loads=1");
+    expect((await run(sidecar, { model: "m1", mode: "live" })).texts[0]).toContain("loads=1");
+    const switched = (await run(sidecar, { model: "m2" })).texts[0];
+    expect(switched).toContain("batched model=m2");
+    expect(switched).toContain("loads=2");
+  });
+
+  test("cancel lands between batched segments", async () => {
+    const sidecar = stubbedSidecar();
+    const controller = new AbortController();
+    const { texts, cancelled } = await run(sidecar, { model: "slow" }, controller.signal, () =>
+      controller.abort(),
+    );
+    expect(cancelled).toBe(true);
+    expect(texts.length).toBeGreaterThanOrEqual(1);
+    expect(texts.length).toBeLessThan(20);
+  });
+
+  test("audio with no speech finishes with zero segments, not an error", async () => {
+    const sidecar = stubbedSidecar();
+    const { texts, cancelled } = await run(sidecar, { model: "silence" });
+    expect(texts).toEqual([]);
+    expect(cancelled).toBe(false);
+  });
+});
+
 // ---------------------------------------------------------------------------
 // Opt-in: real faster-whisper end to end (needs python3 + requirements.txt
 // installed and network/cached weights). See tests/server-engine.md.