feat(webapp,run-engine): Redis waitpoint coordinator arm behind a per-org mint flag

apps/webapp/app/env.server.ts

@@ -2020,6 +2020,12 @@ const EnvironmentSchema = z
     // (stale or fresh) resolves to the same kind for the whole window. See mintFlipGrace.ts.
     RUN_OPS_MINT_FLIP_GRACE_MS: z.coerce.number().int().default(90_000),
 
+    // Per-organization waitpoint coordinator cutover. The org's waitpointSystem flag wins;
+    // this is the fallback when the org has no override. Read only at waitpoint mint time.
+    WAITPOINT_SYSTEM_DEFAULT: z.enum(["legacy", "redis"]).default("legacy"),
+    WAITPOINT_MINT_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
+    WAITPOINT_MINT_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(10_000),
+
     // Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
     // with the runs replicator for leader locking but has its own slot and
     // publication so the two consume independently.

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

@@ -2,6 +2,7 @@ import { isWaitpointOutputTimeout, prettyPrintPacket } from "@trigger.dev/core/v
 import { type PrismaClientOrTransaction } from "~/db.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
+import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { logger } from "~/services/logger.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 import { runStore as defaultRunStore } from "~/v3/runStore.server";
@@ -117,7 +118,14 @@ export class WaitpointPresenter extends BasePresenter {
       }
     }
 
-    const connectedRunIds = await this.#connectedRunFriendlyIds(waitpoint.id);
+    // The connected-runs display is built from a Postgres table that only the Postgres
+    // block-edge write populates. A store-resident waitpoint keeps its edges elsewhere, so
+    // the query would answer an empty list, which reads as "nothing is blocked on this"
+    // rather than "this cannot be shown". Report the difference instead of guessing.
+    const connectedRunsAvailable = parseWaitpointId(waitpoint.id).format === "legacy";
+    const connectedRunIds = connectedRunsAvailable
+      ? await this.#connectedRunFriendlyIds(waitpoint.id)
+      : [];
     const connectedRuns: NextRunListItem[] = [];
 
     if (connectedRunIds.length > 0) {
@@ -164,6 +172,7 @@ export class WaitpointPresenter extends BasePresenter {
       createdAt: waitpoint.createdAt,
       tags: waitpoint.tags,
       connectedRuns,
+      connectedRunsAvailable,
     };
   }
 }

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens.$waitpointParam/route.tsx

@@ -5,6 +5,7 @@ import { z } from "zod";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { LinkButton } from "~/components/primitives/Buttons";
 import { Header2, Header3 } from "~/components/primitives/Headers";
+import { Paragraph } from "~/components/primitives/Paragraph";
 import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
@@ -135,22 +136,28 @@ export default function Page() {
             <Header3>Related runs</Header3>
             <InfoIconTooltip content="These runs have been blocked by this waitpoint." />
           </div>
-          <TaskRunsTable
-            enableSmartColumns={false}
-            total={waitpoint.connectedRuns.length}
-            hasFilters={false}
-            filters={{
-              tasks: [],
-              versions: [],
-              statuses: [],
-              from: undefined,
-              to: undefined,
-            }}
-            runs={waitpoint.connectedRuns}
-            isLoading={false}
-            variant="bright"
-            disableAdjacentRows
-          />
+          {!waitpoint.connectedRunsAvailable ? (
+            <Paragraph variant="small" className="pl-3">
+              Related runs aren't available for this token.
+            </Paragraph>
+          ) : (
+            <TaskRunsTable
+              enableSmartColumns={false}
+              total={waitpoint.connectedRuns.length}
+              hasFilters={false}
+              filters={{
+                tasks: [],
+                versions: [],
+                statuses: [],
+                from: undefined,
+                to: undefined,
+              }}
+              runs={waitpoint.connectedRuns}
+              isLoading={false}
+              variant="bright"
+              disableAdjacentRows
+            />
+          )}
         </div>
       </div>
       {waitpoint.status === "WAITING" && (

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import { z } from "zod";
 import {
   CreateInputStreamWaitpointRequestBody,
@@ -82,7 +83,14 @@ const { action, loader } = createActionApiRoute(
 
       // Create the waitpoint. Co-locate it with the owning run (run-ops split) so a run-ops id
       // run's input-stream waitpoint lands on the run's DB and its block edge resolves.
+      const waitpointMintKind = await resolveWaitpointMintKind({
+        organizationId: authentication.environment.organizationId,
+        id: authentication.environment.id,
+        orgFeatureFlags: authentication.environment.organization.featureFlags,
+      });
+
       const result = await engine.createManualWaitpoint({
+        waitpointMintKind,
         runId: run.id,
         environmentId: authentication.environment.id,
         projectId: authentication.environment.projectId,

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import {
   CreateSessionStreamWaitpointRequestBody,
   type CreateSessionStreamWaitpointResponseBody,
@@ -103,7 +104,14 @@ const { action, loader } = createActionApiRoute(
 
       // Create the waitpoint. Co-locate it with the owning run (run-ops split) so a run-ops id
       // run's session-stream waitpoint lands on the run's DB and its block edge resolves.
+      const waitpointMintKind = await resolveWaitpointMintKind({
+        organizationId: authentication.environment.organizationId,
+        id: authentication.environment.id,
+        orgFeatureFlags: authentication.environment.organization.featureFlags,
+      });
+
       const result = await engine.createManualWaitpoint({
+        waitpointMintKind,
         runId: run.id,
         environmentId: authentication.environment.id,
         projectId: authentication.environment.projectId,

apps/webapp/app/routes/api.v1.waitpoints.tokens.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import {
   CreateWaitpointTokenRequestBody,
   type CreateWaitpointTokenResponseBody,
@@ -104,7 +105,14 @@ const { action } = createActionApiRoute(
         }
       }
 
+      const waitpointMintKind = await resolveWaitpointMintKind({
+        organizationId: authentication.environment.organizationId,
+        id: authentication.environment.id,
+        orgFeatureFlags: authentication.environment.organization.featureFlags,
+      });
+
       const result = await engine.createManualWaitpoint({
+        waitpointMintKind,
         environmentId: authentication.environment.id,
         projectId: authentication.environment.projectId,
         idempotencyKey: body.idempotencyKey,

apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts

@@ -1,4 +1,5 @@
 import type { TypedResponse } from "@remix-run/server-runtime";
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import { json } from "@remix-run/server-runtime";
 import type { WaitForDurationResponseBody } from "@trigger.dev/core/v3";
 import { WaitForDurationRequestBody } from "@trigger.dev/core/v3";
@@ -41,7 +42,14 @@ const { action } = createActionApiRoute(
         ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
         : undefined;
 
+      const waitpointMintKind = await resolveWaitpointMintKind({
+        organizationId: authentication.environment.organizationId,
+        id: authentication.environment.id,
+        orgFeatureFlags: authentication.environment.organization.featureFlags,
+      });
+
       const { waitpoint } = await engine.createDateTimeWaitpoint({
+        waitpointMintKind,
         // Co-locate the waitpoint with the run that blocks on it (run-ops split): a run-ops run lives
         // on the dedicated DB, but the minted waitpoint id is always a cuid, so without the run id
         // the waitpoint would route to the control-plane DB and the block edge would never resolve.

apps/webapp/app/runEngine/services/batchTrigger.server.ts

@@ -1,3 +1,4 @@
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import {
   type BatchTriggerTaskV2RequestBody,
   type BatchTriggerTaskV3RequestBody,
@@ -194,6 +195,11 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
           environmentId: environment.id,
           projectId: environment.projectId,
           organizationId: environment.organizationId,
+          waitpointMintKind: await resolveWaitpointMintKind({
+            organizationId: environment.organizationId,
+            id: environment.id,
+            orgFeatureFlags: environment.organization.featureFlags,
+          }),
         });
       }
 
@@ -285,6 +291,11 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
           environmentId: environment.id,
           projectId: environment.projectId,
           organizationId: environment.organizationId,
+          waitpointMintKind: await resolveWaitpointMintKind({
+            organizationId: environment.organizationId,
+            id: environment.id,
+            orgFeatureFlags: environment.organization.featureFlags,
+          }),
         });
       }
 

apps/webapp/app/runEngine/services/createBatch.server.ts

@@ -1,4 +1,5 @@
 import type { InitializeBatchOptions } from "@internal/run-engine";
+import { resolveWaitpointMintKind } from "~/v3/waitpointMigration/waitpointMintKind.server";
 import { type CreateBatchRequestBody, type CreateBatchResponse } from "@trigger.dev/core/v3";
 import { RunId } from "@trigger.dev/core/v3/isomorphic";
 import { type BatchTaskRun, Prisma } from "@trigger.dev/database";
@@ -137,6 +138,11 @@ export class CreateBatchService extends WithRunEngine {
               environmentId: environment.id,
               projectId: environment.projectId,
               organizationId: environment.organizationId,
+              waitpointMintKind: await resolveWaitpointMintKind({
+                organizationId: environment.organizationId,
+                id: environment.id,
+                orgFeatureFlags: environment.organization.featureFlags,
+              }),
             });
           }
 

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -28,6 +28,10 @@ import { parseDelay } from "~/utils/delays";
 import { removeNullBytesFromKey } from "~/utils/nullBytes";
 import { handleMetadataPacket } from "~/utils/packets";
 import { startSpan } from "~/v3/tracing.server";
+import {
+  resolveWaitpointMintKind,
+  type WaitpointMintKind,
+} from "~/v3/waitpointMigration/waitpointMintKind.server";
 import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
 import { resolveRunMintTarget } from "~/v3/runOpsMigration/resolveRunMintTarget.server";
 import type {
@@ -653,10 +657,16 @@ export class RunEngineTriggerTaskService {
                     event.setAttribute("taskRunId", runFriendlyId);
 
                     const payloadPacket = await this.payloadProcessor.process(triggerRequest);
+                    const waitpointMintKind = await resolveWaitpointMintKind({
+                      organizationId: environment.organizationId,
+                      id: environment.id,
+                      orgFeatureFlags: environment.organization.featureFlags,
+                    });
 
                     const engineTriggerInput = this.#buildEngineTriggerInput({
                       runFriendlyId,
                       environment,
+                      waitpointMintKind,
                       idempotencyKey,
                       idempotencyKeyExpiresAt,
                       body,
@@ -733,10 +743,16 @@ export class RunEngineTriggerTaskService {
                 }
 
                 const payloadPacket = await this.payloadProcessor.process(triggerRequest);
+                const waitpointMintKind = await resolveWaitpointMintKind({
+                  organizationId: environment.organizationId,
+                  id: environment.id,
+                  orgFeatureFlags: environment.organization.featureFlags,
+                });
 
                 const baseEngineInput = this.#buildEngineTriggerInput({
                   runFriendlyId,
                   environment,
+                  waitpointMintKind,
                   idempotencyKey,
                   idempotencyKeyExpiresAt,
                   body,
@@ -899,6 +915,7 @@ export class RunEngineTriggerTaskService {
   #buildEngineTriggerInput(args: {
     runFriendlyId: string;
     environment: AuthenticatedEnvironment;
+    waitpointMintKind: WaitpointMintKind;
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
     body: TriggerTaskRequest["body"];
@@ -988,6 +1005,7 @@ export class RunEngineTriggerTaskService {
         ? { id: args.options.batchId, index: args.options.batchIndex ?? 0 }
         : undefined,
       resumeParentOnCompletion: args.body.options?.resumeParentOnCompletion,
+      waitpointMintKind: args.waitpointMintKind,
       depth: args.depth,
       metadata: args.metadataPacket?.data,
       metadataType: args.metadataPacket?.dataType,

apps/webapp/app/v3/featureFlags.ts

@@ -37,6 +37,9 @@ export const FEATURE_FLAG = {
   runOpsMintShardSetFlippedAt: "runOpsMintShardSetFlippedAt",
   // Fleet-wide pin for the complete cutover. Beats every per-org and per-env pin.
   runOpsMintShardOverride: "runOpsMintShardOverride",
+  // Per-organization waitpoint coordinator selection. Read ONLY at waitpoint mint time;
+  // every later operation on a waitpoint routes by its id shape, never by this flag.
+  waitpointSystem: "waitpointSystem",
   queueMetricsUiEnabled: "queueMetricsUiEnabled",
   // Build path for CLI deploys, resolved by DeploymentService.getDeploySettings.
   deployBuildPath: "deployBuildPath",
@@ -101,6 +104,7 @@ export const FeatureFlagCatalog = {
   // Per-org run-ops-id mint cutover. Defaults to "cuid"; only honored when
   // RUN_OPS_MINT_ENABLED is on AND isSplitEnabled() is true.
   [FEATURE_FLAG.runOpsMintKind]: z.enum(["cuid", "runOpsId"]),
+  [FEATURE_FLAG.waitpointSystem]: z.enum(["legacy", "redis"]),
   // Grace-linger stamp: the previously-effective kind and the flip timestamp, written
   // by stampMintKindFlip on a genuine flip. Display-only (see ORG_LOCKED_FLAGS).
   [FEATURE_FLAG.runOpsMintKindPrev]: z.enum(["cuid", "runOpsId"]),

apps/webapp/app/v3/waitpointMigration/waitpointMintKind.server.ts

@@ -0,0 +1,66 @@
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { BoundedTtlCache } from "~/services/realtime/boundedTtlCache";
+import { singleton } from "~/utils/singleton";
+import { logger } from "~/services/logger.server";
+import { FEATURE_FLAG } from "~/v3/featureFlags";
+import { computeWaitpointMintKind, type WaitpointMintKind } from "./waitpointMintKind.js";
+
+export type { WaitpointMintKind };
+
+// The two unions are declared separately, because the engine never imports from the
+// webapp. Nothing pins them together here on purpose: every call site passes this value
+// into an engine method, so a drift fails at those call sites, where the error is local
+// to the code that actually broke.
+
+type WaitpointSystemFlag = "legacy" | "redis";
+
+const mintCache = singleton(
+  "waitpointMintCache",
+  () =>
+    new BoundedTtlCache<WaitpointSystemFlag | null>(
+      env.WAITPOINT_MINT_FLAG_CACHE_TTL_MS,
+      env.WAITPOINT_MINT_FLAG_CACHE_MAX_ENTRIES
+    )
+);
+
+// ENV-BOUND wrapper — the only place env and $replica are read.
+export async function resolveWaitpointMintKind(environment: {
+  organizationId: string;
+  id: string;
+  /** Pass environment.organization.featureFlags from the call site. */
+  orgFeatureFlags?: unknown;
+}): Promise<WaitpointMintKind> {
+  return computeWaitpointMintKind(environment, {
+    globalDefault: env.WAITPOINT_SYSTEM_DEFAULT,
+    onError: (error) =>
+      logger.error("[waitpointMintKind] flag read failed; minting legacy (fail-safe)", { error }),
+    flag: async (orgId, orgFeatureFlags) => {
+      // null is a cached "this org has no override", which must stay distinct from a miss:
+      // BoundedTtlCache reports a stored undefined as a miss, so never store undefined.
+      const cached = mintCache.get(orgId);
+      if (cached !== undefined) {
+        return cached ?? undefined;
+      }
+
+      // Hot-path pass-through: only read the replica when the caller passed no org flags.
+      const overrides =
+        orgFeatureFlags !== undefined
+          ? orgFeatureFlags
+          : (
+              await $replica.organization.findFirst({
+                where: { id: orgId },
+                select: { featureFlags: true },
+              })
+            )?.featureFlags;
+
+      const value = (overrides as Record<string, unknown> | null | undefined)?.[
+        FEATURE_FLAG.waitpointSystem
+      ];
+      const resolved = value === "redis" || value === "legacy" ? value : null;
+
+      mintCache.set(orgId, resolved);
+      return resolved ?? undefined;
+    },
+  });
+}