feat(streaming): support nested workflow outputs (#7346)

* feat(streaming): support nested workflow outputs

* fix(streaming): reset stale output drilldowns

* fix(streaming): validate nested output selectors

* fix(slack): use public output selectors

* fix(streaming): scope nested outputs by workflow

* fix(cli): align streaming selector help test

Author: TheodoreSpeaks

apps/docs/content/docs/cli/reference.mdx

@@ -5157,7 +5157,7 @@ sim workflows run <workflowId> [options]
 | `--input <json\|@file>` | No | Trigger input as JSON (JSON, or @path / @- to read a file or stdin). |
 | `--async` | No | Queue the run and return immediately. |
 | `--execution-timeout-seconds <value>` | No | Requested server-side timeout for an asynchronous run, in seconds. An upper bound, not the effective timeout: the run uses the smaller of this value and the plan's execution timeout, so requesting more than the plan allows silently yields the plan timeout. Rejected with `400` unless `async` is true. |
-| `--select-output <value...>` | No | Return blockName.field values from the streamed result (e.g. agent_1.content), requires --follow; missing fields are omitted (space-separated, or @path / @- with one value per line; @@value for a literal leading @). |
+| `--select-output <value...>` | No | Return streamed outputs as blockName.path or childWorkflowId.blockName.path; selecting a child workflow applies to every invocation, requires --follow (space-separated, or @path / @- with one value per line; @@value for a literal leading @). |
 | `--include-file-base64` | No | Inline eligible output files as base64 content. Rejected when `async` is true. |
 | `--no-include-file-base64` | No | Send --include-file-base64 as false. |
 | `--base64-max-bytes <value>` | No | Maximum total bytes of file content to inline as base64, lowering but never raising the server limit of 16 MiB. Rejected when `async` is true. |

apps/docs/content/docs/cli/workflows.mdx

@@ -532,7 +532,7 @@ sim workflows run <workflowId> [options]
 | `--input <json\|@file>` | No | Trigger input as JSON (JSON, or @path / @- to read a file or stdin). |
 | `--async` | No | Queue the run and return immediately. |
 | `--execution-timeout-seconds <value>` | No | Requested server-side timeout for an asynchronous run, in seconds. An upper bound, not the effective timeout: the run uses the smaller of this value and the plan's execution timeout, so requesting more than the plan allows silently yields the plan timeout. Rejected with `400` unless `async` is true. |
-| `--select-output <value...>` | No | Return blockName.field values from the streamed result (e.g. agent_1.content), requires --follow; missing fields are omitted (space-separated, or @path / @- with one value per line; @@value for a literal leading @). |
+| `--select-output <value...>` | No | Return streamed outputs as blockName.path or childWorkflowId.blockName.path; selecting a child workflow applies to every invocation, requires --follow (space-separated, or @path / @- with one value per line; @@value for a literal leading @). |
 | `--include-file-base64` | No | Inline eligible output files as base64 content. Rejected when `async` is true. |
 | `--no-include-file-base64` | No | Send --include-file-base64 as false. |
 | `--base64-max-bytes <value>` | No | Maximum total bytes of file content to inline as base64, lowering but never raising the server limit of 16 MiB. Rejected when `async` is true. |

apps/docs/content/docs/integrations/slack.mdx

@@ -1971,7 +1971,7 @@ Trigger from Slack events, interactions, and slash commands
 | `manualChannelFilter` | string | No | Comma-separated channel IDs to restrict to. Set IDs directly here. |
 | `threads` | string | No | Include thread replies, exclude them \(top-level only\), or fire only on thread replies. |
 | `streamResponse` | boolean | No | Create a Slack agent session and stream selected workflow outputs into the conversation that started this run. Custom bots only. |
-| `streamOutputs` | workflow-output-selector | No | Each selected block invocation creates its own Slack response. Agent outputs stream live; other outputs are sent when the block completes. |
+| `streamOutputs` | workflow-output-selector | No | Use `<blockName>.<outputPath>` for this workflow or `<childWorkflowId>.<blockName>.<outputPath>` for a child workflow. Selecting a child workflow applies to every invocation of it. Agent outputs stream live; other outputs are sent when the block completes. |
 | `streamTaskTitle` | string | No | Optional status Slack shows while each selected response is being produced. Leave empty to use Running. |
 | `streamTaskDisplayMode` | string | No | Choose how Slack displays thinking and tool progress. |
 | `streamIncludeThinking` | boolean | No | Show agent thinking as Slack task updates while the response is generated. |

apps/docs/openapi-v2-workflows.json

@@ -8593,6 +8593,10 @@
       "StoredChatDeploymentOutputConfig": {
         "type": "object",
         "properties": {
+          "workflowId": {
+            "description": "Child workflow containing the selected block. Omitted for the deployed workflow.",
+            "type": "string"
+          },
           "blockId": {
             "type": "string",
             "description": "Block whose output the chat streams."
@@ -8902,6 +8906,11 @@
       "ChatDeploymentOutputConfig": {
         "type": "object",
         "properties": {
+          "workflowId": {
+            "description": "Child workflow containing the selected block. Omit for the deployed workflow.",
+            "type": "string",
+            "minLength": 1
+          },
           "blockId": {
             "type": "string",
             "minLength": 1,
@@ -9305,7 +9314,7 @@
             "type": "boolean"
           },
           "selectedOutputs": {
-            "description": "Block output references to include in a streamed response, as `blockId`, `blockId.path`, or `BlockName.path` (resolved against the live workflow). Requires `stream: true` — it shapes the streamed envelope only, so it is rejected on a sync request and when `async` is true. To narrow a finished run, pass `selectedOutputs` to the run resource instead.",
+            "description": "Block output references to include in a streamed response. Use `<blockName>.<outputPath>` for the executed workflow or `<childWorkflowId>.<blockName>.<outputPath>` for a child workflow; block names are normalized workflow reference names. Selecting a child workflow applies to every invocation of it. Requires `stream: true` — it shapes the streamed envelope only, so it is rejected on a sync request and when `async` is true. To narrow a finished run, pass `selectedOutputs` to the run resource instead.",
             "maxItems": 100,
             "type": "array",
             "items": {

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

@@ -21,6 +21,7 @@ import {
   isChatThinkingFrame,
   isChatToolFrame,
 } from '@/lib/workflows/streaming/agent-stream-protocol'
+import { scopeOutputBlockId } from '@/lib/workflows/streaming/output-selector'
 import type {
   ChatFile,
   ChatMessage,
@@ -70,7 +71,7 @@ function extractFilesFromData(
 }
 
 export interface StreamingOptions {
-  outputConfigs?: Array<{ blockId: string; path?: string }>
+  outputConfigs?: Array<{ workflowId?: string; 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.
@@ -430,7 +431,10 @@ export function useChatStreaming() {
 
             if (outputConfigs?.length && finalData.output) {
               for (const config of outputConfigs) {
-                const blockOutputs = finalData.output[config.blockId]
+                const outputBlockId = config.workflowId
+                  ? scopeOutputBlockId(config.workflowId, config.blockId)
+                  : config.blockId
+                const blockOutputs = finalData.output[outputBlockId]
                 if (!blockOutputs) continue
 
                 const value = getOutputValue(blockOutputs, config.path)

apps/sim/app/api/chat/[identifier]/route.ts

@@ -14,6 +14,7 @@ import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { ChatFiles } from '@/lib/uploads'
+import { formatOutputSelector } from '@/lib/workflows/streaming/output-selector'
 import { setChatAuthCookie, validateChatAuth } from '@/app/api/chat/utils'
 import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 
@@ -213,9 +214,11 @@ export const POST = withRouteHandler(
         const selectedOutputs: string[] = []
         if (deployment.outputConfigs && Array.isArray(deployment.outputConfigs)) {
           for (const config of deployment.outputConfigs) {
-            const outputId = config.path
-              ? `${config.blockId}_${config.path}`
-              : `${config.blockId}_content`
+            const outputId = formatOutputSelector(
+              config.blockId,
+              config.path || 'content',
+              config.workflowId
+            )
             selectedOutputs.push(outputId)
           }
         }

apps/sim/app/api/v2/chat-deployments/utils.ts

@@ -51,9 +51,13 @@ function normalizeStoredOutputConfigs(raw: unknown): V2ChatDeploymentOutputConfi
   const configs: V2ChatDeploymentOutputConfig[] = []
   for (const entry of raw) {
     if (!entry || typeof entry !== 'object' || Array.isArray(entry)) continue
-    const { blockId, path } = entry as Record<string, unknown>
+    const { workflowId, blockId, path } = entry as Record<string, unknown>
     if (typeof blockId !== 'string' || blockId.length === 0) continue
-    configs.push({ blockId, path: typeof path === 'string' ? path : '' })
+    configs.push({
+      ...(typeof workflowId === 'string' && workflowId.length > 0 ? { workflowId } : {}),
+      blockId,
+      path: typeof path === 'string' ? path : '',
+    })
   }
   return configs
 }

apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts

@@ -579,6 +579,31 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => {
     )
   })
 
+  it('maps malformed nested output selectors to an input failure', async () => {
+    const result = await executeWorkflowService({
+      workflowId: 'workflow-1',
+      principal: { kind: 'personal_api_key', userId: 'actor-1', keyId: 'key-1' },
+      userId: 'actor-1',
+      input: {},
+      triggerType: 'api',
+      requestId: 'request-1',
+      workflowRecord,
+      selectedOutputs: ['workflow//agent.content'],
+      mode: 'stream',
+      requestHeaders: new Headers(),
+    })
+
+    expect(result).toEqual({
+      ok: false,
+      failure: {
+        kind: 'input',
+        message: 'Invalid selectedOutputs: Invalid output selector: workflow//agent.content',
+        statusCode: 400,
+      },
+    })
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-123')
+  })
+
   it('rejects async manual execution and conflicting mock input before dispatch', async () => {
     authenticatePersonalKey()
 

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

@@ -4,7 +4,8 @@ import { workflow as workflowTable } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { authorizeWorkflowByWorkspacePermission } from '@sim/platform-authz/workflow'
 import { getErrorMessage, toError } from '@sim/utils/errors'
-import { generateId, isValidUuid } from '@sim/utils/id'
+import { generateId } from '@sim/utils/id'
+import type { BlockState } from '@sim/workflow-types/workflow'
 import { eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
@@ -141,6 +142,7 @@ import {
   forwardAgentStreamToExecutionEvents,
   shouldForwardAnswerTextFromSink,
 } from '@/lib/workflows/streaming/forward-agent-stream-events'
+import { resolveOutputSelectors } from '@/lib/workflows/streaming/resolve-output-selectors'
 import {
   agentStreamProtocolResponseHeaders,
   createStreamingResponse,
@@ -152,7 +154,6 @@ import {
   PublicApiNotAllowedError,
   validatePublicApiAllowed,
 } from '@/ee/access-control/utils/permission-check'
-import { normalizeName } from '@/executor/constants'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type {
   BlockCompletionCallbackData,
@@ -298,56 +299,13 @@ function payloadTooLargeResponse(message = 'Workflow execution response exceeds
   )
 }
 
-function resolveOutputIds(
+async function resolveOutputIds(
   selectedOutputs: string[] | undefined,
-  blocks: Record<string, any>
-): string[] | undefined {
-  if (!selectedOutputs || selectedOutputs.length === 0) {
-    return selectedOutputs
-  }
-
-  return selectedOutputs.map((outputId) => {
-    const underscoreIndex = outputId.indexOf('_')
-    const dotIndex = outputId.indexOf('.')
-    if (underscoreIndex > 0) {
-      const maybeUuid = outputId.substring(0, underscoreIndex)
-      if (isValidUuid(maybeUuid)) {
-        return outputId
-      }
-    }
-
-    if (dotIndex > 0) {
-      const maybeUuid = outputId.substring(0, dotIndex)
-      if (isValidUuid(maybeUuid)) {
-        return `${outputId.substring(0, dotIndex)}_${outputId.substring(dotIndex + 1)}`
-      }
-    }
-
-    if (isValidUuid(outputId)) {
-      return outputId
-    }
-
-    if (dotIndex === -1) {
-      logger.warn(`Invalid output ID format (missing dot): ${outputId}`)
-      return outputId
-    }
-
-    const blockName = outputId.substring(0, dotIndex)
-    const path = outputId.substring(dotIndex + 1)
-
-    const normalizedBlockName = normalizeName(blockName)
-    const block = Object.values(blocks).find((b: any) => {
-      return normalizeName(b.name || '') === normalizedBlockName
-    })
-
-    if (!block) {
-      logger.warn(`Block not found for name: ${blockName} (from output ID: ${outputId})`)
-      return outputId
-    }
-
-    const resolvedId = `${block.id}_${path}`
-    logger.debug(`Resolved output ID: ${outputId} -> ${resolvedId}`)
-    return resolvedId
+  blocks: Record<string, BlockState>
+): Promise<string[] | undefined> {
+  return resolveOutputSelectors({
+    selectedOutputs,
+    currentBlocks: blocks,
   })
 }
 
@@ -1671,10 +1629,19 @@ async function handleExecutePost(
     } else {
       reqLogger.info('Using streaming API response')
 
-      const resolvedSelectedOutputs = resolveOutputIds(
-        selectedOutputs,
-        cachedWorkflowData?.blocks || {}
-      )
+      let resolvedSelectedOutputs: string[] | undefined
+      try {
+        resolvedSelectedOutputs = await resolveOutputIds(
+          selectedOutputs,
+          cachedWorkflowData?.blocks || {}
+        )
+      } catch (error) {
+        await releaseExecutionSlot(executionId)
+        return NextResponse.json(
+          { error: `Invalid selectedOutputs: ${getErrorMessage(error)}` },
+          { status: 400 }
+        )
+      }
       const streamVariables = cachedWorkflowData?.variables ?? (workflow as any).variables
       const streamWorkflow = {
         id: workflow.id,