Handle parsing for error when chunked

When the OpenAI API returns an error during streaming, the error JSON
body may be split across multiple HTTP chunks. Previously, the code
would immediately try to raise an error on the first chunk, which would
fail if the JSON was incomplete.

This change accumulates error chunks until a complete, parsable JSON
object is formed, then raises the appropriate Faraday error. It also
prevents the user proc from being called when an error status is
received.

Based on the approach from PR alexrudall#591.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: stevenou

lib/openai/stream.rb

@@ -6,6 +6,7 @@ class Stream
     def initialize(user_proc:, parser: EventStreamParser::Parser.new)
       @user_proc = user_proc
       @parser = parser
+      @accumulated_error = ""
 
       # To be backwards compatible, we need to check how many arguments the user_proc takes.
       @user_proc_arity =
@@ -18,13 +19,16 @@ def initialize(user_proc:, parser: EventStreamParser::Parser.new)
     end
 
     def call(chunk, _bytes, env = nil)
-      handle_http_error(chunk: chunk, env: env) if env && env.status != 200
-
-      parser.feed(chunk) do |event, data|
-        next if data == DONE
-
-        args = [JSON.parse(data), event].first(user_proc_arity)
-        user_proc.call(*args)
+      if env && env.status != 200
+        @accumulated_error += chunk
+        raise_error_when_ready(env)
+      else
+        parser.feed(chunk) do |event, data|
+          next if data == DONE
+
+          args = [JSON.parse(data), event].first(user_proc_arity)
+          user_proc.call(*args)
+        end
       end
     end
 
@@ -34,11 +38,14 @@ def to_proc
 
     private
 
-    attr_reader :user_proc, :parser, :user_proc_arity
+    attr_reader :user_proc, :parser, :user_proc_arity, :accumulated_error
+
+    def raise_error_when_ready(env)
+      parsed_error = try_parse_json(accumulated_error)
+      return if parsed_error.is_a?(String)
 
-    def handle_http_error(chunk:, env:)
       raise_error = Faraday::Response::RaiseError.new
-      raise_error.on_complete(env.merge(body: try_parse_json(chunk)))
+      raise_error.on_complete(env.merge(body: parsed_error))
     end
 
     def try_parse_json(maybe_json)

spec/openai/client/stream_spec.rb

@@ -92,6 +92,56 @@
         end
       end
 
+      context "with a HTTP error response with body containing JSON split across chunks" do
+        let(:error_env) do
+          Faraday::Env.from(
+            method: :post,
+            url: URI("http://example.com"),
+            status: 400,
+            request: {},
+            response: Faraday::Response.new
+          )
+        end
+        let(:expected_body) do
+          {
+            "error" => {
+              "message" => "Test error",
+              "type" => "test_error",
+              "param" => nil,
+              "code" => "test"
+            }
+          }
+        end
+
+        it "raises an error" do
+          json = expected_body.to_json
+          # Split the JSON into two chunks in the middle
+          chunks = [json[0..(json.length / 2)], json[((json.length / 2) + 1)..]]
+
+          expect do
+            chunks.each do |chunk|
+              stream.call(chunk, bytes, error_env)
+            end
+          end.to raise_error(Faraday::BadRequestError) do |e|
+            expect(e.response).to include(status: 400)
+            expect(e.response[:body]).to eq(expected_body)
+          end
+        end
+
+        it "does not call user proc on error" do
+          expect(user_proc).not_to receive(:call)
+
+          json = expected_body.to_json
+          chunks = [json[0..(json.length / 2)], json[((json.length / 2) + 1)..]]
+
+          expect do
+            chunks.each do |chunk|
+              stream.call(chunk, bytes, error_env)
+            end
+          end.to raise_error(Faraday::BadRequestError)
+        end
+      end
+
       context "with a call method that only takes one argument" do
         let(:user_proc) { proc { |data| data } }