fix(run-ops split): route idempotent-run waitpoint block and batch-service prisma to the resolved store

- Block the idempotent parent run's waitpoint via the residency-resolved
  dedup client instead of the fallback prisma, so the write lands on the
  store that owns the parent run.
- Pass the caller-provided _prisma into WithRunEngine so a custom store
  isn't silently overridden by the module singleton.
- Throw when a run-backed alert's environment can't be resolved instead of
  marking it SENT, so a transient replica miss doesn't permanently suppress
  the alert.
- Pin splitEnabled:false in the waitpoint passthrough test so it exercises
  single-DB behaviour rather than relying on ksuid residency.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -287,7 +287,7 @@ export class IdempotencyKeyConcern {
                 : undefined,
               projectId: request.environment.projectId,
               organizationId: request.environment.organizationId,
-              tx: this.prisma,
+              tx: dedupClient,
             });
           }
         );

apps/webapp/app/runEngine/services/batchTrigger.server.ts

@@ -71,7 +71,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
     protected readonly _prisma: PrismaClientOrTransaction = prisma,
     engine?: RunEngine
   ) {
-    super({ prisma, engine });
+    super({ prisma: _prisma, engine });
 
     // Eric note: We need to force sequential processing because when doing parallel, we end up with high-contention on the parent run lock
     // becuase we are triggering a lot of runs at once, and each one is trying to lock the parent run.

apps/webapp/app/v3/services/alerts/deliverAlert.server.ts

@@ -190,26 +190,31 @@ export class DeliverAlertService extends BaseService {
         const env = await this.#controlPlaneResolver.resolveAuthenticatedEnv(
           resolvedTaskRun.runtimeEnvironmentId
         );
+
+        if (!env) {
+          throw new Error(
+            `Could not resolve environment ${resolvedTaskRun.runtimeEnvironmentId} for alert ${alertId}`
+          );
+        }
+
         const lockedWorker = await this.#controlPlaneResolver.resolveRunLockedWorker({
           lockedById: resolvedTaskRun.lockedById,
           lockedToVersionId: resolvedTaskRun.lockedToVersionId,
         });
 
-        if (env) {
-          taskRun = {
-            ...resolvedTaskRun,
-            runtimeEnvironment: { type: env.type, branchName: env.branchName },
-            lockedBy: lockedWorker?.lockedBy
-              ? {
-                  filePath: lockedWorker.lockedBy.filePath,
-                  exportName: lockedWorker.lockedBy.exportName,
-                }
-              : null,
-            lockedToVersion: lockedWorker?.lockedToVersion
-              ? { version: lockedWorker.lockedToVersion.version }
-              : null,
-          };
-        }
+        taskRun = {
+          ...resolvedTaskRun,
+          runtimeEnvironment: { type: env.type, branchName: env.branchName },
+          lockedBy: lockedWorker?.lockedBy
+            ? {
+                filePath: lockedWorker.lockedBy.filePath,
+                exportName: lockedWorker.lockedBy.exportName,
+              }
+            : null,
+          lockedToVersion: lockedWorker?.lockedToVersion
+            ? { version: lockedWorker.lockedToVersion.version }
+            : null,
+        };
       }
     }
 

apps/webapp/test/resolveWaitpointThroughReadThrough.readthrough.test.ts

@@ -230,6 +230,7 @@ describe("resolveWaitpointThroughReadThrough (hetero PG14 legacy + dedicated run
         environmentId: environment.id,
         read: read(id, environment.id),
         deps: {
+          splitEnabled: false,
           newClient: single.handle,
           legacyReplica: legacy.handle,
         },