fix(OpenAI): optimize streamed response processing

src/Responses/StreamResponse.php

@@ -17,6 +17,10 @@
  */
 final class StreamResponse implements ResponseHasMetaInformationContract, ResponseStreamContract
 {
+    private const STREAM_READ_SIZE = 64 * 1024;
+
+    private string $lineBuffer = '';
+
     /**
      * Creates a new Stream Response instance.
      *
@@ -34,64 +38,133 @@ public function __construct(
      */
     public function getIterator(): Generator
     {
-        while (! $this->response->getBody()->eof()) {
-            $line = $this->readLine($this->response->getBody());
+        $body = $this->response->getBody();
+        $event = null;
 
-            $event = null;
+        while (($line = $this->readLine($body)) !== null) {
             if (str_starts_with($line, 'event:')) {
                 $event = trim(substr($line, strlen('event:')));
-                $line = $this->readLine($this->response->getBody());
+
+                unset($line);
+
+                continue;
             }
 
             if (! str_starts_with($line, 'data:')) {
+                $event = null;
+
+                unset($line);
+
                 continue;
             }
 
-            $data = trim(substr($line, strlen('data:')));
+            $data = substr($line, strlen('data:'));
+
+            unset($line);
+
+            if (strlen($data) <= 16 && trim($data) === '[DONE]') {
+                unset($data);
 
-            if ($data === '[DONE]') {
                 break;
             }
 
-            /** @var array{error?: array{message: string|array<int, string>, type: string, code: string}, type?: string} $response */
-            $response = json_decode($data, true, flags: JSON_THROW_ON_ERROR);
+            /** @var array{error?: array{message: string|array<int, string>, type: string, code: string}, type?: string} $attributes */
+            $attributes = json_decode($data, true, flags: JSON_THROW_ON_ERROR);
+
+            unset($data);
 
-            if (isset($response['error'])) {
-                throw new ErrorException($response['error'], $this->response);
+            if (isset($attributes['error'])) {
+                throw new ErrorException($attributes['error'], $this->response);
             }
 
             $skippableTypes = ['ping', 'keepalive', 'response.keep_alive'];
-            if (isset($response['type']) && in_array($response['type'], $skippableTypes, true)) {
+
+            if (isset($attributes['type']) && in_array($attributes['type'], $skippableTypes, true)) {
+                $event = null;
+
+                unset($attributes);
+
                 continue;
             }
 
             if ($event !== null) {
-                $response['__event'] = $event;
+                $attributes['__event'] = $event;
             }
-            $response['__meta'] = $this->meta();
 
-            yield $this->responseClass::from($response);
+            $attributes['__meta'] = $this->meta();
+
+            $streamEvent = $this->responseClass::from($attributes);
+
+            $event = null;
+
+            unset($attributes);
+
+            yield $streamEvent;
+
+            unset($streamEvent);
         }
     }
 
     /**
      * Read a line from the stream.
      */
-    private function readLine(StreamInterface $stream): string
+    private function readLine(StreamInterface $stream): ?string
     {
-        $buffer = '';
+        $newLinePosition = strpos($this->lineBuffer, "\n");
+
+        if ($newLinePosition !== false) {
+            $lineLength = $newLinePosition + 1;
+
+            if ($lineLength === strlen($this->lineBuffer)) {
+                $line = $this->lineBuffer;
+                $this->lineBuffer = '';
+
+                return $line;
+            }
+
+            $line = substr($this->lineBuffer, 0, $lineLength);
+            $this->lineBuffer = substr($this->lineBuffer, $lineLength);
+
+            return $line;
+        }
 
         while (! $stream->eof()) {
-            if ('' === ($byte = $stream->read(1))) {
-                return $buffer;
+            $chunk = $stream->read(self::STREAM_READ_SIZE);
+
+            if ($chunk === '') {
+                continue;
             }
-            $buffer .= $byte;
-            if ($byte === "\n") {
-                break;
+
+            // Split the fresh chunk before appending it to avoid copying a large buffered line.
+            $newLinePosition = strpos($chunk, "\n");
+
+            if ($newLinePosition === false) {
+                $this->lineBuffer .= $chunk;
+
+                continue;
+            }
+
+            $lineLength = $newLinePosition + 1;
+            $line = $this->lineBuffer;
+            $this->lineBuffer = substr($chunk, $lineLength);
+
+            if ($lineLength === strlen($chunk)) {
+                $line .= $chunk;
+            } else {
+                $line .= substr($chunk, 0, $lineLength);
             }
+
+            return $line;
         }
 
-        return $buffer;
+        if ($this->lineBuffer === '') {
+            return null;
+        }
+
+        $line = $this->lineBuffer;
+        $this->lineBuffer = '';
+
+        return $line;
     }
 
     public function meta(): MetaInformation

tests/Responses/StreamResponse.php

@@ -0,0 +1,122 @@
+<?php
+
+use GuzzleHttp\Psr7\Response;
+use GuzzleHttp\Psr7\StreamDecoratorTrait;
+use GuzzleHttp\Psr7\Utils;
+use OpenAI\Responses\Responses\CreateStreamedResponse;
+use OpenAI\Responses\Responses\Output\OutputCompaction;
+use OpenAI\Responses\StreamResponse;
+use Psr\Http\Message\StreamInterface;
+
+final class StreamResponseReadTrackingStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    private StreamInterface $stream;
+
+    public int $readCalls = 0;
+
+    public int $largestRead = 0;
+
+    public ?int $emptyReadCall = null;
+
+    public ?int $maximumBytesPerRead = null;
+
+    public function read($length): string
+    {
+        $this->readCalls++;
+        $this->largestRead = max($this->largestRead, $length);
+
+        if ($this->readCalls === $this->emptyReadCall) {
+            return '';
+        }
+
+        if ($this->maximumBytesPerRead !== null) {
+            $length = min($length, $this->maximumBytesPerRead);
+        }
+
+        return $this->stream->read($length);
+    }
+}
+
+test('reads large SSE events in chunks without losing buffered events', function () {
+    $largeEncryptedContent = str_repeat('a', 128 * 1024);
+
+    $events = [
+        [
+            'type' => 'response.output_item.done',
+            'output_index' => 0,
+            'sequence_number' => 1,
+            'item' => [
+                'id' => 'cmp_large',
+                'encrypted_content' => $largeEncryptedContent,
+                'type' => 'compaction',
+                'created_by' => 'user',
+            ],
+        ],
+        [
+            'type' => 'response.output_item.done',
+            'output_index' => 1,
+            'sequence_number' => 2,
+            'item' => [
+                'id' => 'cmp_buffered',
+                'encrypted_content' => 'buffered content',
+                'type' => 'compaction',
+                'created_by' => 'user',
+            ],
+        ],
+    ];
+
+    $body = implode('', array_map(
+        fn (array $event): string => "event: response.output_item.done\n".
+            'data: '.json_encode($event, flags: JSON_THROW_ON_ERROR)."\n\n",
+        $events,
+    )).'data: [DONE]';
+
+    $stream = new StreamResponseReadTrackingStream(Utils::streamFor($body));
+    $response = new Response(body: $stream);
+    $streamResponse = new StreamResponse(CreateStreamedResponse::class, $response);
+
+    $result = iterator_to_array($streamResponse);
+
+    expect($result)
+        ->toHaveCount(2)
+        ->and($result[0]->response->item)
+        ->toBeInstanceOf(OutputCompaction::class)
+        ->encryptedContent->toBe($largeEncryptedContent)
+        ->and($result[1]->response->item)
+        ->toBeInstanceOf(OutputCompaction::class)
+        ->encryptedContent->toBe('buffered content')
+        ->and($stream->largestRead)->toBe(64 * 1024)
+        ->and($stream->readCalls)->toBeLessThan(10);
+});
+
+test('retries an empty read before EOF without losing the buffered line', function () {
+    $attributes = [
+        'type' => 'response.output_item.done',
+        'output_index' => 0,
+        'sequence_number' => 1,
+        'item' => [
+            'id' => 'cmp_after_empty_read',
+            'encrypted_content' => 'complete content',
+            'type' => 'compaction',
+            'created_by' => 'user',
+        ],
+    ];
+    $body = 'data: '.json_encode($attributes, flags: JSON_THROW_ON_ERROR)."\n";
+
+    $stream = new StreamResponseReadTrackingStream(Utils::streamFor($body));
+    $stream->maximumBytesPerRead = 10;
+    $stream->emptyReadCall = 2;
+
+    $response = new Response(body: $stream);
+    $streamResponse = new StreamResponse(CreateStreamedResponse::class, $response);
+
+    $result = iterator_to_array($streamResponse);
+
+    expect($result)
+        ->toHaveCount(1)
+        ->and($result[0]->response->item)
+        ->toBeInstanceOf(OutputCompaction::class)
+        ->encryptedContent->toBe('complete content');
+});