fix(streaming): validate nested output selectors

Author: TheodoreSpeaks

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

@@ -579,6 +579,32 @@ 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 scoped output selector block ID: workflow//agent',
+        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

@@ -143,7 +143,7 @@ import {
 } from '@/lib/workflows/streaming/forward-agent-stream-events'
 import {
   formatOutputSelector,
-  parseOutputSelector,
+  parseStoredOutputSelector,
 } from '@/lib/workflows/streaming/output-selector'
 import {
   agentStreamProtocolResponseHeaders,
@@ -312,7 +312,7 @@ function resolveOutputIds(
 
   return selectedOutputs.map((outputId) => {
     if (outputId.includes('/')) {
-      const parsed = parseOutputSelector(outputId)
+      const parsed = parseStoredOutputSelector(outputId)
       return formatOutputSelector(parsed.blockId, parsed.path)
     }
 

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

@@ -1352,6 +1352,33 @@ describe('BlockExecutor streaming pump', () => {
     )
   })
 
+  it('forwards the stable block ID for streams from expanded branch nodes', async () => {
+    const handler = createAgentEventsStreamingHandler({
+      events: [{ type: 'text_delta', text: 'branch answer', turn: 'final' }],
+    })
+    const { executor, block, state } = createExecutor(handler)
+    const ctx = createContext(state)
+    const node = createNode(block)
+    node.id = `${block.id}₍0₎`
+    node.metadata = {
+      isParallelBranch: true,
+      subflowId: 'parallel-1',
+      subflowType: 'parallel',
+      originalBlockId: block.id,
+      branchIndex: 0,
+    }
+    let streamedBlockId: string | undefined
+
+    ctx.onStream = async (streamingExec) => {
+      streamedBlockId = streamingExec.blockId
+      await new Response(streamingExec.stream).text()
+    }
+
+    await executor.execute(ctx, node, block)
+
+    expect(streamedBlockId).toBe(block.id)
+  })
+
   it('drains without onStream and still persists answer content', async () => {
     const handler = createAgentEventsStreamingHandler({
       events: [{ type: 'text_delta', text: 'offline answer', turn: 'final' }],

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

@@ -1154,7 +1154,7 @@ export class BlockExecutor {
     selectedOutputs: string[],
     executionOrder?: number
   ): Promise<void> {
-    const blockId = node.id
+    const blockId = node.metadata?.originalBlockId ?? node.id
     const piiEnabled = Boolean(ctx.piiBlockOutputRedaction?.enabled)
     // Live-forward only when a client stream exists and PII redaction is off.
     const forwardToClient = Boolean(ctx.onStream) && !piiEnabled

apps/sim/hooks/queries/chats.test.tsx

@@ -118,4 +118,25 @@ describe('chat mutations invalidate the deployment boundary', () => {
 
     expect(mockInvalidateDeploymentQueries).toHaveBeenCalledWith(expect.anything(), 'wf-1')
   })
+
+  it('normalizes a legacy empty output path to the deployed chat fallback', async () => {
+    const { getResult } = renderHookWithClient(() => useUpdateChat())
+
+    await act(async () => {
+      await getResult().mutateAsync({
+        chatId: 'chat-1',
+        workflowId: 'wf-1',
+        formData: { ...FORM_DATA, selectedOutputBlocks: ['agent-block_'] },
+      })
+    })
+
+    expect(mockRequestJson).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        body: expect.objectContaining({
+          outputConfigs: [{ blockId: 'agent-block', path: 'content' }],
+        }),
+      })
+    )
+  })
 })

apps/sim/hooks/queries/chats.ts

@@ -19,7 +19,7 @@ import {
   updateChatContract,
   verifyChatEmailOtpContract,
 } from '@/lib/api/contracts/chats'
-import { parseOutputSelector } from '@/lib/workflows/streaming/output-selector'
+import { parseInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 import type { OutputConfig } from '@/stores/chat/types'
 import { deploymentKeys, invalidateDeploymentQueries } from './deployments'
 
@@ -228,11 +228,9 @@ function throwUserFriendlyIdentifierError(error: unknown): never {
  */
 function parseOutputConfigs(selectedOutputBlocks: string[]): OutputConfig[] {
   return selectedOutputBlocks.map((outputId) => {
-    const parsed = parseOutputSelector(outputId)
-    if (!parsed.path) {
-      throw new Error(`Chat output selector must include a path: ${outputId}`)
-    }
-    return parsed
+    const normalizedOutputId = outputId.endsWith('_') ? outputId.slice(0, -1) : outputId
+    const parsed = parseInternalOutputSelector(normalizedOutputId)
+    return { ...parsed, path: parsed.path || 'content' }
   })
 }
 

apps/sim/lib/api/contracts/chat-output-config.test.ts

@@ -0,0 +1,31 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { chatOutputConfigSchema } from '@/lib/api/contracts/chats'
+import { v2ChatDeploymentOutputConfigSchema } from '@/lib/api/contracts/v2/chat-deployments'
+
+const OUTPUT_CONFIG_SCHEMAS = [chatOutputConfigSchema, v2ChatDeploymentOutputConfigSchema]
+
+describe('chat output config contracts', () => {
+  it.each(OUTPUT_CONFIG_SCHEMAS)('accepts invocation-scoped output selectors', (schema) => {
+    expect(
+      schema.safeParse({ blockId: 'workflow-block/agent-block', path: 'content.text' }).success
+    ).toBe(true)
+  })
+
+  it.each(OUTPUT_CONFIG_SCHEMAS)(
+    'rejects output selectors the executor cannot format',
+    (schema) => {
+      for (const config of [
+        { blockId: '/agent-block', path: 'content' },
+        { blockId: 'workflow-block/', path: 'content' },
+        { blockId: 'agent-block', path: '.content' },
+        { blockId: 'agent-block', path: 'content.' },
+        { blockId: 'agent-block', path: 'content..text' },
+      ]) {
+        expect(schema.safeParse(config).success).toBe(false)
+      }
+    }
+  )
+})

apps/sim/lib/api/contracts/chats.ts

@@ -1,5 +1,7 @@
+import { getErrorMessage } from '@sim/utils/errors'
 import { z } from 'zod'
 import { defineRouteContract } from '@/lib/api/contracts/types'
+import { formatInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 
 export const chatAuthTypeSchema = z.enum(['public', 'password', 'email', 'sso'])
 export type ChatAuthType = z.output<typeof chatAuthTypeSchema>
@@ -38,10 +40,18 @@ export const chatIdentifierParamsSchema = z.object({
   identifier: z.string().min(1),
 })
 
-export const chatOutputConfigSchema = z.object({
-  blockId: z.string().min(1),
-  path: z.string().min(1),
-})
+export const chatOutputConfigSchema = z
+  .object({
+    blockId: z.string().min(1),
+    path: z.string().min(1),
+  })
+  .superRefine((config, ctx) => {
+    try {
+      formatInternalOutputSelector(config.blockId, config.path)
+    } catch (error) {
+      ctx.addIssue({ code: 'custom', message: getErrorMessage(error, 'Invalid output config') })
+    }
+  })
 
 export const deployedChatOutputConfigSchema = z.object({
   blockId: z.string(),

apps/sim/lib/api/contracts/v2/chat-deployments.ts

@@ -1,3 +1,4 @@
+import { getErrorMessage } from '@sim/utils/errors'
 import { z } from 'zod'
 import { chatAuthTypeSchema, chatDeploymentPasswordSchema } from '@/lib/api/contracts/chats'
 import {
@@ -14,6 +15,7 @@ import {
   v2SortFields,
 } from '@/lib/api/contracts/v2/shared'
 import { v2WorkflowIdParamsSchema } from '@/lib/api/contracts/v2/workflows'
+import { formatInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 
 /**
  * v2 chat-deployment contracts.
@@ -120,6 +122,13 @@ export const v2ChatDeploymentOutputConfigSchema = z
       .describe('Path within that block output.'),
   })
   .strict()
+  .superRefine((config, ctx) => {
+    try {
+      formatInternalOutputSelector(config.blockId, config.path)
+    } catch (error) {
+      ctx.addIssue({ code: 'custom', message: getErrorMessage(error, 'Invalid output config') })
+    }
+  })
   .meta({
     id: 'ChatDeploymentOutputConfig',
     title: 'Chat deployment output config',

apps/sim/lib/webhooks/slack-stream-config.ts

@@ -1,5 +1,5 @@
 import { isRecordLike } from '@sim/utils/object'
-import { parseOutputSelector } from '@/lib/workflows/streaming/output-selector'
+import { parseInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 
 export const SLACK_STREAM_RESPONSE_EVENTS = [
   'message',
@@ -24,7 +24,7 @@ export interface SlackStreamResponseConfig {
 const SLACK_TASK_TITLE_LIMIT = 256
 
 function parseSlackOutputSelector(selector: string): SlackStreamOutputConfig {
-  const parsed = parseOutputSelector(selector)
+  const parsed = parseInternalOutputSelector(selector)
   if (!parsed.path) {
     throw new Error(`Invalid Slack stream output selector: ${selector}`)
   }