fix(server): preserve UTF-8 across stream boundaries

server/src/server/finish_reason.h

@@ -0,0 +1,33 @@
+// Shared client-facing finish-reason policy for HTTP responses and logs.
+#pragma once
+
+#include <string>
+
+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

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, result.ok());
 
     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, result.ok());
     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";
@@ -3370,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);
@@ -3382,7 +3379,7 @@ void HttpServer::process_job(ServerJob * job) {
         } else {
             send_error(fd, status, message);
         }
-        finish_job();
+        if (finish) finish_job();
     };
 
     std::fprintf(stderr,
@@ -3519,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
@@ -3567,15 +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) {
-        auto final_chunks = emitter.emit_finish(completion_tokens, &gen_timings);
+    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.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
@@ -3600,9 +3619,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 "

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 <algorithm>
@@ -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<std::string> 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<std::string> out;
     accumulated_raw_ += piece;
     window_ += piece;
@@ -276,6 +299,10 @@ std::vector<std::string> 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,27 @@ void SseEmitter::emit_content_delta(std::vector<std::string> & out,
 // ─── emit_finish ────────────────────────────────────────────────────────
 
 std::vector<std::string> SseEmitter::emit_finish(int completion_tokens,
-                                                 const GenTimings * timings) {
+                                                 const GenTimings * timings,
+                                                 int generation_cap,
+                                                 bool degenerate_decode_close,
+                                                 bool result_ok) {
     std::vector<std::string> 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.
+    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 +718,10 @@ std::vector<std::string> SseEmitter::emit_finish(int completion_tokens,
         }
     }
 
+    fr = resolve_client_finish_reason(
+        fr, completion_tokens, generation_cap, degenerate_decode_close,
+        result_ok);
+
     // Format-specific final events
     switch (format_) {
     case ApiFormat::OPENAI_CHAT: {
@@ -711,11 +760,12 @@ std::vector<std::string> 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);

server/src/server/sse_emitter.h

@@ -87,7 +87,10 @@ class SseEmitter {
     // the pre-timings API for unit tests that don't exercise that
     // shape.
     std::vector<std::string> emit_finish(int completion_tokens,
-                                         const GenTimings * timings = nullptr);
+                                         const GenTimings * timings = nullptr,
+                                         int generation_cap = -1,
+                                         bool degenerate_decode_close = false,
+                                         bool result_ok = true);
 
     // Get the finish_reason for non-streaming responses.
     std::string finish_reason() const;
@@ -153,6 +156,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;