refactor(webapp,run-engine): route the realtime runs feed through one per-env change channel

Rework the new realtime runs backend (still behind its feature flag, with the
existing backend as the default) so every feed is a predicate over one
self-describing change record, published once per run change to a single
per-environment channel.

A per-instance router indexes the currently-held feeds by run, tag, and batch.
On a change it hydrates the affected runs once and serializes them once, then
fans the result to every matching feed, so one hot shared tag costs a single
query and serialize no matter how many feeds watch it. Newly triggered runs
surface immediately (the run-created event now carries tags and batch) rather
than waiting for a status change. An admission gate bounds how many cold
ClickHouse resolves run concurrently, so a mass reconnect across distinct
filters queues instead of stampeding the database. Live long-polls hold for
about 20 seconds to match the existing backend cadence.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -306,8 +306,10 @@ const EnvironmentSchema = z
     // "1" = run-changed signals are published and the per-org `realtimeBackend`
     // feature flag selects the backend per request.
     REALTIME_NOTIFIER_ENABLED: z.string().default("0"),
-    // Backstop wait before a live notifier request refetches the run (ms).
-    REALTIME_NOTIFIER_LIVE_POLL_TIMEOUT_MS: z.coerce.number().int().default(5_000),
+    // Backstop wait before a live notifier request refetches the run (ms). Matches
+    // Electric's ~20s live long-poll hold so the client polling cadence is unchanged
+    // across backends (a ±15% jitter is applied per request to avoid refetch herds).
+    REALTIME_NOTIFIER_LIVE_POLL_TIMEOUT_MS: z.coerce.number().int().default(20_000),
     // Hard cap on the tag-list snapshot size served by the notifier feed.
     REALTIME_NOTIFIER_MAX_LIST_RESULTS: z.coerce.number().int().default(1_000),
     // Short-TTL coalescing cache for the multi-run (tag-list/batch) resolve+hydrate.
@@ -333,6 +335,11 @@ const EnvironmentSchema = z
     // holding the long-poll (re-resolving cheaply) instead of returning an empty
     // up-to-date the client would immediately re-issue. "0" reverts to per-wake replies.
     REALTIME_NOTIFIER_HOLD_ON_EMPTY: z.string().default("1"),
+    // Max concurrent fresh ClickHouse resolves (cache misses) per instance. Caps the
+    // distinct-filter reconnect stampede: a mass reconnect of N feeds on N different filters
+    // queues to this many concurrent CH queries instead of firing all N at once. Same-filter
+    // bursts collapse via the single-flight cache before taking a permit. 0 disables the gate.
+    REALTIME_NOTIFIER_RESOLVE_ADMISSION_LIMIT: z.coerce.number().int().default(16),
 
     PUBSUB_REDIS_HOST: z
       .string()

apps/webapp/app/models/runtimeEnvironment.server.ts

@@ -237,10 +237,20 @@ export async function findEnvironmentBySlug(
   return environment ? toAuthenticated(environment) : null;
 }
 
+// The authenticated environment plus the run scalars the realtime publish needs.
+// Both come from one taskRun read — see findEnvironmentFromRun.
+export type EnvironmentFromRun = {
+  environment: AuthenticatedEnvironment;
+  runTags: string[];
+  batchId: string | null;
+};
+
 export async function findEnvironmentFromRun(
   runId: string,
   tx?: PrismaClientOrTransaction
-): Promise<AuthenticatedEnvironment | null> {
+): Promise<EnvironmentFromRun | null> {
+  // The include (no select) already pulls every taskRun scalar, so runTags/batchId
+  // ride along for free — no extra query for the realtime publish to send a full record.
   const taskRun = await (tx ?? $replica).taskRun.findFirst({
     where: {
       id: runId,
@@ -249,7 +259,14 @@ export async function findEnvironmentFromRun(
       runtimeEnvironment: { include: authIncludeBase },
     },
   });
-  return taskRun?.runtimeEnvironment ? toAuthenticated(taskRun.runtimeEnvironment) : null;
+  if (!taskRun?.runtimeEnvironment) {
+    return null;
+  }
+  return {
+    environment: toAuthenticated(taskRun.runtimeEnvironment),
+    runTags: taskRun.runTags,
+    batchId: taskRun.batchId,
+  };
 }
 
 export async function createNewSession(

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -12,6 +12,7 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
+import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
 import { applyMetadataMutationToBufferedRun } from "~/v3/mollifier/applyMetadataMutation.server";
@@ -184,6 +185,11 @@ const { action } = createActionApiRoute(
       return json({ error: "Internal Server Error" }, { status: 500 });
     }
     if (pgResult) {
+      // Mid-run metadata flush succeeded: publish a run-changed record so a live single-run
+      // feed reflects metadata.set() without waiting for the next lifecycle event (this
+      // path doesn't otherwise touch the engine event bus). envId is free; partial record,
+      // matched by runId. No-op when disabled.
+      publishChangeRecord({ runId, envId: env.id });
       return json(pgResult, { status: 200 });
     }
 

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -7,7 +7,7 @@ import { MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { logger } from "~/services/logger.server";
-import { publishRunChanged } from "~/services/realtime/runChangeNotifierInstance.server";
+import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 
 // Pull the existing tags out of a buffer entry's serialised payload so
@@ -91,8 +91,13 @@ export async function action({ request, params }: ActionFunctionArgs) {
           },
           data: { runTags: { push: newTags } },
         });
-        // Delegate a run-changed notify (no-op unless enabled).
-        publishRunChanged({ runId: taskRun.id, environmentId: env.id });
+        // Publish a run-changed record with the NEW tag set so tag feeds reindex
+        // (no-op unless enabled).
+        publishChangeRecord({
+          runId: taskRun.id,
+          envId: env.id,
+          tags: existing.concat(newTags),
+        });
         return json({ message: `Successfully set ${newTags.length} new tags.` }, { status: 200 });
       },
       // Buffer-applied patch path. The mutateSnapshot Lua deduplicates

apps/webapp/app/services/realtime/electricStreamProtocol.server.ts

@@ -273,6 +273,26 @@ export function buildRowsBody(changes: RowChange[], skipColumns: string[] = []):
   return JSON.stringify(messages);
 }
 
+/** A row change whose wire `value` was already serialized (once, shared across feeds by
+ * the EnvChangeRouter); the per-feed `operation` is applied here. */
+export type SerializedRowChange = {
+  runId: string;
+  value: Record<string, string | null>;
+  operation: "insert" | "update";
+};
+
+/** Like `buildRowsBody`, but from values serialized once per (runId, columnSet) upstream,
+ * so a run matching many feeds is serialized once and reused across their bodies. */
+export function buildRowsBodyFromSerialized(changes: SerializedRowChange[]): string {
+  const messages: ShapeMessage[] = changes.map((change) => ({
+    key: runShapeKey(change.runId),
+    value: change.value,
+    headers: { operation: change.operation },
+  }));
+  messages.push(UP_TO_DATE);
+  return JSON.stringify(messages);
+}
+
 export const INITIAL_OFFSET = "-1";
 
 /**