@@ -274,7 +316,12 @@ export const ClientChatMessage = memo(
return (
prevProps.message.id === nextProps.message.id &&
prevProps.message.content === nextProps.message.content &&
+ prevProps.message.thinking === nextProps.message.thinking &&
prevProps.message.isStreaming === nextProps.message.isStreaming &&
+ prevProps.message.isThinkingStreaming === nextProps.message.isThinkingStreaming &&
+ prevProps.message.isToolStreaming === nextProps.message.isToolStreaming &&
+ toolCallsFingerprint(prevProps.message.toolCalls) ===
+ toolCallsFingerprint(nextProps.message.toolCalls) &&
prevProps.message.isInitialMessage === nextProps.message.isInitialMessage &&
prevProps.message.files?.length === nextProps.message.files?.length
)
diff --git a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx
new file mode 100644
index 00000000000..d2da6d72ee5
--- /dev/null
+++ b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx
@@ -0,0 +1,570 @@
+/**
+ * @vitest-environment jsdom
+ */
+import { act } from 'react'
+import { createRoot, type Root } from 'react-dom/client'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockReadSSEEvents } = vi.hoisted(() => ({
+ mockReadSSEEvents: vi.fn(),
+}))
+
+vi.mock('@/lib/core/utils/sse', () => ({
+ readSSEEvents: mockReadSSEEvents,
+}))
+
+vi.mock('@sim/utils/id', () => ({
+ generateId: () => 'msg-assistant-1',
+}))
+
+import type { ChatMessage } from '@/app/(interfaces)/chat/components/message/message'
+import {
+ isAnswerChunkFrame,
+ useChatStreaming,
+} from '@/app/(interfaces)/chat/hooks/use-chat-streaming'
+
+describe('isAnswerChunkFrame', () => {
+ it('accepts plain answer chunks without an event type', () => {
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: 'hello' })).toBe(true)
+ })
+
+ it('rejects thinking / stream_error / tool frames even if chunk is present', () => {
+ expect(
+ isAnswerChunkFrame({ blockId: 'a1', chunk: 'leak', event: 'thinking', data: 'thought' })
+ ).toBe(false)
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: 'x', event: 'stream_error' })).toBe(false)
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: 'x', event: 'tool' })).toBe(false)
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: 'x', event: 'tool_call_start' })).toBe(false)
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: 'x', event: 'final' })).toBe(false)
+ })
+
+ it('rejects frames missing blockId or empty chunk', () => {
+ expect(isAnswerChunkFrame({ chunk: 'hello' })).toBe(false)
+ expect(isAnswerChunkFrame({ blockId: 'a1', chunk: '' })).toBe(false)
+ })
+})
+
+interface HookHandle {
+ latest: () => ReturnType
+ unmount: () => void
+}
+
+function renderStreamingHook(): HookHandle {
+ ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
+ const container = document.createElement('div')
+ const root: Root = createRoot(container)
+ let latest!: ReturnType
+
+ function Probe() {
+ latest = useChatStreaming()
+ return null
+ }
+
+ act(() => {
+ root.render(((resolve) => {
+ requestAnimationFrame(() => resolve())
+ })
+ await new Promise((resolve) => {
+ setTimeout(resolve, 60)
+ })
+ })
+}
+
+describe('useChatStreaming thinking + abort (Step 6)', () => {
+ let handle: HookHandle
+ let messages: ChatMessage[]
+ let setMessages: React.Dispatch>
+
+ beforeEach(() => {
+ vi.clearAllMocks()
+ messages = []
+ setMessages = ((updater: React.SetStateAction) => {
+ messages = typeof updater === 'function' ? updater(messages) : updater
+ }) as React.Dispatch>
+ handle = renderStreamingHook()
+
+ // Run rAF immediately so UI batching is deterministic in tests.
+ vi.spyOn(window, 'requestAnimationFrame').mockImplementation((cb) => {
+ cb(performance.now())
+ return 1
+ })
+ })
+
+ afterEach(() => {
+ handle.unmount()
+ vi.restoreAllMocks()
+ })
+
+ it('routes thinking to message.thinking and answer chunks to content only', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'Let me reason. ',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'More thought.',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'Final answer.',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.thinking).toBe('Let me reason. More thought.')
+ expect(assistant?.content).toBe('Final answer.')
+ expect(assistant?.isStreaming).toBe(false)
+ expect(assistant?.isThinkingStreaming).toBe(false)
+ })
+
+ it('surfaces stream_error text in thinking chrome without terminating', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'Working…',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'stream_error',
+ error: 'partial provider glitch',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'Recovered answer.',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.thinking).toContain('Working…')
+ expect(assistant?.thinking).toContain('[Stream error] partial provider glitch')
+ expect(assistant?.content).toBe('Recovered answer.')
+ expect(assistant?.isStreaming).toBe(false)
+ })
+
+ it('clears streaming flags when SSE ends without a terminal final/error frame', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'Halfway…',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'Partial answer',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'start',
+ id: 't1',
+ name: 'search',
+ })
+ // Stream closes abruptly — no final or error event.
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.content).toBe('Partial answer')
+ expect(assistant?.thinking).toBe('Halfway…')
+ expect(assistant?.isStreaming).toBe(false)
+ expect(assistant?.isThinkingStreaming).toBe(false)
+ expect(assistant?.isToolStreaming).toBe(false)
+ expect(assistant?.toolCalls?.some((t) => t.status === 'error')).toBe(true)
+ })
+
+ it('does not append thinking payload into answer when mislabeled as chunk', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ chunk: 'SHOULD_NOT_APPEND',
+ data: 'real thought',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'ok',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.content).toBe('ok')
+ expect(assistant?.thinking).toBe('real thought')
+ })
+
+ it('TTS audioStreamHandler receives answer text only', async () => {
+ const audioStreamHandler = vi.fn().mockResolvedValue(undefined)
+
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'secret internal monologue that must not be spoken.',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'Hello world.',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle
+ .latest()
+ .handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn(), {
+ voiceSettings: {
+ isVoiceEnabled: true,
+ voiceId: 'voice-1',
+ autoPlayResponses: true,
+ },
+ audioStreamHandler,
+ })
+ })
+ await flushUiBatch()
+
+ expect(audioStreamHandler).toHaveBeenCalled()
+ for (const call of audioStreamHandler.mock.calls) {
+ expect(String(call[0])).not.toContain('secret')
+ expect(String(call[0])).not.toContain('monologue')
+ }
+ expect(audioStreamHandler.mock.calls.some((c) => String(c[0]).includes('Hello'))).toBe(true)
+ })
+
+ it('stopStreaming preserves thinking and aborts the shared controller', async () => {
+ const abortController = new AbortController()
+ let resolveStream!: () => void
+ const streamDone = new Promise((resolve) => {
+ resolveStream = resolve
+ })
+
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'thinking',
+ data: 'partial thought',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'partial answer',
+ })
+ // Hold the stream open until Stop aborts.
+ await new Promise((resolve) => {
+ options.signal?.addEventListener('abort', () => resolve(), { once: true })
+ // Also allow test cleanup if abort never fires.
+ streamDone.then(() => resolve())
+ })
+ })
+
+ const streamPromise = act(async () => {
+ await handle
+ .latest()
+ .handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn(), {
+ abortController,
+ })
+ })
+
+ await flushUiBatch()
+ expect(messages.find((m) => m.id === 'msg-assistant-1')?.thinking).toBe('partial thought')
+
+ act(() => {
+ handle.latest().stopStreaming(setMessages)
+ })
+ resolveStream()
+ await streamPromise
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(abortController.signal.aborted).toBe(true)
+ expect(assistant?.thinking).toBe('partial thought')
+ expect(String(assistant?.content)).toContain('partial answer')
+ expect(String(assistant?.content)).toContain('Response stopped by user')
+ expect(assistant?.isStreaming).toBe(false)
+ expect(assistant?.isThinkingStreaming).toBe(false)
+ })
+
+ it('does not replace Stop notice with server Client cancelled request error', async () => {
+ const abortController = new AbortController()
+ let resolveStream!: () => void
+ const streamDone = new Promise((resolve) => {
+ resolveStream = resolve
+ })
+
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'partial answer',
+ })
+ await new Promise((resolve) => {
+ options.signal?.addEventListener(
+ 'abort',
+ () => {
+ // Server still emits terminal cancel error while the reader finishes.
+ void options.onEvent({
+ event: 'error',
+ error: 'Client cancelled request',
+ })
+ resolve()
+ },
+ { once: true }
+ )
+ streamDone.then(() => resolve())
+ })
+ })
+
+ const streamPromise = act(async () => {
+ await handle
+ .latest()
+ .handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn(), {
+ abortController,
+ })
+ })
+
+ await flushUiBatch()
+
+ act(() => {
+ handle.latest().stopStreaming(setMessages)
+ })
+ resolveStream()
+ await streamPromise
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(String(assistant?.content)).toContain('partial answer')
+ expect(String(assistant?.content)).toContain('Response stopped by user')
+ expect(String(assistant?.content)).not.toContain('Client cancelled request')
+ expect(assistant?.isStreaming).toBe(false)
+ })
+
+ it('leaves thinking undefined when no thinking events arrive', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'just text',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.thinking).toBeUndefined()
+ expect(assistant?.content).toBe('just text')
+ })
+})
+
+describe('useChatStreaming tool lifecycle (Step 8)', () => {
+ let handle: HookHandle
+ let messages: ChatMessage[]
+ let setMessages: React.Dispatch>
+
+ beforeEach(() => {
+ vi.clearAllMocks()
+ messages = []
+ setMessages = ((updater: React.SetStateAction) => {
+ messages = typeof updater === 'function' ? updater(messages) : updater
+ }) as React.Dispatch>
+ handle = renderStreamingHook()
+ vi.spyOn(window, 'requestAnimationFrame').mockImplementation((cb) => {
+ cb(performance.now())
+ return 1
+ })
+ })
+
+ afterEach(() => {
+ handle.unmount()
+ vi.restoreAllMocks()
+ })
+
+ it('maps tool start/end into keyed chips without touching answer content', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'start',
+ id: 'toolu_1',
+ name: 'http_request',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'end',
+ id: 'toolu_1',
+ name: 'http_request',
+ status: 'success',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ chunk: 'https://httpbin.org/get',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+ expect(assistant?.content).toBe('https://httpbin.org/get')
+ expect(assistant?.toolCalls).toEqual([
+ {
+ key: 'agent-1:toolu_1',
+ blockId: 'agent-1',
+ id: 'toolu_1',
+ name: 'http_request',
+ displayName: 'Http Request',
+ status: 'success',
+ },
+ ])
+ expect(assistant?.toolCalls?.[0]).not.toHaveProperty('args')
+ expect(assistant?.toolCalls?.[0]).not.toHaveProperty('result')
+ expect(assistant?.isToolStreaming).toBe(false)
+ })
+
+ it('tracks parallel tools and cancels running chips on Stop', async () => {
+ const abortController = new AbortController()
+ let resolveStream!: () => void
+ const streamDone = new Promise((resolve) => {
+ resolveStream = resolve
+ })
+
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'start',
+ id: 'toolu_1',
+ name: 'http_request',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'start',
+ id: 'toolu_2',
+ name: 'function_execute',
+ })
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'end',
+ id: 'toolu_2',
+ name: 'function_execute',
+ status: 'success',
+ })
+ await new Promise((resolve) => {
+ options.signal?.addEventListener('abort', () => resolve(), { once: true })
+ streamDone.then(() => resolve())
+ })
+ })
+
+ const streamPromise = act(async () => {
+ await handle
+ .latest()
+ .handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn(), {
+ abortController,
+ })
+ })
+
+ await flushUiBatch()
+ expect(messages.find((m) => m.id === 'msg-assistant-1')?.toolCalls).toHaveLength(2)
+
+ act(() => {
+ handle.latest().stopStreaming(setMessages)
+ })
+ resolveStream()
+ await streamPromise
+
+ const tools = messages.find((m) => m.id === 'msg-assistant-1')?.toolCalls
+ expect(tools?.find((t) => t.id === 'toolu_1')?.status).toBe('cancelled')
+ expect(tools?.find((t) => t.id === 'toolu_2')?.status).toBe('success')
+ expect(messages.find((m) => m.id === 'msg-assistant-1')?.isToolStreaming).toBe(false)
+ })
+
+ it('settles straggler running tools to success on final', async () => {
+ mockReadSSEEvents.mockImplementation(async (_source, options) => {
+ await options.onEvent({
+ blockId: 'agent-1',
+ event: 'tool',
+ phase: 'start',
+ id: 'toolu_open',
+ name: 'http_request',
+ })
+ await options.onEvent({
+ event: 'final',
+ data: { success: true, output: {} },
+ })
+ })
+
+ await act(async () => {
+ await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+ })
+ await flushUiBatch()
+
+ expect(messages.find((m) => m.id === 'msg-assistant-1')?.toolCalls?.[0]?.status).toBe('success')
+ })
+})
diff --git a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts
index be4b0b2e1ae..50c3764b465 100644
--- a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts
+++ b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts
@@ -3,9 +3,15 @@
import { useRef, useState } from 'react'
import { createLogger } from '@sim/logger'
import { generateId } from '@sim/utils/id'
+import { humanizeToolName } from '@/lib/copilot/tools/tool-display'
import { readSSEEvents } from '@/lib/core/utils/sse'
import { isUserFileWithMetadata } from '@/lib/core/utils/user-file'
-import type { ChatFile, ChatMessage } from '@/app/(interfaces)/chat/components/message/message'
+import type {
+ ChatFile,
+ ChatMessage,
+ ChatToolCall,
+ ChatToolCallStatus,
+} from '@/app/(interfaces)/chat/components/message/message'
import { CHAT_ERROR_MESSAGES } from '@/app/(interfaces)/chat/constants'
const logger = createLogger('UseChatStreaming')
@@ -64,23 +70,99 @@ export interface StreamingOptions {
onAudioEnd?: () => void
audioStreamHandler?: (text: string) => Promise
outputConfigs?: Array<{ blockId: string; path?: string }>
+ /**
+ * Shared AbortController for fetch + SSE body reads. When provided (preferred),
+ * Stop aborts the in-flight request server-side as well as the reader.
+ */
+ abortController?: AbortController
+}
+
+type ChatSseEvent = {
+ blockId?: string
+ chunk?: string
+ event?: string
+ error?: string
+ phase?: 'start' | 'end'
+ id?: string
+ name?: string
+ status?: 'success' | 'error' | 'cancelled'
+ data?:
+ | string
+ | {
+ success: boolean
+ error?: string | { message?: string }
+ output?: Record>
+ }
+}
+
+function toolCallKey(blockId: string, id: string): string {
+ return `${blockId}:${id}`
+}
+
+function settleInFlightTools(
+ map: Map,
+ terminal: Exclude
+): void {
+ for (const [key, tool] of map) {
+ if (tool.status === 'running') {
+ map.set(key, { ...tool, status: terminal })
+ }
+ }
+}
+
+function snapshotToolCalls(
+ order: string[],
+ map: Map
+): ChatToolCall[] | undefined {
+ if (order.length === 0) return undefined
+ return order.map((key) => map.get(key)).filter((t): t is ChatToolCall => Boolean(t))
+}
+
+function anyToolRunning(map: Map): boolean {
+ for (const tool of map.values()) {
+ if (tool.status === 'running') return true
+ }
+ return false
+}
+
+/** Answer text frames never carry an event type (or use a reserved non-answer event). */
+function isAnswerChunkFrame(json: ChatSseEvent): boolean {
+ if (!json.blockId || typeof json.chunk !== 'string' || !json.chunk) return false
+ // Thinking / tools / errors must never use `chunk` for answer accumulation.
+ if (json.event === 'thinking' || json.event === 'stream_error' || json.event === 'error') {
+ return false
+ }
+ if (
+ json.event === 'final' ||
+ json.event === 'tool' ||
+ json.event === 'tool_call_start' ||
+ json.event === 'tool_call_end'
+ ) {
+ return false
+ }
+ return json.event === undefined || json.event === ''
}
export function useChatStreaming() {
const [isStreamingResponse, setIsStreamingResponse] = useState(false)
const abortControllerRef = useRef(null)
const accumulatedTextRef = useRef('')
+ const accumulatedThinkingRef = useRef('')
+ const accumulatedToolCallsRef = useRef([])
const lastStreamedPositionRef = useRef(0)
const audioStreamingActiveRef = useRef(false)
- const lastDisplayedPositionRef = useRef(0) // Track displayed text in synced mode
+ const lastDisplayedPositionRef = useRef(0)
const stopStreaming = (setMessages: React.Dispatch>) => {
if (abortControllerRef.current) {
- // Abort the fetch request
abortControllerRef.current.abort()
abortControllerRef.current = null
const latestContent = accumulatedTextRef.current
+ const latestThinking = accumulatedThinkingRef.current
+ const latestTools = accumulatedToolCallsRef.current.map((tool) =>
+ tool.status === 'running' ? { ...tool, status: 'cancelled' as const } : tool
+ )
setMessages((prev) => {
const lastMessage = prev[prev.length - 1]
@@ -92,7 +174,16 @@ export function useChatStreaming() {
return [
...prev.slice(0, -1),
- { ...lastMessage, content: updatedContent, isStreaming: false },
+ {
+ ...lastMessage,
+ content: updatedContent,
+ // Preserve any thinking / tools received before Stop.
+ thinking: latestThinking || lastMessage.thinking,
+ toolCalls: latestTools.length > 0 ? latestTools : lastMessage.toolCalls,
+ isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
+ },
]
}
@@ -101,6 +192,8 @@ export function useChatStreaming() {
setIsStreamingResponse(false)
accumulatedTextRef.current = ''
+ accumulatedThinkingRef.current = ''
+ accumulatedToolCallsRef.current = []
lastStreamedPositionRef.current = 0
lastDisplayedPositionRef.current = 0
audioStreamingActiveRef.current = false
@@ -112,15 +205,18 @@ export function useChatStreaming() {
setMessages: React.Dispatch>,
setIsLoading: React.Dispatch>,
scrollToBottom: () => void,
- userHasScrolled?: boolean,
streamingOptions?: StreamingOptions
) => {
logger.info('[useChatStreaming] handleStreamedResponse called')
- // Set streaming state
setIsStreamingResponse(true)
- abortControllerRef.current = new AbortController()
- // Check if we should stream audio
+ // Prefer a shared controller from the caller (fetch + reader). Otherwise create one.
+ if (streamingOptions?.abortController) {
+ abortControllerRef.current = streamingOptions.abortController
+ } else if (!abortControllerRef.current) {
+ abortControllerRef.current = new AbortController()
+ }
+
const shouldPlayAudio =
streamingOptions?.voiceSettings?.isVoiceEnabled &&
streamingOptions?.voiceSettings?.autoPlayResponses &&
@@ -133,7 +229,15 @@ export function useChatStreaming() {
}
let accumulatedText = ''
+ let accumulatedThinking = ''
+ let isThinkingStreaming = false
let lastAudioPosition = 0
+ const toolCallsMap = new Map()
+ const toolCallOrder: string[] = []
+
+ const syncToolCallsRef = () => {
+ accumulatedToolCallsRef.current = snapshotToolCalls(toolCallOrder, toolCallsMap) ?? []
+ }
const messageIdMap = new Map()
const messageId = generateId()
@@ -156,14 +260,29 @@ export function useChatStreaming() {
if (!uiDirty) return
uiDirty = false
lastUIFlush = performance.now()
- const snapshot = accumulatedText
+ const contentSnapshot = accumulatedText
+ const thinkingSnapshot = accumulatedThinking
+ const thinkingStreamingSnapshot = isThinkingStreaming
+ const toolCallsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
+ const toolStreamingSnapshot = anyToolRunning(toolCallsMap)
setMessages((prev) =>
prev.map((msg) => {
if (msg.id !== messageId) return msg
if (!msg.isStreaming) return msg
- return { ...msg, content: snapshot }
+ return {
+ ...msg,
+ content: contentSnapshot,
+ thinking: thinkingSnapshot || undefined,
+ isThinkingStreaming: thinkingStreamingSnapshot,
+ toolCalls: toolCallsSnapshot,
+ isToolStreaming: toolStreamingSnapshot,
+ }
})
)
+ // Caller supplies a stick-to-bottom-aware scroller (no-ops if user scrolled away).
+ requestAnimationFrame(() => {
+ scrollToBottom()
+ })
}
const scheduleUIFlush = () => {
@@ -192,20 +311,13 @@ export function useChatStreaming() {
setIsLoading(false)
let terminated = false
+ // Capture before Stop nulls abortControllerRef; needed when the reader
+ // resolves on abort instead of throwing AbortError.
+ const streamAbortSignal = abortControllerRef.current!.signal
try {
- await readSSEEvents<{
- blockId?: string
- chunk?: string
- event?: string
- error?: string
- data?: {
- success: boolean
- error?: string | { message?: string }
- output?: Record>
- }
- }>(response.body, {
- signal: abortControllerRef.current.signal,
+ await readSSEEvents(response.body, {
+ signal: streamAbortSignal,
onParseError: (_data, parseError) => {
logger.error('Error parsing stream data:', parseError)
},
@@ -213,14 +325,47 @@ export function useChatStreaming() {
const { blockId, chunk: contentChunk, event: eventType } = json
if (eventType === 'error' || json.event === 'error') {
+ // User Stop aborts the fetch; the server often still emits a terminal
+ // `{ event: 'error', error: 'Client cancelled request' }` before the
+ // SSE reader finishes. Do not overwrite the stop notice.
+ if (streamAbortSignal.aborted) {
+ settleInFlightTools(toolCallsMap, 'cancelled')
+ syncToolCallsRef()
+ const toolsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
+ setMessages((prev) =>
+ prev.map((msg) =>
+ msg.id === messageId
+ ? {
+ ...msg,
+ isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
+ }
+ : msg
+ )
+ )
+ setIsLoading(false)
+ terminated = true
+ return true
+ }
+
const errorMessage = json.error || CHAT_ERROR_MESSAGES.GENERIC_ERROR
+ settleInFlightTools(toolCallsMap, 'error')
+ syncToolCallsRef()
+ const toolsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
setMessages((prev) =>
prev.map((msg) =>
msg.id === messageId
? {
...msg,
content: errorMessage,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
type: 'assistant' as const,
}
: msg
@@ -231,9 +376,91 @@ export function useChatStreaming() {
return true
}
- if (eventType === 'final' && json.data) {
+ if (eventType === 'stream_error') {
+ const errText =
+ typeof json.error === 'string' && json.error
+ ? json.error
+ : 'A streaming error occurred'
+ logger.warn('[useChatStreaming] Non-terminal stream_error', {
+ blockId,
+ error: errText,
+ })
+ // Non-terminal: keep streaming; surface in thinking chrome so it is visible.
+ accumulatedThinking += `${accumulatedThinking ? '\n\n' : ''}[Stream error] ${errText}`
+ accumulatedThinkingRef.current = accumulatedThinking
+ isThinkingStreaming = true
+ uiDirty = true
+ scheduleUIFlush()
+ return false
+ }
+
+ if (eventType === 'thinking' && blockId && typeof json.data === 'string') {
+ if (!messageIdMap.has(blockId)) {
+ messageIdMap.set(blockId, messageId)
+ }
+ accumulatedThinking += json.data
+ accumulatedThinkingRef.current = accumulatedThinking
+ isThinkingStreaming = true
+ uiDirty = true
+ scheduleUIFlush()
+ return false
+ }
+
+ if (
+ eventType === 'tool' &&
+ blockId &&
+ typeof json.id === 'string' &&
+ json.id &&
+ typeof json.name === 'string' &&
+ json.name
+ ) {
+ if (!messageIdMap.has(blockId)) {
+ messageIdMap.set(blockId, messageId)
+ }
+ const key = toolCallKey(blockId, json.id)
+ if (json.phase === 'start') {
+ if (!toolCallsMap.has(key)) {
+ toolCallOrder.push(key)
+ }
+ toolCallsMap.set(key, {
+ key,
+ blockId,
+ id: json.id,
+ name: json.name,
+ displayName: humanizeToolName(json.name),
+ status: 'running',
+ })
+ } else if (json.phase === 'end') {
+ const endStatus: ChatToolCallStatus =
+ json.status === 'error' || json.status === 'cancelled' ? json.status : 'success'
+ const existing = toolCallsMap.get(key)
+ if (!existing) {
+ toolCallOrder.push(key)
+ toolCallsMap.set(key, {
+ key,
+ blockId,
+ id: json.id,
+ name: json.name,
+ displayName: humanizeToolName(json.name),
+ status: endStatus,
+ })
+ } else {
+ toolCallsMap.set(key, { ...existing, status: endStatus })
+ }
+ }
+ syncToolCallsRef()
+ uiDirty = true
+ scheduleUIFlush()
+ return false
+ }
+
+ if (eventType === 'final' && json.data && typeof json.data === 'object') {
flushUI()
const finalData = json.data
+ isThinkingStreaming = false
+ settleInFlightTools(toolCallsMap, 'success')
+ syncToolCallsRef()
+ const toolsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
const outputConfigs = streamingOptions?.outputConfigs
const formattedOutputs: string[] = []
@@ -358,7 +585,11 @@ export function useChatStreaming() {
? {
...msg,
isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
content: finalContent ?? msg.content,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
files: extractedFiles.length > 0 ? extractedFiles : undefined,
}
: msg
@@ -366,6 +597,8 @@ export function useChatStreaming() {
)
accumulatedTextRef.current = ''
+ accumulatedThinkingRef.current = ''
+ accumulatedToolCallsRef.current = []
lastStreamedPositionRef.current = 0
lastDisplayedPositionRef.current = 0
audioStreamingActiveRef.current = false
@@ -374,11 +607,17 @@ export function useChatStreaming() {
return true
}
- if (blockId && contentChunk) {
- if (!messageIdMap.has(blockId)) {
+ // Answer text only — never append thinking/tool/unknown chunk frames blindly.
+ if (isAnswerChunkFrame(json) && contentChunk) {
+ if (blockId && !messageIdMap.has(blockId)) {
messageIdMap.set(blockId, messageId)
}
+ // First answer chunk settles thinking chrome (still visible, no longer “live”).
+ if (isThinkingStreaming) {
+ isThinkingStreaming = false
+ }
+
accumulatedText += contentChunk
accumulatedTextRef.current = accumulatedText
logger.debug('[useChatStreaming] Received chunk', {
@@ -416,17 +655,47 @@ export function useChatStreaming() {
}
}
}
- } else if (blockId && eventType === 'end') {
- setMessages((prev) =>
- prev.map((msg) => (msg.id === messageId ? { ...msg, isStreaming: false } : msg))
- )
}
},
})
if (!terminated) {
flushUI()
+ // Stream closed without a terminal final/error frame (abrupt disconnect,
+ // or only non-terminal stream_error). Clear live chrome so the UI does not
+ // stay stuck in a streaming/loading state.
+ const wasAborted = streamAbortSignal.aborted
+ settleInFlightTools(toolCallsMap, wasAborted ? 'cancelled' : 'error')
+ syncToolCallsRef()
+ isThinkingStreaming = false
+ const toolsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
+ setMessages((prev) =>
+ prev.map((msg) => {
+ if (msg.id !== messageId) return msg
+ // stopStreaming already wrote the stop notice into content; do not clobber it.
+ if (wasAborted) {
+ return {
+ ...msg,
+ isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
+ }
+ }
+ return {
+ ...msg,
+ isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
+ content: accumulatedText || msg.content,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
+ }
+ })
+ )
if (
+ !wasAborted &&
shouldPlayAudio &&
streamingOptions?.audioStreamHandler &&
accumulatedText.length > lastAudioPosition
@@ -442,10 +711,31 @@ export function useChatStreaming() {
}
}
} catch (error) {
- logger.error('Error processing stream:', error)
+ // Stop / timeout abort the shared fetch controller; body read then throws AbortError.
+ // Match chat.tsx + use-audio-streaming: expected cancel, not a hard failure.
+ if (error instanceof Error && error.name === 'AbortError') {
+ logger.info('Stream aborted by user or timeout')
+ settleInFlightTools(toolCallsMap, 'cancelled')
+ } else {
+ logger.error('Error processing stream:', error)
+ settleInFlightTools(toolCallsMap, 'error')
+ }
+ syncToolCallsRef()
flushUI()
+ const toolsSnapshot = snapshotToolCalls(toolCallOrder, toolCallsMap)
setMessages((prev) =>
- prev.map((msg) => (msg.id === messageId ? { ...msg, isStreaming: false } : msg))
+ prev.map((msg) =>
+ msg.id === messageId
+ ? {
+ ...msg,
+ isStreaming: false,
+ isThinkingStreaming: false,
+ isToolStreaming: false,
+ thinking: accumulatedThinking || msg.thinking,
+ toolCalls: toolsSnapshot ?? msg.toolCalls,
+ }
+ : msg
+ )
)
} finally {
if (uiRAF !== null) cancelAnimationFrame(uiRAF)
@@ -453,11 +743,10 @@ export function useChatStreaming() {
setIsStreamingResponse(false)
abortControllerRef.current = null
- if (!userHasScrolled) {
- setTimeout(() => {
- scrollToBottom()
- }, 300)
- }
+ // Stick-to-bottom-aware; no-ops if the user scrolled away mid-stream.
+ setTimeout(() => {
+ scrollToBottom()
+ }, 300)
if (shouldPlayAudio) {
streamingOptions?.onAudioEnd?.()
@@ -473,3 +762,5 @@ export function useChatStreaming() {
handleStreamedResponse,
}
}
+
+export { isAnswerChunkFrame }
diff --git a/apps/sim/app/api/chat/[identifier]/otp/route.ts b/apps/sim/app/api/chat/[identifier]/otp/route.ts
index 248fcfa84ab..99dfa6069c2 100644
--- a/apps/sim/app/api/chat/[identifier]/otp/route.ts
+++ b/apps/sim/app/api/chat/[identifier]/otp/route.ts
@@ -160,6 +160,7 @@ export const PUT = withRouteHandler(
authType: chat.authType,
password: chat.password,
outputConfigs: chat.outputConfigs,
+ includeThinking: chat.includeThinking,
})
.from(chat)
.where(
@@ -209,6 +210,7 @@ export const PUT = withRouteHandler(
customizations: deployment.customizations,
authType: deployment.authType,
outputConfigs: deployment.outputConfigs,
+ includeThinking: deployment.includeThinking ?? false,
})
setChatAuthCookie(response, deployment.id, deployment.authType, deployment.password)
diff --git a/apps/sim/app/api/chat/[identifier]/route.test.ts b/apps/sim/app/api/chat/[identifier]/route.test.ts
index 9eb104547ff..ea7dfe48b28 100644
--- a/apps/sim/app/api/chat/[identifier]/route.test.ts
+++ b/apps/sim/app/api/chat/[identifier]/route.test.ts
@@ -37,6 +37,7 @@ function createMockNextRequest(
method,
headers: headersObj,
nextUrl: parsedUrl,
+ signal: AbortSignal.timeout(60_000),
cookies: {
get: vi.fn().mockReturnValue(undefined),
},
@@ -114,6 +115,7 @@ vi.mock('@/lib/uploads', () => ({
vi.mock('@/lib/workflows/streaming/streaming', () => ({
createStreamingResponse: vi.fn().mockImplementation(async () => createMockStream()),
+ agentStreamProtocolResponseHeaders: vi.fn().mockReturnValue({}),
}))
vi.mock('@/lib/workflows/executor/execute-workflow', () => ({
@@ -150,6 +152,7 @@ describe('Chat Identifier API Route', () => {
primaryColor: '#000000',
},
outputConfigs: [{ blockId: 'block-1', path: 'output' }],
+ includeThinking: false,
},
]
@@ -443,9 +446,12 @@ describe('Chat Identifier API Route', () => {
expect(createStreamingResponse).toHaveBeenCalledWith(
expect.objectContaining({
executeFn: expect.any(Function),
+ requestSignal: expect.any(AbortSignal),
+ requestHeaders: expect.anything(),
streamConfig: expect.objectContaining({
isSecureMode: true,
workflowTriggerType: 'chat',
+ includeThinking: false,
}),
})
)
diff --git a/apps/sim/app/api/chat/[identifier]/route.ts b/apps/sim/app/api/chat/[identifier]/route.ts
index 819c732db5b..1da492d81ff 100644
--- a/apps/sim/app/api/chat/[identifier]/route.ts
+++ b/apps/sim/app/api/chat/[identifier]/route.ts
@@ -27,6 +27,7 @@ interface ChatConfigSource {
customizations: unknown
authType: string | null
outputConfigs: unknown
+ includeThinking?: boolean | null
}
function toChatConfigResponse(deployment: ChatConfigSource) {
@@ -37,6 +38,7 @@ function toChatConfigResponse(deployment: ChatConfigSource) {
customizations: deployment.customizations,
authType: deployment.authType,
outputConfigs: deployment.outputConfigs,
+ includeThinking: deployment.includeThinking ?? false,
}
}
@@ -80,6 +82,7 @@ export const POST = withRouteHandler(
password: chat.password,
allowedEmails: chat.allowedEmails,
outputConfigs: chat.outputConfigs,
+ includeThinking: chat.includeThinking,
})
.from(chat)
.where(and(eq(chat.identifier, identifier), isNull(chat.archivedAt)))
@@ -216,7 +219,9 @@ export const POST = withRouteHandler(
}
}
- const { createStreamingResponse } = await import('@/lib/workflows/streaming/streaming')
+ const { createStreamingResponse, agentStreamProtocolResponseHeaders } = await import(
+ '@/lib/workflows/streaming/streaming'
+ )
const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow')
const { SSE_HEADERS } = await import('@/lib/core/utils/sse')
@@ -272,17 +277,21 @@ export const POST = withRouteHandler(
variables: (workflowRecord?.variables as Record) ?? undefined,
}
+ const includeThinking = deployment.includeThinking ?? false
const stream = await createStreamingResponse({
requestId,
streamConfig: {
selectedOutputs,
isSecureMode: true,
workflowTriggerType: 'chat',
+ includeThinking,
},
executionId,
workspaceId,
workflowId: deployment.workflowId,
userId: resolvedActorUserId,
+ requestSignal: request.signal,
+ requestHeaders: request.headers,
executeFn: async ({ onStream, onBlockComplete, abortSignal }) =>
executeWorkflow(
workflowForExecution,
@@ -300,6 +309,7 @@ export const POST = withRouteHandler(
abortSignal,
executionMode: 'stream',
billingAttribution,
+ includeThinking,
},
executionId
),
@@ -307,7 +317,13 @@ export const POST = withRouteHandler(
const streamResponse = new NextResponse(stream, {
status: 200,
- headers: SSE_HEADERS,
+ headers: {
+ ...SSE_HEADERS,
+ ...agentStreamProtocolResponseHeaders({
+ includeThinking,
+ requestHeaders: request.headers,
+ }),
+ },
})
return streamResponse
} catch (error: any) {
@@ -344,6 +360,7 @@ export const GET = withRouteHandler(
password: chat.password,
allowedEmails: chat.allowedEmails,
outputConfigs: chat.outputConfigs,
+ includeThinking: chat.includeThinking,
})
.from(chat)
.where(and(eq(chat.identifier, identifier), isNull(chat.archivedAt)))
diff --git a/apps/sim/app/api/chat/manage/[id]/route.ts b/apps/sim/app/api/chat/manage/[id]/route.ts
index 707051d7e9f..44ebc5f4b60 100644
--- a/apps/sim/app/api/chat/manage/[id]/route.ts
+++ b/apps/sim/app/api/chat/manage/[id]/route.ts
@@ -105,6 +105,7 @@ export const PATCH = withRouteHandler(
password,
allowedEmails,
outputConfigs,
+ includeThinking,
} = validatedData
if (workflowId && workflowId !== existingChat[0].workflowId) {
@@ -191,6 +192,10 @@ export const PATCH = withRouteHandler(
updateData.outputConfigs = outputConfigs
}
+ if (includeThinking !== undefined) {
+ updateData.includeThinking = includeThinking
+ }
+
const emailCount = Array.isArray(updateData.allowedEmails)
? updateData.allowedEmails.length
: undefined
@@ -204,6 +209,7 @@ export const PATCH = withRouteHandler(
hasPassword: updateData.password !== undefined,
emailCount,
outputConfigsCount,
+ includeThinking: updateData.includeThinking,
})
await db.update(chat).set(updateData).where(eq(chat.id, chatId))
diff --git a/apps/sim/app/api/chat/route.ts b/apps/sim/app/api/chat/route.ts
index ca1f8019fe7..523efc84c5e 100644
--- a/apps/sim/app/api/chat/route.ts
+++ b/apps/sim/app/api/chat/route.ts
@@ -64,6 +64,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
password,
allowedEmails = [],
outputConfigs = [],
+ includeThinking = false,
} = parsed.data.body
if (authType === 'password' && !password) {
@@ -112,6 +113,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
password,
allowedEmails,
outputConfigs,
+ includeThinking,
workspaceId: workflowRecord.workspaceId,
})
diff --git a/apps/sim/app/api/providers/route.ts b/apps/sim/app/api/providers/route.ts
index 49b90074943..bd7fb52b69a 100644
--- a/apps/sim/app/api/providers/route.ts
+++ b/apps/sim/app/api/providers/route.ts
@@ -24,6 +24,7 @@ import {
} from '@/ee/access-control/utils/permission-check'
import type { StreamingExecution } from '@/executor/types'
import { executeProviderRequest } from '@/providers'
+import { projectStreamingExecutionToByteStream } from '@/providers/stream-pump'
const logger = createLogger('ProvidersAPI')
@@ -238,8 +239,9 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
logger.info(`[${requestId}] Received StreamingExecution from provider`)
// Extract the stream and execution data
- const stream = streamingExec.stream
const executionData = streamingExec.execution
+ // agent-events-v1 is an object stream — project final-turn answer bytes for HTTP.
+ const byteStream = projectStreamingExecutionToByteStream(streamingExec)
// Attach the execution data as a custom header
// We need to safely serialize the execution data to avoid circular references
@@ -288,7 +290,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
}
// Return the stream with execution data in a header
- return new Response(stream, {
+ return new Response(byteStream, {
headers: {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
diff --git a/apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts b/apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts
index 6ea631652f6..102a2f0045a 100644
--- a/apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts
+++ b/apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts
@@ -257,12 +257,15 @@ export const POST = withRouteHandler(
streamConfig: {
selectedOutputs: persistedSnapshot.selectedOutputs,
timeoutMs: preprocessResult.executionTimeout?.sync,
+ includeThinking: persistedSnapshot.metadata.includeThinking === true,
},
executionId: enqueueResult.resumeExecutionId,
workspaceId: workflow.workspaceId || undefined,
workflowId,
userId: enqueueResult.userId,
allowLargeValueWorkflowScope: true,
+ requestSignal: request.signal,
+ requestHeaders: request.headers,
executeFn: async ({ onStream, onBlockComplete, abortSignal }) =>
PauseResumeManager.startResumeExecution({
...resumeArgs,
diff --git a/apps/sim/app/api/workflows/[id]/chat/status/route.test.ts b/apps/sim/app/api/workflows/[id]/chat/status/route.test.ts
index 188a5a47580..6e6cb33da23 100644
--- a/apps/sim/app/api/workflows/[id]/chat/status/route.test.ts
+++ b/apps/sim/app/api/workflows/[id]/chat/status/route.test.ts
@@ -82,6 +82,7 @@ describe('Workflow Chat Status Route', () => {
authType: 'public',
allowedEmails: [],
outputConfigs: [{ blockId: 'agent-1', path: 'content' }],
+ includeThinking: false,
password: 'secret',
isActive: true,
},
@@ -96,5 +97,6 @@ describe('Workflow Chat Status Route', () => {
expect(data.deployment.id).toBe('chat-1')
expect(data.deployment.hasPassword).toBe(true)
expect(data.deployment.outputConfigs).toEqual([{ blockId: 'agent-1', path: 'content' }])
+ expect(data.deployment.includeThinking).toBe(false)
})
})
diff --git a/apps/sim/app/api/workflows/[id]/chat/status/route.ts b/apps/sim/app/api/workflows/[id]/chat/status/route.ts
index 49d555b813d..764f052f8c6 100644
--- a/apps/sim/app/api/workflows/[id]/chat/status/route.ts
+++ b/apps/sim/app/api/workflows/[id]/chat/status/route.ts
@@ -52,6 +52,7 @@ export const GET = withRouteHandler(
authType: chat.authType,
allowedEmails: chat.allowedEmails,
outputConfigs: chat.outputConfigs,
+ includeThinking: chat.includeThinking,
password: chat.password,
isActive: chat.isActive,
})
@@ -71,6 +72,7 @@ export const GET = withRouteHandler(
authType: deploymentResults[0].authType,
allowedEmails: deploymentResults[0].allowedEmails,
outputConfigs: deploymentResults[0].outputConfigs,
+ includeThinking: deploymentResults[0].includeThinking ?? false,
hasPassword: Boolean(deploymentResults[0].password),
}
: null
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts
index a0ca81c0464..c71506f1263 100644
--- a/apps/sim/app/api/workflows/[id]/execute/route.ts
+++ b/apps/sim/app/api/workflows/[id]/execute/route.ts
@@ -87,7 +87,11 @@ import {
loadDeployedWorkflowState,
loadWorkflowFromNormalizedTables,
} from '@/lib/workflows/persistence/utils'
-import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
+import { attachAgentStreamSink } from '@/lib/workflows/streaming/attach-agent-stream-sink'
+import {
+ agentStreamProtocolResponseHeaders,
+ createStreamingResponse,
+} from '@/lib/workflows/streaming/streaming'
import { createHttpResponseFromBlock, workflowHasResponseBlock } from '@/lib/workflows/utils'
import { getWorkspaceBillingSettings } from '@/lib/workspaces/utils'
import { executeWorkflowJob, type WorkflowExecutionPayload } from '@/background/workflow-execution'
@@ -1448,6 +1452,8 @@ async function handleExecutePost(
includeFileBase64,
base64MaxBytes,
timeoutMs: preprocessResult.executionTimeout?.sync,
+ // Workflow API has no chat includeThinking policy — thinking frames stay off.
+ includeThinking: false,
},
executionId,
largeValueExecutionIds,
@@ -1457,6 +1463,8 @@ async function handleExecutePost(
workflowId,
userId: actorUserId,
allowLargeValueWorkflowScope,
+ requestSignal: req.signal,
+ requestHeaders: req.headers,
executeFn: async ({ onStream, onBlockComplete, abortSignal }) =>
executeWorkflow(
streamWorkflow,
@@ -1488,7 +1496,13 @@ async function handleExecutePost(
executionIdClaimCommitted = true
return new NextResponse(stream, {
status: 200,
- headers: SSE_HEADERS,
+ headers: {
+ ...SSE_HEADERS,
+ ...agentStreamProtocolResponseHeaders({
+ includeThinking: false,
+ requestHeaders: req.headers,
+ }),
+ },
})
}
@@ -1529,7 +1543,11 @@ async function handleExecutePost(
event: ExecutionEvent,
terminalStatus?: TerminalExecutionStreamStatus
) => {
- const isBuffered = event.type !== 'stream:chunk' && event.type !== 'stream:done'
+ const isBuffered =
+ event.type !== 'stream:chunk' &&
+ event.type !== 'stream:done' &&
+ event.type !== 'stream:thinking' &&
+ event.type !== 'stream:tool'
let eventToSend = event
if (isBuffered) {
try {
@@ -1727,6 +1745,37 @@ async function handleExecutePost(
const onStream = async (streamingExec: StreamingExecution) => {
const blockId = (streamingExec.execution as any).blockId
+ // Sync window: attach sink before first await so pump delivers thinking/tools.
+ const unsubscribe = attachAgentStreamSink(streamingExec, {
+ onThinkingDelta: async (text) => {
+ await sendEvent({
+ type: 'stream:thinking',
+ timestamp: new Date().toISOString(),
+ executionId,
+ workflowId,
+ data: { blockId, data: text },
+ })
+ },
+ onToolCallStart: async (id, name) => {
+ await sendEvent({
+ type: 'stream:tool',
+ timestamp: new Date().toISOString(),
+ executionId,
+ workflowId,
+ data: { blockId, phase: 'start', id, name },
+ })
+ },
+ onToolCallEnd: async (id, name, status) => {
+ await sendEvent({
+ type: 'stream:tool',
+ timestamp: new Date().toISOString(),
+ executionId,
+ workflowId,
+ data: { blockId, phase: 'end', id, name, status },
+ })
+ },
+ })
+
const reader = streamingExec.stream.getReader()
const decoder = new TextDecoder()
const cancelReader = () => {
@@ -1767,6 +1816,7 @@ async function handleExecutePost(
reqLogger.error('Error streaming block content:', error)
}
} finally {
+ unsubscribe()
timeoutController.signal.removeEventListener('abort', cancelReader)
try {
await reader.cancel().catch(() => {})
diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/components/chat/chat.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/components/chat/chat.tsx
index 43b1ea42d11..9ec3718d571 100644
--- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/components/chat/chat.tsx
+++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/components/chat/chat.tsx
@@ -11,6 +11,7 @@ import {
Label,
Loader,
Skeleton,
+ Switch,
TagInput,
type TagItem,
Textarea,
@@ -83,6 +84,7 @@ const initialFormData: ChatFormData = {
emails: [],
welcomeMessage: 'Hi there! How can I help you today?',
selectedOutputBlocks: [],
+ includeThinking: false,
}
export function ChatDeploy({
@@ -193,6 +195,7 @@ export function ChatDeploy({
(config: { blockId: string; path: string }) => `${config.blockId}_${config.path}`
)
: [],
+ includeThinking: existingChat.includeThinking ?? false,
})
if (existingChat.customizations?.imageUrl) {
@@ -368,6 +371,23 @@ export function ChatDeploy({
)}
+ Allow this chat to stream model thinking when the client opts in. Off by default. +
+