From c2796050809ee3937bad21a5ef794658ca9f3791 Mon Sep 17 00:00:00 2001 From: Drew Schuyler Date: Mon, 3 Aug 2026 11:45:32 -0500 Subject: [PATCH 1/2] fix(server): preserve UTF-8 across stream boundaries --- server/src/server/finish_reason.h | 33 ++++++ server/src/server/http_server.cpp | 42 +++---- server/src/server/sse_emitter.cpp | 59 ++++++++-- server/src/server/sse_emitter.h | 5 +- server/src/server/utf8_utils.h | 134 +++++++++++++++++++--- server/test/test_server_unit.cpp | 183 +++++++++++++++++++++++++++++- 6 files changed, 407 insertions(+), 49 deletions(-) create mode 100644 server/src/server/finish_reason.h diff --git a/server/src/server/finish_reason.h b/server/src/server/finish_reason.h new file mode 100644 index 000000000..0e10d9ba8 --- /dev/null +++ b/server/src/server/finish_reason.h @@ -0,0 +1,33 @@ +// Shared client-facing finish-reason policy for HTTP responses and logs. +#pragma once + +#include + +namespace dflash::common { + +// Resolve the public reason from the emitter's semantic reason and request +// outcome. `generation_cap < 0` disables the cap comparison for callers that +// do not have a generation budget. Disconnect and backend errors retain their +// existing log-only states and take precedence over normal completion. +inline std::string resolve_client_finish_reason( + const std::string & emitter_reason, + int completion_tokens, + int generation_cap, + bool degenerate_decode_close, + bool result_ok = true, + bool client_disconnected = false) { + if (client_disconnected) return "client_disconnect"; + if (!result_ok) return "error"; + + std::string reason = emitter_reason; + if (reason == "stop" && generation_cap >= 0 && + completion_tokens >= generation_cap) { + reason = "length"; + } + // Preserve semantic terminal reasons such as tool_calls. A degenerate + // decode only upgrades an otherwise normal stop to length. + if (degenerate_decode_close && reason == "stop") reason = "length"; + return reason; +} + +} // namespace dflash::common diff --git a/server/src/server/http_server.cpp b/server/src/server/http_server.cpp index 27f421528..4ccf68b9a 100644 --- a/server/src/server/http_server.cpp +++ b/server/src/server/http_server.cpp @@ -19,6 +19,7 @@ #include "http_server.h" #include "admission.h" #include "sse_emitter.h" +#include "finish_reason.h" #include "prompt_normalize.h" #include "tool_hint.h" #include "common/sha1.h" @@ -2091,18 +2092,11 @@ json build_openai_completion_response( message["tool_calls"] = tool_calls; } - // The emitter only knows "stop" / "tool_calls"; it cannot see that the - // daemon hit the n_gen cap. Derive "length" from the committed-token - // count — OpenAI-compatible clients (open-webui, Cline) gate retry - // logic on finish_reason == "length". - std::string finish_reason = emitter.finish_reason(); - if (finish_reason == "stop" && counts.total >= generation_cap) { - finish_reason = "length"; - } - // Degenerate decode (post-close repetition-loop watchdog) also reports - // "length": OpenAI/Anthropic/Gemini all collapse budget-class events - // into one closed enum, with richer signal in sidecar fields below. - if (result.degenerate_decode_close) finish_reason = "length"; + // Resolve the same client-facing policy used by the streaming terminal + // event and the chat DONE log. + const std::string finish_reason = resolve_client_finish_reason( + emitter.finish_reason(), counts.total, generation_cap, + result.degenerate_decode_close); json choice = { {"index", 0}, @@ -2193,12 +2187,14 @@ json build_anthropic_response( } // stop_reason is Anthropic's analog of finish_reason, with the same - // length-vs-EOS distinction — Cline / Anthropic SDK clients gate - // retry logic on stop_reason == "max_tokens". + // shared length-vs-EOS policy as OpenAI and the streaming terminal event. + const std::string finish_reason = resolve_client_finish_reason( + emitter.finish_reason(), counts.total, generation_cap, + result.degenerate_decode_close); std::string stop_reason; - if (emitter.finish_reason() == "tool_calls") { + if (finish_reason == "tool_calls") { stop_reason = "tool_use"; - } else if (counts.total >= generation_cap) { + } else if (finish_reason == "length") { stop_reason = "max_tokens"; } else { stop_reason = "end_turn"; @@ -3568,7 +3564,9 @@ void HttpServer::process_job(ServerJob * job) { client_disconnected = true; } if (req.stream && !client_disconnected) { - auto final_chunks = emitter.emit_finish(completion_tokens, &gen_timings); + auto final_chunks = emitter.emit_finish( + completion_tokens, &gen_timings, n_gen_cap, + result.degenerate_decode_close); for (const auto & chunk : final_chunks) { if (!send_job_bytes(job, chunk.data(), chunk.size())) { client_disconnected = true; @@ -3600,9 +3598,13 @@ void HttpServer::process_job(ServerJob * job) { const double tok_s = elapsed_s > 0.0 ? out_tokens / elapsed_s : 0.0; const double decode_tok_s = result.decode_s > 0.0 ? out_tokens / result.decode_s : 0.0; - const std::string finish = client_disconnected - ? "client_disconnect" - : (result.ok() ? emitter.finish_reason() : "error"); + // Match the counter used by the client-facing path: streaming reports + // completion_tokens, while non-streaming builders use result.tokens. + const int finish_tokens = req.stream ? completion_tokens : result_tokens; + const std::string finish = resolve_client_finish_reason( + result.ok() ? emitter.finish_reason() : "stop", + finish_tokens, n_gen_cap, result.degenerate_decode_close, + result.ok(), client_disconnected); std::fprintf(stderr, "[server] chat DONE %s ok=%s in=%zu effective_in=%zu out=%d " diff --git a/server/src/server/sse_emitter.cpp b/server/src/server/sse_emitter.cpp index c3d53bdd3..2ed02e0cc 100644 --- a/server/src/server/sse_emitter.cpp +++ b/server/src/server/sse_emitter.cpp @@ -1,6 +1,7 @@ // SSE emitter implementation — streaming state machine for all 3 API formats. #include "sse_emitter.h" +#include "finish_reason.h" #include "utf8_utils.h" #include @@ -24,6 +25,26 @@ static bool has_single_request_tool(const json & tools) { return tools.is_array() && tools.size() == 1 && tools[0].is_object(); } +static std::string sanitize_stream_piece(const std::string & raw_piece, + std::string & pending_bytes) { + if (pending_bytes.empty()) { + bool ascii = true; + for (unsigned char c : raw_piece) { + if (c >= 0x80) { + ascii = false; + break; + } + } + if (ascii) return raw_piece; + } + + pending_bytes += raw_piece; + const size_t safe_len = utf8_stream_safe_len(pending_bytes); + std::string complete = pending_bytes.substr(0, safe_len); + pending_bytes.erase(0, safe_len); + return utf8_sanitize(complete); +} + static bool starts_with_potential_bare_json_tool(const std::string & text, const json & tools) { if (!has_single_request_tool(tools)) return false; @@ -238,8 +259,10 @@ std::vector SseEmitter::emit_token(const std::string & raw_piece) { } emit_token_count_++; - // Sanitize input to prevent json::dump() from throwing on invalid UTF-8. - std::string piece = utf8_sanitize(raw_piece); + // Sanitize only complete UTF-8 sequences. Byte-fallback token pieces can + // split one code point across calls, so retain only a valid incomplete + // trailing prefix and let invalid bytes reach the sanitizer promptly. + std::string piece = sanitize_stream_piece(raw_piece, pending_bytes_); std::vector out; accumulated_raw_ += piece; window_ += piece; @@ -276,6 +299,10 @@ std::vector SseEmitter::emit_token(const std::string & raw_piece) { } } window_.clear(); + // Bytes held past the matched stop sequence are not part of the + // response. In particular, do not turn a trailing incomplete + // UTF-8 prefix into U+FFFD during emit_finish(). + pending_bytes_.clear(); stop_hit_ = true; return out; } @@ -500,9 +527,21 @@ void SseEmitter::emit_content_delta(std::vector & out, // ─── emit_finish ──────────────────────────────────────────────────────── std::vector SseEmitter::emit_finish(int completion_tokens, - const GenTimings * timings) { + const GenTimings * timings, + int generation_cap, + bool degenerate_decode_close) { std::vector out; + // A valid but incomplete trailing sequence is only safe to retain until + // the generation is over. At finalization it is genuinely truncated and + // utf8_sanitize() emits one deterministic replacement character. + if (!pending_bytes_.empty()) { + const std::string tail = utf8_sanitize(pending_bytes_); + pending_bytes_.clear(); + accumulated_raw_ += tail; + window_ += tail; + } + // Flush remaining window if (mode_ == StreamMode::REASONING && !window_.empty()) { reasoning_text_ += window_; @@ -673,6 +712,9 @@ std::vector SseEmitter::emit_finish(int completion_tokens, } } + fr = resolve_client_finish_reason( + fr, completion_tokens, generation_cap, degenerate_decode_close); + // Format-specific final events switch (format_) { case ApiFormat::OPENAI_CHAT: { @@ -711,11 +753,12 @@ std::vector SseEmitter::emit_finish(int completion_tokens, json({{"type", "content_block_stop"}, {"index", block_index_}}).dump())); active_kind_.clear(); } - // stop_reason reflects the model's actual finish: "tool_use" when - // any tool calls were emitted (downstream SDKs pivot on this to feed - // tool_result back), else "end_turn". Stop-sequence hits also report - // "end_turn" (Anthropic has no dedicated reason for that case). - const char * stop_reason = tool_calls_.empty() ? "end_turn" : "tool_use"; + // stop_reason follows the same shared policy as the OpenAI terminal + // event: tool calls stay tool_use, budget closure is max_tokens, and + // ordinary completion is end_turn. + const char * stop_reason = + fr == "tool_calls" ? "tool_use" : + fr == "length" ? "max_tokens" : "end_turn"; json anth_usage = {{"output_tokens", completion_tokens}}; if (timings) { anth_usage["timings"] = build_timings_json(*timings, completion_tokens); diff --git a/server/src/server/sse_emitter.h b/server/src/server/sse_emitter.h index 06b4d892e..26e0e283d 100644 --- a/server/src/server/sse_emitter.h +++ b/server/src/server/sse_emitter.h @@ -87,7 +87,9 @@ class SseEmitter { // the pre-timings API for unit tests that don't exercise that // shape. std::vector emit_finish(int completion_tokens, - const GenTimings * timings = nullptr); + const GenTimings * timings = nullptr, + int generation_cap = -1, + bool degenerate_decode_close = false); // Get the finish_reason for non-streaming responses. std::string finish_reason() const; @@ -153,6 +155,7 @@ class SseEmitter { ToolMemory * tool_memory_; StreamMode mode_; + std::string pending_bytes_; // bounded incomplete UTF-8 tail (max 3 bytes) std::string window_; // holdback buffer std::string tool_buffer_; // accumulated tool text bool tool_buffer_fallback_to_content_ = false; diff --git a/server/src/server/utf8_utils.h b/server/src/server/utf8_utils.h index 909107d51..5c994d2f1 100644 --- a/server/src/server/utf8_utils.h +++ b/server/src/server/utf8_utils.h @@ -20,47 +20,147 @@ inline size_t utf8_safe_len(const std::string & s, size_t pos) { return pos; } +inline size_t utf8_sequence_length(uint8_t c) { + if (c < 0x80) return 1; + if (c >= 0xC2 && c <= 0xDF) return 2; + if ((c & 0xF0) == 0xE0) return 3; + if (c >= 0xF0 && c <= 0xF4) return 4; + return 0; +} + +inline bool utf8_valid_trailing_prefix(const std::string & s, + size_t start, size_t seq_len) { + if (seq_len < 2 || start >= s.size()) return false; + for (size_t i = start + 1; i < s.size(); ++i) { + if (((uint8_t)s[i] & 0xC0) != 0x80) return false; + } + if (start + 1 < s.size()) { + const uint8_t lead = (uint8_t)s[start]; + const uint8_t next = (uint8_t)s[start + 1]; + // Once the first continuation byte is present, reject prefixes that + // can only complete as an overlong, surrogate, or out-of-range value. + if (lead == 0xE0 && next < 0xA0) return false; + if (lead == 0xED && next > 0x9F) return false; + if (lead == 0xF0 && next < 0x90) return false; + if (lead == 0xF4 && next > 0x8F) return false; + } + return s.size() - start < seq_len; +} + +// Return the number of bytes that are safe to sanitize now. Only a valid +// incomplete trailing prefix is held back, and that prefix is at most three +// bytes. Invalid bytes are included in the returned prefix so they are +// replaced promptly instead of being retained indefinitely. +inline size_t utf8_stream_safe_len(const std::string & s) { + size_t i = 0; + while (i < s.size()) { + const uint8_t c = (uint8_t)s[i]; + const size_t seq_len = utf8_sequence_length(c); + if (seq_len == 0) { + ++i; + continue; + } + if (seq_len == 1) { + ++i; + continue; + } + if (i + seq_len > s.size()) { + if (utf8_valid_trailing_prefix(s, i, seq_len)) return i; + ++i; + continue; + } + + bool valid = true; + for (size_t j = 1; j < seq_len; ++j) { + if (((uint8_t)s[i + j] & 0xC0) != 0x80) { + valid = false; + break; + } + } + if (!valid) { + ++i; + continue; + } + + uint32_t cp = 0; + if (seq_len == 2) { + cp = ((uint32_t)(c & 0x1F) << 6) | + ((uint32_t)((uint8_t)s[i + 1]) & 0x3F); + if (cp < 0x80) valid = false; + } else if (seq_len == 3) { + cp = ((uint32_t)(c & 0x0F) << 12) | + ((uint32_t)((uint8_t)s[i + 1] & 0x3F) << 6) | + ((uint32_t)((uint8_t)s[i + 2]) & 0x3F); + if (cp < 0x800 || (cp >= 0xD800 && cp <= 0xDFFF)) valid = false; + } else { + cp = ((uint32_t)(c & 0x07) << 18) | + ((uint32_t)((uint8_t)s[i + 1] & 0x3F) << 12) | + ((uint32_t)((uint8_t)s[i + 2] & 0x3F) << 6) | + ((uint32_t)((uint8_t)s[i + 3]) & 0x3F); + if (cp < 0x10000 || cp > 0x10FFFF) valid = false; + } + i += valid ? seq_len : 1; + } + return s.size(); +} + // Sanitize a string for JSON: replace invalid/incomplete UTF-8 with U+FFFD. inline std::string utf8_sanitize(const std::string & s) { + // ASCII is by far the common path for model output. Avoid allocating or + // walking the validation state machine when no UTF-8 work is required. + bool has_high_bit = false; + for (unsigned char c : s) { + if (c >= 0x80) { + has_high_bit = true; + break; + } + } + if (!has_high_bit) return s; + std::string out; out.reserve(s.size()); size_t i = 0; while (i < s.size()) { - uint8_t c = (uint8_t)s[i]; - size_t seq_len = 0; - if (c < 0x80) seq_len = 1; - else if ((c & 0xE0) == 0xC0) seq_len = 2; - else if ((c & 0xF0) == 0xE0) seq_len = 3; - else if ((c & 0xF8) == 0xF0) seq_len = 4; + const uint8_t c = (uint8_t)s[i]; + const size_t seq_len = utf8_sequence_length(c); if (seq_len == 0 || i + seq_len > s.size()) { - out += "\xEF\xBF\xBD"; // U+FFFD - i++; + if (seq_len > 1 && utf8_valid_trailing_prefix(s, i, seq_len)) { + // A valid prefix that reaches end-of-input is one truncated + // sequence, not one independent invalid byte per continuation. + out += "\xEF\xBF\xBD"; + break; + } + out += "\xEF\xBF\xBD"; + ++i; continue; } + bool valid = true; - for (size_t j = 1; j < seq_len; j++) { - if (((uint8_t)s[i + j] & 0xC0) != 0x80) { valid = false; break; } + for (size_t j = 1; j < seq_len; ++j) { + if (((uint8_t)s[i + j] & 0xC0) != 0x80) { + valid = false; + break; + } } if (valid) { - // Decode codepoint and validate range. uint32_t cp = 0; if (seq_len == 1) { cp = c; } else if (seq_len == 2) { - cp = ((uint32_t)(c & 0x1F) << 6) | ((uint32_t)((uint8_t)s[i+1]) & 0x3F); - if (cp < 0x80) valid = false; // overlong + cp = ((uint32_t)(c & 0x1F) << 6) | + ((uint32_t)((uint8_t)s[i+1]) & 0x3F); + if (cp < 0x80) valid = false; } else if (seq_len == 3) { cp = ((uint32_t)(c & 0x0F) << 12) | ((uint32_t)((uint8_t)s[i+1] & 0x3F) << 6) | ((uint32_t)((uint8_t)s[i+2]) & 0x3F); - if (cp < 0x800) valid = false; // overlong - if (cp >= 0xD800 && cp <= 0xDFFF) valid = false; // surrogate + if (cp < 0x800 || (cp >= 0xD800 && cp <= 0xDFFF)) valid = false; } else { cp = ((uint32_t)(c & 0x07) << 18) | ((uint32_t)((uint8_t)s[i+1] & 0x3F) << 12) | ((uint32_t)((uint8_t)s[i+2] & 0x3F) << 6) | ((uint32_t)((uint8_t)s[i+3]) & 0x3F); - if (cp < 0x10000 || cp > 0x10FFFF) valid = false; // overlong or out-of-range + if (cp < 0x10000 || cp > 0x10FFFF) valid = false; } } if (valid) { @@ -68,7 +168,7 @@ inline std::string utf8_sanitize(const std::string & s) { i += seq_len; } else { out += "\xEF\xBF\xBD"; - i++; // only skip lead byte; next byte may be a valid start + ++i; // only skip lead byte; next byte may be a valid start } } return out; diff --git a/server/test/test_server_unit.cpp b/server/test/test_server_unit.cpp index 75389879d..27b68dbd8 100644 --- a/server/test/test_server_unit.cpp +++ b/server/test/test_server_unit.cpp @@ -17,6 +17,7 @@ #include "server/utf8_utils.h" #include "server/api_types.h" #include "server/http_server.h" +#include "server/finish_reason.h" #include "server/chat_template.h" #include "common/sampler.h" #include "common/backend_precision.h" @@ -333,15 +334,191 @@ TEST_CASE(ServerUnitFixture, test_utf8_sanitize_replaces_invalid) { // Truncated 4-byte sequence std::string s2 = "X\xF0\x9F"; std::string out2 = utf8_sanitize(s2); - // Each invalid byte becomes U+FFFD - TEST_ASSERT(out2.find("X") == 0); - TEST_ASSERT(out2.size() > 1); // has replacement(s) + // A valid but truncated prefix collapses to one U+FFFD. + TEST_ASSERT(out2 == "X\xEF\xBF\xBD"); } TEST_CASE(ServerUnitFixture, test_utf8_sanitize_empty) { TEST_ASSERT(utf8_sanitize("") == ""); } +TEST_CASE(ServerUnitFixture, test_finish_reason_resolution_table) { + struct Case { + const char * emitter; + int tokens; + int cap; + bool degenerate; + bool ok; + bool disconnected; + const char * expected; + }; + const std::vector cases = { + {"stop", 3, 8, false, true, false, "stop"}, + {"stop", 8, 8, false, true, false, "length"}, + {"stop", 9, 8, false, true, false, "length"}, + {"stop", 3, 8, true, true, false, "length"}, + {"tool_calls", 8, 8, false, true, false, "tool_calls"}, + {"tool_calls", 8, 8, true, true, false, "tool_calls"}, + {"stop", 3, 8, false, true, true, "client_disconnect"}, + {"stop", 3, 8, false, false, false, "error"}, + }; + for (const auto & c : cases) { + TEST_ASSERT(resolve_client_finish_reason( + c.emitter, c.tokens, c.cap, c.degenerate, + c.ok, c.disconnected) == c.expected); + } +} + +TEST_CASE(ServerUnitFixture, test_streaming_finish_reason_uses_shared_length_policy) { + auto openai = make_emitter(ApiFormat::OPENAI_CHAT); + openai.emit_start(); + openai.emit_token("x"); + const std::string openai_wire = concat(openai.emit_finish(1, nullptr, 1)); + TEST_ASSERT(openai_wire.find("\"finish_reason\":\"length\"") != + std::string::npos); + + auto anthropic = make_emitter(ApiFormat::ANTHROPIC); + anthropic.emit_start(); + anthropic.emit_token("x"); + const std::string anthropic_wire = concat(anthropic.emit_finish(1, nullptr, 1)); + TEST_ASSERT(anthropic_wire.find("\"stop_reason\":\"max_tokens\"") != + std::string::npos); +} + +TEST_CASE(ServerUnitFixture, test_utf8_sanitize_invalid_sequences_are_deterministic) { + const std::string replacement = "\xEF\xBF\xBD"; + + // Overlong U+002F: C0 AF. Each invalid byte remains visible as a + // replacement; the sanitizer is not allowed to pass malformed UTF-8. + TEST_ASSERT(utf8_sanitize("A\xC0\xAFZ") == "A" + replacement + replacement + "Z"); + + // UTF-8 encoding of a surrogate code point: ED A0 80. + TEST_ASSERT(utf8_sanitize("A\xED\xA0\x80Z") == + "A" + replacement + replacement + replacement + "Z"); + + // A lone continuation byte is invalid immediately. + TEST_ASSERT(utf8_sanitize("A\x80Z") == "A" + replacement + "Z"); + + // These incomplete prefixes are already impossible once their first + // continuation byte arrives and must not be retained for another call. + TEST_ASSERT(utf8_stream_safe_len("\xE0\x80") == 2); + TEST_ASSERT(utf8_stream_safe_len("\xED\xA0") == 2); + TEST_ASSERT(utf8_stream_safe_len("\xF0\x80") == 2); + TEST_ASSERT(utf8_stream_safe_len("\xF4\x90") == 2); +} + +TEST_CASE(ServerUnitFixture, test_emitter_replaces_invalid_utf8_without_holding_valid_suffix) { + const std::string replacement = "\xEF\xBF\xBD"; + + auto invalid_continuation = make_emitter(ApiFormat::OPENAI_CHAT); + invalid_continuation.emit_start(); + invalid_continuation.emit_token("A\xE2(Z"); + invalid_continuation.emit_finish(1); + TEST_ASSERT(invalid_continuation.accumulated_text() == + "A" + replacement + "(Z"); + + auto impossible_lead = make_emitter(ApiFormat::OPENAI_CHAT); + impossible_lead.emit_start(); + impossible_lead.emit_token("A\xF5Z"); + impossible_lead.emit_finish(1); + TEST_ASSERT(impossible_lead.accumulated_text() == + "A" + replacement + "Z"); +} + +// Feed one string in two pieces at every byte boundary. This deliberately +// models byte-fallback tokens: a valid code point may be split across calls. +static void assert_emitter_reassembles_at_split(ApiFormat format, + bool reasoning, + const std::string & expected) { + TEST_ASSERT(expected.size() > 1); + for (size_t split = 1; split < expected.size(); ++split) { + auto em = make_emitter(format, json::array(), reasoning); + std::vector chunks = em.emit_start(); + auto append = [&](std::vector more) { + chunks.insert(chunks.end(), more.begin(), more.end()); + }; + append(em.emit_token(expected.substr(0, split))); + append(em.emit_token(expected.substr(split))); + append(em.emit_finish(2)); + + std::string streamed; + for (const auto & chunk : chunks) { + if (chunk.rfind("data: ", 0) != 0) continue; + const size_t end = chunk.find('\n'); + const std::string payload = chunk.substr(6, end - 6); + if (payload == "[DONE]") continue; + const json event = json::parse(payload); + if (!event.contains("choices") || event["choices"].empty()) continue; + const json & delta = event["choices"][0]["delta"]; + const char * field = reasoning ? "reasoning_content" : "content"; + if (delta.contains(field) && delta[field].is_string()) { + streamed += delta[field].get(); + } + } + + const std::string & accumulated = reasoning + ? em.reasoning_text() : em.accumulated_text(); + TEST_ASSERT_MSG(accumulated == expected, + "split UTF-8 text changed before accumulation"); + TEST_ASSERT_MSG(streamed.find("\xEF\xBF\xBD") == std::string::npos, + "valid split UTF-8 emitted U+FFFD"); + TEST_ASSERT_MSG(streamed == expected, + "wire output lost valid split UTF-8"); + } +} + +TEST_CASE(ServerUnitFixture, test_emitter_reassembles_utf8_content_at_every_boundary) { + // 2-, 3-, and 4-byte code points, the exact candidate failure, set + // notation, CJK, emoji, and combining accents. + const std::vector samples = { + "e\xCC\x81", // e + combining acute accent + "caf\xC3\xA9 — \xCF\x80 \xE2\x88\x88 \xE2\x88\x85 \xE2\x8A\x86", + "\xE2\x88\x80\xE2\x88\x83\xE2\x88\x88\xE2\x88\x85\xE2\x8A\x86\xE2\x8A\x87\xE2\x88\xA9\xE2\x89\xA0\xE2\x86\x92", + "\xE6\x97\xA5\xE6\x9C\xAC\xE8\xAA\x9E \xF0\x9F\x9A\xA9", + "Return exactly this text and nothing else: caf\xC3\xA9 — \xCF\x80 \xE2\x88\x88 \xE2\x88\x85 \xE2\x8A\x86", + }; + for (const auto & sample : samples) { + assert_emitter_reassembles_at_split(ApiFormat::OPENAI_CHAT, false, sample); + } +} + +TEST_CASE(ServerUnitFixture, test_emitter_reassembles_utf8_reasoning_at_every_boundary) { + const std::vector samples = { + "caf\xC3\xA9 — \xCF\x80 \xE2\x88\x88 \xE2\x88\x85 \xE2\x8A\x86", + "\xE2\x88\x80\xE2\x88\x83\xE2\x88\x88\xE2\x88\x85\xE2\x8A\x86\xE2\x8A\x87\xE2\x88\xA9\xE2\x89\xA0\xE2\x86\x92", + "\xE6\x97\xA5\xE6\x9C\xAC\xE8\xAA\x9E \xF0\x9F\x98\x80 e\xCC\x81", + }; + for (const auto & sample : samples) { + assert_emitter_reassembles_at_split(ApiFormat::OPENAI_CHAT, true, sample); + } +} + +TEST_CASE(ServerUnitFixture, test_emitter_final_flush_replaces_truncated_utf8_once) { + auto em = make_emitter(ApiFormat::OPENAI_CHAT); + em.emit_start(); + em.emit_token("\xF0"); + em.emit_token("\x9F"); + em.emit_token("\x92"); + em.emit_finish(3); + TEST_ASSERT(em.accumulated_text() == "\xEF\xBF\xBD"); +} + +TEST_CASE(ServerUnitFixture, test_emitter_stop_discards_trailing_incomplete_utf8) { + SseEmitter em(ApiFormat::OPENAI_CHAT, "test_id_001", "test-model", 10, + json::array(), nullptr, {"STOP"}, false); + std::vector chunks = em.emit_start(); + auto append = [&](std::vector more) { + chunks.insert(chunks.end(), more.begin(), more.end()); + }; + append(em.emit_token("beforeSTOP\xF0\x9F\x92")); + append(em.emit_finish(1)); + + const std::string wire = concat(chunks); + TEST_ASSERT(em.accumulated_text() == "before"); + TEST_ASSERT(wire.find("\xEF\xBF\xBD") == std::string::npos); + TEST_ASSERT(wire.find("before") != std::string::npos); +} + // ═══════════════════════════════════════════════════════════════════════ // Reasoning parser tests // ═══════════════════════════════════════════════════════════════════════ From e6ae9ba8ff926b4b8c6476ec819450d47e6d1ae5 Mon Sep 17 00:00:00 2001 From: Drew Schuyler Date: Tue, 4 Aug 2026 08:02:04 -0500 Subject: [PATCH 2/2] fix(server): surface failed generations as errors --- server/src/server/http_server.cpp | 35 ++++++++++++++++++++++++------- server/src/server/sse_emitter.cpp | 11 ++++++++-- server/src/server/sse_emitter.h | 3 ++- server/test/test_server_unit.cpp | 19 +++++++++++++++++ 4 files changed, 58 insertions(+), 10 deletions(-) diff --git a/server/src/server/http_server.cpp b/server/src/server/http_server.cpp index 4ccf68b9a..6226f55e4 100644 --- a/server/src/server/http_server.cpp +++ b/server/src/server/http_server.cpp @@ -2096,7 +2096,7 @@ json build_openai_completion_response( // event and the chat DONE log. const std::string finish_reason = resolve_client_finish_reason( emitter.finish_reason(), counts.total, generation_cap, - result.degenerate_decode_close); + result.degenerate_decode_close, result.ok()); json choice = { {"index", 0}, @@ -2190,7 +2190,7 @@ json build_anthropic_response( // shared length-vs-EOS policy as OpenAI and the streaming terminal event. const std::string finish_reason = resolve_client_finish_reason( emitter.finish_reason(), counts.total, generation_cap, - result.degenerate_decode_close); + result.degenerate_decode_close, result.ok()); std::string stop_reason; if (finish_reason == "tool_calls") { stop_reason = "tool_use"; @@ -3366,7 +3366,8 @@ void HttpServer::process_job(ServerJob * job) { job->done = true; job->cv.notify_one(); }; - auto fail_request = [&](int status, const std::string & message) { + auto fail_request = [&](int status, const std::string & message, + bool finish = true) { std::fprintf(stderr, "[server] request failed: %s\n", message.c_str()); if (req.stream) { stop_job_stream(job); @@ -3378,7 +3379,7 @@ void HttpServer::process_job(ServerJob * job) { } else { send_error(fd, status, message); } - finish_job(); + if (finish) finish_job(); }; std::fprintf(stderr, @@ -3515,6 +3516,26 @@ void HttpServer::process_job(ServerJob * job) { req, prepared, cache, result, completion_tokens, visible_output_seen, client_disconnected); + // GenerateResult is authoritative. Do not finalize a failed generation + // as a normal stop/length/end_turn/completed response. A stream has + // already committed HTTP 200 headers, so fail_request() sends the + // existing SSE error envelope; non-streaming requests get HTTP 500. + // A disconnected client cannot receive either form and must retain the + // existing disconnect-only cleanup/logging path. + if (!result.ok() && !client_disconnected) { + std::string message = "generation failed"; + if (!result.error_detail().empty()) { + message += ": "; + message += result.error_detail(); + } else if (!result.error_code().empty()) { + message += ": "; + message += result.error_code(); + } + // Preserve the authoritative `chat DONE ... finish=error` log below, + // then signal completion once after the common logging path. + fail_request(500, message, false); + } + // Finalize. // Per-request wall-clock timings forwarded to the response's // `usage.timings` (OpenAI Chat usage chunk, Anthropic @@ -3563,17 +3584,17 @@ void HttpServer::process_job(ServerJob * job) { if (job->client_disconnected.load(std::memory_order_acquire)) { client_disconnected = true; } - if (req.stream && !client_disconnected) { + if (result.ok() && req.stream && !client_disconnected) { auto final_chunks = emitter.emit_finish( completion_tokens, &gen_timings, n_gen_cap, - result.degenerate_decode_close); + result.degenerate_decode_close, result.ok()); for (const auto & chunk : final_chunks) { if (!send_job_bytes(job, chunk.data(), chunk.size())) { client_disconnected = true; break; } } - } else if (!req.stream && !client_disconnected) { + } else if (result.ok() && !req.stream && !client_disconnected) { const json response = build_non_streaming_response( req, result, n_gen_cap, gen_timings, tokenizer_, emitter); // Streaming uses non-blocking sends; restore blocking mode before diff --git a/server/src/server/sse_emitter.cpp b/server/src/server/sse_emitter.cpp index 2ed02e0cc..6c3499b13 100644 --- a/server/src/server/sse_emitter.cpp +++ b/server/src/server/sse_emitter.cpp @@ -529,9 +529,15 @@ void SseEmitter::emit_content_delta(std::vector & out, std::vector SseEmitter::emit_finish(int completion_tokens, const GenTimings * timings, int generation_cap, - bool degenerate_decode_close) { + bool degenerate_decode_close, + bool result_ok) { std::vector out; + // Failed generations use HttpServer::fail_request() instead of a + // format-specific terminal success event. Keep this guard here as a + // second line of defense for callers that use the emitter directly. + if (!result_ok) return out; + // A valid but incomplete trailing sequence is only safe to retain until // the generation is over. At finalization it is genuinely truncated and // utf8_sanitize() emits one deterministic replacement character. @@ -713,7 +719,8 @@ std::vector SseEmitter::emit_finish(int completion_tokens, } fr = resolve_client_finish_reason( - fr, completion_tokens, generation_cap, degenerate_decode_close); + fr, completion_tokens, generation_cap, degenerate_decode_close, + result_ok); // Format-specific final events switch (format_) { diff --git a/server/src/server/sse_emitter.h b/server/src/server/sse_emitter.h index 26e0e283d..b6735afe7 100644 --- a/server/src/server/sse_emitter.h +++ b/server/src/server/sse_emitter.h @@ -89,7 +89,8 @@ class SseEmitter { std::vector emit_finish(int completion_tokens, const GenTimings * timings = nullptr, int generation_cap = -1, - bool degenerate_decode_close = false); + bool degenerate_decode_close = false, + bool result_ok = true); // Get the finish_reason for non-streaming responses. std::string finish_reason() const; diff --git a/server/test/test_server_unit.cpp b/server/test/test_server_unit.cpp index 27b68dbd8..840eaf134 100644 --- a/server/test/test_server_unit.cpp +++ b/server/test/test_server_unit.cpp @@ -361,6 +361,8 @@ TEST_CASE(ServerUnitFixture, test_finish_reason_resolution_table) { {"tool_calls", 8, 8, true, true, false, "tool_calls"}, {"stop", 3, 8, false, true, true, "client_disconnect"}, {"stop", 3, 8, false, false, false, "error"}, + {"length", 8, 8, false, false, false, "error"}, + {"tool_calls", 8, 8, true, false, false, "error"}, }; for (const auto & c : cases) { TEST_ASSERT(resolve_client_finish_reason( @@ -369,6 +371,23 @@ TEST_CASE(ServerUnitFixture, test_finish_reason_resolution_table) { } } +TEST_CASE(ServerUnitFixture, test_failed_stream_finish_emits_no_success_terminal_event) { + for (const auto format : {ApiFormat::OPENAI_CHAT, + ApiFormat::ANTHROPIC, + ApiFormat::RESPONSES}) { + auto em = make_emitter(format); + em.emit_start(); + em.emit_token("partial"); + const std::string wire = concat( + em.emit_finish(1, nullptr, 8, false, false)); + TEST_ASSERT(wire.find("finish_reason") == std::string::npos); + TEST_ASSERT(wire.find("stop_reason") == std::string::npos); + TEST_ASSERT(wire.find("response.completed") == std::string::npos); + TEST_ASSERT(wire.find("\"status\":\"completed\"") == + std::string::npos); + } +} + TEST_CASE(ServerUnitFixture, test_streaming_finish_reason_uses_shared_length_policy) { auto openai = make_emitter(ApiFormat::OPENAI_CHAT); openai.emit_start();