diff --git a/backend/cpp/audio-cpp/CMakeLists.txt b/backend/cpp/audio-cpp/CMakeLists.txt index 740d74c32..2bb15b366 100644 --- a/backend/cpp/audio-cpp/CMakeLists.txt +++ b/backend/cpp/audio-cpp/CMakeLists.txt @@ -132,6 +132,8 @@ add_executable(${TARGET} result_map.cpp stem_selection.cpp generation_request.cpp + stream_delta.cpp + wav_header.cpp inference_lane.cpp ) @@ -247,4 +249,32 @@ if(AUDIO_CPP_GRPC_BUILD_TESTS) INSTALL_RPATH "$ORIGIN" BUILD_WITH_INSTALL_RPATH TRUE) add_test(NAME audio_io COMMAND audio_io_ctest) + + # The streaming drivers live in loaded_model.cpp, which links the engine, so + # this cannot be a standalone *_test.cpp. It builds no model and reads no + # file: LoadedModel::Session is a plain struct holding a pointer to an + # engine interface, so the drivers are exercised against fake sessions. + add_executable(streaming_driver_ctest + streaming_driver_ctest.cpp + loaded_model.cpp + capability_routing.cpp + family_gate.cpp + model_options.cpp + inference_lane.cpp) + target_include_directories(streaming_driver_ctest PRIVATE + "${AUDIO_CPP_DIR}/include" + "${CMAKE_CURRENT_SOURCE_DIR}") + target_link_libraries(streaming_driver_ctest PRIVATE + engine_runtime + ggml + Threads::Threads) + target_compile_options(streaming_driver_ctest PRIVATE -Wall -Wextra -Wpedantic) + # No "$ORIGIN" rpath override here, unlike the shipping target and unlike + # audio_io_ctest. loaded_model.cpp reaches make_default_registry, so this + # binary genuinely links libggml, and CMake's own build-tree rpath is what + # finds it: the ggml shared objects land in ${CMAKE_CURRENT_BINARY_DIR}/bin + # while the test binary sits one directory up. A build-host absolute path in + # a test binary is harmless, since package.sh ships only grpc-server, and + # forcing "$ORIGIN" here means ctest cannot start the binary at all. + add_test(NAME streaming_driver COMMAND streaming_driver_ctest) endif() diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index 360f4d0e4..6ab2181da 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -6,8 +6,9 @@ // upstream expects churn, and upstream is Apache-2.0 while LocalAI is MIT. // // This commit adds LoadModel/Free/Status plus the AudioTranscription, VAD, -// Diarize, AudioTransform, TTS and SoundGeneration RPCs. The remaining audio -// RPCs land in later commits. +// Diarize, AudioTransform, TTS, SoundGeneration, TTSStream and +// AudioTranscriptionStream RPCs. The remaining audio RPCs land in later +// commits. #include "backend.pb.h" #include "backend.grpc.pb.h" @@ -21,10 +22,13 @@ #include "model_options.h" #include "result_map.h" #include "stem_selection.h" +#include "stream_delta.h" +#include "wav_header.h" #include #include #include +#include #include #include @@ -415,6 +419,42 @@ float audio_duration_seconds(const engine::runtime::AudioBuffer &audio) { return audiocpp_backend::samples_to_seconds(frames, audio.sample_rate); } +// Refuses a request that routing sent to voice cloning with no speaker +// reference clip. Shared by TTS and TTSStream so the two RPCs cannot drift into +// refusing the same request differently. +// +// Refused FROM THE ROUTE, the same trick AudioTransform uses for params[stem]. +// Voice cloning without a clip is a request the family cannot answer, and every +// cloning family says so only from inside its own prepare(): chatterbox throws +// "Chatterbox prepare requires speaker reference audio", which to_status maps to +// INTERNAL and which names neither the RPC nor the field the caller has to set. +// chatterbox advertises clon and no tts at all, so before this every preset-only +// or voice-less request to it got that engine-internal message. +// +// It cannot misfire on the legitimate case: has_voice_reference is what made +// routing choose VoiceCloning in the first place, so reaching here with a clip +// present is impossible unless the model pinned task:clon, and a clon pin with +// no clip is exactly the misconfiguration worth naming. +// +// Deliberately NOT generalised to "every task whose family declares +// supports_speaker_reference". That flag lives on the engine's CapabilitySet, is +// not carried through this backend's Capabilities mirror, and would need its own +// answer to whether it is advisory or binding before routing could act on it. +// Tracked as a follow-up. +void refuse_cloning_without_a_clip(const audiocpp_backend::LoadedModel &model, + const audiocpp_backend::Route &route, + bool voice_is_file) { + if (route.task != audiocpp_backend::Task::VoiceCloning || voice_is_file) { + return; + } + throw audiocpp_backend::ConfigError( + "audio-cpp: family '" + model.family() + "' routes this request to " + + audiocpp_backend::task_name(route.task) + + ", which needs a speaker reference clip; set TTSRequest.voice to the " + "path of a WAV file (a voice that is not a file on disk is treated as a " + "named preset)"); +} + // Maps a thrown exception onto the gRPC status the client should see. GStatus to_status(const std::exception &err) { if (dynamic_cast(&err) != nullptr) { @@ -1045,37 +1085,7 @@ public: const audiocpp_backend::Route route = model->check_can_serve(audiocpp_backend::Rpc::Tts, shape); - // Refused FROM THE ROUTE, the same trick AudioTransform uses for - // params[stem]. Voice cloning without a clip is a request the family - // cannot answer, and every cloning family says so only from inside - // its own prepare(): chatterbox throws "Chatterbox prepare requires - // speaker reference audio", which to_status maps to INTERNAL and - // which names neither the RPC nor the field the caller has to set. - // chatterbox advertises clon and no tts at all, so before this every - // preset-only or voice-less request to it got that engine-internal - // message. - // - // It cannot misfire on the legitimate case: has_voice_reference is - // what made routing choose VoiceCloning in the first place, so - // reaching here with a clip present is impossible unless the model - // pinned task:clon, and a clon pin with no clip is exactly the - // misconfiguration worth naming. - // - // Deliberately NOT generalised to "every task whose family declares - // supports_speaker_reference". That flag lives on the engine's - // CapabilitySet, is not carried through this backend's Capabilities - // mirror, and would need its own answer to whether it is advisory or - // binding before routing could act on it. Tracked as a follow-up. - if (route.task == audiocpp_backend::Task::VoiceCloning && - !voice_is_file) { - throw audiocpp_backend::ConfigError( - "audio-cpp: family '" + model->family() + - "' routes this request to " + - audiocpp_backend::task_name(route.task) + - ", which needs a speaker reference clip; set TTSRequest.voice" - " to the path of a WAV file (a voice that is not a file on " - "disk is treated as a named preset)"); - } + refuse_cloning_without_a_clip(*model, route, voice_is_file); if (request->dst().empty()) { throw audiocpp_backend::ConfigError( @@ -1218,6 +1228,286 @@ public: return to_status(err); } } + + // The streaming counterpart of TTS, and the first RPC here that writes to + // the wire while the model is still generating. + // + // THE WIRE CONTRACT, which is the thing most likely to go wrong. + // pkg/grpc/server.go's TTSStream wrapper puts every chunk in Reply.audio, + // and core/backend/tts.go's ModelTTSStream forwards those bytes to the HTTP + // response verbatim. There is no framing and no format negotiation, so THE + // FIRST CHUNK MUST BE A WAV HEADER or the client receives raw PCM it has no + // way to interpret; browsers simply refuse the stream. Because the total + // length is unknown while generating, both size fields carry 0xFFFFFFFF, + // which is the convention backend/go/vibevoice-cpp established. + // + // ModelTTSStream will synthesise a header of its own, but ONLY when the + // first Reply carries a non-empty `message` holding JSON with a sample_rate. + // Nothing here ever sets Reply.message, so that branch never fires and there + // is exactly one header on the wire. Setting `message` on this RPC without + // deleting the header below would put a second header 44 bytes into the PCM. + // + // There is no dst: streaming TTS writes to the response, not to a file, and + // core/backend/tts.go sends an empty dst on purpose. That is the one place + // this RPC deliberately diverges from TTS. + GStatus TTSStream(ServerContext *, const backend::TTSRequest *request, + grpc::ServerWriter *writer) override { + try { + GStatus refusal = GStatus::OK; + const auto model = snapshot_for(request, refusal); + if (model == nullptr) { + return refusal; + } + + // The SAME shape builder TTS uses, so the two RPCs cannot describe + // one request differently. pinned_task stays at the call site: it + // comes off the model rather than the request. + audiocpp_backend::RequestShape shape = + audiocpp_backend::build_tts_shape(*request); + const bool voice_is_file = shape.has_voice_reference; + shape.pinned_task = model->pinned_task(); + + // FIRST, before the lane and before any file read, exactly as in + // TTS. Note that this RPC's mode candidates are streaming ONLY: a + // family that can synthesise but cannot stream is refused here + // rather than quietly served a whole buffer at the end, because a + // caller that asked to stream is asking for time to first audio. + const audiocpp_backend::Route route = + model->check_can_serve(audiocpp_backend::Rpc::TtsStream, shape); + refuse_cloning_without_a_clip(*model, route, voice_is_file); + + std::optional reference; + if (voice_is_file) { + // Native rate, native channels: see kVoiceReferenceSampleRate. + reference = audiocpp_backend::read_audio_file( + request->voice(), kVoiceReferenceSampleRate); + } + const auto task = + audiocpp_backend::build_tts_request(*request, std::move(reference)); + + audiocpp_backend::LaneEntry lane = model->acquire(0); + const auto session = + model->session_for(audiocpp_backend::Rpc::TtsStream, shape, lane); + + bool header_sent = false; + bool client_gone = false; + int stream_rate = 0; + int stream_channels = 0; + const auto write_audio = + [&](const engine::runtime::AudioBuffer &audio) { + if (client_gone || audio.samples.empty()) { + return; + } + const int channels = audio.channels > 0 ? audio.channels : 1; + if (!header_sent) { + backend::Reply head; + head.set_audio(audiocpp_backend::streaming_wav_header( + audio.sample_rate, channels)); + if (!writer->Write(head)) { + client_gone = true; + return; + } + header_sent = true; + stream_rate = audio.sample_rate; + stream_channels = channels; + } else if (audio.sample_rate != stream_rate || + channels != stream_channels) { + // The header already went out declaring the first + // chunk's format, and it cannot be taken back. Every + // later byte would be decoded at the wrong rate or the + // wrong interleaving, which plays as a speed or channel + // fault nobody can see in a 200. Fail loudly instead. + throw std::runtime_error( + "audio-cpp: family '" + model->family() + + "' changed its output format mid-stream (" + + std::to_string(stream_rate) + " Hz x" + + std::to_string(stream_channels) + " to " + + std::to_string(audio.sample_rate) + " Hz x" + + std::to_string(channels) + ")"); + } + backend::Reply reply; + reply.set_audio(audiocpp_backend::f32_to_s16le(audio.samples)); + if (!writer->Write(reply)) { + client_gone = true; + } + }; + + // BOTH fields, and named_audio_outputs is the one that carries the + // audio in practice. All three streaming TTS families put their + // chunks there ("chunk_0", "chunk_1", ...) and leave audio_output + // empty until the very end: supertonic and omnivoice build the event + // in next_stream_event, voxcpm2 replays the ones its start_stream + // produced. Reading only audio_output, which is the obvious field, + // yields a stream with no audio in it at all. + const auto emit = [&](const engine::runtime::StreamEvent &event) { + if (event.audio_output.has_value()) { + write_audio(*event.audio_output); + } + for (const auto &named : event.named_audio_outputs) { + write_audio(named.audio); + } + }; + + const auto result = + audiocpp_backend::run_streaming_pull(session, task, emit, lane); + + // The finish_stream result is the session's own MERGED WHOLE, not a + // tail the pull loop missed: supertonic returns its accumulated + // buffer, omnivoice post-processes the merge, voxcpm2 hands back the + // stored result. Emitting it after the chunks would send the entire + // utterance twice. It is therefore used only when the family + // streamed nothing, which is what a family that held everything back + // to finalize looks like. + // + // The cost of that choice, stated plainly: what a client hears is + // the concatenation of the chunks, and for omnivoice that is not + // byte-identical to what the unary TTS RPC returns, because its + // postprocessor runs over the merged buffer at the end. Audio + // already on the wire cannot be post-processed, so any streaming + // implementation has to accept this. + if (!header_sent) { + if (result.audio_output.has_value()) { + write_audio(*result.audio_output); + } else { + for (const auto &named : result.named_audio_outputs) { + write_audio(named.audio); + } + } + } + + if (!header_sent && !client_gone) { + // Routed successfully and produced nothing. Same judgement as + // TTS: from the caller's side the actionable fact is that this + // model does not do this. + throw audiocpp_backend::CapabilityError( + "audio-cpp: family '" + model->family() + + "' produced no audio for the TTSStream RPC"); + } + // client_gone is NOT an error: the client hung up, gRPC already + // knows, and there is nobody left to tell. + return GStatus::OK; + } catch (const std::exception &err) { + // No Result message to fill in on this RPC. A status mid-stream is + // what the client sees, and gRPC delivers it after whatever chunks + // already went out. + return to_status(err); + } + } + + // Server-streaming transcription: zero or more TranscriptStreamResponse + // messages carrying `delta`, then exactly one carrying `final_result`. + // + // THE DELTA CONTRACT. core/backend/transcript.go documents Delta as "an + // incremental text fragment" and the HTTP layer appends them, so a client + // that concatenates every delta must end up with the final text and must + // never see a prefix repeated. The families do not agree on what they report + // (nemotron_asr and vibevoice_asr send fragments, voxtral_realtime sends the + // whole hypothesis every time, and sends it twice), so the reconciliation + // lives in TranscriptDeltaTracker rather than here. See stream_delta.h. + // + // THE OFFLINE FALLBACK. mode_candidates lets this RPC fall back to an + // offline route, so a family with no streaming ASR still answers rather than + // returning UNIMPLEMENTED. It then runs once and the reconciliation below + // emits the whole transcript as a single delta: the same message sequence a + // streaming family produces, with one delta instead of many. That is a + // truthful degradation, and it needs no branch of its own, because a tracker + // that observed nothing reconciles to exactly one fragment. + GStatus AudioTranscriptionStream( + ServerContext *, const backend::TranscriptRequest *request, + grpc::ServerWriter *writer) override { + try { + GStatus refusal = GStatus::OK; + const auto model = snapshot_for(request, refusal); + if (model == nullptr) { + return refusal; + } + + audiocpp_backend::RequestShape shape; + shape.has_prompt_text = !request->prompt().empty(); + shape.pinned_task = model->pinned_task(); + + // Before the lane and before the file read, for the reasons spelled + // out in Diarize. + model->check_can_serve(audiocpp_backend::Rpc::AudioTranscriptionStream, + shape); + + // TranscriptRequest.dst is the INPUT audio path, despite the field + // name; nothing is written back. Read at 16 kHz mono for the reasons + // in kSpeechSampleRate, which apply identically here: the streaming + // sessions build their spans in the same feature domain their + // offline halves do. + auto audio = audiocpp_backend::read_audio_file(request->dst(), + kSpeechSampleRate); + const int sample_rate = audio.sample_rate; + const float duration = audio_duration_seconds(audio); + + audiocpp_backend::LaneEntry lane = model->acquire(0); + const auto session = model->session_for( + audiocpp_backend::Rpc::AudioTranscriptionStream, shape, lane); + + // session.task, not the request: for Asr the prompt is decoding + // context, for Alignment it is the transcript to align. + const auto task = build_transcription_request(*request, session.task, + std::move(audio)); + if (!task.audio_input.has_value()) { + // Unreachable through build_transcription_request, which always + // sets it. Checked because the alternative is dereferencing an + // empty optional below. + throw std::runtime_error( + "audio-cpp: transcription request carries no audio"); + } + + audiocpp_backend::TranscriptDeltaTracker tracker; + bool client_gone = false; + const auto write_delta = [&](const std::string &fragment) { + if (fragment.empty() || client_gone) { + return; + } + backend::TranscriptStreamResponse response; + response.set_delta(fragment); + if (!writer->Write(response)) { + client_gone = true; + } + }; + const auto emit = [&](const engine::runtime::StreamEvent &event) { + if (!event.partial_text.has_value()) { + return; + } + write_delta(tracker.observe(event.partial_text->text)); + }; + + engine::runtime::TaskResult result; + if (session.mode == audiocpp_backend::Mode::Streaming) { + // The buffer inside the request is the chunk source, so the + // audio is held once rather than copied for the feed. + result = audiocpp_backend::run_streaming_audio( + session, task, *task.audio_input, emit, lane); + } else { + result = audiocpp_backend::run_offline(session, task, lane); + } + + // The reconciliation, and the only place the offline fallback needs + // to be thought about: with no partials observed this IS the single + // delta carrying the whole transcript. + if (result.text_output.has_value()) { + write_delta(tracker.reconcile(result.text_output->text)); + } + + backend::TranscriptStreamResponse final_response; + // sample_rate is the BUFFER's, which after read_audio_file is + // kSpeechSampleRate and not necessarily the file's; it is the domain + // the result spans come back in. + audiocpp_backend::fill_transcript_result( + result, sample_rate, duration, + final_response.mutable_final_result()); + if (!client_gone) { + writer->Write(final_response); + } + return GStatus::OK; + } catch (const std::exception &err) { + return to_status(err); + } + } }; void RunServer(const std::string &addr) { diff --git a/backend/cpp/audio-cpp/loaded_model.cpp b/backend/cpp/audio-cpp/loaded_model.cpp index f8888a055..667c817dc 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -4,6 +4,9 @@ #include "engine/framework/assets/tensor_source.h" +#include +#include +#include #include #include @@ -462,4 +465,141 @@ engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, return session.offline->run(request); } +namespace { + +engine::runtime::IStreamingVoiceTaskSession & +require_streaming(const LoadedModel::Session &session) { + if (session.streaming == nullptr) { + throw CapabilityError("audio-cpp: no streaming session for this request"); + } + return *session.streaming; +} + +// Clears the stream event sink on every exit from the driver, including the +// exception path. The session is CACHED and outlives the call that installed +// the sink, so a std::function left behind holding references into that call's +// frame is called with dangling captures by whoever streams next. +class ScopedStreamSink { +public: + ScopedStreamSink(engine::runtime::IStreamingVoiceTaskSession &session, + engine::runtime::StreamEventCallback sink) + : session_(session) { + session_.set_stream_event_sink(std::move(sink)); + } + ~ScopedStreamSink() { session_.set_stream_event_sink(nullptr); } + + ScopedStreamSink(const ScopedStreamSink &) = delete; + ScopedStreamSink &operator=(const ScopedStreamSink &) = delete; + +private: + engine::runtime::IStreamingVoiceTaskSession &session_; +}; + +// Frames per chunk to feed a streaming session, from its own policy. +// +// FRAMES, not floats. preferred_audio_chunk_samples is a per-channel count +// everywhere upstream sets it (nemotron_asr uses its frontend sample rate, +// i.e. one second), and vibevoice_asr refuses a chunk whose float count is not +// divisible by its channel count, so slicing on floats would both mis-size the +// window and hand a family a half frame. +std::int64_t chunk_frames_for(const engine::runtime::StreamingPolicy &policy, + int sample_rate) { + if (policy.preferred_audio_chunk_samples > 0) { + return policy.preferred_audio_chunk_samples; + } + // higgs_audio_stt states its window in seconds (4.0) and leaves the sample + // count at zero, so this branch is real rather than defensive. + if (policy.preferred_audio_chunk_seconds > 0.0 && sample_rate > 0) { + const auto frames = static_cast( + policy.preferred_audio_chunk_seconds * static_cast(sample_rate)); + if (frames > 0) { + return frames; + } + } + // The interface's own default, from IStreamingVoiceTaskSession::streaming_policy. + return 512; +} + +} // namespace + +void begin_stream(const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, LaneEntry &lane) { + // Proof of holding only, as in session_for. + (void)lane; + auto &streaming = require_streaming(session); + // Order is load-bearing: start_stream's reset() is illegal before prepare(). + streaming.prepare(engine::runtime::build_preparation_request(request)); + streaming.start_stream(request); +} + +engine::runtime::TaskResult run_streaming_pull( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const std::function &on_event, + LaneEntry &lane) { + auto &streaming = require_streaming(session); + begin_stream(session, request, lane); + while (const auto event = streaming.next_stream_event()) { + if (on_event) { + on_event(*event); + } + // No pinned family sets is_final on a pulled event, so this is not what + // ends the loop today; the nullopt above is. Honoured anyway, because a + // family that does set it is saying the stream is over and pulling once + // more would be asking a finished session for another chunk. + if (event->is_final) { + break; + } + } + return streaming.finish_stream(); +} + +engine::runtime::TaskResult run_streaming_audio( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const engine::runtime::AudioBuffer &audio, + const std::function &on_event, + LaneEntry &lane) { + auto &streaming = require_streaming(session); + + // Installed BEFORE the stream begins, so a family that reports during + // start_stream is not silently dropped, and destroyed after finish_stream, + // because nemotron_asr emits every one of its partials from inside + // finalize(). + ScopedStreamSink sink(streaming, + [&on_event](const engine::runtime::StreamEvent &event) { + if (on_event) { + on_event(event); + } + }); + + begin_stream(session, request, lane); + + const int channels = audio.channels > 0 ? audio.channels : 1; + const auto total_frames = + static_cast(audio.samples.size() / static_cast(channels)); + const std::int64_t chunk_frames = + chunk_frames_for(streaming.streaming_policy(), audio.sample_rate); + + for (std::int64_t offset = 0; offset < total_frames; offset += chunk_frames) { + const std::int64_t end = std::min(offset + chunk_frames, total_frames); + engine::runtime::AudioChunk chunk; + chunk.sample_rate = audio.sample_rate; + chunk.channels = channels; + // A FRAME index, which is what every span in a returned event is + // expressed in. vibevoice_asr adds the chunk's own frame count to it to + // offset the spans it reports, so a float index here would place every + // span of a stereo stream at twice its real time. + chunk.start_sample = offset; + chunk.samples.assign( + audio.samples.begin() + static_cast(offset * channels), + audio.samples.begin() + static_cast(end * channels)); + const auto event = streaming.process_audio_chunk(chunk); + if (on_event) { + on_event(event); + } + } + return streaming.finish_stream(); +} + } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/loaded_model.h b/backend/cpp/audio-cpp/loaded_model.h index a5e97a3e1..b0a498eff 100644 --- a/backend/cpp/audio-cpp/loaded_model.h +++ b/backend/cpp/audio-cpp/loaded_model.h @@ -13,6 +13,7 @@ #include "engine/framework/runtime/registry.h" #include "engine/framework/runtime/session.h" +#include #include #include #include @@ -161,17 +162,15 @@ public: // previous stream used, carrying that stream's state. session_for hands it // back as it is. // - // Every streaming caller must therefore begin a stream with - // - // session.streaming->prepare(build_preparation_request(...)); - // session.streaming->start_stream(request); - // - // in that order. start_stream's base implementation is a call to reset(), - // which is what clears the previous stream, and reset() is only legal after - // prepare(): silero_vad throws "session prepare() must be called before - // Silero VAD reset()" otherwise. That ordering constraint is also why - // session_for cannot do this for you. Skipping it does not raise an error, - // it silently continues the previous stream. + // Every streaming caller must therefore begin a stream through + // begin_stream() below, which is prepare() then start_stream() in that + // order and is the ONLY implementation of that sequence. start_stream's + // base implementation is a call to reset(), which is what clears the + // previous stream, and reset() is only legal after prepare(): silero_vad + // throws "session prepare() must be called before Silero VAD reset()" + // otherwise. That ordering constraint is also why session_for cannot do + // this for you. Skipping it does not raise an error, it silently continues + // the previous stream. // // Offline sessions need no such care: their interface has no reset and // run() takes a whole request. @@ -256,4 +255,78 @@ engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, const engine::runtime::TaskRequest &request, LaneEntry &lane); +// THE ONE IMPLEMENTATION of the streaming state obligation described in +// session_for's STATE CONTRACT: prepare(), then start_stream(), in that order. +// +// It is a function rather than a comment because the obligation is invisible +// when it is broken. Streaming sessions are CACHED per (task, mode), so the +// object a second stream gets is the warm one the first stream left behind, +// still holding its audio, its tokens and its started flag. What clears it is +// start_stream, whose base implementation IS a reset() and whose seven family +// overrides (nemotron_asr, vibevoice_asr, higgs_audio_stt, voxtral_realtime, +// supertonic, omnivoice, voxcpm2) every one call reset() as their first +// statement, verified in the pinned checkout. Nothing in the type system pins +// that. A future override that dropped the reset would break every call site +// at once with no compile error and no exception, only a second transcript +// that begins with the first one's audio, so the fewer call sites there are to +// break, the better: this is the only one. +// +// prepare() must come first and cannot be folded into session_for, because +// reset() is illegal before prepare() (silero_vad throws "session prepare() +// must be called before Silero VAD reset()"), and because the preparation +// request is derived from the REQUEST, not the model: build_preparation_request +// reads the audio contract, the text and the voice condition off it, so a +// second stream with a different sample rate or length would otherwise run +// against the first stream's contract. +// +// `lane` is a PROOF OF HOLDING, unused at runtime, exactly as in run_offline. +// +// Throws CapabilityError when the session is not a streaming one. +void begin_stream(const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, LaneEntry &lane); + +// Drives a streaming session that takes NO incremental input, which is the TTS +// shape (StreamingInputKind::None, StreamingOutputKind::PullEvents): begin the +// stream, pull events until the session says there are no more, then finish. +// +// NO STREAM EVENT SINK IS INSTALLED HERE, and that is deliberate rather than an +// omission. voxcpm2's start_stream runs the whole synthesis and pushes every +// chunk to the sink, then its next_stream_event replays those same chunks out +// of the stored result, so a sink on this path would put every chunk of audio +// on the wire twice. supertonic and omnivoice ignore set_stream_event_sink +// outright. The pull loop is therefore the single delivery channel. +// +// The returned TaskResult is the session's own merged whole for all three +// families, NOT a tail the pull loop missed. A caller that already emitted the +// pulled events must not also emit its audio; see the TTSStream handler. +engine::runtime::TaskResult run_streaming_pull( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const std::function &on_event, + LaneEntry &lane); + +// Drives a streaming session that CONSUMES audio chunks, which is the ASR shape +// (StreamingInputKind::AudioChunks): begin the stream, feed the buffer in +// policy-sized chunks, then finalize. +// +// A STREAM EVENT SINK IS INSTALLED HERE, and it is not optional: nemotron_asr +// reports its partial text ONLY through the sink, and only from inside +// finalize(), because its decode does not start until the audio is complete. +// Without the sink that family streams a transcript with no partials at all. +// The sink is cleared again before returning, including on the exception path: +// the session is cached and outlives this call, so a sink left holding a +// reference to the caller's frame is a use after free waiting for the next +// stream. +// +// Both delivery channels are consumed, the sink and the value process_audio_chunk +// returns, because the families do not agree on which they use, and +// voxtral_realtime uses BOTH for the same event. The duplicate that produces is +// absorbed by TranscriptDeltaTracker in stream_delta.h rather than here. +engine::runtime::TaskResult run_streaming_audio( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const engine::runtime::AudioBuffer &audio, + const std::function &on_event, + LaneEntry &lane); + } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/stream_delta.cpp b/backend/cpp/audio-cpp/stream_delta.cpp new file mode 100644 index 000000000..6655f3708 --- /dev/null +++ b/backend/cpp/audio-cpp/stream_delta.cpp @@ -0,0 +1,50 @@ +#include "stream_delta.h" + +namespace audiocpp_backend { +namespace { + +// True when `text` begins with `prefix`. An empty prefix matches everything, +// which is what makes the first fragment take the cumulative branch and the +// incremental branch alike: they agree there. +bool starts_with(const std::string &text, const std::string &prefix) { + return text.size() >= prefix.size() && + text.compare(0, prefix.size(), prefix) == 0; +} + +} // namespace + +std::string TranscriptDeltaTracker::observe(const std::string &partial_text) { + if (partial_text.empty()) { + return {}; + } + // Already delivered. Covers the exact repeat voxtral produces on every + // event and a hypothesis that shrank. + if (starts_with(assembled_, partial_text)) { + return {}; + } + if (starts_with(partial_text, assembled_)) { + // Cumulative: the report is the whole transcript so far. + std::string fragment = partial_text.substr(assembled_.size()); + assembled_ = partial_text; + return fragment; + } + // Incremental: the fragment is new text to append. + assembled_ += partial_text; + return partial_text; +} + +std::string TranscriptDeltaTracker::reconcile(const std::string &final_text) { + if (final_text.empty() || final_text == assembled_) { + return {}; + } + if (!starts_with(final_text, assembled_)) { + // Contradicted. Nothing sent can be taken back, so nothing more is + // sent; final_result carries the authoritative text. + return {}; + } + std::string fragment = final_text.substr(assembled_.size()); + assembled_ = final_text; + return fragment; +} + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/stream_delta.h b/backend/cpp/audio-cpp/stream_delta.h new file mode 100644 index 000000000..2de9c13a0 --- /dev/null +++ b/backend/cpp/audio-cpp/stream_delta.h @@ -0,0 +1,80 @@ +#pragma once + +// Turns whatever a streaming session calls a "partial transcript" into the +// incremental deltas AudioTranscriptionStream is contracted to send. Standard +// library only, so it is tested without an audio.cpp checkout. +// +// THIS UNIT EXISTS BECAUSE THE FAMILIES DISAGREE, and the disagreement is +// invisible at the interface: StreamEvent::partial_text is a Transcript either +// way. Read out of the pinned upstream, one family at a time: +// +// nemotron_asr INCREMENTAL. decoder.cpp emits +// current_text.substr(emitted_text.size()) per non-blank +// token, and only through the stream event SINK, during +// finalize(). process_audio_chunk returns empty events. +// vibevoice_asr INCREMENTAL. process_audio_chunk returns +// text.substr(common_prefix_size(...)); the sink is +// deliberately swapped out around its internal run_single, +// so the fragment arrives once, on the return value. +// higgs_audio_stt INCREMENTAL. Same shape as vibevoice_asr. +// voxtral_realtime CUMULATIVE. partial_text is +// tokenizer_.decode(streaming_token_ids_), the whole +// hypothesis so far, and process_available_stream_chunks +// hands the SAME event to the sink AND returns it, so every +// partial arrives TWICE. +// +// Applying either convention to the other family corrupts the transcript: read +// a cumulative report as a delta and the client sees the transcript repeated on +// every event; read an incremental fragment as cumulative and the suffix +// arithmetic eats the front of it. So the tracker decides per fragment, from +// what it has already delivered, and the one rule it enforces is that TEXT THE +// CLIENT HAS ALREADY BEEN SENT IS NEVER SENT AGAIN. + +#include + +namespace audiocpp_backend { + +class TranscriptDeltaTracker { +public: + // Takes one StreamEvent::partial_text and returns the fragment to put on + // the wire, empty when there is nothing new. + // + // The rules, in order: + // 1. An empty partial says nothing. + // 2. A partial the assembly ALREADY STARTS WITH has been delivered: + // nothing is emitted. This is what absorbs voxtral's double delivery + // of every event, and a cumulative hypothesis that shrinks. + // 3. A partial that EXTENDS the assembly is a cumulative report: only its + // new suffix is emitted. + // 4. Anything else is an incremental fragment: it is emitted whole and + // appended. + // + // Rule 3 is the one judgement call, since a fragment that happens to begin + // with the entire transcript so far is indistinguishable from a cumulative + // report. It is read as cumulative because every cumulative family produces + // that shape on EVERY event, while an incremental family produces it only + // when one fragment repeats everything before it, which no tokenizer output + // does in practice. + std::string observe(const std::string &partial_text); + + // Reconciles against TaskResult::text_output, which is authoritative, and + // returns the fragment that makes appending every delta equal it. + // + // This is what makes the OFFLINE FALLBACK a single line rather than its own + // branch: with no partials observed, the assembly is empty and the whole + // final text comes back as one delta. + // + // A final text that CONTRADICTS what was already sent returns empty. A + // fragment on the wire cannot be retracted, so the alternative would be to + // send the transcript a second time and let the client hold it twice. + // final_result carries the authoritative text either way. + std::string reconcile(const std::string &final_text); + + // Everything the client has been sent, concatenated. + const std::string &assembled() const noexcept { return assembled_; } + +private: + std::string assembled_; +}; + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/stream_delta_test.cpp b/backend/cpp/audio-cpp/stream_delta_test.cpp new file mode 100644 index 000000000..07cd1df2f --- /dev/null +++ b/backend/cpp/audio-cpp/stream_delta_test.cpp @@ -0,0 +1,186 @@ +// Unit tests for stream_delta. Standard library only. The harness compiles this +// as a single translation unit, so the implementation is included directly. +// +// The traces below are transcribed from the pinned upstream sessions rather +// than invented, because the whole reason this unit exists is that the four +// streaming ASR families do NOT agree on what partial_text means. + +#include "stream_delta.cpp" + +#include +#include +#include + +static int failures = 0; + +static void check(bool ok, const std::string &name) { + if (!ok) { + failures++; + fprintf(stderr, "FAIL: %s\n", name.c_str()); + } else { + fprintf(stderr, "ok: %s\n", name.c_str()); + } +} + +static void check_eq(const std::string &got, const std::string &want, + const std::string &name) { + check(got == want, name + " (got \"" + got + "\" want \"" + want + "\")"); +} + +using audiocpp_backend::TranscriptDeltaTracker; + +// Feeds a whole trace and returns what a client appending every emitted +// fragment would end up holding, which is the only property that matters. +static std::string client_view(TranscriptDeltaTracker &tracker, + const std::vector &partials, + std::vector *emitted = nullptr) { + std::string view; + for (const auto &partial : partials) { + const std::string fragment = tracker.observe(partial); + if (emitted != nullptr && !fragment.empty()) { + emitted->push_back(fragment); + } + view += fragment; + } + return view; +} + +// nemotron_asr: decoder.cpp emits current_text.substr(emitted_text.size()) on +// every non-blank token, i.e. INCREMENTAL fragments, through the stream event +// sink during finalize(). +static void test_incremental_family() { + TranscriptDeltaTracker tracker; + std::vector emitted; + const std::string view = + client_view(tracker, {"Local", " AI", " now", " speaks."}, &emitted); + check_eq(view, "Local AI now speaks.", "incremental deltas concatenate"); + check(emitted.size() == 4, "incremental family emits one fragment per partial"); + check_eq(tracker.assembled(), "Local AI now speaks.", + "incremental family assembles the whole transcript"); + check_eq(tracker.reconcile("Local AI now speaks."), "", + "a final result the deltas already cover adds nothing"); +} + +// voxtral_realtime: process_one_stream_chunk sets partial_text to +// tokenizer_.decode(streaming_token_ids_), the WHOLE accumulated hypothesis, +// and process_available_stream_chunks then hands the SAME event to both the +// sink and the caller, so every partial arrives twice. +static void test_cumulative_family_with_duplicate_delivery() { + TranscriptDeltaTracker tracker; + std::vector emitted; + const std::string view = client_view( + tracker, {"Local", "Local", "Local AI", "Local AI", "Local AI now", + "Local AI now"}, + &emitted); + check_eq(view, "Local AI now", "cumulative partials are not repeated to the client"); + check(emitted.size() == 3, + "the duplicate delivery of each cumulative event emits nothing twice"); + check_eq(emitted.empty() ? "" : emitted[0], "Local", "first cumulative fragment"); + check_eq(emitted.size() < 2 ? "" : emitted[1], " AI", "second cumulative fragment"); + check_eq(emitted.size() < 3 ? "" : emitted[2], " now", "third cumulative fragment"); +} + +// The rule that separates the two: text the client has ALREADY been sent is +// never sent again, whatever convention produced it. +static void test_already_delivered_text_is_never_resent() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("hello world"), "hello world", "first fragment"); + check_eq(tracker.observe("hello world"), "", "an identical repeat emits nothing"); + check_eq(tracker.observe("hello"), "", + "a shortened hypothesis emits nothing rather than duplicating a prefix"); + check_eq(tracker.assembled(), "hello world", + "a shortened hypothesis does not shrink what the client holds"); +} + +static void test_empty_partials_are_ignored() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe(""), "", "an empty partial emits nothing"); + check_eq(tracker.observe("a"), "a", "a real partial after an empty one still emits"); + check_eq(tracker.observe(""), "", "a later empty partial emits nothing"); + check_eq(tracker.assembled(), "a", "empty partials do not disturb the assembly"); +} + +// The offline fallback: a family with no streaming ASR runs once, so nothing is +// ever observed and the reconciliation IS the single delta the RPC promises. +static void test_offline_fallback_is_one_delta() { + TranscriptDeltaTracker tracker; + check_eq(tracker.reconcile("the whole transcript"), "the whole transcript", + "with no partials the final text is emitted whole"); + check_eq(tracker.assembled(), "the whole transcript", + "the reconciliation is recorded as delivered"); + check_eq(tracker.reconcile("the whole transcript"), "", + "reconciling twice does not duplicate"); +} + +// A streaming family whose partials stopped short of the final text: the tail +// is emitted so that appending every delta still equals final_result.text. +static void test_reconcile_emits_the_tail() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("Local AI"), "Local AI", "partial arrives"); + check_eq(tracker.reconcile("Local AI now speaks."), " now speaks.", + "the untold tail of the final text is emitted"); + check_eq(tracker.assembled(), "Local AI now speaks.", "tail is recorded"); +} + +// Divergence. nemotron's decoder has a rewrite branch: when the new hypothesis +// is NOT an extension of what it already emitted, it emits the whole new text. +// Nothing can retract a fragment already written to the wire, so the tracker +// must not try: it emits nothing further and leaves final_result authoritative. +static void test_divergent_final_text_is_not_appended() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("the cat"), "the cat", "first hypothesis"); + check_eq(tracker.reconcile("the dog"), "", + "a final text that contradicts the deltas is not appended to them"); + check_eq(tracker.assembled(), "the cat", + "a contradicted assembly is left as it was actually sent"); +} + +static void test_empty_final_text() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("something"), "something", "partial arrives"); + check_eq(tracker.reconcile(""), "", "an empty final text emits nothing"); + check_eq(tracker.assembled(), "something", "an empty final text changes nothing"); +} + +// A whitespace-only fragment is real text: the space between two words is +// exactly what an incremental family delivers on its own. +static void test_whitespace_fragments_survive() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("one"), "one", "word"); + check_eq(tracker.observe(" "), " ", "a bare separator is emitted"); + check_eq(tracker.observe("two"), "two", "next word"); + check_eq(tracker.assembled(), "one two", "separator is kept in the assembly"); +} + +// The ambiguity this unit cannot resolve, pinned so that a future reader sees +// the choice rather than rediscovering it: a fragment that EXTENDS everything +// delivered so far is read as a cumulative report, because that is what every +// cumulative family produces on every event, while an incremental family +// producing one is the rare coincidence of a fragment repeating the whole +// transcript so far. +static void test_prefix_extension_is_read_as_cumulative() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("I"), "I", "first fragment"); + check_eq(tracker.observe("I'm"), "'m", + "a fragment extending the assembly is treated as a cumulative report"); + check_eq(tracker.assembled(), "I'm", "cumulative reading assembles once"); +} + +int main() { + test_incremental_family(); + test_cumulative_family_with_duplicate_delivery(); + test_already_delivered_text_is_never_resent(); + test_empty_partials_are_ignored(); + test_offline_fallback_is_one_delta(); + test_reconcile_emits_the_tail(); + test_divergent_final_text_is_not_appended(); + test_empty_final_text(); + test_whitespace_fragments_survive(); + test_prefix_extension_is_read_as_cumulative(); + if (failures) { + fprintf(stderr, "%d check(s) failed\n", failures); + return 1; + } + fprintf(stderr, "all stream_delta checks passed\n"); + return 0; +} diff --git a/backend/cpp/audio-cpp/streaming_driver_ctest.cpp b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp new file mode 100644 index 000000000..6fc225ed0 --- /dev/null +++ b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp @@ -0,0 +1,630 @@ +// Tests for the streaming drivers in loaded_model: begin_stream, +// run_streaming_pull and run_streaming_audio. +// +// Engine-linked, so this runs through ctest rather than +// backend/cpp/run-unit-tests.sh. It builds no model and loads no file: a +// LoadedModel::Session is a plain struct holding a pointer to an engine +// interface, so a fake session exercises the drivers directly, which is the +// only way to assert the STATE OBLIGATION (prepare, then start_stream, on every +// stream) without a GPU and a gigabyte of weights. + +#include "inference_lane.h" +#include "loaded_model.h" + +#include "engine/framework/runtime/session.h" + +#include +#include +#include +#include +#include +#include + +namespace rt = engine::runtime; + +static int failures = 0; + +static void check(bool ok, const std::string &name) { + if (!ok) { + failures++; + fprintf(stderr, "FAIL: %s\n", name.c_str()); + } else { + fprintf(stderr, "ok: %s\n", name.c_str()); + } +} + +static void check_eq(const std::string &got, const std::string &want, + const std::string &name) { + check(got == want, name + " (got \"" + got + "\" want \"" + want + "\")"); +} + +static std::string join(const std::vector &parts) { + std::string out; + for (const auto &part : parts) { + if (!out.empty()) { + out += "|"; + } + out += part; + } + return out; +} + +// -------------------------------------------------------------------------- +// Fakes +// -------------------------------------------------------------------------- + +// Consumes audio chunks, like every streaming ASR family. +// +// It deliberately does NOT override start_stream, so every test that drives it +// also pins the claim the drivers rely on: IStreamingVoiceTaskSession's BASE +// start_stream is a call to reset(). If upstream ever changes that base, the +// replay test below fails rather than the backend silently continuing the +// previous stream. +class FakeAudioSession : public rt::IStreamingVoiceTaskSession { +public: + std::vector calls; + std::vector chunks; + rt::StreamingPolicy policy; + // Set to make process_audio_chunk throw on the nth call (1-based). + int throw_on_chunk = 0; + // Cumulative partial text, which is voxtral_realtime's convention. + bool report_partials = true; + + std::string family() const override { return "fake_audio"; } + rt::VoiceTaskKind task_kind() const override { return rt::VoiceTaskKind::Asr; } + rt::RunMode run_mode() const override { return rt::RunMode::Streaming; } + + void prepare(const rt::SessionPreparationRequest &request) override { + calls.push_back("prepare"); + prepared_ = true; + prepared_rate_ = request.audio.has_value() ? request.audio->sample_rate : 0; + } + + rt::StreamingPolicy streaming_policy() const override { return policy; } + + void set_stream_event_sink(rt::StreamEventCallback sink) override { + calls.push_back(sink ? "sink+" : "sink-"); + sink_ = std::move(sink); + } + + void reset() override { + if (!prepared_) { + // Exactly what silero_vad does, and the reason prepare() has to come + // first rather than being folded into session_for. + throw std::runtime_error("fake: prepare() must be called before reset()"); + } + calls.push_back("reset"); + seen_frames_ = 0; + seen_chunks_ = 0; + text_.clear(); + } + + rt::StreamEvent process_audio_chunk(const rt::AudioChunk &chunk) override { + calls.push_back("chunk"); + chunks.push_back(chunk); + ++seen_chunks_; + if (throw_on_chunk == seen_chunks_) { + throw std::runtime_error("fake: chunk failure"); + } + const int channels = chunk.channels > 0 ? chunk.channels : 1; + seen_frames_ += static_cast(chunk.samples.size()) / channels; + text_ += "w" + std::to_string(seen_chunks_); + rt::StreamEvent event; + if (report_partials) { + event.partial_text = rt::Transcript{text_, "en"}; + } + return event; + } + + rt::TaskResult finalize() override { + calls.push_back("finalize"); + rt::TaskResult result; + result.text_output = rt::Transcript{ + text_ + "/frames=" + std::to_string(seen_frames_), "en"}; + return result; + } + + // Emits through the SINK the way nemotron_asr does, from inside the final + // step rather than from process_audio_chunk. + void emit_through_sink(const std::string &fragment) { + if (!sink_) { + return; + } + rt::StreamEvent event; + event.partial_text = rt::Transcript{fragment, "en"}; + sink_(event); + } + + bool sink_installed() const { return static_cast(sink_); } + int prepared_rate() const { return prepared_rate_; } + +private: + rt::StreamEventCallback sink_; + bool prepared_ = false; + int prepared_rate_ = 0; + std::int64_t seen_frames_ = 0; + int seen_chunks_ = 0; + std::string text_; +}; + +// nemotron_asr's shape: partials arrive only through the sink, and only from +// inside the finalize step. +class SinkOnlyAudioSession : public FakeAudioSession { +public: + SinkOnlyAudioSession() { report_partials = false; } + + rt::TaskResult finalize() override { + emit_through_sink("late "); + emit_through_sink("partial"); + return FakeAudioSession::finalize(); + } +}; + +// Pulls events, like every streaming TTS family. Overrides start_stream the way +// the seven real families do, calling reset() first. +class FakePullSession : public rt::IStreamingVoiceTaskSession { +public: + std::vector calls; + std::size_t event_count = 3; + bool final_on_second = false; + + std::string family() const override { return "fake_pull"; } + rt::VoiceTaskKind task_kind() const override { return rt::VoiceTaskKind::Tts; } + rt::RunMode run_mode() const override { return rt::RunMode::Streaming; } + + void prepare(const rt::SessionPreparationRequest &) override { + calls.push_back("prepare"); + prepared_ = true; + } + + rt::StreamingPolicy streaming_policy() const override { + rt::StreamingPolicy policy; + policy.input = rt::StreamingInputKind::None; + policy.output = rt::StreamingOutputKind::PullEvents; + return policy; + } + + void start_stream(const rt::TaskRequest &request) override { + calls.push_back("start_stream"); + (void)request; + reset(); + } + + void set_stream_event_sink(rt::StreamEventCallback sink) override { + calls.push_back(sink ? "sink+" : "sink-"); + sink_ = std::move(sink); + } + + void reset() override { + if (!prepared_) { + throw std::runtime_error("fake: prepare() must be called before reset()"); + } + calls.push_back("reset"); + emitted_ = 0; + } + + std::optional next_stream_event() override { + if (emitted_ >= event_count) { + return std::nullopt; + } + rt::StreamEvent event; + rt::AudioBuffer audio; + audio.sample_rate = 24000; + audio.channels = 1; + audio.samples.assign(4, 0.25F); + // named_audio_outputs, NOT audio_output: this is where supertonic, + // omnivoice and voxcpm2 all put their streamed chunks. + event.named_audio_outputs.push_back( + {"chunk_" + std::to_string(emitted_), std::move(audio), {}}); + ++emitted_; + if (final_on_second && emitted_ == 2) { + event.is_final = true; + } + calls.push_back("pull"); + return event; + } + + rt::StreamEvent process_audio_chunk(const rt::AudioChunk &) override { + throw std::runtime_error("fake_pull consumes no audio"); + } + + rt::TaskResult finalize() override { + calls.push_back("finalize"); + rt::TaskResult result; + rt::AudioBuffer merged; + merged.sample_rate = 24000; + merged.channels = 1; + merged.samples.assign(4 * emitted_, 0.25F); + result.audio_output = std::move(merged); + return result; + } + + bool sink_installed() const { return static_cast(sink_); } + +private: + rt::StreamEventCallback sink_; + bool prepared_ = false; + std::size_t emitted_ = 0; +}; + +// -------------------------------------------------------------------------- +// Helpers +// -------------------------------------------------------------------------- + +static audiocpp_backend::LoadedModel::Session +streaming_session(rt::IStreamingVoiceTaskSession &fake, + audiocpp_backend::Task task) { + audiocpp_backend::LoadedModel::Session session; + session.task = task; + session.mode = audiocpp_backend::Mode::Streaming; + session.streaming = &fake; + return session; +} + +static rt::TaskRequest audio_request(int sample_rate, int channels, + std::int64_t frames) { + rt::TaskRequest request; + rt::AudioBuffer audio; + audio.sample_rate = sample_rate; + audio.channels = channels; + audio.samples.assign(static_cast(frames * channels), 0.5F); + request.audio_input = std::move(audio); + return request; +} + +// -------------------------------------------------------------------------- +// Tests +// -------------------------------------------------------------------------- + +static void test_begin_stream_prepares_then_starts() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakePullSession fake; + const auto session = streaming_session(fake, audiocpp_backend::Task::Tts); + + rt::TaskRequest request; + audiocpp_backend::begin_stream(session, request, entry); + + check_eq(join(fake.calls), "prepare|start_stream|reset", + "begin_stream prepares before it starts, and start_stream resets"); +} + +// The base implementation of start_stream IS a reset(). FakeAudioSession does +// not override start_stream, so this is that guarantee, read out of the pinned +// header rather than assumed. +static void test_base_start_stream_resets() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + + audiocpp_backend::begin_stream(session, audio_request(16000, 1, 10), entry); + check_eq(join(fake.calls), "prepare|reset", + "the interface's own start_stream resets the session"); +} + +static void test_begin_stream_refuses_a_non_streaming_session() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + audiocpp_backend::LoadedModel::Session session; + session.mode = audiocpp_backend::Mode::Offline; + + bool threw_capability = false; + try { + rt::TaskRequest request; + audiocpp_backend::begin_stream(session, request, entry); + } catch (const audiocpp_backend::CapabilityError &) { + threw_capability = true; + } catch (const std::exception &) { + } + check(threw_capability, + "begin_stream on an offline session throws CapabilityError, not a null deref"); +} + +// THE ONE THIS TASK IS ABOUT. A streaming session is cached, so the second +// stream gets the object the first one left behind. Two identical runs against +// the SAME session must produce identical output. +static void test_a_refetched_session_replays_identically() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 512; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(16000, 1, 1536); + + std::vector first_fragments; + const auto first = audiocpp_backend::run_streaming_audio( + session, request, *request.audio_input, + [&](const rt::StreamEvent &event) { + if (event.partial_text.has_value()) { + first_fragments.push_back(event.partial_text->text); + } + }, + entry); + + std::vector second_fragments; + const auto second = audiocpp_backend::run_streaming_audio( + session, request, *request.audio_input, + [&](const rt::StreamEvent &event) { + if (event.partial_text.has_value()) { + second_fragments.push_back(event.partial_text->text); + } + }, + entry); + + check_eq(join(second_fragments), join(first_fragments), + "a re-fetched streaming session replays the same partials"); + check_eq(second.text_output.has_value() ? second.text_output->text : "", + first.text_output.has_value() ? first.text_output->text : "", + "a re-fetched streaming session replays the same final text"); + check_eq(first.text_output.has_value() ? first.text_output->text : "", + "w1w2w3/frames=1536", + "the first run saw exactly the audio it was given"); + // Not a tautology: without the reset the second run reports six words and + // 3072 frames, and both checks above fail. + check_eq(join(first_fragments), "w1|w1w2|w1w2w3", "cumulative partials"); +} + +static void test_run_streaming_audio_installs_and_clears_the_sink() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + SinkOnlyAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 1024; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(16000, 1, 1024); + + std::vector fragments; + const auto result = audiocpp_backend::run_streaming_audio( + session, request, *request.audio_input, + [&](const rt::StreamEvent &event) { + if (event.partial_text.has_value()) { + fragments.push_back(event.partial_text->text); + } + }, + entry); + + check_eq(join(fragments), "late |partial", + "a family that reports only through the sink is not silent"); + check(!fake.sink_installed(), + "the sink is cleared before returning, so the cached session holds no " + "reference to the caller's frame"); + check_eq(join(fake.calls), "sink+|prepare|reset|chunk|finalize|sink-", + "the sink is installed before the stream begins and cleared after it ends"); + check(result.text_output.has_value(), "the final result still comes back"); +} + +static void test_the_sink_is_cleared_when_the_stream_throws() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 512; + fake.throw_on_chunk = 1; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(16000, 1, 1024); + + bool threw = false; + try { + audiocpp_backend::run_streaming_audio( + session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + } catch (const std::exception &) { + threw = true; + } + check(threw, "a failing chunk propagates"); + check(!fake.sink_installed(), + "the sink is cleared on the exception path too"); +} + +static void test_chunking_honours_the_policy_sample_count() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 16000; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + // 2.5 chunks, so the last one is short. + const auto request = audio_request(16000, 1, 40000); + + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 3, "40000 frames at 16000 per chunk is three chunks"); + if (fake.chunks.size() == 3) { + check(fake.chunks[0].samples.size() == 16000, "first chunk is full"); + check(fake.chunks[1].samples.size() == 16000, "second chunk is full"); + check(fake.chunks[2].samples.size() == 8000, "last chunk is the remainder"); + check(fake.chunks[0].start_sample == 0, "first chunk starts at zero"); + check(fake.chunks[1].start_sample == 16000, "second chunk start index"); + check(fake.chunks[2].start_sample == 32000, "third chunk start index"); + check(fake.chunks[0].sample_rate == 16000, "chunk carries the buffer's rate"); + check(fake.chunks[0].channels == 1, "chunk carries the buffer's channel count"); + } +} + +// higgs_audio_stt states its window in seconds and leaves the sample count at +// zero, so this branch is a real family's path rather than a defensive one. +static void test_chunking_falls_back_to_the_policy_seconds() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 0; + fake.policy.preferred_audio_chunk_seconds = 4.0; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(16000, 1, 96000); // 6 s + + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 2, "6 s at a 4 s window is two chunks"); + if (fake.chunks.size() == 2) { + check(fake.chunks[0].samples.size() == 64000, "first window is 4 s"); + check(fake.chunks[1].samples.size() == 32000, "second window is the 2 s remainder"); + } +} + +static void test_chunking_falls_back_to_the_interface_default() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 0; + fake.policy.preferred_audio_chunk_seconds = 0.0; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(16000, 1, 1024); + + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 2, "a policy naming no window uses the interface's 512"); + if (!fake.chunks.empty()) { + check(fake.chunks[0].samples.size() == 512, "default window is 512 frames"); + } +} + +// A zero sample rate must not turn a seconds-only policy into a zero-length +// chunk, which would loop forever. +static void test_a_seconds_policy_with_no_rate_falls_through() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 0; + fake.policy.preferred_audio_chunk_seconds = 4.0; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(0, 1, 1024); + + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + check(fake.chunks.size() == 2, "a rateless buffer still chunks at the default 512"); +} + +// FRAMES, not floats. vibevoice_asr refuses a chunk whose sample count is not +// divisible by its channel count, and offsets every span it reports by the +// chunk's start_sample. +static void test_stereo_chunks_are_frame_aligned() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 300; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + const auto request = audio_request(48000, 2, 750); + + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 3, "750 frames at 300 frames per chunk is three chunks"); + for (const auto &chunk : fake.chunks) { + check(chunk.samples.size() % 2 == 0, "every stereo chunk is a whole number of frames"); + } + if (fake.chunks.size() == 3) { + check(fake.chunks[0].samples.size() == 600, "300 stereo frames is 600 floats"); + check(fake.chunks[1].start_sample == 300, + "start_sample counts frames, not floats"); + check(fake.chunks[2].samples.size() == 300, "the remainder is 150 frames"); + } +} + +static void test_pull_drains_every_event_and_installs_no_sink() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakePullSession fake; + fake.event_count = 3; + const auto session = streaming_session(fake, audiocpp_backend::Task::Tts); + + std::vector ids; + rt::TaskRequest request; + const auto result = audiocpp_backend::run_streaming_pull( + session, request, + [&](const rt::StreamEvent &event) { + for (const auto &named : event.named_audio_outputs) { + ids.push_back(named.id); + } + }, + entry); + + check_eq(join(ids), "chunk_0|chunk_1|chunk_2", "every pulled event reaches the caller"); + check(!fake.sink_installed(), + "no stream event sink is installed on the pull path, so voxcpm2 cannot " + "deliver every chunk twice"); + check_eq(join(fake.calls), "prepare|start_stream|reset|pull|pull|pull|finalize", + "prepare, start, drain, finish"); + check(result.audio_output.has_value(), "the merged result comes back"); + check(result.audio_output.has_value() && result.audio_output->samples.size() == 12, + "the merged result is the whole synthesis, not a tail"); +} + +static void test_pull_stops_on_a_final_event() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakePullSession fake; + fake.event_count = 5; + fake.final_on_second = true; + const auto session = streaming_session(fake, audiocpp_backend::Task::Tts); + + int events = 0; + rt::TaskRequest request; + audiocpp_backend::run_streaming_pull( + session, request, [&](const rt::StreamEvent &) { ++events; }, entry); + + check(events == 2, "an event marked final ends the pull loop"); +} + +static void test_pull_refuses_a_non_streaming_session() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + audiocpp_backend::LoadedModel::Session session; + + bool threw_capability = false; + try { + rt::TaskRequest request; + audiocpp_backend::run_streaming_pull( + session, request, [](const rt::StreamEvent &) {}, entry); + } catch (const audiocpp_backend::CapabilityError &) { + threw_capability = true; + } catch (const std::exception &) { + } + check(threw_capability, "run_streaming_pull refuses a session with no streaming half"); +} + +// prepare() runs on EVERY stream, not once per session: the preparation request +// is derived from the request (audio contract, text, voice), so a second stream +// at a different rate would otherwise run against the first one's contract. +static void test_prepare_tracks_the_request_not_the_session() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 4096; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + + const auto first = audio_request(16000, 1, 4096); + audiocpp_backend::run_streaming_audio(session, first, *first.audio_input, + [](const rt::StreamEvent &) {}, entry); + check(fake.prepared_rate() == 16000, "the first stream prepares at its own rate"); + + const auto second = audio_request(44100, 1, 4096); + audiocpp_backend::run_streaming_audio(session, second, *second.audio_input, + [](const rt::StreamEvent &) {}, entry); + check(fake.prepared_rate() == 44100, + "the second stream prepares at ITS rate, not the first one's"); +} + +int main() { + test_begin_stream_prepares_then_starts(); + test_base_start_stream_resets(); + test_begin_stream_refuses_a_non_streaming_session(); + test_a_refetched_session_replays_identically(); + test_run_streaming_audio_installs_and_clears_the_sink(); + test_the_sink_is_cleared_when_the_stream_throws(); + test_chunking_honours_the_policy_sample_count(); + test_chunking_falls_back_to_the_policy_seconds(); + test_chunking_falls_back_to_the_interface_default(); + test_a_seconds_policy_with_no_rate_falls_through(); + test_stereo_chunks_are_frame_aligned(); + test_pull_drains_every_event_and_installs_no_sink(); + test_pull_stops_on_a_final_event(); + test_pull_refuses_a_non_streaming_session(); + test_prepare_tracks_the_request_not_the_session(); + if (failures) { + fprintf(stderr, "%d check(s) failed\n", failures); + return 1; + } + fprintf(stderr, "all streaming driver checks passed\n"); + return 0; +} diff --git a/backend/cpp/audio-cpp/wav_header.cpp b/backend/cpp/audio-cpp/wav_header.cpp new file mode 100644 index 000000000..9fb4fb4f9 --- /dev/null +++ b/backend/cpp/audio-cpp/wav_header.cpp @@ -0,0 +1,63 @@ +#include "wav_header.h" + +#include +#include + +namespace audiocpp_backend { +namespace { + +void append_u32(std::string &out, std::uint32_t value) { + out.push_back(static_cast(value & 0xFF)); + out.push_back(static_cast((value >> 8) & 0xFF)); + out.push_back(static_cast((value >> 16) & 0xFF)); + out.push_back(static_cast((value >> 24) & 0xFF)); +} + +void append_u16(std::string &out, std::uint16_t value) { + out.push_back(static_cast(value & 0xFF)); + out.push_back(static_cast((value >> 8) & 0xFF)); +} + +// The unknown-length sentinel, in both the RIFF and the data chunk size. +constexpr std::uint32_t kStreamingSize = 0xFFFFFFFFu; +constexpr std::uint16_t kBitsPerSample = 16; +constexpr std::uint32_t kPcmFmtChunkSize = 16; +constexpr std::uint16_t kFormatTagPcm = 1; +constexpr int kMaxChannels = 65535; + +} // namespace + +std::string streaming_wav_header(int sample_rate, int channels) { + int clamped_channels = channels > 0 ? channels : 1; + if (clamped_channels > kMaxChannels) { + clamped_channels = kMaxChannels; + } + const auto channel_count = static_cast(clamped_channels); + const auto rate = static_cast(sample_rate > 0 ? sample_rate : 0); + const auto block_align = + static_cast(channel_count * (kBitsPerSample / 8)); + // uint32 arithmetic on purpose: 384 kHz by 8 channels is 6.1 MB/s, which + // does not fit the uint16 block align it is derived from. + const std::uint32_t byte_rate = rate * static_cast(block_align); + static_assert(std::numeric_limits::max() >= 0xFFFFFFFFu, + "the streaming sentinel must be representable"); + + std::string header; + header.reserve(44); + header += "RIFF"; + append_u32(header, kStreamingSize); // unknown total length + header += "WAVE"; + header += "fmt "; + append_u32(header, kPcmFmtChunkSize); + append_u16(header, kFormatTagPcm); + append_u16(header, channel_count); + append_u32(header, rate); + append_u32(header, byte_rate); + append_u16(header, block_align); + append_u16(header, kBitsPerSample); + header += "data"; + append_u32(header, kStreamingSize); // unknown payload length + return header; +} + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/wav_header.h b/backend/cpp/audio-cpp/wav_header.h new file mode 100644 index 000000000..42dff40b6 --- /dev/null +++ b/backend/cpp/audio-cpp/wav_header.h @@ -0,0 +1,39 @@ +#pragma once + +// Builds the 44 byte canonical WAV header that precedes a streamed PCM body. +// Standard library only. +// +// TTSStream chunks travel in Reply.audio and the FIRST chunk must be this +// header, or an HTTP client has no format to decode the PCM with. Because the +// total length is unknown while the model is still generating, both size fields +// carry 0xFFFFFFFF; that is the convention backend/go/vibevoice-cpp established +// (govibevoicecpp.go, TTSStream) and the one core/backend/tts.go writes when it +// synthesises a header itself. +// +// WHY THIS BACKEND SENDS THE HEADER RATHER THAN LETTING GO DO IT. +// core/backend/tts.go's ModelTTSStream will emit a header of its own, but only +// when the FIRST Reply carries a non-empty `message` field holding a JSON blob +// with a sample_rate. This backend sends audio and never sets `message`, so +// that branch never fires and there is exactly one header on the wire: this +// one. Do not start setting Reply.message on this RPC without deleting the +// header below, or every stream gains a second header 44 bytes into the PCM. +// +// The header this produces is byte-identical to the one pkg/audio.WAVHeader +// serialises, which is what core/http/endpoints/openai/realtime_model.go +// assumes when it reads the sample rate out of byte offset 24 of the first +// callback. + +#include + +namespace audiocpp_backend { + +// 16-bit PCM, little endian, interleaved. +// +// `channels` is clamped to at least 1 and at most 65535, so a garbage channel +// count can never write a zero block align, which is what a reader divides the +// data size by. `sample_rate` is clamped to at least 0 rather than wrapped: +// a zero rate is visibly wrong to whoever reads the header, whereas the +// 4294967295 an unsigned conversion of -1 would write looks like a real field. +std::string streaming_wav_header(int sample_rate, int channels); + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/wav_header_test.cpp b/backend/cpp/audio-cpp/wav_header_test.cpp new file mode 100644 index 000000000..f86ddfe00 --- /dev/null +++ b/backend/cpp/audio-cpp/wav_header_test.cpp @@ -0,0 +1,163 @@ +// Unit tests for wav_header. Standard library only. The harness compiles this +// as a single translation unit, so the implementation is included directly. + +#include "wav_header.cpp" + +#include +#include +#include + +static int failures = 0; + +static void check(bool ok, const std::string &name) { + if (!ok) { + failures++; + fprintf(stderr, "FAIL: %s\n", name.c_str()); + } else { + fprintf(stderr, "ok: %s\n", name.c_str()); + } +} + +using audiocpp_backend::streaming_wav_header; + +static std::uint32_t read_u32(const std::string &data, size_t offset) { + return static_cast(static_cast(data[offset])) | + (static_cast(static_cast(data[offset + 1])) << 8) | + (static_cast(static_cast(data[offset + 2])) << 16) | + (static_cast(static_cast(data[offset + 3])) << 24); +} + +static std::uint16_t read_u16(const std::string &data, size_t offset) { + return static_cast( + static_cast(static_cast(data[offset])) | + (static_cast(static_cast(data[offset + 1])) << 8)); +} + +static std::string to_hex(const std::string &data) { + static const char *digits = "0123456789abcdef"; + std::string out; + out.reserve(data.size() * 2); + for (const char byte : data) { + const auto value = static_cast(byte); + out.push_back(digits[value >> 4]); + out.push_back(digits[value & 0x0F]); + } + return out; +} + +static void test_header_layout() { + const std::string header = streaming_wav_header(24000, 1); + + check(header.size() == 44, "canonical 44 byte header"); + check(header.compare(0, 4, "RIFF") == 0, "RIFF magic"); + check(header.compare(8, 4, "WAVE") == 0, "WAVE magic"); + check(header.compare(12, 4, "fmt ") == 0, "fmt chunk id"); + check(header.compare(36, 4, "data") == 0, "data chunk id"); + + check(read_u32(header, 16) == 16, "PCM fmt chunk is 16 bytes"); + check(read_u16(header, 20) == 1, "format tag 1 is PCM"); + check(read_u16(header, 22) == 1, "mono channel count"); + check(read_u32(header, 24) == 24000, "sample rate"); + // byte rate = rate * channels * bytes per sample + check(read_u32(header, 28) == 24000 * 1 * 2, "byte rate"); + check(read_u16(header, 32) == 2, "block align for mono 16 bit"); + check(read_u16(header, 34) == 16, "16 bits per sample"); +} + +// The field-wise checks above can all pass while the fields sit in the wrong +// ORDER, since several of them hold the same value. This pins the whole 44 byte +// string against a literal transcribed from the layout in pkg/audio/audio.go, +// which is the struct binary.Write serializes for every Go LocalAI backend. +// Independent of the implementation: it was written out by hand rather than +// captured from a run. +static void test_exact_bytes_match_the_go_layout() { + const std::string expected = + "52494646" // "RIFF" + "ffffffff" // chunk size: streaming sentinel + "57415645" // "WAVE" + "666d7420" // "fmt " + "10000000" // subchunk1 size 16 + "0100" // audio format 1 (PCM) + "0100" // channels 1 + "c05d0000" // sample rate 24000 + "80bb0000" // byte rate 48000 + "0200" // block align 2 + "1000" // bits per sample 16 + "64617461" // "data" + "ffffffff"; // subchunk2 size: streaming sentinel + check(to_hex(streaming_wav_header(24000, 1)) == expected, + "byte for byte match with the canonical mono 24 kHz header"); +} + +// The whole point: a streaming header cannot know the final length, so both +// size fields are the sentinel. A client that sees a real size stops early. +static void test_streaming_sentinels() { + const std::string header = streaming_wav_header(16000, 1); + check(read_u32(header, 4) == 0xFFFFFFFFu, "RIFF chunk size is the sentinel"); + check(read_u32(header, 40) == 0xFFFFFFFFu, "data chunk size is the sentinel"); +} + +static void test_stereo() { + const std::string header = streaming_wav_header(44100, 2); + check(read_u16(header, 22) == 2, "stereo channel count"); + check(read_u32(header, 28) == 44100 * 2 * 2, "stereo byte rate"); + check(read_u16(header, 32) == 4, "block align for stereo 16 bit"); + check(header.size() == 44, "stereo header is still 44 bytes"); +} + +static void test_degenerate_inputs() { + // A zero or negative channel count must not produce a header that divides + // by zero downstream; clamp to mono. + const std::string zero_channels = streaming_wav_header(16000, 0); + check(read_u16(zero_channels, 22) == 1, "zero channels clamps to mono"); + check(read_u16(zero_channels, 32) == 2, "zero channels still block aligns as mono"); + check(read_u32(zero_channels, 28) == 32000, "zero channels byte rate is the mono one"); + + const std::string negative_channels = streaming_wav_header(16000, -3); + check(read_u16(negative_channels, 22) == 1, "negative channels clamps to mono"); + + // A channel count past the field's range must saturate rather than wrap: + // 65536 truncated to uint16 is 0, and a zero channel count writes a zero + // block align, which is what a reader divides the data size by. + const std::string too_many = streaming_wav_header(16000, 65536); + check(read_u16(too_many, 22) == 65535, "an out of range channel count saturates"); + check(read_u16(too_many, 32) != 0, "an out of range channel count never writes a zero block align"); + + // A non-positive rate is written as zero rather than wrapping through the + // unsigned conversion: 0 is visibly wrong to whoever reads the header, + // 4294967295 looks like a plausible field nobody checks. + const std::string zero_rate = streaming_wav_header(0, 1); + check(read_u32(zero_rate, 24) == 0, "zero sample rate stays zero"); + check(read_u32(zero_rate, 28) == 0, "zero sample rate yields a zero byte rate"); + const std::string negative_rate = streaming_wav_header(-48000, 1); + check(read_u32(negative_rate, 24) == 0, "negative sample rate is clamped to zero"); + check(read_u32(negative_rate, 28) == 0, "negative sample rate yields a zero byte rate"); + + // The sentinels are unconditional. A degenerate rate must not turn the + // stream into one a client thinks it can measure. + check(read_u32(zero_rate, 4) == 0xFFFFFFFFu, "degenerate input keeps the RIFF sentinel"); + check(read_u32(zero_rate, 40) == 0xFFFFFFFFu, "degenerate input keeps the data sentinel"); +} + +// Large but legal: 384 kHz 8 channel would overflow a 16 bit byte rate and +// must not overflow the 32 bit one either. +static void test_large_but_legal() { + const std::string header = streaming_wav_header(384000, 8); + check(read_u32(header, 28) == 384000u * 8u * 2u, "high rate multichannel byte rate"); + check(read_u16(header, 32) == 16, "high channel count block align"); +} + +int main() { + test_header_layout(); + test_exact_bytes_match_the_go_layout(); + test_streaming_sentinels(); + test_stereo(); + test_degenerate_inputs(); + test_large_but_legal(); + if (failures) { + fprintf(stderr, "%d check(s) failed\n", failures); + return 1; + } + fprintf(stderr, "all wav_header checks passed\n"); + return 0; +}