* Creates a ReadableStream from a deep research streaming interaction. * * Deep research streaming returns InteractionSSEEvent chunks including: * - interaction.start: initial interaction with ID * - content.delta: incremental text and thought_summary updates * - content.start / content.stop: o
(
stream: AsyncIterable<Interactions.InteractionSSEEvent>,
onComplete?: (
content: string,
usage: {
inputTokens: number
outputTokens: number
reasoningTokens: number
totalTokens: number
},
interactionId?: string
) => void
)
| 578 | * We stream text deltas to the client and track usage from the interaction.complete event. |
| 579 | */ |
| 580 | function createDeepResearchStream( |
| 581 | stream: AsyncIterable<Interactions.InteractionSSEEvent>, |
| 582 | onComplete?: ( |
| 583 | content: string, |
| 584 | usage: { |
| 585 | inputTokens: number |
| 586 | outputTokens: number |
| 587 | reasoningTokens: number |
| 588 | totalTokens: number |
| 589 | }, |
| 590 | interactionId?: string |
| 591 | ) => void |
| 592 | ): ReadableStream<Uint8Array> { |
| 593 | const streamLogger = createLogger('DeepResearchStream') |
| 594 | let fullContent = '' |
| 595 | let completionUsage = { inputTokens: 0, outputTokens: 0, reasoningTokens: 0, totalTokens: 0 } |
| 596 | let completedInteractionId: string | undefined |
| 597 | |
| 598 | return new ReadableStream({ |
| 599 | async start(controller) { |
| 600 | try { |
| 601 | for await (const event of stream) { |
| 602 | if (event.event_type === 'content.delta') { |
| 603 | const delta = (event as Interactions.ContentDelta).delta |
| 604 | if (delta?.type === 'text' && 'text' in delta && delta.text) { |
| 605 | fullContent += delta.text |
| 606 | controller.enqueue(new TextEncoder().encode(delta.text)) |
| 607 | } |
| 608 | } else if (event.event_type === 'interaction.complete') { |
| 609 | const interaction = (event as Interactions.InteractionEvent).interaction |
| 610 | if (interaction?.usage) { |
| 611 | completionUsage = extractInteractionUsage(interaction.usage) |
| 612 | } |
| 613 | completedInteractionId = interaction?.id |
| 614 | } else if (event.event_type === 'interaction.start') { |
| 615 | const interaction = (event as Interactions.InteractionEvent).interaction |
| 616 | if (interaction?.id) { |
| 617 | completedInteractionId = interaction.id |
| 618 | } |
| 619 | } else if (event.event_type === 'error') { |
| 620 | const errorEvent = event as { error?: { code?: string; message?: string } } |
| 621 | const message = errorEvent.error?.message ?? 'Unknown deep research stream error' |
| 622 | streamLogger.error('Deep research stream error', { |
| 623 | code: errorEvent.error?.code, |
| 624 | message, |
| 625 | }) |
| 626 | controller.error(new Error(message)) |
| 627 | return |
| 628 | } |
| 629 | } |
| 630 | |
| 631 | onComplete?.(fullContent, completionUsage, completedInteractionId) |
| 632 | controller.close() |
| 633 | } catch (error) { |
| 634 | streamLogger.error('Error reading deep research stream', { |
| 635 | error: toError(error).message, |
| 636 | }) |
| 637 | controller.error(error) |
no test coverage detected