refactor(run-engine): route checkpoint, delayed, pending-version, and debounce writes through RunStore

Author: d-cs

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

@@ -115,22 +115,20 @@ export class CheckpointSystem {
       }
 
       // Get the run and update the status
-      const run = await this.$.prisma.taskRun.update({
-        where: {
-          id: runId,
-        },
-        data: {
-          status: "WAITING_TO_RESUME",
-        },
-        include: {
-          runtimeEnvironment: {
-            include: {
-              project: true,
-              organization: true,
+      const run = await this.$.runStore.suspendForCheckpoint(
+        runId,
+        {
+          include: {
+            runtimeEnvironment: {
+              include: {
+                project: true,
+                organization: true,
+              },
             },
           },
         },
-      });
+        this.$.prisma
+      );
 
       if (!run) {
         this.$.logger.error("Run not found for createCheckpoint", {
@@ -294,26 +292,24 @@ export class CheckpointSystem {
       }
 
       // Get the run and update the status
-      const run = await this.$.prisma.taskRun.update({
-        where: {
-          id: runId,
-        },
-        data: {
-          status: "EXECUTING",
-        },
-        select: {
-          id: true,
-          status: true,
-          attemptNumber: true,
-          organizationId: true,
-          runtimeEnvironmentId: true,
-          projectId: true,
-          updatedAt: true,
-          createdAt: true,
-          runTags: true,
-          batchId: true,
+      const run = await this.$.runStore.resumeFromCheckpoint(
+        runId,
+        {
+          select: {
+            id: true,
+            status: true,
+            attemptNumber: true,
+            organizationId: true,
+            runtimeEnvironmentId: true,
+            projectId: true,
+            updatedAt: true,
+            createdAt: true,
+            runTags: true,
+            batchId: true,
+          },
         },
-      });
+        this.$.prisma
+      );
 
       if (!run) {
         this.$.logger.error("Run not found for createCheckpoint", {

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

@@ -1160,13 +1160,7 @@ return 0
       updatePayload.runTags = updateData.tags;
     }
 
-    const updatedRun = await prisma.taskRun.update({
-      where: { id: runId },
-      data: updatePayload,
-      include: {
-        associatedWaitpoint: true,
-      },
-    });
+    const updatedRun = await this.$.runStore.rewriteDebouncedRun(runId, updatePayload, prisma);
 
     return updatedRun;
   }

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

@@ -48,26 +48,19 @@ export class DelayedRunSystem {
             throw new ServiceValidationError("Cannot reschedule a run that is not delayed");
           }
 
-          const updatedRun = await prisma.taskRun.update({
-            where: {
-              id: runId,
-            },
-            data: {
+          const updatedRun = await this.$.runStore.rescheduleRun(
+            runId,
+            {
               delayUntil: delayUntil,
-              executionSnapshots: {
-                create: {
-                  engine: "V2",
-                  executionStatus: "DELAYED",
-                  description: "Delayed run was rescheduled to a future date",
-                  runStatus: "DELAYED",
-                  environmentId: snapshot.environmentId,
-                  environmentType: snapshot.environmentType,
-                  projectId: snapshot.projectId,
-                  organizationId: snapshot.organizationId,
-                },
+              snapshot: {
+                environmentId: snapshot.environmentId,
+                environmentType: snapshot.environmentType,
+                projectId: snapshot.projectId,
+                organizationId: snapshot.organizationId,
               },
             },
-          });
+            prisma
+          );
 
           await this.$.worker.reschedule(`enqueueDelayedRun:${updatedRun.id}`, delayUntil);
 
@@ -178,13 +171,13 @@ export class DelayedRunSystem {
 
       const queuedAt = new Date();
 
-      const updatedRun = await this.$.prisma.taskRun.update({
-        where: { id: runId },
-        data: {
-          status: "PENDING",
+      const updatedRun = await this.$.runStore.enqueueDelayedRun(
+        runId,
+        {
           queuedAt,
         },
-      });
+        this.$.prisma
+      );
 
       this.$.eventBus.emit("runEnqueuedAfterDelay", {
         time: new Date(),

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

@@ -129,10 +129,7 @@ export class PendingVersionSystem {
         // Idempotency guard: only flips PENDING_VERSION → PENDING. If another
         // worker already promoted this run between our findMany and the
         // update, count is 0 and we skip the enqueue.
-        const updateResult = await tx.taskRun.updateMany({
-          where: { id: run.id, status: "PENDING_VERSION" },
-          data: { status: "PENDING" },
-        });
+        const updateResult = await this.$.runStore.promotePendingVersionRuns(run.id, tx);
 
         if (updateResult.count === 0) {
           return false;