feat(sdk,core): offload large trigger payloads via object storage

Upload oversized trigger payloads before the API request and send an
application/store pointer instead of embedding large JSON in the trigger body.
Validate pointer payloads in TriggerTaskRequestBody.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 2 people

.changeset/large-trigger-payload-offload.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Offload large trigger payloads to object storage before sending the trigger API request. The SDK uploads packets at or above the existing 128KB limit and sends an `application/store` pointer instead of embedding large JSON in the request body. `TriggerTaskRequestBody` now validates that `application/store` payloads are non-empty storage paths.

packages/core/src/v3/schemas/api-type.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect } from "vitest";
-import { InitializeDeploymentRequestBody } from "./api.js";
+import { InitializeDeploymentRequestBody, TriggerTaskRequestBody } from "./api.js";
 import type { InitializeDeploymentRequestBody as InitializeDeploymentRequestBodyType } from "./api.js";
 
 describe("InitializeDeploymentRequestBody", () => {
@@ -139,3 +139,41 @@ describe("InitializeDeploymentRequestBody", () => {
     });
   });
 });
+
+describe("TriggerTaskRequestBody", () => {
+  it("accepts application/store payload as a non-empty string", () => {
+    const result = TriggerTaskRequestBody.safeParse({
+      payload: "packets/payloads/file.json",
+      context: {},
+      options: {
+        payloadType: "application/store",
+      },
+    });
+
+    expect(result.success).toBe(true);
+  });
+
+  it("rejects application/store payload when payload is not a string", () => {
+    const result = TriggerTaskRequestBody.safeParse({
+      payload: { foo: "bar" },
+      context: {},
+      options: {
+        payloadType: "application/store",
+      },
+    });
+
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects application/store payload when payload is an empty string", () => {
+    const result = TriggerTaskRequestBody.safeParse({
+      payload: "",
+      context: {},
+      options: {
+        payloadType: "application/store",
+      },
+    });
+
+    expect(result.success).toBe(false);
+  });
+});

packages/core/src/v3/schemas/api.ts

@@ -157,11 +157,12 @@ export const IdempotencyKeyOptionsSchema = z.object({
 
 export type IdempotencyKeyOptionsSchema = z.infer<typeof IdempotencyKeyOptionsSchema>;
 
-export const TriggerTaskRequestBody = z.object({
-  payload: z.any(),
-  context: z.any(),
-  options: z
-    .object({
+export const TriggerTaskRequestBody = z
+  .object({
+    payload: z.any(),
+    context: z.any(),
+    options: z
+      .object({
       /** @deprecated engine v1 only */
       dependentAttempt: z.string().optional(),
       /** @deprecated engine v1 only */
@@ -227,9 +228,22 @@ export const TriggerTaskRequestBody = z.object({
           maxDelay: z.string().optional(),
         })
         .optional(),
-    })
-    .optional(),
-});
+      })
+      .optional(),
+  })
+  .superRefine((value, ctx) => {
+    if (value.options?.payloadType !== "application/store") {
+      return;
+    }
+
+    if (typeof value.payload !== "string" || value.payload.length === 0) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: "payload must be a non-empty string when options.payloadType is application/store",
+        path: ["payload"],
+      });
+    }
+  });
 
 export type TriggerTaskRequestBody = z.infer<typeof TriggerTaskRequestBody>;
 

packages/trigger-sdk/src/v3/shared.ts

@@ -5,6 +5,7 @@ import {
   ApiError,
   apiClientManager,
   ApiRequestOptions,
+  conditionallyExportPacket,
   conditionallyImportPacket,
   convertToolParametersToSchema,
   createErrorTaskError,
@@ -2214,6 +2215,10 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
 
   const payloadPacket = await stringifyIO(parsedPayload);
+  const triggerPayloadPacket = await conditionallyExportPacket(
+    payloadPacket,
+    createTriggerPayloadPathPrefix(id)
+  );
 
   // Process idempotency key and extract options for storage
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
@@ -2224,12 +2229,12 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
   const handle = await apiClient.triggerTask(
     id,
     {
-      payload: payloadPacket.data,
+      payload: triggerPayloadPacket.data,
       options: {
         queue: options?.queue ? { name: options.queue } : undefined,
         concurrencyKey: options?.concurrencyKey,
         test: taskContext.ctx?.run.isTest,
-        payloadType: payloadPacket.dataType,
+        payloadType: triggerPayloadPacket.dataType,
         idempotencyKey: processedIdempotencyKey?.toString(),
         idempotencyKeyTTL: options?.idempotencyKeyTTL,
         idempotencyKeyOptions,
@@ -2470,6 +2475,10 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
 
   const payloadPacket = await stringifyIO(parsedPayload);
+  const triggerPayloadPacket = await conditionallyExportPacket(
+    payloadPacket,
+    createTriggerPayloadPathPrefix(id)
+  );
 
   // Process idempotency key and extract options for storage
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
@@ -2483,13 +2492,13 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
       const response = await apiClient.triggerTask(
         id,
         {
-          payload: payloadPacket.data,
+          payload: triggerPayloadPacket.data,
           options: {
             lockToVersion: taskContext.worker?.version, // Lock to current version because we're waiting for it to finish
             queue: options?.queue ? { name: options.queue } : undefined,
             concurrencyKey: options?.concurrencyKey,
             test: taskContext.ctx?.run.isTest,
-            payloadType: payloadPacket.dataType,
+            payloadType: triggerPayloadPacket.dataType,
             delay: options?.delay,
             ttl: options?.ttl,
             tags: options?.tags,
@@ -2556,6 +2565,10 @@ async function triggerAndSubscribe_internal<TIdentifier extends string, TPayload
 
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
   const payloadPacket = await stringifyIO(parsedPayload);
+  const triggerPayloadPacket = await conditionallyExportPacket(
+    payloadPacket,
+    createTriggerPayloadPathPrefix(id)
+  );
 
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
   const idempotencyKeyOptions = processedIdempotencyKey
@@ -2568,13 +2581,13 @@ async function triggerAndSubscribe_internal<TIdentifier extends string, TPayload
       const response = await apiClient.triggerTask(
         id,
         {
-          payload: payloadPacket.data,
+          payload: triggerPayloadPacket.data,
           options: {
             lockToVersion: taskContext.worker?.version,
             queue: options?.queue ? { name: options.queue } : undefined,
             concurrencyKey: options?.concurrencyKey,
             test: taskContext.ctx?.run.isTest,
-            payloadType: payloadPacket.dataType,
+            payloadType: triggerPayloadPacket.dataType,
             delay: options?.delay,
             ttl: options?.ttl,
             tags: options?.tags,
@@ -3074,3 +3087,8 @@ function registerTaskLifecycleHooks<
     });
   }
 }
+
+function createTriggerPayloadPathPrefix(taskId: string): string {
+  const safeTaskId = encodeURIComponent(taskId);
+  return `trigger/${safeTaskId}/${Date.now()}-${Math.random().toString(36).slice(2)}/payload`;
+}