feat(run-engine): mint DATETIME and MANUAL waitpoints by mint kind

The two standalone types are the first creates that can reach the store. Both
engine entry points take the mint kind the caller resolved from the org flag,
and default to legacy when it is absent, so every existing caller is unchanged.

Tested against both arms: the minted id classifies to the expected system, a
repeated idempotency key returns the cached waitpoint either way, and the two
directions that matter for rollout are pinned. A legacy mint stays legacy even
where a store is configured, which is the reversibility claim. A store mint on
a process with no store configured falls back to legacy rather than failing,
which keeps one bad configuration from breaking triggers for a flipped org.

Author: d-cs

internal-packages/run-engine/src/engine/index.ts

@@ -95,6 +95,7 @@ import { LegacyPostgresWaitpointCoordinator } from "./waitpointCoordinator/legac
 import { WaitpointRouterCoordinator } from "./waitpointCoordinator/routerCoordinator.js";
 import { StoreWaitpointCoordinatorArm } from "./waitpointCoordinator/storeArm.js";
 import { WaitpointStoreCoordinator } from "./waitpointCoordinator/storeCoordinator.js";
+import type { WaitpointMintKind } from "./waitpointCoordinator/types.js";
 import type {
   EngineWorker,
   HeartbeatTimeouts,
@@ -1803,6 +1804,7 @@ export class RunEngine {
     completedAfter,
     idempotencyKey,
     idempotencyKeyExpiresAt,
+    waitpointMintKind,
   }: {
     /** The run that will block on this waitpoint. Co-locates the waitpoint with the run's DB. */
     runId?: string;
@@ -1811,6 +1813,8 @@ export class RunEngine {
     completedAfter: Date;
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
+    /** Which coordinator mints this waitpoint. Resolved from the org flag by the caller. */
+    waitpointMintKind?: WaitpointMintKind;
   }) {
     return this.waitpointSystem.createDateTimeWaitpoint({
       runId,
@@ -1819,6 +1823,7 @@ export class RunEngine {
       completedAfter,
       idempotencyKey,
       idempotencyKeyExpiresAt,
+      waitpointMintKind,
     });
   }
 
@@ -1834,6 +1839,7 @@ export class RunEngine {
     timeout,
     tags,
     standaloneResidency,
+    waitpointMintKind,
   }: {
     /** The run that will block on this waitpoint. Co-locates the waitpoint with the run's DB. */
     runId?: string;
@@ -1845,6 +1851,8 @@ export class RunEngine {
     tags?: string[];
     /** Standalone-token residency (no owning run) from the env mint kind; ignored when `runId` is set. */
     standaloneResidency?: "NEW" | "LEGACY";
+    /** Which coordinator mints this waitpoint. Resolved from the org flag by the caller. */
+    waitpointMintKind?: WaitpointMintKind;
   }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
     return this.waitpointSystem.createManualWaitpoint({
       runId,
@@ -1855,6 +1863,7 @@ export class RunEngine {
       timeout,
       tags,
       standaloneResidency,
+      waitpointMintKind,
     });
   }
 

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -12,7 +12,11 @@ import { assertNever } from "assert-never";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
 import { buildCompletedWaitpointRecords } from "../waitpointCoordinator/completedWaitpointRecords.js";
-import type { RunBlockEdge, WaitpointCoordinator } from "../waitpointCoordinator/types.js";
+import type {
+  RunBlockEdge,
+  WaitpointCoordinator,
+  WaitpointMintKind,
+} from "../waitpointCoordinator/types.js";
 import type { EnqueueSystem } from "./enqueueSystem.js";
 import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
 import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
@@ -141,17 +145,18 @@ export class WaitpointSystem {
     completedAfter,
     idempotencyKey,
     idempotencyKeyExpiresAt,
+    waitpointMintKind,
   }: {
     runId?: string;
     projectId: string;
     environmentId: string;
     completedAfter: Date;
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
+    waitpointMintKind?: WaitpointMintKind;
   }) {
     const result = await this.coordinator.createDateTimeWaitpoint({
-      // Pinned until the mint flag reaches this entry point.
-      mintKind: "legacy",
+      mintKind: waitpointMintKind ?? "legacy",
       runId,
       projectId,
       environmentId,
@@ -186,10 +191,12 @@ export class WaitpointSystem {
     timeout,
     tags,
     standaloneResidency,
+    waitpointMintKind,
   }: {
     runId?: string;
     environmentId: string;
     projectId: string;
+    waitpointMintKind?: WaitpointMintKind;
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
     timeout?: Date;
@@ -200,8 +207,7 @@ export class WaitpointSystem {
     standaloneResidency?: "NEW" | "LEGACY";
   }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
     const result = await this.coordinator.createManualWaitpoint({
-      // Pinned until the mint flag reaches this entry point.
-      mintKind: "legacy",
+      mintKind: waitpointMintKind ?? "legacy",
       runId,
       environmentId,
       projectId,

internal-packages/run-engine/src/engine/tests/waitpointStandaloneCreates.test.ts

@@ -0,0 +1,155 @@
+import { containerTest } from "@internal/testcontainers";
+import { trace } from "@internal/tracing";
+import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import type { PrismaClient } from "@trigger.dev/database";
+import type { RedisOptions } from "@internal/redis";
+import { describe, expect } from "vitest";
+import { RunEngine } from "../index.js";
+import { setupAuthenticatedEnvironment } from "./setup.js";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+type Arm = "legacy" | "store";
+
+function engineFor(arm: Arm, prisma: PrismaClient, redisOptions: RedisOptions) {
+  return new RunEngine({
+    prisma,
+    worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
+    queue: { redis: redisOptions },
+    runLock: { redis: redisOptions },
+    // The arm under test is selected by whether a store is configured AT ALL, plus the mint
+    // kind each call passes. Both together are what a flipped organization looks like.
+    waitpointStore: arm === "store" ? { redis: redisOptions } : undefined,
+    machines: {
+      defaultMachine: "small-1x",
+      machines: {
+        "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
+      },
+      baseCostInCents: 0.0001,
+    },
+    tracer: trace.getTracer("test", "0.0.0"),
+  });
+}
+
+const expectedFormat: Record<Arm, string> = { legacy: "legacy", store: "b32hexW" };
+
+describe.each<Arm>(["legacy", "store"])("standalone waitpoint creates (%s arm)", (arm) => {
+  containerTest(
+    "createManualWaitpoint mints into the expected system",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = engineFor(arm, prisma, redisOptions);
+
+      try {
+        const { waitpoint } = await engine.createManualWaitpoint({
+          environmentId: environment.id,
+          projectId: environment.project.id,
+          waitpointMintKind: arm,
+        });
+
+        expect(parseWaitpointId(waitpoint.id).format).toBe(expectedFormat[arm]);
+        expect(waitpoint.status).toBe("PENDING");
+        expect(waitpoint.type).toBe("MANUAL");
+        // Read unconditionally by the debounce path, so it must never be undefined.
+        expect(waitpoint.outputIsError).toBe(false);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
+  containerTest(
+    "a repeated idempotency key returns the cached waitpoint",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = engineFor(arm, prisma, redisOptions);
+
+      try {
+        const args = {
+          environmentId: environment.id,
+          projectId: environment.project.id,
+          idempotencyKey: "same-key",
+          waitpointMintKind: arm,
+        } as const;
+
+        const first = await engine.createManualWaitpoint(args);
+        const second = await engine.createManualWaitpoint(args);
+
+        expect(first.isCached).toBe(false);
+        expect(second.isCached).toBe(true);
+        expect(second.waitpoint.id).toBe(first.waitpoint.id);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
+  containerTest(
+    "createDateTimeWaitpoint mints into the expected system",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = engineFor(arm, prisma, redisOptions);
+
+      try {
+        const { waitpoint } = await engine.createDateTimeWaitpoint({
+          environmentId: environment.id,
+          projectId: environment.project.id,
+          completedAfter: new Date(Date.now() + 60_000),
+          waitpointMintKind: arm,
+        });
+
+        expect(parseWaitpointId(waitpoint.id).format).toBe(expectedFormat[arm]);
+        expect(waitpoint.type).toBe("DATETIME");
+        expect(waitpoint.completedAfter).not.toBeNull();
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+});
+
+describe("standalone waitpoint creates, mint-kind fallback", () => {
+  // Reversibility: clearing the flag must revert the NEXT mint with no deploy, and an
+  // engine that has a store configured must still mint legacy when told to.
+  containerTest(
+    "a legacy mint stays legacy even with a store configured",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = engineFor("store", prisma, redisOptions);
+
+      try {
+        const { waitpoint } = await engine.createManualWaitpoint({
+          environmentId: environment.id,
+          projectId: environment.project.id,
+          waitpointMintKind: "legacy",
+        });
+
+        expect(parseWaitpointId(waitpoint.id).format).toBe("legacy");
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
+  // Fail safe, not fail loud: a store mint on a process with no store configured must not
+  // turn every trigger for a flipped organization into an error.
+  containerTest(
+    "a store mint falls back to legacy when no store is configured",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const engine = engineFor("legacy", prisma, redisOptions);
+
+      try {
+        const { waitpoint } = await engine.createManualWaitpoint({
+          environmentId: environment.id,
+          projectId: environment.project.id,
+          waitpointMintKind: "store",
+        });
+
+        expect(parseWaitpointId(waitpoint.id).format).toBe("legacy");
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+});