From 1e394d661db9d279629fa104c1a069eb82cc0fbe Mon Sep 17 00:00:00 2001 From: Apoorv Darshan Date: Mon, 19 Oct 2026 09:02:17 +0530 Subject: [PATCH 1/3] ci(dart-sdk): run the Dart device SDK tests No workflow or manifest lane ran the tests in sdks/device/dart, so the package's STT fixes were only ever tested on a contributor's machine. The Hygiene manifest runner has no Flutter toolchain, so add a path-scoped workflow like sdk-cpp.yml that sets up the pinned Flutter 3.44.5 and runs flutter test for the package, and record it in the CI budget inventory. Co-Authored-By: Claude Opus 5.5 --- .github/ci-budget.yaml | 40 ++++++++++++++++++++++++++++++++++ .github/workflows/sdk-dart.yml | 34 +++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+) create mode 100644 .github/workflows/sdk-dart.yml diff --git a/.github/ci-budget.yaml b/.github/ci-budget.yaml index fa3ed76738e..b3222acaa5a 100644 --- a/.github/ci-budget.yaml +++ b/.github/ci-budget.yaml @@ -2411,6 +2411,24 @@ }, "runs_on": "ubuntu-latest" }, + "sdk-dart.yml/dart": { + "budget_p50_minutes": 5, + "display_name": "dart", + "expected_runs_class": "unsampled-pr-sentinel", + "expected_runs_per_day": 1, + "if": null, + "matrix": {}, + "matrix_size": 1, + "needs": [], + "owner": "CI maintainers", + "rationale": "2026-10-19 new path-scoped job: two PRs touched sdks/device/dart since 2026-07-30; 5 minutes covers cached Flutter 3.44.5 setup, pub get and the package tests", + "reusable": null, + "runner": "linux", + "runner_counts": { + "linux": 1 + }, + "runs_on": "ubuntu-latest" + }, "sdk-rust.yml/rust": { "budget_p50_minutes": 5, "display_name": "rust", @@ -4872,6 +4890,28 @@ } } }, + "sdk-dart.yml": { + "name": "Dart Device SDK", + "owner": "CI maintainers", + "rationale": "2026-10-19: path-scoped like sdk-cpp.yml; no workflow or manifest lane ran the Dart device SDK tests, and the Hygiene runner has no Flutter toolchain", + "triggers": { + "pull_request": { + "paths": [ + "sdks/device/dart/**", + ".github/workflows/sdk-dart.yml" + ] + }, + "push": { + "branches": [ + "main" + ], + "paths": [ + "sdks/device/dart/**", + ".github/workflows/sdk-dart.yml" + ] + } + } + }, "sdk-rust.yml": { "name": "Rust SDK", "owner": "CI maintainers", diff --git a/.github/workflows/sdk-dart.yml b/.github/workflows/sdk-dart.yml new file mode 100644 index 00000000000..fb2b8e6697d --- /dev/null +++ b/.github/workflows/sdk-dart.yml @@ -0,0 +1,34 @@ +name: Dart Device SDK + +on: + pull_request: + paths: + - "sdks/device/dart/**" + - ".github/workflows/sdk-dart.yml" + push: + branches: [main] + paths: + - "sdks/device/dart/**" + - ".github/workflows/sdk-dart.yml" + +permissions: + contents: read + +jobs: + dart: + runs-on: ubuntu-latest + timeout-minutes: 15 + defaults: + run: + working-directory: sdks/device/dart + steps: + - uses: actions/checkout@v7 + # The Hygiene manifest runner has no Flutter toolchain, so this Flutter + # package runs its own suite, like sdk-cpp.yml and sdk-rust.yml. + - uses: subosito/flutter-action@v2 + with: + channel: stable + flutter-version: 3.44.5 + cache: true + - run: flutter pub get + - run: flutter test From 5492de47756ea00b4bdfb72d1426a9b478a6dc0e Mon Sep 17 00:00:00 2001 From: Apoorv Darshan Date: Mon, 19 Oct 2026 09:02:26 +0530 Subject: [PATCH 2/3] fix(dart-sdk): report a Parakeet stream closed before ready ParakeetTranscriber drops PCM until the server sends ready. The server can accept the socket and then close it before ready (1013 capacity_full or service_not_ready in backend/parakeet/main.py). The stream listener only marked the stream drained, so every later appendPcm dropped its chunk and stop() completed normally: the caller lost the session's audio with no error to retry on or fall back from. When the stream ends before ready and before stop(), keep why it ended (close code and reason, or the stream error) and throw it as an SttStreamClosedException from appendPcm and stop(). A stream that is still waiting for ready keeps dropping PCM. This mirrors the Go SDK fix in #22301. stt.dart and stt_test.dart are reformatted with dart format --line-length 120, which CI's Formatting job applies to them at the latest language version because the package has no resolved config. Co-Authored-By: Claude Opus 5.5 --- ...eadiness-gate-swallows-terminal-close.json | 14 +++ sdks/device/dart/README.md | 10 ++ sdks/device/dart/lib/stt/stt.dart | 53 +++++++-- sdks/device/dart/test/stt_test.dart | 102 ++++++++++-------- 4 files changed, 130 insertions(+), 49 deletions(-) create mode 100644 .github/failure-classes/FC-readiness-gate-swallows-terminal-close.json diff --git a/.github/failure-classes/FC-readiness-gate-swallows-terminal-close.json b/.github/failure-classes/FC-readiness-gate-swallows-terminal-close.json new file mode 100644 index 00000000000..048aec1a3dc --- /dev/null +++ b/.github/failure-classes/FC-readiness-gate-swallows-terminal-close.json @@ -0,0 +1,14 @@ +{ + "schema_version": 1, + "id": "FC-readiness-gate-swallows-terminal-close", + "violated_contract": "A streaming client that holds back writes until the server's ready handshake must report a close that arrives before that handshake. The Go device SDK's Parakeet transcriber returned nil from AppendPCM whenever ready had not arrived and never checked whether its read loop had already ended. When the server accepted the WebSocket and then closed it with 1013 capacity_full or service_not_ready (backend/parakeet/main.py), every PCM chunk of the session was dropped while AppendPCM and Stop both returned nil.", + "canonical_prevention": "The read loop records the error that ended it before signalling completion, and the pre-ready write path checks that signal. A stream whose read loop ended without ready returns the recorded close, with its code still unwrappable, from every write and from Stop. A stream still waiting for ready keeps the documented drop-until-ready behaviour. A loopback test closes the stream before ready and asserts that both calls return that close.", + "canonical_prevention_artifact": [ + "sdks/device/go/omidevice/stt/parakeet_test.go" + ], + "evidence_prs": [], + "scope_hints": [ + "sdks/device/go/omidevice/stt/" + ], + "status": "open" +} diff --git a/sdks/device/dart/README.md b/sdks/device/dart/README.md index 404e55a45cd..324d0dc5361 100644 --- a/sdks/device/dart/README.md +++ b/sdks/device/dart/README.md @@ -29,3 +29,13 @@ Note: the production Omi app uses **native Pigeon BLE** for lifecycle. This pack ## STT See `sdks/device/STT.md`. + +`ParakeetTranscriber` drops PCM until the server sends `ready`. If the server closes the stream before +`ready` (for example 1013 `capacity_full` or `service_not_ready`), `appendPcm` and `stop` throw an +`SttStreamClosedException` carrying the close code and reason, so callers can retry or switch engines. + +## Tests + +Run `flutter test` in this directory. The STT tests drive the transcribers through an in-memory WebSocket +channel, so the suite needs no device, credentials, or transcription service. CI runs it for changes under +`sdks/device/dart/` (`.github/workflows/sdk-dart.yml`). diff --git a/sdks/device/dart/lib/stt/stt.dart b/sdks/device/dart/lib/stt/stt.dart index 9d56537e42f..3f5e6dbd988 100644 --- a/sdks/device/dart/lib/stt/stt.dart +++ b/sdks/device/dart/lib/stt/stt.dart @@ -14,6 +14,25 @@ abstract class StreamingTranscriber { Future stop(); } +/// Thrown by [ParakeetTranscriber.appendPcm] and [ParakeetTranscriber.stop] when the server closed +/// the stream before sending `ready`, e.g. 1013 `capacity_full` or `service_not_ready`. That stream +/// never becomes ready, so callers can retry or switch engines instead of losing the session's audio. +class SttStreamClosedException implements Exception { + const SttStreamClosedException({this.closeCode, this.closeReason, this.error}); + + /// The server's WebSocket close code, e.g. 1013 (try again later). + final int? closeCode; + + /// The server's close reason, e.g. `capacity_full`. + final String? closeReason; + + /// The stream error that ended the connection, when it failed without a close frame. + final Object? error; + + @override + String toString() => 'SttStreamClosedException: stream closed before ready: ${error ?? '$closeCode $closeReason'}'; +} + String deepgramWsUrl({int sampleRate = 16000}) { return 'wss://api.deepgram.com/v1/listen?punctuate=true&model=nova&language=en-US' '&encoding=linear16&sample_rate=$sampleRate&channels=1'; @@ -37,7 +56,8 @@ class DeepgramTranscriber implements StreamingTranscriber { this.drainTimeout = const Duration(seconds: 5), WebSocketChannel? channel, }) { - _channel = channel ?? + _channel = + channel ?? IOWebSocketChannel.connect( Uri.parse(deepgramWsUrl(sampleRate: sampleRate)), headers: {'Authorization': 'Token $apiKey'}, @@ -105,8 +125,7 @@ class ParakeetTranscriber implements StreamingTranscriber { this.drainTimeout = const Duration(seconds: 5), WebSocketChannel? channel, }) { - _channel = channel ?? - WebSocketChannel.connect(Uri.parse(parakeetWsUrl(apiUrl, sampleRate: sampleRate))); + _channel = channel ?? WebSocketChannel.connect(Uri.parse(parakeetWsUrl(apiUrl, sampleRate: sampleRate))); _sub = _channel.stream.listen( (event) { if (event is! String) return; @@ -120,8 +139,8 @@ class ParakeetTranscriber implements StreamingTranscriber { if (text != null && text.isNotEmpty) onTranscript(text); } catch (_) {} }, - onDone: _markDrained, - onError: (_) => _markDrained(), + onDone: _streamEnded, + onError: (Object error) => _streamEnded(error), ); } @@ -132,12 +151,26 @@ class ParakeetTranscriber implements StreamingTranscriber { StreamSubscription? _sub; bool _ready = false; bool _stopped = false; + SttStreamClosedException? _closedBeforeReady; final Completer _drained = Completer(); void _markDrained() { if (!_drained.isCompleted) _drained.complete(); } + // The server can end the stream before ready, e.g. 1013 capacity_full or service_not_ready + // (backend/parakeet/main.py). Such a stream never becomes ready, so keep why it ended. + void _streamEnded([Object? error]) { + if (!_ready && !_stopped) { + _closedBeforeReady ??= SttStreamClosedException( + closeCode: _channel.closeCode, + closeReason: _channel.closeReason, + error: error, + ); + } + _markDrained(); + } + static String? _extract(Map data) { final t = data['text'] ?? data['transcript']; if (t is String && t.isNotEmpty) return t; @@ -146,7 +179,13 @@ class ParakeetTranscriber implements StreamingTranscriber { @override void appendPcm(Uint8List chunk) { - if (_stopped || !_ready) return; + if (_stopped) return; + if (!_ready) { + // PCM before ready is dropped, unless the server already ended the stream. + final closed = _closedBeforeReady; + if (closed != null) throw closed; + return; + } _channel.sink.add(chunk); } @@ -169,6 +208,8 @@ class ParakeetTranscriber implements StreamingTranscriber { try { await _channel.sink.close(); } catch (_) {} + final closed = _closedBeforeReady; + if (closed != null) throw closed; } } diff --git a/sdks/device/dart/test/stt_test.dart b/sdks/device/dart/test/stt_test.dart index cfd050991fd..7fe88ca4cb9 100644 --- a/sdks/device/dart/test/stt_test.dart +++ b/sdks/device/dart/test/stt_test.dart @@ -41,26 +41,20 @@ void main() { test('Deepgram stop sends CloseStream before closing the socket', () async { final channel = RecordingWebSocketChannel(); - final transcriber = DeepgramTranscriber( - apiKey: 'fake-key', - onTranscript: (_) {}, - channel: channel, - ); + final transcriber = DeepgramTranscriber(apiKey: 'fake-key', onTranscript: (_) {}, channel: channel); await transcriber.stop(); expect(channel.events, ['send', 'close']); - expect(channel.sent, [jsonEncode({'type': 'CloseStream'})]); + expect(channel.sent, [ + jsonEncode({'type': 'CloseStream'}), + ]); expect(channel.closed, isTrue); }); test('Deepgram stop still closes when CloseStream send fails', () async { final channel = RecordingWebSocketChannel(sendError: StateError('synthetic send failure')); - final transcriber = DeepgramTranscriber( - apiKey: 'fake-key', - onTranscript: (_) {}, - channel: channel, - ); + final transcriber = DeepgramTranscriber(apiKey: 'fake-key', onTranscript: (_) {}, channel: channel); await transcriber.stop(); @@ -71,11 +65,7 @@ void main() { test('Deepgram still delivers transcripts until stop', () async { final channel = RecordingWebSocketChannel(); final transcripts = []; - DeepgramTranscriber( - apiKey: 'fake-key', - onTranscript: transcripts.add, - channel: channel, - ); + DeepgramTranscriber(apiKey: 'fake-key', onTranscript: transcripts.add, channel: channel); channel.addIncoming( jsonEncode({ @@ -102,11 +92,7 @@ void main() { }), ); final transcripts = []; - final transcriber = DeepgramTranscriber( - apiKey: 'fake-key', - onTranscript: transcripts.add, - channel: channel, - ); + final transcriber = DeepgramTranscriber(apiKey: 'fake-key', onTranscript: transcripts.add, channel: channel); await transcriber.stop(); @@ -131,11 +117,7 @@ void main() { test('Parakeet stop sends finalize before closing the socket', () async { final channel = RecordingWebSocketChannel(); - final transcriber = ParakeetTranscriber( - apiUrl: 'https://parakeet.example', - onTranscript: (_) {}, - channel: channel, - ); + final transcriber = ParakeetTranscriber(apiUrl: 'https://parakeet.example', onTranscript: (_) {}, channel: channel); await transcriber.stop(); @@ -146,11 +128,7 @@ void main() { test('Parakeet stop still closes when finalize send fails', () async { final channel = RecordingWebSocketChannel(sendError: StateError('synthetic send failure')); - final transcriber = ParakeetTranscriber( - apiUrl: 'https://parakeet.example', - onTranscript: (_) {}, - channel: channel, - ); + final transcriber = ParakeetTranscriber(apiUrl: 'https://parakeet.example', onTranscript: (_) {}, channel: channel); await transcriber.stop(); @@ -159,9 +137,7 @@ void main() { }); test('Parakeet stop drains a trailing transcript after finalize', () async { - final channel = RecordingWebSocketChannel( - trailingAfterCloseStream: jsonEncode({'text': 'final para'}), - ); + final channel = RecordingWebSocketChannel(trailingAfterCloseStream: jsonEncode({'text': 'final para'})); final transcripts = []; final transcriber = ParakeetTranscriber( apiUrl: 'https://parakeet.example', @@ -208,7 +184,47 @@ void main() { await transcriber.stop(); await stop; - expect(channel.sent, [Uint8List.fromList([1, 2]), 'finalize']); + expect(channel.sent, [ + Uint8List.fromList([1, 2]), + 'finalize', + ]); + }); + + test('Parakeet reports a stream the server closed before ready', () async { + final channel = RecordingWebSocketChannel(); + final transcriber = ParakeetTranscriber(apiUrl: 'https://parakeet.example', onTranscript: (_) {}, channel: channel); + final pcm = Uint8List.fromList([1, 0, 2, 0]); + + // While ready is pending, PCM is dropped without an error. + transcriber.appendPcm(pcm); + + // The server accepts the socket, then refuses it before ready (backend/parakeet/main.py). + channel.closeCode = 1013; + channel.closeReason = 'capacity_full'; + channel.closeIncoming(); + await Future.delayed(Duration.zero); + + final refused = isA() + .having((e) => e.closeCode, 'closeCode', 1013) + .having((e) => e.closeReason, 'closeReason', 'capacity_full'); + expect(() => transcriber.appendPcm(pcm), throwsA(refused)); + await expectLater(transcriber.stop(), throwsA(refused)); + expect(channel.sent.whereType(), isEmpty); + expect(channel.closed, isTrue); + }); + + test('Parakeet reports a stream that failed before ready', () async { + final channel = RecordingWebSocketChannel(); + final transcriber = ParakeetTranscriber(apiUrl: 'https://parakeet.example', onTranscript: (_) {}, channel: channel); + final failure = StateError('synthetic connection failure'); + + channel.addIncomingError(failure); + channel.closeIncoming(); + await Future.delayed(Duration.zero); + + final failed = isA().having((e) => e.error, 'error', same(failure)); + expect(() => transcriber.appendPcm(Uint8List.fromList([1, 0])), throwsA(failed)); + await expectLater(transcriber.stop(), throwsA(failed)); }); test('Deepgram rejects PCM and a second CloseStream while draining', () async { @@ -227,16 +243,15 @@ void main() { await transcriber.stop(); await stop; - expect(channel.sent, [Uint8List.fromList([1, 2]), jsonEncode({'type': 'CloseStream'})]); + expect(channel.sent, [ + Uint8List.fromList([1, 2]), + jsonEncode({'type': 'CloseStream'}), + ]); }); } class RecordingWebSocketChannel extends StreamChannelMixin implements WebSocketChannel { - RecordingWebSocketChannel({ - this.sendError, - this.trailingAfterCloseStream, - this.completeStreamOnCloseStream = true, - }); + RecordingWebSocketChannel({this.sendError, this.trailingAfterCloseStream, this.completeStreamOnCloseStream = true}); final Object? sendError; final Object? trailingAfterCloseStream; @@ -248,6 +263,8 @@ class RecordingWebSocketChannel extends StreamChannelMixin implements WebSocketC void addIncoming(Object? event) => _incoming.add(event); + void addIncomingError(Object error) => _incoming.addError(error); + void closeIncoming() { if (!_incoming.isClosed) { _incoming.close(); @@ -287,8 +304,7 @@ class _RecordingSink implements WebSocketSink { if (error != null) { throw error; } - if ((event == jsonEncode({'type': 'CloseStream'}) || event == 'finalize') && - _parent.completeStreamOnCloseStream) { + if ((event == jsonEncode({'type': 'CloseStream'}) || event == 'finalize') && _parent.completeStreamOnCloseStream) { scheduleMicrotask(() { final trailing = _parent.trailingAfterCloseStream; if (trailing != null) { From 51f305b55c8ad69b254dec174dccce3fc8929151 Mon Sep 17 00:00:00 2001 From: Apoorv Darshan Date: Mon, 19 Oct 2026 09:05:03 +0530 Subject: [PATCH 3/3] ci: run with ci:full Co-Authored-By: Claude Opus 5.5