fix(agent): bound structured output generation

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

@@ -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'
@@ -1224,6 +1225,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 +1281,9 @@ describe('BlockExecutor streaming pump', () => {
             if (options.attachThinkingOnDrain) {
               timeSegment.thinkingContent = options.attachThinkingOnDrain
             }
+            if (options.finishReason) {
+              timeSegment.finishReason = options.finishReason
+            }
             controller.close()
           },
         })
@@ -1534,6 +1539,106 @@ 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('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<string, unknown>),
+          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({

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,
@@ -71,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'
 
@@ -228,7 +234,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 +280,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 +409,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 +425,7 @@ export class BlockExecutor {
           inputDisplayRegistry,
           isSentinel,
           'execution',
-          streamingPartialOutput
+          failureDiagnosticOutput
         )
       } finally {
         commitBlockRegistry()
@@ -548,7 +557,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 +633,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) {
+        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: 'scrub',
+      })
     }
+    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 +1128,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({
@@ -1220,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,
@@ -1235,9 +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
         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',