From 9eab8660b6f737bab115d3413c5af82c21ba81af Mon Sep 17 00:00:00 2001 From: Maksym Naboka Date: Tue, 28 Jul 2026 18:12:51 -0700 Subject: [PATCH] fix(parakeet): restart capture on mid-session input-device configuration change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reported: mid-dictation, Parakeet stops producing words (~a minute in), then "comes back" later — with the gap permanently missing from the transcript. Investigation found NO regression in the decode path: FluidAudio has been pinned at 0.15.5 since the original integration (#163), the recent Parakeet changes (#218, #222) only touch callback snapshots and the final-transcript artifact strip, and a real-model harness run (parakeet-unified-320ms over 2.5 min of continuous fixture speech) sustained RTFx 5.5 with zero dropouts and no transcript gaps. The decoder does not go deaf on its own. What CAN produce exactly this symptom is a mid-session AVAudioEngineConfigurationChange: the input device disconnects, switches, or renegotiates its format, AVAudioEngine stops rendering, and — because the streaming engines never observe the notification — the session keeps showing "Listening…" while capturing NOTHING. Every word until the user gives up and re-dictates is silently lost. AudioRecorder has handled precisely this since the AirPods-disconnect hang ("the session keeps showing 'recording' while capturing nothing"); the streaming engines were never given the observer, and Parakeet only became the default (and long dictations routine) recently — which is why it reads as a new bug. Unlike the recorder's fail-only handling, Parakeet can recover seamlessly: the decode session and feed stream survive the mic teardown, so the engine rebuilds a routed AVAudioEngine + tap onto the SAME continuation and the dictation continues — everything already captured stays, and only the glitch itself is lost. If the rebuild fails (no input device left), the session fails LOUDLY through the error path instead of eating speech. The observer is selector-based because the block API takes a hard @Sendable closure and every capture the restart needs is non-Sendable. The decision gates live in core (CaptureConfigChangePolicy) so swift test pins them: restart only when the session generation matches and no stop ran; stale/superseded notifications touch nothing. Route/format setup is extracted into makeRoutedEngine()/startCapture() shared by session start and restart. swift test: 1904 passing (4 new). ./build.sh green, no new warnings. Caveat (same as #222): the restart path only exercises against a real device change with a live mic — worth a live pass (unplug/replug the pinned mic mid-dictation) before the next release. Co-Authored-By: Claude Fable 5 --- .../Services/CaptureConfigChangePolicy.swift | 44 ++++ .../Services/ParakeetStreamingEngine.swift | 213 ++++++++++++++---- Package.swift | 1 + .../CaptureConfigChangePolicyTests.swift | 47 ++++ 4 files changed, 264 insertions(+), 41 deletions(-) create mode 100644 OpenWhisp/Services/CaptureConfigChangePolicy.swift create mode 100644 Tests/OpenWhispCoreTests/CaptureConfigChangePolicyTests.swift diff --git a/OpenWhisp/Services/CaptureConfigChangePolicy.swift b/OpenWhisp/Services/CaptureConfigChangePolicy.swift new file mode 100644 index 0000000..30e301e --- /dev/null +++ b/OpenWhisp/Services/CaptureConfigChangePolicy.swift @@ -0,0 +1,44 @@ +import Foundation + +/// Pure decision logic for a mid-session `AVAudioEngineConfigurationChange` on a +/// streaming capture engine (Parakeet/WhisperKit/Apple Speech own their own +/// `AVAudioEngine` taps). +/// +/// When the input device is disconnected, switched, or renegotiates its format +/// mid-dictation, AVAudioEngine STOPS rendering. Without a reaction the session +/// keeps showing "Listening…" while capturing nothing — every word spoken after +/// the change is silently lost until the user gives up and starts a new session. +/// `AudioRecorder` has handled this since the AirPods-disconnect hang; the +/// streaming engines never got the observer. +/// +/// Foundation-only (no AVFoundation) so it lives in OpenWhispCore and +/// `swift test` pins the gates; the notification observer + tap rebuild live +/// app-side in the engine. +enum CaptureConfigChangePolicy { + + /// What the engine should do when the notification fires. + enum Action: Equatable { + /// Stale notification — a stop or a newer session already superseded the + /// capture this observer was armed for. Touch nothing: the successor + /// session owns the mic now. + case ignore + /// The session is still live: rebuild the engine + tap and keep feeding + /// the SAME decode session. Audio captured before the change is already + /// in the feed stream, so the transcript loses only the glitch itself. + case restartCapture + } + + /// `generationMatches` is the engine's session-generation fence (the same + /// gate the partial/EOU callbacks use); `didStop` is the engine's stop flag. + static func action(generationMatches: Bool, didStop: Bool) -> Action { + (generationMatches && !didStop) ? .restartCapture : .ignore + } + + /// User-facing error when the capture rebuild failed (no input device left + /// after the change, or the tap/engine start threw). Surfaced through the + /// session's error path so the session fails LOUDLY instead of silently + /// eating speech for its remainder. + static func restartFailedMessage(underlying: String) -> String { + "Microphone changed mid-dictation and capture couldn't be restarted: \(underlying)" + } +} diff --git a/OpenWhisp/Services/ParakeetStreamingEngine.swift b/OpenWhisp/Services/ParakeetStreamingEngine.swift index 40812ea..af15242 100644 --- a/OpenWhisp/Services/ParakeetStreamingEngine.swift +++ b/OpenWhisp/Services/ParakeetStreamingEngine.swift @@ -21,7 +21,7 @@ import AVFoundation /// **Build:** real implementation only under `#if PARAKEET` (build.sh /// PARAKEET=1); otherwise a stub that reports unavailability, so the default /// build is unaffected. See docs/PARAKEET.md. -final class ParakeetStreamingEngine: StreamingTranscriptionEngine { +final class ParakeetStreamingEngine: NSObject, StreamingTranscriptionEngine { var onPartial: ((String) -> Void)? var onFinal: ((String) -> Void)? var onError: ((String) -> Void)? @@ -48,6 +48,7 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { init(variantID: String = ParakeetCatalog.defaultVariantID) { self.variantID = ParakeetCatalog.normalize(variantID) + super.init() } #if PARAKEET @@ -67,6 +68,16 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { /// exists ONLY so teardown (runStop / feed-loop error) can finish() it. @MainActor private var feedContinuation: AsyncStream.Continuation? @MainActor private var feedTask: Task? + /// What the config-change restart needs to rebuild capture into the live + /// session (see `armConfigChangeObserver`). Non-nil exactly while the + /// observer is armed; cleared on every teardown path. + @MainActor private var restartContext: RestartContext? + + private struct RestartContext { + let continuation: AsyncStream.Continuation + let callbacks: SessionCallbacks + let myGeneration: Int + } @MainActor private var lastPartial = "" @MainActor private var didStop = false @@ -140,34 +151,14 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { return } - let engine = AVAudioEngine() - let input = engine.inputNode - - // Route to the selected input device BEFORE reading the format. - // A pinned but disconnected device falls back to the system default; - // a connected device that fails to route stays a hard error (same - // policy as Apple Speech). - switch AudioInputRoutingPolicy.decide( - microphoneID: selectedDeviceID, - deviceResolved: AudioInputRouter.canResolve(uid: selectedDeviceID) - ) { - case .systemDefault: - break - case .useDevice(let uid): - guard let device = AudioInputRouter.resolve(uid: uid), - AudioInputRouter.apply(device, to: engine) else { - callbacks.error?(AudioInputRoutingPolicy.unresolvedMessage(uid: uid)) - return - } - case .fallbackToDefault(let uid): - NSLog("[ParakeetStreamingEngine] pinned mic '%@' disconnected — capturing system default", uid) - } - - let format = input.outputFormat(forBus: 0) - // 0 Hz / 0 ch (no input device) would make installTap raise an ObjC - // NSException that Swift can't catch — guard it out. - guard format.sampleRate > 0, format.channelCount > 0 else { - callbacks.error?("No audio input device available.") + // Route + format-check BEFORE the (potentially slow) model load, so a + // missing input device fails fast instead of after a first-run download. + let engine: AVAudioEngine + let format: AVAudioFormat + do { + (engine, format) = try makeRoutedEngine() + } catch let setupError as CaptureSetupError { + callbacks.error?(setupError.message) return } @@ -241,6 +232,7 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { // The consumer is dead — tear the mic down too, or the // tap keeps the mic hot and the continuation buffers // yielded audio unbounded until the next session. + self.removeConfigChangeObserver() if let audioEngine = self.audioEngine { audioEngine.inputNode.removeTap(onBus: 0) audioEngine.stop() @@ -253,18 +245,10 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { } } - // The tap closure runs on an audio thread but touches no session - // state — AsyncStream.Continuation.yield is Sendable + thread-safe, - // so it yields straight from the audio thread (no main-actor hop - // per buffer) and publishes a level. - input.installTap(onBus: 0, bufferSize: 1024, format: format) { [weak self] buffer, _ in - continuation.yield(buffer) - self?.publishLevel(from: buffer) - } - - audioEngine = engine - engine.prepare() - try engine.start() + try startCapture( + engine: engine, format: format, + continuation: continuation, callbacks: callbacks, + myGeneration: myGeneration) // Tap installed + AVAudioEngine running: audio is flowing into the // feed stream. Everything above (model load/first-run download, // reset, callback wiring) was the arming gap this signal closes. @@ -276,6 +260,152 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { } } + /// Setup failure with a user-facing message (routing refused, no input + /// device). `LocalizedError` so a generic `localizedDescription` still + /// reads as the message. + private struct CaptureSetupError: Error, LocalizedError { + let message: String + var errorDescription: String? { message } + } + + /// Build a fresh AVAudioEngine routed to the pinned (or default) input and + /// return it with its tap format. Shared by session start and the + /// mid-session configuration-change restart. + @MainActor + private func makeRoutedEngine() throws -> (AVAudioEngine, AVAudioFormat) { + let engine = AVAudioEngine() + let input = engine.inputNode + + // Route to the selected input device BEFORE reading the format. + // A pinned but disconnected device falls back to the system default; + // a connected device that fails to route stays a hard error (same + // policy as Apple Speech). + switch AudioInputRoutingPolicy.decide( + microphoneID: selectedDeviceID, + deviceResolved: AudioInputRouter.canResolve(uid: selectedDeviceID) + ) { + case .systemDefault: + break + case .useDevice(let uid): + guard let device = AudioInputRouter.resolve(uid: uid), + AudioInputRouter.apply(device, to: engine) else { + throw CaptureSetupError(message: AudioInputRoutingPolicy.unresolvedMessage(uid: uid)) + } + case .fallbackToDefault(let uid): + NSLog("[ParakeetStreamingEngine] pinned mic '%@' disconnected — capturing system default", uid) + } + + let format = input.outputFormat(forBus: 0) + // 0 Hz / 0 ch (no input device) would make installTap raise an ObjC + // NSException that Swift can't catch — guard it out. + guard format.sampleRate > 0, format.channelCount > 0 else { + throw CaptureSetupError(message: "No audio input device available.") + } + return (engine, format) + } + + /// Install the feed tap on `engine`, start it, and arm the + /// configuration-change observer. Shared by session start and the + /// mid-session restart — both feed the SAME continuation, so the decode + /// session sees one continuous sample stream. + @MainActor + private func startCapture( + engine: AVAudioEngine, + format: AVAudioFormat, + continuation: AsyncStream.Continuation, + callbacks: SessionCallbacks, + myGeneration: Int + ) throws { + // The tap closure runs on an audio thread but touches no session + // state — AsyncStream.Continuation.yield is Sendable + thread-safe, + // so it yields straight from the audio thread (no main-actor hop + // per buffer) and publishes a level. + engine.inputNode.installTap(onBus: 0, bufferSize: 1024, format: format) { [weak self] buffer, _ in + continuation.yield(buffer) + self?.publishLevel(from: buffer) + } + audioEngine = engine + engine.prepare() + try engine.start() + armConfigChangeObserver( + for: engine, continuation: continuation, + callbacks: callbacks, myGeneration: myGeneration) + } + + /// Watch for the input device disconnecting / switching / renegotiating its + /// format mid-session. AVAudioEngine STOPS rendering when that happens, so + /// without this the session keeps showing "Listening…" while capturing + /// nothing — every word after the change is silently lost (AudioRecorder + /// has the same observer; the streaming engines lacked it). Unlike the + /// recorder's fail-only handling, capture is REBUILT onto the same feed + /// stream: the decode session keeps everything already captured and the + /// dictation continues, losing only the glitch itself. Rebuild failure + /// (e.g. no input device left) surfaces on the session error path. + @MainActor + private func armConfigChangeObserver( + for engine: AVAudioEngine, + continuation: AsyncStream.Continuation, + callbacks: SessionCallbacks, + myGeneration: Int + ) { + removeConfigChangeObserver() + restartContext = RestartContext( + continuation: continuation, callbacks: callbacks, myGeneration: myGeneration) + // Selector-based on purpose: the block observer API takes a hard + // `@Sendable` closure, and every capture it needs (self, the session + // callbacks) is non-Sendable — the selector target keeps the compile + // warning-free and the context lives in `restartContext` instead. + NotificationCenter.default.addObserver( + self, + selector: #selector(inputConfigurationDidChange(_:)), + name: .AVAudioEngineConfigurationChange, + object: engine) + } + + @MainActor + private func removeConfigChangeObserver() { + NotificationCenter.default.removeObserver( + self, name: .AVAudioEngineConfigurationChange, object: nil) + restartContext = nil + } + + /// Fires on the notification-posting thread; hop to the main actor where + /// the session state lives. Same hop shape as the partial callback. + @objc private func inputConfigurationDidChange(_ note: Notification) { + DispatchQueue.main.async { + MainActor.assumeIsolated { self.handleInputConfigurationChange() } + } + } + + @MainActor + private func handleInputConfigurationChange() { + // Same session fence as the partial/EOU hops: a stop or a newer session + // owns the mic now — touch nothing. (Teardown also clears the context.) + guard let context = restartContext, CaptureConfigChangePolicy.action( + generationMatches: generation == context.myGeneration, + didStop: didStop + ) == .restartCapture else { return } + NSLog("[Parakeet] input device configuration changed — restarting capture") + removeConfigChangeObserver() + // The session fence held, so audioEngine is the engine the observer was + // armed for (re-arms replace the observer and the context together). + if let stale = audioEngine { + stale.inputNode.removeTap(onBus: 0) + stale.stop() + } + audioEngine = nil + do { + let (fresh, format) = try makeRoutedEngine() + try startCapture( + engine: fresh, format: format, + continuation: context.continuation, callbacks: context.callbacks, + myGeneration: context.myGeneration) + } catch { + context.callbacks.error?(CaptureConfigChangePolicy.restartFailedMessage( + underlying: error.localizedDescription)) + } + } + func stop(cancel: Bool) { MainActor.assumeIsolated { // Snapshot the final callback at ENQUEUE time, mirroring start()'s @@ -308,6 +438,7 @@ final class ParakeetStreamingEngine: StreamingTranscriptionEngine { // Supersede the session so late partial hops are dropped. generation += 1 + removeConfigChangeObserver() if let audioEngine { audioEngine.inputNode.removeTap(onBus: 0) audioEngine.stop() diff --git a/Package.swift b/Package.swift index a30a45a..4b7e013 100644 --- a/Package.swift +++ b/Package.swift @@ -84,6 +84,7 @@ let package = Package( "AudioCapture.swift", "FileAudioCapture.swift", "AudioInputRoutingPolicy.swift", + "CaptureConfigChangePolicy.swift", "OverlayPhase.swift", "LiveChunkPipeline.swift", "ConfigBundle.swift", diff --git a/Tests/OpenWhispCoreTests/CaptureConfigChangePolicyTests.swift b/Tests/OpenWhispCoreTests/CaptureConfigChangePolicyTests.swift new file mode 100644 index 0000000..0bd5c21 --- /dev/null +++ b/Tests/OpenWhispCoreTests/CaptureConfigChangePolicyTests.swift @@ -0,0 +1,47 @@ +import XCTest +@testable import OpenWhispCore + +/// Pins the gates for the mid-session `AVAudioEngineConfigurationChange` +/// recovery (Parakeet streaming): a live session restarts capture; a stale or +/// stopped session ignores the notification. Without the restart, a device +/// disconnect/format renegotiation leaves the session showing "Listening…" +/// while capturing NOTHING — words are silently lost for the session's +/// remainder (the bug this policy exists to close). +final class CaptureConfigChangePolicyTests: XCTestCase { + + func testLiveSessionRestartsCapture() { + XCTAssertEqual( + CaptureConfigChangePolicy.action(generationMatches: true, didStop: false), + .restartCapture + ) + } + + func testStoppedSessionIgnores() { + // stop() ran but the generation hasn't rotated yet (runStop is queued + // behind the chain): the mic is being torn down — don't rebuild it. + XCTAssertEqual( + CaptureConfigChangePolicy.action(generationMatches: true, didStop: true), + .ignore + ) + } + + func testSupersededSessionIgnores() { + // A newer session bumped the generation: the notification belongs to a + // torn-down capture and must not touch the successor's mic. + XCTAssertEqual( + CaptureConfigChangePolicy.action(generationMatches: false, didStop: false), + .ignore + ) + XCTAssertEqual( + CaptureConfigChangePolicy.action(generationMatches: false, didStop: true), + .ignore + ) + } + + func testRestartFailedMessageCarriesUnderlyingReason() { + let message = CaptureConfigChangePolicy.restartFailedMessage( + underlying: "No audio input device available.") + XCTAssertTrue(message.contains("Microphone changed mid-dictation")) + XCTAssertTrue(message.contains("No audio input device available.")) + } +}