fix(core,webapp): coerce numeric concurrencyKey to string

`concurrencyKey` validation accepted only `z.string().optional()` on the
single-trigger and V2/V3 batch endpoints, and the Phase-2 streaming
NDJSON endpoint accepted `z.record(z.unknown()).optional()` for the
entire `options` field. Callers passing `concurrencyKey: someNumericId`
(e.g. `payload.userId`) either failed schema validation on the first
two paths or sailed through on Phase-2 and then failed downstream at
`prisma.taskRun.create` with `Argument concurrencyKey: Expected String
or Null, provided Int`.

The schema now accepts `string | number` for `concurrencyKey` and
stringifies on the way in, across all three paths. The Phase-2 NDJSON
`options` is tightened to reuse the strict `BatchTriggerTaskItem.options`
shape so it validates identically to the V2/V3 batch endpoints.

A defensive `typeof === "number"` coercion at the `engine.trigger` call
site in `RunEngineTriggerTaskService` covers in-flight Redis-stored
batch items enqueued before the schema fix — those items are rebuilt
from a `Record<string, unknown>` shape that bypasses the new schema
and would otherwise continue failing for up to their TTL.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

.changeset/coerce-concurrency-key-to-string.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Coerce numeric `concurrencyKey` values to string at the API boundary across `tasks.trigger`, `tasks.batchTrigger`, and the Phase-2 streaming batch endpoint.

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -467,7 +467,14 @@ export class RunEngineTriggerTaskService {
                 taskVersion: lockedToBackgroundWorker?.version,
                 sdkVersion: lockedToBackgroundWorker?.sdkVersion,
                 cliVersion: lockedToBackgroundWorker?.cliVersion,
-                concurrencyKey: body.options?.concurrencyKey,
+                // Schema-level coercion now lands `body.options.concurrencyKey`
+                // as `string` on the API path, but the BatchQueue worker rebuilds
+                // body.options from Redis-stored items (Record<string, unknown>),
+                // which can still carry the pre-fix shape from in-flight batches.
+                concurrencyKey:
+                  typeof body.options?.concurrencyKey === "number"
+                    ? String(body.options.concurrencyKey)
+                    : body.options?.concurrencyKey,
                 queue: queueName,
                 lockedQueueId,
                 workerQueue,

apps/webapp/test/engine/triggerTask.test.ts

@@ -253,6 +253,76 @@ describe("RunEngineTriggerTaskService", () => {
     await engine.quit();
   });
 
+  // The BatchQueue worker rebuilds body.options from Redis-stored items
+  // (Record<string, unknown>), so the Phase-2 schema coercion doesn't apply
+  // to in-flight items enqueued before the schema fix. The defensive
+  // `typeof === "number"` coercion at the engine.trigger call site is what
+  // prevents these from failing at prisma.taskRun.create with
+  // "Argument concurrencyKey: Expected String or Null, provided Int".
+  containerTest(
+    "coerces a numeric concurrencyKey to a string at the engine.trigger boundary",
+    async ({ prisma, redisOptions }) => {
+      const engine = new RunEngine({
+        prisma,
+        worker: {
+          redis: redisOptions,
+          workers: 1,
+          tasksPerWorker: 10,
+          pollIntervalMs: 100,
+        },
+        queue: { redis: redisOptions },
+        runLock: { redis: redisOptions },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": {
+              name: "small-1x" as const,
+              cpu: 0.5,
+              memory: 0.5,
+              centsPerMs: 0.0001,
+            },
+          },
+          baseCostInCents: 0.0005,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "test-task";
+      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
+
+      const triggerTaskService = new RunEngineTriggerTaskService({
+        engine,
+        prisma,
+        payloadProcessor: new MockPayloadProcessor(),
+        queueConcern: new DefaultQueueManager(prisma, engine),
+        idempotencyKeyConcern: new IdempotencyKeyConcern(
+          prisma,
+          engine,
+          new MockTraceEventConcern()
+        ),
+        validator: new MockTriggerTaskValidator(),
+        traceEventConcern: new MockTraceEventConcern(),
+        tracer: trace.getTracer("test", "0.0.0"),
+        metadataMaximumSize: 1024 * 1024 * 1,
+      });
+
+      const result = await triggerTaskService.call({
+        taskId: taskIdentifier,
+        environment: authenticatedEnvironment,
+        // Cast through `any` to simulate the in-flight Redis batch-item shape
+        // (Record<string, unknown>) that bypasses the BatchItemNDJSON schema.
+        body: { payload: { userId: 51262 }, options: { concurrencyKey: 51262 as any } },
+      });
+
+      expect(result).toBeDefined();
+      const run = await prisma.taskRun.findUnique({ where: { id: result!.run.id } });
+      expect(run?.concurrencyKey).toBe("51262");
+
+      await engine.quit();
+    }
+  );
+
   containerTest("should handle idempotency keys correctly", async ({ prisma, redisOptions }) => {
     const engine = new RunEngine({
       prisma,

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

@@ -157,6 +157,14 @@ export const IdempotencyKeyOptionsSchema = z.object({
 
 export type IdempotencyKeyOptionsSchema = z.infer<typeof IdempotencyKeyOptionsSchema>;
 
+// Coerces user-supplied concurrencyKey values to string. The downstream Prisma
+// 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));
+
 export const TriggerTaskRequestBody = z.object({
   payload: z.any(),
   context: z.any(),
@@ -195,7 +203,7 @@ export const TriggerTaskRequestBody = z.object({
           concurrencyLimit: z.number().int().optional(),
         })
         .optional(),
-      concurrencyKey: z.string().optional(),
+      concurrencyKey: ConcurrencyKeySchema.optional(),
       delay: z.string().or(z.coerce.date()).optional(),
       idempotencyKey: z
         .string()
@@ -253,7 +261,7 @@ export const BatchTriggerTaskItem = z.object({
   context: z.any(),
   options: z
     .object({
-      concurrencyKey: z.string().optional(),
+      concurrencyKey: ConcurrencyKeySchema.optional(),
       delay: z.string().or(z.coerce.date()).optional(),
       idempotencyKey: z
         .string()
@@ -401,7 +409,12 @@ export type CreateBatchResponse = z.infer<typeof CreateBatchResponse>;
 
 /**
  * Phase 2: Individual item in the NDJSON stream
- * Each line in the NDJSON body should match this schema
+ * Each line in the NDJSON body should match this schema.
+ *
+ * `options` reuses the strict shape from BatchTriggerTaskItem so that the
+ * Phase-2 streaming path validates option fields identically to the V2/V3
+ * batch trigger endpoints — historically this used z.record(z.unknown()) and
+ * let invalid values (e.g. numeric concurrencyKey) reach Prisma.
  */
 export const BatchItemNDJSON = z.object({
   /** Zero-based index of this item (used for idempotency and ordering) */
@@ -411,7 +424,7 @@ export const BatchItemNDJSON = z.object({
   /** The payload for this task run */
   payload: z.unknown().optional(),
   /** Options for this specific item */
-  options: z.record(z.unknown()).optional(),
+  options: BatchTriggerTaskItem.shape.options,
 });
 
 export type BatchItemNDJSON = z.infer<typeof BatchItemNDJSON>;

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

@@ -0,0 +1,88 @@
+import { describe, it, expect } from "vitest";
+import { BatchItemNDJSON, BatchTriggerTaskItem, TriggerTaskRequestBody } from "./api.js";
+
+describe("concurrencyKey coercion", () => {
+  // Phase-2 NDJSON used to accept arbitrary shapes for `options`, so a numeric
+  // concurrencyKey (a common foot-gun when callers pass
+  // `concurrencyKey: payload.userId`) reached Prisma untouched and failed
+  // there with PrismaClientValidationError. The schema now coerces
+  // number → string at the API boundary across every trigger path.
+  describe("BatchItemNDJSON", () => {
+    it("coerces a numeric concurrencyKey to a string", () => {
+      const result = BatchItemNDJSON.safeParse({
+        index: 0,
+        task: "user-workflow-tick",
+        payload: { json: { userId: 51262 } },
+        options: { concurrencyKey: 51262 },
+      });
+
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.options?.concurrencyKey).toBe("51262");
+      }
+    });
+
+    it("accepts a string concurrencyKey unchanged", () => {
+      const result = BatchItemNDJSON.safeParse({
+        index: 0,
+        task: "user-workflow-tick",
+        payload: { json: { userId: 51262 } },
+        options: { concurrencyKey: "user-51262" },
+      });
+
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.options?.concurrencyKey).toBe("user-51262");
+      }
+    });
+
+    it("accepts an item with no options", () => {
+      const result = BatchItemNDJSON.safeParse({
+        index: 0,
+        task: "user-workflow-tick",
+        payload: { json: { userId: 51262 } },
+      });
+
+      expect(result.success).toBe(true);
+    });
+
+    it("rejects a non-numeric, non-string concurrencyKey", () => {
+      const result = BatchItemNDJSON.safeParse({
+        index: 0,
+        task: "user-workflow-tick",
+        options: { concurrencyKey: { nested: "object" } },
+      });
+
+      expect(result.success).toBe(false);
+    });
+  });
+
+  describe("BatchTriggerTaskItem", () => {
+    it("coerces a numeric concurrencyKey to a string", () => {
+      const result = BatchTriggerTaskItem.safeParse({
+        task: "user-workflow-tick",
+        payload: { userId: 51262 },
+        options: { concurrencyKey: 51262 },
+      });
+
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.options?.concurrencyKey).toBe("51262");
+      }
+    });
+  });
+
+  describe("TriggerTaskRequestBody", () => {
+    it("coerces a numeric concurrencyKey to a string", () => {
+      const result = TriggerTaskRequestBody.safeParse({
+        payload: { userId: 51262 },
+        options: { concurrencyKey: 51262 },
+      });
+
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.options?.concurrencyKey).toBe("51262");
+      }
+    });
+  });
+});