wip: incremental ASR

src/client.ts

@@ -288,10 +288,10 @@ export class RealtimeAsrClient {
 	 * `is_last_result` never flips true, so after commit we wait until no new segment for quietMs.
 	 */
 	async recordMic(opts: {
-				device?: string | null; frameMs?: number;
+		device?: string | null; frameMs?: number; commitMs?: number;
 		quietMs?: number; emptyGraceMs?: number; finalTimeout?: number; tailMs?: number;
 		sampleRate?: number; channels?: number; session?: Record<string, any>;
-		onDelta?: (t: string) => void; onLevel?: (peak: number) => void; onStart?: (d: { device: string | null }) => void;
+		onDelta?: (t: string) => void; onLive?: (live: string) => void; onLevel?: (peak: number) => void; onStart?: (d: { device: string | null }) => void;
 		signal?: AbortSignal;
 	} = {}): Promise<MicRecording> {
 		this.requireKey();
@@ -304,26 +304,57 @@ export class RealtimeAsrClient {
 		handle.send({ type: "transcription_session.update", session: this.sessionConfig(minted, opts.session ?? {}) });
 		await handle.expect(EV.sessionUpdated, 20000);
 
+			const norm = (s: string): string => (s ?? "").replace(/\s+/g, " ").trim();
 		const deltas: string[] = [];
 		const finals: Map<string, any> = new Map();
 		const failures: unknown[] = [];
+		// Per-segment live model. The endpoint only runs its decoder on `commit` (a bare `append` emits
+		// nothing until you commit), so the pump below commits periodically. Each item_id accumulates a
+		// cumulative `.delta` and later a `.completed` whose transcript equals that delta. We render the
+		// longer of (partial, final) per id, in first-seen order — so a segment can never be duplicated
+		// and the live view never stalls even if a `.completed` for it arrives before the next delta.
+		const segOrder: string[] = [];
+		const segPartial: Map<string, string> = new Map();
+		const segFinal: Map<string, string> = new Map();
+		const touchSeg = (id: string): void => { if (!segOrder.includes(id)) segOrder.push(id); };
+		let lastLive = "";
+		const liveText = (): string => {
+			const parts: string[] = [];
+			for (const id of segOrder) {
+				const p = norm(segPartial.get(id) ?? "");
+				const f = norm(segFinal.get(id) ?? "");
+				const best = f.length > p.length ? f : p;   // the cumulative delta already contains the final text
+				if (best) parts.push(best);
+			}
+			return norm(parts.join(" "));
+		};
+		const emitLive = (): void => { const s = liveText(); if (s !== lastLive) { lastLive = s; opts.onLive?.(s); } };
 		const offs = [
-			this.on(EV.delta, e => { const t = e.delta ?? ""; deltas.push(t); rec.partial = t; opts.onDelta?.(t); }),
+			this.on(EV.delta, (e: any) => {
+				const id = e.item_id ?? "_", t = e.delta ?? "";
+				touchSeg(id); segPartial.set(id, t); deltas.push(t);
+				rec.partial = t; opts.onDelta?.(t); emitLive();
+			}),
 			this.on(EV.completed, (e: any) => {
 				const key = e.item_id ?? "_";
 				const prev = finals.get(key);
 				const nw = e.words_info?.words?.length ?? 0, pw = prev?.words_info?.words?.length ?? 0;
 				finals.set(key, prev && pw > nw && prev.transcript ? { ...e, words_info: prev.words_info } : e);
+				const t = norm(e.transcript ?? "");
+				if (t && t.length > (segFinal.get(key)?.length ?? 0)) segFinal.set(key, t);
+				emitLive();
 			}),
 			this.on(EV.failed, (e: any) => failures.push(e)),
-		];
+	];
 
 		let pending: Buffer[] = [], carry: Buffer = Buffer.alloc(0), total = 0, peak = 0, cancelled = false;
 		let drainP: Promise<TranscriptResult> | null = null;
+		let commitTimer: ReturnType<typeof setInterval> | null = null;
 		const rec: MicRecording = {
 			partial: "", bytes: 0, seconds: 0, level: 0, error: null,
+			liveText,
 			stop: () => (drainP ??= drain()),
-			cancel: () => { cancelled = true; try { handle.send?.({ type: "input_audio_buffer.clear" }); } catch {} for (const f of offs) f(); handle.close(); },
+			cancel: () => { cancelled = true; if (commitTimer) { clearInterval(commitTimer); commitTimer = null; } try { handle.send?.({ type: "input_audio_buffer.clear" }); } catch {} for (const f of offs) f(); handle.close(); },
 		};
 
 		function flush(force = false): void {
@@ -351,9 +382,23 @@ export class RealtimeAsrClient {
 
 		opts.onStart?.({ device: opts.device ?? null });
 
+		// Drive the decoder: the server stays silent on `append` alone and only produces `.delta`/`.completed`
+		// once a `commit` is received, so we commit whenever new audio has arrived since the last commit.
+		// Each commit flushes a fresh cumulative `.delta` of everything heard so far — that is what makes the
+		// live transcript actually advance while you speak. drain()'s final commit catches the tail.
+		const commitMs = Math.max(250, opts.commitMs ?? 500);
+		let lastCommitBytes = 0;
+		commitTimer = setInterval(() => {
+			if (cancelled || drainP) return;
+			if (total <= lastCommitBytes) return;   // no new audio since last commit (silence) — skip the empty decode
+			lastCommitBytes = total;
+			try { handle.send({ type: "input_audio_buffer.commit" }); } catch {}
+		}, commitMs);
+
 		function stopCapture(): void { try { child.kill("SIGINT"); } catch {} }
 
 		async function drain(): Promise<TranscriptResult> {
+			if (commitTimer) { clearInterval(commitTimer); commitTimer = null; }
 			stopCapture();
 			// let stdout finish flushing the tail, then force-send the last partial frame
 			await Promise.race([new Promise(r => child.once("close", r)), sleep(400), capEnded.catch(() => 0)]);
@@ -608,6 +653,10 @@ export async function listAudioInputs(): Promise<AudioDevice[]> {
 
 /** `-i none:<device|default>` selects the audio half of the [video]:[audio] AVFoundation spec. */
 function micArgs(device: string | null | undefined, { sampleRate = 16000, channels = 1 }: { sampleRate?: number; channels?: number } = {}): string[] {
+	// Testing / power-user override: ASR_MIC_ARGS replaces the ffmpeg argument list verbatim (the `ffmpeg`
+	// binary itself is still used). e.g. feed a clip in real-time via `-re` as a stand-in for a live mic.
+	const override = (process.env.ASR_MIC_ARGS ?? "").trim();
+	if (override) return override.split(/\s+/g).filter(Boolean);
 	const spec = device && device.trim() && device.trim().toLowerCase() !== "default" ? `none:${device.trim()}` : "none:default";
 	return ["-hide_banner", "-v", "error", "-f", "avfoundation", "-i", spec, "-ac", String(channels), "-ar", String(sampleRate), "-f", "s16le", "-"];
 }
@@ -630,6 +679,8 @@ export interface MicRecording {
 	level: number;
 	/** Capture error surfaced by ffmpeg (missing binary / blocked microphone), if any. */
 	error: string | null;
+	/** Deduped, composited live text: finalized segments + the in-progress cumulative partial. */
+	liveText(): string;
 	/** Stop capture, commit, flush, and resolve with the final transcript. Idempotent. */
 	stop(): Promise<TranscriptResult>;
 	/** Abort without waiting for the final segment (drops the session). */