feat: add streaming cloud ASR, polish routing, and settings card layout

Unify Bailian/Volcengine/OpenAI realtime streaming, ABE polish routing with
fun styles, and a shared card-page Settings hierarchy; bump to 1.1 (build 32).

OSGKeyboardShared/Services/CloudASR/BailianRealtimeASRClient.swift

@@ -2,12 +2,14 @@
 // OSGKeyboard · Shared
 //
 // Alibaba Cloud Bailian / DashScope realtime ASR over the classic inference
-// WebSocket (`/api-ws/v1/inference`). Matches OpenLess' `bailian.rs` wire
-// protocol: run-task → PCM binary frames → finish-task → result events.
+// WebSocket (`/api-ws/v1/inference`). Utterance-level duplex session with
+// interim `result-generated` partials; batch `transcribe(samples:)` remains
+// for connection probes and chunk fallback.
 
 import Foundation
+import os
 
-struct BailianRealtimeASRClient: CloudASRTranscribing {
+struct BailianRealtimeASRClient: CloudASRTranscribing, CloudASRStreamingCapable {
     let apiKey: String
     let endpoint: String
     let model: String
@@ -15,28 +17,22 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
     let session: URLSession
 
     /// 100 ms of 16 kHz / 16-bit / mono PCM.
-    private static let targetChunkBytes = 3_200
-    private static let startTimeout: TimeInterval = 8
-    private static let finalTimeout: TimeInterval = 12
+    static let targetChunkBytes = 3_200
+    static let startTimeout: TimeInterval = 8
+    static let finalTimeout: TimeInterval = 12
     private static let sessionTimeout: TimeInterval = startTimeout + finalTimeout + 4
 
     func prepare(dictionary: PersonalDictionary) async throws {}
 
-    func transcribe(
-        samples: [Float],
-        sampleRate: Int,
+    func openStreamingSession(
         locale: Locale,
-        dictionary: PersonalDictionary
-    ) async throws -> String {
+        dictionary: PersonalDictionary,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async throws -> any CloudASRStreamingSession {
+        _ = locale
+        _ = dictionary
         guard !apiKey.isEmpty else { throw CloudASRError.noAPIKey }
-        guard sampleRate == 16_000 else {
-            throw CloudASRError.transport("Bailian realtime expects 16 kHz audio")
-        }
-        guard !samples.isEmpty else { throw CloudASRError.emptyTranscript }
-
         let url = try resolvedEndpointURL()
-        let pcm = Self.pcm16Data(samples: samples)
-        let taskID = UUID().uuidString.replacingOccurrences(of: "-", with: "")
         let resolvedModel = model.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
             ? CloudASRModelCatalog.alibabaFunASRRealtime
             : model.trimmingCharacters(in: .whitespacesAndNewlines)
@@ -50,44 +46,40 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
 
         let wsTask = session.webSocketTask(with: request)
         wsTask.resume()
+        let live = BailianStreamingSession(
+            wsTask: wsTask,
+            model: resolvedModel,
+            vocabularyID: vocabularyID,
+            onPartial: onPartial
+        )
+        try await live.start()
+        return live
+    }
 
-        return try await withThrowingTaskGroup(of: String.self) { group in
-            let events = BailianEventStream(task: wsTask)
-
-            group.addTask {
-                defer { events.cancel() }
-                return try await Self.runSession(
-                    taskID: taskID,
-                    model: resolvedModel,
-                    pcm: pcm,
-                    wsTask: wsTask,
-                    events: events
-                )
-            }
-
-            group.addTask {
-                try await Task.sleep(nanoseconds: UInt64(Self.sessionTimeout * 1_000_000_000))
-                events.cancel()
-                wsTask.cancel(with: .goingAway, reason: nil)
-                throw CloudASRError.transport("session timed out")
-            }
-
-            guard let result = try await group.next() else {
-                throw CloudASRError.emptyTranscript
-            }
-            group.cancelAll()
-            return result.trimmingCharacters(in: .whitespacesAndNewlines)
+    func transcribe(
+        samples: [Float],
+        sampleRate: Int,
+        locale: Locale,
+        dictionary: PersonalDictionary
+    ) async throws -> String {
+        guard sampleRate == 16_000 else {
+            throw CloudASRError.transport("Bailian realtime expects 16 kHz audio")
         }
+        guard !samples.isEmpty else { throw CloudASRError.emptyTranscript }
+
+        let session = try await openStreamingSession(
+            locale: locale,
+            dictionary: dictionary,
+            onPartial: { _ in }
+        )
+        try await session.append(samples: samples)
+        let text = try await session.finish()
+        let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
+        guard !trimmed.isEmpty else { throw CloudASRError.emptyTranscript }
+        return trimmed
     }
 
     /// Settings connection probe: handshake to `task-started` only.
-    ///
-    /// Reaching `task-started` proves endpoint + `Authorization` + model are
-    /// all valid — which is exactly what "validate connection" must check.
-    /// It deliberately sends NO audio: DashScope realtime rejects a short
-    /// silent probe with a `task-failed: emptyAudio`, which is a false
-    /// negative for a connectivity test. A real auth/quota/model failure
-    /// still arrives as `task-failed` before `task-started` and surfaces.
     func probeConnection() async throws {
         guard !apiKey.isEmpty else { throw CloudASRError.noAPIKey }
 
@@ -108,17 +100,23 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
         wsTask.resume()
 
         try await withThrowingTaskGroup(of: Void.self) { group in
-            let events = BailianEventStream(task: wsTask)
+            let events = BailianEventStream(task: wsTask, onPartial: nil)
 
             group.addTask {
                 defer { events.cancel() }
-                try await Self.sendText(
-                    Self.runTaskMessage(taskID: taskID, model: resolvedModel, vocabularyID: nil),
+                try await BailianRealtimeASRClient.sendText(
+                    BailianRealtimeASRClient.runTaskMessage(
+                        taskID: taskID,
+                        model: resolvedModel,
+                        vocabularyID: nil
+                    ),
                     task: wsTask
                 )
                 try await events.waitForStarted(timeout: Self.startTimeout)
-                // Politely end the task; the connection is already proven.
-                try? await Self.sendText(Self.finishTaskMessage(taskID: taskID), task: wsTask)
+                try? await BailianRealtimeASRClient.sendText(
+                    BailianRealtimeASRClient.finishTaskMessage(taskID: taskID),
+                    task: wsTask
+                )
             }
 
             group.addTask {
@@ -133,37 +131,6 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
         }
     }
 
-    private static func runSession(
-        taskID: String,
-        model: String,
-        pcm: Data,
-        wsTask: URLSessionWebSocketTask,
-        events: BailianEventStream
-    ) async throws -> String {
-        try await sendText(
-            runTaskMessage(taskID: taskID, model: model, vocabularyID: nil),
-            task: wsTask
-        )
-
-        try await events.waitForStarted(timeout: startTimeout)
-
-        var offset = 0
-        while offset < pcm.count {
-            let end = min(offset + targetChunkBytes, pcm.count)
-            try await sendBinary(pcm.subdata(in: offset..<end), task: wsTask)
-            offset = end
-        }
-
-        // Let the server register the final frames before ending the task.
-        // Sending `finish-task` in the same instant as the last binary frame
-        // races the server's audio buffering (root cause of `emptyAudio` on
-        // very short clips).
-        try? await Task.sleep(nanoseconds: 120_000_000)
-
-        try await sendText(finishTaskMessage(taskID: taskID), task: wsTask)
-        return try await events.waitForFinalText(timeout: finalTimeout)
-    }
-
     private func resolvedEndpointURL() throws -> URL {
         let raw = endpoint.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
             ? CloudASRModelCatalog.bailianDefaultEndpoint
@@ -172,7 +139,7 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
         return url
     }
 
-    private static func sendText(_ text: String, task: URLSessionWebSocketTask) async throws {
+    static func sendText(_ text: String, task: URLSessionWebSocketTask) async throws {
         do {
             try await task.send(.string(text))
         } catch {
@@ -180,7 +147,7 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
         }
     }
 
-    private static func sendBinary(_ data: Data, task: URLSessionWebSocketTask) async throws {
+    static func sendBinary(_ data: Data, task: URLSessionWebSocketTask) async throws {
         do {
             try await task.send(.data(data))
         } catch {
@@ -188,18 +155,6 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
         }
     }
 
-    private static func pcm16Data(samples: [Float]) -> Data {
-        var data = Data()
-        data.reserveCapacity(samples.count * 2)
-        for sample in samples {
-            let scaled = sample * 32_767.0
-            let clipped = Swift.max(-32_768.0, Swift.min(32_767.0, scaled))
-            var littleEndian = Int16(clipped.rounded()).littleEndian
-            withUnsafeBytes(of: &littleEndian) { data.append(contentsOf: $0) }
-        }
-        return data
-    }
-
     /// Overlap-aware join to avoid cumulative duplicate text from interim replays.
     static func mergeSegments(_ segments: [String]) -> String {
         var result = ""
@@ -275,18 +230,104 @@ struct BailianRealtimeASRClient: CloudASRTranscribing {
     }
 }
 
+// MARK: - Utterance session
+
+private final class BailianStreamingSession: CloudASRStreamingSession, @unchecked Sendable {
+    private let wsTask: URLSessionWebSocketTask
+    private let model: String
+    private let vocabularyID: String?
+    private let onPartial: @Sendable (String) -> Void
+    private let events: BailianEventStream
+    private let taskID: String
+    private let lock = OSAllocatedUnfairLock()
+    private var started = false
+    private var pcmBuffer = Data()
+
+    init(
+        wsTask: URLSessionWebSocketTask,
+        model: String,
+        vocabularyID: String?,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) {
+        self.wsTask = wsTask
+        self.model = model
+        self.vocabularyID = vocabularyID
+        self.onPartial = onPartial
+        self.taskID = UUID().uuidString.replacingOccurrences(of: "-", with: "")
+        self.events = BailianEventStream(task: wsTask, onPartial: onPartial)
+    }
+
+    func start() async throws {
+        try await BailianRealtimeASRClient.sendText(
+            BailianRealtimeASRClient.runTaskMessage(
+                taskID: taskID,
+                model: model,
+                vocabularyID: vocabularyID
+            ),
+            task: wsTask
+        )
+        try await events.waitForStarted(timeout: BailianRealtimeASRClient.startTimeout)
+        lock.withLock { started = true }
+    }
+
+    func append(samples: [Float]) async throws {
+        guard lock.withLock({ started }) else {
+            throw CloudASRError.transport("Bailian session not started")
+        }
+        let pcm = CloudASRStreamingPCM.pcm16LE(samples: samples)
+        let frames: [Data] = lock.withLock {
+            pcmBuffer.append(pcm)
+            var frames: [Data] = []
+            while pcmBuffer.count >= BailianRealtimeASRClient.targetChunkBytes {
+                let frame = pcmBuffer.prefix(BailianRealtimeASRClient.targetChunkBytes)
+                frames.append(Data(frame))
+                pcmBuffer.removeFirst(BailianRealtimeASRClient.targetChunkBytes)
+            }
+            return frames
+        }
+        for frame in frames {
+            try await BailianRealtimeASRClient.sendBinary(frame, task: wsTask)
+        }
+    }
+
+    func finish() async throws -> String {
+        // Flush remaining PCM (pad short last frame as-is — server tolerates).
+        let trailing: Data = lock.withLock {
+            let data = pcmBuffer
+            pcmBuffer.removeAll(keepingCapacity: false)
+            return data
+        }
+        if !trailing.isEmpty {
+            try await BailianRealtimeASRClient.sendBinary(trailing, task: wsTask)
+        }
+        // Avoid emptyAudio race on very short clips.
+        try? await Task.sleep(nanoseconds: 120_000_000)
+        try await BailianRealtimeASRClient.sendText(
+            BailianRealtimeASRClient.finishTaskMessage(taskID: taskID),
+            task: wsTask
+        )
+        return try await events.waitForFinalText(timeout: BailianRealtimeASRClient.finalTimeout)
+    }
+
+    func cancel() {
+        events.cancel()
+    }
+}
+
 // MARK: - Concurrent read loop
 
 private final class BailianEventStream: @unchecked Sendable {
     private let task: URLSessionWebSocketTask
-    private let lock = NSLock()
+    private let onPartial: (@Sendable (String) -> Void)?
+    private let lock = OSAllocatedUnfairLock()
     private var started = false
     private var finalText: String?
     private var failure: Error?
     private var readTask: Task<Void, Never>?
 
-    init(task: URLSessionWebSocketTask) {
+    init(task: URLSessionWebSocketTask, onPartial: (@Sendable (String) -> Void)?) {
         self.task = task
+        self.onPartial = onPartial
         readTask = Task { [weak self] in
             await self?.readLoop()
         }
@@ -320,21 +361,15 @@ private final class BailianEventStream: @unchecked Sendable {
     }
 
     private func snapshotStarted() -> Bool {
-        lock.lock()
-        defer { lock.unlock() }
-        return started
+        lock.withLock { started }
     }
 
     private func snapshotFinalText() -> String? {
-        lock.lock()
-        defer { lock.unlock() }
-        return finalText
+        lock.withLock { finalText }
     }
 
     private func snapshotFailure() -> Error? {
-        lock.lock()
-        defer { lock.unlock() }
-        return failure
+        lock.withLock { failure }
     }
 
     private func readLoop() async {
@@ -395,6 +430,21 @@ private final class BailianEventStream: @unchecked Sendable {
                 } else {
                     partialSegments[sentenceID] = trimmed
                 }
+
+                var displayParts: [String] = []
+                let ids = Set(finalSegments.keys).union(partialSegments.keys).sorted()
+                for id in ids {
+                    if let committed = finalSegments[id] {
+                        displayParts.append(committed)
+                    } else if let live = partialSegments[id] {
+                        displayParts.append(live)
+                    }
+                }
+                let display = BailianRealtimeASRClient.mergeSegments(displayParts)
+                    .trimmingCharacters(in: .whitespacesAndNewlines)
+                if !display.isEmpty {
+                    onPartial?(display)
+                }
             case "task-finished":
                 if finalSegments.isEmpty {
                     publishFinal(lastResultText)
@@ -414,21 +464,15 @@ private final class BailianEventStream: @unchecked Sendable {
     }
 
     private func publishStarted() {
-        lock.lock()
-        started = true
-        lock.unlock()
+        lock.withLock { started = true }
     }
 
     private func publishFinal(_ text: String) {
-        lock.lock()
-        finalText = text
-        lock.unlock()
+        lock.withLock { finalText = text }
     }
 
     private func publishFailure(_ error: Error) {
-        lock.lock()
-        failure = error
-        lock.unlock()
+        lock.withLock { failure = error }
         cancel()
     }
 }

OSGKeyboardShared/Services/CloudASR/CloudASRClients.swift

@@ -83,6 +83,14 @@ public enum CloudASRClientFactory {
                 resourceID: asrModel,
                 session: session
             )
+        case .openaiRealtimeStreaming:
+            return OpenAIRealtimeASRClient(
+                apiKey: store.asrApiKey,
+                endpoint: store.asrBaseURL,
+                model: asrModel,
+                batchBaseURL: LLMProvider.provider(id: "openai").defaultBaseURL,
+                session: session
+            )
         case .localFallback:
             return UnsupportedCloudASRClient(providerId: providerId)
         }

OSGKeyboardShared/Services/CloudASR/CloudASRService.swift

@@ -1,8 +1,9 @@
 // CloudASRService.swift
 // OSGKeyboard · Shared
 //
-// Cloud-engine ASR: uploads PCM chunks to the user's configured provider
-// with personal-dictionary bias. Moonshot falls back to on-device ASR.
+// Cloud-engine ASR: uploads PCM to the user's configured provider with
+// personal-dictionary bias. Streaming-capable providers use one utterance
+// WebSocket; others stay on chunked batch. Moonshot falls back to on-device ASR.
 
 import Foundation
 import os
@@ -16,6 +17,7 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
     private var usesLocalFallback = false
     private var boundProviderId: String?
     private var cancelled = false
+    private var streamingPipeline: StreamingUtterancePipeline?
 
     public init(
         store: any ConfigurationStore = AppGroupStore(),
@@ -29,6 +31,11 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
         self.localFallback = localFallback ?? SpeechAnalyzerASR()
     }
 
+    /// Whether Flow should prefer utterance-level true streaming for the bound provider.
+    public var supportsUtteranceStreaming: Bool {
+        CloudASRModelCatalog.supportsTrueStreamingASR(for: store.asrProviderId)
+    }
+
     public func resetForNewUtterance() {
         lock.withLock { cancelled = false }
         if usesLocalFallback {
@@ -79,6 +86,55 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
         }
     }
 
+    /// Utterance-level streaming; if the session cannot start, fall back to
+    /// chunked batch on the same mic stream. Mid-stream failures surface as
+    /// errors (finalize still has PCM batch fallback).
+    public func transcribeUtteranceStreaming(
+        stream: AsyncStream<AudioBufferSnapshot>,
+        locale: Locale,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async -> ChunkedUtterancePipelineOutcome {
+        bindClientIfNeeded()
+        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 {
+            let pipeline = ChunkedUtterancePipeline(asr: self, locale: locale)
+            return await pipeline.transcribe(stream: stream, onPartial: onPartial)
+        }
+
+        let session: any CloudASRStreamingSession
+        do {
+            session = try await streamingClient.openStreamingSession(
+                locale: locale,
+                dictionary: store.personalDictionary,
+                onPartial: onPartial
+            )
+        } catch {
+            OSGLog.asr.warning(
+                "streaming ASR session open failed, using chunked batch: \(error.localizedDescription, privacy: .public)"
+            )
+            let pipeline = ChunkedUtterancePipeline(asr: self, locale: locale)
+            return await pipeline.transcribe(stream: stream, onPartial: onPartial)
+        }
+
+        let pipeline = StreamingUtterancePipeline(
+            client: streamingClient,
+            locale: locale,
+            dictionary: store.personalDictionary
+        )
+        lock.withLock { streamingPipeline = pipeline }
+        let outcome = await pipeline.transcribe(
+            stream: stream,
+            onPartial: onPartial,
+            preopenedSession: session
+        )
+        lock.withLock { streamingPipeline = nil }
+        return outcome
+    }
+
     public func transcribe(
         stream: AsyncStream<AudioBufferSnapshot>,
         locale: Locale
@@ -88,6 +144,34 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
             return localFallback.transcribe(stream: stream, locale: locale)
         }
 
+        if supportsUtteranceStreaming, lock.withLock({ client is CloudASRStreamingCapable }) {
+            return AsyncStream { continuation in
+                continuation.yield(.capability(onDeviceSupported: false))
+                let task = Task {
+                    let outcome = await self.transcribeUtteranceStreaming(
+                        stream: stream,
+                        locale: locale,
+                        onPartial: { partial in
+                            continuation.yield(.partial(partial))
+                        }
+                    )
+                    switch outcome {
+                    case .success(let success):
+                        continuation.yield(.final(success.text))
+                    case .failure(let message):
+                        continuation.yield(.error(message))
+                    case .cancelled:
+                        break
+                    }
+                    continuation.finish()
+                }
+                continuation.onTermination = { @Sendable _ in
+                    task.cancel()
+                    self.cancel()
+                }
+            }
+        }
+
         return AsyncStream { continuation in
             continuation.yield(.capability(onDeviceSupported: false))
             let task = Task {
@@ -129,6 +213,8 @@ public final class CloudASRService: ASRService, @unchecked Sendable {
 
     public func cancel() {
         lock.withLock { cancelled = true }
+        let pipeline = lock.withLock { streamingPipeline }
+        Task { await pipeline?.cancel() }
         localFallback.cancel()
     }
 

OSGKeyboardShared/Services/CloudASR/CloudASRStreaming.swift

@@ -0,0 +1,135 @@
+// CloudASRStreaming.swift
+// OSGKeyboard · Shared
+//
+// Utterance-scoped cloud ASR sessions: one long-lived connection per press,
+// streaming PCM up and interim text down. Chunked batch ASR remains the
+// fallback for providers without a true streaming protocol.
+
+import Foundation
+
+/// Long-lived cloud ASR session for one Flow utterance.
+public protocol CloudASRStreamingSession: Sendable {
+    /// Append 16 kHz mono Float32 PCM captured while the mic is open.
+    func append(samples: [Float]) async throws
+    /// Signal end-of-audio and wait for the polish-ready final transcript.
+    func finish() async throws -> String
+    func cancel()
+}
+
+/// Providers that can open an utterance-level streaming session.
+public protocol CloudASRStreamingCapable: CloudASRTranscribing {
+    func openStreamingSession(
+        locale: Locale,
+        dictionary: PersonalDictionary,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async throws -> any CloudASRStreamingSession
+}
+
+/// Feeds a live mic stream into a cloud streaming session and mirrors the
+/// existing `ChunkedUtterancePipelineOutcome` surface for Flow.
+public actor StreamingUtterancePipeline {
+    private let client: any CloudASRStreamingCapable
+    private let locale: Locale
+    private let dictionary: PersonalDictionary
+    private var cancelled = false
+    private var activeSession: (any CloudASRStreamingSession)?
+
+    public init(
+        client: any CloudASRStreamingCapable,
+        locale: Locale,
+        dictionary: PersonalDictionary
+    ) {
+        self.client = client
+        self.locale = locale
+        self.dictionary = dictionary
+    }
+
+    public func cancel() {
+        cancelled = true
+        activeSession?.cancel()
+    }
+
+    public func transcribe(
+        stream: AsyncStream<AudioBufferSnapshot>,
+        onPartial: @Sendable @escaping (String) -> Void,
+        preopenedSession: (any CloudASRStreamingSession)? = nil
+    ) async -> ChunkedUtterancePipelineOutcome {
+        cancelled = false
+        do {
+            let session: any CloudASRStreamingSession
+            if let preopenedSession {
+                session = preopenedSession
+            } else {
+                session = try await client.openStreamingSession(
+                    locale: locale,
+                    dictionary: dictionary,
+                    onPartial: onPartial
+                )
+            }
+            activeSession = session
+
+            for await snap in stream {
+                if cancelled || Task.isCancelled {
+                    session.cancel()
+                    return .cancelled
+                }
+                guard !snap.samples.isEmpty else { continue }
+                try await session.append(samples: snap.samples)
+            }
+
+            if cancelled || Task.isCancelled {
+                session.cancel()
+                return .cancelled
+            }
+
+            let finalText = try await session.finish()
+                .trimmingCharacters(in: .whitespacesAndNewlines)
+            activeSession = nil
+            guard !finalText.isEmpty else {
+                return .failure(SharedL10n.string("error.asr.noSpeech"))
+            }
+            return .success(ChunkedUtteranceSuccess(text: finalText))
+        } catch is CancellationError {
+            activeSession?.cancel()
+            activeSession = nil
+            return .cancelled
+        } catch {
+            activeSession?.cancel()
+            activeSession = nil
+            if cancelled || Task.isCancelled { return .cancelled }
+            return .failure(error.localizedDescription)
+        }
+    }
+}
+
+/// Shared PCM helpers for streaming cloud clients.
+enum CloudASRStreamingPCM {
+    static func pcm16LE(samples: [Float]) -> Data {
+        var data = Data()
+        data.reserveCapacity(samples.count * 2)
+        for sample in samples {
+            let scaled = sample * 32_767.0
+            let clipped = Swift.max(-32_768.0, Swift.min(32_767.0, scaled))
+            var littleEndian = Int16(clipped.rounded()).littleEndian
+            withUnsafeBytes(of: &littleEndian) { data.append(contentsOf: $0) }
+        }
+        return data
+    }
+
+    /// Linear upsample 16 kHz → 24 kHz for OpenAI Realtime PCM input.
+    static func upsample16kTo24k(_ samples: [Float]) -> [Float] {
+        guard !samples.isEmpty else { return [] }
+        let outCount = max(1, samples.count * 3 / 2)
+        var output = [Float]()
+        output.reserveCapacity(outCount)
+        let lastIndex = samples.count - 1
+        for i in 0..<outCount {
+            let src = Double(i) * 16.0 / 24.0
+            let i0 = min(Int(src), lastIndex)
+            let i1 = min(i0 + 1, lastIndex)
+            let frac = Float(src - Double(i0))
+            output.append(samples[i0] + (samples[i1] - samples[i0]) * frac)
+        }
+        return output
+    }
+}

OSGKeyboardShared/Services/CloudASR/OpenAIRealtimeASRClient.swift

@@ -0,0 +1,391 @@
+// OpenAIRealtimeASRClient.swift
+// OSGKeyboard · Shared
+//
+// OpenAI Realtime transcription (WebSocket). Streams PCM and transcript
+// deltas for utterance-level ASR. Batch `/audio/transcriptions` remains the
+// fallback path when realtime is unavailable.
+
+import Foundation
+import os
+
+struct OpenAIRealtimeASRClient: CloudASRTranscribing, CloudASRStreamingCapable {
+    let apiKey: String
+    let endpoint: String
+    let model: String
+    let session: URLSession
+    /// Used when streaming fails and Flow falls back to chunked batch ASR.
+    private let batchClient: PromptCloudASRClient
+
+    static let appendChunkBytes = 4_800 // 100 ms @ 24 kHz / 16-bit mono.
+    static let finalTimeout: TimeInterval = 15
+
+    init(
+        apiKey: String,
+        endpoint: String,
+        model: String,
+        batchBaseURL: String,
+        session: URLSession
+    ) {
+        self.apiKey = apiKey
+        self.endpoint = endpoint
+        self.model = model
+        self.session = session
+        self.batchClient = PromptCloudASRClient(
+            providerId: "openai",
+            baseURL: batchBaseURL.isEmpty ? "https://api.openai.com/v1" : batchBaseURL,
+            apiKey: apiKey,
+            model: Self.batchModel(from: model),
+            session: session
+        )
+    }
+
+    func prepare(dictionary: PersonalDictionary) async throws {}
+
+    func openStreamingSession(
+        locale: Locale,
+        dictionary: PersonalDictionary,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async throws -> any CloudASRStreamingSession {
+        _ = dictionary
+        guard !apiKey.isEmpty else { throw CloudASRError.noAPIKey }
+        let url = try resolvedEndpointURL()
+        var request = URLRequest(url: url)
+        request.timeoutInterval = 8
+        request.setValue(
+            "Bearer \(apiKey.trimmingCharacters(in: .whitespacesAndNewlines))",
+            forHTTPHeaderField: "Authorization"
+        )
+
+        let wsTask = session.webSocketTask(with: request)
+        wsTask.resume()
+        let live = OpenAIRealtimeStreamingSession(
+            wsTask: wsTask,
+            model: resolvedRealtimeModel,
+            locale: locale,
+            onPartial: onPartial
+        )
+        try await live.start()
+        return live
+    }
+
+    func transcribe(
+        samples: [Float],
+        sampleRate: Int,
+        locale: Locale,
+        dictionary: PersonalDictionary
+    ) async throws -> String {
+        try await batchClient.transcribe(
+            samples: samples,
+            sampleRate: sampleRate,
+            locale: locale,
+            dictionary: dictionary
+        )
+    }
+
+    func probeConnection() async throws {
+        do {
+            let session = try await openStreamingSession(
+                locale: Locale(identifier: "zh-CN"),
+                dictionary: .empty,
+                onPartial: { _ in }
+            )
+            session.cancel()
+        } catch {
+            try await batchClient.probeConnection()
+        }
+    }
+
+    private var resolvedRealtimeModel: String {
+        let trimmed = model.trimmingCharacters(in: .whitespacesAndNewlines)
+        if trimmed.isEmpty || trimmed.hasPrefix("gpt-4o") || trimmed == "whisper-1" {
+            return CloudASRModelCatalog.openAIRealtimeWhisper
+        }
+        return trimmed
+    }
+
+    private func resolvedEndpointURL() throws -> URL {
+        let raw = endpoint.trimmingCharacters(in: .whitespacesAndNewlines)
+        if raw.hasPrefix("wss://") || raw.hasPrefix("ws://") {
+            guard let url = URL(string: raw) else { throw CloudASRError.invalidURL }
+            return url
+        }
+        guard let url = URL(string: CloudASRModelCatalog.openAIRealtimeEndpoint) else {
+            throw CloudASRError.invalidURL
+        }
+        return url
+    }
+
+    private static func batchModel(from model: String) -> String {
+        let trimmed = model.trimmingCharacters(in: .whitespacesAndNewlines)
+        if trimmed.isEmpty || trimmed.contains("realtime") {
+            return CloudASRModelCatalog.openAITranscribe
+        }
+        return trimmed
+    }
+}
+
+// MARK: - Utterance session
+
+private final class OpenAIRealtimeStreamingSession: CloudASRStreamingSession, @unchecked Sendable {
+    private let wsTask: URLSessionWebSocketTask
+    private let model: String
+    private let locale: Locale
+    private let onPartial: @Sendable (String) -> Void
+    private let lock = OSAllocatedUnfairLock()
+    private var receiveTask: Task<Void, Never>?
+    private var failure: Error?
+    private var sessionReady = false
+    private var finished = false
+    private var pcmBuffer = Data()
+    private var partialByItem: [String: String] = [:]
+    private var completedByItem: [String: String] = [:]
+    private var itemOrder: [String] = []
+    private var awaitingCommit = false
+
+    init(
+        wsTask: URLSessionWebSocketTask,
+        model: String,
+        locale: Locale,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) {
+        self.wsTask = wsTask
+        self.model = model
+        self.locale = locale
+        self.onPartial = onPartial
+    }
+
+    func start() async throws {
+        receiveTask = Task { [weak self] in
+            await self?.receiveLoop()
+        }
+        let language = Self.languageHint(from: locale)
+        var transcription: [String: Any] = [
+            "model": model,
+            "delay": "low",
+        ]
+        if let language {
+            transcription["language"] = language
+        }
+        var input: [String: Any] = [
+            "format": [
+                "type": "audio/pcm",
+                "rate": 24_000,
+            ],
+            "transcription": transcription,
+        ]
+        input["turn_detection"] = NSNull()
+        let update: [String: Any] = [
+            "type": "session.update",
+            "session": [
+                "type": "transcription",
+                "audio": [
+                    "input": input,
+                ],
+            ],
+        ]
+        try await sendJSON(update)
+        let deadline = Date().addingTimeInterval(8)
+        while Date() < deadline {
+            try throwIfFailed()
+            if lock.withLock({ sessionReady }) { return }
+            try await Task.sleep(nanoseconds: 20_000_000)
+        }
+        cancel()
+        throw CloudASRError.transport("OpenAI realtime session timed out")
+    }
+
+    func append(samples: [Float]) async throws {
+        try throwIfFailed()
+        let upsampled = CloudASRStreamingPCM.upsample16kTo24k(samples)
+        let pcm = CloudASRStreamingPCM.pcm16LE(samples: upsampled)
+        let frames: [Data] = lock.withLock {
+            pcmBuffer.append(pcm)
+            var frames: [Data] = []
+            while pcmBuffer.count >= OpenAIRealtimeASRClient.appendChunkBytes {
+                let frame = pcmBuffer.prefix(OpenAIRealtimeASRClient.appendChunkBytes)
+                frames.append(Data(frame))
+                pcmBuffer.removeFirst(OpenAIRealtimeASRClient.appendChunkBytes)
+            }
+            return frames
+        }
+        for frame in frames {
+            try await sendAppend(frame)
+        }
+    }
+
+    func finish() async throws -> String {
+        try throwIfFailed()
+        let trailing: Data = lock.withLock {
+            let data = pcmBuffer
+            pcmBuffer.removeAll(keepingCapacity: false)
+            awaitingCommit = true
+            return data
+        }
+        if !trailing.isEmpty {
+            try await sendAppend(trailing)
+        }
+        try await sendJSON(["type": "input_audio_buffer.commit"])
+
+        let deadline = Date().addingTimeInterval(OpenAIRealtimeASRClient.finalTimeout)
+        while Date() < deadline {
+            try throwIfFailed()
+            let snapshot = lock.withLock { (awaitingCommit, composedFinal(), composedDisplay()) }
+            if !snapshot.0 {
+                let text = snapshot.1.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
+                    ? snapshot.2
+                    : snapshot.1
+                let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
+                cancel()
+                if trimmed.isEmpty { throw CloudASRError.emptyTranscript }
+                return trimmed
+            }
+            let settled = lock.withLock {
+                !completedByItem.isEmpty && partialByItem.isEmpty && !awaitingCommit
+            }
+            if settled {
+                let text = lock.withLock { composedFinal() }
+                    .trimmingCharacters(in: .whitespacesAndNewlines)
+                cancel()
+                if text.isEmpty { throw CloudASRError.emptyTranscript }
+                return text
+            }
+            try await Task.sleep(nanoseconds: 20_000_000)
+        }
+        let fallback = lock.withLock {
+            let final = composedFinal()
+            return final.isEmpty ? composedDisplay() : final
+        }
+        .trimmingCharacters(in: .whitespacesAndNewlines)
+        cancel()
+        if fallback.isEmpty {
+            throw CloudASRError.transport("OpenAI realtime final timed out")
+        }
+        return fallback
+    }
+
+    func cancel() {
+        receiveTask?.cancel()
+        wsTask.cancel(with: .normalClosure, reason: nil)
+        lock.withLock { finished = true }
+    }
+
+    private func receiveLoop() async {
+        while !Task.isCancelled {
+            let message: URLSessionWebSocketTask.Message
+            do {
+                message = try await wsTask.receive()
+            } catch {
+                publishFailure(CloudASRError.transport(error.localizedDescription))
+                return
+            }
+            let text: String
+            switch message {
+            case .string(let value):
+                text = value
+            case .data(let data):
+                text = String(data: data, encoding: .utf8) ?? ""
+            @unknown default:
+                continue
+            }
+            guard let json = try? JSONSerialization.jsonObject(with: Data(text.utf8)) as? [String: Any],
+                  let type = json["type"] as? String else {
+                continue
+            }
+
+            switch type {
+            case "session.created", "session.updated":
+                lock.withLock { sessionReady = true }
+            case "conversation.item.input_audio_transcription.delta":
+                let itemID = json["item_id"] as? String ?? "default"
+                let delta = json["delta"] as? String ?? ""
+                guard !delta.isEmpty else { continue }
+                let display = lock.withLock { () -> String in
+                    if partialByItem[itemID] == nil, completedByItem[itemID] == nil {
+                        itemOrder.append(itemID)
+                    }
+                    partialByItem[itemID, default: ""] += delta
+                    return composedDisplay()
+                }
+                if !display.isEmpty { onPartial(display) }
+            case "conversation.item.input_audio_transcription.completed":
+                let itemID = json["item_id"] as? String ?? "default"
+                let transcript = (json["transcript"] as? String ?? "")
+                    .trimmingCharacters(in: .whitespacesAndNewlines)
+                let display = lock.withLock { () -> String in
+                    if !itemOrder.contains(itemID) {
+                        itemOrder.append(itemID)
+                    }
+                    if !transcript.isEmpty {
+                        completedByItem[itemID] = transcript
+                    }
+                    partialByItem.removeValue(forKey: itemID)
+                    awaitingCommit = false
+                    return composedDisplay()
+                }
+                if !display.isEmpty { onPartial(display) }
+            case "error":
+                let message = ((json["error"] as? [String: Any])?["message"] as? String)
+                    ?? "OpenAI realtime error"
+                publishFailure(CloudASRError.transport(message))
+                return
+            default:
+                break
+            }
+        }
+    }
+
+    private func composedDisplay() -> String {
+        itemOrder.compactMap { id in
+            completedByItem[id] ?? partialByItem[id]
+        }
+        .joined(separator: " ")
+        .trimmingCharacters(in: .whitespacesAndNewlines)
+    }
+
+    private func composedFinal() -> String {
+        itemOrder.compactMap { completedByItem[$0] }
+            .joined(separator: " ")
+            .trimmingCharacters(in: .whitespacesAndNewlines)
+    }
+
+    private func sendAppend(_ pcm: Data) async throws {
+        let audio = pcm.base64EncodedString()
+        try await sendJSON([
+            "type": "input_audio_buffer.append",
+            "audio": audio,
+        ])
+    }
+
+    private func sendJSON(_ body: [String: Any]) async throws {
+        guard JSONSerialization.isValidJSONObject(body),
+              let data = try? JSONSerialization.data(withJSONObject: body),
+              let string = String(data: data, encoding: .utf8) else {
+            throw CloudASRError.decoding("invalid realtime payload")
+        }
+        do {
+            try await wsTask.send(.string(string))
+        } catch {
+            throw CloudASRError.transport(error.localizedDescription)
+        }
+    }
+
+    private func throwIfFailed() throws {
+        let (error, done) = lock.withLock { (failure, finished) }
+        if let error { throw error }
+        if done { throw CloudASRError.transport("OpenAI realtime session cancelled") }
+    }
+
+    private func publishFailure(_ error: Error) {
+        lock.withLock { failure = error }
+        cancel()
+    }
+
+    private static func languageHint(from locale: Locale) -> String? {
+        let id = locale.identifier.lowercased()
+        if id.hasPrefix("zh") { return "zh" }
+        if id.hasPrefix("en") { return "en" }
+        if id.hasPrefix("ja") { return "ja" }
+        if id.hasPrefix("ko") { return "ko" }
+        return nil
+    }
+}

OSGKeyboardShared/Services/CloudASR/VolcengineCloudASRClient.swift

@@ -1,37 +1,36 @@
 // VolcengineCloudASRClient.swift
 // OSGKeyboard · Shared
 //
-// Volcengine SAUC bigmodel ASR client. The service uses a WebSocket with a
-// small custom binary frame wrapper; this file keeps that protocol isolated
-// from the HTTP-style cloud ASR clients.
+// Volcengine SAUC bigmodel ASR client. Utterance-level WebSocket session with
+// enable_nonstream (official two-pass): interim text for on-screen partials,
+// definite utterances for polish-ready finals.
 
 import Foundation
+import os
 
-struct VolcengineCloudASRClient: CloudASRTranscribing {
+struct VolcengineCloudASRClient: CloudASRTranscribing, CloudASRStreamingCapable {
     let apiKey: String
     let endpoint: String
     let resourceID: String
     let session: URLSession
 
-    private static let targetChunkBytes = 6_400 // 200 ms @ 16 kHz, 16-bit, mono.
-    private static let finalTimeout: TimeInterval = 12
+    static let targetChunkBytes = 6_400 // 200 ms @ 16 kHz, 16-bit, mono.
+    static let finalTimeout: TimeInterval = 12
     private static let hotwordCap = 80
 
     func prepare(dictionary: PersonalDictionary) async throws {}
 
-    func transcribe(
-        samples: [Float],
-        sampleRate: Int,
+    func openStreamingSession(
         locale: Locale,
-        dictionary: PersonalDictionary
-    ) async throws -> String {
-        guard !samples.isEmpty else { throw CloudASRError.emptyTranscript }
+        dictionary: PersonalDictionary,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) async throws -> any CloudASRStreamingSession {
+        _ = locale
         let credentials = try VolcengineCredentials.parse(
             apiKey: apiKey,
             fallbackResourceID: resolvedResourceID
         )
         let url = try resolvedEndpointURL()
-        let pcm = Self.pcm16Data(samples: samples)
         let connectID = UUID().uuidString
 
         var request = URLRequest(url: url)
@@ -43,52 +42,31 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
 
         let task = session.webSocketTask(with: request)
         task.resume()
-        defer {
-            task.cancel(with: .normalClosure, reason: nil)
-        }
-
-        let firstPayload = try Self.firstFramePayload(connectID: connectID, dictionary: dictionary)
-        try await send(
-            VolcengineFrame.build(
-                messageType: .fullClientRequest,
-                flags: .positiveSequence,
-                serialization: .json,
-                payload: firstPayload,
-                sequence: 1
-            ),
-            task: task
+        let live = VolcengineStreamingSession(
+            wsTask: task,
+            connectID: connectID,
+            dictionary: dictionary,
+            onPartial: onPartial
         )
+        try await live.start()
+        return live
+    }
 
-        var sequence = 2
-        var offset = 0
-        while offset < pcm.count {
-            let end = min(offset + Self.targetChunkBytes, pcm.count)
-            try await send(
-                VolcengineFrame.build(
-                    messageType: .audioOnlyRequest,
-                    flags: .positiveSequence,
-                    serialization: .none,
-                    payload: pcm.subdata(in: offset..<end),
-                    sequence: Int32(sequence)
-                ),
-                task: task
-            )
-            sequence += 1
-            offset = end
-        }
-
-        try await send(
-            VolcengineFrame.build(
-                messageType: .audioOnlyRequest,
-                flags: .negativeSequence,
-                serialization: .none,
-                payload: Data(),
-                sequence: -Int32(sequence)
-            ),
-            task: task
+    func transcribe(
+        samples: [Float],
+        sampleRate: Int,
+        locale: Locale,
+        dictionary: PersonalDictionary
+    ) async throws -> String {
+        _ = sampleRate
+        guard !samples.isEmpty else { throw CloudASRError.emptyTranscript }
+        let session = try await openStreamingSession(
+            locale: locale,
+            dictionary: dictionary,
+            onPartial: { _ in }
         )
-
-        let text = try await receiveFinalText(task: task)
+        try await session.append(samples: samples)
+        let text = try await session.finish()
         let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
         guard !trimmed.isEmpty else { throw CloudASRError.emptyTranscript }
         return trimmed
@@ -108,57 +86,7 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
         return url
     }
 
-    private func send(_ data: Data, task: URLSessionWebSocketTask) async throws {
-        do {
-            try await task.send(.data(data))
-        } catch {
-            throw CloudASRError.transport(error.localizedDescription)
-        }
-    }
-
-    private func receiveFinalText(task: URLSessionWebSocketTask) async throws -> String {
-        try await withThrowingTaskGroup(of: String.self) { group in
-            group.addTask {
-                var lastPartial = ""
-                while true {
-                    let message = try await task.receive()
-                    let data: Data
-                    switch message {
-                    case .data(let payload):
-                        data = payload
-                    case .string(let string):
-                        data = Data(string.utf8)
-                    @unknown default:
-                        continue
-                    }
-
-                    guard let frame = VolcengineFrame.parse(data) else { continue }
-                    if frame.messageType == .errorMessage {
-                        let body = String(data: frame.payload, encoding: .utf8) ?? ""
-                        let code = frame.errorCode ?? 0
-                        throw CloudASRError.transport("ASR error \(code): \(body)")
-                    }
-                    guard frame.messageType == .fullServerResponse else { continue }
-                    let parsedText = Self.text(from: frame.payload)
-                    if !parsedText.isEmpty {
-                        lastPartial = parsedText
-                    }
-                    if frame.isFinal {
-                        return parsedText.isEmpty ? lastPartial : parsedText
-                    }
-                }
-            }
-            group.addTask {
-                try await Task.sleep(nanoseconds: UInt64(Self.finalTimeout * 1_000_000_000))
-                throw CloudASRError.transport("Volcengine final result timed out")
-            }
-            let result = try await group.next()!
-            group.cancelAll()
-            return result
-        }
-    }
-
-    private static func firstFramePayload(
+    static func firstFramePayload(
         connectID: String,
         dictionary: PersonalDictionary
     ) throws -> Data {
@@ -168,6 +96,11 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
             "enable_punc": true,
             "show_utterances": true,
             "enable_speaker_info": true,
+            // Official two-pass: stream interim for UI, nostream re-decode per
+            // VAD sentence for definite polish-ready text (scheme A).
+            "enable_nonstream": true,
+            "end_window_size": 800,
+            "force_to_speech_time": 1_000,
         ]
         if let context = hotwordContext(dictionary: dictionary) {
             request["context"] = context
@@ -206,19 +139,7 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
         return String(data: data, encoding: .utf8)
     }
 
-    private static func pcm16Data(samples: [Float]) -> Data {
-        var data = Data()
-        data.reserveCapacity(samples.count * 2)
-        for sample in samples {
-            let scaled = sample * 32_767.0
-            let clipped = Swift.max(-32_768.0, Swift.min(32_767.0, scaled))
-            var littleEndian = Int16(clipped.rounded()).littleEndian
-            withUnsafeBytes(of: &littleEndian) { data.append(contentsOf: $0) }
-        }
-        return data
-    }
-
-    private static func text(from payload: Data) -> String {
+    static func displayText(from payload: Data) -> String {
         guard let json = try? JSONSerialization.jsonObject(with: payload) as? [String: Any],
               let result = normalizedResult(from: json) else {
             return ""
@@ -232,6 +153,22 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
         return result["text"] as? String ?? ""
     }
 
+    /// Prefer definite (two-pass) utterance text for polish input.
+    static func committedText(from payload: Data) -> String {
+        guard let json = try? JSONSerialization.jsonObject(with: payload) as? [String: Any],
+              let result = normalizedResult(from: json),
+              let utterances = result["utterances"] as? [[String: Any]],
+              !utterances.isEmpty else {
+            return ""
+        }
+        let definite = utterances.compactMap { utterance -> String? in
+            let isDefinite = utterance["definite"] as? Bool ?? false
+            guard isDefinite else { return nil }
+            return utterance["text"] as? String
+        }
+        return definite.joined()
+    }
+
     private static func normalizedResult(from json: [String: Any]) -> [String: Any]? {
         if let result = json["result"] as? [String: Any] {
             return result
@@ -246,6 +183,222 @@ struct VolcengineCloudASRClient: CloudASRTranscribing {
     }
 }
 
+// MARK: - Utterance session
+
+private final class VolcengineStreamingSession: CloudASRStreamingSession, @unchecked Sendable {
+    private let wsTask: URLSessionWebSocketTask
+    private let connectID: String
+    private let dictionary: PersonalDictionary
+    private let onPartial: @Sendable (String) -> Void
+    private let lock = OSAllocatedUnfairLock()
+    private var sequence: Int32 = 1
+    private var pcmBuffer = Data()
+    private var receiveTask: Task<Void, Never>?
+    private var failure: Error?
+    private var finished = false
+    private var lastDisplay = ""
+    private var lastCommitted = ""
+    private var sawServerFinal = false
+
+    init(
+        wsTask: URLSessionWebSocketTask,
+        connectID: String,
+        dictionary: PersonalDictionary,
+        onPartial: @escaping @Sendable (String) -> Void
+    ) {
+        self.wsTask = wsTask
+        self.connectID = connectID
+        self.dictionary = dictionary
+        self.onPartial = onPartial
+    }
+
+    func start() async throws {
+        let firstPayload = try VolcengineCloudASRClient.firstFramePayload(
+            connectID: connectID,
+            dictionary: dictionary
+        )
+        try await send(
+            VolcengineFrame.build(
+                messageType: .fullClientRequest,
+                flags: .positiveSequence,
+                serialization: .json,
+                payload: firstPayload,
+                sequence: 1
+            )
+        )
+        sequence = 2
+        receiveTask = Task { [weak self] in
+            await self?.receiveLoop()
+        }
+    }
+
+    func append(samples: [Float]) async throws {
+        try throwIfFailed()
+        let pcm = CloudASRStreamingPCM.pcm16LE(samples: samples)
+        let (frames, nextSequences): ([Data], [Int32]) = lock.withLock {
+            pcmBuffer.append(pcm)
+            var frames: [Data] = []
+            while pcmBuffer.count >= VolcengineCloudASRClient.targetChunkBytes {
+                let frame = pcmBuffer.prefix(VolcengineCloudASRClient.targetChunkBytes)
+                frames.append(Data(frame))
+                pcmBuffer.removeFirst(VolcengineCloudASRClient.targetChunkBytes)
+            }
+            let nextSequences: [Int32] = frames.indices.map { _ in
+                let seq = sequence
+                sequence += 1
+                return seq
+            }
+            return (frames, nextSequences)
+        }
+
+        for (frame, seq) in zip(frames, nextSequences) {
+            try await send(
+                VolcengineFrame.build(
+                    messageType: .audioOnlyRequest,
+                    flags: .positiveSequence,
+                    serialization: .none,
+                    payload: frame,
+                    sequence: seq
+                )
+            )
+        }
+    }
+
+    func finish() async throws -> String {
+        try throwIfFailed()
+        let (trailing, endSequence): (Data, Int32) = lock.withLock {
+            let trailing = pcmBuffer
+            pcmBuffer.removeAll(keepingCapacity: false)
+            let endSequence = sequence
+            sequence += 1
+            return (trailing, endSequence)
+        }
+
+        if !trailing.isEmpty {
+            try await send(
+                VolcengineFrame.build(
+                    messageType: .audioOnlyRequest,
+                    flags: .positiveSequence,
+                    serialization: .none,
+                    payload: trailing,
+                    sequence: endSequence
+                )
+            )
+        }
+
+        let negativeSeq = lock.withLock { () -> Int32 in
+            let seq = sequence
+            sequence += 1
+            return seq
+        }
+        try await send(
+            VolcengineFrame.build(
+                messageType: .audioOnlyRequest,
+                flags: .negativeSequence,
+                serialization: .none,
+                payload: Data(),
+                sequence: -negativeSeq
+            )
+        )
+
+        let deadline = Date().addingTimeInterval(VolcengineCloudASRClient.finalTimeout)
+        while Date() < deadline {
+            try throwIfFailed()
+            let snapshot = lock.withLock { (sawServerFinal, lastCommitted, lastDisplay) }
+            if snapshot.0 {
+                let text = snapshot.1.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
+                    ? snapshot.2
+                    : snapshot.1
+                let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
+                cancel()
+                if trimmed.isEmpty { throw CloudASRError.emptyTranscript }
+                return trimmed
+            }
+            try await Task.sleep(nanoseconds: 20_000_000)
+        }
+        cancel()
+        throw CloudASRError.transport("Volcengine final result timed out")
+    }
+
+    func cancel() {
+        receiveTask?.cancel()
+        wsTask.cancel(with: .normalClosure, reason: nil)
+        lock.withLock { finished = true }
+    }
+
+    private func receiveLoop() async {
+        while !Task.isCancelled {
+            let message: URLSessionWebSocketTask.Message
+            do {
+                message = try await wsTask.receive()
+            } catch {
+                publishFailure(CloudASRError.transport(error.localizedDescription))
+                return
+            }
+
+            let data: Data
+            switch message {
+            case .data(let payload):
+                data = payload
+            case .string(let string):
+                data = Data(string.utf8)
+            @unknown default:
+                continue
+            }
+
+            guard let frame = VolcengineFrame.parse(data) else { continue }
+            if frame.messageType == .errorMessage {
+                let body = String(data: frame.payload, encoding: .utf8) ?? ""
+                let code = frame.errorCode ?? 0
+                publishFailure(CloudASRError.transport("ASR error \(code): \(body)"))
+                return
+            }
+            guard frame.messageType == .fullServerResponse else { continue }
+
+            let display = VolcengineCloudASRClient.displayText(from: frame.payload)
+                .trimmingCharacters(in: .whitespacesAndNewlines)
+            let committed = VolcengineCloudASRClient.committedText(from: frame.payload)
+                .trimmingCharacters(in: .whitespacesAndNewlines)
+
+            let emit = lock.withLock { () -> String in
+                if !display.isEmpty {
+                    lastDisplay = display
+                }
+                if !committed.isEmpty {
+                    lastCommitted = committed
+                }
+                if frame.isFinal {
+                    sawServerFinal = true
+                }
+                return lastDisplay
+            }
+
+            if !emit.isEmpty {
+                onPartial(emit)
+            }
+        }
+    }
+
+    private func send(_ data: Data) async throws {
+        do {
+            try await wsTask.send(.data(data))
+        } catch {
+            throw CloudASRError.transport(error.localizedDescription)
+        }
+    }
+
+    private func throwIfFailed() throws {
+        let (error, done) = lock.withLock { (failure, finished) }
+        if let error { throw error }
+        if done { throw CloudASRError.transport("Volcengine session cancelled") }
+    }
+
+    private func publishFailure(_ error: Error) {
+        lock.withLock { failure = error }
+        cancel()
+    }
+}
+
 private struct VolcengineCredentials {
     let appID: String
     let accessToken: String