Method stream

backend/src/workflows/workflows.controller.ts:863–1116

(
    @Param('runId') runId: string,
    @Query(new ZodValidationPipe(StreamRunQuerySchema)) query: StreamRunQueryDto,
    @CurrentAuth() auth: AuthContext | null,
    @Res() res: Response,
    @Req() req: Request,
  )

  @Get('/runs/:runId/stream')
  @ApiOkResponse({ description: 'Server-sent events stream for workflow run updates' })
  async stream(
    @Param('runId') runId: string,
    @Query(new ZodValidationPipe(StreamRunQuerySchema)) query: StreamRunQueryDto,
    @CurrentAuth() auth: AuthContext | null,
    @Res() res: Response,
    @Req() req: Request,
  ): Promise<void> {
    // Auth is now handled via headers (Authorization and X-Organization-Id)
    // using a fetch-based SSE client that supports custom headers
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    if (typeof (res as any).flushHeaders === 'function') {
      (res as any).flushHeaders();
    }

    await this.workflowsService.ensureRunAccess(runId, auth);

    let lastSequence = Number.parseInt(query.cursor ?? '0', 10);
    let terminalCursor = query.terminalCursor;
    let lastLogCursor = query.logCursor ?? null;
    if (Number.isNaN(lastSequence) || lastSequence < 0) {
      lastSequence = 0;
    }

    let active = true;
    let lastStatusSignature: string | null = null;
    let intervalId: NodeJS.Timeout | undefined;
    // eslint-disable-next-line prefer-const -- initialized later after cleanup function is defined
    let heartbeatId: NodeJS.Timeout | undefined;
    let earliestEventTimestamp: number | null = null;
    let latestEventTimestamp: number | null = null;

    const send = (event: string, payload: unknown) => {
      if (!active) {
        return;
      }
      try {
        res.write(`event: ${event}\n`);
        res.write(`data: ${JSON.stringify(payload)}\n\n`);
        // Flush headers if available (helps with immediate delivery)
        if (typeof (res as any).flush === 'function') {
          (res as any).flush();
        }
      } catch (error) {
        // Connection closed or error writing
        console.warn(`Failed to send SSE event ${event}:`, error);
        if (!active) {
          return;
        }
        active = false;
        void cleanup();
      }
    };

    const cleanup = async () => {
      if (!active) {
        return;