From 7db79a139b2ed8e0ad80512d5efc875562872c94 Mon Sep 17 00:00:00 2001 From: Bill Leoutsakos Date: Sat, 29 Aug 2026 11:56:58 -0700 Subject: [PATCH 1/2] fix(agent): bound structured output generation --- apps/sim/blocks/blocks/agent.ts | 2 + .../executor/execution/block-executor.test.ts | 31 ++++++++++ apps/sim/executor/execution/block-executor.ts | 46 +++++++++++--- .../executor/execution/block-retry.test.ts | 5 ++ .../handlers/agent/agent-handler.test.ts | 61 +++++++++++++++++++ .../executor/handlers/agent/agent-handler.ts | 17 +++++- .../handlers/shared/response-format.ts | 49 +++++++++++++++ apps/sim/providers/anthropic/core.ts | 5 +- apps/sim/providers/anthropic/utils.test.ts | 2 + apps/sim/providers/anthropic/utils.ts | 6 ++ 10 files changed, 212 insertions(+), 12 deletions(-) diff --git a/apps/sim/blocks/blocks/agent.ts b/apps/sim/blocks/blocks/agent.ts index b09eae6e203..27ed1fd91b8 100644 --- a/apps/sim/blocks/blocks/agent.ts +++ b/apps/sim/blocks/blocks/agent.ts @@ -399,6 +399,8 @@ Return ONLY the JSON array.`, title: 'Max Output Tokens', type: 'short-input', placeholder: 'Enter max tokens (e.g., 4096)...', + description: + 'Maximum response length. When blank, structured responses use 4,096 tokens; other responses use the model default.', mode: 'advanced', condition: { field: 'model', diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index c0758935e81..1c424e163c9 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/apps/sim/executor/execution/block-executor.test.ts @@ -1224,6 +1224,7 @@ describe('BlockExecutor streaming pump', () => { failAfterText?: string streamError?: Error onFullContent?: (content: string) => void | Promise + finishReason?: string resolvedSecret?: { name: string; value: string } separateResultRegistry?: boolean }): BlockHandler { @@ -1279,6 +1280,9 @@ describe('BlockExecutor streaming pump', () => { if (options.attachThinkingOnDrain) { timeSegment.thinkingContent = options.attachThinkingOnDrain } + if (options.finishReason) { + timeSegment.finishReason = options.finishReason + } controller.close() }, }) @@ -1534,6 +1538,33 @@ describe('BlockExecutor streaming pump', () => { expect(callbackError.message).toContain(secret) }) + it('fails token-limited structured streams before downstream completion', async () => { + const onFullContent = vi.fn() + const handler = createAgentEventsStreamingHandler({ + events: [{ type: 'text_delta', text: '{"answer":"unfinished', turn: 'final' }], + finishReason: 'max_tokens', + onFullContent, + }) + const { executor, block, state } = createExecutor(handler) + block.config.params = { + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + } + const ctx = createContext(state) + + await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow( + /maximum output-token limit/i + ) + + expect(onFullContent).not.toHaveBeenCalled() + expect(state.getBlockOutput(block.id)).toMatchObject({ + content: '{"answer":"unfinished', + error: expect.stringMatching(/maximum output-token limit/i), + }) + expect(state.getBlockOutput(block.id)?.providerTiming?.timeSegments?.[0]?.finishReason).toBe( + 'max_tokens' + ) + }) + it('soft-completes on user abort with drained answer text (no failed block)', async () => { const abortController = new AbortController() const handler = createAgentEventsStreamingHandler({ diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index bc4241208fd..81793b86a76 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -34,6 +34,11 @@ import type { ContextExtensions, WorkflowNodeMetadata, } from '@/executor/execution/types' +import { + assertStructuredOutputNotTokenLimited, + parseResponseFormat, + StructuredOutputTokenLimitError, +} from '@/executor/handlers/shared/response-format' import { generatePauseContextId, mapNodeMetadataToPauseScopes, @@ -228,7 +233,7 @@ export class BlockExecutor { } cleanupSelfReference?.() - let streamingPartialOutput: Record | undefined + let failureDiagnosticOutput: Record | undefined try { /** * Only the handler call is retried. A streaming handler returns before any @@ -274,7 +279,7 @@ export class BlockExecutor { blockCtx.resolvedSecretTraceRegistry = resultRegistry?.forkForPropagatedEntries() // Timeout / drain failures may still have projected answer text — keep it // for the failed block output so logs match what the client already saw. - streamingPartialOutput = streamingExec.execution?.output + failureDiagnosticOutput = streamingExec.execution?.output throw streamError } @@ -403,6 +408,9 @@ export class BlockExecutor { commitBlockRegistry() return stateOutput } catch (error) { + if (!failureDiagnosticOutput && error instanceof StructuredOutputTokenLimitError) { + failureDiagnosticOutput = error.diagnosticOutput + } try { return await this.handleBlockError( error, @@ -416,7 +424,7 @@ export class BlockExecutor { inputDisplayRegistry, isSentinel, 'execution', - streamingPartialOutput + failureDiagnosticOutput ) } finally { commitBlockRegistry() @@ -548,7 +556,7 @@ export class BlockExecutor { inputDisplayRegistry: ResolvedSecretTraceRegistry | undefined, isSentinel: boolean, phase: 'input_resolution' | 'execution', - streamingPartialOutput?: Record + failureDiagnosticOutput?: Record ): Promise { const endedAt = new Date().toISOString() const duration = performance.now() - startTime @@ -624,12 +632,25 @@ export class BlockExecutor { error: errorMessage, } - // Keep any answer text already drained before timeout/failure so logs match - // what was projected to the client. - const partialContent = streamingPartialOutput?.content - if (typeof partialContent === 'string' && partialContent) { - errorOutput.content = partialContent + // Retain completed provider diagnostics for observability and billing while + // keeping the block failed so normal downstream execution cannot consume it. + let providerDiagnostics: Record = {} + for (const key of ['content', 'model', 'tokens', 'toolCalls', 'providerTiming', 'cost']) { + const value = failureDiagnosticOutput?.[key] + if (value !== undefined && (key !== 'content' || value !== '')) { + providerDiagnostics[key] = value + } } + if (ctx.piiBlockOutputRedaction?.enabled && Object.keys(providerDiagnostics).length > 0) { + stripThinkingContentFromOutput(providerDiagnostics) + providerDiagnostics = await redactObjectStrings(providerDiagnostics, { + entityTypes: ctx.piiBlockOutputRedaction.entityTypes, + language: ctx.piiBlockOutputRedaction.language, + customPatterns: ctx.piiBlockOutputRedaction.customPatterns, + onFailure: 'throw', + }) + } + Object.assign(errorOutput, providerDiagnostics) // Only real workflow blocks surface a child workflow name. A custom block's // source workflow is never named to its consumer — and before the handler @@ -1106,6 +1127,7 @@ export class BlockExecutor { resolvedInputs?.responseFormat ?? (block.config?.params as Record | undefined)?.responseFormat ?? (block.config as Record | undefined)?.responseFormat + const parsedResponseFormat = parseResponseFormat(responseFormat) const streamFormat = streamingExec.streamFormat ?? 'text' const pump = createAgentStreamPump({ @@ -1238,6 +1260,12 @@ export class BlockExecutor { if (executionOutput && typeof executionOutput === 'object') { let parsedForFormat = false if (responseFormat) { + // Retain the drained text for failed-block diagnostics, but reject it + // before parsing so truncated structured data never reaches downstream. + executionOutput.content = fullContent + if (parsedResponseFormat) { + assertStructuredOutputNotTokenLimited(executionOutput.providerTiming, executionOutput) + } try { const parsed = JSON.parse(fullContent.trim()) streamingExec.execution.output = { diff --git a/apps/sim/executor/execution/block-retry.test.ts b/apps/sim/executor/execution/block-retry.test.ts index d362391e1f4..f0ed11a1135 100644 --- a/apps/sim/executor/execution/block-retry.test.ts +++ b/apps/sim/executor/execution/block-retry.test.ts @@ -13,6 +13,7 @@ import { describe, expect, it } from 'vitest' import { BlockType } from '@/executor/constants' import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' import { isRetryableBlockError, resolveBlockRetryPolicy } from '@/executor/execution/block-retry' +import { StructuredOutputTokenLimitError } from '@/executor/handlers/shared/response-format' import type { SerializedBlock } from '@/serializer/types' function block( @@ -162,6 +163,10 @@ describe('isRetryableBlockError', () => { ).toBe(false) }) + it('never replays token-limited structured output', () => { + expect(isRetryableBlockError(new StructuredOutputTokenLimitError())).toBe(false) + }) + it('finds a deliberate stop that a provider rewrapped, since name is overwritten', () => { const abort = new Error('aborted') abort.name = 'AbortError' diff --git a/apps/sim/executor/handlers/agent/agent-handler.test.ts b/apps/sim/executor/handlers/agent/agent-handler.test.ts index 5d7591cc673..c237f2edbd3 100644 --- a/apps/sim/executor/handlers/agent/agent-handler.test.ts +++ b/apps/sim/executor/handlers/agent/agent-handler.test.ts @@ -1967,6 +1967,67 @@ describe('AgentBlockHandler', () => { }) }) + it.each([ + { + name: 'defaults structured output to 4,096 tokens', + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + maxTokens: undefined, + expectedMaxTokens: 4096, + }, + { + name: 'keeps an explicit structured-output limit', + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + maxTokens: '512', + expectedMaxTokens: 512, + }, + { + name: 'leaves an unstructured output limit unset', + responseFormat: undefined, + maxTokens: undefined, + expectedMaxTokens: undefined, + }, + ])('$name', async ({ responseFormat, maxTokens, expectedMaxTokens }) => { + await handler.execute(mockContext, mockBlock, { + model: 'gpt-4o', + userPrompt: 'Return an answer.', + responseFormat, + maxTokens, + }) + + expect(mockExecuteProviderRequest.mock.calls[0][1].maxTokens).toBe(expectedMaxTokens) + }) + + it('fails non-streaming structured output that reaches its token limit', async () => { + mockExecuteProviderRequest.mockResolvedValueOnce({ + content: '{"answer":"unfinished', + model: 'mock-model', + tokens: { input: 10, output: 4096, total: 4106 }, + timing: { + timeSegments: [ + { + type: 'model', + startTime: 1, + endTime: 2, + duration: 1, + finishReason: 'max_tokens', + }, + ], + }, + toolCalls: [], + }) + + await expect( + handler.execute(mockContext, mockBlock, { + model: 'claude-sonnet-5', + userPrompt: 'Return an answer.', + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + }) + ).rejects.toMatchObject({ + code: 'structured_output_token_limit', + retryable: false, + }) + }) + it('keeps an ordinary response format unchanged without resolver-recorded lineage', async () => { const responseFormat = { name: 'response_schema', diff --git a/apps/sim/executor/handlers/agent/agent-handler.ts b/apps/sim/executor/handlers/agent/agent-handler.ts index 3664fb82227..381876a9af1 100644 --- a/apps/sim/executor/handlers/agent/agent-handler.ts +++ b/apps/sim/executor/handlers/agent/agent-handler.ts @@ -60,7 +60,11 @@ import type { StreamingConfig, ToolInput, } from '@/executor/handlers/agent/types' -import { parseResponseFormat } from '@/executor/handlers/shared/response-format' +import { + assertStructuredOutputNotTokenLimited, + DEFAULT_STRUCTURED_OUTPUT_MAX_TOKENS, + parseResponseFormat, +} from '@/executor/handlers/shared/response-format' import type { BlockHandler, ExecutionContext, StreamingExecution } from '@/executor/types' import { collectBlockData } from '@/executor/utils/block-data' import { stringifyJSON } from '@/executor/utils/json' @@ -2426,7 +2430,11 @@ export class AgentBlockHandler implements BlockHandler { ? Number(inputs.temperature) : undefined, maxTokens: - inputs.maxTokens != null && inputs.maxTokens !== '' ? Number(inputs.maxTokens) : undefined, + inputs.maxTokens != null && inputs.maxTokens !== '' + ? Number(inputs.maxTokens) + : responseFormat + ? DEFAULT_STRUCTURED_OUTPUT_MAX_TOKENS + : undefined, apiKey: inputs.apiKey, azureEndpoint: inputs.azureEndpoint, azureApiVersion: inputs.azureApiVersion, @@ -2759,6 +2767,11 @@ export class AgentBlockHandler implements BlockHandler { ): BlockOutput { const content = result.content + assertStructuredOutputNotTokenLimited(result.timing, { + content, + ...this.createResponseMetadata(result), + }) + try { const extractedJson = JSON.parse(content.trim()) return { diff --git a/apps/sim/executor/handlers/shared/response-format.ts b/apps/sim/executor/handlers/shared/response-format.ts index 9f1ad68a5ef..983a652b1ed 100644 --- a/apps/sim/executor/handlers/shared/response-format.ts +++ b/apps/sim/executor/handlers/shared/response-format.ts @@ -1,9 +1,58 @@ import { createLogger } from '@sim/logger' +import { NonRetryableExecutionError } from '@/lib/execution/non-retryable-error' import type { BlockOutput } from '@/blocks/types' import { REFERENCE } from '@/executor/constants' const logger = createLogger('SharedResponseFormat') +export const DEFAULT_STRUCTURED_OUTPUT_MAX_TOKENS = 4096 + +const TOKEN_LIMIT_FINISH_REASONS = new Set(['max_tokens', 'max_output_tokens', 'length']) + +interface ProviderTimingLike { + timeSegments?: Array<{ + type?: string + finishReason?: string + }> +} + +export class StructuredOutputTokenLimitError extends NonRetryableExecutionError { + readonly code = 'structured_output_token_limit' as const + readonly diagnosticOutput?: Record + + constructor(diagnosticOutput?: Record) { + super( + 'Structured output reached the maximum output-token limit before completion. Increase Max Output Tokens or reduce the requested response size.' + ) + this.name = 'StructuredOutputTokenLimitError' + this.diagnosticOutput = diagnosticOutput + Object.defineProperty(this, 'diagnosticOutput', { enumerable: false }) + } +} + +/** + * Rejects explicitly token-limited structured generations before their partial + * content can be parsed or exposed as a successful block output. + */ +export function assertStructuredOutputNotTokenLimited( + timing?: ProviderTimingLike, + diagnosticOutput?: Record +): void { + const segments = timing?.timeSegments + if (!Array.isArray(segments)) return + + for (let index = segments.length - 1; index >= 0; index--) { + const segment = segments[index] + if (segment?.type !== 'model') continue + + const finishReason = segment.finishReason?.trim().toLowerCase() + if (finishReason && TOKEN_LIMIT_FINISH_REASONS.has(finishReason)) { + throw new StructuredOutputTokenLimitError(diagnosticOutput) + } + return + } +} + /** * Parse a raw responseFormat value (string or object) into a usable schema. * diff --git a/apps/sim/providers/anthropic/core.ts b/apps/sim/providers/anthropic/core.ts index 041af3c51f4..4b0c26e02f9 100644 --- a/apps/sim/providers/anthropic/core.ts +++ b/apps/sim/providers/anthropic/core.ts @@ -512,7 +512,7 @@ export async function executeAnthropicProviderRequest( createStream: ({ output, finalizeTiming }) => createReadableStreamFromAnthropicStream( streamResponse as AsyncIterable, - ({ content, usage, thinking }) => { + ({ content, usage, thinking, finishReason }) => { const tokens = buildAnthropicUsageTokens(usage) const cost = buildAnthropicUsageCost(request.model, usage) output.content = content @@ -531,6 +531,9 @@ export async function executeAnthropicProviderRequest( if (thinking) { segment.thinkingContent = thinking } + if (finishReason) { + segment.finishReason = finishReason + } } finalizeTiming() diff --git a/apps/sim/providers/anthropic/utils.test.ts b/apps/sim/providers/anthropic/utils.test.ts index 6c284067bdd..011e5bff4d2 100644 --- a/apps/sim/providers/anthropic/utils.test.ts +++ b/apps/sim/providers/anthropic/utils.test.ts @@ -85,6 +85,7 @@ describe('createReadableStreamFromAnthropicStream', () => { } yield { type: 'message_delta', + delta: { stop_reason: 'max_tokens', stop_sequence: null }, usage: { input_tokens: 10, output_tokens: 40, @@ -105,6 +106,7 @@ describe('createReadableStreamFromAnthropicStream', () => { cacheWriteFiveMinute: 10, cacheWriteOneHour: 20, }) + expect(onComplete.mock.calls[0][0].finishReason).toBe('max_tokens') }) it('records [redacted] for redacted_thinking blocks and streams text', async () => { diff --git a/apps/sim/providers/anthropic/utils.ts b/apps/sim/providers/anthropic/utils.ts index ad6ac2958f9..3c01edd3465 100644 --- a/apps/sim/providers/anthropic/utils.ts +++ b/apps/sim/providers/anthropic/utils.ts @@ -16,6 +16,7 @@ export interface AnthropicStreamComplete { usage: AnthropicUsageAccumulator /** Assembled thinking text for traces (redacted blocks become `[redacted]`). */ thinking: string + finishReason?: string } /** @@ -38,6 +39,7 @@ export function createReadableStreamFromAnthropicStream( const thinkingBlocks: string[] = [] let currentThinking = '' let usageSnapshot: AnthropicUsageLike = {} + let finishReason: string | undefined const flushThinkingBlock = () => { if (currentThinking) { @@ -57,6 +59,9 @@ export function createReadableStreamFromAnthropicStream( } if (event.type === 'message_delta') { + if (typeof event.delta.stop_reason === 'string') { + finishReason = event.delta.stop_reason + } usageSnapshot = { ...usageSnapshot, input_tokens: event.usage.input_tokens ?? usageSnapshot.input_tokens, @@ -114,6 +119,7 @@ export function createReadableStreamFromAnthropicStream( content: fullContent, usage, thinking: thinkingBlocks.filter(Boolean).join('\n\n'), + finishReason, }) } From 2b3c495d916002736fa49ec0b0feb7a8156d8266 Mon Sep 17 00:00:00 2001 From: Bill Leoutsakos Date: Sat, 29 Aug 2026 12:22:32 -0700 Subject: [PATCH 2/2] fix(agent): harden structured stream guardrails --- .../executor/execution/block-executor.test.ts | 74 +++++++++++++++++++ apps/sim/executor/execution/block-executor.ts | 29 +++++--- .../providers/bedrock/utils.stream.test.ts | 8 +- apps/sim/providers/bedrock/utils.ts | 7 ++ .../sim/providers/google/utils.stream.test.ts | 2 + apps/sim/providers/google/utils.ts | 10 ++- .../openai-compat/stream-events.test.ts | 8 +- .../providers/openai-compat/stream-events.ts | 3 + .../sim/providers/openai/utils.stream.test.ts | 5 +- apps/sim/providers/openai/utils.ts | 5 ++ apps/sim/providers/stream-events.test.ts | 3 +- apps/sim/providers/stream-events.ts | 7 +- apps/sim/providers/stream-pump.test.ts | 3 +- apps/sim/providers/stream-pump.ts | 8 ++ 14 files changed, 156 insertions(+), 16 deletions(-) diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index 1c424e163c9..f515b0fd01e 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/apps/sim/executor/execution/block-executor.test.ts @@ -6,6 +6,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache' import { isLargeArrayManifest } from '@/lib/execution/payloads/large-array-manifest-metadata' import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref' +import { REDACTION_FAILED_MARKER, redactObjectStrings } from '@/lib/logs/execution/pii-redaction' import { projectTraceSpansForSecrets } from '@/lib/logs/execution/trace-secret-projection' import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' import { BlockType, EDGE } from '@/executor/constants' @@ -1565,6 +1566,79 @@ describe('BlockExecutor streaming pump', () => { ) }) + it('fails empty structured streams when the terminal event reports a token limit', async () => { + const onFullContent = vi.fn() + const handler = createAgentEventsStreamingHandler({ + events: [{ type: 'turn_end', turn: 'final', finishReason: 'length' }], + onFullContent, + }) + const { executor, block, state } = createExecutor(handler) + block.config.params = { + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + } + const ctx = createContext(state) + + await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow( + /maximum output-token limit/i + ) + + expect(onFullContent).not.toHaveBeenCalled() + expect(state.getBlockOutput(block.id)).toMatchObject({ + content: '', + error: expect.stringMatching(/maximum output-token limit/i), + }) + expect(state.getBlockOutput(block.id)?.providerTiming?.timeSegments?.[0]?.finishReason).toBe( + 'length' + ) + }) + + it('preserves the original failure when provider-diagnostic redaction scrubs', async () => { + const redactor = vi.mocked(redactObjectStrings) + redactor + .mockImplementationOnce(async (value, options) => { + expect(options.onFailure).toBe('throw') + return value as never + }) + .mockImplementationOnce(async (value, options) => { + expect(options.onFailure).toBe('scrub') + return { + ...(value as Record), + content: REDACTION_FAILED_MARKER, + model: REDACTION_FAILED_MARKER, + } as never + }) + + const onFullContent = vi.fn() + const handler = createAgentEventsStreamingHandler({ + events: [{ type: 'text_delta', text: '{"answer":"unfinished', turn: 'final' }], + finishReason: 'max_tokens', + onFullContent, + }) + const { executor, block, state } = createExecutor(handler) + block.config.params = { + responseFormat: { type: 'object', properties: { answer: { type: 'string' } } }, + } + const ctx = createContext(state) + ctx.piiBlockOutputRedaction = { + enabled: true, + entityTypes: ['EMAIL_ADDRESS'], + language: 'en', + } + + await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow( + /maximum output-token limit/i + ) + + expect(redactor).toHaveBeenCalledTimes(2) + expect(onFullContent).not.toHaveBeenCalled() + expect(state.getBlockOutput(block.id)).toMatchObject({ + content: REDACTION_FAILED_MARKER, + model: REDACTION_FAILED_MARKER, + tokens: { input: 1, output: 2, total: 3 }, + error: expect.stringMatching(/maximum output-token limit/i), + }) + }) + it('soft-completes on user abort with drained answer text (no failed block)', async () => { const abortController = new AbortController() const handler = createAgentEventsStreamingHandler({ diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index 81793b86a76..949c6dabfc4 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -76,6 +76,7 @@ import { type VariableResolver, } from '@/executor/variables/resolver' import { createAgentStreamPump } from '@/providers/stream-pump' +import { enrichLastModelSegment } from '@/providers/trace-enrichment' import type { SerializedBlock } from '@/serializer/types' import { SYSTEM_SUBBLOCK_IDS } from '@/triggers/constants' @@ -637,7 +638,7 @@ export class BlockExecutor { let providerDiagnostics: Record = {} for (const key of ['content', 'model', 'tokens', 'toolCalls', 'providerTiming', 'cost']) { const value = failureDiagnosticOutput?.[key] - if (value !== undefined && (key !== 'content' || value !== '')) { + if (value !== undefined) { providerDiagnostics[key] = value } } @@ -647,7 +648,7 @@ export class BlockExecutor { entityTypes: ctx.piiBlockOutputRedaction.entityTypes, language: ctx.piiBlockOutputRedaction.language, customPatterns: ctx.piiBlockOutputRedaction.customPatterns, - onFailure: 'throw', + onFailure: 'scrub', }) } Object.assign(errorOutput, providerDiagnostics) @@ -1242,11 +1243,8 @@ export class BlockExecutor { } let fullContent = pumpResult.answerText - if (!fullContent) { - return - } - if (piiEnabled && ctx.piiBlockOutputRedaction) { + if (fullContent && piiEnabled && ctx.piiBlockOutputRedaction) { // Mask before writing to `execution.output` or `onFullContent`. fullContent = await redactObjectStrings(fullContent, { entityTypes: ctx.piiBlockOutputRedaction.entityTypes, @@ -1257,15 +1255,28 @@ export class BlockExecutor { } const executionOutput = streamingExec.execution?.output + if (pumpResult.finishReason && executionOutput?.providerTiming?.timeSegments) { + enrichLastModelSegment(executionOutput.providerTiming.timeSegments, { + finishReason: pumpResult.finishReason, + }) + } + if (executionOutput && typeof executionOutput === 'object' && parsedResponseFormat) { + // Retain even empty content for failed-block diagnostics, but reject it + // before parsing so token-limited structured data never reaches downstream. + executionOutput.content = fullContent + assertStructuredOutputNotTokenLimited(executionOutput.providerTiming, executionOutput) + } + + if (!fullContent) { + return + } + if (executionOutput && typeof executionOutput === 'object') { let parsedForFormat = false if (responseFormat) { // Retain the drained text for failed-block diagnostics, but reject it // before parsing so truncated structured data never reaches downstream. executionOutput.content = fullContent - if (parsedResponseFormat) { - assertStructuredOutputNotTokenLimited(executionOutput.providerTiming, executionOutput) - } try { const parsed = JSON.parse(fullContent.trim()) streamingExec.execution.output = { diff --git a/apps/sim/providers/bedrock/utils.stream.test.ts b/apps/sim/providers/bedrock/utils.stream.test.ts index b6b3c6b3d41..ce04800ebb3 100644 --- a/apps/sim/providers/bedrock/utils.stream.test.ts +++ b/apps/sim/providers/bedrock/utils.stream.test.ts @@ -36,12 +36,18 @@ describe('createReadableStreamFromBedrockStream', () => { yield { metadata: { usage: { inputTokens: 2, outputTokens: 3 } }, } as any + yield { + messageStop: { stopReason: 'max_tokens' }, + } as any })(), onComplete ) const events = await collectEvents(stream) - expect(events).toEqual([{ type: 'text_delta', text: 'Done', turn: 'final' }]) + expect(events).toEqual([ + { type: 'text_delta', text: 'Done', turn: 'final' }, + { type: 'turn_end', turn: 'final', finishReason: 'max_tokens' }, + ]) expect(events.some((e) => e.type === 'thinking_delta')).toBe(false) expect(events.some((e) => e.type === 'tool_call_start')).toBe(false) expect(onComplete).toHaveBeenCalledWith('Done', { inputTokens: 2, outputTokens: 3 }) diff --git a/apps/sim/providers/bedrock/utils.ts b/apps/sim/providers/bedrock/utils.ts index a8e837b1142..759fb1e9800 100644 --- a/apps/sim/providers/bedrock/utils.ts +++ b/apps/sim/providers/bedrock/utils.ts @@ -42,6 +42,7 @@ export function createReadableStreamFromBedrockStream( let fullContent = '' let inputTokens = 0 let outputTokens = 0 + let finishReason: string | undefined let cancelled = false let streamIterator: AsyncIterator | undefined @@ -55,6 +56,9 @@ export function createReadableStreamFromBedrockStream( const event = next.value const streamError = getBedrockStreamError(event) if (streamError) throw streamError + if (event.messageStop?.stopReason) { + finishReason = event.messageStop.stopReason + } if (event.contentBlockDelta?.delta?.text) { const text = event.contentBlockDelta.delta.text fullContent += text @@ -69,6 +73,9 @@ export function createReadableStreamFromBedrockStream( if (onComplete) { onComplete(fullContent, { inputTokens, outputTokens }) } + if (finishReason) { + controller.enqueue({ type: 'turn_end', turn: 'final', finishReason }) + } controller.close() } catch (err) { diff --git a/apps/sim/providers/google/utils.stream.test.ts b/apps/sim/providers/google/utils.stream.test.ts index 1b2dfaa1c79..f988946aeed 100644 --- a/apps/sim/providers/google/utils.stream.test.ts +++ b/apps/sim/providers/google/utils.stream.test.ts @@ -26,6 +26,7 @@ describe('createReadableStreamFromGeminiStream', () => { yield { candidates: [ { + finishReason: 'MAX_TOKENS', content: { parts: [ { text: 'Reasoning step. ', thought: true }, @@ -48,6 +49,7 @@ describe('createReadableStreamFromGeminiStream', () => { expect(events).toEqual([ { type: 'thinking_delta', text: 'Reasoning step. ' }, { type: 'text_delta', text: 'Final answer.', turn: 'final' }, + { type: 'turn_end', turn: 'final', finishReason: 'MAX_TOKENS' }, ]) expect(onComplete).toHaveBeenCalledWith( 'Final answer.', diff --git a/apps/sim/providers/google/utils.ts b/apps/sim/providers/google/utils.ts index 63822da75d0..b2470205913 100644 --- a/apps/sim/providers/google/utils.ts +++ b/apps/sim/providers/google/utils.ts @@ -252,6 +252,7 @@ export function createReadableStreamFromGeminiStream( cachedContentTokenCount: 0, totalTokenCount: 0, } + let finishReason: string | undefined let cancelled = false let streamIterator: AsyncIterator | undefined @@ -275,8 +276,12 @@ export function createReadableStreamFromGeminiStream( if (chunk.usageMetadata) { usage = convertUsageMetadata(chunk.usageMetadata) } + const candidate = chunk.candidates?.[0] + if (candidate?.finishReason) { + finishReason = String(candidate.finishReason) + } - const parts = chunk.candidates?.[0]?.content?.parts + const parts = candidate?.content?.parts if (Array.isArray(parts)) { for (const part of parts) { if (!part.text) continue @@ -301,6 +306,9 @@ export function createReadableStreamFromGeminiStream( if (cancelled) return onComplete?.(fullContent, usage, fullThinking || undefined) + if (finishReason) { + controller.enqueue({ type: 'turn_end', turn: 'final', finishReason }) + } controller.close() } catch (error) { if (!cancelled) { diff --git a/apps/sim/providers/openai-compat/stream-events.test.ts b/apps/sim/providers/openai-compat/stream-events.test.ts index 74fb5305d2e..279c505cf8d 100644 --- a/apps/sim/providers/openai-compat/stream-events.test.ts +++ b/apps/sim/providers/openai-compat/stream-events.test.ts @@ -57,7 +57,9 @@ describe('createOpenAICompatibleAgentEventStream', () => { { providerName: 'Groq' } ) const events = await collectEvents(stream) - expect(events.every((e) => e.type === 'text_delta')).toBe(true) + expect(events.filter((e) => e.type !== 'turn_end').every((e) => e.type === 'text_delta')).toBe( + true + ) expect(events.some((e) => e.type === 'thinking_delta')).toBe(false) }) @@ -141,6 +143,9 @@ describe('createOpenAICompatibleAgentEventStream', () => { const stream = createOpenAICompatibleAgentEventStream( (async function* () { yield* openaiCompatTextOnlyChunks as any + yield { + choices: [{ delta: {}, finish_reason: 'length' }], + } as any })(), { providerName: 'DeepSeek', onComplete } ) @@ -152,6 +157,7 @@ describe('createOpenAICompatibleAgentEventStream', () => { .join('') ).toBe('Hello world') expect(onComplete.mock.calls[0][0].content).toBe('Hello world') + expect(events).toContainEqual({ type: 'turn_end', turn: 'final', finishReason: 'length' }) }) it('surfaces documented in-band provider errors', async () => { diff --git a/apps/sim/providers/openai-compat/stream-events.ts b/apps/sim/providers/openai-compat/stream-events.ts index 9d5ef438279..acdaa963b85 100644 --- a/apps/sim/providers/openai-compat/stream-events.ts +++ b/apps/sim/providers/openai-compat/stream-events.ts @@ -270,6 +270,9 @@ export function createOpenAICompatibleAgentEventStream( ...(finishReason ? { finishReason } : {}), }) } + if (finishReason) { + controller.enqueue({ type: 'turn_end', turn, finishReason }) + } controller.close() } catch (error) { diff --git a/apps/sim/providers/openai/utils.stream.test.ts b/apps/sim/providers/openai/utils.stream.test.ts index bed841d345a..c02f865b587 100644 --- a/apps/sim/providers/openai/utils.stream.test.ts +++ b/apps/sim/providers/openai/utils.stream.test.ts @@ -126,7 +126,10 @@ describe('createReadableStreamFromResponses', () => { const events = await collectEvents(createReadableStreamFromResponses(response, onComplete)) - expect(events).toEqual([{ type: 'text_delta', text: 'Truncated answer', turn: 'final' }]) + expect(events).toEqual([ + { type: 'text_delta', text: 'Truncated answer', turn: 'final' }, + { type: 'turn_end', turn: 'final', finishReason: 'max_output_tokens' }, + ]) expect(onComplete).toHaveBeenCalledWith( 'Truncated answer', { diff --git a/apps/sim/providers/openai/utils.ts b/apps/sim/providers/openai/utils.ts index 0586bf8741f..d8ebf0212d9 100644 --- a/apps/sim/providers/openai/utils.ts +++ b/apps/sim/providers/openai/utils.ts @@ -438,6 +438,7 @@ export function createReadableStreamFromResponses( let fullThinking = '' let finalUsage: ResponsesUsageTokens | undefined let completed = false + let finishReason: string | undefined let sawFunctionCall = false try { @@ -464,6 +465,7 @@ export function createReadableStreamFromResponses( throw new Error(`OpenAI Responses stream incomplete: ${reason}`) } finalUsage = parseResponsesUsage(event.response.usage) + finishReason = reason completed = true continue } @@ -499,6 +501,9 @@ export function createReadableStreamFromResponses( } onComplete?.(fullContent, finalUsage, fullThinking || undefined) + if (finishReason) { + controller.enqueue({ type: 'turn_end', turn: 'final', finishReason }) + } controller.close() } catch (error) { if (!streamAbortController.signal.aborted) { diff --git a/apps/sim/providers/stream-events.test.ts b/apps/sim/providers/stream-events.test.ts index 76c93cd3373..127ccfe47b3 100644 --- a/apps/sim/providers/stream-events.test.ts +++ b/apps/sim/providers/stream-events.test.ts @@ -20,7 +20,7 @@ describe('stream-events contract', () => { { type: 'text_delta', text: 'bye', turn: 'final' }, { type: 'text_delta', text: 'live', turn: 'pending' }, { type: 'turn_end', turn: 'intermediate' }, - { type: 'turn_end', turn: 'final' }, + { type: 'turn_end', turn: 'final', finishReason: 'length' }, { type: 'thinking_delta', text: 'hmm' }, { type: 'tool_call_start', id: 't1', name: 'search' }, { type: 'tool_call_end', id: 't1', name: 'search', status: 'success' }, @@ -41,6 +41,7 @@ describe('stream-events contract', () => { // turn_end classifies a settled turn — 'pending' is not a valid classification. expect(isAgentStreamEvent({ type: 'turn_end', turn: 'pending' })).toBe(false) expect(isAgentStreamEvent({ type: 'turn_end' })).toBe(false) + expect(isAgentStreamEvent({ type: 'turn_end', turn: 'final', finishReason: 1 })).toBe(false) expect(isAgentStreamEvent({ type: 'tool_call_start', id: 't1' })).toBe(false) expect( isAgentStreamEvent({ type: 'tool_call_end', id: 't1', name: 'search', status: 'ok' }) diff --git a/apps/sim/providers/stream-events.ts b/apps/sim/providers/stream-events.ts index 3180ffe1fb6..8881f042f4b 100644 --- a/apps/sim/providers/stream-events.ts +++ b/apps/sim/providers/stream-events.ts @@ -40,6 +40,8 @@ export type AgentStreamEvent = */ type: 'turn_end' turn: TextDeltaTurn + /** Explicit provider termination reason for this model turn, when available. */ + finishReason?: string } | { type: 'thinking_delta'; text: string } | { type: 'tool_call_start'; id: string; name: string } @@ -81,7 +83,10 @@ export function isAgentStreamEvent(value: unknown): value is AgentStreamEvent { (value.turn === undefined || isTextDeltaClassification(value.turn)) ) case 'turn_end': - return isTextDeltaTurn(value.turn) + return ( + isTextDeltaTurn(value.turn) && + (value.finishReason === undefined || typeof value.finishReason === 'string') + ) case 'thinking_delta': return typeof value.text === 'string' case 'tool_call_start': diff --git a/apps/sim/providers/stream-pump.test.ts b/apps/sim/providers/stream-pump.test.ts index ecd7aa64b08..1ec4a4c1da9 100644 --- a/apps/sim/providers/stream-pump.test.ts +++ b/apps/sim/providers/stream-pump.test.ts @@ -70,7 +70,7 @@ describe('createAgentStreamPump', () => { { type: 'tool_call_end', id: '1', name: 'search', status: 'success' }, { type: 'text_delta', text: 'Answer ', turn: 'pending' }, { type: 'text_delta', text: 'here.', turn: 'pending' }, - { type: 'turn_end', turn: 'final' }, + { type: 'turn_end', turn: 'final', finishReason: 'length' }, ] const pump = createAgentStreamPump({ @@ -86,6 +86,7 @@ describe('createAgentStreamPump', () => { // Intermediate turn discarded; only the final turn reaches the answer. expect(result.answerText).toBe('Answer here.') + expect(result.finishReason).toBe('length') expect(text).toBe('Answer here.') // Sinks see the full live timeline including pending deltas + boundaries. expect(seen).toEqual(events) diff --git a/apps/sim/providers/stream-pump.ts b/apps/sim/providers/stream-pump.ts index 22d4fcf0548..5fd61db2a0a 100644 --- a/apps/sim/providers/stream-pump.ts +++ b/apps/sim/providers/stream-pump.ts @@ -46,6 +46,8 @@ export interface CreateAgentStreamPumpOptions { export interface AgentStreamPumpResult { /** Final-turn answer text only (`turn: 'intermediate'` excluded). */ answerText: string + /** Last explicit provider termination reason observed while draining. */ + finishReason?: string fullyDrained: boolean cancelled: boolean cancelReason?: AgentStreamPumpCancelReason @@ -109,6 +111,7 @@ export function createAgentStreamPump(options: CreateAgentStreamPumpOptions): Ag let closedTextStream = false let answerText = '' + let finishReason: string | undefined let thinkingCharsForwarded = 0 let textController: ReadableStreamDefaultController | null = null @@ -269,6 +272,9 @@ export function createAgentStreamPump(options: CreateAgentStreamPumpOptions): Ag if (event.type === 'turn_end') { await dispatchToSinks(event) + if (event.finishReason) { + finishReason = event.finishReason + } const buffered = pendingTurnText pendingTurnText = '' if (buffered && event.turn === 'final') { @@ -411,6 +417,7 @@ export function createAgentStreamPump(options: CreateAgentStreamPumpOptions): Ag closeTextStream() return { answerText, + ...(finishReason ? { finishReason } : {}), fullyDrained: false, cancelled: true, cancelReason: cancelReason ?? 'unknown', @@ -420,6 +427,7 @@ export function createAgentStreamPump(options: CreateAgentStreamPumpOptions): Ag closeTextStream() return { answerText, + ...(finishReason ? { finishReason } : {}), fullyDrained, cancelled: false, }