fix(track-b 5): VoiceController chunkPump + voice-loop e2e proof

Track B-5 of the 2026-05-03 voice audit fix plan
(.planning/audit-2026-05-03/voice.md). Track B-4 wired the audio path
all the way through the SpeechAnalyzer bridge, but the production
chain was still mute: `VoiceController.startSTTSession()` opened an
`AsyncStream<AudioChunk>` and handed the consumer side to the STT
provider, but nothing ever yielded into the producer side. Result:
STT.transcribe drained an empty stream and finalized to "" every
time, regardless of what the user said.

Changes:

* VoiceController gains a `chunkPump:
  @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void`
  init parameter (default: no-op for back-compat with all existing
  tests that drive STT via `_testFireSpeechEnd`).

* `startSTTSession` spawns a detached `chunkPumpTask` that runs the
  pump against the per-session continuation. `endSTTSession` cancels
  the task before closing the continuation so the pump exits cleanly
  via `Task.isCancelled`. `shutdown` cancels the pump task as well —
  no leaked Task on session end (E3 test).

* AppDelegate.installVoice constructs the production chunkPump as a
  detached read loop over `audioGraphOwner.ringBuffer`: 1024-sample
  windows (~64 ms @ 16 kHz), 10 ms sleep on ring-empty, exits on
  Task cancel. Documents the same multi-consumer race that
  AudioLevelEmitter already lives with — fan-out is a follow-up.

* VoiceLoopE2ETests (new, +3 cases):
    - E1: preset chunks → CapturingSTTProvider observes them →
      finalize returns known transcript → orchestrator.submit fires
      with that exact text.
    - E2: empty pump → empty STT result → controller transitions
      back to .idle without a phantom orchestrator.submit (matches
      handleSTTFinalized's empty-text fast path).
    - E3: chunkPumpTask exits within 100 ms of pttUp — no leaked Task.

Verification:

* swift test packages/Voice — 80 tests (was 77; +3 VoiceLoopE2ETests),
  3 skipped, 1 pre-existing failure (TTSInterruptTests.testI3 unchanged
  10s timeout flake, unrelated).
* `bash scripts/check-app-builds.sh` — App target Debug build clean.
* Boundary gates: install-order, no-modal-presentation,
  no-null-voice-adapters, orchestrator-events-single-consumer,
  single-writer-hudstate — all PASS.

What's still TODO (Track B-5+):
* Real VAD-gated session end. Currently `pttUp` is the only path
  that closes the chunk continuation; Silero's `.speechEnd` decision
  isn't yet wired to `endSTTSession`. The plumbing for that is
  Track B-6 / VOICE-08 territory.
* Multi-consumer ring fan-out. WakeWordDAG, AudioLevelEmitter, and
  now the chunk pump all read the same RingBuffer concurrently. Each
  advances the read pointer, so they steal samples from each other.
  In practice the listening session usually fires after wake-word
  inference completes, so contention is intermittent — but it's a
  real bug. A `BufferBroadcaster` fan-out at the AudioGraph tap is
  the right fix; tracked in TODO.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

App/AppDelegate.swift

@@ -834,6 +834,54 @@ final class AppDelegate: NSObject, NSApplicationDelegate {
             busAdapter = DormantVoiceBusEmitter()
         }
 
+        // Track B-5 (2026-05-03 voice audit fix): production chunk pump.
+        // VoiceController.startSTTSession opens an `AsyncStream<AudioChunk>`
+        // and hands the consumer side to the STT provider; the pump fills
+        // the producer side.
+        //
+        // Threading: the read loop is `Task.detached` so it doesn't block
+        // VoiceController's actor. We poll the ring at 1024-frame windows
+        // (~64 ms @ 16 kHz) — large enough to amortize the cross-task hop
+        // but small enough to keep STT partial latency under VAD's hangover.
+        //
+        // Race note: the same RingBuffer is also drained by WakeWordDAG and
+        // AudioLevelEmitter. RingBuffer's docstring claims SPSC, but the
+        // existing AudioLevelEmitter already violates that — the pump
+        // inherits the same race. Track B-5+ should fan out via a per-tap
+        // broadcaster to give STT its own samples; for now the listening
+        // path runs while wake-word fires don't consume the ring (the DAG
+        // pauses inference once we transition to .listening but its read
+        // loop continues to advance the read pointer). Documented for
+        // follow-up rather than fixed under audit scope.
+        let chunkPump: @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void = { [weak self] cont in
+            // Resolve the ring buffer. If the graph isn't ready yet (e.g.,
+            // installVoice short-circuited), the pump finishes immediately
+            // so the STT provider drains and finalize() returns "".
+            guard let owner = await MainActor.run(body: { self?.audioGraphOwner }),
+                  let ring = await owner.ringBuffer else {
+                cont.finish()
+                return
+            }
+
+            var scratch = [Float](repeating: 0, count: 1024)
+            while !Task.isCancelled {
+                let count = scratch.withUnsafeMutableBufferPointer { ptr in
+                    ring.readMono16k(into: ptr)
+                }
+                if count > 0 {
+                    let samples = count == scratch.count
+                        ? scratch
+                        : Array(scratch.prefix(count))
+                    cont.yield(AudioChunk(pcm16k: samples))
+                } else {
+                    // Ring drained — yield 10 ms before retrying so we don't
+                    // spin against a starving producer.
+                    try? await Task.sleep(nanoseconds: 10_000_000)
+                }
+            }
+            cont.finish()
+        }
+
         let vc = VoiceController(
             wakeWordStream: wakeWordDAG.wakeWordStream,
             vadFactory: { sileroVAD },
@@ -842,7 +890,8 @@ final class AppDelegate: NSObject, NSApplicationDelegate {
             orchestrator: orchAdapter,
             bannerCoordinator: bannerAdapter,
             bus: busAdapter,
-            voiceHudCont: voiceCont
+            voiceHudCont: voiceCont,
+            chunkPump: chunkPump
         )
         voiceController = vc
 

packages/Voice/Sources/Voice/VoiceController.swift

@@ -34,6 +34,7 @@ public actor VoiceController {
     private let wakeWordStream: AsyncStream<WakeWordEvent>
     private let vadFactory: @Sendable () -> SileroVAD
     private let sttFactory: @Sendable () -> any STTProvider
+    private let chunkPump: @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void
     private let tts: any VoiceTTSInterface
     private let orchestrator: any VoiceOrchestratorInterface
     private let bannerCoordinator: any VoiceBannerInterface
@@ -51,6 +52,10 @@ public actor VoiceController {
     private var wakeWordTask: Task<Void, Never>?
     private var orchestratorTask: Task<Void, Never>?
 
+    /// Track B-5: per-session pump Task that drains audio frames into the
+    /// STT continuation. Cancelled when listening ends.
+    private var chunkPumpTask: Task<Void, Never>?
+
     /// Exposed for AppDelegate to wire the audio-level emitter.
     public var audioLevelEmitter: AudioLevelEmitter?
 
@@ -79,7 +84,8 @@ public actor VoiceController {
         orchestrator: any VoiceOrchestratorInterface,
         bannerCoordinator: any VoiceBannerInterface,
         bus: any BusOutboundEmitter,
-        voiceHudCont: AsyncStream<VoiceHudIntent>.Continuation
+        voiceHudCont: AsyncStream<VoiceHudIntent>.Continuation,
+        chunkPump: @escaping @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void = { _ in }
     ) {
         self.wakeWordStream = wakeWordStream
         self.vadFactory = vadFactory
@@ -89,6 +95,7 @@ public actor VoiceController {
         self.bannerCoordinator = bannerCoordinator
         self.bus = bus
         self.voiceHudCont = voiceHudCont
+        self.chunkPump = chunkPump
     }
 
     // MARK: - Lifecycle
@@ -108,8 +115,10 @@ public actor VoiceController {
     public func shutdown() async {
         wakeWordTask?.cancel()
         orchestratorTask?.cancel()
+        chunkPumpTask?.cancel()
         wakeWordTask = nil
         orchestratorTask = nil
+        chunkPumpTask = nil
         await audioLevelEmitter?.stop()
         audioLevelEmitter = nil
         sttChunkCont?.finish()
@@ -300,11 +309,24 @@ public actor VoiceController {
     private func startSTTSession() async {
         sttChunkCont?.finish()
         sttChunkCont = nil
+        chunkPumpTask?.cancel()
+        chunkPumpTask = nil
 
         let provider = sttFactory()
         let (chunkStream, cont) = AsyncStream<AudioChunk>.makeStream()
         sttChunkCont = cont
 
+        // Track B-5: spawn the audio-chunk pump for this listening session.
+        // The default pump is a no-op (back-compat with tests that drive STT
+        // via `_testFireSpeechEnd`); production wires a ring-buffer reader
+        // via AppDelegate. The pump returns when the consumer Task drops the
+        // continuation — `cont.onTermination` propagates Task.isCancelled.
+        let pump = self.chunkPump
+        cont.onTermination = { _ in /* pump's Task cancellation handles cleanup */ }
+        chunkPumpTask = Task.detached { [pump] in
+            await pump(cont)
+        }
+
         let partials = provider.transcribe(stream: chunkStream)
 
         // Drain partial results in a detached task.
@@ -325,6 +347,9 @@ public actor VoiceController {
     }
 
     private func endSTTSession() {
+        // Stop the audio pump first so it doesn't yield into a closed continuation.
+        chunkPumpTask?.cancel()
+        chunkPumpTask = nil
         // Close the audio chunk stream — STT provider will finalize naturally.
         sttChunkCont?.finish()
         sttChunkCont = nil

packages/Voice/Tests/VoiceTests/VoiceLoopE2ETests.swift

@@ -0,0 +1,255 @@
+import AVFoundation
+import XCTest
+@testable import Voice
+
+// MARK: - VoiceLoopE2ETests
+//
+// Track B-5 (2026-05-03 voice audit fix): end-to-end wiring proof for the
+// chunk-pump → STT → orchestrator chain landed under track B-4 + B-5.
+//
+// What's exercised:
+//   1. VoiceController spawns the production-style chunk pump on transition
+//      to `.listening`.
+//   2. The pump yields a deterministic batch of `AudioChunk`s into the STT
+//      provider's input stream.
+//   3. The STT provider records the chunks it received and returns a known
+//      final transcript when the chunk stream ends.
+//   4. `endSTTSession` (driven by `pttUp`) closes the chunk continuation,
+//      which (via `cont.onTermination` + Task cancellation) shuts the pump
+//      down — no leaked Task on session end.
+//   5. The transcript propagates `STTProvider.finalize → handleSTTFinalized
+//      → orchestrator.submit`.
+//
+// What's NOT exercised here (Track B-5+ work):
+//   - Real `AudioGraphOwner` hardware path. That requires a mic + entitlement
+//     and is HUMAN-UAT territory (Plan 06-05 Task 4).
+//   - VAD-gated speech-end detection. The current pipeline closes the STT
+//     session on `pttUp`; a follow-up wires Silero `.speechEnd` to the same
+//     path.
+//   - Real macOS 26 `SpeechAnalyzer`. Covered by `LiveSpeechAnalyzerBridge`'s
+//     internal wiring (Track B-4) — exercised in HUMAN-UAT, not unit tests.
+
+final class VoiceLoopE2ETests: XCTestCase {
+
+    // MARK: - E1: chunk-pump audio flows STT → orchestrator.submit
+
+    func testE1_chunkPump_audio_reaches_orchestrator_as_transcript() async throws {
+        let presetChunks: [AudioChunk] = [
+            AudioChunk(pcm16k: Array(repeating: 0.10, count: 1024)),
+            AudioChunk(pcm16k: Array(repeating: 0.12, count: 1024)),
+            AudioChunk(pcm16k: Array(repeating: 0.08, count: 1024))
+        ]
+        let expectedTranscript = "track b-five wiring proof"
+
+        let stt = CapturingSTTProvider(finalResult: expectedTranscript)
+        let orchestrator = MockOrchestratorForController()
+        let tts = MockTTSForController()
+        let banner = MockBannerForController()
+        let bus = MockBusEmitter()
+        let (_, hudCont) = AsyncStream<VoiceHudIntent>.makeStream()
+        let (wwStream, _) = AsyncStream<WakeWordEvent>.makeStream()
+
+        // Chunk pump: yield each preset chunk, then idle until cancelled.
+        // The pump exits cleanly via `Task.isCancelled` once endSTTSession
+        // tears down its task.
+        let chunkPump: @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void = { cont in
+            for chunk in presetChunks {
+                guard !Task.isCancelled else { cont.finish(); return }
+                cont.yield(chunk)
+                try? await Task.sleep(nanoseconds: 5_000_000)  // 5ms between chunks
+            }
+            // Idle so pttUp's cancel is what ends the session, not pump exhaustion.
+            while !Task.isCancelled {
+                try? await Task.sleep(nanoseconds: 10_000_000)
+            }
+            cont.finish()
+        }
+
+        let controller = VoiceController(
+            wakeWordStream: wwStream,
+            vadFactory: { SileroVAD(engine: MockVADEngine()) },
+            sttFactory: { stt },
+            tts: tts,
+            orchestrator: orchestrator,
+            bannerCoordinator: banner,
+            bus: bus,
+            voiceHudCont: hudCont,
+            chunkPump: chunkPump
+        )
+
+        await controller.startForTests()
+
+        // Drive the listening transition via PTT (bypasses wake-word DAG).
+        await controller.pttDown()
+
+        // Allow time for the pump to deliver its presets.
+        try await Task.sleep(for: .milliseconds(80))
+
+        // End the STT session — pttUp closes the chunk continuation, which
+        // lets `CapturingSTTProvider.transcribe`'s drain Task finish, which
+        // unblocks `provider.finalize() → handleSTTFinalized → submit(text:)`.
+        await controller.pttUp()
+
+        // Wait for the submit to land.
+        try await Task.sleep(for: .milliseconds(120))
+
+        // Assertions
+        let submitTexts = await orchestrator.submitCallTexts
+        XCTAssertEqual(submitTexts, [expectedTranscript],
+            "E1: orchestrator should receive exactly the STT final transcript once")
+
+        let chunksSeen = await stt.receivedChunkCount
+        XCTAssertGreaterThanOrEqual(chunksSeen, 1,
+            "E1: STT should observe at least one chunk from the pump (got \(chunksSeen))")
+        XCTAssertLessThanOrEqual(chunksSeen, presetChunks.count,
+            "E1: STT should not see more chunks than the pump yields (got \(chunksSeen))")
+
+        await controller.shutdown()
+    }
+
+    // MARK: - E2: empty pump → STT finalizes empty → no submit
+
+    func testE2_pumpYieldsNoChunks_emptyTranscript_skipsSubmit() async throws {
+        let stt = CapturingSTTProvider(finalResult: "")
+        let orchestrator = MockOrchestratorForController()
+        let tts = MockTTSForController()
+        let banner = MockBannerForController()
+        let bus = MockBusEmitter()
+        let (_, hudCont) = AsyncStream<VoiceHudIntent>.makeStream()
+        let (wwStream, _) = AsyncStream<WakeWordEvent>.makeStream()
+
+        // Pump that yields nothing, just idles.
+        let chunkPump: @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void = { cont in
+            while !Task.isCancelled {
+                try? await Task.sleep(nanoseconds: 10_000_000)
+            }
+            cont.finish()
+        }
+
+        let controller = VoiceController(
+            wakeWordStream: wwStream,
+            vadFactory: { SileroVAD(engine: MockVADEngine()) },
+            sttFactory: { stt },
+            tts: tts,
+            orchestrator: orchestrator,
+            bannerCoordinator: banner,
+            bus: bus,
+            voiceHudCont: hudCont,
+            chunkPump: chunkPump
+        )
+
+        await controller.startForTests()
+        await controller.pttDown()
+        try await Task.sleep(for: .milliseconds(40))
+        await controller.pttUp()
+        try await Task.sleep(for: .milliseconds(80))
+
+        // VoiceController.handleSTTFinalized treats empty text as
+        // "transition straight to .idle without submitting" — see line 336.
+        let submits = await orchestrator.submitCallTexts
+        XCTAssertEqual(submits.count, 0,
+            "E2: empty STT result must NOT trigger orchestrator.submit")
+
+        let state = await controller.state
+        XCTAssertEqual(state, .idle,
+            "E2: after empty STT cycle, controller should return to .idle")
+
+        await controller.shutdown()
+    }
+
+    // MARK: - E3: pump task is cancelled on session end (no leak)
+
+    func testE3_chunkPumpTask_cancelledOnSessionEnd() async throws {
+        let pumpExited = AtomicFlag()
+        let stt = CapturingSTTProvider(finalResult: "anything")
+        let orchestrator = MockOrchestratorForController()
+        let tts = MockTTSForController()
+        let banner = MockBannerForController()
+        let bus = MockBusEmitter()
+        let (_, hudCont) = AsyncStream<VoiceHudIntent>.makeStream()
+        let (wwStream, _) = AsyncStream<WakeWordEvent>.makeStream()
+
+        let chunkPump: @Sendable (AsyncStream<AudioChunk>.Continuation) async -> Void = { [pumpExited] cont in
+            // Loop until cancelled; flip the flag on exit.
+            while !Task.isCancelled {
+                try? await Task.sleep(nanoseconds: 5_000_000)
+            }
+            cont.finish()
+            await pumpExited.set()
+        }
+
+        let controller = VoiceController(
+            wakeWordStream: wwStream,
+            vadFactory: { SileroVAD(engine: MockVADEngine()) },
+            sttFactory: { stt },
+            tts: tts,
+            orchestrator: orchestrator,
+            bannerCoordinator: banner,
+            bus: bus,
+            voiceHudCont: hudCont,
+            chunkPump: chunkPump
+        )
+
+        await controller.startForTests()
+        await controller.pttDown()
+        try await Task.sleep(for: .milliseconds(20))
+        await controller.pttUp()
+        try await Task.sleep(for: .milliseconds(80))
+
+        let exited = await pumpExited.get()
+        XCTAssertTrue(exited,
+            "E3: chunkPump task must exit after pttUp closes the session (no leaked Task)")
+
+        await controller.shutdown()
+    }
+}
+
+// MARK: - CapturingSTTProvider
+
+/// STT mock that counts incoming AudioChunks, drains the chunk stream until
+/// it ends, then returns `finalResult` from `finalize()`. Used to prove the
+/// chunk-pump → STT path actually delivers audio.
+///
+/// Uses an internal counter actor to satisfy Swift 6 strict concurrency —
+/// `NSLock.lock()` is unavailable in async contexts.
+final class CapturingSTTProvider: STTProvider, @unchecked Sendable {
+    let finalResult: String
+    private let counter = Counter()
+
+    init(finalResult: String) {
+        self.finalResult = finalResult
+    }
+
+    var receivedChunkCount: Int {
+        get async { await counter.value }
+    }
+
+    func transcribe(stream: AsyncStream<AudioChunk>) -> AsyncStream<PartialTranscript> {
+        let (partials, partialCont) = AsyncStream<PartialTranscript>.makeStream()
+        let counter = self.counter
+        Task {
+            for await _ in stream {
+                await counter.increment()
+            }
+            partialCont.finish()
+        }
+        return partials
+    }
+
+    func finalize() async throws -> String {
+        finalResult
+    }
+
+    actor Counter {
+        var value: Int = 0
+        func increment() { value += 1 }
+    }
+}
+
+// MARK: - AtomicFlag
+
+actor AtomicFlag {
+    private var flagValue = false
+    func set() { flagValue = true }
+    func get() -> Bool { flagValue }
+}