From 65855ace6c5d4120a7715b53a57a8d99f390bb2d Mon Sep 17 00:00:00 2001 From: Iulian Date: Wed, 9 Sep 2026 20:31:03 +0300 Subject: [PATCH] wip: incremental ASR --- extensions/asr.ts | 31 +++++++++++++----------- src/client.ts | 61 +++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 73 insertions(+), 19 deletions(-) diff --git a/extensions/asr.ts b/extensions/asr.ts index d69aa67..819d046 100644 --- a/extensions/asr.ts +++ b/extensions/asr.ts @@ -309,8 +309,10 @@ export default function asrExtension(pi: ExtensionAPI) { * gives no key-release event for hold-to-talk without the Kitty keyboard protocol. * Shared module state so the /dictate command and the ctrl+shift+d shortcut drive one session. */ - let active: { rec: MicRecording; submit: boolean } | null = null; + let active: { rec: MicRecording; submit: boolean; base: string } | null = null; let starting = false; + /** Compose the composer text from the user's pre-existing draft + the dictated tail (idempotent: never appended). */ + const joinText = (base: string, t: string): string => { const tail = t.trim(); return base ? (tail ? `${base} ${tail}` : base) : tail; }; async function resolveDevice(d?: string | null): Promise { if (!d) return null; @@ -325,23 +327,20 @@ export default function asrExtension(pi: ExtensionAPI) { async function toggleDictation(ctx: ExtensionContext, opts: { device?: string | null; language?: string; submit?: boolean } = {}): Promise { if (!ctx.hasUI) { ctx.ui.notify("dictation needs the interactive TUI", "error"); return; } - // second press while live => stop, flush, paste the final transcript. + // second press while live => stop, flush, finalise the composer with the deduped transcript. if (active) { - const { rec, submit } = active; active = null; + const { rec, submit, base } = active; active = null; ctx.ui.setStatus("asr", undefined); try { const res = await rec.stop(); const text = res.text.trim(); - if (rec.error) { ctx.ui.notify(`dictation error: ${rec.error}`, "error"); return; } - if (!text) { ctx.ui.notify("no speech detected (speak a little longer, or run asr_health)", "warning"); return; } - if (submit) pi.sendUserMessage(text); - else { - const cur = (ctx.ui.getEditorText() ?? "").replace(/\s+$/, ""); - ctx.ui.setEditorText(cur ? `${cur} ${text}` : text); - } + if (rec.error) { ctx.ui.setEditorText(base); ctx.ui.notify(`dictation error: ${rec.error}`, "error"); return; } + if (!text) { ctx.ui.setEditorText(base); ctx.ui.notify("no speech detected (speak a little longer, or run asr_health)", "warning"); return; } + if (submit) { ctx.ui.setEditorText(base); pi.sendUserMessage(text); } + else ctx.ui.setEditorText(joinText(base, text)); // identical to the last live paint — replace, not append ctx.ui.notify(`dictated ${wordCount(text)} words (${res.seconds}s) → ${submit ? "sent" : "prompt"}`, "info"); } catch (e: any) { - rec.cancel(); + rec.cancel(); ctx.ui.setEditorText(base); ctx.ui.notify(`dictation failed: ${e?.message ?? e}`, "error"); } return; @@ -354,13 +353,16 @@ export default function asrExtension(pi: ExtensionAPI) { try { device = await resolveDevice(opts.device); } catch (e: any) { starting = false; ctx.ui.notify(`dictation: ${e?.message ?? e}`, "error"); return; } + const base = (ctx.ui.getEditorText() ?? "").replace(/\s+$/, ""); const client = new RealtimeAsrClient({ language, punctuation: true }); ctx.ui.setStatus("asr", "● listening …"); let rec: MicRecording; try { rec = await client.recordMic({ device, - onDelta: (t) => { if (active) ctx.ui.setStatus("asr", `● ${t || "…listening"}`); }, + // stream word-by-word into the composer: setEditorText REPLACES (never appends), and the client's + // liveText is a per-segment cumulative partial, so each paint just overwrites the growing tail — no dupes. + onLive: (live) => { if (active) { const tail = live.trim(); ctx.ui.setEditorText(joinText(base, tail)); ctx.ui.setStatus("asr", `● ${tail.slice(-70) || "…listening"}`); } }, }); } catch (e: any) { starting = false; ctx.ui.setStatus("asr", undefined); @@ -368,8 +370,9 @@ export default function asrExtension(pi: ExtensionAPI) { return; } starting = false; - active = { rec, submit: !!opts.submit }; - ctx.ui.notify(`🎙 recording${device ? ` from “${device}”` : " (default mic)"} — press ctrl+shift+d or /dictate again to stop`, "info"); + active = { rec, submit: !!opts.submit, base }; + ctx.ui.setEditorText(joinText(base, rec.liveText())); // show current state immediately (deltas are cumulative, so nothing is lost) + ctx.ui.notify(`🎙 recording${device ? ` from “${device}”` : " (default mic)"} — speech streams into the prompt; ctrl+shift+d or /dictate to stop`, "info"); } pi.registerShortcut("ctrl+shift+d", { diff --git a/src/client.ts b/src/client.ts index 74063db..2e7f0f4 100644 --- a/src/client.ts +++ b/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; - 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 { 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 = 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 = new Map(); + const segFinal: Map = 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 | null = null; + let commitTimer: ReturnType | 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 { + 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 { /** `-i none:` 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; /** Abort without waiting for the final segment (drops the session). */