fix(webapp,run-ops-database): keep run-ops batch items co-resident with their batch (#4178)

## Summary

Three fixes to the run-ops database split (the Cloud-only mode where
run-lifecycle rows live on a dedicated Postgres). All are inert in the
default single-database deployment.

The main fix: on the batch trigger paths, a parentless batch's item runs
chose their physical store from a fresh per-org mint-flag read at
processing time, so flipping an org's flag mid-batch could land an item
in a different store than its batch, breaking the `TaskRun.batchId`
foreign key (or silently orphaning the item). The other two harden the
split's safety nets: the schema-parity test now actually compares
columns, and the read fan-out gate now signals when it has been silently
disabled.

## Batch item residency

`RunEngineBatchTriggerService` (api.v2) and the BatchQueue item callback
(api.v3) now anchor each item's id mint on the batch's own friendlyId,
mirroring the already-safe `BatchTriggerV3Service`. Residency is a pure
id-shape check, so an item can no longer diverge from its batch across a
mid-batch flag flip. The pre-failed-run fallback is anchored the same
way (it also sets `batchId`), and the shared mint branch is consolidated
into one helper so every mint path stays in lockstep. No new database
queries; single-database mode is unchanged (a cuid-shaped batch
friendlyId yields a cuid item).

## Schema parity test

The parity test previously read only the dedicated schema and matched
model headers with regexes, so it never compared columns and could not
catch a run-subgraph column that diverged between the two physical
schemas. It now parses both schemas and asserts bidirectional
scalar-column parity (type, nullability, array-ness, default) across the
run-subgraph models, and fails on any field line it can't parse. Scoped
to the run-subgraph models so unrelated control-plane edits don't break
it.

## Read fan-out signal

The split read fan-out gate is decided by the object identity of the NEW
vs control-plane clients. It now warns when both run-ops URLs are set
but the NEW client isn't a distinct instance (fan-out silently off), and
a new test exercises the real topology-into-gate wiring so a future
refactor that aliases the clients can't disable fan-out unnoticed.

## Verification

New unit and glue tests cover all three changes; the DB-backed
residency, store-routing, and topology suites pass against real
Postgres; `typecheck` is clean for both packages.

.server-changes/fix-batch-item-mint-anchor.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Anchor batch item run-ops residency on the batch's own friendlyId (not a fresh per-org flag read) in the run-engine batch trigger service and the BatchQueue item callback, on both the success path and the pre-failed-run failure path, so a mid-batch mint-flag flip can no longer mint an item (or a pre-failed item) into a different physical store than its BatchTaskRun row.

apps/webapp/app/db.server.ts

@@ -290,6 +290,7 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   controlPlaneReplica: $replica,
   hasNewUrl: !!env.RUN_OPS_DATABASE_URL,
   hasLegacyUrl: !!env.RUN_OPS_LEGACY_DATABASE_URL,
+  logger,
 });
 
 // Boot-time interlock: if the flag is on but the distinct-DB sentinel does not

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

@@ -18,6 +18,7 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { batchTriggerWorker } from "~/v3/batchTriggerWorker.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { mintAnchoredRunFriendlyId } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
 import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.server";
 import {
   downloadPacketFromObjectStore,
@@ -588,6 +589,9 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
           parentRunId,
           resumeParentOnCompletion,
           batch: { id: batch.id, index: workingIndex },
+          // Anchor the pre-failed run on the BATCH's residency (same as the happy path) so it
+          // co-resides with its BatchTaskRun row regardless of a mid-batch mint-flag flip.
+          runFriendlyId: mintAnchoredRunFriendlyId(batch.friendlyId, item.options?.region),
           options: item.options as Record<string, unknown>,
           traceContext: options?.traceContext as Record<string, unknown> | undefined,
           spanParentAsLink: options?.spanParentAsLink,
@@ -677,6 +681,12 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
 
     const triggerTaskService = new TriggerTaskService();
 
+    // Anchor the item's mint on the BATCH's own friendlyId (not a fresh per-org flag read) so
+    // an org's mint flag flipping between batch creation and this (possibly much later,
+    // worker-processed) item never splits the item from its BatchTaskRun row. See
+    // mintAnchoredRunFriendlyId.server.ts.
+    const runFriendlyId = mintAnchoredRunFriendlyId(batch.friendlyId, item.options?.region);
+
     const result = await triggerTaskService.call(
       item.task,
       environment,
@@ -695,6 +705,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         spanParentAsLink: options?.spanParentAsLink,
         batchId: batch.id,
         batchIndex: currentIndex,
+        runFriendlyId,
         realtimeStreamsVersion: options?.realtimeStreamsVersion,
         triggerSource: options?.triggerSource ?? "api",
         triggerAction: options?.triggerAction ?? "trigger",

apps/webapp/app/runEngine/services/triggerFailedTask.server.test.ts

@@ -0,0 +1,42 @@
+import { describe, expect, it, vi } from "vitest";
+
+// Empty singletons satisfy the module-level wiring imports; the mint method under test is driven
+// directly via (service as any) and never touches the DB (same boundary as triggerTask.server.test.ts).
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+  runOpsNewPrisma: {},
+  runOpsLegacyPrisma: {},
+  runOpsNewReplica: {},
+  runOpsLegacyReplica: {},
+}));
+vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false }));
+
+import { classifyKind, generateRunOpsId, RunId } from "@trigger.dev/core/v3/isomorphic";
+import { TriggerFailedTaskService } from "./triggerFailedTask.server";
+
+function buildService() {
+  return new TriggerFailedTaskService({ prisma: {} as any, engine: {} as any });
+}
+
+describe("TriggerFailedTaskService.mintFailedRunFriendlyId", () => {
+  it("returns the caller-supplied runFriendlyId verbatim (override wins over any mint)", async () => {
+    const override = RunId.toFriendlyId(generateRunOpsId());
+    const minted = await (buildService() as any).mintFailedRunFriendlyId({
+      organizationId: "org_1",
+      environmentId: "env_1",
+      runFriendlyId: override,
+    });
+    expect(minted).toBe(override);
+  });
+
+  it("without an override, still inherits a run-ops (NEW) parent by id-shape", async () => {
+    const parentRunFriendlyId = RunId.toFriendlyId(generateRunOpsId());
+    const minted = await (buildService() as any).mintFailedRunFriendlyId({
+      organizationId: "org_1",
+      environmentId: "env_1",
+      parentRunFriendlyId,
+    });
+    expect(classifyKind(minted)).toBe("runOpsId");
+  });
+});

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

@@ -43,6 +43,9 @@ export type TriggerFailedTaskRequest = {
   spanParentAsLink?: boolean;
 
   errorCode?: TaskRunErrorCodes;
+
+  /** Pre-minted friendlyId; when set it wins over the mint. Batch callers pass a batch-anchored id. */
+  runFriendlyId?: string;
 };
 
 /**
@@ -86,14 +89,20 @@ export class TriggerFailedTaskService {
 
   // Mint a failed run's friendlyId. The id-kind decides which store the run is
   // born in (cuid → legacy store, run-ops id → new store); the whole subgraph of a
-  // run must agree. Root failed runs mint by the environment's setting; child
-  // failed runs inherit the parent's current store so they never split.
+  // run must agree. A caller-supplied runFriendlyId (batch-anchored id) wins verbatim;
+  // otherwise root failed runs mint by the environment's setting and child failed runs
+  // inherit the parent's current store so they never split.
   private async mintFailedRunFriendlyId(args: {
     organizationId: string;
     environmentId: string;
     orgFeatureFlags?: unknown;
     parentRunFriendlyId?: string;
+    runFriendlyId?: string;
   }): Promise<string> {
+    if (args.runFriendlyId) {
+      return args.runFriendlyId;
+    }
+
     const mintKind = args.parentRunFriendlyId
       ? resolveInheritedMintKind(args.parentRunFriendlyId)
       : await resolveRunIdMintKind({
@@ -125,6 +134,7 @@ export class TriggerFailedTaskService {
         environmentId: request.environment.id,
         orgFeatureFlags: request.environment.organization.featureFlags,
         parentRunFriendlyId: request.parentRunId,
+        runFriendlyId: request.runFriendlyId,
       });
       mintedFriendlyId = failedRunFriendlyId;
 
@@ -321,6 +331,8 @@ export class TriggerFailedTaskService {
     resumeParentOnCompletion?: boolean;
     batch?: { id: string; index: number };
     errorCode?: TaskRunErrorCodes;
+    /** Pre-minted friendlyId; when set it wins over the mint. Batch callers pass a batch-anchored id. */
+    runFriendlyId?: string;
   }): Promise<string | null> {
     // Held for the catch's log line; the in-try `const` is what consumers use.
     let mintedFriendlyId: string | undefined;
@@ -335,6 +347,7 @@ export class TriggerFailedTaskService {
         // single replica lookup by organizationId only when there is no parent.
         orgFeatureFlags: undefined,
         parentRunFriendlyId: opts.parentRunId,
+        runFriendlyId: opts.runFriendlyId,
       });
       mintedFriendlyId = failedRunFriendlyId;
 

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

@@ -14,7 +14,6 @@ import {
   TriggerTraceContext,
 } from "@trigger.dev/core/v3";
 import {
-  generateRunOpsId,
   parseTraceparent,
   RunId,
   serializeTraceparent,
@@ -28,6 +27,7 @@ import { handleMetadataPacket } from "~/utils/packets";
 import { startSpan } from "~/v3/tracing.server";
 import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
 import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server";
+import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
 import type {
   TriggerTaskServiceOptions,
   TriggerTaskServiceResult,
@@ -153,9 +153,7 @@ export class RunEngineTriggerTaskService {
           orgFeatureFlags: environment.organization.featureFlags,
         });
 
-    return mintKind === "runOpsId"
-      ? RunId.toFriendlyId(generateRunOpsId(region))
-      : RunId.generate().friendlyId;
+    return mintFriendlyIdForKind(mintKind, region);
   }
 
   public async call({

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -28,6 +28,7 @@ import { getEventRepositoryForStore, recordRunDebugLog } from "./eventRepository
 import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server";
 import { engine } from "./runEngine.server";
 import { runStore } from "./runStore.server";
+import { mintAnchoredRunFriendlyId } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
 import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server";
 import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
 import {
@@ -808,6 +809,14 @@ export function setupBatchQueueCallbacks() {
           },
         },
         async (span) => {
+          // Anchor every item mint on the BATCH's friendlyId so a mid-batch mint-flag flip
+          // can't split an item (or pre-failed item) from its BatchTaskRun row.
+          const mintItemRunFriendlyId = () =>
+            mintAnchoredRunFriendlyId(
+              friendlyId,
+              (item.options as { region?: string } | undefined)?.region
+            );
+
           const triggerFailedTaskService = new TriggerFailedTaskService({
             prisma,
             engine,
@@ -837,6 +846,7 @@ export function setupBatchQueueCallbacks() {
                 parentRunId: meta.parentRunId,
                 resumeParentOnCompletion: meta.resumeParentOnCompletion,
                 batch: { id: batchId, index: itemIndex },
+                runFriendlyId: mintItemRunFriendlyId(),
                 traceContext: meta.traceContext as Record<string, unknown> | undefined,
                 spanParentAsLink: meta.spanParentAsLink,
               });
@@ -874,6 +884,8 @@ export function setupBatchQueueCallbacks() {
             // Normalize payload - for application/store (R2 paths), this passes through as-is
             const payload = normalizePayload(item.payload, item.payloadType);
 
+            const runFriendlyId = mintItemRunFriendlyId();
+
             const result = await triggerTaskService.call(
               item.task,
               environment,
@@ -893,6 +905,7 @@ export function setupBatchQueueCallbacks() {
                 spanParentAsLink: meta.spanParentAsLink,
                 batchId,
                 batchIndex: itemIndex,
+                runFriendlyId,
                 realtimeStreamsVersion: meta.realtimeStreamsVersion,
                 planType: meta.planType,
                 triggerSource: meta.parentRunId ? "sdk" : (meta.triggerSource ?? "api"),
@@ -929,6 +942,7 @@ export function setupBatchQueueCallbacks() {
                   parentRunId: meta.parentRunId,
                   resumeParentOnCompletion: meta.resumeParentOnCompletion,
                   batch: { id: batchId, index: itemIndex },
+                  runFriendlyId: mintItemRunFriendlyId(),
                   options: item.options as Record<string, unknown>,
                   traceContext: meta.traceContext as Record<string, unknown> | undefined,
                   spanParentAsLink: meta.spanParentAsLink,
@@ -1009,6 +1023,7 @@ export function setupBatchQueueCallbacks() {
                 parentRunId: meta.parentRunId,
                 resumeParentOnCompletion: meta.resumeParentOnCompletion,
                 batch: { id: batchId, index: itemIndex },
+                runFriendlyId: mintItemRunFriendlyId(),
                 options: item.options as Record<string, unknown>,
                 traceContext: meta.traceContext as Record<string, unknown> | undefined,
                 spanParentAsLink: meta.spanParentAsLink,

apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts

@@ -0,0 +1,31 @@
+import {
+  BatchId,
+  classifyKind,
+  generateRunOpsId,
+  parseRunId,
+  REGION_CODES,
+} from "@trigger.dev/core/v3/isomorphic";
+import { describe, expect, it } from "vitest";
+import { mintAnchoredRunFriendlyId } from "./mintAnchoredRunFriendlyId.server";
+
+describe("mintAnchoredRunFriendlyId", () => {
+  it("a run-ops (NEW) batch anchor yields a run-ops (NEW) item friendlyId", () => {
+    const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId);
+    expect(classifyKind(itemFriendlyId)).toBe("runOpsId");
+  });
+
+  it("a cuid (LEGACY) batch anchor yields a cuid (LEGACY) item friendlyId", () => {
+    const batchFriendlyId = BatchId.generate().friendlyId;
+    const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId);
+    expect(classifyKind(itemFriendlyId)).toBe("cuid");
+  });
+
+  it("stamps the requested region char into a run-ops id", () => {
+    const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId, "us-east-1");
+    const parsed = parseRunId(itemFriendlyId);
+    expect(parsed.format).toBe("b32hex");
+    expect(parsed.format === "b32hex" && parsed.region).toBe(REGION_CODES["us-east-1"]);
+  });
+});

apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts

@@ -0,0 +1,15 @@
+import { generateRunOpsId, RunId, type ResidencyKind } from "@trigger.dev/core/v3/isomorphic";
+import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server";
+
+// Shared id-generation branch for every run-mint path: "runOpsId" -> NEW store, "cuid" -> LEGACY.
+export function mintFriendlyIdForKind(mintKind: ResidencyKind, region?: string): string {
+  return mintKind === "runOpsId"
+    ? RunId.toFriendlyId(generateRunOpsId(region))
+    : RunId.generate().friendlyId;
+}
+
+// Anchor a batch item's mint on the BATCH's friendlyId (id-shape, zero I/O), never the per-org
+// flag, so the item and its BatchTaskRun stay co-resident across a mid-batch flag flip.
+export function mintAnchoredRunFriendlyId(batchFriendlyId: string, region?: string): string {
+  return mintFriendlyIdForKind(resolveInheritedMintKind(batchFriendlyId), region);
+}

apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts

@@ -6,9 +6,21 @@ export function computeRunOpsSplitReadEnabled(args: {
   controlPlaneReplica: unknown;
   hasNewUrl: boolean;
   hasLegacyUrl: boolean;
+  logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void };
 }): boolean {
   const newIsDistinctDedicatedClient =
     args.newReplica !== args.controlPlaneWriter && args.newReplica !== args.controlPlaneReplica;
 
-  return newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl;
+  const enabled = newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl;
+
+  // Configured for split but the identity check failed: fan-out is being silently disabled.
+  if (!newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl) {
+    args.logger?.warn(
+      "run-ops split read fan-out is configured (RUN_OPS_DATABASE_URL and " +
+        "RUN_OPS_LEGACY_DATABASE_URL are both set) but the NEW client is not a distinct " +
+        "instance from the control-plane client; read fan-out is silently disabled."
+    );
+  }
+
+  return enabled;
 }

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

@@ -12,7 +12,6 @@ import {
   Prisma,
 } from "@trigger.dev/database";
 import type { RunStore } from "@internal/run-store";
-import { generateRunOpsId, RunId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
 import type { PrismaClientOrTransaction } from "~/db.server";
 import { prisma } from "~/db.server";
@@ -26,6 +25,7 @@ import { getEntitlement } from "~/services/platform.v3.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 import { resolveRunIdMintKind, type RunIdMintKind } from "~/v3/engineVersion.server";
 import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server";
+import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
 import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.server";
 import { batchTriggerWorker } from "../batchTriggerWorker.server";
 import { legacyRunEngineWorker } from "../legacyRunEngineWorker.server";
@@ -361,9 +361,7 @@ export class BatchTriggerV3Service extends BaseService {
           orgFeatureFlags: environment.organization.featureFlags,
         });
 
-    return mintKind === "runOpsId"
-      ? RunId.toFriendlyId(generateRunOpsId(region))
-      : RunId.generate().friendlyId;
+    return mintFriendlyIdForKind(mintKind, region);
   }
 
   async #prepareRunData(

apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts

@@ -0,0 +1,224 @@
+// REGRESSION: the 2-phase batch API (createBatch.server.ts + streamBatchItems.server.ts, wired
+// through BatchQueue to setupBatchQueueCallbacks) must anchor each item's mint on the BATCH's own
+// friendlyId, like batchTriggerV3.server.ts's mintChildFriendlyId does — not re-resolve the
+// per-org mint flag, which can flip between batch creation and this async callback. Covers the
+// happy path (both residency directions) and all three pre-failed-run branches: trigger returned
+// undefined, pre-marked error item, and a thrown trigger error.
+
+import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  findEnvironmentById: vi.fn(),
+  triggerTaskServiceCall: vi.fn(),
+  triggerFailedTaskCall: vi.fn(),
+  setBatchProcessItemCallback: vi.fn(),
+}));
+
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+  runOpsNewPrisma: {},
+  runOpsLegacyPrisma: {},
+  runOpsNewReplica: {},
+  runOpsLegacyReplica: {},
+}));
+
+vi.mock("~/models/runtimeEnvironment.server", () => ({
+  findEnvironmentById: mocks.findEnvironmentById,
+  findEnvironmentFromRun: vi.fn(),
+}));
+
+vi.mock("~/v3/services/triggerTask.server", () => ({
+  TriggerTaskService: class {
+    call(...args: unknown[]) {
+      return mocks.triggerTaskServiceCall(...args);
+    }
+  },
+}));
+
+vi.mock("~/runEngine/services/triggerFailedTask.server", () => ({
+  TriggerFailedTaskService: class {
+    call(...args: unknown[]) {
+      return mocks.triggerFailedTaskCall(...args);
+    }
+  },
+}));
+
+vi.mock("~/v3/runEngine.server", () => ({
+  engine: {
+    setBatchProcessItemCallback: mocks.setBatchProcessItemCallback,
+    setBatchCompletionCallback: vi.fn(),
+    tryCompleteBatch: vi.fn(),
+  },
+}));
+
+vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false }));
+
+import {
+  BatchId,
+  classifyKind,
+  generateRunOpsId,
+  ownerEngine,
+} from "@trigger.dev/core/v3/isomorphic";
+import { setupBatchQueueCallbacks } from "~/v3/runEngineHandlers.server";
+
+vi.setConfig({ testTimeout: 30_000 });
+
+type ProcessItemCallback = (args: {
+  batchId: string;
+  friendlyId: string;
+  itemIndex: number;
+  item: Record<string, unknown>;
+  meta: Record<string, unknown>;
+  attempt: number;
+  isFinalAttempt: boolean;
+}) => Promise<unknown>;
+
+let processItemCallback: ProcessItemCallback;
+
+beforeAll(() => {
+  setupBatchQueueCallbacks();
+  processItemCallback = mocks.setBatchProcessItemCallback.mock.calls[0][0] as ProcessItemCallback;
+});
+
+beforeEach(() => {
+  mocks.findEnvironmentById.mockReset().mockResolvedValue({
+    id: "env_1",
+    organizationId: "org_1",
+    organization: { featureFlags: {} },
+  });
+  mocks.triggerTaskServiceCall.mockReset();
+  mocks.triggerFailedTaskCall.mockReset();
+});
+
+async function runItem(
+  friendlyId: string,
+  isFinalAttempt = false,
+  itemOptions: Record<string, unknown> = {}
+) {
+  await processItemCallback({
+    batchId: "batch_internal_1",
+    friendlyId,
+    itemIndex: 0,
+    item: {
+      task: "some-task",
+      payload: "{}",
+      payloadType: "application/json",
+      options: itemOptions,
+    },
+    meta: { environmentId: "env_1" },
+    attempt: 1,
+    isFinalAttempt,
+  });
+}
+
+describe("setupBatchQueueCallbacks — batch item residency anchoring", () => {
+  // Split is off in this test, so re-resolving the per-org flag would yield cuid — a NEW anchor
+  // yielding runOpsId proves the anchor wins over what the flag would say.
+  it("happy path: a run-ops (NEW) batch anchor mints a run-ops item", async () => {
+    const friendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(friendlyId)).toBe("NEW");
+    mocks.triggerTaskServiceCall.mockResolvedValue({
+      run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" },
+      isCached: false,
+    });
+
+    await runItem(friendlyId);
+
+    expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1);
+    const [, , , options] = mocks.triggerTaskServiceCall.mock.calls[0] as [
+      unknown,
+      unknown,
+      unknown,
+      { runFriendlyId?: string } | undefined,
+    ];
+    expect(options?.runFriendlyId).toBeDefined();
+    expect(classifyKind(options!.runFriendlyId!)).toBe("runOpsId");
+  });
+
+  // The cuid direction catches a "always mint run-ops id regardless of anchor" bug.
+  it("happy path: a cuid (LEGACY) batch anchor mints a cuid item", async () => {
+    const friendlyId = BatchId.generate().friendlyId;
+    expect(ownerEngine(friendlyId)).toBe("LEGACY");
+    mocks.triggerTaskServiceCall.mockResolvedValue({
+      run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" },
+      isCached: false,
+    });
+
+    await runItem(friendlyId);
+
+    expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1);
+    const [, , , options] = mocks.triggerTaskServiceCall.mock.calls[0] as [
+      unknown,
+      unknown,
+      unknown,
+      { runFriendlyId?: string } | undefined,
+    ];
+    expect(options?.runFriendlyId).toBeDefined();
+    expect(classifyKind(options!.runFriendlyId!)).toBe("cuid");
+  });
+
+  // The pre-failed run (TriggerTaskService returned undefined on the final attempt) carries
+  // batchId (→ live TaskRun.batchId FK), so it too must be anchored to the batch's residency.
+  it("failure branch: a run-ops (NEW) batch anchors the pre-failed run to the batch, not the flag", async () => {
+    const friendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(friendlyId)).toBe("NEW");
+    mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await runItem(friendlyId, true);
+
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId");
+  });
+
+  it("failure branch: a cuid (LEGACY) batch anchors the pre-failed run to a cuid id", async () => {
+    const friendlyId = BatchId.generate().friendlyId;
+    mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await runItem(friendlyId, true);
+
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("cuid");
+  });
+
+  // Pre-marked error items (e.g. oversized payloads) are pre-failed before the trigger runs; that
+  // pre-failed run also carries batchId, so it must anchor to the batch too.
+  it("pre-marked error branch: a run-ops (NEW) batch anchors the pre-failed run to the batch", async () => {
+    const friendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(friendlyId)).toBe("NEW");
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await runItem(friendlyId, false, {
+      __error: "payload too large",
+      __errorCode: "PAYLOAD_TOO_LARGE",
+    });
+
+    expect(mocks.triggerTaskServiceCall).not.toHaveBeenCalled();
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId");
+  });
+
+  // The catch branch (the item trigger throws) creates a pre-failed run on the final attempt; it
+  // carries batchId, so it must anchor to the batch too.
+  it("catch branch: a run-ops (NEW) batch anchors the pre-failed run when the trigger throws", async () => {
+    const friendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(friendlyId)).toBe("NEW");
+    mocks.triggerTaskServiceCall.mockRejectedValue(new Error("trigger boom"));
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await runItem(friendlyId, true);
+
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId");
+  });
+});

apps/webapp/test/runEngineBatchTriggerResidencyAnchoring.test.ts

@@ -0,0 +1,185 @@
+// REGRESSION: batch + items must be co-resident (TaskRun.batchId FKs into BatchTaskRun on the
+// run-ops NEW DB). RunEngineBatchTriggerService (api.v2.tasks.batch.ts) must anchor each item's
+// mint on the BATCH's own friendlyId, like batchTriggerV3.server.ts's mintChildFriendlyId does,
+// not re-resolve the per-org flag (which can flip between batch creation and async processing).
+// Covers BOTH the happy path (item minted directly) AND the failure branch (pre-failed run via
+// TriggerFailedTaskService), in BOTH residency directions. Drives the real `processBatchTaskRun`
+// entrypoint with a fake `_engine` (no DB/Redis needed) and captures the trigger-pipeline inputs.
+
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  findEnvironmentById: vi.fn(),
+  triggerTaskServiceCall: vi.fn(),
+  triggerFailedTaskCall: vi.fn(),
+}));
+
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+  runOpsNewPrisma: {},
+  runOpsLegacyPrisma: {},
+  runOpsNewReplica: {},
+  runOpsLegacyReplica: {},
+}));
+
+vi.mock("~/models/runtimeEnvironment.server", () => ({
+  findEnvironmentById: mocks.findEnvironmentById,
+}));
+
+vi.mock("~/v3/services/triggerTask.server", () => ({
+  TriggerTaskService: class {
+    call(...args: unknown[]) {
+      return mocks.triggerTaskServiceCall(...args);
+    }
+  },
+}));
+
+vi.mock("~/runEngine/services/triggerFailedTask.server", () => ({
+  TriggerFailedTaskService: class {
+    call(...args: unknown[]) {
+      return mocks.triggerFailedTaskCall(...args);
+    }
+  },
+}));
+
+import {
+  BatchId,
+  classifyKind,
+  generateRunOpsId,
+  ownerEngine,
+} from "@trigger.dev/core/v3/isomorphic";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { RunEngineBatchTriggerService } from "~/runEngine/services/batchTrigger.server";
+
+vi.setConfig({ testTimeout: 30_000 });
+
+const environment = {
+  id: "env_1",
+  organizationId: "org_1",
+  organization: { featureFlags: {} },
+} as unknown as AuthenticatedEnvironment;
+
+function makeBatch(batchFriendlyId: string) {
+  return {
+    id: "batch_internal_1",
+    friendlyId: batchFriendlyId,
+    runtimeEnvironmentId: "env_1",
+    runCount: 1,
+    payload: JSON.stringify([{ task: "some-task", payload: "{}" }]),
+    payloadType: "application/json",
+    options: {},
+  };
+}
+
+function makeFakeEngine(batch: ReturnType<typeof makeBatch>) {
+  return {
+    runStore: {
+      findBatchTaskRunById: vi.fn().mockResolvedValue(batch),
+      updateBatchTaskRun: vi.fn().mockResolvedValue({ processingJobsCount: 1, runCount: 1 }),
+    },
+    tryCompleteBatch: vi.fn().mockResolvedValue(undefined),
+  };
+}
+
+async function processBatch(batchFriendlyId: string) {
+  const batch = makeBatch(batchFriendlyId);
+  const service = new RunEngineBatchTriggerService(
+    "sequential",
+    {} as any,
+    makeFakeEngine(batch) as any
+  );
+  await service.processBatchTaskRun({
+    batchId: batch.id,
+    processingId: "0",
+    range: { start: 0, count: 50 },
+    attemptCount: 0,
+    strategy: "sequential",
+  });
+}
+
+function anchoredRunFriendlyIdArg(mock: ReturnType<typeof vi.fn>): string | undefined {
+  const [, , , options] = mock.mock.calls[0] as [
+    unknown,
+    unknown,
+    unknown,
+    { runFriendlyId?: string } | undefined,
+  ];
+  return options?.runFriendlyId;
+}
+
+beforeEach(() => {
+  mocks.findEnvironmentById.mockReset().mockResolvedValue(environment);
+  mocks.triggerTaskServiceCall.mockReset();
+  mocks.triggerFailedTaskCall.mockReset();
+});
+
+describe("RunEngineBatchTriggerService — batch item residency anchoring", () => {
+  // Split is off in this test, so re-resolving the per-org flag would yield cuid — a NEW anchor
+  // yielding runOpsId proves the anchor wins over what the flag would say.
+  it("happy path: a run-ops (NEW) batch anchor mints a run-ops item, overriding what the flag would resolve", async () => {
+    const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(batchFriendlyId)).toBe("NEW");
+
+    mocks.triggerTaskServiceCall.mockResolvedValue({
+      run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" },
+      isCached: false,
+    });
+
+    await processBatch(batchFriendlyId);
+
+    expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1);
+    const runFriendlyId = anchoredRunFriendlyIdArg(mocks.triggerTaskServiceCall);
+    expect(runFriendlyId).toBeDefined();
+    expect(classifyKind(runFriendlyId!)).toBe("runOpsId");
+  });
+
+  // The cuid direction catches a "always mint run-ops id regardless of anchor" bug.
+  it("happy path: a cuid (LEGACY) batch anchor mints a cuid item", async () => {
+    const batchFriendlyId = BatchId.generate().friendlyId; // cuid → LEGACY
+    expect(ownerEngine(batchFriendlyId)).toBe("LEGACY");
+
+    mocks.triggerTaskServiceCall.mockResolvedValue({
+      run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" },
+      isCached: false,
+    });
+
+    await processBatch(batchFriendlyId);
+
+    expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1);
+    const runFriendlyId = anchoredRunFriendlyIdArg(mocks.triggerTaskServiceCall);
+    expect(runFriendlyId).toBeDefined();
+    expect(classifyKind(runFriendlyId!)).toBe("cuid");
+  });
+
+  // The pre-failed run created by TriggerFailedTaskService still carries batchId (→ live
+  // TaskRun.batchId FK), so it too must be anchored to the batch's residency, not the per-org flag.
+  it("failure branch: a run-ops (NEW) batch anchors the pre-failed run to the batch, not the flag", async () => {
+    const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId());
+    expect(ownerEngine(batchFriendlyId)).toBe("NEW");
+
+    mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await processBatch(batchFriendlyId);
+
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId");
+  });
+
+  it("failure branch: a cuid (LEGACY) batch anchors the pre-failed run to a cuid id", async () => {
+    const batchFriendlyId = BatchId.generate().friendlyId; // cuid → LEGACY
+
+    mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails
+    mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake");
+
+    await processBatch(batchFriendlyId);
+
+    expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1);
+    const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }];
+    expect(request.runFriendlyId).toBeDefined();
+    expect(classifyKind(request.runFriendlyId!)).toBe("cuid");
+  });
+});

apps/webapp/test/runOpsSplitReadGate.glue.test.ts

@@ -0,0 +1,87 @@
+import { describe, expect, it, vi } from "vitest";
+import { selectRunOpsTopology } from "~/db.server";
+import { computeRunOpsSplitReadEnabled } from "~/v3/runOpsMigration/runOpsSplitReadGate";
+
+// Glue test: nothing previously asserted that selectRunOpsTopology's ACTUAL output, fed into
+// computeRunOpsSplitReadEnabled, yields the right gate. Each unit was only tested against
+// hand-rolled objects, so a refactor that made the NEW client alias a control-plane client
+// (even with correct URLs) would silently disable read fan-out with zero test failure.
+const cp = { writer: { __tag: "cp-writer" } as any, replica: { __tag: "cp-replica" } as any };
+
+describe("selectRunOpsTopology -> computeRunOpsSplitReadEnabled (seam)", () => {
+  it("split-configured with a genuinely distinct NEW client: gate TRUE, no warn", () => {
+    const dedicatedNew = { __tag: "dedicated-new" } as any;
+    const topo = selectRunOpsTopology(
+      { splitEnabled: true, legacyUrl: "postgres://legacy", newUrl: "postgres://new" },
+      {
+        controlPlane: cp,
+        buildNewWriter: vi.fn().mockReturnValue(dedicatedNew),
+        buildNewReplica: vi.fn(),
+      }
+    );
+    const warn = vi.fn();
+
+    const enabled = computeRunOpsSplitReadEnabled({
+      newReplica: topo.newRunOps.replica,
+      controlPlaneWriter: topo.controlPlane.writer,
+      controlPlaneReplica: topo.controlPlane.replica,
+      hasNewUrl: true,
+      hasLegacyUrl: true,
+      logger: { warn },
+    });
+
+    expect(enabled).toBe(true);
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  it("regression: both URLs set but the client factory aliases the control-plane instance -> gate FALSE, warn fires", () => {
+    // Stand-in for the bug this test guards against: a builder refactor that accidentally
+    // returns the shared control-plane client instead of opening a dedicated connection.
+    const topo = selectRunOpsTopology(
+      { splitEnabled: true, legacyUrl: "postgres://legacy", newUrl: "postgres://new" },
+      {
+        controlPlane: cp,
+        buildNewWriter: vi.fn().mockReturnValue(cp.replica),
+        buildNewReplica: vi.fn(),
+      }
+    );
+    const warn = vi.fn();
+
+    const enabled = computeRunOpsSplitReadEnabled({
+      newReplica: topo.newRunOps.replica,
+      controlPlaneWriter: topo.controlPlane.writer,
+      controlPlaneReplica: topo.controlPlane.replica,
+      hasNewUrl: true,
+      hasLegacyUrl: true,
+      logger: { warn },
+    });
+
+    expect(enabled).toBe(false);
+    expect(warn).toHaveBeenCalledTimes(1);
+  });
+
+  it("single mode (URLs unset): topology naturally collapses to control-plane refs -> gate FALSE, no warn", () => {
+    const topo = selectRunOpsTopology(
+      { splitEnabled: false },
+      {
+        controlPlane: cp,
+        buildNewWriter: vi.fn(),
+        buildNewReplica: vi.fn(),
+      }
+    );
+    const warn = vi.fn();
+
+    const enabled = computeRunOpsSplitReadEnabled({
+      newReplica: topo.newRunOps.replica,
+      controlPlaneWriter: topo.controlPlane.writer,
+      controlPlaneReplica: topo.controlPlane.replica,
+      hasNewUrl: false,
+      hasLegacyUrl: false,
+      logger: { warn },
+    });
+
+    expect(enabled).toBe(false);
+    expect(warn).not.toHaveBeenCalled();
+    expect(topo.newRunOps.replica).toBe(cp.replica); // sanity: genuinely the shared instance
+  });
+});

apps/webapp/test/runOpsSplitReadGate.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import { computeRunOpsSplitReadEnabled } from "~/v3/runOpsMigration/runOpsSplitReadGate";
 
 // Distinct sentinel objects standing in for the prisma client singletons.
@@ -72,4 +72,96 @@ describe("computeRunOpsSplitReadEnabled", () => {
       false
     );
   });
+
+  // Observability regression guard: split-configured (both URLs set) but the NEW client is not a
+  // distinct instance must WARN loudly. Without this signal, an accidental refactor that makes the
+  // NEW client alias a control-plane client silently disables read fan-out with zero error/warning.
+  describe("warn signal when configured-but-aliased", () => {
+    it("warns when both URLs are set but NEW aliases the control-plane replica", () => {
+      const warn = vi.fn();
+      const enabled = computeRunOpsSplitReadEnabled({
+        newReplica: cpReplica, // aliasing regression: NEW === control-plane replica
+        controlPlaneWriter: cpWriter,
+        controlPlaneReplica: cpReplica,
+        hasNewUrl: true,
+        hasLegacyUrl: true,
+        logger: { warn },
+      });
+
+      expect(enabled).toBe(false);
+      expect(warn).toHaveBeenCalledTimes(1);
+      expect(warn.mock.calls[0][0]).toMatch(/split.*configured/i);
+    });
+
+    it("warns when both URLs are set but NEW aliases the control-plane writer", () => {
+      const warn = vi.fn();
+      const enabled = computeRunOpsSplitReadEnabled({
+        newReplica: cpWriter, // aliasing regression: NEW === control-plane writer
+        controlPlaneWriter: cpWriter,
+        controlPlaneReplica: cpReplica,
+        hasNewUrl: true,
+        hasLegacyUrl: true,
+        logger: { warn },
+      });
+
+      expect(enabled).toBe(false);
+      expect(warn).toHaveBeenCalledTimes(1);
+    });
+
+    it("does NOT warn in ordinary single mode (both URLs unset, clients naturally aliased)", () => {
+      const warn = vi.fn();
+      const enabled = computeRunOpsSplitReadEnabled({
+        newReplica: cpReplica,
+        controlPlaneWriter: cpWriter,
+        controlPlaneReplica: cpReplica,
+        hasNewUrl: false,
+        hasLegacyUrl: false,
+        logger: { warn },
+      });
+
+      expect(enabled).toBe(false);
+      expect(warn).not.toHaveBeenCalled();
+    });
+
+    it("does NOT warn when only one URL is set (not truly configured for split)", () => {
+      const warn = vi.fn();
+      computeRunOpsSplitReadEnabled({
+        newReplica: cpReplica,
+        controlPlaneWriter: cpWriter,
+        controlPlaneReplica: cpReplica,
+        hasNewUrl: true,
+        hasLegacyUrl: false,
+        logger: { warn },
+      });
+
+      expect(warn).not.toHaveBeenCalled();
+    });
+
+    it("does NOT warn when the NEW client is genuinely distinct (healthy split)", () => {
+      const warn = vi.fn();
+      const enabled = computeRunOpsSplitReadEnabled({
+        newReplica: dedicatedNew,
+        controlPlaneWriter: cpWriter,
+        controlPlaneReplica: cpReplica,
+        hasNewUrl: true,
+        hasLegacyUrl: true,
+        logger: { warn },
+      });
+
+      expect(enabled).toBe(true);
+      expect(warn).not.toHaveBeenCalled();
+    });
+
+    it("does not throw when no logger is supplied (logger stays optional)", () => {
+      expect(() =>
+        computeRunOpsSplitReadEnabled({
+          newReplica: cpReplica,
+          controlPlaneWriter: cpWriter,
+          controlPlaneReplica: cpReplica,
+          hasNewUrl: true,
+          hasLegacyUrl: true,
+        })
+      ).not.toThrow();
+    });
+  });
 });

internal-packages/run-ops-database/prisma/schema.parity.test.ts

@@ -2,11 +2,24 @@ import { describe, expect, it } from "vitest";
 import { readFileSync } from "node:fs";
 import { resolve } from "node:path";
 
-// Asserts every scalar column of the run-subgraph models in the control-plane
-// schema also exists in the dedicated schema (so the dedicated DB can hold the
-// same row shape), and that the dedicated schema contains NO reference to a
-// control-plane model name.
-const CONTROL_PLANE_MODELS = [
+// Parses both the control-plane schema (`@trigger.dev/database`) and this
+// dedicated run-ops schema, then diffs every SCALAR field of each run-subgraph
+// model BIDIRECTIONALLY: a field present in either schema must exist in the
+// other with matching type/nullability/array-ness/`@default`, since a run row
+// must be writable to either physical database.
+//
+// Relation fields are excluded by design: the dedicated schema drops
+// control-plane `@relation`s to control-plane-owned models and replaces the
+// implicit many-to-many waitpoint relations with explicit FK-free join models
+// (`TaskRunWaitpoint`, `WaitpointRunConnection`, `CompletedWaitpoint`). Only
+// the scalar FK columns backing those relations (e.g. `projectId`) are
+// compared; the relation object fields (e.g. `project`) are skipped.
+
+const CONTROL_PLANE_SCHEMA_PATH = "../../database/prisma/schema.prisma";
+const DEDICATED_SCHEMA_PATH = "./schema.prisma";
+
+// Must never appear as a relation target in the dedicated schema.
+const CONTROL_PLANE_ONLY_MODELS = [
   "Organization",
   "OrgMember",
   "Project",
@@ -19,73 +32,232 @@ const CONTROL_PLANE_MODELS = [
   "TaskQueue",
 ];
 
-function readSchema(rel: string) {
+// Must have every scalar column reproduced in the dedicated schema.
+const RUN_SUBGRAPH_MODELS = [
+  "TaskRun",
+  "TaskRunAttempt",
+  "TaskRunExecutionSnapshot",
+  "TaskRunWaitpoint",
+  "TaskRunCheckpoint",
+  "CheckpointRestoreEvent",
+  "TaskRunTag",
+  "Waitpoint",
+  "WaitpointTag",
+  "BatchTaskRun",
+  "TaskRunDependency",
+  "BatchTaskRunItem",
+  "BatchTaskRunError",
+  "Checkpoint",
+];
+
+function readSchema(rel: string): string {
   return readFileSync(resolve(__dirname, rel), "utf8");
 }
 
-// Prisma comments (`///` docs and `//` lines) may legitimately mention
-// control-plane model names in prose, which would false-match the drift
-// regexes below. Strip them so parity assertions only see real schema syntax.
-function stripComments(schema: string) {
-  return schema.replace(/\/\/.*$/gm, "");
+// Strips `/* */` block comments and `//`/`///` line comments so prose mentioning
+// model names can't false-match below. (Neither schema has `/*` inside a string.)
+function stripComments(schema: string): string {
+  return schema.replace(/\/\*[\s\S]*?\*\//g, "").replace(/\/\/.*$/gm, "");
 }
 
+type FieldInfo = {
+  type: string;
+  optional: boolean;
+  array: boolean;
+  default: string | null;
+};
+
+type ModelFields = Map<string, FieldInfo>;
+
+// Prisma model blocks don't nest, so a lazy match to the next `}`-only line suffices.
+function extractModelBlocks(schema: string): Map<string, string> {
+  const blocks = new Map<string, string>();
+  const re = /^model\s+(\w+)\s*\{([\s\S]*?)\n\}/gm;
+  let match: RegExpExecArray | null;
+  while ((match = re.exec(schema))) {
+    blocks.set(match[1], match[2]);
+  }
+  return blocks;
+}
+
+// Raw argument of `@default(...)`, honoring nested parens (`@default(cuid())`). Null if absent.
+function extractDefault(attrs: string): string | null {
+  const marker = "@default(";
+  const idx = attrs.indexOf(marker);
+  if (idx === -1) return null;
+  const start = idx + marker.length;
+  let depth = 0;
+  for (let i = start; i < attrs.length; i++) {
+    if (attrs[i] === "(") {
+      depth++;
+    } else if (attrs[i] === ")") {
+      if (depth === 0) return attrs.slice(start, i);
+      depth--;
+    }
+  }
+  return attrs.slice(start);
+}
+
+// Excludes relation fields: anything carrying `@relation`, plus back-relation fields
+// (e.g. `checkpoints Checkpoint[]`) identified by their type being another model name.
+// `unparsed` collects any content line the field regex fails to match, so a future
+// multi-line field can never silently vanish from the diff.
+function parseScalarFields(
+  body: string,
+  modelNames: Set<string>
+): { fields: ModelFields; unparsed: string[] } {
+  const fields: ModelFields = new Map();
+  const unparsed: string[] = [];
+  for (const rawLine of body.split("\n")) {
+    const line = rawLine.trim();
+    if (!line || line.startsWith("@@")) continue;
+    const match = line.match(/^(\w+)\s+([A-Za-z_]\w*)(\[\])?(\?)?\s*(.*)$/);
+    if (!match) {
+      unparsed.push(line);
+      continue;
+    }
+    const [, name, type, arrayMark, optionalMark, rest] = match;
+    if (rest.includes("@relation")) continue;
+    if (modelNames.has(type)) continue;
+    fields.set(name, {
+      type,
+      array: arrayMark === "[]",
+      optional: optionalMark === "?",
+      default: extractDefault(rest),
+    });
+  }
+  return { fields, unparsed };
+}
+
+function parseSchema(schema: string) {
+  const stripped = stripComments(schema);
+  const modelBlocks = extractModelBlocks(stripped);
+  const modelNames = new Set(modelBlocks.keys());
+  const models = new Map<string, ModelFields>();
+  const unparsed: string[] = [];
+  for (const [name, body] of modelBlocks) {
+    const parsed = parseScalarFields(body, modelNames);
+    models.set(name, parsed.fields);
+    for (const line of parsed.unparsed) {
+      unparsed.push(`${name}: ${line}`);
+    }
+  }
+  return { models, modelNames, unparsed };
+}
+
+function describeField(field: FieldInfo | undefined): string {
+  if (!field) return "<missing>";
+  const array = field.array ? "[]" : "";
+  const optional = field.optional ? "?" : "";
+  const withDefault = field.default !== null ? ` @default(${field.default})` : "";
+  return `${field.type}${array}${optional}${withDefault}`;
+}
+
+const controlPlane = parseSchema(readSchema(CONTROL_PLANE_SCHEMA_PATH));
+const dedicated = parseSchema(readSchema(DEDICATED_SCHEMA_PATH));
+
 describe("dedicated run-ops schema parity", () => {
+  it("includes all 14 run-subgraph models in both schemas", () => {
+    for (const model of RUN_SUBGRAPH_MODELS) {
+      expect(Array.from(dedicated.modelNames)).toContain(model);
+      expect(Array.from(controlPlane.modelNames)).toContain(model);
+    }
+  });
+
+  it("fails on any content line the field parser doesn't recognise (no silent skips)", () => {
+    // Scope the control-plane check to run-subgraph models so an unrelated control-plane schema
+    // edit can't break this run-ops test; the dedicated schema contains only run-ops models.
+    const inRunSubgraph = (entry: string) =>
+      RUN_SUBGRAPH_MODELS.some((model) => entry.startsWith(`${model}: `));
+    expect(controlPlane.unparsed.filter(inRunSubgraph)).toEqual([]);
+    expect(dedicated.unparsed).toEqual([]);
+  });
+
+  it("reproduces every scalar column of each run-subgraph model bidirectionally with matching type/nullability/array/default", () => {
+    const mismatches: string[] = [];
+
+    for (const modelName of RUN_SUBGRAPH_MODELS) {
+      const controlFields = controlPlane.models.get(modelName);
+      const dedicatedFields = dedicated.models.get(modelName);
+      if (!controlFields || !dedicatedFields) {
+        mismatches.push(`${modelName}: model not found in one of the two schemas`);
+        continue;
+      }
+
+      const fieldNames = new Set([...controlFields.keys(), ...dedicatedFields.keys()]);
+      for (const fieldName of fieldNames) {
+        const controlField = controlFields.get(fieldName);
+        const dedicatedField = dedicatedFields.get(fieldName);
+
+        if (!controlField) {
+          mismatches.push(
+            `${modelName}.${fieldName}: dedicated has ${describeField(
+              dedicatedField
+            )} but the control-plane schema has no such scalar field`
+          );
+          continue;
+        }
+        if (!dedicatedField) {
+          mismatches.push(
+            `${modelName}.${fieldName}: control-plane has ${describeField(
+              controlField
+            )} but the dedicated schema has no such scalar field`
+          );
+          continue;
+        }
+
+        const matches =
+          dedicatedField.type === controlField.type &&
+          dedicatedField.array === controlField.array &&
+          dedicatedField.optional === controlField.optional &&
+          dedicatedField.default === controlField.default;
+
+        if (!matches) {
+          mismatches.push(
+            `${modelName}.${fieldName}: control-plane=${describeField(
+              controlField
+            )} dedicated=${describeField(dedicatedField)}`
+          );
+        }
+      }
+    }
+
+    expect(mismatches).toEqual([]);
+  });
+
   it("references no control-plane model as a relation target", () => {
-    const dedicated = stripComments(readSchema("./schema.prisma"));
-    for (const model of CONTROL_PLANE_MODELS) {
+    const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH));
+    for (const model of CONTROL_PLANE_ONLY_MODELS) {
       // A relation target appears as `  fieldName  Model @relation(...)`. A bare
       // scalar column like `projectId String` is fine; the model TYPE must be absent.
       const relationTarget = new RegExp(
         `@relation[^\\n]*\\b${model}\\b|\\b${model}\\b[^\\n]*@relation`
       );
-      expect(dedicated).not.toMatch(relationTarget);
-      expect(dedicated).not.toMatch(new RegExp(`\\s${model}(\\?|\\[\\])?\\s`));
-    }
-  });
-
-  it("includes all 14 run-subgraph models", () => {
-    const dedicated = readSchema("./schema.prisma");
-    for (const m of [
-      "TaskRun",
-      "TaskRunAttempt",
-      "TaskRunExecutionSnapshot",
-      "TaskRunWaitpoint",
-      "TaskRunCheckpoint",
-      "CheckpointRestoreEvent",
-      "TaskRunTag",
-      "Waitpoint",
-      "WaitpointTag",
-      "BatchTaskRun",
-      "TaskRunDependency",
-      "BatchTaskRunItem",
-      "BatchTaskRunError",
-      "Checkpoint",
-    ]) {
-      expect(dedicated).toMatch(new RegExp(`model ${m} \\{`));
+      expect(dedicatedText).not.toMatch(relationTarget);
+      expect(dedicatedText).not.toMatch(new RegExp(`\\s${model}(\\?|\\[\\])?\\s`));
     }
   });
 
   it("keeps the group-(A) waitpoint-block references FK-FREE (scalar columns / explicit FK-free join models)", () => {
-    const dedicated = stripComments(readSchema("./schema.prisma"));
+    const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH));
     // TaskRunWaitpoint must NOT carry a `@relation` to Waitpoint/TaskRun/BatchTaskRun.
-    const trw = dedicated.match(/model TaskRunWaitpoint \{[\s\S]*?\n\}/)![0];
+    const trw = dedicatedText.match(/model TaskRunWaitpoint \{[\s\S]*?\n\}/)![0];
     expect(trw).not.toMatch(/@relation/);
     expect(trw).toMatch(/waitpointId\s+String/);
     expect(trw).toMatch(/taskRunId\s+String/);
     // The two implicit M2M sets are replaced by explicit FK-free join models.
-    expect(dedicated).toMatch(/model WaitpointRunConnection \{/);
-    expect(dedicated).toMatch(/model CompletedWaitpoint \{/);
-    const wrc = dedicated.match(/model WaitpointRunConnection \{[\s\S]*?\n\}/)![0];
+    expect(dedicatedText).toMatch(/model WaitpointRunConnection \{/);
+    expect(dedicatedText).toMatch(/model CompletedWaitpoint \{/);
+    const wrc = dedicatedText.match(/model WaitpointRunConnection \{[\s\S]*?\n\}/)![0];
     expect(wrc).not.toMatch(/@relation/);
     // Waitpoint completion back-refs are scalar, not relations.
-    const wp = dedicated.match(/model Waitpoint \{[\s\S]*?\n\}/)![0];
+    const wp = dedicatedText.match(/model Waitpoint \{[\s\S]*?\n\}/)![0];
     expect(wp).not.toMatch(/completedByTaskRun\s+TaskRun\s*\?\s*@relation/);
   });
 
   it("keeps the group-(B) co-resident references as real FKs (e.g. TaskRunAttempt.taskRun)", () => {
-    const dedicated = stripComments(readSchema("./schema.prisma"));
-    const attempt = dedicated.match(/model TaskRunAttempt \{[\s\S]*?\n\}/)![0];
+    const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH));
+    const attempt = dedicatedText.match(/model TaskRunAttempt \{[\s\S]*?\n\}/)![0];
     // The attempt->run relation stays a real FK (always co-resident).
     expect(attempt).toMatch(/taskRun\s+TaskRun\s+@relation/);
   });