fix(asr): keep recordings when streaming recognition fails or stalls

- Keep recording after a streaming ASR error and retry the captured PCM through batch ASR before reporting a failure. - Treat an expired ASR wait as a stall so batch fallback still runs. - Prime the managed Volcengine stream on mic touch and adopt it only while fresh and open. - Auto-scroll the live transcript to the newest recognized text.
2026-10-20 20:05:08 +02:00
parent f7a031cc1d
commit be56d2210c
9 changed files with 518 additions and 20 deletions
@@ -11,11 +11,43 @@ import os
 import OSGKeyboardShared
 #endif

+/// A primed streaming session is opened before the utterance has an owner.
+/// Route its callbacks through a mutable relay, then bind the real partial
+/// consumer when the mic gesture is committed to an utterance.
+private final class CloudASRPartialRelay: @unchecked Sendable {
+    private let lock = OSAllocatedUnfairLock<
+        (@Sendable (String) -> Void)?
+    >(initialState: nil)
+
+    func bind(_ consumer: @escaping @Sendable (String) -> Void) {
+        lock.withLock { $0 = consumer }
+    }
+
+    func publish(_ text: String) {
+        let consumer = lock.withLock { $0 }
+        consumer?(text)
+    }
+}
+
 /// Uploads PCM only on the user-selected cloud engine path; provider clients
 /// reject missing credentials before network transmission. Personal dictionary
 /// entries are sent as recognition bias. Mutable client/cancellation state is
 /// lock-protected, which is the basis for `@unchecked Sendable`.
 public final class CloudASRService: ASRService, @unchecked Sendable {
+    private struct PrimedStreamingSession {
+        let id: UUID
+        let selection: String
+        let localeID: String
+        let createdAt: ContinuousClock.Instant
+        let relay: CloudASRPartialRelay
+        let task: Task<any CloudASRStreamingSession, Error>
+    }
+
+    /// The gateway closes a session that receives no audio for 15 s (counted
+    /// from WebSocket connect). A primed session older than this is never
+    /// adopted, leaving headroom for the first audio frame to arrive.
+    public static let defaultPrimedSessionMaxAge: Duration = .seconds(10)
+
    private let store: any ConfigurationStore
    private let session: URLSession
    private let localFallback: ASRService
@@ -23,24 +55,28 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
    /// provider-specific local-fallback selection unchanged.
    private let managedClient: (any CloudASRTranscribing)?
    private let managedGrants: GatewayGrantCoordinator?
+    private let primedSessionMaxAge: Duration
    private let lock = OSAllocatedUnfairLock()
    private var client: CloudASRTranscribing?
    private var usesLocalFallback = false
    private var boundClientSelection: String?
    private var cancelled = false
    private var streamingPipeline: StreamingUtterancePipeline?
+    private var primedStreamingSession: PrimedStreamingSession?

    public init(
        store: any ConfigurationStore = AppGroupStore(),
        session: URLSession = .shared,
        localFallback: ASRService? = nil,
        managedClient: (any CloudASRTranscribing)? = nil,
-        managedGrants: GatewayGrantCoordinator? = nil
+        managedGrants: GatewayGrantCoordinator? = nil,
+        primedSessionMaxAge: Duration = CloudASRService.defaultPrimedSessionMaxAge
    ) {
        self.store = store
        self.session = session
        self.managedClient = managedClient
        self.managedGrants = managedGrants
+        self.primedSessionMaxAge = primedSessionMaxAge
        // `SpeechAnalyzerASR` is internal, so it can't appear in a public
        // default argument value — resolve the fallback in the body instead.
        self.localFallback = localFallback ?? SpeechAnalyzerASR()
@@ -77,6 +113,143 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
        }
    }

+    /// Starts the account-managed Volcengine reservation + WebSocket handshake
+    /// while the user's finger is still down. The utterance adopts this session
+    /// moments later, removing those network round trips from first-partial
+    /// latency without sending audio before recording actually begins.
+    public func primeUtteranceStreaming(locale: Locale) {
+        bindClientIfNeeded()
+        guard store.credentialSource == .managed,
+              let selection = lock.withLock({ boundClientSelection }),
+              let streamingClient = lock.withLock({
+                  client as? any CloudASRStreamingCapable
+              }) else {
+            return
+        }
+
+        let localeID = locale.identifier(.bcp47)
+        let alreadyPrimed = lock.withLock { () -> Bool in
+            guard let current = primedStreamingSession else { return false }
+            return current.selection == selection
+                && current.localeID == localeID
+                && !isExpired(current)
+        }
+        guard !alreadyPrimed else { return }
+
+        cancelPrimedUtteranceStreaming()
+        let id = UUID()
+        let relay = CloudASRPartialRelay()
+        let dictionary = store.personalDictionary
+        let task = Task.detached(priority: .userInitiated) {
+            try await streamingClient.openStreamingSession(
+                locale: locale,
+                dictionary: dictionary,
+                onPartial: { relay.publish($0) }
+            )
+        }
+        lock.withLock {
+            primedStreamingSession = PrimedStreamingSession(
+                id: id,
+                selection: selection,
+                localeID: localeID,
+                createdAt: .now,
+                relay: relay,
+                task: task
+            )
+        }
+        // Self-expire: the keyboard's cancel command may never arrive (the
+        // extension can be dismissed, or a start can fail before adoption).
+        let maxAge = primedSessionMaxAge
+        Task { [weak self] in
+            try? await Task.sleep(for: maxAge)
+            guard let self, let expired = self.takePrimedSession(id: id) else { return }
+            FlowTrace.asr("cloud.stream.primeDiscarded", "reason=expired")
+            Self.close(expired)
+        }
+        FlowTrace.asr(
+            "cloud.stream.primeStarted",
+            "locale=\(localeID) selection=\(selection)"
+        )
+    }
+
+    public func cancelPrimedUtteranceStreaming() {
+        guard let primed = takePrimedSession() else { return }
+        Self.close(primed)
+    }
+
+    /// Removes the primed session; with an `id`, only if it is still that one.
+    private func takePrimedSession(id: UUID? = nil) -> PrimedStreamingSession? {
+        lock.withLock {
+            guard let current = primedStreamingSession,
+                  id == nil || current.id == id else {
+                return nil
+            }
+            primedStreamingSession = nil
+            return current
+        }
+    }
+
+    private func isExpired(_ primed: PrimedStreamingSession) -> Bool {
+        ContinuousClock.now - primed.createdAt >= primedSessionMaxAge
+    }
+
+    /// Closes a primed session without cancelling its open task. Cancelling
+    /// mid-reservation can leave a reserved, never-connected gateway session
+    /// holding credits and a concurrency slot until its connect timeout;
+    /// letting the open finish and then closing releases it immediately.
+    private static func close(_ primed: PrimedStreamingSession) {
+        Task {
+            if let session = try? await primed.task.value {
+                session.cancel()
+            }
+        }
+    }
+
+    /// Returns the primed session only if it still matches this utterance and
+    /// can accept audio. Anything else is closed and the caller opens a fresh
+    /// session, so priming can never make an utterance fail.
+    private func adoptPrimedSession(
+        _ primed: PrimedStreamingSession?,
+        selection: String?,
+        localeID: String,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async -> (any CloudASRStreamingSession)? {
+        guard let primed else { return nil }
+        guard primed.selection == selection, primed.localeID == localeID else {
+            FlowTrace.asr("cloud.stream.primeDiscarded", "reason=mismatch")
+            Self.close(primed)
+            return nil
+        }
+        guard !isExpired(primed) else {
+            FlowTrace.asr("cloud.stream.primeDiscarded", "reason=expired")
+            Self.close(primed)
+            return nil
+        }
+
+        primed.relay.bind(onPartial)
+        let session: any CloudASRStreamingSession
+        do {
+            session = try await primed.task.value
+        } catch {
+            FlowTrace.asr(
+                "cloud.stream.primeDiscarded",
+                "reason=openFailed \(CloudASRLogMetadata.describe(error))"
+            )
+            return nil
+        }
+        guard session.acceptsAudio else {
+            FlowTrace.asr("cloud.stream.primeDiscarded", "reason=closed")
+            session.cancel()
+            return nil
+        }
+        let ageMs = Int((ContinuousClock.now - primed.createdAt) / .milliseconds(1))
+        FlowTrace.asr(
+            "cloud.stream.primeAdopted",
+            "locale=\(localeID) ageMs=\(ageMs)"
+        )
+        return session
+    }
+
    public func transcribeChunk(samples: [Float], locale: Locale) async -> ASRChunkResult {
        guard !samples.isEmpty else { return .success("") }
        if Task.isCancelled || lock.withLock({ cancelled }) { return .cancelled }
@@ -129,23 +302,42 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
        onPartial: @escaping @Sendable (String) -> Void
    ) async -> ChunkedUtterancePipelineOutcome {
        bindClientIfNeeded()
+        // A previous utterance's abort/failure leaves `cancelled` set, which
+        // would make this utterance's batch fallback (`transcribeChunk`)
+        // return `.cancelled` for the rest of the Flow session.
+        lock.withLock { cancelled = false }
        if usesLocalFallback {
            let pipeline = ChunkedUtterancePipeline(asr: localFallback, locale: locale)
            return await pipeline.transcribe(stream: stream, onPartial: onPartial)
        }

        guard let streamingClient = lock.withLock({ client as? CloudASRStreamingCapable }) else {
+            cancelPrimedUtteranceStreaming()
            let pipeline = ChunkedUtterancePipeline(asr: self, locale: locale)
            return await pipeline.transcribe(stream: stream, onPartial: onPartial)
        }

+        let selection = lock.withLock { boundClientSelection }
+        let localeID = locale.identifier(.bcp47)
+        // Every utterance either adopts the primed session or closes it.
+        let primed = takePrimedSession()
+
        let session: any CloudASRStreamingSession
        do {
-            session = try await streamingClient.openStreamingSession(
-                locale: locale,
-                dictionary: store.personalDictionary,
+            if let adopted = await adoptPrimedSession(
+                primed,
+                selection: selection,
+                localeID: localeID,
                onPartial: onPartial
-            )
+            ) {
+                session = adopted
+            } else {
+                session = try await streamingClient.openStreamingSession(
+                    locale: locale,
+                    dictionary: store.personalDictionary,
+                    onPartial: onPartial
+                )
+            }
        } catch where ProviderToolCancellation.matches(error) {
            return .cancelled
        } catch {
@@ -249,6 +441,7 @@ public final class CloudASRService: ASRService, @unchecked Sendable {

    public func cancel() {
        lock.withLock { cancelled = true }
+        cancelPrimedUtteranceStreaming()
        let pipeline = lock.withLock { streamingPipeline }
        Task { await pipeline?.cancel() }
        localFallback.cancel()
@@ -259,8 +452,10 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
        let strategy = CloudASRModelCatalog.strategy(for: providerId)
        let credentialSource = store.credentialSource
        let selection = "\(credentialSource.rawValue):\(providerId)"
-        lock.withLock {
-            guard boundClientSelection != selection else { return }
+        let stalePrime = lock.withLock { () -> PrimedStreamingSession? in
+            guard boundClientSelection != selection else { return nil }
+            let stalePrime = primedStreamingSession
+            primedStreamingSession = nil
            boundClientSelection = selection
            if credentialSource == .managed {
                usesLocalFallback = false
@@ -269,12 +464,16 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
                    session: session,
                    managedGrants: managedGrants
                )
-                return
+                return stalePrime
            }
            usesLocalFallback = strategy == .localFallback
            client = usesLocalFallback
                ? nil
                : CloudASRClientFactory.make(store: store, session: session)
+            return stalePrime
+        }
+        if let stalePrime {
+            Self.close(stalePrime)
        }
    }
 }
@@ -51,6 +51,13 @@ public protocol CloudASRStreamingSession: Sendable {
    /// Signal end-of-audio and wait for the polish-ready final transcript.
    func finish() async throws -> String
    func cancel()
+    /// False once the session has failed, closed, or been cancelled, so a
+    /// pre-opened session is not adopted after its transport already died.
+    var acceptsAudio: Bool { get }
+}
+
+public extension CloudASRStreamingSession {
+    var acceptsAudio: Bool { true }
 }

 /// Providers that can open an utterance-level streaming session.
@@ -765,6 +765,12 @@ private final class ManagedVolcengineASRSession: CloudASRStreamingSession, @unch
        }
    }

+    var acceptsAudio: Bool {
+        lock.withLock { state in
+            !state.cancelled && state.failure == nil && !state.receiveClosed && !state.endSent
+        }
+    }
+
    func cancel() {
        let changed = lock.withLock { state -> Bool in
            guard !state.cancelled else { return false }