fix: read-your-writes + global-scope idempotency correctness under the run-ops split (#4284)

## What & why

Two related correctness fixes for the run-ops DB split. Under the split,
run-store reads can route to a **lagging read replica**; a just-written
run/waitpoint/batch can then be missed, causing a wrong decision.

**1. Read-your-writes → owning primary.** Surfaced first as an
intermittent `wait.until({ idempotencyKey })` re-wait on retry. Auditing
the run-store read surface found the same class at sibling sites (some
gating mutations or returning spurious 404s, others
tolerable/self-healing). Reads that must observe their own writes now
route to the owning **primary**
(`findRun`/`findWaitpoint`/`findBatchTaskRunByFriendlyId` →
`*OnPrimary`, a primary re-read on a miss, or a retryable 404 where the
SDK polls). Read-view reads stay on the replica. All additive — the
happy path is unchanged.

**2. Global-scope idempotency across the split.** A `global`-scope key
carries no per-run salt, so the same `(env, task, key)` triggered
concurrently from parents resident on **different** run-ops DBs could
dedup-miss on each DB and create a duplicate (the per-DB unique index
can't enforce cross-DB uniqueness). Such triggers (global scope, or
scope-absent, while split is active) are serialized through the existing
Redis idempotency claim, the loser resolves the winner by id across both
DBs, and the claim is reacquired on the expired/failed
clear-and-recreate path. `run`/`attempt` scope embed the run id and
never contend.

## Stacked for review

This is the **base** of a 2-PR stack, split so review is easier:
- **This PR** — production code only (34 files).
- **Stacked tests PR →
https://github.com/triggerdotdev/trigger.dev/pull/4285** — the
caller-driven guards (55 test files) on top of this branch.

## Validation

Local run-ops split, **both 2-DB and 3-DB**, fresh boot on this branch:
SDK canary 64/71 (only the known concurrency/input-streams/s3 failures),
quarantine sweep **0 unexpected** (340 pass / 16 known / 4 local) in
each topology, dashboard e2e 0 failed. No product regressions.

apps/webapp/app/env.server.ts

@@ -1311,6 +1311,9 @@ const EnvironmentSchema = z
     // claim TTL), how long a waiter blocks before timing out, and the
     // waiter poll interval.
     TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS: z.coerce.number().int().positive().default(30),
+    // Pipeline floor: the claim never shrinks below this even for a short customer key TTL, so it
+    // can't expire mid-pipeline and let a loser re-claim (cross-DB duplicate under the split).
+    TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS: z.coerce.number().int().positive().default(5),
     TRIGGER_MOLLIFIER_CLAIM_WAIT_MS: z.coerce.number().int().positive().default(5_000),
     TRIGGER_MOLLIFIER_CLAIM_POLL_MS: z.coerce.number().int().positive().default(25),
 

apps/webapp/app/models/runtimeEnvironment.server.ts

@@ -274,19 +274,17 @@ export async function findEnvironmentFromRun(
 ): Promise<EnvironmentFromRun | null> {
   // Run-ops scalars (runTags/batchId/runtimeEnvironmentId) from the run store; the env half is
   // resolved via the control-plane resolver so the run-ops DB can split without a cross-DB join.
-  const taskRun = await runStore.findRun(
-    {
-      id: runId,
-    },
-    {
-      select: {
-        runTags: true,
-        batchId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    tx ?? $replica
-  );
+  const select = {
+    runTags: true,
+    batchId: true,
+    runtimeEnvironmentId: true,
+  } as const;
+  let taskRun = await runStore.findRun({ id: runId }, { select }, tx ?? $replica);
+  if (!taskRun) {
+    // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary before
+    // treating it as absent, so runMetadataUpdated doesn't drop a live run's final metadata + publish.
+    taskRun = await runStore.findRun({ id: runId }, { select }, prisma);
+  }
   if (!taskRun) {
     return null;
   }

apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts

@@ -42,29 +42,36 @@ export class ApiWaitpointPresenter extends BasePresenter {
     return this.trace("call", async (span) => {
       // The store routes by the waitpointId's residency (id shape) and reads the owning
       // store's replica. waitpointId is pre-decoded from the friendlyId via WaitpointId.toId.
-      const waitpoint = await this.runStore.findWaitpoint({
-        where: {
-          id: waitpointId,
-          environmentId: environment.id,
-        },
-        select: {
-          id: true,
-          friendlyId: true,
-          type: true,
-          status: true,
-          idempotencyKey: true,
-          userProvidedIdempotencyKey: true,
-          idempotencyKeyExpiresAt: true,
-          inactiveIdempotencyKey: true,
-          output: true,
-          outputType: true,
-          outputIsError: true,
-          completedAfter: true,
-          completedAt: true,
-          createdAt: true,
-          tags: true,
-        },
-      });
+      const where = {
+        id: waitpointId,
+        environmentId: environment.id,
+      };
+      const select = {
+        id: true,
+        friendlyId: true,
+        type: true,
+        status: true,
+        idempotencyKey: true,
+        userProvidedIdempotencyKey: true,
+        idempotencyKeyExpiresAt: true,
+        inactiveIdempotencyKey: true,
+        output: true,
+        outputType: true,
+        outputIsError: true,
+        completedAfter: true,
+        completedAt: true,
+        createdAt: true,
+        tags: true,
+      } as const;
+
+      let waitpoint = await this.runStore.findWaitpoint({ where, select });
+
+      // Read-your-writes on a public GET: a just-minted token may not be on the owning store's
+      // replica yet, so a replica miss would 404 a live token. Re-read the owning primary before
+      // concluding it doesn't exist (mirrors the metadata GET loader + the complete/callback paths).
+      if (!waitpoint) {
+        waitpoint = await this.runStore.findWaitpointOnPrimary({ where, select });
+      }
 
       if (!waitpoint) {
         logger.error(`WaitpointPresenter: Waitpoint not found`, {

apps/webapp/app/presenters/v3/BatchPresenter.server.ts

@@ -47,13 +47,25 @@ export class BatchPresenter extends BasePresenter {
     // The BatchTaskRun (run-ops) is read through the run store, which routes by residency. The
     // runtimeEnvironment (control-plane) is resolved separately because the cross-seam FK is
     // dropped, so the batch row cannot single-SQL join to control-plane RuntimeEnvironment.
-    const batch = await this.runStore.findBatchTaskRunByFriendlyId(
+    let batch = await this.runStore.findBatchTaskRunByFriendlyId(
       batchId,
       environmentId,
       { include: BATCH_INCLUDE },
       this._replica
     );
 
+    // Read-your-writes: findBatchTaskRunByFriendlyId defaults to (and here reads) the replica, so a
+    // batch created within the replica's apply window returns null under lag. Re-read from the owning
+    // primary on a miss so a live batch's detail page never spuriously 404s ("Batch not found").
+    if (!batch) {
+      batch = await this.runStore.findBatchTaskRunByFriendlyId(
+        batchId,
+        environmentId,
+        { include: BATCH_INCLUDE },
+        this._prisma
+      );
+    }
+
     if (!batch) {
       throw new Error("Batch not found");
     }

apps/webapp/app/routes/api.v1.batches.$batchId.ts

@@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
+    // A just-created batch may not yet have replicated to the read replica this client-less
+    // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through
+    // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes,
+    // e.g. api.v3.runs.$runId).
+    shouldRetryNotFound: true,
     findResource: (params, auth) => {
       return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, {
         include: { errors: true },

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts

@@ -33,21 +33,23 @@ const { action, loader } = createActionApiRoute(
   },
   async ({ authentication, body, params }) => {
     try {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runFriendlyId,
-          runtimeEnvironmentId: authentication.environment.id,
+      const where = {
+        friendlyId: params.runFriendlyId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          realtimeStreamsVersion: true,
+          streamBasinName: true,
         },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            realtimeStreamsVersion: true,
-            streamBasinName: true,
-          },
-        },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run
+      // that exists. Re-read the owning primary on a replica miss.
+      const run =
+        (await runStore.findRun(where, args, $replica)) ??
+        (await runStore.findRunOnPrimary(where, args));
 
       if (!run) {
         return json({ error: "Run not found" }, { status: 404 });

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -39,20 +39,22 @@ const { action, loader } = createActionApiRoute(
   },
   async ({ authentication, body, params }) => {
     try {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runFriendlyId,
-          runtimeEnvironmentId: authentication.environment.id,
+      const where = {
+        friendlyId: params.runFriendlyId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          realtimeStreamsVersion: true,
         },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            realtimeStreamsVersion: true,
-          },
-        },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run
+      // that exists. Re-read the owning primary on a replica miss.
+      const run =
+        (await runStore.findRun(where, args, $replica)) ??
+        (await runStore.findRunOnPrimary(where, args));
 
       if (!run) {
         return json({ error: "Run not found" }, { status: 404 });

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -64,6 +64,19 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     );
   }
 
+  // Read-your-writes: a run drained from the buffer to the primary but not yet replicated misses
+  // both the replica read and the buffer. Re-read the owning primary before 404ing.
+  const primaryRun = await runStore.findRunOnPrimary(
+    { friendlyId: parsed.data.runId, runtimeEnvironmentId: env.id },
+    { select: { metadata: true, metadataType: true } }
+  );
+  if (primaryRun) {
+    return json(
+      { metadata: primaryRun.metadata, metadataType: primaryRun.metadataType },
+      { status: 200 }
+    );
+  }
+
   return json({ error: "Run not found" }, { status: 404 });
 }
 

apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts

@@ -75,7 +75,7 @@ const { action, loader } = createActionApiRoute(
     // SDK exposes via `ctx.run.id`). Internally `Session.currentRunId`
     // stores the TaskRun.id cuid, so resolve before handing to the
     // optimistic-claim service.
-    const callingRun = await runStore.findRun(
+    let callingRun = await runStore.findRun(
       {
         friendlyId: body.callingRunId,
         runtimeEnvironmentId: authentication.environment.id,
@@ -83,6 +83,19 @@ const { action, loader } = createActionApiRoute(
       { select: { id: true } },
       $replica
     );
+    if (!callingRun) {
+      // Replica lag: `callingRunId` is the agent's own live run (it is executing this request), so it
+      // exists on the owning primary even when the read replica has not caught up. Re-read the primary
+      // before 404ing — otherwise a lagging replica turns a legitimate handoff into a spurious
+      // "callingRunId not found in this environment".
+      callingRun = await runStore.findRunOnPrimary(
+        {
+          friendlyId: body.callingRunId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+        { select: { id: true } }
+      );
+    }
     if (!callingRun) {
       return json({ error: "callingRunId not found in this environment" }, { status: 404 });
     }

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts

@@ -36,13 +36,22 @@ export async function action({ request, params }: ActionFunctionArgs) {
     // Resolve wherever the waitpoint resides. The store routes by the waitpoint id's residency
     // (id-shape) and probes both run-ops DBs, so a token on either store resolves; the env is
     // resolved below from the row via the control-plane resolver.
-    const waitpoint = await runStore.findWaitpoint({
+    let waitpoint = await runStore.findWaitpoint({
       where: {
         id: waitpointId,
       },
       select: { id: true, status: true, environmentId: true },
     });
 
+    if (!waitpoint) {
+      // Read-your-writes: a token whose callback fires right after mint may not have replicated
+      // yet. Re-read the owning primary before 404ing (mirrors complete.ts's primary fallback).
+      waitpoint = await runStore.findWaitpointOnPrimary({
+        where: { id: waitpointId },
+        select: { id: true, status: true, environmentId: true },
+      });
+    }
+
     if (!waitpoint) {
       return json({ error: "Waitpoint not found" }, { status: 404 });
     }

apps/webapp/app/routes/api.v2.batches.$batchId.ts

@@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
+    // A just-created batch may not yet have replicated to the read replica this client-less
+    // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through
+    // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes,
+    // e.g. api.v3.runs.$runId).
+    shouldRetryNotFound: true,
     findResource: (params, auth) => {
       return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, {
         include: { errors: true },

apps/webapp/app/routes/realtime.v1.batches.$batchId.ts

@@ -2,7 +2,7 @@ import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { resolveRealtimeStreamClient } from "~/services/realtime/resolveRealtimeStreamClient.server";
 import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { runStore } from "~/v3/runStore.server";
+import { resolveBatchTaskRunForRealtime } from "~/v3/realtime/resolveBatchForRealtime.server";
 
 const ParamsSchema = z.object({
   batchId: z.string(),
@@ -13,9 +13,13 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
-    findResource: (params, auth) => {
-      return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id);
-    },
+    // A just-created batch may not yet have replicated to the read replica the client-less lookup uses.
+    // shouldRetryNotFound stamps a retryable 404 for the zodfetch GET; the realtime resolver ALSO
+    // re-reads the owning primary on a replica miss, so the Electric ShapeStream consumer (which ignores
+    // x-should-retry) doesn't strand a live batch on a permanent 404. Mirrors the run-get routes.
+    shouldRetryNotFound: true,
+    findResource: (params, auth) =>
+      resolveBatchTaskRunForRealtime(params.batchId, auth.environment.id),
     authorization: {
       action: "read",
       // See sibling note in api.v1.batches.$batchId.ts — `{type: "runs"}`

apps/webapp/app/routes/realtime.v1.runs.$runId.ts

@@ -15,22 +15,24 @@ export const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, authentication) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: authentication.environment.id,
-        },
-        {
-          include: {
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a run that already exists on the owning primary. A spurious 404
+      // here permanently fails the client's realtime subscription (the SSE client treats 404 as
+      // "stream gone" — nonRetryableStatuses). Re-read the primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -16,29 +16,29 @@ export const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, auth) => {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: auth.environment.id,
-        },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            taskIdentifier: true,
-            runTags: true,
-            realtimeStreamsVersion: true,
-            streamBasinName: true,
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: auth.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          taskIdentifier: true,
+          runTags: true,
+          realtimeStreamsVersion: true,
+          streamBasinName: true,
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
-      return run;
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the SSE subscription
+      // (the client treats 404 as "stream gone"). Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts

@@ -27,29 +27,31 @@ const { action } = createActionApiRoute(
     maxContentLength: MAX_APPEND_BODY_BYTES,
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
-      },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          parentTaskRun: {
-            select: {
-              friendlyId: true,
-            },
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        parentTaskRun: {
+          select: {
+            friendlyId: true,
           },
-          rootTaskRun: {
-            select: {
-              friendlyId: true,
-            },
+        },
+        rootTaskRun: {
+          select: {
+            friendlyId: true,
           },
         },
       },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the append. Re-read the
+    // owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts

@@ -19,32 +19,34 @@ const { action } = createActionApiRoute(
     params: ParamsSchema,
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
-      },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          streamBasinName: true,
-          parentTaskRun: {
-            select: {
-              friendlyId: true,
-              streamBasinName: true,
-            },
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        streamBasinName: true,
+        parentTaskRun: {
+          select: {
+            friendlyId: true,
+            streamBasinName: true,
           },
-          rootTaskRun: {
-            select: {
-              friendlyId: true,
-              streamBasinName: true,
-            },
+        },
+        rootTaskRun: {
+          select: {
+            friendlyId: true,
+            streamBasinName: true,
           },
         },
       },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the ingest client.
+    // Re-read the owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return new Response("Run not found", { status: 404 });
@@ -154,32 +156,33 @@ const loader = createLoaderApiRoute(
     allowJWT: false,
     corsStrategy: "none",
     findResource: async (params, authentication) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: authentication.environment.id,
-        },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            streamBasinName: true,
-            parentTaskRun: {
-              select: {
-                friendlyId: true,
-                streamBasinName: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          streamBasinName: true,
+          parentTaskRun: {
+            select: {
+              friendlyId: true,
+              streamBasinName: true,
             },
-            rootTaskRun: {
-              select: {
-                friendlyId: true,
-                streamBasinName: true,
-              },
+          },
+          rootTaskRun: {
+            select: {
+              friendlyId: true,
+              streamBasinName: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the HEAD probe.
+      // Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
   },
   async ({ request, params, resource: run, authentication }) => {

apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts

@@ -39,22 +39,24 @@ const { action } = createActionApiRoute(
     },
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        completedAt: true,
+        realtimeStreamsVersion: true,
+        streamBasinName: true,
       },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          completedAt: true,
-          realtimeStreamsVersion: true,
-          streamBasinName: true,
-        },
-      },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the input send. Re-read
+    // the owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return json({ ok: false, error: "Run not found" }, { status: 404 });
@@ -133,22 +135,23 @@ const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, auth) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: auth.environment.id,
-        },
-        {
-          include: {
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: auth.environment.id,
+      };
+      const args = {
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the SSE tail (the
+      // client treats 404 as "stream gone"). Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx

@@ -12,20 +12,19 @@ export const action: ActionFunction = async ({ request, params }) => {
   const { projectParam, organizationSlug, envParam, runParam } = v3RunParamsSchema.parse(params);
 
   try {
-    const taskRun = await runStore.findRun(
-      {
-        friendlyId: runParam,
-      },
-      {
-        select: {
-          id: true,
-          idempotencyKey: true,
-          taskIdentifier: true,
-          projectId: true,
-          runtimeEnvironmentId: true,
-        },
-      }
-    );
+    const resetSelect = {
+      id: true,
+      idempotencyKey: true,
+      taskIdentifier: true,
+      projectId: true,
+      runtimeEnvironmentId: true,
+    };
+    let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: resetSelect });
+    if (!taskRun) {
+      // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+      // before 404ing — this null gates the reset mutation below (mirrors cancel/replay).
+      taskRun = await runStore.findRunOnPrimary({ friendlyId: runParam }, { select: resetSelect });
+    }
 
     if (!taskRun) {
       return jsonWithErrorMessage({}, request, "Run not found");

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts

@@ -1,6 +1,6 @@
 import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import { runStore } from "~/v3/runStore.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
@@ -8,7 +8,7 @@ import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
 import {
   canonicalSessionAddressingKey,
-  resolveSessionByIdOrExternalId,
+  resolveSessionWithWriterFallback,
 } from "~/services/realtime/sessions.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
@@ -51,22 +51,27 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
   // Verify the run lives in this environment — keeps callers from
   // subscribing to arbitrary sessions via `/runs/$runParam/...`.
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-      runtimeEnvironmentId: environment.id,
-    },
-    {
-      select: { id: true, friendlyId: true },
-    },
-    $replica
-  );
+  const runWhere = {
+    friendlyId: runParam,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: { id: true, friendlyId: true },
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
+  // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });
   }
 
-  const session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionId);
+  // Replica lag can null out a just-created session; a spurious 404 breaks the dashboard Agent tab
+  // subscription (useRealtimeStream surfaces the error and does not auto-retry). Resolve replica-first
+  // with a writer fallback — the same helper the sibling `.in/append` route uses.
+  const session = await resolveSessionWithWriterFallback(environment.id, sessionId);
 
   if (!session) {
     return new Response("Session not found", { status: 404 });
@@ -76,10 +81,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // this environment is enough to subscribe to any session in the same
   // environment — defeats the point of scoping subscriptions through the
   // run route. SessionRun.runId is indexed (@unique), so this is cheap.
-  const linkedSessionRun = await $replica.sessionRun.findFirst({
-    where: { runId: run.id, sessionId: session.id },
-    select: { id: true },
-  });
+  // Replica lag can null out the just-created run↔session linkage row; a spurious 404 breaks the
+  // dashboard Agent tab subscription (client does not auto-retry). Re-read the primary on a miss.
+  const linkWhere = { runId: run.id, sessionId: session.id };
+  const linkedSessionRun =
+    (await $replica.sessionRun.findFirst({ where: linkWhere, select: { id: true } })) ??
+    (await prisma.sessionRun.findFirst({ where: linkWhere, select: { id: true } }));
 
   if (!linkedSessionRun) {
     return new Response("Session not found for run", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts

@@ -45,21 +45,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Environment not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runId,
-      runtimeEnvironmentId: environment.id,
+  const runWhere = {
+    friendlyId: runId,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
     },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-      },
-    },
-    $replica
-  );
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
+  // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts

@@ -47,21 +47,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Environment not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runId,
-      runtimeEnvironmentId: environment.id,
+  const runWhere = {
+    friendlyId: runId,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
     },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-      },
-    },
-    $replica
-  );
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream
+  // subscription (useRealtimeStream surfaces the error, no auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx

@@ -60,18 +60,22 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     throw new Response("Not Found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    { friendlyId: runParam, projectId: project.id },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-        runtimeEnvironmentId: true,
-      },
-    }
-  );
+  const runWhere = { friendlyId: runParam, projectId: project.id };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
+      runtimeEnvironmentId: true,
+    },
+  };
+  // Client-less findRun defaults to the read replica; replica lag can null out a live run and 404 a
+  // valid stream-viewer request (useRealtimeStream surfaces the error, no auto-retry). Re-read the
+  // owning primary on a replica miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     throw new Response("Not Found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx

@@ -81,7 +81,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
 
     const waitpointId = WaitpointId.toId(waitpointFriendlyId);
 
-    const waitpoint = await runStore.findWaitpoint({
+    let waitpoint = await runStore.findWaitpoint({
       select: {
         projectId: true,
         environmentId: true,
@@ -90,6 +90,14 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
         id: waitpointId,
       },
     });
+    if (!waitpoint) {
+      // Read-your-writes: a just-minted token may not have replicated. Re-read the owning primary
+      // before the auth guard / "No waitpoint found" (mirrors the token complete/callback routes).
+      waitpoint = await runStore.findWaitpointOnPrimary({
+        select: { projectId: true, environmentId: true },
+        where: { id: waitpointId },
+      });
+    }
 
     if (waitpoint?.projectId !== project.id) {
       return redirectWithErrorMessage(

apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts

@@ -80,21 +80,26 @@ export const action = dashboardAction(
       // The project-scope + membership auth is a control-plane concern resolved
       // separately below; joining project/organization here is a cross-DB join
       // that returns nothing once the run lives in the run-ops DB.
-      const taskRun = await runStore.findRun(
-        { friendlyId: runParam },
-        {
-          select: {
-            id: true,
-            engine: true,
-            status: true,
-            friendlyId: true,
-            taskEventStore: true,
-            createdAt: true,
-            completedAt: true,
-            projectId: true,
-          },
-        }
-      );
+      const cancelRunSelect = {
+        id: true,
+        engine: true,
+        status: true,
+        friendlyId: true,
+        taskEventStore: true,
+        createdAt: true,
+        completedAt: true,
+        projectId: true,
+      };
+      let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: cancelRunSelect });
+      if (!taskRun) {
+        // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+        // before treating it as absent (mirrors resolveRunOrganizationId's primary fallback above).
+        taskRun = await runStore.findRun(
+          { friendlyId: runParam },
+          { select: cancelRunSelect },
+          prisma
+        );
+      }
 
       // Project-scope + membership auth is control-plane only — keyed by the
       // run's projectId. A miss is treated as not-found (mirrors the old where).

apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts

@@ -323,7 +323,12 @@ export const action = dashboardAction(
     try {
       // Run-ops read keyed by friendlyId only; membership auth is re-checked on the
       // control plane below, keyed off the resolved run's projectId.
-      const pgRun = await runStore.findRun({ friendlyId: runParam });
+      let pgRun = await runStore.findRun({ friendlyId: runParam });
+      if (!pgRun) {
+        // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+        // before falling back to the mollifier buffer (mirrors resolveRunOrganizationId above).
+        pgRun = await runStore.findRun({ friendlyId: runParam }, prisma);
+      }
 
       // Mollifier read-fallback: if the original isn't in PG yet, synthesise a
       // TaskRun from the buffered snapshot. Needs project/org/env slugs for the

apps/webapp/app/routes/sync.traces.runs.$traceId.ts

@@ -23,7 +23,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       return new Response("No user found in cookie", { status: 401 });
     }
 
-    const run = await runStore.findRun(
+    let run = await runStore.findRun(
       {
         traceId,
       },
@@ -35,6 +35,15 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       $replica
     );
 
+    if (!run) {
+      // Read-your-writes: a just-created run may not have replicated yet. Re-read the owning
+      // primary before 404ing so a live run's realtime trace feed isn't spuriously not-found.
+      run = await runStore.findRunOnPrimary(
+        { traceId },
+        { select: { runtimeEnvironmentId: true } }
+      );
+    }
+
     if (!run) {
       return new Response("No run found", { status: 404 });
     }

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

@@ -1,5 +1,5 @@
-import { RunId } from "@trigger.dev/core/v3/isomorphic";
-import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database";
+import { ownerEngine, RunId } from "@trigger.dev/core/v3/isomorphic";
+import type { PrismaClientOrTransaction, TaskRun, Waitpoint } from "@trigger.dev/database";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
@@ -8,7 +8,8 @@ import type { RunEngine } from "~/v3/runEngine.server";
 import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus";
 import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
-import { claimOrAwait } from "~/v3/mollifier/idempotencyClaim.server";
+import { claimOrAwait, resetResolvedClaim } from "~/v3/mollifier/idempotencyClaim.server";
+import { computeClaimTtlSeconds } from "~/v3/mollifier/claimTtl";
 import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server";
 import { runStore } from "~/v3/runStore.server";
 import { runOpsLegacyPrisma, runOpsNewPrisma } from "~/db.server";
@@ -24,6 +25,13 @@ import type { TraceEventConcern, TriggerTaskRequest } from "../types";
 // handleTriggerRequest.
 const resolveOrgMollifierFlag = makeResolveMollifierFlag();
 
+// Cap on the claim-loser recreate re-acquisition loop (see
+// reacquireClearedGlobalWinner). Each pass reopens a stale resolved slot and
+// re-enters the claim; bounded so a pathological stream of expired/failed
+// winners can't spin forever. On exhaustion we fall open to the create with
+// PG's unique index as the backstop.
+const MAX_CLEARED_WINNER_REACQUIRES = 5;
+
 // Claim ownership context returned to the caller when the
 // IdempotencyKeyConcern won a pre-gate claim. Caller MUST publish the
 // winning runId on pipeline success (`publishClaim`) or release the
@@ -173,6 +181,14 @@ export class IdempotencyKeyConcern {
       }
     );
 
+    // `global`-scope (or scope-absent) keys under the split have no per-run salt, so the Redis claim is
+    // their only cross-DB dedup mutex. Computed here (not just in the claim block below) because the
+    // expired/failed clear-and-recreate path must serialise through it too.
+    const idempotencyKeyScope = request.body.options?.idempotencyKeyOptions?.scope;
+    const globalUnderSplit =
+      (idempotencyKeyScope === "global" || idempotencyKeyScope === undefined) &&
+      (await isSplitEnabled());
+
     const existingRun = idempotencyKey
       ? await runStore.findRun(
           {
@@ -209,107 +225,37 @@ export class IdempotencyKeyConcern {
     }
 
     if (existingRun) {
-      // The idempotency key has expired
-      if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
-        logger.debug("[TriggerTaskService][call] Idempotency key has expired", {
-          idempotencyKey: request.options?.idempotencyKey,
-          run: existingRun,
+      const handled = await this.handleExistingRun(request, parentStore, existingRun, {
+        idempotencyKey,
+        idempotencyKeyExpiresAt,
+        dedupClient,
+      });
+      // A LIVE cached hit (or andWait waitpoint wiring) is terminal.
+      if (handled.isCached) {
+        return handled;
+      }
+      // isCached === false → the existing run was EXPIRED/FAILED, so handleExistingRun cleared its key
+      // and we must recreate. For a global-scope key under split that recreate has to be claim-
+      // serialised too — otherwise two concurrent cross-residency recreates each create a run the
+      // per-DB unique index can't dedup (the same hole reacquireClearedGlobalWinner closes on the
+      // claim-loser path). Non-split / non-global: the plain unserialised recreate is safe.
+      if (globalUnderSplit) {
+        return await this.reacquireClearedGlobalWinner(request, parentStore, {
+          idempotencyKey,
+          idempotencyKeyExpiresAt,
+          dedupClient,
+          ttlSeconds: computeClaimTtlSeconds({
+            keyExpiresAt: idempotencyKeyExpiresAt,
+            now: Date.now(),
+            minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS,
+            maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
+          }),
+          clearedRunId: existingRun.friendlyId,
+          safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS,
+          pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
         });
-
-        // Update the existing run to remove the idempotency key
-        await runStore.clearIdempotencyKey(
-          { byId: { runId: existingRun.id, idempotencyKey } },
-          dedupClient
-        );
-
-        return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
       }
-
-      // If the existing run failed or was expired, we clear the key and do a new run
-      if (shouldIdempotencyKeyBeCleared(existingRun.status)) {
-        logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", {
-          idempotencyKey: request.options?.idempotencyKey,
-          runStatus: existingRun.status,
-          runId: existingRun.id,
-        });
-
-        // Update the existing run to remove the idempotency key
-        await runStore.clearIdempotencyKey(
-          { byId: { runId: existingRun.id, idempotencyKey } },
-          dedupClient
-        );
-
-        return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
-      }
-
-      // We have an idempotent run, so we return it
-      const parentRunId = request.body.options?.parentRunId;
-      const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
-
-      //We're using `andWait` so we need to block the parent run with a waitpoint
-      if (resumeParentOnCompletion && parentRunId) {
-        // `parentRunId` comes from the request body and isn't re-validated
-        // here, so confirm the parent run is in the caller's environment
-        // before wiring a waitpoint against it.
-        const parentRunInternalId = RunId.fromFriendlyId(parentRunId);
-        const parentRunInCallerEnv = await runStore.findRun(
-          {
-            id: parentRunInternalId,
-            runtimeEnvironmentId: request.environment.id,
-          },
-          { select: { id: true } },
-          this.prisma
-        );
-        if (!parentRunInCallerEnv) {
-          throw new ServiceValidationError("Parent run not found in the calling environment", 404);
-        }
-
-        // Get or create waitpoint lazily (existing run may not have one if it was standalone)
-        let associatedWaitpoint = existingRun.associatedWaitpoint;
-        if (!associatedWaitpoint) {
-          associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({
-            runId: existingRun.id,
-            projectId: request.environment.projectId,
-            environmentId: request.environment.id,
-          });
-        }
-
-        await this.traceEventConcern.traceIdempotentRun(
-          request,
-          parentStore,
-          {
-            existingRun,
-            idempotencyKey,
-            incomplete: associatedWaitpoint.status === "PENDING",
-            isError: associatedWaitpoint.outputIsError,
-          },
-          async (event) => {
-            const spanId =
-              request.options?.parentAsLinkType === "replay"
-                ? event.spanId
-                : event.traceparent?.spanId
-                  ? `${event.traceparent.spanId}:${event.spanId}`
-                  : event.spanId;
-
-            await this.engine.blockRunWithWaitpoint({
-              runId: parentRunInternalId,
-              waitpoints: associatedWaitpoint!.id,
-              spanIdToComplete: spanId,
-              batch: request.options?.batchId
-                ? {
-                    id: request.options.batchId,
-                    index: request.options.batchIndex ?? 0,
-                  }
-                : undefined,
-              projectId: request.environment.projectId,
-              organizationId: request.environment.organizationId,
-              tx: dedupClient,
-            });
-          }
-        );
-      }
-
-      return { isCached: true, run: existingRun };
+      return handled;
     }
 
     // Pre-gate claim — closes the PG+buffer race during gate transition.
@@ -325,28 +271,41 @@ export class IdempotencyKeyConcern {
     // claim's Redis SETNX keeps its RTT off the hot path for those requests
     // during staged rollout. The org-flag check is a pure in-memory read of
     // `Organization.featureFlags`, no DB query.
+    //
+    // Under the run-ops split the claim ALSO acts as the only cross-DB mutex a
+    // `global`-scope key has: that key is per (environment, task), so it can be
+    // triggered concurrently from two parents on DIFFERENT physical DBs where
+    // each probe misses and the per-DB unique index can't enforce uniqueness.
+    // For that case the claim is eligible regardless of the per-org flag AND of
+    // resumeParentOnCompletion (the loser wires its parent waitpoint against the
+    // winner in the resolved branch below). An absent scope (pre-hashed key /
+    // older SDK) is treated conservatively as possibly-global — harmless for a
+    // real run/attempt key, whose hash already embeds the parent id so two
+    // parents mint DISTINCT keys that never share a claim slot.
+    // (idempotencyKeyScope / globalUnderSplit are computed above — they also gate the expired/failed
+    // recreate serialisation.)
     const claimEligible =
-      !request.body.options?.resumeParentOnCompletion &&
       !request.body.options?.debounce &&
       !request.options?.oneTimeUseToken &&
-      (await resolveOrgMollifierFlag({
-        envId: request.environment.id,
-        orgId: request.environment.organizationId,
-        taskId: request.taskId,
-        orgFeatureFlags:
-          (request.environment.organization?.featureFlags as
-            | Record<string, unknown>
-            | null
-            | undefined) ?? null,
-      }));
+      (globalUnderSplit ||
+        (!request.body.options?.resumeParentOnCompletion &&
+          (await resolveOrgMollifierFlag({
+            envId: request.environment.id,
+            orgId: request.environment.organizationId,
+            taskId: request.taskId,
+            orgFeatureFlags:
+              (request.environment.organization?.featureFlags as
+                | Record<string, unknown>
+                | null
+                | undefined) ?? null,
+          }))));
     if (claimEligible) {
-      const ttlSeconds = Math.max(
-        1,
-        Math.min(
-          env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
-          Math.ceil((idempotencyKeyExpiresAt.getTime() - Date.now()) / 1000)
-        )
-      );
+      const ttlSeconds = computeClaimTtlSeconds({
+        keyExpiresAt: idempotencyKeyExpiresAt,
+        now: Date.now(),
+        minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS,
+        maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
+      });
       const outcome = await claimOrAwait({
         envId: request.environment.id,
         taskIdentifier: request.taskId,
@@ -356,6 +315,44 @@ export class IdempotencyKeyConcern {
         pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
       });
       if (outcome.kind === "resolved") {
+        // Global-under-split loser: the winner lives on ITS parent's DB, which
+        // may differ from this loser's parent DB. Resolve the winner by id
+        // across both DBs (classify the winner friendlyId → NEW/LEGACY client,
+        // then let the router route+fall-back by id-shape) and feed it through
+        // the same existing-run handling the PG-hit path uses — so expiry-clear,
+        // status-clear, and the resumeParentOnCompletion waitpoint wiring all
+        // apply to the loser exactly as they would to a plain cached hit.
+        if (globalUnderSplit) {
+          const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id);
+          if (winner) {
+            const resolved = await this.handleExistingRun(request, parentStore, winner, {
+              idempotencyKey,
+              idempotencyKeyExpiresAt,
+              dedupClient,
+            });
+            // A LIVE winner (cached hit, or andWait waitpoint wired) is terminal.
+            if (resolved.isCached) {
+              return resolved;
+            }
+            // CRITICAL (CodeRabbit): the resolved winner was EXPIRED or FAILED, so
+            // handleExistingRun cleared its key and would have us CREATE a new run.
+            // The initial create is serialised by the claim, but this clear-and-
+            // recreate is not — and under the split the per-DB unique index can't
+            // dedup a cross-residency recreate, so two concurrent losers clearing
+            // the SAME winner would each create → duplicate run. Re-serialise the
+            // recreate through the claim so exactly one caller recreates (and
+            // publishes) while the rest resolve to the fresh run.
+            return await this.reacquireClearedGlobalWinner(request, parentStore, {
+              idempotencyKey,
+              idempotencyKeyExpiresAt,
+              dedupClient,
+              ttlSeconds,
+              clearedRunId: outcome.runId,
+              safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS,
+              pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
+            });
+          }
+        }
         // Another concurrent trigger committed first. Re-resolve via the
         // existing checks: writer-side PG findFirst first (defeats
         // replica lag), then buffer fallback for the buffered case.
@@ -421,4 +418,238 @@ export class IdempotencyKeyConcern {
 
     return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
   }
+
+  // Resolve an already-existing idempotent run: honour key expiry / status
+  // clearing, and for `andWait` (resumeParentOnCompletion) block the calling
+  // parent on the run's waitpoint. Extracted so both the PG-hit path and the
+  // cross-DB claim-loser path resolve an existing run identically.
+  private async handleExistingRun(
+    request: TriggerTaskRequest,
+    parentStore: string | undefined,
+    existingRun: TaskRun & { associatedWaitpoint?: Waitpoint | null },
+    ctx: {
+      idempotencyKey: string;
+      idempotencyKeyExpiresAt: Date;
+      dedupClient: PrismaClientOrTransaction;
+    }
+  ): Promise<IdempotencyKeyConcernResult> {
+    const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient } = ctx;
+
+    // The idempotency key has expired
+    if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
+      logger.debug("[TriggerTaskService][call] Idempotency key has expired", {
+        idempotencyKey: request.options?.idempotencyKey,
+        run: existingRun,
+      });
+
+      // Update the existing run to remove the idempotency key
+      await runStore.clearIdempotencyKey(
+        { byId: { runId: existingRun.id, idempotencyKey } },
+        dedupClient
+      );
+
+      return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+    }
+
+    // If the existing run failed or was expired, we clear the key and do a new run
+    if (shouldIdempotencyKeyBeCleared(existingRun.status)) {
+      logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", {
+        idempotencyKey: request.options?.idempotencyKey,
+        runStatus: existingRun.status,
+        runId: existingRun.id,
+      });
+
+      // Update the existing run to remove the idempotency key
+      await runStore.clearIdempotencyKey(
+        { byId: { runId: existingRun.id, idempotencyKey } },
+        dedupClient
+      );
+
+      return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+    }
+
+    // We have an idempotent run, so we return it
+    const parentRunId = request.body.options?.parentRunId;
+    const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
+
+    //We're using `andWait` so we need to block the parent run with a waitpoint
+    if (resumeParentOnCompletion && parentRunId) {
+      // `parentRunId` comes from the request body and isn't re-validated
+      // here, so confirm the parent run is in the caller's environment
+      // before wiring a waitpoint against it.
+      const parentRunInternalId = RunId.fromFriendlyId(parentRunId);
+      const parentRunInCallerEnv = await runStore.findRun(
+        {
+          id: parentRunInternalId,
+          runtimeEnvironmentId: request.environment.id,
+        },
+        { select: { id: true } },
+        this.prisma
+      );
+      if (!parentRunInCallerEnv) {
+        throw new ServiceValidationError("Parent run not found in the calling environment", 404);
+      }
+
+      // Get or create waitpoint lazily (existing run may not have one if it was standalone)
+      let associatedWaitpoint = existingRun.associatedWaitpoint;
+      if (!associatedWaitpoint) {
+        associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({
+          runId: existingRun.id,
+          projectId: request.environment.projectId,
+          environmentId: request.environment.id,
+        });
+      }
+
+      await this.traceEventConcern.traceIdempotentRun(
+        request,
+        parentStore,
+        {
+          existingRun,
+          idempotencyKey,
+          incomplete: associatedWaitpoint.status === "PENDING",
+          isError: associatedWaitpoint.outputIsError,
+        },
+        async (event) => {
+          const spanId =
+            request.options?.parentAsLinkType === "replay"
+              ? event.spanId
+              : event.traceparent?.spanId
+                ? `${event.traceparent.spanId}:${event.spanId}`
+                : event.spanId;
+
+          await this.engine.blockRunWithWaitpoint({
+            runId: parentRunInternalId,
+            waitpoints: associatedWaitpoint!.id,
+            spanIdToComplete: spanId,
+            batch: request.options?.batchId
+              ? {
+                  id: request.options.batchId,
+                  index: request.options.batchIndex ?? 0,
+                }
+              : undefined,
+            projectId: request.environment.projectId,
+            organizationId: request.environment.organizationId,
+            tx: dedupClient,
+          });
+        }
+      );
+    }
+
+    return { isCached: true, run: existingRun };
+  }
+
+  // Re-serialise a cross-DB recreate through the claim after a claim-loser's
+  // resolved winner turned out to be EXPIRED / FAILED (its key was cleared by
+  // handleExistingRun). Without this, concurrent losers clearing the same
+  // winner each create a new run on their own DB — the per-DB unique index
+  // can't dedup a cross-residency pair, so the initial-create's serialisation
+  // is lost on the recreate. Each pass: compare-and-delete the stale resolved
+  // slot (keyed on the cleared runId — never an unconditional DEL, so a
+  // reacquirer that already re-published a NEW winner is not wiped), then
+  // re-enter claimOrAwait. Exactly one caller wins the re-claim and recreates
+  // (returning its claim to publish); the rest resolve to the fresh run. If a
+  // re-claim resolves to ANOTHER cleared winner we advance and loop, bounded
+  // by MAX_CLEARED_WINNER_REACQUIRES; on exhaustion (or an unfindable
+  // resolution) we fail CLOSED with a retryable 503 so the SDK retry re-serialises,
+  // rather than fall open to an unserialised cross-DB create.
+  private async reacquireClearedGlobalWinner(
+    request: TriggerTaskRequest,
+    parentStore: string | undefined,
+    ctx: {
+      idempotencyKey: string;
+      idempotencyKeyExpiresAt: Date;
+      dedupClient: PrismaClientOrTransaction;
+      ttlSeconds: number;
+      clearedRunId: string;
+      safetyNetMs: number;
+      pollStepMs: number;
+    }
+  ): Promise<IdempotencyKeyConcernResult> {
+    const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient, ttlSeconds } = ctx;
+    const claimInput = {
+      envId: request.environment.id,
+      taskIdentifier: request.taskId,
+      idempotencyKey,
+    };
+    let staleRunId = ctx.clearedRunId;
+    for (let attempt = 0; attempt < MAX_CLEARED_WINNER_REACQUIRES; attempt++) {
+      await resetResolvedClaim({ ...claimInput, runId: staleRunId });
+
+      const outcome = await claimOrAwait({
+        ...claimInput,
+        ttlSeconds,
+        safetyNetMs: ctx.safetyNetMs,
+        pollStepMs: ctx.pollStepMs,
+      });
+
+      if (outcome.kind === "timed_out") {
+        throw new ServiceValidationError("Idempotency claim resolution timed out", 503);
+      }
+      if (outcome.kind === "claimed") {
+        // We own the recreate. Caller MUST publish the new runId / release on error.
+        return {
+          isCached: false,
+          idempotencyKey,
+          idempotencyKeyExpiresAt,
+          claim: { ...claimInput, token: outcome.token },
+        };
+      }
+      // resolved: another caller won the recreate. Honour it like any cached
+      // hit (incl. andWait wiring). If it too was cleared, advance and loop.
+      const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id);
+      if (!winner) {
+        logger.warn("idempotency reacquire resolved but runId not findable", {
+          envId: request.environment.id,
+          taskIdentifier: request.taskId,
+          claimedRunId: outcome.runId,
+        });
+        break;
+      }
+      const resolved = await this.handleExistingRun(request, parentStore, winner, {
+        idempotencyKey,
+        idempotencyKeyExpiresAt,
+        dedupClient,
+      });
+      if (resolved.isCached) {
+        return resolved;
+      }
+      staleRunId = outcome.runId;
+    }
+    // Exhausted the bounded reacquires (or the winner was unfindable). Rather than fall through to an
+    // UNSERIALISED create — which under global-scope-split can dual-create across DBs (the per-DB unique
+    // index can't dedup cross-residency) — fail closed with a retryable 503 so the SDK retry re-serialises
+    // through a fresh claim (mirrors the timed_out branch above).
+    throw new ServiceValidationError(
+      "Idempotency claim could not be re-serialised after repeated cleared winners",
+      503
+    );
+  }
+
+  // Resolve a claim winner (a run friendlyId) across both split DBs. Classify
+  // the id-shape to pick the writer client for read-your-writes, then read by
+  // id — the routing store routes to the owning store and falls back to the
+  // other, so a winner on either DB is found. Returns null when the id can't be
+  // classified or the row genuinely isn't there (caller falls through).
+  private async resolveWinnerAcrossDbs(
+    winnerFriendlyId: string,
+    environmentId: string
+  ): Promise<(TaskRun & { associatedWaitpoint?: Waitpoint | null }) | null> {
+    let internalId: string;
+    try {
+      internalId = RunId.fromFriendlyId(winnerFriendlyId);
+    } catch {
+      return null;
+    }
+    let client: PrismaClientOrTransaction;
+    try {
+      client = ownerEngine(internalId) === "NEW" ? runOpsNewPrisma : runOpsLegacyPrisma;
+    } catch {
+      client = this.prisma;
+    }
+    return runStore.findRun(
+      { id: internalId, runtimeEnvironmentId: environmentId },
+      { include: { associatedWaitpoint: true } },
+      client
+    );
+  }
 }

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

@@ -752,7 +752,7 @@ export class RunEngineTriggerTaskService {
       // Pipeline returned successfully — publish the claim if we held
       // one. Waiters polling for our key resolve to this runId.
       if (idempotencyClaim && result?.run?.friendlyId) {
-        await publishMollifierClaim({
+        const published = await publishMollifierClaim({
           envId: idempotencyClaim.envId,
           taskIdentifier: idempotencyClaim.taskIdentifier,
           idempotencyKey: idempotencyClaim.idempotencyKey,
@@ -760,6 +760,17 @@ export class RunEngineTriggerTaskService {
           runId: result.run.friendlyId,
           ttlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
         });
+        if (!published) {
+          // Our claim expired mid-pipeline and another claimant took it, so this publish no-op'd: a
+          // different run is now canonical for this key while we return ours (a cross-DB dup under the
+          // split). Rare now the claim TTL is floored (C1); surfaced for monitoring pending auto-
+          // convergence (re-resolve the current winner + cancel this orphan).
+          logger.warn("mollifier claim publish no-op'd; winner lost the claim mid-pipeline", {
+            envId: idempotencyClaim.envId,
+            taskIdentifier: idempotencyClaim.taskIdentifier,
+            runId: result.run.friendlyId,
+          });
+        }
       }
       return result;
     } catch (err) {

apps/webapp/app/services/dashboardAgent.server.ts

@@ -212,7 +212,9 @@ export async function resolveRunCommit(
   environmentId: string,
   runFriendlyId: string
 ): Promise<{ sha: string; version: string; dirty: boolean } | null> {
-  const run = await runStore.findRun(
+  // Read-your-writes: a just-locked run's lockedToVersionId may not have replicated. Read the owning
+  // primary so a live, pinned run resolves its commit instead of silently falling back to branch head.
+  const run = await runStore.findRunOnPrimary(
     { friendlyId: runFriendlyId, runtimeEnvironmentId: environmentId },
     { select: { lockedToVersionId: true } }
   );

apps/webapp/app/services/realtime/sessionRunManager.server.ts

@@ -478,8 +478,10 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
 async function getRunStatusAndFriendlyId(
   runId: string
 ): Promise<{ status: TaskRunStatus; friendlyId: string } | null> {
-  // Use the read replica — this is a hot-path probe and stale-by-ms is
-  // fine. The append handler re-checks if it ends up reusing the runId.
+  // Use the read replica — hot-path probe, stale-by-ms is fine. The dangerous
+  // stale shape (looks-vanished → double-trigger) is closed by the writer re-probe
+  // in ensureRunForSession; a stale-non-final reuse self-heals via the durable S2
+  // stream + next-append re-probe (there is no append-handler re-check).
   // `friendlyId` is fetched alongside `status` so the dead-run-detection
   // branch in `ensureRunForSession` can forward the public-form id as
   // `payload.previousRunId` without a second read. `Session.currentRunId`

apps/webapp/app/v3/mollifier/claimTtl.ts

@@ -0,0 +1,12 @@
+// The claim is a serialization lock that must outlive the winner's create-and-publish pipeline. A short
+// customer key TTL must NOT shrink it below a pipeline floor (else the claim expires mid-pipeline and a
+// polling loser re-claims → cross-DB duplicate). Floor at `minTtlSeconds` independent of key TTL; cap at max.
+export function computeClaimTtlSeconds(input: {
+  keyExpiresAt: Date;
+  now: number;
+  minTtlSeconds: number;
+  maxTtlSeconds: number;
+}): number {
+  const keyTtlSeconds = Math.ceil((input.keyExpiresAt.getTime() - input.now) / 1000);
+  return Math.min(input.maxTtlSeconds, Math.max(input.minTtlSeconds, keyTtlSeconds));
+}

apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts

@@ -163,12 +163,14 @@ export async function publishClaim(input: {
   runId: string;
   ttlSeconds?: number;
   buffer?: MollifierBuffer | null;
-}): Promise<void> {
+}): Promise<boolean> {
   const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer;
-  if (!buffer) return;
+  if (!buffer) return true;
   const ttlSeconds = input.ttlSeconds ?? DEFAULT_CLAIM_TTL_SECONDS;
   try {
-    await buffer.publishClaim({
+    // false = compare-and-set no-op: a stale claimant (our TTL expired) already moved in, so our
+    // publish did NOT set the winner. The caller decides how to converge.
+    return await buffer.publishClaim({
       envId: input.envId,
       taskIdentifier: input.taskIdentifier,
       idempotencyKey: input.idempotencyKey,
@@ -182,6 +184,8 @@ export async function publishClaim(input: {
       taskIdentifier: input.taskIdentifier,
       err: err instanceof Error ? err.message : String(err),
     });
+    // Unknown publish state (transient error) — the claim TTL is the safety net; don't signal a no-op.
+    return true;
   }
 }
 
@@ -213,6 +217,39 @@ export async function releaseClaim(input: {
   }
 }
 
+// Reopen a stale RESOLVED claim slot whose winner was cleared (expired /
+// failed), so the claim-loser reacquire path can re-serialise a cross-DB
+// recreate through a fresh claim. Compare-and-delete on the observed winner
+// runId (buffer-side) so a concurrent reacquirer's freshly published winner
+// is never wiped. Best-effort: on buffer-null / error we no-op and let the
+// reacquire's claimOrAwait proceed (fail-open — the caller caps attempts and
+// ultimately falls through to the create, PG unique index as backstop).
+export async function resetResolvedClaim(input: {
+  envId: string;
+  taskIdentifier: string;
+  idempotencyKey: string;
+  runId: string;
+  buffer?: MollifierBuffer | null;
+}): Promise<boolean> {
+  const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer;
+  if (!buffer) return false;
+  try {
+    return await buffer.resetResolvedClaim({
+      envId: input.envId,
+      taskIdentifier: input.taskIdentifier,
+      idempotencyKey: input.idempotencyKey,
+      expectedRunId: input.runId,
+    });
+  } catch (err) {
+    logger.warn("idempotency resolved-claim reset failed", {
+      envId: input.envId,
+      taskIdentifier: input.taskIdentifier,
+      err: err instanceof Error ? err.message : String(err),
+    });
+    return false;
+  }
+}
+
 function defaultSleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }

apps/webapp/app/v3/realtime/resolveBatchForRealtime.server.ts

@@ -0,0 +1,24 @@
+import { prisma } from "~/db.server";
+import { runStore } from "~/v3/runStore.server";
+
+type BatchStore = Pick<typeof runStore, "findBatchTaskRunByFriendlyId">;
+
+// The realtime batch route reads the batch client-less (replica), which can miss a just-created batch
+// under replica lag. `shouldRetryNotFound` covers the zodfetch GET, but the Electric ShapeStream
+// consumer (self-hosters) ignores `x-should-retry`, so re-read the owning primary on a miss — passing a
+// non-replica writer flips each store leg to its own primary — to avoid a permanent 404.
+export function resolveBatchTaskRunForRealtime(
+  friendlyId: string,
+  environmentId: string,
+  deps?: { store?: BatchStore; writer?: unknown }
+) {
+  const store = deps?.store ?? runStore;
+  const writer = deps?.writer ?? prisma;
+  return store
+    .findBatchTaskRunByFriendlyId(friendlyId, environmentId)
+    .then(
+      (onReplica) =>
+        onReplica ??
+        store.findBatchTaskRunByFriendlyId(friendlyId, environmentId, undefined, writer as never)
+    );
+}

apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -195,18 +195,23 @@ export class BatchTriggerV3Service extends BaseService {
           span.setAttribute("batchId", batchId);
 
           const dependentAttempt = body?.dependentAttempt
-            ? await this.runStore.findTaskRunAttempt({
-                // Scope to the caller's environment (see dependentAttemptWhere).
-                where: dependentAttemptWhere(body.dependentAttempt, environment.id),
-                include: {
-                  taskRun: {
-                    select: {
-                      id: true,
-                      status: true,
+            ? await this.runStore.findTaskRunAttempt(
+                {
+                  // Scope to the caller's environment (see dependentAttemptWhere).
+                  where: dependentAttemptWhere(body.dependentAttempt, environment.id),
+                  include: {
+                    taskRun: {
+                      select: {
+                        id: true,
+                        status: true,
+                      },
                     },
                   },
                 },
-              })
+                // Read-your-writes: the dependent parent attempt's terminal state must be seen even
+                // when the read replica lags, so route this read to the owning primary.
+                this._prisma
+              )
             : undefined;
 
           if (

apps/webapp/test/mollifierIdempotencyClaim.test.ts

@@ -181,13 +181,13 @@ describe("publishClaim", () => {
     expect(buffer.publishClaim).toHaveBeenCalledOnce();
   });
 
-  it("no-op when buffer is null", async () => {
+  it("no buffer → true (nothing to converge, not a no-op'd publish)", async () => {
     await expect(
       publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer: null })
-    ).resolves.toBeUndefined();
+    ).resolves.toBe(true);
   });
 
-  it("swallows errors so trigger pipeline isn't broken by Redis hiccups", async () => {
+  it("swallows errors so trigger pipeline isn't broken by Redis hiccups (returns true, unknown state)", async () => {
     const buffer = {
       publishClaim: vi.fn(async () => {
         throw new Error("ECONNREFUSED");
@@ -195,7 +195,7 @@ describe("publishClaim", () => {
     } as unknown as MollifierBuffer;
     await expect(
       publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer })
-    ).resolves.toBeUndefined();
+    ).resolves.toBe(true);
   });
 });
 

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

@@ -170,7 +170,9 @@ export class RunAttemptSystem {
   }
 
   public async resolveTaskRunContext(runId: string): Promise<TaskRunContext> {
-    const run = await this.$.runStore.findRun(
+    // read-your-writes: a just-created/locked run may not be on a replica yet; read the owning primary
+    // so a live run's execution context resolves instead of throwing a spurious 404.
+    const run = await this.$.runStore.findRunOnPrimary(
       {
         id: runId,
       },
@@ -711,8 +713,8 @@ export class RunAttemptSystem {
 
           const completedAt = new Date();
 
-          // Read current usage values to calculate new totals (safe under runLock)
-          const currentRun = await this.$.runStore.findRun(
+          // Read current usage totals on the owning primary (read-your-writes; a replica lags)
+          const currentRun = await this.$.runStore.findRunOnPrimary(
             { id: runId },
             {
               select: {
@@ -904,7 +906,8 @@ export class RunAttemptSystem {
           const failedAt = new Date();
 
           const retryResult = await retryOutcomeFromCompletion(
-            this.$.readOnlyPrisma,
+            // read-your-writes: lock-time maxAttempts/lockedRetryConfig may not be on a replica yet
+            this.$.prisma,
             this.$.runStore,
             {
               runId,
@@ -917,7 +920,9 @@ export class RunAttemptSystem {
 
           // Force requeue means it was crashed so the attempt span needs to be closed
           if (forceRequeue) {
-            const minimalRun = await this.$.runStore.findRun(
+            // read-your-writes: the run was just written in this flow; read the owning primary so the
+            // requeue event re-read cannot false-miss on a lagging replica (mirrors the :906 read).
+            const minimalRun = await this.$.runStore.findRunOnPrimary(
               {
                 id: runId,
               },
@@ -1375,7 +1380,7 @@ export class RunAttemptSystem {
         // Calculate updated usage if we have attempt duration data
         let usageUpdate: { usageDurationMs: number; costInCents: number } | undefined;
         if (attemptDurationMs !== undefined) {
-          const currentRun = await this.$.runStore.findRun(
+          const currentRun = await this.$.runStore.findRunOnPrimary(
             { id: runId },
             {
               select: {
@@ -1589,8 +1594,8 @@ export class RunAttemptSystem {
 
       const truncatedError = this.#truncateTaskRunError(error);
 
-      // Read current usage values to calculate new totals
-      const currentRun = await this.$.runStore.findRun(
+      // Read current usage totals on the owning primary (read-your-writes; a replica lags)
+      const currentRun = await this.$.runStore.findRunOnPrimary(
         { id: runId },
         {
           select: {

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

@@ -158,22 +158,26 @@ export class TtlSystem {
       const skipped: { runId: string; reason: string }[] = [];
 
       // Fetch all runs in a single query (no snapshot data needed)
-      const runs = await this.$.runStore.findRuns({
-        where: { id: { in: runIds } },
-        select: {
-          id: true,
-          spanId: true,
-          status: true,
-          lockedAt: true,
-          ttl: true,
-          taskEventStore: true,
-          createdAt: true,
-          associatedWaitpoint: { select: { id: true } },
-          organizationId: true,
-          projectId: true,
-          runtimeEnvironmentId: true,
+      const runs = await this.$.runStore.findRuns(
+        {
+          where: { id: { in: runIds } },
+          select: {
+            id: true,
+            spanId: true,
+            status: true,
+            lockedAt: true,
+            ttl: true,
+            taskEventStore: true,
+            createdAt: true,
+            associatedWaitpoint: { select: { id: true } },
+            organizationId: true,
+            projectId: true,
+            runtimeEnvironmentId: true,
+          },
+          // read-your-writes: the queue slot is already claimed; a lagging replica would orphan the run
         },
-      });
+        this.$.prisma
+      );
 
       // Filter runs that can be expired
       const runsToExpire: typeof runs = [];

internal-packages/run-store/src/runOpsStore.residencyMismatchFallback.test.ts

@@ -1,296 +0,0 @@
-// RED→GREEN lock for RoutingRunStore.findRun's ON-MISS FAN-OUT for a CLASSIFIABLE id.
-//
-// THE BUG: findRun for a classifiable id routed to the SINGLE owning store (by id-shape
-// classification) and returned whatever it gave — no on-miss fallback. When a run's PHYSICAL
-// residency does not match its id-shape classification (e.g. a pre-#4154 27-char base62 run that
-// lives on the NEW store but now classifies LEGACY), findRun routed to the wrong store, missed,
-// and returned null → a spurious 404 — even though the run is physically present on the OTHER DB
-// (and runs.list surfaces it). The unclassifiable path already fanned out NEW→LEGACY; this makes
-// the classifiable path equally robust.
-//
-// Uses the REAL two-physical-DB split (heteroRunOpsPostgresTest: prisma14 = full/legacy on PG14,
-// prisma17 = dedicated run-ops subset on PG17). NEVER mocked. The residency/classification MISMATCH
-// is simulated deterministically by injecting a custom `classify` fn into the RoutingRunStore
-// constructor — the physical row is written to the NEW store while `classify` reports its id LEGACY.
-
-import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
-import { ownerEngine, type Residency } from "@trigger.dev/core/v3/isomorphic";
-import type { PrismaClient } from "@trigger.dev/database";
-import type { RunOpsPrismaClient } from "@internal/run-ops-database";
-import { describe, expect } from "vitest";
-import { PostgresRunStore } from "./PostgresRunStore.js";
-import { RoutingRunStore } from "./runOpsStore.js";
-import type { CreateRunInput, RunStore } from "./types.js";
-
-// ownerEngine classifies by internal-id LENGTH/version char: 25 chars → cuid → LEGACY,
-// a v1 body (version "1" at index 25) → run-ops id → NEW.
-function cuidLegacy(seed: string): string {
-  return (seed + "c".repeat(25)).slice(0, 25);
-}
-function runOpsNew(seed: string): string {
-  return (seed.replace(/[^0-9a-v]/g, "0") + "k".repeat(24)).slice(0, 24) + "01";
-}
-
-function makeDedicatedStore(prisma17: RunOpsPrismaClient) {
-  return new PostgresRunStore({
-    prisma: prisma17 as never,
-    readOnlyPrisma: prisma17 as never,
-    schemaVariant: "dedicated",
-  });
-}
-
-function makeLegacyStore(prisma14: PrismaClient) {
-  return new PostgresRunStore({
-    prisma: prisma14,
-    readOnlyPrisma: prisma14,
-    schemaVariant: "legacy",
-  });
-}
-
-// Wrap a real store so findRun/findRunOnPrimary calls are COUNTED while every method still delegates
-// to the REAL PostgresRunStore (this is instrumentation, not a behavior mock — the underlying reads,
-// writes, getters all run for real). Lets us assert the FAST PATH does not touch the other store.
-function countingReads(
-  inner: RunStore,
-  counts: { findRun: number; findRunOnPrimary: number }
-): RunStore {
-  return new Proxy(inner, {
-    get(target, prop) {
-      // Read via target[prop] so getters (e.g. primaryReadClient) run with `this` = the real store.
-      const value = (target as unknown as Record<string | symbol, unknown>)[prop];
-      if (typeof value !== "function") return value;
-      if (prop === "findRun" || prop === "findRunOnPrimary") {
-        return (...args: unknown[]) => {
-          counts[prop as "findRun" | "findRunOnPrimary"] += 1;
-          return (value as (...a: unknown[]) => unknown).apply(target, args);
-        };
-      }
-      return (value as (...a: unknown[]) => unknown).bind(target);
-    },
-  }) as unknown as RunStore;
-}
-
-async function seedLegacyEnvironment(prisma14: PrismaClient, suffix: string) {
-  const organization = await prisma14.organization.create({
-    data: { title: `Org ${suffix}`, slug: `org-${suffix}` },
-  });
-  const project = await prisma14.project.create({
-    data: {
-      name: `Project ${suffix}`,
-      slug: `project-${suffix}`,
-      externalRef: `proj_${suffix}`,
-      organizationId: organization.id,
-    },
-  });
-  const environment = await prisma14.runtimeEnvironment.create({
-    data: {
-      type: "DEVELOPMENT",
-      slug: "dev",
-      projectId: project.id,
-      organizationId: organization.id,
-      apiKey: `tr_dev_${suffix}`,
-      pkApiKey: `pk_dev_${suffix}`,
-      shortcode: `short_${suffix}`,
-    },
-  });
-  return {
-    organizationId: organization.id,
-    projectId: project.id,
-    runtimeEnvironmentId: environment.id,
-    environmentId: environment.id,
-  };
-}
-
-function buildCreateRunInput(params: {
-  runId: string;
-  friendlyId: string;
-  organizationId: string;
-  projectId: string;
-  runtimeEnvironmentId: string;
-}): CreateRunInput {
-  return {
-    data: {
-      id: params.runId,
-      engine: "V2",
-      status: "PENDING",
-      friendlyId: params.friendlyId,
-      runtimeEnvironmentId: params.runtimeEnvironmentId,
-      environmentType: "DEVELOPMENT",
-      organizationId: params.organizationId,
-      projectId: params.projectId,
-      taskIdentifier: "my-task",
-      payload: "{}",
-      payloadType: "application/json",
-      traceContext: {},
-      traceId: `trace_${params.runId}`,
-      spanId: `span_${params.runId}`,
-      queue: "task/my-task",
-      isTest: false,
-      taskEventStore: "taskEvent",
-      depth: 0,
-    },
-    snapshot: {
-      engine: "V2",
-      executionStatus: "RUN_CREATED",
-      description: "Run was created",
-      runStatus: "PENDING",
-      environmentId: params.runtimeEnvironmentId,
-      environmentType: "DEVELOPMENT",
-      projectId: params.projectId,
-      organizationId: params.organizationId,
-    },
-  };
-}
-
-// Insert a TaskRun row DIRECTLY onto the NEW (dedicated) store, bypassing routing, so we can force a
-// residency/classification MISMATCH: the row is physically on #new while `classify` calls its id LEGACY.
-async function insertRunOnNewStore(
-  prisma17: RunOpsPrismaClient,
-  params: {
-    runId: string;
-    friendlyId: string;
-    environmentId: string;
-    organizationId: string;
-    projectId: string;
-  }
-) {
-  await prisma17.taskRun.create({
-    data: {
-      id: params.runId,
-      engine: "V2",
-      status: "PENDING",
-      friendlyId: params.friendlyId,
-      runtimeEnvironmentId: params.environmentId,
-      environmentType: "DEVELOPMENT",
-      organizationId: params.organizationId,
-      projectId: params.projectId,
-      taskIdentifier: "my-task",
-      payload: "{}",
-      payloadType: "application/json",
-      traceContext: {},
-      traceId: `trace_${params.runId}`,
-      spanId: `span_${params.runId}`,
-      queue: "task/my-task",
-      isTest: false,
-      taskEventStore: "taskEvent",
-      depth: 0,
-    },
-  });
-}
-
-describe("RoutingRunStore.findRun — on-miss fan-out for a classifiable id (residency ≠ classification)", () => {
-  // ── THE BUG: a run physically on #new whose id classifies LEGACY must still be found ──
-  // Without the on-miss fallback, findRun routes to #legacy (per classification), misses, returns null.
-  heteroRunOpsPostgresTest(
-    "returns a #new-resident run whose id classifies LEGACY (owning-store miss → other-store fallback)",
-    async ({ prisma14, prisma17 }) => {
-      const env = await seedLegacyEnvironment(prisma14, "mm1");
-      const newStore = makeDedicatedStore(prisma17);
-      const legacyStore = makeLegacyStore(prisma14);
-
-      // A run-ops-shaped id (so ownerEngine would say NEW), but we FORCE classify → LEGACY to model
-      // a residency/classification mismatch: physically on #new, classified LEGACY.
-      const mismatchId = runOpsNew("mm1");
-      const classify = (id: string): Residency => (id === mismatchId ? "LEGACY" : ownerEngine(id));
-      const router = new RoutingRunStore({ new: newStore, legacy: legacyStore, classify });
-
-      await insertRunOnNewStore(prisma17, {
-        runId: mismatchId,
-        friendlyId: "run_mm1",
-        environmentId: env.environmentId,
-        organizationId: env.organizationId,
-        projectId: env.projectId,
-      });
-
-      // Physical residency sanity: on #new only.
-      expect(await prisma17.taskRun.findUnique({ where: { id: mismatchId } })).not.toBeNull();
-      expect(await prisma14.taskRun.findUnique({ where: { id: mismatchId } })).toBeNull();
-
-      // classify → LEGACY routes to #legacy (miss); the fix falls back to #new and finds the run.
-      const byId = (await router.findRun({ id: mismatchId }, { select: { id: true } })) as Record<
-        string,
-        unknown
-      > | null;
-      expect(byId?.id).toBe(mismatchId);
-
-      // Same on the read-your-writes primary variant (a caller-passed writer → findRunOnPrimary).
-      const byIdPrimary = (await router.findRun(
-        { id: mismatchId },
-        { select: { id: true } },
-        prisma14
-      )) as Record<string, unknown> | null;
-      expect(byIdPrimary?.id).toBe(mismatchId);
-    }
-  );
-
-  // ── FAST PATH: a run found in its CLASSIFIED store is a SINGLE read (no second-store probe) ──
-  heteroRunOpsPostgresTest(
-    "does NOT read the other store when the classified (owning) store hits",
-    async ({ prisma14, prisma17 }) => {
-      const env = await seedLegacyEnvironment(prisma14, "mm2");
-
-      // NEW-resident run-ops-id run: owning store = #new. Wrap #legacy to catch any stray probe.
-      const newCounts = { findRun: 0, findRunOnPrimary: 0 };
-      const legacyCounts = { findRun: 0, findRunOnPrimary: 0 };
-      const newStore = countingReads(makeDedicatedStore(prisma17), newCounts);
-      const legacyStore = countingReads(makeLegacyStore(prisma14), legacyCounts);
-      const router = new RoutingRunStore({ new: newStore, legacy: legacyStore });
-
-      const newId = runOpsNew("mm2n"); // classifies NEW
-      await router.createRun(
-        buildCreateRunInput({
-          runId: newId,
-          friendlyId: "run_mm2_new",
-          organizationId: env.organizationId,
-          projectId: env.projectId,
-          runtimeEnvironmentId: env.runtimeEnvironmentId,
-        })
-      );
-
-      const hit = (await router.findRun({ id: newId }, { select: { id: true } })) as Record<
-        string,
-        unknown
-      > | null;
-      expect(hit?.id).toBe(newId);
-      // Owning store read exactly once; the other store NOT touched (fast path preserved).
-      expect(newCounts.findRun).toBe(1);
-      expect(legacyCounts.findRun).toBe(0);
-      expect(legacyCounts.findRunOnPrimary).toBe(0);
-
-      // Symmetric: a cuid run whose owning store is #legacy must not probe #new on a hit.
-      const legacyId = cuidLegacy("mm2l"); // classifies LEGACY
-      await router.createRun(
-        buildCreateRunInput({
-          runId: legacyId,
-          friendlyId: "run_mm2_legacy",
-          organizationId: env.organizationId,
-          projectId: env.projectId,
-          runtimeEnvironmentId: env.runtimeEnvironmentId,
-        })
-      );
-      newCounts.findRun = 0;
-      legacyCounts.findRun = 0;
-
-      const hitLegacy = (await router.findRun(
-        { id: legacyId },
-        { select: { id: true } }
-      )) as Record<string, unknown> | null;
-      expect(hitLegacy?.id).toBe(legacyId);
-      expect(legacyCounts.findRun).toBe(1);
-      expect(newCounts.findRun).toBe(0);
-    }
-  );
-
-  // ── A genuine miss on BOTH stores still returns null (fan-out exhausted) ──
-  heteroRunOpsPostgresTest(
-    "returns null when the run is on neither store",
-    async ({ prisma14, prisma17 }) => {
-      const newStore = makeDedicatedStore(prisma17);
-      const legacyStore = makeLegacyStore(prisma14);
-      const router = new RoutingRunStore({ new: newStore, legacy: legacyStore });
-      expect(
-        await router.findRun({ id: cuidLegacy("ghost") }, { select: { id: true } })
-      ).toBeNull();
-    }
-  );
-});

internal-packages/run-store/src/runOpsStore.ts

@@ -138,20 +138,6 @@ export class RoutingRunStore implements RunStore {
     return this.#routeOrNew(runId).runInTransaction(runId, fn);
   }
 
-  // A waitpoint WRITE co-locates with its run by id-shape (cuid → LEGACY, run-ops id → NEW,
-  // unclassifiable → LEGACY), mirroring how `blockRunWithWaitpointEdges` routes the edge by
-  // run id. The caller's `tx` is never forwarded: a routed write must run on the OWNING store's
-  // OWN client so the row lands in that store's database (same-store atomicity comes from the
-  // owning store opening its own transaction, never from a caller-supplied one).
-  #routeWaitpointWrite(
-    id: string | undefined,
-    _tx?: PrismaClientOrTransaction
-  ): { store: RunStore; tx?: PrismaClientOrTransaction } {
-    const store =
-      typeof id === "string" && this.#classifySafe(id) === "NEW" ? this.#new : this.#legacy;
-    return { store, tx: undefined };
-  }
-
   // Resolve which store ACTUALLY holds a waitpoint id: drain-on-read can relocate a cuid
   // waitpoint onto NEW while keeping its id, so probe the id-shape's home then the other.
   // `onPrimary` probes each store's own primary (read-your-writes callers; a fresh row may not
@@ -243,9 +229,11 @@ export class RoutingRunStore implements RunStore {
     const onPrimary = readYourWrites(argsOrClient, _client);
     const id = idFromWhere(where);
     if (id !== undefined) {
-      // Residency-classifiable (id/friendlyId): route to the owning store, then fall back to the
-      // OTHER store on a miss.
-      return this.#findRunRouted(id, where, args, onPrimary);
+      // Residency-classifiable (id/friendlyId): id-shape is destiny for a run's whole life — a run-ops
+      // id lives on NEW, a cuid on LEGACY — so read the one owning store (no cross-store fan-out).
+      const store = this.#routeOrNew(id);
+      const method = onPrimary ? "findRunOnPrimary" : "findRun";
+      return (store[method] as (...rest: unknown[]) => Promise<unknown>)(where, args);
     }
     // Unclassifiable where (e.g. spanId, idempotencyKey): the run may live on either DB,
     // so fan out NEW-first then LEGACY rather than defaulting to NEW — defaulting silently
@@ -253,31 +241,6 @@ export class RoutingRunStore implements RunStore {
     return this.#findRunUnrouted(where, args, onPrimary);
   }
 
-  // A classifiable id names its OWNING store by id-shape, but physical residency can diverge from
-  // classification (a pre-#4154 base62 run lives on #new yet classifies LEGACY). Read the owning
-  // store first — a hit is a SINGLE read (the fast path) — then, ONLY on a miss, probe the OTHER
-  // store before returning null, so a run whose residency ≠ its id-shape is found rather than 404'd.
-  // Both legs run the SAME index-covered TaskRun lookup; the fan-out cost is paid only on the (rare)
-  // miss. Mirrors #findRunUnrouted's shape but keyed on the classified owner.
-  async #findRunRouted(
-    id: string,
-    where: Prisma.TaskRunWhereInput,
-    args: unknown,
-    onPrimary: boolean
-  ): Promise<unknown> {
-    const method = onPrimary ? "findRunOnPrimary" : "findRun";
-    const owning = this.#routeOrNew(id);
-    const fromOwning = await (owning[method] as (...rest: unknown[]) => Promise<unknown>)(
-      where,
-      args
-    );
-    if (fromOwning != null) {
-      return fromOwning;
-    }
-    const other = owning === this.#new ? this.#legacy : this.#new;
-    return (other[method] as (...rest: unknown[]) => Promise<unknown>)(where, args);
-  }
-
   async #findRunUnrouted(
     where: Prisma.TaskRunWhereInput,
     args: unknown,
@@ -1187,7 +1150,7 @@ export class RoutingRunStore implements RunStore {
       opts?.residency,
       RoutingRunStore.#waitpointId(data)
     );
-    // Never forward the caller's tx into a routed write (matches #routeWaitpointWrite).
+    // Never forward the caller's tx into a routed write (it runs on the owning store's own client).
     return store.createWaitpoint(args, undefined);
   }
 
@@ -1244,6 +1207,11 @@ export class RoutingRunStore implements RunStore {
       args as Record<string, unknown>
     );
     const id = RoutingRunStore.#waitpointId((args as { where?: unknown }).where);
+    // A colocated lookup (no id, resolved via `coLocateWithRunId`) is the (env,idempotencyKey) dedup
+    // probe of createDateTimeWaitpoint/createManualWaitpoint: read-your-writes within the owning
+    // run/tree. On a retry it must observe attempt 1's just-written waitpoint to short-circuit, so it
+    // reads the owning store's PRIMARY — the replica can lag and miss it, re-arming/re-blocking the run.
+    const coLocatedDedup = id === undefined && opts?.coLocateWithRunId !== undefined;
     const store =
       id !== undefined
         ? await this.#resolveWaitpointStore(id, client !== undefined)
@@ -1254,7 +1222,7 @@ export class RoutingRunStore implements RunStore {
       store !== undefined
         ? ((await store.findWaitpoint(
             scalarArgs as typeof args,
-            RoutingRunStore.#ownPrimary(store, client)
+            coLocatedDedup ? store.primaryReadClient : RoutingRunStore.#ownPrimary(store, client)
           )) as Record<string, unknown> | null)
         : (((await this.#new.findWaitpoint(
             scalarArgs as typeof args,
@@ -1701,7 +1669,7 @@ export class RoutingRunStore implements RunStore {
   ): Promise<BatchTaskRun> {
     // Route by the batch's classifiable internal id: run-ops id→NEW, cuid→LEGACY. The caller's
     // `tx` is never forwarded — the create runs on the owning store's own client so the batch and
-    // its co-resident child runs/items land on the same DB. Mirrors #routeWaitpointWrite /
+    // its co-resident child runs/items land on the same DB. Mirrors the by-id waitpoint-write routing /
     // updateBatchTaskRun.
     const store = await this.#routeOrNewForWrite(data.id);
     return store.createBatchTaskRun(data, undefined);
@@ -1718,7 +1686,7 @@ export class RoutingRunStore implements RunStore {
     const id =
       typeof args.where.id === "string" ? args.where.id : (args.where.friendlyId ?? undefined);
     // The caller's `tx` is never forwarded — the update runs on the owning store's own client so
-    // it targets the DB the batch actually lives on. Mirrors #routeWaitpointWrite.
+    // it targets the DB the batch actually lives on. Mirrors the by-id waitpoint-write routing.
     const store = this.#routeOrNew(id);
     return store.updateBatchTaskRun(args, undefined);
   }
@@ -1880,7 +1848,7 @@ export class RoutingRunStore implements RunStore {
   // WaitpointTag — a standalone entity (no run/waitpoint FK) keyed by (environmentId, name).
   // ---------------------------------------------------------------------------
 
-  // Callers never mint a tag id (defaults to cuid), so #routeWaitpointWrite always resolves LEGACY
+  // Callers never mint a tag id (defaults to cuid), so a tag write always resolves LEGACY
   // today — deliberately single-homed, like standalone waitpoint tokens. If tag-id minting is ever made
   // residency-aware, findManyWaitpointTags must de-dupe by (environmentId, name) or names will duplicate.
   upsertWaitpointTag(

packages/redis-worker/src/mollifier/buffer.ts

@@ -462,6 +462,23 @@ export class MollifierBuffer {
     await this.redis.releaseMollifierClaim(claimKey, `${PENDING_PREFIX}${input.token}`);
   }
 
+  // Reopen a RESOLVED claim slot whose winner was cleared, so the claim-loser
+  // reacquire path can re-serialise a cross-DB recreate through a fresh claim.
+  // Compare-and-delete on the observed `expectedRunId` (never an unconditional
+  // DEL) so a concurrent reacquirer that already re-published a NEW winner is
+  // never wiped. Reuses the lookup self-heal's compare-and-delete Lua. Returns
+  // true if the stale slot was cleared.
+  async resetResolvedClaim(
+    input: IdempotencyLookupInput & { expectedRunId: string }
+  ): Promise<boolean> {
+    const claimKey = makeIdempotencyClaimKey(input);
+    const deleted = (await this.redis.delMollifierKeyIfEquals(
+      claimKey,
+      input.expectedRunId
+    )) as number;
+    return deleted === 1;
+  }
+
   // Read the current claim value, used by the wait/poll loop on losers
   // to detect "pending" → "resolved" transitions and timeouts.
   async readClaim(input: IdempotencyLookupInput): Promise<IdempotencyClaimResult | null> {