merge: tail audio ASR fix (P0+P1) into feature/polish-style-packs

Co-authored-by: Rocky <hkgood@users.noreply.github.com>
2026-07-26 10:34:05 +00:00
parent 9d8914fbf3 f0240c3088
commit f108460302
20 changed files with 741 additions and 45 deletions
@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 ## [Unreleased]

+### Fixed
+- **Flow tail ASR drop**: after mic stop, iOS Flow now uses a longer silence drain (350 ms), a fixed 150 ms post-roll, expanded final-chunk ASR recovery, and a partial transcript guard so weak trailing syllables are less likely to disappear from the result. / **Flow 尾音识别丢失**：松手后 iOS Flow 采用更长的静音排空（350 ms）、固定 150 ms 尾音保留、增强末块 ASR 恢复与 partial 兜底，降低弱尾音从结果中消失的概率。
+- **Flow batch ASR fallback**: when pipelined chunk output is clearly shorter than the live partial, the host re-transcribes the full utterance PCM captured during recording (Mac-style safety net). / **Flow 整句 ASR 兜底**：流水线拼接结果明显短于实时 partial 时，主 App 对录音期间累积的整段 PCM 重新识别（对齐 Mac 双保险）。
+
+### Changed
+- **Mac MLX tail drain**: streaming capture now uses the shared `FlowUtteranceEndCoordinator` (silence drain + post-roll) instead of an inline poll loop. / **Mac MLX 尾音排空**：流式采集改用 Shared 层 `FlowUtteranceEndCoordinator`（静音排空 + post-roll），替代内联轮询循环。
+
 ### Added
 - **Polish style packs**: choose a complete writing personality from the new iOS tab or Mac sidebar, create custom prompts, and sync selections and custom styles through iCloud. / **润色风格包**：可在 iOS 新 Tab 或 Mac 侧栏选择完整写作人格、创建自定义提示词，并通过 iCloud 同步选择与自定义风格。
 - **Mac MLX streaming ASR**: local dictation uses Qwen3-ASR via mlx-audio-swift with overlay partial preview, tail drain, vocabulary prompt, and polish-before-insert. / **Mac MLX 流式 ASR**：本地听写改用 mlx-audio-swift 的 Qwen3-ASR，支持浮层 partial 预览、尾部截断、词库 prompt 与润色后再插入。
@@ -64,6 +64,10 @@ final class FlowSessionManager: ObservableObject {
    private var chunkedPipeline: ChunkedUtterancePipeline?
    private var currentPartial = ""
    private var lastFinal = ""
+    /// Partial stitched text captured when the user stops recording.
+    private var bestPartialSnapshot = ""
+    /// Full utterance PCM for batch ASR fallback after pipelined chunking.
+    private var utterancePCMSamples: [Float] = []
    private var chunkWarnings: [String] = []
    private var lastReadyTraceSignature = ""
    private var lastCommandFingerprint = ""
@@ -356,6 +360,8 @@ final class FlowSessionManager: ObservableObject {
        sessionWarning = nil
        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []
        FlowSessionBridge.setHostReady(false)
    }
@@ -1101,6 +1107,8 @@ final class FlowSessionManager: ObservableObject {
        currentCommandSeq = commandSeq
        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []

        let localeId = store.localeId
@@ -1194,6 +1202,9 @@ final class FlowSessionManager: ObservableObject {
        refreshHostReady()
        FlowLiveActivityController.update(phase: .processing)

+        // Snapshot pipelined partial before drain — fallback if the final chunk ASR drops tail text.
+        bestPartialSnapshot = currentPartial.trimmingCharacters(in: .whitespacesAndNewlines)
+
        // Do NOT cancel `asrTask` or `asr` — drain trailing PCM, then finalize.

        // Capture ids now: a cancelled finalize must still clear *this*
@@ -1207,6 +1218,7 @@ final class FlowSessionManager: ObservableObject {
            guard let self else { return }
            let drainReport = await self.capture.endUtteranceAndDrain()
            FlowDiagnostics.logDrain(drainReport)
+            self.utterancePCMSamples = self.capture.consumeUtteranceSamples()
            await self.finalizeUtterance(
                sessionId: drainingSessionId,
                utteranceId: drainingUtteranceId,
@@ -1231,6 +1243,8 @@ final class FlowSessionManager: ObservableObject {
        capture.cancelUtterance()
        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []
        currentUtteranceId = nil
        currentCommandSeq = 0
@@ -1257,6 +1271,8 @@ final class FlowSessionManager: ObservableObject {
        capture.cancelUtterance()
        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []
        storeCurrentError(message, kind: kind)
        currentUtteranceId = nil
@@ -1279,6 +1295,8 @@ final class FlowSessionManager: ObservableObject {
        chunkedPipeline = nil
        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []
        storeCurrentError(message, kind: kind)
        currentUtteranceId = nil
@@ -1331,10 +1349,21 @@ final class FlowSessionManager: ObservableObject {
        let asrElapsed = Date().timeIntervalSince(pipelineStarted)
        FlowDiagnostics.log("ASR phase done in \(String(format: "%.1f", asrElapsed))s finalLen=\(lastFinal.count)")

-        var text = lastFinal.trimmingCharacters(in: .whitespacesAndNewlines)
+        var text = UtteranceTranscriptGuard.resolve(
+            stitchedFinal: lastFinal,
+            partialSnapshot: bestPartialSnapshot
+        )
        if text.isEmpty {
            text = currentPartial.trimmingCharacters(in: .whitespacesAndNewlines)
        }
+
+        if UtteranceBatchFallbackPolicy.shouldRunBatchFallback(
+            stitchedFinal: lastFinal,
+            partialSnapshot: bestPartialSnapshot
+        ), !utterancePCMSamples.isEmpty {
+            text = await runBatchASRFallback(currentText: text)
+        }
+        utterancePCMSamples = []
        guard !text.isEmpty else {
            let key = (asrTask?.isCancelled == true || Task.isCancelled)
                ? "flow.error.recognitionInterrupted"
@@ -1427,6 +1456,8 @@ final class FlowSessionManager: ObservableObject {

        currentPartial = ""
        lastFinal = ""
+        bestPartialSnapshot = ""
+        utterancePCMSamples = []
        chunkWarnings = []
        chunkedPipeline = nil
        debug("utterance finalized length=\(text.count)")
@@ -1555,6 +1586,49 @@ final class FlowSessionManager: ObservableObject {
        )
    }

+    /// Re-transcribe the full utterance PCM when pipelined chunking likely dropped tail text.
+    private func runBatchASRFallback(currentText: String) async -> String {
+        let samples = utterancePCMSamples
+        guard !samples.isEmpty else { return currentText }
+
+        let locale = SpeechLocaleResolver.resolve(store.localeId)
+        let stitched = lastFinal.trimmingCharacters(in: .whitespacesAndNewlines)
+        let partial = bestPartialSnapshot.trimmingCharacters(in: .whitespacesAndNewlines)
+
+        FlowDiagnostics.log(
+            "batch fallback start samples=\(samples.count) stitchedLen=\(stitched.count) partialLen=\(partial.count)"
+        )
+
+        let asrService = asr
+        let result = await Task.detached(priority: .userInitiated) { [asrService] in
+            await asrService.transcribeChunk(samples: samples, locale: locale)
+        }.value
+
+        switch result {
+        case .success(let batchText):
+            let trimmedBatch = batchText.trimmingCharacters(in: .whitespacesAndNewlines)
+            guard !trimmedBatch.isEmpty else { return currentText }
+            let resolved = UtteranceBatchFallbackPolicy.preferredTranscript(
+                batch: trimmedBatch,
+                stitchedFinal: stitched,
+                partialSnapshot: partial,
+                current: currentText
+            )
+            FlowPipelineDiagnostics.logBatchFallback(
+                sampleCount: samples.count,
+                stitchedLength: stitched.count,
+                partialLength: partial.count,
+                batchLength: trimmedBatch.count
+            )
+            return resolved
+        case .failure(let message):
+            FlowDiagnostics.log("batch fallback failed: \(message)")
+            return currentText
+        case .cancelled:
+            return currentText
+        }
+    }
+
    private func asrWaitTimeout() -> TimeInterval {
        // v0.2.0: local engine is iOS `SpeechAnalyzer` only, so the
        // previous Qwen3-specific timeout collapses into the shared
@@ -7,11 +7,7 @@ import Foundation
 import os

 enum MacMLXLiveCapture {
-    private static let tailDrainPolicy = FlowCaptureTailDrainPolicy(
-        silenceRMSThreshold: 0.015,
-        silenceDurationSeconds: 0.35,
-        maxDrainSeconds: 0.75
-    )
+    private static let tailDrainPolicy = FlowCaptureTailDrainPolicy.macMLX

    /// Runs MLX streaming ASR until `finishSignal` fires, then tail-drains and finalizes.
    static func run(
@@ -47,6 +43,7 @@ enum MacMLXLiveCapture {

            let drainTracker = FlowCaptureDrainTracker()
            let draining = OSAllocatedUnfairLock(initialState: false)
+            let drainComplete = OSAllocatedUnfairLock(initialState: false)
            let pendingFeed = OSAllocatedUnfairLock(initialState: [Float]())
            let feedIntervalSamples = 1_600 // 100 ms @ 16 kHz

@@ -60,6 +57,11 @@ enum MacMLXLiveCapture {
                    for await _ in finishSignal {
                        draining.withLock { $0 = true }
                        drainTracker.beginDrain()
+                        _ = await FlowUtteranceEndCoordinator.awaitTailCapture(
+                            tracker: drainTracker,
+                            policy: tailDrainPolicy
+                        )
+                        drainComplete.withLock { $0 = true }
                        break
                    }
                }
@@ -67,10 +69,12 @@ enum MacMLXLiveCapture {
                group.addTask {
                    for await snapshot in audioStream {
                        if Task.isCancelled { break }
+                        if drainComplete.withLock({ $0 }) { break }
                        if draining.withLock({ $0 }) {
-                            drainTracker.noteAudio(samples: snapshot.samples, policy: tailDrainPolicy)
-                            let decision = drainTracker.shouldFinish(policy: tailDrainPolicy)
-                            if decision.finished { break }
+                            drainTracker.noteAudio(
+                                samples: snapshot.samples,
+                                policy: tailDrainPolicy
+                            )
                        }
                        pendingFeed.withLock { buffer in
                            buffer.append(contentsOf: snapshot.samples)
@@ -101,6 +101,7 @@ public actor ChunkedUtterancePipeline {
        var processedChunks = 0
        var previousChunkSamples: [Float] = []
        var lastChunkSamples = 0
+        var didRetryEmptyFinal = false

        let feeder = Task {
            for await chunk in UtteranceStreamChunker.chunks(from: stream, config: config) {
@@ -118,20 +119,28 @@ public actor ChunkedUtterancePipeline {

            guard let chunk = await queue.dequeue() else { break }

+            if chunk.isLast && chunk.samples.isEmpty {
+                continue
+            }
+
            processedChunks += 1
            lastChunkSamples = chunk.samples.count

-            if chunk.isLast,
-               chunk.samples.count < config.minFinalChunkSamples,
-               processedChunks > 1,
-               !previousChunkSamples.isEmpty {
-                let mergedSamples = Array(previousChunkSamples.suffix(config.overlapSamples))
-                    + chunk.samples
-                let mergedResult = await transcribeChunk(samples: mergedSamples)
+            if let preMerge = FinalChunkRecovery.preMergePlan(
+                chunk: chunk,
+                processedChunks: processedChunks,
+                previousChunkSamples: previousChunkSamples,
+                config: config
+            ) {
+                FlowPipelineDiagnostics.logFinalChunkRecovery(
+                    action: "preMerge",
+                    chunkIndex: chunk.index
+                )
+                let mergedResult = await transcribeChunk(samples: preMerge.samples)
                switch mergedResult {
                case .success(let text):
                    stitcher.removeLastSegment()
-                    stitcher.append(index: max(0, chunk.index - 1), text: text)
+                    stitcher.append(index: preMerge.stitchIndex, text: text)
                    publishPartial(from: stitcher, onPartial: onPartial)
                case .failure(let message):
                    failedChunks += 1
@@ -153,8 +162,52 @@ public actor ChunkedUtterancePipeline {
            let result = await transcribeChunk(samples: chunk.samples)
            switch result {
            case .success(let text):
-                stitcher.append(index: chunk.index, text: text)
-                publishPartial(from: stitcher, onPartial: onPartial)
+                if chunk.isLast,
+                   !didRetryEmptyFinal,
+                   let retry = FinalChunkRecovery.emptyResultRetryPlan(
+                       chunk: chunk,
+                       previousChunkSamples: previousChunkSamples,
+                       config: config,
+                       asrText: text
+                   ) {
+                    didRetryEmptyFinal = true
+                    FlowPipelineDiagnostics.logFinalChunkRecovery(
+                        action: "emptyRetry",
+                        chunkIndex: chunk.index
+                    )
+                    let retryResult = await transcribeChunk(samples: retry.samples)
+                    switch retryResult {
+                    case .success(let retryText):
+                        let trimmed = retryText.trimmingCharacters(in: .whitespacesAndNewlines)
+                        if !trimmed.isEmpty {
+                            if retry.stitchIndex < chunk.index {
+                                stitcher.removeLastSegment()
+                            }
+                            stitcher.append(index: retry.stitchIndex, text: retryText)
+                            publishPartial(from: stitcher, onPartial: onPartial)
+                        } else {
+                            stitcher.append(index: chunk.index, text: text)
+                            publishPartial(from: stitcher, onPartial: onPartial)
+                        }
+                    case .failure(let message):
+                        stitcher.append(index: chunk.index, text: text)
+                        publishPartial(from: stitcher, onPartial: onPartial)
+                        failedChunks += 1
+                        chunkWarnings.append(
+                            SharedL10n.format(
+                                "error.asr.chunkFailed",
+                                chunk.index + 1,
+                                message
+                            )
+                        )
+                    case .cancelled:
+                        feeder.cancel()
+                        return .cancelled
+                    }
+                } else {
+                    stitcher.append(index: chunk.index, text: text)
+                    publishPartial(from: stitcher, onPartial: onPartial)
+                }
            case .failure(let message):
                failedChunks += 1
                chunkWarnings.append(
@@ -281,6 +281,9 @@ public final class FlowContinuousCapture {
    private let gate = OSAllocatedUnfairLock(initialState: UtteranceGatePhase.idle)
    private let drainTracker = FlowCaptureDrainTracker()
    private let tailSampleCounter = OSAllocatedUnfairLock(initialState: 0)
+    private let utterancePCMStore = FlowUtterancePCMStore(
+        maxSampleCount: Int(FlowSessionKeys.maxUtteranceDuration) * 16_000
+    )

    private var downsampler: AdaptiveDownsampler?
    private var targetFormat: AVAudioFormat?
@@ -412,6 +415,7 @@ public final class FlowContinuousCapture {
        let proof = audioProofStore
        let tracker = drainTracker
        let tailCounter = tailSampleCounter
+        let pcmStore = utterancePCMStore
        let policy = drainPolicy
        let tap = Self.makeAudioTapBlock(
            downsampler: downsampler,
@@ -422,6 +426,7 @@ public final class FlowContinuousCapture {
            streamRelay: relay,
            drainTracker: tracker,
            tailSampleCounter: tailCounter,
+            utterancePCMStore: pcmStore,
            drainPolicy: policy
        )
        // `format: nil` binds the tap to the input node's *live* format. Passing
@@ -645,6 +650,7 @@ public final class FlowContinuousCapture {
        let (stream, continuation) = AsyncStream<AudioBufferSnapshot>.makeStream()
        drainTracker.reset()
        tailSampleCounter.withLock { $0 = 0 }
+        utterancePCMStore.reset()
        // Bind the consumer before opening the gate so early tap frames
        // are not dropped on the floor.
        streamRelay.bind(continuation)
@@ -666,16 +672,11 @@ public final class FlowContinuousCapture {
        gate.withLock { $0 = .draining }
        drainTracker.beginDrain()

-        var endedBySilence = false
-        while true {
-            let decision = drainTracker.shouldFinish(policy: policy)
-            if decision.finished {
-                endedBySilence = decision.endedBySilence
-                break
-            }
-            if Task.isCancelled { break }
-            try? await Task.sleep(nanoseconds: FlowCaptureConstants.drainPollIntervalNs)
-        }
+        let timing = await FlowUtteranceEndCoordinator.awaitTailCapture(
+            tracker: drainTracker,
+            policy: policy,
+            pollIntervalNs: FlowCaptureConstants.drainPollIntervalNs
+        )

        // NOTE: We intentionally do NOT signal `.endOfStream` to the shared
        // downsampling converter here. `AVAudioConverter` is stateful: once its
@@ -692,8 +693,9 @@ public final class FlowContinuousCapture {
        let tailSamples = tailSampleCounter.withLock { $0 }
        let report = FlowCaptureDrainReport(
            drainDurationSeconds: drainTracker.elapsedSeconds(),
-            endedBySilence: endedBySilence,
-            tailSampleCount: tailSamples
+            endedBySilence: timing.endedBySilence,
+            tailSampleCount: tailSamples,
+            postRollDurationSeconds: timing.postRollDurationSeconds
        )
        drainTracker.reset()
        tailSampleCounter.withLock { $0 = 0 }
@@ -701,11 +703,17 @@ public final class FlowContinuousCapture {
        return report
    }

+    /// Returns the utterance PCM accumulated during the last recording cycle.
+    public func consumeUtteranceSamples() -> [Float] {
+        utterancePCMStore.consume()
+    }
+
    /// Immediate stop without tail drain (abort / session teardown).
    public func cancelUtterance() {
        gate.withLock { $0 = .idle }
        drainTracker.reset()
        tailSampleCounter.withLock { $0 = 0 }
+        utterancePCMStore.reset()
        streamRelay.finish()
    }

@@ -724,6 +732,7 @@ public final class FlowContinuousCapture {
        streamRelay: FlowCaptureStreamRelay,
        drainTracker: FlowCaptureDrainTracker,
        tailSampleCounter: OSAllocatedUnfairLock<Int>,
+        utterancePCMStore: FlowUtterancePCMStore,
        drainPolicy: FlowCaptureTailDrainPolicy
    ) -> @Sendable (AVAudioPCMBuffer, AVAudioTime) -> Void {
        return { buffer, _ in
@@ -743,6 +752,7 @@ public final class FlowContinuousCapture {
            let phase = gate.withLock { $0 }
            switch phase {
            case .recording, .draining:
+                utterancePCMStore.append(snapshot.samples)
                streamRelay.yield(snapshot)
                if phase == .draining {
                    drainTracker.noteAudio(samples: snapshot.samples, policy: drainPolicy)
@@ -558,12 +558,10 @@ public final class LiveDictationController: ObservableObject {
        drainTracker.beginDrain()

        let policy = FlowCaptureTailDrainPolicy.flowDefault
-        while true {
-            let decision = drainTracker.shouldFinish(policy: policy)
-            if decision.finished { break }
-            if Task.isCancelled { break }
-            try? await Task.sleep(nanoseconds: 20_000_000)
-        }
+        _ = await FlowUtteranceEndCoordinator.awaitTailCapture(
+            tracker: drainTracker,
+            policy: policy
+        )

        // Trailing speech is preserved by the live `.draining` forwarding
        // loop above. We deliberately do NOT signal `.endOfStream` to the
@@ -0,0 +1,47 @@
+// FinalChunkRecovery.swift
+// OSGKeyboard · Shared
+//
+// Recovery plans for pipelined utterance ASR when the final chunk is short,
+// empty, or straddles a chunk boundary.
+
+import Foundation
+
+public enum FinalChunkRecovery {
+
+    /// Samples and stitch index when the final chunk should be merged with
+    /// prior overlap *before* the first ASR pass.
+    public static func preMergePlan(
+        chunk: UtteranceAudioChunk,
+        processedChunks: Int,
+        previousChunkSamples: [Float],
+        config: FlowUtteranceChunkConfig
+    ) -> (samples: [Float], stitchIndex: Int)? {
+        guard chunk.isLast, !chunk.samples.isEmpty, !previousChunkSamples.isEmpty else {
+            return nil
+        }
+        guard chunk.samples.count < config.minFinalChunkSamples else { return nil }
+
+        let merged = Array(previousChunkSamples.suffix(config.overlapSamples)) + chunk.samples
+        return (merged, max(0, chunk.index - 1))
+    }
+
+    /// Retry plan when the final chunk had audio but ASR returned empty text.
+    public static func emptyResultRetryPlan(
+        chunk: UtteranceAudioChunk,
+        previousChunkSamples: [Float],
+        config: FlowUtteranceChunkConfig,
+        asrText: String
+    ) -> (samples: [Float], stitchIndex: Int)? {
+        guard chunk.isLast, !chunk.samples.isEmpty else { return nil }
+        guard asrText.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty else {
+            return nil
+        }
+
+        if previousChunkSamples.isEmpty {
+            return (chunk.samples, chunk.index)
+        }
+
+        let merged = Array(previousChunkSamples.suffix(config.overlapSamples)) + chunk.samples
+        return (merged, max(0, chunk.index - 1))
+    }
+}
@@ -16,22 +16,39 @@ public struct FlowCaptureTailDrainPolicy: Sendable, Equatable {
    public let silenceDurationSeconds: TimeInterval
    /// Hard cap so noisy environments cannot stall finalize forever.
    public let maxDrainSeconds: TimeInterval
+    /// Fixed post-roll after silence drain; independent of RMS (captures weak tails).
+    public let postRollSeconds: TimeInterval

    public init(
        silenceRMSThreshold: Float,
        silenceDurationSeconds: TimeInterval,
-        maxDrainSeconds: TimeInterval
+        maxDrainSeconds: TimeInterval,
+        postRollSeconds: TimeInterval = 0
    ) {
        self.silenceRMSThreshold = silenceRMSThreshold
        self.silenceDurationSeconds = silenceDurationSeconds
        self.maxDrainSeconds = maxDrainSeconds
+        self.postRollSeconds = postRollSeconds
    }

-    public static let flowDefault = FlowCaptureTailDrainPolicy(
+    /// iOS Flow host + keyboard utterance capture.
+    public static let iosFlow = FlowCaptureTailDrainPolicy(
        silenceRMSThreshold: 0.015,
-        silenceDurationSeconds: 0.25,
-        maxDrainSeconds: 1.5
+        silenceDurationSeconds: 0.35,
+        maxDrainSeconds: 1.5,
+        postRollSeconds: 0.15
    )
+
+    /// Mac MLX streaming live capture.
+    public static let macMLX = FlowCaptureTailDrainPolicy(
+        silenceRMSThreshold: 0.015,
+        silenceDurationSeconds: 0.35,
+        maxDrainSeconds: 0.75,
+        postRollSeconds: 0.15
+    )
+
+    /// Backward-compatible alias for iOS Flow defaults.
+    public static let flowDefault = iosFlow
 }

 /// Metrics emitted when tail drain completes (for diagnostics and tests).
@@ -39,21 +56,25 @@ public struct FlowCaptureDrainReport: Sendable, Equatable {
    public let drainDurationSeconds: Double
    public let endedBySilence: Bool
    public let tailSampleCount: Int
+    public let postRollDurationSeconds: Double

    public init(
        drainDurationSeconds: Double,
        endedBySilence: Bool,
-        tailSampleCount: Int
+        tailSampleCount: Int,
+        postRollDurationSeconds: Double = 0
    ) {
        self.drainDurationSeconds = drainDurationSeconds
        self.endedBySilence = endedBySilence
        self.tailSampleCount = tailSampleCount
+        self.postRollDurationSeconds = postRollDurationSeconds
    }

    public static let skipped = FlowCaptureDrainReport(
        drainDurationSeconds: 0,
        endedBySilence: false,
-        tailSampleCount: 0
+        tailSampleCount: 0,
+        postRollDurationSeconds: 0
    )
 }

@@ -9,7 +9,31 @@ import os
 public enum FlowPipelineDiagnostics {
    public static func logDrain(_ report: FlowCaptureDrainReport) {
        OSGLog.flow.info(
-            "tailDrain duration=\(report.drainDurationSeconds, format: .fixed(precision: 2))s silenceEnd=\(report.endedBySilence) tailSamples=\(report.tailSampleCount)"
+            "tailDrain duration=\(report.drainDurationSeconds, format: .fixed(precision: 2))s " +
+            "postRoll=\(report.postRollDurationSeconds, format: .fixed(precision: 2))s " +
+            "silenceEnd=\(report.endedBySilence) tailSamples=\(report.tailSampleCount)"
+        )
+    }
+
+    public static func logTranscriptGuardUsedPartial(finalLength: Int, partialLength: Int) {
+        OSGLog.flow.warning(
+            "transcriptGuard partial preferred finalLen=\(finalLength) partialLen=\(partialLength)"
+        )
+    }
+
+    public static func logFinalChunkRecovery(action: String, chunkIndex: Int) {
+        OSGLog.asr.info("finalChunkRecovery \(action) chunk=\(chunkIndex)")
+    }
+
+    public static func logBatchFallback(
+        sampleCount: Int,
+        stitchedLength: Int,
+        partialLength: Int,
+        batchLength: Int
+    ) {
+        OSGLog.flow.info(
+            "batchFallback samples=\(sampleCount) stitchedLen=\(stitchedLength) " +
+            "partialLen=\(partialLength) batchLen=\(batchLength)"
        )
    }

@@ -0,0 +1,66 @@
+// FlowUtteranceEndCoordinator.swift
+// OSGKeyboard · Shared
+//
+// Unified tail-drain + post-roll orchestration after the user stops recording.
+// Keeps the mic gate open through silence detection, then a fixed post-roll
+// window that does not depend on RMS (captures weak trailing syllables).
+
+import Foundation
+
+/// Outcome of `FlowUtteranceEndCoordinator.awaitTailCapture`.
+public struct FlowUtteranceEndTiming: Sendable, Equatable {
+    public let endedBySilence: Bool
+    public let postRollDurationSeconds: Double
+
+    public init(endedBySilence: Bool, postRollDurationSeconds: Double) {
+        self.endedBySilence = endedBySilence
+        self.postRollDurationSeconds = postRollDurationSeconds
+    }
+}
+
+public enum FlowUtteranceEndCoordinator {
+    /// Poll interval while waiting for silence / max drain (20 ms).
+    public static let pollIntervalNs: UInt64 = 20_000_000
+
+    /// Waits until trailing speech drains (silence or max cap), then sleeps
+    /// through a fixed post-roll window so weak tail audio still reaches ASR.
+    ///
+    /// Callers must keep forwarding PCM from the audio tap while this runs
+    /// (gate `.draining` or equivalent).
+    public static func awaitTailCapture(
+        tracker: FlowCaptureDrainTracker,
+        policy: FlowCaptureTailDrainPolicy,
+        pollIntervalNs: UInt64 = pollIntervalNs
+    ) async -> FlowUtteranceEndTiming {
+        var endedBySilence = false
+        while true {
+            let decision = tracker.shouldFinish(policy: policy)
+            if decision.finished {
+                endedBySilence = decision.endedBySilence
+                break
+            }
+            if Task.isCancelled { break }
+            try? await Task.sleep(nanoseconds: pollIntervalNs)
+        }
+
+        let postRoll = await runPostRoll(policy: policy, pollIntervalNs: pollIntervalNs)
+        return FlowUtteranceEndTiming(
+            endedBySilence: endedBySilence,
+            postRollDurationSeconds: postRoll
+        )
+    }
+
+    private static func runPostRoll(
+        policy: FlowCaptureTailDrainPolicy,
+        pollIntervalNs: UInt64
+    ) async -> Double {
+        guard policy.postRollSeconds > 0 else { return 0 }
+        let started = Date()
+        let deadline = started.addingTimeInterval(policy.postRollSeconds)
+        while Date() < deadline {
+            if Task.isCancelled { break }
+            try? await Task.sleep(nanoseconds: pollIntervalNs)
+        }
+        return max(0, Date().timeIntervalSince(started))
+    }
+}
@@ -0,0 +1,46 @@
+// FlowUtterancePCMStore.swift
+// OSGKeyboard · Shared
+//
+// Thread-safe rolling buffer of 16 kHz mono utterance PCM for whole-utterance
+// batch ASR fallback when pipelined chunking drops weak tail segments.
+
+import Foundation
+
+public final class FlowUtterancePCMStore: @unchecked Sendable {
+    private let lock = OSAllocatedUnfairLock()
+    private var samples: [Float] = []
+    private let maxSampleCount: Int
+
+    public init(maxSampleCount: Int) {
+        self.maxSampleCount = max(1, maxSampleCount)
+    }
+
+    public func reset() {
+        lock.withLock {
+            samples.removeAll(keepingCapacity: false)
+        }
+    }
+
+    public func append(_ chunk: [Float]) {
+        guard !chunk.isEmpty else { return }
+        lock.withLock {
+            samples.append(contentsOf: chunk)
+            if samples.count > maxSampleCount {
+                samples.removeFirst(samples.count - maxSampleCount)
+            }
+        }
+    }
+
+    public var sampleCount: Int {
+        lock.withLock { samples.count }
+    }
+
+    /// Returns accumulated samples and clears the store.
+    public func consume() -> [Float] {
+        lock.withLock {
+            let out = samples
+            samples.removeAll(keepingCapacity: false)
+            return out
+        }
+    }
+}
@@ -0,0 +1,41 @@
+// UtteranceBatchFallbackPolicy.swift
+// OSGKeyboard · Shared
+//
+// Decides when to re-run ASR on the full utterance PCM after pipelined
+// chunking, and how to pick the best transcript among candidates.
+
+import Foundation
+
+public enum UtteranceBatchFallbackPolicy {
+    public static let defaultCharacterAdvantage = UtteranceTranscriptGuard.defaultPartialAdvantage
+
+    /// True when chunked output likely lost trailing content vs the live partial.
+    public static func shouldRunBatchFallback(
+        stitchedFinal: String,
+        partialSnapshot: String,
+        minimumCharacterAdvantage: Int = defaultCharacterAdvantage
+    ) -> Bool {
+        let final = stitchedFinal.trimmingCharacters(in: .whitespacesAndNewlines)
+        let partial = partialSnapshot.trimmingCharacters(in: .whitespacesAndNewlines)
+
+        if final.isEmpty, !partial.isEmpty { return true }
+        if partial.isEmpty { return false }
+        return partial.count >= final.count + minimumCharacterAdvantage
+    }
+
+    /// Prefer the longest non-empty transcript after batch ASR completes.
+    public static func preferredTranscript(
+        batch: String,
+        stitchedFinal: String,
+        partialSnapshot: String,
+        current: String
+    ) -> String {
+        let candidates = [batch, current, stitchedFinal, partialSnapshot]
+            .map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
+            .filter { !$0.isEmpty }
+        guard let best = candidates.max(by: { $0.count < $1.count }) else {
+            return current.trimmingCharacters(in: .whitespacesAndNewlines)
+        }
+        return best
+    }
+}
@@ -56,7 +56,11 @@ public enum UtteranceStreamChunker {
                } else if chunkIndex == 0 {
                    // Empty utterance — no chunks.
                } else {
-                    // Stream ended exactly on boundary; mark prior path complete.
+                    // Stream ended exactly on a chunk boundary; prior emit holds
+                    // all tail audio. Marker so FinalChunkRecovery paths run.
+                    continuation.yield(
+                        UtteranceAudioChunk(index: chunkIndex, samples: [], isLast: true)
+                    )
                }

                continuation.finish()
@@ -0,0 +1,35 @@
+// UtteranceTranscriptGuard.swift
+// OSGKeyboard · Shared
+//
+// Chooses the best available transcript when pipelined final text may have
+// dropped a weak tail segment.
+
+import Foundation
+
+public enum UtteranceTranscriptGuard {
+    /// Partial must exceed final by at least this many characters to win.
+    public static let defaultPartialAdvantage = 8
+
+    /// Prefer `stitchedFinal` unless empty or clearly shorter than the live
+    /// partial snapshot taken at mic stop.
+    public static func resolve(
+        stitchedFinal: String,
+        partialSnapshot: String,
+        minimumPartialAdvantage: Int = defaultPartialAdvantage
+    ) -> String {
+        let final = stitchedFinal.trimmingCharacters(in: .whitespacesAndNewlines)
+        let partial = partialSnapshot.trimmingCharacters(in: .whitespacesAndNewlines)
+
+        if final.isEmpty { return partial }
+        if partial.isEmpty { return final }
+
+        if partial.count >= final.count + minimumPartialAdvantage {
+            FlowPipelineDiagnostics.logTranscriptGuardUsedPartial(
+                finalLength: final.count,
+                partialLength: partial.count
+            )
+            return partial
+        }
+        return final
+    }
+}
@@ -115,6 +115,34 @@ final class ChunkedUtterancePipelineTests: XCTestCase {
        }
        XCTAssertTrue(success.text.contains("merged"))
    }
+
+    func testPipelineRetriesEmptyFinalChunkWithOverlap() async {
+        let config = FlowUtteranceChunkConfig(
+            maxChunkDurationSeconds: 0.05,
+            overlapDurationSeconds: 10,
+            pauseExtensionMaxSeconds: 0,
+            pauseRMSThreshold: 0.02,
+            minFinalChunkDurationSeconds: 0.05,
+            sampleRate: 1_000
+        )
+        let asr = EmptyFinalRetryStubASR()
+        let pipeline = ChunkedUtterancePipeline(
+            asr: asr,
+            locale: Locale(identifier: "zh-Hans"),
+            config: config
+        )
+
+        let (stream, continuation) = AsyncStream<AudioBufferSnapshot>.makeStream()
+        continuation.yield(AudioBufferSnapshot(samples: [Float](repeating: 0.1, count: 80), sampleRate: 1_000))
+        continuation.yield(AudioBufferSnapshot(samples: [Float](repeating: 0.1, count: 80), sampleRate: 1_000))
+        continuation.finish()
+
+        let outcome = await pipeline.transcribe(stream: stream) { _ in }
+        guard case .success(let success) = outcome else {
+            return XCTFail("expected success, got \(outcome)")
+        }
+        XCTAssertTrue(success.text.contains("recovered-tail"))
+    }
 }

 private struct FailingSecondChunkASR: ASRService, @unchecked Sendable {
@@ -171,3 +199,32 @@ private struct ShortFinalMergeStubASR: ASRService, @unchecked Sendable {
        return .success("short")
    }
 }
+
+private struct EmptyFinalRetryStubASR: ASRService, @unchecked Sendable {
+    private let callIndex = OSAllocatedUnfairLock(initialState: 0)
+
+    func transcribe(
+        stream: AsyncStream<AudioBufferSnapshot>,
+        locale: Locale
+    ) -> AsyncStream<ASREvent> {
+        AsyncStream { $0.finish() }
+    }
+
+    func cancel() {}
+
+    func transcribeChunk(samples: [Float], locale: Locale) async -> ASRChunkResult {
+        _ = locale
+        let current = callIndex.withLock { state in
+            let value = state
+            state += 1
+            return value
+        }
+        if current == 0 {
+            return .success("head")
+        }
+        if current == 1 {
+            return .success("")
+        }
+        return .success("recovered-tail")
+    }
+}
@@ -0,0 +1,74 @@
+// FinalChunkRecoveryTests.swift
+// OSGKeyboardTests
+
+import XCTest
+@testable import OSGKeyboardShared
+
+final class FinalChunkRecoveryTests: XCTestCase {
+
+    private let config = FlowUtteranceChunkConfig(
+        maxChunkDurationSeconds: 5.0,
+        overlapDurationSeconds: 0.5,
+        pauseExtensionMaxSeconds: 2,
+        pauseRMSThreshold: 0.015,
+        minFinalChunkDurationSeconds: 0.8,
+        sampleRate: 16_000
+    )
+
+    func testPreMergePlanForShortFinalChunk() {
+        let chunk = UtteranceAudioChunk(
+            index: 1,
+            samples: [Float](repeating: 0.1, count: 4_000),
+            isLast: true
+        )
+        let previous = [Float](repeating: 0.2, count: 80_000)
+
+        let plan = FinalChunkRecovery.preMergePlan(
+            chunk: chunk,
+            processedChunks: 2,
+            previousChunkSamples: previous,
+            config: config
+        )
+
+        XCTAssertNotNil(plan)
+        XCTAssertGreaterThan(plan?.samples.count ?? 0, chunk.samples.count)
+        XCTAssertEqual(plan?.stitchIndex, 0)
+    }
+
+    func testEmptyResultRetryPlanUsesOverlapWhenPriorChunkExists() {
+        let chunk = UtteranceAudioChunk(
+            index: 1,
+            samples: [Float](repeating: 0.1, count: 20_000),
+            isLast: true
+        )
+        let previous = [Float](repeating: 0.2, count: 80_000)
+
+        let plan = FinalChunkRecovery.emptyResultRetryPlan(
+            chunk: chunk,
+            previousChunkSamples: previous,
+            config: config,
+            asrText: "  "
+        )
+
+        XCTAssertNotNil(plan)
+        XCTAssertGreaterThan(plan?.samples.count ?? 0, chunk.samples.count)
+    }
+
+    func testEmptyResultRetryPlanRetriesSingleChunkSamples() {
+        let chunk = UtteranceAudioChunk(
+            index: 0,
+            samples: [Float](repeating: 0.1, count: 20_000),
+            isLast: true
+        )
+
+        let plan = FinalChunkRecovery.emptyResultRetryPlan(
+            chunk: chunk,
+            previousChunkSamples: [],
+            config: config,
+            asrText: ""
+        )
+
+        XCTAssertEqual(plan?.samples.count, chunk.samples.count)
+        XCTAssertEqual(plan?.stitchIndex, 0)
+    }
+}
@@ -0,0 +1,35 @@
+// FlowUtteranceEndCoordinatorTests.swift
+// OSGKeyboardTests
+
+import XCTest
+@testable import OSGKeyboardShared
+
+final class FlowUtteranceEndCoordinatorTests: XCTestCase {
+
+    func testAwaitTailCaptureRunsPostRollAfterSilenceDrain() async {
+        let policy = FlowCaptureTailDrainPolicy(
+            silenceRMSThreshold: 0.02,
+            silenceDurationSeconds: 0.05,
+            maxDrainSeconds: 1.0,
+            postRollSeconds: 0.08
+        )
+        let tracker = FlowCaptureDrainTracker()
+        let start = Date().timeIntervalSince1970
+        tracker.beginDrain(now: start)
+
+        let timing = await FlowUtteranceEndCoordinator.awaitTailCapture(
+            tracker: tracker,
+            policy: policy,
+            pollIntervalNs: 5_000_000
+        )
+
+        XCTAssertTrue(timing.endedBySilence)
+        XCTAssertGreaterThanOrEqual(timing.postRollDurationSeconds, 0.07)
+    }
+
+    func testIOSFlowPresetUsesLongerSilenceAndPostRoll() {
+        XCTAssertEqual(FlowCaptureTailDrainPolicy.iosFlow.silenceDurationSeconds, 0.35)
+        XCTAssertEqual(FlowCaptureTailDrainPolicy.iosFlow.postRollSeconds, 0.15)
+        XCTAssertEqual(FlowCaptureTailDrainPolicy.flowDefault, FlowCaptureTailDrainPolicy.iosFlow)
+    }
+}
@@ -0,0 +1,23 @@
+// FlowUtterancePCMStoreTests.swift
+// OSGKeyboardTests
+
+import XCTest
+@testable import OSGKeyboardShared
+
+final class FlowUtterancePCMStoreTests: XCTestCase {
+
+    func testAppendAndConsume() {
+        let store = FlowUtterancePCMStore(maxSampleCount: 100)
+        store.append([1, 2, 3])
+        store.append([4, 5])
+        XCTAssertEqual(store.sampleCount, 5)
+        XCTAssertEqual(store.consume(), [1, 2, 3, 4, 5])
+        XCTAssertEqual(store.sampleCount, 0)
+    }
+
+    func testTrimsOldestWhenOverCap() {
+        let store = FlowUtterancePCMStore(maxSampleCount: 4)
+        store.append([1, 2, 3, 4, 5])
+        XCTAssertEqual(store.consume(), [2, 3, 4, 5])
+    }
+}
@@ -0,0 +1,45 @@
+// UtteranceBatchFallbackPolicyTests.swift
+// OSGKeyboardTests
+
+import XCTest
+@testable import OSGKeyboardShared
+
+final class UtteranceBatchFallbackPolicyTests: XCTestCase {
+
+    func testShouldRunWhenPartialClearlyLonger() {
+        XCTAssertTrue(
+            UtteranceBatchFallbackPolicy.shouldRunBatchFallback(
+                stitchedFinal: "今天很好",
+                partialSnapshot: "今天很好，我们一起去公园吧"
+            )
+        )
+    }
+
+    func testShouldRunWhenFinalEmptyButPartialPresent() {
+        XCTAssertTrue(
+            UtteranceBatchFallbackPolicy.shouldRunBatchFallback(
+                stitchedFinal: "",
+                partialSnapshot: "最后一段"
+            )
+        )
+    }
+
+    func testShouldNotRunWhenPartialNotLonger() {
+        XCTAssertFalse(
+            UtteranceBatchFallbackPolicy.shouldRunBatchFallback(
+                stitchedFinal: "今天很好，我们一起去公园吧",
+                partialSnapshot: "今天很好"
+            )
+        )
+    }
+
+    func testPreferredTranscriptPicksLongestCandidate() {
+        let resolved = UtteranceBatchFallbackPolicy.preferredTranscript(
+            batch: "今天很好，我们一起去公园吧",
+            stitchedFinal: "今天很好",
+            partialSnapshot: "今天很好，我们",
+            current: "今天很好，我们"
+        )
+        XCTAssertEqual(resolved, "今天很好，我们一起去公园吧")
+    }
+}
@@ -0,0 +1,32 @@
+// UtteranceTranscriptGuardTests.swift
+// OSGKeyboardTests
+
+import XCTest
+@testable import OSGKeyboardShared
+
+final class UtteranceTranscriptGuardTests: XCTestCase {
+
+    func testResolvePrefersPartialWhenClearlyLonger() {
+        let resolved = UtteranceTranscriptGuard.resolve(
+            stitchedFinal: "今天天气很好",
+            partialSnapshot: "今天天气很好，我们一起去公园吧"
+        )
+        XCTAssertEqual(resolved, "今天天气很好，我们一起去公园吧")
+    }
+
+    func testResolveKeepsFinalWhenPartialIsNotLonger() {
+        let resolved = UtteranceTranscriptGuard.resolve(
+            stitchedFinal: "今天天气很好，我们一起去公园吧",
+            partialSnapshot: "今天天气很好"
+        )
+        XCTAssertEqual(resolved, "今天天气很好，我们一起去公园吧")
+    }
+
+    func testResolveUsesPartialWhenFinalEmpty() {
+        let resolved = UtteranceTranscriptGuard.resolve(
+            stitchedFinal: "",
+            partialSnapshot: "最后一段 partial"
+        )
+        XCTAssertEqual(resolved, "最后一段 partial")
+    }
+}