From 4e6cf185fa6c5bc9a8d87703215f64b1510ab0b8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?F=C3=A9lix=20Malfait?= <felix.malfait@gmail.com>
Date: Fri, 10 Apr 2026 13:14:19 +0200
Subject: [PATCH] Fix error handling in stream agent chat job (#19550)

## Summary
Improved error handling in the StreamAgentChatJob to properly propagate
and reject errors that occur during the agent chat stream processing.

## Key Changes
- Added error re-throw in the catch block to ensure errors are not
silently swallowed
- Introduced `streamError` variable to capture errors from the stream's
`onError` callback
- Modified the promise resolution logic to reject with the captured
stream error instead of silently resolving on success, ensuring proper
error propagation to callers

## Implementation Details
The changes ensure that errors occurring during stream processing are
properly captured and propagated:
1. Stream errors are now captured in the `onError` callback and stored
in the `streamError` variable
2. After the stream completes and the message is persisted to the
database, the promise is rejected if a stream error occurred
3. The outer catch block now re-throws errors to prevent silent failures

This fix prevents scenarios where stream processing errors would be
masked by a successful database persistence operation.

https://claude.ai/code/session_016DQodL32HYv6CR1cMASNHZ

---------

Co-authored-by: Claude <noreply@anthropic.com>
---
 .../ai/ai-chat/jobs/stream-agent-chat.job.ts  | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
index 6c08f30e2f..ee69d8a370 100644
--- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
+++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
@@ -96,6 +96,7 @@ export class StreamAgentChatJob {
           },
         })
         .catch(() => {});
+      throw error;
     } finally {
       await this.cancelSubscriberService.unsubscribe(cancelChannel);
       await this.threadRepository
@@ -190,6 +191,7 @@ export class StreamAgentChatJob {
       };
       let lastStepConversationSize = 0;
       let totalCacheCreationTokens = 0;
+      let streamError: unknown;
 
       // onFinish fires before the uiStream is fully drained. We use this
       // promise to coordinate: the IIFE waits for DB persist to complete
@@ -247,6 +249,8 @@ export class StreamAgentChatJob {
           writer.merge(
             stream.toUIMessageStream({
               onError: (error) => {
+                streamError = error;
+
                 return error instanceof Error ? error.message : String(error);
               },
               sendStart: false,
@@ -307,13 +311,16 @@ export class StreamAgentChatJob {
 
           await streamFinishedPromise;
 
-          await this.eventPublisherService.publish({
-            threadId: data.threadId,
-            workspaceId: data.workspaceId,
-            event: { type: 'message-persisted', messageId: data.threadId },
-          });
-
-          resolve();
+          if (streamError) {
+            reject(streamError);
+          } else {
+            await this.eventPublisherService.publish({
+              threadId: data.threadId,
+              workspaceId: data.workspaceId,
+              event: { type: 'message-persisted', messageId: data.threadId },
+            });
+            resolve();
+          }
         } catch (error) {
           reject(error);
         }