test(run-engine): pin dequeue-emitted gauges so a sampling regression fails the suite

The gauge assertions were all satisfiable by enqueue-emitted gauges, so
breaking the dequeue scripts' sample-at-return wrapper left the suite
green. The base test now requires the post-admission reading (running 1,
queued 0) and the CK test requires the wildcard aggregate only the CK
dequeue emits. Verified by mutation: disabling the wrapper fails both.

Author: matt-aitken

internal-packages/run-engine/src/run-queue/metrics.test.ts

@@ -123,7 +123,10 @@ describe("RunQueue queue-metrics emission", () => {
 
       const entries = await waitForEntries(redis, definition, (es) => {
         const seen = es.map((e) => e.fields.op);
-        return ["enqueue", "gauge", "started", "ack"].every((o) => seen.includes(o));
+        if (!["enqueue", "gauge", "started", "ack"].every((o) => seen.includes(o))) return false;
+        return es.some(
+          (e) => e.fields.op === "gauge" && e.fields.cc === "1" && e.fields.ql === "0"
+        );
       });
       const ops = entries.map((e) => e.fields.op);
       expect(ops).toContain("enqueue");
@@ -141,6 +144,14 @@ describe("RunQueue queue-metrics emission", () => {
       expect(gauge!.fields.ckq).toBeUndefined();
       expect(gauge!.fields.ckw).toBeUndefined();
 
+      // Pins the dequeue script's sample-at-return wrapper: only the dequeue emits the
+      // post-admission reading (running 1, queued 0); the enqueue gauge sees the inverse.
+      const dequeueGauge = entries.find(
+        (e) => e.fields.op === "gauge" && e.fields.cc === "1" && e.fields.ql === "0"
+      );
+      assertGauge(dequeueGauge);
+      expect(dequeueGauge!.fields.q).toContain("task/my-task");
+
       // The first counter emission also seeds a cum=0 baseline (no wait); the real reading
       // carries wait. Pick the reading (cum > 0).
       const started = entries.find((e) => e.fields.op === "started" && Number(e.fields.cum) > 0);
@@ -283,14 +294,13 @@ describe("RunQueue queue-metrics emission", () => {
       expect(dequeued?.messageId).toBe(message.runId);
 
       const entries = await waitForEntries(redis, definition, (es) =>
-        es.some(
-          (e) => e.fields.op === "gauge" && e.fields.q.includes(":ck:") && e.fields.thr === "0"
-        )
+        es.some((e) => e.fields.op === "gauge" && e.fields.q.includes(":ck:*"))
       );
       const gauges = entries.filter((e) => e.fields.op === "gauge");
       expect(gauges.length).toBeGreaterThan(0);
-      // The aggregate CK dequeue gauge targets the CK wildcard and never sets thr.
-      const aggregate = gauges.find((e) => e.fields.q.includes(":ck:") && e.fields.thr === "0");
+      // The aggregate gauge targets the CK wildcard and only the CK dequeue script emits
+      // it, so this pins that script's sample-at-return wrapper.
+      const aggregate = gauges.find((e) => e.fields.q.includes(":ck:*"));
       assertGauge(aggregate);
       expect(Number(aggregate!.fields.ql)).toBeGreaterThanOrEqual(0);
       expect(Number(aggregate!.fields.cc)).toBeGreaterThanOrEqual(0);