Function start

apps/sim/lib/workflows/streaming/streaming.ts:364–629

(controller)

  return new ReadableStream({
    async start(controller) {
      const state: StreamingState = {
        streamedChunks: new Map(),
        processedOutputs: new Set(),
        streamCompletionTimes: new Map(),
        completedBlockIds: new Set(),
        selectedOutputBytes: 0,
        streamedSelectedOutputKeys: new Set(),
      }

      const sendChunk = (
        blockId: string,
        content: string,
        options: { selectedOutputKey?: string; selectedOutputBytes?: number } = {}
      ) => {
        const separator = state.processedOutputs.size > 0 ? '\n\n' : ''
        const chunk = separator + content
        if (options.selectedOutputKey) {
          const selectedOutputBytes =
            options.selectedOutputBytes ?? Buffer.byteLength(chunk, 'utf8')
          const nextSelectedOutputBytes = state.selectedOutputBytes + selectedOutputBytes
          assertInlineMaterializationSize(nextSelectedOutputBytes, MAX_INLINE_MATERIALIZATION_BYTES)
          state.selectedOutputBytes = nextSelectedOutputBytes
          state.streamedSelectedOutputKeys.add(options.selectedOutputKey)
        }
        controller.enqueue(encodeSSE({ blockId, chunk }))
        state.processedOutputs.add(blockId)
      }

      /**
       * Callback for handling streaming execution events.
       */
      const onStreamCallback = async (streamingExec: StreamingExecutionWithBlockId) => {
        const blockId = streamingExec.execution?.blockId
        if (!blockId) {
          logger.warn(`[${requestId}] Streaming execution missing blockId`)
          return
        }

        const reader = streamingExec.stream.getReader()
        const decoder = new TextDecoder()
        let isFirstChunk = true

        try {
          while (true) {
            const { done, value } = await reader.read()
            if (done) {
              state.streamCompletionTimes.set(blockId, Date.now())
              break
            }

            const textChunk = decoder.decode(value, { stream: true })
            if (!state.streamedChunks.has(blockId)) {
              state.streamedChunks.set(blockId, [])
            }
            state.streamedChunks.get(blockId)!.push(textChunk)

            if (isFirstChunk) {