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:?}" + ); + } + } +}