fix(run-engine): stop counting the wait for a deployment as queue scheduling delay

The scheduling-delay anchor was inferred from includeTtl, the flag that
decides whether to arm a run's TTL. That conflated two independent
questions, and they disagree on one path: when a run held in
PENDING_VERSION is promoted, the promotion is its first real entry into
the queue, so arming TTL there is right, but the run was not runnable
while it waited for a worker version and its queueTimestamp still holds
the original trigger time. Anchoring there billed the entire
wait-for-deployment period, minutes to hours, as queue scheduling delay,
which then fed the delay charts and the health report's flow verdict.

Eligibility anchoring is now its own opt-in, set by the two paths where a
run genuinely was runnable from its queue position: the trigger and a
delayed run coming due. The pending-version promotion anchors at the
promotion instead. TTL behaviour is unchanged.

Anchoring is kept separate from queueTimestamp on purpose: that field is
also the queue ordering score, so moving it would cost a run its place
in line to fix a metric.

Author: ericallam

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

@@ -1169,6 +1169,7 @@ export class RunEngine {
               tx: prisma,
               skipRunLock: true,
               includeTtl: true,
+              anchorEligibilityAtQueuePosition: true,
               enableFastPath,
             });
           } catch (enqueueError) {

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

@@ -164,6 +164,7 @@ export class DelayedRunSystem {
         batchId: run.batchId ?? undefined,
         skipRunLock: true,
         includeTtl: true,
+        anchorEligibilityAtQueuePosition: true,
       });
 
       const queuedAt = new Date();

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

@@ -37,6 +37,7 @@ export class EnqueueSystem {
     runnerId,
     skipRunLock,
     includeTtl = false,
+    anchorEligibilityAtQueuePosition = false,
     enableFastPath = false,
     store,
   }: {
@@ -58,8 +59,25 @@ export class EnqueueSystem {
     workerId?: string;
     runnerId?: string;
     skipRunLock?: boolean;
-    /** When true, include TTL in the queued message (only for first enqueue from trigger). Default false. */
+    /**
+     * When true, arm the run's TTL on the queued message. Set by every path that is the run's
+     * first real entry into the queue: trigger, a delayed run coming due, and the pending-version
+     * promotion. Waitpoint and checkpoint re-enqueues must not re-arm it. Default false.
+     */
     includeTtl?: boolean;
+    /**
+     * When true, the scheduling-delay clock starts at the run's queue position (its
+     * `queueTimestamp`, which for a delayed run is `delayUntil`) rather than now. Set only where
+     * the run was genuinely eligible to execute from that moment: trigger and a delayed run
+     * coming due.
+     *
+     * Deliberately separate from `includeTtl`, because the two disagree on the pending-version
+     * promotion: that promotion is the run's first entry into the queue (so TTL arms there), but
+     * the run was NOT runnable while it waited for a worker version, and its `queueTimestamp`
+     * still holds the original trigger time. Anchoring there would bill the whole wait-for-deploy
+     * period as queue scheduling delay. Default false, so a new call site has to opt in.
+     */
+    anchorEligibilityAtQueuePosition?: boolean;
     /** When true, allow the queue to push directly to worker queue if concurrency is available. */
     enableFastPath?: boolean;
     /**
@@ -98,16 +116,10 @@ export class EnqueueSystem {
       // Force development runs to use the environment id as the worker queue.
       const workerQueue = env.type === "DEVELOPMENT" ? env.id : run.workerQueue;
 
-      // Ordering keeps the run's original position; the scheduling-delay anchor is the
-      // trigger/delay time only on first enqueue (includeTtl). Re-enqueues anchor to now,
-      // else the wait metric absorbs the whole waitpoint/checkpoint duration.
       const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime();
       const timestamp = queuePositionMs - run.priorityMs;
-      const eligibleAtMs = includeTtl ? queuePositionMs : Date.now();
+      const eligibleAtMs = anchorEligibilityAtQueuePosition ? queuePositionMs : Date.now();
 
-      // Include TTL only when explicitly requested: the first enqueue from trigger, the
-      // delayed-run system, and the pending-version promotion (each is a run's first real
-      // entry into the queue). Waitpoint and checkpoint re-enqueues must not add TTL.
       let ttlExpiresAt: number | undefined;
       if (includeTtl && run.ttl) {
         const expireAt = parseNaturalLanguageDuration(run.ttl);

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

@@ -314,7 +314,7 @@ describe("RunEngine pending version", () => {
   );
 
   containerTest(
-    "PENDING_VERSION re-enqueue arms TTL on the queued message",
+    "PENDING_VERSION re-enqueue arms TTL but anchors the scheduling-delay clock at the promotion",
     async ({ prisma, redisOptions }) => {
       // When a run enters PENDING_VERSION (background worker doesn't yet have
       // the task), the first enqueue happens but the message is dequeued and
@@ -407,6 +407,7 @@ describe("RunEngine pending version", () => {
 
         // Now a worker arrives WITH the task — pendingVersionSystem
         // re-enqueues the run.
+        const beforePromotion = Date.now();
         await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
 
         await setTimeout(1000);
@@ -424,6 +425,21 @@ describe("RunEngine pending version", () => {
         assertNonNullable(message);
         expect(message.ttlExpiresAt).toBeDefined();
         expect(typeof message.ttlExpiresAt).toBe("number");
+
+        const queuePositionMs = (run.queueTimestamp ?? run.createdAt).getTime();
+
+        expect(
+          message.eligibleAtMs,
+          "scheduling-delay clock must start at the promotion, since the run was not runnable while it waited for a worker version"
+        ).toBeGreaterThanOrEqual(beforePromotion);
+        expect(
+          message.eligibleAtMs,
+          "anchoring at the original queue position would bill the whole wait-for-deploy period as queue scheduling delay"
+        ).toBeGreaterThan(queuePositionMs);
+        expect(
+          message.timestamp,
+          "ordering keeps the original queue position, so waiting for a deployment must not cost the run its place in line"
+        ).toBe(queuePositionMs - run.priorityMs);
       } finally {
         await engine.quit();
       }