Method chat

backend/src/agents/agents.controller.ts:75–129

(
    @Param('agentRunId') agentRunId: string,
    @Body(new ZodValidationPipe(AgentChatRequestSchema)) body: AgentChatRequestDto,
    @CurrentAuth() auth: AuthContext | null,
    @Res() res: Response,
    @Req() req: Request,
  )

  @Post('/:agentRunId/chat')
  @ApiOkResponse({ description: 'AI SDK-compatible SSE for agent run' })
  async chat(
    @Param('agentRunId') agentRunId: string,
    @Body(new ZodValidationPipe(AgentChatRequestSchema)) body: AgentChatRequestDto,
    @CurrentAuth() auth: AuthContext | null,
    @Res() res: Response,
    @Req() req: Request,
  ): Promise<void> {
    const metadata = await this.agentTraceService.getRunMetadata(agentRunId);
    if (!metadata) {
      throw new NotFoundException(`Agent run ${agentRunId} not found`);
    }
    await this.workflowsService.ensureRunAccess(metadata.workflowRunId, auth);

    let lastSequence = typeof body?.cursor === 'number' ? body.cursor : 0;
    let seenFinish = false;
    let aborted = false;

    req.on('close', () => {
      aborted = true;
    });

    const stream = createUIMessageStream({
      execute: async ({ writer }) => {
        while (!seenFinish && !aborted) {
          const events = await this.agentTraceService.list(agentRunId, lastSequence);
          if (events.length > 0) {
            events.forEach((event) => {
              const chunk = convertAgentTraceToUiChunk(event);
              if (chunk) {
                writer.write(chunk);
                if (chunk.type === 'finish') {
                  seenFinish = true;
                }
              }
            });
            lastSequence = events[events.length - 1]?.sequence ?? lastSequence;
            continue;
          }
          await sleep(1000);
        }
      },
      onError: (error) => {
        this.logger.error(
          `Agent chat stream failed for agent ${agentRunId}`,
          error instanceof Error ? error.stack : String(error),
        );
        return error instanceof Error ? error.message : String(error);
      },
    });

    pipeUIMessageStreamToResponse({
      response: res,
      stream,
    });
  }
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));