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

Ekaterina Bulatova · Ekaterina Bulatova · commit ac30f578a520 · 2026-06-03T13:34:57.000+02:00
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.
diff --git a/.changeset/large-trigger-payload-offload.md b/.changeset/large-trigger-payload-offload.md
@@ -0,0 +1,8 @@
+---
+"@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.
+
+Payload uploads use the same resolved `ApiClient` as the trigger call (including `requestOptions.clientConfig`), not only the global `apiClientManager.client` — so custom `baseURL`, access token, and preview branch apply to both presign and trigger.
diff --git a/packages/core/src/v3/schemas/api-type.test.ts b/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);
+  });
+});
diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts
@@ -161,83 +161,95 @@ export type IdempotencyKeyOptionsSchema = z.infer<typeof IdempotencyKeyOptionsSc
 // column is String?, so passing a number (a common foot-gun when callers do
 // `concurrencyKey: payload.userId`) used to fail at `prisma.taskRun.create`
 // with PrismaClientValidationError. Accept the intent and stringify here.
-const ConcurrencyKeySchema = z
-  .union([z.string(), z.number()])
-  .transform((value) => String(value));
+const ConcurrencyKeySchema = z.union([z.string(), z.number()]).transform((value) => String(value));
 
-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 */
-      parentAttempt: z.string().optional(),
-      /** @deprecated engine v1 only */
-      dependentBatch: z.string().optional(),
-      /**
-       * If triggered in a batch, this is the BatchTaskRun id
-       */
-      parentBatch: z.string().optional(),
-      /**
-       * RunEngine v2
-       * If triggered inside another run, the parentRunId is the friendly ID of the parent run.
-       */
-      parentRunId: z.string().optional(),
-      /**
-       * RunEngine v2
-       * Should be `true` if `triggerAndWait` or `batchTriggerAndWait`
-       */
-      resumeParentOnCompletion: z.boolean().optional(),
-      /**
-       * Locks the version to the passed value.
-       * Automatically set when using `triggerAndWait` or `batchTriggerAndWait`
-       */
-      lockToVersion: z.string().optional(),
+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 */
+        parentAttempt: z.string().optional(),
+        /** @deprecated engine v1 only */
+        dependentBatch: z.string().optional(),
+        /**
+         * If triggered in a batch, this is the BatchTaskRun id
+         */
+        parentBatch: z.string().optional(),
+        /**
+         * RunEngine v2
+         * If triggered inside another run, the parentRunId is the friendly ID of the parent run.
+         */
+        parentRunId: z.string().optional(),
+        /**
+         * RunEngine v2
+         * Should be `true` if `triggerAndWait` or `batchTriggerAndWait`
+         */
+        resumeParentOnCompletion: z.boolean().optional(),
+        /**
+         * Locks the version to the passed value.
+         * Automatically set when using `triggerAndWait` or `batchTriggerAndWait`
+         */
+        lockToVersion: z.string().optional(),
+
+        queue: z
+          .object({
+            name: z.string(),
+            // @deprecated, this is now specified on the queue
+            concurrencyLimit: z.number().int().optional(),
+          })
+          .optional(),
+        concurrencyKey: ConcurrencyKeySchema.optional(),
+        delay: z.string().or(z.coerce.date()).optional(),
+        idempotencyKey: z
+          .string()
+          // Caps user-supplied keys before they reach the unique idempotency index
+          // on the underlying table — values past this fail at the database layer
+          // rather than returning a clean 400.
+          .max(2048, "idempotencyKey must be 2048 characters or less")
+          .optional(),
+        idempotencyKeyTTL: z.string().optional(),
+        /** The original user-provided idempotency key and scope */
+        idempotencyKeyOptions: IdempotencyKeyOptionsSchema.optional(),
+        machine: MachinePresetName.optional(),
+        maxAttempts: z.number().int().optional(),
+        maxDuration: z.number().optional(),
+        metadata: z.any(),
+        metadataType: z.string().optional(),
+        payloadType: z.string().optional(),
+        tags: RunTags.optional(),
+        test: z.boolean().optional(),
+        ttl: z.string().or(z.number().nonnegative().int()).optional(),
+        priority: z.number().optional(),
+        bulkActionId: z.string().optional(),
+        region: z.string().optional(),
+        debounce: z
+          .object({
+            key: z.string().max(512),
+            delay: z.string(),
+            mode: z.enum(["leading", "trailing"]).optional(),
+            maxDelay: z.string().optional(),
+          })
+          .optional(),
+      })
+      .optional(),
+  })
+  .superRefine((value, ctx) => {
+    if (value.options?.payloadType !== "application/store") {
+      return;
+    }
 
-      queue: z
-        .object({
-          name: z.string(),
-          // @deprecated, this is now specified on the queue
-          concurrencyLimit: z.number().int().optional(),
-        })
-        .optional(),
-      concurrencyKey: ConcurrencyKeySchema.optional(),
-      delay: z.string().or(z.coerce.date()).optional(),
-      idempotencyKey: z
-        .string()
-        // Caps user-supplied keys before they reach the unique idempotency index
-        // on the underlying table — values past this fail at the database layer
-        // rather than returning a clean 400.
-        .max(2048, "idempotencyKey must be 2048 characters or less")
-        .optional(),
-      idempotencyKeyTTL: z.string().optional(),
-      /** The original user-provided idempotency key and scope */
-      idempotencyKeyOptions: IdempotencyKeyOptionsSchema.optional(),
-      machine: MachinePresetName.optional(),
-      maxAttempts: z.number().int().optional(),
-      maxDuration: z.number().optional(),
-      metadata: z.any(),
-      metadataType: z.string().optional(),
-      payloadType: z.string().optional(),
-      tags: RunTags.optional(),
-      test: z.boolean().optional(),
-      ttl: z.string().or(z.number().nonnegative().int()).optional(),
-      priority: z.number().optional(),
-      bulkActionId: z.string().optional(),
-      region: z.string().optional(),
-      debounce: z
-        .object({
-          key: z.string().max(512),
-          delay: z.string(),
-          mode: z.enum(["leading", "trailing"]).optional(),
-          maxDelay: z.string().optional(),
-        })
-        .optional(),
-    })
-    .optional(),
-});
+    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>;
 
@@ -1658,9 +1670,7 @@ export const EndAndContinueSessionResponseBody = z.object({
    */
   swapped: z.boolean(),
 });
-export type EndAndContinueSessionResponseBody = z.infer<
-  typeof EndAndContinueSessionResponseBody
->;
+export type EndAndContinueSessionResponseBody = z.infer<typeof EndAndContinueSessionResponseBody>;
 
 export const UpdateSessionRequestBody = z.object({
   tags: z.array(z.string().max(128)).max(10).optional(),
@@ -1712,13 +1722,10 @@ export const ListSessionsQueryParams = z
     "filter[createdAt][from]": z.coerce.number().int().optional(),
     "filter[createdAt][to]": z.coerce.number().int().optional(),
   })
-  .refine(
-    (value) => !(value["page[after]"] && value["page[before]"]),
-    {
-      message: "Cannot pass both page[after] and page[before] on the same request",
-      path: ["page[before]"],
-    }
-  );
+  .refine((value) => !(value["page[after]"] && value["page[before]"]), {
+    message: "Cannot pass both page[after] and page[before] on the same request",
+    path: ["page[before]"],
+  });
 export type ListSessionsQueryParams = z.infer<typeof ListSessionsQueryParams>;
 
 /**
@@ -2111,7 +2118,9 @@ export type UpdatePromptOverrideRequestBody = z.infer<typeof UpdatePromptOverrid
 export const ReactivatePromptOverrideRequestBody = z.object({
   version: z.number().int().positive(),
 });
-export type ReactivatePromptOverrideRequestBody = z.infer<typeof ReactivatePromptOverrideRequestBody>;
+export type ReactivatePromptOverrideRequestBody = z.infer<
+  typeof ReactivatePromptOverrideRequestBody
+>;
 
 export const PromptOkResponseBody = z.object({ ok: z.boolean() });
 export type PromptOkResponseBody = z.infer<typeof PromptOkResponseBody>;
diff --git a/packages/core/src/v3/utils/ioSerialization.ts b/packages/core/src/v3/utils/ioSerialization.ts
@@ -100,22 +100,33 @@ export async function stringifyIO(value: any): Promise<IOPacket> {
   }
 }
 
+/**
+ * Offloads a packet to object storage when it exceeds the size limit.
+ *
+ * @param client - Optional API client to use for the upload presign request. When
+ * omitted, falls back to the global `apiClientManager.client`. Pass an explicit client
+ * (e.g. one built from a custom `clientConfig`) to ensure the payload is uploaded using
+ * the same configuration as the accompanying trigger API call.
+ */
 export async function conditionallyExportPacket(
   packet: IOPacket,
   pathPrefix: string,
-  tracer?: TriggerTracer
+  tracer?: TriggerTracer,
+  client?: ApiClient
 ): Promise<IOPacket> {
-  if (apiClientManager.client) {
+  const $client = client ?? apiClientManager.client;
+
+  if ($client) {
     const { needsOffloading, size } = packetRequiresOffloading(packet);
 
     if (needsOffloading) {
       if (!tracer) {
-        return await exportPacket(packet, pathPrefix);
+        return await exportPacket(packet, pathPrefix, $client);
       } else {
         const result = await tracer.startActiveSpan(
           "store.uploadOutput",
           async (span) => {
-            return await exportPacket(packet, pathPrefix);
+            return await exportPacket(packet, pathPrefix, $client);
           },
           {
             attributes: {
@@ -163,11 +174,15 @@ const ioRetryOptions = {
   randomize: true,
 } satisfies RetryOptions;
 
-async function exportPacket(packet: IOPacket, pathPrefix: string): Promise<IOPacket> {
+async function exportPacket(
+  packet: IOPacket,
+  pathPrefix: string,
+  client: ApiClient
+): Promise<IOPacket> {
   // Offload the output
   const filename = `${pathPrefix}.${getPacketExtension(packet.dataType)}`;
 
-  const presignedResponse = await apiClientManager.client!.createUploadPayloadUrl(filename);
+  const presignedResponse = await client.createUploadPayloadUrl(filename);
 
   if (!presignedResponse.storagePath) {
     throw new Error(
diff --git a/packages/core/test/ioSerialization.test.ts b/packages/core/test/ioSerialization.test.ts
diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts
