refactor(run-engine): move the BATCH waitpoint create onto the coordinator seam

blockRunWithCreatedBatch built its waitpoint with runStore.createWaitpoint
directly, so it had no arm to route to. It now goes through the coordinator.

The P2002 catch moves to the legacy arm, where it belongs: it is the
duplicate-batch contract for a unique index, and it is dead against a store that
reports a duplicate through NX instead. Leaving it wrapped around a store create
would read a genuine store error as a duplicate batch.

The block step keeps its own P2002 catch. The previous shape wrapped the create
and the block in one try, so a P2002 from either returned null; narrowing that
here would be a behaviour change smuggled into an extraction.

Seam also gains the mint kind on the create params and batchWaitpointId on the
lockless params. Both are pinned to their legacy values at every call site, so
behaviour is unchanged.

Author: d-cs

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

@@ -26,7 +26,6 @@ import {
   generateInternalId,
   parseNaturalLanguageDurationInMs,
   RunId,
-  WaitpointId,
 } from "@trigger.dev/core/v3/isomorphic";
 import {
   type PrismaClient,
@@ -1849,22 +1848,19 @@ export class RunEngine {
     organizationId: string;
     tx?: PrismaClientOrTransaction;
   }): Promise<Waitpoint | null> {
-    try {
-      const waitpoint = await this.runStore.createWaitpoint(
-        {
-          data: {
-            ...WaitpointId.generate(),
-            type: "BATCH",
-            idempotencyKey: batchId,
-            userProvidedIdempotencyKey: false,
-            completedByBatchId: batchId,
-            environmentId,
-            projectId,
-          },
-        },
-        tx
-      );
+    const waitpoint = await this.waitpointSystem.createBatchWaitpoint({
+      batchId,
+      environmentId,
+      projectId,
+      tx,
+    });
 
+    // Duplicate batch: the coordinator already reported it.
+    if (!waitpoint) {
+      return null;
+    }
+
+    try {
       await this.blockRunWithWaitpoint({
         runId,
         waitpoints: waitpoint.id,
@@ -1873,19 +1869,17 @@ export class RunEngine {
         batch: { id: batchId },
         // No tx: the block edge routes to the run's owning DB, not the control-plane tx.
       });
-
-      return waitpoint;
     } catch (error) {
-      if (error instanceof Prisma.PrismaClientKnownRequestError) {
-        // duplicate idempotency key
-        if (error.code === "P2002") {
-          return null;
-        } else {
-          throw error;
-        }
+      // The previous shape wrapped the create AND the block in one catch, so a P2002 from
+      // the block step also returned null. Kept deliberately: narrowing it here would be a
+      // behaviour change smuggled into an extraction.
+      if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {
+        return null;
       }
       throw error;
     }
+
+    return waitpoint;
   }
 
   async tryCompleteBatch({ batchId }: { batchId: string }): Promise<void> {

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

@@ -153,6 +153,8 @@ export class WaitpointSystem {
     idempotencyKeyExpiresAt?: Date;
   }) {
     const result = await this.coordinator.createDateTimeWaitpoint({
+      // Pinned until the mint flag reaches this entry point.
+      mintKind: "legacy",
       runId,
       projectId,
       environmentId,
@@ -201,6 +203,8 @@ 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",
       runId,
       environmentId,
       projectId,
@@ -731,6 +735,21 @@ export class WaitpointSystem {
     }); // end of runlock
   }
 
+  /**
+   * The BATCH waitpoint for a batch. Returns null when the batch already has one.
+   *
+   * mintKind is pinned to legacy until the mint flag is threaded through the batch entry
+   * point; the store arm is unreachable from here until then.
+   */
+  public async createBatchWaitpoint(params: {
+    batchId: string;
+    environmentId: string;
+    projectId: string;
+    tx?: PrismaClientOrTransaction;
+  }): Promise<Waitpoint | null> {
+    return this.coordinator.createBatchWaitpoint({ ...params, mintKind: "legacy" });
+  }
+
   public buildRunAssociatedWaitpoint({
     projectId,
     environmentId,

internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts

@@ -8,6 +8,7 @@ import { nanoid } from "nanoid";
 import { UnclassifiableWaitpointId } from "../errors.js";
 import type {
   AssociatedWaitpointData,
+  CreateBatchWaitpointParams,
   ClearRunBlockStateParams,
   CompleteParams,
   CompleteResult,
@@ -332,6 +333,43 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     return { kind: "created", waitpoint };
   }
 
+  /**
+   * The BATCH waitpoint for a batch, keyed on the batch id as its idempotency key.
+   *
+   * The P2002 catch IS the duplicate-batch contract: a second call for the same batch
+   * collides on the idempotencyKey unique index, and null is the caller's "this batch
+   * already has one" signal rather than an error. It stays on this arm because the code
+   * is dead against a non-Postgres store, where NX reports the duplicate instead.
+   */
+  async createBatchWaitpoint({
+    batchId,
+    environmentId,
+    projectId,
+    tx,
+  }: CreateBatchWaitpointParams): Promise<Waitpoint | null> {
+    try {
+      return await this.runStore.createWaitpoint(
+        {
+          data: {
+            ...WaitpointId.generate(),
+            type: "BATCH",
+            idempotencyKey: batchId,
+            userProvidedIdempotencyKey: false,
+            completedByBatchId: batchId,
+            environmentId,
+            projectId,
+          },
+        },
+        tx
+      );
+    } catch (error) {
+      if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {
+        return null;
+      }
+      throw error;
+    }
+  }
+
   async createManualWaitpoint({
     runId,
     environmentId,

internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts

@@ -24,6 +24,7 @@ export type WaitpointCoordinator = {
   complete(params: CompleteParams): Promise<CompleteResult>;
   createDateTimeWaitpoint(params: CreateDateTimeWaitpointParams): Promise<CreateWaitpointResult>;
   createManualWaitpoint(params: CreateManualWaitpointParams): Promise<CreateWaitpointResult>;
+  createBatchWaitpoint(params: CreateBatchWaitpointParams): Promise<Waitpoint | null>;
   mintAssociatedWaitpointData(params: {
     projectId: string;
     environmentId: string;
@@ -34,6 +35,23 @@ export type WaitpointCoordinator = {
   }): Promise<Waitpoint>;
 };
 
+/**
+ * Which coordinator mints a NEW waitpoint. Structurally identical to the webapp's own
+ * WaitpointMintKind; re-declared because the engine never imports from the webapp.
+ *
+ * Read at the mint and never again — every later operation routes by the minted id's shape.
+ */
+export type WaitpointMintKind = "legacy" | "store";
+
+export type CreateBatchWaitpointParams = {
+  batchId: string;
+  environmentId: string;
+  projectId: string;
+  mintKind: WaitpointMintKind;
+  /** Legacy arm only: the create may join a caller transaction. A store arm ignores it. */
+  tx?: PrismaClientOrTransaction;
+};
+
 export type ReadCompletionEnvelopesParams = {
   runId: string;
   /** The DISTINCT completed waitpoint ids to source. Result order is not meaningful. */
@@ -110,7 +128,15 @@ export type RegisterBlocksParams = {
  * The lockless variant writes the edge and does not count. Two methods rather than
  * one method with a flag, so "the batch path issues no extra query" is structural.
  */
-export type RegisterBlocksLocklessParams = Omit<RegisterBlocksParams, "client">;
+export type RegisterBlocksLocklessParams = Omit<RegisterBlocksParams, "client"> & {
+  /**
+   * The parent's BATCH waitpoint id. A store arm asserts it is present and PENDING on the
+   * run's shard before writing any item edge, so the run's pending set can never be
+   * momentarily empty mid-absorb. Neither TLA+ campaign models this, so the assertion is
+   * the only protection. A legacy arm ignores it.
+   */
+  batchWaitpointId?: string;
+};
 
 export type CompleteParams = {
   waitpointId: string;
@@ -143,6 +169,7 @@ export type CreateWaitpointResult =
   | { kind: "created"; waitpoint: Waitpoint };
 
 export type CreateDateTimeWaitpointParams = {
+  mintKind: WaitpointMintKind;
   /** When set, the waitpoint co-locates with this run's DB and the dedup probe targets it. */
   runId?: string;
   projectId: string;
@@ -153,6 +180,7 @@ export type CreateDateTimeWaitpointParams = {
 };
 
 export type CreateManualWaitpointParams = {
+  mintKind: WaitpointMintKind;
   runId?: string;
   environmentId: string;
   projectId: string;