feat(supervisor): set op + kind on all wide events

Author: nicktrn

.server-changes/supervisor-op-field-coverage.md

@@ -0,0 +1,6 @@
+---
+area: supervisor
+type: improvement
+---
+
+Tag every supervisor structured event with `op` and `kind` fields for consistent filtering and aggregation.

apps/supervisor/src/index.ts

@@ -269,6 +269,8 @@ class ManagedSupervisor {
                 setMeta(state, "deployment_id", message.deployment.friendlyId);
               }
               setMeta(state, "machine_preset", message.run.machine.name);
+              state.extras.op = "dequeue";
+              state.extras.kind = "inbound";
               state.extras.iteration = "dequeue";
               state.extras.dequeue_response_ms = dequeueResponseMs;
               state.extras.polling_interval_ms = pollingIntervalMs;

apps/supervisor/src/services/computeSnapshotService.ts

@@ -78,6 +78,7 @@ export class ComputeSnapshotService {
       ...this.wideEventOpts,
       populate: (state) => {
         state.extras.op = "snapshot.schedule";
+        state.extras.kind = "event";
         state.meta.run_id = runFriendlyId;
         state.meta.snapshot_id = data.snapshotFriendlyId;
         state.extras.runner_id = data.runnerId;
@@ -99,6 +100,7 @@ export class ComputeSnapshotService {
         ...this.wideEventOpts,
         populate: (state) => {
           state.extras.op = "snapshot.canceled";
+          state.extras.kind = "event";
           state.meta.run_id = runFriendlyId;
         },
       });
@@ -247,6 +249,7 @@ export class ComputeSnapshotService {
         ...this.wideEventOpts,
         setup: (state) => {
           state.extras.op = "snapshot.dispatch";
+          state.extras.kind = "scheduled";
           state.meta.run_id = snapshot.runFriendlyId;
           state.meta.snapshot_id = snapshot.snapshotFriendlyId;
           state.extras.runner_id = snapshot.runnerId;

apps/supervisor/src/workloadServer/index.ts

@@ -196,6 +196,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
    */
   private wideRoute<T>(
     ctx: { req: IncomingMessage; res: ServerResponse; params?: unknown },
+    op: string,
     route: string,
     method: string,
     fn: () => Promise<T> | T,
@@ -211,7 +212,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         method,
         traceparent: this.headerValueFromRequest(ctx.req, "traceparent"),
         inboundRequestId: this.headerValueFromRequest(ctx.req, "x-request-id"),
-        setup: (state) => this.attachRouteMeta(state, ctx.params),
+        setup: (state) => {
+          state.extras.op = op;
+          state.extras.kind = "inbound";
+          this.attachRouteMeta(state, ctx.params);
+        },
       },
       fn,
       (state) => {
@@ -243,6 +248,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "attempt.start",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/attempts/start",
               "POST",
               async () => {
@@ -278,6 +284,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "attempt.complete",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/attempts/complete",
               "POST",
               async () => {
@@ -315,6 +322,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "heartbeat",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/heartbeat",
               "POST",
               async () => {
@@ -351,6 +359,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "suspend",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/suspend",
               "GET",
               async () => {
@@ -446,6 +455,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "continue",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/continue",
               "GET",
               async () => {
@@ -487,6 +497,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           handler: async (ctx) =>
             this.wideRoute(
               ctx,
+              "snapshots.since",
               "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/since/:snapshotFriendlyId",
               "GET",
               async () => {
@@ -522,6 +533,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         handler: async (ctx) =>
           this.wideRoute(
             ctx,
+            "deployment.dequeue",
             "/api/v1/workload-actions/deployments/:deploymentId/dequeue",
             "GET",
             async () => {
@@ -555,6 +567,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         handler: async (ctx) =>
           this.wideRoute(
             ctx,
+            "logs.debug",
             "/api/v1/workload-actions/runs/:runFriendlyId/logs/debug",
             "POST",
             async () => {
@@ -576,6 +589,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         handler: async (ctx) =>
           this.wideRoute(
             ctx,
+            "logs.debug",
             "/api/v1/workload-actions/runs/:runFriendlyId/logs/debug",
             "POST",
             async () => {
@@ -590,7 +604,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
     httpServer.route("/api/v1/compute/snapshot-complete", "POST", {
       bodySchema: SnapshotCallbackPayloadSchema,
       handler: async (ctx) =>
-        this.wideRoute(ctx, "/api/v1/compute/snapshot-complete", "POST", async () => {
+        this.wideRoute(ctx, "snapshot.callback", "/api/v1/compute/snapshot-complete", "POST", async () => {
           const { reply, body } = ctx;
           if (!this.snapshotService) {
             reply.empty(404);
@@ -680,6 +694,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         emitOneShot({
           ...this.wideEventOpts,
           populate: (state) => {
+            state.extras.op = event === "run_connected" ? "socket.run.connected" : "socket.run.disconnected";
+            state.extras.kind = "event";
             state.extras.event = event;
             setMeta(state, "run_id", friendlyId);
             if (socket.data.deploymentId) {