From 9409e16176c2d6ffab4e483e9af0a13e722f194c Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 18:32:53 -0600 Subject: [PATCH 1/7] parakeet_tdt: publish partials as increments, not the whole transcript A partial is the text decoded since the last one. The CLI's PartialTextRenderer says so and appends them into a scrolling transcript; the server forwards each one as an OpenAI-shaped `transcript.text.delta`, which is incremental by specification. Every other streaming ASR family emits the increment -- vibevoice_asr and higgs_audio_stt via a common-prefix diff, voxtral_realtime via a published-bytes offset. This family published `merged_decode()`, which re-renders the whole transcript from every token decoded so far. So a client that does what the contract says built the transcript back up quadratically: partial_text=Some call me nat partial_text=Some call me nature. Others call me partial_text=Some call me nature. Others call me Mother Nature. I' ... appending to "Some call me natSome call me nature. Others call me...". Now: partial_text=Some call me nat partial_text=ure. Others call me partial_text= Mother Nature. I' partial_text=ve been here for over four point partial_text= five billion years partial_text=, twenty two thousand five partial_text= hundred times longer than you. which concatenates byte for byte to the `text_output` the same run reports. Diffed against what was already published rather than tracking a byte count, because re-decoding with more right context can revise earlier text rather than only extending it; the common prefix is backed off to a UTF-8 boundary so a revision landing mid-character cannot split a code point across two deltas. No server change: `run_transcription_stream` already skips an event with no partial text, and `emit_if_nonempty` only drops an event when every field is empty, so a window that decodes no new text still delivers its word timestamps. Verified against Parakeet-TDT-0.6B-v3 q8_0 on assets/resources/sample_16k.wav, the before and after above coming from that run. --- .../community_models/parakeet_tdt/session.h | 4 +++ src/community_models/parakeet_tdt/session.cpp | 29 ++++++++++++++++++- 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/include/engine/community_models/parakeet_tdt/session.h b/include/engine/community_models/parakeet_tdt/session.h index 8292a00ef..9148c2e69 100644 --- a/include/engine/community_models/parakeet_tdt/session.h +++ b/include/engine/community_models/parakeet_tdt/session.h @@ -125,6 +125,10 @@ class ParakeetTDTStreamingSession final std::vector token_frame_indices_; std::vector token_durations_; runtime::StreamEventCallback stream_event_sink_; + // The transcript already published as partials. merged_decode() returns the + // whole transcript each time, so this is what turns it back into the + // increment a partial is contracted to be. + std::string emitted_text_; bool stream_started_ = false; bool finalized_ = false; }; diff --git a/src/community_models/parakeet_tdt/session.cpp b/src/community_models/parakeet_tdt/session.cpp index 2f6232583..c6b1272ac 100644 --- a/src/community_models/parakeet_tdt/session.cpp +++ b/src/community_models/parakeet_tdt/session.cpp @@ -217,6 +217,23 @@ int64_t seconds_to_samples(float seconds, int sample_rate) { static_cast(seconds) * static_cast(sample_rate))); } +// Longest common prefix of the published transcript and a fresh decode, backed +// off to a UTF-8 boundary. Re-decoding with more right context can revise what +// was already sent rather than only extending it, and without the boundary +// check a revision landing mid-character would split a code point across two +// deltas. Parakeet v3 is multilingual, so that is reachable rather than +// theoretical. +size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { + size_t size = 0; + while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { + ++size; + } + while (size > 0 && (static_cast(rhs[size]) & 0xC0) == 0x80) { + --size; + } + return size; +} + } // namespace ParakeetTDTSessionBase::ParakeetTDTSessionBase( @@ -789,6 +806,7 @@ void ParakeetTDTStreamingSession::reset() { token_ids_.clear(); token_frame_indices_.clear(); token_durations_.clear(); + emitted_text_.clear(); decoder_->reset_state(); stream_started_ = true; finalized_ = false; @@ -912,7 +930,16 @@ runtime::StreamEvent ParakeetTDTStreamingSession::process_ready_windows(bool flu runtime::StreamEvent event; if (changed && !token_ids_.empty()) { auto decoded = merged_decode(); - event.partial_text = runtime::Transcript{decoded.text, ""}; + // A partial is the text decoded since the last one: the CLI appends + // them into a scrolling transcript and the server forwards each as a + // transcript.text.delta. merged_decode() re-renders the whole + // transcript from every token so far, so publishing it unchanged made + // an appending client build "Some call meSome call me nature". + const size_t published = common_prefix_size(emitted_text_, decoded.text); + if (published < decoded.text.size()) { + event.partial_text = runtime::Transcript{decoded.text.substr(published), ""}; + } + emitted_text_ = decoded.text; event.word_timestamps = std::move(decoded.word_timestamps); // The last word has no following word boundary yet, so it remains // provisional and is withheld from the finalized timestamp list. From ee5338b6796a024ee6b6107d210dfe9e0615f2b1 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 18:36:25 -0600 Subject: [PATCH 2/7] parakeet_tdt: hold back a partial that ends mid-character Review follow-up to the commit before it. Two defects in that change. **A delta could end part way through a character.** The UTF-8 backoff only guarded the point where the diff diverges -- the *start* of a delta -- and did nothing about its end. The tokenizer falls back to bytes for text its vocabulary does not cover, so a decode can stop mid-sequence: with `emitted_text_` at "abc" and a decode of "abc\xE4", the prefix is 3, \xE4 is a lead byte rather than a continuation so nothing was backed off, and the lone \xE4 went out as the delta. That is invalid UTF-8 by the time it reaches the SSE JSON. `complete_utf8_end` now bounds a delta at the last complete sequence, and `emitted_text_` records what actually went out, so a held-back character is reconsidered against the decode that completes it rather than skipped. The final `text_output` is unaffected either way, so nothing is lost if a stream ends with a character still held. **The two fields in the event have different shapes, and nothing said so.** `partial_text` is now what is new while `word_timestamps` stays cumulative. That is deliberate -- word_timestamps is not a delta field, it is the finalized set so far, which is why the provisional last word is dropped -- but with the text field changing shape it is worth stating rather than leaving to be rediscovered. Comment only. Helpers exercised directly over a replay of decode sequences: ASCII growth, a 3-byte character arriving one byte at a time, Cyrillic growth, an incomplete tail, and a shrinking decode. Re-ran Parakeet-TDT-0.6B-v3 q8_0 on assets/resources/sample_16k.wav: partials unchanged and still concatenating byte for byte to text_output. --- src/community_models/parakeet_tdt/session.cpp | 56 ++++++++++++++++--- 1 file changed, 48 insertions(+), 8 deletions(-) diff --git a/src/community_models/parakeet_tdt/session.cpp b/src/community_models/parakeet_tdt/session.cpp index c6b1272ac..c2df3351b 100644 --- a/src/community_models/parakeet_tdt/session.cpp +++ b/src/community_models/parakeet_tdt/session.cpp @@ -218,11 +218,9 @@ int64_t seconds_to_samples(float seconds, int sample_rate) { } // Longest common prefix of the published transcript and a fresh decode, backed -// off to a UTF-8 boundary. Re-decoding with more right context can revise what -// was already sent rather than only extending it, and without the boundary -// check a revision landing mid-character would split a code point across two -// deltas. Parakeet v3 is multilingual, so that is reachable rather than -// theoretical. +// off to a UTF-8 boundary. Re-decoding can revise what was already published +// rather than only extending it, and without the backoff a divergence landing +// mid-character would start a delta half way through a code point. size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { size_t size = 0; while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { @@ -234,6 +232,38 @@ size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { return size; } +// How much of a decode is safe to publish: everything up to the last complete +// UTF-8 sequence. The tokenizer falls back to bytes for text its vocabulary +// does not cover, so a decode can end part way through a character -- and a +// delta cut there is invalid UTF-8 by the time it reaches the SSE JSON. The +// remainder is held back and goes out with the window that completes it. +size_t complete_utf8_end(const std::string & text) { + size_t lead = text.size(); + while (lead > 0 && (static_cast(text[lead - 1]) & 0xC0) == 0x80) { + --lead; + } + if (lead == 0) { + return text.size(); + } + --lead; + const auto first = static_cast(text[lead]); + size_t needed = 0; + if ((first & 0x80) == 0x00) { + needed = 1; + } else if ((first & 0xE0) == 0xC0) { + needed = 2; + } else if ((first & 0xF0) == 0xE0) { + needed = 3; + } else if ((first & 0xF8) == 0xF0) { + needed = 4; + } else { + // Not a lead byte at all. Nothing sensible to hold back, and dropping + // bytes would lose text, so publish it and let the consumer see it. + return text.size(); + } + return (text.size() - lead) < needed ? lead : text.size(); +} + } // namespace ParakeetTDTSessionBase::ParakeetTDTSessionBase( @@ -935,11 +965,21 @@ runtime::StreamEvent ParakeetTDTStreamingSession::process_ready_windows(bool flu // transcript.text.delta. merged_decode() re-renders the whole // transcript from every token so far, so publishing it unchanged made // an appending client build "Some call meSome call me nature". + const size_t publishable = complete_utf8_end(decoded.text); const size_t published = common_prefix_size(emitted_text_, decoded.text); - if (published < decoded.text.size()) { - event.partial_text = runtime::Transcript{decoded.text.substr(published), ""}; + if (published < publishable) { + event.partial_text = runtime::Transcript{ + decoded.text.substr(published, publishable - published), ""}; } - emitted_text_ = decoded.text; + // What has actually gone out, so a character held back above is + // reconsidered against the next decode rather than skipped. + emitted_text_.assign(decoded.text, 0, publishable); + // word_timestamps stays cumulative, deliberately. It is not a delta + // field: it is the finalized set so far, which is why the provisional + // last word is dropped below rather than carried. So the two fields in + // this event have different shapes on purpose -- partial_text is what + // is new, word_timestamps is everything settled -- because each matches + // its own contract rather than each other. event.word_timestamps = std::move(decoded.word_timestamps); // The last word has no following word boundary yet, so it remains // provisional and is withheld from the finalized timestamp list. From 6c82c192c02b878a57a02f514415beaf4b422041 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 18:43:49 -0600 Subject: [PATCH 3/7] runtime: one shared publisher for streaming partial text Scope expansion on the two commits before it, which fixed parakeet_tdt alone. The reason parakeet was wrong is that there was nothing to be right with: every family that publishes partial text carried its own copy of the arithmetic, and the copies disagreed. Three shapes across seven families: - a common-prefix diff, duplicated verbatim in higgs_audio_stt and vibevoice_asr, helper and caller both - a published-bytes offset, open-coded in voxtral_realtime, qwen3_asr and sense_asr - no diffing at all -- parakeet_tdt and kroko_asr published the whole running transcript as the partial kroko_asr is the second instance of the bug #68 reported, found by sweeping for the pattern rather than by hitting it: `event.partial_text = result.text_output` where result is the combined decode of everything so far. Same corruption, same cause, and it would have survived a parakeet-only fix. engine::runtime::PartialTextPublisher is now the single implementation and all seven use it. It keeps what has actually been published and returns the increment, with the two properties none of the copies had: - a delta never ends part way through a UTF-8 sequence. A tokenizer falls back to bytes for text its vocabulary does not cover, so a decode can stop mid-character; publishing that puts half a code point on the wire and it reaches the SSE JSON encoder as invalid UTF-8. The tail is held for the update that completes it. - a delta never starts inside one either, when a decode revises published text rather than only extending it. Nothing can retract a delta already sent, so the consumer is wrong either way on a real revision; this keeps it from also being spliced into the middle of a character. tests/unittests/test_partial_text.cpp covers growth, 2/3/4-byte characters arriving one byte at a time, a revision, a shrinking decode, reset, and the empty edges. Verified: partial_text_test passes; Parakeet-TDT-0.6B-v3 q8_0 on assets/resources/sample_16k.wav still emits seven partials concatenating byte for byte to text_output; and the AudioCpp-Bindings suite runs green against this build across kokoro_tts, citrinet_asr, parakeet_tdt, sortformer_diar and bs_roformer, C and C# agreeing on every reported value. --- CMakeLists.txt | 3 + .../community_models/kroko_asr/session.h | 2 + .../community_models/parakeet_tdt/session.h | 9 +- .../community_models/sense_asr/session.h | 3 +- .../engine/framework/runtime/partial_text.h | 57 +++++++++ include/engine/models/qwen3_asr/session.h | 3 +- .../engine/models/voxtral_realtime/session.h | 3 +- src/community_models/kroko_asr/session.cpp | 17 ++- src/community_models/parakeet_tdt/session.cpp | 60 +--------- src/community_models/sense_asr/session.cpp | 8 +- src/framework/runtime/partial_text.cpp | 59 ++++++++++ src/models/higgs_audio_stt/session.cpp | 27 ++--- src/models/qwen3_asr/session.cpp | 7 +- src/models/vibevoice_asr/session.cpp | 27 ++--- src/models/voxtral_realtime/session.cpp | 8 +- tests/unittests/test_partial_text.cpp | 108 ++++++++++++++++++ 16 files changed, 289 insertions(+), 112 deletions(-) create mode 100644 include/engine/framework/runtime/partial_text.h create mode 100644 src/framework/runtime/partial_text.cpp create mode 100644 tests/unittests/test_partial_text.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 5fd53e108..3a6ca6ec8 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -412,6 +412,7 @@ add_library(engine_core OBJECT src/framework/model_spec/metadata.cpp src/framework/runtime/session.cpp src/framework/runtime/task_vocabulary.cpp + src/framework/runtime/partial_text.cpp src/framework/runtime/artifacts.cpp src/framework/runtime/cache.cpp src/framework/runtime/graph_executor.cpp @@ -2485,6 +2486,8 @@ if (ENGINE_BUILD_TESTS OR ENGINE_BUILD_EXTENDED_TESTS OR ENGINE_BUILD_MODEL_TEST add_engine_unittest(audio_chunking_test tests/unittests/test_audio_chunking.cpp) add_test(NAME audio_chunking_test COMMAND audio_chunking_test) + add_engine_unittest(partial_text_test tests/unittests/test_partial_text.cpp) + add_test(NAME partial_text_test COMMAND partial_text_test) add_engine_unittest(wav_writer_formats_test tests/unittests/test_wav_writer_formats.cpp) add_test(NAME wav_writer_formats_test COMMAND wav_writer_formats_test) diff --git a/include/engine/community_models/kroko_asr/session.h b/include/engine/community_models/kroko_asr/session.h index 6611e9327..d317b9b7a 100644 --- a/include/engine/community_models/kroko_asr/session.h +++ b/include/engine/community_models/kroko_asr/session.h @@ -5,6 +5,7 @@ #include "engine/community_models/kroko_asr/encoder.h" #include "engine/community_models/kroko_asr/tokenizer.h" #include "engine/community_models/kroko_asr/zipformer.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/model_spec/metadata.h" #include "engine/framework/runtime/session_base.h" @@ -83,6 +84,7 @@ class KrokoASRSession final std::vector streaming_resampler_source_; int64_t processed_feature_offset_ = 0; int64_t streaming_total_samples_ = 0; + runtime::PartialTextPublisher streaming_partials_; int64_t streaming_source_offset_ = 0; int64_t streaming_source_frames_ = 0; int64_t streaming_next_output_sample_ = 0; diff --git a/include/engine/community_models/parakeet_tdt/session.h b/include/engine/community_models/parakeet_tdt/session.h index 9148c2e69..5f1706551 100644 --- a/include/engine/community_models/parakeet_tdt/session.h +++ b/include/engine/community_models/parakeet_tdt/session.h @@ -13,6 +13,8 @@ #include #include #include + +#include "engine/framework/runtime/partial_text.h" #include #include @@ -125,10 +127,9 @@ class ParakeetTDTStreamingSession final std::vector token_frame_indices_; std::vector token_durations_; runtime::StreamEventCallback stream_event_sink_; - // The transcript already published as partials. merged_decode() returns the - // whole transcript each time, so this is what turns it back into the - // increment a partial is contracted to be. - std::string emitted_text_; + // merged_decode() returns the whole transcript each time; this turns it + // back into the increment a partial is contracted to be. + runtime::PartialTextPublisher partials_; bool stream_started_ = false; bool finalized_ = false; }; diff --git a/include/engine/community_models/sense_asr/session.h b/include/engine/community_models/sense_asr/session.h index 385a78c95..2eb85a988 100644 --- a/include/engine/community_models/sense_asr/session.h +++ b/include/engine/community_models/sense_asr/session.h @@ -4,6 +4,7 @@ #include "engine/community_models/sense_asr/encoder.h" #include "engine/community_models/sense_asr/frontend.h" #include "engine/community_models/sense_asr/types.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/assets/tensor_source.h" #include "engine/framework/model_spec/metadata.h" #include "engine/framework/runtime/session_base.h" @@ -87,7 +88,7 @@ class SenseAsrSession final : public runtime::RuntimeSessionBase, runtime::AudioBuffer streaming_audio_; size_t streaming_audio_offset_values_ = 0; std::string streaming_text_; - size_t streaming_published_bytes_ = 0; + runtime::PartialTextPublisher streaming_partials_; int64_t streaming_windows_processed_ = 0; runtime::StreamEventCallback stream_event_sink_; bool stream_started_ = false; diff --git a/include/engine/framework/runtime/partial_text.h b/include/engine/framework/runtime/partial_text.h new file mode 100644 index 000000000..c2ad2cd24 --- /dev/null +++ b/include/engine/framework/runtime/partial_text.h @@ -0,0 +1,57 @@ +#pragma once + +#include + +namespace engine::runtime { + +// Turns a running transcript into the increments a partial is contracted to be. +// +// A streaming session's partial_text is the text decoded since the last one. +// The CLI appends them into a scrolling transcript (see PartialTextRenderer) +// and the reference server forwards each as an OpenAI-shaped +// `transcript.text.delta`, which is incremental by specification. A family that +// holds a whole running transcript therefore has to publish the difference, and +// every family that did so carried its own copy of the arithmetic. +// +// The copies disagreed and two of them were wrong, which is why this is shared +// rather than duplicated once more: +// +// - A decode can end part way through a UTF-8 sequence, because a tokenizer +// falls back to bytes for text its vocabulary does not cover. Publishing +// that puts half a code point on the wire, and it reaches a JSON encoder +// as invalid UTF-8. The tail is held back for the update that completes it. +// +// - A decode can revise text already published rather than only extending it. +// Nothing can retract a delta that has gone out, so the client's transcript +// is wrong either way; starting the next delta at the divergence at least +// keeps it from being spliced into the middle of a character. +// +// Not thread safe: a session owns one of these and publishes from the thread +// driving the stream. +class PartialTextPublisher { +public: + // The text newly decoded since the last call. Empty when nothing is new, or + // when everything new is an incomplete character still being held. + std::string publish(const std::string & transcript); + + // Forgets what has gone out, for a session starting a fresh stream. + void reset() { published_.clear(); } + + // What has actually been published, which trails `transcript` whenever a + // character is being held back. + const std::string & published() const { return published_; } + +private: + std::string published_; +}; + +// Longest common prefix of two transcripts, backed off so the result never +// lands inside a UTF-8 sequence. Exposed for tests and for callers that need +// the offset rather than the text. +size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs); + +// How much of `text` is safe to publish: everything up to the last complete +// UTF-8 sequence. Equal to text.size() unless the tail is a partial character. +size_t transcript_publishable_end(const std::string & text); + +} // namespace engine::runtime diff --git a/include/engine/models/qwen3_asr/session.h b/include/engine/models/qwen3_asr/session.h index ccf7f5b89..f76d2e14d 100644 --- a/include/engine/models/qwen3_asr/session.h +++ b/include/engine/models/qwen3_asr/session.h @@ -1,5 +1,6 @@ #pragma once +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/assets/tensor_source.h" #include "engine/framework/runtime/session_base.h" #include "engine/models/qwen3_asr/assets.h" @@ -88,7 +89,7 @@ class Qwen3ASRSession final runtime::AudioBuffer streaming_audio_; size_t streaming_audio_offset_values_ = 0; std::string streaming_text_; - size_t streaming_published_bytes_ = 0; + runtime::PartialTextPublisher streaming_partials_; int64_t streaming_windows_processed_ = 0; runtime::StreamEventCallback stream_event_sink_; bool stream_started_ = false; diff --git a/include/engine/models/voxtral_realtime/session.h b/include/engine/models/voxtral_realtime/session.h index 03c934438..a5a43f616 100644 --- a/include/engine/models/voxtral_realtime/session.h +++ b/include/engine/models/voxtral_realtime/session.h @@ -1,5 +1,6 @@ #pragma once +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include "engine/models/voxtral_realtime/assets.h" #include "engine/models/voxtral_realtime/audio_encoder.h" @@ -81,7 +82,7 @@ class VoxtralRealtimeSession final // The transcript decoded so far, and how much of it has already gone out as a partial. Every // partial is the suffix between the two, so the deltas concatenate to exactly this string. std::string streaming_text_; - size_t streaming_published_bytes_ = 0; + runtime::PartialTextPublisher streaming_partials_; int64_t streaming_token_count_ = 0; int32_t previous_stream_token_ = 0; bool stream_started_ = false; diff --git a/src/community_models/kroko_asr/session.cpp b/src/community_models/kroko_asr/session.cpp index 7fd812149..efc769264 100644 --- a/src/community_models/kroko_asr/session.cpp +++ b/src/community_models/kroko_asr/session.cpp @@ -627,6 +627,7 @@ void KrokoASRSession::reset() { endpoint_segments_.clear(); processed_feature_offset_ = 0; streaming_total_samples_ = 0; + streaming_partials_.reset(); streaming_source_offset_ = 0; streaming_source_frames_ = 0; streaming_next_output_sample_ = 0; @@ -903,7 +904,21 @@ runtime::StreamEvent KrokoASRSession::process_streaming_audio( decoded, streaming_total_samples_, streaming_language_); - event.partial_text = result.text_output; + // A partial is the text decoded since the last one, not the whole + // transcript re-rendered: combined_decoded() restates everything + // decoded so far, so an appending consumer built it up quadratically. + // word_timestamps stays cumulative -- it is the finalized set, not a + // delta -- so the two fields differ on purpose. + if (result.text_output.has_value()) { + std::string delta = + streaming_partials_.publish( + result.text_output->text); + if (!delta.empty()) { + event.partial_text = runtime::Transcript{ + std::move(delta), + result.text_output->language}; + } + } event.word_timestamps = result.word_timestamps; } return event; diff --git a/src/community_models/parakeet_tdt/session.cpp b/src/community_models/parakeet_tdt/session.cpp index c2df3351b..005e91f7b 100644 --- a/src/community_models/parakeet_tdt/session.cpp +++ b/src/community_models/parakeet_tdt/session.cpp @@ -217,53 +217,6 @@ int64_t seconds_to_samples(float seconds, int sample_rate) { static_cast(seconds) * static_cast(sample_rate))); } -// Longest common prefix of the published transcript and a fresh decode, backed -// off to a UTF-8 boundary. Re-decoding can revise what was already published -// rather than only extending it, and without the backoff a divergence landing -// mid-character would start a delta half way through a code point. -size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { - size_t size = 0; - while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { - ++size; - } - while (size > 0 && (static_cast(rhs[size]) & 0xC0) == 0x80) { - --size; - } - return size; -} - -// How much of a decode is safe to publish: everything up to the last complete -// UTF-8 sequence. The tokenizer falls back to bytes for text its vocabulary -// does not cover, so a decode can end part way through a character -- and a -// delta cut there is invalid UTF-8 by the time it reaches the SSE JSON. The -// remainder is held back and goes out with the window that completes it. -size_t complete_utf8_end(const std::string & text) { - size_t lead = text.size(); - while (lead > 0 && (static_cast(text[lead - 1]) & 0xC0) == 0x80) { - --lead; - } - if (lead == 0) { - return text.size(); - } - --lead; - const auto first = static_cast(text[lead]); - size_t needed = 0; - if ((first & 0x80) == 0x00) { - needed = 1; - } else if ((first & 0xE0) == 0xC0) { - needed = 2; - } else if ((first & 0xF0) == 0xE0) { - needed = 3; - } else if ((first & 0xF8) == 0xF0) { - needed = 4; - } else { - // Not a lead byte at all. Nothing sensible to hold back, and dropping - // bytes would lose text, so publish it and let the consumer see it. - return text.size(); - } - return (text.size() - lead) < needed ? lead : text.size(); -} - } // namespace ParakeetTDTSessionBase::ParakeetTDTSessionBase( @@ -836,7 +789,7 @@ void ParakeetTDTStreamingSession::reset() { token_ids_.clear(); token_frame_indices_.clear(); token_durations_.clear(); - emitted_text_.clear(); + partials_.reset(); decoder_->reset_state(); stream_started_ = true; finalized_ = false; @@ -965,15 +918,10 @@ runtime::StreamEvent ParakeetTDTStreamingSession::process_ready_windows(bool flu // transcript.text.delta. merged_decode() re-renders the whole // transcript from every token so far, so publishing it unchanged made // an appending client build "Some call meSome call me nature". - const size_t publishable = complete_utf8_end(decoded.text); - const size_t published = common_prefix_size(emitted_text_, decoded.text); - if (published < publishable) { - event.partial_text = runtime::Transcript{ - decoded.text.substr(published, publishable - published), ""}; + std::string delta = partials_.publish(decoded.text); + if (!delta.empty()) { + event.partial_text = runtime::Transcript{std::move(delta), ""}; } - // What has actually gone out, so a character held back above is - // reconsidered against the next decode rather than skipped. - emitted_text_.assign(decoded.text, 0, publishable); // word_timestamps stays cumulative, deliberately. It is not a delta // field: it is the finalized set so far, which is why the provisional // last word is dropped below rather than carried. So the two fields in diff --git a/src/community_models/sense_asr/session.cpp b/src/community_models/sense_asr/session.cpp index 7dc25041d..e784856a0 100644 --- a/src/community_models/sense_asr/session.cpp +++ b/src/community_models/sense_asr/session.cpp @@ -402,7 +402,7 @@ void SenseAsrSession::reset() { streaming_audio_ = runtime::AudioBuffer{}; streaming_audio_offset_values_ = 0; streaming_text_.clear(); - streaming_published_bytes_ = 0; + streaming_partials_.reset(); streaming_windows_processed_ = 0; stream_started_ = false; stream_wall_start_ = {}; @@ -758,12 +758,12 @@ SenseAsrSession::process_one_stream_chunk(const runtime::AudioBuffer &audio) { streaming_result_.text_output->language = item.text_output->language; } streaming_result_.text_output->text = streaming_text_; - if (streaming_published_bytes_ < streaming_text_.size()) { + if (std::string partial = streaming_partials_.publish(streaming_text_); + !partial.empty()) { event.partial_text = runtime::Transcript{ - streaming_text_.substr(streaming_published_bytes_), + std::move(partial), streaming_result_.text_output->language, }; - streaming_published_bytes_ = streaming_text_.size(); } return event; } diff --git a/src/framework/runtime/partial_text.cpp b/src/framework/runtime/partial_text.cpp new file mode 100644 index 000000000..4c1e23d36 --- /dev/null +++ b/src/framework/runtime/partial_text.cpp @@ -0,0 +1,59 @@ +#include "engine/framework/runtime/partial_text.h" + +namespace engine::runtime { + +size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs) { + size_t size = 0; + while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { + ++size; + } + // A divergence inside a character would otherwise start the next delta on a + // continuation byte. + while (size > 0 && (static_cast(rhs[size]) & 0xC0) == 0x80) { + --size; + } + return size; +} + +size_t transcript_publishable_end(const std::string & text) { + size_t lead = text.size(); + while (lead > 0 && (static_cast(text[lead - 1]) & 0xC0) == 0x80) { + --lead; + } + if (lead == 0) { + // All continuation bytes, or empty: nothing to anchor a decision on. + return text.size(); + } + --lead; + const auto first = static_cast(text[lead]); + size_t needed = 0; + if ((first & 0x80) == 0x00) { + needed = 1; + } else if ((first & 0xE0) == 0xC0) { + needed = 2; + } else if ((first & 0xF0) == 0xE0) { + needed = 3; + } else if ((first & 0xF8) == 0xF0) { + needed = 4; + } else { + // Not a lead byte at all, so this is not text this can reason about. + // Holding bytes back would lose them; publish and let the consumer see. + return text.size(); + } + return (text.size() - lead) < needed ? lead : text.size(); +} + +std::string PartialTextPublisher::publish(const std::string & transcript) { + const size_t publishable = transcript_publishable_end(transcript); + const size_t already = transcript_common_prefix(published_, transcript); + std::string delta; + if (already < publishable) { + delta = transcript.substr(already, publishable - already); + } + // What actually went out, so a held-back character is reconsidered against + // the update that completes it rather than skipped. + published_.assign(transcript, 0, publishable); + return delta; +} + +} // namespace engine::runtime diff --git a/src/models/higgs_audio_stt/session.cpp b/src/models/higgs_audio_stt/session.cpp index 12b4d4041..6016a24bf 100644 --- a/src/models/higgs_audio_stt/session.cpp +++ b/src/models/higgs_audio_stt/session.cpp @@ -1,5 +1,6 @@ #include "engine/models/higgs_audio_stt/session.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/audio/chunking.h" #include "engine/framework/debug/profiler.h" #include "engine/framework/runtime/options.h" @@ -55,33 +56,23 @@ int64_t audio_frame_count(const runtime::AudioBuffer & audio) { return static_cast(audio.samples.size() / static_cast(audio.channels)); } -size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { - const size_t limit = std::min(lhs.size(), rhs.size()); - size_t size = 0; - while (size < limit && lhs[size] == rhs[size]) { - ++size; - } - return size; -} - void emit_transcript_delta( const runtime::StreamEventCallback & sink, const runtime::Transcript & transcript, - std::string & emitted_text) { + runtime::PartialTextPublisher & partials) { if (!sink || transcript.text.empty()) { return; } - const size_t prefix_size = common_prefix_size(emitted_text, transcript.text); - if (prefix_size == transcript.text.size()) { - emitted_text = transcript.text; + std::string delta = partials.publish(transcript.text); + if (delta.empty()) { return; } runtime::StreamEvent event; - event.partial_text = runtime::Transcript{transcript.text.substr(prefix_size), transcript.language}; + event.partial_text = runtime::Transcript{std::move(delta), transcript.language}; sink(event); - emitted_text = transcript.text; } + std::string append_streaming_transcript( runtime::TaskResult & total, const runtime::Transcript & chunk) { @@ -390,7 +381,7 @@ runtime::TaskResult HiggsAudioSTTSession::run_single(const HiggsAudioSTTRequest const auto audio_embeddings = audio_encoder_.encode(features); const auto encoder_end = Clock::now(); const auto text_decoder_start = Clock::now(); - std::string emitted_text; + runtime::PartialTextPublisher partials; HiggsAudioSTTTokenCallback token_callback; if (task_.mode == runtime::RunMode::Streaming && stream_event_sink_ != nullptr) { token_callback = [&](const HiggsAudioSTTGeneratedTokens & partial_tokens) { @@ -398,7 +389,7 @@ runtime::TaskResult HiggsAudioSTTSession::run_single(const HiggsAudioSTTRequest emit_transcript_delta( stream_event_sink_, runtime::Transcript{partial.text, partial.language}, - emitted_text); + partials); }; } const auto tokens = text_decoder_.generate(prompt, audio_embeddings, asr_request.generation, token_callback); @@ -409,7 +400,7 @@ runtime::TaskResult HiggsAudioSTTSession::run_single(const HiggsAudioSTTRequest emit_transcript_delta( stream_event_sink_, runtime::Transcript{decoded.text, decoded.language}, - emitted_text); + partials); } const auto postprocess_end = Clock::now(); diff --git a/src/models/qwen3_asr/session.cpp b/src/models/qwen3_asr/session.cpp index 5358a9ed0..1dadfb1eb 100644 --- a/src/models/qwen3_asr/session.cpp +++ b/src/models/qwen3_asr/session.cpp @@ -407,7 +407,7 @@ void Qwen3ASRSession::reset() { streaming_audio_ = runtime::AudioBuffer{}; streaming_audio_offset_values_ = 0; streaming_text_.clear(); - streaming_published_bytes_ = 0; + streaming_partials_.reset(); streaming_windows_processed_ = 0; stream_started_ = false; stream_wall_start_ = {}; @@ -619,12 +619,11 @@ runtime::StreamEvent Qwen3ASRSession::process_one_stream_chunk(const runtime::Au streaming_result_.text_output->language = item.text_output->language; } streaming_result_.text_output->text = streaming_text_; - if (streaming_published_bytes_ < streaming_text_.size()) { + if (std::string partial = streaming_partials_.publish(streaming_text_); !partial.empty()) { event.partial_text = runtime::Transcript{ - streaming_text_.substr(streaming_published_bytes_), + std::move(partial), streaming_result_.text_output->language, }; - streaming_published_bytes_ = streaming_text_.size(); } return event; } diff --git a/src/models/vibevoice_asr/session.cpp b/src/models/vibevoice_asr/session.cpp index a1172fff2..295452f69 100644 --- a/src/models/vibevoice_asr/session.cpp +++ b/src/models/vibevoice_asr/session.cpp @@ -1,5 +1,6 @@ #include "engine/models/vibevoice_asr/session.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/audio/chunking.h" #include "engine/framework/debug/profiler.h" #include "engine/framework/debug/trace.h" @@ -130,33 +131,23 @@ runtime::AudioBuffer pad_audio_tail(runtime::AudioBuffer audio, int64_t target_f return audio; } -size_t common_prefix_size(const std::string & lhs, const std::string & rhs) { - const size_t limit = std::min(lhs.size(), rhs.size()); - size_t size = 0; - while (size < limit && lhs[size] == rhs[size]) { - ++size; - } - return size; -} - void emit_transcript_delta( const runtime::StreamEventCallback & sink, const runtime::Transcript & transcript, - std::string & emitted_text) { + runtime::PartialTextPublisher & partials) { if (!sink || transcript.text.empty()) { return; } - const size_t prefix_size = common_prefix_size(emitted_text, transcript.text); - if (prefix_size == transcript.text.size()) { - emitted_text = transcript.text; + std::string delta = partials.publish(transcript.text); + if (delta.empty()) { return; } runtime::StreamEvent event; - event.partial_text = runtime::Transcript{transcript.text.substr(prefix_size), transcript.language}; + event.partial_text = runtime::Transcript{std::move(delta), transcript.language}; sink(event); - emitted_text = transcript.text; } + std::string append_streaming_transcript( runtime::TaskResult & total, const runtime::Transcript & chunk) { @@ -1293,7 +1284,7 @@ runtime::TaskResult VibeVoiceASRSession::run_single(const VibeVoiceASRRequest & text_decoder_.set_pinned_prefix_steps(0); auto prefill = text_decoder_.prefill_prompt(prompt.input_ids, speech.values, prompt.speech_positions); const uint64_t rng_call_offset = (speech.next_rng_index + 3ull) / 4ull; - std::string emitted_text; + runtime::PartialTextPublisher partials; std::function &)> token_callback; if (task_.mode == runtime::RunMode::Streaming && stream_event_sink_ != nullptr) { token_callback = [&](const std::vector & partial_tokens) { @@ -1302,7 +1293,7 @@ runtime::TaskResult VibeVoiceASRSession::run_single(const VibeVoiceASRRequest & emit_transcript_delta( stream_event_sink_, runtime::Transcript{partial_text, request.language}, - emitted_text); + partials); }; } auto generated = generate_tokens(request, prompt, std::move(prefill), rng_call_offset, token_callback); @@ -1314,7 +1305,7 @@ runtime::TaskResult VibeVoiceASRSession::run_single(const VibeVoiceASRRequest & emit_transcript_delta( stream_event_sink_, runtime::Transcript{decoded.text, request.language}, - emitted_text); + partials); } const auto post_end = Clock::now(); diff --git a/src/models/voxtral_realtime/session.cpp b/src/models/voxtral_realtime/session.cpp index 715e1856e..ae0a7c397 100644 --- a/src/models/voxtral_realtime/session.cpp +++ b/src/models/voxtral_realtime/session.cpp @@ -277,7 +277,7 @@ void VoxtralRealtimeSession::reset() { frontend_stream_state_ = VoxtralRealtimeFrontendStreamState{}; audio_stream_state_ = audio_encoder_.make_stream_state(); streaming_text_.clear(); - streaming_published_bytes_ = 0; + streaming_partials_.reset(); streaming_token_count_ = 0; previous_stream_token_ = 0; first_stream_chunk_ = true; @@ -498,11 +498,11 @@ void VoxtralRealtimeSession::take_stream_delta(runtime::StreamEvent & event) { // Partials carry only the text decoded since the last one, as the other streaming ASR sessions // already emit. Restating the transcript is quadratic in its length and hands a consumer of // transcript.text.delta text it was already given. - if (streaming_published_bytes_ >= streaming_text_.size()) { + std::string delta = streaming_partials_.publish(streaming_text_); + if (delta.empty()) { return; } - event.partial_text = runtime::Transcript{streaming_text_.substr(streaming_published_bytes_), ""}; - streaming_published_bytes_ = streaming_text_.size(); + event.partial_text = runtime::Transcript{std::move(delta), ""}; } } // namespace engine::models::voxtral_realtime diff --git a/tests/unittests/test_partial_text.cpp b/tests/unittests/test_partial_text.cpp new file mode 100644 index 000000000..c58b2429e --- /dev/null +++ b/tests/unittests/test_partial_text.cpp @@ -0,0 +1,108 @@ +#include "engine/framework/runtime/partial_text.h" + +#include "test_assert.h" + +#include +#include +#include + +namespace { + +using engine::runtime::PartialTextPublisher; +using engine::test::require_eq; + +// What a consumer assembles by appending every delta, which is the contract: +// partials concatenate into the transcript. +std::string appended(const std::vector & decodes) { + PartialTextPublisher publisher; + std::string client; + for (const auto & decode : decodes) { + client += publisher.publish(decode); + } + return client; +} + +void test_growing_transcript_yields_increments() { + PartialTextPublisher publisher; + require_eq(publisher.publish("Some call me nat"), std::string("Some call me nat"), "first"); + require_eq(publisher.publish("Some call me nature. Others"), std::string("ure. Others"), "second"); + require_eq(publisher.publish("Some call me nature. Others"), std::string(""), "unchanged"); +} + +// The reason this is shared rather than reimplemented per family: a tokenizer +// falls back to bytes for text its vocabulary does not cover, so a decode can +// stop part way through a character. Publishing that puts half a code point on +// the wire, where it reaches a JSON encoder as invalid UTF-8. +void test_partial_character_is_held_until_complete() { + const std::string cjk = "\xE4\xB8\x80"; // U+4E00, three bytes + PartialTextPublisher publisher; + require_eq(publisher.publish("ab" + cjk.substr(0, 1)), std::string("ab"), "lead byte held"); + require_eq(publisher.publish("ab" + cjk.substr(0, 2)), std::string(""), "still incomplete"); + require_eq(publisher.publish("ab" + cjk), cjk, "released whole"); + require_eq(appended({"ab" + cjk.substr(0, 1), "ab" + cjk.substr(0, 2), "ab" + cjk}), + "ab" + cjk, "assembled"); +} + +void test_two_byte_characters_are_held_too() { + const std::string ru = "\xD0\x9F\xD1\x80\xD0\xB8"; // При + require_eq(appended({ru.substr(0, 1), ru.substr(0, 3), ru.substr(0, 5), ru}), ru, "cyrillic"); +} + +void test_four_byte_character_is_held_until_complete() { + const std::string emoji = "\xF0\x9F\x8E\xB5"; // U+1F3B5 + PartialTextPublisher publisher; + require_eq(publisher.publish(emoji.substr(0, 3)), std::string(""), "incomplete"); + require_eq(publisher.publish(emoji), emoji, "released whole"); +} + +// A revision cannot be retracted -- the delta has already gone out -- but the +// next one must still start on a character boundary rather than inside one. +void test_revision_resumes_on_a_character_boundary() { + const std::string cjk = "\xE4\xB8\x80"; + PartialTextPublisher publisher; + require_eq(publisher.publish("ab" + cjk), "ab" + cjk, "published"); + // Same first byte of the character, different continuation. + const std::string revised = "ab\xE4\xB8\x81"; + const std::string delta = publisher.publish(revised); + require_eq(delta, std::string("\xE4\xB8\x81"), "whole character re-sent"); +} + +void test_shrinking_transcript_publishes_nothing() { + PartialTextPublisher publisher; + require_eq(publisher.publish("abcdef"), std::string("abcdef"), "first"); + require_eq(publisher.publish("abc"), std::string(""), "shrunk"); +} + +void test_reset_forgets_the_published_prefix() { + PartialTextPublisher publisher; + require_eq(publisher.publish("hello"), std::string("hello"), "first"); + publisher.reset(); + require_eq(publisher.published(), std::string(""), "cleared"); + require_eq(publisher.publish("hello"), std::string("hello"), "republished after reset"); +} + +void test_empty_and_ascii_edges() { + PartialTextPublisher publisher; + require_eq(publisher.publish(""), std::string(""), "empty"); + require_eq(publisher.publish("a"), std::string("a"), "single byte"); +} + +} // namespace + +int main() { + try { + test_growing_transcript_yields_increments(); + test_partial_character_is_held_until_complete(); + test_two_byte_characters_are_held_too(); + test_four_byte_character_is_held_until_complete(); + test_revision_resumes_on_a_character_boundary(); + test_shrinking_transcript_publishes_nothing(); + test_reset_forgets_the_published_prefix(); + test_empty_and_ascii_edges(); + std::cout << "partial_text_test passed\n"; + } catch (const std::exception & ex) { + std::cerr << "partial_text_test failed: " << ex.what() << "\n"; + return 1; + } + return 0; +} From 11042875e8e7f1bf4d73cebdb9b9bdf18ad8b008 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 18:48:52 -0600 Subject: [PATCH 4/7] kroko_asr: build the final transcript from the decode, not from the partial Caught by running the family rather than reading it. finalize() did: auto event = process_streaming_audio(true); result.text_output = event.partial_text; which returned the whole transcript only because partial_text restated the whole transcript every time. With partials now being increments, that reported the final window alone as the result: a stream that had correctly emitted eleven deltas ended with text_output = " times longer than you". So this family's final result depended on the bug the previous commits fixed. Nothing in the types said so, and it compiles either way. finalize() now builds its result the way the offline path already does, from combined_decoded() through make_result(), which also carries the speech segments and word timestamps it was assembling by hand. Verified against Kroko-ASR-GGUF community q8_0 on assets/resources/sample_16k.wav: eleven partials whose concatenation equals text_output exactly. Before the whole series of commits, the same run emitted eleven copies of a growing transcript. Also ran sense_asr (SenseVoice-Small q8) over the same clip: one window, one partial, matching text_output. --- src/community_models/kroko_asr/session.cpp | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/community_models/kroko_asr/session.cpp b/src/community_models/kroko_asr/session.cpp index efc769264..8bdb57013 100644 --- a/src/community_models/kroko_asr/session.cpp +++ b/src/community_models/kroko_asr/session.cpp @@ -955,10 +955,14 @@ runtime::TaskResult KrokoASRSession::finalize() { if (stream_event_sink_) { stream_event_sink_(event); } - runtime::TaskResult result; - result.text_output = event.partial_text; - result.speech_segments = endpoint_segments_; - result.word_timestamps = std::move(event.word_timestamps); + // Built from the combined decode, not from event.partial_text. The partial + // is the increment since the last one now, so reusing it here would report + // only the final window as the transcript -- this read the whole transcript + // out of the partial only because the partial restated it every time. + runtime::TaskResult result = make_result( + combined_decoded(), + streaming_total_samples_, + streaming_language_); engine::debug::timing_log_scalar( "kroko_asr.session_ms", engine::debug::elapsed_ms(stream_start_, Clock::now())); From d8f515c9ce7cfa6123e8c6cf338ec06867ddc233 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 19:26:31 -0600 Subject: [PATCH 5/7] runtime: never un-publish text a consumer has already been given Review follow-up. Three fixes, one of them a real bug in the publisher. **A held-back character could be sent twice.** publish() assigned `published_ = transcript[0, publishable)` unconditionally, including when nothing was publishable. A decode that truncates mid-character therefore shortened `published_` past text that had already gone out, and the decode restoring the character sent it a second time: publish("ab\xE4\xB8\x80") -> "ab\xE4\xB8\x80" published_ = 5 bytes publish("ab\xE4\xB8") -> "" published_ = 2 bytes <-- rewound publish("ab\xE4\xB8\x80") -> "\xE4\xB8\x80" <-- consumer sees it twice `published_` is what the consumer has, and that cannot be taken back, so it now only ever moves forward: when nothing whole is new, publish() returns early and leaves it alone. Covered by two new cases -- a truncated decode that regrows, and a shrinking transcript whose text returns. **`std::size_t` and ``.** The header declared bare `size_t`, relying on it leaking out of ``. That holds on this libstdc++ and is not guaranteed; the CI matrix includes Windows and macOS. **Include placement.** partial_text.h had landed mid-way through a standard library block in one header and ahead of alphabetically earlier entries in four others, plus a stray double blank line in higgs_audio_stt. Cosmetic only. Re-ran parakeet_tdt (7 partials), kroko_asr (11) and voxtral_realtime (33) after the change: all unchanged, all still concatenating to text_output. --- .../community_models/kroko_asr/session.h | 2 +- .../community_models/parakeet_tdt/session.h | 2 +- .../community_models/sense_asr/session.h | 2 +- .../engine/framework/runtime/partial_text.h | 5 ++-- include/engine/models/qwen3_asr/session.h | 2 +- src/framework/runtime/partial_text.cpp | 28 +++++++++++-------- src/models/higgs_audio_stt/session.cpp | 3 +- src/models/vibevoice_asr/session.cpp | 3 +- tests/unittests/test_partial_text.cpp | 17 +++++++++++ 9 files changed, 42 insertions(+), 22 deletions(-) diff --git a/include/engine/community_models/kroko_asr/session.h b/include/engine/community_models/kroko_asr/session.h index d317b9b7a..593188bf8 100644 --- a/include/engine/community_models/kroko_asr/session.h +++ b/include/engine/community_models/kroko_asr/session.h @@ -5,8 +5,8 @@ #include "engine/community_models/kroko_asr/encoder.h" #include "engine/community_models/kroko_asr/tokenizer.h" #include "engine/community_models/kroko_asr/zipformer.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/model_spec/metadata.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include diff --git a/include/engine/community_models/parakeet_tdt/session.h b/include/engine/community_models/parakeet_tdt/session.h index 5f1706551..9f9faac45 100644 --- a/include/engine/community_models/parakeet_tdt/session.h +++ b/include/engine/community_models/parakeet_tdt/session.h @@ -1,6 +1,7 @@ #pragma once #include "engine/framework/model_spec/metadata.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include "engine/community_models/parakeet_tdt/assets.h" #include "engine/community_models/parakeet_tdt/decoder.h" @@ -14,7 +15,6 @@ #include #include -#include "engine/framework/runtime/partial_text.h" #include #include diff --git a/include/engine/community_models/sense_asr/session.h b/include/engine/community_models/sense_asr/session.h index 2eb85a988..1a7484e76 100644 --- a/include/engine/community_models/sense_asr/session.h +++ b/include/engine/community_models/sense_asr/session.h @@ -4,9 +4,9 @@ #include "engine/community_models/sense_asr/encoder.h" #include "engine/community_models/sense_asr/frontend.h" #include "engine/community_models/sense_asr/types.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/assets/tensor_source.h" #include "engine/framework/model_spec/metadata.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include diff --git a/include/engine/framework/runtime/partial_text.h b/include/engine/framework/runtime/partial_text.h index c2ad2cd24..d29a1d9f2 100644 --- a/include/engine/framework/runtime/partial_text.h +++ b/include/engine/framework/runtime/partial_text.h @@ -1,5 +1,6 @@ #pragma once +#include #include namespace engine::runtime { @@ -48,10 +49,10 @@ class PartialTextPublisher { // Longest common prefix of two transcripts, backed off so the result never // lands inside a UTF-8 sequence. Exposed for tests and for callers that need // the offset rather than the text. -size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs); +std::size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs); // How much of `text` is safe to publish: everything up to the last complete // UTF-8 sequence. Equal to text.size() unless the tail is a partial character. -size_t transcript_publishable_end(const std::string & text); +std::size_t transcript_publishable_end(const std::string & text); } // namespace engine::runtime diff --git a/include/engine/models/qwen3_asr/session.h b/include/engine/models/qwen3_asr/session.h index f76d2e14d..2499f1ad2 100644 --- a/include/engine/models/qwen3_asr/session.h +++ b/include/engine/models/qwen3_asr/session.h @@ -1,7 +1,7 @@ #pragma once -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/assets/tensor_source.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include "engine/models/qwen3_asr/assets.h" #include "engine/models/qwen3_asr/audio_encoder.h" diff --git a/src/framework/runtime/partial_text.cpp b/src/framework/runtime/partial_text.cpp index 4c1e23d36..40fca1ccc 100644 --- a/src/framework/runtime/partial_text.cpp +++ b/src/framework/runtime/partial_text.cpp @@ -2,8 +2,8 @@ namespace engine::runtime { -size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs) { - size_t size = 0; +std::size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs) { + std::size_t size = 0; while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { ++size; } @@ -15,8 +15,8 @@ size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs return size; } -size_t transcript_publishable_end(const std::string & text) { - size_t lead = text.size(); +std::size_t transcript_publishable_end(const std::string & text) { + std::size_t lead = text.size(); while (lead > 0 && (static_cast(text[lead - 1]) & 0xC0) == 0x80) { --lead; } @@ -26,7 +26,7 @@ size_t transcript_publishable_end(const std::string & text) { } --lead; const auto first = static_cast(text[lead]); - size_t needed = 0; + std::size_t needed = 0; if ((first & 0x80) == 0x00) { needed = 1; } else if ((first & 0xE0) == 0xC0) { @@ -44,14 +44,18 @@ size_t transcript_publishable_end(const std::string & text) { } std::string PartialTextPublisher::publish(const std::string & transcript) { - const size_t publishable = transcript_publishable_end(transcript); - const size_t already = transcript_common_prefix(published_, transcript); - std::string delta; - if (already < publishable) { - delta = transcript.substr(already, publishable - already); + const std::size_t publishable = transcript_publishable_end(transcript); + const std::size_t already = transcript_common_prefix(published_, transcript); + if (already >= publishable) { + // Nothing new that is whole. Leave `published_` alone rather than + // shortening it to this decode: a decode that truncates mid-character + // would otherwise un-publish the character it cut, and the decode that + // restores it would send it a second time. + return {}; } - // What actually went out, so a held-back character is reconsidered against - // the update that completes it rather than skipped. + std::string delta = transcript.substr(already, publishable - already); + // What has actually gone out, so a held-back character is reconsidered + // against the update that completes it rather than skipped. published_.assign(transcript, 0, publishable); return delta; } diff --git a/src/models/higgs_audio_stt/session.cpp b/src/models/higgs_audio_stt/session.cpp index 6016a24bf..0dfe16476 100644 --- a/src/models/higgs_audio_stt/session.cpp +++ b/src/models/higgs_audio_stt/session.cpp @@ -1,9 +1,9 @@ #include "engine/models/higgs_audio_stt/session.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/audio/chunking.h" #include "engine/framework/debug/profiler.h" #include "engine/framework/runtime/options.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/spec_backed_model.h" #include @@ -72,7 +72,6 @@ void emit_transcript_delta( sink(event); } - std::string append_streaming_transcript( runtime::TaskResult & total, const runtime::Transcript & chunk) { diff --git a/src/models/vibevoice_asr/session.cpp b/src/models/vibevoice_asr/session.cpp index 295452f69..54fd4be73 100644 --- a/src/models/vibevoice_asr/session.cpp +++ b/src/models/vibevoice_asr/session.cpp @@ -1,10 +1,10 @@ #include "engine/models/vibevoice_asr/session.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/audio/chunking.h" #include "engine/framework/debug/profiler.h" #include "engine/framework/debug/trace.h" #include "engine/framework/runtime/options.h" +#include "engine/framework/runtime/partial_text.h" #include "engine/framework/sampling/torch_random.h" #include "engine/models/silero_vad/session.h" @@ -147,7 +147,6 @@ void emit_transcript_delta( sink(event); } - std::string append_streaming_transcript( runtime::TaskResult & total, const runtime::Transcript & chunk) { diff --git a/tests/unittests/test_partial_text.cpp b/tests/unittests/test_partial_text.cpp index c58b2429e..292390410 100644 --- a/tests/unittests/test_partial_text.cpp +++ b/tests/unittests/test_partial_text.cpp @@ -67,10 +67,26 @@ void test_revision_resumes_on_a_character_boundary() { require_eq(delta, std::string("\xE4\xB8\x81"), "whole character re-sent"); } +// A decode that truncates mid-character must not un-publish the character it +// cut: the decode that restores it would then send it twice, and a consumer +// that appends every delta would show it twice. +void test_truncated_decode_does_not_resend() { + const std::string cjk = "\xE4\xB8\x80"; + PartialTextPublisher publisher; + require_eq(publisher.publish("ab" + cjk), "ab" + cjk, "published whole"); + require_eq(publisher.publish("ab" + cjk.substr(0, 2)), std::string(""), "truncated"); + require_eq(publisher.publish("ab" + cjk), std::string(""), "not resent"); + require_eq(appended({"ab" + cjk, "ab" + cjk.substr(0, 2), "ab" + cjk}), + "ab" + cjk, "consumer sees it once"); +} + void test_shrinking_transcript_publishes_nothing() { PartialTextPublisher publisher; require_eq(publisher.publish("abcdef"), std::string("abcdef"), "first"); require_eq(publisher.publish("abc"), std::string(""), "shrunk"); + // And the text it dropped is not re-sent when it comes back, because the + // consumer was never told to remove it. + require_eq(publisher.publish("abcdef"), std::string(""), "not resent"); } void test_reset_forgets_the_published_prefix() { @@ -96,6 +112,7 @@ int main() { test_two_byte_characters_are_held_too(); test_four_byte_character_is_held_until_complete(); test_revision_resumes_on_a_character_boundary(); + test_truncated_decode_does_not_resend(); test_shrinking_transcript_publishes_nothing(); test_reset_forgets_the_published_prefix(); test_empty_and_ascii_edges(); From 1ea5f9f5a6b001beaef302ce0851d19b068efba0 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Mon, 14 Sep 2026 21:38:31 -0600 Subject: [PATCH 6/7] runtime: make the publisher O(1) per update, not O(n) Review feedback on the PR: replacing a byte-offset with a diff that rescans the whole transcript every update is a performance regression, and a fair objection. Measured it rather than argued: over a 10,000-update session the publisher cost 7.3 us per update against 0.004 us for the offset, and grew with transcript length. Both causes were incidental rather than inherent to the shape. - The prefix check walked byte by byte. memcmp over the shared span instead: 7.3 us -> 0.49 us. - `published_` was reassigned from the whole transcript on every update, copying it each time. Appended in the common case instead. - The remaining O(n) was the prefix check itself, and it was buying nothing: a delta that has gone out cannot be retracted, so a divergence far behind the publish point is not something this can act on even when it finds one. Checking agreement over a bounded window instead makes publish() O(1). Now flat at ~0.025 us per update whether the transcript is 5 KB or 200 KB, against ~0.007 us for a byte-offset that allocates the same delta string. Same complexity class; the difference is one bounded memcmp. The two tests added state what the window buys and what it gives up: a revision near the end -- where a streaming decode actually revises -- is seen and the delta resumes from it; one further back than the window is not, and cannot be, because the text it would correct has already gone out. Verified byte-identical output against the byte-offset implementations on the three families that have one, driving each from the CLI on the same clip: voxtral_realtime 33 partials IDENTICAL qwen3_asr 1 partial IDENTICAL sense_asr 1 partial IDENTICAL and parakeet_tdt (7), kroko_asr (11) and vibevoice_asr_streaming (5) unchanged, each still concatenating to its own text_output. --- src/framework/runtime/partial_text.cpp | 52 ++++++++++++++++++++++++-- tests/unittests/test_partial_text.cpp | 26 +++++++++++++ 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/src/framework/runtime/partial_text.cpp b/src/framework/runtime/partial_text.cpp index 40fca1ccc..d294dc0e7 100644 --- a/src/framework/runtime/partial_text.cpp +++ b/src/framework/runtime/partial_text.cpp @@ -1,11 +1,22 @@ #include "engine/framework/runtime/partial_text.h" +#include +#include + namespace engine::runtime { std::size_t transcript_common_prefix(const std::string & lhs, const std::string & rhs) { + const std::size_t limit = std::min(lhs.size(), rhs.size()); + // The overwhelmingly common case is that the update only extended what was + // published, so test that wholesale before walking byte by byte: a single + // compare over the shared span instead of a per-byte loop. std::size_t size = 0; - while (size < lhs.size() && size < rhs.size() && lhs[size] == rhs[size]) { - ++size; + if (limit > 0 && std::memcmp(lhs.data(), rhs.data(), limit) == 0) { + size = limit; + } else { + while (size < limit && lhs[size] == rhs[size]) { + ++size; + } } // A divergence inside a character would otherwise start the next delta on a // continuation byte. @@ -43,9 +54,31 @@ std::size_t transcript_publishable_end(const std::string & text) { return (text.size() - lead) < needed ? lead : text.size(); } +// How much of `published_` the update still agrees with. +// +// Checked over a bounded window rather than the whole transcript. A decode +// revises what it has just heard, not text from minutes ago -- and a delta that +// has gone out cannot be retracted anyway, so a divergence behind the window is +// not something this could act on even if it found it. Bounding the check keeps +// publish() O(1) in the length of the transcript instead of O(n), which is what +// makes it as cheap as the byte offset it replaces on a long session. +std::size_t agreed_prefix(const std::string & published, const std::string & transcript) { + constexpr std::size_t kWindow = 256; + if (transcript.size() >= published.size()) { + const std::size_t start = published.size() > kWindow ? published.size() - kWindow : 0; + if (std::memcmp(published.data() + start, transcript.data() + start, + published.size() - start) == 0) { + return published.size(); + } + } + // Disagreed inside the window, or the transcript shrank: fall back to the + // exact answer, which is rare enough to afford. + return transcript_common_prefix(published, transcript); +} + std::string PartialTextPublisher::publish(const std::string & transcript) { const std::size_t publishable = transcript_publishable_end(transcript); - const std::size_t already = transcript_common_prefix(published_, transcript); + const std::size_t already = agreed_prefix(published_, transcript); if (already >= publishable) { // Nothing new that is whole. Leave `published_` alone rather than // shortening it to this decode: a decode that truncates mid-character @@ -56,7 +89,18 @@ std::string PartialTextPublisher::publish(const std::string & transcript) { std::string delta = transcript.substr(already, publishable - already); // What has actually gone out, so a held-back character is reconsidered // against the update that completes it rather than skipped. - published_.assign(transcript, 0, publishable); + // + // Appended rather than reassigned in the common case. A streaming + // transcript is rebuilt from scratch on every update, so assigning the + // whole thing here copies the entire transcript once per update -- O(n^2) + // over a session, which is a real cost on a long one and the reason a + // byte-offset is cheaper. When the update only extended what was already + // published, appending the delta is O(delta) instead. + if (already == published_.size()) { + published_.append(delta); + } else { + published_.assign(transcript, 0, publishable); + } return delta; } diff --git a/tests/unittests/test_partial_text.cpp b/tests/unittests/test_partial_text.cpp index 292390410..7145bfa94 100644 --- a/tests/unittests/test_partial_text.cpp +++ b/tests/unittests/test_partial_text.cpp @@ -80,6 +80,30 @@ void test_truncated_decode_does_not_resend() { "ab" + cjk, "consumer sees it once"); } +// The agreement check is bounded, so state what that buys and what it gives up. +// A revision near the end -- where a streaming decode actually revises -- is +// seen, and the delta resumes from it. +void test_revision_within_the_window_is_seen() { + PartialTextPublisher publisher; + require_eq(publisher.publish("the quick brown fox"), std::string("the quick brown fox"), "first"); + const std::string delta = publisher.publish("the quick brown dog"); + require_eq(delta, std::string("dog"), "resumes at the divergence"); +} + +// A revision further back than the window is not seen, and cannot be: the text +// it would correct has already gone out and nothing can retract it. The +// publisher keeps going forward instead of re-sending a transcript the consumer +// cannot un-append. +void test_revision_behind_the_window_does_not_resend_history() { + PartialTextPublisher publisher; + std::string transcript(600, 'a'); + require_eq(publisher.publish(transcript).size(), transcript.size(), "first"); + std::string revised = transcript; + revised[0] = 'b'; // far behind the 256-byte window + revised += "tail"; + require_eq(publisher.publish(revised), std::string("tail"), "only the new tail"); +} + void test_shrinking_transcript_publishes_nothing() { PartialTextPublisher publisher; require_eq(publisher.publish("abcdef"), std::string("abcdef"), "first"); @@ -113,6 +137,8 @@ int main() { test_four_byte_character_is_held_until_complete(); test_revision_resumes_on_a_character_boundary(); test_truncated_decode_does_not_resend(); + test_revision_within_the_window_is_seen(); + test_revision_behind_the_window_does_not_resend_history(); test_shrinking_transcript_publishes_nothing(); test_reset_forgets_the_published_prefix(); test_empty_and_ascii_edges(); From a01e18c9caf971e02368cc24d5c8b3c0c5a9ba11 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Tue, 15 Sep 2026 06:57:43 -0600 Subject: [PATCH 7/7] Scope to the four families that were broken Per review: voxtral_realtime, qwen3_asr and sense_asr keep their own byte-offset implementations. Those were optimised by the teams that own them and were already publishing partials correctly, so migrating them was consistency rather than a fix -- and consistency is not worth touching working, tuned code that someone else is responsible for. Reverted to origin/main exactly; the net diff no longer touches those six files. The shared publisher stays for the four that needed it: parakeet_tdt and kroko_asr were restating the whole transcript as a partial, higgs_audio_stt and vibevoice_asr carried duplicate copies of the same prefix diff between them. Re-verified after narrowing, driving each from the CLI on the same clip: parakeet_tdt 7 partials concatenate to text_output kroko_asr 11 partials concatenate to text_output higgs_audio_stt 4 partials concatenate to text_output voxtral_realtime 33 partials concatenate to text_output (own implementation) qwen3_asr 1 partial concatenate to text_output (own implementation) sense_asr 1 partial concatenate to text_output (own implementation) partial_text_test still passes; the O(1) work in the previous commit stands, and still matters for the four that keep the publisher. --- include/engine/community_models/sense_asr/session.h | 3 +-- include/engine/models/qwen3_asr/session.h | 3 +-- include/engine/models/voxtral_realtime/session.h | 3 +-- src/community_models/sense_asr/session.cpp | 8 ++++---- src/models/qwen3_asr/session.cpp | 7 ++++--- src/models/voxtral_realtime/session.cpp | 8 ++++---- 6 files changed, 15 insertions(+), 17 deletions(-) diff --git a/include/engine/community_models/sense_asr/session.h b/include/engine/community_models/sense_asr/session.h index 1a7484e76..385a78c95 100644 --- a/include/engine/community_models/sense_asr/session.h +++ b/include/engine/community_models/sense_asr/session.h @@ -6,7 +6,6 @@ #include "engine/community_models/sense_asr/types.h" #include "engine/framework/assets/tensor_source.h" #include "engine/framework/model_spec/metadata.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include @@ -88,7 +87,7 @@ class SenseAsrSession final : public runtime::RuntimeSessionBase, runtime::AudioBuffer streaming_audio_; size_t streaming_audio_offset_values_ = 0; std::string streaming_text_; - runtime::PartialTextPublisher streaming_partials_; + size_t streaming_published_bytes_ = 0; int64_t streaming_windows_processed_ = 0; runtime::StreamEventCallback stream_event_sink_; bool stream_started_ = false; diff --git a/include/engine/models/qwen3_asr/session.h b/include/engine/models/qwen3_asr/session.h index 2499f1ad2..ccf7f5b89 100644 --- a/include/engine/models/qwen3_asr/session.h +++ b/include/engine/models/qwen3_asr/session.h @@ -1,7 +1,6 @@ #pragma once #include "engine/framework/assets/tensor_source.h" -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include "engine/models/qwen3_asr/assets.h" #include "engine/models/qwen3_asr/audio_encoder.h" @@ -89,7 +88,7 @@ class Qwen3ASRSession final runtime::AudioBuffer streaming_audio_; size_t streaming_audio_offset_values_ = 0; std::string streaming_text_; - runtime::PartialTextPublisher streaming_partials_; + size_t streaming_published_bytes_ = 0; int64_t streaming_windows_processed_ = 0; runtime::StreamEventCallback stream_event_sink_; bool stream_started_ = false; diff --git a/include/engine/models/voxtral_realtime/session.h b/include/engine/models/voxtral_realtime/session.h index a5a43f616..03c934438 100644 --- a/include/engine/models/voxtral_realtime/session.h +++ b/include/engine/models/voxtral_realtime/session.h @@ -1,6 +1,5 @@ #pragma once -#include "engine/framework/runtime/partial_text.h" #include "engine/framework/runtime/session_base.h" #include "engine/models/voxtral_realtime/assets.h" #include "engine/models/voxtral_realtime/audio_encoder.h" @@ -82,7 +81,7 @@ class VoxtralRealtimeSession final // The transcript decoded so far, and how much of it has already gone out as a partial. Every // partial is the suffix between the two, so the deltas concatenate to exactly this string. std::string streaming_text_; - runtime::PartialTextPublisher streaming_partials_; + size_t streaming_published_bytes_ = 0; int64_t streaming_token_count_ = 0; int32_t previous_stream_token_ = 0; bool stream_started_ = false; diff --git a/src/community_models/sense_asr/session.cpp b/src/community_models/sense_asr/session.cpp index e784856a0..7dc25041d 100644 --- a/src/community_models/sense_asr/session.cpp +++ b/src/community_models/sense_asr/session.cpp @@ -402,7 +402,7 @@ void SenseAsrSession::reset() { streaming_audio_ = runtime::AudioBuffer{}; streaming_audio_offset_values_ = 0; streaming_text_.clear(); - streaming_partials_.reset(); + streaming_published_bytes_ = 0; streaming_windows_processed_ = 0; stream_started_ = false; stream_wall_start_ = {}; @@ -758,12 +758,12 @@ SenseAsrSession::process_one_stream_chunk(const runtime::AudioBuffer &audio) { streaming_result_.text_output->language = item.text_output->language; } streaming_result_.text_output->text = streaming_text_; - if (std::string partial = streaming_partials_.publish(streaming_text_); - !partial.empty()) { + if (streaming_published_bytes_ < streaming_text_.size()) { event.partial_text = runtime::Transcript{ - std::move(partial), + streaming_text_.substr(streaming_published_bytes_), streaming_result_.text_output->language, }; + streaming_published_bytes_ = streaming_text_.size(); } return event; } diff --git a/src/models/qwen3_asr/session.cpp b/src/models/qwen3_asr/session.cpp index 1dadfb1eb..5358a9ed0 100644 --- a/src/models/qwen3_asr/session.cpp +++ b/src/models/qwen3_asr/session.cpp @@ -407,7 +407,7 @@ void Qwen3ASRSession::reset() { streaming_audio_ = runtime::AudioBuffer{}; streaming_audio_offset_values_ = 0; streaming_text_.clear(); - streaming_partials_.reset(); + streaming_published_bytes_ = 0; streaming_windows_processed_ = 0; stream_started_ = false; stream_wall_start_ = {}; @@ -619,11 +619,12 @@ runtime::StreamEvent Qwen3ASRSession::process_one_stream_chunk(const runtime::Au streaming_result_.text_output->language = item.text_output->language; } streaming_result_.text_output->text = streaming_text_; - if (std::string partial = streaming_partials_.publish(streaming_text_); !partial.empty()) { + if (streaming_published_bytes_ < streaming_text_.size()) { event.partial_text = runtime::Transcript{ - std::move(partial), + streaming_text_.substr(streaming_published_bytes_), streaming_result_.text_output->language, }; + streaming_published_bytes_ = streaming_text_.size(); } return event; } diff --git a/src/models/voxtral_realtime/session.cpp b/src/models/voxtral_realtime/session.cpp index ae0a7c397..715e1856e 100644 --- a/src/models/voxtral_realtime/session.cpp +++ b/src/models/voxtral_realtime/session.cpp @@ -277,7 +277,7 @@ void VoxtralRealtimeSession::reset() { frontend_stream_state_ = VoxtralRealtimeFrontendStreamState{}; audio_stream_state_ = audio_encoder_.make_stream_state(); streaming_text_.clear(); - streaming_partials_.reset(); + streaming_published_bytes_ = 0; streaming_token_count_ = 0; previous_stream_token_ = 0; first_stream_chunk_ = true; @@ -498,11 +498,11 @@ void VoxtralRealtimeSession::take_stream_delta(runtime::StreamEvent & event) { // Partials carry only the text decoded since the last one, as the other streaming ASR sessions // already emit. Restating the transcript is quadratic in its length and hands a consumer of // transcript.text.delta text it was already given. - std::string delta = streaming_partials_.publish(streaming_text_); - if (delta.empty()) { + if (streaming_published_bytes_ >= streaming_text_.size()) { return; } - event.partial_text = runtime::Transcript{std::move(delta), ""}; + event.partial_text = runtime::Transcript{streaming_text_.substr(streaming_published_bytes_), ""}; + streaming_published_bytes_ = streaming_text_.size(); } } // namespace engine::models::voxtral_realtime