From bbda5b0ccd941b7229263b5fb5d386d99116d43d Mon Sep 17 00:00:00 2001 From: Alexander Morozov Date: Mon, 6 Jul 2026 14:57:36 +0400 Subject: [PATCH 1/2] Support streaming compaction output --- .../Responses/Streaming/OutputItem.php | 7 +++++-- tests/Fixtures/Responses.php | 5 +++++ .../ResponseOutputItemCompactionDone.txt | 1 + .../Responses/CreateStreamedResponse.php | 17 +++++++++++++++++ 4 files changed, 28 insertions(+), 2 deletions(-) create mode 100644 tests/Fixtures/Streams/ResponseOutputItemCompactionDone.txt diff --git a/src/Responses/Responses/Streaming/OutputItem.php b/src/Responses/Responses/Streaming/OutputItem.php index 1223c0e3b..68e037452 100644 --- a/src/Responses/Responses/Streaming/OutputItem.php +++ b/src/Responses/Responses/Streaming/OutputItem.php @@ -10,6 +10,7 @@ use OpenAI\Responses\Concerns\HasMetaInformation; use OpenAI\Responses\Meta\MetaInformation; use OpenAI\Responses\Responses\Output\OutputCodeInterpreterToolCall; +use OpenAI\Responses\Responses\Output\OutputCompaction; use OpenAI\Responses\Responses\Output\OutputComputerToolCall; use OpenAI\Responses\Responses\Output\OutputFileSearchToolCall; use OpenAI\Responses\Responses\Output\OutputFunctionToolCall; @@ -34,8 +35,9 @@ * @phpstan-import-type OutputMcpApprovalRequestType from OutputMcpApprovalRequest * @phpstan-import-type OutputMcpCallType from OutputMcpCall * @phpstan-import-type OutputCodeInterpreterToolCallType from OutputCodeInterpreterToolCall + * @phpstan-import-type OutputCompactionType from OutputCompaction * - * @phpstan-type OutputItemType array{type: string, output_index: int, sequence_number: int, item: OutputCodeInterpreterToolCallType|OutputComputerToolCallType|OutputFileSearchToolCallType|OutputFunctionToolCallType|OutputMessageType|OutputReasoningType|OutputWebSearchToolCallType|OutputMcpListToolsType|OutputMcpApprovalRequestType|OutputMcpCallType|OutputImageGenerationToolCallType} + * @phpstan-type OutputItemType array{type: string, output_index: int, sequence_number: int, item: OutputCodeInterpreterToolCallType|OutputComputerToolCallType|OutputFileSearchToolCallType|OutputFunctionToolCallType|OutputMessageType|OutputReasoningType|OutputWebSearchToolCallType|OutputMcpListToolsType|OutputMcpApprovalRequestType|OutputMcpCallType|OutputImageGenerationToolCallType|OutputCompactionType} * * @implements ResponseContract */ @@ -53,7 +55,7 @@ private function __construct( public readonly string $type, public readonly int $outputIndex, public readonly int $sequenceNumber, - public readonly OutputMessage|OutputCodeInterpreterToolCall|OutputFileSearchToolCall|OutputFunctionToolCall|OutputWebSearchToolCall|OutputComputerToolCall|OutputReasoning|OutputMcpListTools|OutputMcpApprovalRequest|OutputMcpCall|OutputImageGenerationToolCall $item, + public readonly OutputMessage|OutputCodeInterpreterToolCall|OutputFileSearchToolCall|OutputFunctionToolCall|OutputWebSearchToolCall|OutputComputerToolCall|OutputReasoning|OutputMcpListTools|OutputMcpApprovalRequest|OutputMcpCall|OutputImageGenerationToolCall|OutputCompaction $item, private readonly MetaInformation $meta, ) {} @@ -74,6 +76,7 @@ public static function from(array $attributes, MetaInformation $meta): self 'mcp_approval_request' => OutputMcpApprovalRequest::from($attributes['item']), 'mcp_call' => OutputMcpCall::from($attributes['item']), 'code_interpreter_call' => OutputCodeInterpreterToolCall::from($attributes['item']), + 'compaction' => OutputCompaction::from($attributes['item']), }; return new self( diff --git a/tests/Fixtures/Responses.php b/tests/Fixtures/Responses.php index 36fd6a2a2..4da44a0f7 100644 --- a/tests/Fixtures/Responses.php +++ b/tests/Fixtures/Responses.php @@ -1042,3 +1042,8 @@ function responseRateLimitsUpdatedEvent() { return fopen(__DIR__.'/Streams/ResponseRateLimitsUpdated.txt', 'r'); } + +function responseOutputItemCompactionDoneEvent() +{ + return fopen(__DIR__.'/Streams/ResponseOutputItemCompactionDone.txt', 'r'); +} diff --git a/tests/Fixtures/Streams/ResponseOutputItemCompactionDone.txt b/tests/Fixtures/Streams/ResponseOutputItemCompactionDone.txt new file mode 100644 index 000000000..c8d409118 --- /dev/null +++ b/tests/Fixtures/Streams/ResponseOutputItemCompactionDone.txt @@ -0,0 +1 @@ +data: {"type":"response.output_item.done","output_index":0,"sequence_number":11,"item":{"id":"cmp_67ccf18f64008190a39b619f4c8455ef087bb177ab789d5c","encrypted_content":"encrypted_string_value","type":"compaction","created_by":"user_123"}} diff --git a/tests/Responses/Responses/CreateStreamedResponse.php b/tests/Responses/Responses/CreateStreamedResponse.php index e303d37f7..907243fbd 100644 --- a/tests/Responses/Responses/CreateStreamedResponse.php +++ b/tests/Responses/Responses/CreateStreamedResponse.php @@ -2,6 +2,8 @@ use OpenAI\Responses\Responses\CreateResponse; use OpenAI\Responses\Responses\CreateStreamedResponse; +use OpenAI\Responses\Responses\Output\OutputCompaction; +use OpenAI\Responses\Responses\Streaming\OutputItem; use OpenAI\Responses\Responses\Streaming\RateLimits; use OpenAI\Responses\Responses\Streaming\ReasoningTextDelta; use OpenAI\Responses\Responses\Streaming\ReasoningTextDone; @@ -68,3 +70,18 @@ 'type' => 'response.rate_limits.updated', ]); }); + +test('output item done event with compaction item', function () { + $response = CreateStreamedResponse::fake(responseOutputItemCompactionDoneEvent()); + + expect($response->getIterator()->current()) + ->toBeInstanceOf(CreateStreamedResponse::class) + ->event->toBe('response.output_item.done') + ->response->toBeInstanceOf(OutputItem::class) + ->response->outputIndex->toBe(0) + ->response->sequenceNumber->toBe(11) + ->response->item->toBeInstanceOf(OutputCompaction::class) + ->response->item->id->toBe('cmp_67ccf18f64008190a39b619f4c8455ef087bb177ab789d5c') + ->response->item->encryptedContent->toBe('encrypted_string_value') + ->response->item->createdBy->toBe('user_123'); +}); From 2cf9c1d196a9cec77c838880d055b75c2e1133f1 Mon Sep 17 00:00:00 2001 From: Alexander Morozov Date: Tue, 11 Aug 2026 21:42:57 +0400 Subject: [PATCH 2/2] Optimize streamed response parsing --- src/Responses/StreamResponse.php | 117 +++++++++++++++++++++------ tests/Responses/StreamResponse.php | 122 +++++++++++++++++++++++++++++ 2 files changed, 217 insertions(+), 22 deletions(-) create mode 100644 tests/Responses/StreamResponse.php diff --git a/src/Responses/StreamResponse.php b/src/Responses/StreamResponse.php index cddda7aa7..d4ce383a0 100644 --- a/src/Responses/StreamResponse.php +++ b/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, type: string, code: string}, type?: string} $response */ - $response = json_decode($data, true, flags: JSON_THROW_ON_ERROR); + /** @var array{error?: array{message: string|array, 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 diff --git a/tests/Responses/StreamResponse.php b/tests/Responses/StreamResponse.php new file mode 100644 index 000000000..71b4323b4 --- /dev/null +++ b/tests/Responses/StreamResponse.php @@ -0,0 +1,122 @@ +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'); +});