fix(streaming): address validated review findings across provider gating and reset reconciliation

Three-reviewer pass over the branch, findings validated against staging:

- agent-handler forwards agentEvents to executeProviderRequest — the flag was
  computed but dropped in the field-by-field copy, so provider-side thinking
  requests (OpenAI summaries, Gemini includeThoughts, Anthropic summarized
  display) never activated on opted-in runs
- openai: restore summary:'auto' alongside explicit reasoning effort — staging
  always paired them; gating summary purely on agentEvents changed legacy
  payloads
- gemini: Gemini 2 + tools + responseFormat falls back to the silent path;
  the live loop never applied the deferred responseSchema for AUTO tools
- openai-compat loop: malformed tool-argument JSON fails the call instead of
  executing with defaulted {} args (staging parsed inside the execution try)
- openai-compat parser: a vendor id arriving after a synthesized start no
  longer renames the call (start/end ids stayed consistent)
- stream-pump: abort closes the byte projection so a drain blocked on
  backpressure cannot deadlock teardown
- chunk_reset removes the block from the client text order (deployed chat +
  panel chat) so a reset block re-registers at arrival position — fixes
  separator/order corruption when parallel blocks stream around a reset
- resume route echoes the negotiated X-Sim-Stream-Protocol response header
  (parity with the chat route); docs: [DONE] wire shape + final-vs-error
  terminal semantics corrected

Author: icecrasher321

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

@@ -35,10 +35,10 @@ Answer text stays on `chunk`. Thinking and tools never reuse `chunk` (so older c
 | `{ "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": "final", "data": … }` | Terminal result envelope for a settled execution. `data.success` may be `false` with `data.error` when the workflow itself failed |
+| `{ "event": "error", "error": "…" }` | Terminal stream failure (timeout, client abort, processing error) — 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) |
+| `data: "[DONE]"` | Stream closed (JSON-encoded sentinel; always follows the terminal `final` or `error` frame) |
 
 ### Live answer text and intermediate turns
 

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

@@ -186,6 +186,30 @@ describe('useChatStreaming thinking + abort', () => {
     expect(assistant?.content).not.toContain('Let me check')
   })
 
+  it('re-registers a reset block at the end so multi-block order matches arrival', async () => {
+    mockReadSSEEvents.mockImplementation(async (_source, options) => {
+      // Agent A streams provisional text, then resets (tools follow).
+      await options.onEvent({ blockId: 'agent-a', chunk: 'Checking the weather…' })
+      await options.onEvent({ blockId: 'agent-a', event: 'chunk_reset' })
+      // Another block streams while A's tools run.
+      await options.onEvent({ blockId: 'block-b', chunk: 'B output' })
+      // A's final turn re-streams; the server bakes in the cross-block separator.
+      await options.onEvent({ blockId: 'agent-a', chunk: '\n\nIt is 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('B output\n\nIt is 68°F.')
+  })
+
   it('settles thinking chrome when a tool starts', async () => {
     let midStreamThinking: boolean | undefined
     mockReadSSEEvents.mockImplementation(async (_source, options) => {

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

@@ -552,9 +552,16 @@ export function useChatStreaming() {
           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.
+            // Remove the block from the order too: its re-streamed text
+            // re-registers at the end, keeping render order = arrival order
+            // (the server re-computes the cross-block separator on re-stream).
             const { blockId } = json
             if (blockTextSegments.has(blockId)) {
-              blockTextSegments.set(blockId, '')
+              blockTextSegments.delete(blockId)
+              const orderIndex = blockTextOrder.indexOf(blockId)
+              if (orderIndex !== -1) {
+                blockTextOrder.splice(orderIndex, 1)
+              }
               recomputeAccumulatedText()
               // Spoken audio cannot be unplayed; clamp so slicing stays valid.
               lastAudioPosition = Math.min(lastAudioPosition, accumulatedText.length)

apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts

@@ -20,7 +20,10 @@ import { getBaseUrl } from '@/lib/core/utils/urls'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
-import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
+import {
+  agentStreamProtocolResponseHeaders,
+  createStreamingResponse,
+} from '@/lib/workflows/streaming/streaming'
 import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
 import type { ResumeExecutionPayload } from '@/background/resume-execution'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
@@ -278,6 +281,11 @@ export const POST = withRouteHandler(
         return new NextResponse(stream, {
           headers: {
             ...SSE_HEADERS,
+            // Echo the negotiated stream protocol (same as the public chat route).
+            ...agentStreamProtocolResponseHeaders({
+              includeThinking: persistedSnapshot.metadata.includeThinking === true,
+              requestHeaders: request.headers,
+            }),
             'X-Execution-Id': enqueueResult.resumeExecutionId,
           },
         })

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

@@ -567,8 +567,15 @@ export function Chat() {
             }
 
             if (event === 'chunk_reset' && blockId) {
+              // Drop the block's provisional text and its order slot — the
+              // final turn re-registers at the end, keeping render order =
+              // arrival order (separators are recomputed on re-stream).
               if (blockSegments.has(blockId)) {
-                blockSegments.set(blockId, '')
+                blockSegments.delete(blockId)
+                const orderIndex = blockOrder.indexOf(blockId)
+                if (orderIndex !== -1) {
+                  blockOrder.splice(orderIndex, 1)
+                }
                 recomputeContent()
                 contentDirty = true
                 scheduleFlush()

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts

@@ -634,12 +634,13 @@ export function useWorkflowExecution() {
 
             /**
              * Intermediate-turn reconciliation: drop the block's streamed text
-             * (chunk_reset frame) and re-arm separator bookkeeping so the
-             * final turn's text starts a clean segment.
+             * (chunk_reset frame) and remove its bookkeeping entirely so
+             * separator counting ignores it and the final turn (or, if none
+             * re-streams, onBlockComplete's output fallback) starts clean.
              */
             const onStreamReset = (blockId: string) => {
               if (!streamedChunks.has(blockId)) return
-              streamedChunks.set(blockId, [])
+              streamedChunks.delete(blockId)
               processedFirstChunk.delete(blockId)
               safeEnqueue(encodeSSE({ blockId, event: 'chunk_reset' }))
             }
@@ -1249,7 +1250,7 @@ export function useWorkflowExecution() {
             onStreamChunkReset: (data) => {
               // Live-streamed text belonged to an intermediate turn (tools
               // follow); the final turn re-streams as regular chunks.
-              streamedChunks.set(data.blockId, [])
+              streamedChunks.delete(data.blockId)
               if (onStreamReset && isExecutingFromChat) {
                 onStreamReset(data.blockId)
               }

apps/sim/executor/handlers/agent/agent-handler.test.ts

@@ -1875,6 +1875,52 @@ describe('AgentBlockHandler', () => {
       expect(providerCallArgs.billingAttribution).toEqual(billingAttribution)
     })
 
+    it('forwards agentEvents to executeProviderRequest on opted-in streaming runs', async () => {
+      const inputs = {
+        model: 'gpt-4o',
+        userPrompt: 'Stream this',
+        apiKey: 'test-api-key',
+      }
+
+      const streamingContext = {
+        ...mockContext,
+        stream: true,
+        selectedOutputs: ['test-agent-block'],
+        metadata: { ...mockContext.metadata, agentEvents: true },
+      } as ExecutionContext
+
+      mockGetProviderFromModel.mockReturnValue('openai')
+
+      await handler.execute(streamingContext, mockBlock, inputs)
+
+      expect(mockExecuteProviderRequest).toHaveBeenCalled()
+      const providerCallArgs = mockExecuteProviderRequest.mock.calls[0][1]
+      expect(providerCallArgs.stream).toBe(true)
+      expect(providerCallArgs.agentEvents).toBe(true)
+    })
+
+    it('does not set agentEvents on runs without the run-level opt-in', async () => {
+      const inputs = {
+        model: 'gpt-4o',
+        userPrompt: 'Stream this',
+        apiKey: 'test-api-key',
+      }
+
+      const streamingContext = {
+        ...mockContext,
+        stream: true,
+        selectedOutputs: ['test-agent-block'],
+      } as ExecutionContext
+
+      mockGetProviderFromModel.mockReturnValue('openai')
+
+      await handler.execute(streamingContext, mockBlock, inputs)
+
+      expect(mockExecuteProviderRequest).toHaveBeenCalled()
+      const providerCallArgs = mockExecuteProviderRequest.mock.calls[0][1]
+      expect(providerCallArgs.agentEvents).toBe(false)
+    })
+
     it('should handle multiple MCP tools from the same server efficiently', async () => {
       const fetchCalls: any[] = []
 

apps/sim/executor/handlers/agent/agent-handler.ts

@@ -1042,6 +1042,7 @@ export class AgentBlockHandler implements BlockHandler {
         verbosity: providerRequest.verbosity,
         thinkingLevel: providerRequest.thinkingLevel,
         previousInteractionId: providerRequest.previousInteractionId,
+        agentEvents: providerRequest.agentEvents,
         streamToolCalls: providerRequest.streamToolCalls,
         abortSignal: ctx.abortSignal,
       })

apps/sim/providers/gemini/core.ts

@@ -1024,7 +1024,16 @@ export async function executeGeminiRequest(
     }
 
     const initialCallTime = Date.now()
-    const shouldStreamToolCalls = request.streamToolCalls ?? false
+    /**
+     * Gemini 2 cannot combine responseSchema with tools, so structured output
+     * is applied on a final schema-configured request after tools settle — the
+     * silent path does this; the live loop would break as soon as a turn has
+     * no calls and skip the schema. Gemini 3 carries responseJsonSchema
+     * alongside tools, so its live loop keeps structured output.
+     */
+    const responseFormatNeedsFinalPass = Boolean(request.responseFormat) && !isGemini3Model(model)
+    const shouldStreamToolCalls =
+      (request.streamToolCalls ?? false) && !responseFormatNeedsFinalPass
     const shouldStream = request.stream && !tools?.length
 
     // Live streaming tool loop

apps/sim/providers/openai-compat/stream-events.test.ts

@@ -95,6 +95,47 @@ describe('createOpenAICompatibleAgentEventStream', () => {
     ])
   })
 
+  it('keeps a synthesized tool id stable when the vendor id arrives after start', async () => {
+    const onComplete = vi.fn()
+    const stream = createOpenAICompatibleAgentEventStream(
+      (async function* () {
+        // Name first (no id) → start emits with a synthesized id.
+        yield {
+          choices: [
+            {
+              delta: {
+                tool_calls: [{ index: 0, function: { name: 'lookup', arguments: '{"q"' } }],
+              },
+            },
+          ],
+        } as any
+        // Vendor id arrives late — must not rename the started call.
+        yield {
+          choices: [
+            {
+              delta: {
+                tool_calls: [{ index: 0, id: 'call_real', function: { arguments: ':1}' } }],
+              },
+            },
+          ],
+          usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
+        } as any
+      })(),
+      { providerName: 'Loose', emitToolCallStarts: true, onComplete }
+    )
+
+    const events = await collectEvents(stream)
+    const start = events.find((e) => e.type === 'tool_call_start')
+    expect(start).toBeDefined()
+    expect(start!.id).not.toBe('call_real')
+
+    // The assembled call keeps the same id as the emitted start.
+    const assembled = onComplete.mock.calls[0][0].toolCalls
+    expect(assembled).toHaveLength(1)
+    expect(assembled[0].id).toBe(start!.id)
+    expect(assembled[0].function).toEqual({ name: 'lookup', arguments: '{"q":1}' })
+  })
+
   it('always enqueues text deltas live and assembles content for onComplete', async () => {
     const onComplete = vi.fn()
     const stream = createOpenAICompatibleAgentEventStream(