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/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/.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 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) {