fix: preserve UTF-8 SSE stream boundaries

lib/openai/internal/util.rb

@@ -781,41 +781,54 @@ def chain_fused(enum, &blk)
       class << self
         # @api private
         #
-        # Assumes Strings have been forced into having `Encoding::BINARY`.
-        #
-        # This decoder is responsible for reassembling lines split across multiple
-        # fragments.
+        # This decoder is responsible for reassembling bytes split across multiple
+        # String fragments without depending on each fragment's encoding label
+        # while preserving the String encoding callers supplied.
         #
         # @param enum [Enumerable<String>]
         #
         # @return [Enumerable<String>]
         def decode_lines(enum)
           re = /(\r\n|\r|\n)/
-          buffer = String.new
+          buffer = String.new(encoding: Encoding::BINARY)
+          encoding = nil
           cr_seen = nil
 
           chain_fused(enum) do |y|
             enum.each do |row|
+              encoding = row.encoding if !row.empty? && (encoding.nil? || encoding == Encoding::US_ASCII)
               offset = buffer.bytesize
-              buffer << row
+              buffer << (row.encoding == Encoding::BINARY ? row : row.b)
               while (match = re.match(buffer, cr_seen&.to_i || offset))
                 case [match.captures.first, cr_seen]
                 in ["\r", nil]
                   cr_seen = match.end(1)
                   next
                 in ["\r" | "\r\n", Integer]
-                  y << buffer.slice!(..(cr_seen.pred))
+                  line = buffer.slice!(..(cr_seen.pred))
+                  line.force_encoding(encoding) unless encoding.nil?
+                  y << line
                 else
-                  y << buffer.slice!(..(match.end(1).pred))
+                  line = buffer.slice!(..(match.end(1).pred))
+                  line.force_encoding(encoding) unless encoding.nil?
+                  y << line
                 end
 
                 offset = 0
                 cr_seen = nil
               end
             end
 
-            y << buffer.slice!(..(cr_seen.pred)) unless cr_seen.nil?
-            y << buffer unless buffer.empty?
+            unless cr_seen.nil?
+              line = buffer.slice!(..(cr_seen.pred))
+              line.force_encoding(encoding) unless encoding.nil?
+              y << line
+            end
+
+            unless buffer.empty?
+              buffer.force_encoding(encoding) unless encoding.nil?
+              y << buffer
+            end
           end
         end
 
@@ -833,8 +846,14 @@ def decode_sse(lines)
           chain_fused(lines) do |y|
             blank = {event: nil, data: nil, id: nil, retry: nil}
             current = {}
+            first_line = true
 
             lines.each do |line|
+              if first_line
+                line = line.byteslice(3..) || "" if line.byteslice(0, 3)&.bytes == [0xEF, 0xBB, 0xBF]
+                first_line = false
+              end
+
               case line.sub(/\R$/, "")
               in ""
                 next if current.empty?

test/openai/sse_utf8_boundary_test.rb

@@ -0,0 +1,104 @@
+# frozen_string_literal: true
+
+require_relative "test_helper"
+
+class OpenAI::Test::SseUtf8BoundaryTest < Minitest::Test
+  extend Minitest::Serial
+
+  def test_one_leading_bom_matches_plain_sse_for_raw_and_helper_streams
+    wire = chat_stream_wire(content: "start ", terminal_content: "end")
+    bom = "\xEF\xBB\xBF".b
+    plain = consume_stream([wire])
+
+    assert_equal(plain, consume_stream([bom + wire]))
+    assert_equal(plain, consume_stream([bom.byteslice(0, 1), bom.byteslice(1, 1), bom.byteslice(2, 1), wire]))
+  end
+
+  def test_utf8_labeled_fragments_match_binary_fragments_across_multibyte_boundaries
+    wire = chat_stream_wire(content: "你好世界", terminal_content: " end")
+    split = wire.index("你".b).succ
+    binary_fragments = [wire.byteslice(...split), wire.byteslice(split...)]
+    utf8_fragments = binary_fragments.map { _1.dup.force_encoding(Encoding::UTF_8) }
+    snapshots = utf8_fragments.map { [_1.bytes, _1.encoding] }
+
+    assert_equal(consume_stream(binary_fragments), consume_stream(utf8_fragments))
+    assert_equal(snapshots, utf8_fragments.map { [_1.bytes, _1.encoding] })
+  end
+
+  def test_ascii_prefix_then_utf8_content_matches_binary_sse
+    wire = chat_stream_wire(content: "你好世界", terminal_content: " end")
+    split = wire.index("你".b)
+    mixed_fragments = [
+      wire.byteslice(...split).force_encoding(Encoding::US_ASCII),
+      wire.byteslice(split...).force_encoding(Encoding::UTF_8)
+    ]
+
+    assert_equal(consume_stream([wire]), consume_stream(mixed_fragments))
+  end
+
+  def test_ascii_prefix_then_utf8_content_preserves_jsonl_decoding
+    jsonl = "{\"content\":\"你好世界\"}\n"
+    split = jsonl.index("你")
+    fragments = [
+      jsonl.byteslice(...split).force_encoding(Encoding::US_ASCII),
+      jsonl.byteslice(split...).force_encoding(Encoding::UTF_8)
+    ]
+
+    line = OpenAI::Internal::Util.decode_lines(fragments).to_a.fetch(0)
+
+    assert_equal("你好世界", JSON.parse(line).fetch("content"))
+  end
+
+  def test_only_a_leading_bom_is_ignored_and_crlf_still_decodes
+    embedded_bom = "before \uFEFF after"
+    wire = chat_stream_wire(content: embedded_bom, terminal_content: "", eol: "\r\n")
+    raw, helper = consume_stream([wire])
+
+    assert_equal([[embedded_bom, nil], ["", :stop]], raw)
+    assert_equal([embedded_bom, :stop], helper)
+  end
+
+  private
+
+  def chat_stream_wire(content:, terminal_content:, eol: "\n")
+    base = {id: "chatcmpl_synthetic", object: "chat.completion.chunk", created: 1, model: "test"}
+    first = base.merge(choices: [{index: 0, delta: {role: "assistant", content: content}, finish_reason: nil}])
+    last = base.merge(choices: [{index: 0, delta: {content: terminal_content}, finish_reason: "stop"}])
+    [first, last].map { "data: #{JSON.generate(_1)}#{eol}#{eol}" }.join.b + "data: [DONE]#{eol}#{eol}".b
+  end
+
+  def consume_stream(fragments)
+    [:stream_raw, :stream].map do |method|
+      transport = OpenAI::HTTPClient.new
+      response = OpenAI::HTTPClient::Response.new(
+        status: 200,
+        headers: {"content-type" => "text/event-stream"},
+        body: fragments
+      )
+
+      transport.stub(:execute, response) do
+        client = OpenAI::Client.new(
+          api_key: "synthetic-key",
+          base_url: "https://sdk.example.test",
+          http_client: transport
+        )
+        stream = client.chat.completions.public_send(
+          method,
+          model: "test",
+          messages: [{role: :user, content: "synthetic"}]
+        )
+        events = stream.to_a
+
+        if method == :stream_raw
+          events.flat_map(&:choices).map { [_1.delta.content, _1.finish_reason] }
+        else
+          completion = stream.get_final_completion
+          [completion.choices.first.message.content, completion.choices.first.finish_reason]
+        end
+
+      ensure
+        stream&.close
+      end
+    end
+  end
+end