fix(agent): bound structured output generation

Author: Bill Leoutsakos

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',

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<void>
+    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({

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<string, any> | undefined
+    let failureDiagnosticOutput: Record<string, any> | 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<string, any>
+    failureDiagnosticOutput?: Record<string, any>
   ): Promise<NormalizedBlockOutput> {
     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<string, unknown> = {}
+    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<string, any> | undefined)?.responseFormat ??
       (block.config as Record<string, any> | 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 = {

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'

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',

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 {

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<string, unknown>
+
+  constructor(diagnosticOutput?: Record<string, unknown>) {
+    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<string, unknown>
+): 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.
  *

apps/sim/providers/anthropic/core.ts

@@ -512,7 +512,7 @@ export async function executeAnthropicProviderRequest(
       createStream: ({ output, finalizeTiming }) =>
         createReadableStreamFromAnthropicStream(
           streamResponse as AsyncIterable<RawMessageStreamEvent>,
-          ({ 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()

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 () => {