From a09fa5614fa0dd73127f51c1998982dcc930fd90 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 18 Oct 2026 14:05:13 -0700 Subject: [PATCH] fix(streaming): stop writing workflow stream frames after the client cancels --- .../lib/workflows/streaming/streaming.test.ts | 31 +++++++++++++ apps/sim/lib/workflows/streaming/streaming.ts | 45 +++++++++++-------- 2 files changed, 57 insertions(+), 19 deletions(-) diff --git a/apps/sim/lib/workflows/streaming/streaming.test.ts b/apps/sim/lib/workflows/streaming/streaming.test.ts index 7d399c8787f..11ba07fe0f5 100644 --- a/apps/sim/lib/workflows/streaming/streaming.test.ts +++ b/apps/sim/lib/workflows/streaming/streaming.test.ts @@ -1417,6 +1417,37 @@ describe('createStreamingResponse agent-events-v1', () => { expect(events).toContainEqual({ event: 'error', error: 'Client cancelled request' }) }) + it('stops writing frames once the client cancels the response stream', async () => { + let finishExecution!: (result: ExecutionResult) => void + let completeBlock!: (blockId: string, output: unknown) => Promise + const stream = await createStreamingResponse({ + requestId: 'request-client-cancel', + streamConfig: { selectedOutputs: ['block_answer'], includeFileBase64: false }, + executeFn: async ({ onBlockComplete }) => { + completeBlock = onBlockComplete + return await new Promise((resolve) => { + finishExecution = resolve + }) + }, + }) + const reader = stream.getReader() + await reader.read() + await reader.cancel() + + await expect(completeBlock('block', { answer: 'late answer' })).resolves.toBeUndefined() + finishExecution({ success: false, status: 'cancelled', output: {}, logs: [] }) + + await vi.waitFor(() => + expect(workflowStreamingLogger.info).toHaveBeenCalledWith( + '[request-client-cancel] Streaming execution aborted by client disconnect' + ) + ) + expect(workflowStreamingLogger.error).not.toHaveBeenCalledWith( + '[request-client-cancel] Stream error', + expect.anything() + ) + }) + it('thinking never enters streamedChunks / log content rewrite', async () => { const headers = new Headers({ 'x-sim-stream-protocol': 'agent-events-v1', diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts index a739dbc09a0..0aff9846de3 100644 --- a/apps/sim/lib/workflows/streaming/streaming.ts +++ b/apps/sim/lib/workflows/streaming/streaming.ts @@ -523,6 +523,7 @@ export async function createStreamingResponse( const maxThinkingChars = DEFAULT_MAX_THINKING_CHARS let requestAborted = false + let streamCancelled = false const onRequestAbort = () => { requestAborted = true timeoutController.abort() @@ -549,8 +550,13 @@ export async function createStreamingResponse( return new ReadableStream({ async start(controller) { + /** After `cancel()` the controller is closed and enqueue throws; the client is gone. */ + const enqueue = (bytes: Uint8Array) => { + if (!streamCancelled) controller.enqueue(bytes) + } + /** Flush headers promptly and keep silent blocks alive through idle-limited proxies. */ - controller.enqueue(encodeSSEComment('keepalive')) + enqueue(encodeSSEComment('keepalive')) keepaliveId = setInterval(() => { try { controller.enqueue(encodeSSEComment('keepalive')) @@ -586,7 +592,7 @@ export async function createStreamingResponse( state.streamedSelectedOutputKeys.add(options.selectedOutputKey) } const frame: ChatStreamChunkFrame = { blockId, chunk } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) state.processedOutputs.add(blockId) } @@ -601,7 +607,7 @@ export async function createStreamingResponse( event: 'thinking', data: forwarded, } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } const sendTool = ( @@ -619,7 +625,7 @@ export async function createStreamingResponse( name, ...(phase === 'end' && status ? { status } : {}), } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } /** @@ -667,7 +673,7 @@ export async function createStreamingResponse( emittedSinceReset = true } else { const frame: ChatStreamChunkFrame = { blockId, chunk: text } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } } @@ -688,7 +694,7 @@ export async function createStreamingResponse( } else if (sinkAnswerText && event.type === 'turn_end') { if (event.turn === 'intermediate' && emittedSinceReset) { const frame: ChatStreamChunkResetFrame = { blockId, event: 'chunk_reset' } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) // Re-arm separator bookkeeping so re-streamed text starts clean. emittedSinceReset = false state.processedOutputs.delete(blockId) @@ -729,7 +735,7 @@ export async function createStreamingResponse( blockId, error: getErrorMessage(error, 'Stream reading error'), } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } finally { unsubscribe?.() } @@ -746,7 +752,7 @@ export async function createStreamingResponse( const selectedOutputBlockId = outputBlockId ?? blockId state.completedBlockIds.add(selectedOutputBlockId) - if (!streamConfig.selectedOutputs?.length) { + if (streamCancelled || !streamConfig.selectedOutputs?.length) { return } @@ -776,7 +782,7 @@ export async function createStreamingResponse( : output for (const descriptor of matchingOutputs) { - if (state.selectedOutputError) { + if (streamCancelled || state.selectedOutputError) { break } try { @@ -837,7 +843,7 @@ export async function createStreamingResponse( event: 'output', data: hydratedOutput, } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) state.selectedOutputBytes = nextSelectedOutputBytes state.streamedSelectedOutputKeys.add(descriptor.key) state.processedOutputs.add(selectedOutputBlockId) @@ -869,7 +875,7 @@ export async function createStreamingResponse( blockId: selectedOutputBlockId, error: errorMessage, } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) break } } @@ -908,7 +914,7 @@ export async function createStreamingResponse( await result._streamingMetadata.loggingSession.markAsFailed(timeoutErrorMessage) } const frame: ChatStreamErrorFrame = { event: 'error', error: timeoutErrorMessage } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } else if (result.status === 'cancelled' && requestAborted) { logger.info(`[${requestId}] Streaming execution aborted by client disconnect`) if (result._streamingMetadata?.loggingSession) { @@ -917,7 +923,7 @@ export async function createStreamingResponse( } // No `final` after abort; clients that already disconnected ignore these. const frame: ChatStreamErrorFrame = { event: 'error', error: 'Client cancelled request' } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } else { await completeLoggingSession(result) @@ -953,18 +959,18 @@ export async function createStreamingResponse( ...(result.status === 'paused' && { status: 'paused' }), }, } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) } } // Terminal marker: always follows success `final` or a single terminal `error`. - controller.enqueue(encodeSSE('[DONE]')) + enqueue(encodeSSE('[DONE]')) if (executionId) { await cleanupExecutionBase64Cache(executionId) } - controller.close() + if (!streamCancelled) controller.close() } catch (error) { logger.error( `[${requestId}] Stream error`, @@ -975,15 +981,15 @@ export async function createStreamingResponse( ? SELECTED_OUTPUT_TOO_LARGE_MESSAGE : getErrorMessage(error, 'Stream processing error') const frame: ChatStreamErrorFrame = { event: 'error', error: errorMessage } - controller.enqueue(encodeSSE(frame)) + enqueue(encodeSSE(frame)) // Same terminal rule as timeout/abort: one error, then [DONE], never `final`. - controller.enqueue(encodeSSE('[DONE]')) + enqueue(encodeSSE('[DONE]')) if (executionId) { await cleanupExecutionBase64Cache(executionId) } - controller.close() + if (!streamCancelled) controller.close() } finally { stopKeepalive() cleanupRequestAbort() @@ -991,6 +997,7 @@ export async function createStreamingResponse( } }, async cancel(reason) { + streamCancelled = true logger.info( `[${requestId}] Streaming response cancelled`, projectResolvedSecretDiagnosticError(reason, undefined)