feat(supervisor): add dequeue and warm start timing to wide event

Thread timing context from queue consumer through to the compute
workload manager's wide event:

- dequeueResponseMs: platform dequeue HTTP round-trip
- pollingIntervalMs: which polling interval was active (idle vs active)
- warmStartCheckMs: warm start check duration

All fields are optional to avoid breaking existing consumers.

Author: nicktrn

apps/supervisor/src/index.ts

@@ -195,7 +195,7 @@ class ManagedSupervisor {
       this.workloadServer.notifyRun({ run });
     });
 
-    this.workerSession.on("runQueueMessage", async ({ time, message }) => {
+    this.workerSession.on("runQueueMessage", async ({ time, message, dequeueResponseMs, pollingIntervalMs }) => {
       this.logger.log(`Received message with timestamp ${time.toLocaleString()}`, message);
 
       if (message.completedWaitpoints.length > 0) {
@@ -244,7 +244,9 @@ class ManagedSupervisor {
 
       this.logger.log("Scheduling run", { runId: message.run.id });
 
+      const warmStartStart = performance.now();
       const didWarmStart = await this.tryWarmStart(message);
+      const warmStartCheckMs = Math.round(performance.now() - warmStartStart);
 
       if (didWarmStart) {
         this.logger.log("Warm start successful", { runId: message.run.id });
@@ -260,6 +262,9 @@ class ManagedSupervisor {
 
         await this.workloadManager.create({
           dequeuedAt: message.dequeuedAt,
+          dequeueResponseMs,
+          pollingIntervalMs,
+          warmStartCheckMs,
           envId: message.environment.id,
           envType: message.environment.type,
           image: message.image,

apps/supervisor/src/workloadManager/compute.ts

@@ -93,6 +93,10 @@ export class ComputeWorkloadManager implements WorkloadManager {
       machine: opts.machine.name,
       // Environment
       instanceName: env.TRIGGER_WORKER_INSTANCE_NAME,
+      // Supervisor timing
+      dequeueResponseMs: opts.dequeueResponseMs,
+      pollingIntervalMs: opts.pollingIntervalMs,
+      warmStartCheckMs: opts.warmStartCheckMs,
       // Request
       image: imageRef,
       url,

apps/supervisor/src/workloadManager/types.ts

@@ -24,6 +24,10 @@ export interface WorkloadManagerCreateOptions {
   nextAttemptNumber?: number;
   dequeuedAt: Date;
   placementTags?: PlacementTag[];
+  // Timing context (populated by supervisor handler, included in wide event)
+  dequeueResponseMs?: number;
+  pollingIntervalMs?: number;
+  warmStartCheckMs?: number;
   // identifiers
   envId: string;
   envType: EnvironmentType;

packages/core/src/v3/runEngineWorker/supervisor/consumerPool.ts

@@ -351,12 +351,12 @@ export class RunQueueConsumerPool {
 
       const consumer = this.consumerFactory({
         ...this.consumerOptions,
-        onDequeue: async (messages) => {
+        onDequeue: async (messages, timing) => {
           // Always update queue length, default to 0 for empty dequeues or missing value
           this.updateQueueLength(messages[0]?.workerQueueLength ?? 0);
 
           // Forward to the original handler
-          await this.consumerOptions.onDequeue(messages);
+          await this.consumerOptions.onDequeue(messages, timing);
         },
       });
 

packages/core/src/v3/runEngineWorker/supervisor/events.ts

@@ -6,6 +6,8 @@ export type WorkerEvents = {
     {
       time: Date;
       message: DequeuedMessage;
+      dequeueResponseMs?: number;
+      pollingIntervalMs?: number;
     },
   ];
   requestRunAttemptStart: [

packages/core/src/v3/runEngineWorker/supervisor/queueConsumer.ts

@@ -15,21 +15,22 @@ export type RunQueueConsumerOptions = {
   preDequeue?: PreDequeueFn;
   preSkip?: PreSkipFn;
   maxRunCount?: number;
-  onDequeue: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
+  onDequeue: (messages: WorkerApiDequeueResponseBody, timing?: { dequeueResponseMs: number; pollingIntervalMs: number }) => Promise<void>;
 };
 
 export class RunQueueConsumer implements QueueConsumer {
   private readonly client: SupervisorHttpClient;
   private readonly preDequeue?: PreDequeueFn;
   private readonly preSkip?: PreSkipFn;
   private readonly maxRunCount?: number;
-  private readonly onDequeue: (messages: WorkerApiDequeueResponseBody) => Promise<void>;
+  private readonly onDequeue: (messages: WorkerApiDequeueResponseBody, timing?: { dequeueResponseMs: number; pollingIntervalMs: number }) => Promise<void>;
 
   private readonly logger = new SimpleStructuredLogger("queue-consumer");
 
   private intervalMs: number;
   private idleIntervalMs: number;
   private isEnabled: boolean;
+  private lastScheduledIntervalMs: number;
 
   constructor(opts: RunQueueConsumerOptions) {
     this.isEnabled = false;
@@ -38,6 +39,7 @@ export class RunQueueConsumer implements QueueConsumer {
     this.preDequeue = opts.preDequeue;
     this.preSkip = opts.preSkip;
     this.maxRunCount = opts.maxRunCount;
+    this.lastScheduledIntervalMs = opts.idleIntervalMs;
     this.onDequeue = opts.onDequeue;
     this.client = opts.client;
   }
@@ -111,16 +113,18 @@ export class RunQueueConsumer implements QueueConsumer {
     let nextIntervalMs = this.idleIntervalMs;
 
     try {
+      const dequeueStart = performance.now();
       const response = await this.client.dequeue({
         maxResources: preDequeueResult?.maxResources,
         maxRunCount: this.maxRunCount,
       });
+      const dequeueResponseMs = Math.round(performance.now() - dequeueStart);
 
       if (!response.success) {
         this.logger.error("Failed to dequeue", { error: response.error });
       } else {
         try {
-          await this.onDequeue(response.data);
+          await this.onDequeue(response.data, { dequeueResponseMs, pollingIntervalMs: this.lastScheduledIntervalMs });
 
           if (response.data.length > 0) {
             nextIntervalMs = this.intervalMs;
@@ -141,6 +145,7 @@ export class RunQueueConsumer implements QueueConsumer {
       this.logger.verbose("scheduled dequeue with idle interval", { delayMs });
     }
 
+    this.lastScheduledIntervalMs = delayMs;
     setTimeout(this.dequeue.bind(this), delayMs);
   }
 }

packages/core/src/v3/runEngineWorker/supervisor/session.ts

@@ -80,13 +80,15 @@ export class SupervisorSession extends EventEmitter<WorkerEvents> {
     });
   }
 
-  private async onDequeue(messages: WorkerApiDequeueResponseBody): Promise<void> {
+  private async onDequeue(messages: WorkerApiDequeueResponseBody, timing?: { dequeueResponseMs: number; pollingIntervalMs: number }): Promise<void> {
     this.logger.verbose("Dequeued messages with contents", { count: messages.length, messages });
 
     for (const message of messages) {
       this.emit("runQueueMessage", {
         time: new Date(),
         message,
+        dequeueResponseMs: timing?.dequeueResponseMs,
+        pollingIntervalMs: timing?.pollingIntervalMs,
       });
     }
   }