feat(ai): surface AI chat stream failures through one typed error channel (#22434)

## Context Investigating a report where the AI chat showed only a `...` spinner while the network response clearly contained `No AI models are available`. Root cause: terminal stream failures reach the client on **two mismatched channels**. | Representation | Persisted (survives reload) | Rendered by client | |---|---|---| | AI-SDK `error` chunk (inside `stream-chunk`) | ✅ RPUSH'd to Redis | ❌ dropped by `readUIMessageStream` (no message part, no error state) | | typed `stream-error` event | ❌ never persisted | ✅ sets the error atom | Live, the `stream-error` event renders. But on reload, `chatStreamCatchupChunks` replays only the persisted **error chunk** — which the reducer discards — and the streaming indicator never clears. ## Change Collapse to a single typed error contract: - **Suppress the opaque `error` chunk** in the stream job; every failure is surfaced through the typed `stream-error` event. Errors are mapped via `mapErrorToStreamError` so an `AiException` keeps its `AiExceptionCode` (e.g. `API_KEY_NOT_CONFIGURED` → the existing "AI not configured" banner) instead of leaking a raw string. - **Persist the terminal error** next to the accumulated chunks and expose it as an explicit `error { code message }` field on `ChatStreamCatchupChunks`, so a client catching up after a reload recovers it — no dependency on the AI SDK's internal chunk shape. - **Reset per-thread stream state at job start**, so a failed turn's leftover chunks/error never replay on the next stream. - **Client replays the catchup error** as a terminal `stream-error` event, which clears the streaming indicator and renders the error (fixes the infinite spinner on a stream that ended in error). ## Notes - `ChatStreamError` is a new metadata GraphQL type; generated types (twenty-front metadata + client-sdk) were hand-updated to keep the tree consistent and will be reconciled by CI's `graphql:generate` check if anything differs. - Server unit test added for the error mapping. No schema/DB migration. ## Test plan - [ ] With no AI provider configured, send a chat message → error renders immediately (not a spinner). - [ ] Reload the thread → the error still renders (recovered from catchup), indicator not spinning. - [ ] Configure a provider and send again → normal streaming; no stale error from the previous failed turn.  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22434?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-02 14:50:57 +02:00
parent 47689e676b
commit d709467902
30 changed files with 965 additions and 243 deletions
@@ -2,6 +2,8 @@ import { Field, Int, ObjectType } from '@nestjs/graphql';

 import GraphQLJSON from 'graphql-type-json';

+import { ChatStreamErrorDTO } from 'src/engine/metadata-modules/ai/ai-chat/dtos/chat-stream-error.dto';
+
@ObjectType('ChatStreamCatchupChunks')
 export class ChatStreamCatchupChunksDTO {
  @Field(() => [GraphQLJSON])
@@ -9,4 +11,7 @@ export class ChatStreamCatchupChunksDTO {

  @Field(() => Int)
  maxSeq: number;
+
+  @Field(() => ChatStreamErrorDTO, { nullable: true })
+  error: ChatStreamErrorDTO | null;
 }
@@ -0,0 +1,10 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+
+@ObjectType('ChatStreamError')
+export class ChatStreamErrorDTO {
+  @Field(() => String)
+  code: string;
+
+  @Field(() => String)
+  message: string;
+}
@@ -10,9 +10,12 @@ import {
  UpdateDateColumn,
 } from 'typeorm';

+import { ADD_LAST_STREAM_ERROR_TO_AGENT_CHAT_THREAD_UPGRADE_COMMAND_NAME } from 'src/database/commands/upgrade-version-command/2-19/add-last-stream-error-to-agent-chat-thread-upgrade-command-name.constant';
+import { WasIntroducedInUpgrade } from 'src/engine/core-modules/upgrade/decorators/was-introduced-in-upgrade.decorator';
 import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
 import { AgentMessageEntity } from 'src/engine/metadata-modules/ai/ai-agent-execution/entities/agent-message.entity';
 import { AgentTurnEntity } from 'src/engine/metadata-modules/ai/ai-agent-execution/entities/agent-turn.entity';
+import { type AgentChatThreadLastStreamError } from 'src/engine/metadata-modules/ai/ai-chat/types/agent-chat-thread-last-stream-error.type';
 import type { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { EntityRelation } from 'src/engine/workspace-manager/workspace-migration/types/entity-relation.interface';

@@ -70,6 +73,13 @@ export class AgentChatThreadEntity {
  @Column({ type: 'varchar', nullable: true })
  activeStreamId: string | null;

+  @WasIntroducedInUpgrade({
+    upgradeCommandName:
+      ADD_LAST_STREAM_ERROR_TO_AGENT_CHAT_THREAD_UPGRADE_COMMAND_NAME,
+  })
+  @Column({ type: 'jsonb', nullable: true })
+  lastStreamError: AgentChatThreadLastStreamError | null;
+
  @OneToMany(() => AgentTurnEntity, (turn) => turn.thread)
  turns: EntityRelation<AgentTurnEntity[]>;

@@ -28,6 +28,7 @@ import { AgentChatStreamingService } from 'src/engine/metadata-modules/ai/ai-cha
 import { AgentChatService } from 'src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service';
 import { ChatExecutionService } from 'src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service';
 import { getCancelChannel } from 'src/engine/metadata-modules/ai/ai-chat/utils/get-cancel-channel.util';
+import { mapErrorToStreamError } from 'src/engine/metadata-modules/ai/ai-chat/utils/map-error-to-stream-error.util';
 import type { AiModelConfig } from 'src/engine/metadata-modules/ai/ai-models/types/ai-model-config.type';
 import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator';
 import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
@@ -55,6 +56,8 @@ export class StreamAgentChatJob {

  @Process(STREAM_AGENT_CHAT_JOB_NAME)
  async handle(data: StreamAgentChatJobData): Promise<void> {
+    await this.eventPublisherService.resetStreamState(data.threadId);
+
    const workspace = await this.workspaceRepository.findOne({
      where: { id: data.workspaceId },
    });
@@ -87,17 +90,33 @@ export class StreamAgentChatJob {
      this.logger.error(
        `Stream ${data.streamId} failed: ${error instanceof Error ? error.message : String(error)}`,
      );
+      const streamError = mapErrorToStreamError(error);
+
+      await this.threadRepository
+        .update(
+          data.workspaceId,
+          { id: data.threadId },
+          {
+            lastStreamError: {
+              ...streamError,
+              failedAt: new Date().toISOString(),
+            },
+          },
+        )
+        .catch((persistError) => {
+          this.logger.error(
+            `Failed to persist stream error for thread ${data.threadId}: ${persistError instanceof Error ? persistError.message : String(persistError)}`,
+          );
+        });
+
      await this.eventPublisherService
        .publish({
          threadId: data.threadId,
          workspaceId: data.workspaceId,
          event: {
            type: 'stream-error',
-            code: 'STREAM_EXECUTION_FAILED',
-            message:
-              error instanceof Error
-                ? error.message
-                : 'Stream execution failed',
+            code: streamError.code,
+            message: streamError.message,
          },
        })
        .catch(() => {});
@@ -197,6 +216,7 @@ export class StreamAgentChatJob {
      let lastStepConversationSize = 0;
      let totalCacheCreationTokens = 0;
      let streamError: unknown;
+      let streamFinishError: unknown;
      let checkHasNoMoreAvailableCredits: () => boolean = () => false;

      // onFinish fires before the uiStream is fully drained. We use this
@@ -282,6 +302,7 @@ export class StreamAgentChatJob {
                });
              },
              onFinish: async ({ responseMessage, isAborted }) => {
+                // Rejecting here would race chunks still draining.
                try {
                  await this.handleStreamFinish({
                    assistantMessageId,
@@ -299,15 +320,23 @@ export class StreamAgentChatJob {
                    userMessagePromise,
                  });
                  await titleWritePromise;
-                  resolveStreamFinished();
                } catch (error) {
-                  reject(error);
+                  streamFinishError = error;
+                } finally {
+                  resolveStreamFinished();
                }
              },
              sendReasoning: true,
            }),
          );
        },
+        // Errors thrown before the model stream merges never reach onFinish.
+        onError: (error) => {
+          streamError = error;
+          resolveStreamFinished();
+
+          return error instanceof Error ? error.message : String(error);
+        },
      });

      // Publish all chunks first, then signal completion. This guarantees
@@ -315,6 +344,10 @@ export class StreamAgentChatJob {
      void (async () => {
        try {
          for await (const chunk of uiStream) {
+            if ((chunk as { type?: string }).type === 'error') {
+              continue;
+            }
+
            await this.eventPublisherService.publish({
              threadId: data.threadId,
              workspaceId: data.workspaceId,
@@ -329,6 +362,8 @@ export class StreamAgentChatJob {

          if (streamError) {
            reject(streamError);
+          } else if (streamFinishError) {
+            reject(streamFinishError);
          } else if (checkHasNoMoreAvailableCredits()) {
            await this.eventPublisherService.publish({
              threadId: data.threadId,
@@ -545,6 +580,7 @@ export class StreamAgentChatJob {
          `"totalCacheCreationTokens" + ${totalCacheCreationTokens}`,
        contextWindowTokens: modelConfig.contextWindowTokens,
        conversationSize: lastStepConversationSize,
+        lastStreamError: null,
      },
    );

@@ -102,13 +102,25 @@ export class AgentChatResolver {
    @AuthUserWorkspaceId() userWorkspaceId: string,
    @AuthWorkspace() { id: workspaceId }: WorkspaceEntity,
  ) {
-    await this.agentChatService.getThreadById({
+    const thread = await this.agentChatService.getThreadById({
      threadId,
      userWorkspaceId,
      workspaceId,
    });

-    return this.eventPublisherService.getAccumulatedChunks(threadId);
+    const { chunks, maxSeq } =
+      await this.eventPublisherService.getAccumulatedChunks(threadId);
+
+    return {
+      chunks,
+      maxSeq,
+      error: thread.lastStreamError
+        ? {
+            code: thread.lastStreamError.code,
+            message: thread.lastStreamError.message,
+          }
+        : null,
+    };
  }

  @Mutation(() => AgentChatThreadDTO)
@@ -211,6 +223,42 @@ export class AgentChatResolver {
    };
  }

+  @Mutation(() => SendChatMessageResultDTO)
+  async retryChatMessage(
+    @Args('threadId', { type: () => UUIDScalarType }) threadId: string,
+    @Args('modelId', { type: () => String, nullable: true })
+    modelId: string | undefined,
+    @AuthUserWorkspaceId() userWorkspaceId: string,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ): Promise<SendChatMessageResultDTO> {
+    if (this.aiModelRegistryService.getAvailableModels().length === 0) {
+      throw new AiException(
+        'No AI models are available. Configure at least one AI provider.',
+        AiExceptionCode.API_KEY_NOT_CONFIGURED,
+      );
+    }
+
+    this.aiModelRegistryService.validateModelAvailability(
+      modelId ?? workspace.smartModel,
+      workspace,
+    );
+
+    await this.billingUsageService.hasAvailableCreditsOrThrow(workspace.id);
+
+    const result = await this.agentChatStreamingService.retryLastFailedTurn({
+      threadId,
+      userWorkspaceId,
+      workspace,
+      modelId,
+    });
+
+    return {
+      messageId: result.messageId,
+      queued: false,
+      streamId: result.streamId,
+    };
+  }
+
  @Mutation(() => Boolean)
  async stopAgentChatStream(
    @Args('threadId', { type: () => UUIDScalarType }) threadId: string,
@@ -0,0 +1,142 @@
+import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import {
+  AgentMessageRole,
+  AgentMessageStatus,
+  type AgentMessageEntity,
+} from 'src/engine/metadata-modules/ai/ai-agent-execution/entities/agent-message.entity';
+import { type AgentChatThreadEntity } from 'src/engine/metadata-modules/ai/ai-chat/entities/agent-chat-thread.entity';
+import { AgentChatStreamingService } from 'src/engine/metadata-modules/ai/ai-chat/services/agent-chat-streaming.service';
+import { AiExceptionCode } from 'src/engine/metadata-modules/ai/ai.exception';
+
+describe('AgentChatStreamingService.retryLastFailedTurn', () => {
+  const workspace = { id: 'workspace-id' } as WorkspaceEntity;
+
+  const failedThread = {
+    id: 'thread-id',
+    title: 'Thread title',
+    conversationSize: 42,
+    activeStreamId: null,
+    lastStreamError: {
+      code: 'STREAM_EXECUTION_FAILED',
+      message: 'Provider timed out',
+      failedAt: '2026-01-01T00:00:00.000Z',
+    },
+  } as unknown as AgentChatThreadEntity;
+
+  const userMessageEntity = {
+    id: 'user-message-id',
+    role: AgentMessageRole.USER,
+    status: AgentMessageStatus.SENT,
+    parts: [{ type: 'text', textContent: 'hello', orderIndex: 0 }],
+    createdAt: new Date('2026-01-01T00:00:00.000Z'),
+  } as unknown as AgentMessageEntity;
+
+  const buildService = ({
+    thread = failedThread,
+    lastUserMessage = { id: 'user-message-id', turnId: 'turn-id' },
+    threadMessages = [userMessageEntity],
+  } = {}) => {
+    const threadRepository = {
+      findOne: jest.fn().mockResolvedValue(thread),
+      update: jest.fn().mockResolvedValue(undefined),
+    };
+    const messageQueueService = { add: jest.fn().mockResolvedValue(undefined) };
+    const agentChatService = {
+      findLatestSentUserMessage: jest.fn().mockResolvedValue(lastUserMessage),
+      deleteAssistantMessagesForTurn: jest.fn().mockResolvedValue(undefined),
+      getMessagesForThread: jest.fn().mockResolvedValue(threadMessages),
+    };
+
+    const service = new AgentChatStreamingService(
+      threadRepository as never,
+      { find: jest.fn() } as never,
+      messageQueueService as never,
+      agentChatService as never,
+      { publish: jest.fn() } as never,
+      { signFileByIdUrl: jest.fn() } as never,
+    );
+
+    return { service, threadRepository, messageQueueService, agentChatService };
+  };
+
+  const retryArguments = {
+    threadId: 'thread-id',
+    userWorkspaceId: 'user-workspace-id',
+    workspace,
+  };
+
+  it('rejects when the thread has no persisted stream error', async () => {
+    const { service, messageQueueService } = buildService({
+      thread: { ...failedThread, lastStreamError: null },
+    });
+
+    await expect(
+      service.retryLastFailedTurn(retryArguments),
+    ).rejects.toMatchObject({
+      code: AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+    });
+    expect(messageQueueService.add).not.toHaveBeenCalled();
+  });
+
+  it('rejects when a stream is already active', async () => {
+    const { service, messageQueueService } = buildService({
+      thread: { ...failedThread, activeStreamId: 'stream-id' },
+    });
+
+    await expect(
+      service.retryLastFailedTurn(retryArguments),
+    ).rejects.toMatchObject({
+      code: AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+    });
+    expect(messageQueueService.add).not.toHaveBeenCalled();
+  });
+
+  it('rejects without clearing state when a newer message exists', async () => {
+    const newerAssistantMessage = {
+      ...userMessageEntity,
+      id: 'newer-message-id',
+      role: AgentMessageRole.ASSISTANT,
+    } as unknown as AgentMessageEntity;
+    const { service, threadRepository } = buildService({
+      threadMessages: [userMessageEntity, newerAssistantMessage],
+    });
+
+    await expect(
+      service.retryLastFailedTurn(retryArguments),
+    ).rejects.toMatchObject({
+      code: AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+    });
+    expect(threadRepository.update).not.toHaveBeenCalled();
+  });
+
+  it('drops the failed output, re-enqueues the turn, and clears the error', async () => {
+    const { service, threadRepository, messageQueueService, agentChatService } =
+      buildService();
+
+    const result = await service.retryLastFailedTurn({
+      ...retryArguments,
+      modelId: 'model-id',
+    });
+
+    expect(
+      agentChatService.deleteAssistantMessagesForTurn,
+    ).toHaveBeenCalledWith({ turnId: 'turn-id', workspaceId: 'workspace-id' });
+    expect(messageQueueService.add).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.objectContaining({
+        threadId: 'thread-id',
+        existingTurnId: 'turn-id',
+        lastUserMessageText: 'hello',
+        modelId: 'model-id',
+        hasTitle: true,
+        conversationSizeTokens: 42,
+      }),
+    );
+    expect(threadRepository.update).toHaveBeenCalledWith(
+      'workspace-id',
+      { id: 'thread-id' },
+      { activeStreamId: result.streamId, lastStreamError: null },
+    );
+    expect(result.messageId).toBe('user-message-id');
+  });
+});
@@ -55,6 +55,12 @@ export class AgentChatEventPublisherService {
    });
  }

+  async resetStreamState(threadId: string): Promise<void> {
+    const redis = this.redisClientService.getClient();
+
+    await redis.del(this.getStreamChunksKey(threadId));
+  }
+
  async getAccumulatedChunks(threadId: string): Promise<{
    chunks: Record<string, unknown>[];
    maxSeq: number;
@@ -7,6 +7,7 @@ import {
  isExtendedFileUIPart,
 } from 'twenty-shared/ai';
 import { FileFolder } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
 import { In, Like } from 'typeorm';

 import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
@@ -139,12 +140,108 @@ export class AgentChatStreamingService {
    await this.threadRepository.update(
      workspace.id,
      { id: thread.id },
-      { activeStreamId: streamId },
+      { activeStreamId: streamId, lastStreamError: null },
    );

    return { streamId, messageId: savedUserMessage.id };
  }

+  async retryLastFailedTurn({
+    threadId,
+    userWorkspaceId,
+    workspace,
+    modelId,
+  }: {
+    threadId: string;
+    userWorkspaceId: string;
+    workspace: WorkspaceEntity;
+    modelId?: string;
+  }): Promise<{ streamId: string; messageId: string }> {
+    const thread = await this.threadRepository.findOne(workspace.id, {
+      where: { id: threadId, userWorkspaceId },
+    });
+
+    if (!thread) {
+      throw new AiException(
+        'Thread not found',
+        AiExceptionCode.THREAD_NOT_FOUND,
+      );
+    }
+
+    if (
+      !isDefined(thread.lastStreamError) ||
+      isDefined(thread.activeStreamId)
+    ) {
+      throw new AiException(
+        'There is no failed turn to retry on this thread',
+        AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+      );
+    }
+
+    const lastUserMessage =
+      await this.agentChatService.findLatestSentUserMessage({
+        threadId,
+        workspaceId: workspace.id,
+      });
+
+    if (!isDefined(lastUserMessage) || !isDefined(lastUserMessage.turnId)) {
+      throw new AiException(
+        'There is no failed turn to retry on this thread',
+        AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+      );
+    }
+
+    await this.agentChatService.deleteAssistantMessagesForTurn({
+      turnId: lastUserMessage.turnId,
+      workspaceId: workspace.id,
+    });
+
+    const messages = await this.loadMessagesFromDB(
+      threadId,
+      userWorkspaceId,
+      workspace.id,
+    );
+
+    const retriedMessage = messages[messages.length - 1];
+
+    if (!retriedMessage || retriedMessage.id !== lastUserMessage.id) {
+      throw new AiException(
+        'There is no failed turn to retry on this thread',
+        AiExceptionCode.NO_FAILED_TURN_TO_RETRY,
+      );
+    }
+
+    const textPart = retriedMessage.parts.find((part) => part.type === 'text');
+
+    const streamId = generateId();
+
+    await this.messageQueueService.add<StreamAgentChatJobData>(
+      STREAM_AGENT_CHAT_JOB_NAME,
+      {
+        threadId,
+        streamId,
+        userWorkspaceId,
+        workspaceId: workspace.id,
+        messages,
+        browsingContext: null,
+        modelId,
+        lastUserMessageText: textPart?.text ?? '',
+        lastUserMessageParts: retriedMessage.parts,
+        hasTitle: !!thread.title,
+        conversationSizeTokens: thread.conversationSize,
+        existingTurnId: lastUserMessage.turnId,
+      },
+    );
+
+    await this.threadRepository.update(
+      workspace.id,
+      { id: threadId },
+      { activeStreamId: streamId, lastStreamError: null },
+    );
+
+    return { streamId, messageId: lastUserMessage.id };
+  }
+
  async flushNextQueuedMessage(
    threadId: string,
    userWorkspaceId: string,
@@ -252,7 +349,7 @@ export class AgentChatStreamingService {
    await this.threadRepository.update(
      workspaceId,
      { id: threadId },
-      { activeStreamId: streamId },
+      { activeStreamId: streamId, lastStreamError: null },
    );
  }

@@ -260,6 +260,37 @@ export class AgentChatService {
    } as AgentMessageEntity;
  }

+  async findLatestSentUserMessage({
+    threadId,
+    workspaceId,
+  }: {
+    threadId: string;
+    workspaceId: string;
+  }): Promise<Pick<AgentMessageEntity, 'id' | 'turnId'> | null> {
+    return this.messageRepository.findOne(workspaceId, {
+      where: {
+        threadId,
+        role: AgentMessageRole.USER,
+        status: AgentMessageStatus.SENT,
+      },
+      order: { createdAt: 'DESC', id: 'DESC' },
+      select: ['id', 'turnId'],
+    });
+  }
+
+  async deleteAssistantMessagesForTurn({
+    turnId,
+    workspaceId,
+  }: {
+    turnId: string;
+    workspaceId: string;
+  }): Promise<void> {
+    await this.messageRepository.delete(workspaceId, {
+      turnId,
+      role: AgentMessageRole.ASSISTANT,
+    });
+  }
+
  async hasAssistantMessageForTurn({
    turnId,
    workspaceId,
@@ -0,0 +1,5 @@
+import { type StreamErrorPayload } from 'src/engine/metadata-modules/ai/ai-chat/utils/map-error-to-stream-error.util';
+
+export type AgentChatThreadLastStreamError = StreamErrorPayload & {
+  failedAt: string;
+};
@@ -0,0 +1,49 @@
+import {
+  AiException,
+  AiExceptionCode,
+} from 'src/engine/metadata-modules/ai/ai.exception';
+import {
+  STREAM_EXECUTION_FAILED_CODE,
+  mapErrorToStreamError,
+} from 'src/engine/metadata-modules/ai/ai-chat/utils/map-error-to-stream-error.util';
+
+describe('mapErrorToStreamError', () => {
+  it('maps an AiException to its typed code and message', () => {
+    const error = new AiException(
+      'No AI models are available. Configure at least one AI provider.',
+      AiExceptionCode.API_KEY_NOT_CONFIGURED,
+    );
+
+    expect(mapErrorToStreamError(error)).toEqual({
+      code: AiExceptionCode.API_KEY_NOT_CONFIGURED,
+      message:
+        'No AI models are available. Configure at least one AI provider.',
+    });
+  });
+
+  it('collapses a generic Error to the fallback code but keeps its message', () => {
+    expect(mapErrorToStreamError(new Error('Provider timed out'))).toEqual({
+      code: STREAM_EXECUTION_FAILED_CODE,
+      message: 'Provider timed out',
+    });
+  });
+
+  it('handles non-Error values with a stable fallback', () => {
+    expect(mapErrorToStreamError('boom')).toEqual({
+      code: STREAM_EXECUTION_FAILED_CODE,
+      message: 'Stream execution failed',
+    });
+  });
+
+  it('truncates oversized provider messages before they are persisted', () => {
+    const result = mapErrorToStreamError(new Error('x'.repeat(10_000)));
+
+    expect(result.code).toBe(STREAM_EXECUTION_FAILED_CODE);
+    expect(result.message.length).toBe(2001);
+    expect(result.message.endsWith('…')).toBe(true);
+  });
+
+  it('leaves short messages untouched', () => {
+    expect(mapErrorToStreamError(new Error('short')).message).toBe('short');
+  });
+});
@@ -0,0 +1,28 @@
+import { AiException } from 'src/engine/metadata-modules/ai/ai.exception';
+
+export const STREAM_EXECUTION_FAILED_CODE = 'STREAM_EXECUTION_FAILED';
+
+const STREAM_ERROR_MESSAGE_MAX_LENGTH = 2000;
+
+export type StreamErrorPayload = {
+  code: string;
+  message: string;
+};
+
+const truncateMessage = (message: string): string =>
+  message.length > STREAM_ERROR_MESSAGE_MAX_LENGTH
+    ? `${message.slice(0, STREAM_ERROR_MESSAGE_MAX_LENGTH)}…`
+    : message;
+
+export const mapErrorToStreamError = (error: unknown): StreamErrorPayload => {
+  if (error instanceof AiException) {
+    return { code: error.code, message: truncateMessage(error.message) };
+  }
+
+  return {
+    code: STREAM_EXECUTION_FAILED_CODE,
+    message: truncateMessage(
+      error instanceof Error ? error.message : 'Stream execution failed',
+    ),
+  };
+};
@@ -17,6 +17,7 @@ export enum AiExceptionCode {
  USER_WORKSPACE_ID_NOT_FOUND = 'USER_WORKSPACE_ID_NOT_FOUND',
  ROLE_NOT_FOUND = 'ROLE_NOT_FOUND',
  ROLE_CANNOT_BE_ASSIGNED_TO_AGENTS = 'ROLE_CANNOT_BE_ASSIGNED_TO_AGENTS',
+  NO_FAILED_TURN_TO_RETRY = 'NO_FAILED_TURN_TO_RETRY',
 }

 const getAiExceptionUserFriendlyMessage = (code: AiExceptionCode) => {
@@ -45,6 +46,8 @@ const getAiExceptionUserFriendlyMessage = (code: AiExceptionCode) => {
      return msg`Role not found.`;
    case AiExceptionCode.ROLE_CANNOT_BE_ASSIGNED_TO_AGENTS:
      return msg`This role cannot be assigned to agents.`;
+    case AiExceptionCode.NO_FAILED_TURN_TO_RETRY:
+      return msg`There is no failed message to retry.`;
    default:
      assertUnreachable(code);
  }
@@ -30,6 +30,7 @@ export const aiGraphqlApiExceptionHandler = (error: Error) => {
      case AiExceptionCode.INVALID_CHAT_THREAD_TITLE:
        throw new UserInputError(error);
      case AiExceptionCode.AGENT_ALREADY_EXISTS:
+      case AiExceptionCode.NO_FAILED_TURN_TO_RETRY:
        throw new ConflictError(error);
      case AiExceptionCode.AGENT_IS_STANDARD:
      case AiExceptionCode.ROLE_CANNOT_BE_ASSIGNED_TO_AGENTS: