@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Override a concurrency limit's `perKey` for a single `concurrencyKey`: `concurrencyLimits.override(name, { concurrencyKey, perKey, ttl })` caps that one key under that one limit for a while (7 days by default, 30 at most), `reset(name, { concurrencyKey })` clears it, and `retrieve(name, { concurrencyKey })` shows the key's bound and when the override expires.
@@ -1464,6 +1464,7 @@ const EnvironmentSchema = z
       .default("info"),
     RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED: z.string().default("1"),
     RUN_ENGINE_QUEUE_GATES_ENABLED: z.string().default("0"),
+    RUN_ENGINE_CONCURRENCY_KEY_LIMITS_ENABLED: z.string().default("0"),
     RUN_ENGINE_QUEUE_RECONCILE_ENABLED: z.string().default("1"),
     RUN_ENGINE_QUEUE_RECONCILE_SCAN_COUNT: z.coerce.number().int().positive().default(100),
     RUN_ENGINE_QUEUE_RECONCILE_LOCK_TTL_SECONDS: z.coerce.number().int().positive().default(10),
 
@@ -1013,6 +1013,7 @@ function KeyStatsTable({
   const { data, isLoading } = useConcurrencyKeys({ ids, timeRange, queueName, search, page });
 
   const rows: ConcurrencyKeyRow[] = data?.success ? data.rows : [];
+  const loadedAt = data?.success ? data.loadedAt : 0;
   const total = data?.success ? data.total : 0;
   const perPage = data?.success ? data.perPage : 25;
   const totalPages = Math.max(1, Math.ceil(total / perPage));
@@ -1046,6 +1047,7 @@ function KeyStatsTable({
             <TableHeaderCell>Key</TableHeaderCell>
             <TableHeaderCell alignment="right">Queued now</TableHeaderCell>
             <TableHeaderCell alignment="right">Running now</TableHeaderCell>
+            <TableHeaderCell alignment="right">Override</TableHeaderCell>
             <TableHeaderCell alignment="right">Oldest wait</TableHeaderCell>
             <TableHeaderCell alignment="right">Started</TableHeaderCell>
             <TableHeaderCell alignment="right">Peak backlog</TableHeaderCell>
@@ -1054,11 +1056,11 @@ function KeyStatsTable({
         </TableHeader>
         <TableBody>
           {showLoading ? (
-            <TableBlankRow colSpan={7} className="text-text-dimmed">
+            <TableBlankRow colSpan={8} className="text-text-dimmed">
               Loading…
             </TableBlankRow>
           ) : rows.length === 0 ? (
-            <TableBlankRow colSpan={7} className="text-text-dimmed">
+            <TableBlankRow colSpan={8} className="text-text-dimmed">
               {search ? `No keys match “${search}”` : "No concurrency keys"}
             </TableBlankRow>
           ) : (
@@ -1072,6 +1074,21 @@ function KeyStatsTable({
                 <TableCell>{row.key}</TableCell>
                 <TableCell alignment="right">{row.queued.toLocaleString()}</TableCell>
                 <TableCell alignment="right">{row.running.toLocaleString()}</TableCell>
+                <TableCell alignment="right">
+                  {row.override === null ? (
+                    <span className="text-text-dimmed">–</span>
+                  ) : (
+                    <span
+                      title={
+                        row.overrideExpiresAt === null
+                          ? "Per-key override"
+                          : `Per-key override, expires in ${formatWaitMs(Math.max(0, row.overrideExpiresAt - loadedAt))}`
+                      }
+                    >
+                      {row.override.toLocaleString()}
+                    </span>
+                  )}
+                </TableCell>
                 <TableCell alignment="right">
                   {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)}
                 </TableCell>
 
@@ -19,26 +19,40 @@ const route = createActionApiRoute(
     corsStrategy: "all",
   },
   async ({ params, body, authentication }) => {
-    return concurrencyLimitsSystem.limits
-      .override(authentication.environment, params.name, body)
-      .match(
-        (limit) => json(limit),
-        (error) => {
-          switch (error.type) {
-            case "limit_not_found":
-              return json({ error: "Concurrency limit not found" }, { status: 404 });
-            case "conflict":
-              return json(
-                { error: "The limit changed concurrently; retry the request" },
-                { status: 409 }
-              );
-            case "invalid_override":
-              return json({ error: error.message }, { status: 400 });
-            default:
-              return json({ error: "Failed to override concurrency limit" }, { status: 500 });
-          }
+    const result =
+      body.concurrencyKey !== undefined
+        ? concurrencyLimitsSystem.limits.overrideKey(authentication.environment, params.name, {
+            concurrencyKey: body.concurrencyKey,
+            perKey: body.perKey,
+            ttl: body.ttl,
+          })
+        : concurrencyLimitsSystem.limits.override(authentication.environment, params.name, {
+            perKey: body.perKey,
+            total: body.total,
+          });
+    return result.match(
+      (limit) => json(limit),
+      (error) => {
+        switch (error.type) {
+          case "limit_not_found":
+            return json({ error: "Concurrency limit not found" }, { status: 404 });
+          case "conflict":
+            return json(
+              { error: "The limit changed concurrently; retry the request" },
+              { status: 409 }
+            );
+          case "invalid_override":
+            return json({ error: error.message }, { status: 400 });
+          case "feature_disabled":
+            return json(
+              { error: "Per-key concurrency overrides are not enabled on this server" },
+              { status: 400 }
+            );
+          default:
+            return json({ error: "Failed to override concurrency limit" }, { status: 500 });
         }
-      );
+      }
+    );
   }
 );
 
 
@@ -1,6 +1,8 @@
 import { json } from "@remix-run/server-runtime";
+import { ResetConcurrencyLimitRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { safeJsonParse } from "~/utils/json";
 import { concurrencyLimitsSystem } from "~/v3/services/concurrencyLimitsSystemInstance.server";
 
 const ParamsSchema = z.object({
@@ -16,8 +18,29 @@ const route = createActionApiRoute(
     },
     corsStrategy: "all",
   },
-  async ({ params, authentication }) => {
-    return concurrencyLimitsSystem.limits.reset(authentication.environment, params.name).match(
+  async ({ params, request, authentication }) => {
+    /** The body is optional: an empty POST resets the limit itself, as it always has. */
+    const rawBody = await request.text();
+    let concurrencyKey: string | undefined;
+    if (rawBody.length > 0) {
+      const parsed = ResetConcurrencyLimitRequestBody.safeParse(safeJsonParse(rawBody));
+      if (!parsed.success) {
+        return json(
+          { error: "Invalid body: expected { concurrencyKey?: string }" },
+          { status: 400 }
+        );
+      }
+      concurrencyKey = parsed.data.concurrencyKey;
+    }
+    const result =
+      concurrencyKey !== undefined
+        ? concurrencyLimitsSystem.limits.resetKey(
+            authentication.environment,
+            params.name,
+            concurrencyKey
+          )
+        : concurrencyLimitsSystem.limits.reset(authentication.environment, params.name);
+    return result.match(
       (limit) => json(limit),
       (error) => {
         switch (error.type) {
 
@@ -7,18 +7,31 @@ const ParamsSchema = z.object({
   name: z.string(),
 });
 
+const SearchParamsSchema = z.object({
+  concurrencyKey: z.string().min(1).max(128).optional(),
+});
+
 export const loader = createLoaderApiRoute(
   {
     params: ParamsSchema,
+    searchParams: SearchParamsSchema,
     findResource: async () => 1,
     authorization: {
       action: "read",
       resource: () => ({ type: "queues" }),
     },
     corsStrategy: "all",
   },
-  async ({ params, authentication }) => {
-    return concurrencyLimitsSystem.limits.retrieve(authentication.environment, params.name).match(
+  async ({ params, searchParams, authentication }) => {
+    const result =
+      searchParams.concurrencyKey !== undefined
+        ? concurrencyLimitsSystem.limits.retrieveKey(
+            authentication.environment,
+            params.name,
+            searchParams.concurrencyKey
+          )
+        : concurrencyLimitsSystem.limits.retrieve(authentication.environment, params.name);
+    return result.match(
       (limit) => json(limit),
       (error) => {
         if (error.type === "limit_not_found") {
 
@@ -38,6 +38,9 @@ export type ConcurrencyKeyRow = {
   key: string;
   queued: number;
   running: number;
+  /** A per-key override of the queue's perKey for this key, and when it lapses; null when none. */
+  override: number | null;
+  overrideExpiresAt: number | null;
   oldestWaitMs: number | null;
   started: number;
   peakBacklog: number;
@@ -163,6 +166,8 @@ export const action = async ({ request }: ActionFunctionArgs) => {
         key: r.concurrency_key,
         queued: l?.queued ?? 0,
         running: l?.running ?? 0,
+        override: l?.limit ?? null,
+        overrideExpiresAt: l?.limitExpiresAt?.getTime() ?? null,
         oldestWaitMs,
         started: r.started,
         peakBacklog: r.peak_backlog,
 
@@ -64,6 +64,7 @@ function createRunEngine() {
       defaultEnvConcurrencyBurstFactor: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR,
       totalConcurrencyEnabled: env.RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED === "1",
       gatesEnabled: env.RUN_ENGINE_QUEUE_GATES_ENABLED === "1",
+      concurrencyKeyLimitsEnabled: env.RUN_ENGINE_CONCURRENCY_KEY_LIMITS_ENABLED === "1",
       reconcile: {
         enabled: env.RUN_ENGINE_QUEUE_RECONCILE_ENABLED === "1",
         scanCount: env.RUN_ENGINE_QUEUE_RECONCILE_SCAN_COUNT,
 
@@ -64,3 +64,38 @@ export async function removeQueueConcurrencyLimits(
 ) {
   await engine.runQueue.removeQueueConcurrencyLimits(environment, queueName);
 }
+
+/** Sets a temporary per-key override of a queue's perKey for one concurrencyKey, expiring after ttlSeconds */
+export async function updateQueueConcurrencyKeyLimit(
+  environment: AuthenticatedEnvironment,
+  queueName: string,
+  concurrencyKey: string,
+  concurrency: number,
+  ttlSeconds: number
+) {
+  await engine.runQueue.updateQueueConcurrencyKeyLimit(
+    environment,
+    queueName,
+    concurrencyKey,
+    concurrency,
+    ttlSeconds
+  );
+}
+
+/** Removes a per-key override of a queue's perKey */
+export async function removeQueueConcurrencyKeyLimit(
+  environment: AuthenticatedEnvironment,
+  queueName: string,
+  concurrencyKey: string
+) {
+  await engine.runQueue.removeQueueConcurrencyKeyLimit(environment, queueName, concurrencyKey);
+}
+
+/** Reads a per-key override of a queue's perKey, with when it expires */
+export async function getQueueConcurrencyKeyLimit(
+  environment: AuthenticatedEnvironment,
+  queueName: string,
+  concurrencyKey: string
+) {
+  return engine.runQueue.getQueueConcurrencyKeyLimit(environment, queueName, concurrencyKey);
+}