fix(chat): reconcile structured output delivery

Author: icecrasher321

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

@@ -144,6 +144,30 @@ describe('useChatStreaming thinking + abort', () => {
     expect(messages[0].isStreaming).toBe(false)
   })
 
+  it('projects file metadata to the fields used by the chat', async () => {
+    mockReadSSEEvents.mockImplementation(async (_source, options) => {
+      await options.onEvent({
+        blockId: 'agent-1',
+        event: 'output',
+        data: {
+          ...imageFile,
+          providerFileId: 'provider-file-1',
+          providerFileUri: 'provider://file-1',
+          remoteUrl: 'https://files.example.com/signed',
+          internalMetadata: { source: 'provider' },
+        },
+      })
+      await options.onEvent({ event: 'final', data: { success: true, output: {} } })
+    })
+
+    await act(async () => {
+      await handle.latest().handleStreamedResponse(makeSseResponse(), setMessages, vi.fn(), vi.fn())
+    })
+
+    expect(messages[0].files).toEqual([imageFile])
+    expect(messages[0].content).toBe('')
+  })
+
   it.each([{ data: [] }, { data: null }, { data: { files: [] } }])(
     'keeps empty structured outputs invisible: $data',
     async ({ data }) => {

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

@@ -37,7 +37,16 @@ const logger = createLogger('UseChatStreaming')
 function extractChatOutput(value: unknown, files: Map<string, ChatFile>): unknown {
   if (value === null || value === undefined) return value
   if (isUserFileWithMetadata(value)) {
-    files.set(value.id, value)
+    files.set(value.id, {
+      id: value.id,
+      name: value.name,
+      url: value.url,
+      key: value.key,
+      size: value.size,
+      type: value.type,
+      context: value.context,
+      base64: value.base64,
+    })
     return undefined
   }
   if (Array.isArray(value)) {

apps/sim/lib/workflows/streaming/streaming.test.ts

@@ -152,6 +152,54 @@ describe('createStreamingResponse', () => {
     expect(events).toContainEqual({ blockId: 'agent', event: 'output', data: [file] })
   })
 
+  it('emits composite response-format selections once as structured outputs', async () => {
+    const result = {
+      content: 'Your image.',
+      files: [
+        {
+          id: 'file-image',
+          name: 'image.png',
+          size: 3,
+          type: 'image/png',
+          key: 'execution/image.png',
+          url: '/api/files/serve/execution%2Fimage.png',
+          base64: 'YWJj',
+        },
+      ],
+      count: 1,
+    }
+    const stream = await createStreamingResponse({
+      requestId: 'request-chat-composite',
+      requestHeaders: new Headers({ [AGENT_STREAM_PROTOCOL_HEADER]: CHAT_OUTPUT_PROTOCOL_V1 }),
+      streamConfig: {
+        selectedOutputs: ['agent_result'],
+        workflowTriggerType: 'chat',
+        includeFileBase64: false,
+      },
+      executeFn: async ({ onStream, onBlockComplete }) => {
+        await onStream({
+          blockId: 'agent',
+          clientStreamTransformed: true,
+          stream: new ReadableStream({
+            start(controller) {
+              controller.enqueue(new TextEncoder().encode(JSON.stringify(result)))
+              controller.close()
+            },
+          }),
+          execution: { success: true, output: {} },
+        })
+        await onBlockComplete('agent', { result })
+        return { success: true, output: {}, logs: [], metadata: { duration: 1 } }
+      },
+    })
+
+    const events = await collectSSEEvents(stream)
+    expect(events.filter((event) => 'chunk' in event)).toEqual([])
+    expect(events.filter((event) => event.event === 'output')).toEqual([
+      { blockId: 'agent', event: 'output', data: result },
+    ])
+  })
+
   it('enforces the aggregate inline byte limit for structured chat outputs', async () => {
     const value = { text: 'x'.repeat(9 * 1024 * 1024) }
     const stream = await createStreamingResponse({

apps/sim/lib/workflows/streaming/streaming.ts

@@ -130,6 +130,7 @@ interface StreamingState {
   processedOutputs: Set<string>
   streamCompletionTimes: Map<string, number>
   completedBlockIds: Set<string>
+  deferredOutputBlocks: Set<string>
   selectedOutputBytes: number
   streamedSelectedOutputKeys: Set<string>
   selectedOutputError?: string
@@ -563,6 +564,7 @@ export async function createStreamingResponse(
         processedOutputs: new Set(),
         streamCompletionTimes: new Map(),
         completedBlockIds: new Set(),
+        deferredOutputBlocks: new Set(),
         selectedOutputBytes: 0,
         streamedSelectedOutputKeys: new Set(),
       }
@@ -632,6 +634,11 @@ export async function createStreamingResponse(
           return
         }
 
+        /** Response-format streams contain complete selected values; send their typed outputs once. */
+        const deferSelectedOutputs =
+          emitStructuredOutputs && streamingExec.clientStreamTransformed === true
+        if (deferSelectedOutputs) state.deferredOutputBlocks.add(blockId)
+
         /**
          * Negotiated clients get answer text live from the sink (pending deltas
          * stream as the model generates; `chunk_reset` clears an intermediate
@@ -641,8 +648,8 @@ export async function createStreamingResponse(
          * only writes once the turn is classified — correct for a consumer that
          * cannot retract, at the cost of arriving in one piece.
          *
-         * Response-format projections rewrite the bytes, so those blocks keep
-         * the byte stream as the frame source either way.
+         * Response-format projections rewrite the bytes. Typed-output clients
+         * receive those selections from onBlockComplete instead.
          */
         const sinkAnswerText =
           clientAcceptsProtocol &&
@@ -708,7 +715,7 @@ export async function createStreamingResponse(
             }
             state.streamedChunks.get(blockId)!.push(textChunk)
 
-            if (!sinkAnswerText) {
+            if (!sinkAnswerText && !deferSelectedOutputs) {
               emitAnswerChunk(textChunk)
             }
           }
@@ -743,7 +750,9 @@ export async function createStreamingResponse(
           return
         }
 
-        const hasStreamedText = state.streamedChunks.has(selectedOutputBlockId)
+        const hasStreamedText =
+          state.streamedChunks.has(selectedOutputBlockId) &&
+          !state.deferredOutputBlocks.has(selectedOutputBlockId)
         if (hasStreamedText && !emitStructuredOutputs) {
           return
         }