refactor(run-engine): add WaitpointCoordinator seam with clearRunBlockState

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`, {

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);
+  }
+}

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;
+};