From bae0e5c611b8800aba43f129d63fcdee0181b289 Mon Sep 17 00:00:00 2001 From: zaneni6 Date: Tue, 20 Oct 2026 13:33:08 -0700 Subject: [PATCH] fix(server): stop truncated UTF-8 from hanging streamed responses A generation cut off by the length limit or a cancel can stop partway through a multi-byte character. finalize() sent those bytes, json::dump() threw on the invalid UTF-8, and the error went out as a plain response after the stream had started, so the stream was never terminated and the client waited until its own timeout. This is what made qwen3-tk:4b time out after 600 s in the model sweep. - Streaming (OpenAI chat/completions, Ollama chat/generate): drop an incomplete trailing UTF-8 sequence at finalize. - OpenAI streaming handlers: errors after the first chunk now go out as an error event plus [DONE], ending the stream; finalize failures are logged. - Non-streaming chat (OpenAI and Ollama): trim the incomplete tail from the generated text before building the JSON response, instead of failing the request. incomplete_utf8_tail moves from the Whisper transcript header to wstream_buf.hpp so both share it. Model sweep: - Raise the default max_completion_tokens from 512 to 1024; thinking models were spending the whole budget on reasoning. - Give qwen3.8-mtp:27b a 1200 s request timeout via a per-model override table, since it decodes at ~1.7 tok/s. Co-Authored-By: Claude --- .github/script/sweep_common.py | 34 +++++++++--- .github/workflows/model-sweep.yml | 4 +- src/include/whisper/transcript_format.hpp | 14 +---- src/include/wstream_buf.hpp | 17 ++++++ src/server/rest_handler.cpp | 64 +++++++++++++++-------- src/server/streaming_ostream.hpp | 27 +++++----- src/server/streaming_ostream_openai.hpp | 21 +++++--- 7 files changed, 115 insertions(+), 66 deletions(-) diff --git a/.github/script/sweep_common.py b/.github/script/sweep_common.py index 8ff91b083..b25725ff1 100644 --- a/.github/script/sweep_common.py +++ b/.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,16 +701,20 @@ 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: @@ -706,7 +722,7 @@ def collect_stream(self, response) -> tuple[str, str]: 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,6 +746,7 @@ 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, @@ -737,9 +754,10 @@ def chat(self, model: str, messages: list, stream: bool = True) -> tuple[str, st # 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 "" diff --git a/.github/workflows/model-sweep.yml b/.github/workflows/model-sweep.yml index 8f10cae98..0c5b01e86 100644 --- a/.github/workflows/model-sweep.yml +++ b/.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' }} diff --git a/src/include/whisper/transcript_format.hpp b/src/include/whisper/transcript_format.hpp index c0d5214f2..592619038 100644 --- a/src/include/whisper/transcript_format.hpp +++ b/src/include/whisper/transcript_format.hpp @@ -9,6 +9,7 @@ #include #include #include +#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 diff --git a/src/include/wstream_buf.hpp b/src/include/wstream_buf.hpp index e6e2ca126..616bd372c 100644 --- a/src/include/wstream_buf.hpp +++ b/src/include/wstream_buf.hpp @@ -7,8 +7,25 @@ #pragma once #include +#include #include +/// 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 { diff --git a/src/server/rest_handler.cpp b/src/server/rest_handler.cpp index b37acca08..ab0235160 100644 --- a/src/server/rest_handler.cpp +++ b/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(); } diff --git a/src/server/streaming_ostream.hpp b/src/server/streaming_ostream.hpp index eca3fe56f..edbb0146d 100644 --- a/src/server/streaming_ostream.hpp +++ b/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& 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 diff --git a/src/server/streaming_ostream_openai.hpp b/src/server/streaming_ostream_openai.hpp index 62e0a56cd..dcff26440 100644 --- a/src/server/streaming_ostream_openai.hpp +++ b/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.