feat(webapp,run-store): route run-graph reads and writes through the run-store router (#4237)

## Summary

Run-graph data (runs, batches, waitpoints, and their related tables) can
now live in a database separate from the control plane, with every read
and write routed to the correct database by each run's residency. This
makes reading and writing run data more reliable once the two are split,
and is a no-op for single-database installs.

## Design

- Run-graph table access goes through the run-store router, which
selects the legacy or the new run-ops store per run instead of assuming
one shared client.
- The legacy run-ops client is now independently pointable, so legacy
run data can be served from its own database (and replica) rather than
the control-plane connection.
- Run-graph writes go straight to the run-graph database instead of
being forwarded through the control plane, and replication targets are
split so runs in the new database still replicate to analytics without
under-counting.
- Read-through slots refuse the control-plane client, so a missing
residency fails loudly instead of silently reading the wrong database.
- Migration `20260710120000_drop_remaining_run_graph_seam_foreign_keys`
drops the foreign keys that still crossed the run-graph / control-plane
seam, which is what lets the two live in separate databases.

The split stays off unless explicitly enabled and the two databases are
confirmed physically distinct; startup fails closed otherwise.

Verified by running the full dashboard end-to-end suite against both a
single-database configuration and a three-database configuration
(control plane, the new database, and a physically separate legacy
database), with runs on both residencies. No misrouted reads in either
configuration.

Author: d-cs

.server-changes/runops-run-graph-client-routing.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Improved the reliability of how run data is read and written.

apps/webapp/app/db.server.ts

@@ -25,6 +25,7 @@ import {
   assertSplitRealtimeInterlock,
 } from "./v3/runOpsMigration/splitMode.server";
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
+import { assertControlPlaneCoresidencyAdvisory } from "./v3/runOpsMigration/controlPlaneCoresidencySentinel.server";
 import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server";
 import type { Span } from "@opentelemetry/api";
 import { context, trace } from "@opentelemetry/api";
@@ -188,13 +189,18 @@ export type RunOpsTopology = {
 export type SelectRunOpsTopologyConfig = {
   splitEnabled: boolean;
   legacyUrl?: string;
+  legacyReplicaUrl?: string;
   newUrl?: string;
   newReplicaUrl?: string;
 };
 export type RunOpsClientBuilders = {
   controlPlane: RunOpsClients;
   buildNewWriter: (url: string, clientType: string) => RunOpsPrismaClient;
   buildNewReplica: (url: string, clientType: string) => RunOpsPrismaClient;
+  // Legacy builders return the same PrismaClient/PrismaReplicaClient types as the control plane (no
+  // RunOpsPrismaClient double-cast needed): the legacy DB carries the full control-plane schema.
+  buildLegacyWriter: (url: string, clientType: string) => PrismaClient;
+  buildLegacyReplica: (url: string, clientType: string) => PrismaReplicaClient;
 };
 
 // Pure run-ops client selector. No env, no isSplitEnabled() — those
@@ -220,7 +226,15 @@ export function selectRunOpsTopology(
     return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane };
   }
 
-  const legacyRunOps = controlPlane;
+  // Track 2: build an INDEPENDENT legacy client from its own DSN instead of aliasing the control
+  // plane. legacyUrl is guaranteed present (the missing-URL branch above aliases and returns).
+  const legacyWriter = builders.buildLegacyWriter(config.legacyUrl, "run-ops-legacy-writer");
+  // Mirror the NEW replica + control-plane $replica fallback: brand a real replica (in the builder),
+  // otherwise reuse the legacy WRITER so replica reads fall back to the legacy primary — unbranded.
+  const legacyReplica: PrismaReplicaClient = config.legacyReplicaUrl
+    ? builders.buildLegacyReplica(config.legacyReplicaUrl, "run-ops-legacy-reader")
+    : legacyWriter;
+  const legacyRunOps: RunOpsClients = { writer: legacyWriter, replica: legacyReplica };
 
   const newWriter = builders.buildNewWriter(config.newUrl, "run-ops-new-writer");
   const newReplica: RunOpsPrismaClient = config.newReplicaUrl
@@ -246,10 +260,19 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
   // Gate on the opt-in flag too: the distinct-DB sentinel only runs when the flag is on.
   const splitEnabled = env.RUN_OPS_SPLIT_ENABLED && !!newUrl && !!env.RUN_OPS_LEGACY_DATABASE_URL;
 
+  // Without a dedicated legacy replica URL, legacy reads fall back to the legacy WRITER (primary).
+  // Surface that so a prod misdeploy is observable instead of a silent load shift onto the primary.
+  if (splitEnabled && !env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL) {
+    logger.warn(
+      "RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL is unset while split is enabled; legacy reads will hit the legacy primary"
+    );
+  }
+
   return selectRunOpsTopology(
     {
       splitEnabled,
       legacyUrl: env.RUN_OPS_LEGACY_DATABASE_URL,
+      legacyReplicaUrl: env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL,
       newUrl,
       newReplicaUrl: env.RUN_OPS_DATABASE_READ_REPLICA_URL,
     },
@@ -268,6 +291,18 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
             tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType }))
           )
         ),
+      // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full
+      // control-plane schema); markReadReplicaClient only on a real replica URL, as with the NEW replica.
+      buildLegacyWriter: (url, clientType) =>
+        captureInfrastructureErrors(
+          tagDatasource("writer", buildWriterClient({ url, clientType }))
+        ),
+      buildLegacyReplica: (url, clientType) =>
+        markReadReplicaClient(
+          captureInfrastructureErrors(
+            tagDatasource("replica", buildReplicaClient({ url, clientType }))
+          )
+        ),
     }
   );
 });
@@ -281,8 +316,17 @@ export const runOpsNewPrisma: PrismaClient = runOpsTopology.newRunOps
   .writer as unknown as PrismaClient;
 export const runOpsNewReplica: PrismaReplicaClient = runOpsTopology.newRunOps
   .replica as unknown as PrismaReplicaClient;
+// Track 2: under split-on these point at the INDEPENDENT legacy client (its own DSN); under split-off
+// or missing URLs they still alias the control-plane client, so single-DB installs are unchanged.
 export const runOpsLegacyPrisma: PrismaClient = runOpsTopology.legacyRunOps.writer;
 export const runOpsLegacyReplica: PrismaReplicaClient = runOpsTopology.legacyRunOps.replica;
+// Branded legacy handles typed as RunOpsPrismaClient for the run-store boundary — same underlying
+// legacy writer/replica as runOpsLegacyPrisma/runOpsLegacyReplica above, but carrying the run-ops
+// brand so the guard classifies provably-legacy access as `runops`, not `cp`.
+export const runOpsLegacyPrismaClient: RunOpsPrismaClient = runOpsTopology.legacyRunOps
+  .writer as unknown as RunOpsPrismaClient;
+export const runOpsLegacyReplicaClient: RunOpsPrismaClient = runOpsTopology.legacyRunOps
+  .replica as unknown as RunOpsPrismaClient;
 
 export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   newReplica: runOpsNewReplicaClient,
@@ -295,8 +339,8 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
 
 // Boot-time interlock: if the flag is on but the distinct-DB sentinel does not
 // confirm two physically-distinct run-ops DBs, refuse to enable split (data-loss
-// interlock). Async, so it cannot live in the synchronous singleton factory —
-// call it from the eager-boot path before any run-ops routing is wired.
+// interlock). Async, so it cannot live in the synchronous singleton factory — called
+// fire-and-forget from the eager-boot path (routing is wired synchronously at module load).
 export async function assertRunOpsSplitSentinel(): Promise<void> {
   if (!env.RUN_OPS_SPLIT_ENABLED) return;
   // Realtime interlock (synchronous): Electric replicates only from the control-plane
@@ -312,6 +356,9 @@ export async function assertRunOpsSplitSentinel(): Promise<void> {
       "RUN_OPS_SPLIT_ENABLED is on but the distinct-DB sentinel did not confirm two physically-distinct run-ops DBs; refusing to enable split (data-loss interlock)."
     );
   }
+  // Advisory-only (T2.3): observe legacy vs control-plane co-residency. Emits a metric + log and only
+  // throws when RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on AND co-residency is positively confirmed.
+  await assertControlPlaneCoresidencyAdvisory();
 }
 
 function getClient() {

apps/webapp/app/env.server.ts

@@ -138,8 +138,10 @@ const EnvironmentSchema = z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_URL is invalid")
       .optional(),
-    // The LEGACY run-ops DB (the control-plane DB during the transition). When unset, legacy
-    // run-ops reuses the existing DATABASE_URL (legacy run-ops == control-plane DB initially).
+    // The LEGACY run-ops DB. Now a CONNECTED DSN (Track 2): when split is on and this is set it builds
+    // an INDEPENDENT legacy Prisma client, no longer an alias of the control-plane client (nor merely
+    // the sentinel's probe target). Unset -> legacy reuses the control-plane client / DATABASE_URL, so
+    // single-DB and self-host installs boot byte-identical.
     RUN_OPS_LEGACY_DATABASE_URL: z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_URL is invalid")
@@ -151,6 +153,22 @@ const EnvironmentSchema = z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_READ_REPLICA_URL is invalid")
       .optional(),
+    // The LEGACY run-ops DB read replica (Track 2). Unset -> the legacy replica handle falls back to the
+    // legacy WRITER (as $replica does with no CP replica). Set in production so legacy reads hit the reader.
+    RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL is invalid")
+      .optional(),
+    // Direct DSN for applying the full @trigger.dev/database migrations to the LEGACY run-ops DB, keeping
+    // its schema current after the control plane moves off it. Direct, not pooled — migrations never run
+    // over a pooler. Optional; unset -> the entrypoint's legacy migrate step is skipped.
+    RUN_OPS_LEGACY_DIRECT_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DIRECT_URL is invalid")
+      .optional(),
+    // Advisory control-plane co-residency sentinel enforcement (Track 2, T2.3). Default OFF; the advisory
+    // arm always emits its metric, this only turns a still-co-resident pair into a hard boot failure.
+    RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT: BoolEnv.default(false),
     // --- Control-plane datasource repoint. Additive-only. ---
     // Optional control-plane DB. Unset (self-host/single-DB) -> getClient()/getReplicaClient() fall back to
     // DATABASE_URL/DATABASE_READ_REPLICA_URL, so boot is byte-identical. When set, these point at the
@@ -1618,6 +1636,14 @@ const EnvironmentSchema = z
     RUN_REPLICATION_DISABLE_PAYLOAD_INSERT: z.string().default("0"),
     RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING: z.string().default("0"),
 
+    // Connection URL for the LEGACY runs-replication source (the runs-CDC slot on the legacy runs DB, plus
+    // the admin recovery route). Direct, not pooled: replication can't run over a pooler. Optional; unset ->
+    // falls back to DATABASE_URL, so nothing changes today.
+    RUN_REPLICATION_LEGACY_DATABASE_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_REPLICATION_LEGACY_DATABASE_URL is invalid")
+      .optional(),
+
     // --- Run-ops DB split — second replication source (the NEW dedicated run-ops DB). ---
     // Cloud-only; only consulted when isSplitEnabled() is true. Self-host never sets these.
     // Connection URL for the run-ops DB used by the runs-replication source. Required when the split is
@@ -1658,6 +1684,12 @@ const EnvironmentSchema = z
     SESSION_REPLICATION_PUBLICATION_NAME: z
       .string()
       .default("sessions_to_clickhouse_v1_publication"),
+    // Connection URL for the sessions-replication slot. Direct, not pooled: replication can't run over a
+    // pooler. Optional; unset -> falls back to DATABASE_URL, so nothing changes today.
+    SESSION_REPLICATION_DATABASE_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "SESSION_REPLICATION_DATABASE_URL is invalid")
+      .optional(),
     SESSION_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(1),
     SESSION_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     SESSION_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),

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

@@ -1,5 +1,5 @@
 import { Prisma } from "@trigger.dev/database";
-import { prisma } from "~/db.server";
+import { runStore } from "~/v3/runStore.server";
 
 export const MAX_TAGS_PER_WAITPOINT = 10;
 const MAX_RETRIES = 3;
@@ -19,19 +19,10 @@ export async function createWaitpointTag({
 
   while (attempts < MAX_RETRIES) {
     try {
-      return await prisma.waitpointTag.upsert({
-        where: {
-          environmentId_name: {
-            environmentId,
-            name: tag,
-          },
-        },
-        create: {
-          name: tag,
-          environmentId,
-          projectId,
-        },
-        update: {},
+      return await runStore.upsertWaitpointTag({
+        environmentId,
+        name: tag,
+        projectId,
       });
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {

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

@@ -2,13 +2,14 @@ import type { TaskRunExecutionResult } from "@trigger.dev/core/v3";
 import type { PrismaClientOrTransaction, PrismaReplicaClient } from "~/db.server";
 import { executionResultForTaskRun } from "~/models/taskRun.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
+import { runStore as defaultRunStore } from "~/v3/runStore.server";
 import { BasePresenter } from "./basePresenter.server";
 
 type ApiRunResultReadThroughDeps = {
   splitEnabled?: boolean;
   newClient?: PrismaReplicaClient;
-  // LEGACY RUN-OPS READ REPLICA ONLY (never a writer/primary); defaults to this._replica.
+  // LEGACY RUN-OPS READ REPLICA ONLY (never a writer/primary); defaults to runOpsLegacyReplica
+  // (the Aurora legacy read replica), never the control-plane replica.
   legacyReplica?: PrismaReplicaClient;
   isPastRetention?: (runId: string) => boolean;
 };
@@ -17,7 +18,8 @@ export class ApiRunResultPresenter extends BasePresenter {
   constructor(
     prisma?: PrismaClientOrTransaction,
     replica?: PrismaClientOrTransaction,
-    private readonly _readThrough?: ApiRunResultReadThroughDeps
+    private readonly _readThrough?: ApiRunResultReadThroughDeps,
+    private readonly runStore = defaultRunStore
   ) {
     super(prisma, replica);
   }
@@ -27,31 +29,14 @@ export class ApiRunResultPresenter extends BasePresenter {
     env: AuthenticatedEnvironment
   ): Promise<TaskRunExecutionResult | undefined> {
     return this.traceWithEnv("call", env, async (span) => {
-      const findRun = (client: PrismaReplicaClient) =>
-        client.taskRun.findFirst({
-          where: { friendlyId, runtimeEnvironmentId: env.id },
-          include: { attempts: { orderBy: { createdAt: "desc" } } },
-        });
-
-      // Single-run result poll routed through run-ops read-through. Split on: primary store first,
-      // then the secondary read replica for runs that miss on new; past-retention ids return
-      // undefined -> the route's normal 404. Split off (single-DB / self-host): readThroughRun does
-      // one plain findFirst against the single client (passthrough).
-      const result = await readThroughRun({
-        runId: friendlyId,
-        environmentId: env.id,
-        readNew: findRun,
-        readLegacy: findRun,
-        deps: {
-          splitEnabled: this._readThrough?.splitEnabled,
-          newClient: this._readThrough?.newClient ?? (this._prisma as PrismaReplicaClient),
-          legacyReplica: this._readThrough?.legacyReplica ?? (this._replica as PrismaReplicaClient),
-          isPastRetention: this._readThrough?.isPastRetention,
-        },
-      });
-
-      const taskRun =
-        result.source === "new" || result.source === "legacy-replica" ? result.value : undefined;
+      // Single-run result poll routed through the run store, which selects the owning DB by
+      // run-id residency (id shape): a run-ops (NEW) id reads the new store, a cuid (LEGACY) id
+      // reads the legacy store. Single-DB / self-host collapses to one plain findFirst against the
+      // one store (passthrough). The identical TaskRun(+attempts) lookup runs inside the router.
+      const taskRun = await this.runStore.findRun(
+        { friendlyId, runtimeEnvironmentId: env.id },
+        { include: { attempts: { orderBy: { createdAt: "desc" } } } }
+      );
 
       if (!taskRun) {
         return undefined;

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

@@ -5,11 +5,11 @@ import { BasePresenter } from "./basePresenter.server";
 import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
 import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import type { PrismaClientOrTransaction, PrismaReplicaClient } from "~/db.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
+import { runStore as defaultRunStore } from "~/v3/runStore.server";
 
-// When omitted, clients default to the inherited _replica handle => passthrough reads the
-// replica exactly as today. isPastRetention is injectable for tests. Typed PrismaReplicaClient
-// to match readThroughRun's readNew/readLegacy + deps.
+// Retained only to preserve the public constructor signature the route passes. Run-ops routing
+// (NEW vs LEGACY residency, replica reads) is now handled inside the injected `runStore`, so
+// these deps are no longer consulted for the read.
 type ApiWaitpointPresenterReadThroughDeps = {
   newClient?: PrismaReplicaClient;
   legacyReplica?: PrismaReplicaClient;
@@ -21,7 +21,8 @@ export class ApiWaitpointPresenter extends BasePresenter {
   constructor(
     prismaClient?: PrismaClientOrTransaction,
     replicaClient?: PrismaClientOrTransaction,
-    private readonly readThroughDeps?: ApiWaitpointPresenterReadThroughDeps
+    private readonly readThroughDeps?: ApiWaitpointPresenterReadThroughDeps,
+    private readonly runStore = defaultRunStore
   ) {
     super(prismaClient, replicaClient);
   }
@@ -39,56 +40,32 @@ export class ApiWaitpointPresenter extends BasePresenter {
     waitpointId: string
   ) {
     return this.trace("call", async (span) => {
-      // Public waitpoint retrieve. Split on: new run-ops client first, then the LEGACY
-      // RUN-OPS READ REPLICA ONLY on a new-probe miss — never the legacy primary.
-      // Split off (single-DB / self-host): one plain waitpoint.findFirst against the replica
-      // (passthrough). The waitpointId is the residency-classifiable run-ops id (the route
-      // pre-decodes the friendlyId via WaitpointId.toId).
-      const hydrate = (client: PrismaReplicaClient) =>
-        client.waitpoint.findFirst({
-          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 result = await readThroughRun({
-        runId: waitpointId,
-        environmentId: environment.id,
-        readNew: (client) => hydrate(client),
-        readLegacy: (replica) => hydrate(replica),
-        deps: {
-          splitEnabled: this.readThroughDeps?.splitEnabled,
-          // Default both clients to the inherited _replica handle (declared
-          // PrismaClientOrTransaction but $replica at runtime) so passthrough reads the replica
-          // as today; split mode injects a distinct newClient.
-          newClient: this.readThroughDeps?.newClient ?? (this._replica as PrismaReplicaClient),
-          legacyReplica:
-            this.readThroughDeps?.legacyReplica ?? (this._replica as PrismaReplicaClient),
-          isPastRetention: this.readThroughDeps?.isPastRetention,
+      // 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 waitpoint =
-        result.source === "new" || result.source === "legacy-replica" ? result.value : null;
-
       if (!waitpoint) {
         logger.error(`WaitpointPresenter: Waitpoint not found`, {
           id: waitpointId,