Function createSSEStream

apps/sim/lib/copilot/request/lifecycle/start.ts:76–409

(params: StreamingOrchestrationParams)

}

export function createSSEStream(params: StreamingOrchestrationParams): ReadableStream {
  const {
    requestPayload,
    userId,
    streamId,
    executionId,
    runId,
    chatId,
    currentChat,
    isNewChat,
    message,
    titleModel,
    titleProvider,
    requestId,
    workspaceId,
    orchestrateOptions,
    otelRoot,
  } = params

  // Reuse caller's root if provided; otherwise start our own.
  const activeOtelRoot =
    otelRoot ??
    startCopilotOtelRoot({
      requestId,
      route: orchestrateOptions.goRoute,
      chatId,
      workflowId: orchestrateOptions.workflowId,
      executionId,
      runId,
      streamId,
      transport: CopilotTransport.Stream,
    })

  const abortController = new AbortController()
  registerActiveStream(streamId, abortController)

  const publisher = new StreamWriter({ streamId, chatId, requestId })

  // Classify cancel: signal.reason (explicit-stop set) wins, then
  // clientDisconnected, else Unknown (latent contract bug — log it).
  const recordCancelled = (errorMessage?: string): CopilotRequestCancelReasonValue => {
    const rawReason = abortController.signal.reason
    let cancelReason: CopilotRequestCancelReasonValue
    if (isExplicitStopReason(rawReason)) {
      cancelReason = CopilotRequestCancelReason.ExplicitStop
    } else if (publisher.clientDisconnected) {
      cancelReason = CopilotRequestCancelReason.ClientDisconnect
    } else {
      cancelReason = CopilotRequestCancelReason.Unknown
      const serializedReason =
        rawReason === undefined
          ? 'undefined'
          : rawReason instanceof Error
            ? `${rawReason.name}: ${rawReason.message}`
            : typeof rawReason === 'string'
              ? rawReason
              : (() => {
                  try {