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.")) + } +}