[OpenAI Bridge] commentary phase is treated as TextDelta, causing duplicate assistant output in streaming responses

[wimwinterberg]

Describe Your Problem

When using the OpenAI Responses streaming API via the Symfony AI OpenAI bridge, assistant output can be duplicated.

This happens because messages with phase: "commentary" are currently handled the same way as phase: "final_answer". As a result, both are emitted as TextDelta, which leads to duplicated visible output.

Expected behavior

• Only phase: "final_answer" should be emitted as TextDelta
• phase: "commentary" should either:
  • be ignored, or
  • be mapped to ThinkingDelta (similar to reasoning_summary_text)

Actual behavior

Both commentary and final_answer phases are treated as regular output text and emitted as TextDelta, causing duplicated responses in the final output.

---

Steps to reproduce

1. Use the OpenAI Responses API with streaming enabled
2. Trigger a response where the model emits both:
   • phase: "commentary"
   • phase: "final_answer"
3. Observe that the same content is appended twice in the final output

---

Technical details

In ResultConverter::convertStream():

if (str_contains($type, 'output_text') && isset($event['delta'])) {
    yield new TextDelta($event['delta']);
}

This logic does not differentiate between phases.

However, OpenAI emits response.output_text.delta for both:

• commentary
• final_answer

The distinction is only available via:

event.item_id → response.output_item → phase

---

Suggested solution

Track phase per item_id using response.output_item.added events:

$itemPhases[$event['item']['id']] = $event['item']['phase'] ?? null;

Then update handling of response.output_text.delta:

$phase = $itemPhases[$event['item_id']] ?? null;

if ($phase === 'final_answer') {
    yield new TextDelta($event['delta']);
}

if ($phase === 'commentary') {
    // optionally map to ThinkingDelta or ignore
}

Also consider replacing:

str_contains($type, 'output_text')

with a stricter check:

$type === 'response.output_text.delta'

---

References

• OpenAI Responses API (assistant message phase field):
  https://platform.openai.com/docs/api-reference/responses

---

Notes

• reasoning_summary_text.* is already correctly handled by the bridge
• commentary is different: it is emitted as a normal message with a phase attribute
• This issue only affects streaming responses

---

[wimwinterberg]

possible fix?

    private function convertStream(RawResultInterface|RawHttpResult $result): \Generator
    {
        $itemPhases = [];
        $currentReasoningSummary = null;
        $currentCommentary = null;

        foreach ($result->getDataStream() as $event) {
            $type = $event['type'] ?? '';

            if ('error' === $type && isset($event['error'])) {
                throw new RuntimeException($this->generateErrorMessage($event['error']));
            }

            if (isset($event['response']['usage'])) {
                yield $this->getTokenUsageExtractor()->fromDataArray($event['response']);
            }

            // track phase for each output item so output_text.delta can be routed correctly.
            if ('response.output_item.added' === $type && isset($event['item']['id'])) {
                $itemPhases[$event['item']['id']] = $event['item']['phase'] ?? null;
            }

            // dedicated buffer for reasoning summary so it cannot collide with commentary.
            if ('response.reasoning_summary_text.delta' === $type && isset($event['delta'])) {
                if (null === $currentReasoningSummary) {
                    $currentReasoningSummary = '';
                    yield new ThinkingStart();
                }

                $currentReasoningSummary .= $event['delta'];
                yield new ThinkingDelta($event['delta']);
                continue;
            }

            if ('response.reasoning_summary_text.done' === $type) {
                yield new ThinkingComplete($currentReasoningSummary ?? '');
                $currentReasoningSummary = null;
                continue;
            }

            // commentary and final_answer both arrive as output_text.delta; route by tracked phase.
            // when item_id is absent we keep backward-compatible TextDelta emission rather than dropping the delta.
            //
            // assumption: at most one commentary item is open at a time. The OpenAI Responses API
            // serializes commentary/final_answer per output index, so $currentCommentary does not
            // need to be keyed by item_id. If that changes, key it like $itemPhases to avoid
            // merging deltas from two concurrent commentary items into a single ThinkingComplete.
            if ('response.output_text.delta' === $type && isset($event['delta'])) {
                $itemId = $event['item_id'] ?? null;
                $phase = null !== $itemId ? ($itemPhases[$itemId] ?? null) : null;

                if ('commentary' === $phase) {
                    if (null === $currentCommentary) {
                        $currentCommentary = '';
                        yield new ThinkingStart();
                    }

                    $currentCommentary .= $event['delta'];
                    yield new ThinkingDelta($event['delta']);
                    continue;
                }

                // final_answer, unknown phase, or missing item_id: keep visible TextDelta.
                yield new TextDelta($event['delta']);
                continue;
            }

            if ('response.output_item.done' === $type && isset($event['item']['id'])) {
                $itemId = $event['item']['id'];
                $phase = $itemPhases[$itemId] ?? ($event['item']['phase'] ?? null);

                if ('commentary' === $phase && null !== $currentCommentary) {
                    yield new ThinkingComplete($currentCommentary);
                    $currentCommentary = null;
                }

                unset($itemPhases[$itemId]);
                continue;
            }

            if (!str_contains($type, 'completed')) {
                continue;
            }

            [$toolCallResult] = $this->extractFunctionCalls($event['response'][self::KEY_OUTPUT] ?? []);

            if ($toolCallResult && 'response.completed' === $type) {
                yield new ToolCallComplete(...$toolCallResult->getContent());
            }
        }
    }

<?php

declare(strict_types=1);

namespace App\Tests\AI\OpenAi\Gpt;

use PHPUnit\Framework\Attributes\CoversClass;
use PHPUnit\Framework\TestCase;
use Symfony\AI\Platform\Bridge\OpenAi\Gpt\ResultConverter;
use Symfony\AI\Platform\Exception\ExceptionInterface;
use Symfony\AI\Platform\Result\Stream\Delta\DeltaInterface;
use Symfony\AI\Platform\Result\Stream\Delta\TextDelta;
use Symfony\AI\Platform\Result\Stream\Delta\ThinkingComplete;
use Symfony\AI\Platform\Result\Stream\Delta\ThinkingDelta;
use Symfony\AI\Platform\Result\Stream\Delta\ThinkingStart;
use Symfony\AI\Platform\Result\StreamResult;
use Symfony\Component\HttpClient\Response\MockResponse;

/**
 * Asserts phase-aware routing of OpenAI Responses streaming events.
 *
 * The OpenAI Responses API can emit both commentary and final_answer as
 * "response.output_text.delta" events. They can only be distinguished by the
 * item's phase tracked via "response.output_item.added" with a matching item_id.
 * The converter must not emit commentary as visible text, otherwise it can
 * duplicate output when final_answer repeats the same content.
 *
 * @see https://developers.openai.com/api/reference/resources/responses#(resource)%20conversations%20%3E%20(model)%20message%20%3E%20(schema)%20%3E%20(property)%20phase
 */
#[CoversClass(ResultConverter::class)]
final class ResultConverterStreamTest extends TestCase
{
    /**
     * @throws ExceptionInterface
     */
    public function testCommentaryPhaseIsNotEmittedAsVisibleText(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_commentary_1', 'phase' => 'commentary'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_commentary_1',
                'delta'   => 'Thinking about the problem...',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_commentary_1', 'phase' => 'commentary'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame([], self::extractTextDeltas($deltas), 'Commentary must never surface as TextDelta.');

        $thinkingStarts = array_filter($deltas, static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingStart);
        self::assertCount(1, $thinkingStarts, 'Commentary should open exactly one ThinkingStart.');

        $thinkingDeltaFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingDelta;
        $thinkingDeltas      = array_values(array_filter($deltas, $thinkingDeltaFilter));
        self::assertCount(1, $thinkingDeltas);
        self::assertInstanceOf(ThinkingDelta::class, $thinkingDeltas[0]);
        self::assertSame('Thinking about the problem...', $thinkingDeltas[0]->getThinking());

        $thinkingCompleteFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingComplete;
        $thinkingCompletes      = array_values(array_filter($deltas, $thinkingCompleteFilter));
        self::assertCount(1, $thinkingCompletes, 'Commentary must finalize on response.output_item.done.');
        self::assertInstanceOf(ThinkingComplete::class, $thinkingCompletes[0]);
        self::assertSame('Thinking about the problem...', $thinkingCompletes[0]->getThinking());
        // endregion
    }

    /**
     * @throws ExceptionInterface
     */
    public function testFinalAnswerPhaseStillEmitsTextDelta(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_final_1', 'phase' => 'final_answer'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_final_1',
                'delta'   => 'Hello, ',
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_final_1',
                'delta'   => 'world!',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_final_1', 'phase' => 'final_answer'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame(['Hello, ', 'world!'], self::extractTextDeltas($deltas));

        $thinkingDeltas = array_filter(
            $deltas,
            static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingDelta
                                                       || $delta instanceof ThinkingStart
                                                       || $delta instanceof ThinkingComplete,
        );
        self::assertSame([], array_values($thinkingDeltas), 'Final answer must not emit thinking deltas.');
        // endregion
    }

    /**
     * @throws ExceptionInterface
     */
    public function testUnknownPhaseKeepsBackwardCompatibleTextDelta(): void
    {
        // region setup
        $events    = [
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_unknown',
                'delta'   => 'legacy text',
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame(['legacy text'], self::extractTextDeltas($deltas));
        // endregion
    }

    /**
     * Regression for "delta with no item_id at all". The previous guard
     * (`isset($event['delta'], $event['item_id'])`) silently dropped such deltas.
     * They must instead surface as TextDelta for backward compatibility.
     *
     * @throws ExceptionInterface
     */
    public function testDeltaWithoutItemIdStillEmitsTextDelta(): void
    {
        // region setup
        $events    = [
            [
                'type'  => 'response.output_text.delta',
                'delta' => 'orphan delta',
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame(['orphan delta'], self::extractTextDeltas($deltas));
        // endregion
    }

    /**
     * @throws ExceptionInterface
     */
    public function testMultipleCommentaryDeltasAccumulateIntoThinkingComplete(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_c',
                'delta'   => 'part 1 ',
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_c',
                'delta'   => 'part 2 ',
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_c',
                'delta'   => 'part 3',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame([], self::extractTextDeltas($deltas));

        $thinkingDeltaFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingDelta;
        $thinkingDeltas      = array_values(array_filter($deltas, $thinkingDeltaFilter));
        self::assertCount(3, $thinkingDeltas);

        $thinkingCompleteFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingComplete;
        $thinkingCompletes      = array_values(array_filter($deltas, $thinkingCompleteFilter));
        self::assertCount(1, $thinkingCompletes);
        self::assertInstanceOf(ThinkingComplete::class, $thinkingCompletes[0]);
        self::assertSame('part 1 part 2 part 3', $thinkingCompletes[0]->getThinking());
        // endregion
    }

    /**
     * A commentary item that is opened and closed without any delta in between
     * must not emit a stray ThinkingStart/Complete pair.
     *
     * @throws ExceptionInterface
     */
    public function testCommentaryItemDoneWithoutDeltasEmitsNothing(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame([], $deltas, 'No deltas must be emitted for an empty commentary item.');
        // endregion
    }

    /**
     * Guards against regressing the long-standing reasoning_summary_text.* path.
     *
     * @throws ExceptionInterface
     */
    public function testReasoningSummaryStreamStillEmitsThinkingDeltas(): void
    {
        // region setup
        $events    = [
            [
                'type'  => 'response.reasoning_summary_text.delta',
                'delta' => 'reasoning step 1. ',
            ],
            [
                'type'  => 'response.reasoning_summary_text.delta',
                'delta' => 'reasoning step 2.',
            ],
            [
                'type' => 'response.reasoning_summary_text.done',
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame([], self::extractTextDeltas($deltas));

        $types = array_map(static fn (DeltaInterface $delta): string => $delta::class, $deltas);
        self::assertSame(
            [
                ThinkingStart::class,
                ThinkingDelta::class,
                ThinkingDelta::class,
                ThinkingComplete::class,
            ],
            $types,
        );

        $thinkingCompleteFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingComplete;
        $thinkingCompletes      = array_values(array_filter($deltas, $thinkingCompleteFilter));
        self::assertInstanceOf(ThinkingComplete::class, $thinkingCompletes[0]);
        self::assertSame('reasoning step 1. reasoning step 2.', $thinkingCompletes[0]->getThinking());
        // endregion
    }

    /**
     * @throws ExceptionInterface
     */
    public function testInterleavedCommentaryAndFinalAnswerDoesNotDuplicateText(): void
    {
        // region setup
        // mirrors the real-world bug: commentary and final_answer carry the same content,
        // but only final_answer should be surfaced as visible text.
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_c',
                'delta'   => 'The answer is 42.',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_f', 'phase' => 'final_answer'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_f',
                'delta'   => 'The answer is 42.',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_f', 'phase' => 'final_answer'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame(['The answer is 42.'], self::extractTextDeltas($deltas), 'Visible text must appear exactly once.');
        // endregion
    }

    /**
     * A truly interleaved stream: both items are opened first, then deltas alternate
     * between them. Routing must follow the item_id, not arrival order.
     *
     * @throws ExceptionInterface
     */
    public function testTrulyInterleavedCommentaryAndFinalAnswerAreRoutedByItemId(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_f', 'phase' => 'final_answer'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_c',
                'delta'   => 'thinking...',
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_f',
                'delta'   => 'visible',
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_c', 'phase' => 'commentary'],
            ],
            [
                'type' => 'response.output_item.done',
                'item' => ['id' => 'item_f', 'phase' => 'final_answer'],
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        self::assertSame(['visible'], self::extractTextDeltas($deltas));

        $thinkingDeltaFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingDelta;
        $thinkingDeltas      = array_values(array_filter($deltas, $thinkingDeltaFilter));
        self::assertCount(1, $thinkingDeltas);
        self::assertInstanceOf(ThinkingDelta::class, $thinkingDeltas[0]);
        self::assertSame('thinking...', $thinkingDeltas[0]->getThinking());

        $thinkingCompleteFilter = static fn (DeltaInterface $delta): bool => $delta instanceof ThinkingComplete;
        $thinkingCompletes      = array_values(array_filter($deltas, $thinkingCompleteFilter));
        self::assertCount(1, $thinkingCompletes, 'item_c.done must finalize commentary; item_f.done must not emit a second ThinkingComplete.');
        self::assertInstanceOf(ThinkingComplete::class, $thinkingCompletes[0]);
        self::assertSame('thinking...', $thinkingCompletes[0]->getThinking());
        // endregion
    }

    /**
     * Streams can be terminated (network abort, client disconnect) before the
     * `response.output_item.done` event arrives. Partial commentary must still
     * never surface as visible text, and ThinkingComplete must be absent so the
     * consumer can tell the thinking block was aborted rather than finalized.
     *
     * @throws ExceptionInterface
     */
    public function testCommentaryWithoutDoneDoesNotLeakVisibleText(): void
    {
        // region setup
        $events    = [
            [
                'type' => 'response.output_item.added',
                'item' => ['id' => 'item_commentary_1', 'phase' => 'commentary'],
            ],
            [
                'type'    => 'response.output_text.delta',
                'item_id' => 'item_commentary_1',
                'delta'   => 'partial commentary',
            ],
        ];
        $converter = new ResultConverter();
        // endregion

        // region act
        $deltas = self::collectDeltas($converter, $events);
        // endregion

        // region assert
        $types = array_map(static fn (DeltaInterface $delta): string => $delta::class, $deltas);
        self::assertSame(
            [ThinkingStart::class, ThinkingDelta::class],
            $types,
            'Aborted commentary must emit ThinkingStart + ThinkingDelta but no TextDelta and no ThinkingComplete.',
        );
        // endregion
    }

    /**
     * @param list<array<string, mixed>> $events
     * @throws ExceptionInterface
     * @return list<DeltaInterface>
     */
    private static function collectDeltas(ResultConverter $converter, array $events): array
    {
        $result = $converter->convert(new MockRawStreamResult(new MockResponse(), $events), ['stream' => true]);
        self::assertInstanceOf(StreamResult::class, $result);

        $retValue = [];
        foreach ($result->getContent() as $delta) {
            self::assertInstanceOf(DeltaInterface::class, $delta);
            $retValue[] = $delta;
        }

        return $retValue;
    }

    /**
     * @param list<DeltaInterface> $deltas
     * @return list<string>
     */
    private static function extractTextDeltas(array $deltas): array
    {
        $retValue = [];
        foreach ($deltas as $delta) {
            if ($delta instanceof TextDelta) {
                $retValue[] = $delta->getText();
            }
        }

        return $retValue;
    }
}

<?php

declare(strict_types=1);

namespace App\Tests\AI\OpenAi\Gpt;

use Generator;
use LogicException;
use Symfony\AI\Platform\Result\RawResultInterface;
use Symfony\Contracts\HttpClient\ResponseInterface;

/**
 * Minimal test double for streaming tests that bypass HTTP/SSE transport.
 *
 * Feed it pre-parsed event arrays as they would arrive from
 * EventSourceHttpClient::getDataStream(). The wrapped ResponseInterface is
 * only used by ResultConverter::convert() to fetch the HTTP status code
 * before dispatching to the stream path.
 */
final readonly class MockRawStreamResult implements RawResultInterface
{
    /**
     * @param list<array<string, mixed>> $events
     */
    public function __construct(
        private ResponseInterface $response,
        private array $events
    ) {
    }

    public function getObject(): ResponseInterface
    {
        return $this->response;
    }

    /**
     * Stream tests never inspect the non-stream payload; failing loudly if they
     * do surfaces accidental misuse faster than returning a silent empty array.
     *
     * @return array<string, mixed>
     */
    public function getData(): array
    {
        throw new LogicException('MockRawStreamResult::getData() is not supported in stream tests.');
    }

    /**
     * @return Generator<array<string, mixed>>
     */
    public function getDataStream(): Generator
    {
        foreach ($this->events as $event) {
            yield $event;
        }
    }
}

[OskarStark]

Can you provide a PR? Thanks

[Amoifr]

Hi @wimwinterberg @OskarStark! Opened #2006 with the fix.

The PR routes commentary and final_answer independently using the per-item_id phase tracking you sketched in this thread, @wimwinterberg. Backward-compat is preserved (deltas without an item_id keep emitting TextDelta), and the reasoning_summary_text.* path got its own buffer so it can no longer collide with commentary tracking. Ships with 9 unit tests, including the real-world duplicate scenario and aborted-stream behavior.

Hopefully this addresses the original report — happy to iterate on review feedback. 🙏