From 41b4e720418c64c73bf0386fdd177fc80438e8a4 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 26 Jul 2026 22:35:43 +0000 Subject: [PATCH] backend(audio-cpp): serve the TTSStream and AudioTranscriptionStream RPCs TTSStream leads with a streaming WAV header carrying 0xFFFFFFFF sizes, matching the convention backend/go/vibevoice-cpp established, so an HTTP client can start playback before the full PCM exists. Its chunks are read from StreamEvent::named_audio_outputs and not audio_output: supertonic, omnivoice and voxcpm2 all put their streamed audio there and leave audio_output empty until the very end, so reading the obvious field yields a stream with no audio in it. The finish_stream result is the family's own merged whole rather than a tail, so it is emitted only when nothing was streamed. Streaming transcription sends incremental deltas and degrades to a single delta plus the final result on families that offer no streaming ASR, which is the same message sequence with fewer deltas. The four streaming ASR families disagree on what partial_text means: nemotron_asr, vibevoice_asr and higgs_audio_stt report incremental fragments while voxtral_realtime reports the whole hypothesis and reports it twice, so the reconciliation lives in one tested unit rather than in the handler. nemotron_asr reports only through the stream event sink, and only from inside finalize, so the audio driver installs one and clears it again before returning: the session is cached and a sink left holding the caller's frame is a use after free waiting for the next stream. begin_stream is now the only implementation of the streaming state obligation, prepare then start_stream. Streaming sessions are cached, and what clears the previous stream is start_stream's reset; a family override that dropped it would break every call site with no compile error, so there is one call site. Assisted-by: Claude:claude-opus-5 [Claude Code] Signed-off-by: Ettore Di Giacinto --- backend/cpp/audio-cpp/CMakeLists.txt | 30 + backend/cpp/audio-cpp/grpc-server.cpp | 356 +++++++++- backend/cpp/audio-cpp/loaded_model.cpp | 140 ++++ backend/cpp/audio-cpp/loaded_model.h | 95 ++- backend/cpp/audio-cpp/stream_delta.cpp | 50 ++ backend/cpp/audio-cpp/stream_delta.h | 80 +++ backend/cpp/audio-cpp/stream_delta_test.cpp | 186 ++++++ .../cpp/audio-cpp/streaming_driver_ctest.cpp | 630 ++++++++++++++++++ backend/cpp/audio-cpp/wav_header.cpp | 63 ++ backend/cpp/audio-cpp/wav_header.h | 39 ++ backend/cpp/audio-cpp/wav_header_test.cpp | 163 +++++ 11 files changed, 1788 insertions(+), 44 deletions(-) create mode 100644 backend/cpp/audio-cpp/stream_delta.cpp create mode 100644 backend/cpp/audio-cpp/stream_delta.h create mode 100644 backend/cpp/audio-cpp/stream_delta_test.cpp create mode 100644 backend/cpp/audio-cpp/streaming_driver_ctest.cpp create mode 100644 backend/cpp/audio-cpp/wav_header.cpp create mode 100644 backend/cpp/audio-cpp/wav_header.h create mode 100644 backend/cpp/audio-cpp/wav_header_test.cpp 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; +}