test(run-store,run-engine): cover dedicated snapshot-write atomicity and drop an unused engine

Add atomicity coverage for the dedicated (#new) leg of createExecutionSnapshot
and lockRunToWorker: both must roll the snapshot back when the CompletedWaitpoint
join insert fails, so a lagging replica can never serve a waitpoint-less resume.
The existing coverage only exercised the legacy _completedWaitpoints path.

Also drop an unused RunEngine from the duplicate-order repair test; it drives
getExecutionSnapshotsSince directly and needs no Redis or worker resources.

Author: d-cs

internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts

@@ -1473,24 +1473,10 @@ describe("RunEngine getSnapshotsSince", () => {
   // fires on every poll for such a snapshot even against a caught-up replica - silently defeating offload.
   containerTest(
     "does not repair from the primary when completedWaitpointOrder has duplicates but the join is complete",
-    async ({ prisma, redisOptions }) => {
+    async ({ prisma }) => {
       const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
-      const engine = new RunEngine({
-        prisma,
-        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
-        queue: { redis: redisOptions },
-        runLock: { redis: redisOptions },
-        machines: {
-          defaultMachine: "small-1x",
-          machines: {
-            "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
-          },
-          baseCostInCents: 0.0001,
-        },
-        tracer: trace.getTracer("test", "0.0.0"),
-      });
-
-      try {
+      // Drives getExecutionSnapshotsSince directly, so it needs no RunEngine (Redis/worker) at all.
+      {
         const scenario = await setupTestScenario(prisma, authenticatedEnvironment, {
           totalWaitpoints: 1,
           outputSizeKB: 1,
@@ -1538,8 +1524,6 @@ describe("RunEngine getSnapshotsSince", () => {
         expect(latest.completedWaitpoints.map((w) => w.id)).toEqual([waitpointId, waitpointId]);
         // No spurious primary read: distinct(order) === join count, so the repair must not fire.
         expect(repairCalls.queryRaw).toBe(0);
-      } finally {
-        await engine.quit();
       }
     }
   );

internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts

@@ -594,3 +594,115 @@ describe("createRun / createFailedRun write the run and its associated waitpoint
     }
   );
 });
+
+// The dedicated (#new) leg connects completed waitpoints through the `CompletedWaitpoint` join table
+// (createMany), where the legacy leg uses the implicit `_completedWaitpoints` M2M. Both must commit the
+// snapshot and its links together: a snapshot that commits before its links can be read waitpoint-less
+// from a lagging replica, and the runner's EXECUTING branch no-ops on an empty set -> the resume hangs.
+describe("createExecutionSnapshot / lockRunToWorker write the snapshot and its links atomically (dedicated)", () => {
+  heteroRunOpsPostgresTest(
+    "createExecutionSnapshot rolls the snapshot back if the CompletedWaitpoint insert fails",
+    async ({ prisma17 }) => {
+      const newStore = makeDedicatedStore(prisma17);
+      const env = await seedEnvironment(prisma17, "dedicated", "ces_ded");
+      const runId = `run_${NEW_ID_26}`;
+      await newStore.createRun(
+        buildCreateRunInput({
+          runId,
+          friendlyId: "run_ces_ded",
+          organizationId: env.organization.id,
+          projectId: env.project.id,
+          runtimeEnvironmentId: env.environment.id,
+        })
+      );
+
+      // Force the dedicated join insert (completedWaitpoint.createMany) to fail mid-write.
+      await prisma17.$executeRawUnsafe('DROP TABLE "CompletedWaitpoint"');
+
+      await expect(
+        // Base client as `tx` = how the engine threads its base prisma through
+        // (continueRunIfUnblocked -> executionSnapshotSystem.createExecutionSnapshot(prisma, ...)).
+        // It is NOT an interactive transaction, so the store must still open its own to stay atomic.
+        newStore.createExecutionSnapshot(
+          {
+            run: { id: runId, status: "EXECUTING", attemptNumber: 1 },
+            snapshot: {
+              executionStatus: "EXECUTING_WITH_WAITPOINTS",
+              description: "Run was blocked by a waitpoint.",
+            },
+            environmentId: env.environment.id,
+            environmentType: "DEVELOPMENT",
+            projectId: env.project.id,
+            organizationId: env.project.id,
+            completedWaitpoints: [{ id: `wp_${NEW_ID_26}`, index: 0 }],
+          },
+          prisma17 as never
+        )
+      ).rejects.toThrow();
+
+      const snap = await prisma17.taskRunExecutionSnapshot.findFirst({
+        where: { runId, executionStatus: "EXECUTING_WITH_WAITPOINTS" },
+      });
+      expect(snap).toBeNull();
+    }
+  );
+
+  heteroRunOpsPostgresTest(
+    "lockRunToWorker rolls the snapshot and run lock back if the CompletedWaitpoint insert fails",
+    async ({ prisma17 }) => {
+      const newStore = makeDedicatedStore(prisma17);
+      const env = await seedEnvironment(prisma17, "dedicated", "lock_ded");
+      const runId = `run_${NEW_ID_26}`;
+      await newStore.createRun(
+        buildCreateRunInput({
+          runId,
+          friendlyId: "run_lock_ded",
+          organizationId: env.organization.id,
+          projectId: env.project.id,
+          runtimeEnvironmentId: env.environment.id,
+        })
+      );
+      const prior = await prisma17.taskRunExecutionSnapshot.findFirstOrThrow({ where: { runId } });
+
+      await prisma17.$executeRawUnsafe('DROP TABLE "CompletedWaitpoint"');
+
+      const snapshotId = `snap_${NEW_ID_26}`;
+      await expect(
+        // lockedById/lockedToVersionId/lockedQueueId are FK-free scalars on the dedicated subset, so
+        // synthetic ids are fine; the base client as `tx` mirrors the dequeue path (no interactive tx).
+        newStore.lockRunToWorker(
+          runId,
+          {
+            lockedAt: new Date(),
+            lockedById: `bwt_${NEW_ID_26}`,
+            lockedToVersionId: `bw_${NEW_ID_26}`,
+            lockedQueueId: `queue_${NEW_ID_26}`,
+            startedAt: new Date(),
+            baseCostInCents: 5,
+            machinePreset: "small-1x",
+            taskVersion: "20260601.1",
+            sdkVersion: "3.0.0",
+            cliVersion: "3.0.0",
+            maxDurationInSeconds: null,
+            snapshot: {
+              id: snapshotId,
+              previousSnapshotId: prior.id,
+              environmentId: env.environment.id,
+              environmentType: "DEVELOPMENT",
+              projectId: env.project.id,
+              organizationId: env.project.id,
+              completedWaitpointIds: [`wp_${NEW_ID_26}`],
+              completedWaitpointOrder: [`wp_${NEW_ID_26}`],
+            },
+          },
+          prisma17 as never
+        )
+      ).rejects.toThrow();
+
+      const snap = await prisma17.taskRunExecutionSnapshot.findUnique({ where: { id: snapshotId } });
+      expect(snap).toBeNull();
+      const run = await prisma17.taskRun.findUniqueOrThrow({ where: { id: runId } });
+      expect(run.status).not.toBe("DEQUEUED");
+    }
+  );
+});