ASR: real native streaming for Nemotron 3.5 + Granite Speech; granite encoder graph cache

app/server/http.cpp

@@ -164,6 +164,21 @@ void send_all(SocketHandle socket, const std::string & data) {
     }
 }
 
+// Streaming replies interleave small SSE events with the client's ongoing
+// upload. With Nagle enabled each small event waits for the peer's ACK of the
+// previous segment (delayed-ACK stretches that to hundreds of milliseconds),
+// so partials queue up behind the upload instead of arriving while the user
+// speaks. Disable Nagle on accepted sockets.
+void set_no_delay(SocketHandle socket) {
+#ifdef _WIN32
+    constexpr int value = 1;
+    setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<const char *>(&value), sizeof(value));
+#else
+    constexpr int value = 1;
+    setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value));
+#endif
+}
+
 // An SSE body is written while a model lock is held, so an unbounded blocking
 // send() lets a client that uploads but never reads fill the kernel send buffer
 // and pin that model indefinitely. A send timeout turns it into a failed write.
@@ -787,6 +802,7 @@ void handle_client(SocketHandle client, IHttpHandler & handler, uint64_t max_req
             // stops reading can pin a model. Applying it server-wide would risk
             // truncating a large ordinary response to a merely slow client.
             set_send_timeout(socket.get(), limits.send_timeout_ms);
+            set_no_delay(socket.get());
         }
         // Constructed unconditionally so it outlives the handler call, but only
         // published on `request` when the client actually declared a chunked body.

app/server/runtime.cpp

@@ -2992,6 +2992,23 @@ HttpResponse ServerState::handle_transcription_live(const HttpRequest & request)
             task_request.options["language"] = language;
             task_request.text_input = engine::runtime::Transcript{std::string(), language};
         }
+        // Streaming latency knobs. lookahead_tokens picks the model's right-context
+        // (chunk duration = (lookahead + 1) x 80 ms on nemotron; the session validates
+        // the value against the model's supported set). stream_chunk_ms bounds how much
+        // audio the ingest layer assembles before handing it to the session — the
+        // policy default batches a full second, which delays every partial by that
+        // much regardless of the model window.
+        // lookahead 0 (80 ms chunks) is rejected even though the GGUF declares it
+        // supported: a single-frame first window is not covered by the reference
+        // chunking validation and decodes garbage — fail loudly instead.
+        if (!query_param(request.query, "lookahead_tokens").empty()) {
+            const int64_t lookahead = parse_bounded_int("lookahead_tokens", 3, 0, 13);
+            task_request.options["lookahead_tokens"] = std::to_string(lookahead);
+        }
+        if (!query_param(request.query, "stream_chunk_ms").empty()) {
+            task_request.options["stream_chunk_ms"] = std::to_string(
+                parse_bounded_int("stream_chunk_ms", 1000, 10, 4000));
+        }
         task_request = apply_default_request_options(model, std::move(task_request));
     } catch (const std::runtime_error & ex) {
         // Deliberately runtime_error and not exception: every rejection above is

app/streaming/streaming.cpp

@@ -1,9 +1,12 @@
 #include "streaming.h"
 
+#include "engine/framework/runtime/options.h"
+
 #include <algorithm>
 #include <cstddef>
 #include <cmath>
 #include <stdexcept>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -37,6 +40,31 @@ int64_t resolve_chunk_samples(
     return policy.preferred_audio_chunk_samples;
 }
 
+// Request-level override for the ingest batching granularity ("stream_chunk_ms",
+// milliseconds of audio per process_audio_chunk call). The session policy's
+// preferred chunk is sized for throughput (nemotron batches a full second), which
+// delays every partial by that much; low-latency clients ask for smaller reads so
+// the model window fills as soon as enough bytes have arrived. Ignored when the
+// option is absent or malformed beyond recognition — a numeric parse failure is
+// the caller's bug and throws.
+int64_t chunk_samples_for_request(
+    const engine::runtime::TaskRequest & request,
+    const engine::runtime::StreamingPolicy & policy,
+    const AudioStreamFormat & format) {
+    const auto override_ms = engine::runtime::find_option(request.options, {"stream_chunk_ms"});
+    if (!override_ms || override_ms->empty()) {
+        return resolve_chunk_samples(policy, format);
+    }
+    size_t consumed = 0;
+    const long long ms = std::stoll(*override_ms, &consumed);
+    if (consumed != override_ms->size() || ms <= 0) {
+        throw std::runtime_error("stream_chunk_ms must be a positive integer (milliseconds)");
+    }
+    const auto samples = static_cast<int64_t>(std::llround(
+        static_cast<double>(ms) * static_cast<double>(format.sample_rate) / 1000.0));
+    return std::max<int64_t>(samples, 1) * static_cast<int64_t>(format.channels);
+}
+
 void feed_audio_stream(
     engine::runtime::IStreamingVoiceTaskSession & session,
     const AudioChunkStream & stream,
@@ -128,14 +156,15 @@ engine::runtime::TaskResult run_stream(
                     "streaming audio input mode requires samples in audio_input, or an "
                     "AudioChunkStream for a live source");
             }
-            feed_audio_stream(session, *stream, resolve_chunk_samples(policy, stream->format), sink);
+            feed_audio_stream(
+                session, *stream, chunk_samples_for_request(request, policy, stream->format), sink);
         }
-        if (policy.output == engine::runtime::StreamingOutputKind::PullEvents) {
-            pull_stream_events(session, sink);
-        }
-        auto result = session.finish_stream();
-        session.set_stream_event_sink(nullptr);
-        return result;
+		if (policy.output == engine::runtime::StreamingOutputKind::PullEvents) {
+			pull_stream_events(session, sink);
+		}
+		auto result = session.finish_stream();
+		session.set_stream_event_sink(nullptr);
+		return result;
     } catch (...) {
         session.set_stream_event_sink(nullptr);
         throw;

docs/nemotron_streaming_end_of_turn.md

@@ -0,0 +1,73 @@
+# Nemotron streaming: end-of-turn latency — verified analysis (reference-validated)
+
+All findings below were established against the official reference
+implementation (transformers `nemotron3_5_asr`, f32, NVIDIA-contributed)
+running the same audio through its own encoder, its own decode loop, and
+its own official `generate()` streaming path.
+
+## The end-of-turn latency budget (x2 CPU, per phrase)
+
+EOT->final = flush encode (~100-175 ms) +, for pocketed utterances, the
+padded-offline fallback (~250-330 ms). The flush encode is the encoder
+running one 64-mel window; it is exact vs the reference (cos 0.999,
+frame-for-frame, verified on multiple shapes).
+
+## Trailing-token mechanism (verified)
+
+The RNNT fires a word's trailing tokens only after a per-utterance amount
+of silence follows the word in the encoded stream. The silence CONTENT is
+irrelevant (digital zeros = room tone = dither; verified). What matters is
+duration, and it interacts with the cut position:
+
+- hello.wav cut at 0.65 s: decodes in streaming at the 64-mel window.
+- hi.wav cut at 0.67 s: decodes in streaming ('Hi. ').
+- hi.wav cut at 0.57 s: does NOT decode in streaming AT ALL.
+
+## The hi.wav 0.57 s cut: streaming-mode failure, not a bug
+
+Exhaustive shape matrix against the reference, same audio (hi.wav[:9120]),
+la0, prompt aligned:
+
+    (8,32)            5 f -> ''
+    (8,32,32)         9 f -> ''
+    (8,32,64)        13 f -> ''
+    (8,32,96)        17 f -> ''
+    (8,32,32,32)     13 f -> ''
+    (8,32,32,32,32)  17 f -> ''
+    (8,32x7)         25 f -> ''
+
+Every streaming shape fails, including 1.6 s of trailing synthetic silence.
+The reference's own official streaming `generate()` also returns ''.
+The OFFLINE encoder on the same audio + >=0.5 s of padding decodes 'Hi. '.
+
+Conclusion: the streaming causal-conv frame grid fails this utterance at
+this cut regardless of windowing; the offline grid succeeds. audio.cpp's
+padded-offline fallback is the only correct mechanism and matches the
+model's own semantics. It is not a workaround to remove - it is the fix.
+
+## Dead ends (do not retry)
+
+- Longer single flush windows (64/72/96 mel): the 0.57 s hi cut fails at
+  every length; longer windows only add encode time.
+- Iterative flush chunks: same frames as the long window, same failure.
+- Silence dither instead of zero padding: identical results.
+- Per-sample-max energy gates for the speculative trigger: decay
+  transients count as speech; use 10 ms windowed RMS.
+
+## Remaining latency levers (ranked)
+
+1. Speculative offline fallback: when the speculative flush's decode comes
+   back blank (decode is cheap, ~10 ms), run the padded-offline encode
+   during the tail wait too. Requires snapshot/restore of the decoder RNNT
+   state alongside the encoder state. Saves the fallback's ~250-330 ms from
+   the EOT->final path for pocketed utterances.
+2. q4_K GGUF regeneration: the encoder is weight-bandwidth-bound; halves
+   the per-chunk cost on CPU.
+3. Vulkan: the flush hides entirely; EOT->final 60-90 ms already.
+
+## Verified harness artifacts
+
+- bench_tmp/ref_nemotron_stream.py: reference runner (offline/official
+  streaming/manual chunked with dumps).
+- bench_tmp/dump/ref96/*: reference frames for the shape matrix above.
+- NEMOTRON_DUMP_CHUNKS / NEMOTRON_DUMP_CACHES: audio.cpp-side dumps.

include/engine/community_models/granite5asr/encoder.h

@@ -55,18 +55,27 @@ class Granite5EncoderRuntime {
         engine::core::ExecutionContext & execution_context,
         assets::TensorStorageType storage_type,
         size_t graph_arena_bytes = 1024ull * 1024ull * 1024ull);
+    ~Granite5EncoderRuntime();
 
     std::vector<int32_t> transcribe_features(
         const Granite5FrontendFeatures & features);
 
     const Granite5ASRAssets & assets() const noexcept { return *assets_; }
 
 private:
+    struct GraphCacheEntry;
+    struct GraphCache;
+    GraphCacheEntry & ensure_graph_entry(int64_t input_frames, int64_t feature_dim);
+
     std::shared_ptr<const Granite5ASRAssets> assets_;
     engine::core::ExecutionContext * execution_context_ = nullptr;
     engine::core::BackendWeightStore weight_store_;
     Granite5EncoderWeights weights_;
     size_t graph_arena_bytes_;
+    // Shape-keyed encoder graphs: building the 16-block conformer graph costs
+    // real milliseconds and chunked streaming re-decodes the SAME window shape
+    // every chunk, so graphs are cached per input length (small LRU).
+    std::unique_ptr<GraphCache> graph_cache_;
 };
 
 }  // namespace engine::community_models::granite5asr

include/engine/community_models/granite5asr/session.h

@@ -82,6 +82,18 @@ class Granite5ASRStreamingSession final
     runtime::StreamEventCallback stream_event_sink_;
     runtime::AudioBuffer streaming_audio_;
     runtime::TaskRequest streaming_request_;
+
+    // Windowed chunked-streaming state (the publisher's chunked TurboCTC recipe):
+    // every center chunk is (re-)encoded together with its left-context window and
+    // CTC-greedy decoded continuously, so partials stream as chunks land.
+    bool decode_next_center_window(bool flush_tail, std::string & delta_out);
+    int64_t stream_center_samples_ = 0;
+    int64_t stream_left_context_samples_ = 0;
+    int64_t stream_next_center_start_ = 0;
+    int32_t stream_last_raw_token_ = -1;
+    std::vector<int32_t> stream_collapsed_ids_;
+    std::string stream_emitted_text_;
+    bool stream_finalized_ = false;
 };
 
 class Granite5ASRLoadedModel final : public runtime::ILoadedVoiceModel {

include/engine/models/nemotron_asr/decoder.h

@@ -45,6 +45,18 @@ class NemotronDecoderRuntime {
         const std::function<bool(NemotronEncodedAudio &)> & next_chunk,
         const NemotronTextDeltaCallback & on_text_delta = nullptr);
 
+    // Incremental chunk-at-a-time decoding: begin resets the prediction-network
+    // state, decode_stream_chunk consumes the frames of one encoded chunk while
+    // continuing that state (firing on_text_delta for newly emitted text), and
+    // finish finalizes the transcript and word timestamps. decode_streaming() is
+    // implemented on top of these three; sessions that receive audio live (SSE
+    // /live, WebSocket, ...) call them directly so partials stream per chunk.
+    void begin_stream_decode(const NemotronDecodeOptions & options);
+    void decode_stream_chunk(
+        const NemotronEncodedAudio & chunk,
+        const NemotronTextDeltaCallback & on_text_delta = nullptr);
+    NemotronDecodedText finish_stream_decode();
+
 private:
     struct Graph;
     struct JointGraph;
@@ -68,6 +80,17 @@ class NemotronDecoderRuntime {
     std::vector<float> logits_scratch_;
     std::vector<float> hidden_read_scratch_;
     std::vector<float> cell_read_scratch_;
+
+    // Persistent state across decode_stream_chunk() calls.
+    bool stream_decode_active_ = false;
+    NemotronDecodeOptions stream_decode_options_;
+    int64_t stream_frame_index_ = 0;
+    int64_t stream_symbols_at_frame_ = 0;
+    int32_t stream_input_token_ = 0;
+    bool stream_decoder_cache_initialized_ = false;
+    std::vector<int32_t> stream_token_ids_;
+    std::vector<int32_t> stream_durations_;
+    std::string stream_emitted_text_;
 };
 
 }  // namespace engine::models::nemotron_asr

include/engine/models/nemotron_asr/encoder.h

@@ -27,6 +27,7 @@ struct NemotronEncoderStreamState {
     bool first_chunk = true;
     bool backend_cache_valid = false;
     const void * backend_cache_owner = nullptr;
+    int64_t dump_seq = 0;
 };
 
 class NemotronEncoderRuntime {
@@ -35,7 +36,8 @@ class NemotronEncoderRuntime {
         std::shared_ptr<const NemotronASRAssets> assets,
         std::shared_ptr<const NemotronWeights> weights,
         engine::core::ExecutionContext & execution_context,
-        size_t graph_arena_bytes);
+        size_t graph_arena_bytes,
+        size_t stream_graph_arena_bytes = 0);
     ~NemotronEncoderRuntime();
 
     void prepare_capacity(int64_t input_frames, int64_t feature_dim, int64_t lookahead_tokens);
@@ -65,13 +67,16 @@ class NemotronEncoderRuntime {
     std::shared_ptr<const NemotronWeights> weights_;
     engine::core::ExecutionContext * execution_context_ = nullptr;
     size_t graph_arena_bytes_ = 0;
+    size_t stream_graph_arena_bytes_ = 0;
     std::unique_ptr<Graph> graph_;
     std::vector<std::unique_ptr<Graph>> stream_graphs_;
     std::vector<float> input_scratch_;
     std::vector<float> output_scratch_;
     std::vector<float> prompt_scratch_;
     std::vector<int32_t> mask_scratch_;
     std::vector<float> attention_mask_scratch_;
+    std::vector<float> attention_key_scratch_;
+    std::vector<float> attention_value_scratch_;
     std::unordered_map<int64_t, std::vector<float>> relative_positional_encoding_cache_;
 };
 

include/engine/models/nemotron_asr/session.h

@@ -93,6 +93,39 @@ class NemotronASRStreamingSession final
 
 private:
     runtime::StreamEventCallback stream_event_sink_;
+
+    // Native chunked-streaming pipeline state. Audio is encoded window-by-window
+    // through encode_stream_chunk (cache-aware, publisher chunk contract) and the
+    // decoder runs incrementally across chunks, so partials stream as chunks land
+    // instead of only at finalize.
+    bool encode_and_decode_next_chunk(bool flush_tail, std::string & delta_out);
+    // Speculative flush support (see the member docs below).
+    int64_t flush_window_mel() const;
+    void build_flush_window(int64_t total, std::vector<float> & window) const;
+    void maybe_speculative_flush();
+    void discard_speculative_flush();
+    int64_t stream_lookahead_ = 0;
+    int64_t stream_prompt_id_ = 0;
+    NemotronDecodeOptions stream_decode_options_;
+    NemotronEncoderStreamState encoder_stream_state_;
+    int64_t stream_first_samples_ = 0;
+    int64_t stream_samples_per_chunk_ = 0;
+    int64_t stream_first_mel_frames_ = 0;
+    int64_t stream_mel_frames_per_chunk_ = 0;
+    int64_t stream_next_chunk_start_ = 0;
+    bool stream_await_first_chunk_ = true;
+    bool stream_tail_encoded_ = false;
+    bool stream_decode_active_ = false;
+    int64_t stream_dump_chunk_seq_ = 0;
+    // Speculative flush: while the turn is still open, the flush window is
+    // encoded as soon as the ingest sees enough trailing silence, and finalize
+    // reuses the result unless speech followed (then the state snapshot is
+    // restored and the normal flush runs). See session.cpp for the knobs.
+    bool spec_valid_ = false;
+    NemotronEncodedAudio spec_frames_;
+    NemotronEncoderStreamState spec_state_snapshot_{};
+    int64_t spec_last_loud_sample_ = 0;
+    int64_t spec_audio_mark_ = 0;
 };
 
 }  // namespace engine::models::nemotron_asr

model_specs/granite5asr.json

@@ -73,6 +73,22 @@
         "description": "Path to the Silero VAD model directory used by automatic audio chunking.",
         "required": false,
         "default": "assets/framework/models/silero_vad"
+      },
+      {
+        "name": "center_chunk_sec",
+        "type": "float",
+        "description": "Center-chunk duration in seconds for buffered streaming; each chunk is decoded as soon as it is complete, which sets the partial cadence.",
+        "required": false,
+        "min": 0.05,
+        "default": 1.0
+      },
+      {
+        "name": "left_context_sec",
+        "type": "float",
+        "description": "Left-context audio re-encoded with every center chunk to give the block-attention CTC model its past context.",
+        "required": false,
+        "min": 0.0,
+        "default": 2.0
       }
     ],
     "load": []
@@ -176,4 +192,4 @@
       }
     }
   ]
-}
+}