feat(streaming): stream answer text live during tool loops via turn_end protocol

The live tool loops buffered all answer text per model turn (classification
of intermediate vs final is only known at turn end), so gated surfaces saw
thinking stream, then dead air with the thinking chrome stuck open, then the
whole answer at once.

Loops now emit text deltas live as `turn: 'pending'` plus a `turn_end`
event per turn. The pump buffers pending text and projects it to the byte
path (answerText/logs/memory/legacy clients) only on a final turn_end, so
all settled semantics are unchanged. Gated surfaces render the pending text
as it streams and reconcile with a reset when a turn resolves to tools:

- public chat: live `chunk` frames from the sink + dual-gated `chunk_reset`;
  byte-path frame emission is suppressed to avoid duplicates (kept for
  response-format transformed streams via clientStreamTransformed)
- canvas: forwarder emits live `stream:chunk` + `stream:chunk_reset`; the
  execute route and HITL resume readers stop re-emitting byte chunks; panel
  chat tracks per-block segments and replaces content on flush
- chat client: per-block text segments, chunk_reset handling, and thinking
  chrome now settles on tool start as well as first answer chunk

Author: icecrasher321

apps/docs/content/docs/en/workflows/deployment/agent-events.mdx

@@ -31,33 +31,40 @@ Answer text stays on `chunk`. Thinking and tools never reuse `chunk` (so older c
 
 | Frame | Meaning |
 |-------|---------|
-| `{ "blockId", "chunk": "…" }` | Final-turn answer text only |
+| `{ "blockId", "chunk": "…" }` | Answer text. Legacy clients receive settled final-turn text; opted-in clients receive it live as the model generates (see below) |
+| `{ "blockId", "event": "chunk_reset" }` | Opted-in only: discard the block’s streamed answer text — it belonged to a turn that resolved to tool calls |
 | `{ "blockId", "event": "thinking", "data": "…" }` | Thinking / reasoning summary delta |
 | `{ "blockId", "event": "tool", "phase": "start"\|"end", "id", "name", "status?" }` | Tool lifecycle (no args / results) |
 | `{ "event": "final", "data": … }` | Terminal success envelope |
 | `{ "event": "error", "error": "…" }` | Terminal failure — followed by `[DONE]`, never by `final` |
 | `{ "event": "stream_error", "blockId?", "error" }` | Non-terminal mid-block read issue; the stream keeps going |
 | `data: [DONE]` | Stream closed (always follows the terminal `final` or `error` frame) |
 
-### Intermediate-turn text
+### Live answer text and intermediate turns
 
-During a live tool loop, the model may emit preamble text before calling a tool. That text is tagged **intermediate** internally and is **not** projected into the user-facing answer channel. Only the **final** turn’s text appears as `chunk`.
+During a live tool loop, the model can’t be classified mid-turn: text it emits may turn out to be the final answer or preamble before a tool call (the stop reason arrives only at turn end).
+
+- **Opted-in clients** (protocol header + `includeThinking`) receive answer text as `chunk` frames **live**, token by token. If the turn then resolves to tool calls, a `chunk_reset` frame tells the client to discard that block’s streamed text — the final turn re-streams live after tools settle. Append `chunk`, honor `chunk_reset`, and the displayed answer always converges to the block’s final content.
+- **Legacy clients** (no header) never see provisional text: only settled final-turn text is emitted as `chunk`, delivered when the turn completes.
+
+Logs, memory, and the block’s `content` output always contain final-turn text only — intermediate preamble is never persisted.
 
 ### Abort
 
 Client disconnect or Stop aborts the provider stream. In-flight tools settle as `cancelled`. Cancel is distinct from execution timeout.
 
 ### Reconnect
 
-Canvas execution-events `stream:chunk`, `stream:thinking`, and `stream:tool` are **live-only** (not buffered for reconnect replay), same as answer chunks. Guaranteed `seq` replay is out of scope.
+Canvas execution-events `stream:chunk`, `stream:chunk_reset`, `stream:thinking`, and `stream:tool` are **live-only** (not buffered for reconnect replay), same as answer chunks. Guaranteed `seq` replay is out of scope.
 
 ## Canvas (draft Run)
 
 When you click **Run** in the builder, the execution-events SSE path forwards the same sink. The canvas is always opted in — it does not send (or need) the `X-Sim-Stream-Protocol` header; the dual gate applies only to the public chat / simple SSE surface:
 
 - `stream:thinking` — `{ blockId, text }`
 - `stream:tool` — `{ blockId, phase, id, name, status? }`
-- `stream:chunk` — answer text (unchanged)
+- `stream:chunk` — answer text, live on agent-events runs
+- `stream:chunk_reset` — `{ blockId }`; discard the block’s streamed text (intermediate turn)
 
 The terminal output panel shows Thinking / Tools chrome above the block output when those events arrive. PII redaction on block output still disables live forwarding (executor rule).
 

apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx

@@ -147,6 +147,82 @@ describe('useChatStreaming thinking + abort', () => {
     expect(assistant?.isThinkingStreaming).toBe(false)
   })
 
+  it('clears a block’s live text on chunk_reset and keeps the re-streamed final turn', async () => {
+    mockReadSSEEvents.mockImplementation(async (_source, options) => {
+      // Turn 1: live preamble, then tools follow → reset.
+      await options.onEvent({ blockId: 'agent-1', chunk: 'Let me check the weather…' })
+      await options.onEvent({
+        blockId: 'agent-1',
+        event: 'tool',
+        phase: 'start',
+        id: 't1',
+        name: 'get_weather',
+      })
+      await options.onEvent({ blockId: 'agent-1', event: 'chunk_reset' })
+      await options.onEvent({
+        blockId: 'agent-1',
+        event: 'tool',
+        phase: 'end',
+        id: 't1',
+        name: 'get_weather',
+        status: 'success',
+      })
+      // Turn 2: final answer streams live.
+      await options.onEvent({ blockId: 'agent-1', chunk: 'It is ' })
+      await options.onEvent({ blockId: 'agent-1', chunk: '68°F.' })
+      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('It is 68°F.')
+    expect(assistant?.content).not.toContain('Let me check')
+  })
+
+  it('settles thinking chrome when a tool starts', async () => {
+    let midStreamThinking: boolean | undefined
+    mockReadSSEEvents.mockImplementation(async (_source, options) => {
+      await options.onEvent({ blockId: 'agent-1', event: 'thinking', data: 'planning…' })
+      await options.onEvent({
+        blockId: 'agent-1',
+        event: 'tool',
+        phase: 'start',
+        id: 't1',
+        name: 'get_weather',
+      })
+      // UI flush is synchronous in tests (rAF mocked) — capture mid-stream state.
+      midStreamThinking = messages.find((m) => m.type === 'assistant')?.isThinkingStreaming
+      await options.onEvent({
+        blockId: 'agent-1',
+        event: 'tool',
+        phase: 'end',
+        id: 't1',
+        name: 'get_weather',
+        status: 'success',
+      })
+      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(midStreamThinking).toBe(false)
+    const assistant = messages.find((m) => m.id === 'msg-assistant-1')
+    expect(assistant?.thinking).toBe('planning…')
+  })
+
   it('ignores non-terminal stream_error frames and keeps streaming', async () => {
     mockReadSSEEvents.mockImplementation(async (_source, options) => {
       await options.onEvent({

apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts

@@ -14,6 +14,7 @@ import { readSSEEvents } from '@/lib/core/utils/sse'
 import { isUserFileWithMetadata } from '@/lib/core/utils/user-file'
 import {
   isChatChunkFrame,
+  isChatChunkResetFrame,
   isChatErrorFrame,
   isChatFinalFrame,
   isChatStreamErrorFrame,
@@ -182,7 +183,19 @@ export function useChatStreaming() {
       return
     }
 
+    /**
+     * Answer text tracked per block so a `chunk_reset` (dual-gated streams:
+     * a live-streamed turn resolved to tool calls) can clear one block's
+     * contribution. `accumulatedText` is re-derived on every mutation —
+     * cross-block separators arrive baked into the chunks.
+     */
+    const blockTextOrder: string[] = []
+    const blockTextSegments = new Map<string, string>()
     let accumulatedText = ''
+    const recomputeAccumulatedText = () => {
+      accumulatedText = blockTextOrder.map((id) => blockTextSegments.get(id) ?? '').join('')
+      accumulatedTextRef.current = accumulatedText
+    }
     let accumulatedThinking = ''
     let isThinkingStreaming = false
     let lastAudioPosition = 0
@@ -356,6 +369,11 @@ export function useChatStreaming() {
             if (!messageIdMap.has(blockId)) {
               messageIdMap.set(blockId, messageId)
             }
+            // Tools starting means the turn's thinking phase is over — settle
+            // the thinking chrome (it re-opens if more thinking streams later).
+            if (json.phase === 'start' && isThinkingStreaming) {
+              isThinkingStreaming = false
+            }
             applyToolCallPhase(
               toolCallsMap,
               toolCallOrder,
@@ -531,6 +549,21 @@ export function useChatStreaming() {
             return true
           }
 
+          if (isChatChunkResetFrame(json)) {
+            // The block's live-streamed text belonged to an intermediate turn
+            // (tool calls follow); drop it — the final turn re-streams after.
+            const { blockId } = json
+            if (blockTextSegments.has(blockId)) {
+              blockTextSegments.set(blockId, '')
+              recomputeAccumulatedText()
+              // Spoken audio cannot be unplayed; clamp so slicing stays valid.
+              lastAudioPosition = Math.min(lastAudioPosition, accumulatedText.length)
+              uiDirty = true
+              scheduleUIFlush()
+            }
+            return false
+          }
+
           // Answer text only — never append thinking/tool/unknown chunk frames blindly.
           if (isChatChunkFrame(json)) {
             const { blockId, chunk: contentChunk } = json
@@ -543,8 +576,12 @@ export function useChatStreaming() {
               isThinkingStreaming = false
             }
 
-            accumulatedText += contentChunk
-            accumulatedTextRef.current = accumulatedText
+            if (!blockTextSegments.has(blockId)) {
+              blockTextOrder.push(blockId)
+              blockTextSegments.set(blockId, '')
+            }
+            blockTextSegments.set(blockId, blockTextSegments.get(blockId)! + contentChunk)
+            recomputeAccumulatedText()
             logger.debug('[useChatStreaming] Received chunk', {
               blockId,
               chunkLength: contentChunk.length,

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -88,7 +88,10 @@ import {
   loadWorkflowDeploymentVersionState,
   loadWorkflowFromNormalizedTables,
 } from '@/lib/workflows/persistence/utils'
-import { forwardAgentStreamToExecutionEvents } from '@/lib/workflows/streaming/forward-agent-stream-events'
+import {
+  forwardAgentStreamToExecutionEvents,
+  shouldForwardAnswerTextFromSink,
+} from '@/lib/workflows/streaming/forward-agent-stream-events'
 import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
 import { createHttpResponseFromBlock, workflowHasResponseBlock } from '@/lib/workflows/utils'
 import { getWorkspaceBillingSettings } from '@/lib/workspaces/utils'
@@ -1725,12 +1728,19 @@ async function handleExecutePost(
           const onStream = async (streamingExec: StreamingExecution) => {
             const blockId = (streamingExec.execution as any).blockId
 
+            // Live answer text rides the sink when available (pending deltas
+            // stream as the model generates; chunk_reset clears intermediate
+            // turns). The byte stream is then drained without re-emitting
+            // chunks — its text is the same final-turn content.
+            const answerTextFromSink = shouldForwardAnswerTextFromSink(streamingExec)
+
             // Sync window: attach sink before first await so pump delivers thinking/tools.
             const unsubscribe = forwardAgentStreamToExecutionEvents(streamingExec, {
               blockId,
               executionId,
               workflowId,
               sendEvent,
+              forwardAnswerText: answerTextFromSink,
             })
 
             const reader = streamingExec.stream.getReader()
@@ -1749,6 +1759,8 @@ async function handleExecutePost(
                 if (timeoutController.signal.aborted || isStreamClosed) break
                 if (done) break
 
+                if (answerTextFromSink) continue
+
                 const chunk = decoder.decode(value, { stream: true })
                 await sendEvent({
                   type: 'stream:chunk',

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/chat.tsx

@@ -238,6 +238,7 @@ export function Chat() {
     selectedWorkflowOutputs,
     setSelectedWorkflowOutput,
     appendMessageContent,
+    setMessageContent,
     finalizeMessageStream,
     getConversationId,
     clearChat,
@@ -256,6 +257,7 @@ export function Chat() {
       selectedWorkflowOutputs: s.selectedWorkflowOutputs,
       setSelectedWorkflowOutput: s.setSelectedWorkflowOutput,
       appendMessageContent: s.appendMessageContent,
+      setMessageContent: s.setMessageContent,
       finalizeMessageStream: s.finalizeMessageStream,
       getConversationId: s.getConversationId,
       clearChat: s.clearChat,
@@ -495,10 +497,21 @@ export function Chat() {
     async (stream: ReadableStream<Uint8Array>, responseMessageId: string) => {
       const reader = stream.getReader()
       streamReaderRef.current = reader
+
+      /**
+       * Answer text tracked per block so a `chunk_reset` frame (a live-streamed
+       * turn resolved to tool calls) can drop one block's contribution. Each
+       * flush replaces the message content with the joined segments.
+       */
+      const blockOrder: string[] = []
+      const blockSegments = new Map<string, string>()
       let accumulatedContent = ''
+      const recomputeContent = () => {
+        accumulatedContent = blockOrder.map((id) => blockSegments.get(id) ?? '').join('')
+      }
 
       const BATCH_MAX_MS = 50
-      let pendingChunks = ''
+      let contentDirty = false
       let batchRAF: number | null = null
       let batchTimer: ReturnType<typeof setTimeout> | null = null
       let lastFlush = 0
@@ -512,9 +525,9 @@ export function Chat() {
           clearTimeout(batchTimer)
           batchTimer = null
         }
-        if (pendingChunks) {
-          appendMessageContent(responseMessageId, pendingChunks)
-          pendingChunks = ''
+        if (contentDirty) {
+          setMessageContent(responseMessageId, accumulatedContent)
+          contentDirty = false
         }
         lastFlush = performance.now()
       }
@@ -534,12 +547,17 @@ export function Chat() {
 
       let finalError: string | null = null
       try {
-        await readSSEEvents<{ event?: string; data?: ExecutionResult; chunk?: string }>(reader, {
+        await readSSEEvents<{
+          event?: string
+          data?: ExecutionResult
+          chunk?: string
+          blockId?: string
+        }>(reader, {
           onParseError: (_data, e) => {
             logger.error('Error parsing stream data:', e)
           },
           onEvent: (json) => {
-            const { event, data: eventData, chunk: contentChunk } = json
+            const { event, data: eventData, chunk: contentChunk, blockId } = json
 
             if (event === 'final' && eventData) {
               if ('success' in eventData && !eventData.success) {
@@ -548,9 +566,25 @@ export function Chat() {
               return true
             }
 
+            if (event === 'chunk_reset' && blockId) {
+              if (blockSegments.has(blockId)) {
+                blockSegments.set(blockId, '')
+                recomputeContent()
+                contentDirty = true
+                scheduleFlush()
+              }
+              return
+            }
+
             if (contentChunk) {
-              accumulatedContent += contentChunk
-              pendingChunks += contentChunk
+              const segmentKey = blockId ?? ''
+              if (!blockSegments.has(segmentKey)) {
+                blockOrder.push(segmentKey)
+                blockSegments.set(segmentKey, '')
+              }
+              blockSegments.set(segmentKey, blockSegments.get(segmentKey)! + contentChunk)
+              recomputeContent()
+              contentDirty = true
               scheduleFlush()
             }
           },
@@ -578,7 +612,14 @@ export function Chat() {
         focusInput(100)
       }
     },
-    [appendMessageContent, finalizeMessageStream, focusInput, selectedOutputs, activeWorkflowId]
+    [
+      appendMessageContent,
+      setMessageContent,
+      finalizeMessageStream,
+      focusInput,
+      selectedOutputs,
+      activeWorkflowId,
+    ]
   )
 
   /**