feat(run-store, webapp): wire server-only execution snapshot rollout

Enables gradual execution snapshot storage rollout without requiring
changes to task SDKs or worker requests. Running tasks retain their
original snapshot storage location.

Mono-RevId: 9bb14e5ab2496009717a14b6081db5e12f160595

Author: d-cs

apps/supervisor/src/index.ts

@@ -742,8 +742,6 @@ class ManagedSupervisor {
       nextAttemptNumber: message.run.attemptNumber,
       snapshotId: message.snapshot.id,
       snapshotFriendlyId: message.snapshot.friendlyId,
-      // Carry the run's storage route to the runner pod so its start request echoes it back.
-      snapshotRoute: message.snapshotRoute,
       placementTags: message.placementTags,
       traceContext: message.run.traceContext,
       annotations: message.run.annotations,
@@ -767,7 +765,6 @@ class ManagedSupervisor {
         {
           runFriendlyId: message.run.friendlyId,
           snapshotFriendlyId: message.snapshot.friendlyId,
-          snapshotRoute: message.snapshotRoute,
         },
         restore
       );
@@ -811,7 +808,6 @@ class ManagedSupervisor {
   private async reportRestoreFailure({
     runFriendlyId,
     snapshotFriendlyId,
-    snapshotRoute,
     runnerId,
     outcome,
   }: RestoreFailure) {
@@ -830,7 +826,7 @@ class ManagedSupervisor {
       return;
     }
     const settled = await settleRestoreFailure(
-      { runnerId, outcome, snapshotRoute },
+      { runnerId, outcome },
       {
         deleteRunner: (name, uid) => manager.deleteRestoreRunner(name, uid),
         report: (body) =>

apps/supervisor/src/services/restoreWatcher.test.ts

@@ -3,12 +3,6 @@ import { Registry } from "prom-client";
 import type { RestoreWatchResult } from "../workloadManager/runCrd.js";
 import { RestoreWatcher, type RestoreFailure } from "./restoreWatcher.js";
 
-const route = (organizationId: string) => ({
-  version: 1 as const,
-  residency: "mirrored" as const,
-  organizationId,
-});
-
 const failed: RestoreWatchResult = {
   ok: false,
   reason: "StartError",
@@ -38,11 +32,10 @@ function setup(opts: { reportFailure?: (failure: RestoreFailure) => Promise<void
 const target = {
   runFriendlyId: "run_a",
   snapshotFriendlyId: "snapshot_1",
-  snapshotRoute: route("org_1"),
 };
 
 describe("RestoreWatcher", () => {
-  it("reports a failed restore with the dequeued snapshot and its route", async () => {
+  it("reports a failed restore with the dequeued snapshot", async () => {
     const { watcher, outcomes, reportFailure } = setup();
 
     const watched = watcher.watch(target, { runnerId: "runner-a", uid: "uid-a" });
@@ -52,42 +45,28 @@ describe("RestoreWatcher", () => {
     expect(reportFailure).toHaveBeenCalledWith({
       runFriendlyId: "run_a",
       snapshotFriendlyId: "snapshot_1",
-      snapshotRoute: route("org_1"),
       runnerId: "runner-a",
       outcome: failed,
     });
   });
 
-  it("names the snapshot and route dequeued last when the restore is redelivered", async () => {
+  it("names the snapshot dequeued last when the restore is redelivered", async () => {
     const { watcher, outcomes, awaitRestore, reportFailure } = setup();
 
     const watched = watcher.watch(target, { runnerId: "runner-a", uid: "uid-a" });
     await watcher.watch(
-      { runFriendlyId: "run_a", snapshotFriendlyId: "snapshot_2", snapshotRoute: route("org_2") },
+      { runFriendlyId: "run_a", snapshotFriendlyId: "snapshot_2" },
       { runnerId: "runner-a", uid: "uid-a" }
     );
     outcomes[0]!.resolve(failed);
     await watched;
 
     expect(awaitRestore).toHaveBeenCalledTimes(1);
     expect(reportFailure).toHaveBeenCalledWith(
-      expect.objectContaining({ snapshotFriendlyId: "snapshot_2", snapshotRoute: route("org_2") })
+      expect.objectContaining({ snapshotFriendlyId: "snapshot_2" })
     );
   });
 
-  it("reports a restore adopted after a restart without a route", async () => {
-    const { watcher, outcomes, reportFailure } = setup();
-
-    const watched = watcher.watch(
-      { runFriendlyId: "run_a", snapshotFriendlyId: "snapshot_1" },
-      { runnerId: "runner-a", uid: "uid-a" }
-    );
-    outcomes[0]!.resolve(failed);
-    await watched;
-
-    expect(reportFailure.mock.calls[0]![0].snapshotRoute).toBeUndefined();
-  });
-
   it("reports nothing for a watch the shutdown ended, whatever it resolved with", async () => {
     const { watcher, outcomes, reportFailure } = setup();
 

apps/supervisor/src/services/restoreWatcher.ts

@@ -1,4 +1,3 @@
-import type { SnapshotRouteWire } from "@trigger.dev/core/v3";
 import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 import { Counter, Gauge, Histogram, type Registry } from "prom-client";
 import {
@@ -12,14 +11,12 @@ import {
 type RestoreWatchTarget = {
   runFriendlyId: string;
   snapshotFriendlyId: string;
-  snapshotRoute?: SnapshotRouteWire;
 };
 
 /** A failed restore, with what the watch or the Runner knew of its run. */
 export type RestoreFailure = {
   runFriendlyId?: string;
   snapshotFriendlyId?: string;
-  snapshotRoute?: SnapshotRouteWire;
   runnerId: string;
   outcome: RestoreWatchResult & { ok: false };
 };
@@ -93,7 +90,6 @@ export class RestoreWatcher {
     const watched = this.watches.get(runnerId);
     if (watched && (watched.uid === undefined || uid === undefined || watched.uid === uid)) {
       watched.snapshotFriendlyId = target.snapshotFriendlyId;
-      watched.snapshotRoute = target.snapshotRoute;
       return;
     }
     // The redelivery replaced the Runner, so the watch on the old one has nothing left to say.
@@ -140,7 +136,6 @@ export class RestoreWatcher {
     await this.report({
       runFriendlyId,
       snapshotFriendlyId: entry.snapshotFriendlyId,
-      snapshotRoute: entry.snapshotRoute,
       runnerId,
       outcome,
     });

apps/supervisor/src/workloadManager/compute.ts

@@ -157,7 +157,6 @@ export class ComputeWorkloadManager implements WorkloadManager {
       TRIGGER_DEPLOYMENT_VERSION: opts.deploymentVersion,
       TRIGGER_RUN_ID: opts.runFriendlyId,
       TRIGGER_SNAPSHOT_ID: opts.snapshotFriendlyId,
-      ...(opts.snapshotRoute ? { TRIGGER_SNAPSHOT_ROUTE: JSON.stringify(opts.snapshotRoute) } : {}),
       TRIGGER_SUPERVISOR_API_PROTOCOL: this.opts.workloadApiProtocol,
       TRIGGER_SUPERVISOR_API_PORT: String(this.opts.workloadApiPort),
       TRIGGER_SUPERVISOR_API_DOMAIN: this.opts.workloadApiDomain ?? "",

apps/supervisor/src/workloadManager/docker.ts

@@ -78,9 +78,6 @@ export class DockerWorkloadManager implements WorkloadManager {
       `TRIGGER_DEPLOYMENT_VERSION=${opts.deploymentVersion}`,
       `TRIGGER_RUN_ID=${opts.runFriendlyId}`,
       `TRIGGER_SNAPSHOT_ID=${opts.snapshotFriendlyId}`,
-      ...(opts.snapshotRoute
-        ? [`TRIGGER_SNAPSHOT_ROUTE=${JSON.stringify(opts.snapshotRoute)}`]
-        : []),
       `TRIGGER_SUPERVISOR_API_PROTOCOL=${this.opts.workloadApiProtocol}`,
       `TRIGGER_SUPERVISOR_API_PORT=${this.opts.workloadApiPort}`,
       `TRIGGER_SUPERVISOR_API_DOMAIN=${this.opts.workloadApiDomain ?? getDockerHostDomain()}`,

apps/supervisor/src/workloadManager/kubernetes.ts

@@ -215,14 +215,6 @@ export class KubernetesWorkloadManager implements WorkloadManager {
                     name: "TRIGGER_SNAPSHOT_ID",
                     value: opts.snapshotFriendlyId,
                   },
-                  ...(opts.snapshotRoute
-                    ? [
-                        {
-                          name: "TRIGGER_SNAPSHOT_ROUTE",
-                          value: JSON.stringify(opts.snapshotRoute),
-                        },
-                      ]
-                    : []),
                   {
                     name: "TRIGGER_SUPERVISOR_API_PROTOCOL",
                     value: this.opts.workloadApiProtocol,

apps/supervisor/src/workloadManager/runCrd.test.ts

@@ -401,7 +401,6 @@ describe("run-crd carries every shared create-option or excludes it on purpose",
     "runFriendlyId",
     "snapshotId",
     "snapshotFriendlyId",
-    "snapshotRoute",
     "traceContext",
     "annotations",
     "hasPrivateLink",
@@ -417,8 +416,6 @@ describe("run-crd carries every shared create-option or excludes it on purpose",
    * creation by another route, such as the runner's name or its token Secret.
    */
   const RUN_CRD_EXCLUDED: Partial<Record<ListedKey, string>> = {
-    snapshotRoute:
-      "Superseded transport, not a run-crd gap: snapshot routing is becoming server-owned (a run's residency is decided from its stored state, not a runner-provided field), so the runner-facing route is being removed rather than built into the Runner. This entry, the field, and the pod backends that set it come out together when that lands.",
     nextAttemptNumber:
       "Not a spec field: it selects the runner's name via getRunnerId, so the attempt lives in the object's name rather than in the spec runnerBodyFor builds.",
     deploymentToken:
@@ -459,9 +456,6 @@ describe("run-crd carries every shared create-option or excludes it on purpose",
     runFriendlyId: { runFriendlyId: "run_other" },
     snapshotId: { snapshotId: "snapshot_other_internal" },
     snapshotFriendlyId: { snapshotFriendlyId: "snapshot_other" },
-    snapshotRoute: {
-      snapshotRoute: { version: 1, residency: "redis-primary", organizationId: "org_abc" },
-    },
     traceContext: { traceContext: { traceparent: "00-probe-probe-01" } },
     annotations: {
       annotations: {
@@ -1685,21 +1679,6 @@ describe("settleRestoreFailure", () => {
     });
   });
 
-  it("sends the dequeued run's snapshot route with the report", async () => {
-    const d = deps();
-    const snapshotRoute = {
-      version: 1 as const,
-      residency: "mirrored" as const,
-      organizationId: "org_1",
-    };
-
-    await settleRestoreFailure({ ...failure("SnapshotNodeGone"), snapshotRoute }, d);
-
-    expect(d.report).toHaveBeenCalledWith(
-      expect.objectContaining({ outcome: "fail", snapshotRoute })
-    );
-  });
-
   // The Runner may be running a guest the watch never saw start.
   it("neither deletes nor reports a Runner the watch timed out on", async () => {
     const d = deps();

apps/supervisor/src/workloadManager/runCrd.ts

@@ -8,12 +8,7 @@ import {
   type ObjectCache,
 } from "@kubernetes/client-node";
 import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
-import type {
-  CheckpointType,
-  EnvironmentType,
-  MachinePreset,
-  SnapshotRouteWire,
-} from "@trigger.dev/core/v3";
+import type { CheckpointType, EnvironmentType, MachinePreset } from "@trigger.dev/core/v3";
 import { type K8sApi, createK8sApi } from "../clients/kubernetes.js";
 import { ReconnectingInformer } from "../clients/reconnectingInformer.js";
 import { getRestoreRunnerId, getRunnerId } from "../util.js";
@@ -1093,7 +1088,6 @@ export type RestoreOutcomeReport = (body: {
   outcome: "requeue" | "fail";
   reason: string;
   message?: string;
-  snapshotRoute?: SnapshotRouteWire;
 }) => Promise<{ success: true } | { success: false; error: string; statusCode?: number }>;
 
 export type RestoreFailureSettlement =
@@ -1118,8 +1112,6 @@ export async function settleRestoreFailure(
   failure: {
     runnerId: string;
     outcome: RestoreWatchResult & { ok: false };
-    /** The dequeued run's route, which spares the platform a read to find it. */
-    snapshotRoute?: SnapshotRouteWire;
   },
   deps: {
     deleteRunner: (runnerId: string, uid: string) => Promise<void>;
@@ -1149,7 +1141,6 @@ export async function settleRestoreFailure(
     outcome: action.outcome,
     reason,
     message: message ?? error,
-    ...(failure.snapshotRoute ? { snapshotRoute: failure.snapshotRoute } : {}),
   });
   const settled: Extract<RestoreFailureSettlement, { action: "reported" }> = result.success
     ? { action: "reported", outcome: action.outcome, result: "ok" }

apps/supervisor/src/workloadManager/types.ts

@@ -3,7 +3,6 @@ import type {
   MachinePreset,
   PlacementTag,
   RunAnnotations,
-  SnapshotRouteWire,
 } from "@trigger.dev/core/v3";
 
 export interface WorkloadManagerOptions {
@@ -51,8 +50,6 @@ export interface WorkloadManagerCreateOptions {
   runFriendlyId: string;
   snapshotId: string;
   snapshotFriendlyId: string;
-  // The run's storage route, materialized as the TRIGGER_SNAPSHOT_ROUTE env var for a cold start.
-  snapshotRoute?: SnapshotRouteWire;
   // Trace context for OTel span emission (W3C format: { traceparent: "00-...", tracestate?: "..." })
   traceContext?: Record<string, unknown>;
   annotations?: RunAnnotations;

apps/webapp/app/env.server.ts

@@ -1381,6 +1381,9 @@ const EnvironmentSchema = z
       .string()
       .default(process.env.REDIS_TLS_DISABLED ?? "false"),
 
+    // Infrastructure endpoint only. Modes and organization overrides are polled feature flags.
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_URL: z.string().optional(),
+
     RUN_ENGINE_RUN_LOCK_REDIS_HOST: z
       .string()
       .optional()