From 8778600295ecd85301685c7311cfa92ea98b9710 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 27 Jul 2026 00:27:08 +0000 Subject: [PATCH] backend(audio-cpp): serve the AudioTranscriptionLive RPC The one bidirectional stream this backend serves. The client sends a TranscriptLiveConfig, then TranscriptLiveAudio frames; the server acknowledges with ready, emits deltas as the audio arrives, and sends final_result once the read side closes. There is no offline fallback: live transcription has to consume audio incrementally, so a family with no streaming ASR is refused rather than served a batch run, which is what this RPC's Streaming-only mode_candidates list already says. The driver is a new sibling of run_streaming_audio, run_streaming_live, because the audio does not exist yet: instead of slicing a buffer it pulls frames from the caller until the read side closes. It installs the same ScopedStreamSink in the same order, which is not optional, since nemotron_asr returns a bare event from process_audio_chunk and reports every partial through the sink from inside finalize(). It buffers the wire's frames up to the family's own preferred window rather than feeding whatever size the client's audio callback produced, and it does not call finish_stream at all when no audio arrived, because nemotron_asr throws "finalize requires streamed audio" and an empty transcript is the truthful answer to transcribing nothing. Three things the handler had to get right and one it cannot: - The audio contract. A live request carries no samples, but nemotron_asr's streaming prepare() throws without an audio contract, and build_preparation_request derives it from TaskRequest::audio_input, so that field is an EMPTY buffer holding only the rate and the channel count. - 16 kHz or a refusal. The families express their spans in their own 16 kHz feature domain whatever the input was, and live frames cannot be resampled on the way in the way a file can, so an 8 kHz session would return timestamps 2x off with a 200. core/backend hardcodes 16000 anyway. - A mid-stream Config is refused. backend.proto calls it a decoder reset, but deltas already on the wire cannot be retracted, so a reset would leave the final text contradicting the transcript the client assembled. Ignoring the message would hand a client that believes it reset the decoder a transcript that silently continues the audio it thought it discarded. - The stale-route identity check cannot run here: TranscriptLiveRequest carries no ModelIdentity in either arm of its oneof, so snapshot_for does not instantiate for it. snapshot_unchecked's comment now names that as a second legitimate class of caller and says the fix is a proto change. eou and eob stay false. They exist for cache-aware models that emit end-of-utterance and end-of-backchannel tokens; audio.cpp's StreamEvent has no equivalent signal, and a client uses eou to decide the speaker yielded the turn, so a guess inferred from silence cuts people off mid-sentence. The lane is held for the whole stream, which is as long as the user keeps talking: the streaming session is stateful and cached, so a concurrent run would interleave two callers' audio and corrupt both transcripts. Verified against nemotron_asr over a real connection with a 14 s WAV in 512-sample frames: ready first, 59 incremental deltas with no repeated prefix, concat(deltas) equal to final_result.text, word timestamps in nanoseconds, eou and eob false. citrinet_asr answers UNIMPLEMENTED naming the family and listing asr/offline. A config followed by a close returns an empty final_result rather than hanging, and a first message that is not a config is INVALID_ARGUMENT. Two concurrent streams both return the complete transcript. Two cleanups on lines Task 12 touched, folded in. The DtypeAllowList terminator is now asserted at compile time: the reported out-of-bounds read did not exist, the single entry does terminate, but the loops have no other bound and any edit that widened an entry would walk off the end. And the dtype guard now short-circuits on "is there a table entry" through a new predicate rather than on the emptiness of the description string, which would have skipped the check on an entry with an empty allow list, i.e. on precisely the entry that refuses every dtype. Assisted-by: Claude:claude-opus-5 [Claude Code] Signed-off-by: Ettore Di Giacinto --- backend/cpp/audio-cpp/family_gate.cpp | 29 +- backend/cpp/audio-cpp/family_gate.h | 9 + backend/cpp/audio-cpp/family_gate_test.cpp | 16 +- backend/cpp/audio-cpp/grpc-server.cpp | 286 +++++++++++++++++- backend/cpp/audio-cpp/loaded_model.cpp | 109 ++++++- backend/cpp/audio-cpp/loaded_model.h | 36 +++ .../cpp/audio-cpp/streaming_driver_ctest.cpp | 237 ++++++++++++++- 7 files changed, 705 insertions(+), 17 deletions(-) diff --git a/backend/cpp/audio-cpp/family_gate.cpp b/backend/cpp/audio-cpp/family_gate.cpp index 8f9ed4b29..83ccd42fa 100644 --- a/backend/cpp/audio-cpp/family_gate.cpp +++ b/backend/cpp/audio-cpp/family_gate.cpp @@ -1,6 +1,7 @@ #include "family_gate.h" #include +#include namespace audiocpp_backend { namespace { @@ -83,14 +84,36 @@ namespace { // widen an entry without running that, because what it prevents is a process // death rather than a wrong answer. struct DtypeAllowList { + // NULL TERMINATED, and the terminator occupies one of these slots: both + // loops below stop at the first nullptr and have no other bound, so an entry + // that named three dtypes would leave them reading past the end of the + // array. That is undefined behaviour rather than a wrong answer, and it is + // one keystroke away from any edit that widens an entry, so the terminator + // is asserted at compile time below rather than trusted. + static constexpr std::size_t kSlots = 3; + const char *family; - const char *allowed[3]; + const char *allowed[kSlots]; }; constexpr DtypeAllowList kDtypeAllowLists[] = { {"supertonic", {"f32", "i64", nullptr}}, }; +constexpr bool allow_lists_are_terminated() { + for (const auto &entry : kDtypeAllowLists) { + if (entry.allowed[DtypeAllowList::kSlots - 1] != nullptr) { + return false; + } + } + return true; +} + +static_assert(allow_lists_are_terminated(), + "every DtypeAllowList must leave its last slot null: the lookups " + "below stop at the first nullptr and would otherwise read past " + "the end of the array"); + const DtypeAllowList *find_allow_list(const std::string &family) { for (const auto &entry : kDtypeAllowLists) { if (family == entry.family) { @@ -102,6 +125,10 @@ const DtypeAllowList *find_allow_list(const std::string &family) { } // namespace +bool family_has_weight_dtype_allow_list(const std::string &family) { + return find_allow_list(family) != nullptr; +} + bool weight_dtype_is_supported(const std::string &family, const std::string &dtype) { const DtypeAllowList *list = find_allow_list(family); if (list == nullptr) { diff --git a/backend/cpp/audio-cpp/family_gate.h b/backend/cpp/audio-cpp/family_gate.h index 7c40da48d..7e8210fd2 100644 --- a/backend/cpp/audio-cpp/family_gate.h +++ b/backend/cpp/audio-cpp/family_gate.h @@ -54,6 +54,15 @@ FamilyDecision decide_family(bool path_is_gguf, const std::string &embedded_fami // synthesising, so that step stays a documented manual one at the table itself. bool weight_dtype_is_supported(const std::string &family, const std::string &dtype); +// True when `family` has an entry in the table at all, which is the question a +// caller deciding whether to OPEN THE FILE has to ask. Distinct from +// "supported_weight_dtypes(family) is empty": that string is also empty for an +// entry with an empty allow list, and such an entry means "this family can run +// nothing", which weight_dtype_is_supported already answers by refusing every +// dtype. Deciding from the string would skip the check on precisely the entry +// that most needs it. +bool family_has_weight_dtype_allow_list(const std::string &family); + // The dtypes `family` is restricted to, as "f32, i64", or empty when it is not // restricted at all. For the refusal message, so the operator is told what to // look for rather than only what is wrong. diff --git a/backend/cpp/audio-cpp/family_gate_test.cpp b/backend/cpp/audio-cpp/family_gate_test.cpp index 1c6f90aea..e57995973 100644 --- a/backend/cpp/audio-cpp/family_gate_test.cpp +++ b/backend/cpp/audio-cpp/family_gate_test.cpp @@ -146,8 +146,20 @@ static void test_weight_dtype_allow_list() { check_eq(supported_weight_dtypes("supertonic"), "f32, i64", "the refusal can name what to look for"); check_eq(supported_weight_dtypes("nemotron_asr"), "", - "an unlisted family reports no restriction, which is what makes " - "the caller skip the whole check"); + "an unlisted family reports no restriction"); + + // What the caller actually decides on, and it is a DIFFERENT question from + // "is the description empty": an entry with an empty allow list would + // describe itself as "" while refusing every dtype, so a caller that skipped + // the file read on the empty string would skip the check on the one entry + // that refuses everything. + check(family_has_weight_dtype_allow_list("supertonic"), + "a listed family has an allow list"); + check(!family_has_weight_dtype_allow_list("nemotron_asr"), + "an unlisted family has none, which is what lets the caller skip " + "opening the file at all"); + check(!family_has_weight_dtype_allow_list(""), + "an empty family name has no allow list"); } int main() { diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index 3365d6b14..e63dd6052 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -6,9 +6,8 @@ // 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, SoundGeneration, TTSStream and -// AudioTranscriptionStream RPCs. The remaining audio RPCs land in later -// commits. +// Diarize, AudioTransform, TTS, SoundGeneration, TTSStream, +// AudioTranscriptionStream and AudioTranscriptionLive RPCs. #include "backend.pb.h" #include "backend.grpc.pb.h" @@ -82,11 +81,23 @@ std::shared_ptr g_model; // Never return LoadedModel& from here: that is the shape that reintroduces the // use-after-free. // -// _unchecked because it performs NO request-level validation. Status is its -// only legitimate caller, because Status takes a HealthMessage, which carries -// no ModelIdentity to check. Every RPC that acts on a model must call -// snapshot_for instead; the name is deliberately unpleasant so that reaching -// for it is a visible decision rather than an omission. +// _unchecked because it performs NO request-level validation. There are exactly +// two legitimate classes of caller, and both are "there is no identity to +// check" rather than "the check was skipped": +// +// 1. Status, which takes a HealthMessage. It carries no ModelIdentity field. +// 2. An RPC whose request message has no ModelIdentity field either, which +// today is AudioTranscriptionLive: TranscriptLiveRequest is a oneof of +// TranscriptLiveConfig and TranscriptLiveAudio and neither carries one, so +// snapshot_for does not even instantiate for it. The stale-route hazard +// #10952 describes is therefore unguarded on that RPC, and the fix has to +// be in backend.proto rather than here: nothing this process can read off +// the request says which model the caller thought it was reaching. When +// that field lands, these handlers move to snapshot_for and this clause +// goes away. +// +// Every OTHER RPC must call snapshot_for; the name is deliberately unpleasant so +// that reaching for it is a visible decision rather than an omission. std::shared_ptr snapshot_unchecked() { std::lock_guard lock(g_model_mu); return g_model; @@ -482,8 +493,9 @@ public: GStatus Status(ServerContext *, const backend::HealthMessage *, backend::StatusResponse *response) override { - // snapshot_unchecked is right here, and is the only place it is: - // HealthMessage carries no ModelIdentity, so there is nothing to check. + // snapshot_unchecked is right here: HealthMessage carries no + // ModelIdentity, so there is nothing to check. See case 1 at the + // function. response->set_state(snapshot_unchecked() ? backend::StatusResponse::READY : backend::StatusResponse::UNINITIALIZED); @@ -1519,6 +1531,260 @@ public: return to_status(err); } } + + // The one BIDIRECTIONAL stream this backend serves: live microphone ASR. + // The client sends a TranscriptLiveConfig, then TranscriptLiveAudio frames; + // this answers with a ready ack, then deltas and words as the audio + // arrives, then one message carrying final_result once the read side + // closes. + // + // THERE IS NO OFFLINE FALLBACK, unlike AudioTranscriptionStream. Live + // transcription has to consume audio incrementally, so a family with no + // streaming ASR is refused rather than served a batch run at the end; the + // Streaming-only mode_candidates list for this RPC is where that is + // expressed, and this handler needs no branch for it. + // + // THE READY ACK IS A CONTRACT, not a courtesy. core/backend/transcript_live.go + // sends the config and then BLOCKS on Recv, treating an Unimplemented there + // as "this backend cannot do live transcription, degrade to the file path" + // and a first response carrying data as a broken backend. So routing must be + // refused before the ack, and the ack must go out before the first audio + // frame is read, or the client never sends one and both sides wait. + // + // eou AND eob STAY FALSE. TranscriptLiveResponse carries them for + // cache-aware models that emit end-of-utterance and end-of-backchannel + // tokens; audio.cpp's StreamEvent has no equivalent signal, and inferring + // one from silence would be a guess wearing a protocol flag's clothes. A + // client uses eou to decide the speaker yielded the turn, so a wrong one + // cuts people off mid-sentence. False is not a degradation here, it is the + // truth: this backend does not know. + GStatus AudioTranscriptionLive( + ServerContext *, + grpc::ServerReaderWriter *stream) override { + try { + // snapshot_unchecked, which every other model-touching handler is + // forbidden to call. TranscriptLiveRequest carries NO ModelIdentity + // field, in either arm of its oneof, so snapshot_for does not even + // instantiate for it: there is nothing to compare. See case 2 at + // snapshot_unchecked, including why the fix is a proto change. + const auto model = snapshot_unchecked(); + if (model == nullptr) { + return GStatus(grpc::StatusCode::FAILED_PRECONDITION, + "audio-cpp: no model is loaded; call LoadModel first"); + } + + backend::TranscriptLiveRequest incoming; + if (!stream->Read(&incoming)) { + // Opened and closed with nothing said. A clean exit, not an + // error: there is nothing to transcribe and nobody to tell. + return GStatus::OK; + } + if (!incoming.has_config()) { + return GStatus(grpc::StatusCode::INVALID_ARGUMENT, + "audio-cpp: the first AudioTranscriptionLive " + "message must carry a Config"); + } + const backend::TranscriptLiveConfig config = incoming.config(); + + // 16 kHz mono or nothing, and this is a REFUSAL rather than a + // resample for the reason spelled out at kSpeechSampleRate: the + // streaming families build their spans in their own 16 kHz feature + // domain whatever the input rate was, so honouring an 8 kHz session + // would return word timestamps 2x off with a 200 and no diagnostic. + // The file-fed handlers resample their input to make the two + // domains the same one; live audio arrives a frame at a time from a + // client that already picked a rate, and read_audio_file's + // resampler cannot be applied to it incrementally, so the only + // honest answers are this refusal or a wrong timestamp. + // + // Costs nothing today: core/backend/transcript_live.go hardcodes + // liveSampleRate = 16000, and backend.proto documents 0 as 16000 and + // says backends may reject other rates. + const int sample_rate = + config.sample_rate() > 0 ? config.sample_rate() : kSpeechSampleRate; + if (sample_rate != kSpeechSampleRate) { + return GStatus(grpc::StatusCode::INVALID_ARGUMENT, + "audio-cpp: live transcription accepts " + + std::to_string(kSpeechSampleRate) + + " Hz mono PCM only, got " + + std::to_string(sample_rate) + + " Hz; resample on the client side"); + } + + audiocpp_backend::RequestShape shape; + // No request signal chooses the task here: TranscriptLiveConfig has + // no prompt field, so has_prompt_text stays false and the only + // candidate is Asr. pinned_task is still copied, because without it + // the model's `task:` option is dead on this RPC. + shape.pinned_task = model->pinned_task(); + + // Before the lane, like every other handler: a family that cannot + // stream ASR is answering a question about itself and must not + // queue behind somebody else's run to be told no. This is also what + // produces the UNIMPLEMENTED the client reads as "degrade to the + // file path", so it has to happen before the ack below. + model->check_can_serve(audiocpp_backend::Rpc::AudioTranscriptionLive, + shape); + + // THE LANE IS HELD FOR THE WHOLE STREAM, which is longer than any + // other handler holds it: as long as the user keeps talking. The + // streaming session is stateful and cached, so a concurrent run on + // the same model would interleave its audio with this one's and + // corrupt both transcripts. Named local because LaneEntry is + // immovable and has to span session_for, every chunk and finalize. + audiocpp_backend::LaneEntry lane = model->acquire(0); + const auto session = model->session_for( + audiocpp_backend::Rpc::AudioTranscriptionLive, shape, lane); + + engine::runtime::TaskRequest task; + // An EMPTY buffer carrying the CONTRACT only: the rate and the + // channel count of the frames about to arrive, and no samples, + // because none exist yet. Not decoration. build_preparation_request + // derives the audio contract from this field, and nemotron_asr's + // streaming prepare() throws "Nemotron ASR streaming prepare() + // requires an audio contract" when it is absent, so a live stream + // without it fails on the first model that serves it. + engine::runtime::AudioBuffer contract; + contract.sample_rate = sample_rate; + // Mono, from the proto: TranscriptLiveAudio.pcm is documented as + // "mono PCM in [-1,1] at config.sample_rate" and there is no channel + // field to say otherwise. + contract.channels = 1; + task.audio_input = std::move(contract); + + if (!config.language().empty()) { + task.options["language"] = config.language(); + } + for (const auto ¶m : config.params()) { + task.options[param.first] = param.second; + } + + bool client_gone = false; + backend::TranscriptLiveResponse ack; + ack.set_ready(true); + if (!stream->Write(ack)) { + // The client hung up between opening the stream and the ack. + // Nothing was consumed and nobody is listening. + return GStatus::OK; + } + + audiocpp_backend::TranscriptDeltaTracker tracker; + const auto write_delta = [&](const std::string &fragment) { + if (fragment.empty() || client_gone) { + return; + } + backend::TranscriptLiveResponse response; + response.set_delta(fragment); + if (!stream->Write(response)) { + client_gone = true; + } + }; + + // The SAME reconciliation AudioTranscriptionStream uses, and for the + // same reason: the families disagree about whether partial_text is a + // fragment or the whole hypothesis, one of them delivers the same + // event twice, and a delta that ends inside a UTF-8 sequence makes + // the Go client's Unmarshal fail and costs the client every + // remaining message. None of that is re-derived here. See + // stream_delta.h. + const auto emit = [&](const engine::runtime::StreamEvent &event) { + if (event.partial_text.has_value()) { + write_delta(tracker.observe(event.partial_text->text)); + } + if (event.word_timestamps.empty() || client_gone) { + return; + } + // FORWARD TOLERANT, and dead against the pinned upstream: no + // family puts word timestamps on a StreamEvent, only on the + // TaskResult, so today every word a live client sees arrives in + // final_result below. Kept because the field exists, a family + // that fills it then works with no change here, and the + // conversion is the same one final_result gets. + backend::TranscriptLiveResponse response; + for (const auto &word : event.word_timestamps) { + auto *out = response.add_words(); + // NANOSECONDS. TranscriptWord is the only unit in this proto + // that is not seconds, and core/backend reads it straight + // into a time.Duration. + out->set_start(audiocpp_backend::samples_to_nanoseconds( + word.span.start_sample, sample_rate)); + out->set_end(audiocpp_backend::samples_to_nanoseconds( + word.span.end_sample, sample_rate)); + out->set_text(word.word); + } + if (!stream->Write(response)) { + client_gone = true; + } + }; + + std::int64_t consumed_frames = 0; + const auto next_frames = [&](std::vector &out) { + while (stream->Read(&incoming)) { + if (incoming.has_config()) { + // backend.proto says a second Config resets the decode + // session. audio.cpp cannot do that truthfully: a reset + // would clear the session's audio while the deltas + // already on the wire cannot be taken back, so the final + // text would then describe only the audio after the + // reset and would CONTRADICT the transcript the client + // assembled. Refused loudly rather than ignored: a + // client that believes it reset the decoder and did not + // is handed a transcript that silently continues the + // audio it thought it discarded. + throw audiocpp_backend::ConfigError( + "audio-cpp: family '" + model->family() + + "' cannot reset a live session mid-stream; close " + "this stream and open a new one"); + } + if (!incoming.has_audio()) { + // Neither arm of the oneof is set. Nothing to feed, and + // not worth failing a live session over. + continue; + } + const auto &pcm = incoming.audio().pcm(); + if (pcm.empty()) { + continue; + } + out.assign(pcm.begin(), pcm.end()); + // Mono, so floats and frames are the same count. Only used + // for the duration reported in final_result. + consumed_frames += static_cast(pcm.size()); + return true; + } + return false; + }; + + // The driver owns the streaming state obligation and the stream + // event sink: nemotron_asr, the only family that serves this RPC + // today, reports EVERY partial through the sink from inside + // finalize(), and its process_audio_chunk returns a bare event. It + // also buffers the wire's frames up to the family's own preferred + // window, which is one second for nemotron_asr and nothing like the + // 512-sample frames a client's audio callback produces. + const auto result = audiocpp_backend::run_streaming_live( + session, task, next_frames, emit, lane); + + // Makes concatenating every delta equal final_result's text, which + // is what the HTTP layer above assembles. Also what flushes a + // held-back partial UTF-8 sequence. + if (result.text_output.has_value()) { + write_delta(tracker.reconcile(result.text_output->text)); + } + + backend::TranscriptLiveResponse final_response; + audiocpp_backend::fill_transcript_result( + result, sample_rate, + audiocpp_backend::samples_to_seconds(consumed_frames, sample_rate), + final_response.mutable_final_result()); + if (!client_gone) { + stream->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 46a2c0886..f38a1dfc9 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -176,8 +176,14 @@ std::string require_family(const std::string &resolved_path, // through rather than guessed at. void require_supported_weight_dtypes(const std::string &family, const std::string &resolved_path) { - const std::string allowed_names = supported_weight_dtypes(family); - if (allowed_names.empty() || !path_looks_like_gguf(resolved_path)) { + // Asked as "is there an entry", not as "is the description non-empty": an + // entry with an empty allow list describes a family that can run nothing, + // and reading the description would skip the check on exactly that entry + // while weight_dtype_is_supported refused every dtype. No such entry exists + // today; the two questions are different ones and only one of them is this + // guard's. + if (!family_has_weight_dtype_allow_list(family) || + !path_looks_like_gguf(resolved_path)) { return; } @@ -212,7 +218,7 @@ void require_supported_weight_dtypes(const std::string &family, resolved_path + "); it aborts the backend process on the first request rather than " "failing the request. Use the 'orig' GGUF package, whose weights are " + - allowed_names + "."); + supported_weight_dtypes(family) + "."); } } // namespace @@ -684,4 +690,101 @@ engine::runtime::TaskResult run_streaming_audio( return streaming.finish_stream(); } +engine::runtime::TaskResult run_streaming_live( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const std::function &)> &next_frames, + const std::function &on_event, + LaneEntry &lane) { + auto &streaming = require_streaming(session); + + // The contract is the only thing that says what rate and layout the frames + // about to arrive are in, and prepare() needs it: see the header. + if (!request.audio_input.has_value()) { + throw ConfigError( + "audio-cpp: a live streaming request carries no audio contract"); + } + const int sample_rate = request.audio_input->sample_rate; + const int channels = + request.audio_input->channels > 0 ? request.audio_input->channels : 1; + + // Installed BEFORE the stream begins and cleared on every exit, including + // the exception path, for the reasons spelled out in run_streaming_audio. + ScopedStreamSink sink(streaming, + [&on_event](const engine::runtime::StreamEvent &event) { + if (on_event) { + on_event(event); + } + }); + + begin_stream(session, request, lane); + + const std::int64_t chunk_frames = + chunk_frames_for(streaming.streaming_policy(), sample_rate); + // chunk_frames_for never returns a non-positive count, so this is never + // zero and the accumulation loop below always terminates. + const std::size_t chunk_floats = static_cast(chunk_frames) * + static_cast(channels); + + std::int64_t fed_frames = 0; + const auto feed = [&](std::vector samples) { + engine::runtime::AudioChunk chunk; + chunk.sample_rate = sample_rate; + chunk.channels = channels; + // A FRAME index, counted across the whole stream: vibevoice_asr offsets + // every span it reports by it, so restarting it per chunk would put + // every word at the top of the recording. + chunk.start_sample = fed_frames; + chunk.samples = std::move(samples); + fed_frames += + static_cast(chunk.samples.size()) / channels; + const auto event = streaming.process_audio_chunk(chunk); + if (on_event) { + on_event(event); + } + }; + + std::vector pending; + std::vector incoming; + while (true) { + incoming.clear(); + if (!next_frames(incoming)) { + break; + } + pending.insert(pending.end(), incoming.begin(), incoming.end()); + while (pending.size() >= chunk_floats) { + std::vector window(pending.begin(), + pending.begin() + + static_cast(chunk_floats)); + pending.erase(pending.begin(), + pending.begin() + + static_cast(chunk_floats)); + feed(std::move(window)); + } + } + + if (!pending.empty()) { + // The tail is whatever did not fill a window. Refused rather than + // truncated when it is not a whole number of frames, exactly as in + // run_streaming_audio: the division above would drop the stray floats + // from the transcript with no diagnostic. Unreachable for a mono live + // stream, which is every live stream today. + if (pending.size() % static_cast(channels) != 0) { + throw ConfigError( + "audio-cpp: live stream ended mid-frame: " + + std::to_string(pending.size()) + " trailing samples across " + + std::to_string(channels) + " channels"); + } + feed(std::move(pending)); + } + + if (fed_frames == 0) { + // Nothing was spoken. See the header: finalizing an empty stream is not + // legal for every family, and an empty transcript is the truthful + // answer rather than an engine-internal INTERNAL. + return engine::runtime::TaskResult{}; + } + 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 b0a498eff..9222ab3fc 100644 --- a/backend/cpp/audio-cpp/loaded_model.h +++ b/backend/cpp/audio-cpp/loaded_model.h @@ -329,4 +329,40 @@ engine::runtime::TaskResult run_streaming_audio( const std::function &on_event, LaneEntry &lane); +// Drives the same ASR shape as run_streaming_audio when the audio DOES NOT +// EXIST YET, which is the live-microphone case: instead of slicing a buffer it +// pulls frames from the caller until the input side closes. +// +// `next_frames` fills `out` with interleaved float PCM and returns true, or +// returns false when there is no more input. It is expected to BLOCK, since the +// only real implementation is a gRPC stream Read, and it may throw: a request +// the handler has to refuse mid-stream unwinds through here, and the sink is +// cleared on that path like every other. +// +// The audio contract comes from `request.audio_input`, which for a live stream +// is an EMPTY buffer carrying only the sample rate and channel count. It is not +// optional: nemotron_asr's streaming prepare() throws "Nemotron ASR streaming +// prepare() requires an audio contract" without one, and there is no buffer to +// derive it from here. +// +// Frames are BUFFERED to the family's own preferred window rather than fed in +// whatever sizes the wire delivered them in, because that window is a family's +// statement about what it can decode (nemotron_asr asks for one second, higgs +// for four), and a 512-sample gRPC frame is a property of the client's audio +// callback rather than of the model. The tail shorter than a window is fed at +// the end. +// +// A stream that carried NO AUDIO returns an empty TaskResult and never calls +// finish_stream. Finalizing an empty stream is not universally legal: +// nemotron_asr throws "Nemotron ASR finalize requires streamed audio", so a +// client that opens a session and closes it without speaking would receive an +// INTERNAL naming an engine internal instead of an empty transcript, which is +// the truthful answer to "transcribe nothing". +engine::runtime::TaskResult run_streaming_live( + const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request, + const std::function &)> &next_frames, + const std::function &on_event, + LaneEntry &lane); + } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/streaming_driver_ctest.cpp b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp index 469cce2b2..95b49c0ef 100644 --- a/backend/cpp/audio-cpp/streaming_driver_ctest.cpp +++ b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp @@ -1,5 +1,5 @@ // Tests for the streaming drivers in loaded_model: begin_stream, -// run_streaming_pull and run_streaming_audio. +// run_streaming_pull, run_streaming_audio and run_streaming_live. // // Engine-linked, so this runs through ctest rather than // backend/cpp/run-unit-tests.sh. It builds no model and loads no file: a @@ -13,7 +13,9 @@ #include "engine/framework/runtime/session.h" +#include #include +#include #include #include #include @@ -638,6 +640,231 @@ static void test_prepare_tracks_the_request_not_the_session() { "the second stream prepares at ITS rate, not the first one's"); } +// -------------------------------------------------------------------------- +// run_streaming_live +// -------------------------------------------------------------------------- + +// Hands the driver a fixed list of wire frames, the way a client's audio +// callback would, and then closes. +static std::function &)> +frames_from(const std::vector &sizes) { + auto index = std::make_shared(0); + auto list = std::make_shared>(sizes); + return [index, list](std::vector &out) { + if (*index >= list->size()) { + return false; + } + out.assign((*list)[*index], 0.5F); + ++*index; + return true; + }; +} + +static rt::TaskRequest live_request(int sample_rate, int channels) { + rt::TaskRequest request; + rt::AudioBuffer contract; + contract.sample_rate = sample_rate; + contract.channels = channels; + request.audio_input = std::move(contract); // no samples: none exist yet + return request; +} + +// The wire's frame size is a property of the client's audio callback. The +// family's window is a statement about what it can decode. The driver feeds the +// second, not the first. +static void test_live_buffers_wire_frames_into_policy_windows() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 1600; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + + // Ten 512-sample frames: 5120 samples, i.e. three full 1600 windows and a + // 320 sample tail. + const auto result = audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), + frames_from(std::vector(10, 512)), + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 4, + "5120 wire samples at a 1600 frame window is three windows and a tail"); + if (fake.chunks.size() == 4) { + check(fake.chunks[0].samples.size() == 1600, "first window is full"); + check(fake.chunks[2].samples.size() == 1600, "third window is full"); + check(fake.chunks[3].samples.size() == 320, "the tail is what was left"); + check(fake.chunks[0].start_sample == 0, "the first window starts at zero"); + check(fake.chunks[1].start_sample == 1600, "start_sample counts frames"); + check(fake.chunks[3].start_sample == 4800, "the tail is offset by all of it"); + check(fake.chunks[0].sample_rate == 16000, "the chunk carries the session rate"); + } + check(result.text_output.has_value() && + result.text_output->text == "w1w2w3w4/frames=5120", + "every wire sample reaches the family exactly once"); +} + +// nemotron_asr's shape: no partials from process_audio_chunk, every one of them +// through the sink from inside finalize. +static void test_live_installs_and_clears_the_sink() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + SinkOnlyAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 512; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + + std::vector fragments; + audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), frames_from({512}), + [&](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 live either"); + check(!fake.sink_installed(), + "the sink is cleared before returning, so the cached session holds no " + "reference to this call's frame"); + check_eq(join(fake.calls), "sink+|prepare|reset|chunk|finalize|sink-", + "sink installed before the stream begins, cleared after it ends"); +} + +// A client that opens a session and closes it without speaking. finalize is NOT +// called: nemotron_asr throws "finalize requires streamed audio", and an empty +// transcript is the truthful answer to transcribing nothing. +static void test_live_with_no_audio_never_finalizes() { + 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 result = audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), frames_from({}), + [](const rt::StreamEvent &) {}, entry); + + check_eq(join(fake.calls), "sink+|prepare|reset|sink-", + "an empty live stream begins and ends without a chunk or a finalize"); + check(!result.text_output.has_value(), + "an empty live stream reports no transcript rather than an error"); +} + +// A tail shorter than a window is still fed. Without this the last fragment of +// speech never reaches the model, and nothing says so. +static void test_live_feeds_a_short_tail() { + 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); + + audiocpp_backend::run_streaming_live(session, live_request(16000, 1), + frames_from({100, 200}), + [](const rt::StreamEvent &) {}, entry); + + check(fake.chunks.size() == 1, + "300 samples against a 16000 frame window is one short chunk, not none"); + if (!fake.chunks.empty()) { + check(fake.chunks[0].samples.size() == 300, "the tail carries everything fed"); + } +} + +// A live request carries no samples, so the CONTRACT is the only thing that says +// what rate the frames are in, and prepare() needs it: nemotron_asr's streaming +// prepare throws without one. +static void test_live_prepares_at_the_contract_rate() { + 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); + + audiocpp_backend::run_streaming_live(session, live_request(16000, 1), + frames_from({512}), + [](const rt::StreamEvent &) {}, entry); + check(fake.prepared_rate() == 16000, + "the empty contract buffer still carries the rate into prepare()"); + + bool threw_config = false; + try { + rt::TaskRequest bare; // no audio_input at all + audiocpp_backend::run_streaming_live(session, bare, frames_from({512}), + [](const rt::StreamEvent &) {}, entry); + } catch (const audiocpp_backend::ConfigError &) { + threw_config = true; + } catch (const std::exception &) { + } + check(threw_config, "a live request with no audio contract is refused"); +} + +// The pull function is the gRPC read, and a request the handler has to refuse +// mid-stream unwinds through the driver. The sink must not survive it. +static void test_live_clears_the_sink_when_the_puller_throws() { + 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); + + bool threw = false; + try { + audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), + [](std::vector &) -> bool { + throw std::runtime_error("fake: the client vanished"); + }, + [](const rt::StreamEvent &) {}, entry); + } catch (const std::exception &) { + threw = true; + } + check(threw, "a failing pull propagates"); + check(!fake.sink_installed(), "the sink is cleared on the pull's exception path"); +} + +// Two live streams over the SAME cached session must not run into each other. +static void test_live_replays_identically_on_a_refetched_session() { + 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 first = audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), frames_from({512, 512}), + [](const rt::StreamEvent &) {}, entry); + const auto second = audiocpp_backend::run_streaming_live( + session, live_request(16000, 1), frames_from({512, 512}), + [](const rt::StreamEvent &) {}, entry); + + check_eq(second.text_output.has_value() ? second.text_output->text : "", + first.text_output.has_value() ? first.text_output->text : "", + "a re-fetched live session replays the same transcript"); + // Not a tautology: without the reset the second run reports w1..w4 and 2048 + // frames. + check_eq(first.text_output.has_value() ? first.text_output->text : "", + "w1w2/frames=1024", "the first live run saw exactly what was fed"); +} + +static void test_live_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 { + audiocpp_backend::run_streaming_live(session, live_request(16000, 1), + frames_from({512}), + [](const rt::StreamEvent &) {}, entry); + } catch (const audiocpp_backend::CapabilityError &) { + threw_capability = true; + } catch (const std::exception &) { + } + check(threw_capability, + "run_streaming_live refuses a session with no streaming half"); +} + int main() { test_begin_stream_prepares_then_starts(); test_base_start_stream_resets(); @@ -655,6 +882,14 @@ int main() { test_pull_stops_on_a_final_event(); test_pull_refuses_a_non_streaming_session(); test_prepare_tracks_the_request_not_the_session(); + test_live_buffers_wire_frames_into_policy_windows(); + test_live_installs_and_clears_the_sink(); + test_live_with_no_audio_never_finalizes(); + test_live_feeds_a_short_tail(); + test_live_prepares_at_the_contract_rate(); + test_live_clears_the_sink_when_the_puller_throws(); + test_live_replays_identically_on_a_refetched_session(); + test_live_refuses_a_non_streaming_session(); if (failures) { fprintf(stderr, "%d check(s) failed\n", failures); return 1;