feat(streaming): stream custom block outputs in chat and Slack

apps/docs/content/docs/platform/enterprise/custom-blocks.mdx

@@ -72,6 +72,14 @@ Pick which of the workflow's outputs consumers can use, and give each one a name
 - **At least one output is required.**
 - Each exposed output needs a unique **name** (max 60 characters) — that's the name consumers reference in their workflows.
 
+For an Agent or Pi block's plain-text `content` output, choose **Live** to publish it as a streaming answer. Outputs default to **Final**, which returns the value when the custom block finishes. Structured response formats and other source output types cannot be marked live.
+
+In the consuming workflow, select that public field in the deployed chat or Slack streaming output picker. Live fields are labeled **(live)**. References stay the same: a block named `Research` with an output named `answer` uses `research.answer`. Consumers do not need to select or access the source workflow's internal blocks.
+
+Live output exposes only the selected answer text. It works independently of **Trace runs in consumer logs**, and does not expose the source's thinking or tool events. Chat can update a provisional answer during tool use; Slack waits until a turn is known to be the final answer because it cannot retract streamed text. Slack marks the response complete when the custom block finishes, and marks it as failed if the block fails after text has arrived. Downstream workflow blocks still receive the completed output value.
+
+Streaming mappings are checked against the latest deployment when saving and executing. If you remove a live source or give it a structured response format, update the custom block's output configuration before consumers run it again.
+
 <Image src="/static/enterprise/custom-blocks-form.png" alt="Create block form filled in: Workspace and Workflow selectors, an uploaded icon, Name and Description fields, an expanded input with a placeholder, and two selected outputs each given a name" width={900} height={570} />
 
 ### 6. Choose whether runs are traced

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

@@ -29,6 +29,14 @@ Agent events are governed by two things that do not depend on each other.
 X-Sim-Stream-Protocol: agent-events-v1
 ```
 
+To receive provisional custom-block answers, also declare `scoped-output-v1`:
+
+```http
+X-Sim-Stream-Protocol: agent-events-v1, scoped-output-v1
+```
+
+With this capability, key answer chunks and `chunk_reset` frames by `streamId` when present, otherwise by `blockId`. Each custom-block field stream has its own opaque ID, so resetting one answer preserves other fields and invocations. Clients that omit this capability receive settled custom-block text without custom-block retractions. The deployed chat client negotiates both capabilities automatically.
+
 It does two things. It switches answer text to live token-by-token `chunk` frames that `chunk_reset` can retract, and it is **required** for any `thinking` or `tool` frame — a client that never declared a version has no contract for their shape, so it keeps the text-only stream it already understands.
 
 Omitting the header is always valid and always safe: you get settled final-turn text and no agent-event frames, which is what every pre-existing integration receives. The response echoes the header back when the protocol was negotiated.

apps/sim/app/(interfaces)/chat/[identifier]/chat.tsx

@@ -6,6 +6,7 @@ import { generateId } from '@sim/utils/id'
 import {
   AGENT_STREAM_PROTOCOL_HEADER,
   AGENT_STREAM_PROTOCOL_V1,
+  SCOPED_OUTPUT_STREAM_PROTOCOL_V1,
 } from '@/lib/workflows/streaming/agent-stream-protocol'
 import { DesktopTitleBarLane } from '@/app/_shell/desktop-title-bar'
 import {
@@ -236,7 +237,7 @@ export default function ChatClient({ identifier }: { identifier: string }) {
         headers: {
           'Content-Type': 'application/json',
           'X-Requested-With': 'XMLHttpRequest',
-          [AGENT_STREAM_PROTOCOL_HEADER]: AGENT_STREAM_PROTOCOL_V1,
+          [AGENT_STREAM_PROTOCOL_HEADER]: `${AGENT_STREAM_PROTOCOL_V1}, ${SCOPED_OUTPUT_STREAM_PROTOCOL_V1}`,
         },
         body: JSON.stringify(payload),
         credentials: 'same-origin',

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

@@ -210,6 +210,23 @@ describe('useChatStreaming thinking + abort', () => {
     expect(assistant?.content).toBe('B output\n\nIt is 68°F.')
   })
 
+  it('retracts only the matching custom field stream and preserves sibling invocations', async () => {
+    mockReadSSEEvents.mockImplementation(async (_source, options) => {
+      await options.onEvent({ blockId: 'custom', streamId: 'first', chunk: 'Provisional' })
+      await options.onEvent({ blockId: 'custom', streamId: 'second', chunk: 'Other answer' })
+      await options.onEvent({ blockId: 'custom', streamId: 'first', event: 'chunk_reset' })
+      await options.onEvent({ blockId: 'custom', streamId: 'first', chunk: '\n\nFinal 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()
+    expect(messages.find((message) => message.id === 'msg-assistant-1')?.content).toBe(
+      'Other answer\n\nFinal answer'
+    )
+  })
+
   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

@@ -525,7 +525,7 @@ export function useChatStreaming() {
             // 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
+            const blockId = json.streamId ?? json.blockId
             if (blockTextSegments.has(blockId)) {
               blockTextSegments.delete(blockId)
               const orderIndex = blockTextOrder.indexOf(blockId)
@@ -541,7 +541,8 @@ export function useChatStreaming() {
 
           // Answer text only — never append thinking/tool/unknown chunk frames blindly.
           if (isChatChunkFrame(json)) {
-            const { blockId, chunk: contentChunk } = json
+            const { chunk: contentChunk } = json
+            const blockId = json.streamId ?? json.blockId
 
             // First answer chunk settles thinking chrome (still visible, no longer “live”).
             if (isThinkingStreaming) {

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

@@ -294,7 +294,7 @@ function OutputSelectMenu({
     ),
     items: [
       ...node.outputs.map((output) => ({
-        label: output.path,
+        label: output.streaming ? `${output.path} (live)` : output.path,
         value: getOutputValue(output, valueMode),
       })),
       ...(node.children.length > 0 ? [folderOption(node)] : []),

apps/sim/blocks/custom/build-config.test.ts

@@ -111,6 +111,22 @@ describe('buildCustomBlockConfig', () => {
     expect(config.outputs.childTraceSpans).toBeUndefined()
   })
 
+  it('advertises a live public field as text for deployment output pickers', () => {
+    const config = buildCustomBlockConfig(
+      {
+        ...row,
+        exposedOutputs: [{ blockId: 'agent', path: 'content', name: 'answer', streaming: true }],
+      },
+      [],
+      { icon }
+    )
+    expect(config.outputs.answer).toEqual({
+      type: 'string',
+      description: 'Streaming text output',
+      streaming: true,
+    })
+  })
+
   it('exposes only curated outputs as named fields', () => {
     const config = buildCustomBlockConfig(
       { ...row, exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'email' }] },

apps/sim/blocks/custom/build-config.ts

@@ -37,6 +37,8 @@ export interface CustomBlockOutput {
   blockId: string
   path: string
   name: string
+  /** Expose the source's answer text while this invocation is running. */
+  streaming?: boolean
 }
 
 /**
@@ -63,7 +65,7 @@ export interface CustomBlockRow {
   workflowId: string
   /** Source workflow's home workspace name, to disambiguate same-named env copies. */
   workspaceName?: string | null
-  /** Curated exposed outputs; empty/absent exposes the child's whole `result`. */
+  /** Curated public outputs; legacy empty definitions expose no data fields. */
   exposedOutputs?: CustomBlockOutput[]
 }
 
@@ -237,9 +239,8 @@ export function buildCustomBlockConfig(
 }
 
 /**
- * The block's declared outputs. Internal plumbing (child workflow id/name, trace
- * spans) is never exposed. With curated `exposedOutputs`, each becomes its own
- * named output; otherwise the whole child `result` is exposed.
+ * Public outputs contain status fields and publisher-curated data only.
+ * Legacy definitions without curated outputs expose no child data and fail at invocation.
  */
 function buildOutputs(exposed: CustomBlockOutput[] | undefined): BlockConfig['outputs'] {
   const outputs: BlockConfig['outputs'] = {
@@ -248,12 +249,12 @@ function buildOutputs(exposed: CustomBlockOutput[] | undefined): BlockConfig['ou
     errorType: { type: 'string', description: 'Machine-readable failure class' },
     errorRef: { type: 'string', description: 'Opaque reference to the failed run' },
   }
-  // No whole-`result` fallback: curation is required at publish, so every
-  // consumer-visible field is one the publisher chose. A legacy row with no
-  // curated outputs advertises no data fields and fails loudly at invocation
-  // rather than silently reverting to exposing the child's raw terminal state.
   for (const out of exposed ?? []) {
-    outputs[out.name] = { type: 'json', description: `Output: ${out.path}` }
+    outputs[out.name] = {
+      type: out.streaming ? 'string' : 'json',
+      description: out.streaming ? 'Streaming text output' : `Output: ${out.path}`,
+      ...(out.streaming ? { streaming: true } : {}),
+    }
   }
   return outputs
 }

apps/sim/ee/custom-blocks/components/custom-block-detail.tsx

@@ -8,6 +8,7 @@ import {
   ChipConfirmModal,
   ChipInput,
   ChipModalField,
+  ChipSwitch,
   ChipTextarea,
   type ComboboxOptionGroup,
   cn,
@@ -27,6 +28,7 @@ import {
   flattenWorkflowOutputs,
 } from '@/lib/workflows/blocks/flatten-outputs'
 import { extractInputFieldsFromBlocks } from '@/lib/workflows/input-format'
+import { isCustomBlockStreamSource } from '@/lib/workflows/streaming/custom-block-output'
 import { UnsavedChangesModal } from '@/app/workspace/[workspaceId]/components/credential-detail'
 import { DropZone } from '@/app/workspace/[workspaceId]/components/drop-zone'
 import { SettingsPanel } from '@/app/workspace/[workspaceId]/settings/components/settings-panel'
@@ -266,7 +268,7 @@ export function CustomBlockDetail({ blockId, workspaceId, onBack }: CustomBlockD
   const visibleOutputs = useMemo(
     () =>
       deployedLoaded
-        ? outputs.filter((o) => labelByKey.has(encodeOutput(o.blockId, o.path)))
+        ? outputs.filter((o) => o.streaming || labelByKey.has(encodeOutput(o.blockId, o.path)))
         : outputs,
     [outputs, deployedLoaded, labelByKey]
   )
@@ -685,7 +687,7 @@ export function CustomBlockDetail({ blockId, workspaceId, onBack }: CustomBlockD
 
           <SettingRow
             label='Outputs'
-            description='Pick which workflow outputs consumers see and name each one. At least one is required.'
+            description='Pick which workflow outputs consumers see and name each one. Enable streaming for text answers to use them live in deployed chat and Slack.'
           >
             <ChipCombobox
               multiSelect
@@ -728,6 +730,31 @@ export function CustomBlockDetail({ blockId, workspaceId, onBack }: CustomBlockD
                         maxLength={60}
                         disabled={!canManageBlock}
                       />
+                      {canManageBlock &&
+                      (o.streaming ||
+                        isCustomBlockStreamSource(deployed.data?.blocks?.[o.blockId], o.path)) ? (
+                        <ChipSwitch
+                          value={o.streaming ? 'live' : 'final'}
+                          aria-label={`Stream ${o.name}`}
+                          options={[
+                            { value: 'final', label: 'Final' },
+                            { value: 'live', label: 'Live' },
+                          ]}
+                          onChange={(value) =>
+                            setOutputs((current) =>
+                              current.map((output) =>
+                                encodeOutput(output.blockId, output.path) === key
+                                  ? { ...output, streaming: value === 'live' }
+                                  : output
+                              )
+                            )
+                          }
+                        />
+                      ) : (
+                        <span className='text-[var(--text-muted)] text-caption'>
+                          {o.streaming ? 'Live' : 'Final'}
+                        </span>
+                      )}
                     </div>
                   )
                 })}

apps/sim/executor/execution/block-executor.test.ts

@@ -1303,6 +1303,58 @@ describe('BlockExecutor streaming pump', () => {
     }
   }
 
+  it('fails the block when required stream delivery fails', async () => {
+    const handler = createAgentEventsStreamingHandler({
+      events: [{ type: 'text_delta', text: 'answer', turn: 'final' }],
+    })
+    const { executor, block, state } = createExecutor(handler)
+    const ctx = createContext(state)
+    ctx.onStream = async () => {
+      throw new Error('Stream delivery failed')
+    }
+    await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow(
+      'Stream delivery failed'
+    )
+  })
+
+  it('cancels an ongoing provider stream when required stream delivery fails', async () => {
+    const cancel = vi.fn()
+    const handler: BlockHandler = {
+      canHandle: () => true,
+      execute: async () => ({
+        stream: new ReadableStream({
+          start(controller) {
+            controller.enqueue({ type: 'text_delta', text: 'answer', turn: 'final' })
+          },
+          cancel,
+        }),
+        streamFormat: 'agent-events-v1',
+        execution: { success: true, output: { content: '' }, logs: [] },
+      }),
+    }
+    const { executor, block, state } = createExecutor(handler)
+    const ctx = createContext(state)
+    const workflowController = new AbortController()
+    ctx.abortSignal = workflowController.signal
+    const deliveryError = new Error('Stream delivery failed')
+    ctx.onStream = async ({ stream }) => {
+      const reader = stream.getReader()
+      try {
+        const { value } = await reader.read()
+        expect(new TextDecoder().decode(value)).toBe('answer')
+        throw deliveryError
+      } finally {
+        reader.releaseLock()
+      }
+    }
+
+    await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow(
+      'Stream delivery failed'
+    )
+    expect(cancel).toHaveBeenCalledExactlyOnceWith(deliveryError)
+    expect(workflowController.signal.aborted).toBe(false)
+  })
+
   it('projects answer text to onStream and content; sink gets full timeline', async () => {
     const onFullContent = vi.fn()
     const handler = createAgentEventsStreamingHandler({

apps/sim/executor/execution/block-executor.ts

@@ -1,4 +1,5 @@
 import { createLogger, type Logger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { isRecordLike } from '@sim/utils/object'
 import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types'
@@ -1177,15 +1178,19 @@ export class BlockExecutor {
       (block.config as Record<string, any> | undefined)?.responseFormat
 
     const streamFormat = streamingExec.streamFormat ?? 'text'
+    const streamDeliveryController = new AbortController()
     const pump = createAgentStreamPump({
       source: streamingExec.stream,
       streamFormat,
       // No live consumer → sink-mode so we never buffer into an unread text stream.
       sinkMode: !forwardToClient,
-      abortSignal: ctx.abortSignal,
+      abortSignal: ctx.abortSignal
+        ? AbortSignal.any([ctx.abortSignal, streamDeliveryController.signal])
+        : streamDeliveryController.signal,
     })
 
     let onStreamPromise: Promise<void> | undefined
+    let streamDeliveryError: Error | undefined
     let processedClientStream: ReadableStream<Uint8Array> | undefined
 
     if (forwardToClient && ctx.onStream && pump.textStream) {
@@ -1218,6 +1223,8 @@ export class BlockExecutor {
             ctx.resolvedSecretTraceRegistry?.exportCommittedProvenanceForValue(resolvedInputs),
         })
         .catch(async (error) => {
+          streamDeliveryError = toError(error)
+          streamDeliveryController.abort(streamDeliveryError)
           this.execLogger.error('Error in onStream callback', {
             blockId,
             ...projectStreamDiagnosticError(error),
@@ -1243,6 +1250,7 @@ export class BlockExecutor {
     if (onStreamPromise) {
       await onStreamPromise
     }
+    if (streamDeliveryError) throw streamDeliveryError
 
     // Timeout still fails the block, but keep any drained answer text so logs
     // match what was already projected to the client before the deadline.