From 93535805080f8ac8b1372bd400dfe0c9c0fd9c08 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 13:17:12 +0100
Subject: [PATCH 1/8] refactor(run-engine): add WaitpointCoordinator seam with
 clearRunBlockState

---
 .../src/engine/systems/waitpointSystem.ts     | 27 +++++-----
 .../legacyPostgresCoordinator.ts              | 51 +++++++++++++++++++
 .../src/engine/waitpointCoordinator/types.ts  | 28 ++++++++++
 3 files changed, 92 insertions(+), 14 deletions(-)
 create mode 100644 internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
 create mode 100644 internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts

diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index 5d5a80772a6..d43e811c6fd 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -7,13 +7,15 @@ import type {
   TaskRunExecutionStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import { Prisma, boundedIn } from "@trigger.dev/database";
+import { Prisma } from "@trigger.dev/database";
 import type { RunStore } from "@internal/run-store";
 import { assertNever } from "assert-never";
 import { nanoid } from "nanoid";
 import { UnclassifiableWaitpointId } from "../errors.js";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
+import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
+import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js";
 import type { EnqueueSystem } from "./enqueueSystem.js";
 import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
 import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
@@ -45,11 +47,17 @@ export class WaitpointSystem {
   private readonly $: SystemResources;
   private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
   private readonly enqueueSystem: EnqueueSystem;
+  private readonly coordinator: WaitpointCoordinator;
 
   constructor(private readonly options: WaitpointSystemOptions) {
     this.$ = options.resources;
     this.executionSnapshotSystem = options.executionSnapshotSystem;
     this.enqueueSystem = options.enqueueSystem;
+    this.coordinator = new LegacyPostgresWaitpointCoordinator({
+      runStore: this.$.runStore,
+      prisma: this.$.prisma,
+      logger: this.$.logger,
+    });
   }
 
   public async clearBlockingWaitpoints({
@@ -59,14 +67,7 @@ export class WaitpointSystem {
     runId: string;
     tx?: PrismaClientOrTransaction;
   }) {
-    // A run's edges co-locate with the run (the edge write routes by runId), so the router routes this
-    // taskRunId-keyed delete to the run's store rather than fanning out. The caller's `tx` is not
-    // forwarded — the delete runs on the owning store's own client (the router never threads a
-    // control-plane tx into a routed write).
-    const deleted = await this.$.runStore.deleteManyTaskRunWaitpoints(
-      { where: { taskRunId: runId } },
-      tx
-    );
+    const deleted = await this.coordinator.clearRunBlockState({ runId, tx });
 
     return deleted.count;
   }
@@ -926,11 +927,9 @@ export class WaitpointSystem {
 
       if (blockingWaitpoints.length > 0) {
         //5. Remove the blocking waitpoints
-        await this.$.runStore.deleteManyTaskRunWaitpoints({
-          where: {
-            taskRunId: runId,
-            id: { in: boundedIn(blockingWaitpoints.map((b) => b.id)) },
-          },
+        await this.coordinator.clearRunBlockState({
+          runId,
+          edgeIds: blockingWaitpoints.map((b) => b.id),
         });
 
         this.$.logger.debug(`continueRunIfUnblocked: removed blocking waitpoints`, {
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
new file mode 100644
index 00000000000..b2e6aadb26b
--- /dev/null
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -0,0 +1,51 @@
+import type { RunStore } from "@internal/run-store";
+import type { Logger } from "@trigger.dev/core/logger";
+import type { PrismaClient } from "@trigger.dev/database";
+import { boundedIn } from "@trigger.dev/database";
+import type { ClearRunBlockStateParams, WaitpointCoordinator } from "./types.js";
+
+export type LegacyPostgresWaitpointCoordinatorOptions = {
+  runStore: RunStore;
+  prisma: PrismaClient;
+  logger: Logger;
+};
+
+/**
+ * Waitpoint coordination against Postgres, through the run-ops store.
+ *
+ * Dependencies are deliberately narrow: no run lock, no worker, no event bus.
+ * That makes "this owns waitpoint state only" structural rather than a convention.
+ */
+export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator {
+  private readonly runStore: RunStore;
+  private readonly prisma: PrismaClient;
+  private readonly logger: Logger;
+
+  constructor(options: LegacyPostgresWaitpointCoordinatorOptions) {
+    this.runStore = options.runStore;
+    this.prisma = options.prisma;
+    this.logger = options.logger;
+  }
+
+  async clearRunBlockState({
+    runId,
+    edgeIds,
+    tx,
+  }: ClearRunBlockStateParams): Promise<{ count: number }> {
+    if (edgeIds) {
+      // Bounded delete of named edges, on the unblock path. No tx: that path is not inside a
+      // caller transaction, and boundedIn caps the id-list arity for Prisma.
+      return this.runStore.deleteManyTaskRunWaitpoints({
+        where: {
+          taskRunId: runId,
+          id: { in: boundedIn(edgeIds) },
+        },
+      });
+    }
+
+    // A run's edges co-locate with the run (the edge write routes by runId), so the router routes
+    // this taskRunId-keyed delete to the run's store rather than fanning out. The caller's `tx` is
+    // passed through: a routing store strips it, and a single store joins it.
+    return this.runStore.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, tx);
+  }
+}
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
new file mode 100644
index 00000000000..8db4b986bbe
--- /dev/null
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -0,0 +1,28 @@
+import type { PrismaClientOrTransaction } from "@trigger.dev/database";
+
+/**
+ * The waitpoint and edge state operations that `WaitpointSystem` delegates.
+ *
+ * Orchestration stays in `WaitpointSystem`: the run lock, snapshot transitions,
+ * worker-job enqueues, event emissions and racepoints. This owns waitpoint and
+ * edge state only, so a non-Postgres implementation can replace it without any
+ * caller learning that it changed.
+ *
+ * The residency hints and `tx` are opaque pass-throughs. Opaque does not mean
+ * type-free — a Prisma type appears here — it means a non-Postgres implementation
+ * never reads the value.
+ */
+export type WaitpointCoordinator = {
+  clearRunBlockState(params: ClearRunBlockStateParams): Promise<{ count: number }>;
+};
+
+export type ClearRunBlockStateParams = {
+  runId: string;
+  /** Edge ids to delete. Omit to clear every edge for the run. */
+  edgeIds?: string[];
+  /**
+   * Forwarded verbatim on the full-clear leg only, and never on the bounded leg
+   * or an edge write. A routing store strips it; a single store joins it.
+   */
+  tx?: PrismaClientOrTransaction;
+};

From a724f32588ef5bb9c8b96fc2c403f474c1d12562 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 13:26:28 +0100
Subject: [PATCH 2/8] refactor(run-engine): move the run block-state read
 behind the coordinator

---
 .../src/engine/systems/waitpointSystem.ts     | 15 +--------------
 .../legacyPostgresCoordinator.ts              | 19 ++++++++++++++++++-
 .../src/engine/waitpointCoordinator/types.ts  | 16 +++++++++++++++-
 3 files changed, 34 insertions(+), 16 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index d43e811c6fd..752dbe90da4 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -683,20 +683,7 @@ export class WaitpointSystem {
 
     return await this.$.runLock.lock("continueRunIfUnblocked", [runId], async () => {
       // 1. Get the any blocking waitpoints
-      const blockingWaitpoints = await this.$.runStore.findManyTaskRunWaitpoints(
-        {
-          where: { taskRunId: runId },
-          select: {
-            id: true,
-            batchId: true,
-            batchIndex: true,
-            waitpoint: {
-              select: { id: true, status: true, type: true, completedAfter: true },
-            },
-          },
-        },
-        this.$.prisma
-      );
+      const blockingWaitpoints = await this.coordinator.readRunBlockState(runId);
 
       // 2. There are blockers still, so do nothing
       if (blockingWaitpoints.some((w) => w.waitpoint.status !== "COMPLETED")) {
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
index b2e6aadb26b..b7ca49f1541 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -2,7 +2,7 @@ import type { RunStore } from "@internal/run-store";
 import type { Logger } from "@trigger.dev/core/logger";
 import type { PrismaClient } from "@trigger.dev/database";
 import { boundedIn } from "@trigger.dev/database";
-import type { ClearRunBlockStateParams, WaitpointCoordinator } from "./types.js";
+import type { ClearRunBlockStateParams, RunBlockEdge, WaitpointCoordinator } from "./types.js";
 
 export type LegacyPostgresWaitpointCoordinatorOptions = {
   runStore: RunStore;
@@ -48,4 +48,21 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     // passed through: a routing store strips it, and a single store joins it.
     return this.runStore.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, tx);
   }
+
+  async readRunBlockState(runId: string): Promise<RunBlockEdge[]> {
+    return this.runStore.findManyTaskRunWaitpoints(
+      {
+        where: { taskRunId: runId },
+        select: {
+          id: true,
+          batchId: true,
+          batchIndex: true,
+          waitpoint: {
+            select: { id: true, status: true, type: true, completedAfter: true },
+          },
+        },
+      },
+      this.prisma
+    );
+  }
 }
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index 8db4b986bbe..f4c63063893 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -1,4 +1,4 @@
-import type { PrismaClientOrTransaction } from "@trigger.dev/database";
+import type { PrismaClientOrTransaction, Waitpoint } from "@trigger.dev/database";
 
 /**
  * The waitpoint and edge state operations that `WaitpointSystem` delegates.
@@ -14,6 +14,7 @@ import type { PrismaClientOrTransaction } from "@trigger.dev/database";
  */
 export type WaitpointCoordinator = {
   clearRunBlockState(params: ClearRunBlockStateParams): Promise<{ count: number }>;
+  readRunBlockState(runId: string): Promise<RunBlockEdge[]>;
 };
 
 export type ClearRunBlockStateParams = {
@@ -26,3 +27,16 @@ export type ClearRunBlockStateParams = {
    */
   tx?: PrismaClientOrTransaction;
 };
+
+/**
+ * One block edge, with the fields the unblock decision reads.
+ *
+ * `batchId` is read by no logic. It rides inside two `logger.debug` payloads
+ * (`waitpointSystem.ts:702-705` and `:936-939`), so removing it changes log output.
+ */
+export type RunBlockEdge = {
+  id: string;
+  batchId: string | null;
+  batchIndex: number | null;
+  waitpoint: Pick<Waitpoint, "id" | "status" | "type" | "completedAfter">;
+};

From e73577ae5f9bb69631503623992b318c4c02cd26 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 13:37:49 +0100
Subject: [PATCH 3/8] refactor(run-engine): move block-edge registration behind
 the coordinator

---
 .../src/engine/systems/waitpointSystem.ts     | 24 +++-----
 .../legacyPostgresCoordinator.ts              | 59 ++++++++++++++++++-
 .../src/engine/waitpointCoordinator/types.ts  | 25 ++++++++
 3 files changed, 92 insertions(+), 16 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index 752dbe90da4..2a9b1d55ac9 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -490,25 +490,19 @@ export class WaitpointSystem {
         this.$.runStore
       );
 
-      // Insert the blocking + historical connections via the run-ops store, routed by the owning
-      // run id so the edge co-resides with the run. Never pinned to the caller's control-plane tx:
-      // that joined `Waitpoint` on the wrong DB and wrote 0 edges. The pending check stays a
-      // SEPARATE store call so it gets its own READ COMMITTED snapshot (see the doc comment above).
-      await this.$.runStore.blockRunWithWaitpointEdges({
+      // Insert the blocking + historical connections and re-check the pending count. The
+      // coordinator keeps these as two separate store statements, in this order, for the READ
+      // COMMITTED reason documented on the method and in the doc comment above.
+      const { pendingCount } = await this.coordinator.registerBlocks({
         runId,
         waitpointIds: $waitpoints,
         projectId,
         spanIdToComplete,
         batchId: batch?.id,
         batchIndex: batch?.index,
+        client: prisma,
       });
 
-      // Check if the run is actually blocked using a separate query (see above). Pass the writer so the
-      // pending re-read is read-your-writes on the owning PRIMARY (a lagging replica can strand the run).
-      // Route by the blocked run id: its blocking waitpoints co-locate with the run, so the router
-      // counts on the run's store and only falls back to the other DB for a cross-tree token.
-      const pendingCount = await this.$.runStore.countPendingWaitpoints($waitpoints, prisma, runId);
-
       const isRunBlocked = pendingCount > 0;
 
       let newStatus: TaskRunExecutionStatus = "SUSPENDED";
@@ -606,10 +600,10 @@ export class WaitpointSystem {
   }): Promise<void> {
     const $waitpoints = typeof waitpoints === "string" ? [waitpoints] : waitpoints;
 
-    // Same routed edge write as blockRunWithWaitpoint, routed by the owning run id. No lock
-    // needed: ON CONFLICT DO NOTHING makes concurrent inserts safe, and the parent snapshot is
-    // already EXECUTING_WITH_WAITPOINTS from blockRunWithCreatedBatch.
-    await this.$.runStore.blockRunWithWaitpointEdges({
+    // Same routed edge write as blockRunWithWaitpoint. No lock needed: ON CONFLICT DO NOTHING
+    // makes concurrent inserts safe, and the parent snapshot is already
+    // EXECUTING_WITH_WAITPOINTS from blockRunWithCreatedBatch. No pending count here.
+    await this.coordinator.registerBlocksLockless({
       runId,
       waitpointIds: $waitpoints,
       projectId,
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
index b7ca49f1541..d310c4e2b9f 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -2,7 +2,13 @@ import type { RunStore } from "@internal/run-store";
 import type { Logger } from "@trigger.dev/core/logger";
 import type { PrismaClient } from "@trigger.dev/database";
 import { boundedIn } from "@trigger.dev/database";
-import type { ClearRunBlockStateParams, RunBlockEdge, WaitpointCoordinator } from "./types.js";
+import type {
+  ClearRunBlockStateParams,
+  RegisterBlocksLocklessParams,
+  RegisterBlocksParams,
+  RunBlockEdge,
+  WaitpointCoordinator,
+} from "./types.js";
 
 export type LegacyPostgresWaitpointCoordinatorOptions = {
   runStore: RunStore;
@@ -65,4 +71,55 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
       this.prisma
     );
   }
+
+  async registerBlocks({
+    client,
+    ...edge
+  }: RegisterBlocksParams): Promise<{ pendingCount: number }> {
+    await this.#writeBlockEdges(edge);
+
+    // Check if the run is actually blocked using a separate query. The separate statement is the
+    // point: under PostgreSQL READ COMMITTED each statement gets its own snapshot, so a
+    // concurrent completion that commits between the edge write and this check is still seen.
+    // It queries ALL requested ids, not just inserted ones: a row that already existed (ON
+    // CONFLICT skipped the insert) but is still PENDING must still block. Pass the caller's
+    // client so the re-read is read-your-writes on the owning PRIMARY, and pass the run id so
+    // the router counts on the run's store instead of fanning out to both DBs.
+    const pendingCount = await this.runStore.countPendingWaitpoints(
+      edge.waitpointIds,
+      client,
+      edge.runId
+    );
+
+    return { pendingCount };
+  }
+
+  async registerBlocksLockless(params: RegisterBlocksLocklessParams): Promise<void> {
+    await this.#writeBlockEdges(params);
+  }
+
+  /**
+   * The edge write, shared by both register paths so they cannot drift.
+   *
+   * Routed by the owning run id so the edge co-resides with the run. Never pinned to a caller
+   * transaction: that joined `Waitpoint` on the wrong DB, wrote 0 edges, and silently never
+   * suspended the parent. The write is idempotent (ON CONFLICT DO NOTHING).
+   */
+  #writeBlockEdges({
+    runId,
+    waitpointIds,
+    projectId,
+    spanIdToComplete,
+    batchId,
+    batchIndex,
+  }: RegisterBlocksLocklessParams): Promise<void> {
+    return this.runStore.blockRunWithWaitpointEdges({
+      runId,
+      waitpointIds,
+      projectId,
+      spanIdToComplete,
+      batchId,
+      batchIndex,
+    });
+  }
 }
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index f4c63063893..faaa77f40ed 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -1,3 +1,4 @@
+import type { ReadClient } from "@internal/run-store";
 import type { PrismaClientOrTransaction, Waitpoint } from "@trigger.dev/database";
 
 /**
@@ -15,6 +16,8 @@ import type { PrismaClientOrTransaction, Waitpoint } from "@trigger.dev/database
 export type WaitpointCoordinator = {
   clearRunBlockState(params: ClearRunBlockStateParams): Promise<{ count: number }>;
   readRunBlockState(runId: string): Promise<RunBlockEdge[]>;
+  registerBlocks(params: RegisterBlocksParams): Promise<{ pendingCount: number }>;
+  registerBlocksLockless(params: RegisterBlocksLocklessParams): Promise<void>;
 };
 
 export type ClearRunBlockStateParams = {
@@ -40,3 +43,25 @@ export type RunBlockEdge = {
   batchIndex: number | null;
   waitpoint: Pick<Waitpoint, "id" | "status" | "type" | "completedAfter">;
 };
+
+export type RegisterBlocksParams = {
+  runId: string;
+  waitpointIds: string[];
+  projectId: string;
+  spanIdToComplete?: string;
+  batchId?: string;
+  batchIndex?: number;
+  /**
+   * Read client for the pending count only. The caller resolves `tx ?? prisma` once
+   * and passes the result, so the writer is used when the caller is inside a
+   * transaction and the pending re-read is read-your-writes on the owning primary.
+   * Never forwarded to the edge write.
+   */
+  client: ReadClient;
+};
+
+/**
+ * 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">;

From c91593fc7d07d371d50f9b7793d48848ed001b32 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 13:47:49 +0100
Subject: [PATCH 4/8] refactor(run-engine): move waitpoint completion behind
 the coordinator

---
 .../src/engine/systems/waitpointSystem.ts     | 81 ++-----------------
 .../legacyPostgresCoordinator.ts              | 78 ++++++++++++++++++
 .../src/engine/waitpointCoordinator/types.ts  | 22 +++++
 3 files changed, 106 insertions(+), 75 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index 2a9b1d55ac9..fc2372b53a1 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -1,4 +1,4 @@
-import { timeoutError, tryCatch } from "@trigger.dev/core/v3";
+import { timeoutError } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import type {
   PrismaClientOrTransaction,
@@ -8,10 +8,8 @@ import type {
   Waitpoint,
 } from "@trigger.dev/database";
 import { Prisma } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
 import { assertNever } from "assert-never";
 import { nanoid } from "nanoid";
-import { UnclassifiableWaitpointId } from "../errors.js";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
 import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
@@ -85,86 +83,19 @@ export class WaitpointSystem {
       isError: boolean;
     };
   }): Promise<Waitpoint> {
-    // Residency store-selection guard. completeWaitpoint arrives with only
-    // (waitpointId, output) — no run id — so the owning run-ops store is selected
-    // by the waitpoint's own residency. In single-DB this is the one store
-    // (no classification). An unclassifiable id throws loud — never default-routes.
-    let store: RunStore;
-    try {
-      store = await this.$.runStore.forWaitpointCompletion(id, { routeKind: "MANUAL" });
-    } catch (error) {
-      this.$.logger.error("completeWaitpoint: unclassifiable waitpointId", {
-        waitpointId: id,
-        error,
-      });
-      throw new UnclassifiableWaitpointId(id, { cause: error });
-    }
-
-    // 1. Complete the Waitpoint (if not completed)
-    const [updateError, updateResult] = await tryCatch(
-      store.updateManyWaitpoints({
-        where: { id, status: "PENDING" },
-        data: {
-          status: "COMPLETED",
-          completedAt: new Date(),
-          output: output?.value,
-          outputType: output?.type,
-          outputIsError: output?.isError,
-        },
-      })
-    );
-
-    if (updateError) {
-      this.$.logger.error("completeWaitpoint: error updating waitpoint:", { updateError });
-      throw updateError;
-    }
-
-    if (updateResult.count === 0) {
-      this.$.logger.info(
-        "completeWaitpoint: attempted to complete a waitpoint that is not PENDING",
-        { waitpointId: id }
-      );
-    }
-
-    // Re-read the just-written row from the RESOLVED store's PRIMARY: the replica (findWaitpoint's
-    // default) can miss it under lag → false "not found" → the parent hangs; this.$.prisma would
-    // instead hit the wrong DB. findWaitpointOnPrimary reads the owning store's primary.
-    const waitpoint = await store.findWaitpointOnPrimary({
-      where: { id },
+    const { waitpoint, blockedRuns } = await this.coordinator.complete({
+      waitpointId: id,
+      output,
     });
 
-    if (!waitpoint) {
-      this.$.logger.error("completeWaitpoint: waitpoint not found", { waitpointId: id });
-      throw new Error("Waitpoint not found");
-    }
-
-    if (waitpoint.status !== "COMPLETED") {
-      this.$.logger.error(`completeWaitpoint: waitpoint is not completed`, {
-        waitpointId: id,
-      });
-      throw new Error("Waitpoint not completed");
-    }
-
-    // 2. Find the TaskRuns blocked by this waitpoint. The edge (TaskRunWaitpoint) co-locates
-    // with its RUN, not this token, so it can live on the OTHER run-ops DB: read via the router
-    // (which fans the waitpointId lookup across both DBs) rather than the token's own `store`,
-    // or a cross-DB blocked run is never found and hangs forever.
-    const affectedTaskRuns = await this.$.runStore.findManyTaskRunWaitpoints(
-      {
-        where: { waitpointId: id },
-        select: { taskRunId: true, spanIdToComplete: true, createdAt: true },
-      },
-      this.$.prisma
-    );
-
-    if (affectedTaskRuns.length === 0) {
+    if (blockedRuns.length === 0) {
       this.$.logger.debug(`completeWaitpoint: no TaskRunWaitpoints found for waitpoint`, {
         waitpointId: id,
       });
     }
 
     // 3. Schedule trying to continue the runs
-    for (const run of affectedTaskRuns) {
+    for (const run of blockedRuns) {
       const jobId = `continueRunIfUnblocked:${run.taskRunId}`;
       //50ms in the future
       const availableAt = new Date(Date.now() + 50);
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
index d310c4e2b9f..de1cebb51c7 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -1,9 +1,13 @@
 import type { RunStore } from "@internal/run-store";
+import { tryCatch } from "@trigger.dev/core/v3";
 import type { Logger } from "@trigger.dev/core/logger";
 import type { PrismaClient } from "@trigger.dev/database";
 import { boundedIn } from "@trigger.dev/database";
+import { UnclassifiableWaitpointId } from "../errors.js";
 import type {
   ClearRunBlockStateParams,
+  CompleteParams,
+  CompleteResult,
   RegisterBlocksLocklessParams,
   RegisterBlocksParams,
   RunBlockEdge,
@@ -98,6 +102,80 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     await this.#writeBlockEdges(params);
   }
 
+  async complete({ waitpointId, output }: CompleteParams): Promise<CompleteResult> {
+    // Residency store-selection guard. complete arrives with only (waitpointId, output) — no run
+    // id — so the owning run-ops store is selected by the waitpoint's own residency. In single-DB
+    // this is the one store (no classification). An unclassifiable id throws loud — never
+    // default-routes. The try wraps ONLY the resolve: widening it would swallow the
+    // "Waitpoint not found" path that a single store relies on.
+    let store: RunStore;
+    try {
+      store = await this.runStore.forWaitpointCompletion(waitpointId, { routeKind: "MANUAL" });
+    } catch (error) {
+      this.logger.error("completeWaitpoint: unclassifiable waitpointId", {
+        waitpointId,
+        error,
+      });
+      throw new UnclassifiableWaitpointId(waitpointId, { cause: error });
+    }
+
+    // 1. Complete the Waitpoint (if not completed)
+    const [updateError, updateResult] = await tryCatch(
+      store.updateManyWaitpoints({
+        where: { id: waitpointId, status: "PENDING" },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(),
+          output: output?.value,
+          outputType: output?.type,
+          outputIsError: output?.isError,
+        },
+      })
+    );
+
+    if (updateError) {
+      this.logger.error("completeWaitpoint: error updating waitpoint:", { updateError });
+      throw updateError;
+    }
+
+    if (updateResult.count === 0) {
+      this.logger.info("completeWaitpoint: attempted to complete a waitpoint that is not PENDING", {
+        waitpointId,
+      });
+    }
+
+    // Re-read the just-written row from the RESOLVED store's PRIMARY: the replica (findWaitpoint's
+    // default) can miss it under lag → false "not found" → the parent hangs. Going back through
+    // the router would re-resolve the store and change the routing, so use the handle.
+    const waitpoint = await store.findWaitpointOnPrimary({
+      where: { id: waitpointId },
+    });
+
+    if (!waitpoint) {
+      this.logger.error("completeWaitpoint: waitpoint not found", { waitpointId });
+      throw new Error("Waitpoint not found");
+    }
+
+    if (waitpoint.status !== "COMPLETED") {
+      this.logger.error(`completeWaitpoint: waitpoint is not completed`, { waitpointId });
+      throw new Error("Waitpoint not completed");
+    }
+
+    // 2. Find the TaskRuns blocked by this waitpoint. The edge (TaskRunWaitpoint) co-locates
+    // with its RUN, not this token, so it can live on the OTHER run-ops DB: read via the router
+    // (which fans the waitpointId lookup across both DBs) rather than the token's own `store`,
+    // or a cross-DB blocked run is never found and hangs forever.
+    const blockedRuns = await this.runStore.findManyTaskRunWaitpoints(
+      {
+        where: { waitpointId },
+        select: { taskRunId: true, spanIdToComplete: true, createdAt: true },
+      },
+      this.prisma
+    );
+
+    return { waitpoint, blockedRuns };
+  }
+
   /**
    * The edge write, shared by both register paths so they cannot drift.
    *
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index faaa77f40ed..e7b9e53a1a3 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -18,6 +18,7 @@ export type WaitpointCoordinator = {
   readRunBlockState(runId: string): Promise<RunBlockEdge[]>;
   registerBlocks(params: RegisterBlocksParams): Promise<{ pendingCount: number }>;
   registerBlocksLockless(params: RegisterBlocksLocklessParams): Promise<void>;
+  complete(params: CompleteParams): Promise<CompleteResult>;
 };
 
 export type ClearRunBlockStateParams = {
@@ -65,3 +66,24 @@ export type RegisterBlocksParams = {
  * one method with a flag, so "the batch path issues no extra query" is structural.
  */
 export type RegisterBlocksLocklessParams = Omit<RegisterBlocksParams, "client">;
+
+export type CompleteParams = {
+  waitpointId: string;
+  output?: {
+    value: string;
+    type?: string;
+    isError: boolean;
+  };
+};
+
+/** One run blocked by the completed waitpoint, with the fields the caller's fan-out loop reads. */
+export type BlockedRun = {
+  taskRunId: string;
+  spanIdToComplete: string | null;
+  createdAt: Date;
+};
+
+export type CompleteResult = {
+  waitpoint: Waitpoint;
+  blockedRuns: BlockedRun[];
+};

From 0909db83d33b302ae6d695d94bff7aee32e7be1b Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 13:59:27 +0100
Subject: [PATCH 5/8] refactor(run-engine): move waitpoint creation and minting
 behind the coordinator

---
 .../src/engine/systems/waitpointSystem.ts     | 232 +++--------------
 .../legacyPostgresCoordinator.ts              | 237 +++++++++++++++++-
 .../src/engine/waitpointCoordinator/types.ts  |  57 +++++
 3 files changed, 333 insertions(+), 193 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index fc2372b53a1..3dbed999445 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -1,5 +1,4 @@
 import { timeoutError } from "@trigger.dev/core/v3";
-import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import type {
   PrismaClientOrTransaction,
   TaskRun,
@@ -7,9 +6,7 @@ import type {
   TaskRunExecutionStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import { Prisma } from "@trigger.dev/database";
 import { assertNever } from "assert-never";
-import { nanoid } from "nanoid";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
 import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
@@ -152,81 +149,27 @@ export class WaitpointSystem {
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
   }) {
-    // Co-location invariant: a DATETIME wait waitpoint lives on the same run-ops DB as the run that
-    // blocks on it (so the block edge's local `Waitpoint` join resolves and completion/resume stay
-    // local). The minted waitpoint id is always a cuid, so without `coLocateWithRunId` the upsert
-    // would always route to LEGACY and a run-ops run on NEW would hang. The (env,idempotencyKey) dedup
-    // is within the owning run/tree (co-resident on one DB), so the dedup probe + rotation target the
-    // SAME store. With no run id (a standalone token has no owning run yet) the lookup falls back to
-    // a cross-DB NEW-then-LEGACY scan and the upsert routes by id-shape. Always routed through the
-    // run store (never a caller tx) so it can never bypass residency onto the wrong DB.
-    const colocate = runId ? { coLocateWithRunId: runId } : undefined;
-    const existingWaitpoint = idempotencyKey
-      ? await this.$.runStore.findWaitpoint(
-          {
-            where: {
-              environmentId,
-              idempotencyKey,
-            },
-          },
-          undefined,
-          colocate
-        )
-      : undefined;
-
-    if (existingWaitpoint) {
-      if (
-        existingWaitpoint.idempotencyKeyExpiresAt &&
-        new Date() > existingWaitpoint.idempotencyKeyExpiresAt
-      ) {
-        //the idempotency key has expired
-        //remove the waitpoint idempotencyKey
-        const rotateArgs = {
-          where: {
-            id: existingWaitpoint.id,
-          },
-          data: {
-            idempotencyKey: nanoid(24),
-            inactiveIdempotencyKey: existingWaitpoint.idempotencyKey,
-          },
-        };
-        await this.$.runStore.updateWaitpoint(rotateArgs, undefined, colocate);
+    const result = await this.coordinator.createDateTimeWaitpoint({
+      runId,
+      projectId,
+      environmentId,
+      completedAfter,
+      idempotencyKey,
+      idempotencyKeyExpiresAt,
+    });
 
-        //let it fall through to create a new waitpoint
-      } else {
-        return { waitpoint: existingWaitpoint, isCached: true };
-      }
+    if (result.kind === "cached") {
+      return { waitpoint: result.waitpoint, isCached: true };
     }
 
-    const upsertArgs = {
-      where: {
-        environmentId_idempotencyKey: {
-          environmentId,
-          idempotencyKey: idempotencyKey ?? nanoid(24),
-        },
-      },
-      create: {
-        ...WaitpointId.generate(),
-        type: "DATETIME" as const,
-        idempotencyKey: idempotencyKey ?? nanoid(24),
-        idempotencyKeyExpiresAt,
-        userProvidedIdempotencyKey: !!idempotencyKey,
-        environmentId,
-        projectId,
-        completedAfter,
-      },
-      update: {},
-    };
-    const waitpoint = await this.$.runStore.upsertWaitpoint(upsertArgs, undefined, colocate);
-
     await this.$.worker.enqueue({
-      id: `finishWaitpoint.${waitpoint.id}`,
+      id: `finishWaitpoint.${result.waitpoint.id}`,
       job: "finishWaitpoint",
-      payload: { waitpointId: waitpoint.id },
+      payload: { waitpointId: result.waitpoint.id },
       availableAt: completedAfter,
     });
 
-    return { waitpoint, isCached: false };
+    return { waitpoint: result.waitpoint, isCached: false };
   }
 
   /** This creates a MANUAL waitpoint, that can be explicitly completed (or failed).
@@ -254,117 +197,35 @@ export class WaitpointSystem {
     // to LEGACY by its cuid id-shape. Ignored when `runId` is set (co-location wins).
     standaloneResidency?: "NEW" | "LEGACY";
   }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
-    // Co-location invariant (see createDateTimeWaitpoint): when a `runId` is supplied the waitpoint
-    // co-locates with that run's DB and the (env,idempotencyKey) dedup is per-run (co-resident). A
-    // standalone token (api.v1.waitpoints.tokens.ts) passes no run id — it is created without an
-    // owner, blocked later by whichever run waits on it (possibly cross-DB, resolved by the
-    // run-co-resident block edge + completion fan-out). With no owner it reads the env mint kind via
-    // `standaloneResidency` so a minted-new env keeps its tokens on NEW; unset, it routes by id-shape. No tx here.
-    const colocate = runId
-      ? { coLocateWithRunId: runId }
-      : standaloneResidency
-        ? { residency: standaloneResidency }
-        : undefined;
-    const existingWaitpoint = idempotencyKey
-      ? await this.$.runStore.findWaitpoint(
-          {
-            where: {
-              environmentId,
-              idempotencyKey,
-            },
-          },
-          undefined,
-          colocate
-        )
-      : undefined;
-
-    if (existingWaitpoint) {
-      if (
-        existingWaitpoint.idempotencyKeyExpiresAt &&
-        new Date() > existingWaitpoint.idempotencyKeyExpiresAt
-      ) {
-        //the idempotency key has expired
-        //remove the waitpoint idempotencyKey
-        await this.$.runStore.updateWaitpoint(
-          {
-            where: {
-              id: existingWaitpoint.id,
-            },
-            data: {
-              idempotencyKey: nanoid(24),
-              inactiveIdempotencyKey: existingWaitpoint.idempotencyKey,
-            },
-          },
-          undefined,
-          colocate
-        );
+    const result = await this.coordinator.createManualWaitpoint({
+      runId,
+      environmentId,
+      projectId,
+      idempotencyKey,
+      idempotencyKeyExpiresAt,
+      timeout,
+      tags,
+      standaloneResidency,
+    });
 
-        //let it fall through to create a new waitpoint
-      } else {
-        return { waitpoint: existingWaitpoint, isCached: true };
-      }
+    if (result.kind === "cached") {
+      return { waitpoint: result.waitpoint, isCached: true };
     }
 
-    const maxRetries = 5;
-    let attempts = 0;
-
-    while (attempts < maxRetries) {
-      try {
-        const waitpoint = await this.$.runStore.upsertWaitpoint(
-          {
-            where: {
-              environmentId_idempotencyKey: {
-                environmentId,
-                idempotencyKey: idempotencyKey ?? nanoid(24),
-              },
-            },
-            create: {
-              ...WaitpointId.generate(),
-              type: "MANUAL",
-              idempotencyKey: idempotencyKey ?? nanoid(24),
-              idempotencyKeyExpiresAt,
-              userProvidedIdempotencyKey: !!idempotencyKey,
-              environmentId,
-              projectId,
-              completedAfter: timeout,
-              tags,
-            },
-            update: {},
-          },
-          undefined,
-          colocate
-        );
-
-        //schedule the timeout
-        if (timeout) {
-          await this.$.worker.enqueue({
-            id: `finishWaitpoint.${waitpoint.id}`,
-            job: "finishWaitpoint",
-            payload: {
-              waitpointId: waitpoint.id,
-              error: JSON.stringify(timeoutError(timeout)),
-            },
-            availableAt: timeout,
-          });
-        }
-
-        return { waitpoint, isCached: false };
-      } catch (error) {
-        if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {
-          // Handle unique constraint violation (conflict)
-          attempts++;
-          if (attempts >= maxRetries) {
-            throw new Error(
-              `Failed to create waitpoint after ${maxRetries} attempts due to conflicts.`
-            );
-          }
-        } else {
-          throw error; // Re-throw other errors
-        }
-      }
+    //schedule the timeout
+    if (timeout) {
+      await this.$.worker.enqueue({
+        id: `finishWaitpoint.${result.waitpoint.id}`,
+        job: "finishWaitpoint",
+        payload: {
+          waitpointId: result.waitpoint.id,
+          error: JSON.stringify(timeoutError(timeout)),
+        },
+        availableAt: timeout,
+      });
     }
 
-    throw new Error(`Failed to create waitpoint after ${maxRetries} attempts due to conflicts.`);
+    return { waitpoint: result.waitpoint, isCached: false };
   }
 
   /**
@@ -864,15 +725,7 @@ export class WaitpointSystem {
     projectId: string;
     environmentId: string;
   }) {
-    return {
-      ...WaitpointId.generate(),
-      type: "RUN" as const,
-      status: "PENDING" as const,
-      idempotencyKey: nanoid(24),
-      userProvidedIdempotencyKey: false,
-      projectId,
-      environmentId,
-    };
+    return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId });
   }
 
   /**
@@ -956,12 +809,9 @@ export class WaitpointSystem {
       // Create waitpoint and link to run atomically
       const waitpointData = this.buildRunAssociatedWaitpoint({ projectId, environmentId });
 
-      // RUN-type within-tree waitpoint that belongs to runId; routes by owning run id.
-      const waitpoint = await this.$.runStore.createWaitpoint({
-        data: {
-          ...waitpointData,
-          completedByTaskRunId: runId,
-        },
+      const waitpoint = await this.coordinator.createAssociatedWaitpoint({
+        runId,
+        data: waitpointData,
       });
 
       // If run has already finished (per snapshot), complete the waitpoint immediately so the parent can resume
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
index de1cebb51c7..97dc8055445 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -1,13 +1,19 @@
 import type { RunStore } from "@internal/run-store";
 import { tryCatch } from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import type { Logger } from "@trigger.dev/core/logger";
-import type { PrismaClient } from "@trigger.dev/database";
-import { boundedIn } from "@trigger.dev/database";
+import type { PrismaClient, Waitpoint } from "@trigger.dev/database";
+import { boundedIn, Prisma } from "@trigger.dev/database";
+import { nanoid } from "nanoid";
 import { UnclassifiableWaitpointId } from "../errors.js";
 import type {
+  AssociatedWaitpointData,
   ClearRunBlockStateParams,
   CompleteParams,
   CompleteResult,
+  CreateDateTimeWaitpointParams,
+  CreateManualWaitpointParams,
+  CreateWaitpointResult,
   RegisterBlocksLocklessParams,
   RegisterBlocksParams,
   RunBlockEdge,
@@ -176,6 +182,233 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     return { waitpoint, blockedRuns };
   }
 
+  async createDateTimeWaitpoint({
+    runId,
+    projectId,
+    environmentId,
+    completedAfter,
+    idempotencyKey,
+    idempotencyKeyExpiresAt,
+  }: CreateDateTimeWaitpointParams): Promise<CreateWaitpointResult> {
+    // Co-location invariant: a DATETIME wait waitpoint lives on the same run-ops DB as the run
+    // that blocks on it. The minted waitpoint id is always a cuid, so without `coLocateWithRunId`
+    // the upsert would always route to LEGACY and a run-ops run on NEW would hang. The
+    // (env,idempotencyKey) dedup is within the owning run/tree, so the dedup probe + rotation
+    // target the SAME store. With no run id the lookup falls back to a cross-DB NEW-then-LEGACY
+    // scan and the upsert routes by id-shape. Always routed through the run store (never a caller
+    // tx) so it can never bypass residency onto the wrong DB.
+    const colocate = runId ? { coLocateWithRunId: runId } : undefined;
+    const existingWaitpoint = idempotencyKey
+      ? await this.runStore.findWaitpoint(
+          {
+            where: {
+              environmentId,
+              idempotencyKey,
+            },
+          },
+          undefined,
+          colocate
+        )
+      : undefined;
+
+    if (existingWaitpoint) {
+      if (
+        existingWaitpoint.idempotencyKeyExpiresAt &&
+        new Date() > existingWaitpoint.idempotencyKeyExpiresAt
+      ) {
+        //the idempotency key has expired
+        //remove the waitpoint idempotencyKey
+        const rotateArgs = {
+          where: {
+            id: existingWaitpoint.id,
+          },
+          data: {
+            idempotencyKey: nanoid(24),
+            inactiveIdempotencyKey: existingWaitpoint.idempotencyKey,
+          },
+        };
+        await this.runStore.updateWaitpoint(rotateArgs, undefined, colocate);
+
+        //let it fall through to create a new waitpoint
+      } else {
+        return { kind: "cached", waitpoint: existingWaitpoint };
+      }
+    }
+
+    // The two `nanoid(24)` calls below are deliberately separate and produce DIFFERENT values:
+    // the upsert `where` key must not match the `create` key, or a guaranteed-miss upsert becomes
+    // a possible update. Do not hoist either to a shared constant.
+    const upsertArgs = {
+      where: {
+        environmentId_idempotencyKey: {
+          environmentId,
+          idempotencyKey: idempotencyKey ?? nanoid(24),
+        },
+      },
+      create: {
+        ...WaitpointId.generate(),
+        type: "DATETIME" as const,
+        idempotencyKey: idempotencyKey ?? nanoid(24),
+        idempotencyKeyExpiresAt,
+        userProvidedIdempotencyKey: !!idempotencyKey,
+        environmentId,
+        projectId,
+        completedAfter,
+      },
+      update: {},
+    };
+    const waitpoint = await this.runStore.upsertWaitpoint(upsertArgs, undefined, colocate);
+
+    return { kind: "created", waitpoint };
+  }
+
+  async createManualWaitpoint({
+    runId,
+    environmentId,
+    projectId,
+    idempotencyKey,
+    idempotencyKeyExpiresAt,
+    timeout,
+    tags,
+    standaloneResidency,
+  }: CreateManualWaitpointParams): Promise<CreateWaitpointResult> {
+    // Co-location invariant (see createDateTimeWaitpoint): when a `runId` is supplied the
+    // waitpoint co-locates with that run's DB and the (env,idempotencyKey) dedup is per-run. A
+    // standalone token passes no run id — it is created without an owner, blocked later by
+    // whichever run waits on it (possibly cross-DB, resolved by the run-co-resident block edge +
+    // completion fan-out). With no owner it reads the env mint kind via `standaloneResidency` so
+    // a minted-new env keeps its tokens on NEW; unset, it routes by id-shape. No tx here.
+    const colocate = runId
+      ? { coLocateWithRunId: runId }
+      : standaloneResidency
+        ? { residency: standaloneResidency }
+        : undefined;
+    const existingWaitpoint = idempotencyKey
+      ? await this.runStore.findWaitpoint(
+          {
+            where: {
+              environmentId,
+              idempotencyKey,
+            },
+          },
+          undefined,
+          colocate
+        )
+      : undefined;
+
+    if (existingWaitpoint) {
+      if (
+        existingWaitpoint.idempotencyKeyExpiresAt &&
+        new Date() > existingWaitpoint.idempotencyKeyExpiresAt
+      ) {
+        //the idempotency key has expired
+        //remove the waitpoint idempotencyKey
+        await this.runStore.updateWaitpoint(
+          {
+            where: {
+              id: existingWaitpoint.id,
+            },
+            data: {
+              idempotencyKey: nanoid(24),
+              inactiveIdempotencyKey: existingWaitpoint.idempotencyKey,
+            },
+          },
+          undefined,
+          colocate
+        );
+
+        //let it fall through to create a new waitpoint
+      } else {
+        return { kind: "cached", waitpoint: existingWaitpoint };
+      }
+    }
+
+    const maxRetries = 5;
+    let attempts = 0;
+
+    while (attempts < maxRetries) {
+      try {
+        // As in createDateTimeWaitpoint, the two `nanoid(24)` calls are deliberately separate and
+        // differ. Both, and `WaitpointId.generate()`, are re-evaluated on every attempt: that is
+        // what makes a retry after a unique-constraint conflict try a fresh key.
+        const waitpoint = await this.runStore.upsertWaitpoint(
+          {
+            where: {
+              environmentId_idempotencyKey: {
+                environmentId,
+                idempotencyKey: idempotencyKey ?? nanoid(24),
+              },
+            },
+            create: {
+              ...WaitpointId.generate(),
+              type: "MANUAL",
+              idempotencyKey: idempotencyKey ?? nanoid(24),
+              idempotencyKeyExpiresAt,
+              userProvidedIdempotencyKey: !!idempotencyKey,
+              environmentId,
+              projectId,
+              completedAfter: timeout,
+              tags,
+            },
+            update: {},
+          },
+          undefined,
+          colocate
+        );
+
+        return { kind: "created", waitpoint };
+      } catch (error) {
+        if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {
+          // Handle unique constraint violation (conflict)
+          attempts++;
+          if (attempts >= maxRetries) {
+            throw new Error(
+              `Failed to create waitpoint after ${maxRetries} attempts due to conflicts.`
+            );
+          }
+        } else {
+          throw error; // Re-throw other errors
+        }
+      }
+    }
+
+    throw new Error(`Failed to create waitpoint after ${maxRetries} attempts due to conflicts.`);
+  }
+
+  mintAssociatedWaitpointData({
+    projectId,
+    environmentId,
+  }: {
+    projectId: string;
+    environmentId: string;
+  }): AssociatedWaitpointData {
+    return {
+      ...WaitpointId.generate(),
+      type: "RUN" as const,
+      status: "PENDING" as const,
+      idempotencyKey: nanoid(24),
+      userProvidedIdempotencyKey: false,
+      projectId,
+      environmentId,
+    };
+  }
+
+  async createAssociatedWaitpoint({
+    runId,
+    data,
+  }: {
+    runId: string;
+    data: AssociatedWaitpointData;
+  }): Promise<Waitpoint> {
+    // RUN-type within-tree waitpoint that belongs to runId; routes by owning run id.
+    return this.runStore.createWaitpoint({
+      data: {
+        ...data,
+        completedByTaskRunId: runId,
+      },
+    });
+  }
+
   /**
    * The edge write, shared by both register paths so they cannot drift.
    *
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index e7b9e53a1a3..36b0ea4b315 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -19,6 +19,16 @@ export type WaitpointCoordinator = {
   registerBlocks(params: RegisterBlocksParams): Promise<{ pendingCount: number }>;
   registerBlocksLockless(params: RegisterBlocksLocklessParams): Promise<void>;
   complete(params: CompleteParams): Promise<CompleteResult>;
+  createDateTimeWaitpoint(params: CreateDateTimeWaitpointParams): Promise<CreateWaitpointResult>;
+  createManualWaitpoint(params: CreateManualWaitpointParams): Promise<CreateWaitpointResult>;
+  mintAssociatedWaitpointData(params: {
+    projectId: string;
+    environmentId: string;
+  }): AssociatedWaitpointData;
+  createAssociatedWaitpoint(params: {
+    runId: string;
+    data: AssociatedWaitpointData;
+  }): Promise<Waitpoint>;
 };
 
 export type ClearRunBlockStateParams = {
@@ -87,3 +97,50 @@ export type CompleteResult = {
   waitpoint: Waitpoint;
   blockedRuns: BlockedRun[];
 };
+
+/**
+ * Discriminated on purpose. The caller enqueues the `finishWaitpoint` job only in the
+ * `created` branch, because today's create methods return before their enqueue on the
+ * cached path. A boolean would let a later edit enqueue on both branches.
+ */
+export type CreateWaitpointResult =
+  | { kind: "cached"; waitpoint: Waitpoint }
+  | { kind: "created"; waitpoint: Waitpoint };
+
+export type CreateDateTimeWaitpointParams = {
+  /** When set, the waitpoint co-locates with this run's DB and the dedup probe targets it. */
+  runId?: string;
+  projectId: string;
+  environmentId: string;
+  completedAfter: Date;
+  idempotencyKey?: string;
+  idempotencyKeyExpiresAt?: Date;
+};
+
+export type CreateManualWaitpointParams = {
+  runId?: string;
+  environmentId: string;
+  projectId: string;
+  idempotencyKey?: string;
+  idempotencyKeyExpiresAt?: Date;
+  timeout?: Date;
+  tags?: string[];
+  /**
+   * For a STANDALONE token (no owning `runId`): the residency the env's mint kind resolves
+   * to. Ignored when `runId` is set, because co-location wins. Only a Postgres
+   * implementation reads this.
+   */
+  standaloneResidency?: "NEW" | "LEGACY";
+};
+
+/** The RUN-waitpoint row data. Pure — no store touch — so the mint is coordinator-owned. */
+export type AssociatedWaitpointData = {
+  id: string;
+  friendlyId: string;
+  type: "RUN";
+  status: "PENDING";
+  idempotencyKey: string;
+  userProvidedIdempotencyKey: false;
+  projectId: string;
+  environmentId: string;
+};

From 43e24de84b4d528287c8d8e5a990e423ce5c9554 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 14:09:25 +0100
Subject: [PATCH 6/8] refactor(run-engine): restore dropped residency comment
 clauses

---
 .../legacyPostgresCoordinator.ts              | 27 ++++++++++---------
 .../src/engine/waitpointCoordinator/types.ts  |  5 ++--
 2 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
index 97dc8055445..d1e48fa4f8d 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
@@ -190,13 +190,14 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     idempotencyKey,
     idempotencyKeyExpiresAt,
   }: CreateDateTimeWaitpointParams): Promise<CreateWaitpointResult> {
-    // Co-location invariant: a DATETIME wait waitpoint lives on the same run-ops DB as the run
-    // that blocks on it. The minted waitpoint id is always a cuid, so without `coLocateWithRunId`
-    // the upsert would always route to LEGACY and a run-ops run on NEW would hang. The
-    // (env,idempotencyKey) dedup is within the owning run/tree, so the dedup probe + rotation
-    // target the SAME store. With no run id the lookup falls back to a cross-DB NEW-then-LEGACY
-    // scan and the upsert routes by id-shape. Always routed through the run store (never a caller
-    // tx) so it can never bypass residency onto the wrong DB.
+    // Co-location invariant: a DATETIME wait waitpoint lives on the same run-ops DB as the run that
+    // blocks on it (so the block edge's local `Waitpoint` join resolves and completion/resume stay
+    // local). The minted waitpoint id is always a cuid, so without `coLocateWithRunId` the upsert
+    // would always route to LEGACY and a run-ops run on NEW would hang. The (env,idempotencyKey) dedup
+    // is within the owning run/tree (co-resident on one DB), so the dedup probe + rotation target the
+    // SAME store. With no run id (a standalone token has no owning run yet) the lookup falls back to
+    // a cross-DB NEW-then-LEGACY scan and the upsert routes by id-shape. Always routed through the
+    // run store (never a caller tx) so it can never bypass residency onto the wrong DB.
     const colocate = runId ? { coLocateWithRunId: runId } : undefined;
     const existingWaitpoint = idempotencyKey
       ? await this.runStore.findWaitpoint(
@@ -272,12 +273,12 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
     tags,
     standaloneResidency,
   }: CreateManualWaitpointParams): Promise<CreateWaitpointResult> {
-    // Co-location invariant (see createDateTimeWaitpoint): when a `runId` is supplied the
-    // waitpoint co-locates with that run's DB and the (env,idempotencyKey) dedup is per-run. A
-    // standalone token passes no run id — it is created without an owner, blocked later by
-    // whichever run waits on it (possibly cross-DB, resolved by the run-co-resident block edge +
-    // completion fan-out). With no owner it reads the env mint kind via `standaloneResidency` so
-    // a minted-new env keeps its tokens on NEW; unset, it routes by id-shape. No tx here.
+    // Co-location invariant (see createDateTimeWaitpoint): when a `runId` is supplied the waitpoint
+    // co-locates with that run's DB and the (env,idempotencyKey) dedup is per-run (co-resident). A
+    // standalone token (api.v1.waitpoints.tokens.ts) passes no run id — it is created without an
+    // owner, blocked later by whichever run waits on it (possibly cross-DB, resolved by the
+    // run-co-resident block edge + completion fan-out). With no owner it reads the env mint kind via
+    // `standaloneResidency` so a minted-new env keeps its tokens on NEW; unset, it routes by id-shape. No tx here.
     const colocate = runId
       ? { coLocateWithRunId: runId }
       : standaloneResidency
diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index 36b0ea4b315..9b89a065c1c 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -126,9 +126,8 @@ export type CreateManualWaitpointParams = {
   timeout?: Date;
   tags?: string[];
   /**
-   * For a STANDALONE token (no owning `runId`): the residency the env's mint kind resolves
-   * to. Ignored when `runId` is set, because co-location wins. Only a Postgres
-   * implementation reads this.
+   * See the `standaloneResidency` param doc on `WaitpointSystem.createManualWaitpoint` for the
+   * full rationale. Only a Postgres implementation reads this.
    */
   standaloneResidency?: "NEW" | "LEGACY";
 };

From bf1c63b2e1b78b0f20b71961221b319f4ba471d5 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 14:34:04 +0100
Subject: [PATCH 7/8] refactor(run-engine): cite the batchId log sites by
 symbol, not line number

The RunBlockEdge comment pointed at stale waitpointSystem.ts line
numbers that no longer match the file after this branch shrank it.
Name the continueRunIfUnblocked method instead so the citation can't
drift again.
---
 .../run-engine/src/engine/waitpointCoordinator/types.ts       | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index 9b89a065c1c..bc2f02e264f 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -45,8 +45,8 @@ export type ClearRunBlockStateParams = {
 /**
  * One block edge, with the fields the unblock decision reads.
  *
- * `batchId` is read by no logic. It rides inside two `logger.debug` payloads
- * (`waitpointSystem.ts:702-705` and `:936-939`), so removing it changes log output.
+ * `batchId` is read by no logic. It rides inside the two `logger.debug` payloads in
+ * `continueRunIfUnblocked`, so removing it changes log output.
  */
 export type RunBlockEdge = {
   id: string;

From b2d56a9c03c9fac1f4f23aa1fac58f3060c87290 Mon Sep 17 00:00:00 2001
From: Dan Sutton <dansutton@trigger.dev>
Date: Fri, 21 Aug 2026 16:10:58 +0100
Subject: [PATCH 8/8] refactor(run-engine): stop exporting the internal
 BlockedRun type
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`BlockedRun` is only named inside types.ts, by CompleteResult. The repo's
knip gate rejects unused exports, so drop the export keyword rather than
add a knip.json exception — nothing outside this file needs the name yet.
---
 .../run-engine/src/engine/waitpointCoordinator/types.ts         | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
index bc2f02e264f..8a50abb7d1c 100644
--- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
+++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
@@ -87,7 +87,7 @@ export type CompleteParams = {
 };
 
 /** One run blocked by the completed waitpoint, with the fields the caller's fan-out loop reads. */
-export type BlockedRun = {
+type BlockedRun = {
   taskRunId: string;
   spanIdToComplete: string | null;
   createdAt: Date;