diff --git a/CLAUDE.md b/CLAUDE.md index 2b2ad4a..11b0e9a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -197,9 +197,10 @@ fast enough for one-subprocess-per-segment. Switching models is a runtime flag Three concurrency/signal-handling bugs were found and fixed via real testing (not just code review): -- `transcribeTrack` was using the SIGINT-canceled context for still-queued transcriptions +- `transcribeWorker` (then still per-track, named `transcribeTrack`) was using the + SIGINT-canceled context for still-queued transcriptions (fixed with a separate background context for ASR calls). -- `main()` wasn't waiting for `transcribeTrack` goroutines to finish before exiting (fixed +- `main()` wasn't waiting for the ASR worker goroutine(s) to finish before exiting (fixed with a `sync.WaitGroup`). - `internal/capture.Stop()` called `cmd.Wait()` before the pipe-reading goroutines had finished draining, which `os/exec` docs call out as incorrect and can truncate the last bit @@ -242,6 +243,27 @@ review): waiting for in-flight transcriptions..." message and a second Ctrl+C now force-exits immediately instead of making the user wait through the full drain. +**ASR memory: single shared worker instead of one per track (2026-08-10).** The 1.7B model +("large" in `transcriptor-ui`) has no quantized-weights option — `qwen_asr` only supports +BF16 (see `third_party/qwen-asr/README.md`, "Memory Requirements": ~6.9GiB static footprint +for the 1.7B model). Before this fix, `cmd/transcriptor-ai/main.go` ran one `transcribeTrack` +goroutine per track (system, mic), each independently spawning `qwen_asr` subprocesses — so +whenever both tracks had speech at the same time, two 1.7B instances ran concurrently, +measured peaking at ~13GiB combined, over half of the 24GB target machine's unified memory +(see "Target hardware"). Fixed by merging both tracks onto one shared worker/channel +(`asrSegs`, `transcribeWorker` — renamed from `transcribeTrack` since it's no longer +per-track): `dispatch()` now sends every enabled track's segments onto the same channel +instead of two separate ones, so `qwen_asr` invocations are strictly serialized regardless of +which track they came from. Per-track chronological order is preserved (VAD's own output +channel is already chronological across both tracks; the worker just drains it in that +order). Trade-off: if both tracks talk at once, one track's transcription now waits for the +other's to finish instead of running in parallel — accepted, since the memory pressure was +the more pressing constraint on the 24GB target machine. Verified empirically, not just "it +compiles": ran the real pipeline with both tracks capturing simultaneous audio (a video +through system output, `say` for spoken input) and polled `pgrep -x qwen_asr` at 0.3s +intervals throughout — max concurrent `qwen_asr` processes observed was 1, confirmed never 2, +while both tracks still produced correctly-ordered transcript segments. + Not yet implemented: post-processing (deferred, see "Out of scope for now"), and anything from `audiotee`'s CONTEXT.md §6.4-adjacent concerns beyond what's listed here. diff --git a/cmd/transcriptor-ai/main.go b/cmd/transcriptor-ai/main.go index 44d62aa..dd9b0aa 100644 --- a/cmd/transcriptor-ai/main.go +++ b/cmd/transcriptor-ai/main.go @@ -4,11 +4,13 @@ // architecture and the reasoning behind it. // // This is a capture+VAD+ASR+SSE test harness: it runs audiotee, feeds -// detected speech segments to qwen_asr (one subprocess call per segment, one -// worker per track), and publishes the results both to stdout and to an SSE -// endpoint (`curl -N http:///events` is the terminal client for -// now — see CLAUDE.md on why SSE, not WebSocket). Post-processing (fuzzy -// glossary correction, LLM reread) isn't wired in yet. +// detected speech segments (from both tracks, chronologically) to a single +// shared qwen_asr worker (one subprocess call per segment), and publishes +// the results both to stdout and to an SSE endpoint (`curl -N +// http:///events` is the terminal client for now — see +// CLAUDE.md on why SSE, not WebSocket, and why one shared worker rather +// than one per track). Post-processing (fuzzy glossary correction, LLM +// reread) isn't wired in yet. package main import ( @@ -121,30 +123,30 @@ func run() error { detector := vad.New(vad.DefaultConfig(sampleRate)) segments := detector.Run(tee) - // Two independent transcribers (2 tracks, 2 ASR engines — see - // CLAUDE.md), each processing its own track's segments sequentially. A - // fresh qwen_asr subprocess per segment is fast enough (~0.15-0.2x - // realtime measured) that this doesn't need to be more concurrent than - // that. + // A single shared worker across both tracks, not one per track. + // qwen_asr has no quantized-weights option (BF16 only — see + // third_party/qwen-asr/README.md "Memory Requirements"); its static + // footprint alone is ~6.9GiB for the 1.7B model. Two per-track workers + // meant two concurrent qwen_asr subprocesses whenever both tracks had + // speech at once, doubling peak RSS to ~13GiB — over half the 24GB + // target machine's unified memory (see "Target hardware" in + // CLAUDE.md). Serializing onto one worker caps peak RSS to ~1 model + // instance, trading it for latency: if both tracks talk at the same + // time, one track's segment now waits for the other's to finish + // transcribing instead of running in parallel. Per-track order is + // still preserved, since `segments` (VAD's output) is itself already + // chronological and this worker drains it in that same order. transcriber := asr.New(asr.Config{BinaryPath: *asrBinary, ModelDir: *asrModelDir, Prompt: *prompt}) - // transcribeTrack uses its own background context, not the + // transcribeWorker uses its own background context, not the // shutdown-signal ctx above: segments still queued when Ctrl+C arrives // (including the ones flushed by c.Stop() below) must still get // transcribed during the graceful drain, not fail with "context // canceled" because the shutdown signal already canceled ctx. asrCtx := context.Background() - var systemSegs, micSegs chan vad.Segment + asrSegs := make(chan vad.Segment, 16) var wg sync.WaitGroup - if *captureSystem { - systemSegs = make(chan vad.Segment, 8) - wg.Add(1) - go func() { defer wg.Done(); transcribeTrack(asrCtx, transcriber, systemSegs, broadcaster, fileWriter) }() - } - if *captureMic { - micSegs = make(chan vad.Segment, 8) - wg.Add(1) - go func() { defer wg.Done(); transcribeTrack(asrCtx, transcriber, micSegs, broadcaster, fileWriter) }() - } + wg.Add(1) + go func() { defer wg.Done(); transcribeWorker(asrCtx, transcriber, asrSegs, broadcaster, fileWriter) }() loop: for { @@ -154,7 +156,7 @@ loop: break loop } logSegment(seg) - dispatch(seg, systemSegs, micSegs) + dispatch(seg, asrSegs, *captureSystem, *captureMic) case err := <-c.Errs(): fmt.Fprintln(os.Stderr, "capture error:", err) case err := <-httpErrs: @@ -182,17 +184,9 @@ loop: // Drain any segments flushed by Stop() closing the chunk stream. for seg := range segments { logSegment(seg) - dispatch(seg, systemSegs, micSegs) - } - - // Closing a nil channel panics — only close the ones actually created - // (i.e. for tracks that were enabled). - if systemSegs != nil { - close(systemSegs) - } - if micSegs != nil { - close(micSegs) + dispatch(seg, asrSegs, *captureSystem, *captureMic) } + close(asrSegs) wg.Wait() // let in-flight/queued transcriptions finish before exiting, and // before their results are published — otherwise a connected SSE client // would miss the tail end of the transcript. @@ -204,21 +198,19 @@ loop: return nil } -// dispatch routes seg to the channel for its track. Sending to a nil -// channel (a disabled track — see -capture-system/-capture-mic) blocks -// forever in Go, so it must be checked, not just sent to unconditionally; -// a disabled track's segments are silently dropped (VAD still runs on -// every track's chunks regardless — negligible cost — but a disabled -// track never reaches ASR). -func dispatch(seg vad.Segment, systemSegs, micSegs chan<- vad.Segment) { +// dispatch sends seg to the single shared ASR worker if its track is +// enabled. A disabled track's segments are silently dropped (VAD still +// runs on every track's chunks regardless — negligible cost — but a +// disabled track never reaches ASR). +func dispatch(seg vad.Segment, asrSegs chan<- vad.Segment, captureSystem, captureMic bool) { switch seg.Track { case capture.TrackSystem: - if systemSegs != nil { - systemSegs <- seg + if captureSystem { + asrSegs <- seg } case capture.TrackMic: - if micSegs != nil { - micSegs <- seg + if captureMic { + asrSegs <- seg } } } @@ -238,7 +230,7 @@ func httpAddrForDisplay(addr string) string { return addr } -func transcribeTrack(ctx context.Context, t *asr.Transcriber, segs <-chan vad.Segment, b *transcript.Broadcaster, fw *transcript.FileWriter) { +func transcribeWorker(ctx context.Context, t *asr.Transcriber, segs <-chan vad.Segment, b *transcript.Broadcaster, fw *transcript.FileWriter) { for seg := range segs { text, err := t.Transcribe(ctx, seg.Data) if err != nil {