Prevent audio-device changes from stalling dictation hotkeys

Sources/Fluid/Services/ASRService.swift

@@ -665,14 +665,50 @@ final class ASRService: ObservableObject {
         self.audioEngineRetirementDrain.schedule(token)
     }
 
-    /// Route recovery and engine retry paths use this completion barrier so the
-    /// old AVAudioEngine and its AVAudioIOUnit are fully deallocated before a
-    /// replacement can touch Core Audio.
-    private func retireAudioEngineAndWait(reason: String) async {
+    private struct AudioEngineRetirementTimeout: LocalizedError {
+        let reason: String
+
+        var errorDescription: String? {
+            "The previous audio engine did not stop in time (\(self.reason))."
+        }
+    }
+
+    private static let audioEngineRetirementTimeout: TimeInterval = 3
+
+    /// Route recovery and engine retry paths use this bounded completion barrier
+    /// so a wedged AVAudioIOUnit cannot block every future recording attempt.
+    private func retireAudioEngineAndWait(reason: String) async throws {
+        let completed: Bool
         if let token = self.detachAudioEngineForRetirement(reason: reason) {
-            await self.audioEngineRetirementDrain.releaseAndWait(token)
+            completed = await self.audioEngineRetirementDrain.releaseAndWait(
+                token,
+                timeout: Self.audioEngineRetirementTimeout
+            )
         } else {
-            await self.audioEngineRetirementDrain.waitForScheduledReleases()
+            completed = await self.audioEngineRetirementDrain.waitForScheduledReleases(
+                timeout: Self.audioEngineRetirementTimeout
+            )
+        }
+
+        guard completed else {
+            DebugLogger.shared.error(
+                "Audio engine retirement timed out (\(reason))",
+                source: "ASRService"
+            )
+            throw AudioEngineRetirementTimeout(reason: reason)
+        }
+    }
+
+    private func waitForAudioEngineRetirement(reason: String) async throws {
+        let completed = await self.audioEngineRetirementDrain.waitForScheduledReleases(
+            timeout: Self.audioEngineRetirementTimeout
+        )
+        guard completed else {
+            DebugLogger.shared.error(
+                "Audio engine retirement barrier timed out (\(reason))",
+                source: "ASRService"
+            )
+            throw AudioEngineRetirementTimeout(reason: reason)
         }
     }
 
@@ -809,11 +845,6 @@ final class ASRService: ObservableObject {
     }
 
     private func startPreferredAudioCapture() async throws {
-        // A non-route path may have scheduled a fire-and-forget retirement. Do
-        // not let capture startup overlap any final AVAudioEngine release that is
-        // already queued.
-        await self.audioEngineRetirementDrain.waitForScheduledReleases()
-
         let directCaptureEnabled = SettingsStore.shared.experimentalDirectAudioCaptureEnabled
         if directCaptureEnabled,
            self.prepareDirectAudioInputIfPossible(reason: "recording_start"),
@@ -852,7 +883,7 @@ final class ASRService: ObservableObject {
     }
 
     private func startCompatibilityAudioCapture(reason: String) async throws {
-        await self.audioEngineRetirementDrain.waitForScheduledReleases()
+        try await self.waitForAudioEngineRetirement(reason: "compatibility_start:\(reason)")
         self.benchmarkLog("audio_backend kind=av_audio_engine_fallback reason=\(reason)")
         try self.configureSession()
         try await self.startEngine()
@@ -1424,7 +1455,7 @@ final class ASRService: ObservableObject {
             self.audioCapturePipeline.setRecordingEnabled(false)
             self.isRunning = false
             self.stopActiveAudioCapture()
-            await self.retireAudioEngineAndWait(reason: "start_failed")
+            try? await self.retireAudioEngineAndWait(reason: "start_failed")
             DebugLogger.shared.error("Failed to start ASR session: \(error)", source: "ASRService")
 
             // Resume media if we paused it before the failure
@@ -1892,7 +1923,7 @@ final class ASRService: ObservableObject {
         DebugLogger.shared.debug("Audio capture stopped", source: "ASRService")
 
         // Cancel/no-transcription paths stay conservative and retire the engine.
-        await self.retireAudioEngineAndWait(reason: "stop_without_transcription")
+        try? await self.retireAudioEngineAndWait(reason: "stop_without_transcription")
 
         // CRITICAL FIX: Await completion of streaming task AND any pending transcriptions
         // This prevents use-after-free crashes (EXC_BAD_ACCESS) when clearing buffer
@@ -2278,7 +2309,7 @@ final class ASRService: ObservableObject {
                 // If this isn't the last attempt, recreate engine and reconfigure
                 if attempts < 3 {
                     DebugLogger.shared.debug("⚠️ Start failed, recreating engine for retry...", source: "ASRService")
-                    await self.retireAudioEngineAndWait(reason: "start_retry")
+                    try await self.retireAudioEngineAndWait(reason: "start_retry")
                     // Need to reconfigure the new engine
                     try? self.configureSession()
                     DebugLogger.shared.debug("✅ Engine recreated and reconfigured, will retry", source: "ASRService")
@@ -2442,7 +2473,7 @@ final class ASRService: ObservableObject {
         _ = await task?.result
         self.audioRouteRecoveryTask = nil
         self.isRecoveringAudioRoute = false
-        await self.audioEngineRetirementDrain.waitForScheduledReleases()
+        try? await self.waitForAudioEngineRetirement(reason: "cancel_route_recovery")
     }
 
     private func performAudioRouteRecovery(_ request: AudioRouteRecoveryRequest) async {
@@ -2488,7 +2519,15 @@ final class ASRService: ObservableObject {
             reason: request.reason,
             requiresIdlePrewarm: true
         )
-        await self.retireAudioEngineAndWait(reason: "idle_route_change:\(request.reason)")
+        do {
+            try await self.retireAudioEngineAndWait(reason: "idle_route_change:\(request.reason)")
+        } catch {
+            DebugLogger.shared.error(
+                "Idle audio route recovery stopped: \(error.localizedDescription)",
+                source: "ASRService"
+            )
+            return
+        }
 
         guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return }
         self.prewarmAudioEngineIfPossible(
@@ -2507,11 +2546,10 @@ final class ASRService: ObservableObject {
         self.audioCapturePipeline.setRecordingEnabled(false)
         self.stopMonitoringDevice()
         self.stopActiveAudioCapture()
-        await self.retireAudioEngineAndWait(reason: "audio_route_recovery")
-
-        guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return }
-
         do {
+            try await self.retireAudioEngineAndWait(reason: "audio_route_recovery")
+            guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return }
+
             self.audioCapturePipeline.setRecordingEnabled(
                 true,
                 sessionID: self.benchmarkSessionID,

Sources/Fluid/Services/AudioEngineRetirementDrain.swift

@@ -1,5 +1,25 @@
 import Foundation
 
+private final nonisolated class AudioEngineRetirementWait: @unchecked Sendable {
+    private let lock = NSLock()
+    private var continuation: CheckedContinuation<Bool, Never>?
+
+    init(_ continuation: CheckedContinuation<Bool, Never>) {
+        self.continuation = continuation
+    }
+
+    func finish(completed: Bool) {
+        self.lock.lock()
+        guard let continuation = self.continuation else {
+            self.lock.unlock()
+            return
+        }
+        self.continuation = nil
+        self.lock.unlock()
+        continuation.resume(returning: completed)
+    }
+}
+
 /// Owns the last strong reference to an audio engine while it is waiting to be
 /// released on the dedicated retirement queue.
 ///
@@ -36,24 +56,39 @@ final nonisolated class AudioEngineRetirementDrain: @unchecked Sendable {
         }
     }
 
-    func releaseAndWait(_ token: AudioEngineRetirementToken) async {
-        await self.enqueueAndWait {
+    @discardableResult
+    func releaseAndWait(
+        _ token: AudioEngineRetirementToken,
+        timeout: TimeInterval
+    ) async -> Bool {
+        await self.enqueueAndWait(timeout: timeout) {
             token.releaseEngine()
         }
     }
 
     /// Waits for every release already submitted to this drain. This is used at
     /// capture start so a fire-and-forget retirement from a non-route path cannot
     /// overlap construction of the next AVAudioEngine.
-    func waitForScheduledReleases() async {
-        await self.enqueueAndWait {}
+    @discardableResult
+    func waitForScheduledReleases(timeout: TimeInterval) async -> Bool {
+        await self.enqueueAndWait(timeout: timeout) {}
     }
 
-    private func enqueueAndWait(_ operation: @escaping @Sendable () -> Void) async {
-        await withCheckedContinuation { (continuation: CheckedContinuation<Void, Never>) in
+    private func enqueueAndWait(
+        timeout: TimeInterval,
+        operation: @escaping @Sendable () -> Void
+    ) async -> Bool {
+        precondition(timeout > 0, "Audio engine retirement timeout must be positive")
+
+        return await withCheckedContinuation { (continuation: CheckedContinuation<Bool, Never>) in
+            let wait = AudioEngineRetirementWait(continuation)
             self.queue.async {
                 operation()
-                continuation.resume()
+                wait.finish(completed: true)
+            }
+
+            DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + timeout) {
+                wait.finish(completed: false)
             }
         }
     }

Tests/FluidDictationIntegrationTests/AudioEngineRetirementDrainTests.swift

@@ -10,7 +10,7 @@ final class AudioEngineRetirementDrainTests: XCTestCase {
         let token = AudioEngineRetirementToken(try XCTUnwrap(probe))
         probe = nil
 
-        await drain.releaseAndWait(token)
+        await drain.releaseAndWait(token, timeout: 1)
 
         XCTAssertEqual(
             recorder.events,
@@ -29,7 +29,7 @@ final class AudioEngineRetirementDrainTests: XCTestCase {
         second = nil
 
         drain.schedule(firstToken)
-        await drain.releaseAndWait(secondToken)
+        await drain.releaseAndWait(secondToken, timeout: 1)
 
         XCTAssertEqual(
             recorder.events,
@@ -39,6 +39,44 @@ final class AudioEngineRetirementDrainTests: XCTestCase {
             ]
         )
     }
+
+    func testWaitForScheduledReleasesTimesOutWhenEngineDeinitIsBlocked() async throws {
+        let drain = AudioEngineRetirementDrain(label: "test.audio-engine-retirement.timeout")
+        let releaseStarted = DispatchSemaphore(value: 0)
+        let allowRelease = DispatchSemaphore(value: 0)
+        var probe: BlockingDeinitProbe? = BlockingDeinitProbe(
+            releaseStarted: releaseStarted,
+            allowRelease: allowRelease
+        )
+        let token = AudioEngineRetirementToken(try XCTUnwrap(probe))
+        probe = nil
+
+        drain.schedule(token)
+        XCTAssertEqual(releaseStarted.wait(timeout: .now() + 1), .success)
+
+        let completedBeforeTimeout = await drain.waitForScheduledReleases(timeout: 0.01)
+        XCTAssertFalse(completedBeforeTimeout)
+
+        allowRelease.signal()
+        let completedAfterRelease = await drain.waitForScheduledReleases(timeout: 1)
+        XCTAssertTrue(completedAfterRelease)
+    }
+
+}
+
+private final class BlockingDeinitProbe {
+    private let releaseStarted: DispatchSemaphore
+    private let allowRelease: DispatchSemaphore
+
+    init(releaseStarted: DispatchSemaphore, allowRelease: DispatchSemaphore) {
+        self.releaseStarted = releaseStarted
+        self.allowRelease = allowRelease
+    }
+
+    deinit {
+        self.releaseStarted.signal()
+        self.allowRelease.wait()
+    }
 }
 
 private final class DeinitProbe {