From f81a053d543e272b6d9d8614d3199ed18df012aa Mon Sep 17 00:00:00 2001 From: Flaviu-Gheorghe Grosan Date: Sun, 16 Aug 2026 18:44:28 +0100 Subject: [PATCH] Fixed character encoding for arch-specific generate loops. Per-token tokenizer.decode() ran from_utf8_lossy over half a character, so emoji and byte-fallback CJK reached the client as replacement chars. Adds TokenTextStream (holds back the incomplete tail, flushes at end of stream) and routes every arch decode loop through it. qwen35's daemon path was already correct, which is why this stayed hidden. --- .../hipfire-arch-cohere2moe/src/spec_emit.rs | 90 +- .../hipfire-arch-deepseek4/src/spec_emit.rs | 40 +- crates/hipfire-cli/src/main.rs | 24 +- crates/hipfire-runtime/examples/daemon.rs | 1002 ++++++++++++----- crates/hipfire-runtime/src/lib.rs | 6 +- .../src/streaming_decode_guard.rs | 244 ++++ crates/hipfire-runtime/src/tokenizer.rs | 526 ++++++++- 7 files changed, 1637 insertions(+), 295 deletions(-) create mode 100644 crates/hipfire-runtime/src/streaming_decode_guard.rs diff --git a/crates/hipfire-arch-cohere2moe/src/spec_emit.rs b/crates/hipfire-arch-cohere2moe/src/spec_emit.rs index fe70cf507..d615b502f 100644 --- a/crates/hipfire-arch-cohere2moe/src/spec_emit.rs +++ b/crates/hipfire-arch-cohere2moe/src/spec_emit.rs @@ -27,7 +27,7 @@ use hipfire_runtime::prompt_frame::ToolCall; use hipfire_runtime::spec::{ ClientEvent, EmitOutcome, FinishSummary, SpecEmit, SpecEmitCtx, StopReason, }; -use hipfire_runtime::tokenizer::Tokenizer; +use hipfire_runtime::tokenizer::{TokenTextStream, Tokenizer}; const MAX_EOS_SUPPRESS: usize = 3; @@ -66,6 +66,31 @@ pub struct Cohere2MoeEmit<'a> { tool_calls_emitted: bool, /// Tokens the loop must force after the current step (drained by `take_forced`). forced: Vec, + /// Incremental token → text decoder. Byte-level BPE and byte-fallback + /// spread one character across several tokens, so decoding per token would + /// surface a U+FFFD per fragment instead of the character. Holds back only + /// the trailing incomplete codepoint; drained in `finish`. + text_stream: TokenTextStream, +} + +/// True when a single token decodes to a bare `<|UPPER_SNAKE|>` structural +/// marker. +/// +/// Checked on the token's own bytes rather than on a `TokenTextStream` +/// fragment: a fragment can carry the completion bytes of a preceding split +/// character, whose leading char would defeat the `<|…|>` shape match. Uses +/// `decode_bytes`, not the lossy per-token `decode(&[tok])`. +fn is_bracketed_marker_token(tokenizer: &Tokenizer, token: u32) -> bool { + let bytes = tokenizer.decode_bytes(&[token]); + let Ok(s) = std::str::from_utf8(&bytes) else { + return false; + }; + s.len() > 4 + && s.starts_with("<|") + && s.ends_with("|>") + && s[2..s.len() - 2] + .chars() + .all(|c| c.is_ascii_uppercase() || c == '_') } impl<'a> Cohere2MoeEmit<'a> { @@ -110,6 +135,7 @@ impl<'a> Cohere2MoeEmit<'a> { tool_calls_buf: Vec::new(), tool_calls_emitted: false, forced: Vec::new(), + text_stream: TokenTextStream::new(), }) } @@ -125,6 +151,26 @@ impl<'a> Cohere2MoeEmit<'a> { self.emitted_count += 1; } + /// Route one decoded text fragment down whichever agentic channel is + /// currently open. Shared by `process` and the end-of-stream flush in + /// `finish`, so a character that completed only at the flush lands on the + /// channel the turn ended on rather than being dropped — dropping it is a + /// silent truncation. + fn route(&mut self, frag: String, events: &mut Vec) { + match self.sec { + Sec::Action => self.action_buf.push_str(&frag), + Sec::Think => { + events.push(ClientEvent::Reasoning(frag)); + self.think_count += 1; + } + Sec::Text | Sec::Pre => { + self.vis_buf.push_str(&frag); + events.push(ClientEvent::Token(frag)); + self.emitted_visible = true; + } + } + } + /// The shared begin/observe body: run one committed token through the marker /// state machine, returning the events + any stop/forced request. fn process(&mut self, token: u32) -> EmitOutcome { @@ -179,30 +225,24 @@ impl<'a> Cohere2MoeEmit<'a> { self.sec = Sec::Pre; self.committed(token, &mut events); } else { - let frag = self.tokenizer.decode(&[token]); // Defense-in-depth: never surface a Cohere structural marker the id // state machine missed (START_OF_TURN_TOKEN, CHATBOT_TOKEN, …). The // token is still committed (target advanced over it); only its emit // is dropped, so a state-machine miss can never leak a marker. - let is_marker = frag.len() > 4 - && frag.starts_with("<|") - && frag.ends_with("|>") - && frag[2..frag.len() - 2] - .chars() - .all(|c| c.is_ascii_uppercase() || c == '_'); - if !is_marker { - match self.sec { - Sec::Action => self.action_buf.push_str(&frag), - Sec::Think => { - events.push(ClientEvent::Reasoning(frag)); - self.think_count += 1; - } - Sec::Text | Sec::Pre => { - self.vis_buf.push_str(&frag); - events.push(ClientEvent::Token(frag)); - self.emitted_visible = true; - } - } + // + // Tested on the token's OWN bytes, not on the stream fragment: a + // fragment can carry the completion bytes of a preceding split + // character, whose leading char would defeat the `<|…|>` match. + let frag = if is_bracketed_marker_token(self.tokenizer, token) { + String::new() + } else { + // Incremental decode with holdback — never + // `self.tokenizer.decode(&[token])`, which is lossy per token + // and splits multi-token UTF-8 into FFFD. Empty on a holdback. + self.text_stream.push(self.tokenizer, token) + }; + if !frag.is_empty() { + self.route(frag, &mut events); } self.committed(token, &mut events); } @@ -238,6 +278,14 @@ impl<'a> SpecEmit for Cohere2MoeEmit<'a> { fn finish(mut self: Box) -> FinishSummary { let mut events = Vec::new(); + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Route the held-back tail down whichever + // channel was open when the turn ended, BEFORE the recovery below so + // it sees the complete `vis_buf`. Dropping it is a silent truncation. + let tail = self.text_stream.flush(); + if !tail.is_empty() { + self.route(tail, &mut events); + } // Tool-call-as-text recovery: a non-Cohere harness can prime North to // write a tool-call JSON array as TEXT instead of via <|START_ACTION|>. if !self.tool_calls_emitted { diff --git a/crates/hipfire-arch-deepseek4/src/spec_emit.rs b/crates/hipfire-arch-deepseek4/src/spec_emit.rs index 9c1cfc510..593316bcf 100644 --- a/crates/hipfire-arch-deepseek4/src/spec_emit.rs +++ b/crates/hipfire-arch-deepseek4/src/spec_emit.rs @@ -17,7 +17,7 @@ use hipfire_runtime::prompt_frame::{ThinkMode, ToolCall}; use hipfire_runtime::spec::{ ClientEvent, EmitOutcome, FinishSummary, SpecEmit, SpecEmitCtx, SpecGrammar, StopReason, }; -use hipfire_runtime::tokenizer::Tokenizer; +use hipfire_runtime::tokenizer::{TokenTextStream, Tokenizer}; pub struct Deepseek4Emit<'a> { tokenizer: &'a Tokenizer, @@ -39,6 +39,11 @@ pub struct Deepseek4Emit<'a> { /// matcher and never calls `grammar()`). The matcher advances inside the /// spec step ONLY — `observe` must NOT touch it (single-advance invariant). grammar: Option, + /// Incremental token → text decoder. Byte-level BPE and byte-fallback + /// spread one character across several tokens, so decoding per token would + /// hand the DSML parser (and the client) a U+FFFD per fragment. Holds back + /// only the trailing incomplete codepoint; drained in `finish`. + text_stream: TokenTextStream, } /// Map a visible DSML channel event into a client event. ToolCalls/Malformed @@ -120,6 +125,7 @@ impl<'a> Deepseek4Emit<'a> { streamed_tokens: Vec::new(), visible_acc: String::new(), grammar, + text_stream: TokenTextStream::new(), }) } @@ -131,13 +137,19 @@ impl<'a> Deepseek4Emit<'a> { fn feed_and_emit(&mut self, token: u32) -> Vec { let mut events = Vec::new(); self.streamed_tokens.push(token); - let frag = self.tokenizer.decode(&[token]); - for ev in self.deferred.absorb_all(self.parser.feed(&frag)) { - if let Some(ce) = visible_client_event(ev) { - if let ClientEvent::Token(ref t) = ce { - self.visible_acc.push_str(t); + // Never `self.tokenizer.decode(&[token])` — that is lossy per token and + // turns any character whose UTF-8 spans several tokens (emoji, + // byte-fallback CJK) into U+FFFD. `frag` is empty while a character is + // still split; the tail is drained in `finish`. + let frag = self.text_stream.push(self.tokenizer, token); + if !frag.is_empty() { + for ev in self.deferred.absorb_all(self.parser.feed(&frag)) { + if let Some(ce) = visible_client_event(ev) { + if let ClientEvent::Token(ref t) = ce { + self.visible_acc.push_str(t); + } + events.push(ce); } - events.push(ce); } } events.push(ClientEvent::Committed { @@ -200,6 +212,20 @@ impl<'a> SpecEmit for Deepseek4Emit<'a> { // held finish events for the wrapper — the generic generate_spec core // must NOT render them before length/malformed is known. let mut events = Vec::new(); + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Feed the held-back tail through the DSML + // parser before finishing it, or those bytes vanish silently. + let tail = self.text_stream.flush(); + if !tail.is_empty() { + for ev in self.deferred.absorb_all(self.parser.feed(&tail)) { + if let Some(ce) = visible_client_event(ev) { + if let ClientEvent::Token(ref t) = ce { + self.visible_acc.push_str(t); + } + events.push(ce); + } + } + } let parser = std::mem::replace(&mut self.parser, dsml::StreamParser::new()); for ev in self.deferred.absorb_all(parser.finish()) { if let Some(ce) = visible_client_event(ev) { diff --git a/crates/hipfire-cli/src/main.rs b/crates/hipfire-cli/src/main.rs index 9d3545287..2ff50582f 100644 --- a/crates/hipfire-cli/src/main.rs +++ b/crates/hipfire-cli/src/main.rs @@ -5147,6 +5147,12 @@ fn complete_request_slots( // and there is no opening marker in the generated stream to detect. let mut think = ThinkOutputRouter::new(matches!(prefix, AssistantPrefix::OpenThink)); let mut routed: Vec = Vec::new(); + // Incremental token -> text decoder. Byte-level BPE and byte-fallback + // spread one character across several tokens, so decoding per token would + // hand the think router (and the HTTP client) a U+FFFD per fragment -- + // "emoji: " arriving as "emoji: ???". Holds back only the trailing + // incomplete codepoint; drained after the receive loop. + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); let mut drain_routed = |routed: &mut Vec, content: &mut String, @@ -5172,7 +5178,9 @@ fn complete_request_slots( match ev { Event::Accepted { .. } => {} Event::Token { id } => { - let text = backend.tokenizer.decode(&[id]); + // Never `backend.tokenizer.decode(&[id])` -- lossy per token, + // splits multi-token UTF-8 into FFFD. Empty on a holdback step. + let text = text_stream.push(&backend.tokenizer, id); if !text.is_empty() { think.push_into(&text, &mut routed); drain_routed( @@ -5199,6 +5207,13 @@ fn complete_request_slots( } } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Route the held-back tail through the think router + // before finishing it, or those bytes are dropped silently. + let tail = text_stream.flush(); + if !tail.is_empty() { + think.push_into(&tail, &mut routed); + } // Flush any trailing partial marker as ordinary text in its channel. think.finish_into(&mut routed); drain_routed( @@ -9676,7 +9691,12 @@ mod tests { let registry = RegistryV1::parse(raw, "test").unwrap(); // Exact Qwen families get VMM + 262144 + 81920 - for tag in ["qwen3.5:4b", "qwen3.6:35b-a3b", "qwen3.8:27b", "qwen3.8:27b-fast"] { + for tag in [ + "qwen3.5:4b", + "qwen3.6:35b-a3b", + "qwen3.8:27b", + "qwen3.8:27b-fast", + ] { let (_, entry) = registry.model(tag).unwrap(); let resolved = resolved_for_model(&paths, tag, Some(tag), Some(entry)).unwrap(); assert_eq!( diff --git a/crates/hipfire-runtime/examples/daemon.rs b/crates/hipfire-runtime/examples/daemon.rs index 3acbfe856..2714bfc07 100644 --- a/crates/hipfire-runtime/examples/daemon.rs +++ b/crates/hipfire-runtime/examples/daemon.rs @@ -9112,6 +9112,69 @@ fn emit_visible_token(stdout: &mut impl std::io::Write, id: &str, text: &str) { let _ = stdout.flush(); } +/// Emit one streamed text fragment, skipping holdback steps. +/// +/// `TokenTextStream::push` returns `""` while a multi-byte character is still +/// split across tokens (every emoji, CJK under byte-fallback, `∑∫`). Emitting +/// that verbatim would put a contentless `{"type":"token"}` chunk on the wire +/// for every byte of every such character, so the plain JSONL emit sites route +/// their per-token fragment through here instead of repeating the check. +/// +/// Paths that classify a fragment before emitting (reasoning vs visible, +/// harmony channels, DSML sections) keep their own routing and only need the +/// same `is_empty()` skip. +fn emit_text_fragment(stdout: &mut impl std::io::Write, id: &str, text: &str) { + if text.is_empty() { + return; + } + emit_visible_token(stdout, id, text); +} + +/// True when a single token decodes to a bare EOS-class marker string. +/// +/// The id-based stop sets miss `<|endoftext|>` on several vocabs — encoding +/// the literal string doesn't round-trip to the special-token id (it yields +/// subwords), so the real id is never in the set and the decode loops also +/// guard on the decoded text. +/// +/// Deliberately checked on the token's OWN bytes rather than on a +/// `TokenTextStream` fragment: a fragment can carry the completion bytes of a +/// preceding split character, and the leading replacement char would defeat +/// the `trim()` match. Uses `decode_bytes`, so it is not the lossy per-token +/// `decode(&[tok])` this module bans for client output. +fn is_eos_class_marker_token( + tokenizer: &hipfire_runtime::tokenizer::Tokenizer, + token: u32, +) -> bool { + let bytes = tokenizer.decode_bytes(&[token]); + matches!( + std::str::from_utf8(&bytes).map(str::trim), + Ok("<|endoftext|>" | "" | "<|im_end|>") + ) +} + +/// True when a single token decodes to a bare `<|UPPER_SNAKE|>` structural +/// marker (`<|START_OF_TURN_TOKEN|>`, `<|CHATBOT_TOKEN|>`, …). +/// +/// Like [`is_eos_class_marker_token`], checked on the token's own bytes rather +/// than on a `TokenTextStream` fragment, which can carry a preceding split +/// character's completion bytes and so no longer match the `<|…|>` shape. +fn is_bracketed_marker_token( + tokenizer: &hipfire_runtime::tokenizer::Tokenizer, + token: u32, +) -> bool { + let bytes = tokenizer.decode_bytes(&[token]); + let Ok(s) = std::str::from_utf8(&bytes) else { + return false; + }; + s.len() > 4 + && s.starts_with("<|") + && s.ends_with("|>") + && s[2..s.len() - 2] + .chars() + .all(|c| c.is_ascii_uppercase() || c == '_') +} + /// Emit one producer-classified reasoning fragment. fn emit_reasoning_token(stdout: &mut impl std::io::Write, id: &str, text: &str) { let envelope = serde_json::json!({ @@ -17852,14 +17915,21 @@ fn ep_emit_token( stop: &[String], ) -> bool { text_acc.push_str(piece); - let _ = writeln!( - stdout, - r#"{{"type":"token","id":"{}","text":{},"attempt_id":{}}}"#, - id, - serde_json::to_string(piece).unwrap_or_else(|_| "\"\"".to_string()), - active_attempt_id() - ); - let _ = stdout.flush(); + // Holdback step: `TokenTextStream::push` returns "" while a character's + // UTF-8 is still split across tokens. Skip the wire event — emitting it + // would put a contentless chunk on the stream for every byte of every + // emoji. The stop-string check still runs; `text_acc` is unchanged, so it + // yields the same answer it did on the previous token. + if !piece.is_empty() { + let _ = writeln!( + stdout, + r#"{{"type":"token","id":"{}","text":{},"attempt_id":{}}}"#, + id, + serde_json::to_string(piece).unwrap_or_else(|_| "\"\"".to_string()), + active_attempt_id() + ); + let _ = stdout.flush(); + } stop.iter().any(|s| !s.is_empty() && text_acc.ends_with(s)) } @@ -18311,6 +18381,7 @@ fn ep_serve_ds4( let mut pos = prompt_n; let mut text_acc = String::new(); let mut local_emitted_ids: Vec = Vec::new(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); while generated < max_tokens { // FIX #3 (ep-no-abort): client cancel mid-decode → emit aborted+done, // reset EP cursors, stop. Without this a Pi/CLI cancel leaves the EP @@ -18336,10 +18407,17 @@ fn ep_serve_ds4( if next == eos_tok { break; } - let piece = m.tokenizer.as_ref().unwrap().decode(&[next]); - for ev in parser.feed(&piece) { - absorb_event(&ev); - emit_stream_event(stdout, id, ev); + // Incremental decode with holdback — never `decode(&[next])`, which is + // lossy per token and feeds the DSML parser FFFD for every byte of a + // multi-token character. `piece` is "" while a character is still + // split; the grammar matcher shares it and simply sees the completed + // character one step later instead of a replacement char. + let piece = text_stream.push(m.tokenizer.as_ref().unwrap(), next); + if !piece.is_empty() { + for ev in parser.feed(&piece) { + absorb_event(&ev); + emit_stream_event(stdout, id, ev); + } } emit_committed_event( stdout, @@ -18349,7 +18427,7 @@ fn ep_serve_ds4( t_decode.elapsed().as_millis() as u64, ); let _ = stdout.flush(); - if grammar_active { + if grammar_active && !piece.is_empty() { matcher.advance(&piece); } local_emitted_ids.push(next); @@ -18409,6 +18487,17 @@ fn ep_serve_ds4( None => break, }; } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Feed the held-back tail through the DSML parser + // before finishing it — dropping it is a silent truncation. + let tail = text_stream.flush(); + if !tail.is_empty() { + text_acc.push_str(&tail); + for ev in parser.feed(&tail) { + absorb_event(&ev); + emit_stream_event(stdout, id, ev); + } + } for ev in parser.finish() { absorb_event(&ev); emit_stream_event(stdout, id, ev); @@ -18715,6 +18804,7 @@ fn ep_serve_minimax( let mut generated = 0usize; let mut pos = prompt_n; let mut text_acc = String::new(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); while generated < max_tokens { // FIX #3 (ep-no-abort): client cancel mid-decode → emit aborted+done, // reset EP cursors, stop. @@ -18736,7 +18826,11 @@ fn ep_serve_minimax( if next == eos_tok { break; } - let piece = m.tokenizer.as_ref().unwrap().decode(&[next]); + // Incremental decode with holdback — never `decode(&[next])`, which + // is lossy per token and splits multi-token UTF-8 into FFFD. `piece` + // is "" while a character is still split; `ep_emit_token` skips the + // wire event for those but still evaluates the stop strings. + let piece = text_stream.push(m.tokenizer.as_ref().unwrap(), next); generated += 1; m.conversation_tokens.push(next); if ep_emit_token(stdout, id, &piece, &mut text_acc, stop) { @@ -18789,6 +18883,12 @@ fn ep_serve_minimax( } }; } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Without this flush those bytes are dropped. + let tail = text_stream.flush(); + if !tail.is_empty() { + ep_emit_token(stdout, id, &tail, &mut text_acc, stop); + } ep_emit_done( stdout, id, @@ -28755,6 +28855,7 @@ fn generate_deepseek4( &mut dsml_malformed, ); }; + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); while generated_count < max_tokens && next_tok != eos_tok { if check_abort(id) { @@ -28772,10 +28873,19 @@ fn generate_deepseek4( ); return; } - let frag = tokenizer.decode(&[next_tok]); - for ev in parser.feed(&frag) { - absorb_event(&ev); - emit_stream_event(stdout, id, ev); + // Incremental decode with holdback. Never + // `tokenizer.decode(&[next_tok])` — that is lossy per token and + // turns any character whose UTF-8 spans several tokens (emoji, + // byte-fallback CJK) into FFFD on the way to the DSML parser and + // the client. `frag` is "" while a character is still split; the + // grammar matcher shares it and simply sees the completed + // character one step later instead of a replacement char. + let frag = text_stream.push(tokenizer, next_tok); + if !frag.is_empty() { + for ev in parser.feed(&frag) { + absorb_event(&ev); + emit_stream_event(stdout, id, ev); + } } emit_committed_event( stdout, @@ -28786,7 +28896,7 @@ fn generate_deepseek4( ); let _ = stdout.flush(); m.conversation_tokens.push(next_tok); - if grammar_active { + if grammar_active && !frag.is_empty() { matcher.advance(&frag); } generated_count += 1; @@ -28812,6 +28922,17 @@ fn generate_deepseek4( } } } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Feed the held-back tail through the DSML + // parser before finishing it — dropping it here is a silent + // truncation of the visible turn. + let tail = text_stream.flush(); + if !tail.is_empty() { + for ev in parser.feed(&tail) { + absorb_event(&ev); + emit_stream_event(stdout, id, ev); + } + } // Flush any buffered partial markers / content. for ev in parser.finish() { absorb_event(&ev); @@ -29151,17 +29272,23 @@ fn generate_deepseek4_heterogeneous( let mut emit_text_buf = String::new(); let mut emit_tool_calls_buf = Vec::new(); let mut dsml_malformed = None; + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); while generated < max_tokens && next_tok != eos_tok { - let fragment = tokenizer.decode(&[next_tok]); - for event in parser.feed(&fragment) { - ds4_absorb_stream_event( - &event, - &mut emit_text_buf, - &mut emit_tool_calls_buf, - &mut dsml_malformed, - ); - emit_stream_event(stdout, id, event); + // Incremental decode with holdback — never + // `tokenizer.decode(&[next_tok])`, which is lossy per token and feeds + // the DSML parser FFFD for every byte of a multi-token character. + let fragment = text_stream.push(tokenizer, next_tok); + if !fragment.is_empty() { + for event in parser.feed(&fragment) { + ds4_absorb_stream_event( + &event, + &mut emit_text_buf, + &mut emit_tool_calls_buf, + &mut dsml_malformed, + ); + emit_stream_event(stdout, id, event); + } } emit_committed_event( stdout, @@ -29211,6 +29338,20 @@ fn generate_deepseek4_heterogeneous( } next_tok = deepseek4::sampling::sample_token(&logits, temp, top_k, top_p, &mut rng); } + // Generation can stop mid-character; route the held-back tail through the + // DSML parser before finishing it, or those bytes vanish silently. + let tail = text_stream.flush(); + if !tail.is_empty() { + for event in parser.feed(&tail) { + ds4_absorb_stream_event( + &event, + &mut emit_text_buf, + &mut emit_tool_calls_buf, + &mut dsml_malformed, + ); + emit_stream_event(stdout, id, event); + } + } for event in parser.finish() { ds4_absorb_stream_event( &event, @@ -29591,6 +29732,7 @@ fn generate_gemma4( let mut stop = false; let mut ttft_ms: Option = None; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); while !stop && generated_count < max_tokens { let committed_len = bundle.state.n_tokens; // KV/seq bound: the verify block occupies [L-1, L-1+draft_len+1). @@ -29641,17 +29783,22 @@ fn generate_gemma4( if ttft_ms.is_none() { ttft_ms = Some(t0.elapsed().as_secs_f64() * 1000.0); } + // Incremental decode with holdback — never `decode(&[t])`, + // which is lossy per token and splits multi-token UTF-8 into + // FFFD. `frag` is "" while a character is still incomplete. let frag = { let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[t]) + text_stream.push(tokenizer, t) }; - let envelope = serde_json::json!({ - "type": "token", - "id": id, - "text": frag, - }); - let _ = writeln!(stdout, "{}", envelope); - let _ = stdout.flush(); + if !frag.is_empty() { + let envelope = serde_json::json!({ + "type": "token", + "id": id, + "text": frag, + }); + let _ = writeln!(stdout, "{}", envelope); + let _ = stdout.flush(); + } m.conversation_tokens.push(t); generated_count += 1; if generated_count >= max_tokens { @@ -29670,6 +29817,20 @@ fn generate_gemma4( } } + // Generation can stop mid-character (max_tokens or EOS landing between + // two byte-fallback tokens). Without this flush those bytes are + // dropped silently. + let tail = text_stream.flush(); + if !tail.is_empty() { + let envelope = serde_json::json!({ + "type": "token", + "id": id, + "text": tail, + }); + let _ = writeln!(stdout, "{}", envelope); + let _ = stdout.flush(); + } + // ── Cursor settle. `spec_step` leaves n_tokens = L+accept_len+1 with // the final bonus's KV slot unwritten, and an EOS mid-block leaves // committed-but-not-emitted tokens counted. Re-anchor the cursor to @@ -29744,6 +29905,7 @@ fn generate_gemma4( let mut generated_count: usize = 0; let mut ttft_ms: Option = None; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); loop { if generated_count >= max_tokens { break; @@ -29757,18 +29919,13 @@ fn generate_gemma4( ttft_ms = Some(t0.elapsed().as_secs_f64() * 1000.0); } + // Incremental decode with holdback — never `decode(&[next_tok])`, + // which is lossy per token and splits multi-token UTF-8 into FFFD. let frag = { let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[next_tok]) + text_stream.push(tokenizer, next_tok) }; - let envelope = serde_json::json!({ - "type": "token", - "id": id, - "text": frag, - "attempt_id": active_attempt_id(), - }); - let _ = writeln!(stdout, "{}", envelope); - let _ = stdout.flush(); + emit_text_fragment(stdout, id, &frag); m.conversation_tokens.push(next_tok); generated_count += 1; @@ -29797,6 +29954,11 @@ fn generate_gemma4( } } + // Generation can stop mid-character (max_tokens or the KV-capacity guard + // firing between two byte-fallback tokens). Without this flush those bytes + // are dropped silently. + emit_text_fragment(stdout, id, &text_stream.flush()); + m.seq_pos = bundle.state.n_tokens; let decode_ms = decode_t0.elapsed().as_millis().max(1); @@ -29889,6 +30051,89 @@ enum GlimmerEmit { Tool(String), } +/// Route one decoded fragment through the harmony channel router, emit +/// whatever it yields, and record it. Returns the router's stop signal. +/// +/// Every glimmer emit site (AR seed, AR loop, spec fallback, accepted block, +/// stop-token flush, profit-probe prediction) funnels through here. They were +/// seven verbatim copies of this body, which is exactly how the per-token +/// `decode(&[tok])` bug survived in some of them — a fix applied idiom-by- +/// idiom skips the copies it doesn't recognise. +/// +/// `frag` is "" on a `TokenTextStream` holdback step; the router already +/// treats an empty fragment as a no-op and the recorder's text append is +/// likewise empty, while `token` is still recorded so `token_ids` stays +/// complete. +#[allow(clippy::too_many_arguments)] +fn glimmer_route_fragment( + stdout: &mut impl std::io::Write, + id: &str, + router: &mut GlimmerHarmonyRouter, + recorder: &mut GlimmerChannelRecorder, + visible_acc: &mut String, + tool_acc: &mut String, + token: u32, + frag: &str, +) -> bool { + let (events, should_stop) = router.push(frag); + for ev in events { + match ev { + GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), + GlimmerEmit::Token(text) => { + visible_acc.push_str(&text); + emit_visible_token(stdout, id, &text); + } + GlimmerEmit::Tool(text) => { + tool_acc.push_str(&text); + } + } + } + if router.just_forced() { + recorder.mark_forced_reasoning_close(); + } + recorder.push(token, frag); + should_stop +} + +/// End-of-stream counterpart to [`glimmer_route_fragment`]: route the +/// `TokenTextStream` tail down whichever harmony channel was open when +/// generation ended. +/// +/// Generation can stop mid-character (max_tokens or a stop token landing +/// between two byte-fallback tokens); dropping the tail is a silent +/// truncation of the visible turn. The tail carries no token id of its own — +/// see [`GlimmerChannelRecorder::push_trailing_text`]. +fn glimmer_route_trailing_fragment( + stdout: &mut impl std::io::Write, + id: &str, + router: &mut GlimmerHarmonyRouter, + recorder: &mut GlimmerChannelRecorder, + visible_acc: &mut String, + tool_acc: &mut String, + frag: &str, +) { + if frag.is_empty() { + return; + } + let (events, _) = router.push(frag); + for ev in events { + match ev { + GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), + GlimmerEmit::Token(text) => { + visible_acc.push_str(&text); + emit_visible_token(stdout, id, &text); + } + GlimmerEmit::Tool(text) => { + tool_acc.push_str(&text); + } + } + } + if router.just_forced() { + recorder.mark_forced_reasoning_close(); + } + recorder.push_trailing_text(frag); +} + /// Map the request's think budget onto Muse Glimmer's `Reasoning strength:` control. /// /// The model card names exactly four levels — `low | medium | high | xhigh` — and specifies @@ -30433,6 +30678,24 @@ impl GlimmerChannelRecorder { _ => {} } } + /// Append end-of-stream text that has no owning token id. + /// + /// Used only for a `TokenTextStream::flush`: generation stopped + /// mid-character, so those bytes belong to a token whose id was already + /// recorded (its `push` carried an empty fragment while the character was + /// still split across tokens). Routing the tail back through `push` with a + /// synthetic id would append a phantom token to `token_ids` and corrupt + /// the verbatim-splice replay. + fn push_trailing_text(&mut self, decoded_frag: &str) { + if decoded_frag.is_empty() { + return; + } + match &mut self.state { + GlimmerRecorderState::Body(body) => body.text.push_str(decoded_frag), + GlimmerRecorderState::Header { decoded, .. } => decoded.push_str(decoded_frag), + _ => {} + } + } fn mark_forced_reasoning_close(&mut self) { if matches!(self.state, GlimmerRecorderState::Refused(_)) { return; @@ -31679,6 +31942,11 @@ fn generate_muse_glimmer( let mut glimmer_emitted_ids: Vec = Vec::new(); let mut glimmer_visible_acc: String = String::new(); let mut glimmer_tool_acc: String = String::new(); + // One incremental decoder for the whole turn — every emit site (AR seed, + // AR loop, spec fallback, accepted block, stop flush, profit probe) shares + // it, so a character split across the boundary between two of those paths + // still reassembles. + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); // Speculative path: greedy is the validated 3.17x path; sampled chain-sample // (temp>0) is gated on dflash_fast_sample + batched logits + !min_p. // The drafter presence is gated by HIPFIRE_DFLASH_DRAFT / dflash_mode=off @@ -31781,25 +32049,20 @@ fn generate_muse_glimmer( // commit the stop token into KV/capture — fall through to shared post-spec finalization. let mut skip_spec_loop = false; if !stop_set.contains(&last_pick) && generated_count < max_tokens { - let frag = m.tokenizer.as_ref().unwrap().decode(&[last_pick]); + // Incremental decode with holdback — never `decode(&[last_pick])`, + // which is lossy per token and splits multi-token UTF-8 into FFFD. // Feed through harmony router; header fragments produce no visible event - let (events, should_stop_seed) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(last_pick, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), last_pick); + let should_stop_seed = glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + last_pick, + &frag, + ); // Seed was routed through harmony; if it indicated stop, we still count it but don't enter spec loop m.conversation_tokens.push(last_pick); glimmer_emitted_ids.push(last_pick); @@ -32078,26 +32341,17 @@ fn generate_muse_glimmer( last_pick = next_tok; break; } - let frag = m.tokenizer.as_ref().unwrap().decode(&[next_tok]); - let (events, should_stop_fb) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => { - emit_reasoning_token(stdout, id, &text) - } - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(next_tok, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), next_tok); + let should_stop_fb = glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + next_tok, + &frag, + ); m.conversation_tokens.push(next_tok); glimmer_emitted_ids.push(next_tok); generated_count += 1; @@ -32433,26 +32687,17 @@ fn generate_muse_glimmer( break; } if stop_set.contains(&tok) { - let frag = m.tokenizer.as_ref().unwrap().decode(&[tok]); - let (events, _) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => { - emit_reasoning_token(stdout, id, &text) - } - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(tok, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), tok); + glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + tok, + &frag, + ); last_pick = tok; m.conversation_tokens.push(tok); glimmer_emitted_ids.push(tok); @@ -32460,24 +32705,17 @@ fn generate_muse_glimmer( should_stop_block = true; break; } - let frag = m.tokenizer.as_ref().unwrap().decode(&[tok]); - let (events, should_stop) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(tok, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), tok); + let should_stop = glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + tok, + &frag, + ); m.conversation_tokens.push(tok); glimmer_emitted_ids.push(tok); generated_count += 1; @@ -32629,26 +32867,17 @@ fn generate_muse_glimmer( if generated_count >= max_tokens { break; } - let frag = m.tokenizer.as_ref().unwrap().decode(&[pred]); - let (events, should_stop_pred) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => { - emit_reasoning_token(stdout, id, &text) - } - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(pred, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), pred); + let should_stop_pred = glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + pred, + &frag, + ); m.conversation_tokens.push(pred); glimmer_emitted_ids.push(pred); generated_count += 1; @@ -32729,49 +32958,35 @@ fn generate_muse_glimmer( // eos/eot always stop; eom is handled via router text marker if stop_set.contains(&next_tok) { // Flush any pending via router marker path then stop - let frag = m.tokenizer.as_ref().unwrap().decode(&[next_tok]); - let (events, _) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(next_tok, &frag); + let frag = text_stream.push(m.tokenizer.as_ref().unwrap(), next_tok); + glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + next_tok, + &frag, + ); // Push stop token to KV? No — mirror prior break-before-push for eos. break; } let frag = { let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[next_tok]) + text_stream.push(tokenizer, next_tok) }; - let (events, should_stop) = harmony_router.push(&frag); - for ev in events { - match ev { - GlimmerEmit::Reasoning(text) => emit_reasoning_token(stdout, id, &text), - GlimmerEmit::Token(text) => { - glimmer_visible_acc.push_str(&text); - emit_visible_token(stdout, id, &text) - } - GlimmerEmit::Tool(text) => { - glimmer_tool_acc.push_str(&text); - } - } - } - if harmony_router.just_forced() { - glimmer_recorder.mark_forced_reasoning_close(); - } - glimmer_recorder.push(next_tok, &frag); + let should_stop = glimmer_route_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + next_tok, + &frag, + ); if should_stop { break; } @@ -32810,6 +33025,19 @@ fn generate_muse_glimmer( } } } + // Generation can stop mid-character; route the held-back tail down the + // open channel BEFORE draining the router, so it is classified like + // every other fragment rather than dropped. + let tail = text_stream.flush(); + glimmer_route_trailing_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + &tail, + ); // Flush any trailing incomplete channel text for ev in harmony_router.flush() { match ev { @@ -32822,6 +33050,20 @@ fn generate_muse_glimmer( } } + // A pure-spec turn (`ar_pending` never set) skips the AR tail above and so + // never reached that flush. `TokenTextStream::flush` is idempotent — this + // is a no-op when the AR tail already drained it. + let tail = text_stream.flush(); + glimmer_route_trailing_fragment( + stdout, + id, + &mut harmony_router, + &mut glimmer_recorder, + &mut glimmer_visible_acc, + &mut glimmer_tool_acc, + &tail, + ); + // A parse failure here used to vanish into `unwrap_or_default()`: the model // emits a whole turn into the tool channel, ATEM refuses it, and the caller // gets an empty content with no tool_calls and no reason why. Keep the @@ -33225,6 +33467,7 @@ fn generate_lfm2moe( let mut generated_count: usize = 0; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); loop { if check_abort(id) { let ep = production_fail_closed_rollback(m, gpu, None, None); @@ -33239,27 +33482,22 @@ fn generate_lfm2moe( break; } - let frag = { - let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[next_tok]) - }; // String-level EOS-class guard. The id-based `stop_toks` above misses // `<|endoftext|>` because encoding the literal STRING doesn't round-trip // to the special-token id (it yields subwords), so the real token id is - // never in the set. The daemon decodes one token at a time, so the - // leaking turn-end token arrives as its own frag — catch it on the - // decoded text and stop WITHOUT emitting (was: "...Paris.<|endoftext|>"). - if matches!(frag.trim(), "<|endoftext|>" | "" | "<|im_end|>") { + // never in the set. Catch it on the token's decoded text and stop + // WITHOUT emitting (was: "...Paris.<|endoftext|>"). + if is_eos_class_marker_token(m.tokenizer.as_ref().unwrap(), next_tok) { break; } - let envelope = serde_json::json!({ - "type": "token", - "id": id, - "text": frag, - "attempt_id": active_attempt_id(), - }); - let _ = writeln!(stdout, "{}", envelope); - let _ = stdout.flush(); + // Incremental decode with holdback: a character whose UTF-8 spans + // several tokens yields "" until its last byte arrives. Never + // `decode(&[next_tok])` — lossy per token, splits UTF-8 into FFFD. + let frag = { + let tokenizer = m.tokenizer.as_ref().unwrap(); + text_stream.push(tokenizer, next_tok) + }; + emit_text_fragment(stdout, id, &frag); m.conversation_tokens.push(next_tok); generated_count += 1; @@ -33297,6 +33535,10 @@ fn generate_lfm2moe( return; } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Without this flush those bytes are dropped. + emit_text_fragment(stdout, id, &text_stream.flush()); + m.seq_pos = m.lfm2moe().unwrap().state.n_tokens; let decode_ms = decode_t0.elapsed().as_millis().max(1); @@ -33682,6 +33924,7 @@ fn generate_minimax( let mut generated_count: usize = 0; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); loop { if generated_count >= max_tokens { break; @@ -33692,20 +33935,17 @@ fn generate_minimax( break; } - // Emit the text fragment. Build through serde_json so a user-supplied - // `id` or arbitrary-UTF-8 fragment can't corrupt the JSONL line. + // Emit the text fragment through the incremental decoder, which holds + // back a character whose UTF-8 spans several tokens until its last + // byte arrives (never `decode(&[next_tok])` — lossy per token, splits + // UTF-8 into FFFD). `emit_text_fragment` builds through serde_json so + // a user-supplied `id` or arbitrary-UTF-8 fragment can't corrupt the + // JSONL line, and skips the "" holdback steps. let frag = { let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[next_tok]) + text_stream.push(tokenizer, next_tok) }; - let envelope = serde_json::json!({ - "type": "token", - "id": id, - "text": frag, - "attempt_id": active_attempt_id(), - }); - let _ = writeln!(stdout, "{}", envelope); - let _ = stdout.flush(); + emit_text_fragment(stdout, id, &frag); m.conversation_tokens.push(next_tok); generated_count += 1; @@ -33730,6 +33970,10 @@ fn generate_minimax( } } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Without this flush those bytes are dropped. + emit_text_fragment(stdout, id, &text_stream.flush()); + m.seq_pos = m.minimax().unwrap().state.n_tokens; let decode_ms = decode_t0.elapsed().as_millis().max(1); @@ -34122,6 +34366,55 @@ fn generate_cohere2moe( Text, Action, } + /// Route one decoded text fragment down whichever agentic channel is + /// currently open. Shared by the decode loop and the end-of-stream flush, + /// so a character that completed only at the flush lands on the same + /// channel the turn ended on instead of being dropped — dropping it is a + /// silent truncation. + /// + /// Skips empty fragments: `TokenTextStream::push` returns "" on every + /// holdback step, and emitting those would put a contentless chunk on the + /// wire for each byte of each emoji. + #[allow(clippy::too_many_arguments)] + fn route_fragment( + stdout: &mut impl std::io::Write, + id: &str, + sec: Sec, + frag: &str, + action_buf: &mut String, + vis_buf: &mut String, + think_count: &mut usize, + emitted_visible: &mut bool, + ) { + if frag.is_empty() { + return; + } + match sec { + Sec::Action => action_buf.push_str(frag), + Sec::Think => { + // Reasoning channel: tagged so clients can fold it; the CLI + // (ignoring unknown fields) shows it inline. + let _ = writeln!( + stdout, + "{}", + serde_json::json!({"type": "token", "id": id, "text": frag, "reasoning": true, "attempt_id": active_attempt_id()}) + ); + let _ = stdout.flush(); + *think_count += 1; + } + Sec::Text | Sec::Pre => { + vis_buf.push_str(frag); + let _ = writeln!( + stdout, + "{}", + serde_json::json!({"type": "token", "id": id, "text": frag, "attempt_id": active_attempt_id()}) + ); + let _ = stdout.flush(); + *emitted_visible = true; + } + } + } + let mut sec = Sec::Pre; let mut action_buf = String::new(); @@ -34208,6 +34501,7 @@ fn generate_cohere2moe( let mut generated_count: usize = 0; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); loop { if generated_count >= max_tokens { break; @@ -34335,12 +34629,6 @@ fn generate_cohere2moe( } sec = Sec::Pre; } else { - // Build the fragment through serde_json so arbitrary UTF-8 can't - // corrupt the JSONL line. - let frag = { - let tokenizer = m.tokenizer.as_ref().unwrap(); - tokenizer.decode(&[next_tok]) - }; // Defense-in-depth: never emit a Cohere structural marker into // visible output / the action buffer. The ID state machine above // handles the 6 THINKING/TEXT/ACTION markers; this catches any OTHER @@ -34348,40 +34636,32 @@ fn generate_cohere2moe( // CHATBOT_TOKEN, START_TOOL_RESULT, …) — each decodes to a full // `<|MARKER|>`. The token is still fed to decode_step below; only its // emit is dropped, so a state-machine miss can never leak a marker. - let is_marker = frag.len() > 4 - && frag.starts_with("<|") - && frag.ends_with("|>") - && frag[2..frag.len() - 2] - .chars() - .all(|c| c.is_ascii_uppercase() || c == '_'); - if is_marker { - // suppressed - } else { - match sec { - Sec::Action => action_buf.push_str(&frag), - Sec::Think => { - // Reasoning channel: tagged so clients can fold it; the CLI - // (ignoring unknown fields) shows it inline. - let _ = writeln!( - stdout, - "{}", - serde_json::json!({"type": "token", "id": id, "text": frag, "reasoning": true, "attempt_id": active_attempt_id()}) - ); - let _ = stdout.flush(); - think_count += 1; - } - Sec::Text | Sec::Pre => { - vis_buf.push_str(&frag); - let _ = writeln!( - stdout, - "{}", - serde_json::json!({"type": "token", "id": id, "text": frag, "attempt_id": active_attempt_id()}) - ); - let _ = stdout.flush(); - emitted_visible = true; - } + // + // Tested on the token's OWN bytes, not on the stream fragment: a + // fragment can carry the completion bytes of a preceding split + // character, whose leading char would defeat the `<|…|>` shape + // match. + let frag = { + let tokenizer = m.tokenizer.as_ref().unwrap(); + if is_bracketed_marker_token(tokenizer, next_tok) { + String::new() + } else { + // Incremental decode with holdback — never + // `decode(&[next_tok])`, which is lossy per token and + // splits multi-token UTF-8 into FFFD. + text_stream.push(tokenizer, next_tok) } - } + }; + route_fragment( + stdout, + id, + sec, + &frag, + &mut action_buf, + &mut vis_buf, + &mut think_count, + &mut emitted_visible, + ); } // Advance one step on the freshly sampled token (plain eager decode — @@ -34403,6 +34683,23 @@ fn generate_cohere2moe( } } + // Generation can stop mid-character (max_tokens or a degenerate-output + // guard firing between two byte-fallback tokens). Route the held-back + // tail down whichever channel was open when the turn ended — dropping it + // is a silent truncation, and it must reach `vis_buf` / `action_buf` too + // so the tool-call recovery below sees the complete text. + let tail = text_stream.flush(); + route_fragment( + stdout, + id, + sec, + &tail, + &mut action_buf, + &mut vis_buf, + &mut think_count, + &mut emitted_visible, + ); + m.seq_pos = m.cohere2moe().unwrap().state.n_tokens; // Tool-call-as-text recovery: if the model never emitted a <|START_ACTION|> @@ -34647,6 +34944,7 @@ fn generate_qwen2( let mut generated_count: usize = 0; let eos_set: &[u32] = &cfg.eos_token_ids; let decode_t0 = Instant::now(); + let mut text_stream = hipfire_runtime::tokenizer::TokenTextStream::new(); let mut next_tok = match gpu.argmax_f32(&state.logits, cfg.vocab_size) { Ok(t) => t, Err(e) => { @@ -34663,21 +34961,16 @@ fn generate_qwen2( if eos_set.contains(&next_tok) { break; } - // Emit text fragment for this token. Tokenizer.decode handles - // BPE byte-fragment reassembly; for special tokens that decode - // to an empty string we still advance the loop. Build through - // serde_json so `id` (user-supplied) and `frag` (arbitrary - // UTF-8 with possible `"` / `\` / control chars) can't corrupt - // the JSONL line. - let frag = tokenizer.decode(&[next_tok]); - let envelope = serde_json::json!({ - "type": "token", - "id": id, - "text": frag, - "attempt_id": active_attempt_id(), - }); - let _ = writeln!(stdout, "{}", envelope); - let _ = stdout.flush(); + // Emit text fragment for this token through the incremental + // decoder: a character whose UTF-8 spans several tokens (emoji, + // byte-fallback CJK) yields "" until its last byte arrives, and + // `emit_text_fragment` skips those holdback steps. Never + // `tokenizer.decode(&[next_tok])` here — that is lossy per token + // and splits UTF-8 into FFFD. The envelope is built through + // serde_json so `id` (user-supplied) and the text (arbitrary UTF-8 + // with possible `"` / `\` / control chars) can't corrupt the line. + let frag = text_stream.push(tokenizer, next_tok); + emit_text_fragment(stdout, id, &frag); m.conversation_tokens.push(next_tok); generated_count += 1; @@ -34691,6 +34984,10 @@ fn generate_qwen2( } } + // Generation can stop mid-character (max_tokens reached between two + // byte-fallback tokens). Without this flush those bytes are dropped. + emit_text_fragment(stdout, id, &text_stream.flush()); + // Daemon bookkeeping: seq_pos matches Qwen2State's internal cursor. m.seq_pos = state.next_pos; @@ -43694,3 +43991,192 @@ mod serve_fault_inject_tests { assert!(!model_retry_reset_eligible(0)); // llama } } + +/// Emit-level coverage for the incremental token → text decode. +/// +/// The helper's own invariants live in `hipfire-runtime`'s +/// `token_text_stream_tests`; these drive the daemon's *wiring* — the JSONL +/// events a client actually receives — with a synthetic byte-fallback vocab, no +/// GPU and no model file. What they pin that nothing else does: no `U+FFFD` on +/// the wire, and no contentless `{"type":"token"}` chunk per holdback step. +#[cfg(test)] +mod streaming_emit_tests { + use super::*; + + /// Synthetic SentencePiece vocab whose ids `0..=255` are the `<0xHH>` + /// byte-fallback tokens. Built through the public `from_hf_json` path (no + /// `Ġ` in the vocab ⇒ the SentencePiece branch), so a character's UTF-8 + /// spreads across one token per byte — exactly the shape that produced + /// `"emoji: ���������"`. + fn byte_fallback_tokenizer() -> hipfire_runtime::tokenizer::Tokenizer { + let mut vocab = serde_json::Map::new(); + for b in 0u32..=255 { + vocab.insert(format!("<0x{b:02X}>"), serde_json::json!(b)); + } + let json = serde_json::json!({ + "model": { "type": "Unigram", "vocab": vocab }, + }) + .to_string(); + hipfire_runtime::tokenizer::Tokenizer::from_hf_json(&json) + .expect("synthetic byte-fallback vocab must load") + } + + /// Split `text` into one byte-fallback token per UTF-8 byte. + fn byte_tokens(text: &str) -> Vec { + text.as_bytes().iter().map(|&b| b as u32).collect() + } + + /// Parse captured JSONL into the `(type, text)` pairs a client would see. + fn events(buf: &[u8]) -> Vec<(String, String)> { + String::from_utf8(buf.to_vec()) + .expect("emitted JSONL must be valid UTF-8") + .lines() + .filter(|l| !l.trim().is_empty()) + .map(|l| { + let v: serde_json::Value = + serde_json::from_str(l).unwrap_or_else(|e| panic!("bad JSONL {l:?}: {e}")); + ( + v["type"].as_str().unwrap_or_default().to_string(), + v["text"].as_str().unwrap_or_default().to_string(), + ) + }) + .collect() + } + + /// Drive the plain JSONL emit path exactly as an arch decode loop does. + fn drive_plain(text: &str) -> Vec<(String, String)> { + let tk = byte_fallback_tokenizer(); + let mut out: Vec = Vec::new(); + let mut stream = hipfire_runtime::tokenizer::TokenTextStream::new(); + for tok in byte_tokens(text) { + let frag = stream.push(&tk, tok); + emit_text_fragment(&mut out, "req-1", &frag); + } + emit_text_fragment(&mut out, "req-1", &stream.flush()); + events(&out) + } + + #[test] + fn emitted_tokens_carry_no_replacement_chars() { + for text in ["emoji: 🎉🔥🚀", "math ∑∫≈", "中文测试", "café naïve"] { + let evs = drive_plain(text); + for (_, t) in &evs { + assert!( + !t.contains('\u{FFFD}'), + "emitted {t:?} contains U+FFFD for input {text:?}" + ); + } + } + } + + #[test] + fn concatenated_token_events_equal_the_source_text() { + for text in ["emoji: 🎉🔥🚀", "math ∑∫≈", "中文测试", "plain ascii", "🎉"] + { + let joined: String = drive_plain(text).into_iter().map(|(_, t)| t).collect(); + assert_eq!(joined, text, "round-trip failed for {text:?}"); + } + } + + /// The holdback contract. Without it every byte of every emoji puts a + /// contentless chunk on the wire — the one thing no helper-level test + /// covers, because it is about the wiring rather than the decoder. + #[test] + fn no_empty_text_events_are_emitted() { + let evs = drive_plain("emoji: 🎉🔥🚀 中文 ∑"); + assert!(!evs.is_empty(), "expected at least one token event"); + for (ty, t) in &evs { + assert_eq!(ty, "token"); + assert!(!t.is_empty(), "holdback step leaked an empty token event"); + } + } + + /// Baseline: the same loop written the old way really does corrupt, so + /// these tests would fail against the pre-fix daemon. + #[test] + fn per_token_decode_baseline_is_corrupt() { + let tk = byte_fallback_tokenizer(); + let corrupt: String = byte_tokens("emoji: 🎉") + .iter() + .map(|&t| tk.decode(&[t])) + .collect(); + assert!(corrupt.contains('\u{FFFD}')); + assert_ne!(corrupt, "emoji: 🎉"); + } + + /// Generation stopping mid-character must not silently drop the tail. + #[test] + fn truncated_generation_flushes_its_tail() { + let tk = byte_fallback_tokenizer(); + let all = byte_tokens("ok 🎉"); + let truncated = &all[..all.len() - 1]; // cut inside the emoji + let mut out: Vec = Vec::new(); + let mut stream = hipfire_runtime::tokenizer::TokenTextStream::new(); + for &tok in truncated { + let frag = stream.push(&tk, tok); + emit_text_fragment(&mut out, "req-1", &frag); + } + let joined_before: String = events(&out).into_iter().map(|(_, t)| t).collect(); + assert_eq!( + joined_before, "ok ", + "partial char must not be emitted early" + ); + + emit_text_fragment(&mut out, "req-1", &stream.flush()); + let joined: String = events(&out).into_iter().map(|(_, t)| t).collect(); + assert!( + joined.len() > joined_before.len(), + "flush dropped the truncated tail entirely" + ); + assert!(joined.starts_with("ok ")); + } + + /// The glimmer harmony router path: same guarantees, but the fragment is + /// channel-classified before it reaches the wire. + #[test] + fn glimmer_router_path_reassembles_multi_token_characters() { + let tk = byte_fallback_tokenizer(); + let body = "Hello 🎉 中文 ∑!"; + let text = format!(" to=user<|message|>{body}"); + let mut out: Vec = Vec::new(); + let mut router = GlimmerHarmonyRouter::new(0); + let mut recorder = GlimmerChannelRecorder::new(); + let mut visible = String::new(); + let mut tool = String::new(); + let mut stream = hipfire_runtime::tokenizer::TokenTextStream::new(); + + for tok in byte_tokens(&text) { + let frag = stream.push(&tk, tok); + glimmer_route_fragment( + &mut out, + "req-1", + &mut router, + &mut recorder, + &mut visible, + &mut tool, + tok, + &frag, + ); + } + let tail = stream.flush(); + glimmer_route_trailing_fragment( + &mut out, + "req-1", + &mut router, + &mut recorder, + &mut visible, + &mut tool, + &tail, + ); + + assert!( + !visible.contains('\u{FFFD}'), + "harmony-routed visible text mojibaked: {visible:?}" + ); + assert_eq!(visible, body, "harmony router lost or corrupted characters"); + for (_, t) in events(&out) { + assert!(!t.is_empty(), "holdback step leaked an empty token event"); + assert!(!t.contains('\u{FFFD}')); + } + } +} diff --git a/crates/hipfire-runtime/src/lib.rs b/crates/hipfire-runtime/src/lib.rs index b4300fbc6..ae1bd54c9 100644 --- a/crates/hipfire-runtime/src/lib.rs +++ b/crates/hipfire-runtime/src/lib.rs @@ -54,8 +54,8 @@ pub mod sampler; pub mod serve; pub mod spec; -pub mod spec_ngram; pub mod ngram_mod; +pub mod spec_ngram; pub mod swap; pub mod tp_shard; #[cfg(feature = "deltanet")] @@ -68,6 +68,10 @@ pub mod eos_filter; pub mod prompt_frame; pub mod semantic; pub mod session_table; +/// Source guard: no per-token `decode(&[tok])` on a client-output path. +/// Test-only; see the module docs for why the guard exists at all. +#[cfg(test)] +mod streaming_decode_guard; pub mod tokenizer; pub mod tool_call; diff --git a/crates/hipfire-runtime/src/streaming_decode_guard.rs b/crates/hipfire-runtime/src/streaming_decode_guard.rs new file mode 100644 index 000000000..2c75fecf4 --- /dev/null +++ b/crates/hipfire-runtime/src/streaming_decode_guard.rs @@ -0,0 +1,244 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Kaden Schutt +// hipfire — see LICENSE and NOTICE in the project root. + +//! Source guard: no per-token `decode(&[tok])` on a client-output path. +//! +//! `Tokenizer::decode()` reassembles UTF-8 byte fragments only within a single +//! call, so calling it once per streamed token runs `from_utf8_lossy` over a +//! partial sequence. Byte-level BPE and SentencePiece byte-fallback spread one +//! character across several tokens (`中` = `<0xE4> <0xB8> <0xAD>`, every emoji, +//! `∑∫`), so the client receives `���` where the model emitted an emoji. +//! [`crate::tokenizer::TokenTextStream`] is the fix; this module checks that it +//! is actually *used*. +//! +//! Every other test around this fix verifies the helper. This one verifies the +//! wiring — without it, an arch added next month reintroduces the bug on its +//! own decode loop and every other test still passes. That is exactly how the +//! bug survived on eight archs while the mainline qwen35 path was correct. +//! +//! Precedent for this style of guard in this repo: `scripts/check-fmt-bomb.sh`, +//! `scripts/test-gpu-lock.sh`. + +#![cfg(test)] + +use std::path::{Path, PathBuf}; + +/// Repository root, derived from this crate's manifest directory. +fn repo_root() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .canonicalize() + .expect("repo root must resolve from CARGO_MANIFEST_DIR") +} + +/// Source files the guard covers. +/// +/// `daemon.rs` and the per-arch crates are where every streamed decode loop +/// lives. `hipfire-cli/src/main.rs` is included because the multi-slot serve +/// backend does its own token → text loop there — that path was one of the +/// affected sites and nothing else would keep it fixed. +fn guarded_files(root: &Path) -> Vec { + let mut files = vec![ + root.join("crates/hipfire-runtime/examples/daemon.rs"), + root.join("crates/hipfire-cli/src/main.rs"), + ]; + let crates_dir = root.join("crates"); + let mut arch_dirs: Vec = std::fs::read_dir(&crates_dir) + .expect("crates/ must be readable") + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .filter(|p| { + p.file_name() + .and_then(|n| n.to_str()) + .is_some_and(|n| n.starts_with("hipfire-arch-")) + }) + .map(|p| p.join("src")) + .filter(|p| p.is_dir()) + .collect(); + arch_dirs.sort(); + for dir in arch_dirs { + collect_rs(&dir, &mut files); + } + files.sort(); + files +} + +fn collect_rs(dir: &Path, out: &mut Vec) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + let mut paths: Vec = entries.filter_map(|e| e.ok()).map(|e| e.path()).collect(); + paths.sort(); + for path in paths { + if path.is_dir() { + collect_rs(&path, out); + } else if path.extension().and_then(|e| e.to_str()) == Some("rs") { + out.push(path); + } + } +} + +/// Line ranges (1-based, inclusive) covered by a `#[cfg(test)]` item. +/// +/// An attribute at indent N marks the item that follows; that item's body ends +/// at the first line that is exactly `}` (or `};`) at the same indent. rustfmt +/// guarantees that shape for every item this repo compiles, and the guard +/// self-tests below fail loudly if the assumption ever stops holding. +fn cfg_test_regions(lines: &[&str]) -> Vec<(usize, usize)> { + let mut regions = Vec::new(); + let mut i = 0usize; + while i < lines.len() { + if lines[i].trim() == "#[cfg(test)]" { + let indent = lines[i].len() - lines[i].trim_start().len(); + let close = format!("{}}}", " ".repeat(indent)); + let close_semi = format!("{close};"); + let mut j = i + 1; + while j < lines.len() && lines[j] != close && lines[j] != close_semi { + j += 1; + } + regions.push((i + 1, j + 1)); + i = j; + } + i += 1; + } + regions +} + +/// Why a `decode(&[` occurrence is permitted, or `None` if it is a violation. +/// +/// Kept deliberately narrow. A bare pattern with no rationale invites the first +/// person who hits a failure to widen it until it stops complaining, and the +/// guard becomes decorative. +fn allowed_reason(lines: &[&str], idx: usize, in_test: bool) -> Option<&'static str> { + let line = lines[idx]; + let trimmed = line.trim_start(); + if trimmed.starts_with("//") { + return Some("comment, not code"); + } + if in_test { + return Some("#[cfg(test)] — test fixture, not client output"); + } + // Grammar matchers drive a text state machine, not client output, and are + // deliberately out of scope (they do still see U+FFFD for non-ASCII, which + // makes grammar-constrained decoding over non-ASCII a separate open + // question). The decode and the `advance` are sometimes split across lines, + // so look at a small window. + let window_end = (idx + 3).min(lines.len()); + let window = lines[idx..window_end].join(" "); + if window.contains("matcher.advance") || window.contains("matcher.is_token_allowed") { + return Some("grammar matcher — text state machine, not client output"); + } + // Whole-vocab dumps decode every id independently on purpose; there is no + // stream to reassemble across. + if line.contains("(0..") && line.contains(".map(") { + return Some("whole-vocab dump — per-id decode is the intent"); + } + None +} + +/// Scan one file, returning `file:line: source` for every violation. +fn violations_in(path: &Path, source: &str) -> Vec { + let lines: Vec<&str> = source.split('\n').collect(); + let regions = cfg_test_regions(&lines); + let mut out = Vec::new(); + for (idx, line) in lines.iter().enumerate() { + if !line.contains("decode(&[") { + continue; + } + let lineno = idx + 1; + let in_test = regions.iter().any(|&(a, b)| a <= lineno && lineno <= b); + if allowed_reason(&lines, idx, in_test).is_none() { + out.push(format!("{}:{}: {}", path.display(), lineno, line.trim())); + } + } + out +} + +/// The guard itself. +#[test] +fn no_per_token_decode_on_client_output_paths() { + let root = repo_root(); + let files = guarded_files(&root); + assert!( + files.len() > 5, + "guard scanned only {} files — the path set is wrong, not the code clean", + files.len() + ); + + let mut scanned_matches = 0usize; + let mut violations = Vec::new(); + for path in &files { + let Ok(source) = std::fs::read_to_string(path) else { + continue; + }; + scanned_matches += source.matches("decode(&[").count(); + violations.extend(violations_in(path, &source)); + } + + // Guard against the guard silently matching nothing (a moved file, a + // renamed API): the legitimate grammar/vocab/test uses are still there, so + // a zero total means the scan itself broke. + assert!( + scanned_matches > 5, + "guard found only {scanned_matches} `decode(&[` occurrences across {} files — \ + the scan is broken, not the tree clean", + files.len() + ); + + assert!( + violations.is_empty(), + "per-token `decode(&[tok])` on a client-output path — a character whose UTF-8 \ + spans several tokens (emoji, byte-fallback CJK, ∑∫) reaches the client as U+FFFD.\n\ + Use `hipfire_runtime::tokenizer::TokenTextStream` (push per token, flush at end of \ + stream) instead.\n\n{}", + violations.join("\n") + ); +} + +/// The guard must actually reject the thing it exists to reject. +#[test] +fn guard_flags_a_reintroduced_per_token_decode() { + let bad = "fn generate_newarch() {\n let frag = tokenizer.decode(&[next_tok]);\n emit(&frag);\n}\n"; + let found = violations_in(Path::new("synthetic.rs"), bad); + assert_eq!( + found.len(), + 1, + "guard failed to flag a fresh per-token decode: {found:?}" + ); + assert!(found[0].contains("synthetic.rs:2")); +} + +/// The allowlist must stay narrow — and must still permit exactly the three +/// documented legitimate shapes. +#[test] +fn guard_allowlist_covers_only_the_documented_shapes() { + let grammar = "fn f() {\n grammar_matcher.advance(&tokenizer.decode(&[t]));\n}\n"; + assert!(violations_in(Path::new("g.rs"), grammar).is_empty()); + + let grammar_split = + "fn f() {\n let text = tokenizer.decode(&[t]);\n grammar_matcher.advance(&text);\n}\n"; + assert!(violations_in(Path::new("g.rs"), grammar_split).is_empty()); + + let vocab_dump = + "fn f() {\n let v: Vec = (0..n).map(|id| tokenizer.decode(&[id])).collect();\n}\n"; + assert!(violations_in(Path::new("v.rs"), vocab_dump).is_empty()); + + let in_test = "#[cfg(test)]\nmod tests {\n fn t() {\n let s = tok.decode(&[id]);\n }\n}\n"; + assert!(violations_in(Path::new("t.rs"), in_test).is_empty()); + + // …but not a plain emit dressed up next to unrelated text. + let sneaky = "fn f() {\n let frag = tokenizer.decode(&[t]);\n buf.push_str(&frag);\n}\n"; + assert_eq!(violations_in(Path::new("s.rs"), sneaky).len(), 1); +} + +/// `#[cfg(test)]` region detection must not swallow the rest of the file — if +/// it did, every real violation after the first test module would be excused. +#[test] +fn cfg_test_region_ends_at_the_matching_close() { + let src = "#[cfg(test)]\nmod tests {\n fn t() {}\n}\n\nfn real() {\n let f = tok.decode(&[t]);\n}\n"; + let lines: Vec<&str> = src.split('\n').collect(); + let regions = cfg_test_regions(&lines); + assert_eq!(regions, vec![(1, 4)], "region must close at the top-level }}"); + assert_eq!(violations_in(Path::new("r.rs"), src).len(), 1); +} diff --git a/crates/hipfire-runtime/src/tokenizer.rs b/crates/hipfire-runtime/src/tokenizer.rs index 5509e0b08..8daec876c 100644 --- a/crates/hipfire-runtime/src/tokenizer.rs +++ b/crates/hipfire-runtime/src/tokenizer.rs @@ -317,7 +317,6 @@ fn sp_dummy_prefix_from_hf_json(tok: &serde_json::Value) -> bool { || pre_tokenizer.map(pretokenizer_prepends).unwrap_or(false) } - impl Tokenizer { /// Load tokenizer from GGUF metadata. pub fn from_gguf(gguf: &GgufFile) -> Result { @@ -1731,6 +1730,133 @@ fn needs_trailing_ws_strip(s: &str) -> bool { false } +/// Incremental token → text decoder for streaming emit paths. +/// +/// `Tokenizer::decode()` reassembles byte fragments only *within a single +/// call*, so calling it once per streamed token (`decode(&[tok])`) runs +/// `from_utf8_lossy` over a partial sequence. Byte-level BPE and +/// SentencePiece byte-fallback both spread one character across several +/// tokens (`中` = `<0xE4> <0xB8> <0xAD>`, every emoji, `∑∫`), so per-token +/// decode yields U+FFFD for each fragment and the client sees `���` where +/// the model emitted an emoji. +/// +/// `TokenTextStream` holds back only the trailing *incomplete* codepoint and +/// emits everything before it. `Tokenizer::decode_bytes` is exactly +/// concatenative over tokens, so this needs a pending-byte buffer and nothing +/// else: **O(1) per token**, with the buffer never exceeding 3 bytes between +/// `push` calls. Do not implement streaming decode by re-decoding the whole +/// streamed-token vector each step — that is quadratic and unnecessary. +/// +/// ```ignore +/// let mut stream = TokenTextStream::new(); +/// for tok in tokens { +/// let frag = stream.push(&tokenizer, tok); // "" while a char is split +/// if !frag.is_empty() { emit(&frag); } +/// } +/// let tail = stream.flush(); // mid-character truncation only +/// if !tail.is_empty() { emit(&tail); } +/// ``` +/// +/// Two requirements that are easy to miss: +/// +/// 1. **`flush()` at end of stream.** Generation can stop mid-character (max +/// tokens reached between two byte-fallback tokens). Without the flush +/// those bytes are dropped silently. +/// 2. **A holdback step returns `""`.** Emit sites must skip empty fragments, +/// or every byte of every emoji puts a contentless chunk on the wire. +#[derive(Debug, Clone, Default)] +pub struct TokenTextStream { + /// Trailing bytes of an incomplete UTF-8 codepoint, awaiting continuation. + /// Never longer than 3 bytes once `push_bytes` returns. + pending: Vec, +} + +impl TokenTextStream { + pub fn new() -> Self { + Self { + pending: Vec::with_capacity(4), + } + } + + /// Feed one token; return the text that became complete because of it. + /// Returns `""` when the token only extended a partial codepoint. + pub fn push(&mut self, tokenizer: &Tokenizer, token: u32) -> String { + self.push_bytes(&tokenizer.decode_bytes(&[token])) + } + + /// Feed several tokens at once (accepted spec-decode block, batch step). + /// Equivalent to `push` per token, concatenated. + pub fn push_tokens(&mut self, tokenizer: &Tokenizer, tokens: &[u32]) -> String { + self.push_bytes(&tokenizer.decode_bytes(tokens)) + } + + /// Feed raw decoded bytes. The byte-level entry point; `push` / + /// `push_tokens` are thin wrappers over it. + pub fn push_bytes(&mut self, bytes: &[u8]) -> String { + if bytes.is_empty() { + return String::new(); + } + self.pending.extend_from_slice(bytes); + let mut out = String::new(); + loop { + match std::str::from_utf8(&self.pending) { + Ok(s) => { + out.push_str(s); + self.pending.clear(); + return out; + } + Err(e) => { + let valid = e.valid_up_to(); + if valid > 0 { + // `valid_up_to` guarantees this prefix is well-formed. + out.push_str( + std::str::from_utf8(&self.pending[..valid]).unwrap_or_default(), + ); + } + match e.error_len() { + // Incomplete trailing sequence — hold it back and wait + // for the next token to supply the continuation bytes. + None => { + self.pending.drain(..valid); + return out; + } + // Genuinely invalid bytes: they can never become valid, + // and holding them stalls the buffer and mutes the rest + // of the turn — worse than the mojibake being fixed. + // Emit one replacement char and step past. + Some(bad) => { + out.push(char::REPLACEMENT_CHARACTER); + self.pending.drain(..valid + bad); + } + } + } + } + } + } + + /// Drain any held-back bytes at end of stream. Only ever non-empty when + /// generation stopped mid-character; the truncated bytes decode lossily + /// rather than vanishing. Call once, after the decode loop. + pub fn flush(&mut self) -> String { + if self.pending.is_empty() { + return String::new(); + } + let tail = String::from_utf8_lossy(&self.pending).into_owned(); + self.pending.clear(); + tail + } + + /// True when nothing is held back (no partial codepoint in flight). + pub fn is_empty(&self) -> bool { + self.pending.is_empty() + } + + /// Number of held-back bytes. Always `0..=3` outside of `push_bytes`. + pub fn pending_len(&self) -> usize { + self.pending.len() + } +} + #[cfg(test)] mod bpe_tests { use super::*; @@ -2454,7 +2580,6 @@ mod prompt_norm_tests { } } - #[cfg(test)] mod sp_dummy_prefix_tests { //! Config-driven SP dummy-prefix coverage (gemma4 first-word bug, @@ -2528,8 +2653,7 @@ mod sp_dummy_prefix_tests { #[test] fn gemma4_no_dummy_prefix_first_word_matches_hf() { - let t = Tokenizer::from_hfq_metadata(&gemma4_fixture_metadata()) - .expect("fixture parses"); + let t = Tokenizer::from_hfq_metadata(&gemma4_fixture_metadata()).expect("fixture parses"); assert_eq!(t.bos_id, 2, "generation_config bos override"); let mut ids = vec![t.bos_id]; ids.extend(t.encode("The capital of France is")); @@ -2538,8 +2662,7 @@ mod sp_dummy_prefix_tests { #[test] fn gemma4_chat_tail_thought_channel_matches_hf() { - let t = Tokenizer::from_hfq_metadata(&gemma4_fixture_metadata()) - .expect("fixture parses"); + let t = Tokenizer::from_hfq_metadata(&gemma4_fixture_metadata()).expect("fixture parses"); assert_eq!( t.encode("<|channel>thought\nThe capital of France is"), vec![100, 45518, 107, 101, 818, 5279, 529, 7001, 563], @@ -2731,3 +2854,394 @@ mod lfm2_bos_tests { ); } } + +#[cfg(test)] +mod token_text_stream_tests { + use super::*; + + /// Synthetic SentencePiece vocab with full byte-fallback coverage: ids + /// `0..=255` are the `<0xHH>` fallback tokens, followed by a few + /// whole-character literals. No model file needed — this is the vocab + /// shape that produces the bug (one character spread over several + /// tokens), reproduced in-test so the invariant runs in CI. + fn synth_byte_fallback() -> Tokenizer { + let mut vocab: Vec = (0u32..=255).map(|b| format!("<0x{b:02X}>")).collect(); + // Whole-character / multi-char literals, so a sequence mixes + // fallback runs with tokens that are already complete UTF-8. + for lit in LITERALS { + vocab.push((*lit).to_string()); + } + let token_to_id: HashMap = vocab + .iter() + .enumerate() + .map(|(i, s)| (s.clone(), i as u32)) + .collect(); + Tokenizer { + vocab, + token_to_id, + merges: Vec::new(), + merge_pair_rank: HashMap::new(), + byte_to_id: None, + special_tokens: Vec::new(), + bos_id: 0, + eos_id: 0, + add_bos: false, + eot_id: None, + is_gpt2_bpe: false, + sp_dummy_prefix: false, + } + } + + const LITERALS: &[&str] = &["hello", "中", "🎉", " world", "∑", "!"]; + + /// Token id of the `<0xHH>` fallback token for byte `b`. + fn byte_tok(b: u8) -> u32 { + b as u32 + } + + /// Encode a string as a pure byte-fallback token run. + fn as_byte_tokens(s: &str) -> Vec { + s.as_bytes().iter().map(|&b| byte_tok(b)).collect() + } + + /// Feed a whole sequence through the stream and return the concatenation + /// of every fragment plus the flushed tail. + fn stream_all(tk: &Tokenizer, seq: &[u32]) -> String { + let mut stream = TokenTextStream::new(); + let mut out = String::new(); + for &t in seq { + out.push_str(&stream.push(tk, t)); + assert!( + stream.pending_len() <= 3, + "holdback buffer must never exceed 3 bytes, got {}", + stream.pending_len() + ); + } + out.push_str(&stream.flush()); + assert!(stream.is_empty(), "flush must drain the buffer"); + out + } + + /// Deterministic xorshift64* — keeps the randomised tests reproducible + /// and dependency-free. + struct Rng(u64); + + impl Rng { + fn next(&mut self) -> u64 { + let mut x = self.0; + x ^= x >> 12; + x ^= x << 25; + x ^= x >> 27; + self.0 = x; + x.wrapping_mul(0x2545_F491_4F6C_DD1D) + } + + fn below(&mut self, n: usize) -> usize { + (self.next() % n as u64) as usize + } + } + + /// The bug this helper exists to fix, pinned as a test: per-token + /// `decode()` mojibakes anything whose UTF-8 spans several tokens, and + /// the stream does not. + #[test] + fn per_token_decode_is_lossy_but_stream_is_not() { + let tk = synth_byte_fallback(); + let text = "emoji: 🎉🔥🚀"; + let seq = as_byte_tokens(text); + + let per_token: String = seq.iter().map(|&t| tk.decode(&[t])).collect(); + assert!( + per_token.contains('\u{FFFD}'), + "baseline: per-token decode must reproduce the reported corruption" + ); + assert_ne!(per_token, text); + + assert_eq!(stream_all(&tk, &seq), text); + assert_eq!(stream_all(&tk, &seq), tk.decode(&seq)); + } + + /// The core invariant: streaming decode is exactly equivalent to + /// whole-sequence decode, for any token sequence. + #[test] + fn streaming_equals_batch_decode_over_random_sequences() { + let tk = synth_byte_fallback(); + let vocab_len = 256 + LITERALS.len(); + let mut rng = Rng(0x5EED_1234_ABCD_0001); + for _ in 0..2000 { + let len = 1 + rng.below(24); + let seq: Vec = (0..len).map(|_| rng.below(vocab_len) as u32).collect(); + assert_eq!( + stream_all(&tk, &seq), + tk.decode(&seq), + "streaming != batch for {seq:?}" + ); + } + } + + /// Same invariant, but over sequences built from real text rather than + /// uniform-random ids — guarantees the well-formed case dominates. + #[test] + fn streaming_equals_batch_decode_over_text_sequences() { + let tk = synth_byte_fallback(); + for text in [ + "emoji: 🎉🔥🚀", + "math ∑∫≈", + "中文测试", + "café naïve", + "plain ascii", + "", + "🎉", + ] { + let seq = as_byte_tokens(text); + assert_eq!(stream_all(&tk, &seq), text, "text {text:?}"); + assert_eq!(stream_all(&tk, &seq), tk.decode(&seq)); + } + } + + /// Byte-split fuzz: split a UTF-8 string's bytes at every possible + /// combination of boundaries, feed the pieces, assert reassembly. + #[test] + fn byte_split_at_every_boundary_reassembles() { + let text = "a🎉b∑"; + let bytes = text.as_bytes(); + let n = bytes.len(); + assert!(n <= 12, "keep the 2^(n-1) enumeration cheap"); + for mask in 0u32..(1 << (n - 1)) { + let mut stream = TokenTextStream::new(); + let mut out = String::new(); + let mut start = 0usize; + for i in 0..n { + let cut = i == n - 1 || (mask >> i) & 1 == 1; + if cut { + out.push_str(&stream.push_bytes(&bytes[start..=i])); + start = i + 1; + } + } + out.push_str(&stream.flush()); + assert_eq!(out, text, "mask {mask:#b} lost bytes"); + } + } + + /// A split mid-character yields `""` — the holdback contract emit sites + /// depend on. + #[test] + fn holdback_returns_empty_mid_character() { + let mut stream = TokenTextStream::new(); + // U+1F389 🎉 = F0 9F 8E 89. + assert_eq!(stream.push_bytes(&[0xF0]), ""); + assert_eq!(stream.push_bytes(&[0x9F]), ""); + assert_eq!(stream.push_bytes(&[0x8E]), ""); + assert_eq!(stream.push_bytes(&[0x89]), "🎉"); + assert!(stream.is_empty()); + } + + /// Requirement 1: generation stopping mid-character must not silently + /// drop the trailing bytes. + #[test] + fn flush_recovers_truncated_tail() { + let tk = synth_byte_fallback(); + let mut stream = TokenTextStream::new(); + // "a" then a truncated 🎉 (max tokens hit between fallback tokens). + assert_eq!(stream.push(&tk, byte_tok(b'a')), "a"); + assert_eq!(stream.push(&tk, byte_tok(0xF0)), ""); + assert_eq!(stream.push(&tk, byte_tok(0x9F)), ""); + let tail = stream.flush(); + assert!(!tail.is_empty(), "truncated bytes must not vanish silently"); + assert_eq!(tail, String::from_utf8_lossy(&[0xF0, 0x9Fu8])); + assert!(stream.is_empty()); + } + + /// Requirement 2, the subtlest part of the fix: genuinely invalid bytes + /// can never become valid, so the stream must emit a replacement char + /// and step past instead of holding them. A stall here would mute the + /// rest of the turn — worse than the mojibake being fixed. + #[test] + fn invalid_bytes_never_stall_the_stream() { + let mut rng = Rng(0xD1CE_0000_0BAD_F00D); + for _ in 0..3000 { + let mut stream = TokenTextStream::new(); + let len = 1 + rng.below(16); + let junk: Vec = (0..len).map(|_| (rng.next() & 0xFF) as u8).collect(); + for (i, &b) in junk.iter().enumerate() { + stream.push_bytes(&[b]); + assert!( + stream.pending_len() <= 3, + "stream stalled at byte {i} of {junk:?} with {} pending", + stream.pending_len() + ); + } + // Any held-back bytes must drain within 3 further pushes of a + // byte that cannot continue a multi-byte sequence. + for _ in 0..3 { + if stream.is_empty() { + break; + } + stream.push_bytes(b"a"); + } + assert!( + stream.is_empty(), + "stream failed to drain after junk {junk:?}" + ); + } + } + + /// A lone continuation byte (0x80) is invalid on its own, not + /// incomplete — it must not be held back at all. + #[test] + fn lone_continuation_byte_emits_immediately() { + let mut stream = TokenTextStream::new(); + assert_eq!(stream.push_bytes(&[0x80]), "\u{FFFD}"); + assert!(stream.is_empty()); + assert_eq!(stream.push_bytes(&[0xFF, 0xFE]), "\u{FFFD}\u{FFFD}"); + assert!(stream.is_empty()); + } + + /// `push_tokens` (accepted spec-decode block) is equivalent to pushing + /// each token individually. + #[test] + fn push_tokens_matches_per_token_pushes() { + let tk = synth_byte_fallback(); + let seq = as_byte_tokens("🎉 中 ∑"); + let mut a = TokenTextStream::new(); + let batched = { + let mut s = a.push_tokens(&tk, &seq); + s.push_str(&a.flush()); + s + }; + assert_eq!(batched, stream_all(&tk, &seq)); + assert_eq!(batched, tk.decode(&seq)); + } +} + +/// Full-vocab sweep against a real model's tokenizer. +/// +/// This is what catches vocab-specific surprises — precisely the kind that let +/// CJK pass while emoji failed on the qwen3.5-9b vocab (whole-character CJK +/// tokens, byte-fallback emoji). Ignored by default and gated on a model file, +/// so CI stays model-free: +/// +/// ```text +/// HIPFIRE_TEST_MODEL=/path/to/model.hfq \ +/// cargo test -p hipfire-runtime --lib --release full_vocab -- --ignored --nocapture +/// ``` +#[cfg(test)] +mod token_text_stream_vocab_sweep { + use super::*; + + fn load_tokenizer() -> Option { + let path = std::env::var("HIPFIRE_TEST_MODEL").ok()?; + let path = std::path::Path::new(&path); + let json = if path.extension().and_then(|e| e.to_str()) == Some("json") { + std::fs::read_to_string(path).ok()? + } else { + let hfq = crate::hfq::HfqFile::open(path) + .unwrap_or_else(|e| panic!("HIPFIRE_TEST_MODEL {path:?} did not open: {e}")); + use crate::model_source::ModelSource; + hfq.metadata_json().to_string() + }; + let tk = if path.extension().and_then(|e| e.to_str()) == Some("json") { + Tokenizer::from_hf_json(&json) + } else { + Tokenizer::from_hfq_metadata(&json) + }; + Some(tk.unwrap_or_else(|e| panic!("tokenizer from {path:?} failed: {e:?}"))) + } + + /// Feed a sequence through the stream, returning text + flushed tail. + fn streamed(tk: &Tokenizer, seq: &[u32]) -> String { + let mut s = TokenTextStream::new(); + let mut out = s.push_tokens(tk, seq); + out.push_str(&s.flush()); + out + } + + /// Every single token id: streaming decode must equal batch decode. + #[test] + #[ignore = "requires HIPFIRE_TEST_MODEL"] + fn full_vocab_singles_match_batch_decode() { + let Some(tk) = load_tokenizer() else { + eprintln!("HIPFIRE_TEST_MODEL unset — skipping"); + return; + }; + let n = tk.vocab.len(); + let mut fffd_only_in_stream = 0usize; + for id in 0..n as u32 { + let batch = tk.decode(&[id]); + let stream = streamed(&tk, &[id]); + assert_eq!(stream, batch, "id {id} diverged"); + if stream.contains('\u{FFFD}') && !batch.contains('\u{FFFD}') { + fffd_only_in_stream += 1; + } + } + assert_eq!(fffd_only_in_stream, 0); + eprintln!("swept {n} single ids"); + } + + /// Adjacent pairs. Only a token whose bytes end mid-codepoint can start a + /// split, so the exhaustive cross is `openers × whole vocab` rather than + /// `vocab²` — the same coverage for the failure mode, without the + /// quadratic blow-up. The opener count is printed so a truncated sweep is + /// never mistaken for a clean one. + #[test] + #[ignore = "requires HIPFIRE_TEST_MODEL"] + fn full_vocab_adjacent_pairs_match_batch_decode() { + let Some(tk) = load_tokenizer() else { + eprintln!("HIPFIRE_TEST_MODEL unset — skipping"); + return; + }; + let n = tk.vocab.len(); + let openers: Vec = (0..n as u32) + .filter(|&id| { + let bytes = tk.decode_bytes(&[id]); + !bytes.is_empty() && std::str::from_utf8(&bytes).is_err() + }) + .collect(); + eprintln!( + "vocab {n}, {} openers (tokens ending mid-codepoint) → {} pairs", + openers.len(), + openers.len() * n + ); + assert!( + !openers.is_empty(), + "no byte-fallback openers in this vocab — the pair sweep would be vacuous; \ + use a vocab with byte-level fallback (emoji coverage) instead" + ); + for &a in &openers { + for b in 0..n as u32 { + let seq = [a, b]; + let batch = tk.decode(&seq); + let stream = streamed(&tk, &seq); + assert_eq!(stream, batch, "pair ({a}, {b}) diverged"); + } + } + } + + /// End-to-end on real text: encode, stream-decode per token, compare. + /// Always includes emoji — a CJK-only reproduction passes on a fully + /// broken path when the vocab has whole-character CJK tokens. + #[test] + #[ignore = "requires HIPFIRE_TEST_MODEL"] + fn real_text_round_trips_through_the_stream() { + let Some(tk) = load_tokenizer() else { + eprintln!("HIPFIRE_TEST_MODEL unset — skipping"); + return; + }; + for text in [ + "emoji: 🎉🔥🚀", + "math ∑∫≈", + "中文测试 with English", + "café naïve — em dash", + "mixed 🎉 中 ∑ ascii", + ] { + let ids = tk.encode(text); + let batch = tk.decode(&ids); + let stream = streamed(&tk, &ids); + assert_eq!(stream, batch, "streaming diverged for {text:?}"); + assert!( + !stream.contains('\u{FFFD}') || batch.contains('\u{FFFD}'), + "streaming introduced U+FFFD for {text:?}" + ); + } + } +}