From 202c1716156dc5dd5af440f5ab31d9f3f5c355f5 Mon Sep 17 00:00:00 2001 From: Maksym Naboka Date: Tue, 28 Jul 2026 20:02:11 -0700 Subject: [PATCH 1/2] fix(streaming): stop fallback no longer clobbers a slow engine finalize MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2.0s stuck-session fallback in stopAppleSpeech was a one-shot deadline. Parakeet's runStop drains the entire queued audio feed through the decoder before session.finish() — on a long dictation that exceeds 2s, so the timer fired first with the STALE streamingText, set appleDidCompleteFinal, and the engine's genuine final (holding the tail words) was dropped by the completion guard. Net effect: the last words of a long dictation silently lost even though the engine decoded them. WhisperKit's finalizeTail re-decode has the same window. Fix: the fallback is now a re-arming poll (StreamingRoutePolicy.runStopFallback, pure + tested) gated on a new StreamingTranscriptionEngine.isFinalizing signal: - While the engine reports finalizing, the fallback re-arms instead of firing. - One grace poll after finalizing ends covers the final's main-actor Task hop into handleAppleSpeechFinal (completing in that gap would still clobber it). - A 30s hard cap fires the stale-partial fallback anyway if the finalize hangs, preserving the stuck-session guarantee the timer exists for. - Engines without the signal (Apple Speech, SpeechAnalyzer, stubs, fakes) keep the default false — first-poll fallback, identical to the old 2s behavior. Parakeet and WhisperKit report finalizing via a pending-non-cancel-stop counter (not a bool: on a fast stop→start→stop, session N's completion must not clear session N+1's already-counted finalize), incremented at stop ENQUEUE (runStop can sit behind a slow predecessor in the lifecycle chain) and balanced on every runStop exit path. Tests: StreamingStopFallbackTests drives the loop with a scripted slow engine — the drain/no-clobber case fails against the old one-shot semantics; grace, hard-cap, fence, and default-signal cases pinned alongside. Co-Authored-By: Claude --- OpenWhisp/Models/AppState.swift | 26 ++-- .../Services/ParakeetStreamingEngine.swift | 25 +++ OpenWhisp/Services/StreamingRoutePolicy.swift | 71 +++++++++ OpenWhisp/Services/TranscriptionEngine.swift | 17 +++ .../Services/WhisperKitStreamingEngine.swift | 21 +++ .../StreamingStopFallbackTests.swift | 143 ++++++++++++++++++ 6 files changed, 294 insertions(+), 9 deletions(-) create mode 100644 Tests/OpenWhispCoreTests/StreamingStopFallbackTests.swift diff --git a/OpenWhisp/Models/AppState.swift b/OpenWhisp/Models/AppState.swift index fb48919..eff1a4b 100644 --- a/OpenWhisp/Models/AppState.swift +++ b/OpenWhisp/Models/AppState.swift @@ -3474,17 +3474,25 @@ class AppState: ObservableObject { // Session-scoped fallback: a rapid follow-up session could otherwise be // finalized early by THIS session's leftover timer. // - // 2.0s (not 0.9s): AppleSpeechEngine's own 0.8s fallback guarantees a - // final for the Apple path, so this is a stuck-session guard (WhisperKit - // streaming teardown, engine deallocated) — with a wide margin so a busy - // main thread can't let this fire first and clobber the engine's genuine - // final (which arrives via an extra Task hop) with a stale partial. + // Re-arming poll, not a one-shot deadline: while the engine reports + // `isFinalizing` (Parakeet draining its decode backlog, WhisperKit's + // finalizeTail), the fallback keeps waiting for the genuine final — a + // one-shot fired first on long dictations and its stale partial made the + // completion guard drop the real final's tail words. Interval/grace/ + // hard-cap semantics live (tested) in StreamingRoutePolicy. let sessionID = activeSessionID + let engine = activeStreamingEngine Task { @MainActor in - try? await Task.sleep(nanoseconds: 2_000_000_000) - if sessionID == self.activeSessionID && self.isAppleSpeechSession && self.isTranscribing && !self.appleDidCompleteFinal { - self.handleAppleSpeechFinal(self.streamingText, sessionID: sessionID) - } + await StreamingRoutePolicy.runStopFallback( + isSessionStillWaiting: { + sessionID == self.activeSessionID && self.isAppleSpeechSession + && self.isTranscribing && !self.appleDidCompleteFinal + }, + isEngineFinalizing: { engine.isFinalizing }, + completeFallback: { + self.handleAppleSpeechFinal(self.streamingText, sessionID: sessionID) + } + ) } } diff --git a/OpenWhisp/Services/ParakeetStreamingEngine.swift b/OpenWhisp/Services/ParakeetStreamingEngine.swift index af15242..ccf03fe 100644 --- a/OpenWhisp/Services/ParakeetStreamingEngine.swift +++ b/OpenWhisp/Services/ParakeetStreamingEngine.swift @@ -81,6 +81,19 @@ final class ParakeetStreamingEngine: NSObject, StreamingTranscriptionEngine { @MainActor private var lastPartial = "" @MainActor private var didStop = false + /// Non-cancel stops enqueued whose `runStop` hasn't completed yet — the + /// engine is still working toward a genuine final (feed drain + `finish()` + /// flush). A counter, not a bool: on a fast stop→start→stop, session N+1's + /// stop is already counted while session N's runStop is still queued, and + /// N's completion must not clear N+1's pending finalize. + @MainActor private var pendingFinalizeStops = 0 + + /// True while a non-cancel stop is still draining/flushing toward its final. + /// Read by AppState's stuck-session fallback; main-actor callers only (same + /// contract as `selectDevice`). + var isFinalizing: Bool { + MainActor.assumeIsolated { pendingFinalizeStops > 0 } + } /// Session generation, bumped on every start and stop — late partial /// callbacks from a torn-down session fail the gate instead of leaking into /// the next one (same rationale as the other streaming engines). @@ -415,6 +428,13 @@ final class ParakeetStreamingEngine: NSObject, StreamingTranscriptionEngine { // session's final with the successor's sessionID, straight through // AppState's session fence. let final = self.onFinal + // Report "finalizing" from ENQUEUE until runStop completes, so the + // stuck-session fallback (StreamingRoutePolicy.runStopFallback) keeps + // waiting for the genuine final instead of completing the session + // with the stale last partial while the feed drain / finish() flush + // runs — enqueue time matters because runStop can sit behind a slow + // predecessor in the chain. + if !cancel { self.pendingFinalizeStops += 1 } self.lifecycle.enqueue { await self.runStop(cancel: cancel, final: final) } @@ -434,6 +454,11 @@ final class ParakeetStreamingEngine: NSObject, StreamingTranscriptionEngine { @MainActor private func runStop(cancel: Bool, final deliverFinal: ((String) -> Void)?) async { + // Balance stop(cancel:false)'s increment on EVERY exit path (no-session + // early return, finish() error): the fallback poll must see finalizing + // end even when no final gets delivered — its grace poll then fires the + // stale-partial completion instead of wedging the session. + defer { if !cancel { pendingFinalizeStops = max(0, pendingFinalizeStops - 1) } } didStop = true // Supersede the session so late partial hops are dropped. generation += 1 diff --git a/OpenWhisp/Services/StreamingRoutePolicy.swift b/OpenWhisp/Services/StreamingRoutePolicy.swift index c085871..7105e4d 100644 --- a/OpenWhisp/Services/StreamingRoutePolicy.swift +++ b/OpenWhisp/Services/StreamingRoutePolicy.swift @@ -105,6 +105,77 @@ public enum StreamingRoutePolicy { return pendingStop ? .beginListeningThenStop : .beginListening } + /// Stuck-session stop-fallback poll interval (seconds). 2.0 (not 0.9): + /// AppleSpeechEngine's own 0.8s fallback guarantees a final for the Apple + /// path, so the fallback is a stuck-session guard — with a wide margin so a + /// busy main thread can't let it fire first and clobber the engine's genuine + /// final (which arrives via an extra Task hop) with a stale partial. + public static let stopFallbackInterval: TimeInterval = 2 + /// Hard cap (seconds) on how long the stop fallback keeps waiting for an + /// engine that reports `isFinalizing`. A finalize that outlasts this is + /// treated as hung and the session completes with the last partial — the + /// stuck-session guarantee the fallback exists for. Generous on purpose: a + /// real decode-backlog drain is far faster than realtime (Parakeet RTFx ~5), + /// so only a genuine hang ever reaches it. + public static let stopFallbackMaxWait: TimeInterval = 30 + + /// The stuck-session fallback armed by a streaming stop (AppState's + /// `stopAppleSpeech`): if the engine's genuine final never arrives, complete + /// the session with the last streamed partial rather than wedging at + /// "Finalizing...". + /// + /// Re-arming poll loop, NOT a one-shot deadline — the one-shot was the + /// tail-word-loss bug: a long dictation leaves Parakeet's `runStop` draining + /// the queued audio feed through the decoder (then `finish()`), which can + /// outlast any fixed window; the timer fired first with the STALE partial, + /// set the did-complete flag, and the genuine final (holding the tail words) + /// was dropped by the completion guard. While the engine reports + /// `isFinalizing`, the fallback re-arms instead of firing. + /// + /// One extra GRACE poll after finalizing ends: the engine clears its + /// finalizing state when its stop chain returns, but the final it delivered + /// still rides a main-actor Task hop into the completion handler — completing + /// in that gap would clobber it. By the next poll the hop has long landed and + /// `isSessionStillWaiting` reports the session done. + /// + /// - Parameters: + /// - sleep: injectable wait (tests script it; production sleeps for real). + /// - isSessionStillWaiting: the caller's session fence — still the same + /// session, still finalizing, no final handled yet. `false` ends the loop + /// silently (the timer's job is done or moot). + /// - isEngineFinalizing: the active engine's `isFinalizing`. + /// - completeFallback: complete the session with the stale partial. Called + /// at most once, as the loop's last act. + @MainActor + public static func runStopFallback( + interval: TimeInterval = stopFallbackInterval, + maxWait: TimeInterval = stopFallbackMaxWait, + sleep: (TimeInterval) async -> Void = { + try? await Task.sleep(nanoseconds: UInt64($0 * 1_000_000_000)) + }, + isSessionStillWaiting: () -> Bool, + isEngineFinalizing: () -> Bool, + completeFallback: () -> Void + ) async { + var elapsed: TimeInterval = 0 + var wasFinalizing = false + while true { + await sleep(interval) + elapsed += interval + guard isSessionStillWaiting() else { return } + let finalizing = isEngineFinalizing() + if (finalizing || wasFinalizing) && elapsed < maxWait { + // `wasFinalizing` is what grants the single grace poll: it holds + // fire for exactly one interval after finalizing ends, then (if + // the final still never landed) the fallback fires after all. + wasFinalizing = finalizing + continue + } + completeFallback() + return + } + } + /// Arming-timeout fallback (seconds): if `onStarted` hasn't fired this long /// after `start()` was issued, flip to Listening anyway so a signal-wiring /// bug can't wedge the session at "Starting…" forever. Deliberately generous diff --git a/OpenWhisp/Services/TranscriptionEngine.swift b/OpenWhisp/Services/TranscriptionEngine.swift index 6d2e97f..567ee76 100644 --- a/OpenWhisp/Services/TranscriptionEngine.swift +++ b/OpenWhisp/Services/TranscriptionEngine.swift @@ -128,6 +128,18 @@ public protocol StreamingTranscriptionEngine: AnyObject { /// non-default mic still recorded the built-in one. func selectDevice(_ deviceID: String) + /// True while a non-cancel `stop` is still actively producing the session's + /// genuine final — e.g. Parakeet draining its queued audio feed through the + /// decoder before `finish()`, or WhisperKit's `finalizeTail` re-decode. The + /// stuck-session fallback (`StreamingRoutePolicy.runStopFallback`) polls this + /// instead of firing at a fixed deadline: a long dictation's decode backlog + /// can outlast any fixed window, and completing the session with the stale + /// last partial would silently drop the tail words the engine is about to + /// deliver. Engines whose final is synchronous or internally guaranteed + /// (Apple Speech's 0.8s synthesized final, SpeechAnalyzer's stop-time + /// delivery) keep the default `false`. + var isFinalizing: Bool { get } + /// Begin streaming recognition for `language` ("auto" = current locale). /// /// `prompt` carries the session's vocabulary/bias context (the same @@ -149,6 +161,11 @@ public protocol StreamingTranscriptionEngine: AnyObject { } extension StreamingTranscriptionEngine { + /// Default: the engine's final needs no extended wait (it arrives promptly or + /// is internally guaranteed), so the stuck-session fallback keeps its original + /// single-deadline behavior. + public var isFinalizing: Bool { false } + /// Convenience for call sites (and the test fake) that carry no bias context. /// Not a way to smuggle a prompt past the capability gate — it forwards the /// empty string, which every engine honors trivially (it's a no-op). diff --git a/OpenWhisp/Services/WhisperKitStreamingEngine.swift b/OpenWhisp/Services/WhisperKitStreamingEngine.swift index 7c0f3e1..a5004bc 100644 --- a/OpenWhisp/Services/WhisperKitStreamingEngine.swift +++ b/OpenWhisp/Services/WhisperKitStreamingEngine.swift @@ -52,6 +52,19 @@ final class WhisperKitStreamingEngine: StreamingTranscriptionEngine { // unconfirmed/hypothesis tail. @MainActor private var lastConfirmedText: String = "" @MainActor private var didFinish: Bool = false + /// Non-cancel stops enqueued whose `runStop` hasn't completed yet — the + /// engine is still working toward a genuine final (the finalizeTail + /// re-decode). A counter, not a bool: on a fast stop→start→stop, session + /// N+1's stop is already counted while session N's runStop is still queued, + /// and N's completion must not clear N+1's pending finalize. Mirrors + /// ParakeetStreamingEngine. + @MainActor private var pendingFinalizeStops = 0 + + /// True while a non-cancel stop is still flushing toward its final. Read by + /// AppState's stuck-session fallback; main-actor callers only. + var isFinalizing: Bool { + MainActor.assumeIsolated { pendingFinalizeStops > 0 } + } /// Whether this stream's `onStarted` already fired. AudioStreamTranscriber /// installs the tap inside `handle.start()`, which doesn't return until the /// stream ENDS — so the first state diff (they arrive per buffer, energy @@ -198,6 +211,11 @@ final class WhisperKitStreamingEngine: StreamingTranscriptionEngine { // final with the successor's sessionID, straight through AppState's // session fence. let final = self.onFinal + // Report "finalizing" from ENQUEUE until runStop completes, so the + // stuck-session fallback (StreamingRoutePolicy.runStopFallback) keeps + // waiting for the genuine final instead of completing the session + // with the stale last partial while finalizeTail decodes the tail. + if !cancel { self.pendingFinalizeStops += 1 } // Strong capture (see start): teardown must run even if the caller drops // its reference to this engine immediately after enqueueing the stop. self.lifecycle.enqueue { @@ -211,6 +229,9 @@ final class WhisperKitStreamingEngine: StreamingTranscriptionEngine { /// before any queued start proceeds. @MainActor private func runStop(cancel: Bool, final deliverFinal: ((String) -> Void)?) async { + // Balance stop(cancel:false)'s increment on every exit path — the + // fallback poll must see finalizing end even when no final is delivered. + defer { if !cancel { pendingFinalizeStops = max(0, pendingFinalizeStops - 1) } } let handle = transcriber transcriber = nil didFinish = true diff --git a/Tests/OpenWhispCoreTests/StreamingStopFallbackTests.swift b/Tests/OpenWhispCoreTests/StreamingStopFallbackTests.swift new file mode 100644 index 0000000..25aa4fc --- /dev/null +++ b/Tests/OpenWhispCoreTests/StreamingStopFallbackTests.swift @@ -0,0 +1,143 @@ +import XCTest +@testable import OpenWhispCore + +/// The stuck-session stop fallback (`StreamingRoutePolicy.runStopFallback`) — +/// the guard timer AppState.stopAppleSpeech arms when a streaming session stops. +/// +/// The regression under test (parakeet-dropout follow-up): the fallback used to +/// be a one-shot 2s deadline. A long dictation leaves Parakeet's runStop +/// draining the queued audio feed through the decoder before `finish()`, which +/// can outlast that window — the timer fired first with the STALE streamingText, +/// set appleDidCompleteFinal, and the engine's genuine final (holding the tail +/// words) was dropped by the completion guard. The loop must instead RE-ARM +/// while the engine reports `isFinalizing`. +/// +/// Each test scripts the probes the way AppState wires them: `sleep` counts +/// polls instead of waiting, `isSessionStillWaiting` is the session fence +/// (flips false once the genuine final was handled or the session rotated), and +/// `isEngineFinalizing` is the engine's drain signal. +@MainActor +final class StreamingStopFallbackTests: XCTestCase { + + /// Baseline (pre-existing behavior preserved): an engine that never reports + /// finalizing — Apple Speech, or a torn-down/deallocated engine — falls back + /// on the FIRST poll, so genuinely stuck sessions still recover at the + /// original 2s latency. + func testStuckSessionCompletesOnFirstPoll() async { + var completions = 0 + var polls = 0 + await StreamingRoutePolicy.runStopFallback( + sleep: { _ in polls += 1 }, + isSessionStillWaiting: { true }, + isEngineFinalizing: { false }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 1) + XCTAssertEqual(polls, 1, "stuck-session recovery latency must stay one interval") + } + + /// THE bug: engine drains a decode backlog for several polls (a long + /// dictation), then delivers its genuine final — the fence flips to done. + /// The fallback must never have fired; firing would have completed the + /// session with the stale partial and the guard would drop the real final's + /// tail words. Fails with the old one-shot behavior (completion at poll 1). + func testRearmsThroughSlowEngineDrainAndNeverClobbersGenuineFinal() async { + var completions = 0 + var polls = 0 + await StreamingRoutePolicy.runStopFallback( + sleep: { _ in polls += 1 }, + // Polls 1–3: still finalizing (6s of drain, past the old 2s one-shot). + // Poll 4: the genuine final landed — session no longer waiting. + isSessionStillWaiting: { polls < 4 }, + isEngineFinalizing: { true }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 0, "the genuine final must win — no stale-partial completion") + XCTAssertEqual(polls, 4) + } + + /// The engine's stop chain finished (finalizing flips false) but its final + /// still rides a main-actor Task hop into the completion handler. The single + /// grace poll must hold fire across that gap; by the next poll the final has + /// landed and the fence ends the loop silently. + func testGracePollCoversFinalDeliveryHop() async { + var completions = 0 + var polls = 0 + await StreamingRoutePolicy.runStopFallback( + sleep: { _ in polls += 1 }, + // Poll 3: the hopped final was handled — session done. + isSessionStillWaiting: { polls < 3 }, + // Finalizing only on poll 1; cleared by poll 2 (runStop returned). + isEngineFinalizing: { polls == 1 }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 0, "completing inside the delivery-hop gap clobbers the final") + } + + /// The grace is ONE poll, not forever: if finalizing ends and no final ever + /// arrives (an exit path that delivers nothing), the fallback still fires — + /// the session must not wedge at "Finalizing...". + func testGraceIsSinglePollThenFallbackFires() async { + var completions = 0 + var polls = 0 + await StreamingRoutePolicy.runStopFallback( + sleep: { _ in polls += 1 }, + isSessionStillWaiting: { true }, + isEngineFinalizing: { polls == 1 }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 1) + XCTAssertEqual(polls, 3, "re-arm (finalizing), grace, fire") + } + + /// Hard cap: an engine hung inside its finalize (reports finalizing forever) + /// must not wedge the session — the fallback fires once maxWait elapses. + func testHardCapFiresEvenWhileEngineStillReportsFinalizing() async { + var completions = 0 + var polls = 0 + await StreamingRoutePolicy.runStopFallback( + interval: 2, maxWait: 30, + sleep: { _ in polls += 1 }, + isSessionStillWaiting: { true }, + isEngineFinalizing: { true }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 1) + XCTAssertEqual(polls, 15, "15 × 2s polls = the 30s cap") + } + + /// Session fence: once the session rotated (or the final was already + /// handled), the loop ends without touching anything — the engine probe must + /// not even be consulted (a rapid follow-up session could otherwise be + /// finalized early by this session's leftover timer). + func testCollapsedFenceEndsLoopWithoutFiring() async { + var completions = 0 + await StreamingRoutePolicy.runStopFallback( + sleep: { _ in }, + isSessionStillWaiting: { false }, + isEngineFinalizing: { + XCTFail("fence must short-circuit before the engine probe") + return true + }, + completeFallback: { completions += 1 } + ) + XCTAssertEqual(completions, 0) + } + + /// The protocol default: engines that don't implement the signal (Apple + /// Speech, SpeechAnalyzer, the lean-build stubs, test fakes) report + /// not-finalizing, which preserves the original one-deadline behavior. + func testEngineIsFinalizingDefaultsToFalse() { + final class BareEngine: StreamingTranscriptionEngine { + var onPartial: ((String) -> Void)? + var onFinal: ((String) -> Void)? + var onError: ((String) -> Void)? + var onLevelChanged: ((_ display: Float, _ vad: Float) -> Void)? + var onStarted: (() -> Void)? + func selectDevice(_ deviceID: String) {} + func start(language: String, prompt: String) throws {} + func stop(cancel: Bool) {} + } + XCTAssertFalse(BareEngine().isFinalizing) + } +} From bf8370694da1db29de486041aaa15b7d36b23b2c Mon Sep 17 00:00:00 2001 From: Maksym Naboka Date: Thu, 30 Jul 2026 12:37:24 -0700 Subject: [PATCH 2/2] Compact the stop-fallback call site to keep the AppState LOC ratchet green The runStopFallback arming replaced a 7-line one-shot with a 15-line block and tripped the MAK-32 ratchet (+8 over budget). Tighten the comment and closure formatting so the change is LOC-neutral; no behavior change. Co-Authored-By: Claude --- OpenWhisp/Models/AppState.swift | 26 +++++++++----------------- 1 file changed, 9 insertions(+), 17 deletions(-) diff --git a/OpenWhisp/Models/AppState.swift b/OpenWhisp/Models/AppState.swift index eff1a4b..7fe5283 100644 --- a/OpenWhisp/Models/AppState.swift +++ b/OpenWhisp/Models/AppState.swift @@ -3471,28 +3471,20 @@ class AppState: ObservableObject { // at the true end (insert / finishSessionUI). activeStreamingEngine.stop(cancel: false) - // Session-scoped fallback: a rapid follow-up session could otherwise be - // finalized early by THIS session's leftover timer. - // - // Re-arming poll, not a one-shot deadline: while the engine reports - // `isFinalizing` (Parakeet draining its decode backlog, WhisperKit's - // finalizeTail), the fallback keeps waiting for the genuine final — a - // one-shot fired first on long dictations and its stale partial made the - // completion guard drop the real final's tail words. Interval/grace/ - // hard-cap semantics live (tested) in StreamingRoutePolicy. + // Session-scoped fallback (a rapid follow-up session must not be + // finalized by THIS session's leftover timer). Re-arming poll, not a + // one-shot: while the engine reports `isFinalizing` (Parakeet draining + // its decode backlog, WhisperKit's finalizeTail) keep waiting for the + // genuine final — a one-shot fired first on long dictations and its stale + // partial dropped the real final's tail words; see StreamingRoutePolicy. let sessionID = activeSessionID let engine = activeStreamingEngine Task { @MainActor in await StreamingRoutePolicy.runStopFallback( - isSessionStillWaiting: { - sessionID == self.activeSessionID && self.isAppleSpeechSession - && self.isTranscribing && !self.appleDidCompleteFinal - }, + isSessionStillWaiting: { sessionID == self.activeSessionID && self.isAppleSpeechSession + && self.isTranscribing && !self.appleDidCompleteFinal }, isEngineFinalizing: { engine.isFinalizing }, - completeFallback: { - self.handleAppleSpeechFinal(self.streamingText, sessionID: sessionID) - } - ) + completeFallback: { self.handleAppleSpeechFinal(self.streamingText, sessionID: sessionID) }) } }