fix(parakeet): restart capture on mid-session input-device configuration change

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

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<AVAudioPCMBuffer>.Continuation?
     @MainActor private var feedTask: Task<Void, Never>?
+    /// 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<AVAudioPCMBuffer>.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<AVAudioPCMBuffer>.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<AVAudioPCMBuffer>.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()

Package.swift

@@ -84,6 +84,7 @@ let package = Package(
                 "AudioCapture.swift",
                 "FileAudioCapture.swift",
                 "AudioInputRoutingPolicy.swift",
+                "CaptureConfigChangePolicy.swift",
                 "OverlayPhase.swift",
                 "LiveChunkPipeline.swift",
                 "ConfigBundle.swift",

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