Function start

apps/sim/app/api/workflows/[id]/execute/route.ts:1224–1775

(controller)

    const stream = new ReadableStream<Uint8Array>({
      async start(controller) {
        let finalMetaStatus: 'complete' | 'error' | 'cancelled' | null = null

        registerManualExecutionAborter(executionId, timeoutController.abort)
        isManualAbortRegistered = true

        let terminalEventPublished = false
        const sendEvent = async (
          event: ExecutionEvent,
          terminalStatus?: TerminalExecutionStreamStatus
        ) => {
          const isBuffered = event.type !== 'stream:chunk' && event.type !== 'stream:done'
          let eventToSend = event
          if (isBuffered) {
            try {
              const entry = terminalStatus
                ? await eventWriter.writeTerminal(event, terminalStatus)
                : await eventWriter.write(event)
              eventToSend = entry.event
              eventToSend.eventId = entry.eventId
              terminalEventPublished ||= Boolean(terminalStatus)
            } catch (e) {
              // The event buffer (Redis replay store) rejected this event — e.g. the flush
              // batch exceeds the per-write byte cap for large block outputs. The buffer only
              // backs reconnect/replay; the live SSE stream is the primary delivery. Fall
              // through to enqueue the event live (below) instead of throwing, so terminal
              // events still reach the active client and the UI doesn't hang on "running".
              // Marking a terminal event delivered-live as published lets finalization close
              // the stream cleanly instead of aborting it with controller.error().
              reqLogger.warn('Event buffer write failed; delivering event over live stream only', {
                eventType: event.type,
                terminal: Boolean(terminalStatus),
                error: toError(e).message,
              })
              terminalEventPublished ||= Boolean(terminalStatus)
            }
          }
          if (!isStreamClosed) {
            try {
              controller.enqueue(encodeSSEEvent(eventToSend))
            } catch {
              isStreamClosed = true
            }
          }
        }

        try {
          const startTime = new Date()

          await sendEvent({
            type: 'execution:started',
            timestamp: startTime.toISOString(),
            executionId,
            workflowId,
            data: {
              startTime: startTime.toISOString(),
            },
          })