feat(run-ops): run-store routing seam + run-engine read seams (#4116)

## What

Introduces the run-store routing seam and the run-engine read seams that
let run lifecycle operations be dispatched to either the control-plane
database or a separately-generated run-ops database, depending on where
a run/batch resides.

- **run-store** (`internal-packages/run-store`): adds `runOpsStore.ts`
and substantially expands `PostgresRunStore.ts` so the store can resolve
residency and route reads/writes to the correct backing client.
`types.ts` grows the routing/residency types; `NoopRunStore.ts` is
removed.
- **run-engine** (`internal-packages/run-engine`): adds
`engine/controlPlaneResolver.ts` and routes the per-system read paths
(dequeue, enqueue, waitpoint, checkpoint, run-attempt, ttl, delayed-run,
execution-snapshot, pending-version, debounce, batch) through the
resolver/store instead of talking to a single Prisma client directly.
`engine/errors.ts`, `engine/types.ts`, and `engine/index.ts` are
extended to support injecting the store/resolver.

Three fixes are included on top of the seam work:

- `c6cadd85f` — routes read-your-writes to the owning store's
**writer**, not its lagging replica, so an operation immediately reading
back what it just wrote sees a consistent result.
- `05c912e05` — normalizes run-ops-generation Prisma errors to the
control-plane error class at the store **write boundary**, so
`instanceof` checks and the `P2002` → 422 handling continue to work
across the separately-generated run-ops Prisma client.
- `88d12907f` — resolves NEW-resident batches in
`ApiBatchResultsPresenter` by routing the batch read through the store,
so a dedicated-DB batch resolves instead of returning 404.

The change is heavily test-first: the bulk of the diff is new
unit/integration coverage for the store routing, residency, and each
run-engine system's control-plane resolver path.

## Why

PR4 of the run-ops split stack (PR1–PR3 land the ClickHouse
test-container and earlier plumbing). This PR is the read-path
foundation: it adds the seam and read-routing but leaves the write path
to route through the same seam in a later PR. Behavior-changing where
the three fixes above touch existing read-your-writes /
error-normalization / batch-resolution paths; otherwise additive (new
store module, new resolver, injectable dependencies with existing
single-client behavior preserved when no dedicated store is configured).

## Tests

Extensive new vitest coverage under `run-store/src/*.test.ts` (routing,
residency, dual-schema select, cross-generation error normalization,
read-after-write, idempotency dedup, mixed residency, waitpoint
co-location) and `run-engine/src/engine/**/*.test.ts` (per-system
`controlPlaneResolver` tests, injectability, block-edge residency,
waitpoint read residency, trigger-create routing, lifecycle router).
Testcontainers-backed; no mocks.

## Notes

Draft, **stacked on #4114** (`runops/pr03-clickhouse-tc`). Review that
first; this diff is against it.

Server-change / changeset note to be added at stack-assembly time.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

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

Author: d-cs

.server-changes/run-engine-store-routing-seam.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Route run engine lifecycle operations through the run store and a control-plane resolver so run data can live on a dedicated backing store separate from the control plane.

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

@@ -11,19 +11,22 @@ export class ApiBatchResultsPresenter extends BasePresenter {
     env: AuthenticatedEnvironment
   ): Promise<BatchTaskRunExecutionResult | undefined> {
     return this.traceWithEnv("call", env, async (span) => {
-      const batchRun = await this._prisma.batchTaskRun.findFirst({
-        where: {
-          friendlyId,
-          runtimeEnvironmentId: env.id,
-        },
-        include: {
-          items: {
-            select: {
-              taskRunId: true,
+      // Route through the store so a NEW-resident batch resolves under the run-ops split (the
+      // router probes NEW→LEGACY and drops this client hint) instead of 404ing on a control-plane read.
+      const batchRun = await runStore.findBatchTaskRunByFriendlyId(
+        friendlyId,
+        env.id,
+        {
+          include: {
+            items: {
+              select: {
+                taskRunId: true,
+              },
             },
           },
         },
-      });
+        this._prisma
+      );
 
       if (!batchRun) {
         return undefined;

apps/webapp/test/presenters/ApiBatchResultsPresenter.split.test.ts

@@ -0,0 +1,97 @@
+// Run-ops split resolution LOCK for ApiBatchResultsPresenter.
+//
+// GET /api/v1/batches/:id/results constructs the presenter BARE (no injected client), so it must
+// resolve a batch that lives in the NEW run-ops DB on its own. The presenter routes the batch-row
+// lookup through the `runStore` singleton, whose split router probes NEW→LEGACY. This drives a
+// NEW-resident (ksuid) batch through a REAL two-physical-DB split router and asserts the bare
+// presenter finds it. Fails before the fix (the presenter read the control-plane DB directly and
+// 404'd on a NEW-resident batch).
+
+import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
+import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import type { RunOpsPrismaClient } from "@internal/run-ops-database";
+import type { Organization, PrismaClient, Project } from "@trigger.dev/database";
+import { generateKsuidId } from "@trigger.dev/core/v3/isomorphic";
+import { expect, vi } from "vitest";
+import { ApiBatchResultsPresenter } from "~/presenters/v3/ApiBatchResultsPresenter.server";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+
+// The split router built over the two testcontainer DBs; injected in place of the db.server-backed
+// singleton the presenter imports. Populated per-test before the presenter is constructed.
+let testRunStore: RoutingRunStore;
+
+// Presenter reads the batch row via `runStore`; child-run reads also go through it. Neutralize the
+// real db.server singleton (no env DB) and the runStore singleton (use the split router below).
+// The getter defers to `testRunStore` so each test can set its own router before constructing.
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+vi.mock("~/v3/runStore.server", () => ({
+  get runStore() {
+    return testRunStore;
+  },
+}));
+
+vi.setConfig({ testTimeout: 60_000 });
+
+function makeSplitRouter(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) {
+  const legacyStore = new PostgresRunStore({
+    prisma: prisma14,
+    readOnlyPrisma: prisma14,
+    schemaVariant: "legacy",
+  });
+  const newStore = new PostgresRunStore({
+    prisma: prisma17 as never,
+    readOnlyPrisma: prisma17 as never,
+    schemaVariant: "dedicated",
+  });
+  return new RoutingRunStore({ new: newStore, legacy: legacyStore });
+}
+
+function authEnv(environmentId: string): AuthenticatedEnvironment {
+  return {
+    id: environmentId,
+    type: "DEVELOPMENT",
+    project: { id: "proj_split" } as Project,
+    organization: { id: "org_split" } as Organization,
+    orgMember: null,
+  } as unknown as AuthenticatedEnvironment;
+}
+
+heteroRunOpsPostgresTest(
+  "a bare ApiBatchResultsPresenter resolves a NEW-resident (ksuid) batch under the split",
+  async ({ prisma14, prisma17 }) => {
+    testRunStore = makeSplitRouter(prisma14, prisma17);
+
+    const environmentId = "env_split_res";
+    // ksuid internal id → classifies to the NEW store, seeded in the NEW (prisma17) DB. The
+    // friendlyId probe fans out NEW→LEGACY regardless of id shape, so the NEW seed is what matters.
+    const batchInternalId = generateKsuidId();
+    const batchFriendlyId = `batch_${generateKsuidId()}`;
+
+    await prisma17.batchTaskRun.create({
+      data: {
+        id: batchInternalId,
+        friendlyId: batchFriendlyId,
+        runtimeEnvironmentId: environmentId,
+      },
+    });
+
+    // Bare construction — exactly how the results route builds it.
+    const presenter = new ApiBatchResultsPresenter();
+    const result = await presenter.call(batchFriendlyId, authEnv(environmentId));
+
+    // Before the fix this 404s (undefined) because a control-plane read misses the NEW-resident batch.
+    expect(result).toEqual({ id: batchFriendlyId, items: [] });
+  }
+);
+
+heteroRunOpsPostgresTest(
+  "a bare ApiBatchResultsPresenter still returns undefined for a genuinely missing batch",
+  async ({ prisma14, prisma17 }) => {
+    testRunStore = makeSplitRouter(prisma14, prisma17);
+
+    const presenter = new ApiBatchResultsPresenter();
+    const result = await presenter.call("batch_does_not_exist", authEnv("env_none"));
+
+    expect(result).toBeUndefined();
+  }
+);