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>
This commit is contained in:
+14
-7
@@ -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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user