From 9488faf0087bafa1e7c53912b003d54a28fda279 Mon Sep 17 00:00:00 2001 From: pushkraj-space Date: Mon, 28 Sep 2026 21:06:48 +0530 Subject: [PATCH] feat(dart): add streaming transcription provider contract Define a provider-neutral boundary between Murmur audio and speech recognition, plus a deterministic fake provider for coordinator tests. It supplies the provider seam used by the capture coordinator (#36). Contract (lib/src/provider.dart, exported from murmur_protocol): - VoiceProvider exposes a stable id, fixed capabilities, and a synchronous, inert createSession(ProviderSessionRequest), so the caller subscribes before start and cannot miss startup events. An unsupported format throws VoiceError(unsupported_format). - ProviderSessionRequest carries only a session id and AudioFormat. It has no credentials, capture mode, device source, or metadata map. Providers stay mode-neutral, and credentials enter only through adapter config. - ProviderSession: events, start, addFrame, mute, unmute, finalize, stop. Sessions are one-shot and never reconnect; a retry is a new session. - Events: partials replace the open hypothesis, finals are append-only verbatim fragments whose concatenation is the utterance, and rejected segments are never assembled. Readiness, amplitude, and a terminal ProviderFailure complete the set. Text lives only on the event stream. - finalize is graceful: the gate closes synchronously, accepted audio is decoded, and the remaining finals are emitted. stop is a hard cancel that settles pending work with cancelled. A state matrix specifies every operation after stop, failure, or finalize. - Optional capabilities: immediateMute, warmUnmute (requires immediateMute), realAudioReadiness, amplitude, and rejectedFinals. Mute, unmute, and finalize are serialized and single-flight. - The caller owns pre-open buffering. At most one addFrame is outstanding. Adapter queues are bounded and documented. An empty queue always admits a valid frame, and an oversized frame throws. - Documented error codes: cancelled, unsupported_format, unauthorized, rate_limited (retryAfterMs), and provider_disconnected. Test support (test/support/fake_voice_provider.dart) is a manually driven fake with no timers. It enforces the contract by default and can delay or fail startup, emit partials, finals, and rejections, finalize late or empty, fail, disconnect, and ignore cancellation. The terminal state is recorded before listeners are notified, so reentrant callbacks see the ended session. Rewrite the VoiceProvider section of docs/voice-runtime.md and update the package README and CHANGELOG. No proto, fixture, dependency, or other-SDK changes. Refs #7 Co-Authored-By: Claude Opus 5.5 --- docs/voice-runtime.md | 41 +- sdks/dart/murmur_protocol/CHANGELOG.md | 1 + sdks/dart/murmur_protocol/README.md | 6 + .../murmur_protocol/lib/murmur_protocol.dart | 1 + .../murmur_protocol/lib/src/provider.dart | 315 +++++ .../murmur_protocol/test/provider_test.dart | 1069 +++++++++++++++++ .../test/support/fake_voice_provider.dart | 443 +++++++ 7 files changed, 1866 insertions(+), 10 deletions(-) create mode 100644 sdks/dart/murmur_protocol/lib/src/provider.dart create mode 100644 sdks/dart/murmur_protocol/test/provider_test.dart create mode 100644 sdks/dart/murmur_protocol/test/support/fake_voice_provider.dart diff --git a/docs/voice-runtime.md b/docs/voice-runtime.md index 0b61cb1..4a109a5 100644 --- a/docs/voice-runtime.md +++ b/docs/voice-runtime.md @@ -37,20 +37,41 @@ should preserve these concepts and pass the shared conformance suite: ### `VoiceProvider` -A live provider session starts with callbacks or an event stream and has an -idempotent hard stop. Optional capabilities are declared instead of assumed: - -- partial and final transcript delivery -- immediate input mute and warm unmute -- graceful finalization of a server-side decoder -- real-audio readiness and normalized amplitude +A provider creates one-shot recognition sessions. A session is created inert, +the caller subscribes to its event stream, and then starts it. Partial and final +transcripts, graceful finalization, an idempotent hard stop, and frame +backpressure are baseline behavior. Optional capabilities are declared instead +of assumed or simulated: + +- immediate input mute: the input gate closes at once while the session stays + warm and the already accepted tail is flushed +- warm unmute of the same session, which requires immediate mute +- real-audio readiness +- normalized amplitude - rejected-final reporting, for example when optional speaker verification rejects a segment -- provider close and typed failure events `stop` means cancel and release resources. `finalize` means close the input -gate, let already accepted audio finish decoding, return the final text, and -then release resources. They are deliberately different operations. +gate, let already accepted audio finish decoding, emit the remaining finals on +the event stream, and then release resources. They are deliberately different +operations. + +- Delivery order is the total order of a session. A partial is the complete, + replaceable hypothesis of the open segment. Finals are append-only, never + revised, and carry their own separators, so an utterance is the exact + concatenation of its finals. A rejected segment is never assembled. +- Mute, unmute, and finalize are serialized and single-flight, and none is + legal before start completes. Stop is always allowed and overtakes any + pending operation with `cancelled`. +- The caller owns pre-open buffering: a bounded pre-roll drained once start or + unmute completes. At most one frame write is outstanding, and each adapter + bounds and documents its internal queue. +- A startup failure, such as a rate limit, is reported by start. A later + failure ends the session with one typed failure event, and stream + completion means the session is closed. Sessions never reconnect; a retry is + a new session, and finals already emitted stay valid. +- Credentials enter only through adapter configuration and never appear in + session requests, events, errors, or logs. ### `VoiceCaptureCoordinator` diff --git a/sdks/dart/murmur_protocol/CHANGELOG.md b/sdks/dart/murmur_protocol/CHANGELOG.md index dea8294..b7acc0e 100644 --- a/sdks/dart/murmur_protocol/CHANGELOG.md +++ b/sdks/dart/murmur_protocol/CHANGELOG.md @@ -3,6 +3,7 @@ - Add typed audio formats and session commands with presence-preserving ProtoJSON serialization. - Add framework-neutral voice connector, session, state, and error interfaces. +- Add the provider-neutral streaming transcription contract. ## 0.1.0 diff --git a/sdks/dart/murmur_protocol/README.md b/sdks/dart/murmur_protocol/README.md index 55ee53c..8259c1b 100644 --- a/sdks/dart/murmur_protocol/README.md +++ b/sdks/dart/murmur_protocol/README.md @@ -15,6 +15,12 @@ Typed `AudioFormat`, `AudioFrame`, and `SessionControl` models preserve the `VoiceSession` interfaces let pure-Dart and Flutter hosts use the same explicit discovery, connection, capture, error, and cleanup lifecycle. +`VoiceProvider` and `ProviderSession` define the provider-neutral streaming +transcription boundary. Create a session, subscribe to its events, await +`start`, await each `addFrame`, concatenate finals verbatim, and end with +`finalize` or `stop`. The dartdoc on `ProviderSession` is the normative +contract. + The package source and pub.dev metadata are in place. It remains unpublished while the initial public protocol is reviewed against the shared fixtures in `conformance/`. diff --git a/sdks/dart/murmur_protocol/lib/murmur_protocol.dart b/sdks/dart/murmur_protocol/lib/murmur_protocol.dart index 04e30ca..3a30c79 100644 --- a/sdks/dart/murmur_protocol/lib/murmur_protocol.dart +++ b/sdks/dart/murmur_protocol/lib/murmur_protocol.dart @@ -2,6 +2,7 @@ library; export 'src/events.dart'; export 'src/protocol.dart'; +export 'src/provider.dart'; export 'src/runtime.dart'; export 'src/session.dart'; export 'src/source.dart'; diff --git a/sdks/dart/murmur_protocol/lib/src/provider.dart b/sdks/dart/murmur_protocol/lib/src/provider.dart new file mode 100644 index 0000000..dfc99c0 --- /dev/null +++ b/sdks/dart/murmur_protocol/lib/src/provider.dart @@ -0,0 +1,315 @@ +import 'runtime.dart'; +import 'session.dart'; + +/// An optional behavior a [VoiceProvider] declares instead of simulating. +/// +/// Partial and final transcripts, [ProviderSession.finalize], +/// [ProviderSession.stop], and [ProviderSession.addFrame] backpressure are +/// baseline behavior, not capabilities. An adapter that produces no +/// hypotheses simply emits no [ProviderPartial] events. +enum ProviderCapability { + /// [ProviderSession.mute] closes the input gate synchronously and flushes the + /// already accepted tail while the session stays warm. + immediateMute, + + /// [ProviderSession.unmute] reopens a muted session without reconnecting. + /// + /// This capability is valid only together with [immediateMute]. + warmUnmute, + + /// The session may emit [ProviderReadiness] when real audio is live. + realAudioReadiness, + + /// The session may emit normalized [ProviderAmplitude] levels. + amplitude, + + /// The session may emit [ProviderRejected], for example when optional + /// speaker verification rejects a segment. + rejectedFinals, +} + +/// Provider-neutral metadata for one [ProviderSession]. +/// +/// The request carries no credentials, device identifiers, or capture mode. +/// Providers are mode-neutral, and vendor tuning and credentials belong in the +/// concrete adapter's configuration. +final class ProviderSessionRequest { + /// Creates a session request for audio in [format]. + /// + /// A blank [sessionId] throws [ArgumentError]. + ProviderSessionRequest({required this.sessionId, required this.format}) { + if (sessionId.trim().isEmpty) { + throw ArgumentError.value(sessionId, 'sessionId', 'must not be empty'); + } + } + + /// The caller's identifier for this recognition session. + /// + /// It correlates diagnostics and need not equal [AudioFrame.sessionId], + /// because a connector session may span several recognition sessions. + final String sessionId; + + /// The format of every frame passed to [ProviderSession.addFrame]. + final AudioFormat format; +} + +/// One recognition-side event from a [ProviderSession]. +/// +/// Delivery order on [ProviderSession.events] is the total order of a session, +/// so events carry no session identifier or sequence number. +sealed class ProviderEvent { + const ProviderEvent(); +} + +/// The complete, replaceable hypothesis of the open segment. +/// +/// The open segment is the text since the last [ProviderFinal] or +/// [ProviderRejected]. Each partial replaces the previous one; it is not a +/// delta, and adapters accumulate vendor deltas before emitting. An empty +/// [text] clears the displayed hypothesis. +final class ProviderPartial extends ProviderEvent { + /// Creates a partial hypothesis. + const ProviderPartial(this.text); + + /// The full hypothesis of the open segment. + final String text; +} + +/// A committed transcript fragment that closes the open segment. +/// +/// Finals are append-only, in order, and never revised. An adapter whose vendor +/// rewrites committed text holds that text as a partial until it is stable. +/// [text] is a verbatim fragment that includes any separator it needs, so an +/// utterance is the exact concatenation of its finals. +final class ProviderFinal extends ProviderEvent { + /// Creates a final fragment. + /// + /// An empty [text] throws [ArgumentError]; an operation that recognized + /// nothing completes with zero finals instead. + ProviderFinal(this.text) { + if (text.isEmpty) { + throw ArgumentError.value(text, 'text', 'must not be empty'); + } + } + + /// The verbatim fragment, including any leading separator. + final String text; +} + +/// A committed segment the provider refused. +/// +/// It closes and discards the open segment. Rejected text may be displayed but +/// never enters transcript assembly or dispatch. Emitted only with +/// [ProviderCapability.rejectedFinals]. +final class ProviderRejected extends ProviderEvent { + /// Creates a rejected segment. + /// + /// An empty [text] throws [ArgumentError]. + ProviderRejected(this.text) { + if (text.isEmpty) { + throw ArgumentError.value(text, 'text', 'must not be empty'); + } + } + + /// The rejected text, for display only. + final String text; +} + +/// Whether the provider currently receives real audio. +/// +/// Emitted only with [ProviderCapability.realAudioReadiness]. +final class ProviderReadiness extends ProviderEvent { + /// Creates a readiness event. + const ProviderReadiness({required this.live}); + + /// Whether real audio is live. + final bool live; +} + +/// A normalized input level. +/// +/// Emitted only with [ProviderCapability.amplitude]. +final class ProviderAmplitude extends ProviderEvent { + /// Creates an amplitude event. + /// + /// A [value] outside 0 to 1 inclusive, or NaN, throws [ArgumentError]. + ProviderAmplitude(this.value) { + if (value.isNaN || value < 0 || value > 1) { + throw ArgumentError.value(value, 'value', 'must be between 0 and 1'); + } + } + + /// The level between 0 and 1 inclusive. + final double value; +} + +/// A terminal failure after [ProviderSession.start] completed. +/// +/// It is the last event; the stream is done immediately afterwards. A startup +/// failure is reported only through [ProviderSession.start]. +final class ProviderFailure extends ProviderEvent { + /// Creates a failure event. + const ProviderFailure(this.error); + + /// The cause, which every pending operation also completes with. + final VoiceError error; +} + +/// A streaming transcription adapter that creates [ProviderSession]s. +/// +/// Credentials enter only through the concrete adapter's constructor or factory +/// configuration. No contract type carries a credential, and an adapter must +/// never place credentials, signed URLs, transcript text, audio, or device +/// identifiers in a [VoiceError] message or metadata. +/// +/// The provider object may share safe provider-level resources, such as an +/// authenticated client, a model handle, or rate-limit state, across its +/// sessions. Decoder and transcript state never cross sessions. +abstract interface class VoiceProvider { + /// A stable identifier used for transcript attribution and diagnostics. + String get id; + + /// The stable, unmodifiable set of optional behaviors this provider + /// implements. + Set get capabilities; + + /// Creates an inert session for [request]. + /// + /// This is synchronous, performs no I/O, and emits nothing, so the caller can + /// subscribe to [ProviderSession.events] before calling + /// [ProviderSession.start]. If the adapter cannot decode `request.format`, it + /// throws a [VoiceError] with code `unsupported_format` that is not + /// retryable. + ProviderSession createSession(ProviderSessionRequest request); +} + +/// One one-shot recognition session. +/// +/// ## Lifecycle +/// +/// Create the session, subscribe to [events], await [start], await each +/// [addFrame], then end with [finalize] (graceful) or [stop] (hard cancel). +/// A session never reconnects or restarts; a retry is a new session from +/// [VoiceProvider.createSession], and a failed session is never resumed. +/// +/// ## Transcript assembly +/// +/// A [ProviderPartial] replaces the open segment's hypothesis. A +/// [ProviderFinal] commits it; finals are never revised, and the utterance is +/// the exact concatenation of its finals. A [ProviderRejected] commits and +/// discards the open segment. Text arrives only on [events]: an operation's +/// finals are added before its future completes, and stream done is +/// authoritative. +/// +/// ## Operations by state +/// +/// Operations on the open, started session follow the member docs. Otherwise: +/// +/// | State | [start] | [addFrame] | [mute] / [unmute] | [finalize] | [stop] | +/// | --- | --- | --- | --- | --- | --- | +/// | not started or start pending | starts, or shares the pending start | [StateError] | [StateError] | [StateError] | cancels | +/// | [finalize] pending | [StateError] | discarded | [StateError] | same future | wins with `cancelled` | +/// | ended by [finalize] | [StateError] | discarded | [StateError] | same future | completes | +/// | ended by [stop] | [StateError] | `cancelled` | `cancelled` | `cancelled` | same future | +/// | ended by failure | [StateError] | that error | that error | that error | completes | +/// +/// "Ended by [stop]" and "ended by failure" include a [finalize] that was +/// pending when the session ended: the ending cause decides every outcome. +/// Without its capability, [mute] or [unmute] throws [UnsupportedError] in +/// every state. +/// +/// ## Buffering and backpressure +/// +/// The caller owns pre-open buffering: it keeps a bounded pre-roll and drains +/// it in order once [start] or [unmute] completes. At most one [addFrame] is +/// outstanding. Each adapter bounds its internal queue, documents that bound +/// and any maximum frame size, and keeps a write pending while the queue is +/// full instead of dropping or growing it. An empty queue always admits one +/// valid frame, and a frame over the documented maximum size is rejected +/// synchronously with [ArgumentError]. +/// +/// ## Failures +/// +/// Documented [VoiceError.code] values are `cancelled`, `unsupported_format` +/// (not retryable), `unauthorized` (not retryable), `rate_limited` (retryable, +/// with an optional decimal-string `metadata['retryAfterMs']`), and +/// `provider_disconnected` (retryable). Other codes are adapter-defined. An +/// unexpected transport close is a [ProviderFailure] with +/// `provider_disconnected`. Finals already emitted stay valid; accepted audio +/// that was not yet decoded is lost. +abstract interface class ProviderSession { + /// Transcript, readiness, amplitude, and failure events in delivery order. + /// + /// The stream is single-subscription, and done means the session is closed. + /// After [stop] the adapter adds no new events, but events already queued + /// for a paused listener may still arrive on resume, so consumers must fence + /// callbacks from a superseded session. Closing never waits for an absent or + /// paused listener. + Stream get events; + + /// Opens the session and completes when frames can be accepted. + /// + /// Calls made while the first start is pending share its future. Once startup + /// has settled or the session has ended, calling this throws [StateError]. + /// A startup failure, such as `rate_limited` or `unauthorized`, is reported + /// only through this future; the adapter releases what it acquired and + /// closes [events]. If [stop] wins against a pending start, this completes + /// with a [VoiceError] whose code is `cancelled`. + Future start(); + + /// Offers one frame and completes when the adapter's bounded queue accepts + /// it. + /// + /// Awaiting the future is the backpressure signal. Calling this before + /// [start] completes, while [unmute] is pending, or while another frame is + /// outstanding throws [StateError]. A frame over the adapter's documented + /// maximum size throws [ArgumentError]. An empty queue always accepts a valid + /// frame, so a write that finds it empty never stays pending. Frames must + /// match the request format and have strictly increasing + /// [AudioFrame.sequence]; an adapter may reject a violation with + /// [ArgumentError]. A frame offered + /// while muted, while [finalize] is pending, or after a successful + /// [finalize] is discarded and completes normally. [stop] completes a frame + /// that was not yet accepted with `cancelled`, and a failure completes it + /// with that failure's error. + Future addFrame(AudioFrame frame); + + /// Closes the input gate synchronously and flushes the accepted tail. + /// + /// Requires [ProviderCapability.immediateMute]; otherwise this throws + /// [UnsupportedError]. Audio accepted before the call, including a pending + /// [addFrame], is decoded and its finals are emitted before the future + /// completes. The session stays warm. Muting an already muted session + /// completes immediately. + /// + /// At most one of [mute], [unmute], and [finalize] is in flight: repeating + /// the pending call shares its future, and a different one throws + /// [StateError]. + Future mute(); + + /// Reopens a muted session without reconnecting. + /// + /// Requires [ProviderCapability.warmUnmute]; otherwise this throws + /// [UnsupportedError]. Completes once frames are accepted again. Unmuting an + /// open session completes immediately. Transitions are serialized as + /// described on [mute]. + Future unmute(); + + /// Gracefully finishes the session. + /// + /// The input gate closes synchronously. Audio accepted before the call is + /// decoded, the remaining finals are emitted, resources are released, and + /// [events] closes. Zero finals is a valid outcome. Repeated and concurrent + /// calls return the identical future, which completes with `cancelled` if + /// [stop] wins or with the failure's error if the provider fails. Calling + /// this before [start] completes throws [StateError]. + Future finalize(); + + /// Cancels recognition and releases resources. + /// + /// The input gate closes synchronously, and unaccepted audio and pending + /// hypotheses are discarded. Every unfinished operation completes with + /// `cancelled`. Concurrent and repeated calls share one future, which never + /// completes with an error. + Future stop(); +} diff --git a/sdks/dart/murmur_protocol/test/provider_test.dart b/sdks/dart/murmur_protocol/test/provider_test.dart new file mode 100644 index 0000000..28c74a9 --- /dev/null +++ b/sdks/dart/murmur_protocol/test/provider_test.dart @@ -0,0 +1,1069 @@ +import 'dart:async'; + +import 'package:murmur_protocol/murmur_protocol.dart'; +import 'package:test/test.dart'; + +import 'support/fake_voice_provider.dart'; + +void main() { + group('provider values', () { + test('rejects a blank session id', () { + for (final id in ['', ' ']) { + expect( + () => ProviderSessionRequest(sessionId: id, format: _format()), + throwsArgumentError, + ); + } + }); + + test('rejects empty final and rejected text', () { + expect(() => ProviderFinal(''), throwsArgumentError); + expect(() => ProviderRejected(''), throwsArgumentError); + expect(ProviderFinal(' hi').text, ' hi'); + expect(const ProviderPartial('').text, isEmpty); + }); + + test('bounds amplitude to 0..1 and rejects NaN', () { + expect(ProviderAmplitude(0).value, 0); + expect(ProviderAmplitude(1).value, 1); + for (final value in [-0.01, 1.01, double.nan, double.infinity]) { + expect(() => ProviderAmplitude(value), throwsArgumentError); + } + }); + + test('exposes unmodifiable capabilities and sessions', () { + final capabilities = {ProviderCapability.amplitude}; + final provider = FakeVoiceProvider(capabilities: capabilities); + capabilities.add(ProviderCapability.immediateMute); + + expect(provider.capabilities, {ProviderCapability.amplitude}); + expect( + () => provider.capabilities.add(ProviderCapability.rejectedFinals), + throwsUnsupportedError, + ); + provider.createSession(_request()); + expect(provider.sessions, hasLength(1)); + expect(provider.sessions.clear, throwsUnsupportedError); + }); + }); + + group('request, format, and startup', () { + test('carries every documented audio encoding', () { + final provider = FakeVoiceProvider(); + for (final encoding in AudioEncoding.values) { + final session = provider.createSession( + _request(format: _format(encoding: encoding)), + ); + expect(session.request.format.encoding, encoding); + expect(session.request.sessionId, 'capture-1'); + } + }); + + test('rejects an unsupported format synchronously', () { + final provider = FakeVoiceProvider( + supportedEncodings: {AudioEncoding.pcmS16le}, + ); + expect( + () => provider.createSession( + _request(format: _format(encoding: AudioEncoding.opus)), + ), + throwsA(_voiceError('unsupported_format', retryable: false)), + ); + expect(provider.sessions, isEmpty); + }); + + test('creates an inert session that emits nothing', () async { + final run = _Run(FakeVoiceProvider().createSession(_request())); + await pumpEventQueue(); + + expect(run.events, isEmpty); + expect(run.done, isFalse); + expect(run.session.startCalls, 0); + }); + + test('a held start stays pending and concurrent starts share it', () async { + final run = _Run( + FakeVoiceProvider( + holdStart: true, + capabilities: {ProviderCapability.realAudioReadiness}, + ).createSession(_request()), + ); + var started = false; + final first = run.session.start()..then((_) => started = true); + final second = run.session.start(); + await pumpEventQueue(); + + expect(identical(first, second), isTrue); + expect(run.session.startPending, isTrue); + expect(started, isFalse); + + run.session.emitReadiness(live: false); + run.session.completeStart(); + await first; + expect(started, isTrue); + expect(run.session.startPending, isFalse); + expect(run.session.startCalls, 2); + expect(run.events.single, isA()); + }); + + test('a startup failure surfaces its exact error and closes', () async { + final run = _Run( + FakeVoiceProvider(holdStart: true).createSession(_request()), + ); + final start = run.session.start(); + final rateLimited = VoiceError( + code: 'rate_limited', + message: 'Too many sessions.', + retryable: true, + metadata: {'retryAfterMs': '1500'}, + ); + run.session.failStart(rateLimited); + + await expectLater(start, throwsA(same(rateLimited))); + await pumpEventQueue(); + expect(run.events, isEmpty); + expect(run.done, isTrue); + expect(run.session.isClosed, isTrue); + }); + + test('stop during a pending start cancels it', () async { + final run = _Run( + FakeVoiceProvider(holdStart: true).createSession(_request()), + ); + final start = run.session.start(); + final startExpectation = expectLater( + start, + throwsA(_voiceError('cancelled')), + ); + await run.session.stop(); + await startExpectation; + await pumpEventQueue(); + expect(run.done, isTrue); + // A late acquisition after cancellation is ignored. + run.session.completeStart(); + }); + + test( + 'disconnect during a pending start fails it without an event', + () async { + final run = _Run( + FakeVoiceProvider(holdStart: true).createSession(_request()), + ); + final start = run.session.start(); + run.session.disconnect(); + + await expectLater(start, throwsA(_voiceError('provider_disconnected'))); + await pumpEventQueue(); + expect(run.events, isEmpty); + expect(run.done, isTrue); + }, + ); + + test('start is one-shot after every settled outcome', () async { + final provider = FakeVoiceProvider(holdStart: true); + + final succeeded = await _started(provider); + expect(succeeded.session.start, throwsStateError); + expect(succeeded.session.startCalls, 2); + + final failed = _Run(provider.createSession(_request())); + final failedStart = failed.session.start(); + failed.session.failStart(_failure('unauthorized', retryable: false)); + await expectLater(failedStart, throwsA(_voiceError('unauthorized'))); + expect(failed.session.start, throwsStateError); + + final cancelled = _Run(provider.createSession(_request())); + final cancelledStart = cancelled.session.start(); + final cancelledStartExpectation = expectLater( + cancelledStart, + throwsA(_voiceError('cancelled')), + ); + await cancelled.session.stop(); + await cancelledStartExpectation; + expect(cancelled.session.start, throwsStateError); + + final finalized = await _started(provider); + final finalize = finalized.session.finalize(); + finalized.session.completeFinalize(); + await finalize; + expect(finalized.session.start, throwsStateError); + }); + + test('stop before start takes precedence over pre-start rules', () async { + final run = _Run(FakeVoiceProvider().createSession(_request())); + await run.session.stop(); + + await expectLater( + run.session.finalize(), + throwsA(_voiceError('cancelled')), + ); + await expectLater( + run.session.addFrame(_frame(1)), + throwsA(_voiceError('cancelled')), + ); + expect(run.session.start, throwsStateError); + run.session.emitPartial('late'); + await pumpEventQueue(); + expect(run.events, isEmpty); + expect(run.done, isTrue); + }); + + test('a live session rejects driver events before start', () { + final session = FakeVoiceProvider().createSession(_request()); + + expect(() => session.emitPartial('early'), throwsStateError); + expect(() => session.fail(_failure('boom')), throwsStateError); + expect(session.disconnect, throwsStateError); + }); + + test('rejects frames before start completes', () async { + final session = FakeVoiceProvider( + holdStart: true, + ).createSession(_request()); + expect(() => session.addFrame(_frame(1)), throwsStateError); + + final start = session.start(); + expect(() => session.addFrame(_frame(1)), throwsStateError); + session.completeStart(); + await start; + await session.addFrame(_frame(1)); + expect(session.frames, hasLength(1)); + }); + }); + + group('partials', () { + test('each partial replaces the open hypothesis in order', () async { + final run = await _started(FakeVoiceProvider()); + run.session + ..emitPartial('he') + ..emitPartial('hello') + ..emitPartial('') + ..emitFinal('Hello.') + ..emitPartial(' wor'); + + expect(run.partials, ['he', 'hello', '', ' wor']); + expect(run.openHypothesis, ' wor'); + expect(run.transcript, 'Hello.'); + }); + + test('a final clears the open hypothesis', () async { + final run = await _started(FakeVoiceProvider()); + run.session + ..emitPartial('hel') + ..emitFinal('Hello'); + + expect(run.openHypothesis, isEmpty); + }); + }); + + group('multi-final assembly', () { + test('concatenates finals verbatim across partials and finalize', () async { + final run = await _started(FakeVoiceProvider()); + run.session + ..emitPartial('Turn') + ..emitFinal('Turn on') + ..emitPartial(' the') + ..emitFinal(' the lights.'); + final finalize = run.session.finalize(); + run.session + ..emitPartial('你') + ..emitFinal('你好') + ..emitFinal('。') + ..completeFinalize(); + await finalize; + + expect(run.transcript, 'Turn on the lights.你好。'); + }); + + test('every final arrives before finalize completes and done', () async { + final run = await _started(FakeVoiceProvider()); + final finalize = run.session.finalize() + ..then((_) => run.order.add('finalized')); + run.session + ..emitFinal('one') + ..emitFinal(' two') + ..completeFinalize(); + await finalize; + await pumpEventQueue(); + + expect(run.order, ['final', 'final', 'done', 'finalized']); + }); + }); + + group('finalization', () { + test('discards frames offered while and after finalizing', () async { + final run = await _started(FakeVoiceProvider()); + await run.session.addFrame(_frame(1)); + final finalize = run.session.finalize(); + + await run.session.addFrame(_frame(2)); + run.session.completeFinalize(); + await finalize; + await run.session.addFrame(_frame(3)); + + expect(run.session.frames.map((frame) => frame.sequence), [BigInt.one]); + }); + + test('a finalized session rejects gate changes and allows stop', () async { + final run = await _started( + FakeVoiceProvider( + capabilities: { + ProviderCapability.immediateMute, + ProviderCapability.warmUnmute, + }, + ), + ); + final finalize = run.session.finalize(); + expect(run.session.mute, throwsStateError); + expect(run.session.unmute, throwsStateError); + run.session.completeFinalize(); + await finalize; + + expect(run.session.mute, throwsStateError); + expect(run.session.unmute, throwsStateError); + expect(run.session.start, throwsStateError); + await run.session.stop(); + await expectLater(run.session.finalize(), completes); + }); + + test('a held frame belongs to the finalized tail', () async { + final run = await _started(FakeVoiceProvider()); + run.session.holdFrames = true; + final tail = run.session.addFrame(_frame(1)); + final finalize = run.session.finalize(); + + expect(run.session.completeFinalize, throwsStateError); + run.session.releaseFrame(); + await tail; + run.session + ..emitFinal('tail') + ..completeFinalize(); + await finalize; + + expect(run.session.frames, hasLength(1)); + expect(run.transcript, 'tail'); + }); + + test('a late finalize stays pending until completed', () async { + final run = await _started(FakeVoiceProvider()); + var finalized = false; + final finalize = run.session.finalize()..then((_) => finalized = true); + await pumpEventQueue(); + + expect(run.session.finalizePending, isTrue); + expect(finalized, isFalse); + expect(run.done, isFalse); + + run.session.completeFinalize(); + await finalize; + expect(finalized, isTrue); + }); + + test('an empty finalize completes with zero finals', () async { + final run = await _started(FakeVoiceProvider()); + final finalize = run.session.finalize(); + run.session.completeFinalize(); + await finalize; + await pumpEventQueue(); + + expect(run.events, isEmpty); + expect(run.transcript, isEmpty); + expect(run.done, isTrue); + }); + + test('finalize before start completes throws', () async { + final session = FakeVoiceProvider( + holdStart: true, + ).createSession(_request()); + expect(session.finalize, throwsStateError); + final start = session.start(); + expect(session.finalize, throwsStateError); + session.completeStart(); + await start; + }); + }); + + group('duplicate finalize', () { + test('replays one terminal future after stop and failures', () async { + final stopped = await _started(FakeVoiceProvider()); + await stopped.session.stop(); + + final failed = await _started(FakeVoiceProvider()); + failed.session.fail(_failure('decoder_crashed')); + + final startFailed = FakeVoiceProvider( + holdStart: true, + ).createSession(_request()); + final start = startFailed.start(); + startFailed.failStart(_failure('unauthorized', retryable: false)); + await expectLater(start, throwsA(_voiceError('unauthorized'))); + + for (final (session, code) in [ + (stopped.session, 'cancelled'), + (failed.session, 'decoder_crashed'), + (startFailed, 'unauthorized'), + ]) { + final first = session.finalize(); + final second = session.finalize(); + expect(identical(first, second), isTrue); + await expectLater(first, throwsA(_voiceError(code))); + await expectLater(second, throwsA(_voiceError(code))); + } + }); + + test('concurrent and repeated calls share one future', () async { + final run = await _started(FakeVoiceProvider()); + final first = run.session.finalize(); + final second = run.session.finalize(); + expect(identical(first, second), isTrue); + + run.session.completeFinalize(); + await first; + final replay = run.session.finalize(); + expect(identical(first, replay), isTrue); + await replay; + expect(run.session.finalizeCalls, 3); + }); + }); + + group('cancellation', () { + test('stop is idempotent, shared, and closes events', () async { + final run = await _started(FakeVoiceProvider()); + final first = run.session.stop(); + final second = run.session.stop(); + expect(identical(first, second), isTrue); + await first; + await pumpEventQueue(); + + expect(run.done, isTrue); + expect(run.session.isClosed, isTrue); + expect(run.session.stopCalls, 2); + }); + + test('stop settles a held frame with cancelled', () async { + final run = await _started(FakeVoiceProvider()); + run.session.holdFrames = true; + final frame = run.session.addFrame(_frame(1)); + expect(run.session.heldFrames, 1); + final frameExpectation = expectLater( + frame, + throwsA(_voiceError('cancelled')), + ); + await run.session.stop(); + await frameExpectation; + expect(run.session.frames, isEmpty); + expect(run.session.heldFrames, 0); + }); + + test('stop settles pending mute, unmute, and finalize', () async { + final provider = _warmProvider(); + final muting = await _started(provider); + final mute = muting.session.mute(); + final muteExpectation = expectLater( + mute, + throwsA(_voiceError('cancelled')), + ); + await muting.session.stop(); + await muteExpectation; + + final unmuting = await _started(provider); + final muted = unmuting.session.mute(); + unmuting.session.completeMute(); + await muted; + final unmute = unmuting.session.unmute(); + final unmuteExpectation = expectLater( + unmute, + throwsA(_voiceError('cancelled')), + ); + await unmuting.session.stop(); + await unmuteExpectation; + + final finalizing = await _started(provider); + final finalize = finalizing.session.finalize(); + final finalizeExpectation = expectLater( + finalize, + throwsA(_voiceError('cancelled')), + ); + await finalizing.session.stop(); + await finalizeExpectation; + }); + + test('operations after stop complete with cancelled', () async { + final run = await _started(_warmProvider()); + await run.session.stop(); + + for (final operation in Function()>[ + () => run.session.addFrame(_frame(1)), + run.session.mute, + run.session.unmute, + run.session.finalize, + ]) { + await expectLater(operation(), throwsA(_voiceError('cancelled'))); + } + expect(run.session.start, throwsStateError); + }); + + test('the ending cause decides after stop overtakes finalize', () async { + final run = await _started(_warmProvider()); + final finalize = run.session.finalize(); + final stop = run.session.stop(); + + await expectLater(finalize, throwsA(_voiceError('cancelled'))); + await expectLater(stop, completes); + await expectLater(run.session.mute(), throwsA(_voiceError('cancelled'))); + expect(identical(run.session.finalize(), finalize), isTrue); + }); + + test('a stop reentered from onDone shares the future', () async { + final session = FakeVoiceProvider().createSession(_request()); + Future? reentered; + session.events.listen(null, onDone: () => reentered = session.stop()); + await session.start(); + + final stop = session.stop(); + expect(reentered, isNotNull); + expect(identical(stop, reentered), isTrue); + await stop; + expect(session.stopCalls, 2); + }); + + test('cleanup completes without a listener', () async { + final session = FakeVoiceProvider().createSession(_request()); + await session.start(); + await session.stop(); + + expect(session.isClosed, isTrue); + }); + + test('cleanup completes with a paused listener', () async { + final run = await _started(FakeVoiceProvider()); + run.subscription.pause(); + await run.session.stop(); + + expect(run.session.isClosed, isTrue); + expect(run.done, isFalse); + run.subscription.resume(); + await pumpEventQueue(); + expect(run.done, isTrue); + }); + }); + + group('late events', () { + test('driver events after termination add nothing', () async { + final provider = FakeVoiceProvider( + capabilities: { + ProviderCapability.amplitude, + ProviderCapability.realAudioReadiness, + ProviderCapability.rejectedFinals, + }, + ); + + final stopped = await _started(provider); + await stopped.session.stop(); + + final finalized = await _started(provider); + final finalize = finalized.session.finalize(); + finalized.session.completeFinalize(); + await finalize; + + final failed = await _started(provider); + failed.session.fail(_failure('boom')); + + for (final run in [stopped, finalized, failed]) { + final before = run.events.length; + run.session + ..emitPartial('late') + ..emitFinal('late') + ..rejectFinal('late') + ..emitReadiness(live: true) + ..emitAmplitude(0.5) + ..emitFinal('') + ..rejectFinal('') + ..emitAmplitude(2) + ..fail(_failure('again')) + ..disconnect(); + await pumpEventQueue(); + expect(run.events, hasLength(before)); + expect(run.done, isTrue); + } + }); + + test('an undeclared late event still throws', () async { + final run = await _started(FakeVoiceProvider()); + await run.session.stop(); + + expect(() => run.session.emitAmplitude(0.5), throwsStateError); + expect(() => run.session.rejectFinal(''), throwsStateError); + }); + + test('stop cannot revoke events queued for a paused listener', () async { + final run = await _started(FakeVoiceProvider()); + run.subscription.pause(); + run.session.emitPartial('queued'); + await run.session.stop(); + run.session.emitFinal('after stop'); + run.subscription.resume(); + await pumpEventQueue(); + + expect(run.partials, ['queued']); + expect(run.transcript, isEmpty); + expect(run.order, ['partial', 'done']); + }); + + test('ignoreCancellation keeps delivering after stop', () async { + final run = await _started(FakeVoiceProvider(ignoreCancellation: true)); + run.session.holdFrames = true; + var frameSettled = false; + var finalizeSettled = false; + unawaited( + run.session.addFrame(_frame(1)).whenComplete(() => frameSettled = true), + ); + unawaited( + run.session.finalize().whenComplete(() => finalizeSettled = true), + ); + + await run.session.stop(); + run.session.emitFinal('still here'); + await pumpEventQueue(); + + expect(run.session.stopCalls, 1); + expect(run.transcript, 'still here'); + expect(run.done, isFalse); + expect(run.session.isClosed, isFalse); + expect(frameSettled, isFalse); + expect(finalizeSettled, isFalse); + }); + }); + + group('failure and disconnect', () { + test('a failure emits exactly one event and then done', () async { + final run = await _started(FakeVoiceProvider()); + final failure = _failure('decoder_crashed'); + run.session.fail(failure); + await pumpEventQueue(); + + expect(run.events, hasLength(1)); + expect((run.events.single as ProviderFailure).error, same(failure)); + expect(run.order, ['failure', 'done']); + }); + + test('a listener reentering on failure sees the failed session', () async { + final session = _warmProvider().createSession(_request()); + final failure = _failure('decoder_crashed'); + final order = []; + final reentered = >[]; + final subscription = session.events.listen((event) { + order.add('failure'); + reentered + ..add(session.addFrame(_frame(1))) + ..add(session.stop()) + ..add(session.finalize()); + }, onDone: () => order.add('done')); + addTearDown(subscription.cancel); + await session.start(); + final mute = session.mute(); + + session.fail(failure); + + expect(order, ['failure', 'done']); + await expectLater(mute, throwsA(same(failure))); + await expectLater(reentered[0], throwsA(same(failure))); + await expectLater(reentered[1], completes); + await expectLater(reentered[2], throwsA(same(failure))); + expect(session.frames, isEmpty); + await expectLater(session.finalize(), throwsA(same(failure))); + await expectLater(session.unmute(), throwsA(same(failure))); + }); + + test('pending operations complete with the failure error', () async { + final run = await _started(_warmProvider()); + run.session.holdFrames = true; + final frame = run.session.addFrame(_frame(1)); + final mute = run.session.mute(); + final failure = _failure('decoder_crashed'); + run.session.fail(failure); + + await expectLater(frame, throwsA(same(failure))); + await expectLater(mute, throwsA(same(failure))); + + final finalizing = await _started(FakeVoiceProvider()); + final finalize = finalizing.session.finalize(); + finalizing.session.fail(failure); + await expectLater(finalize, throwsA(same(failure))); + }); + + test('operations after a failure complete with that error', () async { + final run = await _started(_warmProvider()); + final failure = _failure('decoder_crashed'); + run.session.fail(failure); + + for (final operation in Function()>[ + () => run.session.addFrame(_frame(1)), + run.session.mute, + run.session.unmute, + run.session.finalize, + ]) { + await expectLater(operation(), throwsA(same(failure))); + } + expect(run.session.start, throwsStateError); + await expectLater(run.session.stop(), completes); + }); + + test('operations after a startup failure complete with it', () async { + final session = _warmProvider(holdStart: true).createSession(_request()); + final start = session.start(); + final failure = _failure('unauthorized', retryable: false); + session.failStart(failure); + await expectLater(start, throwsA(same(failure))); + + for (final operation in Function()>[ + () => session.addFrame(_frame(1)), + session.mute, + session.unmute, + session.finalize, + ]) { + await expectLater(operation(), throwsA(same(failure))); + } + await expectLater(session.stop(), completes); + }); + + test('a mid-session rate limit keeps its metadata', () async { + final run = await _started(FakeVoiceProvider()); + run.session.fail( + VoiceError( + code: 'rate_limited', + message: 'Slow down.', + retryable: true, + metadata: {'retryAfterMs': '2000'}, + ), + ); + + final error = (run.events.single as ProviderFailure).error; + expect(error.code, 'rate_limited'); + expect(error.retryable, isTrue); + expect(error.metadata, {'retryAfterMs': '2000'}); + }); + + test('disconnect is a retryable provider_disconnected failure', () async { + final run = await _started(FakeVoiceProvider()); + run.session.disconnect(); + await pumpEventQueue(); + + expect( + (run.events.single as ProviderFailure).error, + _voiceError('provider_disconnected', retryable: true), + ); + expect(run.done, isTrue); + }); + }); + + group('capabilities and the input gate', () { + test('uncapable mute and unmute are unsupported', () async { + final run = await _started(FakeVoiceProvider()); + expect(run.session.mute, throwsUnsupportedError); + expect(run.session.unmute, throwsUnsupportedError); + + final muteOnly = await _started( + FakeVoiceProvider(capabilities: {ProviderCapability.immediateMute}), + ); + expect(muteOnly.session.unmute, throwsUnsupportedError); + await muteOnly.session.stop(); + expect(muteOnly.session.unmute, throwsUnsupportedError); + }); + + test('mute closes the gate and flushes the held tail', () async { + final run = await _started(_warmProvider()); + run.session.holdFrames = true; + final tail = run.session.addFrame(_frame(1)); + var flushed = false; + final mute = run.session.mute()..then((_) => flushed = true); + + expect(run.session.muted, isTrue); + expect(run.session.completeMute, throwsStateError); + run.session.releaseFrame(); + await tail; + run.session.emitFinal('tail'); + await pumpEventQueue(); + expect(flushed, isFalse); + + run.session.completeMute(); + await mute; + expect(run.transcript, 'tail'); + expect(run.session.frames, hasLength(1)); + + run.session.holdFrames = false; + await run.session.addFrame(_frame(2)); + expect(run.session.frames, hasLength(1)); + expect(run.done, isFalse); + }); + + test( + 'a repeated mute shares the flush and a muted mute completes', + () async { + final run = await _started(_warmProvider()); + final first = run.session.mute(); + expect(identical(first, run.session.mute()), isTrue); + run.session.completeMute(); + await first; + await expectLater(run.session.mute(), completes); + expect(run.session.mutePending, isFalse); + }, + ); + + test('an open unmute completes immediately', () async { + final run = await _started(_warmProvider()); + await expectLater(run.session.unmute(), completes); + expect(run.session.unmutePending, isFalse); + }); + + test('conflicting gate transitions throw while one is pending', () async { + final run = await _started(_warmProvider()); + final mute = run.session.mute(); + expect(run.session.unmute, throwsStateError); + expect(run.session.finalize, throwsStateError); + run.session.completeMute(); + await mute; + + final unmute = run.session.unmute(); + expect(identical(unmute, run.session.unmute()), isTrue); + expect(run.session.mute, throwsStateError); + expect(run.session.finalize, throwsStateError); + run.session.completeUnmute(); + await unmute; + + final finalize = run.session.finalize(); + expect(run.session.mute, throwsStateError); + expect(run.session.unmute, throwsStateError); + run.session.completeFinalize(); + await finalize; + }); + + test('frames wait for a pending unmute to complete', () async { + final run = await _started(_warmProvider()); + final mute = run.session.mute(); + run.session.completeMute(); + await mute; + + final unmute = run.session.unmute(); + expect(() => run.session.addFrame(_frame(1)), throwsStateError); + run.session.completeUnmute(); + await unmute; + expect(run.session.muted, isFalse); + + await run.session.addFrame(_frame(1)); + expect(run.session.frames, hasLength(1)); + }); + + test('gate changes before start completes throw', () async { + final session = _warmProvider(holdStart: true).createSession(_request()); + expect(session.mute, throwsStateError); + expect(session.unmute, throwsStateError); + expect(session.finalize, throwsStateError); + + final start = session.start(); + expect(session.mute, throwsStateError); + expect(session.unmute, throwsStateError); + expect(session.finalize, throwsStateError); + session.completeStart(); + await start; + }); + + test('warmUnmute requires immediateMute', () { + expect( + () => FakeVoiceProvider(capabilities: {ProviderCapability.warmUnmute}), + throwsArgumentError, + ); + }); + + test('undeclared optional events throw', () async { + final run = await _started(FakeVoiceProvider()); + expect(() => run.session.emitReadiness(live: true), throwsStateError); + expect(() => run.session.emitAmplitude(0.5), throwsStateError); + expect(() => run.session.rejectFinal('no'), throwsStateError); + expect(() => run.session.emitFinal(''), throwsArgumentError); + expect(run.events, isEmpty); + }); + + test('declared optional events are delivered', () async { + final run = await _started( + FakeVoiceProvider( + capabilities: { + ProviderCapability.realAudioReadiness, + ProviderCapability.amplitude, + }, + ), + ); + run.session + ..emitReadiness(live: true) + ..emitAmplitude(0.25); + + expect((run.events[0] as ProviderReadiness).live, isTrue); + expect((run.events[1] as ProviderAmplitude).value, 0.25); + }); + + test('a rejected final is delivered but never assembled', () async { + final run = await _started( + FakeVoiceProvider(capabilities: {ProviderCapability.rejectedFinals}), + ); + run.session + ..emitFinal('Keep this.') + ..emitPartial(' someone else') + ..rejectFinal(' someone else') + ..emitFinal(' And this.'); + + expect( + run.events.whereType().single.text, + ' someone else', + ); + expect(run.openHypothesis, isEmpty); + expect(run.transcript, 'Keep this. And this.'); + }); + }); + + group('backpressure and validity', () { + test('a held frame stays pending until released', () async { + final run = await _started(FakeVoiceProvider()); + run.session.holdFrames = true; + var accepted = false; + final frame = run.session.addFrame(_frame(1)) + ..then((_) => accepted = true); + await pumpEventQueue(); + + expect(accepted, isFalse); + expect(run.session.frames, isEmpty); + expect(() => run.session.addFrame(_frame(2)), throwsStateError); + + run.session.releaseFrame(); + await frame; + expect(accepted, isTrue); + expect(run.session.frames.single.sequence, BigInt.one); + expect(run.session.releaseFrame, throwsStateError); + }); + + test('rejects a mismatched format or non-increasing sequence', () async { + final run = await _started(FakeVoiceProvider()); + expect( + () => run.session.addFrame( + _frame(1, format: _format(encoding: AudioEncoding.opus)), + ), + throwsArgumentError, + ); + expect( + () => run.session.addFrame(_frame(1, format: _format(channels: 2))), + throwsArgumentError, + ); + + await run.session.addFrame(_frame(2)); + expect(() => run.session.addFrame(_frame(2)), throwsArgumentError); + expect(() => run.session.addFrame(_frame(1)), throwsArgumentError); + await run.session.addFrame(_frame(3)); + expect(run.session.frames.map((frame) => frame.sequence), [ + BigInt.two, + BigInt.from(3), + ]); + }); + }); +} + +/// Records a session's events, their kinds, and stream completion. +final class _Run { + _Run(this.session) { + subscription = session.events.listen( + (event) { + events.add(event); + order.add(switch (event) { + ProviderPartial() => 'partial', + ProviderFinal() => 'final', + ProviderRejected() => 'rejected', + ProviderReadiness() => 'readiness', + ProviderAmplitude() => 'amplitude', + ProviderFailure() => 'failure', + }); + }, + onDone: () { + done = true; + order.add('done'); + }, + ); + addTearDown(subscription.cancel); + } + + final FakeProviderSession session; + final List events = []; + final List order = []; + late final StreamSubscription subscription; + var done = false; + + List get partials => + events.whereType().map((event) => event.text).toList(); + + /// The utterance: the exact concatenation of finals. + String get transcript => + events.whereType().map((event) => event.text).join(); + + /// The hypothesis of the segment opened after the last commit. + String get openHypothesis { + var hypothesis = ''; + for (final event in events) { + hypothesis = switch (event) { + ProviderPartial(:final text) => text, + ProviderFinal() || ProviderRejected() => '', + _ => hypothesis, + }; + } + return hypothesis; + } +} + +Future<_Run> _started(FakeVoiceProvider provider) async { + final run = _Run(provider.createSession(_request())); + final start = run.session.start(); + if (run.session.startPending) run.session.completeStart(); + await start; + return run; +} + +FakeVoiceProvider _warmProvider({bool holdStart = false}) => FakeVoiceProvider( + capabilities: { + ProviderCapability.immediateMute, + ProviderCapability.warmUnmute, + }, + holdStart: holdStart, +); + +ProviderSessionRequest _request({AudioFormat? format}) => + ProviderSessionRequest(sessionId: 'capture-1', format: format ?? _format()); + +AudioFormat _format({ + AudioEncoding encoding = AudioEncoding.pcmS16le, + int channels = 1, +}) => AudioFormat( + sampleRateHz: 16000, + channels: channels, + encoding: encoding, + frameDurationMs: 10, +); + +AudioFrame _frame(int sequence, {AudioFormat? format}) => AudioFrame( + protocol: ProtocolVersion.current, + sessionId: 'connector-session-1', + sequence: BigInt.from(sequence), + monotonicTimeUs: BigInt.from(sequence * 10000), + format: format ?? _format(), + payloadBase64: 'AAA=', +); + +VoiceError _failure(String code, {bool retryable = true}) => + VoiceError(code: code, message: 'Synthetic failure.', retryable: retryable); + +Matcher _voiceError(String code, {bool? retryable}) { + var matcher = isA().having((error) => error.code, 'code', code); + if (retryable != null) { + matcher = matcher.having( + (error) => error.retryable, + 'retryable', + retryable, + ); + } + return matcher; +} diff --git a/sdks/dart/murmur_protocol/test/support/fake_voice_provider.dart b/sdks/dart/murmur_protocol/test/support/fake_voice_provider.dart new file mode 100644 index 0000000..255a608 --- /dev/null +++ b/sdks/dart/murmur_protocol/test/support/fake_voice_provider.dart @@ -0,0 +1,443 @@ +import 'dart:async'; +import 'dart:collection'; + +import 'package:murmur_protocol/murmur_protocol.dart'; + +/// A deterministic, manually driven [VoiceProvider] for tests. +/// +/// Nothing happens on a timer: tests complete held operations and emit events +/// explicitly. By default every session enforces the [ProviderSession] +/// contract. [ignoreCancellation] opts into a broken provider whose [stop] +/// neither closes events nor settles pending operations. +final class FakeVoiceProvider implements VoiceProvider { + /// Creates a fake provider. + /// + /// [supportedEncodings] defaults to every [AudioEncoding]. With [holdStart], + /// each session's start stays pending until [FakeProviderSession.completeStart] + /// or [FakeProviderSession.failStart]. Declaring + /// [ProviderCapability.warmUnmute] without [ProviderCapability.immediateMute] + /// throws [ArgumentError]. + FakeVoiceProvider({ + this.id = 'fake', + Set capabilities = const {}, + Set? supportedEncodings, + this.holdStart = false, + this.ignoreCancellation = false, + }) : capabilities = Set.unmodifiable(capabilities), + _supportedEncodings = Set.unmodifiable( + supportedEncodings ?? AudioEncoding.values, + ) { + if (capabilities.contains(ProviderCapability.warmUnmute) && + !capabilities.contains(ProviderCapability.immediateMute)) { + throw ArgumentError.value( + capabilities, + 'capabilities', + 'warmUnmute requires immediateMute', + ); + } + } + + @override + final String id; + + @override + final Set capabilities; + + final bool holdStart; + final bool ignoreCancellation; + final Set _supportedEncodings; + final List _sessions = []; + + /// Every session created so far, oldest first. + List get sessions => UnmodifiableListView(_sessions); + + @override + FakeProviderSession createSession(ProviderSessionRequest request) { + if (!_supportedEncodings.contains(request.format.encoding)) { + throw VoiceError( + code: 'unsupported_format', + message: 'The fake provider cannot decode this audio format.', + retryable: false, + ); + } + final session = FakeProviderSession._(this, request); + _sessions.add(session); + return session; + } +} + +/// A session of [FakeVoiceProvider] with test controls. +/// +/// Controls that settle an operation throw [StateError] when nothing is +/// pending on a live session, and do nothing after the session ended, which +/// models a late provider callback. +final class FakeProviderSession implements ProviderSession { + FakeProviderSession._(this._provider, this.request); + + final FakeVoiceProvider _provider; + + /// The request this session was created for. + final ProviderSessionRequest request; + + final StreamController _events = + StreamController(sync: true); + final List _frames = []; + + Completer? _start; + var _started = false; + Completer? _heldFrame; + AudioFrame? _heldFrameValue; + BigInt? _lastSequence; + var _muted = false; + Completer? _mute; + Completer? _unmute; + Completer? _finalize; + var _finalized = false; + VoiceError? _endError; + Future? _stop; + + /// Whether each [addFrame] stays pending until [releaseFrame]. + bool holdFrames = false; + + var _startCalls = 0; + var _finalizeCalls = 0; + var _stopCalls = 0; + + /// Every [start] call, including shared and rejected ones. + int get startCalls => _startCalls; + + /// Every [finalize] call, including shared and rejected ones. + int get finalizeCalls => _finalizeCalls; + + /// Every [stop] call, including shared and ignored ones. + int get stopCalls => _stopCalls; + + /// Frames the session accepted, in order. Discarded frames are absent. + List get frames => UnmodifiableListView(_frames); + + /// The number of frames waiting for [releaseFrame], either 0 or 1. + int get heldFrames => _heldFrame == null ? 0 : 1; + + /// Whether the input gate is closed by [mute] and not yet reopened. + bool get muted => _muted; + + /// Whether [events] has been closed. + bool get isClosed => _events.isClosed; + + bool get startPending => _isPending(_start); + bool get mutePending => _isPending(_mute); + bool get unmutePending => _isPending(_unmute); + bool get finalizePending => _isPending(_finalize); + + bool get _ended => _finalized || _endError != null; + + @override + Stream get events => _events.stream; + + @override + Future start() { + _startCalls++; + final pending = _start; + if (pending != null && !pending.isCompleted) return pending.future; + if (pending != null || _ended) { + throw StateError('a provider session starts only once'); + } + final start = _start = Completer(); + if (!_provider.holdStart) completeStart(); + return start.future; + } + + /// Completes a start held by [FakeVoiceProvider.holdStart]. + void completeStart() { + if (_ended) return; + final start = _requirePending(_start, 'start'); + _started = true; + start.complete(); + } + + /// Fails a pending start with [error] and closes [events]. + void failStart(VoiceError error) { + if (_ended) return; + _requirePending(_start, 'start'); + _end(error); + } + + @override + Future addFrame(AudioFrame frame) { + final endError = _endError; + if (endError != null) return Future.error(endError); + if (_finalized) return Future.value(); + if (!_started) { + throw StateError('frames are accepted only after start completes'); + } + if (unmutePending) { + throw StateError('frames are not accepted while unmute is pending'); + } + if (_heldFrame != null) { + throw StateError('another frame is still outstanding'); + } + _validate(frame); + if (_muted || finalizePending) return Future.value(); + if (!holdFrames) { + _frames.add(frame); + return Future.value(); + } + _heldFrameValue = frame; + return (_heldFrame = Completer()).future; + } + + /// Accepts the frame held by [holdFrames]. + void releaseFrame() { + if (_ended) return; + final held = _requirePending(_heldFrame, 'frame'); + _frames.add(_heldFrameValue!); + _heldFrame = null; + _heldFrameValue = null; + held.complete(); + } + + @override + Future mute() { + _requireCapability(ProviderCapability.immediateMute, 'mute'); + final endError = _endError; + if (endError != null) return Future.error(endError); + _requireGateTransition(); + final pending = _mute; + if (pending != null) return pending.future; + if (unmutePending) { + throw StateError('mute cannot start while unmute is pending'); + } + if (_muted) return Future.value(); + _muted = true; + return (_mute = Completer()).future; + } + + /// Completes a pending mute flush. + /// + /// Throws [StateError] while a frame is held, because it belongs to the + /// flushed tail. + void completeMute() { + if (_ended) return; + final mute = _requirePending(_mute, 'mute'); + _requireNoHeldFrame(); + _mute = null; + mute.complete(); + } + + @override + Future unmute() { + _requireCapability(ProviderCapability.warmUnmute, 'unmute'); + final endError = _endError; + if (endError != null) return Future.error(endError); + _requireGateTransition(); + final pending = _unmute; + if (pending != null) return pending.future; + if (mutePending) { + throw StateError('unmute cannot start while a mute flush is pending'); + } + if (!_muted) return Future.value(); + return (_unmute = Completer()).future; + } + + /// Completes a pending unmute and reopens the input gate. + void completeUnmute() { + if (_ended) return; + final unmute = _requirePending(_unmute, 'unmute'); + _unmute = null; + _muted = false; + unmute.complete(); + } + + @override + Future finalize() { + _finalizeCalls++; + final existing = _finalize; + if (existing != null) return existing.future; + final endError = _endError; + if (endError != null) { + // Later calls replay this identical terminal outcome. + return (_finalize = Completer()..completeError(endError)).future; + } + if (!_started) { + throw StateError('finalize requires a started session'); + } + if (mutePending || unmutePending) { + throw StateError('finalize cannot start while a gate change is pending'); + } + return (_finalize = Completer()).future; + } + + /// Completes a pending finalize, then closes [events]. + /// + /// Throws [StateError] while a frame is held, because it belongs to the + /// finalized tail. + void completeFinalize() { + if (_ended) return; + final finalize = _requirePending(_finalize, 'finalize'); + _requireNoHeldFrame(); + _finalized = true; + finalize.complete(); + unawaited(_events.close()); + } + + @override + Future stop() { + _stopCalls++; + if (_provider.ignoreCancellation) return Future.value(); + final existing = _stop; + if (existing != null) return existing; + // Cache first: closing the sync stream notifies listeners reentrantly. + final stop = _stop = Future.value(); + if (!_ended) { + _end( + VoiceError( + code: 'cancelled', + message: 'The provider session was cancelled.', + retryable: true, + ), + ); + } + return stop; + } + + /// Fails the session with [error]. + /// + /// While start is pending this acts like [failStart]. After start it emits + /// one [ProviderFailure], completes pending operations with [error], and + /// closes [events]. + void fail(VoiceError error) { + if (_ended) return; + if (startPending) { + _end(error); + return; + } + _requireStartCalled(); + _end(error, lastEvent: ProviderFailure(error)); + } + + /// Fails the session with a retryable `provider_disconnected` error. + void disconnect() { + fail( + VoiceError( + code: 'provider_disconnected', + message: 'The provider connection closed unexpectedly.', + retryable: true, + ), + ); + } + + // Each emit checks termination before building its event, so a late + // callback with an invalid value adds nothing instead of throwing. + + void emitPartial(String text) { + if (_canEmit()) _events.add(ProviderPartial(text)); + } + + void emitFinal(String text) { + if (_canEmit()) _events.add(ProviderFinal(text)); + } + + void rejectFinal(String text) { + if (_canEmit(ProviderCapability.rejectedFinals)) { + _events.add(ProviderRejected(text)); + } + } + + void emitReadiness({required bool live}) { + if (_canEmit(ProviderCapability.realAudioReadiness)) { + _events.add(ProviderReadiness(live: live)); + } + } + + void emitAmplitude(double value) { + if (_canEmit(ProviderCapability.amplitude)) { + _events.add(ProviderAmplitude(value)); + } + } + + /// Whether an event may be added now. + /// + /// An undeclared [capability] throws [StateError] in every state. + bool _canEmit([ProviderCapability? capability]) { + if (capability != null && !_provider.capabilities.contains(capability)) { + throw StateError('this event requires ${capability.name}'); + } + if (_ended) return false; + _requireStartCalled(); + return true; + } + + /// Ends the session with [error] and settles every pending operation. + /// + /// The terminal state is recorded before [lastEvent] and done are delivered, + /// so a listener that reenters the session observes the ended session. + void _end(VoiceError error, {ProviderEvent? lastEvent}) { + _endError = error; + for (final pending in [_start, _heldFrame, _mute, _unmute, _finalize]) { + if (pending != null && !pending.isCompleted) { + pending.completeError(error); + } + } + _heldFrame = null; + _heldFrameValue = null; + if (lastEvent != null) _events.add(lastEvent); + unawaited(_events.close()); + } + + void _validate(AudioFrame frame) { + final expected = request.format; + final actual = frame.format; + if (actual.sampleRateHz != expected.sampleRateHz || + actual.channels != expected.channels || + actual.encoding != expected.encoding || + actual.frameDurationMs != expected.frameDurationMs) { + throw ArgumentError.value(frame, 'frame', 'does not match the format'); + } + final last = _lastSequence; + if (last != null && frame.sequence <= last) { + throw ArgumentError.value( + frame, + 'frame', + 'sequence must be strictly increasing', + ); + } + _lastSequence = frame.sequence; + } + + void _requireCapability(ProviderCapability capability, String operation) { + if (!_provider.capabilities.contains(capability)) { + throw UnsupportedError('$operation requires ${capability.name}'); + } + } + + void _requireGateTransition() { + if (_finalize != null) { + throw StateError('the input gate cannot change after finalize'); + } + if (!_started) { + throw StateError('the input gate changes only after start completes'); + } + } + + void _requireStartCalled() { + if (_start == null) { + throw StateError('the session has not been started'); + } + } + + void _requireNoHeldFrame() { + if (_heldFrame != null) { + throw StateError('a held frame must be released first'); + } + } + + Completer _requirePending(Completer? operation, String name) { + if (!_isPending(operation)) { + throw StateError('no $name is pending'); + } + return operation!; + } + + static bool _isPending(Completer? operation) => + operation != null && !operation.isCompleted; +}