Function onBlockCompleteCallback

apps/sim/lib/workflows/streaming/streaming.ts:443–534

(blockId: string, output: unknown)

      const base64MaxBytes = streamConfig.base64MaxBytes

      const onBlockCompleteCallback = async (blockId: string, output: unknown) => {
        state.completedBlockIds.add(blockId)

        if (!streamConfig.selectedOutputs?.length) {
          return
        }

        if (state.streamedChunks.has(blockId)) {
          return
        }

        const matchingOutputs = getSelectedOutputDescriptors(streamConfig.selectedOutputs).filter(
          (descriptor) => descriptor.blockId === blockId
        )

        for (const descriptor of matchingOutputs) {
          if (state.selectedOutputError) {
            break
          }
          try {
            const remainingBytes = getRemainingSelectedOutputBytes(state.selectedOutputBytes)
            const extractionContext = {
              requestId,
              workspaceId: options.workspaceId,
              workflowId: options.workflowId,
              executionId,
              largeValueExecutionIds: options.largeValueExecutionIds,
              largeValueKeys: options.largeValueKeys,
              fileKeys: options.fileKeys,
              allowLargeValueWorkflowScope: options.allowLargeValueWorkflowScope,
              userId: options.userId,
              base64MaxBytes: Math.min(
                base64MaxBytes ?? MAX_INLINE_MATERIALIZATION_BYTES,
                getBase64DecodedByteBudget(remainingBytes)
              ),
            }
            const materializationContext = buildMaterializationContext(extractionContext)
            const outputValue = await extractOutputValue(output, descriptor.path, extractionContext)

            if (outputValue !== undefined) {
              const materializedOutput = await materializeInlineExecutionValue(
                outputValue,
                materializationContext,
                { maxBytes: remainingBytes }
              )
              const shouldHydrateOutput = includeFileBase64
              const hydratedOutput = shouldHydrateOutput
                ? await hydrateUserFilesWithBase64(materializedOutput, {
                    requestId,
                    ...materializationContext,
                    maxBytes: Math.min(
                      base64MaxBytes ?? MAX_INLINE_MATERIALIZATION_BYTES,
                      getBase64DecodedByteBudget(remainingBytes)
                    ),
                    preserveLargeValueMetadata: true,
                  })
                : materializedOutput
              await materializeInlineExecutionValue(hydratedOutput, materializationContext, {