From 9071d7e1b8ebd36fd3b56a0913c9dddeaa1f9579 Mon Sep 17 00:00:00 2001 From: Julian Pscheid Date: Thu, 8 Oct 2026 08:38:51 -0700 Subject: [PATCH] fix(asr/nemotron-multilingual): skip unchanged partial callbacks --- ...onMultilingualAsrManager+BlankRescue.swift | 24 ++-- ...eamingNemotronMultilingualAsrManager.swift | 8 +- ...NemotronMultilingualPublicationTests.swift | 114 +++++++++++++++++- 3 files changed, 133 insertions(+), 13 deletions(-) diff --git a/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager+BlankRescue.swift b/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager+BlankRescue.swift index 90be7db11..c4637515a 100644 --- a/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager+BlankRescue.swift +++ b/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager+BlankRescue.swift @@ -88,18 +88,24 @@ extension StreamingNemotronMultilingualAsrManager { ) { guard partialPublicationSuppressionDepth == 0, !inBlankRescue else { return } guard let callback = callback, let tokenizer = tokenizer else { return } - guard !isFinal, Self.blankRescueEnabled, Self.rescueRmsThreshold > 0 else { + guard Self.blankRescueEnabled, Self.rescueRmsThreshold > 0 else { callback(tokenizer.decode(ids: accumulatedTokenIds).text) return } - let ids = Self.partialPublicationTokenIds( - liveIds: accumulatedTokenIds, liveTimings: accumulatedTokenTimings, - langTagTokenIds: config.langTagTokenIds, - openSpan: rescueSpanOpen - ? (rescueSpanStartFrame, rescueSpanLastSpeechFrame, rescueSpanPreRollFrames, rescueSpanOverflowed) - : nil, - nextSpanStartFrame: rescueFrameCursor - rescuePreRollTail.count / ASRConstants.samplesPerEncoderFrame) - callback(tokenizer.decode(ids: ids).text) + let ids = + isFinal + ? accumulatedTokenIds + : Self.partialPublicationTokenIds( + liveIds: accumulatedTokenIds, liveTimings: accumulatedTokenTimings, + langTagTokenIds: config.langTagTokenIds, + openSpan: rescueSpanOpen + ? (rescueSpanStartFrame, rescueSpanLastSpeechFrame, rescueSpanPreRollFrames, rescueSpanOverflowed) + : nil, + nextSpanStartFrame: rescueFrameCursor - rescuePreRollTail.count / ASRConstants.samplesPerEncoderFrame) + let text = tokenizer.decode(ids: ids).text + guard text != lastDeliveredPartialText else { return } + lastDeliveredPartialText = text + callback(text) } /// `processChunk` plus blank-span bookkeeping. All streaming call sites diff --git a/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager.swift b/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager.swift index 26d6676f5..3ebbbeec5 100644 --- a/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager.swift +++ b/Sources/FluidAudio/ASR/Parakeet/Streaming/Nemotron/StreamingNemotronMultilingualAsrManager.swift @@ -3,8 +3,8 @@ import AVFoundation import Foundation /// Callback delivering cumulative transcript text; language tags use `detectedLanguage()`. -/// With blank rescue enabled, text is settled and only extends the previous text; -/// chunks with nothing newly settled may repeat it unchanged. An unresolved blank +/// With blank rescue enabled, text is settled, only extends the previous text, +/// and is delivered only when it changes. An unresolved blank /// span can hold back later text for the 15 s span cap (15.04 s at window granularity) /// plus the rest of that chunk in audio time; processing time is additional. /// `getPartialTranscript()` may already include that text; `finish()` flushes the rest. @@ -269,6 +269,7 @@ public actor StreamingNemotronMultilingualAsrManager { // Callbacks internal var partialCallback: NemotronMultilingualPartialCallback? internal var partialPublicationSuppressionDepth: Int = 0 + internal var lastDeliveredPartialText: String? // Stats internal var processedChunks: Int = 0 @@ -813,6 +814,7 @@ public actor StreamingNemotronMultilingualAsrManager { /// Reset all states for a new transcription session. /// Preserves the currently selected prompt id and ml configuration. public func reset() async { + lastDeliveredPartialText = nil StreamingAsrUtils.resetSharedState( audioBuffer: &audioBuffer, accumulatedTokenIds: &accumulatedTokenIds, @@ -903,6 +905,7 @@ public actor StreamingNemotronMultilingualAsrManager { } internal func resetStates() throws { + lastDeliveredPartialText = nil let cacheConfig = EncoderCacheManager.CacheConfig( channelShape: config.cacheChannelShape, timeShape: config.cacheTimeShape, @@ -1057,6 +1060,7 @@ public actor StreamingNemotronMultilingualAsrManager { lastFinishTokenTimings = accumulatedTokenTimings accumulatedTokenIds.removeAll() accumulatedTokenTimings.removeAll() + lastDeliveredPartialText = nil // The emitted-token tail must follow the accumulated ids it mirrors. vocabularyBias?.resetMatchState() diff --git a/Tests/FluidAudioTests/ASR/Parakeet/Streaming/NemotronMultilingualPublicationTests.swift b/Tests/FluidAudioTests/ASR/Parakeet/Streaming/NemotronMultilingualPublicationTests.swift index cf08fa8cd..8cba951f8 100644 --- a/Tests/FluidAudioTests/ASR/Parakeet/Streaming/NemotronMultilingualPublicationTests.swift +++ b/Tests/FluidAudioTests/ASR/Parakeet/Streaming/NemotronMultilingualPublicationTests.swift @@ -13,6 +13,116 @@ final class NemotronMultilingualPublicationTests: XCTestCase { private let later = 5 private let boundary = 6 + func testRepeatedSettledPublicationDeliversOnce() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, period], timings: [timing(hello, 5), timing(period, 17)], + openSpanStart: 14) + + await manager.publishCurrentState() + await manager.publishCurrentState() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo"]) + } + + func testChangedSettledTextDeliversAnotherUpdate() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, world], timings: [timing(hello, 5), timing(world, 14)]) + await manager.publishCurrentState() + await manager.publishCurrentState() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo", "Hallo Welt"]) + } + + func testEqualDecodedTextAfterNewTokensPreservesRescueDisabledBehavior() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, langTag], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, langTag, boundary], timings: [timing(hello, 5), timing(boundary, 6)]) + await manager.publishCurrentState() + + let rescueEnabled = Manager.blankRescueEnabled && Manager.rescueRmsThreshold > 0 + XCTAssertEqual(updates.withLock { $0 }, rescueEnabled ? ["Hallo"] : ["Hallo", "Hallo", "Hallo"]) + } + + func testResetAllowsTheSameTextToDeliverAgain() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + + await manager.reset() + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + await manager.publishCurrentState() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo", "Hallo"]) + } + + func testResetStatesAllowsTheSameTextToDeliverAgain() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + + try await manager.resetStates() + await manager.setPublicationState(tokenizer: tokenizer, ids: [hello], timings: [timing(hello, 5)]) + await manager.publishCurrentState() + await manager.publishCurrentState() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo", "Hallo"]) + } + + func testFinalFlushDeliversChangedTextOnlyOnce() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, period], timings: [timing(hello, 5), timing(period, 17)], + openSpanStart: 14) + await manager.publishCurrentState() + + try await manager.finalizeRescueSpanIfNeeded() + try await manager.finalizeRescueSpanIfNeeded() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo", "Hallo."]) + } + + func testFinalFlushSkipsTextAlreadyDeliveredLive() async throws { + let tokenizer = try makeTokenizer() + let manager = Manager() + let updates = OSAllocatedUnfairLock<[String]>(initialState: []) + await manager.setPartialCallback { text in updates.withLock { $0.append(text) } } + await manager.setPublicationState( + tokenizer: tokenizer, ids: [hello, period], timings: [timing(hello, 5), timing(period, 17)]) + await manager.publishCurrentState() + + try await manager.finalizeRescueSpanIfNeeded() + + XCTAssertEqual(updates.withLock { $0 }, ["Hallo."]) + } + func testRescueInsertionBeforeAccumulatedPunctuationPublishesOnlyExtensions() throws { let tokenizer = try makeTokenizer() let ids = [langTag, hello, period, later] @@ -198,7 +308,7 @@ final class NemotronMultilingualPublicationTests: XCTestCase { await manager.publishCurrentState() await manager.publishCurrentState() XCTAssertEqual(originalUpdates.withLock { $0 }, []) - XCTAssertEqual(replacementUpdates.withLock { $0 }, ["Hallo.", "Hallo."]) + XCTAssertEqual(replacementUpdates.withLock { $0 }, ["Hallo."]) } func testNestedSuppressionKeepsTheCurrentCallbackUntilTheChunkSettles() async throws { @@ -228,7 +338,7 @@ final class NemotronMultilingualPublicationTests: XCTestCase { let depth = await manager.partialPublicationSuppressionDepth XCTAssertEqual(depth, 0) XCTAssertEqual(originalUpdates.withLock { $0 }, []) - XCTAssertEqual(replacementUpdates.withLock { $0 }, ["Hallo.", "Hallo."]) + XCTAssertEqual(replacementUpdates.withLock { $0 }, ["Hallo."]) } private func timing(_ id: Int, _ frame: Int) -> TokenTiming {