Make local STT a streaming runtime service (#11)

* Separate STT runtime service from stdio transport

* Transcribe PCM chunks without temp WAV files

* Emit local STT runtime event frames

* Add streaming audio chunk command

* Keep STT session configuration in runtime state

* Buffer streaming STT audio in runtime sessions

* Stream local STT audio from Electron

* Emit final STT segment events on session flush

* Forward STT runtime events to capture listeners

* Persist live STT transcript segments

* Wrap whisper STT behind backend interface

* Queue streaming STT transcription jobs

* Report streaming STT progress by audio time

* Prepare STT response sink for async events

* Make streaming STT jobs self-contained

* Isolate streaming STT job execution

* Separate STT backend from runtime state

* Isolate STT request source

* Introduce STT worker boundary

* Route STT worker through owned commands

* Run STT backend on a worker thread

* Submit streaming STT jobs asynchronously

* Drain streaming STT events through runtime service

* Pump STT runtime events from stdio loop

* Drop stale STT job results after cancel

* Verify async STT job submission

* Keep Metal STT backend on runtime thread

* Add opt-in STT worker process boundary

* Enable STT worker process by default

* Warm local STT runtime on startup

* Emit final events from STT worker results

* Refresh STT warmup after model changes

* Ignore STT events outside active session

* Clamp PCM16 normalization range

* Scan WAV chunks without header cap

Author: qyinm

apps/desktop/main.cjs

@@ -52,6 +52,7 @@ const MAX_SECTION_INSTRUCTION_LENGTH = 1200;
 const AUTO_UPDATE_REPO = "qyinm/MirrorNote";
 const AUTO_UPDATE_INTERVAL = "30 minutes";
 const DEFAULT_AUDIO_DEVICE_ID = "default";
+const LOCAL_STT_STREAM_CHUNK_SECONDS = 5;
 
 const LOCAL_STT_MODELS = [
   {
@@ -1210,6 +1211,9 @@ async function downloadLocalSTTModel(modelID, targetWebContents = null) {
     await fs.rm(destinationPath, { force: true });
     await fs.rename(temporaryPath, destinationPath);
     await writeElectronSettings({ selectedLocalSTTModelID: model.id });
+    warmLocalSTTServer().catch((error) => {
+      console.error("Local STT warmup failed after model download:", error);
+    });
     updateTrayMenu();
     sendSTTModelDownloadProgress(targetWebContents, {
       modelID: model.id,
@@ -2387,10 +2391,23 @@ async function retryTranscript(id) {
     throw new Error("This note does not have a playable recording to transcribe.");
   }
 
-  const segments = await transcribeRecordingFile(paths.recordingPath, transcriptionConfiguration);
+  const segments = await transcribeRecordingFile(paths.recordingPath, transcriptionConfiguration, {
+    onSegments: async (liveSegments) => {
+      await writeLiveTranscript(paths, liveSegments);
+    },
+  });
   return saveTranscript(id, segments);
 }
 
+async function writeLiveTranscript(paths, segments) {
+  const normalizedSegments = normalizeTranscriptSegments(segments);
+  await writeFileAtomic(paths.transcriptPath, serializeTranscriptJSONL(normalizedSegments), "utf8");
+
+  const metadata = await readJSONIfExists(paths.metadataPath) || {};
+  metadata.transcriptSegmentCount = normalizedSegments.length;
+  await writeJSONFileAtomic(paths.metadataPath, metadata);
+}
+
 async function promptForFileUpload(targetWindow, title, filters) {
   const result = await dialog.showOpenDialog(targetWindow || undefined, {
     title,
@@ -3154,10 +3171,10 @@ async function fetchAudioDeviceInventory() {
   }
 }
 
-async function transcribeRecordingFile(recordingPath, transcriptionConfiguration) {
+async function transcribeRecordingFile(recordingPath, transcriptionConfiguration, options = {}) {
   const sttInputPath = await prepareLocalSTTAudioInput(recordingPath);
   try {
-    return await transcribeLocalSTTAudioFile(sttInputPath, transcriptionConfiguration);
+    return await transcribeLocalSTTAudioFile(sttInputPath, transcriptionConfiguration, options);
   } finally {
     if (sttInputPath !== recordingPath) {
       await fs.rm(path.dirname(sttInputPath), { recursive: true, force: true }).catch(() => {});
@@ -3205,7 +3222,231 @@ function runProcess(command, args) {
   });
 }
 
-async function transcribeLocalSTTAudioFile(recordingPath, transcriptionConfiguration) {
+async function transcribeLocalSTTAudioFile(recordingPath, transcriptionConfiguration, options = {}) {
+  const sessionID = randomUUID();
+  const streamedSegments = [];
+  const fallbackSegments = [];
+  let lastLiveSegmentCount = 0;
+  const unsubscribe = localSTTServer.onEvent((event) => {
+    if (event?.sessionID !== sessionID) {
+      return;
+    }
+    const captureEvent = captureEventFromLocalSTTEvent(event);
+    if (captureEvent) {
+      broadcastCaptureEvent(captureEvent);
+    }
+    if ((event?.eventType === "partial_segment" || event?.eventType === "final_segment") && event.segment) {
+      streamedSegments.push(event.segment);
+    }
+  });
+  const flushLiveSegments = async () => {
+    if (typeof options.onSegments !== "function" || streamedSegments.length === lastLiveSegmentCount) {
+      return;
+    }
+    lastLiveSegmentCount = streamedSegments.length;
+    await options.onSegments(normalizeTranscriptSegments(streamedSegments));
+  };
+
+  try {
+    await sendLocalSTTServerRequest({
+      version: 4,
+      id: randomUUID(),
+      command: "start_session",
+      sessionID,
+      configuration: transcriptionConfiguration,
+    });
+
+    for await (const chunk of readLocalSTTPcmChunks(recordingPath, LOCAL_STT_STREAM_CHUNK_SECONDS)) {
+      const response = await sendLocalSTTServerRequest({
+        version: 4,
+        id: randomUUID(),
+        command: "push_audio_chunk",
+        sessionID,
+        pcmSamples: chunk.samples,
+        sampleRate: chunk.sampleRate,
+      });
+      if (Array.isArray(response.segments)) {
+        fallbackSegments.push(...response.segments);
+      }
+      await flushLiveSegments();
+    }
+
+    const finalResponse = await sendLocalSTTServerRequest({
+      version: 4,
+      id: randomUUID(),
+      command: "finalize_session",
+      sessionID,
+    });
+    if (Array.isArray(finalResponse.segments)) {
+      fallbackSegments.push(...finalResponse.segments);
+    }
+    await flushLiveSegments();
+
+    return normalizeTranscriptSegments(streamedSegments.length > 0 ? streamedSegments : fallbackSegments);
+  } catch (error) {
+    await sendLocalSTTServerRequest({
+      version: 4,
+      id: randomUUID(),
+      command: "cancel_session",
+      sessionID,
+    }).catch(() => {});
+    throw error;
+  } finally {
+    unsubscribe();
+  }
+}
+
+function captureEventFromLocalSTTEvent(event) {
+  if (!event || typeof event !== "object") {
+    return null;
+  }
+  if ((event.eventType === "partial_segment" || event.eventType === "final_segment") && event.segment) {
+    return {
+      kind: "transcriptionSegment",
+      transcriptionSegment: event.segment,
+    };
+  }
+  if (event.eventType === "progress") {
+    return {
+      kind: "transcriptionProgress",
+      transcriptionProgress: {
+        processedSeconds: Number(event.progress?.processedSeconds || 0),
+        totalSeconds: Number(event.progress?.receivedSeconds || 0),
+      },
+    };
+  }
+  if (event.eventType === "error") {
+    return {
+      kind: "failed",
+      errorMessage: event.error?.message || "Local STT failed.",
+    };
+  }
+  if (event.eventType === "session_finished") {
+    return {
+      kind: "stateChanged",
+      state: "finalizing",
+      detail: "Local transcription finished.",
+    };
+  }
+  return null;
+}
+
+async function* readLocalSTTPcmChunks(recordingPath, chunkDurationSeconds) {
+  const file = await fs.open(recordingPath, "r");
+  try {
+    const header = await readPcm16WavHeader(file);
+    const framesPerChunk = Math.max(1, Math.floor(header.sampleRate * chunkDurationSeconds));
+    const bytesPerChunk = framesPerChunk * header.blockAlign;
+    const buffer = Buffer.alloc(bytesPerChunk);
+    let position = header.dataOffset;
+    let remaining = header.dataSize;
+
+    while (remaining > 0) {
+      const bytesToRead = Math.min(buffer.length, remaining);
+      const alignedBytesToRead = bytesToRead - (bytesToRead % header.blockAlign);
+      if (alignedBytesToRead <= 0) {
+        break;
+      }
+      const { bytesRead } = await file.read(buffer, 0, alignedBytesToRead, position);
+      if (bytesRead <= 0) {
+        break;
+      }
+      const alignedBytesRead = bytesRead - (bytesRead % header.blockAlign);
+      if (alignedBytesRead <= 0) {
+        break;
+      }
+      yield {
+        sampleRate: header.sampleRate,
+        samples: pcm16BufferToMonoFloat32(buffer.subarray(0, alignedBytesRead), header.channels),
+      };
+      position += alignedBytesRead;
+      remaining -= alignedBytesRead;
+    }
+  } finally {
+    await file.close();
+  }
+}
+
+async function readPcm16WavHeader(file) {
+  const stat = await file.stat();
+  const riffHeader = Buffer.alloc(12);
+  const riffRead = await file.read(riffHeader, 0, riffHeader.length, 0);
+  if (riffRead.bytesRead !== riffHeader.length || riffHeader.subarray(0, 4).toString("ascii") !== "RIFF" || riffHeader.subarray(8, 12).toString("ascii") !== "WAVE") {
+    throw new Error("unsupported WAV format for local STT: expected RIFF/WAVE data");
+  }
+
+  let cursor = 12;
+  let sampleRate = 0;
+  let channels = 0;
+  let bitsPerSample = 0;
+  let audioFormat = 0;
+  let dataOffset = 0;
+  let dataSize = 0;
+  const chunkHeader = Buffer.alloc(8);
+
+  while (cursor + chunkHeader.length <= stat.size) {
+    const { bytesRead } = await file.read(chunkHeader, 0, chunkHeader.length, cursor);
+    if (bytesRead !== chunkHeader.length) {
+      break;
+    }
+    const chunkID = chunkHeader.subarray(0, 4).toString("ascii");
+    const chunkSize = chunkHeader.readUInt32LE(4);
+    cursor += 8;
+    if (cursor + chunkSize > stat.size) {
+      break;
+    }
+
+    if (chunkID === "fmt ") {
+      if (chunkSize < 16) {
+        throw new Error("unsupported WAV format for local STT: invalid fmt chunk");
+      }
+      const fmtBuffer = Buffer.alloc(16);
+      const fmtRead = await file.read(fmtBuffer, 0, fmtBuffer.length, cursor);
+      if (fmtRead.bytesRead !== fmtBuffer.length) {
+        throw new Error("unsupported WAV format for local STT: invalid fmt chunk");
+      }
+      audioFormat = fmtBuffer.readUInt16LE(0);
+      channels = fmtBuffer.readUInt16LE(2);
+      sampleRate = fmtBuffer.readUInt32LE(4);
+      bitsPerSample = fmtBuffer.readUInt16LE(14);
+    } else if (chunkID === "data") {
+      dataOffset = cursor;
+      dataSize = chunkSize;
+      break;
+    }
+
+    cursor += chunkSize + (chunkSize % 2);
+  }
+
+  if (audioFormat !== 1 || bitsPerSample !== 16 || channels <= 0 || sampleRate <= 0 || dataOffset <= 0 || dataSize <= 0) {
+    throw new Error("unsupported WAV format for local STT: only PCM16 WAV input is supported");
+  }
+
+  return {
+    sampleRate,
+    channels,
+    blockAlign: channels * 2,
+    dataOffset,
+    dataSize,
+  };
+}
+
+function pcm16BufferToMonoFloat32(buffer, channels) {
+  const blockAlign = channels * 2;
+  const frameCount = Math.floor(buffer.length / blockAlign);
+  const samples = new Array(frameCount);
+  for (let frame = 0; frame < frameCount; frame += 1) {
+    let sum = 0;
+    const frameOffset = frame * blockAlign;
+    for (let channel = 0; channel < channels; channel += 1) {
+      sum += buffer.readInt16LE(frameOffset + channel * 2) / 32768.0;
+    }
+    samples[frame] = sum / channels;
+  }
+  return samples;
+}
+
+async function transcribeLocalSTTAudioFileLegacy(recordingPath, transcriptionConfiguration) {
   const request = {
     version: 4,
     id: randomUUID(),
@@ -3228,6 +3469,7 @@ class LocalSTTServerBridge {
     this.pending = new Map();
     this.stdoutBuffer = Buffer.alloc(0);
     this.stderrTail = [];
+    this.eventHandlers = new Set();
   }
 
   ensureStarted() {
@@ -3291,6 +3533,11 @@ class LocalSTTServerBridge {
         return;
       }
 
+      if (response.event) {
+        this.emitEvent(response);
+        continue;
+      }
+
       const pending = this.pending.get(response.id);
       if (!pending) {
         continue;
@@ -3306,6 +3553,23 @@ class LocalSTTServerBridge {
     }
   }
 
+  emitEvent(response) {
+    for (const handler of this.eventHandlers) {
+      try {
+        handler(response.event, response);
+      } catch (error) {
+        console.error("Local STT event handler failed:", error);
+      }
+    }
+  }
+
+  onEvent(handler) {
+    this.eventHandlers.add(handler);
+    return () => {
+      this.eventHandlers.delete(handler);
+    };
+  }
+
   request(request) {
     this.ensureStarted();
     return new Promise((resolve, reject) => {
@@ -3363,6 +3627,22 @@ function sendLocalSTTServerRequest(request) {
   return localSTTServer.request(request);
 }
 
+async function warmLocalSTTServer() {
+  if (process.env.MIRROR_NOTE_TEST_EXPORTS === "1") {
+    return;
+  }
+  const { selectedModel, transcriptionConfiguration } = await selectedTranscriptionConfiguration();
+  if (!selectedModel || !transcriptionConfiguration) {
+    return;
+  }
+  await sendLocalSTTServerRequest({
+    version: 4,
+    id: `warm-stt-${Date.now()}`,
+    command: "prepare_model",
+    configuration: transcriptionConfiguration,
+  });
+}
+
 async function fetchCaptureCapabilities() {
   const { transcriptionConfiguration } = await selectedTranscriptionConfiguration();
   if (!transcriptionConfiguration) {
@@ -3690,7 +3970,11 @@ function registerIPCHandlers() {
   ipcMain.handle("settings:get", () => getSettings());
   ipcMain.handle("settings:update", async (_event, patch) => {
     await writeElectronSettings(patch || {});
-    return getSettings();
+    const settings = await getSettings();
+    warmLocalSTTServer().catch((error) => {
+      console.error("Local STT warmup failed after settings update:", error);
+    });
+    return settings;
   });
   ipcMain.handle("settings:download-stt-model", (event, modelID) => downloadLocalSTTModel(modelID, event.sender));
   ipcMain.handle("settings:cancel-stt-model-download", (_event, modelID) => cancelLocalSTTModelDownload(modelID));
@@ -3729,6 +4013,9 @@ function startElectronApp() {
     createApplicationMenu();
     createWindow();
     createTray();
+    warmLocalSTTServer().catch((error) => {
+      console.error("Local STT warmup failed:", error);
+    });
   });
 
   app.on("activate", () => {