feat(sdk): chat sessions accept concurrencyKey and trigger-time named limits

.changeset/chat-session-concurrency.md

@@ -0,0 +1,12 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Chat agents can now scope concurrency per session. Pass `concurrencyKey` (for example, your chat ID or tenant ID) and trigger-time named limits via `triggerConfig.concurrency` when starting a chat session, from `chat.createStartSessionAction`, the `AgentChat` client, or a handover. Keys are never defaulted, so a session without one shares the task's keyless pool.
+
+```ts
+const start = chat.createStartSessionAction("support-chat", {
+  triggerConfig: { concurrencyKey: user.id },
+});
+```

apps/webapp/app/services/realtime/sessionRunManager.server.ts

@@ -318,6 +318,8 @@ async function triggerSessionRun(params: {
     options: {
       ...(config.machine ? { machine: config.machine as never } : {}),
       ...(config.queue ? { queue: { name: config.queue } } : {}),
+      ...(config.concurrency ? { concurrency: config.concurrency } : {}),
+      ...(config.concurrencyKey !== undefined ? { concurrencyKey: config.concurrencyKey } : {}),
       ...(config.tags ? { tags: config.tags } : {}),
       ...(config.maxAttempts !== undefined ? { maxAttempts: config.maxAttempts } : {}),
       ...(config.maxDuration !== undefined ? { maxDuration: config.maxDuration } : {}),

apps/webapp/app/v3/webhookEngine.server.ts

@@ -7,6 +7,7 @@ import { env } from "~/env.server";
 import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
 import { logger } from "~/services/logger.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import { SessionTriggerConfig as SessionTriggerConfigSchema } from "@trigger.dev/core/v3";
 import {
   ensureRunForSession,
   type SessionTriggerConfig,
@@ -154,17 +155,6 @@ function createWebhookEngine() {
           return { success: false, errorType: "NOT_FOUND", error: "Environment not found" };
         }
 
-        const template = (triggerConfigTemplate ?? {}) as Partial<SessionTriggerConfig>;
-        const triggerConfig: SessionTriggerConfig = {
-          ...template,
-          basePayload: {
-            messages: [],
-            trigger: "preload",
-            chatId: externalId,
-            ...(template.basePayload ?? {}),
-          },
-        };
-
         // Resume an existing session; otherwise only START one when the event is a session-start
         // (startOn). Resume-only with no session yet -> ignore (no session, no run, no egress).
         const existing = await findSessionByExternalId(environment, externalId);
@@ -175,15 +165,64 @@ function createWebhookEngine() {
             skippedReason: "startOn: not a session-start event",
           };
         }
-        const { session, isCached } = existing
-          ? { session: existing, isCached: true }
-          : await findOrCreateSession({
-              environment,
-              externalId,
-              type: "chat.agent",
-              taskIdentifier,
-              triggerConfig,
-            });
+
+        let session;
+        let isCached;
+        if (existing) {
+          session = existing;
+          isCached = true;
+        } else {
+          /** The template arrives unvalidated (`z.record(z.unknown())` on the routing
+           * target), and continuations re-parse the stored row with a throwing parse —
+           * so anything this path persists must parse, or the session strands forever.
+           * A bad template fails the CREATE delivery terminally; resumes above never
+           * touch the template, so a broken template can't stop existing sessions.
+           * Known fields persist normalized (the parse output) while unknown template
+           * keys are kept as the pre-validation path stored them; a non-object
+           * `basePayload` is rejected rather than spread into index-keyed garbage. */
+          const template = (triggerConfigTemplate ?? {}) as Partial<SessionTriggerConfig>;
+          if (
+            template.basePayload !== undefined &&
+            (typeof template.basePayload !== "object" ||
+              template.basePayload === null ||
+              Array.isArray(template.basePayload))
+          ) {
+            return {
+              success: false,
+              error:
+                "Invalid triggerConfigTemplate on the webhook routing target: basePayload must be an object",
+            };
+          }
+          const assembled = {
+            ...template,
+            basePayload: {
+              messages: [],
+              trigger: "preload",
+              chatId: externalId,
+              ...(template.basePayload ?? {}),
+            },
+          };
+          const parsedTriggerConfig = SessionTriggerConfigSchema.safeParse(assembled);
+          if (!parsedTriggerConfig.success) {
+            return {
+              success: false,
+              error: `Invalid triggerConfigTemplate on the webhook routing target: ${parsedTriggerConfig.error.issues
+                .map((issue) => `${issue.path.join(".")}: ${issue.message}`)
+                .join("; ")}`,
+            };
+          }
+          const triggerConfig: SessionTriggerConfig = {
+            ...assembled,
+            ...parsedTriggerConfig.data,
+          };
+          ({ session, isCached } = await findOrCreateSession({
+            environment,
+            externalId,
+            type: "chat.agent",
+            taskIdentifier,
+            triggerConfig,
+          }));
+        }
 
         if (session.closedAt || (session.expiresAt && session.expiresAt.getTime() < Date.now())) {
           return { success: false, error: "Session is closed or expired" };

docs/ai-chat/client-protocol.mdx

@@ -206,6 +206,8 @@ Pick `"preload"` when the UI has rendered but the user hasn't typed (warms the a
 | `expiresAt` | `string` (ISO date) | Retention cap. |
 | `triggerConfig.machine` | `string` | Machine preset (`micro`, `small-1x`, …) for every run. |
 | `triggerConfig.queue` | `string` | Queue name. |
+| `triggerConfig.concurrency` | `string[]` | Up to two [named concurrency limits](/concurrency#sharing-a-limit-between-tasks) every run holds, replacing the task's declared named limits. |
+| `triggerConfig.concurrencyKey` | `string` | Scopes every run of this session to its own pool under each `perKey` bound it holds. Never defaulted — pass one (e.g. your chat or tenant ID) to isolate sessions from each other. |
 | `triggerConfig.tags` | `string[]` | Tags applied to every run (in addition to session-level `tags`). |
 | `triggerConfig.maxAttempts` | `number` | Per-run retry cap (1–10). |
 | `triggerConfig.maxDuration` | `number` | Per-run wall-clock cap, seconds. |

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

@@ -271,6 +271,20 @@ export type IdempotencyKeyOptionsSchema = z.infer<typeof IdempotencyKeyOptionsSc
 // with PrismaClientValidationError. Accept the intent and stringify here.
 const ConcurrencyKeySchema = z.union([z.string(), z.number()]).transform((value) => String(value));
 
+/**
+ * Trigger-time named concurrency limits. The charset rule matches what the queue
+ * concern enforces, so a bad name is a uniform up-front 400 on every path instead
+ * of a late validation error after the request (or earlier batch items) succeeded.
+ */
+const TriggerConcurrencyLimitsSchema = z
+  .string()
+  .regex(/^[a-zA-Z0-9_-]{1,122}$/, {
+    message:
+      "Concurrency limit names are 1-122 characters using only letters, numbers, underscores and hyphens",
+  })
+  .array()
+  .max(2);
+
 const ExternalDeploymentId = z.preprocess((value) => {
   if (typeof value !== "string") {
     return value;
@@ -336,7 +350,7 @@ export const TriggerTaskRequestBody = z
           )
           .max(3)
           .optional(),
-        concurrency: z.string().min(1).max(128).array().max(2).optional(),
+        concurrency: TriggerConcurrencyLimitsSchema.optional(),
         concurrencyKey: ConcurrencyKeySchema.optional(),
         delay: z.string().or(z.coerce.date()).optional(),
         idempotencyKey: z
@@ -452,7 +466,7 @@ export const BatchTriggerTaskItem = z.object({
         )
         .max(3)
         .optional(),
-      concurrency: z.string().min(1).max(128).array().max(2).optional(),
+      concurrency: TriggerConcurrencyLimitsSchema.optional(),
       tags: RunTags.optional(),
       test: z.boolean().optional(),
       ttl: z.string().or(z.number().nonnegative().int()).optional(),
@@ -1890,6 +1904,12 @@ export const SessionTriggerConfig = z.object({
   basePayload: z.record(z.unknown()),
   machine: MachinePresetName.optional(),
   queue: z.string().max(128).optional(),
+  /** Named concurrency limits every run holds, replacing the task's declared named limits.
+   * The charset rule is enforced here so a bad name is rejected before the session row
+   * persists, instead of surfacing from the trigger after the session already exists. */
+  concurrency: TriggerConcurrencyLimitsSchema.optional(),
+  /** Scopes every run to its own pool under each `perKey` bound it holds. Never defaulted — a session without one shares the keyless pool. */
+  concurrencyKey: ConcurrencyKeySchema.optional(),
   tags: z.array(z.string().max(128)).max(10).optional(),
   maxAttempts: z.number().int().positive().max(10).optional(),
   /** Per-run wall-clock cap (seconds). Forwarded to `TaskRunOptions.maxDuration`. */

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

@@ -119,6 +119,7 @@ import {
   type SessionSubscribeOptions,
 } from "./sessions.js";
 import { createTask } from "./shared.js";
+import { triggerConcurrencyBody } from "./concurrency-shared.js";
 import { markChatAgentRunForStreamsWarning } from "./streams.js";
 import { tracer } from "./tracer.js";
 
@@ -11800,6 +11801,12 @@ function createChatStartSessionAction<TChat extends AnyTask = AnyTask>(
       params.clientData !== undefined ? { metadata: params.clientData } : {};
     const maxAttempts = params.triggerConfig?.maxAttempts ?? options?.triggerConfig?.maxAttempts;
     const maxDuration = params.triggerConfig?.maxDuration ?? options?.triggerConfig?.maxDuration;
+    const concurrency =
+      params.triggerConfig?.concurrency !== undefined
+        ? params.triggerConfig.concurrency
+        : options?.triggerConfig?.concurrency;
+    const concurrencyKey =
+      params.triggerConfig?.concurrencyKey ?? options?.triggerConfig?.concurrencyKey;
     const idleTimeoutInSeconds =
       params.triggerConfig?.idleTimeoutInSeconds ?? options?.triggerConfig?.idleTimeoutInSeconds;
 
@@ -11818,6 +11825,8 @@ function createChatStartSessionAction<TChat extends AnyTask = AnyTask>(
       ...(options?.triggerConfig?.queue || params.triggerConfig?.queue
         ? { queue: params.triggerConfig?.queue ?? options?.triggerConfig?.queue }
         : {}),
+      ...(concurrency !== undefined ? triggerConcurrencyBody(concurrency) : {}),
+      ...(concurrencyKey !== undefined ? { concurrencyKey } : {}),
       tags,
       ...(maxAttempts !== undefined ? { maxAttempts } : {}),
       ...(maxDuration !== undefined ? { maxDuration } : {}),

packages/trigger-sdk/src/v3/chat-client.ts

@@ -29,6 +29,7 @@ import {
 } from "@trigger.dev/core/v3";
 import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js";
 import { chatRunTags, slimSubmitMessageForWire } from "./ai-shared.js";
+import { triggerConcurrencyBody } from "./concurrency-shared.js";
 import { sessions } from "./sessions.js";
 
 // ─── Type inference ────────────────────────────────────────────────
@@ -671,6 +672,12 @@ export class AgentChat<TAgent = unknown> {
       },
       ...(this.triggerConfigDefault?.machine ? { machine: this.triggerConfigDefault.machine } : {}),
       ...(this.triggerConfigDefault?.queue ? { queue: this.triggerConfigDefault.queue } : {}),
+      ...(this.triggerConfigDefault?.concurrency !== undefined
+        ? triggerConcurrencyBody(this.triggerConfigDefault.concurrency)
+        : {}),
+      ...(this.triggerConfigDefault?.concurrencyKey !== undefined
+        ? { concurrencyKey: this.triggerConfigDefault.concurrencyKey }
+        : {}),
       tags: chatRunTags(this.chatId, this.triggerConfigDefault?.tags),
       ...(this.triggerConfigDefault?.maxAttempts !== undefined
         ? { maxAttempts: this.triggerConfigDefault.maxAttempts }

packages/trigger-sdk/src/v3/chat-server.ts

@@ -50,6 +50,7 @@
  * `ai.ts` statically imports `agentSkillsRuntime` (which uses `node:`
  * builtins unfit for some serverless runtimes) and the heavy task
  * runtime. Allowed imports: `./ai-shared.js`, `./chat-client.js`,
+ * `./concurrency-shared.js` (dependency-free validation helpers),
  * `@trigger.dev/core/v3` (api client), `ai` (types + lightweight
  * helpers like `stepCountIs` / `convertToModelMessages`).
  */
@@ -72,6 +73,7 @@ import {
 import type { FinishReason, ModelMessage, Tool, UIMessage, UIMessageChunk } from "ai";
 import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js";
 import { chatRunTags } from "./ai-shared.js";
+import { triggerConcurrencyBody } from "./concurrency-shared.js";
 
 // `StreamTextResult` is defined locally rather than imported from `ai`: its
 // generic arity diverged (v6 `StreamTextResult<TOOLS, OUTPUT>`, v7
@@ -543,6 +545,12 @@ async function openHandoverSession(opts: {
     },
     ...(opts.triggerConfig?.machine ? { machine: opts.triggerConfig.machine } : {}),
     ...(opts.triggerConfig?.queue ? { queue: opts.triggerConfig.queue } : {}),
+    ...(opts.triggerConfig?.concurrency !== undefined
+      ? triggerConcurrencyBody(opts.triggerConfig.concurrency)
+      : {}),
+    ...(opts.triggerConfig?.concurrencyKey !== undefined
+      ? { concurrencyKey: opts.triggerConfig.concurrencyKey }
+      : {}),
     tags,
     ...(opts.triggerConfig?.maxAttempts !== undefined
       ? { maxAttempts: opts.triggerConfig.maxAttempts }

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

@@ -0,0 +1,36 @@
+/**
+ * Concurrency helpers with no runtime dependencies, importable from the lean
+ * browser and route-handler entrypoints (chat-client, chat-server) without
+ * pulling the task runtime's module graph into those bundles.
+ */
+
+/**
+ * Trigger-time named limits: strings only, like `queue`. They replace the task's
+ * declared named limits for this run; the server resolves names to the run's gates.
+ */
+export function triggerConcurrencyBody(concurrency: string | string[] | undefined): {
+  concurrency?: string[];
+} {
+  if (concurrency === undefined) {
+    return {};
+  }
+  const limits = Array.isArray(concurrency) ? concurrency : [concurrency];
+  if (limits.length > 2) {
+    throw new Error("The concurrency option accepts at most two named limits.");
+  }
+  if (limits.some((name) => typeof name !== "string" || name.length === 0)) {
+    throw new Error("The concurrency option takes limit names: non-empty strings.");
+  }
+  for (const name of limits) {
+    validateConcurrencyLimitName(name);
+  }
+  return { concurrency: limits };
+}
+
+export function validateConcurrencyLimitName(name: string): void {
+  if (!/^[a-zA-Z0-9_-]{1,122}$/.test(name)) {
+    throw new Error(
+      `Concurrency limit "${name}": names are 1-122 characters using only letters, numbers, underscores and hyphens.`
+    );
+  }
+}

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

@@ -150,6 +150,98 @@ describe("chat.createStartSessionAction — runtime", () => {
     expect(lastStartBody?.triggerConfig.lockToVersion).toBe("20260101.1");
   });
 
+  it("forwards concurrency and concurrencyKey from triggerConfig, with per-call precedence", async () => {
+    installStartFixture();
+
+    const start = chat.createStartSessionAction("fake-chat", {
+      triggerConfig: {
+        concurrency: ["chats"],
+        concurrencyKey: "tenant-default",
+      },
+    });
+    await start({
+      chatId: "chat-conc",
+      triggerConfig: { concurrencyKey: "tenant-42" },
+    });
+
+    expect(lastStartBody?.triggerConfig.concurrency).toEqual(["chats"]);
+    expect(lastStartBody?.triggerConfig.concurrencyKey).toBe("tenant-42");
+  });
+
+  it("per-call concurrency wins over the action default, and an empty array clears it", async () => {
+    installStartFixture();
+
+    const start = chat.createStartSessionAction("fake-chat", {
+      triggerConfig: { concurrency: ["chats"] },
+    });
+
+    await start({
+      chatId: "chat-conc-override",
+      triggerConfig: { concurrency: ["priority"] },
+    });
+    expect(lastStartBody?.triggerConfig.concurrency).toEqual(["priority"]);
+
+    await start({
+      chatId: "chat-conc-clear",
+      triggerConfig: { concurrency: [] },
+    });
+    expect(lastStartBody?.triggerConfig.concurrency).toEqual([]);
+  });
+
+  it("never defaults concurrencyKey from the chatId", async () => {
+    installStartFixture();
+
+    const start = chat.createStartSessionAction("fake-chat");
+    await start({ chatId: "chat-no-key" });
+
+    expect(lastStartBody?.triggerConfig.concurrencyKey).toBeUndefined();
+    expect(lastStartBody?.triggerConfig.concurrency).toBeUndefined();
+  });
+
+  it("rejects invalid trigger-time limit names before any network call", async () => {
+    installStartFixture();
+
+    const start = chat.createStartSessionAction("fake-chat", {
+      triggerConfig: { concurrency: ["not a valid name!"] },
+    });
+
+    await expect(start({ chatId: "chat-bad-limit" })).rejects.toThrow(
+      /letters, numbers, underscores and hyphens/
+    );
+    expect(lastStartBody).toBeUndefined();
+  });
+
+  it("rejects an empty-string concurrency instead of silently dropping it", async () => {
+    installStartFixture();
+
+    const emptyDefault = chat.createStartSessionAction("fake-chat", {
+      triggerConfig: { concurrency: "" as unknown as string[] },
+    });
+    await expect(emptyDefault({ chatId: "chat-empty-limit" })).rejects.toThrow(
+      /non-empty strings/
+    );
+
+    const emptyPerCall = chat.createStartSessionAction("fake-chat");
+    await expect(
+      emptyPerCall({
+        chatId: "chat-empty-limit-2",
+        triggerConfig: { concurrency: "" as unknown as string[] },
+      })
+    ).rejects.toThrow(/non-empty strings/);
+
+    const nullPerCall = chat.createStartSessionAction("fake-chat", {
+      triggerConfig: { concurrency: ["chats"] },
+    });
+    await expect(
+      nullPerCall({
+        chatId: "chat-null-limit",
+        triggerConfig: { concurrency: null as unknown as string[] },
+      })
+    ).rejects.toThrow(/non-empty strings/);
+
+    expect(lastStartBody).toBeUndefined();
+  });
+
   it("server-mints override tokens for additional API keys", async () => {
     const requests: Array<{ url: string; body: unknown }> = [];
     const start = chat.createStartSessionAction("fake-chat", {