streaming: publish partial text as increments, through one shared publisher

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)

include/engine/community_models/kroko_asr/session.h

@@ -6,6 +6,7 @@
 #include "engine/community_models/kroko_asr/tokenizer.h"
 #include "engine/community_models/kroko_asr/zipformer.h"
 #include "engine/framework/model_spec/metadata.h"
+#include "engine/framework/runtime/partial_text.h"
 #include "engine/framework/runtime/session_base.h"
 
 #include <chrono>
@@ -83,6 +84,7 @@ class KrokoASRSession final
     std::vector<float> 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;

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"
@@ -13,6 +14,7 @@
 #include <filesystem>
 #include <memory>
 #include <string>
+
 #include <unordered_map>
 #include <vector>
 
@@ -125,6 +127,9 @@ class ParakeetTDTStreamingSession final
     std::vector<int32_t> token_frame_indices_;
     std::vector<int32_t> token_durations_;
     runtime::StreamEventCallback stream_event_sink_;
+    // 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;
 };

include/engine/framework/runtime/partial_text.h

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <cstddef>
+#include <string>
+
+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.
+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.
+std::size_t transcript_publishable_end(const std::string & text);
+
+}  // namespace engine::runtime

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;
@@ -940,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()));

src/community_models/parakeet_tdt/session.cpp

@@ -789,6 +789,7 @@ void ParakeetTDTStreamingSession::reset() {
     token_ids_.clear();
     token_frame_indices_.clear();
     token_durations_.clear();
+    partials_.reset();
     decoder_->reset_state();
     stream_started_ = true;
     finalized_ = false;
@@ -912,7 +913,21 @@ 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".
+        std::string delta = partials_.publish(decoded.text);
+        if (!delta.empty()) {
+            event.partial_text = runtime::Transcript{std::move(delta), ""};
+        }
+        // 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.

src/framework/runtime/partial_text.cpp

@@ -0,0 +1,107 @@
+#include "engine/framework/runtime/partial_text.h"
+
+#include <algorithm>
+#include <cstring>
+
+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;
+    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.
+    while (size > 0 && (static_cast<unsigned char>(rhs[size]) & 0xC0) == 0x80) {
+        --size;
+    }
+    return size;
+}
+
+std::size_t transcript_publishable_end(const std::string & text) {
+    std::size_t lead = text.size();
+    while (lead > 0 && (static_cast<unsigned char>(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<unsigned char>(text[lead]);
+    std::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();
+}
+
+// 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 = 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
+        // would otherwise un-publish the character it cut, and the decode that
+        // restores it would send it a second time.
+        return {};
+    }
+    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.
+    //
+    // 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;
+}
+
+}  // namespace engine::runtime

src/models/higgs_audio_stt/session.cpp

@@ -3,6 +3,7 @@
 #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 <algorithm>
@@ -55,31 +56,20 @@ int64_t audio_frame_count(const runtime::AudioBuffer & audio) {
     return static_cast<int64_t>(audio.samples.size() / static_cast<size_t>(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(
@@ -390,15 +380,15 @@ 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) {
             const auto partial = postprocessor_.decode(partial_tokens, asr_request);
             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 +399,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();