diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index 6ab2181da..3365d6b14 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -1365,6 +1365,15 @@ public: // 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. + // + // This guard also depends on the pull loop having DRAINED the + // session. It does today: no family sets is_final on a pulled + // event, so the loop runs to nullopt. If one ever does, the loop + // breaks early, and omnivoice's finish_stream drains and merges the + // chunks that were never emitted (session.cpp:576) into the result + // this branch then discards, silently truncating the audio. Whoever + // teaches a family to end a pull stream early has to revisit the + // pair together. if (!header_sent) { if (result.audio_output.has_value()) { write_audio(*result.audio_output); @@ -1402,8 +1411,10 @@ public: // 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. + // whole hypothesis and repeats the last event of each batch), and a delta + // must additionally be valid UTF-8 or the Go client cannot unmarshal it at + // all, 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 diff --git a/backend/cpp/audio-cpp/loaded_model.cpp b/backend/cpp/audio-cpp/loaded_model.cpp index 667c817dc..fb80bee30 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -164,6 +164,107 @@ std::string require_family(const std::string &resolved_path, return decision.family; } +// Families that CRASH THE PROCESS on a weight dtype they cannot handle, and the +// dtypes they can. +// +// A guard rather than a note, because the failure is not an exception: loading a +// supertonic package whose weights are stored as f16 or q8_0 reaches +// ggml_concat with one f16 operand and one f32 one, and ggml_abort takes the +// whole backend process down with SIGABRT on the FIRST synthesis. Nothing +// upstream of here can catch that, so an operator sees a backend that dies on +// every request with no status and no message, and the model that killed it +// loaded successfully. +// +// Attributed rather than assumed: the abort is identical through the unary TTS +// RPC and through TTSStream, and upstream's own docs/gguf.md:90 records +// supertonic's 16-bit GGUF column as "---", i.e. NOT TESTED, and its q8_0 as +// "No (unsupported weight dtype)". Only the `orig` package, whose 698 weight +// tensors are f32, is marked Pass, and that is the one every verification of +// this backend has used. +// +// i64 is allowed alongside f32 because the orig package itself carries 72 i64 +// tensors (shape and index constants), so refusing them would refuse the +// artifact that works. +// +// TO REMOVE THIS: bump AUDIO_CPP_VERSION past a fix, load a supertonic f16 +// package, and synthesise. If it produces audio, delete the entry. Do not widen +// it without running that, because the failure it prevents is a process death +// rather than a wrong answer. +struct DtypeAllowList { + const char *family; + const char *allowed[3]; +}; + +constexpr DtypeAllowList kDtypeAllowLists[] = { + {"supertonic", {"f32", "i64", nullptr}}, +}; + +// Refuses a GGUF whose weights are stored in a dtype the family cannot survive. +// +// Only GGUF paths are inspected. A directory of safetensors carries its dtypes +// per file and has not been tested against this failure, so it is passed +// through rather than guessed at. +void require_supported_weight_dtypes(const std::string &family, + const std::string &resolved_path) { + const DtypeAllowList *list = nullptr; + for (const auto &entry : kDtypeAllowLists) { + if (family == entry.family) { + list = &entry; + break; + } + } + if (list == nullptr || !path_looks_like_gguf(resolved_path)) { + return; + } + + std::string offending_dtype; + std::string offending_tensor; + try { + const auto source = + engine::assets::open_tensor_source(std::filesystem::path(resolved_path)); + if (source == nullptr) { + return; + } + for (const auto &tensor : source->tensors()) { + bool allowed = false; + for (const char *const *name = list->allowed; *name != nullptr; ++name) { + if (tensor.dtype == *name) { + allowed = true; + break; + } + } + if (!allowed) { + offending_dtype = tensor.dtype; + offending_tensor = tensor.name; + break; + } + } + } catch (const std::exception &) { + // Unreadable as a tensor source. Not this guard's problem to report: + // the registry load below produces a message naming the real fault, and + // refusing here would turn every unusual packaging into this error. + return; + } + + if (offending_dtype.empty()) { + return; + } + std::string allowed_names; + for (const char *const *name = list->allowed; *name != nullptr; ++name) { + if (!allowed_names.empty()) { + allowed_names += ", "; + } + allowed_names += *name; + } + throw ConfigError( + "audio-cpp: family '" + family + "' cannot run weights stored as '" + + offending_dtype + "' (tensor '" + offending_tensor + "' in " + + 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 + "."); +} + } // namespace engine::runtime::VoiceTaskKind to_engine_task(Task task) { @@ -293,6 +394,12 @@ LoadedModel::LoadedModel(const std::string &resolved_path, throw ConfigError("audio-cpp: unknown audio.cpp family '" + family + "'"); } + // Before the load, for the same reason parse_backend_type runs before it: + // a refusal a metadata read can produce should not cost a full model load. + // More importantly it must precede the FIRST REQUEST, since that is where + // an unsupported dtype aborts the process rather than failing. + require_supported_weight_dtypes(family, resolved_path); + // Session options are built BEFORE the load, because parse_backend_type // rejects an unknown backend name. Validating after the load would make // `backend:cudaa` cost a full model load, on a fault a string comparison @@ -562,6 +669,30 @@ engine::runtime::TaskResult run_streaming_audio( LaneEntry &lane) { auto &streaming = require_streaming(session); + const int channels = audio.channels > 0 ? audio.channels : 1; + // REFUSED, not rounded away, and checked before anything is touched so a + // refusal leaves no half-started stream on a cached session. + // + // An interleaved buffer whose float count is not a whole number of frames + // is a truncated input, and the integer division below would silently drop + // the tail floats: they would never be fed, never reach the transcript, and + // nothing would say so. Upstream refuses it from the other side of this + // same call rather than tolerating it: vibevoice_asr's process_audio_chunk + // throws "VibeVoice-ASR streamed audio samples must be divisible by channel + // count" (session.cpp:74-77). + // + // ConfigError, i.e. INVALID_ARGUMENT, because the buffer came from the + // caller's file. read_audio_file's positive-rate path always answers mono + // and so cannot reach this, but its native-rate path passes the reader's + // sample count through unchanged, and a driver does not get to assume which + // path its caller took. + if (audio.samples.size() % static_cast(channels) != 0) { + throw ConfigError( + "audio-cpp: streaming input is not a whole number of frames: " + + std::to_string(audio.samples.size()) + " samples across " + + std::to_string(channels) + " channels"); + } + // 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 @@ -575,7 +706,6 @@ engine::runtime::TaskResult run_streaming_audio( 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 = diff --git a/backend/cpp/audio-cpp/stream_delta.cpp b/backend/cpp/audio-cpp/stream_delta.cpp index 6655f3708..87d5ad64a 100644 --- a/backend/cpp/audio-cpp/stream_delta.cpp +++ b/backend/cpp/audio-cpp/stream_delta.cpp @@ -11,40 +11,129 @@ bool starts_with(const std::string &text, const std::string &prefix) { text.compare(0, prefix.size(), prefix) == 0; } +// Length of the longest prefix of `text` that does NOT end inside a multi-byte +// UTF-8 sequence, i.e. the most that can go on the wire without splitting a +// character in half. +// +// Only the TRAILING sequence is examined. Bytes further back are none of this +// function's business: a family that emitted a malformed sequence in the middle +// of its own transcript cannot be repaired here without deleting part of that +// transcript. +// +// Anything that can never complete is reported as complete, so it goes out +// rather than being held forever: a lone continuation byte with no lead, a lead +// byte the encoding does not define, and a run of five or more continuation +// bytes are all passed through. A tracker that stalled on undecodable input +// would turn one bad byte into a permanently silent stream, which is worse than +// the bad byte. +std::size_t utf8_complete_prefix_length(const std::string &text) { + std::size_t index = text.size(); + std::size_t continuations = 0; + while (index > 0 && continuations < 4) { + const auto byte = static_cast(text[index - 1]); + if ((byte & 0xC0) == 0x80) { + --index; + ++continuations; + continue; + } + std::size_t needed = 1; + if ((byte & 0x80) == 0x00) { + needed = 1; + } else if ((byte & 0xE0) == 0xC0) { + needed = 2; + } else if ((byte & 0xF0) == 0xE0) { + needed = 3; + } else if ((byte & 0xF8) == 0xF0) { + needed = 4; + } else { + // Not a lead byte this encoding defines, so nothing is waiting on + // it and it must not be held. + needed = 1; + } + if (continuations + 1 >= needed) { + return text.size(); + } + // The trailing sequence is short by at least one byte: cut before its + // lead byte and keep the rest for the next fragment. + return index - 1; + } + return text.size(); +} + } // namespace +std::string TranscriptDeltaTracker::release(const std::string &fragment) { + // Held-back bytes go in FRONT of whatever arrived next, or the character + // they begin is reassembled in the wrong order. + const std::string candidate = pending_ + fragment; + const std::size_t cut = utf8_complete_prefix_length(candidate); + pending_ = candidate.substr(cut); + std::string emitted = candidate.substr(0, cut); + assembled_ += emitted; + return emitted; +} + 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)) { + // The comparisons run against everything KNOWN, delivered plus held back, + // rather than against the delivered text alone. Comparing against the + // delivered text would treat the held-back byte as new on the very next + // report and emit it twice. + const std::string known = assembled_ + pending_; + // Already accounted for. Covers the repeat voxtral produces when the sink + // and the return value carry the same event, and a hypothesis that shrank. + if (starts_with(known, partial_text)) { return {}; } - if (starts_with(partial_text, assembled_)) { + if (starts_with(partial_text, known)) { // Cumulative: the report is the whole transcript so far. - std::string fragment = partial_text.substr(assembled_.size()); - assembled_ = partial_text; - return fragment; + return release(partial_text.substr(known.size())); } // Incremental: the fragment is new text to append. - assembled_ += partial_text; - return partial_text; + // + // A family that REWRITES its hypothesis lands here too, and the client's + // view is then wrong in a way nothing downstream can fix. nemotron's + // decoder has such a branch (decoder.cpp:551-554): when the new text is not + // an extension of what it already emitted, it emits the whole new text. So + // "the cat sat" followed by "the cat sap" leaves the client holding + // "the cat satthe cat sap", and reconcile then correctly refuses to append + // to a contradicted assembly, which leaves concat(deltas) != final with no + // signal on the wire. This is NOT repaired here, and the reason is that a + // delta stream has no retraction: emitting only the differing suffix would + // read as "sap" appended to "the cat sat", which is a different wrong + // answer, and emitting a correction would need a wire field that does not + // exist. final_result carries the authoritative text either way. It did not + // fire in a 331 delta run, because an RNN-T decode is monotonic in practice. + return release(partial_text); } std::string TranscriptDeltaTracker::reconcile(const std::string &final_text) { if (final_text.empty() || final_text == assembled_) { + // Nothing further is owed. Held-back bytes are dropped rather than + // flushed: they are not in the authoritative text, so sending them + // would contradict it. + pending_.clear(); 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. + pending_.clear(); return {}; } - std::string fragment = final_text.substr(assembled_.size()); - assembled_ = final_text; - return fragment; + // Compared against the DELIVERED text, so the fragment below already + // contains whatever was held back. pending_ is therefore cleared rather + // than prepended, or those bytes would go out twice. + // + // The fragment ends on a character boundary whenever final_text is + // well-formed, which is the normal case and the reason a held-back sequence + // is always flushed here. It is still cut, so a family handing back a final + // text that is itself truncated mid-character cannot put a partial sequence + // on the wire through this path either. + pending_.clear(); + return release(final_text.substr(assembled_.size())); } } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/stream_delta.h b/backend/cpp/audio-cpp/stream_delta.h index 2de9c13a0..4600fbe87 100644 --- a/backend/cpp/audio-cpp/stream_delta.h +++ b/backend/cpp/audio-cpp/stream_delta.h @@ -19,9 +19,11 @@ // 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. +// hypothesis so far. process_available_stream_chunks hands +// every event it produces to the sink from INSIDE its loop +// (session.cpp:385-386) and RETURNS only the last of the +// batch, so the last event of each batch arrives twice and +// the others arrive once. // // 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 @@ -29,6 +31,22 @@ // 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. +// +// The cumulative reading is provably safe for voxtral, which is the family it +// matters for: its decode is a pure concatenation of per-token byte strings +// (tokenizer_text.cpp:171-183), so decode(ids[0..n]) is an unconditional BYTE +// PREFIX of decode(ids[0..n+1]) and one of its reports can never be mistaken +// for an incremental fragment. +// +// UTF-8 IS THE OTHER HALF OF THAT SAME FACT. Because that decode concatenates +// raw token BYTES, a multi-byte character is split across token boundaries, and +// the difference between two consecutive cumulative reports is then a lone +// continuation byte. TranscriptStreamResponse.delta is a proto3 `string`, whose +// wire format REQUIRES valid UTF-8: the C++ runtime serializes an invalid one +// with at most a warning, but the Go runtime refuses to unmarshal it, and the +// client loses every remaining delta AND the final_result. So no fragment this +// class returns ever ends inside a character; an incomplete trailing sequence is +// held back and merged into the next fragment. #include @@ -39,13 +57,14 @@ 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: + // The rules, in order, all of them against everything KNOWN (delivered + // plus held back), never against the delivered text alone: // 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. + // 2. A partial the known text ALREADY STARTS WITH has been accounted for: + // nothing is emitted. This is what absorbs voxtral's repeat of the + // last event in each batch, and a cumulative hypothesis that shrinks. + // 3. A partial that EXTENDS the known text is a cumulative report: only + // its new suffix is emitted. // 4. Anything else is an incremental fragment: it is emitted whole and // appended. // @@ -55,6 +74,11 @@ public: // 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. + // + // What comes back is the fragment MINUS any incomplete trailing UTF-8 + // sequence, which is carried into the next call. So an empty return can + // also mean "the only new bytes were half a character", and the caller + // needs no knowledge of that: writing nothing is exactly right. std::string observe(const std::string &partial_text); // Reconciles against TaskResult::text_output, which is authoritative, and @@ -64,17 +88,30 @@ public: // branch: with no partials observed, the assembly is empty and the whole // final text comes back as one delta. // + // It is also what FLUSHES a held-back UTF-8 sequence, and it can always do + // so: the final text is complete, so the fragment from the last delivered + // byte to its end ends on a character boundary. + // // 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. + // Everything the client has been sent, concatenated. Held-back bytes are + // deliberately NOT included: this is what the client holds, not what the + // tracker knows. const std::string &assembled() const noexcept { return assembled_; } private: + // Appends the emittable prefix of `fragment` to assembled_ and returns it, + // keeping any incomplete trailing UTF-8 sequence in pending_. + std::string release(const std::string &fragment); + std::string assembled_; + // An incomplete trailing UTF-8 sequence, computed but not sent. Always a + // proper prefix of one character, so at most three bytes. + std::string pending_; }; } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/stream_delta_test.cpp b/backend/cpp/audio-cpp/stream_delta_test.cpp index 07cd1df2f..65b4a400e 100644 --- a/backend/cpp/audio-cpp/stream_delta_test.cpp +++ b/backend/cpp/audio-cpp/stream_delta_test.cpp @@ -166,6 +166,187 @@ static void test_prefix_extension_is_read_as_cumulative() { check_eq(tracker.assembled(), "I'm", "cumulative reading assembles once"); } +// -------------------------------------------------------------------------- +// UTF-8 boundaries +// +// TranscriptStreamResponse.delta is a proto3 `string`, and the wire format +// REQUIRES a string field to be valid UTF-8. The C++ runtime serializes an +// invalid one with at most a warning; the Go runtime refuses to unmarshal it, +// so the client loses every delta AND the final_result still to come. +// +// Not hypothetical. voxtral_realtime reports the whole hypothesis as +// tokenizer_.decode(streaming_token_ids_), and that decode is a pure +// concatenation of raw token BYTES (tokenizer_text.cpp:171-183), so a +// multi-byte character is split across token boundaries and the cumulative +// difference between two consecutive reports is a lone continuation byte. +// -------------------------------------------------------------------------- + +// True when `text` is well-formed UTF-8. Written out here rather than reused +// from the implementation on purpose: a test that shares the implementation's +// idea of a boundary cannot catch the implementation's idea being wrong. +static bool is_valid_utf8(const std::string &text) { + size_t i = 0; + while (i < text.size()) { + const auto lead = static_cast(text[i]); + size_t length = 0; + if ((lead & 0x80) == 0x00) { + length = 1; + } else if ((lead & 0xE0) == 0xC0) { + length = 2; + } else if ((lead & 0xF0) == 0xE0) { + length = 3; + } else if ((lead & 0xF8) == 0xF0) { + length = 4; + } else { + return false; + } + if (i + length > text.size()) { + return false; + } + for (size_t k = 1; k < length; ++k) { + if ((static_cast(text[i + k]) & 0xC0) != 0x80) { + return false; + } + } + i += length; + } + return true; +} + +// Written as byte escapes so the test does not depend on the encoding of this +// source file. +static const std::string kEAcute = "\xC3\xA9"; // 2 bytes +static const std::string kEuro = "\xE2\x82\xAC"; // 3 bytes +static const std::string kEmoji = "\xF0\x9F\x8E\xA7"; // 4 bytes + +// A cumulative family advancing its hypothesis one BYTE at a time, which is +// what voxtral_realtime does across a multi-byte character. +static void test_cumulative_split_multibyte_character() { + TranscriptDeltaTracker tracker; + std::vector emitted; + const std::string full = "5" + kEuro; + std::vector partials; + for (size_t n = 1; n <= full.size(); ++n) { + partials.push_back(full.substr(0, n)); + } + const std::string view = client_view(tracker, partials, &emitted); + + check_eq(view, full, "a byte-at-a-time cumulative report still assembles"); + for (size_t i = 0; i < emitted.size(); ++i) { + check(is_valid_utf8(emitted[i]), + "cumulative fragment " + std::to_string(i) + " is valid UTF-8"); + } + check_eq(tracker.reconcile(full), "", "the final text adds nothing"); +} + +// An incremental family splitting a character across two fragments. +static void test_incremental_split_multibyte_character() { + TranscriptDeltaTracker tracker; + std::vector emitted; + const std::string view = client_view( + tracker, {"caf" + kEAcute.substr(0, 1), kEAcute.substr(1), " au lait"}, + &emitted); + + check_eq(view, "caf" + kEAcute + " au lait", + "an incremental split character still assembles"); + for (size_t i = 0; i < emitted.size(); ++i) { + check(is_valid_utf8(emitted[i]), + "incremental fragment " + std::to_string(i) + " is valid UTF-8"); + } + check(emitted.size() == 3, "one fragment out per partial, none swallowed"); + if (emitted.size() == 3) { + check_eq(emitted[0], "caf", "the lead byte of the character is held back"); + check_eq(emitted[1], kEAcute, + "the held byte is merged into the next fragment, not sent alone"); + check_eq(emitted[2], " au lait", "the rest follows unchanged"); + } +} + +// A 4 byte character split three ways, so the held-back buffer has to survive +// more than one round. +static void test_four_byte_character_split_three_ways() { + TranscriptDeltaTracker tracker; + std::vector emitted; + const std::string view = + client_view(tracker, + {"listen " + kEmoji.substr(0, 1), kEmoji.substr(1, 2), + kEmoji.substr(3), " now"}, + &emitted); + check_eq(view, "listen " + kEmoji + " now", + "a 4 byte character survives three splits"); + for (size_t i = 0; i < emitted.size(); ++i) { + check(is_valid_utf8(emitted[i]), + "4 byte fragment " + std::to_string(i) + " is valid UTF-8"); + } +} + +// The held-back bytes must reach the client. reconcile can always flush them, +// because the final text is complete by construction. +static void test_reconcile_flushes_a_held_back_sequence() { + TranscriptDeltaTracker tracker; + const std::string first = tracker.observe("done" + kEuro.substr(0, 2)); + check_eq(first, "done", "the incomplete trailing sequence is held back"); + check(is_valid_utf8(first), "what was emitted is valid UTF-8"); + const std::string tail = tracker.reconcile("done" + kEuro); + check_eq(tail, kEuro, "reconcile flushes the completed character"); + check(is_valid_utf8(tail), "the flushed tail is valid UTF-8"); + check_eq(tracker.assembled(), "done" + kEuro, "the client holds the whole text"); +} + +// Held-back bytes are not lost track of: the next cumulative report emits the +// whole character rather than only the bytes that just arrived. +static void test_held_bytes_join_the_next_fragment() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("a" + kEuro.substr(0, 1)), "a", + "only the complete prefix goes out"); + check_eq(tracker.observe("a" + kEuro), kEuro, + "the next cumulative report emits the whole character at once"); + check_eq(tracker.assembled(), "a" + kEuro, "assembly is correct"); +} + +// A whole multi-byte character arriving at once must NOT be held back: holding +// a complete sequence would stall every stream by one character. +static void test_a_complete_character_is_not_held() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe("x" + kEuro), "x" + kEuro, + "a fragment ending on a boundary is emitted immediately"); + check_eq(tracker.observe("x" + kEuro + kEmoji), kEmoji, + "and so is the next one"); +} + +// Bytes that can never complete must not be held forever: a stray continuation +// byte or an invalid lead is passed through rather than stalling the stream. +// Repairing a family's malformed output is not something a delta tracker can do +// without altering the transcript. +static void test_undecodable_bytes_are_not_held_forever() { + TranscriptDeltaTracker tracker; + check_eq(tracker.observe(std::string("ok\x80")), std::string("ok\x80"), + "a stray continuation byte is passed through, not held"); + check_eq(tracker.observe(std::string("ok\x80") + "next"), "next", + "the stream continues"); + + // A lead byte the encoding does not define (0xF8 and above). Nothing can + // ever complete it, so holding it would stall the stream for good. + TranscriptDeltaTracker invalid_lead; + check_eq(invalid_lead.observe(std::string("ok\xFE")), std::string("ok\xFE"), + "an undefined lead byte is passed through, not held"); + check_eq(invalid_lead.observe(std::string("ok\xFE") + "more"), "more", + "the stream continues past an undefined lead byte"); + + // The same byte followed by continuation bytes, which is the shape that + // looks most like a real sequence waiting to be completed. + TranscriptDeltaTracker invalid_run; + check_eq(invalid_run.observe(std::string("\xFE\x80\x80")), std::string("\xFE\x80\x80"), + "an undefined lead with continuations is passed through"); + + // Five continuation bytes with no lead in sight: the scan gives up rather + // than walking back through the whole buffer. + TranscriptDeltaTracker orphans; + check_eq(orphans.observe(std::string("\x80\x80\x80\x80\x80")), + std::string("\x80\x80\x80\x80\x80"), + "a run of orphan continuation bytes is passed through"); +} + int main() { test_incremental_family(); test_cumulative_family_with_duplicate_delivery(); @@ -177,6 +358,13 @@ int main() { test_empty_final_text(); test_whitespace_fragments_survive(); test_prefix_extension_is_read_as_cumulative(); + test_cumulative_split_multibyte_character(); + test_incremental_split_multibyte_character(); + test_four_byte_character_split_three_ways(); + test_reconcile_flushes_a_held_back_sequence(); + test_held_bytes_join_the_next_fragment(); + test_a_complete_character_is_not_held(); + test_undecodable_bytes_are_not_held_forever(); if (failures) { fprintf(stderr, "%d check(s) failed\n", failures); return 1; diff --git a/backend/cpp/audio-cpp/streaming_driver_ctest.cpp b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp index 6fc225ed0..469cce2b2 100644 --- a/backend/cpp/audio-cpp/streaming_driver_ctest.cpp +++ b/backend/cpp/audio-cpp/streaming_driver_ctest.cpp @@ -440,6 +440,39 @@ static void test_chunking_honours_the_policy_sample_count() { } } +// A buffer whose float count is not a whole number of frames is REFUSED rather +// than truncated. The integer division would otherwise drop the tail floats +// from the fed audio, and therefore from the transcript, with no diagnostic. +static void test_a_partial_trailing_frame_is_refused() { + audiocpp_backend::InferenceLane lane("test"); + audiocpp_backend::LaneEntry entry(lane, 0); + FakeAudioSession fake; + fake.policy.preferred_audio_chunk_samples = 100; + const auto session = streaming_session(fake, audiocpp_backend::Task::Asr); + + // 501 floats across 2 channels: 250 whole frames and one stray float. + rt::TaskRequest request; + rt::AudioBuffer audio; + audio.sample_rate = 48000; + audio.channels = 2; + audio.samples.assign(501, 0.5F); + request.audio_input = std::move(audio); + + bool threw_config = false; + try { + audiocpp_backend::run_streaming_audio(session, request, *request.audio_input, + [](const rt::StreamEvent &) {}, entry); + } catch (const audiocpp_backend::ConfigError &) { + threw_config = true; + } catch (const std::exception &) { + } + check(threw_config, + "a buffer that is not a whole number of frames is refused with ConfigError"); + check(fake.calls.empty(), + "the refusal precedes every call into the session, so no half-started " + "stream is left on the cached one"); +} + // 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() { @@ -613,6 +646,7 @@ int main() { 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_a_partial_trailing_frame_is_refused(); 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();