feat(run-engine,webapp,sdk,core,cli): queue concurrency limits, gates, overrides, metrics, and management API

Queues get richer concurrency controls: a total concurrency limit across
all concurrency keys of a queue (alongside the existing per-key limit),
named concurrency limits declared in code and shared across tasks, and
trigger-time selection between them. The dashboard gains a Concurrency
page with per-queue metrics charts and an override dialog for per-key
and total bounds, plus a management API for listing, retrieving,
overriding and resetting concurrency limits. Chat sessions can set
concurrency options when triggering. Run admission gates in the engine
enforce the new limits and are off by default behind
RUN_ENGINE_QUEUE_GATES_ENABLED.

Mono-RevId: b8b2c98fae12cf9abb09b3f1349fd828d053ab70

Author: matt-aitken

.changeset/chat-session-concurrency.md

@@ -0,0 +1,12 @@
+---
+"@trigger.dev/sdk": minor
+"@trigger.dev/core": minor
+---
+
+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 },
+});
+```

.changeset/task-concurrency.md

@@ -0,0 +1,25 @@
+---
+"@trigger.dev/sdk": minor
+"@trigger.dev/core": minor
+"@trigger.dev/react-hooks": minor
+---
+
+Control a task's concurrency with the new `concurrency` option, and share limits across tasks with named concurrency limits. An inline shape caps the task itself; `concurrencyLimit()` declares a limit any task can hold (up to two named limits per task), and a trigger call can switch a run's named limits with its own `concurrency` option.
+
+```ts
+import { concurrencyLimit, task } from "@trigger.dev/sdk";
+
+export const openaiLimit = concurrencyLimit({ name: "openai", total: 25 });
+
+export const generateSummary = task({
+  id: "generate-summary",
+  concurrency: [{ perKey: 1, total: 5 }, openaiLimit],
+  run: async (payload) => {},
+});
+```
+
+`perKey` caps each `concurrencyKey` pool and `total` caps across everything, keys or not. The queue-level `concurrencyLimit` option keeps working unchanged and is deprecated in favor of `concurrency`. Enforcement happens server-side; servers without support accept the option but do not enforce it yet.
+
+Manage limits at runtime with the new `concurrencyLimits` namespace: `list()` and `retrieve(name)` report each limit's bounds plus its live `running` and `queued` counts, `override(name, { perKey, total })` changes only the given bounds (overriding `total` to `0` pauses the limit), and `reset(name)` restores the declared values.
+
+Queue reads (`queues.list()` and `queues.retrieve()`) now report a `version` that discriminates the shape: `V1` queues keep today's fields (their own `concurrencyLimit` and its override state), while `V2` queues (tasks declared with `concurrency`) carry no queue-level concurrency, since their limits are read and overridden through `concurrencyLimits` (a task's inline limit under its derived `task/<task-id>` name). Existing reads keep compiling: a `V2` queue reports `concurrencyLimit` as null and `concurrency` as undefined.

apps/webapp/app/components/billing/OrgBanner.tsx

@@ -21,10 +21,10 @@ import { useOptionalProject, useProject } from "~/hooks/useProject";
 import { useShowSelfServe } from "~/hooks/useShowSelfServe";
 import { useCurrentPlan } from "~/routes/_app.orgs.$organizationSlug/route";
 import {
+  concurrencyPath,
   organizationProjectsPath,
   v3BillingLimitsPath,
   v3BillingPath,
-  v3QueuesPath,
 } from "~/utils/pathBuilder";
 import { ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT } from "~/utils/environmentPauseSource";
 
@@ -71,7 +71,7 @@ export function OrgBanner() {
     showSelfServe,
   });
 
-  const hideQueuesButton = location.pathname.endsWith("/queues");
+  const hideConcurrencyButton = location.pathname.endsWith("/concurrency");
   const hideBillingLimitBanner = location.pathname.endsWith("/settings/billing-limits");
 
   switch (bannerKind) {
@@ -89,7 +89,7 @@ export function OrgBanner() {
       return isArchived ? (
         <ArchivedEnvironmentBanner />
       ) : (
-        <PausedEnvironmentBanner hideButton={hideQueuesButton} />
+        <PausedEnvironmentBanner hideButton={hideConcurrencyButton} />
       );
     default:
       return null;
@@ -303,7 +303,7 @@ function PausedEnvironmentBanner({ hideButton }: { hideButton: boolean }) {
         hideButton ? undefined : (
           <LinkButton
             variant="tertiary/small"
-            to={v3QueuesPath(organization, project, environment)}
+            to={concurrencyPath(organization, project, environment)}
           >
             Manage
           </LinkButton>

apps/webapp/app/components/dashboard-agent/DashboardAgentPanel.tsx

@@ -52,7 +52,7 @@ import {
 } from "./unread-counts";
 import { AgentPanelColumn, type DashboardAgentMode, type DragHandleProps } from "./panel-layout";
 import { markerAfterActiveChat, markerAfterActivity } from "./thinking-marker";
-import { concurrencyPath } from "~/utils/pathBuilder";
+import { concurrencyLimitsPath } from "~/utils/pathBuilder";
 import { scopeMatchesPath, sessionPathFor } from "./agent-scope";
 
 function serializePageContext(pageContext: AgentPageContext): string | undefined {
@@ -149,7 +149,7 @@ export function DashboardAgentPanel({
   const entityId = agentPageEntityId(pageContext, location.pathname);
 
   const pagePaths = useMemo<Record<string, string>>(
-    () => ({ raise_env_limit: concurrencyPath(organization, project, environment) }),
+    () => ({ raise_env_limit: concurrencyLimitsPath(organization, project, environment) }),
     [organization, project, environment]
   );
 

apps/webapp/app/components/dashboard-agent/page-label.ts

@@ -24,7 +24,7 @@ const KIND_LABELS: Record<Exclude<AgentPage["kind"], "other">, string> = {
   alerts: "Alerts",
   apikeys: "API keys",
   envvars: "Environment variables",
-  concurrency: "Concurrency",
+  concurrency: "Concurrency limits",
   regions: "Regions",
   settings: "Settings",
   waitpoints: "Waitpoints",
@@ -50,6 +50,7 @@ const SECTION_LABELS: Record<string, string> = {
   "bulk-actions": "Bulk actions",
   branches: "Branches",
   concurrency: "Concurrency",
+  "concurrency-limits": "Concurrency limits",
   dashboards: "Dashboards",
   deployments: "Deployments",
   "dev-branches": "Branches",

apps/webapp/app/components/navigation/favoritePages.tsx

@@ -223,7 +223,8 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
   logs: { icon: "logs", name: "Logs" },
   errors: { icon: "errors", name: "Errors", singular: "Error" },
   query: { icon: "query", name: "Query" },
-  queues: { icon: "queues", name: "Queues", singular: "Queue" },
+  queues: { icon: "queues", name: "Concurrency" },
+  concurrency: { icon: "queues", name: "Concurrency" },
   dashboards: { icon: "dashboards", name: "Dashboards", singular: "Dashboard" },
   deployments: { icon: "deployments", name: "Deploys", singular: "Deploy" },
   "environment-variables": { icon: "environment-variables", name: "Environment variables" },
@@ -234,7 +235,7 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
   "bulk-actions": { icon: "bulk-actions", name: "Bulk actions", singular: "Bulk action" },
   apikeys: { icon: "apikeys", name: "API keys" },
   alerts: { icon: "alerts", name: "Alerts", singular: "Alert" },
-  concurrency: { icon: "concurrency", name: "Concurrency" },
+  "concurrency-limits": { icon: "concurrency", name: "Concurrency limits" },
   limits: { icon: "limits", name: "Limits" },
   schedules: { icon: "schedules", name: "Schedules", singular: "Schedule" },
   test: { icon: "test", name: "Test", singular: "Test" },

apps/webapp/app/components/navigation/sideMenuSections.tsx

@@ -23,7 +23,7 @@ import {
   type OrgForPath,
   type ProjectForPath,
   branchesPath,
-  concurrencyPath,
+  concurrencyLimitsPath,
   limitsPath,
   queryPath,
   regionsPath,
@@ -39,7 +39,7 @@ import {
   v3ProjectAlertsPath,
   v3ProjectSettingsIntegrationsPath,
   v3PromptsPath,
-  v3QueuesPath,
+  concurrencyPath,
   v3WaitpointTokensPath,
 } from "~/utils/pathBuilder";
 import { AlphaBadge, NewBadge } from "../FeatureBadges";
@@ -160,10 +160,10 @@ export function buildSideMenuSections({
               } satisfies SideMenuItemConfig,
               {
                 id: "queues",
-                name: "Queues",
+                name: "Concurrency",
                 icon: QueuesIcon,
                 activeIconColor: "text-queues",
-                to: v3QueuesPath(organization, project, environment),
+                to: concurrencyPath(organization, project, environment),
                 dataAction: "queues",
               } satisfies SideMenuItemConfig,
               {
@@ -269,10 +269,10 @@ export function buildSideMenuSections({
         ? [
             {
               id: "concurrency",
-              name: "Concurrency",
+              name: "Concurrency limits",
               icon: ConcurrencyIcon,
               activeIconColor: "text-text-bright",
-              to: concurrencyPath(organization, project, environment),
+              to: concurrencyLimitsPath(organization, project, environment),
               dataAction: "concurrency",
             } satisfies SideMenuItemConfig,
           ]

apps/webapp/app/components/primitives/charts/MetricChart.tsx

@@ -41,6 +41,8 @@ type MetricChartProps = {
    * are config values that existed all along, so carry the first value backward instead.
    */
   carryBackfill?: string[];
+  /** Column that, when positive on a bucket, exempts it from the carryBackfill overwrite. */
+  carryBackfillGuard?: string;
   /**
    * Line only. Recolour a series' stroke above a threshold with a gradient split (colour only
    * above the line). `value` sets a constant threshold; `valueFromSeries` reads a (roughly
@@ -78,13 +80,22 @@ export function MetricChart({
   timeRange,
   warningOverlay,
   carryBackfill,
+  carryBackfillGuard,
   thresholdStroke,
   onHasDataChange,
   sampleCountColumn,
 }: MetricChartProps) {
   const { points, xKind: resolvedXKind } = useMemo(
-    () => buildMetricPoints(rows, { series, xColumn, xKind, carryBackfill, sampleCountColumn }),
-    [rows, series, xColumn, xKind, carryBackfill, sampleCountColumn]
+    () =>
+      buildMetricPoints(rows, {
+        series,
+        xColumn,
+        xKind,
+        carryBackfill,
+        carryBackfillGuard,
+        sampleCountColumn,
+      }),
+    [rows, series, xColumn, xKind, carryBackfill, carryBackfillGuard, sampleCountColumn]
   );
 
   const chartConfig = useMemo(() => {

apps/webapp/app/components/primitives/charts/metricPoints.ts

@@ -58,12 +58,22 @@ export type BuildMetricPointsOptions = {
   /** Defaults to `time` when every x is a datetime or epoch number, `category` otherwise. */
   xKind?: MetricXKind;
   carryBackfill?: string[];
+  /** Column that, when positive on a bucket, exempts it from the carryBackfill overwrite:
+   * the bucket has a real sampled value, so an earlier bucket's carry must not clobber it. */
+  carryBackfillGuard?: string;
   sampleCountColumn?: string;
 };
 
 export function buildMetricPoints(
   rows: MetricChartRow[],
-  { series, xColumn = "t", xKind, carryBackfill, sampleCountColumn }: BuildMetricPointsOptions
+  {
+    series,
+    xColumn = "t",
+    xKind,
+    carryBackfill,
+    carryBackfillGuard,
+    sampleCountColumn,
+  }: BuildMetricPointsOptions
 ): { points: MetricPoint[]; xKind: MetricXKind } {
   // Rows built by seriesFromRows already carry the coordinate under the reserved key.
   const xOf = (row: MetricChartRow) => (METRIC_X_KEY in row ? row[METRIC_X_KEY] : row[xColumn]);
@@ -80,6 +90,9 @@ export function buildMetricPoints(
       // A bucket with no value for this series is a gap, not a zero.
       point[s.key] = hasSamples && value != null ? toNumber(value) : null;
     }
+    if (carryBackfillGuard) {
+      point[carryBackfillGuard] = toNumber(r[carryBackfillGuard]);
+    }
     // Set last, so a series key colliding with the reserved key can't displace the coordinate.
     point[METRIC_X_KEY] = kind === "time" ? (timeValueMs(xOf(r)) ?? NaN) : String(xOf(r) ?? "");
     return point;
@@ -99,7 +112,10 @@ export function buildMetricPoints(
       const first = points.findIndex((p) => toNumber(p[key]) > 0);
       if (first > 0) {
         const value = points[first]![key]!;
-        for (let i = 0; i < first; i++) points[i]![key] = value;
+        for (let i = 0; i < first; i++) {
+          if (carryBackfillGuard && toNumber(points[i]![carryBackfillGuard]) > 0) continue;
+          points[i]![key] = value;
+        }
       }
     }
   }