fix(run-ops): align single-source leader-lock id, guard shutdown stop, fix test status type

- Register the implicit single source with id "legacy" so its leader-lock key
  matches the id the admin status route probes; otherwise leadership always
  reads false in the non-split config.
- Guard the shutdown-path client.stop() fan-out against re-firing per incoming
  transaction and add a catch so rejections don't surface as unhandled.
- Use the TaskRunStatus type alias (not the const value) for status annotations
  in the dual-source dedup tests.

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

Author: d-cs

apps/webapp/app/services/runsReplicationInstance.server.ts

@@ -138,6 +138,17 @@ function initializeRunsReplicationInstance() {
     pgConnectionUrl: DATABASE_URL,
     slotName: env.RUN_REPLICATION_SLOT_NAME,
     publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
+    // Explicit legacy source so the leader-lock key matches the id the status
+    // route probes from the registry below.
+    sources: [
+      {
+        id: "legacy",
+        pgConnectionUrl: DATABASE_URL,
+        slotName: env.RUN_REPLICATION_SLOT_NAME,
+        publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
+        originGeneration: env.RUN_REPLICATION_LEGACY_ORIGIN_GENERATION,
+      },
+    ],
   });
 
   // Register the live handle so the status route + lifecycle routes can find it.

apps/webapp/app/services/runsReplicationService.server.ts

@@ -162,6 +162,7 @@ export class RunsReplicationService {
   private logger: Logger;
   private _isShuttingDown = false;
   private _isShutDownComplete = false;
+  private _shutdownStopInFlight = false;
   private _tracer: Tracer;
   private _meter: Meter;
   private _acknowledgeTimeoutMs: number;
@@ -631,11 +632,18 @@ export class RunsReplicationService {
 
     if (this._isShuttingDown) {
       // A global shutdown stops every source's client; mark complete once all
-      // have stopped. For a single source this is identical to the prior
-      // "stop the one client, then mark complete" behavior.
-      Promise.all(Array.from(this._sources.values()).map((r) => r.client.stop())).finally(() => {
-        this._isShutDownComplete = true;
-      });
+      // have stopped. Guard against re-firing per incoming transaction, and
+      // swallow client.stop() rejections so they don't surface as unhandled.
+      if (!this._shutdownStopInFlight) {
+        this._shutdownStopInFlight = true;
+        Promise.all(Array.from(this._sources.values()).map((r) => r.client.stop()))
+          .catch((error) => {
+            this.logger.error("Error stopping replication clients during shutdown", { error });
+          })
+          .finally(() => {
+            this._isShutDownComplete = true;
+          });
+      }
     }
 
     // If there are no events, do nothing

apps/webapp/test/runsReplicationService.part8.test.ts

@@ -1,6 +1,6 @@
 import { ClickHouse } from "@internal/clickhouse";
 import { createPostgresContainer, replicationContainerTest } from "@internal/testcontainers";
-import { PrismaClient } from "@trigger.dev/database";
+import { PrismaClient, type TaskRunStatus as TaskRunStatusType } from "@trigger.dev/database";
 import { setTimeout } from "node:timers/promises";
 import { z } from "zod";
 import { TaskRunStatus } from "~/database-types";
@@ -96,7 +96,7 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => {
         const seedFkRows = async (
           client: PrismaClient,
           tag: string,
-          status: TaskRunStatus,
+          status: TaskRunStatusType,
           friendlyId: string
         ) => {
           await client.organization.create({
@@ -275,7 +275,7 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => {
         const seedFkRows = async (
           client: PrismaClient,
           tag: string,
-          status: TaskRunStatus,
+          status: TaskRunStatusType,
           friendlyId: string
         ) => {
           await client.organization.create({
@@ -444,7 +444,7 @@ describe("RunsReplicationService (part 8/8) - dual-source dedup", () => {
           client: PrismaClient,
           tag: string,
           runId: string,
-          status: TaskRunStatus
+          status: TaskRunStatusType
         ) => {
           const orgId = `org_${tag}_${suffix}`;
           const projectId = `proj_${tag}_${suffix}`;