update persistent PCM voice bridge

server/pcm-segmenter.ts

@@ -0,0 +1,331 @@
+/**
+ * pcm-segmenter.ts — Continuous PCM16 utterance segmenter for streaming STT.
+ *
+ * Turns an endless stream of raw PCM16LE mono audio into discrete utterance
+ * segments suitable for a request/response STT API (Whisper), without losing
+ * speech at segment boundaries:
+ *
+ *   - PRE-ROLL: audio captured just BEFORE speech onset is prepended to the
+ *     utterance so soft word onsets are not clipped.
+ *   - HANGOVER: an utterance only ends after a sustained silence, so trailing
+ *     syllables and short pauses inside a sentence are kept.
+ *   - ADAPTIVE VAD: the speech threshold tracks the noise floor, so it works
+ *     for both close-mic speech and external audio (TV, videos, news).
+ *   - FORCED CUTS: continuous speech longer than `maxUtteranceMs` is split at
+ *     the QUIETEST recent window (least likely to be mid-word) instead of an
+ *     arbitrary time boundary.
+ *
+ * The class is pure (no I/O): callers feed PCM buffers in and receive
+ * segment buffers via the `onSegment` callback.
+ */
+
+// ─── WAV encoding ────────────────────────────────────────────────────────────
+
+/** Wrap raw PCM16LE mono samples in a 44-byte WAV header. */
+export function pcm16ToWav(pcm: Buffer, sampleRate: number): Buffer {
+  const header = Buffer.alloc(44);
+  const byteRate = sampleRate * 2; // mono, 16-bit
+  header.write("RIFF", 0);
+  header.writeUInt32LE(36 + pcm.length, 4);
+  header.write("WAVE", 8);
+  header.write("fmt ", 12);
+  header.writeUInt32LE(16, 16); // fmt chunk size
+  header.writeUInt16LE(1, 20); // PCM format
+  header.writeUInt16LE(1, 22); // mono
+  header.writeUInt32LE(sampleRate, 24);
+  header.writeUInt32LE(byteRate, 28);
+  header.writeUInt16LE(2, 32); // block align
+  header.writeUInt16LE(16, 34); // bits per sample
+  header.write("data", 36);
+  header.writeUInt32LE(pcm.length, 40);
+  return Buffer.concat([header, pcm]);
+}
+
+// ─── Segmenter ───────────────────────────────────────────────────────────────
+
+export interface PcmSegmenterOptions {
+  /** Sample rate of the incoming PCM16LE mono stream. Default: 16000 */
+  sampleRate?: number;
+  /** Audio kept from before speech onset. Default: 320ms */
+  preRollMs?: number;
+  /** Sustained silence that ends an utterance. Default: 900ms */
+  hangoverMs?: number;
+  /** Utterances with less speech than this are discarded as noise. Default: 180ms */
+  minSpeechMs?: number;
+  /** Continuous speech is force-cut after this long. Default: 7000ms */
+  maxUtteranceMs?: number;
+  /** Absolute minimum level (dBFS) treated as speech. Default: -48 */
+  speechFloorDb?: number;
+  /** Speech must exceed the noise floor by this margin (dB). Default: 9 */
+  noiseMarginDb?: number;
+}
+
+export interface SegmentInfo {
+  /** True when the segment was cut mid-speech (maxUtteranceMs reached). */
+  forced: boolean;
+  /** Audio duration of the emitted segment in milliseconds. */
+  durationMs: number;
+  /** Milliseconds of the segment classified as speech. */
+  speechMs: number;
+}
+
+interface Window {
+  buf: Buffer;
+  db: number;
+  speech: boolean;
+}
+
+/** VAD analysis window; 20ms is fine-grained enough for word boundaries. */
+const WINDOW_MS = 20;
+/** Trailing silence kept on an emitted segment (helps STT decode the tail). */
+const TAIL_SILENCE_KEEP_MS = 240;
+/** Forced cuts search for the quietest window within this recent span. */
+const FORCED_CUT_SEARCH_MS = 2000;
+/** Forced cuts never split inside the newest audio (still mid-word). */
+const FORCED_CUT_GUARD_MS = 200;
+
+export class PcmSegmenter {
+  private readonly sampleRate: number;
+  private readonly preRollMs: number;
+  private readonly hangoverMs: number;
+  private readonly minSpeechMs: number;
+  private readonly maxUtteranceMs: number;
+  private readonly speechFloorDb: number;
+  private readonly noiseMarginDb: number;
+  private readonly windowBytes: number;
+
+  /** Carry-over for incoming buffers that don't align to whole windows. */
+  private remainder: Buffer = Buffer.alloc(0);
+  /** Rolling buffer of recent windows while idle (pre-roll source). */
+  private preRoll: Window[] = [];
+  /** Windows of the in-progress utterance (empty while idle). */
+  private active: Window[] = [];
+  private activeSpeechMs = 0;
+  private silenceRunMs = 0;
+  /** EMA noise floor in dBFS; rises slowly, falls quickly. */
+  private noiseFloorDb = -65;
+
+  constructor(
+    options: PcmSegmenterOptions,
+    private readonly onSegment: (pcm: Buffer, info: SegmentInfo) => void,
+  ) {
+    this.sampleRate = options.sampleRate ?? 16000;
+    this.preRollMs = options.preRollMs ?? 320;
+    this.hangoverMs = options.hangoverMs ?? 900;
+    this.minSpeechMs = options.minSpeechMs ?? 180;
+    this.maxUtteranceMs = options.maxUtteranceMs ?? 7000;
+    this.speechFloorDb = options.speechFloorDb ?? -48;
+    this.noiseMarginDb = options.noiseMarginDb ?? 9;
+    this.windowBytes = Math.round((this.sampleRate * WINDOW_MS) / 1000) * 2;
+  }
+
+  /** Feed the next PCM16LE buffer from the stream. */
+  feed(pcm: Buffer): void {
+    const data = this.remainder.length
+      ? Buffer.concat([this.remainder, pcm])
+      : pcm;
+    const wholeWindows = Math.floor(data.length / this.windowBytes);
+    const usedBytes = wholeWindows * this.windowBytes;
+    this.remainder = data.subarray(usedBytes);
+
+    for (let i = 0; i < wholeWindows; i++) {
+      const buf = data.subarray(i * this.windowBytes, (i + 1) * this.windowBytes);
+      this.processWindow(buf);
+    }
+  }
+
+  /**
+   * Finalize the in-progress utterance immediately (stream stopped).
+   * Emits the tail segment if it contains enough speech.
+   */
+  flush(): void {
+    if (this.remainder.length >= 4) {
+      // Analyse the sub-window remainder too — a stream can stop mid-window.
+      this.processWindow(this.remainder);
+    }
+    this.remainder = Buffer.alloc(0);
+    this.endUtterance(false);
+  }
+
+  /** Drop all buffered state (new session). Keeps the learned noise floor. */
+  reset(): void {
+    this.remainder = Buffer.alloc(0);
+    this.preRoll = [];
+    this.active = [];
+    this.activeSpeechMs = 0;
+    this.silenceRunMs = 0;
+  }
+
+  /**
+   * Snapshot of the in-progress utterance, for rolling partial transcription.
+   * Returns null while idle or when the utterance is still too short.
+   */
+  getActiveSnapshot(): { pcm: Buffer; durationMs: number; speechMs: number } | null {
+    if (!this.active.length || this.activeSpeechMs < this.minSpeechMs) return null;
+    const pcm = Buffer.concat(this.active.map((w) => w.buf));
+    return {
+      pcm,
+      durationMs: this.msOf(pcm),
+      speechMs: this.activeSpeechMs,
+    };
+  }
+
+  // ── Internals ────────────────────────────────────────────────────────────
+
+  private msOf(buf: Buffer): number {
+    return (buf.length / 2 / this.sampleRate) * 1000;
+  }
+
+  private speechThresholdDb(): number {
+    return Math.max(this.speechFloorDb, this.noiseFloorDb + this.noiseMarginDb);
+  }
+
+  private processWindow(buf: Buffer): void {
+    const db = rmsDb(buf);
+    const speech = db >= this.speechThresholdDb();
+    const win: Window = { buf, db, speech };
+    const winMs = this.msOf(buf);
+
+    // Track the noise floor from non-speech windows only, capped so loud
+    // continuous audio can never drag the floor up into speech territory.
+    if (!speech) {
+      const alpha = db < this.noiseFloorDb ? 0.2 : 0.02;
+      this.noiseFloorDb = Math.min(
+        -35,
+        this.noiseFloorDb + alpha * (db - this.noiseFloorDb),
+      );
+    }
+
+    if (!this.active.length) {
+      if (speech) {
+        // Speech onset: start the utterance with the pre-roll prepended.
+        this.active = [...this.preRoll, win];
+        this.preRoll = [];
+        this.activeSpeechMs = winMs;
+        this.silenceRunMs = 0;
+      } else {
+        this.preRoll.push(win);
+        let total = this.preRoll.reduce((s, w) => s + this.msOf(w.buf), 0);
+        while (this.preRoll.length > 1 && total > this.preRollMs) {
+          total -= this.msOf(this.preRoll[0].buf);
+          this.preRoll.shift();
+        }
+      }
+      return;
+    }
+
+    // Utterance in progress: keep everything (speech AND intra-speech silence).
+    this.active.push(win);
+    if (speech) {
+      this.activeSpeechMs += winMs;
+      this.silenceRunMs = 0;
+    } else {
+      this.silenceRunMs += winMs;
+    }
+
+    if (this.silenceRunMs >= this.hangoverMs) {
+      this.endUtterance(false);
+      return;
+    }
+
+    const activeMs = this.active.reduce((s, w) => s + this.msOf(w.buf), 0);
+    if (activeMs >= this.maxUtteranceMs) {
+      this.forceCut();
+    }
+  }
+
+  private endUtterance(forced: boolean): void {
+    if (!this.active.length) return;
+
+    let windows = this.active;
+    this.active = [];
+    const speechMs = this.activeSpeechMs;
+    this.activeSpeechMs = 0;
+    this.silenceRunMs = 0;
+
+    // Trim trailing silence beyond a short tail; the trimmed silence seeds
+    // the pre-roll so the next onset still has context.
+    let tailSilence = 0;
+    while (tailSilence < windows.length && !windows[windows.length - 1 - tailSilence].speech) {
+      tailSilence++;
+    }
+    const keepTail = Math.ceil(TAIL_SILENCE_KEEP_MS / WINDOW_MS);
+    if (tailSilence > keepTail) {
+      const cut = windows.length - (tailSilence - keepTail);
+      this.preRoll = windows.slice(cut);
+      windows = windows.slice(0, cut);
+    }
+
+    if (speechMs < this.minSpeechMs) {
+      // Noise blip — don't transcribe it, but keep its tail as pre-roll so a
+      // real onset right after still gets its leading audio.
+      this.preRoll = [...windows.slice(-Math.ceil(this.preRollMs / WINDOW_MS)), ...this.preRoll];
+      return;
+    }
+
+    const pcm = Buffer.concat(windows.map((w) => w.buf));
+    this.onSegment(pcm, {
+      forced,
+      durationMs: this.msOf(pcm),
+      speechMs,
+    });
+  }
+
+  /**
+   * Continuous speech reached maxUtteranceMs: split at the quietest window in
+   * the recent past (least likely to be mid-word). Audio after the cut point
+   * becomes the start of the next utterance — nothing is dropped.
+   */
+  private forceCut(): void {
+    const searchWindows = Math.ceil(FORCED_CUT_SEARCH_MS / WINDOW_MS);
+    const guardWindows = Math.ceil(FORCED_CUT_GUARD_MS / WINDOW_MS);
+    const end = Math.max(1, this.active.length - guardWindows);
+    const start = Math.max(1, end - searchWindows);
+
+    let cut = end - 1;
+    for (let i = start; i < end; i++) {
+      if (this.active[i].db < this.active[cut].db) cut = i;
+    }
+
+    const emitted = this.active.slice(0, cut);
+    const carried = this.active.slice(cut);
+    const emittedSpeechMs = emitted.reduce(
+      (s, w) => s + (w.speech ? this.msOf(w.buf) : 0),
+      0,
+    );
+
+    this.active = carried;
+    this.activeSpeechMs = carried.reduce(
+      (s, w) => s + (w.speech ? this.msOf(w.buf) : 0),
+      0,
+    );
+    // silenceRunMs continues counting across the cut; recompute from the tail.
+    let silence = 0;
+    for (let i = carried.length - 1; i >= 0 && !carried[i].speech; i--) {
+      silence += this.msOf(carried[i].buf);
+    }
+    this.silenceRunMs = silence;
+
+    if (emittedSpeechMs < this.minSpeechMs || !emitted.length) return;
+
+    const pcm = Buffer.concat(emitted.map((w) => w.buf));
+    this.onSegment(pcm, {
+      forced: true,
+      durationMs: this.msOf(pcm),
+      speechMs: emittedSpeechMs,
+    });
+  }
+}
+
+/** RMS level of a PCM16LE buffer in dBFS. */
+function rmsDb(buf: Buffer): number {
+  const samples = buf.length >> 1;
+  if (!samples) return -Infinity;
+  let sumSq = 0;
+  for (let i = 0; i < samples; i++) {
+    const s = buf.readInt16LE(i * 2) / 32768;
+    sumSq += s * s;
+  }
+  const rms = Math.sqrt(sumSq / samples);
+  return 20 * Math.log10(rms + 1e-10);
+}

server/voice-ws.ts

@@ -1,810 +1,608 @@
 /**
- * voice-ws.ts — Real-time voice-to-sign WebSocket server
+ * Persistent real-time Arabic speech bridge.
  *
- * Protocol (client → server):
- *   { type: "chunk", audio: "<base64>", mimeType: "audio/m4a", chunkId: N }
- *   { type: "text",  text: "...",       chunkId: N }   ← legacy: pre-recognized text
- *   { type: "stop" }
- *
- * Protocol (server → client):
- *   { type: "ready" }
- *   { type: "queued",             chunkId, queueSize }
- *   { type: "partial_transcript", text, chunkId }
- *   { type: "final_transcript",   text, words, chunkId, rawText }
- *   { type: "words",              words, chunkId, rawText }  ← legacy compat
- *   { type: "error",              message, chunkId? }
- *
- * Algorithms:
- * ─────────────────────────────────────────────────────────────────────────────
- * 1. SLIDING WINDOW CONTEXT (NLP)
- *    Each chunk sent to Whisper includes the last 5 words as prompt context
- *    to prevent word splitting across chunk boundaries.
- *
- * 2. MIN-HEAP ORDERED DELIVERY (Data Structure)
- *    Results stored in a min-heap sorted by chunkId for ordered delivery
- *    even when results arrive out of order.
- *
- * 3. MAXIMUM MUNCH TOKENIZATION (NLP)
- *    Extract all known words from a sentence using greedy longest-match-first.
- *
- * 4. PARALLEL PROCESSING + DROP STRATEGY (Performance)
- *    Process up to 4 chunks in parallel; drop excess to prevent queue buildup.
- *
- * 5. ARABIC TEXT NORMALIZATION (NLP) — internal only
- *    Used only for sign/avatar VOCAB matching. Raw dialect preserved in transcripts.
- *
- * 6. POSITION-BASED DEDUPLICATION
- *    Track sent words by position to prevent immediate repeats.
- * ─────────────────────────────────────────────────────────────────────────────
+ * The mobile/web client streams sequenced PCM16 frames here. One persistent
+ * recognizer WebSocket is retained for each logical session, including across
+ * short client reconnects. No live audio is sent through Whisper HTTP.
  */
-
-import { WebSocketServer, WebSocket } from "ws";
+import { WebSocket, WebSocketServer } from "ws";
 import type { Server } from "http";
-import { ENV } from "./_core/env";
-
-// ─── Message types ───────────────────────────────────────────────────────────
-
-interface ChunkMsg {
-  type: "chunk";
-  audio: string; // base64
-  mimeType?: string;
-  chunkId: number;
-}
-interface TextMsg {
-  type: "text";
-  text: string;
-  chunkId: number;
-}
-interface StopMsg {
-  type: "stop";
+import {
+  OrderedAudioFrames,
+  type BufferedAudioFrame,
+} from "../shared/realtime-voice-protocol";
+
+const SESSION_RETENTION_MS = 30000;
+const UPSTREAM_RECONNECT_INITIAL_MS = 250;
+const UPSTREAM_RECONNECT_MAX_MS = 8000;
+const MAX_UPSTREAM_REPLAY_FRAMES = 600;
+
+type SendMessage = (message: Record<string, unknown>) => void;
+
+interface PcmStartMessage {
+  type: "pcm_start";
+  sessionId: string;
+  sampleRate?: number;
+  firstAudioSequence?: number;
 }
-type ClientMessage = ChunkMsg | TextMsg | StopMsg;
-
-// ─── Arabic text normalization (internal only — for sign/avatar matching) ────
-
-/** Normalize for VOCAB lookup only. Never use on user-facing transcript text. */
-function normalizeAr(s: string): string {
-  return s
-    .replace(/[\u064B-\u065F\u0670]/g, "") // remove diacritics
-    .replace(/[أإآ]/g, "ا") // unify alif
-    .replace(/ة/g, "ه") // taa marbuta → haa
-    .replace(/ى/g, "ي") // alif maqsura → yaa
-    .replace(/[^\u0600-\u06FF\s]/g, " ") // remove non-Arabic
-    .replace(/\s+/g, " ")
-    .trim();
-}
-
-// ─── Vocabulary (normalized → canonical key) ─────────────────────────────────
-
-const VOCAB: Record<string, string> = {
-  انا: "أنا",
-  بحب: "بحب",
-  احب: "بحب",
-  اهواك: "بحب",
-  صباح: "صباح",
-  الخير: "الخير",
-  خير: "الخير",
-  النور: "النور",
-  نور: "النور",
-  المدرسه: "المدرسة",
-  مدرسه: "المدرسة",
-  مدرسة: "المدرسة",
-  التعليم: "المدرسة",
-  الدراسه: "المدرسة",
-  الدراسة: "المدرسة",
-  عشان: "عشان",
-  لان: "عشان",
-  لأن: "عشان",
-  مرحبا: "مرحبا",
-  مرحبه: "مرحبا",
-  اهلا: "مرحبا",
-  هلا: "مرحبا",
-  السلام: "مرحبا",
-  اهلن: "مرحبا",
-  شكرا: "شكراً",
-  شكر: "شكراً",
-  شكرًا: "شكراً",
-  عفوا: "عفواً",
-  عفو: "عفواً",
-  مساء: "مساء",
-  مساءً: "مساء",
-  مساعده: "مساعدة",
-  ساعدني: "مساعدة",
-  نجده: "مساعدة",
-  مساعدة: "مساعدة",
-  النجده: "مساعدة",
-  نعم: "نعم",
-  اه: "نعم",
-  اي: "نعم",
-  صح: "نعم",
-  ايوه: "نعم",
-  لا: "لا",
-  ماء: "ماء",
-  ميه: "ماء",
-  مي: "ماء",
-  ماي: "ماء",
-  ماءً: "ماء",
-  طيب: "طيب",
-  حسنا: "طيب",
-  تمام: "طيب",
-  حسنًا: "طيب",
-  اوكيه: "طيب",
-  سعيد: "سعيد",
-  مبسوط: "سعيد",
-  فرحان: "سعيد",
-  سعيده: "سعيد",
-  مبسوطه: "سعيد",
-};
 
-/**
- * NLP: Extract all known words from a sentence in order.
- * Algorithm: Maximum Munch on single-word tokens.
- */
-function extractAllWords(rawText: string): string[] {
-  const normalized = normalizeAr(rawText);
-  const tokens = normalized.split(/\s+/).filter(Boolean);
-  const result: string[] = [];
-  for (const token of tokens) {
-    const canonical = VOCAB[token];
-    if (canonical) result.push(canonical);
-  }
-  return result;
+interface PcmMessage {
+  type: "pcm";
+  sessionId: string;
+  audioSequence: number;
+  audio: string;
 }
 
-// ─── Min-Heap for ordered delivery ───────────────────────────────────────────
-
-interface HeapItem {
-  chunkId: number;
-  words: string[];
-  rawText: string;
+interface PcmStopMessage {
+  type: "pcm_stop";
+  sessionId: string;
+  lastAudioSequence?: number;
 }
 
-class MinHeap {
-  private data: HeapItem[] = [];
-  push(item: HeapItem) {
-    this.data.push(item);
-    this.bubbleUp(this.data.length - 1);
-  }
-  peek(): HeapItem | undefined {
-    return this.data[0];
-  }
-  pop(): HeapItem | undefined {
-    if (!this.data.length) return undefined;
-    const top = this.data[0];
-    const last = this.data.pop()!;
-    if (this.data.length) {
-      this.data[0] = last;
-      this.sinkDown(0);
-    }
-    return top;
-  }
-  get size() {
-    return this.data.length;
-  }
-  private bubbleUp(i: number) {
-    while (i > 0) {
-      const p = (i - 1) >> 1;
-      if (this.data[p].chunkId <= this.data[i].chunkId) break;
-      [this.data[p], this.data[i]] = [this.data[i], this.data[p]];
-      i = p;
-    }
-  }
-  private sinkDown(i: number) {
-    const n = this.data.length;
-    while (true) {
-      let min = i;
-      const l = 2 * i + 1,
-        r = 2 * i + 2;
-      if (l < n && this.data[l].chunkId < this.data[min].chunkId) min = l;
-      if (r < n && this.data[r].chunkId < this.data[min].chunkId) min = r;
-      if (min === i) break;
-      [this.data[min], this.data[i]] = [this.data[i], this.data[min]];
-      i = min;
-    }
+function sendJson(socket: WebSocket, message: Record<string, unknown>): void {
+  if (socket.readyState === WebSocket.OPEN) {
+    socket.send(JSON.stringify(message));
   }
 }
 
-// ─── Sliding Window Context ──────────────────────────────────────────────────
-
-class SlidingContext {
-  private lastRawWords: string[] = [];
-
-  getPrompt(): string {
-    // Whisper's initial_prompt is designed for *prior speech context only* — NOT
-    // instructions. Passing Arabic instructions causes Whisper to echo their
-    // vocabulary (e.g. "موسيقى") as hallucinations. Only pass the last spoken
-    // words so Whisper maintains continuity across chunk boundaries.
-    if (!this.lastRawWords.length) return "";
-    return this.lastRawWords.slice(-5).join(" ");
+function recognizerWebSocketUrl(): string {
+  const configured =
+    process.env.WHISPER_VAD_WS_URL ??
+    process.env.WHISPER_VAD_URL ??
+    process.env.EXPO_PUBLIC_WHISPER_VAD_URL ??
+    "https://taqarub-vosk-ar.hf.space";
+  const raw = configured.trim().replace(/^['"]|['"]$/g, "").replace(/\/+$/, "");
+  if (!raw) {
+    throw new Error(
+      "Persistent recognizer is not configured. Set WHISPER_VAD_URL.",
+    );
   }
-
-  /** Feed raw (un-normalized) Whisper output to preserve dialect in context */
-  update(rawText: string) {
-    const words = rawText.trim().split(/\s+/).filter(Boolean);
-    this.lastRawWords = [...this.lastRawWords, ...words].slice(-10);
+  if (raw.endsWith("/ws/transcribe")) {
+    return raw.replace(/^https:/, "wss:").replace(/^http:/, "ws:");
   }
+  return `${raw.replace(/^https:/, "wss:").replace(/^http:/, "ws:")}/ws/transcribe`;
+}
 
-  reset() {
-    this.lastRawWords = [];
-  }
+function transcriptWords(text: string): string[] {
+  return text
+    .replace(/[^\p{L}\p{N}]+/gu, " ")
+    .trim()
+    .split(/\s+/)
+    .filter(Boolean);
 }
 
-// ─── Voice Session ───────────────────────────────────────────────────────────
-
-class VoiceSession {
-  private heap = new MinHeap();
-  private nextExpected = 0;
-  private processing = 0;
-  private readonly MAX_CONCURRENT = 4;
-  private backlog: {
-    chunkId: number;
-    audioBase64: string;
-    mimeType: string;
-    ext: string;
-  }[] = [];
-  private context = new SlidingContext();
-  private allSentWords: string[] = [];
-  private totalEnqueued = 0;
-  /** How many consecutive chunks came back empty (silence/hallucination filtered) */
-  private consecutiveSilenceCount = 0;
+class PersistentArabicRecognizer {
+  private socket: WebSocket | null = null;
+  private connectPromise: Promise<void> | null = null;
+  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private reconnectAttempt = 0;
+  private handshakeReady = false;
+  private active = true;
+  private lastAcknowledgedSequence = -1;
+  private replayFrames = new Map<number, BufferedAudioFrame>();
+  private latestRevision = new Map<string, number>();
+  private finalizedUtterances = new Set<string>();
+  private finalEventHistory = new Map<number, Record<string, unknown>>();
+  private latestInterimEvent: Record<string, unknown> | null = null;
+  private nextUtteranceIndex = 0;
+  private upstreamProtocolVersion = 1;
+  private legacyRevision = 0;
+  private legacyInterim = "";
 
   constructor(
-    private ws: WebSocket,
-    private sendMsg: (msg: Record<string, any>) => void,
+    readonly sessionId: string,
+    private readonly sampleRate: number,
+    private readonly emit: SendMessage,
   ) {}
 
-  enqueue(chunk: {
-    chunkId: number;
-    audioBase64: string;
-    mimeType: string;
-    ext: string;
-  }) {
-    this.totalEnqueued++;
-    this.backlog.push(chunk);
-    this.sendMsg({
-      type: "queued",
-      chunkId: chunk.chunkId,
-      queueSize: this.processing + this.backlog.length,
-    });
-
-    this.drainBacklog();
+  start(): Promise<void> {
+    this.active = true;
+    return this.ensureConnected();
   }
 
-  private drainBacklog() {
-    while (this.processing < this.MAX_CONCURRENT && this.backlog.length > 0) {
-      const chunk = this.backlog.shift()!;
-      this.processing++;
-      const prompt = this.context.getPrompt();
-      this.processChunk(chunk, prompt)
-        .catch((err) =>
-          console.error(`[voice-ws] chunk ${chunk.chunkId}:`, err.message),
-        )
-        .finally(() => {
-          this.processing--;
-          this.drainBacklog();
+  accept(frame: BufferedAudioFrame): void {
+    if (frame.audioSequence <= this.lastAcknowledgedSequence) {
+      this.emitAck();
+      return;
+    }
+    if (!this.replayFrames.has(frame.audioSequence)) {
+      if (this.replayFrames.size >= MAX_UPSTREAM_REPLAY_FRAMES) {
+        this.emit({
+          type: "error",
+          sessionId: this.sessionId,
+          message: "Recognizer replay buffer is full",
         });
+        return;
+      }
+      this.replayFrames.set(frame.audioSequence, frame);
+    }
+    this.sendFrame(frame);
+    if (!this.socket || this.socket.readyState !== WebSocket.OPEN) {
+      void this.ensureConnected().catch(() => this.scheduleReconnect());
     }
   }
 
-  private async processChunk(
-    chunk: {
-      chunkId: number;
-      audioBase64: string;
-      mimeType: string;
-      ext: string;
-    },
-    prompt: string,
-  ) {
-    let text: string;
-    try {
-      text = await transcribeWithWhisper(
-        chunk.audioBase64,
-        chunk.ext,
-        chunk.mimeType,
-        prompt,
-      );
-    } catch (err: any) {
-      console.error(
-        `[voice-ws] ❌ transcription failed for chunk ${chunk.chunkId}:`,
-        err.message,
+  stop(): void {
+    this.active = false;
+    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    this.reconnectTimer = null;
+    if (this.socket?.readyState === WebSocket.OPEN) {
+      this.socket.send(
+        JSON.stringify({
+          type: "pcm_stop",
+          sessionId: this.sessionId,
+        }),
       );
-      this.sendMsg({
-        type: "error",
-        message: `Transcription failed: ${err.message}`,
-        chunkId: chunk.chunkId,
-      });
-      // Still add to heap with empty text so ordering isn't broken
-      text = "";
     }
+  }
 
-    // Only update context on real speech.
-    // On silence/hallucination (text === ""), increment the silence streak.
-    // After 2+ consecutive silent chunks reset the context so stale speech
-    // content stops "anchoring" Whisper and pulling it toward hallucinations.
-    if (text) {
-      this.consecutiveSilenceCount = 0;
-      this.context.update(text);
-    } else {
-      this.consecutiveSilenceCount++;
-      if (this.consecutiveSilenceCount >= 1) {
-        this.context.reset();
-        console.log("[voice-ws] 🔄 context reset — silence detected");
-      }
-    }
+  close(): void {
+    this.active = false;
+    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    this.reconnectTimer = null;
+    this.socket?.close();
+    this.socket = null;
+    this.connectPromise = null;
+  }
 
-    // Send partial transcript immediately (before word extraction + ordering)
-    if (text.trim()) {
-      this.sendMsg({
-        type: "partial_transcript",
-        text: text.trim(),
-        chunkId: chunk.chunkId,
-      });
-    }
+  acknowledgedSequence(): number {
+    return this.lastAcknowledgedSequence;
+  }
 
-    const words = extractAllWords(text);
-    this.heap.push({ chunkId: chunk.chunkId, words, rawText: text });
-    this.flushOrdered();
+  replayTranscriptEvents(): void {
+    for (const event of [...this.finalEventHistory.entries()]
+      .sort(([left], [right]) => left - right)
+      .map(([, value]) => value)) {
+      this.emit(event);
+    }
+    if (this.latestInterimEvent) this.emit(this.latestInterimEvent);
   }
 
-  private flushOrdered() {
-    while (
-      this.heap.size > 0 &&
-      this.heap.peek()!.chunkId === this.nextExpected
+  private ensureConnected(): Promise<void> {
+    if (
+      this.socket?.readyState === WebSocket.OPEN &&
+      this.handshakeReady
     ) {
-      const item = this.heap.pop()!;
-      this.nextExpected++;
-
-      // Deduplicate words by position
-      const newWords: string[] = [];
-      if (item.words.length > 0) {
-        for (const w of item.words) {
-          if (this.shouldSend(w)) {
-            newWords.push(w);
-            this.allSentWords.push(w);
-          }
-        }
-      }
-
-      // Send final_transcript with both text and matched words
-      this.sendMsg({
-        type: "final_transcript",
-        text: item.rawText.trim(),
-        words: newWords,
-        chunkId: item.chunkId,
-        rawText: item.rawText.trim(),
-      });
+      return Promise.resolve();
+    }
+    if (this.connectPromise) return this.connectPromise;
 
-      // Also send legacy "words" message for backward compatibility
-      if (newWords.length > 0) {
-        this.sendMsg({
-          type: "words",
-          words: newWords,
-          chunkId: item.chunkId,
-          rawText: item.rawText,
-        });
+    const promise = new Promise<void>((resolve, reject) => {
+      let url: string;
+      try {
+        url = recognizerWebSocketUrl();
+      } catch (error) {
+        reject(error);
+        return;
       }
-    }
-  }
 
-  /**
-   * Should we send this word?
-   * Yes if: never sent before, or 8+ other words have been sent since.
-   */
-  private shouldSend(word: string): boolean {
-    const lastIdx = this.allSentWords.lastIndexOf(word);
-    if (lastIdx === -1) return true;
-    return this.allSentWords.length - lastIdx > 8;
-  }
+      const socket = new WebSocket(url, { perMessageDeflate: false });
+      this.socket = socket;
+      this.handshakeReady = false;
+      let connected = false;
+      const timeout = setTimeout(() => {
+        if (connected) return;
+        socket.close();
+        reject(new Error(`Timed out connecting to Arabic recognizer at ${url}`));
+      }, 8000);
+
+      socket.on("message", (raw) => {
+        let message: Record<string, any>;
+        try {
+          message = JSON.parse(raw.toString());
+        } catch {
+          return;
+        }
 
-  /** Process pre-recognized text (from Web Speech API) — no Whisper needed */
-  processText(chunkId: number, rawText: string) {
-    if (!rawText.trim()) return;
-    this.totalEnqueued++;
-    this.sendMsg({
-      type: "queued",
-      chunkId,
-      queueSize: this.processing,
-    });
+        if (message.type === "ready") {
+          this.upstreamProtocolVersion = Number(
+            message.protocolVersion ?? 1,
+          );
+          const bufferedSequences = [...this.replayFrames.keys()];
+          const firstAudioSequence =
+            bufferedSequences.length > 0
+              ? Math.min(...bufferedSequences)
+              : this.lastAcknowledgedSequence + 1;
+          socket.send(
+            JSON.stringify({
+              type: "pcm_start",
+              sessionId: this.sessionId,
+              sampleRate: this.sampleRate,
+              firstAudioSequence,
+              firstUtteranceIndex: this.nextUtteranceIndex,
+              language: "ar",
+            }),
+          );
+          return;
+        }
 
-    const words = extractAllWords(rawText);
-    this.context.update(rawText);
-    this.heap.push({ chunkId, words, rawText });
-    this.flushOrdered();
-  }
+        if (message.type === "pcm_ready") {
+          connected = true;
+          clearTimeout(timeout);
+          this.handshakeReady = true;
+          this.reconnectAttempt = 0;
+          this.replay();
+          resolve();
+          return;
+        }
 
-  reset() {
-    this.heap = new MinHeap();
-    this.nextExpected = 0;
-    this.processing = 0;
-    this.context.reset();
-    this.allSentWords = [];
-    this.totalEnqueued = 0;
-    this.consecutiveSilenceCount = 0;
-  }
-}
+        if (message.type === "audio_ack") {
+          const sequence = Number(message.audioSequence);
+          if (Number.isFinite(sequence)) {
+            this.lastAcknowledgedSequence = Math.max(
+              this.lastAcknowledgedSequence,
+              sequence,
+            );
+            for (const key of this.replayFrames.keys()) {
+              if (key <= this.lastAcknowledgedSequence) {
+                this.replayFrames.delete(key);
+              }
+            }
+            this.emitAck();
+          }
+          return;
+        }
 
-// ─── Whisper API ─────────────────────────────────────────────────────────────
+        if (message.type === "interim" || message.type === "final") {
+          this.forwardTranscript(message);
+          return;
+        }
 
-// ─── Silence / hallucination detection ───────────────────────────────────────
+        if (
+          message.type === "partial_transcript" ||
+          message.type === "final_transcript"
+        ) {
+          this.forwardLegacyTranscript(message);
+          return;
+        }
 
-/** no_speech_prob threshold: above this → treat chunk as silence.
- *  0.6 is well-calibrated for Arabic speech; 0.35 was too aggressive and
- *  caused real speech with low energy to be silently dropped. */
-const NO_SPEECH_THRESHOLD = 0.6;
+        if (message.type === "error") {
+          this.emit({
+            type: "error",
+            sessionId: this.sessionId,
+            message: String(message.message ?? "Recognizer error"),
+          });
+        }
+      });
 
-/** avg_logprob threshold: below this → model is guessing, not transcribing.
- *  -0.8 avoids over-filtering dialect Arabic which naturally scores lower. */
-const LOW_CONFIDENCE_LOGPROB = -0.8;
+      socket.on("error", (error) => {
+        if (!connected) {
+          clearTimeout(timeout);
+          reject(error);
+        }
+      });
 
-/** compression_ratio threshold: above this → repetitive/looping hallucination */
-const HIGH_COMPRESSION_RATIO = 2.4;
+      socket.on("close", () => {
+        clearTimeout(timeout);
+        if (this.socket === socket) this.socket = null;
+        this.handshakeReady = false;
+        this.connectPromise = null;
+        if (this.active) this.scheduleReconnect();
+      });
+    }).finally(() => {
+      this.connectPromise = null;
+    });
 
-/**
- * Punctuation-only / symbol-only outputs that carry no speech content.
- */
-const HALLUCINATION_RE = /^[\s.،؟!…\-–—\u200B-\u200F♪♫♬]+$/;
+    this.connectPromise = promise;
+    return promise;
+  }
 
-/**
- * Ghost phrases Whisper hallucinates from YouTube/podcast training data.
- * These are STRIPPED from transcripts (not used to drop the whole chunk),
- * so real speech appearing in the same chunk is preserved.
- *
- * Order matters: longer/more-specific phrases must come before their substrings
- * so the longer form is stripped first.
- */
-const GHOST_PHRASES: string[] = [
-  // ── Arabic — longer phrases first ────────────────────────────────────────
-  "شكراً على المشاهدة",
-  "شكرا على المشاهدة",
-  "شكراً على المتابعة",
-  "شكرا على المتابعة",
-  "للمزيد من الفيديوهات",
-  "للمزيد من المحتوى",
-  "لا تنسى الاشتراك",
-  "لا تنسوا الاشتراك",
-  "اشترك معنا",
-  "اللهم صل على محمد",
-  "صلى الله عليه وسلم",
-  "بسم الله الرحمن الرحيم",
-  "الحمد لله رب العالمين",
-  // ── Arabic — single-word ghosts ───────────────────────────────────────────
-  "اشتركوا", // subscribe (plural)  — before "اشترك"
-  "أشترك", // subscribe (hamza)
-  "اشترك", // subscribe
-  "الحمد لله",
-  "سبحان الله",
-  "موسيقى", // music — most common Arabic hallucination on silence
-  "موسيقا",
-  // ── English (training bleed) ─────────────────────────────────────────────
-  "thanks for watching",
-  "thank you for watching",
-  "please subscribe",
-  "see you next time",
-  "background music",
-  "[applause]",
-  "[laughter]",
-  "[silence]",
-  "[inaudible]",
-  "[music]",
-  "subscribe",
-  "music",
-];
+  private scheduleReconnect(): void {
+    if (!this.active || this.reconnectTimer) return;
+    const delay = Math.min(
+      UPSTREAM_RECONNECT_MAX_MS,
+      UPSTREAM_RECONNECT_INITIAL_MS * 2 ** this.reconnectAttempt++,
+    );
+    this.reconnectTimer = setTimeout(() => {
+      this.reconnectTimer = null;
+      void this.ensureConnected().catch(() => this.scheduleReconnect());
+    }, delay);
+  }
 
-/**
- * Strip every ghost phrase from `text`, normalise whitespace, and return
- * the cleaned string.  Returns "" if nothing real remains.
- *
- * Stripping (not dropping) preserves real speech that appeared in the same
- * chunk as a hallucination — the previous approach discarded whole chunks.
- */
-function stripGhostPhrases(text: string): string {
-  let result = text.trim();
-  for (const ghost of GHOST_PHRASES) {
-    // Arabic has no case, but toLowerCase is harmless and handles English ghosts.
-    const re = new RegExp(ghost.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "gi");
-    const before = result;
-    result = result
-      .replace(re, " ")
-      .replace(/\s{2,}/g, " ")
-      .trim();
-    if (result !== before) {
-      console.log(`[voice-ws] ✂️ stripped ghost "${ghost}" → "${result}"`);
+  private sendFrame(frame: BufferedAudioFrame): void {
+    if (
+      !this.handshakeReady ||
+      this.socket?.readyState !== WebSocket.OPEN
+    ) {
+      return;
+    }
+    this.socket.send(
+      JSON.stringify({
+        type: "pcm",
+        sessionId: this.sessionId,
+        audioSequence: frame.audioSequence,
+        chunkId: frame.audioSequence,
+        sampleRate: this.sampleRate,
+        audio: frame.audio,
+      }),
+    );
+    if (this.upstreamProtocolVersion < 2) {
+      this.lastAcknowledgedSequence = Math.max(
+        this.lastAcknowledgedSequence,
+        frame.audioSequence,
+      );
+      this.replayFrames.delete(frame.audioSequence);
+      this.emitAck();
     }
   }
-  if (!result || HALLUCINATION_RE.test(result)) return "";
-  return result;
-}
 
-/**
- * Segment-level stats check: returns true only when the ENTIRE chunk has
- * no real speech (high no_speech_prob, very low logprob, or looping output).
- */
-function isChunkSilence(
-  segments:
-    | {
-        no_speech_prob?: number;
-        avg_logprob?: number;
-        compression_ratio?: number;
-      }[]
-    | undefined,
-): boolean {
-  if (!segments?.length) return false;
-  const n = segments.length;
-
-  const avgNoSpeech =
-    segments.reduce((s, seg) => s + (seg.no_speech_prob ?? 0), 0) / n;
-  if (avgNoSpeech > NO_SPEECH_THRESHOLD) {
-    console.log(
-      `[voice-ws] 🤫 silence (no_speech_prob=${avgNoSpeech.toFixed(2)})`,
+  private replay(): void {
+    const frames = [...this.replayFrames.values()].sort(
+      (left, right) => left.audioSequence - right.audioSequence,
     );
-    return true;
+    for (const frame of frames) this.sendFrame(frame);
   }
 
-  const avgLogprob =
-    segments.reduce((s, seg) => s + (seg.avg_logprob ?? 0), 0) / n;
-  if (avgLogprob < LOW_CONFIDENCE_LOGPROB) {
-    console.log(
-      `[voice-ws] 🎲 low confidence (avg_logprob=${avgLogprob.toFixed(2)})`,
-    );
-    return true;
+  private emitAck(): void {
+    this.emit({
+      type: "audio_ack",
+      sessionId: this.sessionId,
+      audioSequence: this.lastAcknowledgedSequence,
+    });
   }
 
-  const avgCompression =
-    segments.reduce((s, seg) => s + (seg.compression_ratio ?? 0), 0) / n;
-  if (avgCompression > HIGH_COMPRESSION_RATIO) {
-    console.log(
-      `[voice-ws] 🔁 repetitive (compression_ratio=${avgCompression.toFixed(2)})`,
-    );
-    return true;
+  private forwardTranscript(message: Record<string, any>): void {
+    const utteranceId = String(message.utteranceId ?? "");
+    const revision = Number(message.revision);
+    const utteranceIndex = Number(message.utteranceIndex);
+    const text = String(message.text ?? "").trim();
+    if (
+      !utteranceId ||
+      !Number.isFinite(revision) ||
+      !Number.isFinite(utteranceIndex)
+    ) {
+      return;
+    }
+
+    const previousRevision = this.latestRevision.get(utteranceId) ?? -1;
+    if (message.type === "interim") {
+      if (
+        this.finalizedUtterances.has(utteranceId) ||
+        revision <= previousRevision
+      ) {
+        return;
+      }
+    } else {
+      if (
+        this.finalizedUtterances.has(utteranceId) ||
+        revision < previousRevision
+      ) {
+        return;
+      }
+      this.finalizedUtterances.add(utteranceId);
+    }
+
+    this.latestRevision.set(utteranceId, revision);
+    const event: Record<string, unknown> = {
+      type: message.type,
+      sessionId: this.sessionId,
+      utteranceId,
+      utteranceIndex,
+      revision,
+      audioSequence: Number(message.audioSequence),
+      text,
+      words: message.type === "final" ? transcriptWords(text) : undefined,
+    };
+    if (message.type === "final") {
+      this.nextUtteranceIndex = Math.max(
+        this.nextUtteranceIndex,
+        utteranceIndex + 1,
+      );
+      this.finalEventHistory.set(utteranceIndex, event);
+      if (this.latestInterimEvent?.utteranceId === utteranceId) {
+        this.latestInterimEvent = null;
+      }
+    } else {
+      this.latestInterimEvent = event;
+    }
+    this.emit(event);
   }
 
-  return false;
-}
+  private forwardLegacyTranscript(message: Record<string, any>): void {
+    const text = String(message.text ?? "").trim();
+    if (!text) return;
+
+    if (message.type === "partial_transcript") {
+      if (text === this.legacyInterim) return;
+      this.legacyInterim = text;
+      this.legacyRevision++;
+      this.forwardTranscript({
+        type: "interim",
+        utteranceId: `${this.sessionId}:${this.nextUtteranceIndex}`,
+        utteranceIndex: this.nextUtteranceIndex,
+        revision: this.legacyRevision,
+        audioSequence: Number(message.chunkId ?? -1),
+        text,
+      });
+      return;
+    }
 
-/**
- * Master transcript cleaner.
- * 1. Reject pure punctuation/symbol outputs.
- * 2. Strip known ghost phrases (preserves real speech in the same chunk).
- * 3. Drop the chunk only if segment stats confirm total silence.
- *
- * Returns cleaned text, or "" if nothing real remains.
- */
-function cleanTranscript(
-  rawText: string,
-  segments:
-    | {
-        no_speech_prob?: number;
-        avg_logprob?: number;
-        compression_ratio?: number;
-      }[]
-    | undefined,
-): string {
-  // 1. Pure punctuation / symbols
-  if (!rawText.trim() || HALLUCINATION_RE.test(rawText.trim())) return "";
-
-  // 2. Strip ghost phrases — may return "" if nothing real is left
-  const cleaned = stripGhostPhrases(rawText);
-  if (!cleaned) return "";
-
-  // 3. Segment-level silence check
-  if (isChunkSilence(segments)) return "";
-
-  return cleaned;
+    this.legacyRevision++;
+    this.forwardTranscript({
+      type: "final",
+      utteranceId: `${this.sessionId}:${this.nextUtteranceIndex}`,
+      utteranceIndex: this.nextUtteranceIndex,
+      revision: this.legacyRevision,
+      audioSequence: Number(message.chunkId ?? -1),
+      text,
+    });
+    this.legacyRevision = 0;
+    this.legacyInterim = "";
+  }
 }
 
-// ─── Whisper API ─────────────────────────────────────────────────────────────
+class LiveRecognitionSession {
+  private sender: SendMessage = () => {};
+  private orderedAudio: OrderedAudioFrames;
+  private recognizer: PersistentArabicRecognizer;
+  private retentionTimer: ReturnType<typeof setTimeout> | null = null;
 
-async function transcribeWithWhisper(
-  audioBase64: string,
-  ext: string,
-  mimeType: string,
-  prompt: string,
-): Promise<string> {
-  if (!ENV.forgeApiUrl || !ENV.forgeApiKey) {
-    throw new Error(
-      "STT API not configured. Set BUILT_IN_FORGE_API_URL and BUILT_IN_FORGE_API_KEY in .env " +
-        "(supports OpenAI, Groq, or any OpenAI-compatible Whisper API)",
+  constructor(
+    readonly sessionId: string,
+    sampleRate: number,
+    firstAudioSequence: number,
+    private readonly onExpire: () => void,
+  ) {
+    this.orderedAudio = new OrderedAudioFrames(firstAudioSequence);
+    this.recognizer = new PersistentArabicRecognizer(
+      sessionId,
+      sampleRate,
+      (message) => this.sender(message),
     );
   }
 
-  const audioBuffer = Buffer.from(audioBase64, "base64");
-  if (audioBuffer.length < 500) {
-    console.log("[voice-ws] ⚠️ chunk too small (silence), skipping");
-    return "";
+  async bind(sender: SendMessage): Promise<void> {
+    if (this.retentionTimer) clearTimeout(this.retentionTimer);
+    this.retentionTimer = null;
+    this.sender = sender;
+    await this.recognizer.start();
+    this.sender({
+      type: "pcm_ready",
+      sessionId: this.sessionId,
+      acknowledgedAudioSequence: this.recognizer.acknowledgedSequence(),
+    });
+    this.recognizer.replayTranscriptEvents();
   }
 
-  const mimeMap: Record<string, string> = {
-    m4a: "audio/m4a",
-    mp4: "audio/mp4",
-    webm: "audio/webm",
-    wav: "audio/wav",
-    mp3: "audio/mpeg",
-    ogg: "audio/ogg",
-  };
-  const resolvedMime = mimeMap[ext] ?? mimeType ?? "audio/m4a";
-  const blob = new Blob([audioBuffer], { type: resolvedMime });
-
-  const models = ["whisper-large-v3-turbo", "whisper-large-v3", "whisper-1"];
-  for (const model of models) {
-    try {
-      const fd = new FormData();
-      fd.append("file", blob, `audio.${ext}`);
-      fd.append("model", model);
-      fd.append("language", "ar");
-      // verbose_json gives us no_speech_prob per segment for silence detection
-      fd.append("response_format", "verbose_json");
-      fd.append("prompt", prompt);
-      fd.append("temperature", "0"); // 0 = deterministic, no guessing
-
-      const baseUrl = ENV.forgeApiUrl.endsWith("/")
-        ? ENV.forgeApiUrl
-        : `${ENV.forgeApiUrl}/`;
-      const endpoint = new URL("v1/audio/transcriptions", baseUrl).toString();
-
-      console.log(
-        `[voice-ws] 🔊 transcribing with ${model} (${Math.round(audioBuffer.length / 1024)}KB)`,
-      );
-
-      const res = await fetch(endpoint, {
-        method: "POST",
-        headers: {
-          authorization: `Bearer ${ENV.forgeApiKey}`,
-          "Accept-Encoding": "identity",
-        },
-        body: fd,
-        signal: AbortSignal.timeout(10000),
+  receive(message: PcmMessage): void {
+    const result = this.orderedAudio.receive({
+      audioSequence: message.audioSequence,
+      audio: message.audio,
+    });
+    if (result.duplicate) {
+      this.sender({
+        type: "audio_ack",
+        sessionId: this.sessionId,
+        audioSequence: this.recognizer.acknowledgedSequence(),
       });
+      return;
+    }
+    for (const frame of result.ready) this.recognizer.accept(frame);
+  }
 
-      if (!res.ok) {
-        const errText = await res.text();
-        if (
-          res.status === 404 ||
-          errText.includes("model") ||
-          errText.includes("not found")
-        ) {
-          console.warn(`[voice-ws] model ${model} not available, trying next`);
-          continue;
-        }
-        throw new Error(`Whisper ${res.status}: ${errText}`);
-      }
+  stop(): void {
+    this.recognizer.stop();
+  }
 
-      const data = (await res.json()) as {
-        text?: string;
-        segments?: {
-          no_speech_prob?: number;
-          avg_logprob?: number;
-          compression_ratio?: number;
-          text?: string;
-        }[];
-      };
-      const raw = data.text?.trim() ?? "";
-      console.log(`[voice-ws] ✅ ${model} → "${raw}"`);
-
-      const text = cleanTranscript(raw, data.segments);
-      if (text !== raw) {
-        console.log(`[voice-ws] 🧹 cleaned → "${text}"`);
-      }
-      return text;
-    } catch (err: any) {
-      const cause = err.cause?.message ?? err.cause?.code ?? "";
-      if (
-        err.name === "TimeoutError" ||
-        err.message?.includes("not available")
-      ) {
-        console.warn(`[voice-ws] ${model} timeout/unavailable, trying next`);
-        continue;
-      }
-      // "fetch failed" is a generic network error — log the root cause for debugging
-      if (err.message === "fetch failed") {
-        console.error(
-          `[voice-ws] ❌ fetch failed for ${model} — cause: ${cause || "unknown"}`,
-        );
-        throw new Error(`fetch failed (${cause || "network error"})`);
-      }
-      throw err;
-    }
+  unbind(sender: SendMessage): void {
+    if (this.sender !== sender) return;
+    this.sender = () => {};
+    if (this.retentionTimer) clearTimeout(this.retentionTimer);
+    this.retentionTimer = setTimeout(() => {
+      this.recognizer.close();
+      this.onExpire();
+    }, SESSION_RETENTION_MS);
+  }
+
+  closeSoon(): void {
+    if (this.retentionTimer) clearTimeout(this.retentionTimer);
+    this.retentionTimer = setTimeout(() => {
+      this.recognizer.close();
+      this.onExpire();
+    }, 5000);
   }
-  throw new Error("All Whisper models unavailable");
 }
 
-// ─── WebSocket Server ────────────────────────────────────────────────────────
+const sessions = new Map<string, LiveRecognitionSession>();
 
-function sendJson(ws: WebSocket, msg: Record<string, any>) {
-  if (ws.readyState === WebSocket.OPEN) {
-    ws.send(JSON.stringify(msg));
-  }
+function getOrCreateSession(message: PcmStartMessage): LiveRecognitionSession {
+  const existing = sessions.get(message.sessionId);
+  if (existing) return existing;
+
+  const session = new LiveRecognitionSession(
+    message.sessionId,
+    message.sampleRate ?? 16000,
+    message.firstAudioSequence ?? 0,
+    () => sessions.delete(message.sessionId),
+  );
+  sessions.set(message.sessionId, session);
+  return session;
 }
 
-export function registerVoiceWebSocket(httpServer: Server) {
-  const wss = new WebSocketServer({
+export function registerVoiceWebSocket(httpServer: Server): void {
+  const server = new WebSocketServer({
     server: httpServer,
     path: "/ws/voice",
     perMessageDeflate: false,
   });
 
-  // Log STT configuration on startup
-  if (ENV.forgeApiUrl && ENV.forgeApiKey) {
-    console.log(
-      `[voice-ws] ✅ STT API configured: ${ENV.forgeApiUrl} (server-side Whisper)`,
-    );
-  } else {
-    console.warn(
-      "[voice-ws] ⚠️ STT API NOT configured! Set BUILT_IN_FORGE_API_URL and BUILT_IN_FORGE_API_KEY in .env",
-    );
-    console.warn(
-      "[voice-ws]   Free option: Groq API → https://console.groq.com (whisper-large-v3-turbo)",
-    );
-    console.warn(
-      "[voice-ws]   Set BUILT_IN_FORGE_API_URL=https://api.groq.com/openai",
-    );
-    console.warn("[voice-ws]   Set BUILT_IN_FORGE_API_KEY=gsk_...");
-  }
-
-  wss.on("connection", (ws) => {
-    console.log("[voice-ws] 🔌 client connected");
-
-    const send = (msg: Record<string, any>) => sendJson(ws, msg);
-    const session = new VoiceSession(ws, send);
-
-    send({ type: "ready" });
+  server.on("connection", (socket) => {
+    const boundSessionIds = new Set<string>();
+    const send: SendMessage = (message) => sendJson(socket, message);
+    send({ type: "ready", protocolVersion: 2 });
 
-    ws.on("message", (raw) => {
-      let msg: ClientMessage;
+    socket.on("message", (raw) => {
+      let message: Record<string, any>;
       try {
-        msg = JSON.parse(raw.toString());
+        message = JSON.parse(raw.toString());
       } catch {
         return;
       }
 
-      console.log(
-        `[voice-ws] ← ${msg.type}${(msg as any).chunkId != null ? ` #${(msg as any).chunkId}` : ""}`,
-      );
-
-      if (msg.type === "stop") {
-        session.reset();
+      if (message.type === "ping") {
+        send({ type: "pong" });
         return;
       }
 
-      // Pre-recognized text (from Web Speech API or legacy clients)
-      if (msg.type === "text" && msg.text) {
-        session.processText(msg.chunkId, msg.text);
+      if (message.type === "pcm_start") {
+        const start = message as PcmStartMessage;
+        if (!start.sessionId) {
+          send({ type: "error", message: "sessionId is required" });
+          return;
+        }
+        const session = getOrCreateSession(start);
+        boundSessionIds.add(start.sessionId);
+        void session.bind(send).catch((error) => {
+          send({
+            type: "error",
+            sessionId: start.sessionId,
+            message:
+              error instanceof Error ? error.message : String(error),
+          });
+        });
         return;
       }
 
-      // Raw audio chunk — transcribe with Whisper
-      if (msg.type === "chunk" && msg.audio) {
-        // Check STT config before accepting chunks
-        if (!ENV.forgeApiUrl || !ENV.forgeApiKey) {
+      if (message.type === "pcm") {
+        const pcm = message as PcmMessage;
+        const session = sessions.get(pcm.sessionId);
+        if (!session) {
           send({
             type: "error",
-            message:
-              "Server STT not configured. Set BUILT_IN_FORGE_API_URL and BUILT_IN_FORGE_API_KEY.",
-            chunkId: msg.chunkId,
+            sessionId: pcm.sessionId,
+            message: "Unknown PCM session; send pcm_start first",
           });
           return;
         }
+        session.receive(pcm);
+        return;
+      }
 
-        const mimeType = msg.mimeType ?? "audio/m4a";
-        const ext = mimeType.split("/")[1]?.split(";")[0] ?? "m4a";
-        session.enqueue({
-          chunkId: msg.chunkId,
-          audioBase64: msg.audio,
-          mimeType,
-          ext,
+      if (message.type === "pcm_stop") {
+        const stop = message as PcmStopMessage;
+        const session = sessions.get(stop.sessionId);
+        session?.stop();
+        session?.closeSoon();
+        return;
+      }
+
+      if (message.type === "chunk") {
+        send({
+          type: "error",
+          message:
+            "Recorded audio chunks are no longer supported; stream PCM continuously.",
         });
       }
     });
 
-    ws.on("close", () => {
-      console.log("[voice-ws] 🔌 client disconnected");
-      session.reset();
-    });
-    ws.on("error", (err) => {
-      console.error("[voice-ws] ❌ socket error:", err);
+    socket.on("close", () => {
+      for (const sessionId of boundSessionIds) {
+        sessions.get(sessionId)?.unbind(send);
+      }
     });
   });
 
   console.log(
-    "[voice-ws] ✅ WebSocket registered at /ws/voice (server-side STT pipeline)",
+    "[voice-ws] persistent PCM WebSocket registered at /ws/voice",
   );
 }

shared/realtime-voice-protocol.ts

@@ -0,0 +1,212 @@
+export interface BufferedAudioFrame {
+  audioSequence: number;
+  audio: string;
+}
+
+/**
+ * Bounded replay buffer. Frames remain here until the server acknowledges
+ * them, which makes reconnect replay deterministic and idempotent.
+ */
+export class AudioReplayBuffer {
+  private frames: BufferedAudioFrame[] = [];
+
+  constructor(private readonly capacity: number) {
+    if (!Number.isInteger(capacity) || capacity < 1) {
+      throw new Error("AudioReplayBuffer capacity must be a positive integer");
+    }
+  }
+
+  push(frame: BufferedAudioFrame): BufferedAudioFrame | null {
+    if (
+      this.frames.some(
+        (existing) => existing.audioSequence === frame.audioSequence,
+      )
+    ) {
+      return null;
+    }
+
+    this.frames.push(frame);
+    this.frames.sort((a, b) => a.audioSequence - b.audioSequence);
+    return this.frames.length > this.capacity ? this.frames.shift()! : null;
+  }
+
+  acknowledge(audioSequence: number): void {
+    this.frames = this.frames.filter(
+      (frame) => frame.audioSequence > audioSequence,
+    );
+  }
+
+  unacknowledged(): BufferedAudioFrame[] {
+    return [...this.frames];
+  }
+
+  firstSequence(): number | null {
+    return this.frames[0]?.audioSequence ?? null;
+  }
+
+  clear(): void {
+    this.frames = [];
+  }
+
+  get size(): number {
+    return this.frames.length;
+  }
+}
+
+/**
+ * Turns potentially duplicated or reordered frames back into one contiguous
+ * stream. WebSocket normally preserves order, but reconnect replay can overlap
+ * with frames already accepted by the server.
+ */
+export class OrderedAudioFrames {
+  private pending = new Map<number, BufferedAudioFrame>();
+
+  constructor(
+    private nextSequence: number,
+    private readonly capacity = 600,
+  ) {}
+
+  receive(frame: BufferedAudioFrame): {
+    duplicate: boolean;
+    ready: BufferedAudioFrame[];
+  } {
+    if (
+      frame.audioSequence < this.nextSequence ||
+      this.pending.has(frame.audioSequence)
+    ) {
+      return { duplicate: true, ready: [] };
+    }
+
+    this.pending.set(frame.audioSequence, frame);
+    if (this.pending.size > this.capacity) {
+      const highest = Math.max(...this.pending.keys());
+      this.pending.delete(highest);
+    }
+
+    const ready: BufferedAudioFrame[] = [];
+    while (this.pending.has(this.nextSequence)) {
+      ready.push(this.pending.get(this.nextSequence)!);
+      this.pending.delete(this.nextSequence);
+      this.nextSequence++;
+    }
+    return { duplicate: false, ready };
+  }
+
+  expectedSequence(): number {
+    return this.nextSequence;
+  }
+}
+
+export interface RealtimeTranscriptEvent {
+  sessionId: string;
+  utteranceId: string;
+  utteranceIndex: number;
+  revision: number;
+  text: string;
+}
+
+export interface TranscriptUpdate {
+  accepted: boolean;
+  confirmedText: string;
+  interimText: string;
+  appendedFinals: string[];
+}
+
+/**
+ * Applies recognizer revisions while keeping exactly one replaceable interim
+ * hypothesis. Finals are buffered until every earlier utterance is present.
+ */
+export class TranscriptAccumulator {
+  private latestRevision = new Map<string, number>();
+  private finalized = new Set<string>();
+  private pendingFinals = new Map<number, RealtimeTranscriptEvent>();
+  private nextFinalIndex = 0;
+  private activeInterim: RealtimeTranscriptEvent | null = null;
+  private confirmedParts: string[] = [];
+  private sessionId: string | null = null;
+
+  reset(sessionId?: string): void {
+    this.latestRevision.clear();
+    this.finalized.clear();
+    this.pendingFinals.clear();
+    this.nextFinalIndex = 0;
+    this.activeInterim = null;
+    this.confirmedParts = [];
+    this.sessionId = sessionId ?? null;
+  }
+
+  /** Start a replacement audio session without discarding confirmed text. */
+  continueSession(sessionId?: string): void {
+    const confirmedParts = [...this.confirmedParts];
+    this.reset(sessionId);
+    this.confirmedParts = confirmedParts;
+  }
+
+  applyInterim(event: RealtimeTranscriptEvent): TranscriptUpdate {
+    if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
+    if (this.finalized.has(event.utteranceId)) return this.snapshot(false);
+
+    const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
+    if (event.revision <= previousRevision) return this.snapshot(false);
+    if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);
+
+    this.latestRevision.set(event.utteranceId, event.revision);
+    this.activeInterim = event;
+    return this.snapshot(true);
+  }
+
+  applyFinal(event: RealtimeTranscriptEvent): TranscriptUpdate {
+    if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
+    if (this.finalized.has(event.utteranceId)) return this.snapshot(false);
+
+    const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
+    if (event.revision < previousRevision) return this.snapshot(false);
+    if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);
+
+    const pending = this.pendingFinals.get(event.utteranceIndex);
+    if (pending && pending.revision >= event.revision) {
+      return this.snapshot(false);
+    }
+
+    this.latestRevision.set(event.utteranceId, event.revision);
+    this.pendingFinals.set(event.utteranceIndex, event);
+
+    const appendedFinals: string[] = [];
+    while (this.pendingFinals.has(this.nextFinalIndex)) {
+      const next = this.pendingFinals.get(this.nextFinalIndex)!;
+      this.pendingFinals.delete(this.nextFinalIndex);
+      this.finalized.add(next.utteranceId);
+      const text = next.text.replace(/\s+/g, " ").trim();
+      if (text) {
+        this.confirmedParts.push(text);
+        appendedFinals.push(text);
+      }
+      if (this.activeInterim?.utteranceId === next.utteranceId) {
+        this.activeInterim = null;
+      }
+      this.nextFinalIndex++;
+    }
+
+    return this.snapshot(true, appendedFinals);
+  }
+
+  private acceptSession(sessionId: string): boolean {
+    if (this.sessionId === null) {
+      this.sessionId = sessionId;
+      return true;
+    }
+    return this.sessionId === sessionId;
+  }
+
+  private snapshot(
+    accepted: boolean,
+    appendedFinals: string[] = [],
+  ): TranscriptUpdate {
+    return {
+      accepted,
+      confirmedText: this.confirmedParts.join(" "),
+      interimText: this.activeInterim?.text.trim() ?? "",
+      appendedFinals,
+    };
+  }
+}