diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index 250542ee7be..7f9dfb27356 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/apps/sim/executor/execution/block-executor.test.ts @@ -1352,6 +1352,47 @@ describe('BlockExecutor streaming pump', () => { ) }) + it('projects a selected structured string into live sink deltas', async () => { + const handler = createAgentEventsStreamingHandler({ + events: [ + { type: 'text_delta', text: '{"answer":"Hello ', turn: 'pending' }, + { type: 'text_delta', text: 'world","score":1}', turn: 'pending' }, + { type: 'turn_end', turn: 'final' }, + ], + }) + const { executor, block, state } = createExecutor(handler) + block.config.params = { + responseFormat: { + schema: { + type: 'object', + properties: { answer: { type: 'string' }, score: { type: 'number' } }, + }, + }, + } + const ctx = createContext(state) + ctx.stream = true + ctx.selectedOutputs = [`${block.id}_answer`] + const sinkText: string[] = [] + let forwardedText = '' + + ctx.onStream = async (streamingExec) => { + expect(streamingExec.clientStreamTransformed).toBe(true) + expect(streamingExec.clientSinkTransformed).toBe(true) + streamingExec.subscribe?.({ + onEvent: (event) => { + if (event.type === 'text_delta') sinkText.push(event.text) + }, + }) + forwardedText = await new Response(streamingExec.stream).text() + } + + await executor.execute(ctx, createNode(block), block) + + expect(sinkText).toEqual(['Hello ', 'world']) + expect(forwardedText).toBe('Hello world') + expect(state.getBlockOutput(block.id)?.answer).toBe('Hello world') + }) + 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' }], diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index 313989475aa..8f006b8297b 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -1197,6 +1197,15 @@ export class BlockExecutor { selectedOutputs, responseFormat ) + const clientStreamTransformed = processedClientStream !== pump.textStream + const projectedSubscribe = clientStreamTransformed + ? streamingResponseFormatProcessor.processEventSubscription( + pump.subscribe, + blockId, + selectedOutputs, + responseFormat + ) + : undefined // Start onStream without awaiting so a sync `subscribe(sink)` can run before // the first provider pull, then read the projected text stream concurrently @@ -1208,10 +1217,11 @@ export class BlockExecutor { ...(executionOrder !== undefined ? { executionOrder } : {}), stream: processedClientStream, streamFormat: 'text', - subscribe: pump.subscribe, + subscribe: projectedSubscribe ?? pump.subscribe, // processStream returns the input stream identity when no // response-format extraction applies. - clientStreamTransformed: processedClientStream !== pump.textStream, + clientStreamTransformed, + clientSinkTransformed: Boolean(projectedSubscribe), displayResolvedSecretTraceProvenance: ctx.resolvedSecretTraceRegistry?.exportCommittedProvenanceForValue(resolvedInputs), }) diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts index 2cd95e2c907..8b7c0b9b130 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts @@ -2233,6 +2233,9 @@ describe('WorkflowBlockHandler', () => { const childStream = { blockId: 'agent-1', stream: new ReadableStream(), + subscribe: vi.fn(), + clientStreamTransformed: true, + clientSinkTransformed: true, execution: { success: true, output: {} }, } await extensions.onStream(childStream) diff --git a/apps/sim/executor/types.ts b/apps/sim/executor/types.ts index b3e98b2c497..a5527d8f0c9 100644 --- a/apps/sim/executor/types.ts +++ b/apps/sim/executor/types.ts @@ -666,10 +666,11 @@ export interface StreamingExecution { /** * True when {@link stream} is a response-format projection (selected JSON * fields extracted from structured output) rather than raw answer text. Sink - * `text_delta` events then do NOT match the byte stream, so consumers must - * keep sourcing answer text from {@link stream} instead of the sink. + * `text_delta` events match it only when {@link clientSinkTransformed} is true. */ clientStreamTransformed?: boolean + /** True when sink text deltas are projected to match a transformed client stream. */ + clientSinkTransformed?: boolean /** Internal provenance for the exact block input that initiated this live stream. */ displayResolvedSecretTraceProvenance?: ResolvedSecretTraceProvenanceV1 /** Internal source registry retained only for sanitizing failures while the stream drains. */ diff --git a/apps/sim/executor/utils.test.ts b/apps/sim/executor/utils.test.ts index 5769b81ac8a..1c53ca51c65 100644 --- a/apps/sim/executor/utils.test.ts +++ b/apps/sim/executor/utils.test.ts @@ -1,8 +1,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { formatInternalOutputSelector } from '@/lib/workflows/streaming/output-selector' import { StreamingResponseFormatProcessor, streamingResponseFormatProcessor, } from '@/executor/utils' +import type { AgentStreamEvent, AgentStreamSink } from '@/providers/stream-events' describe('StreamingResponseFormatProcessor', () => { let processor: StreamingResponseFormatProcessor @@ -170,6 +172,118 @@ describe('StreamingResponseFormatProcessor', () => { expect(result).toBe('charlie') }) + it('projects a selected string field from live event deltas', async () => { + let sourceSink: AgentStreamSink | undefined + const projectedEvents: AgentStreamEvent[] = [] + const subscribe = processor.processEventSubscription( + (sink) => { + sourceSink = sink + return () => {} + }, + 'block-1', + ['block-1_answer'], + JSON.stringify({ + type: 'object', + properties: { + meta: { type: 'object' }, + answer: { type: 'string' }, + score: { type: 'number' }, + }, + }) + ) + + expect(subscribe).toBeDefined() + subscribe?.({ + onEvent: async (event) => { + projectedEvents.push(event) + }, + }) + + await sourceSink?.onEvent({ + type: 'text_delta', + text: '{"meta":{"source":"test"},"answer":"Hello ', + turn: 'pending', + }) + expect(projectedEvents).toEqual([{ type: 'text_delta', text: 'Hello ', turn: 'pending' }]) + + await sourceSink?.onEvent({ + type: 'text_delta', + text: 'world","score":1}', + turn: 'pending', + }) + await sourceSink?.onEvent({ type: 'thinking_delta', text: 'done' }) + await sourceSink?.onEvent({ type: 'turn_end', turn: 'final' }) + + expect(projectedEvents).toEqual([ + { type: 'text_delta', text: 'Hello ', turn: 'pending' }, + { type: 'text_delta', text: 'world', turn: 'pending' }, + { type: 'thinking_delta', text: 'done' }, + { type: 'turn_end', turn: 'final' }, + ]) + }) + + 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[] = [] + const subscribe = processor.processEventSubscription( + (sink) => { + sourceSink = sink + return () => {} + }, + 'block-1', + ['block-1_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":"line\\', turn: 'final' }) + await sourceSink?.onEvent({ type: 'text_delta', text: 'nnext"}', turn: 'final' }) + + expect(projectedText).toEqual(['line', '\nnext']) + }) + + it('does not claim live sink projection for non-string fields', () => { + const subscribe = processor.processEventSubscription( + () => () => {}, + 'block-1', + ['block-1_score'], + { schema: { properties: { score: { type: 'number' } } } } + ) + + expect(subscribe).toBeUndefined() + }) + it.concurrent('should handle missing fields gracefully', async () => { const mockStream = new ReadableStream({ start(controller) { diff --git a/apps/sim/executor/utils.ts b/apps/sim/executor/utils.ts index c509e4121ca..4b17e0110aa 100644 --- a/apps/sim/executor/utils.ts +++ b/apps/sim/executor/utils.ts @@ -1,8 +1,238 @@ 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' const logger = createLogger('ExecutorUtils') +type AgentStreamSubscribe = (sink: AgentStreamSink) => UnsubscribeAgentStreamSink + +interface JsonStringToken { + end: number + rawValue: string +} + +function selectedFieldsForBlock(blockId: string, selectedOutputs: string[]): string[] { + const prefix = `${formatInternalOutputSelector(blockId)}_` + return selectedOutputs + .filter((outputId) => outputId.startsWith(prefix)) + .map((outputId) => outputId.slice(prefix.length)) +} + +function readJsonStringToken(input: string, start: number): JsonStringToken | null { + if (input[start] !== '"') return null + + let escaped = false + for (let index = start + 1; index < input.length; index++) { + const char = input[index] + if (escaped) { + escaped = false + continue + } + if (char === '\\') { + escaped = true + continue + } + if (char === '"') { + return { end: index + 1, rawValue: input.slice(start + 1, index) } + } + } + + return null +} + +function skipWhitespace(input: string, start: number): number { + let index = start + while (index < input.length && ' \n\r\t'.includes(input[index])) index++ + return index +} + +function findJsonValueEnd(input: string, start: number): number | null { + const first = input[start] + if (first === '"') return readJsonStringToken(input, start)?.end ?? null + + if (first === '{' || first === '[') { + const closingTokens = [first === '{' ? '}' : ']'] + let inString = false + let escaped = false + + for (let index = start + 1; index < input.length; index++) { + const char = input[index] + if (inString) { + if (escaped) { + escaped = false + } else if (char === '\\') { + escaped = true + } else if (char === '"') { + inString = false + } + continue + } + + if (char === '"') { + inString = true + } else if (char === '{') { + closingTokens.push('}') + } else if (char === '[') { + closingTokens.push(']') + } else if (char === '}' || char === ']') { + if (closingTokens.at(-1) !== char) return null + closingTokens.pop() + if (closingTokens.length === 0) return index + 1 + } + } + + return null + } + + for (let index = start; index < input.length; index++) { + if (input[index] === ',' || input[index] === '}') return index + } + return null +} + +function locateTopLevelStringField(input: string, field: string): string | null { + let index = skipWhitespace(input, 0) + if (input[index] !== '{') return null + index++ + let firstProperty = true + + while (index < input.length) { + index = skipWhitespace(input, index) + if (input[index] === '}') return null + if (!firstProperty) { + if (input[index] !== ',') return null + index = skipWhitespace(input, index + 1) + } else if (input[index] === ',') { + return null + } + + const keyStart = index + const keyToken = readJsonStringToken(input, keyStart) + if (!keyToken) return null + + let key: unknown + try { + key = JSON.parse(input.slice(keyStart, keyToken.end)) + } catch { + return null + } + + index = skipWhitespace(input, keyToken.end) + if (input[index] !== ':') return null + index = skipWhitespace(input, index + 1) + if (index >= input.length) return null + + if (key === field) { + if (input[index] !== '"') return null + return readJsonStringToken(input, index)?.rawValue ?? input.slice(index + 1) + } + + const valueEnd = findJsonValueEnd(input, index) + if (valueEnd === null) return null + index = valueEnd + firstProperty = false + } + + return null +} + +function stableJsonStringPrefixLength(rawValue: string): number { + let index = 0 + while (index < rawValue.length) { + if (rawValue[index] !== '\\') { + index++ + continue + } + + const escapeStart = index + if (index + 1 >= rawValue.length) return escapeStart + const escapeType = rawValue[index + 1] + if (escapeType !== 'u') { + if (!'"\\/bfnrt'.includes(escapeType)) return escapeStart + index += 2 + continue + } + + if (index + 6 > rawValue.length) return escapeStart + const encodedCode = rawValue.slice(index + 2, index + 6) + if (!/^[0-9a-fA-F]{4}$/.test(encodedCode)) return escapeStart + const code = Number.parseInt(encodedCode, 16) + if (code >= 0xd800 && code <= 0xdbff) { + if (index + 12 > rawValue.length || rawValue.slice(index + 6, index + 8) !== '\\u') { + return escapeStart + } + const encodedLow = rawValue.slice(index + 8, index + 12) + if (!/^[0-9a-fA-F]{4}$/.test(encodedLow)) return escapeStart + const low = Number.parseInt(encodedLow, 16) + if (low < 0xdc00 || low > 0xdfff) return escapeStart + index += 12 + continue + } + + index += 6 + } + + return index +} + +function decodeStableJsonStringPrefix(rawValue: string): string | null { + const stablePrefix = rawValue.slice(0, stableJsonStringPrefixLength(rawValue)) + try { + return JSON.parse(`"${stablePrefix}"`) + } catch { + return null + } +} + +function declaresTopLevelStringField(responseFormat: unknown, field: string): boolean { + if (field.includes('.')) return false + + let parsed = responseFormat + if (typeof parsed === 'string') { + try { + parsed = JSON.parse(parsed) + } catch { + return false + } + } + if (!isRecordLike(parsed)) return false + + const schema = isRecordLike(parsed.schema) ? parsed.schema : parsed + if (!isRecordLike(schema.properties)) return false + const fieldSchema = schema.properties[field] + return isRecordLike(fieldSchema) && fieldSchema.type === 'string' +} + +class IncrementalJsonStringFieldProjector { + private buffer = '' + private emittedValue = '' + + constructor(private readonly field: string) {} + + push(chunk: string): string { + this.buffer += chunk + const rawValue = locateTopLevelStringField(this.buffer, this.field) + if (rawValue === null) return '' + + const decodedValue = decodeStableJsonStringPrefix(rawValue) + if (decodedValue === null) return '' + if (!decodedValue.startsWith(this.emittedValue)) { + throw new Error(`Structured stream field changed after emission: ${this.field}`) + } + + const delta = decodedValue.slice(this.emittedValue.length) + this.emittedValue = decodedValue + return delta + } + + reset(): void { + this.buffer = '' + this.emittedValue = '' + } +} + /** * Processes a streaming response to extract only the selected response format fields * instead of streaming the full JSON wrapper. @@ -14,26 +244,11 @@ export class StreamingResponseFormatProcessor implements ResponseFormatStreamPro selectedOutputs: string[], responseFormat?: any ): ReadableStream { - const hasResponseFormatSelection = selectedOutputs.some((outputId) => { - const blockIdForOutput = outputId.includes('_') - ? outputId.split('_')[0] - : outputId.split('.')[0] - return blockIdForOutput === blockId && outputId.includes('_') - }) - - if (!hasResponseFormatSelection || !responseFormat) { + const selectedFields = selectedFieldsForBlock(blockId, selectedOutputs) + if (selectedFields.length === 0 || !responseFormat) { return originalStream } - const selectedFields = selectedOutputs - .filter((outputId) => { - const blockIdForOutput = outputId.includes('_') - ? outputId.split('_')[0] - : outputId.split('.')[0] - return blockIdForOutput === blockId && outputId.includes('_') - }) - .map((outputId) => outputId.substring(blockId.length + 1)) - logger.info('Processing streaming response format', { blockId, selectedFields, @@ -44,6 +259,38 @@ export class StreamingResponseFormatProcessor implements ResponseFormatStreamPro return this.createProcessedStream(originalStream, selectedFields, blockId) } + processEventSubscription( + originalSubscribe: AgentStreamSubscribe, + blockId: string, + selectedOutputs: string[], + responseFormat?: unknown + ): AgentStreamSubscribe | undefined { + const selectedFields = selectedFieldsForBlock(blockId, selectedOutputs) + if ( + selectedFields.length !== 1 || + !declaresTopLevelStringField(responseFormat, selectedFields[0]) + ) { + return undefined + } + + const selectedField = selectedFields[0] + return (sink) => { + const projector = new IncrementalJsonStringFieldProjector(selectedField) + return originalSubscribe({ + onEvent: async (event) => { + if (event.type === 'text_delta') { + const text = projector.push(event.text) + if (text) await sink.onEvent({ ...event, text }) + return + } + + await sink.onEvent(event) + if (event.type === 'turn_end') projector.reset() + }, + }) + } + } + private createProcessedStream( originalStream: ReadableStream, selectedFields: string[], @@ -64,6 +311,7 @@ export class StreamingResponseFormatProcessor implements ResponseFormatStreamPro const { done, value } = await reader.read() if (done) { + buffer += decoder.decode() if (buffer.trim() && !hasProcessedComplete) { self.processCompleteJson(buffer, selectedFields, controller) } diff --git a/apps/sim/lib/webhooks/slack-execution-stream.test.ts b/apps/sim/lib/webhooks/slack-execution-stream.test.ts index 8f0c5cc812f..a1fa2ed015d 100644 --- a/apps/sim/lib/webhooks/slack-execution-stream.test.ts +++ b/apps/sim/lib/webhooks/slack-execution-stream.test.ts @@ -366,6 +366,33 @@ describe('SlackExecutionStreamController', () => { }) }) + it('streams projected structured text from a matching transformed event sink', async () => { + const { controller } = await createController() + const subscribe = vi.fn(({ onEvent }) => { + void onEvent({ type: 'text_delta', text: 'Live ', turn: 'pending' }) + void onEvent({ type: 'text_delta', text: 'answer', turn: 'pending' }) + void onEvent({ type: 'turn_end', turn: 'final' }) + return vi.fn() + }) + + await controller.callbacks.onStream?.({ + blockId: 'agent', + executionOrder: 6, + stream: createByteStream('fallback bytes'), + streamFormat: 'text', + clientStreamTransformed: true, + clientSinkTransformed: true, + subscribe, + }) + + const answerText = mockAppendSlackAgentStream.mock.calls + .flatMap((call) => call[3]) + .filter((chunk) => chunk.type === 'markdown_text') + .map((chunk) => chunk.text) + .join('') + expect(answerText).toBe('Live answer') + }) + it('sends a selected nested non-streaming output after block completion', async () => { const config: SlackStreamResponseConfig = { ...BASE_CONFIG, diff --git a/apps/sim/lib/webhooks/slack-execution-stream.ts b/apps/sim/lib/webhooks/slack-execution-stream.ts index 3a9f43b3dbe..0490db67009 100644 --- a/apps/sim/lib/webhooks/slack-execution-stream.ts +++ b/apps/sim/lib/webhooks/slack-execution-stream.ts @@ -22,6 +22,7 @@ import { type SlackStreamSessionTarget, unregisterSlackStreamSession, } from '@/lib/webhooks/slack-stream-sessions' +import { shouldForwardAnswerTextFromSink } from '@/lib/workflows/streaming/forward-agent-stream-events' import { formatOutputSelector, scopeOutputBlockId } from '@/lib/workflows/streaming/output-selector' import type { BlockCompletionCallbackData, ExecutionCallbacks } from '@/executor/execution/types' import type { ExecutionResult, StreamingExecution } from '@/executor/types' @@ -409,7 +410,7 @@ export class SlackExecutionStreamController { ) this.invocations.set(key, invocation) - const answerFromEventSink = Boolean(stream.subscribe) && !stream.clientStreamTransformed + const answerFromEventSink = shouldForwardAnswerTextFromSink(stream) const unsubscribe = stream.subscribe?.({ onEvent: async (event) => { if (!answerFromEventSink && event.type === 'text_delta') return diff --git a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts index efe186f3b05..5c3c2dd7f65 100644 --- a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts +++ b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts @@ -225,7 +225,7 @@ describe('forwardAgentStreamToExecutionEvents', () => { }) describe('shouldForwardAnswerTextFromSink', () => { - it('requires a sink and an untransformed client stream', () => { + it('requires sink text that matches the client stream', () => { const base = { stream: new ReadableStream(), execution: { success: true, output: {} }, @@ -237,5 +237,13 @@ describe('shouldForwardAnswerTextFromSink', () => { expect( shouldForwardAnswerTextFromSink({ ...base, subscribe, clientStreamTransformed: true }) ).toBe(false) + expect( + shouldForwardAnswerTextFromSink({ + ...base, + subscribe, + clientStreamTransformed: true, + clientSinkTransformed: true, + }) + ).toBe(true) }) }) diff --git a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts index 74ace5d93ae..3a5737966c5 100644 --- a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts +++ b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts @@ -24,8 +24,8 @@ export interface ForwardAgentStreamEventsOptions { * When true, answer text deltas forward live as `stream:chunk` events and an * intermediate `turn_end` forwards as `stream:chunk_reset`. The caller MUST * then stop emitting `stream:chunk` from the block's byte stream, or clients - * receive the final turn's text twice. Never enable for response-format - * projected streams ({@link StreamingExecution.clientStreamTransformed}). + * receive the final turn's text twice. Response-format projected streams may + * enable this only when their sink text is projected too. */ forwardAnswerText?: boolean /** Builds the safe display copy without exposing provenance to the stream bridge. */ @@ -51,7 +51,10 @@ async function projectDisplayValue( * `forwardAnswerText`) instead of the block's byte stream. */ export function shouldForwardAnswerTextFromSink(streamingExec: StreamingExecution): boolean { - return Boolean(streamingExec.subscribe) && streamingExec.clientStreamTransformed !== true + return ( + Boolean(streamingExec.subscribe) && + (streamingExec.clientStreamTransformed !== true || streamingExec.clientSinkTransformed === true) + ) } /** diff --git a/apps/sim/lib/workflows/streaming/streaming.test.ts b/apps/sim/lib/workflows/streaming/streaming.test.ts index 1400be73a1a..b8325a7fc09 100644 --- a/apps/sim/lib/workflows/streaming/streaming.test.ts +++ b/apps/sim/lib/workflows/streaming/streaming.test.ts @@ -1436,6 +1436,63 @@ describe('createStreamingResponse agent-events-v1', () => { expect(events.some((event) => event.event === 'chunk_reset')).toBe(false) }) + it('streams projected sink text live when it matches the transformed byte stream', async () => { + const headers = new Headers({ + 'x-sim-stream-protocol': 'agent-events-v1', + }) + const stream = await createStreamingResponse({ + requestId: 'request-1', + requestHeaders: headers, + streamConfig: { + includeThinking: false, + includeToolCalls: false, + selectedOutputs: ['agent-1_answer'], + }, + executeFn: async ({ onStream }) => { + let sink: AgentStreamSink | undefined + const onStreamPromise = onStream({ + blockId: 'agent-1', + stream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('fallback bytes')) + controller.close() + }, + }), + streamFormat: 'text', + subscribe: (nextSink) => { + sink = nextSink + return () => {} + }, + clientStreamTransformed: true, + clientSinkTransformed: true, + execution: { + blockId: 'agent-1', + success: true, + output: { answer: 'Live answer' }, + logs: [], + metadata: {}, + }, + }) + + await sink?.onEvent({ type: 'text_delta', text: 'Live ', turn: 'pending' }) + await sink?.onEvent({ type: 'text_delta', text: 'answer', turn: 'pending' }) + await sink?.onEvent({ type: 'turn_end', turn: 'final' }) + await onStreamPromise + + return { + success: true, + output: { answer: 'Live answer' }, + logs: [], + } + }, + }) + + const events = await collectSSEEvents(stream) + expect(events.filter((event) => event.chunk !== undefined).map((event) => event.chunk)).toEqual( + ['Live ', 'answer'] + ) + }) + it('includeThinking without includeToolCalls does not emit tool frames', async () => { const headers = new Headers({ 'x-sim-stream-protocol': 'agent-events-v1', diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts index dda38e49dab..5a9905d237f 100644 --- a/apps/sim/lib/workflows/streaming/streaming.ts +++ b/apps/sim/lib/workflows/streaming/streaming.ts @@ -37,6 +37,7 @@ import { type ChatStreamToolFrame, clientAcceptsAgentStreamProtocol, } from '@/lib/workflows/streaming/agent-stream-protocol' +import { shouldForwardAnswerTextFromSink } from '@/lib/workflows/streaming/forward-agent-stream-events' import type { BlockLog, ExecutionResult, StreamingExecution } from '@/executor/types' import { projectResolvedSecretDiagnosticError } from '@/executor/utils/resolved-secret-content-projection' import { navigatePathAsync } from '@/executor/variables/resolvers/reference-async.server' @@ -614,9 +615,7 @@ export async function createStreamingResponse( * the byte stream as the frame source either way. */ const sinkAnswerText = - clientAcceptsProtocol && - Boolean(streamingExec.subscribe) && - streamingExec.clientStreamTransformed !== true + clientAcceptsProtocol && shouldForwardAnswerTextFromSink(streamingExec) /** False until the first chunk since block start or since a reset. */ let emittedSinceReset = false