fix(streaming): match encoded output selectors

Author: TheodoreSpeaks

apps/sim/executor/utils.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { formatInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 import {
   StreamingResponseFormatProcessor,
   streamingResponseFormatProcessor,
@@ -221,6 +222,33 @@ describe('StreamingResponseFormatProcessor', () => {
       ])
     })
 
+    it('matches encoded selectors for block IDs containing underscores', async () => {
+      let sourceSink: AgentStreamSink | undefined
+      const projectedText: string[] = []
+      const subscribe = processor.processEventSubscription(
+        (sink) => {
+          sourceSink = sink
+          return () => {}
+        },
+        'answer_agent',
+        [formatInternalOutputSelector('answer_agent', 'answer')],
+        { schema: { properties: { answer: { type: 'string' } } } }
+      )
+
+      subscribe?.({
+        onEvent: (event) => {
+          if (event.type === 'text_delta') projectedText.push(event.text)
+        },
+      })
+      await sourceSink?.onEvent({
+        type: 'text_delta',
+        text: '{"answer":"Streamed"}',
+        turn: 'final',
+      })
+
+      expect(projectedText).toEqual(['Streamed'])
+    })
+
     it('holds incomplete JSON escapes until they can be decoded', async () => {
       let sourceSink: AgentStreamSink | undefined
       const projectedText: string[] = []

apps/sim/executor/utils.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { isRecordLike } from '@sim/utils/object'
+import { formatInternalOutputSelector } from '@/lib/workflows/streaming/output-selector'
 import type { ResponseFormatStreamProcessor } from '@/executor/types'
 import type { AgentStreamSink, UnsubscribeAgentStreamSink } from '@/providers/stream-events'
 
@@ -13,7 +14,7 @@ interface JsonStringToken {
 }
 
 function selectedFieldsForBlock(blockId: string, selectedOutputs: string[]): string[] {
-  const prefix = `${blockId}_`
+  const prefix = `${formatInternalOutputSelector(blockId)}_`
   return selectedOutputs
     .filter((outputId) => outputId.startsWith(prefix))
     .map((outputId) => outputId.slice(prefix.length))