materialize refs before sending in response block

Author: icecrasher321

apps/sim/app/api/workflows/[id]/execute/response-block.test.ts

@@ -8,16 +8,22 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { AuthType } from '@/lib/auth/hybrid'
 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 { compactExecutionPayload } from '@/lib/execution/payloads/serializer'
+import { EXECUTION_RESOURCE_LIMIT_CODE } from '@/lib/execution/resource-errors'
 import type { ExecutionResult } from '@/lib/workflows/types'
 import { createHttpResponseFromBlock, workflowHasResponseBlock } from '@/lib/workflows/utils'
 
 const { mockUploadFile } = vi.hoisted(() => ({
   mockUploadFile: vi.fn(),
 }))
 
+const MATERIALIZATION_CONTEXT = {
+  workspaceId: 'workspace-1',
+  workflowId: 'workflow-1',
+  executionId: 'execution-1',
+  userId: 'user-1',
+}
+
 vi.mock('@/lib/uploads', () => ({
   StorageService: {
     uploadFile: mockUploadFile,
@@ -92,14 +98,14 @@ describe('Response block gating by auth type', () => {
     expect(shouldFormatAsResponseBlock).toBe(false)
   })
 
-  it('should apply Response block formatting for API key callers', () => {
+  it('should apply Response block formatting for API key callers', async () => {
     const authType = AuthType.API_KEY
     const hasResponseBlock = workflowHasResponseBlock(resultWithResponseBlock)
 
     const shouldFormatAsResponseBlock = authType !== AuthType.INTERNAL_JWT && hasResponseBlock
     expect(shouldFormatAsResponseBlock).toBe(true)
 
-    const response = createHttpResponseFromBlock(resultWithResponseBlock)
+    const response = await createHttpResponseFromBlock(resultWithResponseBlock)
     expect(response.status).toBe(200)
   })
 
@@ -112,7 +118,7 @@ describe('Response block gating by auth type', () => {
   })
 
   it('should return raw user data via createHttpResponseFromBlock', async () => {
-    const response = createHttpResponseFromBlock(resultWithResponseBlock)
+    const response = await createHttpResponseFromBlock(resultWithResponseBlock)
     const body = await response.json()
 
     // Response block returns the user-defined data directly (no success/executionId wrapper)
@@ -121,66 +127,88 @@ describe('Response block gating by auth type', () => {
     expect(body.executionId).toBeUndefined()
   })
 
-  it('should respect custom status codes from Response block', () => {
+  it('should respect custom status codes from Response block', async () => {
     const result = buildExecutionResult({
       output: { data: { error: 'Not found' }, status: 404, headers: {} },
     })
 
-    const response = createHttpResponseFromBlock(result)
+    const response = await createHttpResponseFromBlock(result)
     expect(response.status).toBe(404)
   })
 
-  it('should return manifest metadata directly for Response block data', async () => {
+  it('should materialize manifest data for Response block HTTP output', async () => {
+    const rows = Array.from({ length: 100 }, (_, index) => ({
+      key: `SIM-${index}`,
+      payload: 'x'.repeat(100),
+    }))
     const output = await compactExecutionPayload(
       {
-        data: {
-          rows: Array.from({ length: 120_000 }, (_, index) => ({
-            key: `SIM-${index}`,
-            payload: 'x'.repeat(100),
-          })),
-        },
+        data: { rows },
         status: 200,
         headers: {},
       },
       {
-        workspaceId: 'workspace-1',
-        workflowId: 'workflow-1',
-        executionId: 'execution-1',
-        userId: 'user-1',
+        ...MATERIALIZATION_CONTEXT,
         requireDurable: true,
         preserveRoot: true,
+        thresholdBytes: 1024,
       }
     )
-    const response = createHttpResponseFromBlock(buildExecutionResult({ output }))
+    const response = await createHttpResponseFromBlock(
+      buildExecutionResult({ output }),
+      MATERIALIZATION_CONTEXT
+    )
     const body = await response.json()
 
     expect(response.status).toBe(200)
-    expect(isLargeArrayManifest(body.rows)).toBe(true)
+    expect(body.rows).toEqual(rows)
     expect(body.success).toBeUndefined()
   })
 
-  it('should keep large string Response block data bounded as a generic ref', async () => {
+  it('should materialize large string refs for Response block HTTP output', async () => {
+    const text = 'x'.repeat(9 * 1024 * 1024)
     const output = await compactExecutionPayload(
       {
-        data: {
-          text: 'x'.repeat(9 * 1024 * 1024),
-        },
+        data: { text },
         status: 200,
         headers: {},
       },
       {
-        workspaceId: 'workspace-1',
-        workflowId: 'workflow-1',
-        executionId: 'execution-1',
-        userId: 'user-1',
+        ...MATERIALIZATION_CONTEXT,
         requireDurable: true,
         preserveRoot: true,
       }
     )
-    const response = createHttpResponseFromBlock(buildExecutionResult({ output }))
+    const response = await createHttpResponseFromBlock(
+      buildExecutionResult({ output }),
+      MATERIALIZATION_CONTEXT
+    )
     const body = await response.json()
 
     expect(response.status).toBe(200)
-    expect(isLargeValueRef(body.text)).toBe(true)
+    expect(body.text).toBe(text)
+  })
+
+  it('should reject Response block HTTP output that is too large to inline', async () => {
+    const output = await compactExecutionPayload(
+      {
+        data: {
+          text: 'x'.repeat(17 * 1024 * 1024),
+        },
+        status: 200,
+        headers: {},
+      },
+      {
+        ...MATERIALIZATION_CONTEXT,
+        requireDurable: true,
+        preserveRoot: true,
+      }
+    )
+
+    await expect(
+      createHttpResponseFromBlock(buildExecutionResult({ output }), MATERIALIZATION_CONTEXT)
+    ).rejects.toMatchObject({
+      code: EXECUTION_RESOURCE_LIMIT_CODE,
+    })
   })
 })

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -794,7 +794,16 @@ async function handleExecutePost(
             preserveUserFileBase64: true,
             preserveRoot: true,
           })
-          return createHttpResponseFromBlock({ ...result, output: compactResponseBlockOutput })
+          return await createHttpResponseFromBlock(
+            { ...result, output: compactResponseBlockOutput },
+            {
+              workspaceId,
+              workflowId,
+              executionId,
+              userId: actorUserId,
+              allowLargeValueWorkflowScope: Boolean(resolvedRunFromBlock?.sourceSnapshot),
+            }
+          )
         }
 
         const compactOutput = await compactRoutePayload(outputWithBase64, {

apps/sim/lib/workflows/utils.test.ts

@@ -242,14 +242,25 @@ describe('validateWorkflowPermissions', () => {
 })
 
 describe('createHttpResponseFromBlock', () => {
-  it('returns large refs as response metadata without sync materialization', async () => {
-    const response = createHttpResponseFromBlock({
+  it('rejects large refs that cannot be materialized for HTTP response output', async () => {
+    await expect(
+      createHttpResponseFromBlock({
+        output: {
+          data: { issues: largeValueRef },
+          status: 200,
+        },
+      } as any)
+    ).rejects.toThrow('This execution value is too large to inline')
+  })
+
+  it('returns raw response data when no large execution values are present', async () => {
+    const response = await createHttpResponseFromBlock({
       output: {
-        data: { issues: largeValueRef },
+        data: { issues: [] },
         status: 200,
       },
     } as any)
 
-    await expect(response.json()).resolves.toEqual({ issues: largeValueRef })
+    await expect(response.json()).resolves.toEqual({ issues: [] })
   })
 })

apps/sim/lib/workflows/utils.ts

@@ -6,6 +6,19 @@ import { authorizeWorkflowByWorkspacePermission } from '@sim/workflow-authz'
 import { and, asc, eq, inArray, isNull, max, min, sql } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
+import {
+  isLargeArrayManifest,
+  materializeLargeArrayManifest,
+} from '@/lib/execution/payloads/large-array-manifest'
+import {
+  getLargeValueMaterializationError,
+  isLargeValueRef,
+} from '@/lib/execution/payloads/large-value-ref'
+import {
+  type ExecutionMaterializationContext,
+  MAX_INLINE_MATERIALIZATION_BYTES,
+} from '@/lib/execution/payloads/materialization.server'
+import { materializeLargeValueRef } from '@/lib/execution/payloads/store'
 import { getNextWorkflowColor } from '@/lib/workflows/colors'
 import { buildDefaultWorkflowArtifacts } from '@/lib/workflows/defaults'
 import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils'
@@ -315,17 +328,65 @@ export const workflowHasResponseBlock = (
   return responseBlock !== undefined
 }
 
-export const createHttpResponseFromBlock = (
-  executionResult: Pick<ExecutionResult, 'output'>
-): NextResponse => {
+async function materializeHttpResponseValue(
+  value: unknown,
+  context: ExecutionMaterializationContext | undefined,
+  seen = new WeakSet<object>()
+): Promise<unknown> {
+  if (isLargeArrayManifest(value)) {
+    return materializeLargeArrayManifest(value, {
+      ...context,
+      maxBytes: MAX_INLINE_MATERIALIZATION_BYTES,
+    })
+  }
+
+  if (isLargeValueRef(value)) {
+    const materialized = await materializeLargeValueRef(value, {
+      ...context,
+      maxBytes: MAX_INLINE_MATERIALIZATION_BYTES,
+    })
+    if (materialized === undefined) {
+      throw getLargeValueMaterializationError(value)
+    }
+    return materializeHttpResponseValue(materialized, context, seen)
+  }
+
+  if (!value || typeof value !== 'object') {
+    return value
+  }
+
+  if (seen.has(value)) {
+    return value
+  }
+  seen.add(value)
+
+  if (Array.isArray(value)) {
+    return Promise.all(value.map((item) => materializeHttpResponseValue(item, context, seen)))
+  }
+
+  return Object.fromEntries(
+    await Promise.all(
+      Object.entries(value as Record<string, unknown>).map(async ([key, entryValue]) => [
+        key,
+        await materializeHttpResponseValue(entryValue, context, seen),
+      ])
+    )
+  )
+}
+
+export const createHttpResponseFromBlock = async (
+  executionResult: Pick<ExecutionResult, 'output'>,
+  context?: ExecutionMaterializationContext
+): Promise<NextResponse> => {
   const { data = {}, status = 200, headers = {} } = executionResult.output
+  const responseData = await materializeHttpResponseValue(data, context)
 
   const responseHeaders = new Headers({
     'Content-Type': 'application/json',
     ...headers,
   })
 
-  return NextResponse.json(data, {
+  return NextResponse.json(responseData, {
     status: status,
     headers: responseHeaders,
   })