fix(server): stop truncated UTF-8 from hanging streamed responses

.github/script/sweep_common.py

@@ -54,6 +54,17 @@
 
 IS_WINDOWS = os.name == "nt"
 
+# Per-request timeouts, in seconds, for models too slow for --request-timeout.
+# A model listed here gets whichever of the two is longer, so raising
+# --request-timeout still raises it for these models too.
+#
+# qwen3.8-mtp:27b decodes at ~1.7 tok/s on the NPU, so a reply that uses the
+# whole default 1024-token budget takes ~600 s, plus ~50 s to load the model
+# on the first round.
+REQUEST_TIMEOUT_OVERRIDES = {
+    "qwen3.8-mtp:27b": 1200.0,
+}
+
 
 def usage_to_dict(usage) -> dict:
     """Flattens an SDK usage object into a plain dict.
@@ -154,7 +165,7 @@ def build_parser(description: str) -> argparse.ArgumentParser:
     parser.add_argument(
         "--gen-lim",
         type=int,
-        default=512,
+        default=1024,
         help="max_completion_tokens per request. 0 or less leaves it unset.",
     )
     parser.add_argument(
@@ -163,7 +174,8 @@ def build_parser(description: str) -> argparse.ArgumentParser:
         default=600.0,
         help=(
             "Per-request timeout in seconds (default: 600). Applied both as the "
-            "HTTP timeout and as a wall-clock deadline on streamed responses."
+            "HTTP timeout and as a wall-clock deadline on streamed responses. "
+            "Models in REQUEST_TIMEOUT_OVERRIDES get a longer one."
         ),
     )
     parser.add_argument(
@@ -689,24 +701,28 @@ def completion_kwargs(self) -> dict:
             return {"max_completion_tokens": self.args.gen_lim}
         return {}
 
-    def collect_stream(self, response) -> tuple[str, str]:
+    def request_timeout(self, model: str) -> float:
+        """Per-request timeout for one model, see REQUEST_TIMEOUT_OVERRIDES."""
+        return max(self.args.request_timeout, REQUEST_TIMEOUT_OVERRIDES.get(model, 0.0))
+
+    def collect_stream(self, response, timeout: float) -> tuple[str, str]:
         """Accumulates streamed chunks into (reasoning_content, output_content).
 
         The HTTP timeout only bounds the wait for the next chunk, so a model
-        that keeps emitting tokens can stream well past it. This enforces
-        --request-timeout as a deadline on the whole response.
+        that keeps emitting tokens can stream well past it. This enforces the
+        request timeout as a deadline on the whole response.
         """
         reasoning_content, output_content = "", ""
         usage = None
-        deadline = time.monotonic() + self.args.request_timeout
+        deadline = time.monotonic() + timeout
         for chunk in response:
             if time.monotonic() > deadline:
                 try:
                     response.close()
                 except Exception:
                     pass
                 raise TimeoutError(
-                    f"stream exceeded the {self.args.request_timeout:.0f}s "
+                    f"stream exceeded the {timeout:.0f}s "
                     f"request timeout after {len(output_content)} characters"
                 )
             # With include_usage the usage arrives on its own terminal chunk
@@ -730,16 +746,18 @@ def chat(self, model: str, messages: list, stream: bool = True) -> tuple[str, st
         only the reporting cares about.
         """
         self.last_usage = {}
+        timeout = self.request_timeout(model)
         response = self.client.chat.completions.create(
             model=model,
             messages=messages,
             stream=stream,
             # a stream reports usage only when asked, as OpenAI's does
             **({"stream_options": {"include_usage": True}} if stream else {}),
             **self.completion_kwargs(),
+            timeout=timeout,
         )
         if stream:
-            reasoning, output, usage = self.collect_stream(response)
+            reasoning, output, usage = self.collect_stream(response, timeout)
         else:
             message = response.choices[0].message
             reasoning = getattr(message, "reasoning_content", None) or ""

.github/workflows/model-sweep.yml

@@ -42,7 +42,7 @@ on:
       gen_lim:
         description: "max_completion_tokens per request (0 or less leaves it unset)"
         type: string
-        default: "512"
+        default: "1024"
       request_timeout:
         description: "Per-request timeout in seconds"
         type: string
@@ -66,7 +66,7 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  GEN_LIM: ${{ inputs.gen_lim || '512' }}
+  GEN_LIM: ${{ inputs.gen_lim || '1024' }}
   REQUEST_TIMEOUT: ${{ inputs.request_timeout || '600' }}
   # 0 unless dispatched, so the cap only exists when someone asks for one.
   MODEL_LIMIT: ${{ inputs.limit || '0' }}

src/include/whisper/transcript_format.hpp

@@ -9,6 +9,7 @@
 #include <cstdlib>
 #include <string>
 #include <vector>
+#include "wstream_buf.hpp"
 
 namespace transcript {
 
@@ -36,19 +37,6 @@ inline size_t find_time_stamp(const std::string& output, size_t pos, size_t& len
     return std::string::npos;
 }
 
-/// The number of bytes at the end of text that begin a UTF-8 character it does not finish.
-inline size_t incomplete_utf8_tail(const std::string& text) {
-    for (size_t back = 1; back <= 4 && back <= text.size(); back++) {
-        unsigned char c = text[text.size() - back];
-        if ((c & 0xC0) == 0x80) {
-            continue;  // a continuation byte
-        }
-        size_t width = c >= 0xF0 ? 4 : c >= 0xE0 ? 3 : c >= 0xC0 ? 2 : 1;
-        return width > back ? back : 0;
-    }
-    return 0;
-}
-
 /// Splits Whisper::generate's output with return_time_stamp, as in
 /// "<|0.00|> Hello.<|1.50|><|1.50|> Bye.<|3.00|>", into segments. Text starts
 /// at the timestamp before it and ends at the one after it; text with no

src/include/wstream_buf.hpp

@@ -7,8 +7,25 @@
 #pragma once
 
 #include <iostream>
+#include <string>
 #include <vector>
 
+/// The number of bytes at the end of text that begin a UTF-8 character it does not finish.
+/// \note Tokens are bytes, so a generation cut off by the length limit or a cancel
+///       can stop partway through a character. Those bytes are invalid UTF-8, and
+///       json::dump() throws on them.
+inline size_t incomplete_utf8_tail(const std::string& text) {
+    for (size_t back = 1; back <= 4 && back <= text.size(); back++) {
+        unsigned char c = text[text.size() - back];
+        if ((c & 0xC0) == 0x80) {
+            continue;  // a continuation byte
+        }
+        size_t width = c >= 0xF0 ? 4 : c >= 0xE0 ? 3 : c >= 0xC0 ? 2 : 1;
+        return width > back ? back : 0;
+    }
+    return 0;
+}
+
 /// \brief wstream_buf class
 /// \note This class is used to avoid partial UTF-8 characters from the decoder.
 struct wstream_buf : std::streambuf {

src/server/rest_handler.cpp

@@ -912,6 +912,8 @@ void RestHandler::handle_chat(const json& request,
                 this->auto_chat_engine->clear_context();
                 return;
             }
+            // A character the generation stopped partway through can never be finished
+            response_text.resize(response_text.size() - incomplete_utf8_tail(response_text));
             //std::string response_text = chat_engine->generate_with_prompt(meta_info, prompts, length_limit, std::cout, payload);
             auto total_end_time = time_utils::now();
             meta_info.total_duration = (uint64_t)time_utils::duration_ns(total_start_time, total_end_time).first;
@@ -1289,6 +1291,17 @@ void RestHandler::handle_openai_chat_completion(const json& request,
                 };
             streaming_ostream_openai_chat ostream(model, auto_chat_engine.get(), openai_stream_callback, parse_options,
                                                   stream_include_usage(request));  // streaming in chat completion format
+            // Once a chunk is out the headers are sent, so an error can only go
+            // out as one more event, and the stream still has to be ended or the
+            // client waits on it until its own timeout
+            auto send_stream_error = [&](const json& error_response) {
+                if (stream_started) {
+                    send_streaming_response(json("data: " + error_response.dump() + "\n\n"), false);
+                    send_streaming_response(json("data: [DONE]\n\n"), true);
+                } else {
+                    send_response(error_response);
+                }
+                };
 
             header_print("FLM", "Start prefill...");
             try {
@@ -1327,18 +1340,11 @@ void RestHandler::handle_openai_chat_completion(const json& request,
             try {
                 auto_chat_engine->generate(meta_info, length_limit, ostream, [&] { return cancellation_token->cancelled(); });
             } catch (const ModelRequestError& error) {
-                const json error_response = ModelErrorResponse(error);
-                if (stream_started) {
-                    send_streaming_response(json("data: " + error_response.dump() + "\n\n"), false);
-                    send_streaming_response(json("data: [DONE]\n\n"), true);
-                } else {
-                    send_response(error_response);
-                }
+                send_stream_error(ModelErrorResponse(error));
                 if (error.session_cleared()) this->prompt_cache.reset();
                 return;
             } catch (const std::exception& e) {
-                json error_response = ExceptionResponse(e);
-                send_response(error_response);
+                send_stream_error(ExceptionResponse(e));
                 this->auto_chat_engine->clear_context();
                 this->prompt_cache.reset();
                 return;
@@ -1347,8 +1353,13 @@ void RestHandler::handle_openai_chat_completion(const json& request,
                 header_print("❌ ", "Generation Cancelled!");
                 this->prompt_cache.reset();
             }
-
-            ostream.finalize(meta_info);
+
+            try {
+                ostream.finalize(meta_info);
+            } catch (const std::exception& e) {
+                header_print("ERROR", std::string("Finishing the stream failed: ") + e.what());
+                send_stream_error(ExceptionResponse(e));
+            }
         }
         else {
             nullstream nstream;
@@ -1395,6 +1406,8 @@ void RestHandler::handle_openai_chat_completion(const json& request,
                 this->prompt_cache.reset();
                 return;
             }
+            // A character the generation stopped partway through can never be finished
+            response_text.resize(response_text.size() - incomplete_utf8_tail(response_text));
             // check response_text
             json choices = build_nstream_response(response_text, meta_info, parse_options);
             response = {
@@ -1617,6 +1630,17 @@ void RestHandler::handle_openai_completion(const json& request,
                 send_streaming_response(data_json, is_final);
                 };
             streaming_ostream_openai ostream(model, openai_stream_callback, stream_include_usage(request));  // streaming in completion format
+            // Once a chunk is out the headers are sent, so an error can only go
+            // out as one more event, and the stream still has to be ended or the
+            // client waits on it until its own timeout
+            auto send_stream_error = [&](const json& error_response) {
+                if (stream_started) {
+                    send_streaming_response(json("data: " + error_response.dump() + "\n\n"), false);
+                    send_streaming_response(json("data: [DONE]\n\n"), true);
+                } else {
+                    send_response(error_response);
+                }
+                };
             uniformed_input.prompt = prompt;
             try {
                 bool success = auto_chat_engine->insert(meta_info, uniformed_input, [&] { return cancellation_token && cancellation_token->cancelled(); });
@@ -1635,21 +1659,19 @@ void RestHandler::handle_openai_completion(const json& request,
             try {
                 auto_chat_engine->generate(meta_info, length_limit, ostream, [&] { return cancellation_token && cancellation_token->cancelled(); });
             } catch (const ModelRequestError& error) {
-                const json error_response = ModelErrorResponse(error);
-                if (stream_started) {
-                    send_streaming_response(json("data: " + error_response.dump() + "\n\n"), false);
-                    send_streaming_response(json("data: [DONE]\n\n"), true);
-                } else {
-                    send_response(error_response);
-                }
+                send_stream_error(ModelErrorResponse(error));
                 return;
             } catch (const std::exception& e) {
-                json error_response = ExceptionResponse(e);
-                send_response(error_response);
+                send_stream_error(ExceptionResponse(e));
                 this->auto_chat_engine->clear_context();
                 return;
             }
-            ostream.finalize(meta_info);
+            try {
+                ostream.finalize(meta_info);
+            } catch (const std::exception& e) {
+                header_print("ERROR", std::string("Finishing the stream failed: ") + e.what());
+                send_stream_error(ExceptionResponse(e));
+            }
 
             this->auto_chat_engine->clear_context();
         }

src/server/streaming_ostream.hpp

@@ -56,27 +56,26 @@ class streaming_buf : public std::streambuf {
 public:
     ///@brief Call this when generation is complete
     void finalize_chat(chat_meta_info_t& meta_info) {
-        // Send all remaining content, including incomplete sequences
-        if (!buffer.empty()) {
-            send_response(buffer, true);
-            buffer.clear();
-        } else {
-            send_chat_final_response(meta_info);
-        }
+        drop_incomplete_utf8_tail();
+        send_chat_final_response(meta_info);
     }
     ///@brief Call this when generation is complete
     ///@param context the context
     void finalize_generate(chat_meta_info_t& meta_info, std::vector<int>& context) {
-        // Send all remaining content, including incomplete sequences
-        if (!buffer.empty()) {
-            send_response(buffer, true);
-            buffer.clear();
-        } else {
-            send_generate_final_response(meta_info, context);
-        }
+        drop_incomplete_utf8_tail();
+        send_generate_final_response(meta_info, context);
     }
 
 private:
+    ///@brief Send the remaining content but not an incomplete UTF-8 sequence at its end
+    ///@note A generation cut off by the length limit or a cancel can stop partway
+    ///      through a multi-byte character, the missing bytes will never come, and
+    ///      invalid UTF-8 makes json::dump() throw
+    void drop_incomplete_utf8_tail() {
+        flush_complete_utf8_sequences(false);
+        buffer.clear();
+    }
+
     ///@brief Get UTF-8 sequence length from first byte
     ///@param first_byte the first byte
     ///@return sequence length, or 0 if invalid

src/server/streaming_ostream_openai.hpp

@@ -69,11 +69,12 @@ class streaming_buf_openai : public std::streambuf {
 public:
     ///@brief Call this when generation is complete
     void finalize(chat_meta_info_t& meta_info) {
-        // Send all remaining content, including incomplete sequences
-        if (!buffer.empty()) {
-            send_response(buffer, true);
-            buffer.clear();
-        }
+        // Send the remaining content but not an incomplete UTF-8 sequence at
+        // its end: a generation cut off by the length limit or a cancel can
+        // stop partway through a multi-byte character, the missing bytes will
+        // never come, and invalid UTF-8 makes json::dump() throw
+        flush_complete_utf8_sequences(false);
+        buffer.clear();
         send_final_response(meta_info);
     }
 
@@ -331,10 +332,14 @@ class streaming_buf_openai_chat : public std::streambuf {
 public:
     ///@brief Call this when generation is complete
     void finalize(chat_meta_info_t& meta_info) {
-        // Send all remaining content, including incomplete sequences, and let
-        // the parser release whatever it held back
-        send_response(buffer, true);
+        // Send the remaining content but not an incomplete UTF-8 sequence at
+        // its end: a generation cut off by the length limit or a cancel can
+        // stop partway through a multi-byte character, the missing bytes will
+        // never come, and invalid UTF-8 makes json::dump() throw
+        flush_complete_utf8_sequences();
         buffer.clear();
+        // Let the parser release whatever it held back
+        send_response("", true);
         // Only promote to a tool-call finish reason on an otherwise clean stop;
         // length/cancel/error must win, so a truncated or aborted generation is
         // never reported to the client as a complete tool call.