feat(webapp): enable combined concurrency limit enforcement by default

The env var stays as a kill switch: set
RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED=0 to disable enforcement
without redeploying code. Off by default previously meant the SDK,
dashboard, and API surfaced combined limits that nothing enforced.

.changeset/brave-otters-recover.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Fixed a chat agent hanging after an interrupted turn: when a run was killed mid-answer (out of memory, crash, or eviction) and only the one message it was answering was still outstanding, the new run never replied to it. That message is now re-answered on the new run.

.changeset/calm-chat-reconnects.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Browser chats now keep the active turn open across page reloads when older completion records are replayed.

.changeset/chat-custom-agent-end-and-continue.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Add `chat.endAndContinue()` so fully hand-rolled custom chat agents can hand a conversation off to a fresh run on the latest deployed task version while preserving unconsumed Session input.

.changeset/fluffy-pans-argue.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Fix chat transport discarding the next turn after stopping generation. `skipToTurnComplete` is now reset when a new message or action is sent, so a message sent after `stopGeneration` streams normally instead of leaving the chat stuck in a streaming state.

.changeset/local-bundle-deploy.md

@@ -1,6 +0,0 @@
----
-"@trigger.dev/core": patch
-"trigger.dev": patch
----
-
-Add an experimental `--local-bundle` deploy flag that runs the install and bundling steps on your machine and uploads only the build output; the image is still built remotely. Useful when your project's install step needs tooling or credentials that only exist locally.

.changeset/queue-total-concurrency-limit.md

@@ -0,0 +1,18 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Cap a queue's total concurrency across all of its `concurrencyKey` values with the new `totalConcurrencyLimit` queue option. On a keyed queue, `concurrencyLimit` applies to each key value independently, so ten active keys with a limit of 5 can run 50 at once. `totalConcurrencyLimit` bounds the whole queue while each key still gets at most `concurrencyLimit`.
+
+```ts
+import { queue } from "@trigger.dev/sdk";
+
+export const perUserQueue = queue({
+  name: "per-user-queue",
+  concurrencyLimit: 1,
+  totalConcurrencyLimit: 10,
+});
+```
+
+Enforcement happens server-side and only applies to runs triggered with a `concurrencyKey`. Servers that have not enabled total concurrency limits accept the option but do not enforce it yet.

.changeset/quiet-chats-validate.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Custom chat agents now validate and parse client data declared with `chat.withClientData({ schema })` before passing it to agent code.

.changeset/quiet-floors-hold.md

@@ -1,7 +0,0 @@
----
-"@trigger.dev/sdk": patch
----
-
-Fixes a message sent while the agent was mid-answer being lost if the run then crashed. The cursor written at the end of each turn could point past a message that had arrived during that turn but had not been answered yet, so the next boot skipped it and no error was raised anywhere. Such a message is now held until a turn actually takes it.
-
-This also removes the in-memory buffer those messages used to sit in, on both `chat.agent` and `chat.createSession()`, so a message waiting for its turn is durable rather than only present in the worker that received it.

.changeset/quiet-tests-stay-home.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+"@trigger.dev/build": patch
+---
+
+Stop shipping compiled test files in the published packages. The `*.test.ts` sources were being emitted into `dist`, adding dead weight to every install and leaving modules that `require("vitest")` (not a dependency) inside the tarball, which tripped tooling that walks every file in a package.

.changeset/session-side-channels.md

@@ -0,0 +1,16 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Named side channels on a Session: durable, two-way realtime streams that outlive a single run and are shared across runs. Open a channel with `sessions.open(id).channel(name)` (or `chat.channel(name)` inside a `chat.agent`) to get an `.in`/`.out` pair addressed by name rather than the reserved default pair. Writing a side channel's `.in` does not wake or trigger a run, so a channel can carry out-of-band data (a stream of frames, a control signal) that many clients read while the agent produces it.
+
+```ts
+// Inside a chat.agent: stream frames on a named channel, wakes nothing
+const frames = chat.channel("screenshots");
+await frames.out.append(frame);
+frames.in.on((control) => { /* client control, no suspend */ });
+```
+
+Declare channel record types once with `sessions.defineChannel(...)` and infer them on both the producer and the consumer, including `useSessionStreamChannel` in React. Channels get a default retention that keeps them bounded, overridable per channel.

.changeset/session-tags-limit-ten.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Session `triggerConfig.tags` now accepts up to 10 tags, matching the run tag limit. Previously it was capped at 5, which for `chat.agent` left room for only 4 of your own tags after the automatic `chat:{chatId}` tag.

.changeset/spry-steers-defer.md

@@ -1,28 +0,0 @@
----
-"@trigger.dev/sdk": patch
-"@trigger.dev/core": patch
----
-
-A message that arrives mid-turn and is not injected into that turn is now answered as the next turn, instead of being dropped. This is what the `pendingMessages` docs have always described, and it applies to the default too: configuring `pendingMessages` without a `shouldInject` declines every batch, which previously meant every mid-turn message was lost with no error at either end.
-
-```ts
-chat.agent({
-  id: "my-chat",
-  pendingMessages: {
-    onReceived: ({ message }) => logger.info("arrived mid-turn", { id: message.id }),
-    // Only interrupt once the agent has started calling tools.
-    shouldInject: ({ steps }) => steps.length > 0,
-  },
-  run: async ({ messages, signal }) =>
-    streamText({
-      model,
-      messages,
-      abortSignal: signal,
-      // Required for injection. Without it nothing injects, and every
-      // mid-turn message is answered as the next turn instead.
-      ...chat.toStreamTextOptions(),
-    }),
-});
-```
-
-A declined message keeps its place in the queue, so it survives a crash and is answered by whichever run picks the conversation up. An injected one is consumed at the moment it is injected, so it is never also answered as a later turn.

.changeset/tidy-mailboxes-wait.md

@@ -1,25 +0,0 @@
----
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
----
-
-Fixes a case where a chat could silently lose a message. If a message arrived while the agent was between turns and a stop arrived after it, the cursor the next boot resumed from could point past that message, so it was never answered and no error was raised. This affected `chat.agent`, not just custom agents.
-
-Fixes a recovered answer being cut off. After a crash the agent replays the message it had not answered yet, but it was replaying the stop that arrived after that message too, so the turn answering it was aborted the moment it began. A stop is now only applied to the turn that was live when it arrived. That holds however the stop got there: sent after the last completed turn, or sent to a chat whose most recent turn was completed by an older version of the SDK.
-
-One limitation to know about: the recovered answer is persisted correctly, but a chat page that stayed open across the crash keeps showing the partial answer it had already received. Reload the page to see the full recovered answer.
-
-Also fixes a retried send being answered twice. When a send was retried and its idempotency claim was lost, the agent could consume the same message a second time.
-
-Custom agent loops can now inspect pending chat input without consuming it, and consume one record at a time, with `chat.messages.hasPending()` and `chat.messages.next()`. Records carry stable identifiers so a redelivery is recognisable.
-
-```ts
-if (await chat.messages.hasPending()) {
-  const record = await chat.messages.next({ timeoutInSeconds: 0 });
-  if (record) handle(record.payload);
-}
-```
-
-`hasPending()` answers for messages alone, so a message sitting behind a stop, or behind a record this version of the SDK does not recognise, still reports as pending and is still delivered. Anything the agent has no consumer for is discarded rather than left where it would make every message queued behind it undeliverable. `chat.messages.next()` returning `undefined` means no message became consumable before the timeout.
-
-`chat.writeTurnComplete()`'s `sessionInEventId` is the cursor that is safe to resume from, not the sequence of the record the turn answered. It is held back behind any message still waiting to be handled, so a value below the record you just handled is expected.

.changeset/violet-buses-tease.md

@@ -1,5 +0,0 @@
----
-"trigger.dev": patch
----
-
-Send the CLI version header on all API requests so deployments are attributable to a CLI version

.github/workflows/publish.yml

@@ -33,7 +33,18 @@ on:
       - "packages/**"
       - "!packages/**/*.md"
       - "!packages/**/*.eslintrc"
+      # CLI + libraries published to npm; none are built into the webapp/supervisor images.
+      - "!packages/cli-v3/**"
+      - "!packages/build/**"
+      - "!packages/python/**"
+      - "!packages/react-hooks/**"
+      - "!packages/rsc/**"
+      - "!packages/schema-to-json/**"
       - "internal-packages/**"
+      # Test/tooling-only internal packages, never in an image.
+      - "!internal-packages/testcontainers/**"
+      - "!internal-packages/sdk-compat-tests/**"
+      - "!internal-packages/observability-map/**"
       - "apps/**"
       - "!apps/**/*.md"
       - "!apps/**/*.eslintrc"

.server-changes/admin-dashboard-enabled-flag.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: feature
----
-
-Self-hosted instances can now disable the admin dashboard and user impersonation entirely. See the self-hosting docs for the new setting.

.server-changes/appearance-themes-and-options.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: feature
----
-
-The dashboard has two new themes, Black and White, plus appearance options for stronger colors and underlined links.

.server-changes/deployment-logs-follow-scroll.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: feature
----
-
-Deployment logs no longer jump to the bottom while you are reading earlier output. Scroll up to pause auto-scroll, and scroll back down or use the new scroll-to-bottom button in the log header to resume following.

.server-changes/env-var-inputs-autocomplete-off.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-Stop the browser offering to autofill or save environment variable values as saved credentials.

.server-changes/reduce-webapp-cpu-on-worker-routes.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-Cut webapp CPU usage by about a quarter on the routes that workers call most, freeing headroom at the same request rate. Detailed event-loop blocking traces are no longer recorded by default, because producing them was itself a large part of that cost.

.server-changes/runs-list-column-customization.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: feature
----
-
-Customize the runs list: show, hide, and reorder columns, and add smart columns that pull a value straight out of a run's payload, metadata, or output. Your column choices are saved in the page URL, so you can share a view, bookmark it, or save it straight to your favorites.

.server-changes/runs-list-query-limit-error.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-When a runs list or runs.list API request spans too much data to complete, it now returns a clear, actionable error asking you to narrow the time range, instead of failing with a generic error.

.server-changes/runs-list-read-isolation.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-Improved the performance and reliability of the runs list and the runs.list API, especially for large projects and filtered views.

.server-changes/vercel-staging-gating.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-The Staging branch setting now shows an upgrade prompt on plans that don't include a Staging environment, instead of looking editable and then silently doing nothing when saved.

.server-changes/vercel-version-skew-protection.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-New Vercel connections now get version skew protection turned on automatically, so each run uses the task version its deployment shipped with. Automatic atomic deployments are deprecated and no longer offered when you connect a project, but stay available in your Vercel integration settings.

SECURITY.md

@@ -9,7 +9,7 @@ We take the security of Trigger.dev seriously — for both our Cloud service and
 Use one of these private channels instead:
 
 1. **GitHub (preferred):** Open a private report from the repository's **Security** tab — click **"Report a vulnerability"** ([direct link](https://github.com/triggerdotdev/trigger.dev/security/advisories/new)).
-2. **Email:** `security-advisories@trigger.dev`
+2. **Email:** `security@trigger.dev`
 
 Please include as much of the following as you can:
 

apps/webapp/app/components/BlankStatePanels.tsx

@@ -5,11 +5,11 @@ import {
   ChatBubbleLeftRightIcon,
   PlusIcon,
   QuestionMarkCircleIcon,
-  SparklesIcon,
   Squares2X2Icon,
 } from "@heroicons/react/20/solid";
 import { AIChatIcon } from "~/assets/icons/AIChatIcon";
 import { AIPenIcon } from "~/assets/icons/AIPenIcon";
+import { AISparkleIcon } from "~/assets/icons/AISparkleIcon";
 import { BranchEnvironmentIconSmall } from "~/assets/icons/EnvironmentIcons";
 import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
 import openBulkActionsPanel from "~/assets/images/open-bulk-actions-panel.png";
@@ -132,8 +132,8 @@ export function HasNoTasksDev({ initializedAt }: { initializedAt: Date | string
         {!initialized && (
           <>
             <div className="flex flex-col gap-4 rounded-md border border-indigo-400/20 bg-indigo-800/10 p-4 sm:flex-row sm:items-center">
-              <span className="flex size-9 shrink-0 items-center justify-center rounded-md bg-indigo-500/15 text-indigo-400">
-                <SparklesIcon className="size-5" />
+              <span className="flex size-9 shrink-0 items-center justify-center self-start rounded-md bg-indigo-500/15 text-indigo-400">
+                <AISparkleIcon className="size-5" />
               </span>
               <div className="min-w-0 flex-1">
                 <Paragraph className="text-text-bright">Set it up with your AI agent</Paragraph>
@@ -142,7 +142,7 @@ export function HasNoTasksDev({ initializedAt }: { initializedAt: Date | string
                   includes your project reference.
                 </Paragraph>
               </div>
-              <div className="shrink-0">
+              <div className="shrink-0 sm:ml-4">
                 <InitAgentPromptV3 />
               </div>
             </div>
@@ -181,7 +181,6 @@ export function HasNoTasksDev({ initializedAt }: { initializedAt: Date | string
           stepNumber="2"
           title={devConnected ? "Dev server connected" : "Start the dev server"}
           complete={devConnected}
-          displaySpinner={!devConnected}
         />
         <StepContentContainer>
           {devConnected ? (

apps/webapp/app/components/SetupCommands.tsx

@@ -1,4 +1,4 @@
-import { CheckIcon, SparklesIcon } from "@heroicons/react/20/solid";
+import { CheckIcon } from "@heroicons/react/20/solid";
 import { createContext, useContext, useMemo, useRef, useState } from "react";
 import { useAppOrigin } from "~/hooks/useAppOrigin";
 import { useProject } from "~/hooks/useProject";
@@ -180,19 +180,25 @@ export function InitAgentPromptV3() {
     void navigator.clipboard.writeText(prompt).catch(() => {});
   };
 
+  // The idle label is the longest, so reserve its width to stop the button from
+  // resizing when it briefly swaps to the shorter "Copied prompt".
+  const idleLabel = "Copy AI agent prompt";
+
   return (
     <SimpleTooltip
       asChild
       tabbable
       button={
-        <Button
-          type="button"
-          variant="primary/medium"
-          LeadingIcon={copied ? CheckIcon : SparklesIcon}
-          leadingIconClassName={copied ? "text-success" : undefined}
-          onClick={onCopy}
-        >
-          {copied ? "Copied prompt" : "Copy AI agent prompt"}
+        <Button type="button" variant="primary/medium" onClick={onCopy}>
+          <span className="grid justify-items-center">
+            <span className="col-start-1 row-start-1 flex items-center gap-x-1.5">
+              {copied && <CheckIcon className="size-4 shrink-0 text-text-bright" />}
+              <span>{copied ? "Copied prompt" : idleLabel}</span>
+            </span>
+            <span aria-hidden className="invisible col-start-1 row-start-1">
+              {idleLabel}
+            </span>
+          </span>
         </Button>
       }
       content="Copies a setup prompt to paste into Claude Code, Cursor, or any coding agent"

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

@@ -2,7 +2,6 @@ import { cn } from "~/utils/cn";
 import { formatCurrency } from "~/utils/numberFormatter";
 import { Paragraph } from "../primitives/Paragraph";
 import { SimpleTooltip } from "../primitives/Tooltip";
-import { motion } from "framer-motion";
 
 type UsageBarProps = {
   current: number;
@@ -11,8 +10,6 @@ type UsageBarProps = {
   isPaying: boolean;
 };
 
-const startFactor = 4;
-
 export function UsageBar({ current, billingLimit, tierLimit, isPaying }: UsageBarProps) {
   const getLargestNumber = Math.max(current, tierLimit ?? -Infinity, billingLimit ?? -Infinity, 5);
   //creates a maximum range for the progress bar, add 10% to the largest number so the bar doesn't reach the end
@@ -26,13 +23,10 @@ export function UsageBar({ current, billingLimit, tierLimit, isPaying }: UsageBa
   const usageCappedToLimitPercentage = Math.min(usagePercentage, tierRunLimitPercentage);
 
   return (
-    <div className="h-fit w-full py-6">
+    <div className={cn("h-fit w-full pt-6", billingLimit !== undefined ? "pb-12" : "pb-6")}>
       <div className="relative h-3 w-full rounded-sm bg-background-bright">
         {billingLimit !== undefined && (
-          <motion.div
-            initial={{ width: billingLimitPercentage / startFactor + "%" }}
-            animate={{ width: billingLimitPercentage + "%" }}
-            transition={{ duration: 1.5, type: "spring" }}
+          <div
             style={{ width: `${billingLimitPercentage}%` }}
             className="absolute h-3 rounded-l-sm"
           >
@@ -42,12 +36,9 @@ export function UsageBar({ current, billingLimit, tierLimit, isPaying }: UsageBa
               position="bottomRow2"
               percentage={billingLimitPercentage}
             />
-          </motion.div>
+          </div>
         )}
-        <motion.div
-          initial={{ width: usagePercentage / startFactor + "%" }}
-          animate={{ width: usagePercentage + "%" }}
-          transition={{ duration: 1.5, type: "spring" }}
+        <div
           style={{ width: `${usagePercentage}%` }}
           className={cn(
             "absolute h-3 rounded-l-sm",
@@ -60,12 +51,9 @@ export function UsageBar({ current, billingLimit, tierLimit, isPaying }: UsageBa
             position="topRow1"
             percentage={usagePercentage}
           />
-        </motion.div>
+        </div>
         {tierLimit !== undefined && (
-          <motion.div
-            initial={{ width: tierRunLimitPercentage / startFactor + "%" }}
-            animate={{ width: tierRunLimitPercentage + "%" }}
-            transition={{ duration: 1.5, type: "spring" }}
+          <div
             style={{ width: `${tierRunLimitPercentage}%` }}
             className="absolute h-3 rounded-l-sm bg-green-900/20"
           >
@@ -75,12 +63,9 @@ export function UsageBar({ current, billingLimit, tierLimit, isPaying }: UsageBa
               position="bottomRow1"
               percentage={tierRunLimitPercentage}
             />
-          </motion.div>
+          </div>
         )}
-        <motion.div
-          initial={{ width: usageCappedToLimitPercentage / startFactor + "%" }}
-          animate={{ width: usageCappedToLimitPercentage + "%" }}
-          transition={{ duration: 1.5, type: "spring" }}
+        <div
           style={{ width: `${usageCappedToLimitPercentage}%` }}
           className="absolute h-3 rounded-l-sm bg-green-600"
         />

apps/webapp/app/components/layout/MetricsLayout.tsx

@@ -9,6 +9,7 @@
  *   - `Grid` — tiles; columns derived from tile count unless `columns` is set. `kind="charts"`
  *     bakes the fixed chart-row height.
  *   - `Content` — table / tabs below the tiles. Full-bleed by default; `inset` for a padded column.
+ *     `toolbar` adds a bar flush above the content for controls that scope only this region.
  *
  * Optional:
  *   - `Sidebar` — a persistent right-hand panel; fixed `width` or `resizable`. Present ⇒ Root
@@ -22,12 +23,14 @@
  * ```tsx
  * <MetricsLayout.Root>
  *   <MetricsLayout.Filters>
- *     <div className="flex items-center gap-2">…search + TimeFilter…</div>
- *     <PaginationControls … />
+ *     <div className="flex items-center gap-2">…TimeFilter…</div>
+ *     <div className="flex items-center gap-2">…page-wide actions…</div>
  *   </MetricsLayout.Filters>
  *   <MetricsLayout.Grid>…stat tiles…</MetricsLayout.Grid>
  *   <MetricsLayout.Grid kind="charts">…chart tiles…</MetricsLayout.Grid>
- *   <MetricsLayout.Content>…table…</MetricsLayout.Content>
+ *   <MetricsLayout.Content toolbar={<>…search…<PaginationControls … /></>}>
+ *     …table…
+ *   </MetricsLayout.Content>
  * </MetricsLayout.Root>
  * ```
  */
@@ -341,16 +344,44 @@ function MetricsLayoutGrid({
  * spans edge to edge with its own top border; pass `inset` for a padded column (the detail page's
  * tabs + charts). Separation from the tiles above comes from the scroll column's gap alone (no
  * extra top margin), so the tile → content step matches the gap between tile rows.
+ *
+ * Pass `toolbar` for controls that scope this region only (search, pagination) — they sit flush on
+ * top of the content instead of in the page-wide `Filters` bar, so their scope is visible.
  */
 function MetricsLayoutContent({
   children,
   inset = false,
+  toolbar,
 }: {
   children: ReactNode;
   /** Pad the content into a column (page gutter) instead of letting it span edge to edge. */
   inset?: boolean;
+  /**
+   * Controls that act on this region alone. Rendered as a bar directly above the content with no
+   * gap, so it reads as belonging to the table below rather than to the tiles above. Compose
+   * left/right clusters as child divs — `justify-between` spreads them.
+   */
+  toolbar?: ReactNode;
 }) {
-  return <div className={cn("flex flex-col gap-2.5", inset && "px-2.5")}>{children}</div>;
+  const content = <div className={cn("flex flex-col gap-2.5", inset && "px-2.5")}>{children}</div>;
+
+  if (!toolbar) {
+    return content;
+  }
+
+  return (
+    <div className="flex flex-col">
+      <div
+        className={cn(
+          "flex items-center justify-between gap-2 border-t border-grid-dimmed px-1.5 py-1.5",
+          inset && "mx-2.5"
+        )}
+      >
+        {toolbar}
+      </div>
+      {content}
+    </div>
+  );
 }
 
 export const MetricsLayout = {

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

@@ -141,7 +141,8 @@ export function OrganizationSettingsSideMenu({
             badge={
               hasProjectRuntimeUpdate ? (
                 <>
-                  <span aria-hidden className="size-2 shrink-0 rounded-full bg-warning" />
+                  {/* mr-1 lifts the right gap to 12px so it matches the dot's 12px top/bottom inset in the h-8 row */}
+                  <span aria-hidden className="mr-1 size-2 shrink-0 rounded-full bg-warning" />
                   <span className="sr-only">Runtime update available.</span>
                 </>
               ) : undefined

apps/webapp/app/components/runs/v3/RunIcon.tsx

@@ -36,6 +36,7 @@ import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
 import { TraceIcon } from "~/assets/icons/TraceIcon";
 import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
 import { StreamsIcon } from "~/assets/icons/StreamsIcon";
+import { AIChatIcon } from "~/assets/icons/AIChatIcon";
 
 type TaskIconProps = {
   name: string | undefined;
@@ -169,6 +170,8 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
           className={cn(className, "text-text-dimmed group-hover/spannode:text-text-bright")}
         />
       );
+    case "sessions":
+      return <AIChatIcon className={cn(className, "text-sessions")} />;
     case "hero-sparkles":
       return (
         <SparklesIcon

apps/webapp/app/db.server.ts

@@ -28,6 +28,7 @@ import { singleton } from "./utils/singleton";
 import { registerDatabaseMetricsSource } from "./utils/databaseMetrics.server";
 import {
   isSplitEnabled,
+  assertShardsRequireSplit,
   assertSplitRealtimeInterlock,
 } from "./v3/runOpsMigration/splitMode.server";
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
@@ -617,6 +618,12 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
 // interlock). Async, so it cannot live in the synchronous singleton factory — called
 // fire-and-forget from the eager-boot path (routing is wired synchronously at module load).
 export async function assertRunOpsSplitSentinel(): Promise<void> {
+  // Shard interlock first: shard clients are only built on the split-on arm, so this case has to be
+  // checked BEFORE the split-off early return below, which would otherwise skip it in silence.
+  assertShardsRequireSplit({
+    splitFlagEnabled: env.RUN_OPS_SPLIT_ENABLED,
+    shards: env.RUN_OPS_SHARDS,
+  });
   if (!env.RUN_OPS_SPLIT_ENABLED) return;
   // Realtime interlock (synchronous): Electric replicates only from the control-plane
   // DB, so split-on without the native realtime backend leaves NEW-resident runs

apps/webapp/app/env.server.ts

@@ -1377,6 +1377,7 @@ const EnvironmentSchema = z
     RUN_ENGINE_RUN_QUEUE_LOG_LEVEL: z
       .enum(["log", "error", "warn", "info", "debug"])
       .default("info"),
+    RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED: z.string().default("1"),
     RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM: z.string().default("0"),
     RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED: z.string().default("0"),
     RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MIN_MS: z.coerce.number().int().default(50),

apps/webapp/app/presenters/v3/BatchListPresenter.server.ts

@@ -75,6 +75,7 @@ export class BatchListPresenter extends BasePresenter {
       runOpsNew?: RunOpsPrismaClient; // new run-ops client (run-ops brand ⇒ guard classifies as runops)
       runOpsLegacyReplica?: RunOpsPrismaClient; // legacy run-ops READ REPLICA only — never the legacy primary
       controlPlaneReplica?: PrismaClientOrTransaction; // control-plane DB (for project)
+      shardReplicas?: ReadonlyArray<{ key: string; replica: RunOpsPrismaClient }>;
       splitEnabled?: boolean; // resolved boot constant
     }
   ) {
@@ -110,9 +111,11 @@ export class BatchListPresenter extends BasePresenter {
     // unsound across the residency split: legacy cuid ids ("c…") sort ABOVE new run-ops ids ("0…")
     // under id order, so a new-only page can hide pre-flip legacy batches that belong ahead of it.
     // Ordering is by createdAt (id tiebreak), which is chronologically correct across both schemes.
-    const [newRows, legacyRows] = await Promise.all([
+    const shardReplicas = this.readRoute.shardReplicas ?? [];
+    const [newRows, legacyRows, ...shardRows] = await Promise.all([
       scan(this.readRoute.runOpsNew ?? passthrough),
       scan(this.readRoute.runOpsLegacyReplica ?? passthrough),
+      ...shardReplicas.map((shard) => scan(shard.replica)),
     ]);
 
     // De-dupe by id (new wins), re-sort under the page's keyset order, re-apply the over-fetch LIMIT.
@@ -125,6 +128,11 @@ export class BatchListPresenter extends BasePresenter {
         byId.set(row.id, row);
       }
     }
+    for (const rows of shardRows) {
+      for (const row of rows) {
+        byId.set(row.id, row);
+      }
+    }
 
     // forward => newest-first (createdAt DESC), backward => oldest-first (ASC); id is the stable
     // tiebreak (ASCII codepoint, NEVER localeCompare).
@@ -167,7 +175,23 @@ export class BatchListPresenter extends BasePresenter {
     ).batchTaskRun.findFirst({
       where: { runtimeEnvironmentId: environmentId },
     });
-    return Boolean(onLegacy);
+    if (onLegacy) {
+      return true;
+    }
+
+    const shardReplicas = this.readRoute.shardReplicas ?? [];
+    if (shardReplicas.length === 0) {
+      return false;
+    }
+
+    const onShards = await Promise.all(
+      shardReplicas.map((shard) =>
+        shard.replica.batchTaskRun.findFirst({
+          where: { runtimeEnvironmentId: environmentId },
+        })
+      )
+    );
+    return onShards.some(Boolean);
   }
 
   public async call({

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -840,6 +840,45 @@ export class SpanPresenter extends BasePresenter {
           },
         };
       }
+      case "session-stream": {
+        if (!span.entity.id) {
+          logger.error(`SpanPresenter: No session stream id`, {
+            spanId,
+            sessionStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const parts = span.entity.id.split(":");
+        const io = parts.at(-1);
+        const channel = parts.at(-2) ?? "";
+        const sessionId = parts.at(-3);
+
+        if (!sessionId || (io !== "out" && io !== "in")) {
+          logger.error(`SpanPresenter: Invalid session stream id`, {
+            spanId,
+            sessionStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const metadata = span.entity.metadata
+          ? (safeJsonParse(span.entity.metadata) as Record<string, unknown> | undefined)
+          : undefined;
+
+        return {
+          ...data,
+          entity: {
+            type: "session-stream" as const,
+            object: {
+              sessionId,
+              channel: channel.length > 0 ? channel : undefined,
+              io,
+              metadata,
+            },
+          },
+        };
+      }
       case "prompt": {
         const promptData = extractPromptSpanData(span.properties as Record<string, unknown>);
 

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx

@@ -56,6 +56,7 @@ import {
   runOpsSplitReadEnabled,
   type PrismaClientOrTransaction,
 } from "~/db.server";
+import { runOpsNonAliasedShardReplicas } from "~/v3/runOpsMigration/shardHandles.server";
 import {
   docsPath,
   EnvironmentParamSchema,
@@ -104,6 +105,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     runOpsNew: runOpsNewReplicaClient,
     runOpsLegacyReplica: runOpsLegacyReplicaClient,
     controlPlaneReplica: $replica as unknown as PrismaClientOrTransaction,
+    shardReplicas: runOpsNonAliasedShardReplicas,
     splitEnabled: runOpsSplitReadEnabled,
   });
   const list = await presenter.call({

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx

@@ -493,12 +493,10 @@ function QueuesWithMetricsView() {
         </PageAccessories>
       </NavBar>
       <MetricsLayout.Root>
-        {/* Filters — pinned bar directly under the NavBar. Left cluster = search + period; right
-            cluster = pagination. */}
+        {/* Filters — pinned bar directly under the NavBar. This row is page-wide only: Period is
+            the one control that changes the tiles and charts below, so it leads the row. Search
+            and pagination scope the table alone and live in that table's own bar instead. */}
         <MetricsLayout.Filters className="px-2">
-          <div className="flex items-center gap-1.5">
-            <QueueFilters />
-          </div>
           <div className="flex items-center gap-1.5">
             <TimeFilter
               period={timeRange.period ?? undefined}
@@ -507,16 +505,12 @@ function QueuesWithMetricsView() {
               maxPeriodDays={maxPeriodDays}
               shortcut={{ key: "d" }}
             />
+          </div>
+          <div className="flex items-center gap-1.5">
             {environment.runsEnabled &&
             env.pauseSource !== ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT ? (
               <EnvironmentPauseResumeButton env={env} />
             ) : null}
-            <PaginationControls
-              currentPage={pagination.currentPage}
-              totalPages={pagination.mode === "unfiltered" ? pagination.totalPages : 1}
-              hasNextPage={pagination.mode === "filtered" ? pagination.hasMore : undefined}
-              showPageNumbers={false}
-            />
           </div>
         </MetricsLayout.Filters>
 
@@ -688,7 +682,22 @@ function QueuesWithMetricsView() {
           </ChartSyncProvider>
         ) : null}
 
-        <MetricsLayout.Content>
+        <MetricsLayout.Content
+          /* Search + pagination only ever affect the table, so they sit on the table rather than
+             in the page-wide Filters bar, where their position implied they filtered the metrics
+             above. Same left/right split as the classic view's bar. */
+          toolbar={
+            <>
+              <QueueFilters />
+              <PaginationControls
+                currentPage={pagination.currentPage}
+                totalPages={pagination.mode === "unfiltered" ? pagination.totalPages : 1}
+                hasNextPage={pagination.mode === "filtered" ? pagination.hasMore : undefined}
+                showPageNumbers={false}
+              />
+            </>
+          }
+        >
           {/* Default overflow-x-auto container so wide tables still scroll horizontally on
                 narrow viewports; the page (not this region) owns vertical scrolling. */}
           <Table containerClassName="border-t">

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam/route.tsx

@@ -57,6 +57,14 @@ import { redirectWithErrorMessage } from "~/models/message.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { SessionPresenter } from "~/presenters/v3/SessionPresenter.server";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  canonicalSessionAddressingKey,
+  resolveSessionByIdOrExternalId,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { logger } from "~/services/logger.server";
 import {
   type StreamChunk,
   useRealtimeStream,
@@ -115,11 +123,34 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     throw new Response("Session not found", { status: 404 });
   }
 
-  return typedjson({ session, loadedAt: Date.now() });
+  let channels: string[] = [];
+  const streamSessionId = session.agentView?.sessionId;
+  if (streamSessionId) {
+    const [channelsError, listed] = await tryCatch(
+      (async () => {
+        const row = await resolveSessionByIdOrExternalId($replica, environment.id, streamSessionId);
+        if (!row) return [] as string[];
+        const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session: row });
+        if (!(realtimeStream instanceof S2RealtimeStreams)) return [] as string[];
+        const addressingKey = canonicalSessionAddressingKey(row, streamSessionId);
+        return realtimeStream.listSessionChannels(addressingKey);
+      })()
+    );
+    if (channelsError) {
+      logger.warn("Failed to list session channels", {
+        sessionId: streamSessionId,
+        error: channelsError,
+      });
+    } else {
+      channels = listed ?? [];
+    }
+  }
+
+  return typedjson({ session, channels, loadedAt: Date.now() });
 };
 
 export default function Page() {
-  const { session, loadedAt } = useTypedLoaderData<typeof loader>();
+  const { session, channels, loadedAt } = useTypedLoaderData<typeof loader>();
   const organization = useOrganization();
   const project = useProject();
   const environment = useEnvironment();
@@ -158,7 +189,7 @@ export default function Page() {
       <PageBody scrollable={false}>
         <ResizablePanelGroup orientation="horizontal" className="max-h-full">
           <ResizablePanel id="session-conversation" min={"300px"}>
-            <ConversationPane session={session} />
+            <ConversationPane session={session} channels={channels} />
           </ResizablePanel>
           <ResizableHandle id="session-handle" />
           <ResizablePanel
@@ -177,18 +208,49 @@ export default function Page() {
 
 type LoadedSession = ReturnType<typeof useTypedLoaderData<typeof loader>>["session"];
 
-function ConversationPane({ session }: { session: LoadedSession }) {
+function ConversationPane({ session, channels }: { session: LoadedSession; channels: string[] }) {
   const organization = useOrganization();
   const project = useProject();
   const environment = useEnvironment();
   const { value, replace } = useSearchParams();
   const isRaw = value("raw") === "1";
+  const channelParam = value("channel");
+  const activeChannel = channelParam && channels.includes(channelParam) ? channelParam : undefined;
 
   const sessionId = session.agentView.sessionId;
   const encodedSession = encodeURIComponent(sessionId);
   const sessionResourceBase = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodedSession}/realtime/v1`;
 
-  const setView = useCallback((raw: boolean) => replace({ raw: raw ? "1" : undefined }), [replace]);
+  const setView = useCallback(
+    (raw: boolean) => replace({ raw: raw ? "1" : undefined, channel: undefined }),
+    [replace]
+  );
+  const selectChannel = useCallback(
+    (channel: string) => replace({ channel, raw: undefined }),
+    [replace]
+  );
+
+  const utilityBarProps = {
+    channels,
+    activeChannel,
+    onSelectChannel: selectChannel,
+  };
+
+  if (activeChannel) {
+    const channelBase = `${sessionResourceBase}/channels/${encodeURIComponent(activeChannel)}`;
+    return (
+      <div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
+        <RawConversationView
+          key={activeChannel}
+          inResourcePath={`${channelBase}/in`}
+          outResourcePath={`${channelBase}/out`}
+          isRaw={isRaw}
+          onChangeView={setView}
+          {...utilityBarProps}
+        />
+      </div>
+    );
+  }
 
   return (
     <div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
@@ -198,10 +260,11 @@ function ConversationPane({ session }: { session: LoadedSession }) {
           outResourcePath={`${sessionResourceBase}/out`}
           isRaw={isRaw}
           onChangeView={setView}
+          {...utilityBarProps}
         />
       ) : (
         <>
-          <ConversationUtilityBar isRaw={isRaw} onChangeView={setView} />
+          <ConversationUtilityBar isRaw={isRaw} onChangeView={setView} {...utilityBarProps} />
           <div className="min-w-0 flex-1 overflow-y-auto overflow-x-hidden px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-surface-control">
             <AgentView agentView={session.agentView} />
           </div>
@@ -214,29 +277,45 @@ function ConversationPane({ session }: { session: LoadedSession }) {
 function ConversationUtilityBar({
   isRaw,
   onChangeView,
+  channels = [],
+  activeChannel,
+  onSelectChannel,
   right,
 }: {
   isRaw: boolean;
   onChangeView: (raw: boolean) => void;
+  channels?: string[];
+  activeChannel?: string;
+  onSelectChannel?: (channel: string) => void;
   right?: React.ReactNode;
 }) {
   return (
     <div className="flex h-9 items-center justify-between gap-3 border-b border-grid-bright px-3">
       <TabContainer className="-mb-2">
         <TabButton
-          isActive={!isRaw}
+          isActive={!isRaw && !activeChannel}
           layoutId="conversation-view-mode"
           onClick={() => onChangeView(false)}
         >
           Rendered
         </TabButton>
         <TabButton
-          isActive={isRaw}
+          isActive={isRaw && !activeChannel}
           layoutId="conversation-view-mode"
           onClick={() => onChangeView(true)}
         >
           Raw
         </TabButton>
+        {channels.map((channel) => (
+          <TabButton
+            key={channel}
+            isActive={activeChannel === channel}
+            layoutId="conversation-view-mode"
+            onClick={() => onSelectChannel?.(channel)}
+          >
+            {channel}
+          </TabButton>
+        ))}
       </TabContainer>
       {right}
     </div>
@@ -266,11 +345,17 @@ function RawConversationView({
   outResourcePath,
   isRaw,
   onChangeView,
+  channels,
+  activeChannel,
+  onSelectChannel,
 }: {
   inResourcePath: string;
   outResourcePath: string;
   isRaw: boolean;
   onChangeView: (raw: boolean) => void;
+  channels?: string[];
+  activeChannel?: string;
+  onSelectChannel?: (channel: string) => void;
 }) {
   const {
     chunks: inChunks,
@@ -496,7 +581,14 @@ function RawConversationView({
 
   return (
     <>
-      <ConversationUtilityBar isRaw={isRaw} onChangeView={onChangeView} right={controls} />
+      <ConversationUtilityBar
+        isRaw={isRaw}
+        onChangeView={onChangeView}
+        channels={channels}
+        activeChannel={activeChannel}
+        onSelectChannel={onSelectChannel}
+        right={controls}
+      />
       <div className="flex min-h-0 flex-1 flex-col bg-background-deep">
         <div
           ref={scrollRef}

apps/webapp/app/routes/_app.orgs.$organizationSlug.settings.projects/ProjectsPage.tsx

@@ -13,6 +13,7 @@ import {
   SettingsContainer,
   SettingsHeader,
   SettingsRow,
+  SettingsRowDescription,
   SettingsRowTitle,
   SettingsSection,
 } from "~/components/primitives/SettingsLayout";
@@ -52,6 +53,10 @@ export function ProjectsPage({
   otherProjects: ProjectRuntimeRow[];
 }) {
   const count = needsUpdate.length;
+  // "All projects" lists every project — both the ones needing an update and the rest — sorted by name.
+  const allProjects = [...needsUpdate, ...otherProjects].sort((a, b) =>
+    a.name.localeCompare(b.name)
+  );
 
   return (
     <PageContainer>
@@ -64,10 +69,15 @@ export function ProjectsPage({
             <>
               <SettingsSection>
                 <SettingsHeader
-                  title="Runtime update available"
+                  title={
+                    <span className="flex items-center gap-x-2">
+                      <span aria-hidden className="size-2 shrink-0 rounded-full bg-warning" />
+                      Runtime update available
+                    </span>
+                  }
                   description={`${count} ${
                     count === 1 ? "project is" : "projects are"
-                  } still running Node.js ${NODE_RUNTIME_UPDATE_MAJOR} in Production. Update each one to Node.js ${NODE_RUNTIME_TARGET_MAJOR} and deploy a new version.`}
+                  } still running Node.js ${NODE_RUNTIME_UPDATE_MAJOR} in production. Update each one to Node.js ${NODE_RUNTIME_TARGET_MAJOR} and deploy a new version.`}
                 />
 
                 <SettingsRow
@@ -76,11 +86,12 @@ export function ProjectsPage({
                   description={
                     <>
                       Set the runtime in{" "}
-                      <InlineCode variant="extra-small" className="whitespace-nowrap">
+                      <InlineCode variant="extra-extra-small" className="whitespace-nowrap">
                         trigger.config.ts
                       </InlineCode>
-                      , then deploy. <InlineCode variant="extra-small">node-22</InlineCode> and{" "}
-                      <InlineCode variant="extra-small">node-26</InlineCode> are also supported.
+                      , then deploy. <InlineCode variant="extra-extra-small">node-22</InlineCode>{" "}
+                      and <InlineCode variant="extra-extra-small">node-26</InlineCode> are also
+                      supported.
                     </>
                   }
                   action={
@@ -108,7 +119,13 @@ export function ProjectsPage({
               </SettingsSection>
 
               <SettingsSection>
-                <SettingsHeader title="Projects to update" />
+                <SettingsHeader
+                  title={
+                    <span className="text-warning">
+                      {count} {count === 1 ? "project" : "projects"} to update
+                    </span>
+                  }
+                />
                 {needsUpdate.map((project) => (
                   <ProjectRow
                     key={project.ref}
@@ -121,21 +138,34 @@ export function ProjectsPage({
           ) : null}
 
           <SettingsSection>
-            <SettingsHeader title="Projects" />
-            {otherProjects.length === 0 ? (
+            <SettingsHeader title="All projects" />
+            {allProjects.length === 0 ? (
               <SettingsBlock>
-                <Paragraph variant="small">
-                  {count === 0 ? "This organization has no projects yet." : "No other projects."}
-                </Paragraph>
+                <Paragraph variant="small">This organization has no projects yet.</Paragraph>
               </SettingsBlock>
             ) : (
-              otherProjects.map((project) => (
-                <ProjectRow
-                  key={project.ref}
-                  organizationSlug={organizationSlug}
-                  project={project}
-                />
-              ))
+              <>
+                {count === 0 ? (
+                  <SettingsBlock>
+                    <div className="space-y-0.5">
+                      <div className="flex items-center gap-x-2">
+                        <span aria-hidden className="size-2 shrink-0 rounded-full bg-success" />
+                        <SettingsRowTitle>All projects are up to date</SettingsRowTitle>
+                      </div>
+                      <SettingsRowDescription>
+                        Every project is running the latest Node.js version in production.
+                      </SettingsRowDescription>
+                    </div>
+                  </SettingsBlock>
+                ) : null}
+                {allProjects.map((project) => (
+                  <ProjectRow
+                    key={project.ref}
+                    organizationSlug={organizationSlug}
+                    project={project}
+                  />
+                ))}
+              </>
             )}
           </SettingsSection>
         </SettingsContainer>

apps/webapp/app/routes/_app.orgs.$organizationSlug.settings.usage/route.tsx

@@ -1,4 +1,3 @@
-import { InformationCircleIcon } from "@heroicons/react/20/solid";
 import { Await } from "@remix-run/react";
 import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { formatDurationMilliseconds } from "@trigger.dev/core/v3";
@@ -9,17 +8,18 @@ import { UsageBar } from "~/components/billing/UsageBar";
 import { getUsageBarBillingLimitDollars } from "~/components/billing/billingAlertsFormat";
 import { PageContainer } from "~/components/layout/AppLayout";
 import { MetricsLayout } from "~/components/layout/MetricsLayout";
+import { LinkButton } from "~/components/primitives/Buttons";
 import { Card } from "~/components/primitives/charts/Card";
 import type { ChartConfig } from "~/components/primitives/charts/Chart";
 import { Chart } from "~/components/primitives/charts/ChartCompound";
 import { Header2 } from "~/components/primitives/Headers";
-import { InfoPanel } from "~/components/primitives/InfoPanel";
 import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
 import { Paragraph } from "~/components/primitives/Paragraph";
 import { Select, SelectItem } from "~/components/primitives/Select";
 import { Spinner } from "~/components/primitives/Spinner";
 import {
   Table,
+  TableBlankRow,
   TableBody,
   TableCell,
   TableHeader,
@@ -33,8 +33,12 @@ import { UsagePresenter, type UsageSeriesData } from "~/presenters/v3/UsagePrese
 import { getPromoCredits } from "~/services/platform.v3.server";
 import { requireUserId } from "~/services/session.server";
 import { formatCurrency, formatCurrencyAccurate, formatNumber } from "~/utils/numberFormatter";
-import { useBillingLimit } from "~/hooks/useOrganizations";
-import { OrganizationParamsSchema, organizationPath } from "~/utils/pathBuilder";
+import { useBillingLimit, useOrganization } from "~/hooks/useOrganizations";
+import {
+  OrganizationParamsSchema,
+  organizationPath,
+  v3BillingLimitsPath,
+} from "~/utils/pathBuilder";
 import { useCurrentPlan } from "../_app.orgs.$organizationSlug/route";
 import { pageMeta } from "~/utils/pageTitle";
 
@@ -108,7 +112,10 @@ export default function Page() {
   const { usage, tasks, months, isCurrentMonth, promoCredits } =
     useTypedLoaderData<typeof loader>();
   const currentPlan = useCurrentPlan();
+  const organization = useOrganization();
   const billingLimit = useBillingLimit();
+  const hasBillingLimit =
+    billingLimit !== undefined && billingLimit.isConfigured && billingLimit.mode === "custom";
   const planLimitCents = currentPlan?.v3Subscription?.plan?.limits.includedUsage ?? 0;
   // Enterprise bills against prepaid credits, not a per-month included-usage tier,
   // so the "Included usage" marker doesn't apply.
@@ -156,75 +163,89 @@ export default function Page() {
 
         <MetricsLayout.Grid columns={{ base: 1 }}>
           {promoCredits && (
-            <div className="flex flex-col gap-1">
-              <div className="flex items-end gap-8">
-                <div className="flex flex-col gap-1">
-                  <Header2 className="whitespace-nowrap">Credits</Header2>
-                  <p className="whitespace-nowrap text-3xl font-medium text-text-bright">
-                    {formatCurrency(promoCredits.remainingCents / 100, false)}
-                  </p>
-                </div>
-                <div className="flex w-full flex-1 flex-col gap-1 pb-1">
-                  <div className="h-2 w-full overflow-hidden rounded-full bg-background-raised">
-                    <div
-                      className="h-full rounded-full bg-blue-500"
-                      style={{
-                        width: `${
-                          promoCredits.grantedCents > 0
-                            ? Math.min(
-                                100,
-                                Math.max(
-                                  0,
-                                  (promoCredits.remainingCents / promoCredits.grantedCents) * 100
+            <Card className="pb-4">
+              <Card.Content className="pl-4 pr-4">
+                <div className="flex flex-col gap-2">
+                  <div className="flex items-baseline gap-2">
+                    <p className="whitespace-nowrap text-3xl font-medium text-text-bright">
+                      {formatCurrency(promoCredits.remainingCents / 100, false)}
+                    </p>
+                    <Header2 className="whitespace-nowrap">credits</Header2>
+                  </div>
+                  <div className="flex w-full flex-col gap-4">
+                    <div className="h-3 w-full overflow-hidden rounded-sm bg-background-raised">
+                      <div
+                        className="h-full rounded-sm bg-blue-500"
+                        style={{
+                          width: `${
+                            promoCredits.grantedCents > 0
+                              ? Math.min(
+                                  100,
+                                  Math.max(
+                                    0,
+                                    (promoCredits.remainingCents / promoCredits.grantedCents) * 100
+                                  )
                                 )
-                              )
-                            : 0
-                        }%`,
-                      }}
-                    />
-                  </div>
-                  <Paragraph variant="extra-small" className="text-text-dimmed">
-                    {formatCurrency(promoCredits.remainingCents / 100, false)} of{" "}
-                    {formatCurrency(promoCredits.grantedCents / 100, false)} remaining
-                    {promoCredits.expiresAt
-                      ? ` · expires ${creditExpiryFormatter.format(new Date(promoCredits.expiresAt))}`
-                      : ""}
-                  </Paragraph>
-                </div>
-              </div>
-            </div>
-          )}
-          <div className="flex w-full flex-col gap-2">
-            <Suspense fallback={<Spinner />}>
-              <Await
-                resolve={usage}
-                errorElement={
-                  <div className="flex min-h-40 items-center justify-center">
-                    <Paragraph variant="small">Failed to load graph.</Paragraph>
-                  </div>
-                }
-              >
-                {(usage) => (
-                  <div className="flex items-end gap-8">
-                    <div className="flex flex-col gap-1">
-                      <Header2 className="whitespace-nowrap">
-                        {isCurrentMonth ? "Month-to-date" : "Usage"}
-                      </Header2>
-                      <p className="whitespace-nowrap text-3xl font-medium text-text-bright">
-                        {formatCurrency(usage.overall.current, false)}
-                      </p>
+                              : 0
+                          }%`,
+                        }}
+                      />
                     </div>
-                    <UsageBar
-                      current={usage.overall.current}
-                      isPaying={currentPlan?.v3Subscription?.isPaying ?? false}
-                      tierLimit={isCurrentMonth && !isEnterprise ? planLimitCents / 100 : undefined}
-                      billingLimit={billingLimitDollars}
-                    />
+                    <Paragraph variant="extra-small" className="text-text-bright">
+                      {formatCurrency(promoCredits.remainingCents / 100, false)} of{" "}
+                      {formatCurrency(promoCredits.grantedCents / 100, false)} remaining
+                      {promoCredits.expiresAt
+                        ? `. Expires ${creditExpiryFormatter.format(new Date(promoCredits.expiresAt))}`
+                        : ""}
+                    </Paragraph>
                   </div>
-                )}
-              </Await>
-            </Suspense>
-          </div>
+                </div>
+              </Card.Content>
+            </Card>
+          )}
+          <Card className="pb-4">
+            <Card.Content className="pl-4 pr-4">
+              <Suspense fallback={<Spinner />}>
+                <Await
+                  resolve={usage}
+                  errorElement={
+                    <div className="flex min-h-40 items-center justify-center">
+                      <Paragraph variant="small">Failed to load graph.</Paragraph>
+                    </div>
+                  }
+                >
+                  {(usage) => (
+                    <div className="flex flex-col gap-2">
+                      <div className="flex items-center justify-between gap-2">
+                        <div className="flex items-baseline gap-2">
+                          <p className="whitespace-nowrap text-3xl font-medium text-text-bright">
+                            {formatCurrency(usage.overall.current, false)}
+                          </p>
+                          <Header2 className="whitespace-nowrap">
+                            {isCurrentMonth ? "month-to-date" : "usage"}
+                          </Header2>
+                        </div>
+                        <LinkButton
+                          variant="secondary/small"
+                          to={v3BillingLimitsPath(organization)}
+                        >
+                          {hasBillingLimit ? "Update billing limit" : "Set billing limit"}
+                        </LinkButton>
+                      </div>
+                      <UsageBar
+                        current={usage.overall.current}
+                        isPaying={currentPlan?.v3Subscription?.isPaying ?? false}
+                        tierLimit={
+                          isCurrentMonth && !isEnterprise ? planLimitCents / 100 : undefined
+                        }
+                        billingLimit={billingLimitDollars}
+                      />
+                    </div>
+                  )}
+                </Await>
+              </Suspense>
+            </Card.Content>
+          </Card>
         </MetricsLayout.Grid>
 
         <MetricsLayout.Grid>
@@ -253,7 +274,13 @@ export default function Page() {
           </Card>
         </MetricsLayout.Grid>
         <MetricsLayout.Content>
-          <Header2 className="pl-3">Tasks</Header2>
+          <div className="mt-2.5 flex items-baseline justify-between gap-2 pl-3 pr-3">
+            <Header2>Tasks</Header2>
+            <Paragraph variant="extra-small" className="text-right text-text-dimmed">
+              Dev environment runs are excluded from the usage data above, since they do not have an
+              associated compute cost.
+            </Paragraph>
+          </div>
           <Suspense fallback={<Spinner />}>
             <Await
               resolve={tasks}
@@ -263,68 +290,54 @@ export default function Page() {
                 </div>
               }
             >
-              {(tasks) => {
-                return (
-                  <>
-                    <Table>
-                      <TableHeader>
-                        <TableRow>
-                          <TableHeaderCell>Task</TableHeaderCell>
-                          <TableHeaderCell alignment="right">Runs</TableHeaderCell>
-                          <TableHeaderCell alignment="right">Average duration</TableHeaderCell>
-                          <TableHeaderCell alignment="right">Average cost</TableHeaderCell>
-                          <TableHeaderCell alignment="right">Total duration</TableHeaderCell>
-                          <TableHeaderCell alignment="right">Total cost</TableHeaderCell>
+              {(tasks) => (
+                <Table>
+                  <TableHeader>
+                    <TableRow>
+                      <TableHeaderCell>Task</TableHeaderCell>
+                      <TableHeaderCell alignment="right">Runs</TableHeaderCell>
+                      <TableHeaderCell alignment="right">Average duration</TableHeaderCell>
+                      <TableHeaderCell alignment="right">Average cost</TableHeaderCell>
+                      <TableHeaderCell alignment="right">Total duration</TableHeaderCell>
+                      <TableHeaderCell alignment="right">Total cost</TableHeaderCell>
+                    </TableRow>
+                  </TableHeader>
+                  <TableBody>
+                    {tasks.length === 0 ? (
+                      <TableBlankRow colSpan={6}>
+                        <Paragraph className="w-auto" variant="base/bright">
+                          No runs for this period
+                        </Paragraph>
+                      </TableBlankRow>
+                    ) : (
+                      tasks.map((task) => (
+                        <TableRow key={task.taskIdentifier}>
+                          <TableCell>{task.taskIdentifier}</TableCell>
+                          <TableCell alignment="right" className="tabular-nums">
+                            {formatNumber(task.runCount)}
+                          </TableCell>
+                          <TableCell alignment="right">
+                            {formatDurationMilliseconds(task.averageDuration, {
+                              style: "short",
+                            })}
+                          </TableCell>
+                          <TableCell alignment="right" className="tabular-nums">
+                            {formatCurrencyAccurate(task.averageCost)}
+                          </TableCell>
+                          <TableCell alignment="right" className="tabular-nums">
+                            {formatDurationMilliseconds(task.totalDuration, {
+                              style: "short",
+                            })}
+                          </TableCell>
+                          <TableCell alignment="right" className="tabular-nums">
+                            {formatCurrencyAccurate(task.totalCost)}
+                          </TableCell>
                         </TableRow>
-                      </TableHeader>
-                      <TableBody>
-                        {tasks.length === 0 ? (
-                          <TableRow>
-                            <TableCell colSpan={6}>
-                              <div className="flex items-center justify-center py-8">
-                                <Paragraph variant="base/bright">No runs for this period</Paragraph>
-                              </div>
-                            </TableCell>
-                          </TableRow>
-                        ) : (
-                          tasks.map((task) => (
-                            <TableRow key={task.taskIdentifier}>
-                              <TableCell>{task.taskIdentifier}</TableCell>
-                              <TableCell alignment="right" className="tabular-nums">
-                                {formatNumber(task.runCount)}
-                              </TableCell>
-                              <TableCell alignment="right">
-                                {formatDurationMilliseconds(task.averageDuration, {
-                                  style: "short",
-                                })}
-                              </TableCell>
-                              <TableCell alignment="right" className="tabular-nums">
-                                {formatCurrencyAccurate(task.averageCost)}
-                              </TableCell>
-                              <TableCell alignment="right" className="tabular-nums">
-                                {formatDurationMilliseconds(task.totalDuration, {
-                                  style: "short",
-                                })}
-                              </TableCell>
-                              <TableCell alignment="right" className="tabular-nums">
-                                {formatCurrencyAccurate(task.totalCost)}
-                              </TableCell>
-                            </TableRow>
-                          ))
-                        )}
-                      </TableBody>
-                    </Table>
-                    <InfoPanel
-                      icon={InformationCircleIcon}
-                      variant="minimal"
-                      panelClassName="max-w-full"
-                    >
-                      Dev environment runs are excluded from the usage data above, since they do not
-                      have an associated compute cost.
-                    </InfoPanel>
-                  </>
-                );
-              }}
+                      ))
+                    )}
+                  </TableBody>
+                </Table>
+              )}
             </Await>
           </Suspense>
         </MetricsLayout.Content>

apps/webapp/app/routes/api.v1.projects.$projectRef.$env.deploy-settings.ts

@@ -0,0 +1,70 @@
+import { json, type LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { type GetDeploySettingsResponseBody } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { authenticateApiKeyWithScope } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { DeploymentService } from "~/v3/services/deployment.server";
+
+const ParamsSchema = z.object({
+  projectRef: z.string(),
+  env: z.enum(["dev", "staging", "prod", "preview"]),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  try {
+    const authResult = await authenticateApiKeyWithScope(request, {
+      action: "read",
+      resource: { type: "deployments" },
+    });
+
+    if (!authResult.ok) {
+      logger.info("Invalid or missing api key", { url: request.url });
+      return json({ error: authResult.error }, { status: authResult.status });
+    }
+
+    const { environment: authenticatedEnv } = authResult.authentication;
+    const { projectRef, env } = parsedParams.data;
+
+    const deploymentService = new DeploymentService();
+
+    return await deploymentService
+      .getDeploySettings(authenticatedEnv, { projectRef, envSlug: env })
+      .match(
+        ({ buildPath, buildPathSource }) => {
+          logger.info("Resolved deploy build path", {
+            environmentId: authenticatedEnv.id,
+            projectRef,
+            env,
+            buildPath,
+            buildPathSource,
+          });
+
+          return json({ build_path: buildPath } satisfies GetDeploySettingsResponseBody);
+        },
+        (error) => {
+          switch (error.type) {
+            case "environment_mismatch":
+              return json(
+                { error: "API key does not belong to this project environment" },
+                { status: 403 }
+              );
+            case "failed_to_load_global_flags":
+            default:
+              error.type satisfies "failed_to_load_global_flags";
+              logger.error("Failed to load the global feature flags", { error: error.cause });
+              return json({ error: "Internal Server Error" }, { status: 500 });
+          }
+        }
+      );
+  } catch (error) {
+    if (error instanceof Response) throw error;
+    logger.error("Failed to resolve deploy settings", { error });
+    return json({ error: "Internal Server Error" }, { status: 500 });
+  }
+}

apps/webapp/app/routes/api.v1.sessions.ts

@@ -12,6 +12,10 @@ import { $replica, prisma, type PrismaClient } from "~/db.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { logger } from "~/services/logger.server";
 import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
+import {
+  isSafeSessionExternalId,
+  SESSION_CHANNEL_SCOPE_INFIX,
+} from "~/services/realtime/sessionChannels.server";
 import {
   ensureRunForSession,
   type SessionTriggerConfig,
@@ -168,6 +172,13 @@ const { action } = createActionApiRoute(
   },
   async ({ authentication, body }) => {
     try {
+      if (body.externalId && !isSafeSessionExternalId(body.externalId)) {
+        return json(
+          { error: `externalId cannot contain "${SESSION_CHANNEL_SCOPE_INFIX}"` },
+          { status: 422 }
+        );
+      }
+
       // Idempotent on (env, externalId): two concurrent POSTs converge to the same row, and
       // `triggerConfig` is refreshed on the cached path so a redeployed config reaches the next run.
       const { session, isCached } = await findOrCreateSession({

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { STREAM_START_HEADER } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
@@ -185,12 +186,15 @@ const loader = createLoaderApiRoute(
     // turn's first chunk and the SSE closes before records land.
     const peekSettled = request.headers.get("X-Peek-Settled") === "1";
 
+    const startFrom =
+      request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
     return realtimeStream.streamResponseFromSessionStream(
       request,
       resource.addressingKey,
       params.io,
       getRequestAbortSignal(),
-      { lastEventId, timeoutInSeconds, peekSettled }
+      { lastEventId, timeoutInSeconds, peekSettled, startFrom }
     );
   }
 );

apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.append.ts

@@ -0,0 +1,145 @@
+import { json } from "@remix-run/server-runtime";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { nanoid } from "nanoid";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  SESSION_CHANNEL_NAME_REGEX,
+  sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+  canonicalSessionAddressingKey,
+  resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  claimSessionStreamPart,
+  releaseSessionStreamPart,
+} from "~/services/sessionStreamWaitpointCache.server";
+import { anyResource, createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { ServiceValidationError } from "~/v3/services/common.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+const MAX_APPEND_BODY_BYTES = 1024 * 1024;
+
+const { action, loader } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    method: "POST",
+    maxContentLength: MAX_APPEND_BODY_BYTES,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) =>
+      resolveSessionWithWriterFallback(auth.environment.id, params.session),
+    authorization: {
+      action: "write",
+      resource: (params, _s, _h, _b, session) => {
+        const ids = new Set<string>([params.session]);
+        if (session) {
+          ids.add(session.friendlyId);
+          if (session.externalId) ids.add(session.externalId);
+        }
+        return anyResource(sessionChannelResources(params.channel, ids));
+      },
+    },
+  },
+  async ({ request, params, authentication, resource: session }) => {
+    if (!session) {
+      return new Response("Session not found", { status: 404 });
+    }
+
+    if (session.closedAt) {
+      return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
+    }
+
+    if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
+      return json({ ok: false, error: "Cannot append to an expired session" }, { status: 400 });
+    }
+
+    if (params.io === "out" && authentication.type !== "PRIVATE") {
+      return json(
+        { ok: false, error: "Appending to the out channel requires secret key authentication" },
+        { status: 403 }
+      );
+    }
+
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return json(
+        { ok: false, error: "Session channels require the S2 realtime backend" },
+        { status: 501 }
+      );
+    }
+
+    const addressingKey = canonicalSessionAddressingKey(session, params.session);
+    const claimKey = `${addressingKey}:channels:${params.channel}`;
+
+    const part = await request.text();
+
+    const clientPartId = request.headers.get("X-Part-Id");
+    const partId = clientPartId ?? nanoid(7);
+
+    const wonClaim = clientPartId
+      ? await claimSessionStreamPart(
+          authentication.environment.id,
+          claimKey,
+          params.io,
+          clientPartId
+        )
+      : true;
+
+    let appendSeq: number | undefined;
+    if (wonClaim) {
+      const [appendError, seq] = await tryCatch(
+        realtimeStream.appendPartToSessionStream(
+          part,
+          partId,
+          addressingKey,
+          params.io,
+          params.channel
+        )
+      );
+      appendSeq = seq ?? undefined;
+
+      if (appendError) {
+        if (clientPartId) {
+          await releaseSessionStreamPart(
+            authentication.environment.id,
+            claimKey,
+            params.io,
+            clientPartId
+          );
+        }
+        if (appendError instanceof ServiceValidationError) {
+          return json(
+            { ok: false, error: appendError.message },
+            { status: appendError.status ?? 422 }
+          );
+        }
+        logger.error("Failed to append to session channel stream", {
+          sessionId: session.id,
+          io: params.io,
+          channel: params.channel,
+          error: appendError,
+        });
+        return json(
+          { ok: false, error: "Something went wrong, please try again." },
+          { status: 500 }
+        );
+      }
+    }
+
+    return json({ ok: true, seq: appendSeq }, { status: 200 });
+  }
+);
+
+export { action, loader };

apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.records.ts

@@ -0,0 +1,76 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  SESSION_CHANNEL_NAME_REGEX,
+  sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+  canonicalSessionAddressingKey,
+  isSessionFriendlyIdForm,
+  resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+const SearchSchema = z.object({
+  afterEventId: z.string().regex(/^\d+$/).optional(),
+});
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    searchParams: SearchSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      const row = await resolveSessionWithWriterFallback(auth.environment.id, params.session);
+      if (!row && isSessionFriendlyIdForm(params.session)) {
+        return undefined;
+      }
+      return {
+        row,
+        addressingKey: canonicalSessionAddressingKey(row, params.session),
+      };
+    },
+    authorization: {
+      action: "read",
+      resource: ({ row, addressingKey }, params) => {
+        const ids = new Set<string>([addressingKey]);
+        if (row) {
+          ids.add(row.friendlyId);
+          if (row.externalId) ids.add(row.externalId);
+        }
+        return anyResource(sessionChannelResources(params.channel, ids));
+      },
+    },
+  },
+  async ({ params, authentication, resource, searchParams }) => {
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return new Response("Session channels require the S2 realtime backend", { status: 501 });
+    }
+
+    const afterSeqNum =
+      searchParams.afterEventId !== undefined ? Number(searchParams.afterEventId) : undefined;
+
+    const records = await realtimeStream.readSessionStreamRecords(
+      resource.addressingKey,
+      params.io,
+      afterSeqNum,
+      params.channel
+    );
+
+    return json({ records });
+  }
+);

apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.ts

@@ -0,0 +1,157 @@
+import { json } from "@remix-run/server-runtime";
+import { STREAM_START_HEADER } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  SESSION_CHANNEL_NAME_REGEX,
+  sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+  canonicalSessionAddressingKey,
+  isSessionFriendlyIdForm,
+  resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  anyResource,
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+const { action } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    method: "PUT",
+    allowJWT: true,
+    corsStrategy: "all",
+    authorization: {
+      action: "write",
+      resource: (params) => anyResource(sessionChannelResources(params.channel, [params.session])),
+    },
+  },
+  async ({ params, authentication }) => {
+    if (params.io === "out" && authentication.type !== "PRIVATE") {
+      return new Response("Initializing the out channel requires secret key authentication", {
+        status: 403,
+      });
+    }
+
+    const maybeSession = await resolveSessionWithWriterFallback(
+      authentication.environment.id,
+      params.session
+    );
+
+    if (!maybeSession && isSessionFriendlyIdForm(params.session)) {
+      return new Response("Session not found", { status: 404 });
+    }
+
+    if (maybeSession?.closedAt) {
+      return new Response("Cannot initialize a channel on a closed session", { status: 400 });
+    }
+
+    if (maybeSession?.expiresAt && maybeSession.expiresAt.getTime() < Date.now()) {
+      return new Response("Cannot initialize a channel on an expired session", { status: 400 });
+    }
+
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: maybeSession,
+      organization: maybeSession ? null : authentication.environment.organization,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return new Response("Session channels require the S2 realtime backend", { status: 501 });
+    }
+
+    const addressingKey = canonicalSessionAddressingKey(maybeSession, params.session);
+
+    const { responseHeaders } = await realtimeStream.initializeSessionStream(
+      addressingKey,
+      params.io,
+      params.channel
+    );
+
+    return json({ version: "v2" }, { status: 202, headers: responseHeaders });
+  }
+);
+
+const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      const row = await resolveSessionWithWriterFallback(auth.environment.id, params.session);
+      if (!row && isSessionFriendlyIdForm(params.session)) {
+        return undefined;
+      }
+      return {
+        row,
+        addressingKey: canonicalSessionAddressingKey(row, params.session),
+      };
+    },
+    authorization: {
+      action: "read",
+      resource: ({ row, addressingKey }, params) => {
+        const ids = new Set<string>([addressingKey]);
+        if (row) {
+          ids.add(row.friendlyId);
+          if (row.externalId) ids.add(row.externalId);
+        }
+        return anyResource(sessionChannelResources(params.channel, ids));
+      },
+    },
+  },
+  async ({ params, request, authentication, resource }) => {
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return new Response("Session channels require the S2 realtime backend", { status: 501 });
+    }
+
+    if (request.method === "HEAD") {
+      return new Response(null, { status: 200, headers: { "X-Last-Chunk-Index": "0" } });
+    }
+
+    const lastEventId = request.headers.get("Last-Event-ID") ?? undefined;
+
+    const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
+    let timeoutInSeconds: number | undefined;
+    if (timeoutInSecondsRaw) {
+      const parsed = Number(timeoutInSecondsRaw);
+      if (!Number.isFinite(parsed) || !Number.isInteger(parsed)) {
+        return new Response("Invalid timeout seconds", { status: 400 });
+      }
+      if (parsed < 1) {
+        return new Response("Timeout seconds must be greater than 0", { status: 400 });
+      }
+      if (parsed > 600) {
+        return new Response("Timeout seconds must be less than 600", { status: 400 });
+      }
+      timeoutInSeconds = parsed;
+    }
+
+    const startFrom =
+      request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
+    return realtimeStream.streamResponseFromSessionStream(
+      request,
+      resource.addressingKey,
+      params.io,
+      getRequestAbortSignal(),
+      { lastEventId, timeoutInSeconds, startFrom },
+      params.channel
+    );
+  }
+);
+
+export { action, loader };

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -1,3 +1,4 @@
+import { STREAM_START_HEADER } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
@@ -59,6 +60,9 @@ export const loader = createLoaderApiRoute(
     // Get Last-Event-ID header for resuming from a specific position
     const lastEventId = request.headers.get("Last-Event-ID") || undefined;
 
+    const startFrom =
+      request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
     const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
     const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
 
@@ -88,6 +92,7 @@ export const loader = createLoaderApiRoute(
       {
         lastEventId,
         timeoutInSeconds,
+        startFrom,
       }
     );
   }

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx

@@ -123,7 +123,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
               .map((t) => t.trim())
               .filter(Boolean)
           : []),
-      ].slice(0, 5);
+      ].slice(0, 10);
 
       const triggerConfig = {
         basePayload: {

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -1803,6 +1803,21 @@ function SpanEntity({ span }: { span: Span }) {
         />
       );
     }
+    case "session-stream": {
+      const { sessionId, channel, io } = span.entity.object;
+      const base = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodeURIComponent(sessionId)}/realtime/v1`;
+      const resourcePath = channel
+        ? `${base}/channels/${encodeURIComponent(channel)}/${io}`
+        : `${base}/${io}`;
+      const displayName = channel ? `${channel}.${io}` : `${sessionId}.${io}`;
+      return (
+        <RealtimeStreamViewer
+          resourcePath={resourcePath}
+          headerLabel={channel ? "Channel:" : "Session:"}
+          displayName={displayName}
+        />
+      );
+    }
     case "ai-generation":
     case "ai-summary": {
       return (

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam.realtime.v1.$io.ts

@@ -1,13 +1,12 @@
 import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica } from "~/db.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
 import {
   canonicalSessionAddressingKey,
-  resolveSessionByIdOrExternalId,
+  resolveSessionWithWriterFallback,
 } from "~/services/realtime/sessions.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
@@ -45,7 +44,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Environment not found", { status: 404 });
   }
 
-  const session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionParam);
+  const session = await resolveSessionWithWriterFallback(environment.id, sessionParam);
   if (!session) {
     return new Response("Session not found", { status: 404 });
   }

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam.realtime.v1.channels.$channel.$io.ts

@@ -0,0 +1,70 @@
+import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { findProjectBySlug } from "~/models/project.server";
+import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import { SESSION_CHANNEL_NAME_REGEX } from "~/services/realtime/sessionChannels.server";
+import {
+  canonicalSessionAddressingKey,
+  resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { requireUserId } from "~/services/session.server";
+import { EnvironmentParamSchema } from "~/utils/pathBuilder";
+
+const ParamsSchema = z.object({
+  sessionParam: z.string(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  const userId = await requireUserId(request);
+  const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
+  const { sessionParam, channel, io } = ParamsSchema.parse(params);
+
+  const project = await findProjectBySlug(organizationSlug, projectParam, userId);
+  if (!project) {
+    return new Response("Project not found", { status: 404 });
+  }
+
+  const environment = await findEnvironmentBySlug(project.id, envParam, userId);
+  if (!environment) {
+    return new Response("Environment not found", { status: 404 });
+  }
+
+  const session = await resolveSessionWithWriterFallback(environment.id, sessionParam);
+  if (!session) {
+    return new Response("Session not found", { status: 404 });
+  }
+
+  const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
+
+  if (!(realtimeStream instanceof S2RealtimeStreams)) {
+    return new Response("Session channels require the S2 realtime backend", {
+      status: 501,
+    });
+  }
+
+  const lastEventId = request.headers.get("Last-Event-ID") || undefined;
+  const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
+  let timeoutInSeconds: number | undefined;
+  if (timeoutInSecondsRaw !== null) {
+    timeoutInSeconds = Number(timeoutInSecondsRaw);
+    if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
+      return new Response("Invalid timeout", { status: 400 });
+    }
+  }
+
+  const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
+
+  return realtimeStream.streamResponseFromSessionStream(
+    request,
+    addressingKey,
+    io,
+    getRequestAbortSignal(),
+    { lastEventId, timeoutInSeconds },
+    channel
+  );
+}

apps/webapp/app/services/deploymentApiPaths.server.ts

@@ -4,6 +4,7 @@ export const deploymentApiPaths: (RegExp | string)[] = [
   // /current is runtime SDK surface, kept out of the deploy budget
   /^\/api\/v\d+\/deployments(?!\/current$)(\/|$)/,
   /^\/api\/v1\/projects\/[^/]+\/(dev|staging|prod|preview)$/,
+  /^\/api\/v1\/projects\/[^/]+\/(dev|staging|prod|preview)\/deploy-settings$/,
   /^\/api\/v1\/projects\/[^/]+\/envvars$/,
   /^\/api\/v1\/projects\/[^/]+\/envvars\/[^/]+\/import$/,
   /^\/api\/v1\/projects\/[^/]+\/branches$/,

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

@@ -70,8 +70,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
 
     const stream = new ReadableStream<StreamChunk>({
       start: async (controller) => {
-        // Start from lastEventId if provided, otherwise from beginning
-        let lastId = options?.lastEventId ?? "0";
+        let lastId = options?.lastEventId ?? (options?.startFrom === "latest" ? "$" : "0");
         let retryCount = 0;
         const maxRetries = 3;
         let lastDataTime = Date.now();

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

@@ -150,8 +150,14 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
    * the session's `friendlyId` and the I/O direction. Used by the session
    * realtime routes to route traffic to `sessions/{friendlyId}/{out|in}`.
    */
-  public toSessionStreamName(friendlyId: string, io: "out" | "in"): string {
-    return `${this.streamPrefix}/sessions/${friendlyId}/${io}`;
+  public toSessionStreamName(friendlyId: string, io: "out" | "in", channel?: string): string {
+    return `${this.streamPrefix}${this.#sessionStreamRelativeName(friendlyId, io, channel)}`;
+  }
+
+  #sessionStreamRelativeName(friendlyId: string, io: "out" | "in", channel?: string): string {
+    return channel
+      ? `/sessions/${friendlyId}/channels/${channel}/${io}`
+      : `/sessions/${friendlyId}/${io}`;
   }
 
   async initializeStream(
@@ -170,11 +176,12 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
    */
   async initializeSessionStream(
     friendlyId: string,
-    io: "out" | "in"
+    io: "out" | "in",
+    channel?: string
   ): Promise<{ responseHeaders?: Record<string, string> }> {
     return this.#initializeStreamByName(
-      this.toSessionStreamName(friendlyId, io),
-      `/sessions/${friendlyId}/${io}`
+      this.toSessionStreamName(friendlyId, io, channel),
+      this.#sessionStreamRelativeName(friendlyId, io, channel)
     );
   }
 
@@ -217,9 +224,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     part: string,
     partId: string,
     friendlyId: string,
-    io: "out" | "in"
+    io: "out" | "in",
+    channel?: string
   ): Promise<number> {
-    return this.#appendPartByName(part, partId, this.toSessionStreamName(friendlyId, io));
+    return this.#appendPartByName(part, partId, this.toSessionStreamName(friendlyId, io, channel));
   }
 
   async #appendPartByName(part: string, partId: string, s2Stream: string): Promise<number> {
@@ -259,9 +267,62 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   async readSessionStreamRecords(
     friendlyId: string,
     io: "out" | "in",
-    afterSeqNum?: number
+    afterSeqNum?: number,
+    channel?: string
   ): Promise<StreamRecord[]> {
-    return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io), afterSeqNum);
+    return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io, channel), afterSeqNum);
+  }
+
+  async listSessionChannels(friendlyId: string): Promise<string[]> {
+    const prefix = `${this.streamPrefix}/sessions/${friendlyId}/channels/`;
+    const names = await this.#s2ListStreamNames(prefix);
+    const channels = new Set<string>();
+    for (const name of names) {
+      const rest = name.slice(prefix.length);
+      const channel = rest.split("/")[0];
+      if (channel) channels.add(channel);
+    }
+    return [...channels];
+  }
+
+  async #s2ListStreamNames(prefix: string): Promise<string[]> {
+    const names: string[] = [];
+    let startAfter: string | undefined;
+
+    for (let page = 0; page < 100; page++) {
+      const qs = new URLSearchParams();
+      qs.set("prefix", prefix);
+      if (startAfter) qs.set("start_after", startAfter);
+
+      const res = await fetch(`${this.baseUrl}/streams?${qs}`, {
+        method: "GET",
+        headers: {
+          Authorization: `Bearer ${this.token}`,
+          Accept: "application/json",
+          "S2-Basin": this.basin,
+        },
+      });
+
+      if (!res.ok) {
+        if (res.status === 404) return names;
+        const text = await res.text().catch(() => "");
+        throw new Error(`S2 listStreams failed: ${res.status} ${res.statusText} ${text}`);
+      }
+
+      const body = (await res.json()) as {
+        has_more?: boolean;
+        streams?: Array<{ name: string; deleted_at?: string | null }>;
+      };
+      const streams = body.streams ?? [];
+      for (const stream of streams) {
+        if (stream.deleted_at) continue;
+        names.push(stream.name);
+      }
+      if (!body.has_more || streams.length === 0) break;
+      startAfter = streams[streams.length - 1]!.name;
+    }
+
+    return names;
   }
 
   async #readRecordsByName(s2Stream: string, afterSeqNum?: number): Promise<StreamRecord[]> {
@@ -402,9 +463,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     friendlyId: string,
     io: "out" | "in",
     signal: AbortSignal,
-    options?: StreamResponseOptions
+    options?: StreamResponseOptions,
+    channel?: string
   ): Promise<Response> {
-    const s2Stream = this.toSessionStreamName(friendlyId, io);
+    const s2Stream = this.toSessionStreamName(friendlyId, io, channel);
 
     let waitSeconds = options?.timeoutInSeconds ?? this.s2WaitSeconds;
     let settled = false;
@@ -527,12 +589,19 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   ): Promise<Response> {
     const startSeq = this.parseLastEventId(options?.lastEventId);
 
-    this.logger.info(`S2 streaming records from stream`, { stream: s2Stream, startSeq });
+    const tailFromLatest = startSeq == null && options?.startFrom === "latest";
+
+    this.logger.info(`S2 streaming records from stream`, {
+      stream: s2Stream,
+      startSeq,
+      tailFromLatest,
+    });
 
     // Request SSE stream from S2 and return it directly
     const s2Response = await this.s2StreamRecords(s2Stream, {
-      seq_num: startSeq ?? 0,
-      clamp: true,
+      ...(tailFromLatest
+        ? { tail_offset: 1, clamp: true }
+        : { seq_num: startSeq ?? 0, clamp: true }),
       wait: options?.timeoutInSeconds ?? this.s2WaitSeconds, // S2 will keep the connection open and stream new records
       signal, // Pass abort signal so S2 connection is cleaned up when client disconnects
     });
@@ -672,6 +741,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     stream: string,
     opts: {
       seq_num?: number;
+      tail_offset?: number;
       clamp?: boolean;
       wait?: number;
       signal?: AbortSignal;
@@ -680,6 +750,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     // GET /v1/streams/{stream}/records with Accept: text/event-stream for SSE streaming
     const qs = new URLSearchParams();
     if (opts.seq_num != null) qs.set("seq_num", String(opts.seq_num));
+    if (opts.tail_offset != null) qs.set("tail_offset", String(opts.tail_offset));
     if (opts.clamp != null) qs.set("clamp", String(opts.clamp));
     if (opts.wait != null) qs.set("wait", String(opts.wait));
 

apps/webapp/app/services/realtime/sessionChannels.server.test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it } from "vitest";
+import {
+  isSafeSessionExternalId,
+  SESSION_CHANNEL_SCOPE_INFIX,
+  sessionChannelResources,
+} from "./sessionChannels.server";
+
+describe("isSafeSessionExternalId", () => {
+  it("rejects an externalId that collides with the channel-scope fold", () => {
+    expect(isSafeSessionExternalId(`session_abc${SESSION_CHANNEL_SCOPE_INFIX}screencast`)).toBe(
+      false
+    );
+    expect(isSafeSessionExternalId(":channels:")).toBe(false);
+    expect(isSafeSessionExternalId("a:channels:b:channels:c")).toBe(false);
+  });
+
+  it("allows normal externalIds, including single colons that are not the fold infix", () => {
+    expect(isSafeSessionExternalId("chat-3c3a1756-a49a-4c78-891a-51f78596c984")).toBe(true);
+    expect(isSafeSessionExternalId("user:123")).toBe(true);
+    expect(isSafeSessionExternalId("org:abc:chat:1")).toBe(true);
+    expect(isSafeSessionExternalId("channels")).toBe(true);
+    expect(isSafeSessionExternalId("plain")).toBe(true);
+  });
+
+  it("keeps a channel-scoped token's folded id from equaling any allowed session's bare key", () => {
+    const channel = "screencast";
+    const foldedIds = sessionChannelResources(channel, ["session_abc"])
+      .map((r) => r.id)
+      .filter((id) => id.includes(SESSION_CHANNEL_SCOPE_INFIX));
+
+    for (const foldedId of foldedIds) {
+      expect(isSafeSessionExternalId(foldedId)).toBe(false);
+    }
+  });
+});

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

@@ -0,0 +1,39 @@
+import type { RbacResource } from "@trigger.dev/rbac";
+
+/**
+ * Channel names are both a URL path segment and an S2 stream-name segment, and
+ * they fold into the RBAC resource id as `${key}:channels:${channel}`, so a
+ * `/` would break addressing and a `:` would break scope parsing. Constrain to
+ * a safe, bounded alphabet.
+ */
+export const SESSION_CHANNEL_NAME_REGEX = /^[A-Za-z0-9._-]{1,128}$/;
+
+/**
+ * The infix the channel-scope fold uses in the RBAC resource id
+ * (`${key}:channels:${channel}`). A session externalId is used verbatim as a
+ * resource key, so an externalId containing this infix could equal a
+ * channel-scoped token's folded id and collide with it. Reject it at session
+ * creation so a bare session key can never look like a folded channel key.
+ */
+export const SESSION_CHANNEL_SCOPE_INFIX = ":channels:";
+
+export function isSafeSessionExternalId(externalId: string): boolean {
+  return !externalId.includes(SESSION_CHANNEL_SCOPE_INFIX);
+}
+
+/**
+ * Build the authorization resource set for a named channel. For each candidate
+ * session key (URL form, friendlyId, externalId) we authorize BOTH the
+ * channel-folded id (`${key}:channels:${channel}`, matched by a narrow
+ * channel-scoped token) and the bare session id (`${key}`, matched by a
+ * session-wide token so it grants every channel). RBAC matches ids exactly, so
+ * a channel token cannot match the bare session and vice versa.
+ */
+export function sessionChannelResources(channel: string, keys: Iterable<string>): RbacResource[] {
+  const resources: RbacResource[] = [];
+  for (const key of keys) {
+    resources.push({ type: "sessions", id: `${key}:channels:${channel}` });
+    resources.push({ type: "sessions", id: key });
+  }
+  return resources;
+}

apps/webapp/app/services/realtime/types.ts

@@ -36,6 +36,13 @@ export interface StreamIngestor {
 export type StreamResponseOptions = {
   timeoutInSeconds?: number;
   lastEventId?: string;
+  /**
+   * Where a fresh subscription (no `lastEventId`) starts reading. `"latest"`
+   * starts at the current tail so the subscriber sees only records appended
+   * after it connects; `"beginning"` (the default when unset) replays history.
+   * Ignored when `lastEventId` is set.
+   */
+  startFrom?: "beginning" | "latest";
   /**
    * Session-stream-only. When `true`, the responder MAY peek the tail
    * of `.out` and short-circuit to `wait=0` + `X-Session-Settled: true`

apps/webapp/app/v3/featureFlags.ts

@@ -1,4 +1,5 @@
 import { z } from "zod";
+import { DeployBuildPath } from "@trigger.dev/core/v3";
 
 export const FEATURE_FLAG = {
   defaultWorkerInstanceGroupId: "defaultWorkerInstanceGroupId",
@@ -37,6 +38,11 @@ export const FEATURE_FLAG = {
   // Fleet-wide pin for the complete cutover. Beats every per-org and per-env pin.
   runOpsMintShardOverride: "runOpsMintShardOverride",
   queueMetricsUiEnabled: "queueMetricsUiEnabled",
+  // Build path for CLI deploys, resolved by DeploymentService.getDeploySettings.
+  deployBuildPath: "deployBuildPath",
+  deployBuildPathPreview: "deployBuildPathPreview",
+  deployBuildPathStaging: "deployBuildPathStaging",
+  deployBuildPathProduction: "deployBuildPathProduction",
   // Per-organization rollout for creating additional environment API keys.
   additionalApiKeysEnabled: "additionalApiKeysEnabled",
   // System-wide kill switch for issuing additional environment API keys.
@@ -148,6 +154,10 @@ export const FeatureFlagCatalog = {
   // Per-org access to the Queue Metrics dashboard UI (view only; emission is global and
   // separate). Off unless enabled for the org.
   [FEATURE_FLAG.queueMetricsUiEnabled]: z.coerce.boolean(),
+  [FEATURE_FLAG.deployBuildPath]: DeployBuildPath,
+  [FEATURE_FLAG.deployBuildPathPreview]: DeployBuildPath,
+  [FEATURE_FLAG.deployBuildPathStaging]: DeployBuildPath,
+  [FEATURE_FLAG.deployBuildPathProduction]: DeployBuildPath,
   // Strict booleans prevent a stringified "false" from silently enabling API-key
   // creation or lookup. Cold/absent values resolve to the safe `false`.
   [FEATURE_FLAG.additionalApiKeysEnabled]: z.boolean(),

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

@@ -62,6 +62,7 @@ function createRunEngine() {
     queue: {
       defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
       defaultEnvConcurrencyBurstFactor: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR,
+      totalConcurrencyEnabled: env.RUN_ENGINE_TOTAL_CONCURRENCY_LIMITS_ENABLED === "1",
       logLevel: env.RUN_ENGINE_RUN_QUEUE_LOG_LEVEL,
       redis: {
         keyPrefix: "engine:",

apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts

@@ -64,6 +64,48 @@ export async function probeControlPlaneCoresidency(
 
 export type DistinctTarget = { id: string; url: string };
 
+/** Injection seam for the retry tests: no containers, no real waiting. */
+export type DistinctProbeOptions = {
+  logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void };
+  readFingerprint?: (url: string) => Promise<DatabaseFingerprint>;
+  /** Total attempts per target, including the first. Bounded so boot latency stays bounded. */
+  attempts?: number;
+  sleep?: (ms: number) => Promise<void>;
+};
+
+const DEFAULT_PROBE_ATTEMPTS = 3;
+const RETRY_BASE_DELAY_MS = 250;
+
+const defaultSleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
+
+/**
+ * Read one fingerprint, retrying a bounded number of times.
+ *
+ * The probe fails CLOSED, and that must not change: "distinct" is a positive claim a failed probe
+ * cannot support. But failing closed on the first blip means one shard being briefly unreachable
+ * collapses the deployment to single-DB, and the boot interlock then refuses the boot for the whole
+ * fleet. A transient error deserves a retry; a persistent one still fails closed, just later.
+ */
+async function readFingerprintWithRetry(
+  url: string,
+  read: (url: string) => Promise<DatabaseFingerprint>,
+  attempts: number,
+  sleep: (ms: number) => Promise<void>
+): Promise<DatabaseFingerprint> {
+  let lastError: unknown;
+  for (let attempt = 1; attempt <= attempts; attempt++) {
+    try {
+      return await read(url);
+    } catch (error) {
+      lastError = error;
+      if (attempt < attempts) {
+        await sleep(RETRY_BASE_DELAY_MS * attempt);
+      }
+    }
+  }
+  throw lastError;
+}
+
 /**
  * Set uniqueness over every store that owns its own database. Fail-closed: a probe that cannot
  * answer returns NOT distinct, because "distinct" is a positive claim a failed probe cannot support.
@@ -77,14 +119,22 @@ export type DistinctTarget = { id: string; url: string };
  */
 export async function probeDistinctStores(
   targets: DistinctTarget[],
-  opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
+  opts?: DistinctProbeOptions
 ): Promise<{ distinct: true } | { distinct: false; reason: string }> {
   if (targets.length < 2) {
     return { distinct: true };
   }
 
+  const read = opts?.readFingerprint ?? readDatabaseFingerprint;
+  const attempts = opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS;
+  const sleep = opts?.sleep ?? defaultSleep;
+
   try {
-    const fingerprints = await Promise.all(targets.map((t) => readDatabaseFingerprint(t.url)));
+    // Retry per TARGET, not around the whole set: one slow shard must not re-probe the stores that
+    // already answered. A duplicate verdict below is final and is never retried.
+    const fingerprints = await Promise.all(
+      targets.map((t) => readFingerprintWithRetry(t.url, read, attempts, sleep))
+    );
 
     const seen = new Map<string, string>();
     for (const [index, target] of targets.entries()) {
@@ -104,7 +154,9 @@ export async function probeDistinctStores(
 
     return { distinct: true };
   } catch (error) {
-    const reason = `distinct-db sentinel probe failed; failing closed (single-DB). ${String(error)}`;
+    const reason =
+      `distinct-db sentinel probe failed after ${opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS} ` +
+      `attempt(s); failing closed (single-DB). ${String(error)}`;
     opts?.logger?.warn(reason, { error });
     return { distinct: false, reason };
   }
@@ -115,7 +167,7 @@ export async function probeDistinctStores(
 export async function probeDistinctDatabases(
   legacyUrl: string,
   newUrl: string,
-  opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
+  opts?: DistinctProbeOptions
 ): Promise<{ distinct: true } | { distinct: false; reason: string }> {
   return probeDistinctStores(
     [

apps/webapp/app/v3/runOpsMigration/shardHandles.server.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { buildShardHandleMaps } from "./shardHandles.server";
+import { buildShardHandleMaps, nonAliasedShardReplicas } from "./shardHandles.server";
 
 // Two distinct sentinels per shard: the maps must not cross writer and replica.
 function handle(key: string) {
@@ -35,3 +35,26 @@ describe("buildShardHandleMaps", () => {
     expect(replicas.get("a")).not.toEqual({ tag: "a-writer" });
   });
 });
+
+describe("nonAliasedShardReplicas", () => {
+  it("yields an empty list when no shard is configured", () => {
+    expect(nonAliasedShardReplicas([])).toEqual([]);
+  });
+
+  it("keeps the configured order and carries each shard's replica", () => {
+    expect(nonAliasedShardReplicas([handle("b"), handle("a")])).toEqual([
+      { key: "b", replica: { tag: "b-replica" } },
+      { key: "a", replica: { tag: "a-replica" } },
+    ]);
+  });
+
+  it("drops a shard that declares aliasOf", () => {
+    expect(nonAliasedShardReplicas([{ ...handle("a"), aliasOf: "new" }, handle("b")])).toEqual([
+      { key: "b", replica: { tag: "b-replica" } },
+    ]);
+  });
+
+  it("never carries a writer in place of a replica", () => {
+    expect(nonAliasedShardReplicas([handle("a")])[0]?.replica).not.toEqual({ tag: "a-writer" });
+  });
+});

apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts

@@ -5,14 +5,16 @@
  * is what keeps every gen-2 arm unreachable today.
  */
 import type { PrismaClient } from "@trigger.dev/database";
+import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import type { ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaReplicaClient } from "~/db.server";
 import { runOpsShardHandles } from "~/db.server";
 
 type ShardHandle = {
   key: string;
-  writer: unknown;
-  replica: unknown;
+  writer: RunOpsPrismaClient;
+  replica: RunOpsPrismaClient;
+  aliasOf?: string;
 };
 
 export function buildShardHandleMaps(handles: ShardHandle[]): {
@@ -22,8 +24,8 @@ export function buildShardHandleMaps(handles: ShardHandle[]): {
   const replicas = new Map<ShardKey, PrismaReplicaClient>();
   const writers = new Map<ShardKey, PrismaClient>();
   for (const handle of handles) {
-    replicas.set(handle.key, handle.replica as PrismaReplicaClient);
-    writers.set(handle.key, handle.writer as PrismaClient);
+    replicas.set(handle.key, handle.replica as unknown as PrismaReplicaClient);
+    writers.set(handle.key, handle.writer as unknown as PrismaClient);
   }
   return { replicas, writers };
 }
@@ -40,7 +42,17 @@ function resolveShardHandles(): ShardHandle[] {
   }
 }
 
-const maps = buildShardHandleMaps(resolveShardHandles());
+export function nonAliasedShardReplicas<TClient>(
+  handles: ReadonlyArray<{ key: string; replica: TClient; aliasOf?: string }>
+): ReadonlyArray<{ key: string; replica: TClient }> {
+  return handles
+    .filter((handle) => handle.aliasOf === undefined)
+    .map((handle) => ({ key: handle.key, replica: handle.replica }));
+}
+
+const handles = resolveShardHandles();
+const maps = buildShardHandleMaps(handles);
 
 export const runOpsShardReplicas = maps.replicas;
 export const runOpsShardWriters = maps.writers;
+export const runOpsNonAliasedShardReplicas = nonAliasedShardReplicas(handles);

apps/webapp/app/v3/runOpsMigration/splitMode.server.ts

@@ -7,7 +7,11 @@
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { probeDistinctStores as defaultProbe } from "./distinctDbSentinel.server";
-import { nonAliasedShards, type ShardTarget } from "~/v3/runOpsShards.server";
+import {
+  nonAliasedShards,
+  type RunOpsShardDescriptor,
+  type ShardTarget,
+} from "~/v3/runOpsShards.server";
 
 export type SplitModeConfig = {
   flagEnabled: boolean;
@@ -73,6 +77,35 @@ export function assertSplitRealtimeInterlock(config: SplitRealtimeInterlockConfi
   }
 }
 
+export type ShardsRequireSplitConfig = {
+  splitFlagEnabled: boolean;
+  /** Raw descriptors. The alias exemption is applied here so no call site can forget it. */
+  shards: RunOpsShardDescriptor[];
+};
+
+/**
+ * Boot-time shard interlock (pure predicate). Shard clients are only built on the split-on arm of
+ * `selectRunOpsTopology`, so a shard configured while the split flag is off is dropped in silence:
+ * no client, no fan-out leg, and any row already resident on that database vanishes from every
+ * list with no error. The other two ways split can end up disabled (URLs missing, sentinel not
+ * distinct) already refuse to boot; this closes the one that does not.
+ */
+export function assertShardsRequireSplit(config: ShardsRequireSplitConfig): void {
+  if (config.splitFlagEnabled) {
+    return;
+  }
+  // An aliased shard owns no database: it shares its target's client by reference, so its rows are
+  // still read with the split off and nothing is dropped. Exempt here exactly as it is exempt from
+  // the distinctness sentinel, the coresidency loop and replication.
+  const owning = nonAliasedShards(config.shards).map((shard) => shard.key);
+  if (owning.length === 0) {
+    return;
+  }
+  throw new Error(
+    `RUN_OPS_SHARDS configures shard(s) ${owning.join(", ")} but RUN_OPS_SPLIT_ENABLED is off, so no shard client is built and rows on those databases would be silently missing; refusing to start.`
+  );
+}
+
 let cached: Promise<boolean> | undefined;
 
 export function isSplitEnabled(): Promise<boolean> {

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

@@ -40,6 +40,23 @@ export async function updateQueueConcurrencyLimits(
   await engine.runQueue.updateQueueConcurrencyLimits(environment, queueName, concurrency);
 }
 
+/** Updates the RunQueue total concurrency limit for a queue (the cap across all concurrency-key values) */
+export async function updateQueueTotalConcurrencyLimits(
+  environment: AuthenticatedEnvironment,
+  queueName: string,
+  totalConcurrency: number
+) {
+  await engine.runQueue.updateQueueTotalConcurrencyLimits(environment, queueName, totalConcurrency);
+}
+
+/** Removes the RunQueue total concurrency limit for a queue */
+export async function removeQueueTotalConcurrencyLimits(
+  environment: AuthenticatedEnvironment,
+  queueName: string
+) {
+  await engine.runQueue.removeQueueTotalConcurrencyLimits(environment, queueName);
+}
+
 /** Removes the RunQueue limits for a queue */
 export async function removeQueueConcurrencyLimits(
   environment: AuthenticatedEnvironment,

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -33,8 +33,10 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
 import { engine } from "../runEngine.server";
 import {
   removeQueueConcurrencyLimits,
+  removeQueueTotalConcurrencyLimits,
   updateEnvConcurrencyLimits,
   updateQueueConcurrencyLimits,
+  updateQueueTotalConcurrencyLimits,
 } from "../runQueue.server";
 import { scheduleEngine } from "../scheduleEngine.server";
 import { normalizeScheduleWindow } from "../scheduleWindow.server";
@@ -401,6 +403,7 @@ async function createWorkerTask(
         {
           name: task.queue?.name ?? `task/${task.id}`,
           concurrencyLimit: task.queue?.concurrencyLimit,
+          totalConcurrencyLimit: task.queue?.totalConcurrencyLimit,
         },
         task.id,
         task.queue?.name ? "NAMED" : "VIRTUAL",
@@ -552,6 +555,7 @@ async function createWorkerQueue(
   const taskQueue = await upsertWorkerQueueRecord(
     queueName,
     baseConcurrencyLimit ?? null,
+    queue.totalConcurrencyLimit ?? null,
     orderableName,
     queueType,
     worker,
@@ -560,6 +564,21 @@ async function createWorkerQueue(
 
   const newConcurrencyLimit = taskQueue.concurrencyLimit;
 
+  /**
+   * The total limit key is separate from the per-queue limit key that pause zeroes,
+   * so it is safe to sync it regardless of the paused state. The engine clamps it
+   * to the environment limit at read time, so the raw declared value is stored.
+   */
+  if (typeof taskQueue.totalConcurrencyLimit === "number") {
+    await updateQueueTotalConcurrencyLimits(
+      environment,
+      taskQueue.name,
+      taskQueue.totalConcurrencyLimit
+    );
+  } else {
+    await removeQueueTotalConcurrencyLimits(environment, taskQueue.name);
+  }
+
   if (!taskQueue.paused) {
     if (typeof newConcurrencyLimit === "number") {
       logger.debug("createWorkerQueue: updating concurrency limit", {
@@ -598,6 +617,7 @@ async function createWorkerQueue(
 async function upsertWorkerQueueRecord(
   queueName: string,
   concurrencyLimit: number | null,
+  totalConcurrencyLimit: number | null,
   orderableName: string,
   queueType: TaskQueueType,
   worker: BackgroundWorker,
@@ -624,6 +644,7 @@ async function upsertWorkerQueueRecord(
           name: queueName,
           orderableName,
           concurrencyLimit,
+          totalConcurrencyLimit,
           runtimeEnvironmentId: worker.runtimeEnvironmentId,
           projectId: worker.projectId,
           type: queueType,
@@ -648,6 +669,7 @@ async function upsertWorkerQueueRecord(
           // If overridden, keep current limit and update base; otherwise update limit normally
           concurrencyLimit: hasOverride ? undefined : concurrencyLimit,
           concurrencyLimitBase: hasOverride ? concurrencyLimit : undefined,
+          totalConcurrencyLimit,
         },
       });
     }
@@ -659,6 +681,7 @@ async function upsertWorkerQueueRecord(
       return await upsertWorkerQueueRecord(
         queueName,
         concurrencyLimit,
+        totalConcurrencyLimit,
         orderableName,
         queueType,
         worker,

apps/webapp/app/v3/services/deployment.server.ts

@@ -4,16 +4,25 @@ import { errAsync, fromPromise, okAsync, type ResultAsync } from "neverthrow";
 import { Prisma, type WorkerDeployment, type Project, boundedIn } from "@trigger.dev/database";
 import {
   BuildServerMetadata,
+  DeployBuildPath,
   logger,
   type GitMeta,
   type DeploymentEvent,
+  type RuntimeEnvironmentType,
 } from "@trigger.dev/core/v3";
 import { TimeoutDeploymentService } from "./timeoutDeployment.server";
 import { recordDeploymentFinished } from "./recordDeploymentFinished.server";
 import { env } from "~/env.server";
 import { createRemoteImageBuild } from "../remoteImageBuilder.server";
 import { FINAL_DEPLOYMENT_STATUSES } from "./failDeployment.server";
-import { enqueueBuild, generateRegistryCredentials } from "~/services/platform.v3.server";
+import {
+  enqueueBuild,
+  generateRegistryCredentials,
+  isBillingConfigured,
+} from "~/services/platform.v3.server";
+import { FEATURE_FLAG, type FeatureFlagKey } from "../featureFlags";
+import { flags } from "../featureFlags.server";
+import { globalFlagsRegistry } from "../globalFlagsRegistry.server";
 import { AppendInput, AppendRecord, S2 } from "@s2-dev/streamstore";
 import { createRedisClient } from "~/redis.server";
 
@@ -28,6 +37,31 @@ const s2TokenRedis = createRedisClient("s2-token-cache", {
 });
 const s2 = env.S2_ENABLED === "1" ? new S2({ accessToken: env.S2_ACCESS_TOKEN }) : undefined;
 
+const DEPLOY_BUILD_PATH_ENV_FLAG: Partial<Record<RuntimeEnvironmentType, FeatureFlagKey>> = {
+  PREVIEW: FEATURE_FLAG.deployBuildPathPreview,
+  STAGING: FEATURE_FLAG.deployBuildPathStaging,
+  PRODUCTION: FEATURE_FLAG.deployBuildPathProduction,
+};
+
+const DEPLOY_ENV_SLUG_FOR_TYPE: Record<RuntimeEnvironmentType, DeployEnvSlug> = {
+  DEVELOPMENT: "dev",
+  STAGING: "staging",
+  PRODUCTION: "prod",
+  PREVIEW: "preview",
+};
+
+type DeployEnvSlug = "dev" | "staging" | "prod" | "preview";
+
+type DeployBuildPathSource =
+  | "unavailable"
+  | "organization_environment"
+  | "organization"
+  | "global_environment"
+  | "global"
+  | "default";
+
+type DeploySettings = { buildPath: DeployBuildPath; buildPathSource: DeployBuildPathSource };
+
 export class DeploymentService extends BaseService {
   /**
    * Progresses a deployment from PENDING to INSTALLING and then to BUILDING.
@@ -282,6 +316,67 @@ export class DeploymentService extends BaseService {
       .map(() => undefined);
   }
 
+  public getDeploySettings(
+    authenticatedEnv: Pick<AuthenticatedEnvironment, "type" | "organization" | "project">,
+    target: { projectRef: string; envSlug: DeployEnvSlug }
+  ) {
+    const validateTarget = (): ResultAsync<undefined, { type: "environment_mismatch" }> => {
+      if (
+        authenticatedEnv.project.externalRef !== target.projectRef ||
+        DEPLOY_ENV_SLUG_FOR_TYPE[authenticatedEnv.type] !== target.envSlug
+      ) {
+        return errAsync({ type: "environment_mismatch" as const });
+      }
+      return okAsync(undefined);
+    };
+
+    const loadGlobalFlags = () =>
+      fromPromise(Promise.resolve(globalFlagsRegistry.current() ?? flags()), (error) => ({
+        type: "failed_to_load_global_flags" as const,
+        cause: error,
+      }));
+
+    const pickBuildPath = (globalFlagSet: Record<string, unknown>): DeploySettings => {
+      const envKey = DEPLOY_BUILD_PATH_ENV_FLAG[authenticatedEnv.type];
+      const orgFlags = authenticatedEnv.organization.featureFlags;
+      const orgFlagSet: Record<string, unknown> =
+        orgFlags && typeof orgFlags === "object" && !Array.isArray(orgFlags)
+          ? (orgFlags as Record<string, unknown>)
+          : {};
+
+      const candidates: Array<
+        [Record<string, unknown>, FeatureFlagKey | undefined, DeployBuildPathSource]
+      > = [
+        [orgFlagSet, envKey, "organization_environment"],
+        [orgFlagSet, FEATURE_FLAG.deployBuildPath, "organization"],
+        [globalFlagSet, envKey, "global_environment"],
+        [globalFlagSet, FEATURE_FLAG.deployBuildPath, "global"],
+      ];
+
+      for (const [flagSet, key, buildPathSource] of candidates) {
+        if (!key) continue;
+        const parsed = DeployBuildPath.safeParse(flagSet[key]);
+        if (parsed.success) {
+          return { buildPath: parsed.data, buildPathSource };
+        }
+      }
+
+      return { buildPath: "depot", buildPathSource: "default" };
+    };
+
+    const resolveBuildPath = (): ResultAsync<
+      DeploySettings,
+      { type: "failed_to_load_global_flags"; cause: unknown }
+    > => {
+      if (!isBillingConfigured()) {
+        return okAsync({ buildPath: "depot" as const, buildPathSource: "unavailable" as const });
+      }
+      return loadGlobalFlags().map(pickBuildPath);
+    };
+
+    return validateTarget().andThen(resolveBuildPath);
+  }
+
   /**
    * Generates registry credentials for a deployment. Returns an error if the deployment is in a final state.
    *

apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts

@@ -13,6 +13,7 @@ import type {
   StartRunAttemptResult,
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
+import { getMeter } from "@internal/tracing";
 import { SemanticInternalAttributes } from "@trigger.dev/core/v3";
 import { fromFriendlyId } from "@trigger.dev/core/v3/isomorphic";
 import { WORKER_HEADERS, type WorkerQueueClass } from "@trigger.dev/core/v3/workers";
@@ -21,11 +22,9 @@ import { Prisma, WorkerInstanceGroupType } from "@trigger.dev/database";
 import { json } from "@remix-run/server-runtime";
 import { createHash, timingSafeEqual } from "crypto";
 import { customAlphabet } from "nanoid";
-import { Counter } from "prom-client";
 import { z } from "zod";
 import { env } from "~/env.server";
-import { metricsRegister } from "~/metrics.server";
-import { evaluateCreatedAtGate } from "./workloadTokenAuthorization.server";
+import { evaluateCreatedAtGate, runAgeBucket } from "./workloadTokenAuthorization.server";
 import {
   isWorkerQueueDequeueDisabled,
   recordBlockedDequeue,
@@ -62,17 +61,11 @@ if (workloadCreatedAtGateEnabled && !workloadTokenCutoff) {
 
 type WorkloadGateAction = "start" | "complete" | "continue" | "snapshots_since";
 
-// singleton: module-scope registration double-registers under dev HMR
-const workloadAuthGateCounter = singleton(
-  "workloadAuthGateCounter",
-  () =>
-    new Counter({
-      name: "workload_auth_gate_total",
-      help: "Deployment token authorization outcomes on worker actions",
-      labelNames: ["outcome", "action"] as const,
-      registers: [metricsRegister],
-    })
-);
+const meter = getMeter("workload-auth-gate");
+
+const workloadAuthGateCounter = meter.createCounter("workload_auth_gate_total", {
+  description: "Deployment token authorization outcomes on worker actions",
+});
 
 function createAuthenticatedWorkerInstanceCache() {
   return createCache({
@@ -456,7 +449,7 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     if (environmentId) {
       // Scoping is delegated to the engine snapshot read; no run-row read here. Recorded so the
       // platform can see how much traffic is env-scoped as enforcement rolls out.
-      workloadAuthGateCounter.inc({ outcome: "env_scoped", action });
+      workloadAuthGateCounter.add(1, { outcome: "env_scoped", action });
       return;
     }
 
@@ -464,7 +457,10 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
       return;
     }
 
-    const run = await this._engine.runStore.findRun({ id: runId }, { select: { createdAt: true } });
+    const run = await this._engine.runStore.findRun(
+      { id: runId },
+      { select: { createdAt: true, environmentType: true } }
+    );
 
     if (!run) {
       // Let the engine method surface the canonical not-found error.
@@ -476,7 +472,12 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
       cutoff: workloadTokenCutoff,
     });
 
-    workloadAuthGateCounter.inc({ outcome, action });
+    workloadAuthGateCounter.add(1, {
+      outcome,
+      action,
+      env_type: run.environmentType ?? "unknown",
+      run_age_bucket: runAgeBucket(run.createdAt, new Date()),
+    });
 
     if (!allow) {
       logger.warn("[workload-auth] rejecting untokened worker action created after cutoff", {

apps/webapp/app/v3/services/worker/workloadTokenAuthorization.server.ts

@@ -25,3 +25,22 @@ export function evaluateCreatedAtGate(params: {
     ? { outcome: "suppressed", allow: false }
     : { outcome: "grandfathered", allow: true };
 }
+
+const HOUR_MS = 60 * 60 * 1000;
+const DAY_MS = 24 * HOUR_MS;
+
+const RUN_AGE_BUCKETS = [
+  { under: HOUR_MS, label: "lt_1h" },
+  { under: DAY_MS, label: "1h_1d" },
+  { under: 7 * DAY_MS, label: "1d_7d" },
+  { under: 30 * DAY_MS, label: "7d_30d" },
+] as const;
+
+/**
+ * Coarse age of the run behind an untokened worker action. Reading this while the cutoff is set far
+ * in the future answers "what would a cutoff of X reject" without rejecting anything.
+ */
+export function runAgeBucket(runCreatedAt: Date, now: Date): string {
+  const ageMs = now.getTime() - runCreatedAt.getTime();
+  return RUN_AGE_BUCKETS.find((bucket) => ageMs < bucket.under)?.label ?? "gt_30d";
+}

apps/webapp/test/batchListPresenter.readroute.test.ts

@@ -16,14 +16,18 @@ vi.mock("~/db.server", () => ({
 import {
   heteroPostgresTest,
   heteroRunOpsPostgresTest,
+  makeNShardRunOpsPostgresTest,
   postgresTest,
 } from "@internal/testcontainers";
+import { generateRunOpsId, generateRunOpsIdV2 } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
 import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import {
+  type BatchList,
   type BatchListOptions,
   BatchListPresenter,
 } from "~/presenters/v3/BatchListPresenter.server";
+import { nonAliasedShardReplicas } from "~/v3/runOpsMigration/shardHandles.server";
 
 vi.setConfig({ testTimeout: 120_000 });
 
@@ -163,7 +167,7 @@ async function mirrorEnvParents(
 }
 
 async function createBatch(
-  prisma: PrismaClient,
+  prisma: PrismaClient | RunOpsPrismaClient,
   ctx: SeedContext,
   batch: {
     id: string;
@@ -174,7 +178,7 @@ async function createBatch(
     createdAt?: Date;
   }
 ) {
-  return prisma.batchTaskRun.create({
+  return (prisma as PrismaClient).batchTaskRun.create({
     data: {
       id: batch.id,
       friendlyId: batch.friendlyId,
@@ -621,3 +625,289 @@ describe("BatchListPresenter run-ops read routing (PG14 control-plane/legacy + P
     }
   );
 });
+
+describe("BatchListPresenter gen-2 shard legs (legacy PG14 + new PG17 + 2 shard PG17 databases)", () => {
+  const twoShardTest = makeNShardRunOpsPostgresTest(2);
+
+  const shardPresenter = (
+    legacyPrisma: PrismaClient,
+    newPrisma: RunOpsPrismaClient,
+    shardReplicas: ReadonlyArray<{ key: string; replica: RunOpsPrismaClient }>
+  ) =>
+    new BatchListPresenter(legacyPrisma, legacyPrisma, {
+      runOpsNew: newPrisma,
+      runOpsLegacyReplica: legacyPrisma as unknown as RunOpsPrismaClient,
+      controlPlaneReplica: legacyPrisma,
+      splitEnabled: true,
+      shardReplicas,
+    });
+
+  twoShardTest(
+    "a gen-2 batch on its shard appears in the list alongside gen-1 and legacy batches",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const shardA = shardPrismas[0]!;
+      const ctx = await seedParents(legacyPrisma, "gen2-visible");
+
+      const legacyId = "cmm00000000000000000legac";
+      const newId = generateRunOpsId();
+      const shardId = generateRunOpsIdV2("a");
+
+      await createBatch(legacyPrisma, ctx, {
+        id: legacyId,
+        friendlyId: "fr_legacy",
+        createdAt: new Date(Date.now() - 3 * 60_000),
+      });
+      await createBatch(newPrisma, ctx, {
+        id: newId,
+        friendlyId: "fr_new",
+        createdAt: new Date(Date.now() - 2 * 60_000),
+      });
+      await createBatch(shardA, ctx, {
+        id: shardId,
+        friendlyId: "fr_shard_a",
+        createdAt: new Date(Date.now() - 1 * 60_000),
+      });
+
+      const page = await shardPresenter(legacyPrisma, newPrisma, [
+        { key: "a", replica: shardA },
+      ]).call(baseCall(ctx, { pageSize: 10 }));
+
+      expect(page.batches.map((b) => b.id)).toEqual([shardId, newId, legacyId]);
+      expect(page.batches.map((b) => b.friendlyId)).toContain("fr_shard_a");
+
+      const withoutShardLeg = await shardPresenter(legacyPrisma, newPrisma, []).call(
+        baseCall(ctx, { pageSize: 10 })
+      );
+      expect(withoutShardLeg.batches.map((b) => b.id)).toEqual([newId, legacyId]);
+    }
+  );
+
+  twoShardTest(
+    "a page spanning legacy, new and two shards is ordered by createdAt then id across all stores",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const [shardA, shardB] = [shardPrismas[0]!, shardPrismas[1]!];
+      const ctx = await seedParents(legacyPrisma, "gen2-order");
+
+      const t0 = new Date(Date.now() - 10 * 60_000);
+      const at = (minutes: number) => new Date(t0.getTime() + minutes * 60_000);
+
+      const legacyId = "cmm00000000000000000order";
+      const newId = generateRunOpsId();
+      const shardAId = generateRunOpsIdV2("a");
+      const shardBId = generateRunOpsIdV2("b");
+
+      await createBatch(legacyPrisma, ctx, {
+        id: legacyId,
+        friendlyId: "fr_o_legacy",
+        createdAt: at(0),
+      });
+      await createBatch(newPrisma, ctx, { id: newId, friendlyId: "fr_o_new", createdAt: at(1) });
+      await createBatch(shardA, ctx, { id: shardAId, friendlyId: "fr_o_a", createdAt: at(2) });
+      await createBatch(shardB, ctx, { id: shardBId, friendlyId: "fr_o_b", createdAt: at(3) });
+
+      const tieTime = at(4);
+      const tieOnNew = generateRunOpsId();
+      const tieOnShardB = generateRunOpsIdV2("b");
+      await createBatch(newPrisma, ctx, {
+        id: tieOnNew,
+        friendlyId: "fr_tie_new",
+        createdAt: tieTime,
+      });
+      await createBatch(shardB, ctx, {
+        id: tieOnShardB,
+        friendlyId: "fr_tie_b",
+        createdAt: tieTime,
+      });
+
+      const page = await shardPresenter(legacyPrisma, newPrisma, [
+        { key: "a", replica: shardA },
+        { key: "b", replica: shardB },
+      ]).call(baseCall(ctx, { pageSize: 10 }));
+
+      const tieHead = tieOnNew > tieOnShardB ? tieOnNew : tieOnShardB;
+      const tieTail = tieOnNew > tieOnShardB ? tieOnShardB : tieOnNew;
+      expect(page.batches.map((b) => b.id)).toEqual([
+        tieHead,
+        tieTail,
+        shardBId,
+        shardAId,
+        newId,
+        legacyId,
+      ]);
+    }
+  );
+
+  twoShardTest(
+    "paging forward then backward across boundaries that span stores loses and repeats no batch",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const [shardA, shardB] = [shardPrismas[0]!, shardPrismas[1]!];
+      const ctx = await seedParents(legacyPrisma, "gen2-paging");
+
+      const t0 = Date.now() - 60 * 60_000;
+      const seeded: string[] = [];
+      for (let i = 0; i < 8; i++) {
+        const store = [legacyPrisma, newPrisma, shardA, shardB][i % 4]!;
+        const id =
+          i % 4 === 0
+            ? `cmm0000000000000000pag${i}`
+            : i % 4 === 1
+              ? generateRunOpsId()
+              : generateRunOpsIdV2(i % 4 === 2 ? "a" : "b");
+        await createBatch(store, ctx, {
+          id,
+          friendlyId: `fr_pag_${i}`,
+          createdAt: new Date(t0 + i * 60_000),
+        });
+        seeded.push(id);
+      }
+      const newestFirst = [...seeded].reverse();
+
+      const presenter = shardPresenter(legacyPrisma, newPrisma, [
+        { key: "a", replica: shardA },
+        { key: "b", replica: shardB },
+      ]);
+
+      const forward: string[][] = [];
+      let cursor: string | undefined;
+      for (let guard = 0; guard < 10; guard++) {
+        const page = await presenter.call(
+          baseCall(ctx, { pageSize: 3, direction: "forward", cursor })
+        );
+        forward.push(page.batches.map((b) => b.id));
+        if (!page.pagination.next) break;
+        cursor = page.pagination.next;
+      }
+      expect(forward.flat()).toEqual(newestFirst);
+
+      let backCursor: string | undefined;
+      cursor = undefined;
+      for (let guard = 0; guard < 10; guard++) {
+        const page = await presenter.call(
+          baseCall(ctx, { pageSize: 3, direction: "forward", cursor })
+        );
+        backCursor = page.pagination.previous;
+        if (!page.pagination.next) break;
+        cursor = page.pagination.next;
+      }
+
+      const backward: string[][] = [];
+      for (let guard = 0; guard < 10 && backCursor; guard++) {
+        const page: BatchList = await presenter.call(
+          baseCall(ctx, { pageSize: 3, direction: "backward", cursor: backCursor })
+        );
+        backward.unshift(page.batches.map((b) => b.id));
+        backCursor = page.pagination.previous;
+      }
+
+      const backwardIds = backward.flat();
+      expect(backwardIds).toHaveLength(6);
+      expect(new Set(backwardIds).size).toBe(backwardIds.length);
+      expect(backwardIds).toEqual(newestFirst.slice(0, 6));
+    }
+  );
+
+  twoShardTest(
+    "the empty-state probe reports batches present when only a shard holds batches",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const shardB = shardPrismas[1]!;
+      const ctx = await seedParents(legacyPrisma, "gen2-probe");
+
+      await createBatch(shardB, ctx, {
+        id: generateRunOpsIdV2("b"),
+        friendlyId: "fr_probe_shard",
+      });
+
+      const presenter = shardPresenter(legacyPrisma, newPrisma, [
+        { key: "a", replica: shardPrismas[0]! },
+        { key: "b", replica: shardB },
+      ]);
+
+      const page = await presenter.call(baseCall(ctx, { friendlyId: "fr_does_not_exist" }));
+      expect(page.batches).toHaveLength(0);
+      expect(page.hasAnyBatches).toBe(true);
+
+      const withoutShardLeg = await shardPresenter(legacyPrisma, newPrisma, []).call(
+        baseCall(ctx, { friendlyId: "fr_does_not_exist" })
+      );
+      expect(withoutShardLeg.hasAnyBatches).toBe(false);
+    }
+  );
+
+  twoShardTest(
+    "a duplicated id resolves by precedence: a shard copy outranks new, and new outranks legacy",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const shardA = shardPrismas[0]!;
+      const ctx = await seedParents(legacyPrisma, "gen2-precedence");
+
+      const onBothGenOne = "cmm000000000000000000prec";
+      await createBatch(legacyPrisma, ctx, {
+        id: onBothGenOne,
+        friendlyId: "fr_prec_gen1",
+        status: "PENDING",
+        createdAt: new Date(Date.now() - 60_000),
+      });
+      await createBatch(newPrisma, ctx, {
+        id: onBothGenOne,
+        friendlyId: "fr_prec_gen1",
+        status: "COMPLETED",
+        createdAt: new Date(Date.now() - 60_000),
+      });
+
+      const onNewAndShard = generateRunOpsIdV2("a");
+      await createBatch(newPrisma, ctx, {
+        id: onNewAndShard,
+        friendlyId: "fr_prec_shard",
+        status: "PENDING",
+        createdAt: new Date(Date.now() - 30_000),
+      });
+      await createBatch(shardA, ctx, {
+        id: onNewAndShard,
+        friendlyId: "fr_prec_shard",
+        status: "COMPLETED",
+        createdAt: new Date(Date.now() - 30_000),
+      });
+
+      const page = await shardPresenter(legacyPrisma, newPrisma, [
+        { key: "a", replica: shardA },
+      ]).call(baseCall(ctx, { pageSize: 10 }));
+
+      expect(page.batches.map((b) => b.id).filter((id) => id === onBothGenOne)).toHaveLength(1);
+      expect(page.batches.map((b) => b.id).filter((id) => id === onNewAndShard)).toHaveLength(1);
+      expect(page.batches.find((b) => b.id === onBothGenOne)?.status).toBe("COMPLETED");
+      expect(page.batches.find((b) => b.id === onNewAndShard)?.status).toBe("COMPLETED");
+    }
+  );
+
+  twoShardTest(
+    "an aliased shard contributes no leg, and its rows still arrive once via the aliased store",
+    async ({ legacyPrisma, newPrisma, shardPrismas }) => {
+      const shardB = shardPrismas[1]!;
+      const ctx = await seedParents(legacyPrisma, "gen2-alias");
+
+      const soakId = generateRunOpsIdV2("a");
+      const realShardId = generateRunOpsIdV2("b");
+      await createBatch(newPrisma, ctx, {
+        id: soakId,
+        friendlyId: "fr_soak",
+        createdAt: new Date(Date.now() - 60_000),
+      });
+      await createBatch(shardB, ctx, { id: realShardId, friendlyId: "fr_real_shard" });
+
+      const aliasedSpy = spyClient(newPrisma as unknown as PrismaClient);
+
+      const legs = nonAliasedShardReplicas([
+        { key: "a", replica: aliasedSpy.client as unknown as RunOpsPrismaClient, aliasOf: "new" },
+        { key: "b", replica: shardB },
+      ]);
+      expect(legs.map((leg) => leg.key)).toEqual(["b"]);
+
+      const page = await shardPresenter(
+        legacyPrisma,
+        aliasedSpy.client as unknown as RunOpsPrismaClient,
+        legs
+      ).call(baseCall(ctx, { pageSize: 10 }));
+      expect(page.batches.map((b) => b.id)).toEqual([realShardId, soakId]);
+      expect(aliasedSpy.counts.findMany).toBe(1);
+    }
+  );
+});

apps/webapp/test/deploySettingsRoute.test.ts

@@ -0,0 +1,121 @@
+import { errAsync, okAsync } from "neverthrow";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  authenticateApiKeyWithScope: vi.fn<(...args: any[]) => Promise<any>>(),
+  getDeploySettings: vi.fn<(...args: any[]) => any>(),
+}));
+
+vi.mock("~/services/apiAuth.server", () => ({
+  authenticateApiKeyWithScope: mocks.authenticateApiKeyWithScope,
+}));
+vi.mock("~/v3/services/deployment.server", () => ({
+  DeploymentService: class {
+    getDeploySettings = mocks.getDeploySettings;
+  },
+}));
+vi.mock("~/services/logger.server", () => ({
+  logger: { info: vi.fn(), debug: vi.fn(), error: vi.fn() },
+}));
+
+import { loader } from "~/routes/api.v1.projects.$projectRef.$env.deploy-settings";
+
+function environment(overrides: Record<string, unknown> = {}) {
+  return {
+    id: "env_1",
+    type: "PRODUCTION",
+    project: { id: "proj_1", externalRef: "proj_ref" },
+    organization: { featureFlags: {} },
+    ...overrides,
+  };
+}
+
+function load(env = "prod", projectRef = "proj_ref") {
+  return loader({
+    request: new Request(
+      `https://app.example.com/api/v1/projects/${projectRef}/${env}/deploy-settings`
+    ),
+    params: { projectRef, env },
+    context: {},
+  });
+}
+
+describe("deploy settings route", () => {
+  beforeEach(() => {
+    mocks.authenticateApiKeyWithScope.mockReset();
+    mocks.getDeploySettings.mockReset();
+    mocks.authenticateApiKeyWithScope.mockResolvedValue({
+      ok: true,
+      authentication: { environment: environment() },
+    });
+    mocks.getDeploySettings.mockReturnValue(
+      okAsync({ buildPath: "depot", buildPathSource: "default" })
+    );
+  });
+
+  it("rejects an unknown env slug before authenticating", async () => {
+    const response = await load("nope");
+    expect(response.status).toBe(400);
+    expect(mocks.authenticateApiKeyWithScope).not.toHaveBeenCalled();
+  });
+
+  it("passes the auth failure through", async () => {
+    mocks.authenticateApiKeyWithScope.mockResolvedValue({
+      ok: false,
+      status: 401,
+      error: "Invalid API key",
+    });
+    const response = await load();
+    expect(response.status).toBe(401);
+    expect(await response.json()).toEqual({ error: "Invalid API key" });
+    expect(mocks.getDeploySettings).not.toHaveBeenCalled();
+  });
+
+  it("maps an environment mismatch to 403", async () => {
+    mocks.getDeploySettings.mockReturnValue(errAsync({ type: "environment_mismatch" }));
+    const response = await load("prod", "proj_other");
+    expect(response.status).toBe(403);
+    expect(mocks.getDeploySettings).toHaveBeenCalledWith(environment(), {
+      projectRef: "proj_other",
+      envSlug: "prod",
+    });
+  });
+
+  it("returns only the build path, resolved for the authenticated environment", async () => {
+    const env = environment({ type: "PREVIEW" });
+    mocks.authenticateApiKeyWithScope.mockResolvedValue({
+      ok: true,
+      authentication: { environment: env },
+    });
+    mocks.getDeploySettings.mockReturnValue(
+      okAsync({ buildPath: "native", buildPathSource: "organization_environment" })
+    );
+
+    const response = await load("preview");
+    expect(response.status).toBe(200);
+    expect(await response.json()).toEqual({ build_path: "native" });
+    expect(mocks.getDeploySettings).toHaveBeenCalledWith(env, {
+      projectRef: "proj_ref",
+      envSlug: "preview",
+    });
+    expect(mocks.authenticateApiKeyWithScope).toHaveBeenCalledWith(expect.any(Request), {
+      action: "read",
+      resource: { type: "deployments" },
+    });
+  });
+
+  it("returns 500 when the global flags cannot be loaded", async () => {
+    mocks.getDeploySettings.mockReturnValue(
+      errAsync({ type: "failed_to_load_global_flags", cause: new Error("db down") })
+    );
+    const response = await load();
+    expect(response.status).toBe(500);
+    expect(await response.json()).toEqual({ error: "Internal Server Error" });
+  });
+
+  it("rethrows a Response thrown by authentication", async () => {
+    const thrown = new Response(null, { status: 429 });
+    mocks.authenticateApiKeyWithScope.mockRejectedValue(thrown);
+    await expect(load()).rejects.toBe(thrown);
+  });
+});

apps/webapp/test/deploymentServiceDeploySettings.test.ts

@@ -0,0 +1,182 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  isBillingConfigured: vi.fn<() => boolean>(),
+  current: vi.fn<() => Record<string, unknown> | undefined>(),
+  flags: vi.fn<() => Promise<Record<string, unknown>>>(),
+}));
+
+vi.mock("~/services/platform.v3.server", async (importOriginal) => ({
+  ...(await importOriginal<object>()),
+  isBillingConfigured: mocks.isBillingConfigured,
+}));
+vi.mock("~/v3/globalFlagsRegistry.server", () => ({
+  globalFlagsRegistry: { current: mocks.current },
+}));
+vi.mock("~/v3/featureFlags.server", async (importOriginal) => ({
+  ...(await importOriginal<object>()),
+  flags: mocks.flags,
+}));
+
+import { DeploymentService } from "~/v3/services/deployment.server";
+
+type EnvType = "DEVELOPMENT" | "PREVIEW" | "STAGING" | "PRODUCTION";
+type EnvSlug = "dev" | "staging" | "prod" | "preview";
+
+const SLUG: Record<EnvType, EnvSlug> = {
+  DEVELOPMENT: "dev",
+  STAGING: "staging",
+  PRODUCTION: "prod",
+  PREVIEW: "preview",
+};
+
+function resolve(
+  type: EnvType,
+  orgFeatureFlags: unknown = {},
+  target = { projectRef: "proj_ref", envSlug: SLUG[type] }
+) {
+  return new DeploymentService().getDeploySettings(
+    {
+      type,
+      project: { externalRef: "proj_ref" },
+      organization: { featureFlags: orgFeatureFlags },
+    } as any,
+    target
+  );
+}
+
+async function path(type: EnvType, orgFeatureFlags: unknown = {}) {
+  const result = await resolve(type, orgFeatureFlags);
+  if (result.isErr()) throw result.error.cause;
+  return [result.value.buildPath, result.value.buildPathSource];
+}
+
+describe("DeploymentService.getDeploySettings", () => {
+  beforeEach(() => {
+    mocks.isBillingConfigured.mockReset().mockReturnValue(true);
+    mocks.current.mockReset().mockReturnValue({});
+    mocks.flags.mockReset().mockResolvedValue({});
+  });
+
+  it("rejects a target that is not the key's project or environment type", async () => {
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    for (const target of [
+      { projectRef: "proj_other", envSlug: "prod" as const },
+      { projectRef: "proj_ref", envSlug: "staging" as const },
+      { projectRef: "proj_ref", envSlug: "preview" as const },
+    ]) {
+      const result = await resolve("PRODUCTION", {}, target);
+      expect(result.isErr() && result.error).toEqual({ type: "environment_mismatch" });
+    }
+    expect(mocks.flags).not.toHaveBeenCalled();
+  });
+
+  it("accepts every environment type on its own slug", async () => {
+    for (const type of ["DEVELOPMENT", "PREVIEW", "STAGING", "PRODUCTION"] as const) {
+      expect(await path(type)).toEqual(["depot", "default"]);
+    }
+  });
+
+  it("is depot when the native build server is unavailable, whatever the flags say", async () => {
+    mocks.isBillingConfigured.mockReturnValue(false);
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    expect(await path("PRODUCTION", { deployBuildPath: "native" })).toEqual([
+      "depot",
+      "unavailable",
+    ]);
+  });
+
+  it("defaults to depot when nothing is set", async () => {
+    expect(await path("PRODUCTION")).toEqual(["depot", "default"]);
+  });
+
+  it("applies the plain global flag to every environment type", async () => {
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    for (const type of ["DEVELOPMENT", "PREVIEW", "STAGING", "PRODUCTION"] as const) {
+      expect(await path(type)).toEqual(["native", "global"]);
+    }
+  });
+
+  it("prefers the global env-type flag over the plain global flag", async () => {
+    mocks.current.mockReturnValue({
+      deployBuildPath: "native",
+      deployBuildPathProduction: "depot",
+    });
+    expect(await path("PRODUCTION")).toEqual(["depot", "global_environment"]);
+    expect(await path("STAGING")).toEqual(["native", "global"]);
+  });
+
+  it("lets the org plain flag beat every global flag", async () => {
+    mocks.current.mockReturnValue({
+      deployBuildPath: "native",
+      deployBuildPathProduction: "native",
+    });
+    expect(await path("PRODUCTION", { deployBuildPath: "depot" })).toEqual([
+      "depot",
+      "organization",
+    ]);
+
+    mocks.current.mockReturnValue({ deployBuildPath: "depot" });
+    expect(await path("PRODUCTION", { deployBuildPath: "native" })).toEqual([
+      "native",
+      "organization",
+    ]);
+  });
+
+  it("prefers the org env-type flag over the org plain flag", async () => {
+    const org = { deployBuildPath: "native", deployBuildPathPreview: "native_local_bundle" };
+    expect(await path("PREVIEW", org)).toEqual(["native_local_bundle", "organization_environment"]);
+    expect(await path("PRODUCTION", org)).toEqual(["native", "organization"]);
+  });
+
+  it("never lets another environment type's key leak", async () => {
+    mocks.current.mockReturnValue({ deployBuildPathStaging: "native" });
+    expect(await path("PRODUCTION", { deployBuildPathPreview: "native" })).toEqual([
+      "depot",
+      "default",
+    ]);
+    expect(await path("DEVELOPMENT", { deployBuildPathProduction: "native" })).toEqual([
+      "depot",
+      "default",
+    ]);
+  });
+
+  it("skips values the schema rejects instead of treating them as depot", async () => {
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    expect(await path("PRODUCTION", { deployBuildPathProduction: "bogus" })).toEqual([
+      "native",
+      "global",
+    ]);
+    expect(
+      await path("PRODUCTION", { deployBuildPathProduction: null, deployBuildPath: 1 })
+    ).toEqual(["native", "global"]);
+  });
+
+  it("tolerates a malformed org flag blob", async () => {
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    for (const blob of [null, undefined, "native", 42, ["native"]]) {
+      expect(await path("PRODUCTION", blob)).toEqual(["native", "global"]);
+    }
+  });
+
+  it("reads the registry snapshot without calling flags()", async () => {
+    mocks.current.mockReturnValue({ deployBuildPath: "native" });
+    await path("PRODUCTION");
+    expect(mocks.flags).not.toHaveBeenCalled();
+  });
+
+  it("falls back to flags() when the registry is cold", async () => {
+    mocks.current.mockReturnValue(undefined);
+    mocks.flags.mockResolvedValue({ deployBuildPath: "native" });
+    expect(await path("PRODUCTION")).toEqual(["native", "global"]);
+    expect(mocks.flags).toHaveBeenCalledTimes(1);
+  });
+
+  it("returns an error when the global flags cannot be loaded", async () => {
+    mocks.current.mockReturnValue(undefined);
+    mocks.flags.mockRejectedValue(new Error("db down"));
+    const result = await resolve("PRODUCTION");
+    expect(result.isErr()).toBe(true);
+    expect(result.isErr() && result.error).toMatchObject({ type: "failed_to_load_global_flags" });
+  });
+});

apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts

@@ -1,16 +1,15 @@
 // Real heterogeneous legacy + new Postgres proof for the alert-hydration TaskRun read.
-// The DB is never mocked. A test-only RunStore wraps two real PostgresRunStore
-// instances and routes findRun by id residency (run-ops id → NEW, cuid → LEGACY),
-// mirroring the sibling routing suite. The ProjectAlertChannel read must stay control-plane.
+// The DB is never mocked. The REAL RoutingRunStore wraps two real PostgresRunStore instances and
+// routes findRun by id residency, mirroring the sibling routing suite. The ProjectAlertChannel
+// read must stay control-plane.
 //
 // The alert env-type read (parentEnvironment?.type ?? type) is resolved via the app
 // ControlPlaneResolver over a control-plane client DISTINCT from the run-ops store, proving the
 // cross-provider inversion. The prior version co-located env + run and masked it.
 import { heteroPostgresTest, postgresTest } from "@internal/testcontainers";
-import { PostgresRunStore } from "@internal/run-store";
-import type { ReadClient, RunStore } from "@internal/run-store";
-import type { Prisma, PrismaClient } from "@trigger.dev/database";
-import { generateRunOpsId, ownerEngine } from "@trigger.dev/core/v3/isomorphic";
+import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import type { PrismaClient } from "@trigger.dev/database";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import { describe, expect } from "vitest";
 import { ControlPlaneCache } from "~/v3/runOpsMigration/controlPlaneCache.server";
 import { ControlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
@@ -28,145 +27,20 @@ function buildControlPlaneResolver(controlPlane: PrismaClient) {
 
 vi.setConfig({ testTimeout: 60_000 });
 
-// Test-only routing store: resolve findRun by id length (27 → NEW, else LEGACY),
-// dropping any forwarded client so each inner store uses its OWN prisma. NOT a mock —
-// real DB I/O against two PostgresRunStore instances.
-class RoutingRunStore implements RunStore {
-  readonly #newStore: PostgresRunStore;
-  readonly #legacyStore: PostgresRunStore;
-
-  constructor(newStore: PostgresRunStore, legacyStore: PostgresRunStore) {
-    this.#newStore = newStore;
-    this.#legacyStore = legacyStore;
-  }
-
-  #resolveById(runId: string): PostgresRunStore {
-    return ownerEngine(runId) === "NEW" ? this.#newStore : this.#legacyStore;
-  }
-
-  #idFromWhere(where: Prisma.TaskRunWhereInput): string | undefined {
-    const id = (where as { id?: unknown }).id;
-    return typeof id === "string" ? id : undefined;
-  }
-
-  async findRun(
-    where: Prisma.TaskRunWhereInput,
-    argsOrClient?: { select?: Prisma.TaskRunSelect; include?: Prisma.TaskRunInclude } | ReadClient,
-    _client?: ReadClient
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(where);
-    if (id !== undefined) {
-      return (this.#resolveById(id).findRun as any)(where, argsOrClient);
-    }
-    const fromNew = await (this.#newStore.findRun as any)(where, argsOrClient);
-    return fromNew ?? (this.#legacyStore.findRun as any)(where, argsOrClient);
-  }
-
-  // The remaining RunStore methods are not exercised here; delegate to NEW to satisfy
-  // the interface.
-  findRunOrThrow(...a: any[]): any {
-    return (this.#newStore.findRunOrThrow as any)(...a);
-  }
-  findRuns(...a: any[]): any {
-    return (this.#newStore.findRuns as any)(...a);
-  }
-  createRun(p: any, tx?: any): any {
-    return this.#resolveById(p.data.id).createRun(p, tx);
-  }
-  createCancelledRun(p: any, tx?: any): any {
-    return this.#resolveById(p.data.id).createCancelledRun(p, tx);
-  }
-  createFailedRun(p: any, tx?: any): any {
-    return this.#resolveById(p.data.id).createFailedRun(p, tx);
-  }
-  updateMetadata(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).updateMetadata as any)(...[runId, ...a]);
-  }
-  startAttempt(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).startAttempt as any)(runId, ...a);
-  }
-  completeAttemptSuccess(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).completeAttemptSuccess as any)(runId, ...a);
-  }
-  recordRetryOutcome(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).recordRetryOutcome as any)(runId, ...a);
-  }
-  requeueRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).requeueRun as any)(runId, ...a);
-  }
-  recordBulkActionMembership(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).recordBulkActionMembership as any)(runId, ...a);
-  }
-  cancelRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).cancelRun as any)(runId, ...a);
-  }
-  failRunPermanently(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).failRunPermanently as any)(runId, ...a);
-  }
-  expireRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).expireRun as any)(runId, ...a);
-  }
-  expireRunsBatch(runIds: string[], ...a: any[]): any {
-    return (this.#resolveById(runIds[0] ?? "").expireRunsBatch as any)(runIds, ...a);
-  }
-  lockRunToWorker(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).lockRunToWorker as any)(runId, ...a);
-  }
-  parkPendingVersion(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).parkPendingVersion as any)(runId, ...a);
-  }
-  promotePendingVersionRuns(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).promotePendingVersionRuns as any)(runId, ...a);
-  }
-  expireParkedRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).expireParkedRun as any)(runId, ...a);
-  }
-  suspendForCheckpoint(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).suspendForCheckpoint as any)(runId, ...a);
-  }
-  resumeFromCheckpoint(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).resumeFromCheckpoint as any)(runId, ...a);
-  }
-  rescheduleRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).rescheduleRun as any)(runId, ...a);
-  }
-  enqueueDelayedRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).enqueueDelayedRun as any)(runId, ...a);
-  }
-  rewriteDebouncedRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).rewriteDebouncedRun as any)(runId, ...a);
-  }
-  clearIdempotencyKey(params: any, tx?: any): any {
-    const runId = params?.byId?.runId ?? "";
-    return this.#resolveById(runId).clearIdempotencyKey(params, tx);
-  }
-  pushTags(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).pushTags as any)(runId, ...a);
-  }
-  pushRealtimeStream(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).pushRealtimeStream as any)(runId, ...a);
-  }
-  finalizeRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).finalizeRun as any)(runId, ...a);
-  }
-  findManyBatchTaskRunItems(...a: any[]): any {
-    return (this.#newStore.findManyBatchTaskRunItems as any)(...a);
-  }
-  findBatchTaskRunItem(...a: any[]): any {
-    return (this.#newStore.findBatchTaskRunItem as any)(...a);
-  }
-  upsertWaitpointTag(...a: any[]): any {
-    return (this.#newStore.upsertWaitpointTag as any)(...a);
-  }
-  findManyWaitpointTags(...a: any[]): any {
-    return (this.#newStore.findManyWaitpointTags as any)(...a);
-  }
-}
+// The alert-hydration TaskRun read runs through the REAL RoutingRunStore over two real
+// PostgresRunStore instances (NEW = PG17, LEGACY = PG14). The DB is never mocked. The router
+// resolves residency from the id shape — a v1 run-ops id (26 chars, version "1" at index 25) to
+// NEW, a 25-char cuid to LEGACY — and never forwards a caller-passed control-plane client into a
+// routed read, so each store uses its OWN prisma.
 
 function buildRoutingStore(prisma17: PrismaClient, prisma14: PrismaClient) {
-  const newStore = new PostgresRunStore({ prisma: prisma17, readOnlyPrisma: prisma17 });
+  const newStore = new PostgresRunStore({
+    prisma: prisma17,
+    readOnlyPrisma: prisma17,
+    schemaVariant: "dedicated",
+  });
   const legacyStore = new PostgresRunStore({ prisma: prisma14, readOnlyPrisma: prisma14 });
-  return new RoutingRunStore(newStore, legacyStore);
+  return new RoutingRunStore({ new: newStore, legacy: legacyStore });
 }
 
 async function seedProject(prisma: PrismaClient, suffix: string) {

apps/webapp/test/redisRealtimeStreams.test.ts

@@ -1521,4 +1521,146 @@ describe("RedisRealtimeStreams", () => {
       await redis.quit();
     }
   );
+
+  redisTest(
+    "startFrom 'latest' skips the backlog and delivers only new records",
+    { timeout: 30_000 },
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+      const redisRealtimeStreams = new RedisRealtimeStreams({ redis: redisOptions });
+
+      const runId = "run_latest_test";
+      const streamId = "latest-stream";
+      const encoder = new TextEncoder();
+
+      const ingest = async (line: string) => {
+        const stream = new ReadableStream({
+          start(controller) {
+            controller.enqueue(encoder.encode(line + "\n"));
+            controller.close();
+          },
+        });
+        await redisRealtimeStreams.ingestData(stream, runId, streamId, "default");
+      };
+
+      await ingest("old-0");
+      await ingest("old-1");
+
+      const abortController = new AbortController();
+      const response = await redisRealtimeStreams.streamResponse(
+        new Request("http://localhost/test"),
+        runId,
+        streamId,
+        abortController.signal,
+        { startFrom: "latest", timeoutInSeconds: 10 }
+      );
+
+      const reader = response.body!.getReader();
+      const decoder = new TextDecoder();
+      const receivedData: string[] = [];
+
+      const readLoop = (async () => {
+        let done = false;
+        while (!done && receivedData.length < 1) {
+          const { value, done: streamDone } = await reader.read().catch(() => ({
+            value: undefined,
+            done: true,
+          }));
+          done = streamDone;
+          if (value) {
+            const events = decoder
+              .decode(value)
+              .split("\n\n")
+              .filter((event) => event.trim());
+            for (const event of events) {
+              for (const l of event.split("\n")) {
+                if (l.startsWith("data: ")) {
+                  const data = l.substring(6).trim();
+                  if (data) receivedData.push(data);
+                }
+              }
+            }
+          }
+        }
+      })();
+
+      await new Promise((resolve) => setTimeout(resolve, 500));
+      await ingest("new-0");
+
+      await readLoop;
+      abortController.abort();
+      reader.releaseLock();
+
+      expect(receivedData).toContain("new-0");
+      expect(receivedData).not.toContain("old-0");
+      expect(receivedData).not.toContain("old-1");
+
+      await redis.del(`stream:${runId}:${streamId}`);
+      await redis.quit();
+    }
+  );
+
+  redisTest(
+    "default start replays the backlog from the beginning",
+    { timeout: 30_000 },
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+      const redisRealtimeStreams = new RedisRealtimeStreams({ redis: redisOptions });
+
+      const runId = "run_beginning_test";
+      const streamId = "beginning-stream";
+      const encoder = new TextEncoder();
+
+      const ingestStream = new ReadableStream({
+        start(controller) {
+          controller.enqueue(encoder.encode("old-0\n"));
+          controller.enqueue(encoder.encode("old-1\n"));
+          controller.close();
+        },
+      });
+      await redisRealtimeStreams.ingestData(ingestStream, runId, streamId, "default");
+
+      const abortController = new AbortController();
+      const response = await redisRealtimeStreams.streamResponse(
+        new Request("http://localhost/test"),
+        runId,
+        streamId,
+        abortController.signal,
+        { timeoutInSeconds: 10 }
+      );
+
+      const reader = response.body!.getReader();
+      const decoder = new TextDecoder();
+      const receivedData: string[] = [];
+
+      let done = false;
+      while (!done && receivedData.length < 2) {
+        const { value, done: streamDone } = await reader.read();
+        done = streamDone;
+        if (value) {
+          const events = decoder
+            .decode(value)
+            .split("\n\n")
+            .filter((event) => event.trim());
+          for (const event of events) {
+            for (const l of event.split("\n")) {
+              if (l.startsWith("data: ")) {
+                const data = l.substring(6).trim();
+                if (data) receivedData.push(data);
+              }
+            }
+          }
+        }
+      }
+
+      abortController.abort();
+      reader.releaseLock();
+
+      expect(receivedData).toContain("old-0");
+      expect(receivedData).toContain("old-1");
+
+      await redis.del(`stream:${runId}:${streamId}`);
+      await redis.quit();
+    }
+  );
 });

apps/webapp/test/runOpsSplitMode.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest";
 import { PostgreSqlContainer } from "@testcontainers/postgresql";
 import {
   computeSplitEnabled,
+  assertShardsRequireSplit,
   assertSplitRealtimeInterlock,
 } from "~/v3/runOpsMigration/splitMode.server";
 import { probeDistinctDatabases } from "~/v3/runOpsMigration/distinctDbSentinel.server";
@@ -155,6 +156,58 @@ describe("assertSplitRealtimeInterlock (pure)", () => {
   });
 });
 
+describe("assertShardsRequireSplit (pure)", () => {
+  const owning = (key: string) => ({
+    key,
+    region: "local",
+    url: `postgres://${key}`,
+    replication: { slotName: `s_${key}`, publicationName: `p_${key}`, originGeneration: 2 },
+  });
+  const aliased = (key: string) => ({ key, region: "local", aliasOf: "new" as const });
+
+  it("allows shards when the split flag is on", () => {
+    expect(() =>
+      assertShardsRequireSplit({ splitFlagEnabled: true, shards: [owning("a")] })
+    ).not.toThrow();
+  });
+
+  it("allows the split flag off when no shard is configured", () => {
+    expect(() => assertShardsRequireSplit({ splitFlagEnabled: false, shards: [] })).not.toThrow();
+  });
+
+  // Shards are only built on the split-on arm of selectRunOpsTopology, so configuring one while
+  // the split flag is off silently drops it: no client, no leg, and any row already resident on
+  // that database disappears from every list with no error.
+  it("refuses to boot when a shard that owns a database is configured but the split flag is off", () => {
+    expect(() =>
+      assertShardsRequireSplit({ splitFlagEnabled: false, shards: [owning("a"), owning("b")] })
+    ).toThrow(/RUN_OPS_SHARDS/);
+  });
+
+  it("names the dropped shards so the operator can see which ones they are", () => {
+    expect(() =>
+      assertShardsRequireSplit({ splitFlagEnabled: false, shards: [owning("a"), owning("b")] })
+    ).toThrow(/a, b/);
+  });
+
+  // An aliased shard owns no database: it shares its target's client by reference, so its rows are
+  // still read with the split off. Refusing to boot for one is a false positive.
+  it("allows an alias-only config with the split flag off", () => {
+    expect(() =>
+      assertShardsRequireSplit({ splitFlagEnabled: false, shards: [aliased("a")] })
+    ).not.toThrow();
+  });
+
+  it("refuses only for the owning shards when the config mixes both", () => {
+    expect(() =>
+      assertShardsRequireSplit({
+        splitFlagEnabled: false,
+        shards: [aliased("a"), owning("b")],
+      })
+    ).toThrow(/shard\(s\) b /);
+  });
+});
+
 describe("distinct-DB sentinel (real Postgres)", () => {
   it("reports NOT distinct when both URLs hit the same physical cluster", async () => {
     const pg = await new PostgreSqlContainer("docker.io/postgres:14").start();

apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts

@@ -1,243 +1,35 @@
 import { heteroPostgresTest } from "@internal/testcontainers";
-import { PostgresRunStore } from "@internal/run-store";
-import type { ReadClient, RunStore } from "@internal/run-store";
-import type { Prisma, PrismaClient } from "@trigger.dev/database";
+import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import type { PrismaClient } from "@trigger.dev/database";
 import { parsePacket } from "@trigger.dev/core/v3";
-import { generateRunOpsId, ownerEngine } from "@trigger.dev/core/v3/isomorphic";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import { setTimeout } from "timers/promises";
 import { describe, expect } from "vitest";
 import { UpdateMetadataService } from "~/services/metadata/updateMetadata.server";
 
 vi.setConfig({ testTimeout: 60_000 });
 
-/**
- * A test-only RunStore that routes residency-bearing operations to one of two
- * inner PostgresRunStore instances (NEW = PG17, LEGACY = PG14) purely by run-id
- * classification — NOT by whatever client the service forwards as `tx`.
- *
- * This is the load-bearing design point: the UpdateMetadataService forwards
- * `this._prisma` as the tx/client to every findRun/updateMetadata call. To prove
- * STORE residency routing (and not the forwarded prisma), this wrapper IGNORES
- * the forwarded client for residency-bearing calls and resolves to its own inner
- * store by id length, then calls the inner store WITHOUT forwarding the outer tx
- * (passes undefined), so the inner PostgresRunStore uses its own prisma17/prisma14.
- *
- * Classification contract (version char): a v1 id (26 chars, version "1" at index 25) => NEW store;
- * 25-char cuid => LEGACY store.
- */
-class RoutingRunStore implements RunStore {
-  readonly #newStore: PostgresRunStore;
-  readonly #legacyStore: PostgresRunStore;
-
-  constructor(newStore: PostgresRunStore, legacyStore: PostgresRunStore) {
-    this.#newStore = newStore;
-    this.#legacyStore = legacyStore;
-  }
-
-  // Resolve by the version char: a v1 body => NEW, otherwise LEGACY (25-char cuid).
-  #resolveById(runId: string): PostgresRunStore {
-    return ownerEngine(runId) === "NEW" ? this.#newStore : this.#legacyStore;
-  }
-
-  // Extract a classifiable run id from a `where`. Prefers `where.id`; if only a
-  // friendlyId is present the stub does not classify, so the caller falls back
-  // to read-through (try NEW, then LEGACY).
-  #idFromWhere(where: Prisma.TaskRunWhereInput): string | undefined {
-    const id = (where as { id?: unknown }).id;
-    return typeof id === "string" ? id : undefined;
-  }
-
-  // ---- Reads (residency routing; drop forwarded client) ----
-
-  async findRun(
-    where: Prisma.TaskRunWhereInput,
-    argsOrClient?: { select?: Prisma.TaskRunSelect; include?: Prisma.TaskRunInclude } | ReadClient,
-    _client?: ReadClient
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(where);
-    if (id !== undefined) {
-      // Classifiable by id shape — route to the owning store, dropping the
-      // forwarded client so the inner store uses its OWN prisma.
-      return (this.#resolveById(id).findRun as any)(where, argsOrClient);
-    }
-    // Not classifiable (friendlyId-only / other) — read-through: NEW then LEGACY.
-    const fromNew = await (this.#newStore.findRun as any)(where, argsOrClient);
-    if (fromNew) {
-      return fromNew;
-    }
-    return (this.#legacyStore.findRun as any)(where, argsOrClient);
-  }
-
-  async findRunOrThrow(
-    where: Prisma.TaskRunWhereInput,
-    argsOrClient?: { select?: Prisma.TaskRunSelect; include?: Prisma.TaskRunInclude } | ReadClient,
-    _client?: ReadClient
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(where);
-    if (id !== undefined) {
-      return (this.#resolveById(id).findRunOrThrow as any)(where, argsOrClient);
-    }
-    const fromNew = await (this.#newStore.findRun as any)(where, argsOrClient);
-    if (fromNew) {
-      return fromNew;
-    }
-    return (this.#legacyStore.findRunOrThrow as any)(where, argsOrClient);
-  }
-
-  async findRunOnPrimary(
-    where: Prisma.TaskRunWhereInput,
-    args?: { select?: Prisma.TaskRunSelect; include?: Prisma.TaskRunInclude }
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(where);
-    if (id !== undefined) {
-      return (this.#resolveById(id).findRunOnPrimary as any)(where, args);
-    }
-    const fromNew = await (this.#newStore.findRunOnPrimary as any)(where, args);
-    if (fromNew) {
-      return fromNew;
-    }
-    return (this.#legacyStore.findRunOnPrimary as any)(where, args);
-  }
-
-  async findRunOrThrowOnPrimary(
-    where: Prisma.TaskRunWhereInput,
-    args?: { select?: Prisma.TaskRunSelect; include?: Prisma.TaskRunInclude }
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(where);
-    if (id !== undefined) {
-      return (this.#resolveById(id).findRunOrThrowOnPrimary as any)(where, args);
-    }
-    const fromNew = await (this.#newStore.findRunOnPrimary as any)(where, args);
-    if (fromNew) {
-      return fromNew;
-    }
-    return (this.#legacyStore.findRunOrThrowOnPrimary as any)(where, args);
-  }
-
-  async findRuns(
-    args: { where: Prisma.TaskRunWhereInput },
-    _client?: ReadClient
-  ): Promise<unknown> {
-    const id = this.#idFromWhere(args.where);
-    if (id !== undefined) {
-      return (this.#resolveById(id).findRuns as any)(args);
-    }
-    // Read-through across both stores, NEW first.
-    const fromNew = (await (this.#newStore.findRuns as any)(args)) as unknown[];
-    const fromLegacy = (await (this.#legacyStore.findRuns as any)(args)) as unknown[];
-    return [...fromNew, ...fromLegacy];
-  }
-
-  // ---- Field touches (residency routing; drop forwarded tx) ----
-
-  async updateMetadata(
-    runId: string,
-    data: Parameters<RunStore["updateMetadata"]>[1],
-    options: Parameters<RunStore["updateMetadata"]>[2],
-    _tx?: unknown
-  ): Promise<{ count: number }> {
-    // Route by run id, dropping the forwarded tx so the inner store writes to
-    // its OWN prisma — this is what proves the CAS targets the owning store.
-    return this.#resolveById(runId).updateMetadata(runId, data, options);
-  }
-
-  // ---- Everything else: delegate by run id to satisfy the RunStore interface;
-  // not exercised by these tests. ----
-
-  createRun(params: any, _tx?: unknown): any {
-    return this.#resolveById(params.data.id).createRun(params);
-  }
-  createCancelledRun(params: any, _tx?: unknown): any {
-    return this.#resolveById(params.data.id).createCancelledRun(params);
-  }
-  createFailedRun(params: any, _tx?: unknown): any {
-    return this.#resolveById(params.data.id).createFailedRun(params);
-  }
-  startAttempt(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).startAttempt as any)(runId, data, args);
-  }
-  completeAttemptSuccess(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).completeAttemptSuccess as any)(runId, data, args);
-  }
-  recordRetryOutcome(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).recordRetryOutcome as any)(runId, data, args);
-  }
-  requeueRun(runId: string, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).requeueRun as any)(runId, args);
-  }
-  recordBulkActionMembership(runId: string, bulkActionId: string, _tx?: unknown): any {
-    return this.#resolveById(runId).recordBulkActionMembership(runId, bulkActionId);
-  }
-  cancelRun(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).cancelRun as any)(runId, data, args);
-  }
-  failRunPermanently(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).failRunPermanently as any)(runId, data, args);
-  }
-  expireRun(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).expireRun as any)(runId, data, args);
-  }
-  expireRunsBatch(runIds: string[], data: any, _tx?: unknown): any {
-    return this.#resolveById(runIds[0] ?? "").expireRunsBatch(runIds, data);
-  }
-  lockRunToWorker(runId: string, data: any, _tx?: unknown): any {
-    return this.#resolveById(runId).lockRunToWorker(runId, data);
-  }
-  parkPendingVersion(runId: string, data: any, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).parkPendingVersion as any)(runId, data, args);
-  }
-  promotePendingVersionRuns(runId: string, args?: any, _tx?: unknown): any {
-    return this.#resolveById(runId).promotePendingVersionRuns(runId, args);
-  }
-  expireParkedRun(runId: string, data: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).expireParkedRun as any)(runId, data);
-  }
-  suspendForCheckpoint(runId: string, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).suspendForCheckpoint as any)(runId, args);
-  }
-  resumeFromCheckpoint(runId: string, args: any, _tx?: unknown): any {
-    return (this.#resolveById(runId).resumeFromCheckpoint as any)(runId, args);
-  }
-  rescheduleRun(runId: string, data: any, _tx?: unknown): any {
-    return this.#resolveById(runId).rescheduleRun(runId, data);
-  }
-  enqueueDelayedRun(runId: string, data: any, _tx?: unknown): any {
-    return this.#resolveById(runId).enqueueDelayedRun(runId, data);
-  }
-  rewriteDebouncedRun(runId: string, data: any, _tx?: unknown): any {
-    return this.#resolveById(runId).rewriteDebouncedRun(runId, data);
-  }
-  clearIdempotencyKey(params: any, _tx?: unknown): any {
-    const runId = params?.byId?.runId ?? "";
-    return this.#resolveById(runId).clearIdempotencyKey(params);
-  }
-  pushTags(runId: string, tags: string[], where: any, _tx?: unknown): any {
-    return this.#resolveById(runId).pushTags(runId, tags, where);
-  }
-  pushRealtimeStream(runId: string, streamId: string, _tx?: unknown): any {
-    return this.#resolveById(runId).pushRealtimeStream(runId, streamId);
-  }
-  finalizeRun(runId: string, ...a: any[]): any {
-    return (this.#resolveById(runId).finalizeRun as any)(runId, ...a);
-  }
-  findManyBatchTaskRunItems(...a: any[]): any {
-    return (this.#newStore.findManyBatchTaskRunItems as any)(...a);
-  }
-  findBatchTaskRunItem(...a: any[]): any {
-    return (this.#newStore.findBatchTaskRunItem as any)(...a);
-  }
-  upsertWaitpointTag(...a: any[]): any {
-    return (this.#newStore.upsertWaitpointTag as any)(...a);
-  }
-  findManyWaitpointTags(...a: any[]): any {
-    return (this.#newStore.findManyWaitpointTags as any)(...a);
-  }
-}
+// Real heterogeneous NEW + LEGACY Postgres proof for UpdateMetadataService, exercising the REAL
+// RoutingRunStore over two real PostgresRunStore instances (NEW = PG17, LEGACY = PG14). The DB is
+// never mocked.
+//
+// The load-bearing design point: UpdateMetadataService forwards `this._prisma` as the tx/client to
+// every findRun/updateMetadata call. That client is bound to the control plane — the wrong database
+// for a run resident on either store — so the router must never forward it verbatim. It does not:
+// a non-replica client escalates to the OWNING store's own primary, so residency routing is proved
+// rather than the forwarded prisma.
+//
+// Residency comes from the id shape: a v1 run-ops id (26 chars, version "1" at index 25) resolves
+// to NEW, a 25-char cuid to LEGACY.
 
 function buildRoutingStore(prisma17: PrismaClient, prisma14: PrismaClient) {
-  const newStore = new PostgresRunStore({ prisma: prisma17, readOnlyPrisma: prisma17 });
+  const newStore = new PostgresRunStore({
+    prisma: prisma17,
+    readOnlyPrisma: prisma17,
+    schemaVariant: "dedicated",
+  });
   const legacyStore = new PostgresRunStore({ prisma: prisma14, readOnlyPrisma: prisma14 });
-  return new RoutingRunStore(newStore, legacyStore);
+  return new RoutingRunStore({ new: newStore, legacy: legacyStore });
 }
 
 // 25-char cuid-format id (starts with "c"), no v1 version marker.
@@ -466,8 +258,8 @@ describe("UpdateMetadataService store routing (hetero)", () => {
         logLevel: "error",
       });
 
-      // Call WITHOUT an environment arg, so the `where` is just `{ id: runId }` and
-      // the router classifies by id length (25 => LEGACY).
+      // Call WITHOUT an environment arg, so the `where` is just `{ id: runId }` and the router
+      // resolves residency from the id shape (a 25-char cuid is not a v1 body => LEGACY).
       const result = await service.call(runId, {
         operations: [{ type: "set", key: "x", value: 1 }],
       });

apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts

@@ -1,6 +1,6 @@
 import { heteroPostgresTest } from "@internal/testcontainers";
 import { PrismaClient } from "@trigger.dev/database";
-import { describe, expect, vi } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
   probeDistinctDatabases,
   probeDistinctStores,
@@ -66,6 +66,78 @@ describe("probeDistinctDatabases", () => {
   );
 });
 
+// A transient failure on ONE store must not refuse the boot fleet-wide. The probe fails closed,
+// so an unretried blip on any shard collapses the whole deployment to single-DB and the boot
+// interlock then throws. Retry a bounded number of times, then fail closed exactly as before.
+describe("probeDistinctStores bounded retry", () => {
+  const fp = (sysId: string, db: string) => ({ systemIdentifier: sysId, databaseName: db });
+
+  it("recovers when a transient failure clears within the retry budget", async () => {
+    let calls = 0;
+    const readFingerprint = vi.fn(async (url: string) => {
+      calls++;
+      if (calls === 2) throw new Error("ECONNREFUSED");
+      return fp("sys", url);
+    });
+    const result = await probeDistinctStores(
+      [
+        { id: "new", url: "a" },
+        { id: "shard-a", url: "b" },
+      ],
+      { readFingerprint, attempts: 3, sleep: async () => {} }
+    );
+    expect(result).toEqual({ distinct: true });
+  });
+
+  it("fails closed once the retry budget is exhausted", async () => {
+    const readFingerprint = vi.fn(async () => {
+      throw new Error("ECONNREFUSED");
+    });
+    const result = await probeDistinctStores(
+      [
+        { id: "new", url: "a" },
+        { id: "shard-a", url: "b" },
+      ],
+      { readFingerprint, attempts: 3, sleep: async () => {} }
+    );
+    expect(result).toMatchObject({ distinct: false });
+  });
+
+  it("bounds the attempts it makes", async () => {
+    const readFingerprint = vi.fn(async () => {
+      throw new Error("ECONNREFUSED");
+    });
+    await probeDistinctStores(
+      [
+        { id: "a", url: "a" },
+        { id: "b", url: "b" },
+      ],
+      {
+        readFingerprint,
+        attempts: 3,
+        sleep: async () => {},
+      }
+    );
+    // 2 targets x 3 attempts each, and no more.
+    expect(readFingerprint).toHaveBeenCalledTimes(6);
+  });
+
+  // A duplicate is a correct, final answer. Retrying it would delay every boot of a genuinely
+  // misconfigured deployment for no benefit.
+  it("does not retry a genuine duplicate", async () => {
+    const readFingerprint = vi.fn(async () => fp("sys", "same"));
+    const result = await probeDistinctStores(
+      [
+        { id: "new", url: "a" },
+        { id: "shard-a", url: "b" },
+      ],
+      { readFingerprint, attempts: 3, sleep: async () => {} }
+    );
+    expect(result).toMatchObject({ distinct: false });
+    expect(readFingerprint).toHaveBeenCalledTimes(2);
+  });
+});
+
 describe("probeDistinctStores (set uniqueness at N)", () => {
   heteroPostgresTest(
     "reports distinct for two separate physical clusters",

apps/webapp/test/workloadTokenAuthorization.test.ts

@@ -1,5 +1,8 @@
 import { describe, expect, it } from "vitest";
-import { evaluateCreatedAtGate } from "~/v3/services/worker/workloadTokenAuthorization.server";
+import {
+  evaluateCreatedAtGate,
+  runAgeBucket,
+} from "~/v3/services/worker/workloadTokenAuthorization.server";
 
 const cutoff = new Date("2026-07-09T00:00:00.000Z");
 const before = new Date("2026-07-01T00:00:00.000Z");
@@ -24,3 +27,30 @@ describe("evaluateCreatedAtGate", () => {
     expect(result.allow).toBe(true);
   });
 });
+
+describe("runAgeBucket", () => {
+  const now = new Date("2026-08-28T12:00:00.000Z");
+  const agoMs = (ms: number) => new Date(now.getTime() - ms);
+
+  const HOUR = 60 * 60 * 1000;
+  const DAY = 24 * HOUR;
+
+  it.each([
+    [agoMs(0), "lt_1h"],
+    [agoMs(HOUR - 1), "lt_1h"],
+    [agoMs(HOUR), "1h_1d"],
+    [agoMs(DAY - 1), "1h_1d"],
+    [agoMs(DAY), "1d_7d"],
+    [agoMs(7 * DAY - 1), "1d_7d"],
+    [agoMs(7 * DAY), "7d_30d"],
+    [agoMs(30 * DAY - 1), "7d_30d"],
+    [agoMs(30 * DAY), "gt_30d"],
+    [agoMs(365 * DAY), "gt_30d"],
+  ])("buckets %s as %s", (createdAt, expected) => {
+    expect(runAgeBucket(createdAt, now)).toBe(expected);
+  });
+
+  it("puts a future createdAt in the youngest bucket rather than throwing", () => {
+    expect(runAgeBucket(new Date(now.getTime() + DAY), now)).toBe("lt_1h");
+  });
+});

apps/webapp/vitest.config.ts

@@ -17,6 +17,7 @@ export default defineConfig({
       "app/v3/services/bulk/**/*.test.ts",
       "app/runEngine/concerns/**/*.test.ts",
       "app/runEngine/services/**/*.test.ts",
+      "app/services/realtime/**/*.test.ts",
       "app/utils/**/*.test.ts",
       "app/components/code/**/*.test.ts",
       "app/components/runs/**/*.test.ts",

docs/ai-chat/reference.mdx

@@ -490,7 +490,7 @@ Options for [`chat.headStart()`](/ai-chat/fast-starts#head-start), the warm-serv
 | `agentId`              | `string`                                                | required    | The `chat.agent` / `chat.customAgent` id to hand off to                                              |
 | `run`                  | `(args: HeadStartRunArgs) => Promise<StreamTextResult>` | required    | First-turn callback. Call `streamText` and spread `chat.toStreamTextOptions({ tools })`              |
 | `idleTimeoutInSeconds` | `number`                                                | `60`        | How long the agent waits for the handover signal                                                     |
-| `triggerConfig`        | `Partial<SessionTriggerConfig>`                         | `undefined` | Run options (tags, queue, machine, maxAttempts, maxDuration, region, lockToVersion) for the auto-triggered handover-prepare run. The `chat:{chatId}` tag is prepended automatically |
+| `triggerConfig`        | `Partial<SessionTriggerConfig>`                         | `undefined` | Run options (tags, queue, machine, maxAttempts, maxDuration, region, lockToVersion) for the auto-triggered handover-prepare run. The `chat:{chatId}` tag is prepended automatically and counts toward the 10-tag limit |
 
 `chat.headStart(options)` returns the handler `(req: Request) => Promise<Response>`. The `run` callback receives `HeadStartRunArgs`: `{ messages: UIMessage[], signal: AbortSignal, chat: HeadStartChatHelper }`, where the helper exposes `chat.toStreamTextOptions({ tools })` and a `chat.session` escape hatch. See [Head Start](/ai-chat/fast-starts#head-start) for the full guide.
 

docs/ai-chat/sessions.mdx

@@ -111,7 +111,7 @@ const { id, runId, publicAccessToken, isCached } = await sessions.start({
 | `type` | `string` | Free-form discriminator. `chat.agent` uses `"chat.agent"`. |
 | `externalId` | `string?` | Your stable identity. Cannot start with `session_` (reserved). |
 | `taskIdentifier` | `string` | Task this session triggers runs against. |
-| `triggerConfig` | `SessionTriggerConfig` | Trigger options applied to every run: `tags`, `queue`, `machine`, `maxAttempts`, `idleTimeoutInSeconds`, `basePayload`. |
+| `triggerConfig` | `SessionTriggerConfig` | Trigger options applied to every run: `tags` (up to 10, same as [run tags](/tags); the chat helpers such as `chat.createStartSessionAction` and `AgentChat` add a `chat:{chatId}` tag themselves, which uses one slot. Direct `sessions.start` callers get all 10 and must add any chat tag themselves), `queue`, `machine`, `maxAttempts`, `idleTimeoutInSeconds`, `basePayload`. |
 | `tags` | `string[]?` | Up to 10 tags on the Session row (separate from `triggerConfig.tags`). |
 | `metadata` | `Record<string, unknown>?` | Arbitrary JSON. |
 | `expiresAt` | `Date?` | Hard retention deadline. |

docs/ai-chat/side-channels.mdx

@@ -0,0 +1,172 @@
+---
+title: "Side channels"
+sidebarTitle: "Side channels"
+description: "Named, durable stream pairs on a Session, separate from the chat transcript. A side channel outlives a single run, is shared across runs, and its input does not wake a run."
+---
+
+**A side channel is a named `.in`/`.out` stream pair on a [Session](/ai-chat/sessions), separate from the reserved chat transcript.** Like the transcript it is durable and cross-run, but it is addressed by a name, and writing its `.in` does not wake or trigger a run.
+
+Side channels are a Session primitive, not a chat feature. Any Session can carry them: a `chat.agent`, a task-bound Session, or an external process holding your secret key. Use one to stream out-of-band data alongside (or instead of) a transcript: a feed of browser screenshots, progress telemetry, or a control channel the client writes to. Many clients can read the channel live while a run, or your backend, produces it.
+
+```mermaid
+flowchart LR
+  A["chat.agent run"] -- "frames" --> OUT([channel .out])
+  OUT --> C[Browser clients]
+  C -- "control (pause, viewport)" --> IN([channel .in])
+  IN -. "observed, no run wake" .-> A
+```
+
+## Define the channel once
+
+Declare the channel's record types in one shared module with `sessions.defineChannel`, then import it on both the producer and the consumer so the types line up.
+
+```ts /trigger/channels.ts
+import { sessions } from "@trigger.dev/sdk";
+
+export type ScreenshotFrame = { url: string; step: number };
+export type ViewportControl = { paused: boolean };
+
+export const screenshots = sessions.defineChannel<{
+  out: ScreenshotFrame;
+  in: ViewportControl;
+}>("screenshots");
+```
+
+## Produce on `.out` from a chat.agent
+
+Inside a `chat.agent` run, `chat.channel(...)` opens a channel on the current run's Session. Writing `.out` is durable and cross-run, and wakes nothing. The client control arrives on `.in.on(...)` without waking a run:
+
+```ts /trigger/browser-agent.ts
+import { chat } from "@trigger.dev/sdk/ai";
+import { streamText } from "ai";
+import { screenshots } from "./channels";
+
+export const browserAgent = chat.agent({
+  id: "browser-agent",
+  run: async ({ messages, signal }) => {
+    const frames = chat.channel(screenshots);
+
+    frames.in.on((control) => setPaused(control.paused)); // control: ViewportControl
+
+    driveBrowser({
+      signal,
+      onFrame: (frame) => frames.out.append(frame), // frame: ScreenshotFrame
+    });
+
+    return streamText({ model, messages, abortSignal: signal }); // transcript, as usual
+  },
+});
+```
+
+<Note>
+  A side channel's `.in` is subscribe-only from the run's side (`.on` / `.once` / `.peek`). `.wait()`
+  is not supported on a named channel, because a side channel never suspends or wakes a run.
+</Note>
+
+## From a task or your backend
+
+Nothing here needs a `chat.agent`. Open a channel on any Session by id with `sessions.open(sessionId).channel(...)`; the handle exposes the same `.out` (`append` / `pipe` / `writer`) and `.in` (`send` / `on` / `once` / `peek`) surface as the reserved pair. Create the Session with [`sessions.start`](/ai-chat/sessions) bound to any task, then produce from that task's run:
+
+```ts /trigger/render-frames.ts
+import { sessions, task } from "@trigger.dev/sdk";
+import { screenshots } from "./channels";
+
+export const renderFrames = task({
+  id: "render-frames",
+  run: async (payload: { sessionId: string; steps: number }) => {
+    const frames = sessions.open(payload.sessionId).channel(screenshots);
+    for (let step = 1; step <= payload.steps; step++) {
+      frames.in.on((control) => setPaused(control.paused));
+      await frames.out.append({ url: await renderStep(step), step });
+    }
+  },
+});
+```
+
+Or produce from your own backend, which holds the secret key that `.out` writes require:
+
+```ts Your backend code
+import { sessions } from "@trigger.dev/sdk";
+import { screenshots } from "./trigger/channels";
+
+await sessions.open(sessionId).channel(screenshots).out.append({ url, step });
+```
+
+Either way the client reads the channel the same way, below.
+
+## Read `.out` in React
+
+`useSessionStreamChannel` reads one side of a channel and updates a `records` array. Pass the channel definition as the type argument so `records` is typed from it. `from: "latest"` with `maxRecords: 1` gives a live "latest frame" view with bounded memory:
+
+```tsx app/components/Screencast.tsx
+"use client";
+
+import { useSessionStreamChannel } from "@trigger.dev/react-hooks";
+import type { screenshots } from "../trigger/channels";
+
+export function Screencast({ sessionId, accessToken }: { sessionId: string; accessToken: string }) {
+  const { records } = useSessionStreamChannel<typeof screenshots>("screenshots", {
+    sessionId,
+    accessToken,
+    io: "out",
+    from: "latest",
+    maxRecords: 1,
+  });
+
+  const latest = records[0]; // ScreenshotFrame | undefined
+  return latest ? <img src={latest.url} alt={`frame ${latest.step}`} /> : <p>Waiting…</p>;
+}
+```
+
+`useSessionStreamChannel` has the same options and return shape as [`useSessionStream`](/realtime/react-hooks/session-stream) (`io`, `from`, `maxRecords`, `lastEventId`, `onRecords`, `onControl`, `throttleInMs`, `timeoutInSeconds`), plus the typed channel generic. A bare name string works without the generic, with `records` typed `unknown`.
+
+The client writes the `.in` control with a session handle: `sessions.open(sessionId).channel(screenshots).in.send({ paused: true })`. This appends to the channel and does not wake a run.
+
+## From MCP
+
+An MCP client can read and write a session's channels with two [MCP tools](/mcp-tools): `read_session_channel` drains a channel's records (with an optional `timeoutInSeconds` to wait for the next one), and `write_session_channel` appends a record to a channel's `.in` to send control input to a running agent. Reading `.out` gives the producer feed (e.g. the screencast); writing `.in` does not wake a run, and `.out` stays producer-only.
+
+## Retention
+
+A side channel's streams are bounded by the same retention as the rest of your realtime streams: streams are created on demand when first written and age out on your plan's retention window, with empty streams cleaned up automatically. A channel needs no separate setup or trimming.
+
+<Warning>
+  Records are capped at ~1 MiB each. Stream a pointer, not bytes: write large payloads (a screenshot
+  PNG) to object storage and put the URL on the channel. A base64 image inflates ~33% and will exceed
+  the cap. Pointers also keep the channel small.
+</Warning>
+
+## Auth
+
+A side channel is covered by the session's public access token: a token scoped to `read:sessions:{id}` / `write:sessions:{id}` grants every channel of that session. Mint a narrower token scoped to a single channel with `read:sessions:{id}:channels:{name}`. Writing a channel's `.out` requires secret-key auth (only the agent run), so a browser cannot forge frames; `.in` is writable with the session token. See [Realtime auth](/realtime/auth).
+
+### Scope tokens to the channel, not the whole session
+
+Two properties of the session token are worth designing around when a browser only needs one channel:
+
+- **A session-wide token grants every channel, including ones added later.** `read:sessions:{id}` reads the reserved chat transcript and all named channels. If a client should see only the screencast frames and not the chat, give it `read:sessions:{id}:channels:screencast` instead. The channel-scoped token reads only that channel: it cannot read another channel or the reserved transcript.
+- **A session write token can write the reserved `.in` too, not just a channel's.** `write:sessions:{id}` can send a chat message on the reserved `.in`, so a client meant only to send control input on one channel should hold `write:sessions:{id}:channels:{name}`, which confines it to that channel's `.in`.
+
+```ts Mint a channel-scoped token (your backend)
+import { auth } from "@trigger.dev/sdk";
+
+const token = await auth.createPublicToken({
+  scopes: { read: { sessions: `${sessionId}:channels:screencast` } },
+});
+```
+
+<Note>
+  A session's `externalId` cannot contain `:channels:`, since that is the delimiter the channel scope
+  uses. `sessions.start` rejects it. Any other string, including single colons, is fine.
+</Note>
+
+## Next steps
+
+<CardGroup cols={2}>
+  <Card title="Sessions" icon="layer-group" href="/ai-chat/sessions">
+    The durable, cross-run primitive side channels are built on.
+  </Card>
+  <Card title="Read a session channel in React" icon="react" href="/realtime/react-hooks/session-stream">
+    The `useSessionStream` hook `useSessionStreamChannel` mirrors.
+  </Card>
+</CardGroup>

docs/docs.json

@@ -97,6 +97,7 @@
                   "ai-chat/frontend",
                   "ai-chat/server-chat",
                   "ai-chat/sessions",
+                  "ai-chat/side-channels",
                   "ai-chat/chat-local",
                   "ai-chat/types",
                   "ai-chat/custom-agents",
@@ -217,6 +218,7 @@
                   "realtime/react-hooks/triggering",
                   "realtime/react-hooks/subscribe",
                   "realtime/react-hooks/streams",
+                  "realtime/react-hooks/session-stream",
                   "realtime/react-hooks/swr",
                   "realtime/react-hooks/use-wait-token"
                 ]

docs/mcp-tools.mdx

@@ -271,3 +271,40 @@ Close an agent chat conversation. The agent exits its loop gracefully. Without t
 <Callout type="warning">
   The `start_agent_chat`, `send_agent_message`, and `close_agent_chat` tools are write operations and are not available in readonly mode.
 </Callout>
+
+## Session Channel Tools
+
+Read and write a session's realtime streams: a named [side channel](/ai-chat/side-channels) or the reserved chat transcript pair. Use these to observe an agent's out-of-band output (a screencast, telemetry) or to send it control input.
+
+### read_session_channel
+
+Read records from a session's realtime stream. By default it returns the records that exist right now after an optional cursor and closes, so it is a point-in-time drain, not a live subscription. Set `timeoutInSeconds` to wait for the next record when none exist yet.
+
+**Parameters:**
+- `sessionId` (required): the session id (`session_*`) or the externalId it was created with
+- `channel` (optional): the named side channel to read. Omit to read the reserved chat transcript pair
+- `io` (optional, default: `out`): which side to read, `out` (producer feed) or `in` (client input)
+- `afterEventId` (optional): cursor. Only return records after this event id. Use the `nextCursor` from a prior read to page forward
+- `maxRecords` (optional, default: `100`): maximum records to return
+- `timeoutInSeconds` (optional): wait up to this many seconds for at least one record when none exist yet
+
+**Example usage:**
+- `"Read the latest frames on the screencast channel for this session"`
+- `"Wait for the next control message on the session's status channel"`
+
+### write_session_channel
+
+Append one record to a named side channel's `in` stream. Sends control input to a running agent (e.g. a pause command) without waking or triggering a run. The reserved transcript and a channel's `out` side are not writable here; `out` is producer-only.
+
+**Parameters:**
+- `sessionId` (required): the session id or externalId
+- `channel` (required): the named side channel to write to
+- `value` (required): the record to append. Pass an object for a structured record (e.g. `{ paused: true }`) or a string for a raw one
+
+**Example usage:**
+- `"Pause the screencast on this session"`
+- `"Send { paused: true } to the viewport channel"`
+
+<Callout type="warning">
+  `write_session_channel` is a write operation and is not available in readonly mode.
+</Callout>

docs/realtime/auth.mdx

@@ -135,6 +135,24 @@ When using non-root API keys (recommended), the expiration cannot be more than 3
 
 The format used for a time span is the same as the [jose package](https://github.com/panva/jose), which is a number followed by a unit. Valid units are: "sec", "secs", "second", "seconds", "s", "minute", "minutes", "min", "mins", "m", "hour", "hours", "hr", "hrs", "h", "day", "days", "d", "week", "weeks", "w", "year", "years", "yr", "yrs", and "y". It is not possible to specify months. 365.25 days is used as an alias for a year. If the string is suffixed with "ago", or prefixed with a "-", the resulting time span gets subtracted from the current unix timestamp. A "from now" suffix can also be used for readability when adding to the current unix timestamp.
 
+### Refreshing an expired token
+
+A realtime stream subscription can outlive its token. Pass a `refreshAccessToken` callback and a subscription rejected with a 401/403 re-mints once and reconnects, instead of failing. With no refresher, auth errors stay terminal. Mint the fresh token from your backend, where your secret key lives:
+
+```tsx
+import { useRealtimeStream } from "@trigger.dev/react-hooks";
+
+const { parts } = useRealtimeStream<{ url: string }>(runId, "frames", {
+  accessToken,
+  refreshAccessToken: async () => {
+    const res = await fetch("/api/realtime-token"); // your backend calls auth.createPublicToken
+    return (await res.json()).token;
+  },
+});
+```
+
+`refreshAccessToken` is available on every realtime hook, and on `useApiClient` and `TriggerAuthContext` so hooks under a provider share one refresher.
+
 ### Auto-generated tokens
 
 When you [trigger tasks](/triggering) from your backend, the `handle` received includes a `publicAccessToken` field. This token can be used to authenticate real-time requests in your frontend application.

docs/realtime/react-hooks/session-stream.mdx

@@ -0,0 +1,126 @@
+---
+title: "Read a session channel in React"
+sidebarTitle: "Session streams"
+description: "Subscribe to a session's output or input channel from React with useSessionStream: accumulate records, resume from a cursor, and read only the latest."
+---
+
+**`useSessionStream` subscribes to one channel of a [session](/ai-chat/sessions) and updates a `records` array as new records arrive.** It reads the `out` channel by default (the agent's output) or `in` (the input channel). It is read-only; `useSession` is reserved for two-way (read and write) communication.
+
+<Note>
+  Requires a Public Access Token with the `read:sessions:{id}` scope. See [Realtime
+  auth](/realtime/auth) for generating one.
+</Note>
+
+## Basic usage
+
+Pass the session id (or external id) and an `accessToken`. The hook returns the `records` received so far, the last control record, the cursor of the last record seen, and any error:
+
+```tsx
+"use client";
+
+import { useSessionStream } from "@trigger.dev/react-hooks";
+
+export function SessionViewer({
+  sessionId,
+  accessToken,
+}: {
+  sessionId: string;
+  accessToken: string;
+}) {
+  const { records, error } = useSessionStream<string>(sessionId, { accessToken });
+
+  if (error) return <div>Error: {error.message}</div>;
+
+  return <div>{records.join("")}</div>;
+}
+```
+
+## Options
+
+```tsx
+const { records, lastEventId, lastControl, error, stop } = useSessionStream(sessionId, {
+  accessToken: "pk_...", // Required: public access token with read:sessions:{id}
+  io: "out", // Optional: "out" (default) or "in"
+  from: "beginning", // Optional: "beginning" (default) or "latest"
+  maxRecords: 100, // Optional: keep only the most recent N records (default: unbounded)
+  lastEventId: undefined, // Optional: resume cursor
+  timeoutInSeconds: 60, // Optional: close after this long with no new data (default: 60)
+  throttleInMs: 16, // Optional: throttle record updates (default: 16ms)
+  onRecords: (batch) => {}, // Optional: callback per throttled batch, each with its event id
+  onControl: (event) => {}, // Optional: callback for control records (e.g. turn-complete)
+});
+```
+
+The return value:
+
+- **`records`**: every data record received so far, in arrival order. Control records are delivered to `onControl` instead.
+- **`lastEventId`**: the cursor of the last record seen. Persist it and pass it back as the `lastEventId` option to resume.
+- **`lastControl`**: the last control record (for example `turn-complete`).
+- **`stop`**: abort the subscription, keeping the records received so far.
+
+## Start from the latest record
+
+By default the hook replays the channel history, then live-tails. Pass `from: "latest"` to start at the current tail (the latest record, then live updates) instead of replaying, and `maxRecords` to bound memory:
+
+```tsx
+const { records } = useSessionStream<{ url: string }>(sessionId, {
+  accessToken,
+  io: "out",
+  from: "latest", // start at the latest record, then live updates
+  maxRecords: 1, // keep just the most recent record
+});
+```
+
+<Note>
+  `from: "latest"` requires a server that supports it. Against an older server a client that passes
+  it degrades safely to a full replay.
+</Note>
+
+## Resume from a cursor
+
+The hook resumes automatically across a component remount. A full page reload clears in-memory state, so to resume there, persist the returned `lastEventId` and pass it back on the next load. The channel then continues after that record with no replay and no gap:
+
+```tsx
+const cursorKey = `session-cursor:${sessionId}:out`; // scope the key to this session and channel
+const saved = localStorage.getItem(cursorKey) ?? undefined;
+
+const { records, lastEventId } = useSessionStream<string>(sessionId, {
+  accessToken,
+  lastEventId: saved,
+  onRecords: (batch) => localStorage.setItem(cursorKey, batch.at(-1)!.id),
+});
+```
+
+## React to control records
+
+Control records (such as `turn-complete`) never enter `records`. Handle them with `onControl`, or read the latest from `lastControl`:
+
+```tsx
+const { records, lastControl } = useSessionStream<string>(sessionId, {
+  accessToken,
+  onControl: (event) => {
+    if (event.subtype === "turn-complete") {
+      console.log("The turn is complete");
+    }
+  },
+});
+```
+
+For an expiring token on a long-lived subscription, pass `refreshAccessToken` (see [Realtime auth](/realtime/auth)). To read a session channel outside React, use [`session.out.read()`](/ai-chat/sessions).
+
+## Named side channels
+
+`useSessionStream` reads a session's reserved channel. To read a [named side channel](/ai-chat/side-channels) — a durable, cross-run stream separate from the chat transcript — use `useSessionStreamChannel`. It takes the channel name as its first argument and has the same options and return shape, plus a channel-definition type argument that types `records`:
+
+```tsx
+import { useSessionStreamChannel } from "@trigger.dev/react-hooks";
+import type { screenshots } from "../trigger/channels";
+
+const { records } = useSessionStreamChannel<typeof screenshots>("screenshots", {
+  sessionId,
+  accessToken,
+  io: "out",
+  from: "latest",
+  maxRecords: 1,
+});
+```

docs/realtime/react-hooks/streams.mdx

@@ -130,16 +130,84 @@ export function AIStreamViewer({
 The `useRealtimeStream` hook accepts the following options:
 
 ```tsx
-const { parts, error } = useRealtimeStream(streamOrRunId, streamKeyOrOptions, {
+const { parts, lastEventId, error } = useRealtimeStream(streamOrRunId, streamKeyOrOptions, {
   accessToken: "pk_...", // Required: Public access token
   baseURL: "https://api.trigger.dev", // Optional: Custom API URL
   timeoutInSeconds: 60, // Optional: Timeout (default: 60)
-  startIndex: 0, // Optional: Start from specific chunk
+  from: "beginning", // Optional: "beginning" (default) or "latest"
+  maxParts: 100, // Optional: keep only the most recent N parts (default: unbounded)
+  lastEventId: undefined, // Optional: resume cursor (takes precedence over startIndex)
+  startIndex: 0, // Optional: start from a specific chunk index
   throttleInMs: 16, // Optional: Throttle updates (default: 16ms)
-  onData: (chunk) => {}, // Optional: Callback for each chunk
+  onData: (chunk) => {}, // Optional: callback for each chunk
+  onParts: (batch) => {}, // Optional: callback per throttled batch, each with its event id
+  refreshAccessToken: async () => "pk_...", // Optional: mint a fresh token on expiry
 });
 ```
 
+The hook returns `lastEventId`, the cursor of the last part it received. Persist it and pass it back as the `lastEventId` option to resume later.
+
+### Live view: start from the latest record
+
+By default a subscriber replays the whole stream history, then live-tails. Pass `from: "latest"` to start at the current tail (the latest record, then live updates) instead of replaying, and `maxParts` to keep memory bounded. Together they give a last-value view:
+
+```tsx
+"use client";
+
+import { useRealtimeStream } from "@trigger.dev/react-hooks";
+
+export function LatestFrame({ runId, accessToken }: { runId: string; accessToken: string }) {
+  const { parts } = useRealtimeStream<{ url: string }>(runId, "frames", {
+    accessToken,
+    from: "latest", // start at the latest frame, then live updates
+    maxParts: 1, // keep just the most recent frame
+  });
+
+  const frame = parts.at(-1);
+  return frame ? <img src={frame.url} alt="latest frame" /> : null;
+}
+```
+
+<Note>
+  `from: "latest"` requires a server that supports it. Against an older server a client that passes
+  it degrades safely to a full replay.
+</Note>
+
+### Resume across a page reload
+
+The hook resumes automatically across a component remount. A full page reload clears in-memory
+state, so to resume there, persist the returned `lastEventId` and pass it back on the next load. The
+subscription then continues after that record with no replay and no gap:
+
+```tsx
+const cursorKey = `frames-cursor:${runId}`; // scope the key to this stream
+const saved = localStorage.getItem(cursorKey) ?? undefined;
+
+const { parts, lastEventId } = useRealtimeStream<{ url: string }>(runId, "frames", {
+  accessToken,
+  lastEventId: saved, // resume where the previous session left off
+  onParts: (batch) => localStorage.setItem(cursorKey, batch.at(-1)!.id),
+});
+```
+
+### Refresh an expired access token
+
+Public access tokens are short-lived. For a long-running subscription, pass `refreshAccessToken` to
+mint a fresh token when the server rejects the connection with a 401/403. The subscription re-mints
+once and reconnects; with no refresher, auth errors stay terminal:
+
+```tsx
+const { parts } = useRealtimeStream<{ url: string }>(runId, "frames", {
+  accessToken,
+  refreshAccessToken: async () => {
+    const res = await fetch("/api/realtime-token"); // your backend mints a fresh public token
+    return (await res.json()).token;
+  },
+});
+```
+
+`refreshAccessToken` is also available on [`useApiClient` and `TriggerAuthContext`](/realtime/auth), so every hook under a provider shares one refresher.
+
 ### Using Default Stream
 
 You can omit the stream key to use the default stream:

docs/self-hosting/security.mdx

@@ -15,7 +15,7 @@ We take the security of Trigger.dev seriously, for both Cloud and self-hosted de
 <Steps>
   <Step title="Choose a private channel">
     - **GitHub (preferred):** open a private report from the repository's **Security** tab using **"Report a vulnerability"** ([direct link](https://github.com/triggerdotdev/trigger.dev/security/advisories/new)).
-    - **Email:** `security-advisories@trigger.dev`
+    - **Email:** `security@trigger.dev`
   </Step>
   <Step title="Include the details">
     A description and impact, steps to reproduce (a proof of concept helps), affected versions/components, and any suggested fix.

docs/tasks/streams.mdx

@@ -144,9 +144,12 @@ With options:
 const stream = await aiStream.read(runId, {
   timeoutInSeconds: 60, // Stop if no data for 60 seconds
   startIndex: 10, // Start from the 10th chunk
+  from: "latest", // Or skip history and read only new records from now
 });
 ```
 
+Pass `from: "latest"` to start at the current tail and receive only records appended after the read connects, instead of replaying from the beginning.
+
 #### Appending to a Stream
 
 Use the defined stream's `append()` method to add a single chunk:

hosting/k8s/helm/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v2
 name: trigger
 description: The official Trigger.dev Helm chart
 type: application
-version: 4.5.12
-appVersion: v4.5.12
+version: 4.5.14
+appVersion: v4.5.14
 home: https://trigger.dev
 sources:
   - https://github.com/triggerdotdev/trigger.dev

internal-packages/database/prisma/migrations/20260827120000_add_task_queue_total_concurrency_limit/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "TaskQueue" ADD COLUMN "totalConcurrencyLimit" INTEGER;

internal-packages/database/prisma/schema.prisma

@@ -1974,6 +1974,9 @@ model TaskQueue {
   /// percentage (the source of truth). The absolute concurrencyLimit is materialized from it.
   /// Decimal(5,2) allows fractional percentages like 12.50% (0.01–100.00).
   concurrencyLimitOverridePercent Decimal?  @db.Decimal(5, 2)
+  /// Caps total concurrent runs across ALL concurrencyKey values of this queue
+  /// (concurrencyLimit applies per key value). Null = no total cap.
+  totalConcurrencyLimit           Int?
   rateLimit                       Json?
 
   paused Boolean @default(false)

internal-packages/run-engine/src/engine/index.ts

@@ -209,6 +209,7 @@ export class RunEngine {
       queueSelectionStrategy: new FairQueueSelectionStrategy(queueSelectionStrategyOptions),
       defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10,
       defaultEnvConcurrencyBurstFactor: options.queue?.defaultEnvConcurrencyBurstFactor,
+      totalConcurrencyEnabled: options.queue?.totalConcurrencyEnabled,
       logger: new Logger("RunQueue", options.queue?.logLevel ?? "info"),
       redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
       retryOptions: options.queue?.retryOptions,

internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts

@@ -1126,6 +1126,7 @@ export class RunAttemptSystem {
                   orgId: env.organizationId,
                   projectId: env.project.id,
                   timestamp: retryAt.getTime(),
+                  resetQueueAttempts: !forceRequeue,
                   error: {
                     type: "INTERNAL_ERROR",
                     code: "TASK_RUN_DEQUEUED_MAX_RETRIES",
@@ -1249,6 +1250,7 @@ export class RunAttemptSystem {
     checkpointId,
     completedWaitpoints,
     batchId,
+    resetQueueAttempts = false,
     tx,
   }: {
     run: { id: string };
@@ -1269,6 +1271,12 @@ export class RunAttemptSystem {
       index?: number;
     }[];
     batchId?: string;
+    /**
+     * Pass when the worker reported the attempt's failure itself (an ordinary task retry), so the
+     * queue's redelivery budget is reset rather than consumed. Engine-detected stalls and dequeue
+     * failures leave it unset so a run that never comes back healthy is still bounded.
+     */
+    resetQueueAttempts?: boolean;
   }): Promise<{ wasRequeued: boolean } & ExecutionResult> {
     const prisma = tx ?? this.$.prisma;
 
@@ -1278,6 +1286,7 @@ export class RunAttemptSystem {
         orgId,
         messageId: run.id,
         retryAt: timestamp,
+        resetAttemptCount: resetQueueAttempts,
       });
 
       if (!gotRequeued) {

internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts

@@ -491,6 +491,134 @@ describe("RunEngine attempt failures", () => {
     }
   });
 
+  containerTest(
+    "task retries routed through the queue do not consume the queue's nack budget",
+    async ({ prisma, redisOptions }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+      const engine = new RunEngine({
+        prisma,
+        worker: {
+          redis: redisOptions,
+          workers: 1,
+          tasksPerWorker: 10,
+          pollIntervalMs: 100,
+        },
+        queue: {
+          redis: redisOptions,
+          retryOptions: {
+            maxAttempts: 2,
+          },
+          masterQueueConsumersDisabled: true,
+          processWorkerQueueDebounceMs: 50,
+        },
+        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.0001,
+        },
+        retryWarmStartThresholdMs: 0,
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      try {
+        const taskIdentifier = "test-task";
+        const taskMaxAttempts = 4;
+
+        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier, undefined, {
+          maxAttempts: taskMaxAttempts,
+          factor: 1,
+          minTimeoutInMs: 100,
+          maxTimeoutInMs: 100,
+          randomize: false,
+        });
+
+        const run = await engine.trigger(
+          {
+            number: 1,
+            friendlyId: "run_1234",
+            environment: authenticatedEnvironment,
+            taskIdentifier,
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "t12345",
+            spanId: "s12345",
+            workerQueue: "main",
+            queue: "task/test-task",
+            isTest: false,
+            tags: [],
+          },
+          prisma
+        );
+
+        const error = {
+          type: "BUILT_IN_ERROR" as const,
+          name: "Error",
+          message: "boom",
+          stackTrace: "Error: boom",
+        };
+
+        for (let attempt = 1; attempt <= taskMaxAttempts; attempt++) {
+          await setTimeout(500);
+          await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id);
+          const dequeued = await engine.dequeueFromWorkerQueue({
+            consumerId: "test_12345",
+            workerQueue: "main",
+          });
+          expect(dequeued.length).toBe(1);
+
+          const attemptResult = await engine.startRunAttempt({
+            runId: dequeued[0].run.id,
+            snapshotId: dequeued[0].snapshot.id,
+          });
+          expect(attemptResult.run.attemptNumber).toBe(attempt);
+
+          const result = await engine.completeRunAttempt({
+            runId: dequeued[0].run.id,
+            snapshotId: attemptResult.snapshot.id,
+            completion: {
+              ok: false,
+              id: dequeued[0].run.id,
+              error,
+              retry: {
+                timestamp: Date.now() + 100,
+                delay: 100,
+              },
+            },
+          });
+
+          if (attempt < taskMaxAttempts) {
+            expect(result.attemptStatus).toBe("RETRY_QUEUED");
+            expect(result.run.status).toBe("PENDING");
+          } else {
+            expect(result.attemptStatus).toBe("RUN_FINISHED");
+            expect(result.run.status).toBe("COMPLETED_WITH_ERRORS");
+          }
+        }
+
+        const executionData = await engine.getRunExecutionData({ runId: run.id });
+        assertNonNullable(executionData);
+        expect(executionData.run.attemptNumber).toBe(taskMaxAttempts);
+        expect(executionData.run.status).toBe("COMPLETED_WITH_ERRORS");
+        expect(await engine.runQueue.lengthOfDeadLetterQueue(authenticatedEnvironment)).toBe(0);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
   containerTest("OOM retry on larger machine", async ({ prisma, redisOptions }) => {
     //create environment
     const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

internal-packages/run-engine/src/engine/types.ts

@@ -91,6 +91,8 @@ export type RunEngineOptions = {
     defaultEnvConcurrency?: number;
     defaultEnvConcurrencyBurstFactor?: number;
     logLevel?: LogLevel;
+    /** Enforce per-queue total concurrency limits across concurrency-key variants. See RunQueueOptions.totalConcurrencyEnabled. */
+    totalConcurrencyEnabled?: boolean;
     /** Optional queue-metrics emitter; enables gauge + counter emission from the RunQueue. */
     queueMetrics?: RunQueueMetricsEmitter;
     queueSelectionStrategyOptions?: Pick<

internal-packages/run-engine/src/run-queue/index.ts

@@ -179,6 +179,26 @@ export type RunQueueOptions = {
    * CK operation re-anchors from ckIndex. Default: 86400 (24h).
    */
   counterTtlSeconds?: number;
+  /**
+   * When true, concurrency-keyed queues maintain a per-base-queue groupConcurrency SET
+   * (total in-flight across all key variants) and enforce the queue's total concurrency
+   * limit at admit time. Default false: admit paths are byte-identical to before, and
+   * only the release-side SREM mirror runs (a no-op on an absent set), so the flag can
+   * be flipped on a fleet that has fully rolled onto this build without draining queues.
+   *
+   * Runs already in flight when the flag turns on are not in the set, so a queue can
+   * transiently exceed its total limit by the number of those runs. The excess is
+   * one-time and self-corrects as each pre-flag run completes (its release mirror
+   * no-ops), after which the limit is enforced exactly.
+   *
+   * A release path that misses the group mirror (an instance on an older build during
+   * rollout) leaves the member behind, briefly under-admitting. The dequeue gate
+   * reconciles: when a queue sits at its total, members whose message key no longer
+   * exists are pruned, so such leaks clear within seconds instead of blocking the
+   * queue. Enabling only after every instance runs this build avoids the noise but is
+   * no longer load-bearing for correctness.
+   */
+  totalConcurrencyEnabled?: boolean;
   workerOptions?: {
     pollIntervalMs?: number;
     immediatePollIntervalMs?: number;
@@ -464,6 +484,38 @@ export class RunQueue {
     return result ? Number(result) : undefined;
   }
 
+  public async updateQueueTotalConcurrencyLimits(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string,
+    totalConcurrency: number
+  ) {
+    return this.redis.set(this.keys.queueTotalConcurrencyLimitKey(env, queue), totalConcurrency);
+  }
+
+  public async removeQueueTotalConcurrencyLimits(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string
+  ) {
+    return this.redis.del(this.keys.queueTotalConcurrencyLimitKey(env, queue));
+  }
+
+  public async getQueueTotalConcurrencyLimit(env: MinimalAuthenticatedEnvironment, queue: string) {
+    const result = await this.redis.get(this.keys.queueTotalConcurrencyLimitKey(env, queue));
+
+    return result ? Number(result) : undefined;
+  }
+
+  /**
+   * Total in-flight runs across all concurrency-key variants of a queue (the
+   * groupConcurrency SET cardinality). Admits only populate the set while
+   * totalConcurrencyEnabled is on. After the flag is turned off the set drains
+   * to zero through the release-side mirrors, so a nonzero read reflects real
+   * runs admitted while it was on, never stale state.
+   */
+  public async totalConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) {
+    return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue));
+  }
+
   public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) {
     await this.#callUpdateEnvironmentConcurrencyLimits({
       envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
@@ -1116,12 +1168,19 @@ export class RunQueue {
     messageId,
     retryAt,
     incrementAttemptCount = true,
+    resetAttemptCount = false,
     skipDequeueProcessing = false,
   }: {
     orgId: string;
     messageId: string;
     retryAt?: number;
     incrementAttemptCount?: boolean;
+    /**
+     * Zero the message's attempt counter instead of incrementing it. The counter is the budget
+     * for dequeues that never reach execution; a caller that knows an attempt did execute passes
+     * this so an ordinary task retry cannot exhaust it and dead-letter the run.
+     */
+    resetAttemptCount?: boolean;
     skipDequeueProcessing?: boolean;
   }) {
     return this.#trace(
@@ -1148,7 +1207,9 @@ export class RunQueue {
           [SemanticAttributes.WORKER_QUEUE]: this.#getWorkerQueueFromMessage(message),
         });
 
-        if (incrementAttemptCount) {
+        if (resetAttemptCount) {
+          message.attempt = 0;
+        } else if (incrementAttemptCount) {
           message.attempt = message.attempt + 1;
           if (message.attempt >= maxAttempts) {
             await this.#callMoveToDeadLetterQueue({ message });
@@ -1231,6 +1292,7 @@ export class RunQueue {
             this.keys.envCurrentDequeuedKeyFromQueue(message.queue),
             this.keys.queueRunningCounterKeyFromQueue(message.queue),
             this.keys.ckIndexKeyFromQueue(message.queue),
+            this.keys.queueGroupConcurrencyKeyFromQueue(message.queue),
             messageId,
             this.options.redis.keyPrefix ?? "",
             String(this.counterTtlSeconds)
@@ -2180,6 +2242,11 @@ export class RunQueue {
       const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue);
       const baseQueueKey = this.keys.baseQueueKeyFromQueue(message.queue);
       const ckKeyPrefix = this.options.redis.keyPrefix ?? "";
+      const groupConcurrencyKey = this.keys.queueGroupConcurrencyKeyFromQueue(message.queue);
+      const totalConcurrencyLimitKey = this.keys.queueTotalConcurrencyLimitKeyFromQueue(
+        message.queue
+      );
+      const totalConcurrencyEnabledArg = this.options.totalConcurrencyEnabled ? "1" : "0";
 
       if (ttlInfo) {
         result = await this.redis.enqueueMessageWithTtlCkTracked(
@@ -2200,6 +2267,8 @@ export class RunQueue {
           envConcurrencyLimitBurstFactorKey,
           lengthCounterKey,
           baseQueueKey,
+          groupConcurrencyKey,
+          totalConcurrencyLimitKey,
           // args
           queueName,
           messageId,
@@ -2215,6 +2284,7 @@ export class RunQueue {
           enableFastPathArg,
           ckKeyPrefix,
           String(this.counterTtlSeconds),
+          totalConcurrencyEnabledArg,
           metricsGaugeArg
         );
       } else {
@@ -2235,6 +2305,8 @@ export class RunQueue {
           envConcurrencyLimitBurstFactorKey,
           lengthCounterKey,
           baseQueueKey,
+          groupConcurrencyKey,
+          totalConcurrencyLimitKey,
           // args
           queueName,
           messageId,
@@ -2248,6 +2320,7 @@ export class RunQueue {
           enableFastPathArg,
           ckKeyPrefix,
           String(this.counterTtlSeconds),
+          totalConcurrencyEnabledArg,
           metricsGaugeArg
         );
       }
@@ -2513,6 +2586,8 @@ export class RunQueue {
         ttlQueueKey,
         lengthCounterKey,
         runningCounterKey,
+        this.keys.queueGroupConcurrencyKeyFromQueue(ckWildcardQueue),
+        this.keys.queueTotalConcurrencyLimitKeyFromQueue(ckWildcardQueue),
         //args
         ckWildcardQueue,
         String(Date.now()),
@@ -2520,6 +2595,7 @@ export class RunQueue {
         String(this.options.defaultEnvConcurrencyBurstFactor ?? 1),
         this.options.redis.keyPrefix ?? "",
         String(maxCount),
+        this.options.totalConcurrencyEnabled ? "1" : "0",
         metricsGaugeArg
       );
 
@@ -2769,6 +2845,7 @@ export class RunQueue {
         ckIndexKey,
         lengthCounterKey,
         runningCounterKey,
+        this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue),
         messageId,
         messageQueue,
         messageKeyValue,
@@ -2832,6 +2909,7 @@ export class RunQueue {
         envCurrentDequeuedKey,
         this.keys.queueRunningCounterKeyFromQueue(queue),
         this.keys.ckIndexKeyFromQueue(queue),
+        this.keys.queueGroupConcurrencyKeyFromQueue(queue),
         messageId,
         this.options.redis.keyPrefix ?? "",
         String(this.counterTtlSeconds)
@@ -2898,6 +2976,7 @@ export class RunQueue {
         ckIndexKey,
         lengthCounterKey,
         runningCounterKey,
+        this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue),
         //args
         messageId,
         messageQueue,
@@ -2961,6 +3040,7 @@ export class RunQueue {
         ckIndexKey,
         lengthCounterKey,
         runningCounterKey,
+        this.keys.queueGroupConcurrencyKeyFromQueue(messageQueue),
         messageId,
         messageQueue,
         ckWildcardName
@@ -3749,7 +3829,7 @@ return __qmret(0)
     // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear
     // scripts.
     this.redis.defineCommand("enqueueMessageCkTracked", {
-      numberOfKeys: 15,
+      numberOfKeys: 17,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -3768,6 +3848,9 @@ local envConcurrencyLimitBurstFactorKey = KEYS[13]
 -- Counter keys (KEYS 14-15)
 local lengthCounterKey = KEYS[14]
 local baseQueueKey = KEYS[15]
+-- Total-cap keys (KEYS 16-17)
+local groupConcurrencyKey = KEYS[16]
+local totalConcurrencyLimitKey = KEYS[17]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -3784,6 +3867,7 @@ local enableFastPath = ARGV[10]
 local keyPrefix = ARGV[11]
 -- TTL (seconds) applied to counter lazy-init SETs
 local counterTtl = ARGV[12]
+local totalConcurrencyEnabled = ARGV[13] == '1'
 
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 
@@ -3804,15 +3888,34 @@ if enableFastPath == '1' then
       )
 
       if queueCurrent < queueLimit then
-        redis.call('SET', messageKey, messageData)
-        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
-        redis.call('SADD', envCurrentConcurrencyKey, messageId)
-        redis.call('RPUSH', workerQueueKey, messageKeyValue)
+        -- Total-cap gate: a fast-path admit consumes a group slot, so it must
+        -- respect the env-clamped total limit. At the cap we fall through to the
+        -- slow path (the message queues; the dequeue gate holds it).
+        local totalAllowsFastPath = true
+        if totalConcurrencyEnabled then
+          local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey)
+          if rawTotalLimit then
+            local totalLimit = math.min(tonumber(rawTotalLimit), envLimit)
+            if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then
+              totalAllowsFastPath = false
+            end
+          end
+        end
+
+        if totalAllowsFastPath then
+          redis.call('SET', messageKey, messageData)
+          redis.call('SADD', queueCurrentConcurrencyKey, messageId)
+          redis.call('SADD', envCurrentConcurrencyKey, messageId)
+          if totalConcurrencyEnabled then
+            redis.call('SADD', groupConcurrencyKey, messageId)
+          end
+          redis.call('RPUSH', workerQueueKey, messageKeyValue)
 ${QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA}
-        -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged.
-        -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the
-        -- worker pulls the message from the worker queue.
-        return __qmret(1)
+          -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged.
+          -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the
+          -- worker pulls the message from the worker queue.
+          return __qmret(1)
+        end
       end
     end
   end
@@ -3864,8 +3967,12 @@ if queueName ~= ckWildcardName then
   redis.call('ZREM', masterQueueKey, queueName)
 end
 
--- Update the concurrency keys
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+-- Update the concurrency keys. The groupConcurrency SREM mirrors the per-CK SREM
+-- unconditionally (no flag check) so a disabled flag still drains the group set.
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -3876,7 +3983,7 @@ return __qmret(0)
     });
 
     this.redis.defineCommand("enqueueMessageWithTtlCkTracked", {
-      numberOfKeys: 16,
+      numberOfKeys: 18,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -3896,6 +4003,9 @@ local envConcurrencyLimitBurstFactorKey = KEYS[14]
 -- Counter keys (KEYS 15-16)
 local lengthCounterKey = KEYS[15]
 local baseQueueKey = KEYS[16]
+-- Total-cap keys (KEYS 17-18)
+local groupConcurrencyKey = KEYS[17]
+local totalConcurrencyLimitKey = KEYS[18]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -3914,6 +4024,7 @@ local enableFastPath = ARGV[12]
 local keyPrefix = ARGV[13]
 -- TTL (seconds) applied to counter lazy-init SETs
 local counterTtl = ARGV[14]
+local totalConcurrencyEnabled = ARGV[15] == '1'
 
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 
@@ -3934,12 +4045,29 @@ if enableFastPath == '1' then
       )
 
       if queueCurrent < queueLimit then
-        redis.call('SET', messageKey, messageData)
-        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
-        redis.call('SADD', envCurrentConcurrencyKey, messageId)
-        redis.call('RPUSH', workerQueueKey, messageKeyValue)
+        -- Total-cap gate: see enqueueMessageCkTracked.
+        local totalAllowsFastPath = true
+        if totalConcurrencyEnabled then
+          local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey)
+          if rawTotalLimit then
+            local totalLimit = math.min(tonumber(rawTotalLimit), envLimit)
+            if tonumber(redis.call('SCARD', groupConcurrencyKey) or '0') >= totalLimit then
+              totalAllowsFastPath = false
+            end
+          end
+        end
+
+        if totalAllowsFastPath then
+          redis.call('SET', messageKey, messageData)
+          redis.call('SADD', queueCurrentConcurrencyKey, messageId)
+          redis.call('SADD', envCurrentConcurrencyKey, messageId)
+          if totalConcurrencyEnabled then
+            redis.call('SADD', groupConcurrencyKey, messageId)
+          end
+          redis.call('RPUSH', workerQueueKey, messageKeyValue)
 ${QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA}
-        return __qmret(1)
+          return __qmret(1)
+        end
       end
     end
   end
@@ -3987,8 +4115,12 @@ if queueName ~= ckWildcardName then
   redis.call('ZREM', masterQueueKey, queueName)
 end
 
--- Update the concurrency keys
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+-- Update the concurrency keys. The groupConcurrency SREM mirrors the per-CK SREM
+-- unconditionally (no flag check) so a disabled flag still drains the group set.
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -4171,7 +4303,7 @@ for i, member in ipairs(expiredMembers) do
 
       local concurrencyKey = queueKey .. ":currentConcurrency"
       local dequeuedKey = queueKey .. ":currentDequeued"
-      redis.call('SREM', concurrencyKey, runId)
+      local removedFromCurrent = redis.call('SREM', concurrencyKey, runId)
       local removedFromDequeued = redis.call('SREM', dequeuedKey, runId)
 
       local projMatch = string.match(rawQueueKey, ":proj:([^:]+):env:")
@@ -4191,6 +4323,10 @@ for i, member in ipairs(expiredMembers) do
         if removedFromDequeued == 1 then
           decrFloored(runningCounterKey)
         end
+        -- Mirror the per-CK currentConcurrency SREM into the base groupConcurrency set
+        if removedFromCurrent == 1 then
+          redis.call('SREM', keyPrefix .. ckMatch .. ":groupConcurrency", runId)
+        end
 
         local ckIndexKey = keyPrefix .. ckMatch .. ":ckIndex"
         local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
@@ -4505,7 +4641,7 @@ return results
     // (normal dequeue, TTL-expired, or stale-orphan path — all of which were
     // counted at enqueue time).
     this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", {
-      numberOfKeys: 11,
+      numberOfKeys: 13,
       lua: `
 local ckIndexKey = KEYS[1]
 local queueConcurrencyLimitKey = KEYS[2]
@@ -4518,6 +4654,8 @@ local masterQueueKey = KEYS[8]
 local ttlQueueKey = KEYS[9]
 local lengthCounterKey = KEYS[10]
 local runningCounterKey = KEYS[11]
+local groupConcurrencyKey = KEYS[12]
+local totalConcurrencyLimitKey = KEYS[13]
 
 local ckWildcardName = ARGV[1]
 local currentTime = tonumber(ARGV[2])
@@ -4525,6 +4663,7 @@ local defaultEnvConcurrencyLimit = ARGV[3]
 local defaultEnvConcurrencyBurstFactor = ARGV[4]
 local keyPrefix = ARGV[5]
 local maxCount = tonumber(ARGV[6] or '1')
+local totalConcurrencyEnabled = ARGV[7] == '1'
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 ${QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA}
 
@@ -4549,6 +4688,46 @@ local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurren
 local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency
 local actualMaxCount = math.min(maxCount, envAvailableCapacity)
 
+-- Total-cap gate: bound this batch by the remaining headroom across ALL ck
+-- variants (groupConcurrency SCARD vs the env-clamped total limit). Each admit
+-- below SADDs into the group set and bumps dequeuedCount, and dequeuedCount is
+-- bounded by actualMaxCount, so tightening here is sufficient to prevent
+-- over-admitting past the cap within a single batch.
+if totalConcurrencyEnabled then
+  local rawTotalLimit = redis.call('GET', totalConcurrencyLimitKey)
+  if rawTotalLimit then
+    local totalConcurrencyLimit = math.min(tonumber(rawTotalLimit), envConcurrencyLimit)
+    local groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
+
+    -- Self-heal before holding the queue at its limit. A terminal release path
+    -- that misses the group mirror (an older build, or a future script) leaves
+    -- a member behind, but every terminal path deletes the run's message key,
+    -- so a member with no message key is provably dead. Members of re-queued
+    -- runs keep their message key and clear through the mirrored ack when the
+    -- run completes. The short lock bounds a saturated queue to one pass per
+    -- interval, and SSCAN with a persisted cursor bounds each pass to one
+    -- batch so a large set never blocks Redis for a full traversal; successive
+    -- passes cover the whole set.
+    if groupCurrentConcurrency >= totalConcurrencyLimit then
+      local reconcileLockKey = groupConcurrencyKey .. ':reconcileLock'
+      if redis.call('SET', reconcileLockKey, '1', 'NX', 'EX', '10') then
+        local reconcileCursorKey = groupConcurrencyKey .. ':reconcileCursor'
+        local reconcileCursor = redis.call('GET', reconcileCursorKey) or '0'
+        local scanResult = redis.call('SSCAN', groupConcurrencyKey, reconcileCursor, 'COUNT', '500')
+        redis.call('SET', reconcileCursorKey, scanResult[1], 'EX', '3600')
+        for _, groupMemberId in ipairs(scanResult[2]) do
+          if redis.call('EXISTS', messageKeyPrefix .. groupMemberId) == 0 then
+            redis.call('SREM', groupConcurrencyKey, groupMemberId)
+          end
+        end
+        groupCurrentConcurrency = tonumber(redis.call('SCARD', groupConcurrencyKey) or '0')
+      end
+    end
+
+    actualMaxCount = math.min(actualMaxCount, totalConcurrencyLimit - groupCurrentConcurrency)
+  end
+end
+
 if actualMaxCount <= 0 then
   return __qmret(nil)
 end
@@ -4606,6 +4785,9 @@ for _, ckQueueName in ipairs(ckQueues) do
           decrLengthCounter()
           redis.call('SADD', ckConcurrencyKey, messageId)
           redis.call('SADD', envCurrentConcurrencyKey, messageId)
+          if totalConcurrencyEnabled then
+            redis.call('SADD', groupConcurrencyKey, messageId)
+          end
 
           if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then
             local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '')
@@ -5095,7 +5277,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId)
     // removed something) and runningCounter (when SREM currentDequeued actually
     // removed something).
     this.redis.defineCommand("acknowledgeMessageCkTracked", {
-      numberOfKeys: 12,
+      numberOfKeys: 13,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -5110,6 +5292,7 @@ local workerQueueKey = KEYS[9]
 local ckIndexKey = KEYS[10]
 local lengthCounterKey = KEYS[11]
 local runningCounterKey = KEYS[12]
+local groupConcurrencyKey = KEYS[13]
 
 -- Args:
 local messageId = ARGV[1]
@@ -5162,7 +5345,12 @@ end
 
 -- Update the concurrency keys. DECR runningCounter only when SREM
 -- currentDequeued actually removed an entry (the message was in flight).
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+-- The groupConcurrency SREM mirrors the per-CK SREM so the group set drains
+-- on every release path.
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -5181,7 +5369,7 @@ end
     // runningCounter (floored); ZADD back to the variant zset INCRs
     // lengthCounter only when ZADD reported a new entry.
     this.redis.defineCommand("nackMessageCkTracked", {
-      numberOfKeys: 11,
+      numberOfKeys: 12,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -5195,6 +5383,7 @@ local envQueueKey = KEYS[8]
 local ckIndexKey = KEYS[9]
 local lengthCounterKey = KEYS[10]
 local runningCounterKey = KEYS[11]
+local groupConcurrencyKey = KEYS[12]
 
 -- Args:
 local messageId = ARGV[1]
@@ -5220,7 +5409,11 @@ redis.call('SET', messageKey, messageData)
 -- so we skip the eager lazy-init here (unlike releaseConcurrencyTracked, which
 -- mirrors the same DECR pattern with init). A post-TTL nack's floored DECR
 -- no-ops; the next dequeueMessageFromKeyTracked reseeds from current state.
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+-- The groupConcurrency SREM mirrors the per-CK SREM.
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -5276,7 +5469,7 @@ end
     // Tracked variant: same as moveToDeadLetterQueueCk. ZREM may DECR
     // lengthCounter (defensive); SREM currentDequeued may DECR runningCounter.
     this.redis.defineCommand("moveToDeadLetterQueueCkTracked", {
-      numberOfKeys: 12,
+      numberOfKeys: 13,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -5291,6 +5484,7 @@ local deadLetterQueueKey = KEYS[9]
 local ckIndexKey = KEYS[10]
 local lengthCounterKey = KEYS[11]
 local runningCounterKey = KEYS[12]
+local groupConcurrencyKey = KEYS[13]
 
 -- Args:
 local messageId = ARGV[1]
@@ -5340,8 +5534,12 @@ end
 redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)
 
 -- Update the concurrency keys. DECR runningCounter only when SREM
--- currentDequeued actually removed an entry.
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+-- currentDequeued actually removed an entry. The groupConcurrency SREM mirrors
+-- the per-CK SREM.
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -5376,7 +5574,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId)
     // something. Caller should only invoke this variant for CK queues — non-CK
     // queues should keep calling releaseConcurrency.
     this.redis.defineCommand("releaseConcurrencyTracked", {
-      numberOfKeys: 6,
+      numberOfKeys: 7,
       lua: `
 -- Keys:
 local queueCurrentConcurrencyKey = KEYS[1]
@@ -5385,6 +5583,7 @@ local queueCurrentDequeuedKey = KEYS[3]
 local envCurrentDequeuedKey = KEYS[4]
 local runningCounterKey = KEYS[5]
 local ckIndexKey = KEYS[6]
+local groupConcurrencyKey = KEYS[7]
 
 -- Args:
 local messageId = ARGV[1]
@@ -5406,7 +5605,10 @@ if redis.call('EXISTS', runningCounterKey) == 0 then
   redis.call('SET', runningCounterKey, total, 'EX', counterTtl)
 end
 
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -5521,7 +5723,7 @@ redis.call('SREM', envCurrentDequeuedKey, messageId)
     // Tracked variant of clearMessageFromConcurrencySets — see releaseConcurrencyTracked
     // for the contract. Only invoke for CK queues.
     this.redis.defineCommand("clearMessageFromConcurrencySetsTracked", {
-      numberOfKeys: 6,
+      numberOfKeys: 7,
       lua: `
 -- Keys:
 local queueCurrentConcurrencyKey = KEYS[1]
@@ -5530,6 +5732,7 @@ local queueCurrentDequeuedKey = KEYS[3]
 local envCurrentDequeuedKey = KEYS[4]
 local runningCounterKey = KEYS[5]
 local ckIndexKey = KEYS[6]
+local groupConcurrencyKey = KEYS[7]
 
 -- Args:
 local messageId = ARGV[1]
@@ -5547,7 +5750,10 @@ if redis.call('EXISTS', runningCounterKey) == 0 then
   redis.call('SET', runningCounterKey, total, 'EX', counterTtl)
 end
 
-redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+local removedFromCurrentConcurrency = redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+if removedFromCurrentConcurrency == 1 then
+  redis.call('SREM', groupConcurrencyKey, messageId)
+end
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -5941,6 +6147,8 @@ declare module "@internal/redis" {
       envConcurrencyLimitBurstFactorKey: string,
       lengthCounterKey: string,
       baseQueueKey: string,
+      groupConcurrencyKey: string,
+      totalConcurrencyLimitKey: string,
       queueName: string,
       messageId: string,
       messageData: string,
@@ -5953,6 +6161,7 @@ declare module "@internal/redis" {
       enableFastPath: string,
       keyPrefix: string,
       counterTtl: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[number, number[] | null]>
     ): Result<[number, number[] | null], Context>;
@@ -5974,6 +6183,8 @@ declare module "@internal/redis" {
       envConcurrencyLimitBurstFactorKey: string,
       lengthCounterKey: string,
       baseQueueKey: string,
+      groupConcurrencyKey: string,
+      totalConcurrencyLimitKey: string,
       queueName: string,
       messageId: string,
       messageData: string,
@@ -5988,6 +6199,7 @@ declare module "@internal/redis" {
       enableFastPath: string,
       keyPrefix: string,
       counterTtl: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[number, number[] | null]>
     ): Result<[number, number[] | null], Context>;
@@ -6004,12 +6216,15 @@ declare module "@internal/redis" {
       ttlQueueKey: string,
       lengthCounterKey: string,
       runningCounterKey: string,
+      groupConcurrencyKey: string,
+      totalConcurrencyLimitKey: string,
       ckWildcardName: string,
       currentTime: string,
       defaultEnvConcurrencyLimit: string,
       defaultEnvConcurrencyBurstFactor: string,
       keyPrefix: string,
       maxCount: string,
+      totalConcurrencyEnabled: string,
       metricsEnabled: string,
       callback?: Callback<[string[] | null, number[] | null]>
     ): Result<[string[] | null, number[] | null], Context>;
@@ -6034,6 +6249,7 @@ declare module "@internal/redis" {
       ckIndexKey: string,
       lengthCounterKey: string,
       runningCounterKey: string,
+      groupConcurrencyKey: string,
       messageId: string,
       messageQueueName: string,
       messageKeyValue: string,
@@ -6054,6 +6270,7 @@ declare module "@internal/redis" {
       ckIndexKey: string,
       lengthCounterKey: string,
       runningCounterKey: string,
+      groupConcurrencyKey: string,
       messageId: string,
       messageQueueName: string,
       messageData: string,
@@ -6077,6 +6294,7 @@ declare module "@internal/redis" {
       ckIndexKey: string,
       lengthCounterKey: string,
       runningCounterKey: string,
+      groupConcurrencyKey: string,
       messageId: string,
       messageQueueName: string,
       ckWildcardName: string,
@@ -6102,6 +6320,7 @@ declare module "@internal/redis" {
       envCurrentDequeuedKey: string,
       runningCounterKey: string,
       ckIndexKey: string,
+      groupConcurrencyKey: string,
       messageId: string,
       keyPrefix: string,
       counterTtl: string,
@@ -6115,6 +6334,7 @@ declare module "@internal/redis" {
       envCurrentDequeuedKey: string,
       runningCounterKey: string,
       ckIndexKey: string,
+      groupConcurrencyKey: string,
       messageId: string,
       keyPrefix: string,
       counterTtl: string,

internal-packages/run-engine/src/run-queue/keyProducer.ts

@@ -24,6 +24,8 @@ const constants = {
   CK_INDEX_PART: "ckIndex",
   LENGTH_COUNTER_PART: "lengthCounter",
   RUNNING_COUNTER_PART: "runningCounter",
+  GROUP_CONCURRENCY_PART: "groupConcurrency",
+  TOTAL_CONCURRENCY_LIMIT_PART: "totalConcurrency",
 } as const;
 
 export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
@@ -338,6 +340,32 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
     return `${this.baseQueueKeyFromQueue(queue)}:${constants.RUNNING_COUNTER_PART}`;
   }
 
+  /**
+   * SET of in-flight messageIds across ALL concurrency-key variants of a base queue.
+   * SCARD of this set is the queue's total running count, gated by the total
+   * concurrency limit. Lives at the base queue so every ck variant shares it.
+   */
+  queueGroupConcurrencyKey(env: RunQueueKeyProducerEnvironment, queue: string): string {
+    return `${this.queueKey(env, queue)}:${constants.GROUP_CONCURRENCY_PART}`;
+  }
+
+  queueGroupConcurrencyKeyFromQueue(queue: string): string {
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.GROUP_CONCURRENCY_PART}`;
+  }
+
+  /**
+   * String key holding the queue's total concurrency limit (the cap across all
+   * concurrency-key variants). Absent = no total cap. Readers clamp to the
+   * environment limit; the raw requested value is what's stored.
+   */
+  queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string {
+    return `${this.queueKey(env, queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`;
+  }
+
+  queueTotalConcurrencyLimitKeyFromQueue(queue: string): string {
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`;
+  }
+
   isCkWildcard(queue: string): boolean {
     return queue.endsWith(":ck:*");
   }

internal-packages/run-engine/src/run-queue/tests/nack.test.ts

@@ -130,6 +130,81 @@ describe("RunQueue.nackMessage", () => {
     }
   });
 
+  redisTest(
+    "nacking with resetAttemptCount zeroes the counter instead of dead-lettering",
+    async ({ redisContainer }) => {
+      const queue = new RunQueue({
+        ...testOptions,
+        retryOptions: {
+          ...testOptions.retryOptions,
+          maxAttempts: 2,
+        },
+        queueSelectionStrategy: new FairQueueSelectionStrategy({
+          redis: {
+            keyPrefix: "runqueue:test:",
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+          },
+          keys: testOptions.keys,
+        }),
+        redis: {
+          keyPrefix: "runqueue:test:",
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+        },
+      });
+
+      try {
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: messageDev,
+          workerQueue: authenticatedEnvDev.id,
+        });
+
+        await setTimeout(1000);
+
+        const dequeued = await queue.dequeueMessageFromWorkerQueue(
+          "test_12345",
+          authenticatedEnvDev.id
+        );
+        assertNonNullable(dequeued);
+
+        const first = await queue.nackMessage({
+          orgId: messageDev.orgId,
+          messageId: messageDev.runId,
+        });
+        expect(first).toBe(true);
+
+        const afterFirst = await queue.readMessage(messageDev.orgId, messageDev.runId);
+        expect(afterFirst?.attempt).toBe(1);
+
+        await setTimeout(1000);
+
+        const dequeued2 = await queue.dequeueMessageFromWorkerQueue(
+          "test_12345",
+          authenticatedEnvDev.id
+        );
+        assertNonNullable(dequeued2);
+
+        // A plain nack here would hit maxAttempts and dead-letter the run
+        const second = await queue.nackMessage({
+          orgId: messageDev.orgId,
+          messageId: messageDev.runId,
+          resetAttemptCount: true,
+        });
+        expect(second).toBe(true);
+
+        const afterReset = await queue.readMessage(messageDev.orgId, messageDev.runId);
+        expect(afterReset?.attempt).toBe(0);
+
+        expect(await queue.lengthOfEnvQueue(authenticatedEnvDev)).toBe(1);
+        expect(await queue.lengthOfDeadLetterQueue(authenticatedEnvDev)).toBe(0);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest(
     "nacking a message with maxAttempts reached should be moved to dead letter queue",
     async ({ redisContainer }) => {

internal-packages/run-engine/src/run-queue/tests/totalConcurrency.test.ts

@@ -0,0 +1,412 @@
+import { assertNonNullable, redisTest } from "@internal/testcontainers";
+import { trace } from "@internal/tracing";
+import { setTimeout } from "node:timers/promises";
+import { describe } from "vitest";
+import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js";
+import { RunQueue } from "../index.js";
+import { RunQueueFullKeyProducer } from "../keyProducer.js";
+import type { InputPayload } from "../types.js";
+import { Decimal } from "@trigger.dev/database";
+
+const testOptions = {
+  name: "rq",
+  tracer: trace.getTracer("rq"),
+  workers: 1,
+  defaultEnvConcurrency: 25,
+  retryOptions: {
+    maxAttempts: 5,
+    factor: 1.1,
+    minTimeoutInMs: 100,
+    maxTimeoutInMs: 1_000,
+    randomize: true,
+  },
+  keys: new RunQueueFullKeyProducer(),
+};
+
+const authenticatedEnvDev = {
+  id: "e1234",
+  type: "DEVELOPMENT" as const,
+  maximumConcurrencyLimit: 10,
+  concurrencyLimitBurstFactor: new Decimal(2.0),
+  project: { id: "p1234" },
+  organization: { id: "o1234" },
+};
+
+function createQueue(redisContainer: any, totalConcurrencyEnabled: boolean) {
+  return new RunQueue({
+    ...testOptions,
+    totalConcurrencyEnabled,
+    queueSelectionStrategy: new FairQueueSelectionStrategy({
+      redis: {
+        keyPrefix: "runqueue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      keys: testOptions.keys,
+    }),
+    redis: {
+      keyPrefix: "runqueue:test:",
+      host: redisContainer.getHost(),
+      port: redisContainer.getPort(),
+    },
+  });
+}
+
+function makeMessage(overrides: Partial<InputPayload> = {}): InputPayload {
+  return {
+    runId: "r1",
+    taskIdentifier: "task/my-task",
+    orgId: "o1234",
+    projectId: "p1234",
+    environmentId: "e1234",
+    environmentType: "DEVELOPMENT",
+    queue: "task/my-task",
+    timestamp: Date.now(),
+    attempt: 0,
+    ...overrides,
+  };
+}
+
+async function waitFor(condition: () => Promise<boolean>, timeoutMs = 20_000): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await condition()) {
+      return true;
+    }
+    await setTimeout(250);
+  }
+  return condition();
+}
+
+vi.setConfig({ testTimeout: 60_000 });
+
+describe("RunQueue total concurrency limit", () => {
+  redisTest(
+    "caps in-flight runs across concurrency keys at the total limit",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 2);
+
+        const now = Date.now();
+        for (const [i, ck] of ["ck-a", "ck-a", "ck-b", "ck-b"].entries()) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `r${i}`,
+              concurrencyKey: ck,
+              timestamp: now - 1000 + i,
+            }),
+            workerQueue: "main",
+          });
+        }
+
+        const admittedTwo = await waitFor(
+          async () =>
+            (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 2
+        );
+        expect(admittedTwo).toBe(true);
+
+        /**
+         * The remaining two messages must stay queued: give the master consumers a
+         * couple of extra polling cycles to prove the gate holds, not just that it
+         * hadn't caught up yet.
+         */
+        await setTimeout(2000);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(2);
+        expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(2);
+
+        const dequeued1 = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
+        assertNonNullable(dequeued1);
+        const dequeued2 = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
+        assertNonNullable(dequeued2);
+
+        await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, dequeued1.messageId);
+
+        const thirdAdmitted = await waitFor(async () => {
+          const total = await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task");
+          const queued = await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task");
+          return total === 2 && queued === 1;
+        });
+        expect(thirdAdmitted).toBe(true);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "still enforces the per-key limit under the total limit",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 10);
+
+        const now = Date.now();
+        for (const i of [0, 1]) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `r${i}`,
+              concurrencyKey: "ck-a",
+              timestamp: now - 1000 + i,
+            }),
+            workerQueue: "main",
+          });
+        }
+
+        const oneAdmitted = await waitFor(
+          async () =>
+            (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1
+        );
+        expect(oneAdmitted).toBe(true);
+
+        await setTimeout(2000);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+        expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+        expect(
+          await queue.currentConcurrencyOfQueue(authenticatedEnvDev, "task/my-task", "ck-a")
+        ).toBe(1);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "ignores the stored total limit and maintains no group set when disabled",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, false);
+      try {
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+        const now = Date.now();
+        for (const [i, ck] of ["ck-a", "ck-b", "ck-c"].entries()) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `r${i}`,
+              concurrencyKey: ck,
+              timestamp: now - 1000 + i,
+            }),
+            workerQueue: "main",
+          });
+        }
+
+        const allAdmitted = await waitFor(
+          async () => (await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")) === 0
+        );
+        expect(allAdmitted).toBe(true);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(0);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
+  redisTest("enqueue fast path respects the total limit", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer, true);
+    try {
+      await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+      await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+      const now = Date.now();
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r0", concurrencyKey: "ck-a", timestamp: now - 1000 }),
+        workerQueue: "main",
+        enableFastPath: true,
+        skipDequeueProcessing: true,
+      });
+
+      /** The fast path admits synchronously, so the group slot is taken immediately. */
+      expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+      expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(0);
+
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r1", concurrencyKey: "ck-b", timestamp: now - 999 }),
+        workerQueue: "main",
+        enableFastPath: true,
+        skipDequeueProcessing: true,
+      });
+
+      /** At the total limit the fast path must fall through to a normal enqueue. */
+      expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+      expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("nacking releases the total slot", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer, true);
+    try {
+      await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+      await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r0", concurrencyKey: "ck-a", timestamp: Date.now() - 1000 }),
+        workerQueue: "main",
+      });
+
+      const admitted = await waitFor(
+        async () => (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1
+      );
+      expect(admitted).toBe(true);
+
+      /** A second run on another key waits behind the total limit of 1. */
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r1", concurrencyKey: "ck-b", timestamp: Date.now() - 500 }),
+        workerQueue: "main",
+      });
+
+      await setTimeout(2000);
+      expect(await queue.lengthOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+
+      const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
+      assertNonNullable(dequeued);
+      expect(dequeued.messageId).toBe("r0");
+
+      /**
+       * Nack r0 with a far-future retryAt so it cannot immediately reclaim the
+       * slot. If the nack released r0's group slot, r1 is the only eligible run
+       * and must be admitted; if the slot leaked, the queue stays blocked and r1
+       * never surfaces.
+       */
+      await queue.nackMessage({
+        orgId: authenticatedEnvDev.organization.id,
+        messageId: "r0",
+        retryAt: Date.now() + 120_000,
+      });
+
+      const r1Admitted = await waitFor(async () => {
+        const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", {
+          blockingPop: false,
+        });
+        return next?.messageId === "r1";
+      });
+      expect(r1Admitted).toBe(true);
+      expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest(
+    "reconciles a leaked group member instead of blocking the queue",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        const keys = testOptions.keys;
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r0",
+            concurrencyKey: "ck-a",
+            timestamp: Date.now() - 1000,
+          }),
+          workerQueue: "main",
+        });
+
+        const admitted = await waitFor(
+          async () =>
+            (await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")) === 1
+        );
+        expect(admitted).toBe(true);
+
+        const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main");
+        assertNonNullable(dequeued);
+        expect(dequeued.messageId).toBe("r0");
+
+        /**
+         * Simulate a terminal release from a build without the group mirror: the
+         * message key is deleted and the per-key and env sets are cleared, but the
+         * group member is left behind.
+         */
+        await queue.redis.del(keys.messageKey(authenticatedEnvDev.organization.id, "r0"));
+        await queue.redis.srem(
+          keys.queueCurrentConcurrencyKey(authenticatedEnvDev, "task/my-task", "ck-a"),
+          "r0"
+        );
+        await queue.redis.srem(keys.envCurrentConcurrencyKey(authenticatedEnvDev), "r0");
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+
+        /** The next run must still be admitted: the gate prunes the dead member. */
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r1",
+            concurrencyKey: "ck-b",
+            timestamp: Date.now() - 500,
+          }),
+          workerQueue: "main",
+        });
+
+        const r1Admitted = await waitFor(async () => {
+          const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", {
+            blockingPop: false,
+          });
+          return next?.messageId === "r1";
+        });
+        expect(r1Admitted).toBe(true);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "reconciles a large leaked backlog across bounded passes",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer, true);
+      try {
+        const keys = testOptions.keys;
+        const groupKey = keys.queueGroupConcurrencyKey(authenticatedEnvDev, "task/my-task");
+        await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, "task/my-task", 5);
+        await queue.updateQueueTotalConcurrencyLimits(authenticatedEnvDev, "task/my-task", 1);
+
+        /** 1,200 dead members: more than one SSCAN batch, none with a message key. */
+        const dead = Array.from({ length: 1200 }, (_, i) => `dead-${i}`);
+        await queue.redis.sadd(groupKey, ...dead);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1200);
+
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({
+            runId: "r0",
+            concurrencyKey: "ck-a",
+            timestamp: Date.now() - 1000,
+          }),
+          workerQueue: "main",
+        });
+
+        /**
+         * Each dequeue attempt reconciles at most one SSCAN batch behind a 10s
+         * lock; dropping the lock between polls lets the passes run back to
+         * back instead of waiting out the interval.
+         */
+        const r0Admitted = await waitFor(async () => {
+          await queue.redis.del(`${groupKey}:reconcileLock`);
+          const next = await queue.dequeueMessageFromWorkerQueue("consumer-1", "main", {
+            blockingPop: false,
+          });
+          return next?.messageId === "r0";
+        }, 30_000);
+        expect(r0Admitted).toBe(true);
+        expect(await queue.totalConcurrencyOfQueue(authenticatedEnvDev, "task/my-task")).toBe(1);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+});