fix(core,sdk,webapp): allow task-scoped keys to create batches

Author: carderne

.changeset/friendly-keys-batch-tokens.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Allow task-scoped environment API keys to use batch operations for their permitted tasks.

apps/webapp/app/routes/api.v3.batches.ts

@@ -10,14 +10,19 @@ import { getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
-import { anyResource, createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import {
+  anyResource,
+  createActionApiRoute,
+  everyResource,
+} from "~/services/routeBuilders/apiBuilder.server";
 import { batchPublicAccessScopes } from "~/utils/batchItemAuthorization";
 import { canWriteParentRun } from "~/utils/parentRunAuthorization.server";
 import { clientSafeErrorMessage } from "~/utils/prismaErrors";
 import {
   handleRequestIdempotency,
   saveRequestIdempotency,
 } from "~/utils/requestIdempotency.server";
+import { scopeRequestIdempotencyKey } from "~/utils/requestIdempotencyKey";
 import { sanitizeTriggerSource } from "~/utils/triggerSource";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
@@ -40,9 +45,21 @@ const { action, loader } = createActionApiRoute(
     maxContentLength: 131_072, // 128KB is plenty for the batch metadata
     authorization: {
       action: "batchTrigger",
-      // No specific tasks exist yet. This grant only creates the batch shell;
-      // phase two authorizes every streamed task before enqueueing it.
-      resource: () => anyResource([{ type: "batch" }, { type: "tasks" }]),
+      resource: (_params, _searchParams, _headers, body) => {
+        // Newer clients declare the distinct task identifiers before creating
+        // the batch, so selected-task credentials can be authorized before the
+        // shell is created. Older clients omit them and retain the existing
+        // collection-level behavior: broad credentials pass, selected-task
+        // credentials fail closed.
+        if (!body.taskIdentifiers) {
+          return anyResource([{ type: "batch" }, { type: "tasks" }]);
+        }
+
+        return everyResource(
+          body.taskIdentifiers.map((id) => ({ type: "tasks" as const, id })),
+          [{ type: "batch" }, { type: "tasks" }]
+        );
+      },
     },
     corsStrategy: "all",
   },
@@ -96,11 +113,18 @@ const { action, loader } = createActionApiRoute(
       triggerClient,
     });
 
-    // Handle idempotency for the batch creation
+    // Keep create-batch retries isolated by environment and the task set that
+    // was authorized above. Sorting makes the scope stable when callers send
+    // the same identifiers in a different order.
+    const scopedIdempotencyKey = scopeRequestIdempotencyKey(body.idempotencyKey, [
+      authentication.environment.id,
+      ...(body.taskIdentifiers ? [...new Set(body.taskIdentifiers)].sort() : []),
+    ]);
+
     const cachedResponse = await handleRequestIdempotency<
       { friendlyId: string; runCount: number },
       CreateBatchResponse
-    >(body.idempotencyKey, {
+    >(scopedIdempotencyKey, {
       requestType: "create-batch",
       findCachedEntity: async (cachedRequestId) => {
         const batch = await engine.runStore.findBatchTaskRunById(cachedRequestId);
@@ -133,7 +157,7 @@ const { action, loader } = createActionApiRoute(
     const service = new CreateBatchService();
 
     service.onBatchTaskRunCreated.attachOnce(async (batch) => {
-      await saveRequestIdempotency(body.idempotencyKey, "create-batch", batch.id);
+      await saveRequestIdempotency(scopedIdempotencyKey, "create-batch", batch.id);
     });
 
     try {

apps/webapp/test/auth-api.e2e.full.test.ts

@@ -744,13 +744,16 @@ describe("API", () => {
     });
   });
 
-  // v3 batches use a collection-level resource { type: "tasks" } with
-  // no id — items are validated per-row when streamed. So id-specific
-  // scopes (write:tasks:foo) shouldn't grant blanket access; only
-  // type-level write:tasks (or admin/write:all) should.
-  describe("Trigger task — batch v3 (api.v3.batches) collection-level", () => {
+  // v3 batch creation accepts an optional declaration of the distinct task
+  // identifiers that will be streamed. New clients send it so selected-task
+  // credentials can authorize the batch before its shell is created. Older
+  // clients omit it and retain the collection-level, fail-closed behavior.
+  describe("Trigger task — batch v3 (api.v3.batches)", () => {
     const path = "/api/v3/batches";
-    const buildBody = () => ({ runCount: 1 });
+    const buildBody = (taskIdentifiers?: string[]) => ({
+      runCount: taskIdentifiers?.length ?? 1,
+      taskIdentifiers,
+    });
 
     it("missing auth: 401", async () => {
       const server = getTestServer();
@@ -779,6 +782,67 @@ describe("API", () => {
       expect(res.status).not.toBe(403);
     });
 
+    it("JWT with batchTrigger:tasks:taskA + declared taskA: auth passes", async () => {
+      const server = getTestServer();
+      const seed = await seedTestEnvironment(server.prisma);
+      const jwt = await generateJWT({
+        secretKey: seed.apiKey,
+        payload: {
+          pub: true,
+          sub: seed.environment.id,
+          scopes: ["batchTrigger:tasks:taskA"],
+        },
+        expirationTime: "15m",
+      });
+      const res = await server.webapp.fetch(path, {
+        method: "POST",
+        headers: { Authorization: `Bearer ${jwt}`, "Content-Type": "application/json" },
+        body: JSON.stringify(buildBody(["taskA"])),
+      });
+      expect(res.status).not.toBe(401);
+      expect(res.status).not.toBe(403);
+    });
+
+    it("JWT with batchTrigger:tasks:taskA + a mixed declaration: 403", async () => {
+      const server = getTestServer();
+      const seed = await seedTestEnvironment(server.prisma);
+      const jwt = await generateJWT({
+        secretKey: seed.apiKey,
+        payload: {
+          pub: true,
+          sub: seed.environment.id,
+          scopes: ["batchTrigger:tasks:taskA"],
+        },
+        expirationTime: "15m",
+      });
+      const res = await server.webapp.fetch(path, {
+        method: "POST",
+        headers: { Authorization: `Bearer ${jwt}`, "Content-Type": "application/json" },
+        body: JSON.stringify(buildBody(["taskA", "taskB"])),
+      });
+      expect(res.status).toBe(403);
+    });
+
+    it("JWT with batchTrigger:tasks:taskA + no declaration: 403", async () => {
+      const server = getTestServer();
+      const seed = await seedTestEnvironment(server.prisma);
+      const jwt = await generateJWT({
+        secretKey: seed.apiKey,
+        payload: {
+          pub: true,
+          sub: seed.environment.id,
+          scopes: ["batchTrigger:tasks:taskA"],
+        },
+        expirationTime: "15m",
+      });
+      const res = await server.webapp.fetch(path, {
+        method: "POST",
+        headers: { Authorization: `Bearer ${jwt}`, "Content-Type": "application/json" },
+        body: JSON.stringify(buildBody()),
+      });
+      expect(res.status).toBe(403);
+    });
+
     it("JWT with read:tasks: 403", async () => {
       const server = getTestServer();
       const seed = await seedTestEnvironment(server.prisma);

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

@@ -430,6 +430,8 @@ export type BatchTriggerTaskV3Response = z.infer<typeof BatchTriggerTaskV3Respon
 export const CreateBatchRequestBody = z.object({
   /** Expected number of items in the batch */
   runCount: z.number().int().positive(),
+  /** Distinct task identifiers expected in the item stream */
+  taskIdentifiers: z.array(z.string().min(1)).min(1).optional(),
   /** Parent run ID for batchTriggerAndWait (friendly ID) */
   parentRunId: z.string().optional(),
   /** Whether to resume parent on completion (true for batchTriggerAndWait) */

packages/core/src/v3/schemas/idempotencyKey.test.ts

@@ -97,6 +97,24 @@ describe("idempotencyKey length validation", () => {
 
       expect(result.success).toBe(true);
     });
+
+    it("accepts a non-empty task identifier declaration", () => {
+      const result = CreateBatchRequestBody.safeParse({
+        runCount: 2,
+        taskIdentifiers: ["task-a", "task-b"],
+      });
+
+      expect(result.success).toBe(true);
+    });
+
+    it("rejects an empty task identifier declaration", () => {
+      const result = CreateBatchRequestBody.safeParse({
+        runCount: 1,
+        taskIdentifiers: [],
+      });
+
+      expect(result.success).toBe(false);
+    });
   });
 
   describe("CreateWaitpointTokenRequestBody", () => {

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

@@ -1,6 +1,22 @@
 import { ApiClient } from "@trigger.dev/core/v3";
 import { describe, it, expect } from "vitest";
-import { offloadBatchItemPayloads, readableStreamToAsyncIterable } from "./shared.js";
+import {
+  offloadBatchItemPayloads,
+  readableStreamToAsyncIterable,
+  uniqueBatchTaskIdentifiers,
+} from "./shared.js";
+
+describe("uniqueBatchTaskIdentifiers", () => {
+  it("returns a stable, deduplicated declaration for batch creation", () => {
+    expect(
+      uniqueBatchTaskIdentifiers([
+        { index: 0, task: "task-b", payload: "{}" },
+        { index: 1, task: "task-a", payload: "{}" },
+        { index: 2, task: "task-b", payload: "{}" },
+      ])
+    ).toEqual(["task-a", "task-b"]);
+  });
+});
 
 describe("offloadBatchItemPayloads", () => {
   // A real client is required for conditionallyExportPacket's truthy check; small payloads

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

@@ -1644,6 +1644,10 @@ export async function batchTriggerAndWaitTasks<TTasks extends readonly AnyTask[]
  * @param requestOptions - Optional request options
  * @internal
  */
+export function uniqueBatchTaskIdentifiers(items: BatchItemNDJSON[]): string[] {
+  return Array.from(new Set(items.map((item) => item.task))).sort();
+}
+
 async function executeBatchTwoPhase(
   apiClient: ReturnType<typeof apiClientManager.clientOrThrow>,
   items: BatchItemNDJSON[],
@@ -1678,6 +1682,7 @@ async function executeBatchTwoPhase(
     batch = await apiClient.createBatch(
       {
         runCount: items.length,
+        taskIdentifiers: uniqueBatchTaskIdentifiers(items),
         parentRunId: options.parentRunId,
         resumeParentOnCompletion: options.resumeParentOnCompletion,
         idempotencyKey: options.idempotencyKey,