From dc945f13dc98752015310a68e8b68ca88eeeea5f Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:44:49 +0100
Subject: [PATCH 01/13] feat(webapp): derive the non-aliased shard target list
 once

Every boot check that must not treat two handles over one database as two
databases needs the same list. The alias exemption keys on the declared
aliasOf field, never on client object identity: two store objects can sit
over one database, which identity comparison cannot see.
---
 apps/webapp/app/v3/runOpsShards.server.ts | 30 ++++++++++++
 apps/webapp/test/runOpsShards.test.ts     | 56 ++++++++++++++++++++++-
 2 files changed, 85 insertions(+), 1 deletion(-)

diff --git a/apps/webapp/app/v3/runOpsShards.server.ts b/apps/webapp/app/v3/runOpsShards.server.ts
index 23c45efa404..12cc460aa06 100644
--- a/apps/webapp/app/v3/runOpsShards.server.ts
+++ b/apps/webapp/app/v3/runOpsShards.server.ts
@@ -122,3 +122,33 @@ export function validateShardListAgainstNewUrl(
 ): boolean {
   return shards.length === 0 || !!newUrl;
 }
+
+// A shard that owns its own physical database. Every boot check that must not treat two handles
+// over one database as two databases derives its target list from here: the distinctness sentinel,
+// the coresidency loop, the read gate, the replication sources and the migration loop.
+export type ShardTarget = {
+  key: string;
+  url: string;
+  replicaUrl?: string;
+  directUrl?: string;
+};
+
+// An aliased shard shares its target's client BY REFERENCE, so it is never its own database. The
+// exemption keys on the declared `aliasOf` field, never on client object identity: two store objects
+// can sit over one database, which identity comparison cannot see.
+export function nonAliasedShards(shards: RunOpsShardDescriptor[]): ShardTarget[] {
+  const targets: ShardTarget[] = [];
+  for (const shard of shards) {
+    if (shard.aliasOf !== undefined) continue;
+    // Unreachable for a valid descriptor (the schema requires exactly one of url/aliasOf); this is
+    // the type narrowing, not a second policy.
+    if (shard.url === undefined) continue;
+    targets.push({
+      key: shard.key,
+      url: shard.url,
+      ...(shard.replicaUrl !== undefined ? { replicaUrl: shard.replicaUrl } : {}),
+      ...(shard.directUrl !== undefined ? { directUrl: shard.directUrl } : {}),
+    });
+  }
+  return targets;
+}
diff --git a/apps/webapp/test/runOpsShards.test.ts b/apps/webapp/test/runOpsShards.test.ts
index fef7e925e75..c7fd0e0defa 100644
--- a/apps/webapp/test/runOpsShards.test.ts
+++ b/apps/webapp/test/runOpsShards.test.ts
@@ -1,6 +1,11 @@
 import { describe, expect, it } from "vitest";
 import { z } from "zod";
-import { parseRunOpsShards, validateShardListAgainstNewUrl } from "~/v3/runOpsShards.server";
+import {
+  nonAliasedShards,
+  parseRunOpsShards,
+  validateShardListAgainstNewUrl,
+  type RunOpsShardDescriptor,
+} from "~/v3/runOpsShards.server";
 
 function run(raw: string | undefined) {
   const schema = z.string().optional().transform(parseRunOpsShards);
@@ -82,3 +87,52 @@ describe("validateShardListAgainstNewUrl", () => {
     expect(validateShardListAgainstNewUrl([valid as never], undefined)).toBe(false);
   });
 });
+
+describe("nonAliasedShards", () => {
+  const shardA: RunOpsShardDescriptor = {
+    key: "a",
+    region: "us-east-1",
+    url: "postgres://h/a",
+    replication: { slotName: "sa", publicationName: "pa", originGeneration: 2 },
+  };
+  const shardB: RunOpsShardDescriptor = {
+    key: "b",
+    region: "us-west-2",
+    url: "postgres://h/b",
+    replicaUrl: "postgres://h/b-replica",
+    directUrl: "postgres://h/b-direct",
+    replication: { slotName: "sb", publicationName: "pb", originGeneration: 3 },
+  };
+  const aliased: RunOpsShardDescriptor = {
+    key: "z",
+    region: "us-east-1",
+    aliasOf: "new",
+  };
+
+  it("returns [] for no descriptors", () => {
+    expect(nonAliasedShards([])).toEqual([]);
+  });
+
+  it("keeps a shard that owns its own database", () => {
+    expect(nonAliasedShards([shardA])).toEqual([{ key: "a", url: "postgres://h/a" }]);
+  });
+
+  it("carries the replica and direct URLs when the descriptor sets them", () => {
+    expect(nonAliasedShards([shardB])).toEqual([
+      {
+        key: "b",
+        url: "postgres://h/b",
+        replicaUrl: "postgres://h/b-replica",
+        directUrl: "postgres://h/b-direct",
+      },
+    ]);
+  });
+
+  it("drops an aliased shard, because it shares its target's database", () => {
+    expect(nonAliasedShards([aliased])).toEqual([]);
+  });
+
+  it("keeps declaration order across a mixed list", () => {
+    expect(nonAliasedShards([shardA, aliased, shardB]).map((s) => s.key)).toEqual(["a", "b"]);
+  });
+});

From 4b1085bf965c07ddefb69ed95fa978baf16fe79f Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:46:56 +0100
Subject: [PATCH 02/13] feat(webapp): check distinctness as set uniqueness over
 every run-ops store

probeDistinctStores groups every target by its system identifier and database
name, so a duplicate between any two stores blocks the boot, not only a
duplicate between the gen-1 pair. Fail-closed is unchanged: a probe that
cannot answer returns not-distinct.

probeDistinctDatabases stays exported as a delegate over a 2-element list. Its
four existing container tests are the proof that set uniqueness over one pair
is the pairwise compare of today.
---
 .../distinctDbSentinel.server.ts              |  74 +++++++----
 .../distinctDbSentinel.server.test.ts         | 115 +++++++++++++++++-
 2 files changed, 166 insertions(+), 23 deletions(-)

diff --git a/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts b/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts
index 4b2bfd9d986..ed7fb0cb237 100644
--- a/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts
@@ -62,32 +62,46 @@ export async function probeControlPlaneCoresidency(
   }
 }
 
-export async function probeDistinctDatabases(
-  legacyUrl: string,
-  newUrl: string,
+export type DistinctTarget = { id: string; url: string };
+
+/**
+ * Set uniqueness over every store that owns its own database. Fail-closed: a probe that cannot
+ * answer returns NOT distinct, because "distinct" is a positive claim a failed probe cannot support.
+ *
+ * Same-cluster-different-database policy (unchanged from the pairwise probe): two databases inside
+ * the SAME cluster (same system identifier, different current_database()) are reported distinct.
+ * They are genuinely separate Postgres databases with separate WAL-visible state for our purposes.
+ *
+ * An ALIASED shard never appears in `targets`. It shares its target's client by reference, so it is
+ * not its own database and inclusion would guarantee a duplicate. See nonAliasedShards.
+ */
+export async function probeDistinctStores(
+  targets: DistinctTarget[],
   opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
 ): Promise<{ distinct: true } | { distinct: false; reason: string }> {
+  if (targets.length < 2) {
+    return { distinct: true };
+  }
+
   try {
-    const [legacy, next] = await Promise.all([
-      readDatabaseFingerprint(legacyUrl),
-      readDatabaseFingerprint(newUrl),
-    ]);
-    const sameCluster = legacy.systemIdentifier === next.systemIdentifier;
-    const sameDb = sameCluster && legacy.databaseName === next.databaseName;
-    // Same-cluster-different-database policy: two databases inside the SAME cluster
-    // (same system identifier, different current_database()) are reported distinct: true.
-    // That is acceptable — they are genuinely separate Postgres databases with separate
-    // WAL-visible state for our purposes, and the Cloud topology always uses separate
-    // clusters anyway. A stricter "must be a different cluster" policy would gate on
-    // sameCluster alone; that is flagged as an open question, not decided here.
-    if (sameDb) {
-      const reason =
-        "run-ops legacy and new URLs resolve to the SAME physical database " +
-        `(systemIdentifier=${legacy.systemIdentifier}, database=${legacy.databaseName}); ` +
-        "refusing to enable split — pooler/replica likely.";
-      opts?.logger?.warn(reason);
-      return { distinct: false, reason };
+    const fingerprints = await Promise.all(targets.map((t) => readDatabaseFingerprint(t.url)));
+
+    const seen = new Map<string, string>();
+    for (const [index, target] of targets.entries()) {
+      const fingerprint = fingerprints[index];
+      const key = `${fingerprint.systemIdentifier}/${fingerprint.databaseName}`;
+      const first = seen.get(key);
+      if (first !== undefined) {
+        const reason =
+          `run-ops stores "${first}" and "${target.id}" resolve to the SAME physical database ` +
+          `(systemIdentifier=${fingerprint.systemIdentifier}, database=${fingerprint.databaseName}); ` +
+          "refusing to enable split — pooler/replica likely.";
+        opts?.logger?.warn(reason);
+        return { distinct: false, reason };
+      }
+      seen.set(key, target.id);
     }
+
     return { distinct: true };
   } catch (error) {
     const reason = `distinct-db sentinel probe failed; failing closed (single-DB). ${String(error)}`;
@@ -95,3 +109,19 @@ export async function probeDistinctDatabases(
     return { distinct: false, reason };
   }
 }
+
+// The gen-1 pairwise entry point, kept as a thin delegate over a 2-element target list. Set
+// uniqueness over one pair IS the pairwise compare, and this function's tests are the proof.
+export async function probeDistinctDatabases(
+  legacyUrl: string,
+  newUrl: string,
+  opts?: { logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void } }
+): Promise<{ distinct: true } | { distinct: false; reason: string }> {
+  return probeDistinctStores(
+    [
+      { id: "legacy", url: legacyUrl },
+      { id: "new", url: newUrl },
+    ],
+    opts
+  );
+}
diff --git a/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts b/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts
index d2baaa6404a..562d50b63d5 100644
--- a/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts
+++ b/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts
@@ -1,7 +1,10 @@
 import { heteroPostgresTest } from "@internal/testcontainers";
 import { PrismaClient } from "@trigger.dev/database";
 import { describe, expect, vi } from "vitest";
-import { probeDistinctDatabases } from "~/v3/runOpsMigration/distinctDbSentinel.server";
+import {
+  probeDistinctDatabases,
+  probeDistinctStores,
+} from "~/v3/runOpsMigration/distinctDbSentinel.server";
 
 // Spinning up two separate postgres clusters and probing each can exceed the 5s default.
 vi.setConfig({ testTimeout: 60_000 });
@@ -62,3 +65,113 @@ describe("probeDistinctDatabases", () => {
     }
   );
 });
+
+describe("probeDistinctStores (set uniqueness at N)", () => {
+  heteroPostgresTest(
+    "reports distinct for two separate physical clusters",
+    async ({ uri14, uri17 }) => {
+      const result = await probeDistinctStores([
+        { id: "legacy", url: uri14 },
+        { id: "new", url: uri17 },
+      ]);
+      expect(result).toEqual({ distinct: true });
+    }
+  );
+
+  heteroPostgresTest("reports distinct for a single target", async ({ uri14 }) => {
+    const result = await probeDistinctStores([{ id: "legacy", url: uri14 }]);
+    expect(result).toEqual({ distinct: true });
+  });
+
+  heteroPostgresTest("reports distinct for an empty target list", async () => {
+    const result = await probeDistinctStores([]);
+    expect(result).toEqual({ distinct: true });
+  });
+
+  heteroPostgresTest(
+    "reports NOT distinct, naming both ids, when two targets resolve to one database",
+    async ({ uri14, uri17 }) => {
+      const result = await probeDistinctStores([
+        { id: "legacy", url: uri14 },
+        { id: "new", url: uri17 },
+        { id: "shard-a", url: uri14 },
+      ]);
+      expect(result.distinct).toBe(false);
+      if (result.distinct === false) {
+        expect(result.reason).toMatch(/same physical database/i);
+        expect(result.reason).toMatch(/legacy/);
+        expect(result.reason).toMatch(/shard-a/);
+      }
+    }
+  );
+
+  // A pairwise implementation that only ever compares the first two targets passes every other
+  // case in this file and fails this one: legacy vs new is clean, and the duplicate pair is
+  // shard against shard on a third database.
+  heteroPostgresTest(
+    "catches a duplicate between two SHARDS while the gen-1 pair is clean",
+    async ({ postgresContainer14, uri14, uri17 }) => {
+      const shardDb = `sentinel_shard_dupe_${Date.now()}`;
+      const admin = new PrismaClient({
+        datasources: {
+          db: { url: urlWithDatabase(postgresContainer14.getConnectionUri(), "postgres") },
+        },
+      });
+      try {
+        await admin.$executeRawUnsafe(`CREATE DATABASE "${shardDb}"`);
+      } finally {
+        await admin.$disconnect();
+      }
+      const shardUrl = urlWithDatabase(uri14, shardDb);
+
+      const result = await probeDistinctStores([
+        { id: "legacy", url: uri14 },
+        { id: "new", url: uri17 },
+        { id: "shard-a", url: shardUrl },
+        { id: "shard-b", url: shardUrl },
+      ]);
+      expect(result.distinct).toBe(false);
+      if (result.distinct === false) {
+        expect(result.reason).toMatch(/shard-a/);
+        expect(result.reason).toMatch(/shard-b/);
+      }
+    }
+  );
+
+  heteroPostgresTest(
+    "reports distinct for two databases in the SAME cluster",
+    async ({ postgresContainer14, uri14, uri17 }) => {
+      const otherDb = `sentinel_set_other_${Date.now()}`;
+      const admin = new PrismaClient({
+        datasources: {
+          db: { url: urlWithDatabase(postgresContainer14.getConnectionUri(), "postgres") },
+        },
+      });
+      try {
+        await admin.$executeRawUnsafe(`CREATE DATABASE "${otherDb}"`);
+      } finally {
+        await admin.$disconnect();
+      }
+
+      const result = await probeDistinctStores([
+        { id: "legacy", url: uri14 },
+        { id: "new", url: uri17 },
+        { id: "shard-a", url: urlWithDatabase(uri14, otherDb) },
+      ]);
+      expect(result).toEqual({ distinct: true });
+    }
+  );
+
+  heteroPostgresTest(
+    "fails closed to NOT distinct when one target cannot be reached",
+    async ({ uri14, uri17 }) => {
+      const unreachable = "postgresql://nobody:nobody@127.0.0.1:1/does_not_exist";
+      const result = await probeDistinctStores([
+        { id: "legacy", url: uri14 },
+        { id: "new", url: uri17 },
+        { id: "shard-a", url: unreachable },
+      ]);
+      expect(result.distinct).toBe(false);
+    }
+  );
+});

From 8f3aa11376f4f9398ba056584d954e2bbe73aa2f Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:48:05 +0100
Subject: [PATCH 03/13] feat(webapp): give the split gate every store that owns
 its own database

computeSplitEnabled builds one probe target per store and passes them to the
set-uniqueness probe. An aliased shard is already absent from the list, so it
needs no exception. The flag-off short circuit is unchanged, so a single-database
boot still opens no second connection.
---
 .../v3/runOpsMigration/splitMode.server.ts    | 17 ++++-
 apps/webapp/test/runOpsSplitMode.test.ts      | 71 +++++++++++++++++++
 2 files changed, 85 insertions(+), 3 deletions(-)

diff --git a/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts b/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts
index 688f95bac03..b9a4e3dfdf2 100644
--- a/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts
@@ -6,12 +6,15 @@
  */
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
-import { probeDistinctDatabases as defaultProbe } from "./distinctDbSentinel.server";
+import { probeDistinctStores as defaultProbe } from "./distinctDbSentinel.server";
+import { nonAliasedShards, type ShardTarget } from "~/v3/runOpsShards.server";
 
 export type SplitModeConfig = {
   flagEnabled: boolean;
   legacyUrl?: string;
   newUrl?: string;
+  /** Gen-2 shards that own their own database. Empty (the default) is today's gen-1 pair. */
+  shards?: ShardTarget[];
 };
 
 export type SplitModeDeps = {
@@ -34,9 +37,16 @@ export async function computeSplitEnabled(
     );
     return false;
   }
-  // Hard gate #2: runtime sentinel must confirm physically-distinct DBs.
+  // Hard gate #2: runtime sentinel must confirm physically-distinct DBs. At N stores this is set
+  // uniqueness over every store that owns its own database, not a compare of the gen-1 pair. An
+  // aliased shard is already absent from `shards` — it shares its target's client by reference.
   const probe = deps.probe ?? defaultProbe;
-  const result = await probe(config.legacyUrl, config.newUrl, { logger: deps.logger });
+  const targets = [
+    { id: "legacy", url: config.legacyUrl },
+    { id: "new", url: config.newUrl },
+    ...(config.shards ?? []).map((shard) => ({ id: `shard-${shard.key}`, url: shard.url })),
+  ];
+  const result = await probe(targets, { logger: deps.logger });
   return result.distinct === true;
 }
 
@@ -72,6 +82,7 @@ export function isSplitEnabled(): Promise<boolean> {
         flagEnabled: env.RUN_OPS_SPLIT_ENABLED,
         legacyUrl: env.RUN_OPS_LEGACY_DATABASE_URL,
         newUrl: env.RUN_OPS_DATABASE_URL,
+        shards: nonAliasedShards(env.RUN_OPS_SHARDS),
       },
       { logger }
     );
diff --git a/apps/webapp/test/runOpsSplitMode.test.ts b/apps/webapp/test/runOpsSplitMode.test.ts
index 7ce2bec3a5d..fd7da6f356c 100644
--- a/apps/webapp/test/runOpsSplitMode.test.ts
+++ b/apps/webapp/test/runOpsSplitMode.test.ts
@@ -61,6 +61,77 @@ describe("computeSplitEnabled (pure)", () => {
   });
 });
 
+describe("computeSplitEnabled shard targets", () => {
+  const shardA = { key: "a", url: "postgres://shard-a" };
+  const shardB = { key: "b", url: "postgres://shard-b" };
+
+  it("probes the gen-1 pair only when no shard is configured", async () => {
+    const probe = vi.fn().mockResolvedValue({ distinct: true });
+    await computeSplitEnabled(
+      { flagEnabled: true, legacyUrl: "postgres://a", newUrl: "postgres://b" },
+      { probe }
+    );
+    expect(probe).toHaveBeenCalledWith(
+      [
+        { id: "legacy", url: "postgres://a" },
+        { id: "new", url: "postgres://b" },
+      ],
+      expect.anything()
+    );
+  });
+
+  it("appends one target per shard, keyed by shard id", async () => {
+    const probe = vi.fn().mockResolvedValue({ distinct: true });
+    await computeSplitEnabled(
+      {
+        flagEnabled: true,
+        legacyUrl: "postgres://a",
+        newUrl: "postgres://b",
+        shards: [shardA, shardB],
+      },
+      { probe }
+    );
+    expect(probe).toHaveBeenCalledWith(
+      [
+        { id: "legacy", url: "postgres://a" },
+        { id: "new", url: "postgres://b" },
+        { id: "shard-a", url: "postgres://shard-a" },
+        { id: "shard-b", url: "postgres://shard-b" },
+      ],
+      expect.anything()
+    );
+  });
+
+  it("stays single-DB when a shard duplicates another store", async () => {
+    const probe = vi.fn().mockResolvedValue({ distinct: false, reason: "same DB" });
+    expect(
+      await computeSplitEnabled(
+        {
+          flagEnabled: true,
+          legacyUrl: "postgres://a",
+          newUrl: "postgres://b",
+          shards: [shardA],
+        },
+        { probe }
+      )
+    ).toBe(false);
+  });
+
+  it("never probes a shard when the flag is off", async () => {
+    const probe = vi.fn();
+    await computeSplitEnabled(
+      {
+        flagEnabled: false,
+        legacyUrl: "postgres://a",
+        newUrl: "postgres://b",
+        shards: [shardA],
+      },
+      { probe }
+    );
+    expect(probe).not.toHaveBeenCalled();
+  });
+});
+
 describe("assertSplitRealtimeInterlock (pure)", () => {
   it("throws when split is on but the native realtime backend is off", () => {
     expect(() =>

From 45c44507bb30b299c206ca8e86d3991d4048aea2 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:50:55 +0100
Subject: [PATCH 04/13] feat(webapp): run the co-residency advisory per run-ops
 store

Every store that owns its own database is probed against the control plane. The
legacy emission keeps its exact call shape and its untagged metric series, so a
deployment with no shard configured reports what it reports today. A shard
emission carries its shard key.

Every store emits before any enforcement throw, so one offending store never
costs another store its metric. A probe that throws degrades that one store to
unknown.

Two tests read RUN_OPS_LEGACY_DATABASE_URL from the developer's .env, because ??
only guards nullish and they passed undefined. They now pin the value, so they
no longer depend on the local environment.
---
 ...rolPlaneCoresidencySentinel.server.test.ts | 126 +++++++++++++++++-
 .../controlPlaneCoresidencySentinel.server.ts |  92 +++++++++----
 2 files changed, 192 insertions(+), 26 deletions(-)

diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.test.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.test.ts
index 37301f68743..d08f96529c9 100644
--- a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.test.ts
+++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.test.ts
@@ -34,6 +34,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: false,
       probe: async () => ({ coresident: "true" }),
       emit,
@@ -46,6 +47,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     await expect(
       assertControlPlaneCoresidencyAdvisory({
         ...urls,
+        shards: [],
         expectSplit: true,
         probe: async () => ({ coresident: "true" }),
         emit: vi.fn(),
@@ -58,6 +60,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: true,
       probe: async () => ({ coresident: "unknown", reason: "denied" }),
       emit,
@@ -71,6 +74,7 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const warn = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
       ...urls,
+      shards: [],
       expectSplit: true,
       probe: async () => {
         throw new Error("probe blew up");
@@ -86,9 +90,12 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     const emit = vi.fn();
     const probe = vi.fn();
     await assertControlPlaneCoresidencyAdvisory({
-      legacyUrl: undefined,
+      // "" and not undefined: ?? only guards nullish, so undefined would read the ambient
+      // RUN_OPS_LEGACY_DATABASE_URL and this test would depend on the developer's .env.
+      legacyUrl: "",
       controlPlaneUrl: "postgres://cp",
       expectSplit: true,
+      shards: [],
       probe,
       emit,
       log: noopLog,
@@ -97,3 +104,120 @@ describe("assertControlPlaneCoresidencyAdvisory", () => {
     expect(emit).not.toHaveBeenCalled();
   });
 });
+
+describe("assertControlPlaneCoresidencyAdvisory at N shards", () => {
+  const urls = { legacyUrl: "postgres://legacy", controlPlaneUrl: "postgres://cp" };
+  const shardA = { key: "a", url: "postgres://shard-a" };
+  const shardB = { key: "b", url: "postgres://shard-b" };
+
+  it("emits the legacy verdict with NO shard key, so today's series is unchanged", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(1);
+    expect(emit).toHaveBeenCalledWith("false");
+  });
+
+  it("emits one tagged verdict per shard, plus the untagged legacy verdict", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [shardA, shardB],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(3);
+    expect(emit).toHaveBeenCalledWith("false");
+    expect(emit).toHaveBeenCalledWith("false", "a");
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("probes each shard against the control plane", async () => {
+    const probe = vi.fn().mockResolvedValue({ coresident: "false" });
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: false,
+      shards: [shardA],
+      probe,
+      emit: vi.fn(),
+      log: noopLog,
+    });
+    expect(probe).toHaveBeenCalledWith("postgres://legacy", "postgres://cp", expect.anything());
+    expect(probe).toHaveBeenCalledWith("postgres://shard-a", "postgres://cp", expect.anything());
+  });
+
+  it("names the offending shard when enforcement is opted in and a shard is co-resident", async () => {
+    await expect(
+      assertControlPlaneCoresidencyAdvisory({
+        ...urls,
+        expectSplit: true,
+        shards: [shardA],
+        probe: async (url: string) =>
+          url === "postgres://shard-a"
+            ? ({ coresident: "true", reason: "same db" } as const)
+            : ({ coresident: "false" } as const),
+        emit: vi.fn(),
+        log: noopLog,
+      })
+    ).rejects.toThrow(/shard a/i);
+  });
+
+  it("emits every store before it throws, so no store loses its metric", async () => {
+    const emit = vi.fn();
+    await expect(
+      assertControlPlaneCoresidencyAdvisory({
+        ...urls,
+        expectSplit: true,
+        shards: [shardA, shardB],
+        probe: async (url: string) =>
+          url === "postgres://shard-a"
+            ? ({ coresident: "true", reason: "same db" } as const)
+            : ({ coresident: "false" } as const),
+        emit,
+        log: noopLog,
+      })
+    ).rejects.toThrow();
+    expect(emit).toHaveBeenCalledTimes(3);
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("degrades one shard's throwing probe to unknown and still reports the others", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      ...urls,
+      expectSplit: true,
+      shards: [shardA, shardB],
+      probe: async (url: string) => {
+        if (url === "postgres://shard-a") throw new Error("probe blew up");
+        return { coresident: "false" } as const;
+      },
+      emit,
+      log: { info: () => {}, warn: () => {} },
+    });
+    expect(emit).toHaveBeenCalledWith("unknown", "a");
+    expect(emit).toHaveBeenCalledWith("false", "b");
+  });
+
+  it("still probes the shards when there is no legacy DSN", async () => {
+    const emit = vi.fn();
+    await assertControlPlaneCoresidencyAdvisory({
+      legacyUrl: "",
+      controlPlaneUrl: "postgres://cp",
+      expectSplit: false,
+      shards: [shardA],
+      probe: async () => ({ coresident: "false" }),
+      emit,
+      log: noopLog,
+    });
+    expect(emit).toHaveBeenCalledTimes(1);
+    expect(emit).toHaveBeenCalledWith("false", "a");
+  });
+});
diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
index 1fb741b0e67..eab0fe1f034 100644
--- a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
@@ -12,6 +12,7 @@ import type { Counter } from "@opentelemetry/api";
 import { getMeter } from "@internal/tracing";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
+import { nonAliasedShards, type ShardTarget } from "~/v3/runOpsShards.server";
 import {
   probeControlPlaneCoresidency,
   type CoresidencyProbeResult,
@@ -39,12 +40,15 @@ export type CoresidencyEnforcement = { throw: false } | { throw: true; message:
 export function resolveCoresidencyEnforcement(args: {
   coresident: CoresidencyVerdict;
   expectSplit: boolean;
+  /** Omitted for the legacy store, so its message stays exactly as it was. */
+  shardKey?: string;
 }): CoresidencyEnforcement {
   if (args.expectSplit && args.coresident === "true") {
+    const store = args.shardKey === undefined ? "legacy run-ops DB" : `shard ${args.shardKey}`;
     return {
       throw: true,
       message:
-        "RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on but the legacy run-ops DB is still co-resident with the control-plane DB; refusing to start.",
+        `RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on but the ${store} is still co-resident with the control-plane DB; refusing to start.`,
     };
   }
   return { throw: false };
@@ -57,46 +61,84 @@ type AdvisoryLogger = {
 
 export async function assertControlPlaneCoresidencyAdvisory(deps?: {
   probe?: typeof probeControlPlaneCoresidency;
-  emit?: (verdict: CoresidencyVerdict) => void;
+  /** shardKey is omitted for the legacy store, so its metric series is unchanged at N=0. */
+  emit?: (verdict: CoresidencyVerdict, shardKey?: string) => void;
   log?: AdvisoryLogger;
   expectSplit?: boolean;
   legacyUrl?: string;
   controlPlaneUrl?: string;
+  shards?: ShardTarget[];
 }): Promise<void> {
   const log = deps?.log ?? logger;
   const legacyUrl = deps?.legacyUrl ?? env.RUN_OPS_LEGACY_DATABASE_URL;
   const controlPlaneUrl =
     deps?.controlPlaneUrl ?? env.CONTROL_PLANE_DATABASE_URL ?? env.DATABASE_URL;
-  // No legacy DSN (single-DB / self-host) or no control-plane DSN -> nothing to compare.
-  if (!legacyUrl || !controlPlaneUrl) return;
+  const shards = deps?.shards ?? nonAliasedShards(env.RUN_OPS_SHARDS);
+  // No control-plane DSN -> nothing to compare against, for any store.
+  if (!controlPlaneUrl) return;
+
+  // The legacy store carries NO shard key, so its metric series and its message are unchanged.
+  // An aliased shard is already absent from `shards`: it shares its target's database on purpose,
+  // so a co-residency verdict for it would duplicate its target's verdict.
+  const stores: Array<{ url: string; shardKey?: string }> = [
+    ...(legacyUrl ? [{ url: legacyUrl }] : []),
+    ...shards.map((shard) => ({ url: shard.url, shardKey: shard.key })),
+  ];
+  if (stores.length === 0) return;
 
   const probe = deps?.probe ?? probeControlPlaneCoresidency;
   const emit =
     deps?.emit ??
-    ((verdict: CoresidencyVerdict) => getCoresidentCounter().add(1, { result: verdict }));
+    ((verdict: CoresidencyVerdict, shardKey?: string) =>
+      getCoresidentCounter().add(
+        1,
+        shardKey === undefined ? { result: verdict } : { result: verdict, shard: shardKey }
+      ));
   const expectSplit = deps?.expectSplit ?? env.RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT;
 
-  let result: CoresidencyProbeResult;
-  try {
-    result = await probe(legacyUrl, controlPlaneUrl, { logger: log });
-  } catch (error) {
-    // Any unexpected throw still degrades to "unknown" — the advisory arm must never crash boot.
-    log.warn("run-ops control-plane co-residency probe threw; reporting unknown", { error });
-    result = { coresident: "unknown", reason: String(error) };
-  }
+  const results = await Promise.all(
+    stores.map(async (store) => {
+      let result: CoresidencyProbeResult;
+      try {
+        result = await probe(store.url, controlPlaneUrl, { logger: log });
+      } catch (error) {
+        // Any unexpected throw degrades THAT store to "unknown" — the advisory arm must never
+        // crash boot, and one store's denied probe must not hide another store's verdict.
+        log.warn("run-ops control-plane co-residency probe threw; reporting unknown", {
+          error,
+          shard: store.shardKey,
+        });
+        result = { coresident: "unknown", reason: String(error) };
+      }
+      return { store, result };
+    })
+  );
 
-  emit(result.coresident);
-  log.info("run_ops_legacy_control_plane_coresident", {
-    coresident: result.coresident,
-    reason: "reason" in result ? result.reason : undefined,
-    expectSplit,
-  });
+  // Emit and log EVERY store before any enforcement throw, so a failing store never costs
+  // another store its metric.
+  for (const { store, result } of results) {
+    // One argument for the legacy store, so its emission is byte-identical to today's.
+    if (store.shardKey === undefined) {
+      emit(result.coresident);
+    } else {
+      emit(result.coresident, store.shardKey);
+    }
+    log.info("run_ops_legacy_control_plane_coresident", {
+      coresident: result.coresident,
+      reason: "reason" in result ? result.reason : undefined,
+      expectSplit,
+      shard: store.shardKey,
+    });
+  }
 
-  const enforcement = resolveCoresidencyEnforcement({
-    coresident: result.coresident,
-    expectSplit,
-  });
-  if (enforcement.throw) {
-    throw new Error(enforcement.message);
+  for (const { store, result } of results) {
+    const enforcement = resolveCoresidencyEnforcement({
+      coresident: result.coresident,
+      expectSplit,
+      shardKey: store.shardKey,
+    });
+    if (enforcement.throw) {
+      throw new Error(enforcement.message);
+    }
   }
 }

From f8e9f4f834c59a77e7e29370c31b648c0c26dcc7 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:51:30 +0100
Subject: [PATCH 05/13] feat(webapp): warn per shard when a shard handle is not
 a distinct instance

The gate takes the shard replica handles and warns for any non-aliased shard
whose client is not distinct from the control-plane or gen-1 new client. The
returned verdict stays the gen-1 verdict: the distinctness sentinel already
fail-closes the boot on the same condition, and a gen-2 fault must not disable
the proven gen-1 read fan-out on top of that.

An aliased shard shares its target's client on purpose, so identity equality is
its correct state and never a fault.
---
 .../v3/runOpsMigration/runOpsSplitReadGate.ts | 23 +++++
 apps/webapp/test/runOpsSplitReadGate.test.ts  | 87 +++++++++++++++++++
 2 files changed, 110 insertions(+)

diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
index 0872a256508..89066e26bc6 100644
--- a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
+++ b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
@@ -8,6 +8,12 @@ export function computeRunOpsSplitReadEnabled(args: {
   controlPlaneReplica: unknown;
   hasNewUrl: boolean;
   hasLegacyUrl: boolean;
+  /**
+   * Gen-2 shard replica handles. Observability only: a non-distinct shard handle WARNS and never
+   * changes the returned verdict. The distinctness sentinel already fail-closes the boot on the
+   * same condition, and a gen-2 fault must not disable the proven gen-1 read fan-out as well.
+   */
+  shardHandles?: Array<{ key: string; replica: unknown; aliasOf?: "new" }>;
   logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void };
 }): boolean {
   const newIsDistinctDedicatedClient =
@@ -24,5 +30,22 @@ export function computeRunOpsSplitReadEnabled(args: {
     );
   }
 
+  // An aliased shard shares its target's client by reference, so identity equality is its correct
+  // state and never a fault. Keyed on the declared field, not on object identity.
+  for (const shard of args.shardHandles ?? []) {
+    if (shard.aliasOf !== undefined) continue;
+    if (
+      shard.replica === args.controlPlaneWriter ||
+      shard.replica === args.controlPlaneReplica ||
+      shard.replica === args.newReplica
+    ) {
+      args.logger?.warn(
+        `run-ops shard ${shard.key} declares its own database but its replica client is not a ` +
+          "distinct instance from the control-plane or gen-1 new client; reads for that shard " +
+          "would not reach its database."
+      );
+    }
+  }
+
   return enabled;
 }
diff --git a/apps/webapp/test/runOpsSplitReadGate.test.ts b/apps/webapp/test/runOpsSplitReadGate.test.ts
index 4deb0bb5329..0b87e13b905 100644
--- a/apps/webapp/test/runOpsSplitReadGate.test.ts
+++ b/apps/webapp/test/runOpsSplitReadGate.test.ts
@@ -165,3 +165,90 @@ describe("computeRunOpsSplitReadEnabled", () => {
     });
   });
 });
+
+describe("computeRunOpsSplitReadEnabled shard handles", () => {
+  const shardA = { __tag: "shard-a" };
+  const shardB = { __tag: "shard-b" };
+  const base = {
+    newReplica: dedicatedNew,
+    controlPlaneWriter: cpWriter,
+    controlPlaneReplica: cpReplica,
+    hasNewUrl: true,
+    hasLegacyUrl: true,
+  };
+
+  it("does not warn when every shard handle is a distinct instance", () => {
+    const warn = vi.fn();
+    const enabled = computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [
+        { key: "a", replica: shardA },
+        { key: "b", replica: shardB },
+      ],
+      logger: { warn },
+    });
+    expect(enabled).toBe(true);
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  it("warns, naming the shard, when a shard replica aliases a control-plane handle", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [{ key: "a", replica: cpReplica }],
+      logger: { warn },
+    });
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn.mock.calls[0][0]).toMatch(/shard a/i);
+  });
+
+  it("warns when a shard replica aliases the gen-1 new replica", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [{ key: "a", replica: dedicatedNew }],
+      logger: { warn },
+    });
+    expect(warn).toHaveBeenCalledTimes(1);
+  });
+
+  it("does NOT warn for an aliased shard, because sharing is its purpose", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [{ key: "z", replica: dedicatedNew, aliasOf: "new" as const }],
+      logger: { warn },
+    });
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  // The distinctness sentinel already fail-closes the boot on this condition. A gen-2 fault must
+  // not disable the proven gen-1 read fan-out on top of that.
+  it("keeps the gen-1 verdict when a shard handle is not distinct", () => {
+    expect(
+      computeRunOpsSplitReadEnabled({
+        ...base,
+        shardHandles: [{ key: "a", replica: cpReplica }],
+      })
+    ).toBe(true);
+  });
+
+  it("warns once per offending shard", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [
+        { key: "a", replica: cpReplica },
+        { key: "b", replica: cpWriter },
+      ],
+      logger: { warn },
+    });
+    expect(warn).toHaveBeenCalledTimes(2);
+  });
+
+  it("is unchanged when no shard handle is supplied", () => {
+    const warn = vi.fn();
+    expect(computeRunOpsSplitReadEnabled({ ...base, logger: { warn } })).toBe(true);
+    expect(warn).not.toHaveBeenCalled();
+  });
+});

From 8af907d0afdd11ad281075bb14a3aac5bfb9cc7f Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:56:23 +0100
Subject: [PATCH 06/13] feat(webapp): build one replication source per shard
 and refuse an uncovered shard

buildReplicationSources appends a source per shard that owns its own database,
each with the slot, publication and origin generation its descriptor declares.
An aliased shard takes no source: its target's slot already carries its WAL.

assertReplicationCoversSplit now also requires a source per non-aliased shard.
ShardReplicationMisconfiguredError subclasses the split error, so the boot catch
site reaches the same process.exit(1) it reaches today. A shard whose runs never
arrive in ClickHouse must not serve traffic.

RunsReplicationService is untouched. Its own check already rejects a duplicate
source id, slot name or origin generation.
---
 .../runsReplicationInstance.server.ts         |  94 +++++++++-
 .../test/runsReplicationInstance.test.ts      | 164 ++++++++++++++++++
 2 files changed, 249 insertions(+), 9 deletions(-)

diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts
index 164ce07fb92..b732d2c47f4 100644
--- a/apps/webapp/app/services/runsReplicationInstance.server.ts
+++ b/apps/webapp/app/services/runsReplicationInstance.server.ts
@@ -3,6 +3,7 @@ import { env } from "~/env.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
 import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server";
+import { nonAliasedShards } from "~/v3/runOpsShards.server";
 import { meter, provider } from "~/v3/tracer.server";
 import {
   setRunsReplicationConfiguredSources,
@@ -31,6 +32,15 @@ export function buildReplicationSources(args: {
   newSlotName: string;
   newPublicationName: string;
   newOriginGeneration: number;
+  /**
+   * Gen-2 shards that own their own database, each with its own slot, publication and origin
+   * generation. An aliased shard is absent: its target's slot already covers its WAL.
+   */
+  shards?: Array<{
+    key: string;
+    url: string;
+    replication: { slotName: string; publicationName: string; originGeneration: number };
+  }>;
 }): RunsReplicationSource[] {
   const legacy: RunsReplicationSource = {
     id: "legacy",
@@ -54,7 +64,25 @@ export function buildReplicationSources(args: {
     originGeneration: args.newOriginGeneration,
   };
 
-  return [legacy, next];
+  // Shard sources come after the gen-1 pair. Reached only when the new source is on, because
+  // split is the precondition for a shard to exist at all. The origin generations come from the
+  // descriptor, which the boot parser already bounds to 2..255 and checks for duplicates; the
+  // service re-checks uniqueness across every source it is given.
+  const shardSources: RunsReplicationSource[] = (args.shards ?? []).map((shard) => ({
+    id: shardSourceId(shard.key),
+    pgConnectionUrl: shard.url,
+    slotName: shard.replication.slotName,
+    publicationName: shard.replication.publicationName,
+    originGeneration: shard.replication.originGeneration,
+  }));
+
+  return [legacy, next, ...shardSources];
+}
+
+// The replication source id for a shard. It derives the client name and the leader-lock key, so it
+// must be stable and unique across sources.
+export function shardSourceId(key: string): string {
+  return `shard-${key}`;
 }
 
 /**
@@ -66,24 +94,54 @@ export function buildReplicationSources(args: {
  * rather than ship a fleet-wide under-count.
  */
 export class SplitReplicationMisconfiguredError extends Error {
-  constructor() {
+  constructor(message?: string) {
     super(
-      'RUN_OPS_SPLIT_ENABLED is on but the runs-replication sources[] has no "new" source: ' +
-        "run-ops runs on the new DB would not replicate to ClickHouse, under-counting every " +
-        "ClickHouse-fronted aggregate. Enable the new replication source " +
-        "(RUN_REPLICATION_NEW_ENABLED / RUN_REPLICATION_RUN_OPS_DATABASE_URL) or turn the split off."
+      message ??
+        'RUN_OPS_SPLIT_ENABLED is on but the runs-replication sources[] has no "new" source: ' +
+          "run-ops runs on the new DB would not replicate to ClickHouse, under-counting every " +
+          "ClickHouse-fronted aggregate. Enable the new replication source " +
+          "(RUN_REPLICATION_NEW_ENABLED / RUN_REPLICATION_RUN_OPS_DATABASE_URL) or turn the split off."
     );
     this.name = "SplitReplicationMisconfiguredError";
   }
 }
 
+/**
+ * A configured shard with no replication source of its own. Subclasses the split error on purpose:
+ * the boot catch site tests `instanceof SplitReplicationMisconfiguredError` to reach
+ * process.exit(1), and a shard whose runs never reach ClickHouse must take that same exit.
+ */
+export class ShardReplicationMisconfiguredError extends SplitReplicationMisconfiguredError {
+  constructor(shardKey: string) {
+    super(
+      `run-ops shard ${shardKey} is configured but the runs-replication sources[] has no ` +
+        `"${shardSourceId(shardKey)}" source: runs on that shard would not replicate to ` +
+        "ClickHouse, under-counting every ClickHouse-fronted aggregate. Give the shard a " +
+        "replication slot, publication and origin generation, or remove the shard."
+    );
+    this.name = "ShardReplicationMisconfiguredError";
+  }
+}
+
 export function assertReplicationCoversSplit(args: {
   splitEnabled: boolean;
   sources: RunsReplicationSource[];
+  /** Every configured shard, aliased ones included. An aliased shard needs no source of its own. */
+  shards?: Array<{ key: string; aliasOf?: "new" }>;
 }): void {
-  if (args.splitEnabled && !args.sources.some((s) => s.id === "new")) {
+  if (!args.splitEnabled) {
+    return;
+  }
+  if (!args.sources.some((s) => s.id === "new")) {
     throw new SplitReplicationMisconfiguredError();
   }
+  for (const shard of args.shards ?? []) {
+    // An aliased shard shares its target's database, so the target's slot already carries its WAL.
+    if (shard.aliasOf !== undefined) continue;
+    if (!args.sources.some((s) => s.id === shardSourceId(shard.key))) {
+      throw new ShardReplicationMisconfiguredError(shard.key);
+    }
+  }
 }
 
 function initializeRunsReplicationInstance() {
@@ -171,6 +229,18 @@ function initializeRunsReplicationInstance() {
     // The legacy-only instance above is never started in the dual path (no slot/lock
     // taken). runsReplicationService.server.ts is untouched. The create route also calls
     // setRunsReplicationGlobal — last-writer-wins is the existing contract.
+    // An aliased shard replicates through its target's slot, so only the shards that own their own
+    // database take a source. Coverage is then checked against EVERY descriptor, aliased included.
+    // The schema requires `replication` on every non-aliased descriptor, so the guard below is a
+    // type narrowing and not a policy.
+    const shardReplicationByKey = new Map(
+      env.RUN_OPS_SHARDS.flatMap((d) => (d.replication ? [[d.key, d.replication] as const] : []))
+    );
+    const shardsWithReplication = nonAliasedShards(env.RUN_OPS_SHARDS).flatMap((shard) => {
+      const replication = shardReplicationByKey.get(shard.key);
+      return replication ? [{ key: shard.key, url: shard.url, replication }] : [];
+    });
+
     isSplitEnabled()
       .then(async (splitEnabled) => {
         const sources = buildReplicationSources({
@@ -184,10 +254,16 @@ function initializeRunsReplicationInstance() {
           newSlotName: env.RUN_REPLICATION_NEW_SLOT_NAME,
           newPublicationName: env.RUN_REPLICATION_NEW_PUBLICATION_NAME,
           newOriginGeneration: env.RUN_REPLICATION_NEW_ORIGIN_GENERATION,
+          shards: shardsWithReplication,
         });
 
-        // Refuse to start replication if split is on but `#new` is not a source.
-        assertReplicationCoversSplit({ splitEnabled, sources });
+        // Refuse to start replication if split is on but `#new` is not a source, or if any shard
+        // that owns its own database has no source of its own.
+        assertReplicationCoversSplit({
+          splitEnabled,
+          sources,
+          shards: env.RUN_OPS_SHARDS.map((d) => ({ key: d.key, aliasOf: d.aliasOf })),
+        });
 
         if (sources.length > 1) {
           // Release the bootstrap instance's eager replication client (Redis + Redlock)
diff --git a/apps/webapp/test/runsReplicationInstance.test.ts b/apps/webapp/test/runsReplicationInstance.test.ts
index 67f595c597c..f9f1716028b 100644
--- a/apps/webapp/test/runsReplicationInstance.test.ts
+++ b/apps/webapp/test/runsReplicationInstance.test.ts
@@ -202,6 +202,170 @@ describe("assertReplicationCoversSplit (boot gate-coupling)", () => {
   });
 });
 
+describe("replication sources at N shards", () => {
+  const baseArgs = {
+    legacyUrl: "postgres://legacy",
+    legacySlotName: "task_runs_to_clickhouse_v1",
+    legacyPublicationName: "task_runs_to_clickhouse_v1_publication",
+    legacyOriginGeneration: 0,
+    newSlotName: "task_runs_to_clickhouse_v2",
+    newPublicationName: "task_runs_to_clickhouse_v2_publication",
+    newOriginGeneration: 1,
+    splitEnabled: true,
+    newUrl: "postgres://new",
+  };
+
+  const shardA = {
+    key: "a",
+    url: "postgres://shard-a",
+    replication: {
+      slotName: "task_runs_to_clickhouse_shard_a",
+      publicationName: "task_runs_to_clickhouse_shard_a_publication",
+      originGeneration: 2,
+    },
+  };
+  const shardB = {
+    key: "b",
+    url: "postgres://shard-b",
+    replication: {
+      slotName: "task_runs_to_clickhouse_shard_b",
+      publicationName: "task_runs_to_clickhouse_shard_b_publication",
+      originGeneration: 3,
+    },
+  };
+
+  it("appends nothing when no shard is configured", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [] });
+    expect(sources.map((s) => s.id)).toEqual(["legacy", "new"]);
+  });
+
+  it("appends one source per shard, after legacy and new", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA, shardB] });
+    expect(sources.map((s) => s.id)).toEqual(["legacy", "new", "shard-a", "shard-b"]);
+    expect(sources[2]).toEqual({
+      id: "shard-a",
+      pgConnectionUrl: "postgres://shard-a",
+      slotName: "task_runs_to_clickhouse_shard_a",
+      publicationName: "task_runs_to_clickhouse_shard_a_publication",
+      originGeneration: 2,
+    });
+  });
+
+  it("gives each shard its own origin generation, between 2 and 255", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA, shardB] });
+    const gens = sources.map((s) => s.originGeneration);
+    expect(new Set(gens).size).toBe(gens.length);
+    for (const gen of gens.slice(2)) {
+      expect(gen).toBeGreaterThanOrEqual(2);
+      expect(gen).toBeLessThanOrEqual(255);
+    }
+  });
+
+  it("appends no shard source when the new source is off, because split is the precondition", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      splitEnabled: false,
+      shards: [shardA],
+    });
+    expect(sources.map((s) => s.id)).toEqual(["legacy"]);
+  });
+
+  it("throws when a shard that owns its database has no source", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources,
+        shards: [{ key: "a" }],
+      })
+    ).toThrow(SplitReplicationMisconfiguredError);
+  });
+
+  it("names the uncovered shard in the message", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources,
+        shards: [{ key: "a" }, { key: "b" }],
+      })
+    ).toThrow(/shard b/i);
+  });
+
+  it("does NOT throw when every shard has its own source", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA, shardB] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources,
+        shards: [{ key: "a" }, { key: "b" }],
+      })
+    ).not.toThrow();
+  });
+
+  it("does NOT require a source for an aliased shard, because its target's slot covers it", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources,
+        shards: [{ key: "z", aliasOf: "new" }],
+      })
+    ).not.toThrow();
+  });
+
+  it("does NOT check shard coverage when split is off", () => {
+    const sources = buildReplicationSources({ ...baseArgs, splitEnabled: false, shards: [] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: false,
+        sources,
+        shards: [{ key: "a" }],
+      })
+    ).not.toThrow();
+  });
+
+  // The catch site keys on `instanceof SplitReplicationMisconfiguredError` to reach
+  // process.exit(1). A shard with no replication must reach the same exit.
+  it("raises an error the existing exit path recognizes", () => {
+    try {
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources: buildReplicationSources({ ...baseArgs, shards: [] }),
+        shards: [{ key: "a" }],
+      });
+      expect.unreachable("expected a throw");
+    } catch (error) {
+      expect(error).toBeInstanceOf(SplitReplicationMisconfiguredError);
+    }
+  });
+
+  // The service validates sources before it builds a single replication client, so this needs no
+  // container. RunsReplicationService itself is untouched by this change: the check already exists.
+  it("rejects two shards that share an origin generation, via the service's own check", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      shards: [shardA, { ...shardB, replication: { ...shardB.replication, originGeneration: 2 } }],
+    });
+
+    expect(
+      () =>
+        new RunsReplicationService({
+          clickhouseFactory: new TestReplicationClickhouseFactory(
+            new ClickHouse({ url: "http://127.0.0.1:1", name: "unused", logLevel: "warn" })
+          ),
+          serviceName: "runs-replication",
+          pgConnectionUrl: "postgres://legacy",
+          slotName: "unused",
+          publicationName: "unused",
+          redisOptions: { host: "127.0.0.1", port: 1 },
+          sources,
+          logLevel: "warn",
+        })
+    ).toThrow(/duplicate originGeneration/i);
+  });
+});
+
 describe("RunsReplication new-source backfill origin generation (integration)", () => {
   replicationContainerTest(
     "backfill via the new source tags the ClickHouse row with the new origin generation (gen=1), not gen=0",

From 7afd2a167bcdc7ec6d90d65cb69c3eef54b5b562 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:56:23 +0100
Subject: [PATCH 07/13] feat(docker): list the migration DSN of every shard
 that owns its own database

The runner image has no jq, so the entrypoint cannot parse RUN_OPS_SHARDS on its
own. This script prints one DSN per line and takes a unit test, which an inline
node -e string could not.

An unset or blank variable prints nothing, so a single-database install is
unaffected. Invalid JSON exits 1, so a malformed descriptor stops the container
before the migrations run rather than after.
---
 apps/webapp/test/runOpsShardDsns.test.ts | 63 ++++++++++++++++++++++++
 docker/scripts/runOpsShardDsns.mjs       | 60 ++++++++++++++++++++++
 2 files changed, 123 insertions(+)
 create mode 100644 apps/webapp/test/runOpsShardDsns.test.ts
 create mode 100644 docker/scripts/runOpsShardDsns.mjs

diff --git a/apps/webapp/test/runOpsShardDsns.test.ts b/apps/webapp/test/runOpsShardDsns.test.ts
new file mode 100644
index 00000000000..02f3be673fa
--- /dev/null
+++ b/apps/webapp/test/runOpsShardDsns.test.ts
@@ -0,0 +1,63 @@
+import { describe, expect, it } from "vitest";
+// The entrypoint calls this file with plain `node`, so it takes no path alias and no bundler.
+import { shardMigrationDsns } from "../../../docker/scripts/runOpsShardDsns.mjs";
+
+const shardA = {
+  key: "a",
+  region: "us-east-1",
+  url: "postgres://h/a",
+  replication: { slotName: "sa", publicationName: "pa", originGeneration: 2 },
+};
+
+describe("shardMigrationDsns", () => {
+  it("returns nothing when the variable is unset", () => {
+    expect(shardMigrationDsns(undefined)).toEqual([]);
+  });
+
+  it("returns nothing when the variable is blank", () => {
+    expect(shardMigrationDsns("   ")).toEqual([]);
+  });
+
+  it("returns nothing for an empty array", () => {
+    expect(shardMigrationDsns("[]")).toEqual([]);
+  });
+
+  it("throws on invalid JSON, so the entrypoint stops before it migrates", () => {
+    expect(() => shardMigrationDsns("{not json")).toThrow(/not valid JSON/i);
+  });
+
+  it("throws when the value is JSON but not an array", () => {
+    expect(() => shardMigrationDsns('{"key":"a"}')).toThrow(/not a JSON array/i);
+  });
+
+  it("returns the url of a shard that owns its own database", () => {
+    expect(shardMigrationDsns(JSON.stringify([shardA]))).toEqual(["postgres://h/a"]);
+  });
+
+  it("prefers directUrl over url, because migrations must not go through a pooler", () => {
+    const withDirect = { ...shardA, directUrl: "postgres://h/a-direct" };
+    expect(shardMigrationDsns(JSON.stringify([withDirect]))).toEqual(["postgres://h/a-direct"]);
+  });
+
+  it("skips an aliased shard, because its target's invocation already migrates it", () => {
+    const aliased = { key: "z", region: "us-east-1", aliasOf: "new" };
+    expect(shardMigrationDsns(JSON.stringify([shardA, aliased]))).toEqual(["postgres://h/a"]);
+  });
+
+  it("skips a descriptor with neither url nor directUrl", () => {
+    const noUrl = { key: "b", region: "us-east-1" };
+    expect(shardMigrationDsns(JSON.stringify([shardA, noUrl]))).toEqual(["postgres://h/a"]);
+  });
+
+  it("keeps declaration order across several shards", () => {
+    const shardB = { ...shardA, key: "b", url: "postgres://h/b" };
+    expect(shardMigrationDsns(JSON.stringify([shardA, shardB]))).toEqual([
+      "postgres://h/a",
+      "postgres://h/b",
+    ]);
+  });
+
+  it("throws when an entry is not an object", () => {
+    expect(() => shardMigrationDsns('["postgres://h/a"]')).toThrow(/not an object/i);
+  });
+});
diff --git a/docker/scripts/runOpsShardDsns.mjs b/docker/scripts/runOpsShardDsns.mjs
new file mode 100644
index 00000000000..384e543e996
--- /dev/null
+++ b/docker/scripts/runOpsShardDsns.mjs
@@ -0,0 +1,60 @@
+// Print the migration DSN of every run-ops shard that owns its own database, one per line, so
+// entrypoint.sh can loop over them. The runner image has no `jq`, and this script is unit-tested,
+// which an inline `node -e` string could not be.
+//
+// Contract:
+//   RUN_OPS_SHARDS unset or blank -> print nothing, exit 0 (single-DB and gen-1-only installs)
+//   invalid JSON, or not an array -> message on stderr, exit 1 (the app rejects the same value)
+//   a descriptor with `aliasOf`    -> skipped; it shares its target's database
+//   the DSN                        -> `directUrl` if set, else `url`; skipped if neither is set
+//
+// Never print a DSN to stderr or to a log: stdout is consumed by the caller, nothing else.
+
+export function shardMigrationDsns(raw) {
+  if (raw === undefined || raw === null || String(raw).trim() === "") {
+    return [];
+  }
+
+  let parsed;
+  try {
+    parsed = JSON.parse(raw);
+  } catch {
+    throw new Error("RUN_OPS_SHARDS is not valid JSON");
+  }
+
+  if (!Array.isArray(parsed)) {
+    throw new Error("RUN_OPS_SHARDS is not a JSON array");
+  }
+
+  const dsns = [];
+  for (const descriptor of parsed) {
+    if (descriptor === null || typeof descriptor !== "object") {
+      throw new Error("RUN_OPS_SHARDS holds an entry that is not an object");
+    }
+    // An aliased shard is the same database as its target, which is migrated by its own invocation.
+    if (descriptor.aliasOf !== undefined && descriptor.aliasOf !== null) {
+      continue;
+    }
+    const dsn = descriptor.directUrl ?? descriptor.url;
+    if (typeof dsn !== "string" || dsn === "") {
+      continue;
+    }
+    dsns.push(dsn);
+  }
+  return dsns;
+}
+
+// `import.meta.main` is not available on every supported node, so compare argv instead.
+const invokedDirectly =
+  process.argv[1] !== undefined && process.argv[1].endsWith("runOpsShardDsns.mjs");
+
+if (invokedDirectly) {
+  try {
+    for (const dsn of shardMigrationDsns(process.env.RUN_OPS_SHARDS)) {
+      process.stdout.write(`${dsn}\n`);
+    }
+  } catch (error) {
+    process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+    process.exit(1);
+  }
+}

From 16165ce18ed0bb2aba351ba370e208217e052098 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 16:57:52 +0100
Subject: [PATCH 08/13] feat(docker): migrate every run-ops shard that owns its
 own database

The two hardcoded run-ops invocations gain a loop over the shard DSNs. Each
shard runs the identical schema, so this is the existing migrations against a
new DSN, and @internal/run-ops-database needs no change.

A for loop and not a while-read pipeline: a pipeline subshell would swallow a
failed migration on any iteration but the last, so a broken shard would boot.
Tracing stays off across the capture and the loop, because set -x prints an
assignment and the DSN carries credentials.

Installs that never set RUN_OPS_SHARDS skip the block entirely.
---
 docker/scripts/entrypoint.sh | 38 ++++++++++++++++++++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/docker/scripts/entrypoint.sh b/docker/scripts/entrypoint.sh
index 1e5c7c7cab0..4d9899f2732 100755
--- a/docker/scripts/entrypoint.sh
+++ b/docker/scripts/entrypoint.sh
@@ -49,6 +49,44 @@ else
   echo "RUN_OPS_LEGACY_DIRECT_URL not set, skipping legacy run-ops migrations."
 fi
 
+# Run-ops shards: migrate every gen-2 shard that owns its own database. Each shard runs the
+# identical schema, so this is the existing run-ops migrations against a new DSN. An aliased shard is
+# skipped by the DSN script: it IS its target's database. Installs that never set RUN_OPS_SHARDS
+# skip this entirely.
+{ set +x; } 2>/dev/null
+if [ -n "$RUN_OPS_SHARDS" ]; then
+  set -x
+  if [ "$SKIP_RUN_OPS_SHARD_MIGRATIONS" != "1" ]; then
+    echo "Running run-ops shard migrations"
+    # Tracing stays OFF from here to the end of the loop: `set -x` prints an assignment, so
+    # capturing a DSN under tracing would put the credentials in the logs.
+    { set +x; } 2>/dev/null
+    # A malformed descriptor exits 1 here, so the container stops before it migrates anything.
+    shard_dsns=$(node scripts/runOpsShardDsns.mjs)
+    # A `for` loop and NOT `... | while read`: a pipeline subshell would swallow a failed migration
+    # on any iteration but the last. Here `set -e` stops the boot on the first shard that fails.
+    # IFS is newline-only so a DSN is never split on other whitespace, and `set -f` stops a DSN
+    # query string (it holds `?`) from being read as a glob pattern.
+    old_ifs=$IFS
+    IFS='
+'
+    set -f
+    for shard_dsn in $shard_dsns; do
+      # Subshell with tracing off so `set -x` does not print the DSN (with credentials) to the logs.
+      (set +x; RUN_OPS_DATABASE_URL="$shard_dsn" DIRECT_URL="$shard_dsn" pnpm --filter @internal/run-ops-database db:migrate:deploy)
+    done
+    set +f
+    IFS=$old_ifs
+    set -x
+    echo "Run-ops shard migrations done"
+  else
+    echo "SKIP_RUN_OPS_SHARD_MIGRATIONS=1, skipping run-ops shard migrations."
+  fi
+else
+  set -x
+  echo "RUN_OPS_SHARDS not set, skipping run-ops shard migrations."
+fi
+
 if [ "$SKIP_DASHBOARD_AGENT_MIGRATIONS" != "1" ]; then
   echo "Running dashboard agent migrations"
   pnpm --filter @internal/dashboard-agent-db db:migrate:deploy

From 04cc56ac8ff93ab4cc87e84a9b5eef525008a7b4 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 17:00:00 +0100
Subject: [PATCH 09/13] feat(webapp): give the read gate the shard handles and
 the replication builder the descriptors

The read gate receives one handle per shard, with the declared aliasOf, so a
shard whose client is not distinct warns. The verdict it returns is unchanged.

The replication instance passes the shard descriptors to the source builder and
to the coverage check. The distinctness sentinel and the co-residency advisory
read the descriptors themselves, so the boot order is unchanged: the probe still
runs where it ran, with a wider target list.
---
 apps/webapp/app/db.server.ts                             | 9 +++++++++
 .../app/services/runsReplicationInstance.server.ts       | 4 ++--
 .../controlPlaneCoresidencySentinel.server.ts            | 3 +--
 3 files changed, 12 insertions(+), 4 deletions(-)

diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts
index 11a8aa307e9..40feafa3da5 100644
--- a/apps/webapp/app/db.server.ts
+++ b/apps/webapp/app/db.server.ts
@@ -594,6 +594,15 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   controlPlaneReplica: $replica,
   hasNewUrl: !!env.RUN_OPS_DATABASE_URL,
   hasLegacyUrl: !!env.RUN_OPS_LEGACY_DATABASE_URL,
+  // Observability only: a non-distinct shard handle warns and never changes the gen-1 verdict.
+  // Empty unless RUN_OPS_SHARDS is configured.
+  shardHandles: runOpsShardHandles.map((handle) => ({
+    key: handle.key,
+    replica: handle.replica,
+    // The DECLARED field, not client identity: an aliased shard shares its target's client by
+    // reference, so identity comparison cannot tell the two apart.
+    aliasOf: env.RUN_OPS_SHARDS.find((d) => d.key === handle.key)?.aliasOf,
+  })),
   logger,
 });
 
diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts
index b732d2c47f4..6244cccbcdf 100644
--- a/apps/webapp/app/services/runsReplicationInstance.server.ts
+++ b/apps/webapp/app/services/runsReplicationInstance.server.ts
@@ -81,7 +81,7 @@ export function buildReplicationSources(args: {
 
 // The replication source id for a shard. It derives the client name and the leader-lock key, so it
 // must be stable and unique across sources.
-export function shardSourceId(key: string): string {
+function shardSourceId(key: string): string {
   return `shard-${key}`;
 }
 
@@ -111,7 +111,7 @@ export class SplitReplicationMisconfiguredError extends Error {
  * the boot catch site tests `instanceof SplitReplicationMisconfiguredError` to reach
  * process.exit(1), and a shard whose runs never reach ClickHouse must take that same exit.
  */
-export class ShardReplicationMisconfiguredError extends SplitReplicationMisconfiguredError {
+class ShardReplicationMisconfiguredError extends SplitReplicationMisconfiguredError {
   constructor(shardKey: string) {
     super(
       `run-ops shard ${shardKey} is configured but the runs-replication sources[] has no ` +
diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
index eab0fe1f034..870f51c75bc 100644
--- a/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneCoresidencySentinel.server.ts
@@ -47,8 +47,7 @@ export function resolveCoresidencyEnforcement(args: {
     const store = args.shardKey === undefined ? "legacy run-ops DB" : `shard ${args.shardKey}`;
     return {
       throw: true,
-      message:
-        `RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on but the ${store} is still co-resident with the control-plane DB; refusing to start.`,
+      message: `RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on but the ${store} is still co-resident with the control-plane DB; refusing to start.`,
     };
   }
   return { throw: false };

From a684713ee38158568a760b1b104998db06d65646 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 17:14:42 +0100
Subject: [PATCH 10/13] fix(webapp): refuse a replication identity collision at
 the fatal boot gate

The descriptor parser checks uniqueness among shards only, so a shard could take
the slot name, publication name or origin generation of the legacy or the new
source. The service has its own check, but it throws from the constructor, which
the caller reaches only after it has shut the bootstrap instance down. That left
the process up with NO replication at all, legacy included, behind one
console.error. It is the exact silent ClickHouse under-count this family of
errors exists to prevent.

The check now runs in assertReplicationCoversSplit, before anything is torn
down, and raises a subclass the existing catch site already recognizes.

A correct deployment satisfies this today, because two consumers on one slot is
a data race that cannot work.
---
 .../runsReplicationInstance.server.ts         | 37 ++++++++-
 .../test/runsReplicationInstance.test.ts      | 75 ++++++++++++++++---
 2 files changed, 100 insertions(+), 12 deletions(-)

diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts
index 6244cccbcdf..750a8fbf789 100644
--- a/apps/webapp/app/services/runsReplicationInstance.server.ts
+++ b/apps/webapp/app/services/runsReplicationInstance.server.ts
@@ -79,8 +79,9 @@ export function buildReplicationSources(args: {
   return [legacy, next, ...shardSources];
 }
 
-// The replication source id for a shard. It derives the client name and the leader-lock key, so it
-// must be stable and unique across sources.
+// The replication source id for a shard. It derives the per-source client name and the key the
+// status route probes, so it must be stable and unique across sources. The leader lock is keyed on
+// the slot name, not on this id.
 function shardSourceId(key: string): string {
   return `shard-${key}`;
 }
@@ -106,6 +107,25 @@ export class SplitReplicationMisconfiguredError extends Error {
   }
 }
 
+/**
+ * Two sources that share an identity. The descriptor parser checks uniqueness AMONG shards only, so
+ * it cannot see the env-configured legacy and new sources: a shard can collide with either. The
+ * service has its own check, but it throws from the constructor, which the caller reaches only AFTER
+ * it has shut the bootstrap instance down — leaving the process up with NO replication at all, which
+ * is the exact silent under-count this family of errors exists to prevent. So the check runs here,
+ * at the fatal gate, before anything is torn down.
+ */
+class DuplicateReplicationIdentityError extends SplitReplicationMisconfiguredError {
+  constructor(field: string, value: unknown) {
+    super(
+      `the runs-replication sources[] has two sources with the same ${field} "${String(value)}": ` +
+        "two consumers on one WAL stream is a data race, and a shared origin generation defeats the " +
+        "ClickHouse dedup tiebreak. Give every source its own slot, publication and origin generation."
+    );
+    this.name = "DuplicateReplicationIdentityError";
+  }
+}
+
 /**
  * A configured shard with no replication source of its own. Subclasses the split error on purpose:
  * the boot catch site tests `instanceof SplitReplicationMisconfiguredError` to reach
@@ -142,6 +162,19 @@ export function assertReplicationCoversSplit(args: {
       throw new ShardReplicationMisconfiguredError(shard.key);
     }
   }
+
+  // Cross-source identity, over EVERY source and not only the shards. A correct two-source
+  // deployment already satisfies this, because two consumers on one WAL slot is a data race that
+  // cannot work. So this adds a loud failure for a configuration that was already broken silently.
+  for (const field of ["id", "slotName", "publicationName", "originGeneration"] as const) {
+    const seen = new Set<unknown>();
+    for (const source of args.sources) {
+      if (seen.has(source[field])) {
+        throw new DuplicateReplicationIdentityError(field, source[field]);
+      }
+      seen.add(source[field]);
+    }
+  }
 }
 
 function initializeRunsReplicationInstance() {
diff --git a/apps/webapp/test/runsReplicationInstance.test.ts b/apps/webapp/test/runsReplicationInstance.test.ts
index f9f1716028b..edc2f34260f 100644
--- a/apps/webapp/test/runsReplicationInstance.test.ts
+++ b/apps/webapp/test/runsReplicationInstance.test.ts
@@ -251,16 +251,6 @@ describe("replication sources at N shards", () => {
     });
   });
 
-  it("gives each shard its own origin generation, between 2 and 255", () => {
-    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA, shardB] });
-    const gens = sources.map((s) => s.originGeneration);
-    expect(new Set(gens).size).toBe(gens.length);
-    for (const gen of gens.slice(2)) {
-      expect(gen).toBeGreaterThanOrEqual(2);
-      expect(gen).toBeLessThanOrEqual(255);
-    }
-  });
-
   it("appends no shard source when the new source is off, because split is the precondition", () => {
     const sources = buildReplicationSources({
       ...baseArgs,
@@ -340,6 +330,71 @@ describe("replication sources at N shards", () => {
     }
   });
 
+  // F1 class: the descriptor parser checks uniqueness AMONG shards only. It cannot see the
+  // env-configured legacy and new sources, so a shard can collide with them. The service's own
+  // check throws too late: the caller has already shut the bootstrap instance down, so the throw
+  // leaves the process up with NO replication at all. These must fail at the fatal boot gate.
+  it("throws when a shard's slot name collides with the gen-1 new slot", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      shards: [
+        { ...shardA, replication: { ...shardA.replication, slotName: baseArgs.newSlotName } },
+      ],
+    });
+    expect(() =>
+      assertReplicationCoversSplit({ splitEnabled: true, sources, shards: [{ key: "a" }] })
+    ).toThrow(SplitReplicationMisconfiguredError);
+  });
+
+  it("throws when a shard's origin generation collides with the gen-1 new generation", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      newOriginGeneration: 2,
+      shards: [shardA],
+    });
+    expect(() =>
+      assertReplicationCoversSplit({ splitEnabled: true, sources, shards: [{ key: "a" }] })
+    ).toThrow(SplitReplicationMisconfiguredError);
+  });
+
+  it("throws when a shard's publication name collides with the legacy publication", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      shards: [
+        {
+          ...shardA,
+          replication: { ...shardA.replication, publicationName: baseArgs.legacyPublicationName },
+        },
+      ],
+    });
+    expect(() =>
+      assertReplicationCoversSplit({ splitEnabled: true, sources, shards: [{ key: "a" }] })
+    ).toThrow(SplitReplicationMisconfiguredError);
+  });
+
+  it("names the colliding field in the message", () => {
+    const sources = buildReplicationSources({
+      ...baseArgs,
+      shards: [
+        { ...shardA, replication: { ...shardA.replication, slotName: baseArgs.newSlotName } },
+      ],
+    });
+    expect(() =>
+      assertReplicationCoversSplit({ splitEnabled: true, sources, shards: [{ key: "a" }] })
+    ).toThrow(/slotName/);
+  });
+
+  it("does NOT throw when every shard's slot, publication and generation are its own", () => {
+    const sources = buildReplicationSources({ ...baseArgs, shards: [shardA, shardB] });
+    expect(() =>
+      assertReplicationCoversSplit({
+        splitEnabled: true,
+        sources,
+        shards: [{ key: "a" }, { key: "b" }],
+      })
+    ).not.toThrow();
+  });
+
   // The service validates sources before it builds a single replication client, so this needs no
   // container. RunsReplicationService itself is untouched by this change: the check already exists.
   it("rejects two shards that share an origin generation, via the service's own check", () => {

From ca8e79375a4958cf437e14a5c6248ca9322911cb Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 17:14:42 +0100
Subject: [PATCH 11/13] fix(webapp): warn about the read-gate case a shard can
 actually reach

A shard with no replicaUrl takes its own writer as its replica handle, so its
reads go to its primary. That is the per-shard analogue of the existing legacy
warning, and it is reachable today.

The control-plane identity check stays as a regression guard, now marked as
unreachable by construction: a non-aliased shard always gets a freshly built
client. It exists so a future control-plane fallback for shards cannot silently
route a shard's reads to another database.
---
 apps/webapp/app/db.server.ts                  |  1 +
 .../v3/runOpsMigration/runOpsSplitReadGate.ts | 22 +++++++--
 apps/webapp/test/runOpsSplitReadGate.test.ts  | 46 +++++++++++++++++++
 3 files changed, 65 insertions(+), 4 deletions(-)

diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts
index 40feafa3da5..d4f3bd3fa88 100644
--- a/apps/webapp/app/db.server.ts
+++ b/apps/webapp/app/db.server.ts
@@ -598,6 +598,7 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   // Empty unless RUN_OPS_SHARDS is configured.
   shardHandles: runOpsShardHandles.map((handle) => ({
     key: handle.key,
+    writer: handle.writer,
     replica: handle.replica,
     // The DECLARED field, not client identity: an aliased shard shares its target's client by
     // reference, so identity comparison cannot tell the two apart.
diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
index 89066e26bc6..e70dc29f449 100644
--- a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
+++ b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts
@@ -9,11 +9,11 @@ export function computeRunOpsSplitReadEnabled(args: {
   hasNewUrl: boolean;
   hasLegacyUrl: boolean;
   /**
-   * Gen-2 shard replica handles. Observability only: a non-distinct shard handle WARNS and never
-   * changes the returned verdict. The distinctness sentinel already fail-closes the boot on the
-   * same condition, and a gen-2 fault must not disable the proven gen-1 read fan-out as well.
+   * Gen-2 shard handles. Observability only: a non-distinct shard handle WARNS and never changes the
+   * returned verdict. A gen-2 fault must not disable the proven gen-1 read fan-out, and the
+   * distinctness sentinel already fail-closes the boot when two stores share a database.
    */
-  shardHandles?: Array<{ key: string; replica: unknown; aliasOf?: "new" }>;
+  shardHandles?: Array<{ key: string; writer?: unknown; replica: unknown; aliasOf?: "new" }>;
   logger?: { warn: (msg: string, meta?: Record<string, unknown>) => void };
 }): boolean {
   const newIsDistinctDedicatedClient =
@@ -34,6 +34,20 @@ export function computeRunOpsSplitReadEnabled(args: {
   // state and never a fault. Keyed on the declared field, not on object identity.
   for (const shard of args.shardHandles ?? []) {
     if (shard.aliasOf !== undefined) continue;
+
+    // A shard with no replica URL takes its own writer as its replica handle, so its reads go to
+    // its primary. This is the per-shard analogue of the existing legacy-primary warning.
+    if (shard.writer !== undefined && shard.replica === shard.writer) {
+      args.logger?.warn(
+        `run-ops shard ${shard.key} has no read replica handle; reads for that shard will hit the ` +
+          "shard primary. Set the shard's replicaUrl to keep replica reads off its primary."
+      );
+      continue;
+    }
+
+    // Unreachable by construction today: a non-aliased shard always gets a freshly built client.
+    // Kept as a regression guard, so a future control-plane fallback for shards cannot silently
+    // route a shard's reads to another database.
     if (
       shard.replica === args.controlPlaneWriter ||
       shard.replica === args.controlPlaneReplica ||
diff --git a/apps/webapp/test/runOpsSplitReadGate.test.ts b/apps/webapp/test/runOpsSplitReadGate.test.ts
index 0b87e13b905..430abbff859 100644
--- a/apps/webapp/test/runOpsSplitReadGate.test.ts
+++ b/apps/webapp/test/runOpsSplitReadGate.test.ts
@@ -233,6 +233,52 @@ describe("computeRunOpsSplitReadEnabled shard handles", () => {
     ).toBe(true);
   });
 
+  // The reachable case. A shard with no replicaUrl gets its own WRITER as its replica handle, so its
+  // reads go to its primary. selectRunOpsTopology does exactly that (db.server.ts), which makes this
+  // the per-shard analogue of the existing legacy "reads will hit the legacy primary" warning.
+  it("warns when a shard has no distinct replica handle, so its reads hit its primary", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [{ key: "a", writer: shardA, replica: shardA }],
+      logger: { warn },
+    });
+    expect(warn).toHaveBeenCalledTimes(1);
+    expect(warn.mock.calls[0][0]).toMatch(/shard a/i);
+    expect(warn.mock.calls[0][0]).toMatch(/primary/i);
+  });
+
+  it("does not warn when a shard has its own distinct replica handle", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [{ key: "a", writer: shardA, replica: shardB }],
+      logger: { warn },
+    });
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  it("does NOT warn about primary reads for an aliased shard", () => {
+    const warn = vi.fn();
+    computeRunOpsSplitReadEnabled({
+      ...base,
+      shardHandles: [
+        { key: "z", writer: dedicatedNew, replica: dedicatedNew, aliasOf: "new" as const },
+      ],
+      logger: { warn },
+    });
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  it("keeps the gen-1 verdict when a shard reads from its primary", () => {
+    expect(
+      computeRunOpsSplitReadEnabled({
+        ...base,
+        shardHandles: [{ key: "a", writer: shardA, replica: shardA }],
+      })
+    ).toBe(true);
+  });
+
   it("warns once per offending shard", () => {
     const warn = vi.fn();
     computeRunOpsSplitReadEnabled({

From d84017601f4c66a4e95bd3334b540879e70e1af1 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 17:14:42 +0100
Subject: [PATCH 12/13] fix(docker): reject a shard DSN holding a line break,
 and scope the loop's shell options

One DSN per line is the protocol between the script and the entrypoint, and the
URL parser strips ASCII line breaks, so a DSN holding one would split into two
bogus DSNs with nothing upstream to reject it.

The loop now runs in a subshell, so its IFS and noglob changes need no restore
and cannot leak into the rest of the entrypoint.
---
 apps/webapp/test/runOpsShardDsns.test.ts | 14 ++++++++++++++
 docker/scripts/entrypoint.sh             | 22 +++++++++++-----------
 docker/scripts/runOpsShardDsns.mjs       |  5 +++++
 3 files changed, 30 insertions(+), 11 deletions(-)

diff --git a/apps/webapp/test/runOpsShardDsns.test.ts b/apps/webapp/test/runOpsShardDsns.test.ts
index 02f3be673fa..73c6601a6e6 100644
--- a/apps/webapp/test/runOpsShardDsns.test.ts
+++ b/apps/webapp/test/runOpsShardDsns.test.ts
@@ -61,3 +61,17 @@ describe("shardMigrationDsns", () => {
     expect(() => shardMigrationDsns('["postgres://h/a"]')).toThrow(/not an object/i);
   });
 });
+
+describe("shardMigrationDsns line protocol", () => {
+  // One DSN per line is the protocol with entrypoint.sh, so a line break would split one DSN into
+  // two bogus ones. The URL parser strips ASCII line breaks, so nothing upstream rejects this.
+  it("throws when a DSN holds a line break", () => {
+    const bad = { key: "a", region: "r", url: "postgres://h/a\npostgres://evil/db" };
+    expect(() => shardMigrationDsns(JSON.stringify([bad]))).toThrow(/line break/i);
+  });
+
+  it("throws when a directUrl holds a carriage return", () => {
+    const bad = { key: "a", region: "r", url: "postgres://h/a", directUrl: "postgres://h/a\rx" };
+    expect(() => shardMigrationDsns(JSON.stringify([bad]))).toThrow(/line break/i);
+  });
+});
diff --git a/docker/scripts/entrypoint.sh b/docker/scripts/entrypoint.sh
index 4d9899f2732..58d17bc6ce9 100755
--- a/docker/scripts/entrypoint.sh
+++ b/docker/scripts/entrypoint.sh
@@ -65,18 +65,18 @@ if [ -n "$RUN_OPS_SHARDS" ]; then
     shard_dsns=$(node scripts/runOpsShardDsns.mjs)
     # A `for` loop and NOT `... | while read`: a pipeline subshell would swallow a failed migration
     # on any iteration but the last. Here `set -e` stops the boot on the first shard that fails.
-    # IFS is newline-only so a DSN is never split on other whitespace, and `set -f` stops a DSN
-    # query string (it holds `?`) from being read as a glob pattern.
-    old_ifs=$IFS
-    IFS='
+    # The whole loop runs in a subshell, so the IFS and `set -f` changes need no restore and cannot
+    # leak into the rest of the entrypoint. IFS is newline-only so a DSN is never split on other
+    # whitespace, and `set -f` stops a DSN query string (it holds `?`) from acting as a glob.
+    (
+      IFS='
 '
-    set -f
-    for shard_dsn in $shard_dsns; do
-      # Subshell with tracing off so `set -x` does not print the DSN (with credentials) to the logs.
-      (set +x; RUN_OPS_DATABASE_URL="$shard_dsn" DIRECT_URL="$shard_dsn" pnpm --filter @internal/run-ops-database db:migrate:deploy)
-    done
-    set +f
-    IFS=$old_ifs
+      set -f
+      for shard_dsn in $shard_dsns; do
+        # Tracing stays off so `set -x` never prints the DSN (with credentials) to the logs.
+        RUN_OPS_DATABASE_URL="$shard_dsn" DIRECT_URL="$shard_dsn" pnpm --filter @internal/run-ops-database db:migrate:deploy
+      done
+    )
     set -x
     echo "Run-ops shard migrations done"
   else
diff --git a/docker/scripts/runOpsShardDsns.mjs b/docker/scripts/runOpsShardDsns.mjs
index 384e543e996..ccb457b4c34 100644
--- a/docker/scripts/runOpsShardDsns.mjs
+++ b/docker/scripts/runOpsShardDsns.mjs
@@ -39,6 +39,11 @@ export function shardMigrationDsns(raw) {
     if (typeof dsn !== "string" || dsn === "") {
       continue;
     }
+    // One DSN per line IS the protocol with the caller, so a DSN holding a line break would split
+    // into two bogus DSNs. The URL parser strips ASCII line breaks, so nothing upstream rejects it.
+    if (/[\r\n]/.test(dsn)) {
+      throw new Error("RUN_OPS_SHARDS holds a DSN containing a line break");
+    }
     dsns.push(dsn);
   }
   return dsns;

From e6b40d1693025f447700b161e2d472ad6bcd9d5b Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Tue, 25 Aug 2026 18:06:19 +0100
Subject: [PATCH 13/13] fix(docker): hold the shard DSN script to the same
 contract as the boot schema

The script was laxer than the schema that validates the same variable. It
treated any aliasOf value as an alias, so a shard with a typo in that field was
skipped and its database never migrated. It also accepted a shard that owns its
database but declares no replication, which the application rejects at boot: the
entrypoint migrated the database first and the boot failed afterwards.

The script now rejects an unsupported aliasOf value, a descriptor that sets both
url and aliasOf or neither, and a non-aliased descriptor with no replication. So
an invalid descriptor stops the container before any migration runs, which is
what the block exists to guarantee.
---
 apps/webapp/test/runOpsShardDsns.test.ts | 42 +++++++++++++++++++++---
 docker/scripts/runOpsShardDsns.mjs       | 17 +++++++++-
 2 files changed, 54 insertions(+), 5 deletions(-)

diff --git a/apps/webapp/test/runOpsShardDsns.test.ts b/apps/webapp/test/runOpsShardDsns.test.ts
index 73c6601a6e6..a46f160e3fb 100644
--- a/apps/webapp/test/runOpsShardDsns.test.ts
+++ b/apps/webapp/test/runOpsShardDsns.test.ts
@@ -44,9 +44,9 @@ describe("shardMigrationDsns", () => {
     expect(shardMigrationDsns(JSON.stringify([shardA, aliased]))).toEqual(["postgres://h/a"]);
   });
 
-  it("skips a descriptor with neither url nor directUrl", () => {
+  it("rejects a descriptor with neither url nor aliasOf", () => {
     const noUrl = { key: "b", region: "us-east-1" };
-    expect(shardMigrationDsns(JSON.stringify([shardA, noUrl]))).toEqual(["postgres://h/a"]);
+    expect(() => shardMigrationDsns(JSON.stringify([shardA, noUrl]))).toThrow(/exactly one/i);
   });
 
   it("keeps declaration order across several shards", () => {
@@ -66,12 +66,46 @@ describe("shardMigrationDsns line protocol", () => {
   // One DSN per line is the protocol with entrypoint.sh, so a line break would split one DSN into
   // two bogus ones. The URL parser strips ASCII line breaks, so nothing upstream rejects this.
   it("throws when a DSN holds a line break", () => {
-    const bad = { key: "a", region: "r", url: "postgres://h/a\npostgres://evil/db" };
+    const bad = { ...shardA, url: "postgres://h/a\npostgres://evil/db" };
     expect(() => shardMigrationDsns(JSON.stringify([bad]))).toThrow(/line break/i);
   });
 
   it("throws when a directUrl holds a carriage return", () => {
-    const bad = { key: "a", region: "r", url: "postgres://h/a", directUrl: "postgres://h/a\rx" };
+    const bad = { ...shardA, directUrl: "postgres://h/a\rx" };
     expect(() => shardMigrationDsns(JSON.stringify([bad]))).toThrow(/line break/i);
   });
 });
+
+// The script and the boot schema validate the same variable, so they must agree. If the script is
+// laxer, the entrypoint migrates a database and the application then refuses to start, which breaks
+// the fail-before-migration contract the entrypoint exists to hold.
+describe("shardMigrationDsns matches the descriptor contract", () => {
+  it("rejects an aliasOf value the schema does not allow", () => {
+    const bad = [{ key: "a", region: "r", url: "postgres://h/a", aliasOf: "other" }];
+    expect(() => shardMigrationDsns(JSON.stringify(bad))).toThrow(/aliasOf/i);
+  });
+
+  it("rejects a shard that owns its database but declares no replication", () => {
+    const bad = [{ key: "b", region: "r", url: "postgres://h/b" }];
+    expect(() => shardMigrationDsns(JSON.stringify(bad))).toThrow(/replication/i);
+  });
+
+  it("rejects a descriptor that sets both url and aliasOf", () => {
+    const bad = [{ key: "c", region: "r", url: "postgres://h/c", aliasOf: "new" }];
+    expect(() => shardMigrationDsns(JSON.stringify(bad))).toThrow(/exactly one/i);
+  });
+
+  it("rejects a descriptor that sets neither url nor aliasOf", () => {
+    const bad = [{ key: "d", region: "r" }];
+    expect(() => shardMigrationDsns(JSON.stringify(bad))).toThrow(/exactly one/i);
+  });
+
+  it("still accepts a valid aliased descriptor and skips it", () => {
+    const ok = [{ key: "z", region: "r", aliasOf: "new" }];
+    expect(shardMigrationDsns(JSON.stringify(ok))).toEqual([]);
+  });
+
+  it("still accepts a valid owning descriptor", () => {
+    expect(shardMigrationDsns(JSON.stringify([shardA]))).toEqual(["postgres://h/a"]);
+  });
+});
diff --git a/docker/scripts/runOpsShardDsns.mjs b/docker/scripts/runOpsShardDsns.mjs
index ccb457b4c34..009359cb219 100644
--- a/docker/scripts/runOpsShardDsns.mjs
+++ b/docker/scripts/runOpsShardDsns.mjs
@@ -31,8 +31,23 @@ export function shardMigrationDsns(raw) {
     if (descriptor === null || typeof descriptor !== "object") {
       throw new Error("RUN_OPS_SHARDS holds an entry that is not an object");
     }
+    // The boot schema (runOpsShards.server.ts) validates the same variable. This script must not be
+    // laxer: a descriptor it accepts and the application rejects would migrate a database and then
+    // fail the boot, which breaks the fail-before-migration contract.
+    const hasAlias = descriptor.aliasOf !== undefined && descriptor.aliasOf !== null;
+    if (hasAlias && descriptor.aliasOf !== "new") {
+      throw new Error(`RUN_OPS_SHARDS: aliasOf must be "new", got "${descriptor.aliasOf}"`);
+    }
+    const hasUrl = typeof descriptor.url === "string" && descriptor.url !== "";
+    if (hasUrl === hasAlias) {
+      throw new Error("RUN_OPS_SHARDS: exactly one of url or aliasOf is required");
+    }
+    if (!hasAlias && (descriptor.replication === undefined || descriptor.replication === null)) {
+      throw new Error("RUN_OPS_SHARDS: replication is required unless aliasOf is set");
+    }
+
     // An aliased shard is the same database as its target, which is migrated by its own invocation.
-    if (descriptor.aliasOf !== undefined && descriptor.aliasOf !== null) {
+    if (hasAlias) {
       continue;
     }
     const dsn = descriptor.directUrl ?? descriptor.url;