fix(run-engine,webapp): address whole-branch adversarial review

Three-model blind review (no Critical). Fixes:
- Refresh ckVtimeFloor TTL on enqueue/nack, not just dequeue: a dequeue-quiescent
  but enqueue-active base queue could expire the floor key while ckVtime survived,
  making a new variant register at 0 and jump the backlog (fairness inversion).
- tostring() the vtime tag advance so a fractional quantum/weight is not truncated
  to an integer by Redis's Lua-number ZADD conversion (silent tag freeze).
- Validate the env config: enable flag now uses BoolEnv (so =true/1/yes work, not
  only '1'); quantum/windowMultiplier/stateTtlSeconds are int().positive() (a 0 TTL
  errored SET ... EX 0 and stopped all CK dequeues; a 0 multiplier caused a full
  ZRANGE scan). Constructor clamps as defense-in-depth.
- Tests: floor-not-lost regression (H1), large-N (>window) sharding no-starvation,
  tightened ckBalanced no-harm bound, corrected op-count budget (EXISTS = 7 fixed).
- Softened a dangling plan-doc path in a comment.
Old Lua command bodies remain byte-identical (edits are in the ...Vtime... commands
+ env/wiring/tests only).

Author: 1stvamp

apps/webapp/app/env.server.ts

@@ -943,10 +943,10 @@ const EnvironmentSchema = z
 
     // 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),
+    RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED: BoolEnv.default(false),
+    RUN_ENGINE_CK_VTIME_QUANTUM: z.coerce.number().int().positive().default(1),
+    RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER: z.coerce.number().int().positive().default(3),
+    RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS: z.coerce.number().int().positive().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. */

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

@@ -107,7 +107,7 @@ function createRunEngine() {
         batchMaxWaitMs: env.RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS,
       },
       ckVirtualTimeScheduling:
-        env.RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED === "1"
+        env.RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED
           ? {
               enabled: true,
               quantum: env.RUN_ENGINE_CK_VTIME_QUANTUM,

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

@@ -121,8 +121,8 @@ export type RunQueueOptions = {
   /**
    * Fair (virtual-time / SFQ) ordering across concurrency-key variants of a
    * base queue. Off by default; when off, the exact pre-existing Lua commands
-   * run and no vtime keys are created. See docs/superpowers/plans/
-   * 2026-07-23-ck-virtual-time-scheduling-plan.md.
+   * run and no vtime keys are created. See the CK virtual-time scheduling
+   * design for the ordering model.
    */
   ckVirtualTimeScheduling?: {
     enabled: boolean;
@@ -226,9 +226,19 @@ export class RunQueue {
     this.shardCount = options.shardCount ?? 2;
     this.counterTtlSeconds = options.counterTtlSeconds ?? 86400;
     this.#ckVtimeEnabled = options.ckVirtualTimeScheduling?.enabled ?? false;
-    this.#ckVtimeQuantum = options.ckVirtualTimeScheduling?.quantum ?? 1;
-    this.#ckVtimeWindowMultiplier = options.ckVirtualTimeScheduling?.scanWindowMultiplier ?? 3;
-    this.#ckVtimeStateTtl = options.ckVirtualTimeScheduling?.stateTtlSeconds ?? 86400;
+    // Defense-in-depth: clamp so a directly-constructed RunQueue can't get bad
+    // values that would freeze tags (quantum <= 0) or force an O(N) scan /
+    // EX 0 error (multiplier / ttl <= 0).
+    const resolvedQuantum = options.ckVirtualTimeScheduling?.quantum ?? 1;
+    this.#ckVtimeQuantum = resolvedQuantum > 0 ? resolvedQuantum : 1;
+    this.#ckVtimeWindowMultiplier = Math.max(
+      1,
+      Math.floor(options.ckVirtualTimeScheduling?.scanWindowMultiplier ?? 3)
+    );
+    this.#ckVtimeStateTtl = Math.max(
+      1,
+      Math.floor(options.ckVirtualTimeScheduling?.stateTtlSeconds ?? 86400)
+    );
     this.retryOptions = options.retryOptions ?? defaultRetrySettings;
     this.redis = createRedisClient(options.redis, {
       onError: (error) => {
@@ -3918,6 +3928,7 @@ end
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
 redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
+redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
 
 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
@@ -4045,6 +4056,7 @@ end
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
 redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
+redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
 
 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
@@ -4826,7 +4838,7 @@ local function tryServe(ckQueueName)
           local weight = 1
           local tag = tonumber(redis.call('ZSCORE', ckVtimeKey, ckQueueName) or floor)
           if tag < floor then tag = floor end
-          redis.call('ZADD', ckVtimeKey, tag + (quantum / weight), ckQueueName)
+          redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName)
         end
       else
         redis.call('ZREM', fullQueueKey, messageId)
@@ -5577,6 +5589,7 @@ end
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
 redis.call('ZADD', ckVtimeKey, 'NX', vfloor, messageQueueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
+redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
 
 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')

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

@@ -397,6 +397,99 @@ describe("CK virtual-time (SFQ) dequeue", () => {
     }
   );
 
+  // H1 regression: the floor key must not be allowed to expire while ckVtime
+  // survives. Before the fix, only the dequeue command refreshed the floor
+  // key's TTL, so a dequeue-quiescent + enqueue-active base queue let the floor
+  // key expire underneath a live ckVtime; a brand-new variant then read a
+  // missing floor as 0 and jumped ahead of the whole established backlog. The
+  // enqueue/nack registration paths now refresh the floor key TTL too.
+  redisTest(
+    "enqueue refreshes the floor key TTL and a new variant registers at the current floor",
+    async ({ redisContainer }) => {
+      const stateTtlSeconds = 3600;
+      const queue = createQueue(redisContainer, { stateTtlSeconds });
+      try {
+        const t0 = Date.now() - 100_000;
+        const cks = ["a", "b"];
+
+        for (const ck of cks) {
+          for (let i = 0; i < 25; i++) {
+            await queue.enqueueMessage({
+              env: authenticatedEnvDev,
+              message: makeMessage({
+                runId: `r-${ck}-${i}`,
+                concurrencyKey: ck,
+                timestamp: t0 + i,
+              }),
+              workerQueue: authenticatedEnvDev.id,
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+
+        const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(variantName("a"));
+        const ckVtimeFloorKey = testOptions.keys.ckVtimeFloorKeyFromQueue(variantName("a"));
+        const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+
+        // Drive tags and the floor above 0 with a run of serves.
+        for (let call = 0; call < 20; call++) {
+          const messages = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 2);
+          for (const m of messages) {
+            await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, m.messageId, {
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+
+        const floor = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0");
+        expect(floor).toBeGreaterThan(10);
+        expect(await queue.redis.exists(ckVtimeKey)).toBe(1);
+
+        // Simulate the floor key's TTL decaying toward expiry while dequeues are
+        // quiescent. Without the fix, only a dequeue would ever bump it back.
+        await queue.redis.pexpire(ckVtimeFloorKey, 2_000);
+
+        // WITHOUT dequeuing, enqueue several more messages on an existing
+        // variant. The enqueue registration path must refresh the floor key TTL.
+        for (let i = 0; i < 5; i++) {
+          await queue.enqueueMessage({
+            env: authenticatedEnvDev,
+            message: makeMessage({
+              runId: `r-a-more-${i}`,
+              concurrencyKey: "a",
+              timestamp: t0 + 500 + i,
+            }),
+            workerQueue: authenticatedEnvDev.id,
+            skipDequeueProcessing: true,
+          });
+        }
+
+        // The floor key TTL was pushed back up to (about) stateTtl, well above
+        // the 2s decay we forced.
+        const floorPttl = await queue.redis.pttl(ckVtimeFloorKey);
+        expect(floorPttl).toBeGreaterThan(2_000);
+        expect(floorPttl).toBeLessThanOrEqual(stateTtlSeconds * 1000);
+
+        // Enqueue-only activity does not move the floor value itself.
+        const floorAfter = Number((await queue.redis.get(ckVtimeFloorKey)) ?? "0");
+        expect(floorAfter).toBe(floor);
+
+        // A brand-new variant enqueued now registers at the CURRENT floor, so it
+        // cannot leapfrog the established backlog back to 0.
+        await queue.enqueueMessage({
+          env: authenticatedEnvDev,
+          message: makeMessage({ runId: "r-fresh", concurrencyKey: "fresh", timestamp: t0 }),
+          workerQueue: authenticatedEnvDev.id,
+          skipDequeueProcessing: true,
+        });
+        const freshTag = Number(await queue.redis.zscore(ckVtimeKey, variantName("fresh")));
+        expect(freshTag).toBe(floor);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
   redisTest("no service, no advance", async ({ redisContainer }) => {
     const queue = createQueue(redisContainer);
     try {

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

@@ -346,10 +346,10 @@ describe("CK virtual-time concurrency and op-count budget", () => {
       expect(off.served).toBe(cks.length * perKey);
       expect(on.served).toBe(cks.length * perKey);
 
-      // Per dequeue call the vtime path adds at worst: GET floor, ZRANGE min,
-      // ZRANGE window, SET floor, EXPIRE, the pass-2 ZRANGEBYSCORE, plus per
-      // serve one ZSCORE and one ZADD.
-      const budget = dequeueCalls * (6 + 2 * maxCount);
+      // Per dequeue call the vtime path adds at worst 7 fixed ops: GET floor,
+      // ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, SET floor,
+      // EXISTS ckVtime, EXPIRE ckVtime — plus per serve one ZSCORE and one ZADD.
+      const budget = dequeueCalls * (7 + 2 * maxCount);
       expect(
         on.totalCalls,
         `on_total ${on.totalCalls} exceeds off_total ${off.totalCalls} + budget ${budget}`

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

@@ -432,7 +432,68 @@ describe("CK virtual-time fairness on the real batched dequeue path", () => {
       const offMax = maxPerKeyMeanWait(off);
       debugLog("ckBalanced", { onMax, offMax, ratio: onMax / offMax });
 
-      expect(onMax).toBeLessThanOrEqual(1.25 * offMax);
+      // Observed ratio is 1.0 (the fair order is neutral on the symmetric case),
+      // so allow only modest headroom rather than the original 1.25.
+      expect(onMax).toBeLessThanOrEqual(1.1 * offMax);
+    }
+  );
+
+  // ckManyKeys (sharding coverage): cardinality ABOVE the pass-1 fair window.
+  // The batched dequeue uses maxCount 10, so window = actualMaxCount * 3 = 30.
+  // With ~60 attacker keys (all on the same old head) plus 1 light key on a
+  // newer head, 61 variants sit above the 30-wide pass-1 ZRANGE window, so no
+  // single fair pass can even see every key. The property to hold is that this
+  // does NOT permanently starve the light key: as attackers advance their tags
+  // out of the bottom of the window, the light key (still at the floor) rises
+  // into it and gets served, and every message drains exactly once. A bounded
+  // first-serve delay is fine; permanent starvation or a stuck drain is not.
+  redisTest(
+    "ckManyKeys: light key is not starved when cardinality exceeds the fair window",
+    async ({ redisContainer }) => {
+      const t0 = Date.now() - 500_000;
+      const messages: ScenarioMessage[] = [];
+      const attackerCount = 60;
+      for (let i = 0; i < 8; i++) {
+        for (let k = 0; k < attackerCount; k++) {
+          const ck = `att${String(k).padStart(2, "0")}`;
+          // All attackers share the same old head timestamp (tied heads).
+          messages.push({ runId: `${ck}-${i}`, ck, timestamp: t0 });
+        }
+      }
+      for (let i = 0; i < 10; i++) {
+        messages.push({ runId: `light-${i}`, ck: "light", timestamp: t0 + 50_000 + i });
+      }
+      const scenario: Scenario = {
+        name: "ckManyKeys",
+        messages,
+        envConcurrencyLimit: 25,
+        holdSteps: 3,
+        maxSteps: 1_000,
+      };
+
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
+
+      // No loss and no double-serve in either run: the run terminates and every
+      // message (attackers + light) is served exactly once within maxSteps.
+      assertConservation(scenario, on, off);
+
+      const isLight = (ck: string) => ck === "light";
+
+      // The light key IS eventually served (no permanent starvation) in both
+      // runs, and drains fully.
+      const onFirstServe = firstServeStep(on, isLight);
+      const offFirstServe = firstServeStep(off, isLight);
+      expect(on.drainStep).toBeGreaterThanOrEqual(0);
+      expect(off.drainStep).toBeGreaterThanOrEqual(0);
+
+      debugLog("ckManyKeys", {
+        variants: attackerCount + 1,
+        onFirstServe,
+        offFirstServe,
+        onDrainStep: on.drainStep,
+        offDrainStep: off.drainStep,
+      });
     }
   );