From 94c548cd5dc43131e0471b0edabe496dff245862 Mon Sep 17 00:00:00 2001 From: Guilhem Date: Mon, 14 Sep 2026 19:27:27 +0200 Subject: [PATCH] fix: re-attach flow chat to the same job on SSE timeout instead of re-running it (#11122) * fix: re-attach flow chat to the same job on SSE timeout instead of re-running it Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01S4xKKyrBTs35MEScYucgZW * fix: restart the flow chat stream when the streaming sub-job changes across a reconnect Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01S4xKKyrBTs35MEScYucgZW --------- Co-authored-by: Claude Fable 5.1 --- .../conversations/FlowChatManager.svelte.ts | 341 ++++++++++-------- 1 file changed, 185 insertions(+), 156 deletions(-) diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index 7cf2c8d6f7..bc8033e45d 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -18,6 +18,17 @@ export interface ConversationWithDraft extends FlowConversation { isDraft?: boolean } +// Per-turn stream state, kept across SSE reconnects to the same job. +interface StreamTurnState { + accumulatedContent: string + assistantMessageId: string + // Last offset the server reported; sent back on reconnect so the stream resumes + // after the deltas already rendered rather than replaying from the start. + // It indexes the stream of `streamJobId` only. + streamOffset: number | undefined + streamJobId: string | undefined +} + export class FlowChatManager { // State messages = $state([]) @@ -484,171 +495,22 @@ export class FlowChatManager { this.currentEventSource.close() } - // Track stream state for this message - let accumulatedContent = '' - let assistantMessageId = '' - let isCompleted = false - try { const jobId = await this.#onRunFlow?.(messageContent, currentConversationId, additionalInputs) if (!jobId) { console.error('No jobId returned from onRunFlow') return } + this.currentJobId = jobId - // Build the EventSource URL - const streamUrl = `/api/w/${this.#workspace()}/jobs_u/getupdate_sse/${jobId}` - const url = new URL(streamUrl, window.location.origin) - url.searchParams.set('poll_delay_ms', '50') - url.searchParams.set('fast', 'true') - url.searchParams.set('only_result', 'true') - // Create EventSource connection - const eventSource = new EventSource(url.toString()) - this.currentEventSource = eventSource - - // start polling this.startPolling(currentConversationId, isNewConversation) - eventSource.onmessage = async (event) => { - try { - const data = JSON.parse(event.data) - const type = data.type - - // Handle timeout - reconnect to SSE - if (type === 'timeout') { - eventSource.close() - this.currentEventSource = undefined - // Reconnect - this.handleStreamingMessage( - messageContent, - currentConversationId, - isNewConversation, - additionalInputs - ) - return - } - - // Handle ping - just ignore - if (type === 'ping') { - return - } - - // Handle error - if (type === 'error') { - eventSource.close() - this.currentEventSource = undefined - console.error('SSE error:', data) - sendUserToast('Stream error: ' + (data.error || 'Unknown error'), true) - this.cleanup() - return - } - - // Handle not found - if (type === 'not_found') { - eventSource.close() - this.currentEventSource = undefined - console.error('Job not found') - sendUserToast('Job not found', true) - this.cleanup() - return - } - - if (type === 'update') { - if (data.flow_stream_job_id) { - this.currentJobId = data.flow_stream_job_id - } - // Process new stream content - if (data.new_result_stream) { - // Stop polling since we are receiving last step streaming - this.stopPolling() - const { - type, - content: newContent, - success - } = parseStreamDeltas(data.new_result_stream) - accumulatedContent += newContent - - // Create tool message if type is tool_result - if (type === 'tool_result') { - // set last message streaming to false - this.messages = this.messages.map((msg) => - msg.id === this.messages[this.messages.length - 1].id - ? { ...msg, streaming: false } - : msg - ) - - this.messages = [ - ...this.messages, - { - id: 'temp-' + randomUUID(), - content: newContent, - created_at: new Date().toISOString(), - created_seq: 0, - message_type: 'tool', - conversation_id: currentConversationId, - job_id: '', - loading: false, - streaming: false, - success - } - ] - // Reset assistant message ID since we are creating a tool message - assistantMessageId = '' - accumulatedContent = '' - } - - // Create message on first content - else if ( - type === 'message' && - assistantMessageId.length === 0 && - accumulatedContent.length > 0 - ) { - assistantMessageId = 'temp-' + randomUUID() - this.messages = [ - ...this.messages, - { - id: assistantMessageId, - content: accumulatedContent, - created_at: new Date().toISOString(), - created_seq: 0, - message_type: 'assistant', - conversation_id: currentConversationId, - job_id: '', - loading: false, - streaming: true - } - ] - } else { - // Update existing message - this.messages = this.messages.map((msg) => - msg.id === assistantMessageId ? { ...msg, content: accumulatedContent } : msg - ) - } - } - - // Handle completion - if (data.completed) { - isCompleted = true - // Do a final poll to get all messages from database - if (this.selectedConversationId) { - await this.pollConversationMessages(this.selectedConversationId, { - removeTempMessages: true - }) - } - this.cleanup() - } - } - } catch (error) { - console.error('Error processing stream event:', error) - } - } - - eventSource.onerror = (error) => { - if (isCompleted) return - console.error('EventSource error:', error) - sendUserToast('Stream error occurred', true) - this.cleanup() - } + this.#followJob(jobId, currentConversationId, { + accumulatedContent: '', + assistantMessageId: '', + streamOffset: undefined, + streamJobId: undefined + }) } catch (error) { console.error('Stream connection error:', error) sendUserToast('Failed to connect to stream', true) @@ -656,6 +518,173 @@ export class FlowChatManager { } } + // Opens an SSE connection on an already-running job. The server closes every + // stream after TIMEOUT_SSE_STREAM, so a timeout re-enters here with the same + // job and turn state rather than starting a new run. + #followJob(jobId: string, currentConversationId: string, turn: StreamTurnState) { + const streamUrl = `/api/w/${this.#workspace()}/jobs_u/getupdate_sse/${jobId}` + const url = new URL(streamUrl, window.location.origin) + url.searchParams.set('poll_delay_ms', '50') + url.searchParams.set('fast', 'true') + url.searchParams.set('only_result', 'true') + if (turn.streamOffset !== undefined) { + url.searchParams.set('stream_offset', turn.streamOffset.toString()) + } + const eventSource = new EventSource(url.toString()) + this.currentEventSource = eventSource + let isCompleted = false + + eventSource.onmessage = async (event) => { + try { + const data = JSON.parse(event.data) + const type = data.type + + if (type === 'timeout') { + eventSource.close() + this.currentEventSource = undefined + this.#followJob(jobId, currentConversationId, turn) + return + } + + // Handle ping - just ignore + if (type === 'ping') { + return + } + + // Handle error + if (type === 'error') { + eventSource.close() + this.currentEventSource = undefined + console.error('SSE error:', data) + sendUserToast('Stream error: ' + (data.error || 'Unknown error'), true) + this.cleanup() + return + } + + // Handle not found + if (type === 'not_found') { + eventSource.close() + this.currentEventSource = undefined + console.error('Job not found') + sendUserToast('Job not found', true) + this.cleanup() + return + } + + if (type === 'update') { + if (data.flow_stream_job_id) { + this.currentJobId = data.flow_stream_job_id + if (data.flow_stream_job_id !== turn.streamJobId) { + const offsetFromOtherJob = + turn.streamJobId !== undefined && turn.streamOffset !== undefined + turn.streamJobId = data.flow_stream_job_id + if (offsetFromOtherJob) { + // The offset indexes the previous sub-job's stream (a retried last step + // gets a new one), so this connection skipped the new job's first chunks. + // Drop this delta and re-attach from the start of the new sub-job. + turn.streamOffset = undefined + eventSource.close() + this.currentEventSource = undefined + this.#followJob(jobId, currentConversationId, turn) + return + } + } + } + if (data.stream_offset !== undefined) { + turn.streamOffset = data.stream_offset + } + // Process new stream content + if (data.new_result_stream) { + // Stop polling since we are receiving last step streaming + this.stopPolling() + const { type, content: newContent, success } = parseStreamDeltas(data.new_result_stream) + turn.accumulatedContent += newContent + + // Create tool message if type is tool_result + if (type === 'tool_result') { + // set last message streaming to false + this.messages = this.messages.map((msg) => + msg.id === this.messages[this.messages.length - 1].id + ? { ...msg, streaming: false } + : msg + ) + + this.messages = [ + ...this.messages, + { + id: 'temp-' + randomUUID(), + content: newContent, + created_at: new Date().toISOString(), + created_seq: 0, + message_type: 'tool', + conversation_id: currentConversationId, + job_id: '', + loading: false, + streaming: false, + success + } + ] + // Reset assistant message ID since we are creating a tool message + turn.assistantMessageId = '' + turn.accumulatedContent = '' + } + + // Create message on first content + else if ( + type === 'message' && + turn.assistantMessageId.length === 0 && + turn.accumulatedContent.length > 0 + ) { + turn.assistantMessageId = 'temp-' + randomUUID() + this.messages = [ + ...this.messages, + { + id: turn.assistantMessageId, + content: turn.accumulatedContent, + created_at: new Date().toISOString(), + created_seq: 0, + message_type: 'assistant', + conversation_id: currentConversationId, + job_id: '', + loading: false, + streaming: true + } + ] + } else { + // Update existing message + this.messages = this.messages.map((msg) => + msg.id === turn.assistantMessageId + ? { ...msg, content: turn.accumulatedContent } + : msg + ) + } + } + + // Handle completion + if (data.completed) { + isCompleted = true + // Do a final poll to get all messages from database + if (this.selectedConversationId) { + await this.pollConversationMessages(this.selectedConversationId, { + removeTempMessages: true + }) + } + this.cleanup() + } + } + } catch (error) { + console.error('Error processing stream event:', error) + } + } + + eventSource.onerror = (error) => { + if (isCompleted) return + console.error('EventSource error:', error) + sendUserToast('Stream error occurred', true) + this.cleanup() + } + } + private async handlePollingMessage( messageContent: string, currentConversationId: string,