feat: stamp the worker cell and lane on runs at attempt start

Records, per run, which worker cluster (cell) and lane (`container` or
`microvm`) the most recent attempt executed in, so runs can be
attributed to the cluster that served them and queried by it.

The supervisor adds both values to every attempt start it proxies for a
runner, overriding anything the runner sends. The engine merges them
into the run's existing `annotations` inside the update that already
starts the attempt, replacing the previous stamp as a unit. Replication
projects them into two new `task_runs_v2` columns, `worker_cell` and
`worker_lane`. The cell name comes from the new optional
`TRIGGER_WORKER_CELL_NAME` supervisor env var (helm:
`supervisor.config.cellName`); the lane is derived from the selected
backend. Both also go into the worker instance metadata sent on connect,
and admins see them on the run inspector. Neither is exposed to
customers.

Mono-RevId: 87b2c6a5ce7724b4d32c2826fc74c266e3604c3a

Author: myftija

.gitignore

@@ -64,6 +64,7 @@ apps/**/public/build
 .husky
 /packages/react-hooks/src/package.json
 /packages/core/src/package.json
+/packages/build/src/package.json
 /packages/trigger-sdk/src/package.json
 /packages/python/src/package.json
 **/.claude/settings.local.json

apps/supervisor/src/env.test.ts

@@ -184,3 +184,21 @@ describe("Env superRefine - backpressure source awareness", () => {
     ).toThrow();
   });
 });
+
+describe("worker cell name", () => {
+  it("treats an empty value as unset", () => {
+    expect(
+      Env.parse({ ...base, TRIGGER_WORKER_CELL_NAME: "" }).TRIGGER_WORKER_CELL_NAME
+    ).toBeUndefined();
+  });
+
+  it("keeps a well-formed name", () => {
+    expect(
+      Env.parse({ ...base, TRIGGER_WORKER_CELL_NAME: "us-west-2.cell_1" }).TRIGGER_WORKER_CELL_NAME
+    ).toBe("us-west-2.cell_1");
+  });
+
+  it("refuses a malformed name at boot", () => {
+    expect(() => Env.parse({ ...base, TRIGGER_WORKER_CELL_NAME: "bad/name" })).toThrow();
+  });
+});

apps/supervisor/src/env.ts

@@ -14,6 +14,14 @@ export const Env = z
   .object({
     // This will come from `spec.nodeName` in k8s
     TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
+    // Name of the worker cluster (cell) this supervisor serves. Stamped on every run it starts.
+    TRIGGER_WORKER_CELL_NAME: z.preprocess(
+      (value) => (value === "" ? undefined : value),
+      z
+        .string()
+        .regex(/^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/)
+        .optional()
+    ),
     TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().default(30),
 
     // Opt-in, dev-only: stream this process's logs over a local telnet/TCP socket on this port.

apps/supervisor/src/index.ts

@@ -23,6 +23,7 @@ import {
   RunnerRestoreInformer,
   settleRestoreFailure,
 } from "./workloadManager/runCrd.js";
+import { workerLaneFor, type WorkloadManagerBackend } from "./workerLane.js";
 import { RestoreWatcher, type RestoreFailure } from "./services/restoreWatcher.js";
 import { DockerWorkloadManager } from "./workloadManager/docker.js";
 import {
@@ -99,7 +100,7 @@ class ManagedSupervisor {
   private readonly metricsServer?: HttpServer;
   private readonly workloadServer: WorkloadServer;
   private readonly workloadManager: WorkloadManager;
-  private readonly workloadManagerBackend: "kubernetes" | "run-crd" | "docker";
+  private readonly workloadManagerBackend: WorkloadManagerBackend;
   private readonly runCrdManager?: RunCrdWorkloadManager;
   private readonly restoreInformer?: RunnerRestoreInformer;
   private readonly restoreWatcher?: RestoreWatcher;
@@ -339,7 +340,13 @@ class ManagedSupervisor {
 
     const workerToken = getWorkerToken();
 
+    const workerStamp = {
+      workerCell: env.TRIGGER_WORKER_CELL_NAME,
+      workerLane: workerLaneFor(this.workloadManagerBackend, env.KUBERNETES_RUNNER_RUNTIME),
+    };
+
     this.workerSession = new SupervisorSession({
+      metadata: workerStamp,
       resolveResponseSchema,
       workerToken,
       apiUrl: env.TRIGGER_API_URL,
@@ -567,6 +574,7 @@ class ManagedSupervisor {
 
     // Responds to workload requests only
     this.workloadServer = new WorkloadServer({
+      workerStamp,
       port: env.TRIGGER_WORKLOAD_API_PORT_INTERNAL,
       host: env.TRIGGER_WORKLOAD_API_HOST_INTERNAL,
       workerClient: this.workerSession.httpClient,

apps/supervisor/src/workerLane.test.ts

@@ -0,0 +1,14 @@
+import { describe, expect, it } from "vitest";
+import { workerLaneFor } from "./workerLane.js";
+
+describe("workerLaneFor", () => {
+  it("follows the configured runtime on the run-crd backend", () => {
+    expect(workerLaneFor("run-crd", "microvm")).toBe("microvm");
+    expect(workerLaneFor("run-crd", "container")).toBe("container");
+  });
+
+  it("is always container on the pod and docker backends", () => {
+    expect(workerLaneFor("kubernetes", "microvm")).toBe("container");
+    expect(workerLaneFor("docker", "microvm")).toBe("container");
+  });
+});

apps/supervisor/src/workerLane.ts

@@ -0,0 +1,17 @@
+import type { RunnerRuntime } from "./workloadManager/runCrd.js";
+
+export type WorkloadManagerBackend = "kubernetes" | "run-crd" | "docker";
+
+/** The isolation lane every run started by a supervisor on this backend executes in. */
+export function workerLaneFor(
+  backend: WorkloadManagerBackend,
+  runnerRuntime: RunnerRuntime
+): RunnerRuntime {
+  switch (backend) {
+    case "run-crd":
+      return runnerRuntime;
+    case "kubernetes":
+    case "docker":
+      return "container";
+  }
+}

apps/supervisor/src/workloadServer/index.ts

@@ -113,6 +113,8 @@ type WorkloadServerOptions = {
   wideEventOpts: WideEventOptions;
   /** When true, high-frequency HTTP routes also emit wide events. */
   wideEventsNoisyRoutes: boolean;
+  /** Where runs started through this server execute. Overrides anything the runner sends. */
+  workerStamp?: { workerCell?: string; workerLane?: string };
 };
 
 export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
@@ -142,6 +144,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
   >();
 
   private readonly workerClient: SupervisorHttpClient;
+  private readonly workerStamp: { workerCell?: string; workerLane?: string };
 
   constructor(opts: WorkloadServerOptions) {
     super();
@@ -150,6 +153,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
     const port = opts.port;
 
     this.workerClient = opts.workerClient;
+    this.workerStamp = opts.workerStamp ?? {};
     this.checkpointClient = opts.checkpointClient;
     this.wideEventOpts = opts.wideEventOpts;
     this.wideEventsNoisyRoutes = opts.wideEventsNoisyRoutes;
@@ -402,7 +406,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 const startResponse = await this.workerClient.startRunAttempt(
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
-                  body,
+                  {
+                    ...body,
+                    workerCell: this.workerStamp.workerCell,
+                    workerLane: this.workerStamp.workerLane,
+                  },
                   this.runnerIdFromRequest(req),
                   auth.environmentId
                 );

apps/supervisor/src/workloadServer/workloadServer.attemptStart.test.ts

@@ -0,0 +1,79 @@
+import { WORKLOAD_HEADERS } from "@trigger.dev/core/v3/workers";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+vi.mock("std-env", () => ({
+  env: {
+    TRIGGER_API_URL: "http://localhost:3030",
+    TRIGGER_WORKER_TOKEN: "test-token",
+    MANAGED_WORKER_SECRET: "test-secret",
+    OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+  },
+}));
+
+const { WorkloadServer } = await import("./index.js");
+
+const PORT = 18734;
+const BASE = `http://127.0.0.1:${PORT}`;
+
+const startRunAttempt = vi.fn(async () => ({
+  success: true as const,
+  data: { envVars: {} },
+}));
+
+const workerClient = { startRunAttempt } as any;
+
+let server: InstanceType<typeof WorkloadServer>;
+
+beforeAll(async () => {
+  server = new WorkloadServer({
+    port: PORT,
+    workerClient,
+    wideEventOpts: { service: "supervisor", env: { nodeId: "test" }, enabled: false },
+    wideEventsNoisyRoutes: false,
+    workerStamp: { workerCell: "cell-a", workerLane: "microvm" },
+  });
+  await server.start();
+});
+
+afterAll(async () => {
+  await server.stop();
+});
+
+function startAttempt(body: Record<string, unknown>) {
+  return fetch(`${BASE}/api/v1/workload-actions/runs/run_1/snapshots/snap_1/attempts/start`, {
+    method: "POST",
+    headers: {
+      "content-type": "application/json",
+      [WORKLOAD_HEADERS.RUNNER_ID]: "runner_1",
+    },
+    body: JSON.stringify(body),
+  });
+}
+
+describe("WorkloadServer attempt start", () => {
+  it("forwards the runner body with the supervisor's cell and lane stamped on", async () => {
+    const res = await startAttempt({ isWarmStart: true });
+
+    expect(res.status).toBe(200);
+    expect(startRunAttempt).toHaveBeenLastCalledWith(
+      "run_1",
+      "snap_1",
+      { isWarmStart: true, workerCell: "cell-a", workerLane: "microvm" },
+      "runner_1",
+      undefined
+    );
+  });
+
+  it("does not let the runner choose its own cell or lane", async () => {
+    const res = await startAttempt({ workerCell: "spoofed", workerLane: "container" });
+
+    expect(res.status).toBe(200);
+    expect(startRunAttempt).toHaveBeenLastCalledWith(
+      "run_1",
+      "snap_1",
+      { workerCell: "cell-a", workerLane: "microvm" },
+      "runner_1",
+      undefined
+    );
+  });
+});

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -4,6 +4,7 @@ import {
   prettyPrintPacket,
   RunAnnotations,
   SemanticInternalAttributes,
+  WorkerStampAnnotations,
   type TaskRunContext,
   TaskRunError,
   TriggerTraceContext,
@@ -325,6 +326,7 @@ export class SpanPresenter extends BasePresenter {
     const externalTraceId = this.#getExternalTraceId(run.traceContext);
 
     const taskKind = RunAnnotations.safeParse(run.annotations).data?.taskKind;
+    const workerStamp = WorkerStampAnnotations.safeParse(run.annotations).data;
     const isAgentRun = taskKind === "AGENT";
     const isScheduled = taskKind === "SCHEDULED";
     const isWebhook = taskKind === "WEBHOOK";
@@ -467,6 +469,8 @@ export class SpanPresenter extends BasePresenter {
       engine: run.engine,
       region,
       workerQueue: run.workerQueue,
+      workerCell: workerStamp?.workerCell,
+      workerLane: workerStamp?.workerLane,
       traceId: run.traceId,
       spanId: run.spanId,
       isCached: !!linkedRunId,

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts

@@ -26,6 +26,8 @@ export const action = createActionWorkerApiRoute(
       runFriendlyId,
       snapshotFriendlyId,
       isWarmStart: body.isWarmStart,
+      workerCell: body.workerCell,
+      workerLane: body.workerLane,
       runnerId,
       environmentId,
     });