diff --git a/apps/docs/content/docs/platform/enterprise/custom-blocks.mdx b/apps/docs/content/docs/platform/enterprise/custom-blocks.mdx index 45761df87d3..1ba9a83bad2 100644 --- a/apps/docs/content/docs/platform/enterprise/custom-blocks.mdx +++ b/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. + 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 ### 6. Choose whether runs are traced diff --git a/apps/docs/content/docs/workflows/deployment/agent-events.mdx b/apps/docs/content/docs/workflows/deployment/agent-events.mdx index aa14fdbc1e6..83f6bb59fbf 100644 --- a/apps/docs/content/docs/workflows/deployment/agent-events.mdx +++ b/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. diff --git a/apps/sim/app/(interfaces)/chat/[identifier]/chat.tsx b/apps/sim/app/(interfaces)/chat/[identifier]/chat.tsx index f066279140d..3ff128e6cae 100644 --- a/apps/sim/app/(interfaces)/chat/[identifier]/chat.tsx +++ b/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', diff --git a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx index 4362096144c..0e84bc7d8bf 100644 --- a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.test.tsx +++ b/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) => { diff --git a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts b/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts index 8dbac2525a3..bf6f2e25911 100644 --- a/apps/sim/app/(interfaces)/chat/hooks/use-chat-streaming.ts +++ b/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) { diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/components/output-select/output-select.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/components/output-select/output-select.tsx index 62b09885d61..715b5f43bd8 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/chat/components/output-select/output-select.tsx +++ b/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)] : []), diff --git a/apps/sim/blocks/custom/build-config.test.ts b/apps/sim/blocks/custom/build-config.test.ts index 1d5079f88a3..3616380a5ab 100644 --- a/apps/sim/blocks/custom/build-config.test.ts +++ b/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' }] }, diff --git a/apps/sim/blocks/custom/build-config.ts b/apps/sim/blocks/custom/build-config.ts index f7e20d82bdb..5eede08337f 100644 --- a/apps/sim/blocks/custom/build-config.ts +++ b/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 } diff --git a/apps/sim/ee/custom-blocks/components/custom-block-detail.tsx b/apps/sim/ee/custom-blocks/components/custom-block-detail.tsx index 6ee3c467b1e..10998c7b279 100644 --- a/apps/sim/ee/custom-blocks/components/custom-block-detail.tsx +++ b/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 + {canManageBlock && + (o.streaming || + isCustomBlockStreamSource(deployed.data?.blocks?.[o.blockId], o.path)) ? ( + + setOutputs((current) => + current.map((output) => + encodeOutput(output.blockId, output.path) === key + ? { ...output, streaming: value === 'live' } + : output + ) + ) + } + /> + ) : ( + + {o.streaming ? 'Live' : 'Final'} + + )} ) })} diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index 250542ee7be..9c68d46c934 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/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({ diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index 744f6b63123..09898758fe7 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/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 | 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 | undefined + let streamDeliveryError: Error | undefined let processedClientStream: ReadableStream | 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. diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts index 6b1d4283ef3..ad565e07fd2 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { + createBlock, encryptionMockFns, environmentUtilsMockFns, loggerMock, @@ -17,11 +18,12 @@ import { remapCustomBlockInputKeys, WorkflowBlockHandler, } from '@/executor/handlers/workflow/workflow-handler' -import type { ExecutionContext } from '@/executor/types' +import type { ExecutionContext, StreamingExecution } from '@/executor/types' import { ANONYMOUS_SECRET_TRACE_REPLACEMENT, ResolvedSecretTraceRegistry, } from '@/executor/utils/resolved-secret-trace-registry' +import type { AgentStreamEvent, AgentStreamSink } from '@/providers/stream-events' import type { SerializedBlock } from '@/serializer/types' const mockWorkflowLogger = vi.mocked(loggerMock.createLogger).mock.results[ @@ -1398,6 +1400,217 @@ describe('WorkflowBlockHandler', () => { mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } }) }) + describe('public output streaming', () => { + const exposedOutputs = [{ blockId: 'b1', path: 'content', name: 'answer', streaming: true }] + + beforeEach(() => { + mockGetCustomBlockAuthority.mockResolvedValue({ + workflowId: 'source-workflow-id', + organizationId: 'org-1', + ownerUserId: 'owner-9', + exposedOutputs, + requiredInputIds: [], + traceChildRuns: false, + }) + mockReadWorkflowDefinitionAsExecutor.mockResolvedValue({ + workflow: { + id: 'source-workflow-id', + name: 'Private workflow', + workspaceId: 'workspace-source', + variables: {}, + }, + workspaceId: 'workspace-source', + state: { + deploymentVersionId: 'deployment-version-1', + blocks: { b1: createBlock({ id: 'b1', type: 'agent', name: 'Private agent' }) }, + edges: [], + loops: {}, + parallels: {}, + }, + }) + }) + + it('maps a selected public output with tracing off and keeps private metadata out', async () => { + const received: StreamingExecution[] = [] + const chunks: string[] = [] + const onBlockStart = vi.fn() + const onBlockComplete = vi.fn() + const events: AgentStreamEvent[] = [] + const subscribe = vi.fn((sink: AgentStreamSink) => { + void sink.onEvent({ type: 'thinking_delta', text: 'private thinking' }) + void sink.onEvent({ type: 'tool_call_start', id: 'private-id', name: 'private-tool' }) + void sink.onEvent({ type: 'text_delta', text: 'public answer', turn: 'final' }) + void sink.onEvent({ type: 'turn_end', turn: 'final' }) + return vi.fn() + }) + const onStream = vi.fn(async (stream: StreamingExecution) => { + received.push(stream) + stream.subscribe?.({ + onEvent: (event) => { + events.push(event) + }, + }) + chunks.push(await new Response(stream.stream).text()) + }) + mockExecutorExecute.mockImplementation(async () => { + const extensions = executorOptions.at(-1)!.contextExtensions as ExecutionContext + expect(extensions.selectedOutputs).toEqual(['b1_content']) + expect(extensions.stream).toBe(true) + await extensions.onStream?.({ + blockId: 'b1', + executionOrder: 99, + streamFormat: 'text', + subscribe, + stream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('public answer')) + controller.close() + }, + }), + execution: { + success: true, + output: { content: 'public answer', thinking: 'private thinking' }, + }, + displayResolvedSecretTraceProvenance: { + version: 1, + complete: true, + entries: [{ encryptedValue: 'sealed', name: 'PRIVATE_KEY' }], + scope: { userId: 'owner-9' }, + }, + }) + return { + success: true, + output: {}, + logs: [{ blockId: 'b1', success: true, output: { content: 'public answer' } }], + } + }) + const ctx = customBlockContext({ + stream: true, + selectedOutputs: [`${mockBlock.id}_answer`], + onStream, + onBlockStart, + onBlockComplete, + }) + const output = await handler.executeWithNode( + ctx, + customBlock(), + {}, + { nodeId: mockBlock.id, executionOrder: 7 } + ) + + expect(chunks).toEqual(['public answer']) + expect(received[0]).toMatchObject({ + blockId: mockBlock.id, + outputPath: 'answer', + executionOrder: 7, + streamId: expect.any(String), + childWorkflowInstanceId: expect.any(String), + execution: { success: true, output: {} }, + }) + expect(received[0].displayResolvedSecretTraceProvenance).toEqual({ + version: 1, + complete: true, + entries: [{ encryptedValue: 'sealed' }], + }) + expect(events).toEqual([ + { type: 'text_delta', text: 'public answer', turn: 'final' }, + { type: 'turn_end', turn: 'final' }, + ]) + expect(subscribe).toHaveBeenCalledOnce() + expect(onBlockStart).not.toHaveBeenCalled() + expect(onBlockComplete).not.toHaveBeenCalled() + expect(output).toMatchObject({ + answer: 'public answer', + success: true, + _childWorkflowInstanceId: received[0].childWorkflowInstanceId, + }) + expect(output).not.toHaveProperty('childTraceSpans') + }) + + it('does not enable a live source unless its public output is selected', async () => { + await handler.execute( + customBlockContext({ stream: true, selectedOutputs: [`${mockBlock.id}_success`] }), + customBlock(), + {} + ) + expect(executorOptions[0].contextExtensions.stream).toBe(false) + expect(executorOptions[0].contextExtensions.selectedOutputs).toEqual([]) + }) + + it('still rejects selectors targeting private child outputs', async () => { + await expect( + handler.execute( + customBlockContext({ + stream: true, + selectedOutputs: ['source-workflow-id.b1_content'], + }), + customBlock(), + {} + ) + ).rejects.toThrow() + expect(mockExecutorExecute).not.toHaveBeenCalled() + }) + + it('fails if a source deployment changes to an unsupported block', async () => { + mockReadWorkflowDefinitionAsExecutor.mockResolvedValue({ + workflow: { + id: 'source-workflow-id', + name: 'Private workflow', + workspaceId: 'workspace-source', + variables: {}, + }, + workspaceId: 'workspace-source', + state: { + deploymentVersionId: 'deployment-version-1', + blocks: { b1: createBlock({ id: 'b1', type: 'api' }) }, + edges: [], + loops: {}, + parallels: {}, + }, + }) + await expect( + handler.execute( + customBlockContext({ stream: true, selectedOutputs: [`${mockBlock.id}_answer`] }), + customBlock(), + {} + ) + ).rejects.toThrow() + expect(mockExecutorExecute).not.toHaveBeenCalled() + }) + + it('redacts a source stream error at the custom block boundary', async () => { + const onStream = vi.fn(async (stream: StreamingExecution) => { + await expect(new Response(stream.stream).text()).rejects.toThrow( + 'Custom block output stream failed' + ) + }) + mockExecutorExecute.mockImplementation(async () => { + const extensions = executorOptions[0].contextExtensions as ExecutionContext + await extensions.onStream?.({ + blockId: 'b1', + streamFormat: 'text', + stream: new ReadableStream({ + start(controller) { + controller.error(new Error('PRIVATE_PROVIDER_SECRET')) + }, + }), + execution: { success: false, output: {} }, + }) + return { success: true, output: {} } + }) + await handler.execute( + customBlockContext({ + stream: true, + selectedOutputs: [`${mockBlock.id}_answer`], + onStream, + }), + customBlock(), + {} + ) + expect(onStream).toHaveBeenCalledOnce() + }) + }) + describe('live child spans for the terminal reconcile', () => { const rawSpan = { id: 's1', name: 'Agent 1', type: 'agent', blockId: 'b1' } const projectedSpan = { ...rawSpan, input: { key: '{{PUBLISHER_SECRET}}' } } diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.ts b/apps/sim/executor/handlers/workflow/workflow-handler.ts index 3eb08c62779..f0b60bd507d 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.ts @@ -25,6 +25,10 @@ import { type StartBlockRunIdentity, } from '@/lib/workflows/executor/start-run-identity' import { extractInputFieldsFromBlocks } from '@/lib/workflows/input-format' +import { + assertCustomBlockStreamingOutputs, + selectCustomBlockStreamingOutputs, +} from '@/lib/workflows/streaming/custom-block-output' import { scopeOutputBlockId, selectChildOutputSelectors, @@ -507,9 +511,25 @@ export class WorkflowBlockHandler implements BlockHandler { `Selected stream output exceeds the maximum child workflow depth of ${DEFAULTS.MAX_SSE_CHILD_DEPTH}` ) } - const childSelectedOutputs = isCustomBlock ? [] : childOutputSelection.selectedOutputs + if (isCustomBlock) { + assertCustomBlockStreamingOutputs(exposedOutputs, childWorkflow.rawBlocks || {}) + } + const publicOutputSelection = selectCustomBlockStreamingOutputs( + effectiveBlockId, + exposedOutputs, + ctx.selectedOutputs + ) + if (!withinSseChildDepth && publicOutputSelection.selectedOutputs.length > 0) { + throw new BoundarySafeError({ + errorType: 'depth_limit', + message: 'Selected streaming output exceeds the maximum child workflow depth', + }) + } + const childSelectedOutputs = isCustomBlock + ? publicOutputSelection.selectedOutputs + : childOutputSelection.selectedOutputs const shouldStreamChild = - shouldPropagateCallbacks && Boolean(ctx.stream) && childSelectedOutputs.length > 0 + withinSseChildDepth && Boolean(ctx.stream) && childSelectedOutputs.length > 0 if (!withinSseChildDepth && !isCustomBlock) { logger.info('Dropping SSE callbacks beyond max child depth', { @@ -825,26 +845,107 @@ export class WorkflowBlockHandler implements BlockHandler { } } } - if (shouldPropagateCallbacks) { - if (shouldStreamChild) { - childCallbacks.onStream = async (streamingExecution) => { - if (!streamingExecution.blockId) { - throw new Error('Child workflow stream is missing its block ID') - } - if (!ctx.onStream) { - throw new Error('Child workflow stream has no parent stream callback') + if (shouldStreamChild) { + childCallbacks.onStream = async (streamingExecution) => { + if (!streamingExecution.blockId) { + throw new Error('Child workflow stream is missing its block ID') + } + if (!ctx.onStream) { + throw new Error('Child workflow stream has no parent stream callback') + } + if (isCustomBlock) { + const output = publicOutputSelection.outputsByBlockId.get(streamingExecution.blockId) + if (!output) throw new Error('Custom block received an unselected source stream') + if (streamingExecution.streamFormat !== 'text') { + throw new Error('Custom block output requires a projected text stream') } - const selectedBlockRef = - childOutputSelection.selectedBlockRefs.get(streamingExecution.blockId) ?? - streamingExecution.blockId - await ctx.onStream({ - ...streamingExecution, - blockId: scopeOutputBlockId(workflowId, selectedBlockRef), - childWorkflowInstanceId: streamingExecution.childWorkflowInstanceId ?? instanceId, + const reader = streamingExecution.stream.getReader() + let sourceSettled = false + const publicStream = new ReadableStream({ + async pull(controller) { + try { + const { done, value } = await reader.read() + if (done) { + sourceSettled = true + reader.releaseLock() + controller.close() + } else { + controller.enqueue(value) + } + } catch { + sourceSettled = true + reader.releaseLock() + controller.error(new Error('Custom block output stream failed')) + } + }, + async cancel(reason) { + try { + await reader.cancel(reason) + } finally { + reader.releaseLock() + } + }, }) + const provenance = streamingExecution.displayResolvedSecretTraceProvenance + /** Only answer text and turn retractions cross; tool/thinking events stay private. */ + await ctx + .onStream({ + blockId: effectiveBlockId, + outputPath: output.name, + streamId: generateId(), + childWorkflowInstanceId: instanceId, + executionOrder: nodeMetadata?.executionOrder, + stream: publicStream, + streamFormat: 'text', + subscribe: streamingExecution.subscribe + ? (sink) => + streamingExecution.subscribe!({ + onEvent: (event) => { + if (event.type === 'text_delta') { + return sink.onEvent({ + type: 'text_delta', + text: event.text, + turn: event.turn, + }) + } + if (event.type === 'turn_end') { + return sink.onEvent({ type: 'turn_end', turn: event.turn }) + } + }, + }) + : undefined, + displayResolvedSecretTraceProvenance: provenance + ? { + version: 1, + complete: provenance.complete, + entries: provenance.entries.map(({ encryptedValue }) => ({ encryptedValue })), + } + : undefined, + execution: { success: true, output: {} }, + }) + .catch(async (error: unknown) => { + if (!sourceSettled) { + await reader.cancel() + reader.releaseLock() + } + throw error + }) + return } + const selectedBlockRef = + childOutputSelection.selectedBlockRefs.get(streamingExecution.blockId) ?? + streamingExecution.blockId + await ctx.onStream({ + ...streamingExecution, + blockId: scopeOutputBlockId(workflowId, selectedBlockRef), + childWorkflowInstanceId: streamingExecution.childWorkflowInstanceId ?? instanceId, + }) } + } + if (shouldPropagateCallbacks) { childCallbacks.onChildWorkflowInstanceReady = ctx.onChildWorkflowInstanceReady + } + if (shouldPropagateCallbacks || shouldStreamChild) { childCallbacks.childWorkflowContext = { parentBlockId: instanceId, workflowName: childWorkflowName, @@ -992,11 +1093,10 @@ export class WorkflowBlockHandler implements BlockHandler { return { ...exposedOutput, ...buildChildTraceHandle(childExecutionId, traceChildRuns), - // Both are only set while the child is streaming to an identified consumer. The - // instance id is how the terminal correlates the child's live rows back to this - // invocation; the spans let it reconcile a row whose completion event was lost. - // The block executor lifts them onto the block log and strips them from state. - ...(shouldPropagateCallbacks ? { _childWorkflowInstanceId: instanceId } : {}), + /** Correlate public streams and authorized child traces with this invocation. */ + ...(shouldPropagateCallbacks || shouldStreamChild + ? { _childWorkflowInstanceId: instanceId } + : {}), ...(childTraceSpans.length > 0 ? { childTraceSpans } : {}), } } diff --git a/apps/sim/executor/types.ts b/apps/sim/executor/types.ts index 36c7256967b..2d48f6dcd3c 100644 --- a/apps/sim/executor/types.ts +++ b/apps/sim/executor/types.ts @@ -671,6 +671,10 @@ export interface StreamingExecution { blockId?: string /** Internal identity that disambiguates repeated invocations of one child workflow. */ childWorkflowInstanceId?: string + /** Public custom-block field; its stream settles when the enclosing block completes. */ + outputPath?: string + /** Opaque identity for one projected field stream, including repeated source executions. */ + streamId?: string /** Per-run invocation order, unique across loop and parallel executions. */ executionOrder?: number /** diff --git a/apps/sim/lib/api/contracts/custom-blocks.ts b/apps/sim/lib/api/contracts/custom-blocks.ts index 6e0cc76533a..f31c37a2416 100644 --- a/apps/sim/lib/api/contracts/custom-blocks.ts +++ b/apps/sim/lib/api/contracts/custom-blocks.ts @@ -35,6 +35,7 @@ const exposedOutputSchema = z.object({ blockId: z.string().min(1), path: z.string().min(1), name: z.string().min(1).max(60), + streaming: z.boolean().optional(), }) /** @@ -71,7 +72,7 @@ export const customBlockSchema = z.object({ /** Whether this block's runs are joined into consumers' traces, org-wide. */ traceChildRuns: z.boolean(), inputFields: z.array(inputFieldSchema), - /** Curated outputs exposed to consumers; empty = expose the child's whole result. */ + /** Curated public outputs; legacy empty definitions expose no data fields. */ exposedOutputs: z.array(exposedOutputSchema), }) diff --git a/apps/sim/lib/webhooks/slack-execution-stream.test.ts b/apps/sim/lib/webhooks/slack-execution-stream.test.ts index 8f0c5cc812f..6bdb040859c 100644 --- a/apps/sim/lib/webhooks/slack-execution-stream.test.ts +++ b/apps/sim/lib/webhooks/slack-execution-stream.test.ts @@ -98,7 +98,8 @@ async function createController( user: 'U123', user_team_id: 'T123', }, - } + }, + abortSignal?: AbortSignal ) { const loggingSession = createLoggingSession() const controller = await SlackExecutionStreamController.create({ @@ -110,6 +111,7 @@ async function createController( triggerInput, config, loggingSession: loggingSession as never, + abortSignal, }) return { controller, loggingSession } } @@ -127,6 +129,189 @@ describe('SlackExecutionStreamController', () => { }) }) + describe('custom block public fields', () => { + const config: SlackStreamResponseConfig = { + ...BASE_CONFIG, + outputConfigs: [ + { blockId: 'custom', path: 'answer' }, + { blockId: 'custom', path: 'summary' }, + ], + } + const completion = { + output: { success: true, answer: 'Live answer', summary: 'Final summary' }, + executionTime: 10, + startedAt: '2026-01-01T00:00:00.000Z', + endedAt: '2026-01-01T00:00:00.010Z', + executionOrder: 4, + childWorkflowInstanceId: 'custom-invocation', + } + + it('waits for custom completion, then sends other selected fields without duplicating the answer', async () => { + const { controller } = await createController(config) + await controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Live answer'), + streamFormat: 'text', + execution: { success: true, output: {} }, + }) + expect(mockAppendSlackAgentStream.mock.calls.flatMap((call) => call[3])).toContainEqual({ + type: 'markdown_text', + text: 'Live answer', + }) + expect(mockStopSlackAgentStream).not.toHaveBeenCalled() + + await controller.callbacks.onBlockComplete?.( + 'custom', + 'Published block', + 'custom_block_abc', + completion + ) + const chunks = mockAppendSlackAgentStream.mock.calls.flatMap((call) => call[3]) + expect(chunks.filter((chunk) => chunk.type === 'markdown_text')).toEqual([ + { type: 'markdown_text', text: 'Live answer' }, + { type: 'markdown_text', text: 'Final summary' }, + ]) + expect(mockStopSlackAgentStream).toHaveBeenCalledTimes(2) + controller.assertSucceeded() + }) + + it('keeps repeated source streams and custom invocations separate', async () => { + const { controller } = await createController(config) + for (const instance of ['first', 'second']) { + for (const iteration of [1, 2]) { + await controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: `${instance}-${iteration}`, + childWorkflowInstanceId: instance, + executionOrder: 4, + stream: createByteStream(`${instance} ${iteration}`), + execution: { success: true, output: {} }, + }) + } + await controller.callbacks.onBlockComplete?.( + 'custom', + 'Published block', + 'custom_block_abc', + { ...completion, childWorkflowInstanceId: instance } + ) + } + expect(mockStartSlackAgentStream).toHaveBeenCalledTimes(6) + const taskIds = mockStartSlackAgentStream.mock.calls.map((call) => call[2][0].id) + expect(new Set(taskIds).size).toBe(6) + controller.assertSucceeded() + }) + + it('marks streamed output as failed if the custom block fails after the source ends', async () => { + const { controller } = await createController(config) + await controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Partial answer'), + execution: { success: true, output: {} }, + }) + await controller.finalize({ success: false, output: {}, error: 'Custom block failed' }) + const tasks = mockAppendSlackAgentStream.mock.calls + .flatMap((call) => call[3]) + .filter((chunk) => chunk.type === 'task_update') + expect(tasks.map((task) => task.status)).toEqual(['error']) + expect(mockStopSlackAgentStream).toHaveBeenCalledOnce() + }) + + it('settles an error-port completion as failed even without a success flag', async () => { + const { controller } = await createController(config) + await controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Partial answer'), + execution: { success: true, output: {} }, + }) + await controller.callbacks.onBlockComplete?.( + 'custom', + 'Published block', + 'custom_block_abc', + { + ...completion, + output: { error: 'Custom block execution failed', errorType: 'execution_failed' }, + } + ) + const tasks = mockAppendSlackAgentStream.mock.calls + .flatMap((call) => call[3]) + .filter((chunk) => chunk.type === 'task_update') + expect(tasks.map((task) => task.status)).toEqual(['error']) + expect(mockStopSlackAgentStream).toHaveBeenCalledOnce() + }) + + it('closes a partial message after a delivery failure while still reporting the failure', async () => { + const { controller } = await createController(config) + mockAppendSlackAgentStream.mockRejectedValueOnce(new Error('Slack unavailable')) + await expect( + controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Partial answer'), + execution: { success: true, output: {} }, + }) + ).rejects.toThrow('Slack unavailable') + await controller.finalize({ success: false, output: {} }) + expect(mockStopSlackAgentStream).toHaveBeenCalledOnce() + expect(() => controller.assertSucceeded()).toThrow('Slack unavailable') + }) + + it('closes an already streamed message when the parent run is cancelled', async () => { + const abortController = new AbortController() + const { controller } = await createController(config, undefined, abortController.signal) + await controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Partial answer'), + execution: { success: true, output: {} }, + }) + abortController.abort() + await controller.finalize({ success: false, status: 'cancelled', output: {} }) + expect(mockStopSlackAgentStream).toHaveBeenCalledWith( + 'xoxb-token', + 'C123', + '1700000001.000002', + 'processing', + undefined + ) + controller.assertSucceeded() + }) + + it('rejects an unselected public field before opening a Slack message', async () => { + const { controller } = await createController(config) + await expect( + controller.callbacks.onStream?.({ + blockId: 'custom', + outputPath: 'private', + streamId: 'answer-stream', + childWorkflowInstanceId: 'custom-invocation', + executionOrder: 4, + stream: createByteStream('Hidden'), + execution: { success: true, output: {} }, + }) + ).rejects.toThrow('unselected Slack output') + expect(mockStartSlackAgentStream).not.toHaveBeenCalled() + }) + }) + it('streams agent text and task events for each selected invocation', async () => { const { controller } = await createController() const events: AgentStreamEvent[] = [ diff --git a/apps/sim/lib/webhooks/slack-execution-stream.ts b/apps/sim/lib/webhooks/slack-execution-stream.ts index 3a9f43b3dbe..19140237de2 100644 --- a/apps/sim/lib/webhooks/slack-execution-stream.ts +++ b/apps/sim/lib/webhooks/slack-execution-stream.ts @@ -114,6 +114,7 @@ class SlackInvocationStream { private fullAnswer = '' private thinking = '' private emittedAnswer = false + private settled = false private chain: Promise = Promise.resolve() constructor( @@ -239,27 +240,54 @@ class SlackInvocationStream { return this.enqueue(() => this.appendAnswer(text)) } - complete(): Promise { - return this.enqueue(async () => { - await this.flushThinking() - await this.flushAnswer(true) - if (!this.emittedAnswer && this.fullAnswer) { + flush(): Promise { + return this.enqueue(() => this.flushAnswer(true)) + } + + complete(status: 'complete' | 'error' = 'complete'): Promise { + const settle = async (deliveryFailed = false) => { + if (this.settled) return + const aborted = this.signal?.aborted === true + if ((aborted || deliveryFailed) && (!this.channel || !this.ts)) return + if (!aborted && !deliveryFailed) { + await this.flushThinking() + await this.flushAnswer(true) + } + if (!aborted && !deliveryFailed && !this.emittedAnswer && this.fullAnswer) { const projected = await this.projectFinalText(this.fullAnswer) if (projected) { await this.append(splitMarkdown(projected)) this.emittedAnswer = true } } - await this.append([ - { - type: 'task_update', - id: this.taskId, - title: this.title, - status: 'complete', - }, - ]) - await stopSlackAgentStream(this.token, this.channel!, this.ts!, 'processing', this.signal) - }) + if (!aborted) await this.ensureStarted() + const signal = aborted ? undefined : this.signal + await appendSlackAgentStream( + this.token, + this.channel!, + this.ts!, + [ + { + type: 'task_update', + id: this.taskId, + title: this.title, + status, + }, + ], + signal + ) + await stopSlackAgentStream(this.token, this.channel!, this.ts!, 'processing', signal) + this.settled = true + } + /** Close an opened message even when its last write failed; the controller retains that error. */ + this.chain = this.chain.then( + () => settle(), + (error: unknown) => { + if (status !== 'error') throw error + return settle(true) + } + ) + return this.chain } sendSettledOutput(text: string): Promise { @@ -287,6 +315,10 @@ export class SlackExecutionStreamController { private readonly target: SlackReplyTarget private readonly token: string private readonly invocations = new Map() + private readonly projectedInvocations = new Map< + string, + { parentKey: string; path: string; invocation: SlackInvocationStream } + >() private constructor( private readonly options: SlackExecutionStreamControllerOptions, @@ -389,11 +421,22 @@ export class SlackExecutionStreamController { if (this.selectedForBlock(stream.blockId).length === 0) { throw new Error(`Slack streaming received an unselected block: ${stream.blockId}`) } - const key = this.invocationKey( + const parentKey = this.invocationKey( stream.blockId, stream.executionOrder, stream.childWorkflowInstanceId ) + if (stream.outputPath !== undefined) { + if (!stream.streamId || !stream.childWorkflowInstanceId) { + throw new Error('Custom block stream is missing its public invocation identity') + } + if ( + !this.selectedForBlock(stream.blockId).some((output) => output.path === stream.outputPath) + ) { + throw new Error('Custom block streamed an unselected Slack output') + } + } + const key = stream.outputPath !== undefined ? stream.streamId! : parentKey if (this.invocations.has(key)) { throw new Error(`Duplicate Slack stream invocation: ${key}`) } @@ -401,16 +444,21 @@ export class SlackExecutionStreamController { this.token, this.target, this.options.config, - this.taskId(stream.executionOrder, stream.childWorkflowInstanceId), + this.taskId(stream.executionOrder, stream.streamId ?? stream.childWorkflowInstanceId), this.options.config.taskTitle, (text) => this.projectLiveText(text, stream.displayResolvedSecretTraceProvenance), (text) => this.projectFinalText(text, stream.displayResolvedSecretTraceProvenance), this.options.abortSignal ) this.invocations.set(key, invocation) + if (stream.outputPath !== undefined) { + this.projectedInvocations.set(key, { parentKey, path: stream.outputPath, invocation }) + } - const answerFromEventSink = Boolean(stream.subscribe) && !stream.clientStreamTransformed - const unsubscribe = stream.subscribe?.({ + /** Slack cannot retract an intermediate tool turn, so public fields use final-turn bytes. */ + const subscribe = stream.outputPath === undefined ? stream.subscribe : undefined + const answerFromEventSink = Boolean(subscribe) && !stream.clientStreamTransformed + const unsubscribe = subscribe?.({ onEvent: async (event) => { if (!answerFromEventSink && event.type === 'text_delta') return await invocation.onEvent(event) @@ -430,9 +478,14 @@ export class SlackExecutionStreamController { const remainder = decoder.decode() if (remainder) await invocation.appendProjectedBytes(remainder) } - await invocation.complete() + if (stream.outputPath !== undefined) { + await invocation.flush() + } else { + await invocation.complete() + } } finally { unsubscribe?.() + reader.releaseLock() } } catch (error) { throw this.recordFailure(error) @@ -451,12 +504,23 @@ export class SlackExecutionStreamController { ) if (this.invocations.has(key)) return + const streamedPaths = new Set() + for (const [streamId, projected] of this.projectedInvocations) { + if (projected.parentKey !== key) continue + streamedPaths.add(projected.path) + const failed = data.output.success === false || typeof data.output.error === 'string' + await projected.invocation.complete(failed ? 'error' : 'complete') + this.projectedInvocations.delete(streamId) + this.invocations.set(key, projected.invocation) + } + const display = await this.options.loggingSession.projectDisplayContent( { output: data.output }, data.displayResolvedSecretTraceProvenance ) if (!Object.hasOwn(display, 'output')) return const values = selected.flatMap((selection) => { + if (streamedPaths.has(selection.path)) return [] const value = pluckByPath(display.output, selection.path) return value === undefined ? [] : [{ path: selection.path, value }] }) @@ -483,6 +547,18 @@ export class SlackExecutionStreamController { } async finalize(result: ExecutionResult): Promise { + for (const [streamId, projected] of this.projectedInvocations) { + try { + await projected.invocation.complete('error') + if (result.success && result.status !== 'cancelled' && result.status !== 'paused') { + throw new Error('Custom block stream ended without its block completion') + } + } catch (error) { + this.recordFailure(error) + } finally { + this.projectedInvocations.delete(streamId) + } + } const status = result.status === 'cancelled' || (result.success && result.status !== 'paused') ? 'active' diff --git a/apps/sim/lib/workflows/blocks/flatten-outputs.ts b/apps/sim/lib/workflows/blocks/flatten-outputs.ts index 7c3feb35739..5b0145514b4 100644 --- a/apps/sim/lib/workflows/blocks/flatten-outputs.ts +++ b/apps/sim/lib/workflows/blocks/flatten-outputs.ts @@ -31,6 +31,7 @@ export interface FlattenedBlockOutput { /** Type from the block's output schema (e.g. `string`, `number`, `json`). * Used by the table column-sidebar to pick the right column type. */ leafType?: string + streaming?: boolean } /** @@ -113,6 +114,7 @@ export function flattenWorkflowOutputs( blockType: block.type, path: fullPath, leafType: declaredType, + ...(isRecordLike(outputObj) && outputObj.streaming === true ? { streaming: true } : {}), }) return } diff --git a/apps/sim/lib/workflows/custom-blocks/operations.test.ts b/apps/sim/lib/workflows/custom-blocks/operations.test.ts index 1df4482fb55..d849a9b52f5 100644 --- a/apps/sim/lib/workflows/custom-blocks/operations.test.ts +++ b/apps/sim/lib/workflows/custom-blocks/operations.test.ts @@ -58,6 +58,34 @@ describe('custom block entitlement', () => { }) }) +describe('custom block streaming publication', () => { + const exposedOutputs = [{ blockId: 'agent', path: 'content', name: 'answer', streaming: true }] + + it('rejects a streaming source that the active deployment cannot stream', async () => { + queueTableRows(schemaMock.workflow, [{ id: 'wf-1', workspaceId: 'ws-1', isDeployed: true }]) + loadDeployedWorkflowState.mockResolvedValue({ blocks: { agent: { type: 'api' } } }) + await expect(publishCustomBlock({ ...publishParams, exposedOutputs })).rejects.toThrow( + CustomBlockValidationError + ) + expect(loadDeployedWorkflowState).toHaveBeenCalledWith('wf-1') + }) + + it('validates updates against the canonical published workflow', async () => { + queueTableRows(schemaMock.customBlock, [{ workflowId: 'source-workflow' }]) + loadDeployedWorkflowState.mockResolvedValue({ blocks: { agent: { type: 'agent' } } }) + await expect(updateCustomBlock('published-block', { exposedOutputs })).resolves.toBeUndefined() + expect(loadDeployedWorkflowState).toHaveBeenCalledWith('source-workflow') + }) + + it('rejects a stale streaming mapping on update', async () => { + queueTableRows(schemaMock.customBlock, [{ workflowId: 'source-workflow' }]) + loadDeployedWorkflowState.mockResolvedValue({ blocks: {} }) + await expect(updateCustomBlock('published-block', { exposedOutputs })).rejects.toThrow( + CustomBlockValidationError + ) + }) +}) + describe('custom block input hydration', () => { it('passes the joined source workspace to deployed-state loading', async () => { const block = { diff --git a/apps/sim/lib/workflows/custom-blocks/operations.ts b/apps/sim/lib/workflows/custom-blocks/operations.ts index a35f980b7e1..380e36644e2 100644 --- a/apps/sim/lib/workflows/custom-blocks/operations.ts +++ b/apps/sim/lib/workflows/custom-blocks/operations.ts @@ -7,6 +7,7 @@ import { workspace, } from '@sim/db/schema' import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' import { generateId, generateShortId } from '@sim/utils/id' import { and, eq, isNull, ne, sql } from 'drizzle-orm' import { isOrganizationFeatureEntitled } from '@/lib/billing/core/subscription' @@ -16,6 +17,7 @@ import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import type { DbOrTx } from '@/lib/db/types' import { extractInputFieldsFromBlocks, type WorkflowInputField } from '@/lib/workflows/input-format' import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils' +import { assertCustomBlockStreamingOutputs } from '@/lib/workflows/streaming/custom-block-output' import { getWorkspaceWithOwner } from '@/lib/workspaces/permissions/utils' import type { CustomBlockOutput, CustomBlockRow } from '@/blocks/custom/build-config' import { CUSTOM_BLOCK_TYPE_PREFIX, isReservedOutputName } from '@/blocks/custom/build-config' @@ -23,6 +25,19 @@ import { CUSTOM_BLOCK_TYPE_PREFIX, isReservedOutputName } from '@/blocks/custom/ const logger = createLogger('CustomBlocksOperations') const CUSTOM_BLOCK_HYDRATION_CONCURRENCY = 10 +async function validateStreamingOutputs( + workflowId: string, + outputs: readonly CustomBlockOutput[] +): Promise { + if (!outputs.some((output) => output.streaming)) return + const deployed = await loadDeployedWorkflowState(workflowId) + try { + assertCustomBlockStreamingOutputs(outputs, deployed.blocks) + } catch (error) { + throw new CustomBlockValidationError(getErrorMessage(error)) + } +} + /** Whether the deployment permits Custom Blocks surfaces independent of an organization's plan. */ export function isCustomBlocksDeploymentEnabled(): boolean { return isBillingEnabled || isCustomBlocksEnabled @@ -509,6 +524,8 @@ export async function publishCustomBlock(params: { throw new CustomBlockValidationError('You can only publish a workflow from its own workspace') } + await validateStreamingOutputs(workflowId, exposedOutputs) + const id = generateId() const type = `${CUSTOM_BLOCK_TYPE_PREFIX}${generateShortId(10).toLowerCase()}` const now = new Date() @@ -607,6 +624,15 @@ export async function updateCustomBlock( if (updates.exposedOutputs !== undefined) { assertNoReservedOutputNames(updates.exposedOutputs) assertCuratedOutputs(updates.exposedOutputs) + if (updates.exposedOutputs.some((output) => output.streaming)) { + const [published] = await db + .select({ workflowId: customBlock.workflowId }) + .from(customBlock) + .where(eq(customBlock.id, id)) + .limit(1) + if (!published) throw new CustomBlockValidationError('Custom block not found') + await validateStreamingOutputs(published.workflowId, updates.exposedOutputs) + } } const patch: Partial = { updatedAt: new Date() } if (updates.name !== undefined) patch.name = updates.name diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index 241e649be37..3a1f9371081 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -44,7 +44,12 @@ export interface ExecuteWorkflowOptions { blockType: string, executionOrder: number ) => Promise - onBlockComplete?: (blockId: string, output: unknown, outputBlockId?: string) => Promise + onBlockComplete?: ( + blockId: string, + output: unknown, + outputBlockId?: string, + childWorkflowInstanceId?: string + ) => Promise /** Transfers post-execution logging ownership to the streaming caller after execution succeeds. */ skipLoggingComplete?: boolean includeFileBase64?: boolean @@ -212,7 +217,12 @@ export async function executeWorkflow( _blockType: string, data: BlockCompletionCallbackData ) => { - await streamConfig.onBlockComplete!(blockId, data.output, data.outputBlockId) + await streamConfig.onBlockComplete!( + blockId, + data.output, + data.outputBlockId, + data.childWorkflowInstanceId + ) } : undefined, }, diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 4cf96093dc2..78504026236 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -39,6 +39,7 @@ import { } from '@/lib/logs/execution/cancellation' import { LoggingSession } from '@/lib/logs/execution/logging-session' import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server' +import type { ExecuteWorkflowOptions } from '@/lib/workflows/executor/execute-workflow' import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' import { type ExecutionEvent, @@ -418,7 +419,7 @@ interface StartResumeExecutionArgs { userId: string sendEvent?: (event: ExecutionEvent) => void onStream?: (streamingExec: StreamingExecution) => Promise - onBlockComplete?: (blockId: string, output: unknown) => Promise + onBlockComplete?: ExecuteWorkflowOptions['onBlockComplete'] abortSignal?: AbortSignal } @@ -1071,7 +1072,7 @@ export class PauseResumeManager { userId: string sendEvent?: (event: ExecutionEvent) => void onStream?: (streamingExec: StreamingExecution) => Promise - onBlockComplete?: (blockId: string, output: unknown) => Promise + onBlockComplete?: ExecuteWorkflowOptions['onBlockComplete'] abortSignal?: AbortSignal }): Promise { const { @@ -1727,7 +1728,12 @@ export class PauseResumeManager { } as ExecutionEvent) if (externalOnBlockComplete) { - await externalOnBlockComplete(blockId, callbackData.output) + await externalOnBlockComplete( + blockId, + callbackData.output, + callbackData.outputBlockId, + callbackData.childWorkflowInstanceId + ) } }, onChildWorkflowInstanceReady: async ( diff --git a/apps/sim/lib/workflows/streaming/agent-stream-protocol.ts b/apps/sim/lib/workflows/streaming/agent-stream-protocol.ts index f18ef078638..36c47cc8dc7 100644 --- a/apps/sim/lib/workflows/streaming/agent-stream-protocol.ts +++ b/apps/sim/lib/workflows/streaming/agent-stream-protocol.ts @@ -35,6 +35,8 @@ export const AGENT_STREAM_PROTOCOL_HEADER = 'x-sim-stream-protocol' as const export const AGENT_STREAM_PROTOCOL_HEADER_LABEL = 'X-Sim-Stream-Protocol' as const export const AGENT_STREAM_PROTOCOL_V1 = 'agent-events-v1' as const +/** Client keys answer text and retractions by `streamId` when present. */ +export const SCOPED_OUTPUT_STREAM_PROTOCOL_V1 = 'scoped-output-v1' as const export type AgentStreamProtocol = typeof AGENT_STREAM_PROTOCOL_V1 @@ -48,17 +50,19 @@ export type AgentStreamProtocol = typeof AGENT_STREAM_PROTOCOL_V1 */ export interface ChatStreamChunkFrame { blockId: string + /** Separates public fields and repeated custom-block invocations. */ + streamId?: string chunk: string } /** - * Negotiated agent-events streams only: the live-streamed answer text for - * `blockId` belonged to an intermediate turn (tool calls follow). Clients - * discard the block's accumulated answer text; the final turn re-streams after - * tools settle. + * Retracts answer text from an intermediate turn before tools run. + * Negotiated clients discard text keyed by `streamId` when present, otherwise + * by `blockId`; the final turn streams after tools settle. */ export interface ChatStreamChunkResetFrame { blockId: string + streamId?: string event: 'chunk_reset' } @@ -122,6 +126,7 @@ export function isChatChunkFrame(value: unknown): value is ChatStreamChunkFrame if (!isRecordLike(value)) return false return ( typeof value.blockId === 'string' && + (value.streamId === undefined || typeof value.streamId === 'string') && typeof value.chunk === 'string' && value.chunk.length > 0 && value.event === undefined @@ -130,7 +135,11 @@ export function isChatChunkFrame(value: unknown): value is ChatStreamChunkFrame export function isChatChunkResetFrame(value: unknown): value is ChatStreamChunkResetFrame { if (!isRecordLike(value)) return false - return value.event === 'chunk_reset' && typeof value.blockId === 'string' + return ( + value.event === 'chunk_reset' && + typeof value.blockId === 'string' && + (value.streamId === undefined || typeof value.streamId === 'string') + ) } export function isChatThinkingFrame(value: unknown): value is ChatStreamThinkingFrame { @@ -184,6 +193,20 @@ export function isChatStreamErrorFrame(value: unknown): value is ChatStreamStrea */ export function clientAcceptsAgentStreamProtocol( requestHeaders: Headers | { get(name: string): string | null } +): boolean { + return hasStreamProtocol(requestHeaders, AGENT_STREAM_PROTOCOL_V1) +} + +/** Enables retractions scoped to one public field and invocation, instead of an entire block. */ +export function clientAcceptsScopedOutputStreams( + requestHeaders: Headers | { get(name: string): string | null } +): boolean { + return hasStreamProtocol(requestHeaders, SCOPED_OUTPUT_STREAM_PROTOCOL_V1) +} + +function hasStreamProtocol( + requestHeaders: Headers | { get(name: string): string | null }, + protocol: string ): boolean { const raw = requestHeaders.get(AGENT_STREAM_PROTOCOL_HEADER) if (!raw) { @@ -196,7 +219,7 @@ export function clientAcceptsAgentStreamProtocol( .map((token) => token.trim().toLowerCase()) .filter(Boolean) - return tokens.includes(AGENT_STREAM_PROTOCOL_V1) + return tokens.includes(protocol) } /** True when either agent-event policy is on, before protocol negotiation. */ diff --git a/apps/sim/lib/workflows/streaming/custom-block-output.test.ts b/apps/sim/lib/workflows/streaming/custom-block-output.test.ts new file mode 100644 index 00000000000..6a511becbd5 --- /dev/null +++ b/apps/sim/lib/workflows/streaming/custom-block-output.test.ts @@ -0,0 +1,68 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { + assertCustomBlockStreamingOutputs, + isCustomBlockStreamSource, + selectCustomBlockStreamingOutputs, +} from '@/lib/workflows/streaming/custom-block-output' + +const answer = { blockId: 'private-agent', path: 'content', name: 'answer_text', streaming: true } + +describe('custom block streaming outputs', () => { + it('maps only selected public fields, preserving the existing selector syntax', () => { + const selection = selectCustomBlockStreamingOutputs( + 'custom-instance', + [answer, { blockId: 'other-agent', path: 'content', name: 'summary' }], + ['custom-instance_answer_text', 'custom-instance_summary', 'unrelated_answer_text'] + ) + expect(selection.selectedOutputs).toEqual(['private-agent_content']) + expect([...selection.outputsByBlockId.values()]).toEqual([answer]) + expect( + selectCustomBlockStreamingOutputs('custom-instance', [answer], []).selectedOutputs + ).toEqual([]) + }) + + it.each(['agent', 'pi'])('accepts unstructured %s content', (type) => { + expect( + isCustomBlockStreamSource({ type, subBlocks: { responseFormat: { value: '' } } }, 'content') + ).toBe(true) + }) + + it('rejects unsupported, removed, and structured sources', () => { + for (const blocks of [ + {}, + { 'private-agent': { type: 'api' } }, + { + 'private-agent': { + type: 'agent', + subBlocks: { responseFormat: { value: '{"type":"object"}' } }, + }, + }, + ]) { + expect(() => assertCustomBlockStreamingOutputs([answer], blocks)).toThrow('must reference') + } + expect(() => + assertCustomBlockStreamingOutputs([{ ...answer, path: 'thinking' }], { + 'private-agent': { type: 'agent' }, + }) + ).toThrow('must reference') + }) + + it('rejects ambiguous mappings and nested public names', () => { + const blocks = { 'private-agent': { type: 'agent' } } + expect(() => + assertCustomBlockStreamingOutputs([answer, { ...answer, name: 'other' }], blocks) + ).toThrow('only once') + expect(() => + assertCustomBlockStreamingOutputs([{ ...answer, name: 'answer.text' }], blocks) + ).toThrow('single output field') + }) + + it('leaves final-only outputs compatible with arbitrary source types', () => { + expect(() => + assertCustomBlockStreamingOutputs([{ ...answer, streaming: false }], {}) + ).not.toThrow() + }) +}) diff --git a/apps/sim/lib/workflows/streaming/custom-block-output.ts b/apps/sim/lib/workflows/streaming/custom-block-output.ts new file mode 100644 index 00000000000..b0c5eb9a4c0 --- /dev/null +++ b/apps/sim/lib/workflows/streaming/custom-block-output.ts @@ -0,0 +1,74 @@ +import { isRecordLike } from '@sim/utils/object' +import { + formatInternalOutputSelector, + parseInternalOutputSelector, +} from '@/lib/workflows/streaming/output-selector' +import type { CustomBlockOutput } from '@/blocks/custom/build-config' + +interface StreamSourceBlock { + type: string + subBlocks?: Record +} + +/** Public streams currently expose only unstructured Agent/Pi answer text. */ +export function isCustomBlockStreamSource( + block: StreamSourceBlock | undefined, + path: string +): boolean { + if (!block || !['agent', 'pi'].includes(block.type) || path !== 'content') return false + const subBlock = block.subBlocks?.responseFormat + const responseFormat = isRecordLike(subBlock) ? subBlock.value : subBlock + return responseFormat == null || responseFormat === '' +} + +/** Validate against the deployment being published or executed, never the caller's graph. */ +export function assertCustomBlockStreamingOutputs( + outputs: readonly CustomBlockOutput[], + blocks: Readonly> +): void { + const sources = new Set() + for (const output of outputs) { + if (!output.streaming) continue + if (outputs.filter((candidate) => candidate.name === output.name).length !== 1) { + throw new Error('Each streaming output must have a unique public name') + } + if ( + !output.name || + output.name.includes('.') || + output.name.includes('/') || + output.name.trim() !== output.name + ) { + throw new Error('A streaming output name must be a single output field') + } + if (!isCustomBlockStreamSource(blocks[output.blockId], output.path)) { + throw new Error( + `Streaming output "${output.name}" must reference an Agent or Pi content output without a response format` + ) + } + if (sources.has(output.blockId)) { + throw new Error('Each streaming source can be exposed only once') + } + sources.add(output.blockId) + } +} + +/** Translate selected public fields into private child selectors without exposing child IDs. */ +export function selectCustomBlockStreamingOutputs( + blockId: string, + outputs: readonly CustomBlockOutput[], + selectedOutputs: readonly string[] = [] +): { selectedOutputs: string[]; outputsByBlockId: ReadonlyMap } { + const selectedPaths = new Set( + selectedOutputs + .map(parseInternalOutputSelector) + .filter((selector) => !selector.workflowId && selector.blockId === blockId) + .map((selector) => selector.path) + ) + const selected = outputs.filter((output) => output.streaming && selectedPaths.has(output.name)) + return { + selectedOutputs: selected.map((output) => + formatInternalOutputSelector(output.blockId, output.path) + ), + outputsByBlockId: new Map(selected.map((output) => [output.blockId, output])), + } +} diff --git a/apps/sim/lib/workflows/streaming/nested-output-options.ts b/apps/sim/lib/workflows/streaming/nested-output-options.ts index 1ec3ef71908..31bb0070070 100644 --- a/apps/sim/lib/workflows/streaming/nested-output-options.ts +++ b/apps/sim/lib/workflows/streaming/nested-output-options.ts @@ -18,6 +18,7 @@ export interface WorkflowOutputOption { groupKey: string groupLabel: string path: string + streaming?: boolean menuPath: WorkflowOutputMenuSegment[] } @@ -117,6 +118,7 @@ export function buildWorkflowOutputOptions({ groupKey: menuBlockId, groupLabel, path: output.path, + ...(output.streaming ? { streaming: true } : {}), menuPath: [ ...invocationPath, { diff --git a/apps/sim/lib/workflows/streaming/streaming.test.ts b/apps/sim/lib/workflows/streaming/streaming.test.ts index 1400be73a1a..adc5d87e495 100644 --- a/apps/sim/lib/workflows/streaming/streaming.test.ts +++ b/apps/sim/lib/workflows/streaming/streaming.test.ts @@ -102,6 +102,141 @@ describe('createStreamingResponse', () => { clearLargeValueCacheForTests() }) + it('streams custom fields once per invocation and preserves other selected final outputs', async () => { + const stream = await createStreamingResponse({ + requestId: 'custom-stream', + streamConfig: { selectedOutputs: ['custom_answer', 'custom_summary'] }, + executeFn: async ({ onStream, onBlockComplete }) => { + for (const instance of ['first', 'second']) { + await onStream({ + blockId: 'custom', + outputPath: 'answer', + streamId: `${instance}-answer`, + childWorkflowInstanceId: instance, + streamFormat: 'text', + stream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`${instance} answer`)) + controller.close() + }, + }), + execution: { success: true, output: {} }, + }) + await onBlockComplete( + 'custom', + { answer: `${instance} answer`, summary: `${instance} summary` }, + undefined, + instance + ) + } + /** A later invocation may take a branch which never runs the live source. */ + await onBlockComplete( + 'custom', + { answer: 'settled answer', summary: 'settled summary' }, + undefined, + 'third' + ) + return { success: true, output: {}, logs: [] } + }, + }) + const events = await collectSSEEvents(stream) + const chunks = events.filter((event) => typeof event.chunk === 'string') + expect(chunks.map((event) => String(event.chunk).trim())).toEqual([ + 'first answer', + 'first summary', + 'second answer', + 'second summary', + 'settled answer', + 'settled summary', + ]) + expect(new Set(chunks.map((event) => event.streamId)).size).toBe(6) + expect(events.at(-1)?.event).toBe('final') + }) + + it.each([ + { protocol: 'agent-events-v1', live: false }, + { protocol: 'agent-events-v1, scoped-output-v1', live: true }, + ])('negotiates scoped custom answer retractions: $live', async ({ protocol, live }) => { + const stream = await createStreamingResponse({ + requestId: 'live-custom-stream', + requestHeaders: new Headers({ 'x-sim-stream-protocol': protocol }), + streamConfig: { selectedOutputs: ['custom_answer'] }, + executeFn: async ({ onStream, onBlockComplete }) => { + let sink: AgentStreamSink | undefined + let finishBytes: (() => void) | undefined + const pending = onStream({ + blockId: 'custom', + outputPath: 'answer', + streamId: 'public-stream', + childWorkflowInstanceId: 'invocation', + stream: new ReadableStream({ + start(controller) { + finishBytes = () => { + controller.enqueue(new TextEncoder().encode('Final answer')) + controller.close() + } + }, + }), + subscribe: (subscriber) => { + sink = subscriber + return vi.fn() + }, + execution: { success: true, output: {} }, + }) + if (!sink) throw new Error('Subscription was not installed before pumping') + await sink.onEvent({ type: 'text_delta', text: 'Checking', turn: 'pending' }) + await sink.onEvent({ type: 'turn_end', turn: 'intermediate' }) + await sink.onEvent({ type: 'text_delta', text: 'Final answer', turn: 'final' }) + if (!finishBytes) throw new Error('Test byte stream was not initialized') + finishBytes() + await pending + await onBlockComplete( + 'custom', + { success: true, answer: 'Final answer' }, + undefined, + 'invocation' + ) + return { success: true, output: {} } + }, + }) + const events = await collectSSEEvents(stream) + expect(events.filter((event) => event.event !== 'final')).toEqual( + live + ? [ + { blockId: 'custom', streamId: 'public-stream', chunk: 'Checking' }, + { blockId: 'custom', streamId: 'public-stream', event: 'chunk_reset' }, + { blockId: 'custom', streamId: 'public-stream', chunk: 'Final answer' }, + ] + : [{ blockId: 'custom', streamId: 'public-stream', chunk: 'Final answer' }] + ) + }) + + it('fails when a custom stream supplies an unselected public field', async () => { + const stream = await createStreamingResponse({ + requestId: 'unselected-stream', + streamConfig: { selectedOutputs: ['custom_answer'] }, + executeFn: async ({ onStream }) => { + await onStream({ + blockId: 'custom', + outputPath: 'private', + streamId: 'stream', + childWorkflowInstanceId: 'invocation', + stream: new ReadableStream({ + start(controller) { + controller.close() + }, + }), + execution: { success: true, output: {} }, + }) + return { success: true, output: {} } + }, + }) + const events = await collectSSEEvents(stream) + expect(events).toEqual([ + { event: 'error', error: 'Custom block streamed an unselected output' }, + ]) + }) + it('forwards raw execution state to terminal logging', async () => { const safeComplete = vi.fn().mockResolvedValue(undefined) const executionState = { @@ -946,6 +1081,16 @@ describe('agent stream protocol response headers', () => { expect(agentStreamProtocolResponseHeaders({ requestHeaders: new Headers() })).toEqual({}) expect(agentStreamProtocolResponseHeaders({})).toEqual({}) }) + + it('echoes the negotiated capability for field-scoped retractions', () => { + expect( + agentStreamProtocolResponseHeaders({ + requestHeaders: new Headers({ + 'x-sim-stream-protocol': 'agent-events-v1, scoped-output-v1', + }), + }) + ).toEqual({ 'x-sim-stream-protocol': 'agent-events-v1, scoped-output-v1' }) + }) }) describe('createStreamingResponse agent-events-v1', () => { diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts index 7c57eb35586..5db8aeb6972 100644 --- a/apps/sim/lib/workflows/streaming/streaming.ts +++ b/apps/sim/lib/workflows/streaming/streaming.ts @@ -37,6 +37,8 @@ import { type ChatStreamThinkingFrame, type ChatStreamToolFrame, clientAcceptsAgentStreamProtocol, + clientAcceptsScopedOutputStreams, + SCOPED_OUTPUT_STREAM_PROTOCOL_V1, } from '@/lib/workflows/streaming/agent-stream-protocol' import type { BlockLog, ExecutionResult, StreamingExecution } from '@/executor/types' import { projectResolvedSecretDiagnosticError } from '@/executor/utils/resolved-secret-content-projection' @@ -81,7 +83,12 @@ interface StreamingConfig { export type StreamingExecutorFn = (callbacks: { onStream: (streamingExec: StreamingExecution) => Promise - onBlockComplete: (blockId: string, output: unknown, outputBlockId?: string) => Promise + onBlockComplete: ( + blockId: string, + output: unknown, + outputBlockId?: string, + childWorkflowInstanceId?: string + ) => Promise abortSignal: AbortSignal }) => Promise @@ -119,7 +126,11 @@ export function agentStreamProtocolResponseHeaders(options: { if (!clientAcceptsAgentStreamProtocol(options.requestHeaders)) { return {} } - return { [AGENT_STREAM_PROTOCOL_HEADER]: AGENT_STREAM_PROTOCOL_V1 } + return { + [AGENT_STREAM_PROTOCOL_HEADER]: clientAcceptsScopedOutputStreams(options.requestHeaders) + ? `${AGENT_STREAM_PROTOCOL_V1}, ${SCOPED_OUTPUT_STREAM_PROTOCOL_V1}` + : AGENT_STREAM_PROTOCOL_V1, + } } interface StreamingState { @@ -505,6 +516,8 @@ export async function createStreamingResponse( */ const clientAcceptsProtocol = Boolean(options.requestHeaders) && clientAcceptsAgentStreamProtocol(options.requestHeaders!) + const acceptsScopedOutputs = + Boolean(options.requestHeaders) && clientAcceptsScopedOutputStreams(options.requestHeaders!) /** * Frames additionally require the negotiated protocol: a client that never * declared a version has no contract for their shape, so it keeps the text @@ -542,11 +555,16 @@ export async function createStreamingResponse( streamedSelectedOutputKeys: new Set(), } let thinkingCharsEmitted = 0 + const projectedOutputInvocations = new Map>() const sendChunk = ( blockId: string, content: string, - options: { selectedOutputKey?: string; selectedOutputBytes?: number } = {} + options: { + selectedOutputKey?: string + selectedOutputBytes?: number + streamId?: string + } = {} ) => { const separator = state.processedOutputs.size > 0 ? '\n\n' : '' const chunk = separator + content @@ -558,9 +576,13 @@ export async function createStreamingResponse( state.selectedOutputBytes = nextSelectedOutputBytes state.streamedSelectedOutputKeys.add(options.selectedOutputKey) } - const frame: ChatStreamChunkFrame = { blockId, chunk } + const frame: ChatStreamChunkFrame = { + blockId, + chunk, + ...(options.streamId ? { streamId: options.streamId } : {}), + } controller.enqueue(encodeSSE(frame)) - state.processedOutputs.add(blockId) + state.processedOutputs.add(options.streamId ?? blockId) } const sendThinking = (blockId: string, text: string) => { @@ -606,6 +628,21 @@ export async function createStreamingResponse( logger.warn(`[${requestId}] Streaming execution missing blockId`) return } + const { outputPath, streamId, childWorkflowInstanceId } = streamingExec + if (outputPath !== undefined) { + if (!streamId || !childWorkflowInstanceId) { + throw new Error('Custom block stream is missing its public invocation identity') + } + const selected = getSelectedOutputDescriptors(streamConfig.selectedOutputs ?? []).some( + (descriptor) => descriptor.blockId === blockId && descriptor.path === outputPath + ) + if (!selected) throw new Error('Custom block streamed an unselected output') + const invocationKey = `${blockId}\0${childWorkflowInstanceId}` + const paths = projectedOutputInvocations.get(invocationKey) ?? new Set() + paths.add(outputPath) + projectedOutputInvocations.set(invocationKey, paths) + state.streamedSelectedOutputKeys.add(`${blockId}\0${outputPath}`) + } /** * Negotiated clients get answer text live from the sink (pending deltas @@ -621,6 +658,7 @@ export async function createStreamingResponse( */ const sinkAnswerText = clientAcceptsProtocol && + (outputPath === undefined || acceptsScopedOutputs) && Boolean(streamingExec.subscribe) && streamingExec.clientStreamTransformed !== true @@ -631,10 +669,14 @@ export async function createStreamingResponse( if (!text) return if (!emittedSinceReset) { // sendChunk adds the cross-block separator + output bookkeeping. - sendChunk(blockId, text) + sendChunk(blockId, text, { streamId }) emittedSinceReset = true } else { - const frame: ChatStreamChunkFrame = { blockId, chunk: text } + const frame: ChatStreamChunkFrame = { + blockId, + chunk: text, + ...(streamId ? { streamId } : {}), + } controller.enqueue(encodeSSE(frame)) } } @@ -655,11 +697,15 @@ export async function createStreamingResponse( } } else if (sinkAnswerText && event.type === 'turn_end') { if (event.turn === 'intermediate' && emittedSinceReset) { - const frame: ChatStreamChunkResetFrame = { blockId, event: 'chunk_reset' } + const frame: ChatStreamChunkResetFrame = { + blockId, + event: 'chunk_reset', + ...(streamId ? { streamId } : {}), + } controller.enqueue(encodeSSE(frame)) // Re-arm separator bookkeeping so re-streamed text starts clean. emittedSinceReset = false - state.processedOutputs.delete(blockId) + state.processedOutputs.delete(streamId ?? blockId) } } }, @@ -673,21 +719,24 @@ export async function createStreamingResponse( while (true) { const { done, value } = await reader.read() if (done) { - state.streamCompletionTimes.set(blockId, Date.now()) + if (outputPath === undefined) state.streamCompletionTimes.set(blockId, Date.now()) + const remainder = decoder.decode() + if (remainder && !sinkAnswerText) emitAnswerChunk(remainder) break } const textChunk = decoder.decode(value, { stream: true }) - if (!state.streamedChunks.has(blockId)) { - state.streamedChunks.set(blockId, []) + if (outputPath === undefined) { + if (!state.streamedChunks.has(blockId)) state.streamedChunks.set(blockId, []) + state.streamedChunks.get(blockId)!.push(textChunk) } - state.streamedChunks.get(blockId)!.push(textChunk) if (!sinkAnswerText) { emitAnswerChunk(textChunk) } } } catch (error) { + if (outputPath !== undefined) throw error logger.error( `[${requestId}] Error reading stream for block ${blockId}`, projectResolvedSecretDiagnosticError(error, undefined) @@ -700,6 +749,7 @@ export async function createStreamingResponse( controller.enqueue(encodeSSE(frame)) } finally { unsubscribe?.() + reader.releaseLock() } } @@ -709,10 +759,14 @@ export async function createStreamingResponse( const onBlockCompleteCallback = async ( blockId: string, output: unknown, - outputBlockId?: string + outputBlockId?: string, + childWorkflowInstanceId?: string ) => { const selectedOutputBlockId = outputBlockId ?? blockId state.completedBlockIds.add(selectedOutputBlockId) + const invocationKey = `${selectedOutputBlockId}\0${childWorkflowInstanceId}` + const streamedPaths = projectedOutputInvocations.get(invocationKey) + projectedOutputInvocations.delete(invocationKey) if (!streamConfig.selectedOutputs?.length) { return @@ -739,6 +793,7 @@ export async function createStreamingResponse( : output for (const descriptor of matchingOutputs) { + if (streamedPaths?.has(descriptor.path)) continue if (state.selectedOutputError) { break } @@ -799,6 +854,9 @@ export async function createStreamingResponse( sendChunk(selectedOutputBlockId, formattedOutput, { selectedOutputKey: descriptor.key, selectedOutputBytes, + ...(childWorkflowInstanceId + ? { streamId: `${childWorkflowInstanceId}:${descriptor.path}` } + : {}), }) } } catch (error) { diff --git a/packages/db/schema.ts b/packages/db/schema.ts index 58c431eab58..baf0c4e24eb 100644 --- a/packages/db/schema.ts +++ b/packages/db/schema.ts @@ -4231,12 +4231,13 @@ export const customBlock = pgTable( */ inputs: json('inputs').$type>(), /** - * Curated outputs exposed to consumers: `Array<{ blockId, path, name }>`. Each - * maps a child-workflow block output (blockId + dot-path) to a friendly output - * name on the block. Empty/absent → expose the child's whole `result`. Internal - * plumbing (child workflow id, trace spans) is never exposed. + * Publisher-curated mappings from child outputs to public fields. + * Empty legacy definitions expose no child data and fail at invocation. */ - outputs: json('outputs').$type>(), + outputs: + json('outputs').$type< + Array<{ blockId: string; path: string; name: string; streaming?: boolean }> + >(), enabled: boolean('enabled').notNull().default(true), /** * The publisher's org-wide decision on whether this block's runs are joined into diff --git a/packages/workflow-types/src/blocks.ts b/packages/workflow-types/src/blocks.ts index e40cf7441d8..35bd4f40316 100644 --- a/packages/workflow-types/src/blocks.ts +++ b/packages/workflow-types/src/blocks.ts @@ -81,6 +81,8 @@ export type OutputFieldDefinition = description?: string condition?: OutputCondition hiddenFromDisplay?: boolean + /** This public field can supply live answer text to deployment output consumers. */ + streaming?: boolean } export function isHiddenFromDisplay(def: unknown): boolean {