fix: adjudicated PR #4796 review fixes

- Restore process listeners in a finally block so a failed StreamdownRenderer
  test doesn't leak unhandledRejection listeners into later tests.
- Import assertExhaustive from the core/utils subpath instead of the root.
- Correct the envConcurrency.current comment: it's the displayed dequeued
  count, not the gated concurrency count, and can trail it.
- Correct the dashboard-agent token comment: environmentId is the default,
  organizationId is the actual authorization boundary.
- Cap the CK-variant scan in slotHoldersOfQueue's Lua so a queue with many
  concurrency-key variants can't turn a per-request read into an unbounded
  scan; report the cap via the existing truncated signal.
- Carry queuedAt/queueWaitReliable through the run list presenter so
  list_runs computes wait the same way get_run does.
- Print the seed script's ClickHouse fixture command as a heredoc so it
  survives copy-paste.

Author: kathiekiwi

apps/webapp/app/components/code/StreamdownRenderer.test.ts

@@ -128,15 +128,18 @@ describe("loadStreamdownRenderer", () => {
       process.once("unhandledRejection", (err) => resolve(err as Error));
     });
 
-    const mod = await loadStreamdownRenderer(() => Promise.reject(new Error("boom")), [0, 0]);
-    const html = renderToStaticMarkup(createElement(mod.default, null, "hello **world**"));
-    expect(html).toContain("hello");
-
-    const dispatched = await caught;
-    expect(dispatched.message).toMatch(/boom/);
-
-    for (const listener of priorListeners) {
-      process.on("unhandledRejection", listener as NodeJS.UnhandledRejectionListener);
+    try {
+      const mod = await loadStreamdownRenderer(() => Promise.reject(new Error("boom")), [0, 0]);
+      const html = renderToStaticMarkup(createElement(mod.default, null, "hello **world**"));
+      expect(html).toContain("hello");
+
+      const dispatched = await caught;
+      expect(dispatched.message).toMatch(/boom/);
+    } finally {
+      process.removeAllListeners("unhandledRejection");
+      for (const listener of priorListeners) {
+        process.on("unhandledRejection", listener as NodeJS.UnhandledRejectionListener);
+      }
     }
   });
 });

apps/webapp/app/components/runs/v3/runColumns.ts

@@ -62,6 +62,7 @@ const ALWAYS_SELECTED_FIELDS = [
   "status",
   "createdAt",
   "queueTimestamp",
+  "queuedAt",
   "scheduleId",
   "startedAt",
   "lockedAt",

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

@@ -329,6 +329,8 @@ export class ApiRunListPresenter extends BasePresenter {
             startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
             finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
             delayedUntil: run.delayUntil ? new Date(run.delayUntil) : undefined,
+            queuedAt: run.queuedAt ? new Date(run.queuedAt) : undefined,
+            queueWaitReliable: run.queueWaitReliable,
             isTest: run.isTest,
             ttl: run.ttl ?? undefined,
             expiredAt: run.expiredAt ? new Date(run.expiredAt) : undefined,

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

@@ -24,6 +24,7 @@ import { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
+import { STALE_QUEUED_AT_STATUSES } from "~/services/dashboardAgentWatchRunChecks";
 import { runTriggeredAt } from "~/v3/runTimestamps";
 import {
   deriveRunSelect,
@@ -331,6 +332,12 @@ export class NextRunListPresenter {
           updatedAt: run.updatedAt.toISOString(),
           startedAt: startedAt ? startedAt.toISOString() : undefined,
           delayUntil: run.delayUntil ? run.delayUntil.toISOString() : undefined,
+          queuedAt: run.queuedAt ? run.queuedAt.toISOString() : undefined,
+          // A resumed, retried or paused run's stale queuedAt doesn't measure this attempt's wait.
+          queueWaitReliable:
+            run.queuedAt !== null && run.queuedAt !== undefined
+              ? !STALE_QUEUED_AT_STATUSES.has(run.status)
+              : false,
           hasFinished,
           finishedAt: hasFinished
             ? (run.completedAt?.toISOString() ?? run.updatedAt.toISOString())

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

@@ -1,5 +1,5 @@
 import { formatTriggerUri } from "@internal/dashboard-agent-contracts";
-import { assertExhaustive } from "@trigger.dev/core";
+import { assertExhaustive } from "@trigger.dev/core/utils";
 import { type Prettify, type QueueItem, type RetrieveQueueParam } from "@trigger.dev/core/v3";
 import {
   boundedIn,
@@ -21,6 +21,7 @@ export type SlotHolderStatus = TaskRunStatus | "not_found";
 /** Env-scope concurrency, alongside the queue row — the queue can show headroom while the env is saturated. */
 export type EnvConcurrency = {
   limit: number;
+  /** The displayed dequeued count (envCurrentDequeuedKey), not the gated envCurrentConcurrencyKey — can trail it. */
   current: number;
   /** The dequeue gate is `current < limit * burstFactor`, not `current < limit`. */
   burstFactor: number;

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

@@ -59,9 +59,9 @@ export function dashboardAgentUserApiOrigin(): string {
 // mint a token for themselves. The `in` proxy injects this into the turn's
 // metadata so the token reaches the agent without ever touching the browser.
 //
-// Endpoints that bind something to one environment read `environmentId` off the token,
-// so the agent can't name a different one in a request body. `organizationId` is the outer
-// boundary: it never widens what `environmentId` already pins.
+// `environmentId` is the default environment for the turn. `organizationId` is the actual
+// authorization boundary: an org-wide token lets a request body override the environment
+// to any env within that org.
 export function mintDashboardAgentUserActorToken(
   userId: string,
   opts: { environmentId: string; organizationId: string }

apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts

@@ -33,6 +33,7 @@ const LIST_RUN_DEFAULT_SELECT = {
   status: true,
   createdAt: true,
   queueTimestamp: true,
+  queuedAt: true,
   scheduleId: true,
   startedAt: true,
   lockedAt: true,

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

@@ -105,6 +105,7 @@ export type ListedRun = Prisma.TaskRunGetPayload<{
     startedAt: true;
     lockedAt: true;
     delayUntil: true;
+    queuedAt: true;
     updatedAt: true;
     completedAt: true;
     isTest: true;

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

@@ -141,6 +141,7 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({
 });
 
 const DEFAULT_SLOT_HOLDER_LIMIT = 20;
+const DEFAULT_SLOT_HOLDER_MAX_VARIANTS = 50;
 
 /**
  * "admitted": the run holds a concurrency slot (member of currentConcurrency).
@@ -691,19 +692,28 @@ export class RunQueue {
   public async slotHoldersOfQueue(
     env: MinimalAuthenticatedEnvironment,
     queue: string,
-    options?: { limit?: number }
+    options?: { limit?: number; maxVariants?: number }
   ): Promise<QueueSlotHolders> {
     const limit = options?.limit ?? DEFAULT_SLOT_HOLDER_LIMIT;
+    const maxVariants = options?.maxVariants ?? DEFAULT_SLOT_HOLDER_MAX_VARIANTS;
     const baseQueueKey = this.keys.queueKey(env, queue);
 
-    const [admittedCount, dequeuedCount, runningReported, orphanCount, truncated, rawHolders] =
-      await this.redis.slotHoldersOfQueue(
-        baseQueueKey,
-        this.keys.ckIndexKeyFromQueue(baseQueueKey),
-        this.keys.queueRunningCounterKey(env, queue),
-        this.options.redis.keyPrefix ?? "",
-        String(limit)
-      );
+    const [
+      admittedCount,
+      dequeuedCount,
+      runningReported,
+      orphanCount,
+      truncated,
+      rawHolders,
+      skippedVariants,
+    ] = await this.redis.slotHoldersOfQueue(
+      baseQueueKey,
+      this.keys.ckIndexKeyFromQueue(baseQueueKey),
+      this.keys.queueRunningCounterKey(env, queue),
+      this.options.redis.keyPrefix ?? "",
+      String(limit),
+      String(maxVariants)
+    );
 
     const holders = rawHolders.map(([runId, variant, phase]) => ({
       runId,
@@ -716,7 +726,8 @@ export class RunQueue {
       admittedCount,
       dequeuedCount,
       runningReported,
-      truncated: truncated === 1,
+      // A CK-variant scan cap also makes the snapshot incomplete, same as a holder-list cap.
+      truncated: truncated === 1 || skippedVariants > 0,
       unlistedRunning: Math.max(0, runningReported - dequeuedCount),
       consistency:
         dequeuedCount === runningReported && orphanCount === 0 ? "consistent" : "mismatch",
@@ -5637,6 +5648,7 @@ local runningCounterKey = KEYS[3]
 
 local keyPrefix = ARGV[1]
 local maxHolders = tonumber(ARGV[2])
+local maxVariants = tonumber(ARGV[3])
 
 local admittedCount = 0
 local dequeuedCount = 0
@@ -5687,7 +5699,14 @@ end
 
 collect(baseQueueKey, '')
 
-local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
+-- Capped so a queue with many CK variants can't turn this into an unbounded
+-- per-request scan; skippedVariants makes the cap visible to the caller.
+local totalVariants = redis.call('ZCARD', ckIndexKey)
+local variants = redis.call('ZRANGE', ckIndexKey, 0, maxVariants - 1)
+local skippedVariants = totalVariants - #variants
+if skippedVariants < 0 then
+  skippedVariants = 0
+end
 for _, v in ipairs(variants) do
   collect(keyPrefix .. v, v)
 end
@@ -5702,6 +5721,7 @@ return {
   orphanCount,
   truncated,
   holders,
+  skippedVariants,
 }
 `,
     });
@@ -5727,6 +5747,7 @@ type SlotHoldersReply = [
   orphanCount: number,
   truncated: number,
   holders: [runId: string, variant: string, phase: string][],
+  skippedVariants: number,
 ];
 
 declare module "@internal/redis" {
@@ -5840,6 +5861,7 @@ declare module "@internal/redis" {
       // args
       keyPrefix: string,
       maxHolders: string,
+      maxVariants: string,
       callback?: Callback<SlotHoldersReply>
     ): Result<SlotHoldersReply, Context>;
 

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

@@ -261,4 +261,44 @@ describe("RunQueue.slotHoldersOfQueue", () => {
       await queue.quit();
     }
   });
+
+  redisTest(
+    "caps the number of CK variants scanned and reports it as truncated",
+    async ({ redisContainer }) => {
+      const queue = createQueue(redisContainer);
+      try {
+        // Per variant: a fast-path holder plus a slow-path message, so the variant lands
+        // in ckIndex (enumerable) and has one admitted holder.
+        for (let i = 0; i < 3; i++) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({ runId: `r${i}a`, concurrencyKey: `ck-${i}` }),
+            workerQueue: WORKER_QUEUE,
+            skipDequeueProcessing: true,
+            enableFastPath: true,
+          });
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({ runId: `r${i}b`, concurrencyKey: `ck-${i}` }),
+            workerQueue: WORKER_QUEUE,
+            skipDequeueProcessing: true,
+          });
+        }
+
+        const capped = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE, {
+          maxVariants: 2,
+        });
+        expect(capped.holders).toHaveLength(2);
+        expect(capped.truncated).toBe(true);
+
+        const uncapped = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE, {
+          maxVariants: 10,
+        });
+        expect(uncapped.holders).toHaveLength(3);
+        expect(uncapped.truncated).toBe(false);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
 });