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/ApiBatchResultsPresenter.server.ts

@@ -1,5 +1,5 @@
 import type { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
-import { ownerEngine } from "@trigger.dev/core/v3/isomorphic";
+import { resolveShard, type ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import {
   $replica,
   type PrismaClientOrTransaction,
@@ -13,6 +13,8 @@ import { runStore as defaultRunStore } from "~/v3/runStore.server";
 import { BasePresenter } from "./basePresenter.server";
 
 import { boundedIn } from "@trigger.dev/database";
+import { runOpsShardReplicas } from "~/v3/runOpsMigration/shardHandles.server";
+import { logger } from "~/services/logger.server";
 /**
  * Run-ops read-through wiring. All optional; absent (or `splitEnabled` falsy) collapses `call` to
  * passthrough. `legacyReplica` is a READ REPLICA handle only — there is NO legacy-primary field.
@@ -21,6 +23,8 @@ type ApiBatchResultsReadThroughDeps = {
   splitEnabled?: boolean;
   newClient?: PrismaReplicaClient;
   legacyReplica?: PrismaReplicaClient;
+  /** Gen-2 shard replicas by shard char; empty (RUN_OPS_SHARDS unset) keeps today's behaviour. */
+  shardReplicas?: ReadonlyMap<ShardKey, PrismaReplicaClient>;
   isPastRetention?: (runId: string) => boolean;
 };
 
@@ -181,16 +185,57 @@ export class ApiBatchResultsPresenter extends BasePresenter {
 
     const taskRunIds = batchRun.items.map((item) => item.taskRunId);
 
-    const newRows = (await newClient.taskRun.findMany({
-      where: { id: { in: boundedIn(taskRunIds) } },
-      select: memberRunSelect,
-    })) as TaskRunWithAttempts[];
+    // A gen-2 id is directly routable to its own shard, so it must not join the gen-1 read:
+    // it would miss there, and (being dedicated-family) never reach the legacy probe either.
+    const shardReplicas = this.readThrough?.shardReplicas ?? runOpsShardReplicas;
+    const genOneIds: string[] = [];
+    const idsByShard = new Map<ShardKey, string[]>();
+    for (const id of taskRunIds) {
+      const shardKey = resolveShard(id);
+      if (shardKey === "new" || shardKey === "legacy") {
+        genOneIds.push(id);
+      } else if (shardReplicas.has(shardKey)) {
+        const group = idsByShard.get(shardKey);
+        group ? group.push(id) : idsByShard.set(shardKey, [id]);
+      } else {
+        // Not routable and not a gen-1 shape. A gen-1 store is the wrong database, and a
+        // dedicated-family id never reaches the legacy probe, so falling back there would
+        // drop the member silently. Drop it loudly instead.
+        logger.error("ApiBatchResultsPresenter: gen-2 member on an unconfigured shard key", {
+          runId: id,
+          shardKey,
+          configured: [...shardReplicas.keys()],
+        });
+      }
+    }
+
+    const newRows = (
+      genOneIds.length > 0
+        ? ((await newClient.taskRun.findMany({
+            where: { id: { in: boundedIn(genOneIds) } },
+            select: memberRunSelect,
+          })) as TaskRunWithAttempts[])
+        : []
+    ).concat(
+      (
+        await Promise.all(
+          [...idsByShard.entries()].map(
+            async ([shardKey, ids]) =>
+              (await shardReplicas.get(shardKey)!.taskRun.findMany({
+                where: { id: { in: boundedIn(ids) } },
+                select: memberRunSelect,
+              })) as TaskRunWithAttempts[]
+          )
+        )
+      ).flat()
+    );
     const runsById = new Map(newRows.map((run) => [run.id, run]));
 
-    // A run-ops id can only live on NEW, so only misses that AREN'T run-ops-shaped are candidates
-    // for the legacy probe — mirrors readThroughRun's per-id "NEW residency skips legacy" rule.
-    const legacyCandidateIds = taskRunIds.filter(
-      (id) => !runsById.has(id) && ownerEngine(id) !== "NEW"
+    // A dedicated-family id (gen-1 v1 or gen-2) can only live on its own store, so only
+    // misses that AREN'T dedicated-shaped are candidates for the legacy probe — mirrors
+    // readThroughRun's per-id "dedicated residency skips legacy" rule.
+    const legacyCandidateIds = genOneIds.filter(
+      (id) => !runsById.has(id) && resolveShard(id) === "legacy"
     );
     if (legacyCandidateIds.length > 0) {
       const legacyRows = (await legacyReplica.taskRun.findMany({

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,
@@ -93,7 +94,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/routes/engine.v1.runs.$runFriendlyId.waitpoints.tokens.$waitpointFriendlyId.wait.ts

@@ -38,7 +38,14 @@ const { action } = createActionApiRoute(
       });
 
       if (!waitpoint) {
-        throw json({ error: "Waitpoint not found" }, { status: 404 });
+        // Retryable: a miss here can be replica lag. resolveWaitpointThroughReadThrough
+        // deliberately does not read the legacy primary, so it relies on the caller retrying.
+        // A plain 404 is not retried by the SDK, which would turn a transient miss into a
+        // permanent failure.
+        throw json(
+          { error: "Waitpoint not found" },
+          { status: 404, headers: { "x-should-retry": "true" } }
+        );
       }
 
       const _result = await engine.blockRunWithWaitpoint({
@@ -55,6 +62,11 @@ const { action } = createActionApiRoute(
         { status: 200 }
       );
     } catch (error) {
+      // A Response thrown inside the try is a deliberate status (the 404 above), not a
+      // failure. Re-throw it untouched, or every intentional 4xx here becomes a 500.
+      if (error instanceof Response) {
+        throw error;
+      }
       logger.error("Failed to wait for waitpoint", { runId, waitpointId, error });
       throw json({ error: "Failed to wait for waitpoint token" }, { status: 500 });
     }

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -1,4 +1,4 @@
-import { ownerEngine, RunId } from "@trigger.dev/core/v3/isomorphic";
+import { resolveShard, RunId, type ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClientOrTransaction, TaskRun, Waitpoint } from "@trigger.dev/database";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
@@ -13,9 +13,10 @@ import { computeClaimTtlSeconds } from "~/v3/mollifier/claimTtl";
 import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server";
 import { runStore } from "~/v3/runStore.server";
 import { runOpsLegacyPrisma, runOpsNewPrisma } from "~/db.server";
+import { runOpsShardWriters } from "~/v3/runOpsMigration/shardHandles.server";
 import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server";
 import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
-import { resolveIdempotencyDedupClient } from "./idempotencyResidency.server";
+import { clientForShardKey, resolveIdempotencyDedupClient } from "./idempotencyResidency.server";
 import type { TraceEventConcern, TriggerTaskRequest } from "../types";
 
 // In-memory per-org mollifier-enabled check, shared with `evaluateGate`
@@ -32,6 +33,16 @@ const resolveOrgMollifierFlag = makeResolveMollifierFlag();
 // PG's unique index as the backstop.
 const MAX_CLEARED_WINNER_REACQUIRES = 5;
 
+// The store that owns a shard key. A function, not a map: the handles are module constants and
+// `runOpsShardWriters` is already keyed, so a second structure would add an allocation and, if
+// memoised, mutable module state. Reading them lazily also keeps this module importable by
+// triggerTask under a `~/db.server` mock that omits them.
+function idempotencyClientFor(shardKey: ShardKey): PrismaClientOrTransaction | undefined {
+  if (shardKey === "legacy") return runOpsLegacyPrisma;
+  if (shardKey === "new") return runOpsNewPrisma;
+  return runOpsShardWriters.get(shardKey);
+}
+
 // Claim ownership context returned to the caller when the
 // IdempotencyKeyConcern won a pre-gate claim. Caller MUST publish the
 // winning runId on pipeline success (`publishClaim`) or release the
@@ -172,12 +183,9 @@ export class IdempotencyKeyConcern {
       {
         isSplitEnabled,
         fallbackClient: this.prisma,
-        newClient: runOpsNewPrisma,
-        legacyClient: runOpsLegacyPrisma,
+        clientFor: idempotencyClientFor,
         resolveMintKind: resolveRunIdMintKind,
-        // `isMigrated` is intentionally omitted: until a child of a swept
-        // legacy-id parent can be born on the new DB, the swept-marker override
-        // would never change the answer, so a child routes by parent id-shape.
+        logger,
       }
     );
 
@@ -640,12 +648,15 @@ export class IdempotencyKeyConcern {
     } catch {
       return null;
     }
-    let client: PrismaClientOrTransaction;
-    try {
-      client = ownerEngine(internalId) === "NEW" ? runOpsNewPrisma : runOpsLegacyPrisma;
-    } catch {
-      client = this.prisma;
-    }
+    // The routing store routes by id and never forwards this object, so its identity only
+    // signals read-your-writes. Resolving it through the shard map keeps the two idempotency
+    // call sites in agreement and stops this reading as gen-2-unaware.
+    const client = clientForShardKey(
+      resolveShard(internalId),
+      idempotencyClientFor,
+      this.prisma,
+      logger
+    );
     return runStore.findRun(
       { id: internalId, runtimeEnvironmentId: environmentId },
       { include: { associatedWaitpoint: true } },