Fixed character encoding for arch-specific generate loops.

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<u32>,
+    /// 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<ClientEvent>) {
+        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<Self>) -> 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 {

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<Deepseek4SpecGrammar>,
+    /// 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<ClientEvent> {
         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) {

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<ThinkRouteEvent> = 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: <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<ThinkRouteEvent>,
                             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!(