diff --git a/apps/webapp/app/routes/@.runs.$runParam.ts b/apps/webapp/app/routes/@.runs.$runParam.ts
index d8ff7fd49d1..dc2ba8c33a2 100644
--- a/apps/webapp/app/routes/@.runs.$runParam.ts
+++ b/apps/webapp/app/routes/@.runs.$runParam.ts
@@ -8,6 +8,7 @@ import { redirectWithErrorMessage } from "~/models/message.server";
 import { requireUser } from "~/services/session.server";
 import { impersonate, rootPath, v3RunPath, v3RunSpanPath } from "~/utils/pathBuilder";
 import { findBufferedRunRedirectInfo } from "~/v3/mollifier/syntheticRedirectInfo.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -33,17 +34,21 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     );
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-    },
-    {
-      select: {
-        spanId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    prisma
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: runParam,
+        },
+        {
+          select: {
+            spanId: true,
+            runtimeEnvironmentId: true,
+          },
+        },
+        prisma
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts b/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts
index 39eebc2303e..bc459aa516e 100644
--- a/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts
+++ b/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts
@@ -5,6 +5,7 @@ import { runOpsLegacyReplica, runOpsNewReplica, runOpsSplitReadEnabled } from "~
 import { ApiBatchResultsPresenter } from "~/presenters/v3/ApiBatchResultsPresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   /* This is the batch friendly ID */
@@ -42,6 +43,14 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
       return json(result);
     } catch (error) {
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) {
+        logger.warn("Unroutable batch id on batch results", {
+          error: error instanceof Error ? error.message : error,
+        });
+        return unroutable;
+      }
+
       logger.error("Failed to load batch results", { error });
       return json({ error: "Something went wrong, please try again." }, { status: 500 });
     }
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts
index be717efa355..893937d6b4d 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts
@@ -10,6 +10,7 @@ import { logger } from "~/services/logger.server";
 import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 import { runStore } from "~/v3/runStore.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 // Pull the existing tags out of a buffer entry's serialised payload so
 // the buffer-path response can dedup against them, matching the
@@ -137,6 +138,14 @@ export async function action({ request, params }: ActionFunctionArgs) {
     }
     return outcome.response;
   } catch (error) {
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on run tags", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to add run tags", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }
diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts
index dbc521a7424..e6b96dc4d31 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts
@@ -12,6 +12,7 @@ import { RescheduleTaskRunService } from "~/v3/services/rescheduleTaskRun.server
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 import { parseDelay } from "~/utils/delays";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -156,6 +157,14 @@ export async function action({ request, params }: ActionFunctionArgs) {
     if (error instanceof ServiceValidationError) {
       return json({ error: error.message }, { status: 400 });
     }
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on reschedule", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to reschedule run", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }
diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts
index 5df9e496c17..784b0310aff 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts
@@ -5,6 +5,7 @@ import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.ser
 import { runOpsLegacyReplica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   /* This is the run friendly ID */
@@ -41,6 +42,14 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
     return json(result);
   } catch (error) {
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on run result", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to load run result", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts
index 431a7eb2582..39281bde67e 100644
--- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts
+++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts
@@ -6,6 +6,7 @@ import { env } from "~/env.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
 import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 import { engine } from "~/v3/runEngine.server";
 import { runStore } from "~/v3/runStore.server";
@@ -102,6 +103,19 @@ export async function action({ request, params }: ActionFunctionArgs) {
       { status: 200 }
     );
   } catch (error) {
+    // Same as the complete route: the waitpoint id comes off the URL, so an unconfigured shard
+    // key is caller-supplied input and must answer 404 rather than 500. This route is a bare
+    // Remix action, so the api-builder boundary never sees the error — answer it here.
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      // Same reason as the complete route: a silent 404 would hide a dropped shard key.
+      logger.warn("Unroutable waitpoint id on HTTP callback", {
+        waitpointFriendlyId: params.waitpointFriendlyId,
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to complete HTTP callback", { error });
     throw json({ error: "Failed to complete HTTP callback" }, { status: 500 });
   }
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
index 950b36e873b..aff5045fbd2 100644
--- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
+++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
@@ -10,6 +10,7 @@ import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 import { engine } from "~/v3/runEngine.server";
 import { runStore } from "~/v3/runStore.server";
 
@@ -87,6 +88,19 @@ const { action, loader } = createActionApiRoute(
       // client gets the correct status code instead of a 500, and we don't log them as errors.
       if (error instanceof Response) throw error;
 
+      // A caller-supplied id naming a shard this topology has no store for cannot be routed,
+      // so it is a 404 like an absent token — not the 500 this catch would otherwise answer.
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) {
+        // Logged so a shard key dropped from an append-only config still alarms, rather than
+        // every live token on it quietly answering "not found".
+        logger.warn("Unroutable waitpoint id on token completion", {
+          waitpointFriendlyId: params.waitpointFriendlyId,
+          error: error instanceof Error ? error.message : error,
+        });
+        throw unroutable;
+      }
+
       logger.error("Failed to complete waitpoint token", {
         error:
           error instanceof Error
diff --git a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts b/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts
index f2e138861e3..2cb47971979 100644
--- a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts
+++ b/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts
@@ -6,6 +6,7 @@ import { requireUserId } from "~/services/session.server";
 import { ProjectParamSchema, v3RunPath } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamSchema = ProjectParamSchema.extend({
   runParam: z.string(),
@@ -15,16 +16,20 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
   const { organizationSlug, projectParam, runParam } = ParamSchema.parse(params);
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-    },
-    {
-      select: {
-        projectId: true,
-        runtimeEnvironmentId: true,
-      },
-    }
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: runParam,
+        },
+        {
+          select: {
+            projectId: true,
+            runtimeEnvironmentId: true,
+          },
+        }
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts b/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts
index 300caa27ddf..63475d193d4 100644
--- a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts
+++ b/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts
@@ -5,6 +5,7 @@ import { requireUserId } from "~/services/session.server";
 import { v3RunSpanPath } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -36,18 +37,22 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: validatedParams.runParam,
-    },
-    {
-      select: {
-        friendlyId: true,
-        spanId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    prisma
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: validatedParams.runParam,
+        },
+        {
+          select: {
+            friendlyId: true,
+            spanId: true,
+            runtimeEnvironmentId: true,
+          },
+        },
+        prisma
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts
index 062b75503a6..7238310372c 100644
--- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts
@@ -13,6 +13,7 @@ import {
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -61,8 +62,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
   // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
   const run =
-    (await runStore.findRun(runWhere, runArgs, $replica)) ??
-    (await runStore.findRunOnPrimary(runWhere, runArgs));
+    (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), {
+      runParam: params.runParam,
+    })) ??
+    (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), {
+      runParam: params.runParam,
+    }));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts
index 4fbb309e2a8..7a398e91b67 100644
--- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts
@@ -8,6 +8,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -60,8 +61,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
   // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
   const run =
-    (await runStore.findRun(runWhere, runArgs, $replica)) ??
-    (await runStore.findRunOnPrimary(runWhere, runArgs));
+    (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), {
+      runParam: params.runParam,
+    })) ??
+    (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), {
+      runParam: params.runParam,
+    }));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts
index 7eae8f6fdcf..be963701f4c 100644
--- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts
@@ -8,6 +8,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -62,8 +63,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream
   // subscription (useRealtimeStream surfaces the error, no auto-retry). Re-read the primary on a miss.
   const run =
-    (await runStore.findRun(runWhere, runArgs, $replica)) ??
-    (await runStore.findRunOnPrimary(runWhere, runArgs));
+    (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), {
+      runParam: params.runParam,
+    })) ??
+    (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), {
+      runParam: params.runParam,
+    }));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });
diff --git a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts
index 649f2ef268f..86c1d5cbeb4 100644
--- a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts
+++ b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts
@@ -15,6 +15,7 @@ import {
   type TraceExportContext,
 } from "~/v3/eventRepository/traceExport.server";
 import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   const user = await requireUser(request);
@@ -31,20 +32,24 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
   // Run-ops read keyed by friendlyId only (routes to the owning DB by residency). Org
   // membership is a control-plane concern resolved separately below — joining it here is a
   // cross-DB join that returns nothing once the run lives in run-ops.
-  let run = await runStore.findRun(
-    { friendlyId: parsedParams.runParam },
-    {
-      select: {
-        friendlyId: true,
-        traceId: true,
-        organizationId: true,
-        runtimeEnvironmentId: true,
-        createdAt: true,
-        completedAt: true,
-        taskEventStore: true,
-        taskIdentifier: true,
-      },
-    }
+  let run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        { friendlyId: parsedParams.runParam },
+        {
+          select: {
+            friendlyId: true,
+            traceId: true,
+            organizationId: true,
+            runtimeEnvironmentId: true,
+            createdAt: true,
+            completedAt: true,
+            taskEventStore: true,
+            taskIdentifier: true,
+          },
+        }
+      ),
+    { runParam: parsedParams.runParam }
   );
 
   // Authorize on the control-plane DB: the user must be a member of the run's org. A
diff --git a/apps/webapp/app/routes/resources.runs.$runParam.ts b/apps/webapp/app/routes/resources.runs.$runParam.ts
index e4328fe4b37..7cfafb673d4 100644
--- a/apps/webapp/app/routes/resources.runs.$runParam.ts
+++ b/apps/webapp/app/routes/resources.runs.$runParam.ts
@@ -12,72 +12,77 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.
 import { FINAL_ATTEMPT_STATUSES, isFinalRunStatus } from "~/v3/taskStatus";
 
 import { boundedIn } from "@trigger.dev/database";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 export type RunInspectorData = UseDataFunctionReturn<typeof loader>;
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
   const parsedParams = v3RunParamsSchema.pick({ runParam: true }).parse(params);
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: parsedParams.runParam,
-    },
-    {
-      select: {
-        id: true,
-        traceId: true,
-        //metadata
-        number: true,
-        taskIdentifier: true,
-        friendlyId: true,
-        isTest: true,
-        runTags: true,
-        machinePreset: true,
-        runtimeEnvironmentId: true,
-        projectId: true,
-        lockedById: true,
-        lockedToVersionId: true,
-        //status + duration
-        status: true,
-        startedAt: true,
-        createdAt: true,
-        updatedAt: true,
-        queuedAt: true,
-        completedAt: true,
-        logsDeletedAt: true,
-        //idempotency
-        idempotencyKey: true,
-        //delayed
-        delayUntil: true,
-        //ttl
-        ttl: true,
-        expiredAt: true,
-        //queue
-        queue: true,
-        concurrencyKey: true,
-        //schedule
-        scheduleId: true,
-        //usage
-        baseCostInCents: true,
-        costInCents: true,
-        usageDurationMs: true,
-        payload: true,
-        payloadType: true,
-        metadata: true,
-        metadataType: true,
-        maxAttempts: true,
-        parentTaskRun: {
-          select: {
-            friendlyId: true,
-          },
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: parsedParams.runParam,
         },
-        rootTaskRun: {
+        {
           select: {
+            id: true,
+            traceId: true,
+            //metadata
+            number: true,
+            taskIdentifier: true,
             friendlyId: true,
+            isTest: true,
+            runTags: true,
+            machinePreset: true,
+            runtimeEnvironmentId: true,
+            projectId: true,
+            lockedById: true,
+            lockedToVersionId: true,
+            //status + duration
+            status: true,
+            startedAt: true,
+            createdAt: true,
+            updatedAt: true,
+            queuedAt: true,
+            completedAt: true,
+            logsDeletedAt: true,
+            //idempotency
+            idempotencyKey: true,
+            //delayed
+            delayUntil: true,
+            //ttl
+            ttl: true,
+            expiredAt: true,
+            //queue
+            queue: true,
+            concurrencyKey: true,
+            //schedule
+            scheduleId: true,
+            //usage
+            baseCostInCents: true,
+            costInCents: true,
+            usageDurationMs: true,
+            payload: true,
+            payloadType: true,
+            metadata: true,
+            metadataType: true,
+            maxAttempts: true,
+            parentTaskRun: {
+              select: {
+                friendlyId: true,
+              },
+            },
+            rootTaskRun: {
+              select: {
+                friendlyId: true,
+              },
+            },
           },
-        },
-      },
-    }
+        }
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts
index 7275e6d73aa..45aca19269a 100644
--- a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts
+++ b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts
@@ -6,6 +6,7 @@ import { requireUserId } from "~/services/session.server";
 import { engine } from "~/v3/runEngine.server";
 import { runStore } from "~/v3/runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamSchema = z.object({
   runParam: z.string(),
@@ -18,20 +19,24 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // Run-ops read keyed by friendlyId only (routes to the owning DB by residency). The
   // project/org-membership auth is a control-plane concern resolved separately below —
   // joining it here is a cross-DB join that returns nothing once the run lives in run-ops.
-  const run = await runStore.findRun(
-    { friendlyId: runParam },
-    {
-      select: {
-        id: true,
-        engine: true,
-        friendlyId: true,
-        queue: true,
-        concurrencyKey: true,
-        queueTimestamp: true,
-        runtimeEnvironmentId: true,
-        projectId: true,
-      },
-    }
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        { friendlyId: runParam },
+        {
+          select: {
+            id: true,
+            engine: true,
+            friendlyId: true,
+            queue: true,
+            concurrencyKey: true,
+            queueTimestamp: true,
+            runtimeEnvironmentId: true,
+            projectId: true,
+          },
+        }
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/routes/runs.$runParam.ts b/apps/webapp/app/routes/runs.$runParam.ts
index 6723f39a621..f6f9241a449 100644
--- a/apps/webapp/app/routes/runs.$runParam.ts
+++ b/apps/webapp/app/routes/runs.$runParam.ts
@@ -6,6 +6,7 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.
 import { redirectWithErrorMessage } from "~/models/message.server";
 import { requireUser } from "~/services/session.server";
 import { rootPath, v3RunPath } from "~/utils/pathBuilder";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -16,17 +17,21 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
   const { runParam } = ParamsSchema.parse(params);
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-    },
-    {
-      select: {
-        spanId: true,
-        projectId: true,
-        runtimeEnvironmentId: true,
-      },
-    }
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: runParam,
+        },
+        {
+          select: {
+            spanId: true,
+            projectId: true,
+            runtimeEnvironmentId: true,
+          },
+        }
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {
diff --git a/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts b/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts
index 1f0004dda1a..39e1dcef678 100644
--- a/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts
+++ b/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts
@@ -1785,6 +1785,12 @@ export function createLoaderWorkerApiRoute<
         return error;
       }
 
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) {
+        logBoundaryError("Unroutable id", error, request.url);
+        return unroutable;
+      }
+
       logBoundaryError("Error in loader", error, request.url);
 
       return json({ error: "Internal Server Error" }, { status: 500 });
@@ -1954,6 +1960,12 @@ export function createActionWorkerApiRoute<
         return json({ error: error.message }, { status: error.status ?? 422 });
       }
 
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) {
+        logBoundaryError("Unroutable id", error, request.url);
+        return unroutable;
+      }
+
       logBoundaryError("Error in action", error, request.url);
 
       return json({ error: "Internal Server Error" }, { status: 500 });
diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts
index 7c074b1586b..f2bb8de1071 100644
--- a/apps/webapp/app/services/runsReplicationInstance.server.ts
+++ b/apps/webapp/app/services/runsReplicationInstance.server.ts
@@ -9,6 +9,7 @@ import {
   setRunsReplicationConfiguredSources,
   setRunsReplicationGlobal,
 } from "./runsReplicationGlobal.server";
+import { runsReplicationSourceMetrics } from "./runsReplicationMetrics.server";
 import {
   RunsReplicationService,
   type RunsReplicationSource,
@@ -249,6 +250,9 @@ function initializeRunsReplicationInstance() {
     insertStrategy: env.RUN_REPLICATION_INSERT_STRATEGY,
     disablePayloadInsert: env.RUN_REPLICATION_DISABLE_PAYLOAD_INSERT === "1",
     disableErrorFingerprinting: env.RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING === "1",
+    // A source whose publication carries no usable table logs every 30s and replicates nothing.
+    // Boot cannot see it (the source IS configured), so the counter is the alarmable signal.
+    onSourceError: runsReplicationSourceMetrics.recordSourceError,
   };
 
   // Construct the SINGLE legacy source synchronously (the split gate has not resolved
diff --git a/apps/webapp/app/services/runsReplicationMetrics.server.ts b/apps/webapp/app/services/runsReplicationMetrics.server.ts
new file mode 100644
index 00000000000..d8aa3dddf97
--- /dev/null
+++ b/apps/webapp/app/services/runsReplicationMetrics.server.ts
@@ -0,0 +1,38 @@
+/**
+ * A replication source whose publication carries no usable table replicates nothing while the
+ * service stays up and healthy: boot passes, `assertReplicationCoversSplit` only checks that a
+ * source is CONFIGURED, and the client's retry loop logs every 30s. Counting it is what makes it
+ * alarmable — a non-zero rate here means that source's runs are not reaching ClickHouse.
+ */
+import { PublicationMisconfiguredError } from "@internal/replication";
+import { Counter, type Registry, type RegistryContentType } from "prom-client";
+import { metricsRegister } from "~/metrics.server";
+import { singleton } from "~/utils/singleton";
+
+export type RunsReplicationSourceMetrics = {
+  recordSourceError(info: { sourceId: string; error: unknown }): void;
+};
+
+export function buildRunsReplicationSourceMetrics(
+  register: Registry<RegistryContentType>
+): RunsReplicationSourceMetrics {
+  const publicationMisconfigured = new Counter({
+    name: "runs_replication_publication_misconfigured_total",
+    help: "A replication source's publication does not carry the replicated table, so that source replicates nothing.",
+    labelNames: ["source"],
+    registers: [register],
+  });
+
+  return {
+    recordSourceError: ({ sourceId, error }) => {
+      if (error instanceof PublicationMisconfiguredError) {
+        publicationMisconfigured.inc({ source: sourceId });
+      }
+    },
+  };
+}
+
+// singleton: module-scope Counter registration double-registers under dev HMR.
+export const runsReplicationSourceMetrics = singleton("runsReplicationSourceMetrics", () =>
+  buildRunsReplicationSourceMetrics(metricsRegister)
+);
diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts
index ace3029abf1..4047e831603 100644
--- a/apps/webapp/app/services/runsReplicationService.server.ts
+++ b/apps/webapp/app/services/runsReplicationService.server.ts
@@ -116,6 +116,12 @@ export type RunsReplicationServiceOptions = {
   disablePayloadInsert?: boolean;
   disableErrorFingerprinting?: boolean;
   maxPoisonStripsPerBatch?: number;
+  /**
+   * Per-source client error hook. A client error does not stop the service — a misconfigured
+   * publication just replicates nothing while the retry loop logs — so the owner needs a seam to
+   * count it on.
+   */
+  onSourceError?: (info: { sourceId: string; error: unknown }) => void;
 };
 
 type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -470,6 +476,7 @@ export class RunsReplicationService {
         sourceId: source.id,
         error,
       });
+      this.options.onSourceError?.({ sourceId: source.id, error });
     });
 
     client.events.on("start", () => {
diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts
index a4e4a64d36d..16341328a31 100644
--- a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts
+++ b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts
@@ -314,6 +314,39 @@ describe("resolveMintShardWith — cache, read failure and fail-safe", () => {
     expect(["a", "b"]).toContain(await resolveMintShardWith({ id: "env_1" }, deps));
   });
 
+  it("reports an unparseable stored list while still degrading to gen-1", async () => {
+    // The observed failure: `runOpsMintShardSet` saved as "A,B" (uppercase) reverted the whole
+    // fleet to gen-1 minting with ZERO log lines, because the parse throw was swallowed and
+    // `onReadFailed` never fires for it — the read succeeded. The degrade is correct; silence is not.
+    const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "A,B" }) });
+    const readFailures: unknown[] = [];
+    const parseFailures: Array<{ key: string; value: string }> = [];
+    deps.onReadFailed = (error) => readFailures.push(error);
+    deps.onSetParseFailed = ({ key, value }) => parseFailures.push({ key, value });
+
+    expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
+    expect(readFailures).toEqual([]);
+    expect(parseFailures).toEqual([{ key: "runOpsMintShardSet", value: "A,B" }]);
+  });
+
+  it("reports a reserved key in the stored list too", async () => {
+    const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "a,legacy" }) });
+    const parseFailures: Array<{ key: string; value: string }> = [];
+    deps.onSetParseFailed = ({ key, value }) => parseFailures.push({ key, value });
+
+    expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
+    expect(parseFailures).toEqual([{ key: "runOpsMintShardSet", value: "a,legacy" }]);
+  });
+
+  it("stays silent for a stored list that parses", async () => {
+    const deps = wrapperDeps();
+    const parseFailures: unknown[] = [];
+    deps.onSetParseFailed = (failure) => parseFailures.push(failure);
+
+    expect(["a", "b"]).toContain(await resolveMintShardWith({ id: "env_1" }, deps));
+    expect(parseFailures).toEqual([]);
+  });
+
   it("returns gen-1 when the stored list is empty", async () => {
     const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "" }) });
     expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts
index 2855f936249..1687ff76c65 100644
--- a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts
+++ b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts
@@ -8,6 +8,7 @@ import {
   GEN_1_PIN_VALUE,
   isValidPinValue,
   readMintShardSetResolution,
+  type MintShardSetParseFailure,
   type MintShardSetResolution,
 } from "./mintShardGrace";
 
@@ -166,6 +167,9 @@ export type ResolveMintShardDeps = {
   onPinRejected?: (info: { environmentId: string; pin: string; activeSet: string[] }) => void;
   onOverrideRejected?: (info: { override: string; activeSet: string[] }) => void;
   onReadFailed?: (error: unknown) => void;
+  // A stored set that PARSED badly, which is not a read failure: onReadFailed never fires for it,
+  // yet the fleet reverts to gen-1 minting. Reported here so the operator sees the degrade.
+  onSetParseFailed?: (failure: MintShardSetParseFailure) => void;
 };
 
 // The live list is org-independent, so one process-wide entry serves every mint: one query per
@@ -180,7 +184,7 @@ async function refreshConfig(deps: ResolveMintShardDeps): Promise<GlobalShardCon
   try {
     const flags = await deps.readFlags();
     const config: GlobalShardConfig = {
-      resolution: readMintShardSetResolution(flags),
+      resolution: readMintShardSetResolution(flags, deps.onSetParseFailed),
       override: flags[FEATURE_FLAG.runOpsMintShardOverride],
     };
     deps.cache.current = { value: config, expiresAt: deps.nowMs + deps.ttlMs };
diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts
index 65d8af6f417..0172f762977 100644
--- a/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts
+++ b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts
@@ -67,15 +67,25 @@ export function effectiveMintShardSet(
   return nowMs < r.flippedAtMs + graceMs ? r.prevSet : r.set;
 }
 
+/** Reports a stored value that could not be parsed. The module stays pure; the caller logs. */
+export type MintShardSetParseFailure = { key: string; value: string; error: unknown };
+
 // The active set lives in the control-plane database, not in the environment. A deploy rolls
 // for hours, so two pods can hold different environment values at the same time; only a shared
 // row lets every pod agree on one set. Boot may reject a bad environment value, but the mint
 // path must never throw on a bad stored value, so an unreadable list degrades to empty.
-function readStoredCsv(value: unknown): string[] {
+//
+// Degrading is right; degrading SILENTLY is not. An unparseable value (a `"A,B"` typed in
+// uppercase) reverts the whole fleet to gen-1 minting, so the failure is reported to the caller,
+// which owns the logger. The reserved-key and alphabet rejections are the same failure.
+export type OnInvalidShardSet = (failure: MintShardSetParseFailure) => void;
+
+function readStoredCsv(value: unknown, key: string, onInvalid?: OnInvalidShardSet): string[] {
   if (typeof value !== "string") return [];
   try {
     return parseShardCsv(value);
-  } catch {
+  } catch (error) {
+    onInvalid?.({ key, value, error });
     return [];
   }
 }
@@ -84,7 +94,8 @@ function readStoredCsv(value: unknown): string[] {
 // timestamp can never apply, so it is dropped. A timestamp with an EMPTY prevSet is meaningful:
 // it graces a first activation, serving no shards for the window.
 export function readMintShardSetResolution(
-  flags: Record<string, unknown> | null | undefined
+  flags: Record<string, unknown> | null | undefined,
+  onInvalid?: OnInvalidShardSet
 ): MintShardSetResolution {
   const source = flags ?? {};
   const flippedAtRaw = source[SET_FLIPPED_AT_KEY];
@@ -92,8 +103,11 @@ export function readMintShardSetResolution(
   const flippedAtMs = Number.isNaN(parsed) ? undefined : parsed;
 
   return {
-    set: readStoredCsv(source[SET_KEY]),
-    prevSet: flippedAtMs === undefined ? undefined : readStoredCsv(source[SET_PREV_KEY]),
+    set: readStoredCsv(source[SET_KEY], SET_KEY, onInvalid),
+    prevSet:
+      flippedAtMs === undefined
+        ? undefined
+        : readStoredCsv(source[SET_PREV_KEY], SET_PREV_KEY, onInvalid),
     flippedAtMs,
   };
 }
@@ -105,7 +119,8 @@ export function stampMintShardSetFlip(
   existingFlags: Record<string, unknown> | null | undefined,
   outgoingFlags: Record<string, unknown>,
   nowMs: number,
-  graceMs: number
+  graceMs: number,
+  onInvalid?: OnInvalidShardSet
 ): Record<string, unknown> {
   // Only act when the save actually SETS the list. Omitting it must not inject a default.
   if (typeof outgoingFlags[SET_KEY] !== "string") {
@@ -113,11 +128,15 @@ export function stampMintShardSetFlip(
   }
 
   const existing = existingFlags ?? {};
-  const outgoingSet = readStoredCsv(outgoingFlags[SET_KEY]);
-  const storedSet = readStoredCsv(existing[SET_KEY]);
+  const outgoingSet = readStoredCsv(outgoingFlags[SET_KEY], SET_KEY, onInvalid);
+  const storedSet = readStoredCsv(existing[SET_KEY], SET_KEY, onInvalid);
 
   if (outgoingSet.join(",") !== storedSet.join(",")) {
-    const effective = effectiveMintShardSet(readMintShardSetResolution(existing), nowMs, graceMs);
+    const effective = effectiveMintShardSet(
+      readMintShardSetResolution(existing, onInvalid),
+      nowMs,
+      graceMs
+    );
     outgoingFlags[SET_PREV_KEY] = effective.join(",");
     outgoingFlags[SET_FLIPPED_AT_KEY] = new Date(nowMs).toISOString();
     return outgoingFlags;
diff --git a/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts b/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts
index 8a657060ef9..e644ea91fa2 100644
--- a/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts
+++ b/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts
@@ -309,4 +309,70 @@ describe("readThroughRun (legacy replica + new DB)", () => {
       expect(throwingLegacy).not.toHaveBeenCalled();
     }
   );
+  // Which store served a read was a return value only — never emitted — so during a cohort ramp
+  // there was no way to see from outside the process where reads were landing.
+  heteroPostgresTest(
+    "emits the serving source for a gen-2 shard, the gen-1 new store and the legacy replica",
+    async ({ prisma14, prisma17 }) => {
+      const emitted: string[] = [];
+      const deps = {
+        splitEnabled: true,
+        newClient: prisma17 as unknown as PrismaReplicaClient,
+        legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        shardReplicas: new Map([["a", prisma17 as unknown as PrismaReplicaClient]]),
+        onSource: (source: string) => emitted.push(source),
+      };
+
+      await readThroughRun({
+        id: SHARD_A_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, true),
+        readLegacy: (c) => realRead(c, false),
+        deps,
+      });
+      await readThroughRun({
+        id: NEW_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, true),
+        readLegacy: (c) => realRead(c, false),
+        deps,
+      });
+      await readThroughRun({
+        id: LEGACY_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, false),
+        readLegacy: (c) => realRead(c, true),
+        deps,
+      });
+
+      expect(emitted).toEqual(["shard:a", "new", "legacy-replica"]);
+    }
+  );
+
+  heteroPostgresTest(
+    "emits nothing for a miss, so a not-found cannot look like a hit",
+    async ({ prisma14, prisma17 }) => {
+      const emitted: string[] = [];
+
+      const result = await readThroughRun({
+        id: LEGACY_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, false),
+        readLegacy: (c) => realRead(c, false),
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+          onSource: (source: string) => emitted.push(source),
+        },
+      });
+
+      expect(result.found).toBe(false);
+      expect(emitted).toEqual([]);
+    }
+  );
 });
diff --git a/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts b/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts
index 6e1beaae62c..de37b225b85 100644
--- a/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts
@@ -25,11 +25,12 @@ import {
 import { logger as defaultLogger } from "~/services/logger.server";
 import { resolveShard, type ShardKey } from "@trigger.dev/core/v3/isomorphic";
 import { isSplitEnabled } from "./splitMode.server";
+import { recordReadThroughSource } from "./readThroughSourceMetric.server";
 import { runOpsShardReplicas } from "./shardHandles.server";
 
 type ShardSource = `shard:${string}`;
 
-type ReadThroughSource = "new" | "legacy-replica" | ShardSource;
+export type ReadThroughSource = "new" | "legacy-replica" | ShardSource;
 
 /**
  * `found` carries hit/miss STRUCTURALLY. `source` is open-ended once shards exist, so a
@@ -56,6 +57,8 @@ type ReadThroughDeps = {
   logger?: { error: (m: string, meta?: Record<string, unknown>) => void };
   /** Saturation-signal emit hook: called on each legacy-replica hit. */
   onLegacyReplicaRead?: (id: string) => void;
+  /** Per-source emit hook: called with the store that served a HIT. Defaults to the metric. */
+  onSource?: (source: ReadThroughSource) => void;
 };
 
 type ReadThroughRunInput<T> = {
@@ -67,7 +70,12 @@ type ReadThroughRunInput<T> = {
   deps?: ReadThroughDeps;
 };
 
-function hit<T>(source: ReadThroughSource, value: T): ReadThroughResult<T> {
+function hit<T>(
+  source: ReadThroughSource,
+  value: T,
+  onSource: (source: ReadThroughSource) => void
+): ReadThroughResult<T> {
+  onSource(source);
   return { found: true, source, value };
 }
 
@@ -83,13 +91,14 @@ export async function readThroughRun<T>(
   const legacyReplica = deps?.legacyReplica ?? defaultLegacyReplica;
   const shardReplicas = deps?.shardReplicas ?? runOpsShardReplicas;
   const logger = deps?.logger ?? defaultLogger;
+  const onSource = deps?.onSource ?? recordReadThroughSource;
 
   const splitEnabled = deps?.splitEnabled ?? (await isSplitEnabled());
 
   // Passthrough: single plain read against the one collapsed store.
   if (!splitEnabled) {
     const v = await input.readNew(newClient);
-    return v != null ? hit("new", v) : miss("not-found");
+    return v != null ? hit("new", v, onSource) : miss("not-found");
   }
 
   // Total: an unclassifiable id resolves to "legacy" (probe rather than drop a real run).
@@ -111,18 +120,18 @@ export async function readThroughRun<T>(
     }
     // A gen-2 shard is a dedicated-schema store, exactly like `new`, so `readNew` fits.
     const v = await input.readNew(shardReplica);
-    return v != null ? hit(`shard:${shardKey}`, v) : miss("not-found");
+    return v != null ? hit(`shard:${shardKey}`, v, onSource) : miss("not-found");
   }
 
   if (shardKey === "new") {
     const v = await input.readNew(newClient);
-    return v != null ? hit("new", v) : miss("not-found");
+    return v != null ? hit("new", v, onSource) : miss("not-found");
   }
 
   if (idKind === "waitpoint") {
     const v = await input.readNew(newClient);
     if (v != null) {
-      return hit("new", v);
+      return hit("new", v, onSource);
     }
   }
 
@@ -130,7 +139,7 @@ export async function readThroughRun<T>(
   const lv = await input.readLegacy(legacyReplica);
   if (lv != null) {
     deps?.onLegacyReplicaRead?.(id);
-    return hit("legacy-replica", lv);
+    return hit("legacy-replica", lv, onSource);
   }
 
   if (deps?.isPastRetention?.(id)) {
diff --git a/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts
new file mode 100644
index 00000000000..920e2f8ca87
--- /dev/null
+++ b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts
@@ -0,0 +1,35 @@
+// The per-shard half of read-through observability: /metrics carried no `shard` label at all, so a
+// cohort ramp was unobservable. A gen-2 source splits into a constant `source` plus the shard key,
+// so one query sums the whole gen-2 cohort and another breaks it down per shard.
+import { Registry, type RegistryContentType } from "prom-client";
+import { describe, expect, it } from "vitest";
+import { buildReadThroughSourceMetric } from "./readThroughSourceMetric.server";
+
+describe("read-through source metric", () => {
+  it("labels a gen-2 hit with its shard key", async () => {
+    const register = new Registry<RegistryContentType>();
+    const record = buildReadThroughSourceMetric(register);
+
+    record("shard:a");
+    record("shard:a");
+    record("shard:b");
+
+    const exposed = await register.metrics();
+    expect(exposed).toContain('runops_read_through_source_total{source="shard",shard="a"} 2');
+    expect(exposed).toContain('runops_read_through_source_total{source="shard",shard="b"} 1');
+  });
+
+  it("keeps the two gen-1 sources distinguishable and shard-less", async () => {
+    const register = new Registry<RegistryContentType>();
+    const record = buildReadThroughSourceMetric(register);
+
+    record("new");
+    record("legacy-replica");
+
+    const exposed = await register.metrics();
+    expect(exposed).toContain('runops_read_through_source_total{source="new",shard="none"} 1');
+    expect(exposed).toContain(
+      'runops_read_through_source_total{source="legacy-replica",shard="none"} 1'
+    );
+  });
+});
diff --git a/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts
new file mode 100644
index 00000000000..136049c35ce
--- /dev/null
+++ b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts
@@ -0,0 +1,42 @@
+/**
+ * Which store actually served a read-through read: `ReadThroughSource` was an internal return type
+ * only, so a cohort ramp was invisible from outside the process. On the hot path, so each label
+ * child is resolved once and cached — `inc({ source, shard })` hashes a fresh object per call.
+ */
+import { Counter, type Registry, type RegistryContentType } from "prom-client";
+import { metricsRegister } from "~/metrics.server";
+import { singleton } from "~/utils/singleton";
+import type { ReadThroughSource } from "./readThrough.server";
+
+const SHARD_PREFIX = "shard:";
+
+export function buildReadThroughSourceMetric(
+  register: Registry<RegistryContentType>
+): (source: ReadThroughSource) => void {
+  const counter = new Counter({
+    name: "runops_read_through_source_total",
+    help: "Read-through reads served, by the store that served them. `shard` carries the gen-2 shard key.",
+    labelNames: ["source", "shard"],
+    registers: [register],
+  });
+
+  const children = new Map<string, { inc: (value?: number) => void }>();
+
+  return (source) => {
+    let child = children.get(source);
+    if (child === undefined) {
+      // A gen-2 source is split into a constant `source` and the shard key, so one query sums the
+      // whole gen-2 cohort and another breaks it down per shard.
+      child = source.startsWith(SHARD_PREFIX)
+        ? counter.labels("shard", source.slice(SHARD_PREFIX.length))
+        : counter.labels(source, "none");
+      children.set(source, child);
+    }
+    child.inc();
+  };
+}
+
+// singleton: module-scope Counter registration double-registers under dev HMR.
+export const recordReadThroughSource = singleton("readThroughSourceMetric", () =>
+  buildReadThroughSourceMetric(metricsRegister)
+);
diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts
index 360bc9fd863..5b3435b48b1 100644
--- a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts
+++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts
@@ -66,6 +66,27 @@ function reportOverrideRejected(info: { override: string; activeSet: string[] })
   logger.error("[runOpsMintShard] override shard is not in the active set; ignoring it", info);
 }
 
+// Keyed by the offending value, like the override report: one bad stored list applies to the whole
+// fleet. Bounded and TTL'd because the refresh runs once per cache TTL per process, which would
+// otherwise repeat this line for as long as the value stays broken.
+const reportedSetParseFailures = singleton(
+  "runOpsMintShardReportedSetParseFailures",
+  () => new BoundedTtlCache<true>(REPORT_TTL_MS, REPORT_MAX_ENTRIES)
+);
+
+// The stored list degrades to empty on a parse failure, which is the correct fail-safe but reverts
+// the whole fleet to gen-1 minting. `shard-set read failed` never covers this: the read SUCCEEDED.
+function reportSetParseFailed(failure: { key: string; value: string; error: unknown }): void {
+  const cacheKey = `${failure.key}:${failure.value}`;
+  if (reportedSetParseFailures.get(cacheKey) !== undefined) return;
+  reportedSetParseFailures.set(cacheKey, true);
+  logger.error("[runOpsMintShard] stored shard set is unparseable; minting gen-1 (fail-safe)", {
+    key: failure.key,
+    value: failure.value,
+    error: failure.error instanceof Error ? failure.error.message : failure.error,
+  });
+}
+
 /**
  * Which shard an environment mints new roots into. Call only after resolveRunIdMintKind has
  * returned "runOpsId". Returns "new" to mean a gen-1 run-ops id, which is today's behaviour.
@@ -94,5 +115,6 @@ export async function resolveMintShard(environment: {
     onOverrideRejected: reportOverrideRejected,
     onReadFailed: (error) =>
       logger.error("[runOpsMintShard] shard-set read failed; minting gen-1 (fail-safe)", { error }),
+    onSetParseFailed: reportSetParseFailed,
   });
 }
diff --git a/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts
new file mode 100644
index 00000000000..c5f6f03e05f
--- /dev/null
+++ b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts
@@ -0,0 +1,47 @@
+import { UnknownShardKey } from "@internal/run-store";
+import { describe, expect, it } from "vitest";
+import { undefinedOnUnroutableId } from "./unroutableRead.server";
+
+// `RoutingRunStore.findRun` is not async: it resolves the shard and throws before returning
+// anything, so these stand in for a read that fails while the call expression is still being
+// evaluated rather than one that returns a rejected promise.
+function throwsWhileRouting(): Promise<string | null> {
+  throw new UnknownShardKey("q", ["legacy", "new"]);
+}
+
+function rejectsLater(): Promise<string | null> {
+  return Promise.reject(new UnknownShardKey("q", ["legacy", "new"]));
+}
+
+describe("undefinedOnUnroutableId", () => {
+  it("returns undefined when the read throws synchronously while routing", async () => {
+    await expect(
+      undefinedOnUnroutableId(() => throwsWhileRouting(), { runParam: "run_x" })
+    ).resolves.toBeUndefined();
+  });
+
+  it("returns undefined when the read rejects", async () => {
+    await expect(
+      undefinedOnUnroutableId(() => rejectsLater(), { runParam: "run_x" })
+    ).resolves.toBeUndefined();
+  });
+
+  it("passes a successful read through untouched", async () => {
+    await expect(undefinedOnUnroutableId(async () => "found", { runParam: "run_x" })).resolves.toBe(
+      "found"
+    );
+  });
+
+  it("rethrows anything that is not an unroutable id", async () => {
+    const boom = new Error("connection refused");
+
+    await expect(
+      undefinedOnUnroutableId(
+        () => {
+          throw boom;
+        },
+        { runParam: "run_x" }
+      )
+    ).rejects.toBe(boom);
+  });
+});
diff --git a/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts
new file mode 100644
index 00000000000..d209ff46ad6
--- /dev/null
+++ b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts
@@ -0,0 +1,24 @@
+import { UnknownShardKey } from "@internal/run-store";
+import { logger } from "~/services/logger.server";
+
+// An id naming a shard with no configured store cannot locate a row, so the caller's own
+// not-found path is the right answer rather than a 500. Logged so a dropped shard key still
+// alarms instead of reading as an absent run.
+//
+// A thunk, not a promise: `RoutingRunStore.findRun` is not async and routes before returning, so
+// an unroutable id throws while the argument is still being evaluated.
+export async function undefinedOnUnroutableId<T>(
+  read: () => Promise<T>,
+  context: Record<string, unknown>
+): Promise<T | undefined> {
+  try {
+    return await read();
+  } catch (error) {
+    if (error instanceof UnknownShardKey) {
+      logger.warn("Unroutable id treated as not found", { ...context, error: error.message });
+      return undefined;
+    }
+
+    throw error;
+  }
+}
diff --git a/apps/webapp/app/v3/runStore.server.ts b/apps/webapp/app/v3/runStore.server.ts
index 7f51a940cec..8441f82ea1a 100644
--- a/apps/webapp/app/v3/runStore.server.ts
+++ b/apps/webapp/app/v3/runStore.server.ts
@@ -135,9 +135,28 @@ const routingStoreMetrics: RoutingStoreMetrics = singleton("routingStoreMetrics"
     labelNames: ["from", "to"],
     registers: [metricsRegister],
   });
+  // The only per-shard series /metrics carries, so a cohort ramp is visible per shard while it
+  // happens. On every routed operation: the label child is resolved once per shard and cached,
+  // because `inc({ shard })` hashes a fresh label object on each call. Cardinality is bounded by
+  // the shard alphabet plus the two reserved keys.
+  const shardRouted = new Counter({
+    name: "runops_shard_routed_total",
+    help: "Operations routed to a shard by an id that resolves to it alone. Fan-outs, probes and fallback legs resolve no single shard and are excluded.",
+    labelNames: ["shard"],
+    registers: [metricsRegister],
+  });
+  const shardRoutedChildren = new Map<string, { inc: (value?: number) => void }>();
   return {
     recordDuplicateId: (shardKeys) => duplicateId.inc({ shard_keys: shardKeys.join(",") }),
     recordWaitpointProbeFallback: (from, to) => probeFallback.inc({ from, to }),
+    recordShardRouted: (shardKey) => {
+      let child = shardRoutedChildren.get(shardKey);
+      if (child === undefined) {
+        child = shardRouted.labels(shardKey);
+        shardRoutedChildren.set(shardKey, child);
+      }
+      child.inc();
+    },
   };
 });
 
diff --git a/apps/webapp/test/runsReplicationMetrics.test.ts b/apps/webapp/test/runsReplicationMetrics.test.ts
new file mode 100644
index 00000000000..e6f8bf43f97
--- /dev/null
+++ b/apps/webapp/test/runsReplicationMetrics.test.ts
@@ -0,0 +1,44 @@
+// The alarmable half of the "publication carries no tables" failure: the client reports it, and
+// this counter is what turns that report into a series an alert can fire on, per source.
+import { PublicationMisconfiguredError } from "@internal/replication";
+import { Registry, type RegistryContentType } from "prom-client";
+import { describe, expect, it } from "vitest";
+import { buildRunsReplicationSourceMetrics } from "~/services/runsReplicationMetrics.server";
+
+function freshRegister() {
+  return new Registry<RegistryContentType>();
+}
+
+function misconfigured(publicationName: string) {
+  return new PublicationMisconfiguredError(
+    `Publication '${publicationName}' exists but has NO TABLES configured.`,
+    { publicationName, table: "TaskRun" }
+  );
+}
+
+describe("runs-replication source metrics", () => {
+  it("counts a publication misconfiguration against the source that reported it", async () => {
+    const register = freshRegister();
+    const metrics = buildRunsReplicationSourceMetrics(register);
+
+    metrics.recordSourceError({ sourceId: "shard-a", error: misconfigured("runs_shard_a_pub") });
+    metrics.recordSourceError({ sourceId: "shard-a", error: misconfigured("runs_shard_a_pub") });
+    metrics.recordSourceError({ sourceId: "new", error: misconfigured("runs_new_pub") });
+
+    const exposed = await register.metrics();
+    expect(exposed).toContain(
+      'runs_replication_publication_misconfigured_total{source="shard-a"} 2'
+    );
+    expect(exposed).toContain('runs_replication_publication_misconfigured_total{source="new"} 1');
+  });
+
+  it("leaves the counter alone for any other client error", async () => {
+    const register = freshRegister();
+    const metrics = buildRunsReplicationSourceMetrics(register);
+
+    metrics.recordSourceError({ sourceId: "legacy", error: new Error("connection terminated") });
+
+    const exposed = await register.metrics();
+    expect(exposed).not.toContain('source="legacy"');
+  });
+});
diff --git a/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts b/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts
new file mode 100644
index 00000000000..cf33af14a43
--- /dev/null
+++ b/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts
@@ -0,0 +1,86 @@
+// End-to-end for the failure a configured-but-empty publication caused in production: the source
+// replicated nothing, boot passed (a source WAS configured), and the only symptom was a log line
+// every 30s. This drives the real service against a real publication with no tables and asserts the
+// whole chain now produces a number: client -> onSourceError -> the /metrics counter.
+import { ClickHouse } from "@internal/clickhouse";
+import { PublicationMisconfiguredError } from "@internal/replication";
+import { replicationContainerTest } from "@internal/testcontainers";
+import { setTimeout } from "node:timers/promises";
+import { Registry, type RegistryContentType } from "prom-client";
+import { buildRunsReplicationSourceMetrics } from "~/services/runsReplicationMetrics.server";
+import { RunsReplicationService } from "~/services/runsReplicationService.server";
+import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
+
+vi.setConfig({ testTimeout: 90_000 });
+
+describe("RunsReplicationService — a source whose publication carries no tables", () => {
+  replicationContainerTest(
+    "reports the source error, which lands on the alarmable counter",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      // The production shape: the publication exists, so the client adopts it rather than creating
+      // one, and it carries no tables — so this source's WAL never reaches ClickHouse.
+      await prisma.$executeRawUnsafe(`CREATE PUBLICATION empty_pub;`);
+
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication",
+        logLevel: "warn",
+      });
+
+      const register = new Registry<RegistryContentType>();
+      const metrics = buildRunsReplicationSourceMetrics(register);
+      const reported: Array<{ sourceId: string; error: unknown }> = [];
+
+      const service = new RunsReplicationService({
+        clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
+        pgConnectionUrl: postgresContainer.getConnectionUri(),
+        serviceName: "runs-replication",
+        slotName: "empty_pub_slot",
+        publicationName: "empty_pub",
+        redisOptions,
+        flushIntervalMs: 100,
+        flushBatchSize: 1,
+        leaderLockTimeoutMs: 5000,
+        leaderLockExtendIntervalMs: 1000,
+        logLevel: "error",
+        sources: [
+          {
+            id: "shard-a",
+            pgConnectionUrl: postgresContainer.getConnectionUri(),
+            slotName: "empty_pub_slot",
+            publicationName: "empty_pub",
+            originGeneration: 2,
+          },
+        ],
+        onSourceError: (info) => {
+          reported.push(info);
+          metrics.recordSourceError(info);
+        },
+      });
+
+      try {
+        await service.start();
+
+        const deadline = Date.now() + 20_000;
+        while (
+          !reported.some((r) => r.error instanceof PublicationMisconfiguredError) &&
+          Date.now() < deadline
+        ) {
+          await setTimeout(250);
+        }
+      } finally {
+        await service.shutdown();
+      }
+
+      const misconfigured = reported.filter(
+        (r) => r.error instanceof PublicationMisconfiguredError
+      );
+      expect(misconfigured.length).toBeGreaterThan(0);
+      expect(misconfigured[0]?.sourceId).toBe("shard-a");
+
+      expect(await register.metrics()).toContain(
+        'runs_replication_publication_misconfigured_total{source="shard-a"}'
+      );
+    }
+  );
+});
diff --git a/apps/webapp/test/waitpointTokenUnroutableId.test.ts b/apps/webapp/test/waitpointTokenUnroutableId.test.ts
new file mode 100644
index 00000000000..a34dda4d641
--- /dev/null
+++ b/apps/webapp/test/waitpointTokenUnroutableId.test.ts
@@ -0,0 +1,106 @@
+// `resolveShard` is pure id-shape, so any 24-char base32hex core plus `[a-z0-9]` plus "2" parses as
+// a gen-2 id naming a shard — including a shard no topology configures. The run routes already
+// answer 404 for that, but the waitpoint-token routes wrap their body in a catch that turns every
+// non-Response error into a 500, so a caller could induce a 5xx (and trip a canary) at will.
+//
+// Both routes are driven for real: the routing store is a REAL RoutingRunStore (it is the thing that
+// throws UnknownShardKey, from its own id-shape routing — the sub-stores are never reached), and the
+// route code under test is the exported action / the handler the api-builder captured.
+import { describe, expect, vi } from "vitest";
+
+const H = vi.hoisted(() => ({
+  handlers: [] as Array<{ config: any; handler: any }>,
+  store: undefined as any,
+}));
+
+vi.mock("~/v3/runStore.server", () => ({
+  runStore: new Proxy(
+    {},
+    {
+      get(_t, prop) {
+        const store = H.store;
+        if (!store) throw new Error("test bug: H.store not initialised before handler ran");
+        const value = store[prop];
+        return typeof value === "function" ? value.bind(store) : value;
+      },
+    }
+  ),
+}));
+
+vi.mock("~/services/routeBuilders/apiBuilder.server", () => ({
+  anyResource: (x: unknown) => x,
+  createActionApiRoute: (config: any, handler: any) => {
+    H.handlers.push({ config, handler });
+    return { action: vi.fn(), loader: vi.fn() };
+  },
+}));
+
+import { RoutingRunStore, type RunStore } from "@internal/run-store";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { action as callbackAction } from "~/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash";
+
+// 24-char base32hex core + shard char "z" + version "2": a well-formed gen-2 id whose shard the
+// router below has no store for.
+const UNCONFIGURED_SHARD_WAITPOINT_ID = "waitpoint_" + "c".repeat(24) + "z2";
+
+// Only "new" and "legacy" are configured, so a gen-2 id lands on UnknownShardKey inside the router.
+// The sub-stores are unreachable placeholders: the throw happens before any of them is consulted.
+function routerWithNoShards() {
+  const unreachable = new Proxy(
+    {},
+    {
+      get() {
+        throw new Error("no sub-store should be reached for an unconfigured shard key");
+      },
+    }
+  ) as RunStore;
+  return new RoutingRunStore({ new: unreachable, legacy: unreachable });
+}
+
+async function completeHandler() {
+  await import("~/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete");
+  const entry = H.handlers.find((h) => Boolean(h.config?.params?.shape?.waitpointFriendlyId));
+  if (!entry) throw new Error("complete-route handler was not captured");
+  return entry.handler as (args: any) => Promise<Response>;
+}
+
+describe("waitpoint-token routes with an id naming an unconfigured shard", () => {
+  it("answers 404, not 500, on the token completion route", async () => {
+    H.store = routerWithNoShards();
+    const handler = await completeHandler();
+
+    const thrown = await handler({
+      authentication: { environment: { id: "env_1" } },
+      body: { data: { ok: true } },
+      params: { waitpointFriendlyId: WaitpointId.toFriendlyId(UNCONFIGURED_SHARD_WAITPOINT_ID) },
+    }).then(
+      (response) => response,
+      (error) => error
+    );
+
+    expect(thrown).toBeInstanceOf(Response);
+    expect((thrown as Response).status).toBe(404);
+    // Not retryable: no number of retries makes a topology grow a store.
+    expect((thrown as Response).headers.get("x-should-retry")).toBe("false");
+  });
+
+  it("answers 404, not 500, on the HTTP-callback route", async () => {
+    H.store = routerWithNoShards();
+
+    const payload = JSON.stringify({ ok: true });
+    const response = await callbackAction({
+      request: new Request("http://localhost/callback", {
+        method: "POST",
+        headers: { "content-type": "application/json", "content-length": String(payload.length) },
+        body: payload,
+      }),
+      params: {
+        waitpointFriendlyId: WaitpointId.toFriendlyId(UNCONFIGURED_SHARD_WAITPOINT_ID),
+        hash: "whatever",
+      },
+      context: {},
+    } as any);
+
+    expect(response.status).toBe(404);
+  });
+});
diff --git a/internal-packages/replication/src/client.publicationMisconfigured.test.ts b/internal-packages/replication/src/client.publicationMisconfigured.test.ts
new file mode 100644
index 00000000000..0eeb5e6f0e6
--- /dev/null
+++ b/internal-packages/replication/src/client.publicationMisconfigured.test.ts
@@ -0,0 +1,44 @@
+// A publication that exists but carries no tables is the failure that replicated NOTHING in
+// production while every log line said "healthy" and boot passed: the source IS configured, so
+// the boot-time coverage assert cannot see it, and the client just retries every 30s.
+//
+// The client already detects it. What was missing is a way for the owner to COUNT it, which needs
+// the emitted error to be distinguishable from any other client error. This drives the real
+// client against a real publication with no tables and pins that type.
+import { postgresAndRedisTest } from "@internal/testcontainers";
+import { LogicalReplicationClient } from "./client.js";
+import { PublicationMisconfiguredError } from "./errors.js";
+
+describe("publication with no tables", () => {
+  postgresAndRedisTest(
+    "emits a typed PublicationMisconfiguredError rather than a bare client error",
+    async ({ postgresContainer, prisma, redisOptions }) => {
+      // The exact production shape: the publication exists, so the client adopts it instead of
+      // creating one, and validation finds it carries no tables at all.
+      await prisma.$executeRawUnsafe(`CREATE PUBLICATION no_tables_pub;`);
+
+      const client = new LogicalReplicationClient({
+        name: "no-tables",
+        slotName: "no_tables_slot",
+        publicationName: "no_tables_pub",
+        redisOptions,
+        table: "TaskRun",
+        pgConfig: { connectionString: postgresContainer.getConnectionUri() },
+      });
+
+      const errors: unknown[] = [];
+      client.events.on("error", (error) => errors.push(error));
+
+      await client.subscribe();
+      await client.shutdown();
+
+      const misconfigured = errors.filter((e) => e instanceof PublicationMisconfiguredError);
+      expect(misconfigured).toHaveLength(1);
+
+      const error = misconfigured[0] as PublicationMisconfiguredError;
+      expect(error.publicationName).toBe("no_tables_pub");
+      expect(error.table).toBe("TaskRun");
+      expect(error.message).toContain("NO TABLES configured");
+    }
+  );
+});
diff --git a/internal-packages/replication/src/client.ts b/internal-packages/replication/src/client.ts
index 1a128a41e80..a41d9b0c7d4 100644
--- a/internal-packages/replication/src/client.ts
+++ b/internal-packages/replication/src/client.ts
@@ -5,7 +5,7 @@ import { tryCatch } from "@trigger.dev/core/utils";
 import EventEmitter from "node:events";
 import { type ClientConfig, type Connection, Client } from "pg";
 import Redlock, { type Lock } from "redlock";
-import { LogicalReplicationClientError } from "./errors.js";
+import { LogicalReplicationClientError, PublicationMisconfiguredError } from "./errors.js";
 import {
   type PgoutputMessage,
   getPgoutputStartReplicationSQL,
@@ -609,7 +609,13 @@ export class LogicalReplicationClient {
           error: validationError,
         });
 
-        this.events.emit("error", new LogicalReplicationClientError(validationError));
+        this.events.emit(
+          "error",
+          new PublicationMisconfiguredError(validationError, {
+            publicationName: this.options.publicationName,
+            table: this.options.table,
+          })
+        );
         return false;
       }
 
@@ -672,7 +678,14 @@ export class LogicalReplicationClient {
 
   async #validatePublicationConfiguration(): Promise<string | null> {
     if (!this.client) {
-      return "Cannot validate publication configuration: client not connected";
+      // Not a misconfiguration: a disconnected client cannot be inspected, and reporting it as
+      // one would count a connectivity fault on the publication series. Unreachable today, since
+      // both callers guard on `this.client` first.
+      this.logger.debug("Skipping publication validation: client not connected", {
+        name: this.options.name,
+        publicationName: this.options.publicationName,
+      });
+      return null;
     }
 
     // Check if the publication has the correct table
diff --git a/internal-packages/replication/src/errors.ts b/internal-packages/replication/src/errors.ts
index 0521305a5e7..bd702c2bb25 100644
--- a/internal-packages/replication/src/errors.ts
+++ b/internal-packages/replication/src/errors.ts
@@ -3,3 +3,21 @@ export class LogicalReplicationClientError extends Error {
     super(message);
   }
 }
+
+/**
+ * The publication a source subscribes through exists but does not carry the table we replicate —
+ * commonly because it was created with no tables at all. Nothing throws and nothing stops: the
+ * client retries and logs, while that source replicates NOTHING, so every ClickHouse-fronted
+ * aggregate silently under-counts. Typed so a consumer can put a number on it and alarm.
+ */
+export class PublicationMisconfiguredError extends LogicalReplicationClientError {
+  readonly publicationName: string;
+  readonly table: string;
+
+  constructor(message: string, options: { publicationName: string; table: string }) {
+    super(message);
+    this.name = "PublicationMisconfiguredError";
+    this.publicationName = options.publicationName;
+    this.table = options.table;
+  }
+}
diff --git a/internal-packages/run-store/src/routingStoreMetrics.ts b/internal-packages/run-store/src/routingStoreMetrics.ts
index d4aa454d903..85fe89f6402 100644
--- a/internal-packages/run-store/src/routingStoreMetrics.ts
+++ b/internal-packages/run-store/src/routingStoreMetrics.ts
@@ -4,13 +4,21 @@
  *
  * runops_shard_duplicate_id_total       — one id returned by two shards that should be disjoint
  * runops_waitpoint_probe_fallback_total — a waitpoint was not on the store its id named
+ * runops_shard_routed_total             — an operation was routed to a shard, labelled by key
  */
 export type RoutingStoreMetrics = {
   recordDuplicateId(shardKeys: string[]): void;
   recordWaitpointProbeFallback(from: string, to: string): void;
+  /**
+   * Every routed operation, keyed by the shard it landed on — the only per-shard series the
+   * router emits, and what makes a cohort ramp visible while it is happening rather than after.
+   * On the hot path, so an implementation must not build a label object per call.
+   */
+  recordShardRouted(shardKey: string): void;
 };
 
 export const noopRoutingStoreMetrics: RoutingStoreMetrics = {
   recordDuplicateId() {},
   recordWaitpointProbeFallback() {},
+  recordShardRouted() {},
 };
diff --git a/internal-packages/run-store/src/runOpsStore.newMethods.test.ts b/internal-packages/run-store/src/runOpsStore.newMethods.test.ts
index 8a56057a117..d7da3c76fc3 100644
--- a/internal-packages/run-store/src/runOpsStore.newMethods.test.ts
+++ b/internal-packages/run-store/src/runOpsStore.newMethods.test.ts
@@ -229,6 +229,7 @@ for (const topology of TOPOLOGIES) {
       const { router } = buildRouter(topology, {
         recordDuplicateId: (keys) => seen.push(keys),
         recordWaitpointProbeFallback() {},
+        recordShardRouted() {},
       });
       await router.findManyWaitpointTags({ where: { environmentId: "env" } });
       // A duplicate confined to {legacy, new} is the known drain-mirror case and stays silent. Once a
diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts
index aff0d50764e..717cfb3f788 100644
--- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts
+++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts
@@ -349,7 +349,10 @@ describe("RoutingRunStore id-to-shard-key seam", () => {
   });
 });
 
-function buildNShardRouter(shardKeys: string[], opts: { aliasOf?: Record<string, string> } = {}) {
+function buildNShardRouter(
+  shardKeys: string[],
+  opts: { aliasOf?: Record<string, string>; routed?: string[] } = {}
+) {
   const log: Call[] = [];
   const newStore = fakeStore("new", log);
   const legacyStore = fakeStore("legacy", log);
@@ -365,6 +368,15 @@ function buildNShardRouter(shardKeys: string[], opts: { aliasOf?: Record<string,
     legacy: legacyStore,
     shards,
     resolveShard: (id: string) => id.split(":")[0]!,
+    ...(opts.routed
+      ? {
+          metrics: {
+            recordDuplicateId() {},
+            recordWaitpointProbeFallback() {},
+            recordShardRouted: (key: string) => opts.routed!.push(key),
+          },
+        }
+      : {}),
   });
   return { router, log, byKey };
 }
@@ -384,6 +396,62 @@ describe("RoutingRunStore #distinctStores — one entry per database", () => {
     expect(trace(log)).toEqual(["a:findRun"]);
   });
 
+  // Per-shard routing was invisible from outside the process: nothing the router emitted carried
+  // the shard it routed to, so a cohort ramp could only be inferred from the databases themselves.
+  it("counts every routed operation against the shard it landed on", async () => {
+    const routed: string[] = [];
+    const { router } = buildNShardRouter(["a", "b"], { routed });
+
+    await router.findRun({ id: "a:run_1" });
+    await router.findRun({ id: "b:run_1" });
+    await router.findRun({ id: "a:run_2" });
+
+    expect(routed).toEqual(["a", "b", "a"]);
+  });
+
+  // The counter is read as per-shard traffic during a ramp, so a lookup that had to search
+  // rather than route must not be attributed to whichever store answered.
+  it("does not count an unrouted lookup that misses", async () => {
+    const routed: string[] = [];
+    const { router } = buildNShardRouter(["a", "b"], { routed });
+
+    await router.findRun({ spanId: "nope" });
+
+    expect(routed).toEqual([]);
+  });
+
+  it("does not count an id-less operation against the default store", async () => {
+    const routed: string[] = [];
+    const { router } = buildNShardRouter(["a", "b"], { routed });
+
+    await router.findRun({ spanId: "s1" }, undefined);
+    await router.runInTransaction(undefined, async () => undefined);
+
+    expect(routed).toEqual([]);
+  });
+
+  it("does not count the legacy leg of an unrouted probe", async () => {
+    const routed: string[] = [];
+    const { router } = buildNShardRouter(["a", "b"], { routed });
+
+    await router.findRunsByIdempotencyKeys({
+      runtimeEnvironmentId: "env_1",
+      taskIdentifier: "task",
+      idempotencyKeys: ["k1"],
+    });
+
+    expect(routed).toEqual([]);
+  });
+
+  it("does not count anything before an operation is routed", async () => {
+    const routed: string[] = [];
+    buildNShardRouter(["a", "b"], { routed });
+
+    // Constructing the router touches every configured store to build #distinctStores; that is
+    // not traffic, and counting it would put a boot-time step in a per-shard traffic series.
+    expect(routed).toEqual([]);
+  });
+
   it("counts an aliased shard's database ONCE in a sum", async () => {
     // "a" aliases "new": two keys, one database.
     const { router, log } = buildNShardRouter(["a"], { aliasOf: { a: "new" } });
@@ -555,6 +623,7 @@ describe("RoutingRunStore merge precedence and duplicate alarm", () => {
       metrics: {
         recordDuplicateId: (k: string[]) => seen.push(k),
         recordWaitpointProbeFallback() {},
+        recordShardRouted() {},
       },
       seen,
     };
@@ -670,7 +739,15 @@ describe("RoutingRunStore countPendingWaitpoints — disjoint-sum partition", ()
         { key: "b", store: mk("b") },
       ],
       resolveShard: (id: string) => (id.includes(":") ? id.split(":")[0]! : "legacy"),
-      ...(spy ? { metrics: { recordDuplicateId: spy, recordWaitpointProbeFallback() {} } } : {}),
+      ...(spy
+        ? {
+            metrics: {
+              recordDuplicateId: spy,
+              recordWaitpointProbeFallback() {},
+              recordShardRouted() {},
+            },
+          }
+        : {}),
     });
     return { router, log };
   }
@@ -791,6 +868,7 @@ describe("RoutingRunStore waitpoint probes at N", () => {
       metrics: {
         recordDuplicateId() {},
         recordWaitpointProbeFallback: (from, to) => falls.push([from, to]),
+        recordShardRouted() {},
       },
     });
     await router.updateWaitpoint({ where: { id: "cuid_w1" }, data: {} } as never);
@@ -910,7 +988,11 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () =
       legacy: fakeStore("legacy", log, { batch: { id: "batch_dup", from: "legacy" } }),
       shards: [{ key: "a", store: fakeStore("a", log, { batch: { id: "batch_dup", from: "a" } }) }],
       resolveShard: (id: string) => id.split(":")[0]!,
-      metrics: { recordDuplicateId: (k) => seen.push(k), recordWaitpointProbeFallback() {} },
+      metrics: {
+        recordDuplicateId: (k) => seen.push(k),
+        recordWaitpointProbeFallback() {},
+        recordShardRouted() {},
+      },
     });
     const batch = (await router.findBatchTaskRunById("batch_dup")) as { from: string } | null;
     // batchTriggerV3 writes raw to the control plane while runEngine routes by id, so this is a
@@ -928,7 +1010,11 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () =
       legacy: fakeStore("legacy", log, { runs: [{ id: "dup", from: "legacy" }] }),
       shards: [{ key: "a", store: fakeStore("a", log, { runs: [{ id: "dup", from: "a" }] }) }],
       resolveShard: (id: string) => id.split(":")[0]!,
-      metrics: { recordDuplicateId: (k) => seen.push(k), recordWaitpointProbeFallback() {} },
+      metrics: {
+        recordDuplicateId: (k) => seen.push(k),
+        recordWaitpointProbeFallback() {},
+        recordShardRouted() {},
+      },
     });
     await router.findRun({ spanId: "span_x" });
     expect(seen).toEqual([["legacy", "a"]]);
diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts
index fa923244d06..fe2e9fa61a8 100644
--- a/internal-packages/run-store/src/runOpsStore.ts
+++ b/internal-packages/run-store/src/runOpsStore.ts
@@ -170,11 +170,16 @@ export class RoutingRunStore implements RunStore {
     // it shares its target's database, and a second leg over one database double-counts a sum.
     // The discriminator is the DECLARATION, not object identity — the wiring layer may build a
     // second store object over a shared client.
+    // Direct map lookup rather than #shardStore: every key here comes from #precedence, so it is
+    // configured by construction, and #shardStore counts a ROUTED operation — building the list
+    // must not bump the per-shard counter at boot.
+    // Assigned before #distinctStores: reintroducing #shardStore below would otherwise throw on
+    // an unassigned #metrics at construction, in every deployment, at boot.
+    this.#metrics = options.metrics ?? noopRoutingStoreMetrics;
+
     this.#distinctStores = this.#precedence
       .filter((key) => !aliasedKeys.has(key))
-      .map((key) => ({ key, store: this.#shardStore(key) }));
-
-    this.#metrics = options.metrics ?? noopRoutingStoreMetrics;
+      .map((key) => ({ key, store: this.#shards.get(key)! }));
     this.#logger = options.logger ?? new Logger("RoutingRunStore", "warn");
   }
 
@@ -207,6 +212,14 @@ export class RoutingRunStore implements RunStore {
     return store;
   }
 
+  // Counted separately from #shardStore, which probes, fan-outs and fallback legs also go through.
+  // Only a key an id resolved to on its own is traffic for that shard.
+  #shardStoreRoutedById(key: ShardKey): RunStore {
+    const store = this.#shardStore(key);
+    this.#metrics.recordShardRouted(key);
+    return store;
+  }
+
   // A duplicate id is EXPECTED across the gen-1 pair (drain mirrors a token onto both). Any other
   // combination breaks id-determinism: alarm, but keep the deterministic pick.
   #reportDuplicateId(id: string, shardKeys: ShardKey[]): void {
@@ -455,7 +468,7 @@ export class RoutingRunStore implements RunStore {
 
   // Route an existing run-ops id by residency. Throws on an unclassifiable id.
   #route(id: string): RunStore {
-    return this.#shardStore(this.#shardKeyOf(id));
+    return this.#shardStoreRoutedById(this.#shardKeyOf(id));
   }
 
   // Best-effort shard key; falls back to #idlessRouteShard when the id is absent. Classification is
@@ -473,7 +486,10 @@ export class RoutingRunStore implements RunStore {
   }
 
   #routeOrNew(id: string | undefined): RunStore {
-    return this.#shardStore(this.#routeKeyOrDefault(id));
+    const key = this.#routeKeyOrDefault(id);
+    // An absent id fell back to the default store rather than resolving anywhere, so it is not
+    // traffic attributable to that shard.
+    return typeof id === "string" ? this.#shardStoreRoutedById(key) : this.#shardStore(key);
   }
 
   // WRITE routing is pure id-shape (cuid → LEGACY, run-ops id → NEW). A LEGACY-classified id is