diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts
index a6e6199a49..b98f2920b3 100644
--- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts
@@ -126,10 +126,13 @@ function enhanceExecutionSnapshotWithWaitpoints(
 async function getSnapshotWaitpointIds(
   prisma: PrismaClientOrTransaction,
   snapshotId: string,
-  runStore?: RunStore
+  runStore?: RunStore,
+  // The owning run id, so the router can route to the run's store (the completed-waitpoint join
+  // co-locates with the snapshot/run) instead of fanning out to both run-ops DBs.
+  runId?: string
 ): Promise<string[]> {
   if (runStore) {
-    return runStore.findSnapshotCompletedWaitpointIds(snapshotId, prisma);
+    return runStore.findSnapshotCompletedWaitpointIds(snapshotId, prisma, runId);
   }
 
   const result = await prisma.$queryRaw<{ B: string }[]>`
@@ -144,10 +147,12 @@ async function getSnapshotWaitpointIds(
 async function getSnapshotWaitpointIdsWithPresence(
   prisma: PrismaClientOrTransaction,
   snapshotId: string,
-  runStore?: RunStore
+  runStore?: RunStore,
+  // The owning run id, so the router can route to the run's store instead of fanning out.
+  runId?: string
 ): Promise<{ present: boolean; ids: string[] }> {
   if (runStore) {
-    return runStore.findSnapshotCompletedWaitpointIdsWithPresence(snapshotId, prisma);
+    return runStore.findSnapshotCompletedWaitpointIdsWithPresence(snapshotId, prisma, runId);
   }
 
   const rows = await prisma.$queryRaw<{ id: string; B: string | null }[]>`
@@ -170,7 +175,10 @@ async function getSnapshotWaitpointIdsWithPresence(
 async function fetchWaitpointsInChunks(
   prisma: PrismaClientOrTransaction,
   waitpointIds: string[],
-  runStore?: RunStore
+  runStore?: RunStore,
+  // The owning run id, so the router routes to the run's store and only falls back to the other DB
+  // for the rare cross-tree token, instead of fanning every chunk out to both run-ops DBs.
+  runId?: string
 ): Promise<Waitpoint[]> {
   if (waitpointIds.length === 0) return [];
 
@@ -178,7 +186,7 @@ async function fetchWaitpointsInChunks(
   for (let i = 0; i < waitpointIds.length; i += WAITPOINT_CHUNK_SIZE) {
     const chunk = waitpointIds.slice(i, i + WAITPOINT_CHUNK_SIZE);
     const waitpoints = runStore
-      ? await runStore.findManyWaitpoints({ where: { id: { in: chunk } } }, prisma)
+      ? await runStore.findManyWaitpoints({ where: { id: { in: chunk } } }, prisma, runId)
       : await prisma.waitpoint.findMany({
           where: { id: { in: chunk } },
         });
@@ -347,7 +355,8 @@ export async function getExecutionSnapshotsSince(
   const { present, ids } = await getSnapshotWaitpointIdsWithPresence(
     prisma,
     latestSnapshot.id,
-    runStore
+    runStore,
+    runId
   );
   let waitpointIds = ids;
 
@@ -356,7 +365,7 @@ export async function getExecutionSnapshotsSince(
   // authoritative - re-read from the primary so the runner is not handed a waitpoint-less continue
   // (which it silently drops, hanging the run). Single-reader replicas never hit this (present stays true).
   if (repairClient && repairClient !== prisma && !present) {
-    waitpointIds = await getSnapshotWaitpointIds(repairClient, latestSnapshot.id, runStore);
+    waitpointIds = await getSnapshotWaitpointIds(repairClient, latestSnapshot.id, runStore, runId);
     readClient = repairClient;
   }
 
@@ -369,7 +378,12 @@ export async function getExecutionSnapshotsSince(
   // poll even against a caught-up replica.
   const expectedCount = new Set(latestSnapshot.completedWaitpointOrder ?? []).size;
   if (repairClient && repairClient !== prisma && waitpointIds.length < expectedCount) {
-    const repaired = await getSnapshotWaitpointIds(repairClient, latestSnapshot.id, runStore);
+    const repaired = await getSnapshotWaitpointIds(
+      repairClient,
+      latestSnapshot.id,
+      runStore,
+      runId
+    );
     if (repaired.length > waitpointIds.length) {
       waitpointIds = repaired;
       readClient = repairClient;
@@ -377,7 +391,7 @@ export async function getExecutionSnapshotsSince(
   }
 
   // Step 4: Fetch waitpoints in chunks to avoid NAPI string conversion limits
-  const waitpoints = await fetchWaitpointsInChunks(readClient, waitpointIds, runStore);
+  const waitpoints = await fetchWaitpointsInChunks(readClient, waitpointIds, runStore, runId);
 
   // Step 5: Build enhanced snapshots - only latest gets waitpoints, others get empty arrays
   // The runner only uses completedWaitpoints from the latest snapshot anyway
diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
index a0d5b73349..3e71bdbd32 100644
--- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -59,9 +59,10 @@ export class WaitpointSystem {
     runId: string;
     tx?: PrismaClientOrTransaction;
   }) {
-    // Route the delete: a run's edges may live on #new and/or #legacy (mid-drain), so it must fan
-    // across both stores. The caller's `tx` is not forwarded into either leg — each store's delete
-    // runs on its own client (the router never threads a control-plane tx into a routed write).
+    // A run's edges co-locate with the run (the edge write routes by runId), so the router routes this
+    // taskRunId-keyed delete to the run's store rather than fanning out. The caller's `tx` is not
+    // forwarded — the delete runs on the owning store's own client (the router never threads a
+    // control-plane tx into a routed write).
     const deleted = await this.$.runStore.deleteManyTaskRunWaitpoints(
       { where: { taskRunId: runId } },
       tx
@@ -493,7 +494,9 @@ export class WaitpointSystem {
 
       // Check if the run is actually blocked using a separate query (see above). Pass the writer so the
       // pending re-read is read-your-writes on the owning PRIMARY (a lagging replica can strand the run).
-      const pendingCount = await this.$.runStore.countPendingWaitpoints($waitpoints, prisma);
+      // Route by the blocked run id: its blocking waitpoints co-locate with the run, so the router
+      // counts on the run's store and only falls back to the other DB for a cross-tree token.
+      const pendingCount = await this.$.runStore.countPendingWaitpoints($waitpoints, prisma, runId);
 
       const isRunBlocked = pendingCount > 0;
 
diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts
index c206f4605f..0700a4896b 100644
--- a/internal-packages/run-store/src/PostgresRunStore.ts
+++ b/internal-packages/run-store/src/PostgresRunStore.ts
@@ -1897,7 +1897,9 @@ export class PostgresRunStore implements RunStore {
 
   async findSnapshotCompletedWaitpointIds(
     snapshotId: string,
-    client?: ReadClient
+    client?: ReadClient,
+    // `runId` selects residency at the router; a single store has one client and ignores it.
+    _runId?: string
   ): Promise<string[]> {
     const prisma = client ?? this.readOnlyPrisma;
 
@@ -1924,7 +1926,9 @@ export class PostgresRunStore implements RunStore {
   // return the snapshot (via a separate read) while a different, laggier reader returns 0 links.
   async findSnapshotCompletedWaitpointIdsWithPresence(
     snapshotId: string,
-    client?: ReadClient
+    client?: ReadClient,
+    // `runId` selects residency at the router; a single store has one client and ignores it.
+    _runId?: string
   ): Promise<{ present: boolean; ids: string[] }> {
     const prisma = client ?? this.readOnlyPrisma;
 
@@ -2097,7 +2101,12 @@ export class PostgresRunStore implements RunStore {
       SELECT COUNT(*) FROM inserted`;
   }
 
-  async countPendingWaitpoints(waitpointIds: string[], client?: ReadClient): Promise<number> {
+  async countPendingWaitpoints(
+    waitpointIds: string[],
+    client?: ReadClient,
+    // `runId` selects residency at the router; a single store has one client and ignores it.
+    _runId?: string
+  ): Promise<number> {
     const prisma = client ?? this.readOnlyPrisma;
 
     if (waitpointIds.length === 0) {
@@ -2128,6 +2137,35 @@ export class PostgresRunStore implements RunStore {
     return Number(pendingCheck[0]?.pending_count ?? 0);
   }
 
+  // One `SELECT id, status` returns which ids are PENDING and which exist on this store (any status),
+  // so the router can route by run id and only re-count the ids ABSENT here on the other DB — without
+  // undercounting pending (which would prematurely unblock a run) or double-counting a drain-mirrored
+  // id. Reads only id+status, so it stays cheap for a run's small blocking set.
+  async countPendingWaitpointsWithPresence(
+    waitpointIds: string[],
+    client?: ReadClient
+  ): Promise<{ pendingIds: string[]; presentIds: string[] }> {
+    const prisma = client ?? this.readOnlyPrisma;
+
+    if (waitpointIds.length === 0) {
+      return { pendingIds: [], presentIds: [] };
+    }
+
+    const rows =
+      this.schemaVariant === "dedicated"
+        ? await prisma.$queryRaw<{ id: string; status: string }[]>`
+            SELECT id, status FROM "Waitpoint" WHERE id = ANY(${waitpointIds}::text[])
+          `
+        : await prisma.$queryRaw<{ id: string; status: string }[]>`
+            SELECT id, status FROM "Waitpoint" WHERE id IN (${Prisma.join(waitpointIds)})
+          `;
+
+    return {
+      pendingIds: rows.filter((r) => r.status === "PENDING").map((r) => r.id),
+      presentIds: rows.map((r) => r.id),
+    };
+  }
+
   async createWaitpoint<T extends Prisma.WaitpointCreateArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointCreateArgs>,
     tx?: PrismaClientOrTransaction
@@ -2189,7 +2227,9 @@ export class PostgresRunStore implements RunStore {
 
   async findManyWaitpoints<T extends Prisma.WaitpointFindManyArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointFindManyArgs>,
-    client?: ReadClient
+    client?: ReadClient,
+    // `runId` selects residency at the router; a single store has one client and ignores it.
+    _runId?: string
   ): Promise<Prisma.WaitpointGetPayload<T>[]> {
     const prisma = client ?? this.readOnlyPrisma;
 
diff --git a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts
index ac042d544a..a362f8228b 100644
--- a/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts
+++ b/internal-packages/run-store/src/PostgresRunStore.writeAtomicity.test.ts
@@ -748,7 +748,7 @@ describe("RoutingRunStore never threads a caller tx into either sub-store (recor
   );
 
   heteroRunOpsPostgresTest(
-    "deleteManyTaskRunWaitpoints fan-out hands BOTH sub-stores an undefined tx",
+    "deleteManyTaskRunWaitpoints routes by taskRunId to the owning store with an undefined tx",
     async ({ prisma14, prisma17 }) => {
       const legacyCalls: RecordedCall[] = [];
       const newCalls: RecordedCall[] = [];
@@ -769,14 +769,15 @@ describe("RoutingRunStore never threads a caller tx into either sub-store (recor
       );
       await seedLegacyBlockingEdge(prisma14, env, runId, "spy_del");
 
-      // Keyed by taskRunId → the both-stores fan-out branch. Pass the base control-plane client as tx.
+      // Keyed by a classifiable taskRunId (a cuid run → #legacy): routes to the owning store, no
+      // fan-out. Pass the base control-plane client as tx — it must NOT be threaded into the routed leg.
       await router.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, prisma14);
 
       const legacyTx = txArgsFor(legacyCalls, "deleteManyTaskRunWaitpoints");
       const newTx = txArgsFor(newCalls, "deleteManyTaskRunWaitpoints");
       expect(legacyTx.length).toBeGreaterThan(0);
-      expect(newTx.length).toBeGreaterThan(0);
-      for (const arg of [...legacyTx, ...newTx]) expect(arg).toBeUndefined();
+      expect(newTx.length).toBe(0); // routed to #legacy only; the non-owning store is never touched
+      for (const arg of legacyTx) expect(arg).toBeUndefined();
     }
   );
 });
diff --git a/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts b/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts
index ca2027d572..ba87c75ff5 100644
--- a/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts
+++ b/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts
@@ -834,18 +834,20 @@ describe("RoutingRunStore — mixed-residency matrix (cuid #legacy + run-ops id
     }
   );
 
-  // ── Case 11b: deleteManyTaskRunWaitpoints by taskRunId fans out to both and sums (:944) ──
-  // A run's edges can straddle DBs mid-drain; a delete keyed by taskRunId (not waitpointId) must
-  // delete from BOTH DBs and sum the count.
+  // ── Case 11b: deleteManyTaskRunWaitpoints by taskRunId routes to the run's store (no fan-out) ──
+  // An edge always co-locates with its run (blockRunWithWaitpointEdges routes the write by runId), so
+  // a run's edges only ever live on ONE store — a straddle can't arise via the write path. A delete
+  // keyed by a classifiable taskRunId therefore routes to the run's store and must NOT touch the other
+  // DB. To prove the routing (not a fan-out), we manufacture an edge on the non-owning DB too and
+  // assert it survives the delete.
   heteroRunOpsPostgresTest(
-    "case 11b: deleteManyTaskRunWaitpoints by taskRunId deletes edges on both DBs and sums",
+    "case 11b: deleteManyTaskRunWaitpoints by taskRunId routes to the run's store and leaves the other DB untouched",
     async ({ prisma14, prisma17 }) => {
       const { router } = makeSplitRouter(prisma14, prisma17);
       const env = await seedSharedEnv(prisma14, "m11b");
 
-      // ONE logical run id whose edges happen to exist on BOTH DBs (the straddle the fan-out guards).
-      // The edge is FK-free on #new (unnest path) and FK-bound on #legacy, so seed a co-resident
-      // waitpoint + run on #legacy for its edge, and write the #new edge directly.
+      // A run-ops run (→ #new). Its real edge lives on #new; we also plant an edge on #legacy that the
+      // write path could never create, purely to prove the routed delete does not fan out to #legacy.
       const runId = runOpsNew("m11br");
       const legacyToken = cuidLegacy("m11bt");
       await router.createRun(buildCreateRunInput({ runId, friendlyId: "run_m11b", ...env }));
@@ -881,7 +883,7 @@ describe("RoutingRunStore — mixed-residency matrix (cuid #legacy + run-ops id
       await prisma14.$executeRawUnsafe(
         `INSERT INTO "TaskRunWaitpoint" ("id","taskRunId","waitpointId","projectId","createdAt","updatedAt") VALUES (gen_random_uuid(),'${runId}','${legacyToken}','${env.projectId}',NOW(),NOW())`
       );
-      // #new edge (FK-free) pointing at a run-ops token absent locally — drain straddle.
+      // The run's real edge, on its own DB (#new, FK-free unnest path), pointing at a run-ops token.
       const newToken = runOpsNew("m11bn");
       await prisma17.$executeRawUnsafe(
         `INSERT INTO "TaskRunWaitpoint" ("id","taskRunId","waitpointId","projectId","createdAt","updatedAt") VALUES (gen_random_uuid(),'${runId}','${newToken}','${env.projectId}',NOW(),NOW())`
@@ -891,10 +893,10 @@ describe("RoutingRunStore — mixed-residency matrix (cuid #legacy + run-ops id
       expect(await prisma17.taskRunWaitpoint.count({ where: { taskRunId: runId } })).toBe(1);
 
       const { count } = await router.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } });
-      expect(count).toBe(2); // one edge deleted on each DB, summed
-
-      expect(await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId } })).toBe(0);
+      // Routed to #new (the run's store): only its edge is deleted; #legacy is never touched.
+      expect(count).toBe(1);
       expect(await prisma17.taskRunWaitpoint.count({ where: { taskRunId: runId } })).toBe(0);
+      expect(await prisma14.taskRunWaitpoint.count({ where: { taskRunId: runId } })).toBe(1);
     }
   );
 });
diff --git a/internal-packages/run-store/src/runOpsStore.runKeyedRouting.test.ts b/internal-packages/run-store/src/runOpsStore.runKeyedRouting.test.ts
new file mode 100644
index 0000000000..e29ecb11cb
--- /dev/null
+++ b/internal-packages/run-store/src/runOpsStore.runKeyedRouting.test.ts
@@ -0,0 +1,348 @@
+import { describe, expect, it } from "vitest";
+import { RoutingRunStore } from "./runOpsStore.js";
+import type { ReadClient, RunStore } from "./types.js";
+
+// Pure routing unit tests: run-keyed waitpoint/snapshot reads must route by the run id in scope
+// instead of fanning out to BOTH run-ops DBs. No DB: each slot is a fake RunStore backed by a
+// per-slot set of waitpoint rows / snapshot-join ids, so the assertions are purely about WHICH store
+// the router queries (the co-located run's store, never the other) and about the route-then-fallback
+// that keeps a rare cross-tree token visible. Correctness against real two-DB topology is covered by
+// the heteroRunOpsPostgresTest suites (crossDbTokenBlock, snapshotCompletedWaitpoints, …).
+
+type Call = { method: string; args: unknown[] };
+
+type WaitpointRow = { id: string; status: "PENDING" | "COMPLETED" };
+
+type FakeConfig = {
+  // Waitpoint rows resident on this store, keyed by id → status (for findManyWaitpoints /
+  // countPendingWaitpoints[WithPresence]).
+  waitpoints?: WaitpointRow[];
+  // Snapshot-join waitpoint ids resident on this store (for findSnapshotCompletedWaitpointIds).
+  snapshotWaitpointIds?: string[];
+  // Whether this store has the snapshot at all (for the WithPresence variant).
+  snapshotPresent?: boolean;
+  // Edge rows to return from findManyTaskRunWaitpoints, regardless of filter (routing-only).
+  edges?: Array<Record<string, unknown>>;
+};
+
+type FakeStore = RunStore & {
+  slot: "new" | "legacy";
+  calls: Call[];
+  primaryReadClient: { __primary: "new" | "legacy" };
+};
+
+function idsFromWhere(where: unknown): string[] | undefined {
+  const id = (where as { id?: unknown } | undefined)?.id;
+  if (typeof id === "string") return [id];
+  if (id && typeof id === "object") {
+    const inArr = (id as { in?: unknown }).in;
+    if (Array.isArray(inArr)) return inArr.filter((x): x is string => typeof x === "string");
+  }
+  return undefined;
+}
+
+function fakeStore(slot: "new" | "legacy", config: FakeConfig = {}): FakeStore {
+  const calls: Call[] = [];
+  const rows = config.waitpoints ?? [];
+  const byId = new Map(rows.map((r) => [r.id, r]));
+
+  const record = (method: string) => (args: unknown[]) => calls.push({ method, args });
+
+  const store: Partial<FakeStore> = {
+    slot,
+    calls,
+    primaryReadClient: { __primary: slot },
+
+    findManyTaskRunWaitpoints: ((args: unknown, client?: ReadClient) => {
+      record("findManyTaskRunWaitpoints")([args, client]);
+      return Promise.resolve((config.edges ?? []) as never);
+    }) as FakeStore["findManyTaskRunWaitpoints"],
+
+    deleteManyTaskRunWaitpoints: ((args: unknown, tx?: unknown) => {
+      record("deleteManyTaskRunWaitpoints")([args, tx]);
+      return Promise.resolve({ count: rows.length } as never);
+    }) as FakeStore["deleteManyTaskRunWaitpoints"],
+
+    findSnapshotCompletedWaitpointIds: ((snapshotId: string, client?: ReadClient) => {
+      record("findSnapshotCompletedWaitpointIds")([snapshotId, client]);
+      return Promise.resolve(config.snapshotWaitpointIds ?? []);
+    }) as FakeStore["findSnapshotCompletedWaitpointIds"],
+
+    findSnapshotCompletedWaitpointIdsWithPresence: ((snapshotId: string, client?: ReadClient) => {
+      record("findSnapshotCompletedWaitpointIdsWithPresence")([snapshotId, client]);
+      return Promise.resolve({
+        present: config.snapshotPresent ?? false,
+        ids: config.snapshotWaitpointIds ?? [],
+      });
+    }) as FakeStore["findSnapshotCompletedWaitpointIdsWithPresence"],
+
+    findManyWaitpoints: ((args: { where?: unknown }, client?: ReadClient) => {
+      record("findManyWaitpoints")([args, client]);
+      const requested = idsFromWhere(args.where);
+      const result =
+        requested === undefined
+          ? rows
+          : requested.map((id) => byId.get(id)).filter((r): r is WaitpointRow => r != null);
+      return Promise.resolve(result as never);
+    }) as FakeStore["findManyWaitpoints"],
+
+    countPendingWaitpoints: ((waitpointIds: string[], client?: ReadClient) => {
+      record("countPendingWaitpoints")([waitpointIds, client]);
+      const count = waitpointIds.filter((id) => byId.get(id)?.status === "PENDING").length;
+      return Promise.resolve(count);
+    }) as FakeStore["countPendingWaitpoints"],
+
+    countPendingWaitpointsWithPresence: ((waitpointIds: string[], client?: ReadClient) => {
+      record("countPendingWaitpointsWithPresence")([waitpointIds, client]);
+      const presentIds = waitpointIds.filter((id) => byId.has(id));
+      const pendingIds = presentIds.filter((id) => byId.get(id)?.status === "PENDING");
+      return Promise.resolve({ pendingIds, presentIds });
+    }) as FakeStore["countPendingWaitpointsWithPresence"],
+  };
+
+  return store as unknown as FakeStore;
+}
+
+// Deterministic residency by id prefix via the classify seam (no dependence on id-shape rules).
+function buildRouter(newConfig: FakeConfig = {}, legacyConfig: FakeConfig = {}) {
+  const newStore = fakeStore("new", newConfig);
+  const legacyStore = fakeStore("legacy", legacyConfig);
+  const router = new RoutingRunStore({
+    new: newStore,
+    legacy: legacyStore,
+    classify: (id: string) => (id.startsWith("new") ? "NEW" : "LEGACY"),
+  });
+  return { router, newStore, legacyStore };
+}
+
+const WRITER = { __writer: true } as unknown as ReadClient; // non-replica → escalates to own primary
+
+describe("RoutingRunStore.findManyTaskRunWaitpoints — route by taskRunId (no fan-out)", () => {
+  it("routes an edge read keyed by a NEW run id to the new store only", async () => {
+    const { router, newStore, legacyStore } = buildRouter();
+    await router.findManyTaskRunWaitpoints({
+      where: { taskRunId: "new_run" },
+      select: { taskRunId: true },
+    });
+    expect(newStore.calls.map((c) => c.method)).toEqual(["findManyTaskRunWaitpoints"]);
+    expect(legacyStore.calls).toHaveLength(0);
+  });
+
+  it("routes an edge read keyed by a LEGACY run id to the legacy store only", async () => {
+    const { router, newStore, legacyStore } = buildRouter();
+    await router.findManyTaskRunWaitpoints({
+      where: { taskRunId: "legacy_run" },
+      select: { taskRunId: true },
+    });
+    expect(legacyStore.calls.map((c) => c.method)).toEqual(["findManyTaskRunWaitpoints"]);
+    expect(newStore.calls).toHaveLength(0);
+  });
+
+  it("escalates a caller writer client to the owning store's own primary", async () => {
+    const { router, newStore } = buildRouter();
+    await router.findManyTaskRunWaitpoints(
+      { where: { taskRunId: "new_run" }, select: { taskRunId: true } },
+      WRITER
+    );
+    expect(newStore.calls[0]?.args[1]).toEqual({ __primary: "new" });
+  });
+
+  it("still fans out when keyed by waitpointId (no run id in scope)", async () => {
+    const { router, newStore, legacyStore } = buildRouter();
+    await router.findManyTaskRunWaitpoints({
+      where: { waitpointId: "waitpoint_x" },
+      select: { taskRunId: true },
+    });
+    expect(newStore.calls).toHaveLength(1);
+    expect(legacyStore.calls).toHaveLength(1);
+  });
+});
+
+describe("RoutingRunStore.deleteManyTaskRunWaitpoints — route by taskRunId (no fan-out)", () => {
+  it("deletes only on the owning store for a classifiable taskRunId", async () => {
+    const { router, newStore, legacyStore } = buildRouter({ waitpoints: [] });
+    const result = await router.deleteManyTaskRunWaitpoints({
+      where: { taskRunId: "legacy_run", id: { in: ["waitpoint_a"] } },
+    });
+    expect(legacyStore.calls.map((c) => c.method)).toEqual(["deleteManyTaskRunWaitpoints"]);
+    expect(newStore.calls).toHaveLength(0);
+    expect(result.count).toBe(0);
+  });
+
+  it("still fans out and sums when there is no taskRunId in the where", async () => {
+    const { router, newStore, legacyStore } = buildRouter();
+    await router.deleteManyTaskRunWaitpoints({ where: { waitpointId: "waitpoint_x" } });
+    expect(newStore.calls).toHaveLength(1);
+    expect(legacyStore.calls).toHaveLength(1);
+  });
+
+  it("never threads a caller tx into the routed delete", async () => {
+    const { router, legacyStore } = buildRouter();
+    await router.deleteManyTaskRunWaitpoints({ where: { taskRunId: "legacy_run" } }, {
+      $fake: "cp-tx",
+    } as never);
+    expect(legacyStore.calls[0]?.args[1]).toBeUndefined();
+  });
+});
+
+describe("RoutingRunStore.findSnapshotCompletedWaitpointIds — route by runId", () => {
+  it("routes to the run's store when a runId is threaded through", async () => {
+    const { router, newStore, legacyStore } = buildRouter(
+      { snapshotWaitpointIds: ["waitpoint_n"] },
+      { snapshotWaitpointIds: ["waitpoint_l"] }
+    );
+    const ids = await router.findSnapshotCompletedWaitpointIds(
+      "c".repeat(25),
+      undefined,
+      "new_run"
+    );
+    expect(ids).toEqual(["waitpoint_n"]);
+    expect(legacyStore.calls).toHaveLength(0);
+    expect(newStore.calls.map((c) => c.method)).toEqual(["findSnapshotCompletedWaitpointIds"]);
+  });
+
+  it("still fans out and merges when no runId is supplied", async () => {
+    const { router, newStore, legacyStore } = buildRouter(
+      { snapshotWaitpointIds: ["waitpoint_n"] },
+      { snapshotWaitpointIds: ["waitpoint_l"] }
+    );
+    const ids = await router.findSnapshotCompletedWaitpointIds("c".repeat(25));
+    expect(ids.sort()).toEqual(["waitpoint_l", "waitpoint_n"]);
+    expect(newStore.calls).toHaveLength(1);
+    expect(legacyStore.calls).toHaveLength(1);
+  });
+});
+
+describe("RoutingRunStore.findSnapshotCompletedWaitpointIdsWithPresence — route by runId", () => {
+  it("routes to the run's store when a runId is threaded through", async () => {
+    const { router, newStore } = buildRouter(
+      { snapshotWaitpointIds: ["waitpoint_n"], snapshotPresent: true },
+      { snapshotWaitpointIds: ["waitpoint_l"], snapshotPresent: true }
+    );
+    const res = await router.findSnapshotCompletedWaitpointIdsWithPresence(
+      "c".repeat(25),
+      undefined,
+      "legacy_run"
+    );
+    expect(res).toEqual({ present: true, ids: ["waitpoint_l"] });
+    expect(newStore.calls).toHaveLength(0);
+  });
+
+  it("still fans out (present is the OR) when no runId is supplied", async () => {
+    const { router } = buildRouter(
+      { snapshotWaitpointIds: [], snapshotPresent: false },
+      { snapshotWaitpointIds: ["waitpoint_l"], snapshotPresent: true }
+    );
+    const res = await router.findSnapshotCompletedWaitpointIdsWithPresence("c".repeat(25));
+    expect(res).toEqual({ present: true, ids: ["waitpoint_l"] });
+  });
+});
+
+describe("RoutingRunStore.findManyWaitpoints — route by runId then fall back for missing ids", () => {
+  it("queries only the run's store when every requested token co-locates with the run", async () => {
+    const { router, newStore, legacyStore } = buildRouter({
+      waitpoints: [
+        { id: "waitpoint_a", status: "COMPLETED" },
+        { id: "waitpoint_b", status: "COMPLETED" },
+      ],
+    });
+    const rows = (await router.findManyWaitpoints(
+      { where: { id: { in: ["waitpoint_a", "waitpoint_b"] } } },
+      undefined,
+      "new_run"
+    )) as WaitpointRow[];
+    expect(rows.map((r) => r.id).sort()).toEqual(["waitpoint_a", "waitpoint_b"]);
+    expect(legacyStore.calls).toHaveLength(0);
+    expect(newStore.calls).toHaveLength(1);
+  });
+
+  it("falls back to the other store for ONLY the ids missing on the run's store (cross-tree token)", async () => {
+    const { router, legacyStore } = buildRouter(
+      { waitpoints: [{ id: "waitpoint_local", status: "COMPLETED" }] },
+      { waitpoints: [{ id: "waitpoint_crosstree", status: "COMPLETED" }] }
+    );
+    const rows = (await router.findManyWaitpoints(
+      { where: { id: { in: ["waitpoint_local", "waitpoint_crosstree"] } } },
+      undefined,
+      "new_run"
+    )) as WaitpointRow[];
+    expect(rows.map((r) => r.id).sort()).toEqual(["waitpoint_crosstree", "waitpoint_local"]);
+    // The fallback leg is queried with ONLY the missing id, never the whole set.
+    const fallbackCall = legacyStore.calls[0];
+    expect(fallbackCall?.method).toBe("findManyWaitpoints");
+    const fallbackWhere = (fallbackCall!.args[0] as { where?: unknown }).where;
+    expect(idsFromWhere(fallbackWhere)).toEqual(["waitpoint_crosstree"]);
+  });
+
+  it("still fans out (NEW-wins dedup) when no runId is supplied", async () => {
+    const { router, newStore, legacyStore } = buildRouter(
+      { waitpoints: [{ id: "waitpoint_a", status: "COMPLETED" }] },
+      { waitpoints: [{ id: "waitpoint_a", status: "PENDING" }] }
+    );
+    const rows = (await router.findManyWaitpoints({
+      where: { id: { in: ["waitpoint_a"] } },
+    })) as WaitpointRow[];
+    expect(newStore.calls).toHaveLength(1);
+    expect(legacyStore.calls).toHaveLength(1);
+    // NEW-wins: the deduped row is the NEW copy (COMPLETED), not the stale legacy PENDING one.
+    expect(rows).toEqual([{ id: "waitpoint_a", status: "COMPLETED" }]);
+  });
+});
+
+describe("RoutingRunStore.countPendingWaitpoints — route by runId then partition-fallback", () => {
+  it("counts on the run's store only when every waitpoint co-locates with the run", async () => {
+    const { router, newStore, legacyStore } = buildRouter({
+      waitpoints: [
+        { id: "waitpoint_a", status: "PENDING" },
+        { id: "waitpoint_b", status: "COMPLETED" },
+      ],
+    });
+    const count = await router.countPendingWaitpoints(
+      ["waitpoint_a", "waitpoint_b"],
+      undefined,
+      "new_run"
+    );
+    expect(count).toBe(1);
+    expect(legacyStore.calls).toHaveLength(0);
+    expect(newStore.calls.map((c) => c.method)).toEqual(["countPendingWaitpointsWithPresence"]);
+  });
+
+  it("counts a cross-tree pending token via the fallback so a blocked run is not prematurely unblocked", async () => {
+    // The classic crossDbTokenBlock shape: a LEGACY run blocks on a token resident on the NEW DB.
+    const { router, newStore } = buildRouter(
+      { waitpoints: [{ id: "waitpoint_crosstree", status: "PENDING" }] },
+      { waitpoints: [] }
+    );
+    const count = await router.countPendingWaitpoints(
+      ["waitpoint_crosstree"],
+      undefined,
+      "legacy_run"
+    );
+    expect(count).toBe(1);
+    // Fallback queried the other store with ONLY the id missing on the run's store.
+    expect(newStore.calls.map((c) => c.method)).toEqual(["countPendingWaitpoints"]);
+    expect(newStore.calls[0]?.args[0]).toEqual(["waitpoint_crosstree"]);
+  });
+
+  it("trusts the run's store for an id present there (COMPLETED) even if a stale mirror is PENDING elsewhere", async () => {
+    const { router, legacyStore } = buildRouter(
+      { waitpoints: [{ id: "waitpoint_a", status: "COMPLETED" }] },
+      { waitpoints: [{ id: "waitpoint_a", status: "PENDING" }] }
+    );
+    const count = await router.countPendingWaitpoints(["waitpoint_a"], undefined, "new_run");
+    // Present on the run's store → not in the missing set → the other store is never consulted.
+    expect(count).toBe(0);
+    expect(legacyStore.calls).toHaveLength(0);
+  });
+
+  it("still fans out and sums when no runId is supplied", async () => {
+    const { router, newStore, legacyStore } = buildRouter(
+      { waitpoints: [{ id: "waitpoint_a", status: "PENDING" }] },
+      { waitpoints: [{ id: "waitpoint_b", status: "PENDING" }] }
+    );
+    const count = await router.countPendingWaitpoints(["waitpoint_a", "waitpoint_b"]);
+    expect(count).toBe(2);
+    expect(newStore.calls).toHaveLength(1);
+    expect(legacyStore.calls).toHaveLength(1);
+  });
+});
diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts
index 4a98b8f003..d32833ca31 100644
--- a/internal-packages/run-store/src/runOpsStore.ts
+++ b/internal-packages/run-store/src/runOpsStore.ts
@@ -977,13 +977,22 @@ export class RoutingRunStore implements RunStore {
     return store.createExecutionSnapshot(input, undefined);
   }
 
-  // Snapshot ids are cuids (they always classify LEGACY), and a snapshot's CompletedWaitpoint join
-  // co-locates with its run, which may live on either store, so fan out to BOTH and merge (like
-  // findWaitpointCompletedSnapshotIds) rather than route by the un-classifiable snapshot id.
+  // The CompletedWaitpoint join co-locates with the snapshot, which co-locates with its run. When the
+  // caller threads the run id (executionSnapshotSystem has it in scope), route to the run's store — no
+  // fan-out. Snapshot ids are cuids (they always classify LEGACY), so absent a run id we can't route
+  // by the snapshot id and must fan out to BOTH stores and merge (like findWaitpointCompletedSnapshotIds).
   async findSnapshotCompletedWaitpointIds(
     snapshotId: string,
-    client?: ReadClient
+    client?: ReadClient,
+    runId?: string
   ): Promise<string[]> {
+    if (runId !== undefined) {
+      const store = this.#routeOrNew(runId);
+      return store.findSnapshotCompletedWaitpointIds(
+        snapshotId,
+        RoutingRunStore.#ownPrimary(store, client)
+      );
+    }
     const [fromNew, fromLegacy] = await Promise.all([
       this.#new.findSnapshotCompletedWaitpointIds(
         snapshotId,
@@ -997,12 +1006,20 @@ export class RoutingRunStore implements RunStore {
     return uniqueStrings([...fromNew, ...fromLegacy]);
   }
 
-  // Snapshot-id has no residency to route on, so fan out; the snapshot lives on exactly one store, so
-  // `present` is the OR and `ids` the union.
+  // As above: route to the run's store when the run id is threaded through, else fan out (the snapshot
+  // lives on exactly one store, so `present` is the OR and `ids` the union).
   async findSnapshotCompletedWaitpointIdsWithPresence(
     snapshotId: string,
-    client?: ReadClient
+    client?: ReadClient,
+    runId?: string
   ): Promise<{ present: boolean; ids: string[] }> {
+    if (runId !== undefined) {
+      const store = this.#routeOrNew(runId);
+      return store.findSnapshotCompletedWaitpointIdsWithPresence(
+        snapshotId,
+        RoutingRunStore.#ownPrimary(store, client)
+      );
+    }
     const [fromNew, fromLegacy] = await Promise.all([
       this.#new.findSnapshotCompletedWaitpointIdsWithPresence(
         snapshotId,
@@ -1070,20 +1087,74 @@ export class RoutingRunStore implements RunStore {
     return (await this.#routeOrNewForWrite(params.runId)).blockRunWithWaitpointEdges(edges);
   }
 
-  // A run's waitpoints can be scattered across both stores (drain in flight), so count on
-  // each and sum rather than assume one home.
-  async countPendingWaitpoints(waitpointIds: string[], client?: ReadClient): Promise<number> {
+  // A run's blocking waitpoints mostly co-locate with the run; only a cross-tree token (a standalone
+  // MANUAL token, or waitForRun across trees) lives on the other DB. When the run id is threaded
+  // through, route to the run's store and fall back to the other DB for ONLY the ids absent there —
+  // partitioning by found-ness so a present-but-completed id is trusted from the run's store and a
+  // cross-tree pending token is still counted (never undercounted, which would prematurely unblock).
+  // Absent a run id, count on each and sum (a caller with no run id in scope).
+  async countPendingWaitpoints(
+    waitpointIds: string[],
+    client?: ReadClient,
+    runId?: string
+  ): Promise<number> {
+    if (runId === undefined) {
+      const [fromNew, fromLegacy] = await Promise.all([
+        this.#new.countPendingWaitpoints(
+          waitpointIds,
+          RoutingRunStore.#ownPrimary(this.#new, client)
+        ),
+        this.#legacy.countPendingWaitpoints(
+          waitpointIds,
+          RoutingRunStore.#ownPrimary(this.#legacy, client)
+        ),
+      ]);
+      return fromNew + fromLegacy;
+    }
+
+    if (waitpointIds.length === 0) {
+      return 0;
+    }
+    const runStore = this.#routeOrNew(runId);
+    const otherStore = runStore === this.#new ? this.#legacy : this.#new;
+    const { pendingIds, presentIds } = await runStore.countPendingWaitpointsWithPresence(
+      waitpointIds,
+      RoutingRunStore.#ownPrimary(runStore, client)
+    );
+    const present = new Set(presentIds);
+    const missing = waitpointIds.filter((id) => !present.has(id));
+    if (missing.length === 0) {
+      return pendingIds.length;
+    }
+    const otherPending = await otherStore.countPendingWaitpoints(
+      missing,
+      RoutingRunStore.#ownPrimary(otherStore, client)
+    );
+    return pendingIds.length + otherPending;
+  }
+
+  // Fan out and union: an id lives on exactly one store in steady state (a drain-mirror can put it on
+  // both), so the union of pending/present ids dedups a mirror correctly. Not on any hot path — the
+  // router's countPendingWaitpoints routes to a sub-store's variant directly — but required by the
+  // interface and correct for any defensive caller.
+  async countPendingWaitpointsWithPresence(
+    waitpointIds: string[],
+    client?: ReadClient
+  ): Promise<{ pendingIds: string[]; presentIds: string[] }> {
     const [fromNew, fromLegacy] = await Promise.all([
-      this.#new.countPendingWaitpoints(
+      this.#new.countPendingWaitpointsWithPresence(
         waitpointIds,
         RoutingRunStore.#ownPrimary(this.#new, client)
       ),
-      this.#legacy.countPendingWaitpoints(
+      this.#legacy.countPendingWaitpointsWithPresence(
         waitpointIds,
         RoutingRunStore.#ownPrimary(this.#legacy, client)
       ),
     ]);
-    return fromNew + fromLegacy;
+    return {
+      pendingIds: uniqueStrings([...fromNew.pendingIds, ...fromLegacy.pendingIds]),
+      presentIds: uniqueStrings([...fromNew.presentIds, ...fromLegacy.presentIds]),
+    };
   }
 
   // A waitpoint co-locates with the OWNER it points at, in priority order: an explicit
@@ -1177,41 +1248,83 @@ export class RoutingRunStore implements RunStore {
 
   async findManyWaitpoints<T extends Prisma.WaitpointFindManyArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointFindManyArgs>,
-    client?: ReadClient
+    client?: ReadClient,
+    runId?: string
   ): Promise<Prisma.WaitpointGetPayload<T>[]> {
     const { scalarArgs, relations } = splitWaitpointRelationProjection(
       args as Record<string, unknown>
     );
+    const rows = (await this.#collectManyWaitpoints(
+      scalarArgs,
+      client,
+      runId
+    )) as Prisma.WaitpointGetPayload<T>[];
+    for (const row of rows) {
+      await this.#reresolveWaitpointRelationsCrossDb(
+        row as Record<string, unknown>,
+        relations,
+        client
+      );
+    }
+    return rows;
+  }
+
+  // Collect the scalar waitpoint rows (relation re-resolution happens in the caller). With a run id in
+  // scope and a bounded id set to partition on, route to the run's store and fall back to the other DB
+  // for ONLY the ids missing there (a rare cross-tree token) — the two legs are disjoint by
+  // construction, so no dedup is needed. Otherwise fan out to BOTH and dedup by id NEW-wins.
+  async #collectManyWaitpoints(
+    scalarArgs: Record<string, unknown>,
+    client: ReadClient | undefined,
+    runId: string | undefined
+  ): Promise<Record<string, unknown>[]> {
+    if (runId !== undefined) {
+      const requestedIds = idListFromWhere((scalarArgs.where ?? {}) as Prisma.TaskRunWhereInput);
+      if (requestedIds !== undefined) {
+        const runStore = this.#routeOrNew(runId);
+        const fromRun = (await runStore.findManyWaitpoints(
+          scalarArgs as Prisma.WaitpointFindManyArgs,
+          RoutingRunStore.#ownPrimary(runStore, client)
+        )) as Record<string, unknown>[];
+        const foundIds = new Set(
+          fromRun.map((w) => w.id).filter((id): id is string => typeof id === "string")
+        );
+        const missing = requestedIds.filter((id) => !foundIds.has(id));
+        if (missing.length === 0) {
+          return fromRun;
+        }
+        const otherStore = runStore === this.#new ? this.#legacy : this.#new;
+        const fromOther = (await otherStore.findManyWaitpoints(
+          narrowArgsToIds(scalarArgs, missing) as Prisma.WaitpointFindManyArgs,
+          RoutingRunStore.#ownPrimary(otherStore, client)
+        )) as Record<string, unknown>[];
+        return [...fromRun, ...fromOther];
+      }
+      // No bounded id set to partition on → fall through to the fan-out path.
+    }
+
     const [fromNew, fromLegacy] = await Promise.all([
       this.#new.findManyWaitpoints(
-        scalarArgs as typeof args,
+        scalarArgs as Prisma.WaitpointFindManyArgs,
         RoutingRunStore.#ownPrimary(this.#new, client)
-      ),
+      ) as Promise<Record<string, unknown>[]>,
       this.#legacy.findManyWaitpoints(
-        scalarArgs as typeof args,
+        scalarArgs as Prisma.WaitpointFindManyArgs,
         RoutingRunStore.#ownPrimary(this.#legacy, client)
-      ),
+      ) as Promise<Record<string, unknown>[]>,
     ]);
     // A token mirrored onto both DBs during drain appears in BOTH legs; dedup by id with NEW-wins
     // (the NEW copy is authoritative once a run migrates), matching the router's NEW-wins invariant
     // (#findRunsOpen). Without this, edge-waitpoint hydration could read a stale LEGACY status and
     // strand the run. Rows whose projection omits `id` can't be deduped and pass through.
-    const byId = new Map<string, Prisma.WaitpointGetPayload<T>>();
-    const passthrough: Prisma.WaitpointGetPayload<T>[] = [];
+    const byId = new Map<string, Record<string, unknown>>();
+    const passthrough: Record<string, unknown>[] = [];
     for (const w of [...fromLegacy, ...fromNew]) {
-      const id = (w as { id?: unknown }).id;
+      const id = w.id;
       if (typeof id === "string") byId.set(id, w);
       else passthrough.push(w);
     }
-    const rows = [...byId.values(), ...passthrough];
-    for (const row of rows) {
-      await this.#reresolveWaitpointRelationsCrossDb(
-        row as Record<string, unknown>,
-        relations,
-        client
-      );
-    }
-    return rows;
+    return [...byId.values(), ...passthrough];
   }
 
   // Re-resolve a waitpoint's group-A relations across BOTH DBs and attach them to `row`. Each target
@@ -1391,17 +1504,32 @@ export class RoutingRunStore implements RunStore {
       args as Record<string, unknown>
     );
 
-    const [fromNew, fromLegacy] = await Promise.all([
-      this.#new.findManyTaskRunWaitpoints(
-        scalarArgs as typeof args,
-        RoutingRunStore.#ownPrimary(this.#new, client)
-      ),
-      this.#legacy.findManyTaskRunWaitpoints(
+    // An edge always co-locates with its RUN (blockRunWithWaitpointEdges routes the write by runId),
+    // so a read keyed by a classifiable `taskRunId` routes to that run's store — no fan-out, no dedup.
+    // Only a `waitpointId`/`batchId` predicate (no run id) still fans across both stores.
+    const taskRunId = whereFieldString(
+      (args.where as { taskRunId?: Prisma.TaskRunWhereInput["id"] } | undefined)?.taskRunId
+    );
+    let edges: Record<string, unknown>[];
+    if (taskRunId !== undefined) {
+      const store = this.#routeOrNew(taskRunId);
+      edges = (await store.findManyTaskRunWaitpoints(
         scalarArgs as typeof args,
-        RoutingRunStore.#ownPrimary(this.#legacy, client)
-      ),
-    ]);
-    const edges = dedupeEdgesById([...fromNew, ...fromLegacy]) as Record<string, unknown>[];
+        RoutingRunStore.#ownPrimary(store, client)
+      )) as Record<string, unknown>[];
+    } else {
+      const [fromNew, fromLegacy] = await Promise.all([
+        this.#new.findManyTaskRunWaitpoints(
+          scalarArgs as typeof args,
+          RoutingRunStore.#ownPrimary(this.#new, client)
+        ),
+        this.#legacy.findManyTaskRunWaitpoints(
+          scalarArgs as typeof args,
+          RoutingRunStore.#ownPrimary(this.#legacy, client)
+        ),
+      ]);
+      edges = dedupeEdgesById([...fromNew, ...fromLegacy]) as Record<string, unknown>[];
+    }
 
     if (waitpoint) {
       await this.#hydrateEdgeWaitpointsCrossDb(edges, waitpoint, client);
@@ -1470,9 +1598,15 @@ export class RoutingRunStore implements RunStore {
     args: Prisma.TaskRunWaitpointDeleteManyArgs,
     tx?: PrismaClientOrTransaction
   ): Promise<Prisma.BatchPayload> {
-    // Edges co-locate with their RUN, not their waitpoint, so a waitpointId/taskRunId predicate may
-    // match edges on either store; delete from both so no cross-DB edge keeps a run blocked. The
-    // caller's `tx` is never forwarded — each leg deletes on its own store's client.
+    // An edge always co-locates with its RUN, so a delete keyed by a classifiable `taskRunId` routes
+    // to that run's store — no fan-out. Only a `waitpointId`-keyed delete (no run id) still deletes
+    // from both stores. The caller's `tx` is never forwarded — each leg deletes on its own client.
+    const taskRunId = whereFieldString(
+      (args.where as { taskRunId?: Prisma.TaskRunWhereInput["id"] } | undefined)?.taskRunId
+    );
+    if (taskRunId !== undefined) {
+      return (await this.#routeOrNewForWrite(taskRunId)).deleteManyTaskRunWaitpoints(args);
+    }
     const [fromNew, fromLegacy] = await Promise.all([
       this.#new.deleteManyTaskRunWaitpoints(args),
       this.#legacy.deleteManyTaskRunWaitpoints(args),
@@ -1872,6 +2006,15 @@ function narrowToIds(args: FindRunsArgs, ids: string[]): FindRunsArgs {
   return { ...args, where: { ...args.where, id: { in: ids } } };
 }
 
+// Clone find-many args, replacing the `id` filter with `{ in: ids }` while keeping any other `where`
+// conditions and the projection/ordering intact. Used to re-query only the ids missing on the first leg.
+function narrowArgsToIds(args: Record<string, unknown>, ids: string[]): Record<string, unknown> {
+  return {
+    ...args,
+    where: { ...((args.where as Record<string, unknown>) ?? {}), id: { in: ids } },
+  };
+}
+
 // Merge edge rows from both stores, keeping one per edge `id` (NEW seen last wins). Rows whose
 // projection omits `id` can't be deduped, so they pass through unchanged.
 function dedupeEdgesById<R>(rows: R[]): R[] {
diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts
index 399aab5b07..f765ef4d5b 100644
--- a/internal-packages/run-store/src/types.ts
+++ b/internal-packages/run-store/src/types.ts
@@ -674,12 +674,19 @@ export interface RunStore {
   ): Promise<Prisma.TaskRunExecutionSnapshotGetPayload<{ include: { checkpoint: true } }>>;
 
   // Implicit-join group
-  findSnapshotCompletedWaitpointIds(snapshotId: string, client?: ReadClient): Promise<string[]>;
+  /** `runId` (when known) routes to the run's store — the snapshot + its join co-locate with the run;
+   * omit it and the router fans out (the cuid snapshot id alone can't say which store holds the join). */
+  findSnapshotCompletedWaitpointIds(
+    snapshotId: string,
+    client?: ReadClient,
+    runId?: string
+  ): Promise<string[]>;
   /** As above, but reports in the SAME read whether the snapshot is visible on the reader: `present=false`
    * means this reader lacks the snapshot, so its empty id list is not authoritative (repair from primary). */
   findSnapshotCompletedWaitpointIdsWithPresence(
     snapshotId: string,
-    client?: ReadClient
+    client?: ReadClient,
+    runId?: string
   ): Promise<{ present: boolean; ids: string[] }>;
   /** Run ids connected to a waitpoint (WaitpointRunConnection / `_WaitpointRunConnections`), this DB only. */
   findWaitpointConnectedRunIds(waitpointId: string, client?: ReadClient): Promise<string[]>;
@@ -694,7 +701,20 @@ export interface RunStore {
     batchIndex?: number;
     tx?: PrismaClientOrTransaction;
   }): Promise<void>;
-  countPendingWaitpoints(waitpointIds: string[], client?: ReadClient): Promise<number>;
+  /** `runId` (when known) routes to the run's store and falls back to the other DB only for ids absent
+   * there (a rare cross-tree token), instead of fanning the count out to both DBs on every call. */
+  countPendingWaitpoints(
+    waitpointIds: string[],
+    client?: ReadClient,
+    runId?: string
+  ): Promise<number>;
+  /** Which of the given ids are PENDING and which exist on this store (any status), so the router can
+   * route by run id and only re-count the ids absent here on the other DB — without undercounting
+   * pending (which would prematurely unblock a run) or double-counting a drain-mirrored id. */
+  countPendingWaitpointsWithPresence(
+    waitpointIds: string[],
+    client?: ReadClient
+  ): Promise<{ pendingIds: string[]; presentIds: string[] }>;
 
   // Waitpoint group
   createWaitpoint<T extends Prisma.WaitpointCreateArgs>(
@@ -717,9 +737,12 @@ export interface RunStore {
   findWaitpointOnPrimary<T extends Prisma.WaitpointFindFirstArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointFindFirstArgs>
   ): Promise<Prisma.WaitpointGetPayload<T> | null>;
+  /** `runId` (when known) routes to the run's store and falls back to the other DB only for the ids
+   * missing there, instead of fanning every token read out to both DBs. */
   findManyWaitpoints<T extends Prisma.WaitpointFindManyArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointFindManyArgs>,
-    client?: ReadClient
+    client?: ReadClient,
+    runId?: string
   ): Promise<Prisma.WaitpointGetPayload<T>[]>;
   updateWaitpoint<T extends Prisma.WaitpointUpdateArgs>(
     args: Prisma.SelectSubset<T, Prisma.WaitpointUpdateArgs>,