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;