feat(run-engine,webapp): wire ckVirtualTimeScheduling env flag (code-dark)

Task 9. Threads the ckVirtualTimeScheduling option from RunEngineOptions.queue into
the RunQueue constructor, and exposes RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED (+
quantum, window multiplier, state TTL) env vars in the webapp. Off by default at
both layers (env default '0' -> undefined -> today's dequeue path). No behaviour
change until explicitly enabled.

Author: 1stvamp

apps/webapp/app/env.server.ts

@@ -941,6 +941,13 @@ const EnvironmentSchema = z
     RUN_ENGINE_TTL_CONSUMERS_DISABLED: BoolEnv.default(false),
     RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS: z.coerce.number().int().default(5_000),
 
+    // Fair (virtual-time) ordering across concurrency-key variants of a base queue.
+    // Off by default; when off the run queue behaves exactly as before.
+    RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED: z.string().default("0"),
+    RUN_ENGINE_CK_VTIME_QUANTUM: z.coerce.number().default(1),
+    RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER: z.coerce.number().default(3),
+    RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS: z.coerce.number().default(86400),
+
     /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
      *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
     RUN_ENGINE_DEFAULT_MAX_TTL: z.string().optional(),

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

@@ -106,6 +106,15 @@ function createRunEngine() {
         batchMaxSize: env.RUN_ENGINE_TTL_WORKER_BATCH_MAX_SIZE,
         batchMaxWaitMs: env.RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS,
       },
+      ckVirtualTimeScheduling:
+        env.RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED === "1"
+          ? {
+              enabled: true,
+              quantum: env.RUN_ENGINE_CK_VTIME_QUANTUM,
+              scanWindowMultiplier: env.RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER,
+              stateTtlSeconds: env.RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS,
+            }
+          : undefined,
     },
     runLock: {
       redis: {

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

@@ -234,6 +234,7 @@ export class RunEngine {
             workerItemsSuffix: "ttl-worker:{queue:ttl-expiration:}items",
             visibilityTimeoutMs: options.queue?.ttlSystem?.visibilityTimeoutMs ?? 30_000,
           },
+      ckVirtualTimeScheduling: options.queue?.ckVirtualTimeScheduling,
     });
 
     this.worker = new Worker({

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

@@ -16,6 +16,7 @@ import {
 } from "@trigger.dev/redis-worker";
 import type { ControlPlaneResolver } from "./controlPlaneResolver.js";
 import type { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
+import type { RunQueueOptions } from "../run-queue/index.js";
 import type { MinimalAuthenticatedEnvironment } from "../shared/index.js";
 import type { LockRetryConfig } from "./locking.js";
 import type { workerCatalog } from "./workerCatalog.js";
@@ -123,6 +124,9 @@ export type RunEngineOptions = {
       /** Max time (ms) to wait for more items before flushing a batch (default: 5000) */
       batchMaxWaitMs?: number;
     };
+    /** Fair (virtual-time) ordering across concurrency-key variants of a base queue.
+     *  Passed through to RunQueue; off by default (undefined = today's behaviour). */
+    ckVirtualTimeScheduling?: RunQueueOptions["ckVirtualTimeScheduling"];
   };
   runLock: {
     redis: RedisOptions;