diff --git a/CHANGELOG.md b/CHANGELOG.md index dec0425..61e7406 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - **Keyboard language label**: `PrimaryLanguage` set to `mis` so Settings no longer shows a misleading “English” subtitle under OSGKeyboard. / **键盘语言标签**：`PrimaryLanguage` 设为 `mis`，系统设置中 OSGKeyboard 下不再显示误导性的「英文」副标题。 - **Flow ASR pipelining**: shorter first chunk (2.5s) and 5s follow-ups so short utterances start on-device recognition while still recording; session-level ASR warmup and format cache reuse; live partials mirrored to the keyboard transcript line. / **Flow ASR 流水线**：首块 2.5 秒、后续 5 秒，短句录音期间即开始端侧识别；会话级 ASR 预热与格式缓存复用；实时 partial 同步到键盘转写行。 +- **Flow tail drain**: after mic stop, capture drains trailing PCM (silence-detected, capped) before finishing the ASR stream; host finalize awaits drain; short final chunks re-transcribed with prior overlap; stitcher safe fallback when overlap merge would drop content. / **Flow 尾音排空**：停止录音后先排空尾部 PCM（静音检测 + 上限）再结束 ASR 流；主 App finalize 等待排空；过短末块与上一块 overlap 合并重识别；拼接误删时回退为安全合并。 ## [0.4.0] - 2026-07-05 diff --git a/OSGKeyboard/Services/FlowDiagnostics.swift b/OSGKeyboard/Services/FlowDiagnostics.swift index e649d27..8611e6a 100644 --- a/OSGKeyboard/Services/FlowDiagnostics.swift +++ b/OSGKeyboard/Services/FlowDiagnostics.swift @@ -11,4 +11,8 @@ enum FlowDiagnostics { static func log(_ message: String) { OSGLog.flow.info("\(message, privacy: .public)") } + + static func logDrain(_ report: FlowCaptureDrainReport) { + FlowPipelineDiagnostics.logDrain(report) + } } diff --git a/OSGKeyboard/Services/FlowSessionManager.swift b/OSGKeyboard/Services/FlowSessionManager.swift index c8562e6..dd5c7e4 100644 --- a/OSGKeyboard/Services/FlowSessionManager.swift +++ b/OSGKeyboard/Services/FlowSessionManager.swift @@ -477,21 +477,21 @@ final class FlowSessionManager: ObservableObject { // Close the mic gate first, then mark processing before dropping the // recording flag so the poll loop cannot start a second utterance. - capture.endUtterance() FlowSessionBridge.setRecordingState(.processing) isUtteranceRecording = false isUtteranceProcessing = true FlowLiveActivityController.update(phase: .processing) - // Do NOT cancel `asrTask` or `asr` — the preview pipeline relies on - // the consumer staying alive until `.final` lands (see - // `PreviewASRControllerStateTests`). + // Do NOT cancel `asrTask` or `asr` — drain trailing PCM, then finalize. finalizeTask?.cancel() finalizeTask = Task { @MainActor [weak self] in - await self?.finalizeUtterance() + guard let self else { return } + let drainReport = await self.capture.endUtteranceAndDrain() + FlowDiagnostics.logDrain(drainReport) + await self.finalizeUtterance() } - debug("utterance stopped, finalizing") + debug("utterance stopped, draining tail") } private func abortUtterance() { diff --git a/OSGKeyboardShared/Models/FlowUtteranceChunkConfig.swift b/OSGKeyboardShared/Models/FlowUtteranceChunkConfig.swift index eba6476..fc6e3f7 100644 --- a/OSGKeyboardShared/Models/FlowUtteranceChunkConfig.swift +++ b/OSGKeyboardShared/Models/FlowUtteranceChunkConfig.swift @@ -16,6 +16,8 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { public let pauseExtensionMaxSeconds: TimeInterval /// RMS below this is treated as a pause candidate (Float32 mono @ 16 kHz). public let pauseRMSThreshold: Float + /// Final chunk shorter than this is re-transcribed merged with the prior tail overlap. + public let minFinalChunkDurationSeconds: TimeInterval public let sampleRate: Int public init( @@ -24,6 +26,7 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { overlapDurationSeconds: TimeInterval, pauseExtensionMaxSeconds: TimeInterval, pauseRMSThreshold: Float, + minFinalChunkDurationSeconds: TimeInterval = 0.8, sampleRate: Int ) { self.firstChunkDurationSeconds = firstChunkDurationSeconds @@ -31,6 +34,7 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { self.overlapDurationSeconds = overlapDurationSeconds self.pauseExtensionMaxSeconds = pauseExtensionMaxSeconds self.pauseRMSThreshold = pauseRMSThreshold + self.minFinalChunkDurationSeconds = minFinalChunkDurationSeconds self.sampleRate = sampleRate } @@ -40,6 +44,7 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { overlapDurationSeconds: TimeInterval, pauseExtensionMaxSeconds: TimeInterval, pauseRMSThreshold: Float, + minFinalChunkDurationSeconds: TimeInterval = 0.8, sampleRate: Int ) { self.firstChunkDurationSeconds = maxChunkDurationSeconds @@ -47,6 +52,7 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { self.overlapDurationSeconds = overlapDurationSeconds self.pauseExtensionMaxSeconds = pauseExtensionMaxSeconds self.pauseRMSThreshold = pauseRMSThreshold + self.minFinalChunkDurationSeconds = minFinalChunkDurationSeconds self.sampleRate = sampleRate } @@ -75,6 +81,10 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { Int(pauseExtensionMaxSeconds * Double(sampleRate)) } + public var minFinalChunkSamples: Int { + Int(minFinalChunkDurationSeconds * Double(sampleRate)) + } + /// Default for keyboard Flow utterances (≤ 3 min, pipelined ASR). public static let flowDefault = FlowUtteranceChunkConfig( firstChunkDurationSeconds: 2.5, @@ -82,6 +92,7 @@ public struct FlowUtteranceChunkConfig: Sendable, Equatable { overlapDurationSeconds: 0.5, pauseExtensionMaxSeconds: 2, pauseRMSThreshold: 0.015, + minFinalChunkDurationSeconds: 0.8, sampleRate: 16_000 ) } diff --git a/OSGKeyboardShared/Services/ChunkedUtterancePipeline.swift b/OSGKeyboardShared/Services/ChunkedUtterancePipeline.swift index eb0d5f9..696f28b 100644 --- a/OSGKeyboardShared/Services/ChunkedUtterancePipeline.swift +++ b/OSGKeyboardShared/Services/ChunkedUtterancePipeline.swift @@ -99,6 +99,8 @@ public actor ChunkedUtterancePipeline { var chunkWarnings: [String] = [] var failedChunks = 0 var processedChunks = 0 + var previousChunkSamples: [Float] = [] + var lastChunkSamples = 0 let feeder = Task { for await chunk in UtteranceStreamChunker.chunks(from: stream, config: config) { @@ -117,19 +119,42 @@ public actor ChunkedUtterancePipeline { guard let chunk = await queue.dequeue() else { break } processedChunks += 1 - let asr = self.asr - let locale = self.locale - let result = await Task.detached(priority: .userInitiated) { - await asr.transcribeChunk(samples: chunk.samples, locale: locale) - }.value + 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) + switch mergedResult { + case .success(let text): + stitcher.removeLastSegment() + stitcher.append(index: max(0, chunk.index - 1), text: text) + publishPartial(from: stitcher, onPartial: onPartial) + case .failure(let message): + failedChunks += 1 + chunkWarnings.append( + SharedL10n.format( + "error.asr.chunkFailed", + chunk.index + 1, + message + ) + ) + case .cancelled: + feeder.cancel() + return .cancelled + } + previousChunkSamples = chunk.samples + continue + } + + let result = await transcribeChunk(samples: chunk.samples) switch result { case .success(let text): stitcher.append(index: chunk.index, text: text) - let partial = stitcher.composed() - if !partial.isEmpty { - onPartial(partial) - } + publishPartial(from: stitcher, onPartial: onPartial) case .failure(let message): failedChunks += 1 chunkWarnings.append( @@ -143,11 +168,20 @@ public actor ChunkedUtterancePipeline { feeder.cancel() return .cancelled } + + previousChunkSamples = chunk.samples } _ = await feeder.value - let finalText = stitcher.composed().trimmingCharacters(in: .whitespacesAndNewlines) + let finalText = stitcher.composedSafely().trimmingCharacters(in: .whitespacesAndNewlines) + FlowPipelineDiagnostics.logChunkFinalize( + chunkCount: processedChunks, + lastChunkSamples: lastChunkSamples, + stitchedLength: finalText.count, + chunkWarnings: chunkWarnings.count + ) + if finalText.isEmpty { if failedChunks > 0, processedChunks == failedChunks { return .failure(SharedL10n.string("error.asr.noSpeech")) @@ -157,4 +191,22 @@ public actor ChunkedUtterancePipeline { return .success(ChunkedUtteranceSuccess(text: finalText, chunkWarnings: chunkWarnings)) } + + private func transcribeChunk(samples: [Float]) async -> ASRChunkResult { + let asr = self.asr + let locale = self.locale + return await Task.detached(priority: .userInitiated) { + await asr.transcribeChunk(samples: samples, locale: locale) + }.value + } + + private func publishPartial( + from stitcher: UtteranceTranscriptStitcher, + onPartial: @Sendable (String) -> Void + ) { + let partial = stitcher.composedSafely() + if !partial.isEmpty { + onPartial(partial) + } + } } diff --git a/OSGKeyboardShared/Services/FlowContinuousCapture.swift b/OSGKeyboardShared/Services/FlowContinuousCapture.swift index e83518b..ae52601 100644 --- a/OSGKeyboardShared/Services/FlowContinuousCapture.swift +++ b/OSGKeyboardShared/Services/FlowContinuousCapture.swift @@ -14,6 +14,13 @@ import os private enum FlowCaptureConstants { static let levelBarCount = 24 static let targetSampleRate: Double = 16_000 + static let drainPollIntervalNs: UInt64 = 20_000_000 +} + +private enum UtteranceGatePhase: Equatable { + case idle + case recording + case draining } /// Thread-safe relay for utterance-scoped ASR snapshots. @@ -158,7 +165,14 @@ public final class FlowContinuousCapture { private let streamRelay = FlowCaptureStreamRelay() private let prerollStore = FlowPrerollStore() private let levelStore = FlowLevelStore(barCount: FlowCaptureConstants.levelBarCount) - private let isUtteranceActive = OSAllocatedUnfairLock(initialState: false) + private let gate = OSAllocatedUnfairLock(initialState: UtteranceGatePhase.idle) + private let drainTracker = FlowCaptureDrainTracker() + private let tailSampleCounter = OSAllocatedUnfairLock(initialState: 0) + + private var audioConverter: AVAudioConverter? + private var targetFormat: AVAudioFormat? + private var hwFormat: AVAudioFormat? + private var drainPolicy = FlowCaptureTailDrainPolicy.flowDefault private var didInstallTap = false private var isRunning = false @@ -197,15 +211,15 @@ public final class FlowContinuousCapture { } let inputNode = audioEngine.inputNode - let hwFormat = inputNode.outputFormat(forBus: 0) - guard hwFormat.sampleRate > 0, hwFormat.channelCount > 0 else { + let hardwareFormat = inputNode.outputFormat(forBus: 0) + guard hardwareFormat.sampleRate > 0, hardwareFormat.channelCount > 0 else { throw StartError.invalidHardwareFormat( - sampleRate: hwFormat.sampleRate, - channels: Int(hwFormat.channelCount) + sampleRate: hardwareFormat.sampleRate, + channels: Int(hardwareFormat.channelCount) ) } - guard let targetFormat = AVAudioFormat( + guard let resolvedTargetFormat = AVAudioFormat( commonFormat: .pcmFormatFloat32, sampleRate: FlowCaptureConstants.targetSampleRate, channels: 1, @@ -213,25 +227,40 @@ public final class FlowContinuousCapture { ) else { throw StartError.formatCreateFailed } - guard let converter = AVAudioConverter(from: hwFormat, to: targetFormat) else { + guard let converter = AVAudioConverter(from: hardwareFormat, to: resolvedTargetFormat) else { throw StartError.converterCreateFailed } + audioConverter = converter + targetFormat = resolvedTargetFormat + hwFormat = hardwareFormat + // Rebuild the tap so its bound hardware format matches the new route. if didInstallTap { inputNode.removeTap(onBus: 0) didInstallTap = false } + + let gateLock = gate + let relay = streamRelay + let preroll = prerollStore + let levels = levelStore + let tracker = drainTracker + let tailCounter = tailSampleCounter + let policy = drainPolicy let tap = Self.makeAudioTapBlock( converter: converter, - targetFormat: targetFormat, - hwFormat: hwFormat, - utteranceFlag: isUtteranceActive, - levelStore: levelStore, - prerollStore: prerollStore, - streamRelay: streamRelay + targetFormat: resolvedTargetFormat, + hwFormat: hardwareFormat, + gate: gateLock, + levelStore: levels, + prerollStore: preroll, + streamRelay: relay, + drainTracker: tracker, + tailSampleCounter: tailCounter, + drainPolicy: policy ) - inputNode.installTap(onBus: 0, bufferSize: 4096, format: hwFormat, block: tap) + inputNode.installTap(onBus: 0, bufferSize: 4096, format: hardwareFormat, block: tap) didInstallTap = true audioEngine.prepare() @@ -245,7 +274,9 @@ public final class FlowContinuousCapture { /// Tear down the engine and release the audio session. public func stop() { removeSessionObservers() - isUtteranceActive.withLock { $0 = false } + gate.withLock { $0 = .idle } + drainTracker.reset() + tailSampleCounter.withLock { $0 = 0 } streamRelay.finish() if didInstallTap { @@ -256,6 +287,9 @@ public final class FlowContinuousCapture { audioEngine.stop() } isRunning = false + audioConverter = nil + targetFormat = nil + hwFormat = nil try? AVAudioSession.sharedInstance().setActive( false, options: .notifyOthersOnDeactivation @@ -288,8 +322,6 @@ public final class FlowContinuousCapture { object: nil, queue: .main ) { [weak self] note in - // Extract Sendable primitives here (Notification isn't Sendable) - // before hopping onto the main actor. let reasonRaw = note.userInfo?[AVAudioSessionRouteChangeReasonKey] as? UInt MainActor.assumeIsolated { self?.handleRouteChange(reasonRaw: reasonRaw) } } @@ -321,8 +353,6 @@ public final class FlowContinuousCapture { guard isRunning else { return } guard let reasonRaw, let reason = AVAudioSession.RouteChangeReason(rawValue: reasonRaw) else { return } - // Only rebuild for real device swaps (plugging / unplugging a headset - // or AirPods). Ignore `.categoryChange`, which we trigger ourselves. switch reason { case .oldDeviceUnavailable, .newDeviceAvailable: log.info("Audio route changed (\(reasonRaw, privacy: .public)) — rebuilding engine") @@ -337,7 +367,6 @@ public final class FlowContinuousCapture { let type = AVAudioSession.InterruptionType(rawValue: typeRaw) else { return } switch type { case .began: - // The system already paused our engine; wait for `.ended`. log.info("Audio interruption began") case .ended: guard isRunning else { return } @@ -373,38 +402,129 @@ public final class FlowContinuousCapture { /// Begin forwarding downsampled buffers to ASR for one utterance. public func beginUtterance() -> AsyncStream { let (stream, continuation) = AsyncStream.makeStream() + drainTracker.reset() + tailSampleCounter.withLock { $0 = 0 } // Bind the consumer before opening the gate so early tap frames // are not dropped on the floor. streamRelay.bind(continuation) streamRelay.replay(prerollStore.drain()) - isUtteranceActive.withLock { $0 = true } + gate.withLock { $0 = .recording } return stream } - /// Stop forwarding buffers; finishes the ASR stream. - public func endUtterance() { - isUtteranceActive.withLock { $0 = false } + /// Drain trailing PCM after the user stops, then finish the ASR stream. + public func endUtteranceAndDrain( + policy: FlowCaptureTailDrainPolicy = .flowDefault + ) async -> FlowCaptureDrainReport { + let currentPhase = gate.withLock { $0 } + guard currentPhase == .recording else { + return .skipped + } + + drainPolicy = policy + 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 flushSamples = flushConverterTailToStream() + tailSampleCounter.withLock { $0 += flushSamples } + streamRelay.finish() + gate.withLock { $0 = .idle } + + let tailSamples = tailSampleCounter.withLock { $0 } + let report = FlowCaptureDrainReport( + drainDurationSeconds: drainTracker.elapsedSeconds(), + endedBySilence: endedBySilence, + tailSampleCount: tailSamples + ) + drainTracker.reset() + tailSampleCounter.withLock { $0 = 0 } + FlowPipelineDiagnostics.logDrain(report) + return report } + /// Immediate stop without tail drain (abort / session teardown). public func cancelUtterance() { - endUtterance() + gate.withLock { $0 = .idle } + drainTracker.reset() + tailSampleCounter.withLock { $0 = 0 } + streamRelay.finish() } public func currentAudioLevels() -> [Float] { levelStore.snapshot() } + // MARK: - Converter flush + + @discardableResult + private func flushConverterTailToStream() -> Int { + guard let converter = audioConverter, + let targetFormat, + let hwFormat else { + return 0 + } + + var flushedSamples = 0 + let capacity = AVAudioFrameCount(max(512, hwFormat.sampleRate / 20)) + guard let outBuffer = AVAudioPCMBuffer(pcmFormat: targetFormat, frameCapacity: capacity) else { + return 0 + } + + var endOfStreamSignaled = false + while true { + outBuffer.frameLength = 0 + var error: NSError? + let status = converter.convert(to: outBuffer, error: &error) { _, outStatus in + if endOfStreamSignaled { + outStatus.pointee = .noDataNow + return nil + } + endOfStreamSignaled = true + outStatus.pointee = .endOfStream + return nil + } + + if status == .error || error != nil { + break + } + guard status == .haveData, outBuffer.frameLength > 0 else { + break + } + + let snapshot = AudioBufferSnapshot(buffer: outBuffer) + guard !snapshot.samples.isEmpty else { break } + streamRelay.yield(snapshot) + flushedSamples += snapshot.samples.count + } + + return flushedSamples + } + // MARK: - Audio tap (nonisolated — runs on realtime thread) private nonisolated static func makeAudioTapBlock( converter: AVAudioConverter, targetFormat: AVAudioFormat, hwFormat: AVAudioFormat, - utteranceFlag: OSAllocatedUnfairLock, + gate: OSAllocatedUnfairLock, levelStore: FlowLevelStore, prerollStore: FlowPrerollStore, - streamRelay: FlowCaptureStreamRelay + streamRelay: FlowCaptureStreamRelay, + drainTracker: FlowCaptureDrainTracker, + tailSampleCounter: OSAllocatedUnfairLock, + drainPolicy: FlowCaptureTailDrainPolicy ) -> @Sendable (AVAudioPCMBuffer, AVAudioTime) -> Void { return { buffer, _ in levelStore.update(from: buffer, barCount: FlowCaptureConstants.levelBarCount) @@ -426,9 +546,15 @@ public final class FlowContinuousCapture { let snapshot = AudioBufferSnapshot(buffer: outBuffer) guard !snapshot.samples.isEmpty else { return } - if utteranceFlag.withLock({ $0 }) { + let phase = gate.withLock { $0 } + switch phase { + case .recording, .draining: streamRelay.yield(snapshot) - } else { + if phase == .draining { + drainTracker.noteAudio(samples: snapshot.samples, policy: drainPolicy) + tailSampleCounter.withLock { $0 += snapshot.samples.count } + } + case .idle: prerollStore.append(snapshot) } } diff --git a/OSGKeyboardShared/Services/LiveDictationController.swift b/OSGKeyboardShared/Services/LiveDictationController.swift index c07f360..2dd9c4c 100644 --- a/OSGKeyboardShared/Services/LiveDictationController.swift +++ b/OSGKeyboardShared/Services/LiveDictationController.swift @@ -41,6 +41,12 @@ import AVFoundation import Speech import os +private enum LiveCaptureGatePhase: Equatable { + case idle + case recording + case draining +} + /// Thread-safe relay so the AVAudioEngine tap can yield snapshots without /// hopping through `@MainActor` (which adds latency and can reorder frames). private final class CaptureStreamRelay: @unchecked Sendable { @@ -100,6 +106,11 @@ public final class LiveDictationController: ObservableObject { public var asrTask: Task? private let streamRelay = CaptureStreamRelay() private var chunkedPipeline: ChunkedUtterancePipeline? + private let captureGate = OSAllocatedUnfairLock(initialState: LiveCaptureGatePhase.idle) + private let drainTracker = FlowCaptureDrainTracker() + private var audioConverter: AVAudioConverter? + private var targetFormat: AVAudioFormat? + private var hwFormat: AVAudioFormat? private var didConfigureAudioSession = false private var didInstallTap = false @@ -218,10 +229,6 @@ public final class LiveDictationController: ObservableObject { // If a previous `asrTask` is somehow still running (e.g. the // user smashed the disc twice quickly), `start()` cancels it // at the entry point as a safety net. - teardownCapturePipeline() - // Fallback: if we already have a meaningful partial but the - // backend never emits `.final`, promote the partial so the - // preview still inserts text after "停止录音". let partial = currentPartial.trimmingCharacters(in: .whitespacesAndNewlines) if !partial.isEmpty && lastFinal.isEmpty { lastFinal = partial @@ -230,9 +237,10 @@ public final class LiveDictationController: ObservableObject { if phase == .recording { phase = .processing } - // Deactivate so the user's music resumes if the preview is - // dismissed mid-recording. - try? AVAudioSession.sharedInstance().setActive(false, options: .notifyOthersOnDeactivation) + + Task { @MainActor [weak self] in + await self?.drainTailAndTeardownCapture() + } // Safety net: if the ASR pipeline never produces a `.final` // (analyzer hang, system glitch, dropped continuation), force @@ -311,6 +319,12 @@ public final class LiveDictationController: ObservableObject { return } + audioConverter = converter + self.targetFormat = targetFormat + self.hwFormat = hwFormat + drainTracker.reset() + captureGate.withLock { $0 = .recording } + let (stream, continuation) = AsyncStream.makeStream() streamRelay.bind(continuation) @@ -333,8 +347,16 @@ public final class LiveDictationController: ObservableObject { } } let relay = streamRelay + let gate = captureGate + let tracker = drainTracker + let policy = FlowCaptureTailDrainPolicy.flowDefault let onSnapshot: @Sendable (AudioBufferSnapshot) -> Void = { snapshot in + let phase = gate.withLock { $0 } + guard phase == .recording || phase == .draining else { return } relay.yield(snapshot) + if phase == .draining { + tracker.noteAudio(samples: snapshot.samples, policy: policy) + } } let tap = Self.makeAudioTapBlock( converter: converter, @@ -521,7 +543,88 @@ public final class LiveDictationController: ObservableObject { } } - private func teardownCapturePipeline() { + private func drainTailAndTeardownCapture() async { + let beganDrain = captureGate.withLock { phase -> Bool in + switch phase { + case .recording: + phase = .draining + return true + case .draining, .idle: + return false + } + } + guard beganDrain else { return } + + 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) + } + + _ = flushConverterTailToStream() + streamRelay.finish() + teardownCaptureEngine() + captureGate.withLock { $0 = .idle } + drainTracker.reset() + audioConverter = nil + targetFormat = nil + hwFormat = nil + + try? AVAudioSession.sharedInstance().setActive( + false, + options: .notifyOthersOnDeactivation + ) + } + + @discardableResult + private func flushConverterTailToStream() -> Int { + guard let converter = audioConverter, + let targetFormat, + let hwFormat else { + return 0 + } + + var flushedSamples = 0 + let capacity = AVAudioFrameCount(max(512, hwFormat.sampleRate / 20)) + guard let outBuffer = AVAudioPCMBuffer(pcmFormat: targetFormat, frameCapacity: capacity) else { + return 0 + } + + var endOfStreamSignaled = false + while true { + outBuffer.frameLength = 0 + var error: NSError? + let status = converter.convert(to: outBuffer, error: &error) { _, outStatus in + if endOfStreamSignaled { + outStatus.pointee = .noDataNow + return nil + } + endOfStreamSignaled = true + outStatus.pointee = .endOfStream + return nil + } + + if status == .error || error != nil { + break + } + guard status == .haveData, outBuffer.frameLength > 0 else { + break + } + + let snapshot = AudioBufferSnapshot(buffer: outBuffer) + guard !snapshot.samples.isEmpty else { break } + streamRelay.yield(snapshot) + flushedSamples += snapshot.samples.count + } + + return flushedSamples + } + + private func teardownCaptureEngine() { if didInstallTap { audioEngine.inputNode.removeTap(onBus: 0) didInstallTap = false @@ -529,6 +632,10 @@ public final class LiveDictationController: ObservableObject { if audioEngine.isRunning { audioEngine.stop() } + } + + private func teardownCapturePipeline() { + teardownCaptureEngine() streamRelay.finish() } diff --git a/OSGKeyboardShared/Utilities/FlowCaptureTailDrain.swift b/OSGKeyboardShared/Utilities/FlowCaptureTailDrain.swift new file mode 100644 index 0000000..a76485d --- /dev/null +++ b/OSGKeyboardShared/Utilities/FlowCaptureTailDrain.swift @@ -0,0 +1,130 @@ +// FlowCaptureTailDrain.swift +// OSGKeyboard · Shared +// +// Tail-drain policy and silence tracking for utterance end. After the user +// stops recording, capture keeps forwarding PCM until trailing speech drains +// or a safety timeout elapses (symmetric to pre-roll at utterance start). + +import Foundation + +/// Tunable tail-drain policy shared by Flow capture and preview dictation. +public struct FlowCaptureTailDrainPolicy: Sendable, Equatable { + /// RMS below this counts as silence while draining (16 kHz mono Float32). + public let silenceRMSThreshold: Float + /// Finish drain after this much continuous silence. + public let silenceDurationSeconds: TimeInterval + /// Hard cap so noisy environments cannot stall finalize forever. + public let maxDrainSeconds: TimeInterval + + public init( + silenceRMSThreshold: Float, + silenceDurationSeconds: TimeInterval, + maxDrainSeconds: TimeInterval + ) { + self.silenceRMSThreshold = silenceRMSThreshold + self.silenceDurationSeconds = silenceDurationSeconds + self.maxDrainSeconds = maxDrainSeconds + } + + public static let flowDefault = FlowCaptureTailDrainPolicy( + silenceRMSThreshold: 0.015, + silenceDurationSeconds: 0.25, + maxDrainSeconds: 1.5 + ) +} + +/// Metrics emitted when tail drain completes (for diagnostics and tests). +public struct FlowCaptureDrainReport: Sendable, Equatable { + public let drainDurationSeconds: Double + public let endedBySilence: Bool + public let tailSampleCount: Int + + public init( + drainDurationSeconds: Double, + endedBySilence: Bool, + tailSampleCount: Int + ) { + self.drainDurationSeconds = drainDurationSeconds + self.endedBySilence = endedBySilence + self.tailSampleCount = tailSampleCount + } + + public static let skipped = FlowCaptureDrainReport( + drainDurationSeconds: 0, + endedBySilence: false, + tailSampleCount: 0 + ) +} + +/// Thread-safe silence tracker used while draining trailing audio. +public final class FlowCaptureDrainTracker: @unchecked Sendable { + private let lock = OSAllocatedUnfairLock() + private var drainStartedAt: TimeInterval? + private var lastAudibleAt: TimeInterval? + + public init() {} + + public func reset() { + lock.withLock { + drainStartedAt = nil + lastAudibleAt = nil + } + } + + public func beginDrain(now: TimeInterval = Date().timeIntervalSince1970) { + lock.withLock { + drainStartedAt = now + lastAudibleAt = now + } + } + + public func noteAudio( + samples: [Float], + policy: FlowCaptureTailDrainPolicy, + now: TimeInterval = Date().timeIntervalSince1970 + ) { + guard !samples.isEmpty else { return } + let rms = Self.rms(of: samples) + lock.withLock { + guard drainStartedAt != nil else { return } + if rms >= policy.silenceRMSThreshold { + lastAudibleAt = now + } + } + } + + public func shouldFinish( + policy: FlowCaptureTailDrainPolicy, + now: TimeInterval = Date().timeIntervalSince1970 + ) -> (finished: Bool, endedBySilence: Bool) { + lock.withLock { + guard let started = drainStartedAt else { + return (true, false) + } + let audible = lastAudibleAt ?? started + if now - started >= policy.maxDrainSeconds { + return (true, false) + } + if now - audible >= policy.silenceDurationSeconds { + return (true, true) + } + return (false, false) + } + } + + public func elapsedSeconds(now: TimeInterval = Date().timeIntervalSince1970) -> Double { + lock.withLock { + guard let started = drainStartedAt else { return 0 } + return max(0, now - started) + } + } + + public static func rms(of samples: [Float]) -> Float { + guard !samples.isEmpty else { return 0 } + var sum: Float = 0 + for sample in samples { + sum += sample * sample + } + return sqrtf(sum / Float(samples.count)) + } +} diff --git a/OSGKeyboardShared/Utilities/FlowPipelineDiagnostics.swift b/OSGKeyboardShared/Utilities/FlowPipelineDiagnostics.swift new file mode 100644 index 0000000..5c347e9 --- /dev/null +++ b/OSGKeyboardShared/Utilities/FlowPipelineDiagnostics.swift @@ -0,0 +1,34 @@ +// FlowPipelineDiagnostics.swift +// OSGKeyboard · Shared +// +// Structured Flow pipeline metrics for Console.app filtering. + +import Foundation +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)" + ) + } + + public static func logChunkFinalize( + chunkCount: Int, + lastChunkSamples: Int, + stitchedLength: Int, + chunkWarnings: Int + ) { + OSGLog.flow.info( + "chunkPipeline chunks=\(chunkCount) lastChunkSamples=\(lastChunkSamples) " + + "stitchedLen=\(stitchedLength) warnings=\(chunkWarnings)" + ) + } + + public static func logStitcherSafeFallback(naiveLength: Int, mergedLength: Int) { + OSGLog.asr.warning( + "stitcher safe fallback naive=\(naiveLength) merged=\(mergedLength)" + ) + } +} diff --git a/OSGKeyboardShared/Utilities/UtteranceTranscriptStitcher.swift b/OSGKeyboardShared/Utilities/UtteranceTranscriptStitcher.swift index e33b1c6..4be4f5b 100644 --- a/OSGKeyboardShared/Utilities/UtteranceTranscriptStitcher.swift +++ b/OSGKeyboardShared/Utilities/UtteranceTranscriptStitcher.swift @@ -21,6 +21,12 @@ public struct UtteranceTranscriptStitcher: Sendable { } } + /// Drop the highest-index segment (used when re-transcribing a merged tail chunk). + public mutating func removeLastSegment() { + guard !segments.isEmpty else { return } + segments.removeLast() + } + public func composed() -> String { guard let first = segments.first else { return "" } var result = first.text @@ -30,6 +36,21 @@ public struct UtteranceTranscriptStitcher: Sendable { return result } + /// Prefer overlap-aware merge, but fall back to naive join when dedup would drop real content. + public func composedSafely() -> String { + let merged = composed() + guard segments.count >= 2 else { return merged } + let naive = segments.map(\.text).joined(separator: " ") + if merged.count + 16 < naive.count { + FlowPipelineDiagnostics.logStitcherSafeFallback( + naiveLength: naive.count, + mergedLength: merged.count + ) + return naive + } + return merged + } + /// Merge `next` onto `previous`, dropping duplicated suffix/prefix overlap. public static func mergeWithOverlap(previous: String, next: String) -> String { let trimmedNext = next.trimmingCharacters(in: .whitespacesAndNewlines) diff --git a/OSGKeyboardTests/ChunkedUtterancePipelineTests.swift b/OSGKeyboardTests/ChunkedUtterancePipelineTests.swift index efd289d..22263b1 100644 --- a/OSGKeyboardTests/ChunkedUtterancePipelineTests.swift +++ b/OSGKeyboardTests/ChunkedUtterancePipelineTests.swift @@ -87,6 +87,34 @@ final class ChunkedUtterancePipelineTests: XCTestCase { XCTAssertFalse(success.text.isEmpty) XCTAssertEqual(success.chunkWarnings.count, 1) } + + func testPipelineRetranscribesShortFinalChunkWithPriorOverlap() async { + let config = FlowUtteranceChunkConfig( + maxChunkDurationSeconds: 0.05, + overlapDurationSeconds: 10, + pauseExtensionMaxSeconds: 0, + pauseRMSThreshold: 0.02, + minFinalChunkDurationSeconds: 0.05, + sampleRate: 1_000 + ) + let asr = ShortFinalMergeStubASR() + let pipeline = ChunkedUtterancePipeline( + asr: asr, + locale: Locale(identifier: "zh-Hans"), + config: config + ) + + let (stream, continuation) = AsyncStream.makeStream() + continuation.yield(AudioBufferSnapshot(samples: [Float](repeating: 0.1, count: 80), sampleRate: 1_000)) + continuation.yield(AudioBufferSnapshot(samples: [Float](repeating: 0.1, count: 20), 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("merged")) + } } private struct FailingSecondChunkASR: ASRService, @unchecked Sendable { @@ -114,3 +142,32 @@ private struct FailingSecondChunkASR: ASRService, @unchecked Sendable { return .success("seg\(samples.count)") } } + +private struct ShortFinalMergeStubASR: ASRService, @unchecked Sendable { + private let callIndex = OSAllocatedUnfairLock(initialState: 0) + + func transcribe( + stream: AsyncStream, + locale: Locale + ) -> AsyncStream { + 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 samples.count > 20 { + return .success("merged-tail") + } + return .success("short") + } +} diff --git a/OSGKeyboardTests/FlowCaptureTailDrainTests.swift b/OSGKeyboardTests/FlowCaptureTailDrainTests.swift new file mode 100644 index 0000000..6652ed1 --- /dev/null +++ b/OSGKeyboardTests/FlowCaptureTailDrainTests.swift @@ -0,0 +1,51 @@ +// FlowCaptureTailDrainTests.swift +// OSGKeyboardTests + +import XCTest +@testable import OSGKeyboardShared + +final class FlowCaptureTailDrainTests: XCTestCase { + + private let policy = FlowCaptureTailDrainPolicy( + silenceRMSThreshold: 0.02, + silenceDurationSeconds: 0.2, + maxDrainSeconds: 1.0 + ) + + func testDrainFinishesAfterContinuousSilence() { + let tracker = FlowCaptureDrainTracker() + let start = Date().timeIntervalSince1970 + tracker.beginDrain(now: start) + + tracker.noteAudio(samples: [0.001, 0.001], policy: policy, now: start + 0.05) + + let silentAt = start + 0.1 + let decision = tracker.shouldFinish(policy: policy, now: silentAt + policy.silenceDurationSeconds) + XCTAssertTrue(decision.finished) + XCTAssertTrue(decision.endedBySilence) + } + + func testDrainFinishesAtMaxDurationEvenWithoutSilence() { + let tracker = FlowCaptureDrainTracker() + let start = Date().timeIntervalSince1970 + tracker.beginDrain(now: start) + + tracker.noteAudio(samples: [0.5, 0.4], policy: policy, now: start + 0.05) + tracker.noteAudio(samples: [0.45, 0.42], policy: policy, now: start + 0.4) + + let decision = tracker.shouldFinish(policy: policy, now: start + policy.maxDrainSeconds) + XCTAssertTrue(decision.finished) + XCTAssertFalse(decision.endedBySilence) + } + + func testRMSDetectsAudibleSamples() { + XCTAssertGreaterThan( + FlowCaptureDrainTracker.rms(of: [0.2, 0.18, 0.15]), + policy.silenceRMSThreshold + ) + XCTAssertLessThan( + FlowCaptureDrainTracker.rms(of: [0.001, 0.0005]), + policy.silenceRMSThreshold + ) + } +} diff --git a/OSGKeyboardTests/UtteranceStreamChunkerTests.swift b/OSGKeyboardTests/UtteranceStreamChunkerTests.swift index 30bae15..8a93bab 100644 --- a/OSGKeyboardTests/UtteranceStreamChunkerTests.swift +++ b/OSGKeyboardTests/UtteranceStreamChunkerTests.swift @@ -63,4 +63,33 @@ final class UtteranceStreamChunkerTests: XCTestCase { XCTAssertGreaterThanOrEqual(received.count, 2) XCTAssertTrue(received.last?.isLast == true) } + + func testFinalChunkIncludesLateArrivingTailSamples() async { + let config = FlowUtteranceChunkConfig( + firstChunkDurationSeconds: 0.5, + subsequentChunkDurationSeconds: 1.0, + overlapDurationSeconds: 0, + pauseExtensionMaxSeconds: 0, + pauseRMSThreshold: 0.02, + sampleRate: 1_000 + ) + let head = [Float](repeating: 0.05, count: 600) + let tail = [Float](repeating: 0.08, count: 250) + + let (stream, continuation) = AsyncStream.makeStream() + continuation.yield(AudioBufferSnapshot(samples: head, sampleRate: Double(config.sampleRate))) + continuation.yield(AudioBufferSnapshot(samples: tail, sampleRate: Double(config.sampleRate))) + continuation.finish() + + var received: [UtteranceAudioChunk] = [] + for await chunk in UtteranceStreamChunker.chunks(from: stream, config: config) { + received.append(chunk) + } + + guard let last = received.last else { + return XCTFail("expected at least one chunk") + } + XCTAssertTrue(last.isLast) + XCTAssertGreaterThanOrEqual(last.samples.count, tail.count) + } } diff --git a/OSGKeyboardTests/UtteranceTranscriptStitcherTests.swift b/OSGKeyboardTests/UtteranceTranscriptStitcherTests.swift index 85ab076..e09159f 100644 --- a/OSGKeyboardTests/UtteranceTranscriptStitcherTests.swift +++ b/OSGKeyboardTests/UtteranceTranscriptStitcherTests.swift @@ -20,4 +20,24 @@ final class UtteranceTranscriptStitcherTests: XCTestCase { stitcher.append(index: 0, text: "第一段") XCTAssertEqual(stitcher.composed(), "第一段 第二段") } + + func testComposedSafelyFallsBackWhenOverlapMergeShortensTooMuch() { + var stitcher = UtteranceTranscriptStitcher() + stitcher.append(index: 0, text: "今天天气很好我们") + stitcher.append(index: 1, text: "去公园") + let merged = stitcher.composed() + let safe = stitcher.composedSafely() + XCTAssertFalse(merged.isEmpty) + XCTAssertFalse(safe.isEmpty) + XCTAssertTrue(safe.contains("去公园")) + } + + func testRemoveLastSegmentSupportsMergedTailRetranscription() { + var stitcher = UtteranceTranscriptStitcher() + stitcher.append(index: 0, text: "第一段") + stitcher.append(index: 1, text: "第二段") + stitcher.removeLastSegment() + stitcher.append(index: 1, text: "第二段合并") + XCTAssertEqual(stitcher.composed(), "第一段 第二段合并") + } }