From 621023c6385650137d798bdaa480d0a50004a156 Mon Sep 17 00:00:00 2001 From: Fabian Meyer <44942030+dinooo13@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:45:03 +0200 Subject: [PATCH] Coordinator: the feed and warm loops are a RecordingEngineSession The utterance, the feed loop with its live passes, and the warm loop move into a session on the coordinator's clock. It hands back what it ended with: the source to transcribe and the feed to wait for. The coordinator keeps the recording number and the checks after every wait. Co-Authored-By: Claude Opus 5.5 --- .../PladderCore/DictationCoordinator.swift | 115 +++----------- .../PladderCore/RecordingEngineSession.swift | 134 ++++++++++++++++ .../RecordingEngineSessionTests.swift | 143 ++++++++++++++++++ docs/ARCHITECTURE.md | 12 +- 4 files changed, 306 insertions(+), 98 deletions(-) create mode 100644 Sources/PladderCore/RecordingEngineSession.swift create mode 100644 Tests/PladderCoreTests/RecordingEngineSessionTests.swift diff --git a/Sources/PladderCore/DictationCoordinator.swift b/Sources/PladderCore/DictationCoordinator.swift index 195cb29..9c6a10b 100644 --- a/Sources/PladderCore/DictationCoordinator.swift +++ b/Sources/PladderCore/DictationCoordinator.swift @@ -99,25 +99,12 @@ public final class DictationCoordinator { private var pendingUnloads: [any TranscriptionEngine] = [] private var cycleEngine: (any TranscriptionEngine)? - // Cancelling cannot abort a CoreML call already started, so a release landing in - // a warm pass waits for it; the log shows that as `engine` above `engine-time`. - private var warmupTask: Task? - private var stream: Stream? - - private struct Stream { - let engine: any StreamingTranscriptionEngine - let utterance: Utterance - let feed: Task - } + private var engineSession: RecordingEngineSession? private var recordingID = 0 private var cancelCleanup: Task? private var muteRestoreTask: Task? - // Every utterance is padded to the model's full window, so half a second of - // silence costs the same encoder pass as a real call. - private static let warmupSamples = [Float](repeating: 0, count: 8_000) - @ObservationIgnored private(set) var handledHotkeyEvents = 0 public struct CycleTiming: Sendable, Equatable { @@ -387,8 +374,16 @@ public final class DictationCoordinator { onEvent(.recordingStarted) armOutputMute() startKeyDownWork() - await startEngineWork(live: live, recordingID: mine) - guard isCurrent(mine) else { return } + if let engine = cycleEngine { + let session = RecordingEngineSession( + engine: engine, capture: capture, clock: clock, + intervals: .init(feed: feedInterval, livePass: livePassInterval, warmup: warmupInterval), + live: live, onPartial: { [weak self] in self?.partialTranscript = $0 }) + engineSession = session + await session.begin() + guard isCurrent(mine) else { return } + session.startLoops() + } levelTask = Task { [weak self] in for await level in levels { guard let self, self.state.isRecording else { return } @@ -416,62 +411,6 @@ public final class DictationCoordinator { state.isRecording && recordingID == id } - // A streaming engine that cannot begin an utterance is used like a batch engine: - // transcribed whole at release, slower but not lost. - private func startEngineWork(live: Bool, recordingID id: Int) async { - if let streaming = cycleEngine as? (any StreamingTranscriptionEngine), - let utterance = try? await streaming.beginUtterance() { - guard isCurrent(id) else { - await streaming.abandonUtterance(utterance) - return - } - startStreamingFeed(streaming, utterance: utterance, live: live) - // The live loop's pass is the warm pass; a second loop would only compete with it. - if !live { startWarmupLoop { await streaming.warmPass() } } - } else if let engine = cycleEngine { - startWarmupLoop { [warmupSamples = Self.warmupSamples] in - _ = try? await engine.transcribe(warmupSamples) - } - } - } - - // One loop for feed and live passes, so they never drain against each other. At - // release the loop is cancelled and awaited: a chunk drained just before is still - // fed, so every sample reaches the engine once and in order. - private func startStreamingFeed(_ engine: any StreamingTranscriptionEngine, utterance: Utterance, live: Bool) { - let feed = Task { [weak self, clock, feedInterval, livePassInterval] in - var fed = 0 - while !Task.isCancelled { - if !live { try? await clock.sleep(for: feedInterval) } - guard let self, !Task.isCancelled, self.state.isRecording else { return fed } - let chunk = await self.capture.drain() - if !chunk.isEmpty { - fed += chunk.count - await engine.feed(chunk, to: utterance) - } - guard live, !Task.isCancelled else { continue } - let text = await engine.livePass(utterance) - // A pass finishing after the release would put stale text back on screen. - guard !Task.isCancelled, self.state.isRecording else { return fed } - self.partialTranscript = text - try? await clock.sleep(for: livePassInterval) - } - return fed - } - stream = Stream(engine: engine, utterance: utterance, feed: feed) - } - - private func startWarmupLoop(_ warm: @escaping @Sendable () async -> Void) { - warmupTask?.cancel() - warmupTask = Task.detached(priority: .utility) { [clock, interval = warmupInterval] in - while !Task.isCancelled { - await warm() - if Task.isCancelled { return } - try? await clock.sleep(for: interval) - } - } - } - private func armOutputMute() { guard settings.muteOutputWhileDictating, let outputMuter else { return } let session = recordingID @@ -493,7 +432,7 @@ public final class DictationCoordinator { // Synchronous, and every call only flips a flag, cancels a task or spawns one, // so on the release path this costs nothing before `recordingStopped`. @discardableResult - private func endRecording() -> Stream? { + private func endRecording() -> RecordingEngineSession.Ending? { settleTask?.cancel() settleTask = nil gesture.reset() @@ -504,20 +443,17 @@ public final class DictationCoordinator { inputLevel = 0 maxDurationTask?.cancel() maxDurationTask = nil - let ended = stream - ended?.feed.cancel() - stream = nil // Before the state flip, so no further pass is queued ahead of the real call. - warmupTask?.cancel() - warmupTask = nil + let ending = engineSession?.end() + engineSession = nil partialTranscript = nil restoreOutputDevice() - return ended + return ending } public func hotkeyReleased(submit: Bool = false) { guard state.isRecording else { return } - let stream = endRecording() + let ending = endRecording() state = .transcribing // Snapshots anything copied while speaking now, beside the engine pass, not in // the paste. @@ -534,19 +470,14 @@ public final class DictationCoordinator { // The feed loop finishes handing over a chunk drained before the release. What // it waits for is engine work `endUtterance` would wait for anyway: engine time. let handover = ContinuousClock.now - let fed = await stream?.feed.value + let fed = await ending?.feed?.value let stopped = ContinuousClock.now let feedWait = stopped - handover let audio = await self.capture.stop() let captureStop = ContinuousClock.now - stopped - let source: DictationFinish.Source = if let stream { - .utterance(stream.engine, stream.utterance) - } else { - .whole(engine) - } let outcome = await finish.run( DictationFinish.Input( - audio: audio, fedSamples: fed ?? 0, source: source, submit: submit, + audio: audio, fedSamples: fed ?? 0, source: ending?.source ?? .whole(engine), submit: submit, captureStop: captureStop, feedWait: feedWait), onStage: { self.enter($0) }) self.finished(outcome) @@ -594,19 +525,15 @@ public final class DictationCoordinator { // Synchronous, so the state is idle on return; the task stops the microphone. private func beginCancel() -> Task? { guard state.isRecording else { return nil } - let stream = endRecording() + let ending = endRecording() willPolish = false becomeIdle() cycleEngine = nil - // The mic goes off first; the engine drops the utterance only once the feed has - // exited, so no chunk lands after it. Then it waits for the cancel before it. + // The mic goes off first, then the utterance; then it waits for the cancel before it. let previous = cancelCleanup let cleanup = Task { [capture] in _ = await capture.stop() - if let stream { - _ = await stream.feed.value - await stream.engine.abandonUtterance(stream.utterance) - } + await ending?.abandon() await previous?.value } cancelCleanup = cleanup diff --git a/Sources/PladderCore/RecordingEngineSession.swift b/Sources/PladderCore/RecordingEngineSession.swift new file mode 100644 index 0000000..5f6422d --- /dev/null +++ b/Sources/PladderCore/RecordingEngineSession.swift @@ -0,0 +1,134 @@ +import Foundation + +// See docs/ARCHITECTURE.md, "Dictation flow". +@MainActor +final class RecordingEngineSession { + struct Intervals: Sendable { + var feed: Duration + var livePass: Duration + var warmup: Duration + } + + struct Ending: Sendable { + let source: DictationFinish.Source + // Awaited at release: a chunk drained just before is still fed, so every sample + // reaches the engine once and in order. + let feed: Task? + + // The engine drops the utterance only once the feed has exited, so no chunk lands + // after it. + func abandon() async { + guard case .utterance(let engine, let utterance) = source else { return } + _ = await feed?.value + await engine.abandonUtterance(utterance) + } + } + + // Every utterance is padded to the model's full window, so half a second of + // silence costs the same encoder pass as a real call. + static let warmupSamples = [Float](repeating: 0, count: 8_000) + + private let engine: any TranscriptionEngine + private let capture: any AudioCapture + private let clock: any Clock + private let intervals: Intervals + private let live: Bool + private let onPartial: @MainActor (String?) -> Void + private var streaming: (engine: any StreamingTranscriptionEngine, utterance: Utterance)? + private var feed: Task? + // Cancelling cannot abort a CoreML call already started, so a release landing in + // a warm pass waits for it; the log shows that as `engine` above `engine-time`. + private var warmup: Task? + private(set) var isEnded = false + + init( + engine: any TranscriptionEngine, + capture: any AudioCapture, + clock: any Clock, + intervals: Intervals, + live: Bool, + onPartial: @escaping @MainActor (String?) -> Void + ) { + self.engine = engine + self.capture = capture + self.clock = clock + self.intervals = intervals + self.live = live + self.onPartial = onPartial + } + + // A streaming engine that cannot begin an utterance is used like a batch engine: + // transcribed whole at release, slower but not lost. + func begin() async { + guard let engine = engine as? (any StreamingTranscriptionEngine), + let utterance = try? await engine.beginUtterance() + else { return } + guard !isEnded else { + await engine.abandonUtterance(utterance) + return + } + streaming = (engine, utterance) + } + + func startLoops() { + guard !isEnded else { return } + if let streaming { + startFeed(streaming.engine, utterance: streaming.utterance) + // The live loop's pass is the warm pass; a second loop would only compete with it. + if !live { startWarmupLoop { [engine = streaming.engine] in await engine.warmPass() } } + } else { + startWarmupLoop { [engine, warmupSamples = Self.warmupSamples] in + _ = try? await engine.transcribe(warmupSamples) + } + } + } + + // Synchronous: it only cancels, so the release path waits on nothing here. + @discardableResult + func end() -> Ending { + isEnded = true + feed?.cancel() + warmup?.cancel() + warmup = nil + let source: DictationFinish.Source = if let streaming { + .utterance(streaming.engine, streaming.utterance) + } else { + .whole(engine) + } + return Ending(source: source, feed: feed) + } + + // One loop for feed and live passes, so they never drain against each other. + private func startFeed(_ engine: any StreamingTranscriptionEngine, utterance: Utterance) { + feed = Task { [weak self, capture, clock, intervals, live] in + var fed = 0 + while !Task.isCancelled { + if !live { try? await clock.sleep(for: intervals.feed) } + guard let self, !Task.isCancelled, !self.isEnded else { return fed } + let chunk = await capture.drain() + if !chunk.isEmpty { + fed += chunk.count + await engine.feed(chunk, to: utterance) + } + guard live, !Task.isCancelled else { continue } + let text = await engine.livePass(utterance) + // A pass finishing after the release would put stale text back on screen. + guard !Task.isCancelled, !self.isEnded else { return fed } + self.onPartial(text) + try? await clock.sleep(for: intervals.livePass) + } + return fed + } + } + + private func startWarmupLoop(_ warm: @escaping @Sendable () async -> Void) { + warmup?.cancel() + warmup = Task.detached(priority: .utility) { [clock, interval = intervals.warmup] in + while !Task.isCancelled { + await warm() + if Task.isCancelled { return } + try? await clock.sleep(for: interval) + } + } + } +} diff --git a/Tests/PladderCoreTests/RecordingEngineSessionTests.swift b/Tests/PladderCoreTests/RecordingEngineSessionTests.swift new file mode 100644 index 0000000..2ef2cac --- /dev/null +++ b/Tests/PladderCoreTests/RecordingEngineSessionTests.swift @@ -0,0 +1,143 @@ +import Foundation +import PladderTestSupport +import Testing +@testable import PladderCore + +@MainActor +@Suite(.timeLimit(.minutes(1))) struct RecordingEngineSessionTests { + private func session( + _ engine: any TranscriptionEngine, clock: ManualClock, live: Bool = false, + partials: Recorder = Recorder() + ) async -> RecordingEngineSession { + let capture = FakeCapture() + await capture.setDrainSamples(Array(repeating: 0.1, count: 8_000)) + return RecordingEngineSession( + engine: engine, capture: capture, clock: clock, + intervals: .init(feed: .seconds(1), livePass: .milliseconds(500), warmup: .seconds(2)), + live: live, onPartial: { partials.append($0) }) + } + + private func loadedStreaming( + livePassDelay: Duration = .zero, feedDelay: Duration = .zero, beginDelay: Duration = .zero, + beginFails: Bool = false + ) async throws -> FakeStreamingEngine { + let engine = FakeStreamingEngine( + livePassDelay: livePassDelay, feedDelay: feedDelay, beginDelay: beginDelay, beginFails: beginFails) + try await engine.load() + return engine + } + + @Test func theFeedLoopHandsOverEveryDrainedChunkOnce() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming() + let s = await session(engine, clock: clock) + await s.begin() + s.startLoops() + for second in 1...3 { + #expect(await clock.waitForSleeper(at: .seconds(second))) + clock.advance(to: .seconds(second)) + #expect(await waitUntil { engine.feedCounts.count == second }) + } + let ending = s.end() + guard case .utterance = ending.source else { + Issue.record("expected the utterance, got \(ending.source)") + return + } + #expect(await ending.feed?.value == 24_000) + #expect(engine.feedCounts == [8_000, 8_000, 8_000]) + } + + @Test func endCancelsTheWarmLoopBeforeReturning() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming() + let s = await session(engine, clock: clock) + await s.begin() + s.startLoops() + #expect(await waitUntil { engine.warmPassCount == 1 }) + #expect(await clock.waitForSleeper(at: .seconds(2))) + s.end() + #expect(clock.sleeperCount == 0) + clock.advance(by: .seconds(10)) + #expect(engine.warmPassCount == 1) + } + + @Test func aLivePassFinishingAfterEndPublishesNothing() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming(livePassDelay: .milliseconds(50)) + let partials = Recorder() + let s = await session(engine, clock: clock, live: true, partials: partials) + await s.begin() + s.startLoops() + #expect(await waitUntil { partials.all == ["partial 1"] }) + #expect(await clock.waitForSleeper(at: .milliseconds(500))) + clock.advance(to: .milliseconds(500)) + #expect(await waitUntil { engine.livePassCount == 2 }) + let ending = s.end() + _ = await ending.feed?.value + #expect(partials.all == ["partial 1"]) + #expect(engine.warmPassCount == 0) + } + + @Test func aBatchEngineGetsOnlyTheWarmLoop() async throws { + let clock = ManualClock() + let engine = CountingEngine() + try await engine.load() + let s = await session(engine, clock: clock) + await s.begin() + s.startLoops() + #expect(await waitUntil { engine.calls == [8_000] }) + let ending = s.end() + guard case .whole = ending.source else { + Issue.record("expected the whole buffer, got \(ending.source)") + return + } + #expect(ending.feed == nil) + #expect(clock.sleeperCount == 0) + } + + @Test func aStreamingEngineThatCannotBeginIsWarmedAndTranscribedWhole() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming(beginFails: true) + let s = await session(engine, clock: clock) + await s.begin() + s.startLoops() + #expect(await waitUntil { engine.log.contains("transcribe") }) + let ending = s.end() + guard case .whole = ending.source else { + Issue.record("expected the whole buffer, got \(ending.source)") + return + } + #expect(ending.feed == nil) + } + + @Test func abandonWaitsForTheFeedBeforeDroppingTheUtterance() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming(feedDelay: .milliseconds(50)) + let s = await session(engine, clock: clock) + await s.begin() + s.startLoops() + #expect(await clock.waitForSleeper(at: .seconds(1))) + clock.advance(to: .seconds(1)) + #expect(await waitUntil { engine.isFeeding }) + await s.end().abandon() + #expect(Array(engine.log.suffix(2)) == ["fed", "abandon"]) + } + + @Test func aBeginOvertakenByEndDropsItsUtteranceAndStartsNoLoop() async throws { + let clock = ManualClock() + let engine = try await loadedStreaming(beginDelay: .milliseconds(50)) + let s = await session(engine, clock: clock) + let beginning = Task { await s.begin() } + #expect(await waitUntil { engine.log.contains("begin") }) + let ending = s.end() + await beginning.value + s.startLoops() + guard case .whole = ending.source else { + Issue.record("expected the whole buffer, got \(ending.source)") + return + } + #expect(engine.log.last == "abandon") + #expect(clock.sleeperCount == 0) + #expect(engine.warmPassCount == 0) + } +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8632445..70c60d8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -42,9 +42,10 @@ its timers with `ManualClock`. running, and starts the capture; 4. arms the output mute and starts the key-down work: the clipboard snapshot and the polish model's load; -5. starts the engine work. A streaming engine begins an utterance and gets a - feed loop. In the other styles the loop wakes every second and hands the - engine the audio captured since. With the Live Transcript style it skips +5. starts the engine work, a `RecordingEngineSession` on the coordinator's + clock. A streaming engine begins an utterance and gets a feed loop. In + the other styles the loop wakes every second and hands the engine the + audio captured since. With the Live Transcript style it skips that second: each turn hands over what has been captured, asks for the text so far, publishes it as `partialTranscript` and then waits half a second, so audio goes in on the live-pass cadence. Outside Live @@ -53,7 +54,10 @@ its timers with `ManualClock`. Neural Engine from going cold; in Live Transcript the live passes keep it warm instead. A batch engine, or a streaming one that could not begin, gets the warm loop in every style, and nothing else, and is transcribed - whole at release; + whole at release. A session ended while its utterance was still beginning + drops that utterance itself and starts no loop. At release or cancel the + session hands the coordinator what it ended with: the utterance or the + whole buffer to transcribe, and the feed loop to wait for; 6. starts the level meter and the 10 minute cap. **Release** (`hotkeyReleased`). `endRecording()` stops everything the press