fix(webapp): wake unfiltered runs feeds on the native realtime backend

Runs subscriptions with no tag filter were never indexed for change
routing, so they only received updates from the periodic backstop poll
instead of sub-second change notifications. Route every change record
to zero-filter feeds and apply the same rule in the replay path.

Author: ericallam

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

@@ -88,6 +88,8 @@ type EnvState = {
   byBatchId: Map<string, Set<Feed>>;
   /** All tag feeds, for routing partial records (no tags) as hydrate-to-classify candidates. */
   tagFeeds: Set<Feed>;
+  /** Tag feeds with no tag filter — they match every record but are unreachable via byTag. */
+  unfilteredTagFeeds: Set<Feed>;
   /** When this env's channel subscription started (for the gap-coverage check). */
   subscribedAtMs: number;
   /** Latest record per run, insertion-ordered, for replaying inter-poll gaps to newly-armed feeds. */
@@ -241,6 +243,7 @@ export class EnvChangeRouter {
       byTag: new Map(),
       byBatchId: new Map(),
       tagFeeds: new Set(),
+      unfilteredTagFeeds: new Set(),
       subscribedAtMs: Date.now(),
       recent: new Map(),
     };
@@ -309,11 +312,11 @@ export class EnvChangeRouter {
       case "batch":
         return record.batchId != null && record.batchId === feed.filter.batchId;
       case "tag": {
-        // Partial record (no tags) = hydrate-to-classify candidate, like the live path.
-        if (record.tags === undefined) {
+        const tags = feed.filter.tags;
+        // Unfiltered feed matches everything; partial record (no tags) = hydrate-to-classify.
+        if (tags.length === 0 || record.tags === undefined) {
           return true;
         }
-        const tags = feed.filter.tags;
         return record.tags.some((tag) => tags.includes(tag));
       }
     }
@@ -368,6 +371,9 @@ export class EnvChangeRouter {
         break;
       case "tag":
         env.tagFeeds.add(feed);
+        if (feed.filter.tags.length === 0) {
+          env.unfilteredTagFeeds.add(feed);
+        }
         for (const tag of feed.filter.tags) {
           addToIndex(env.byTag, tag, feed);
         }
@@ -385,6 +391,7 @@ export class EnvChangeRouter {
         break;
       case "tag":
         env.tagFeeds.delete(feed);
+        env.unfilteredTagFeeds.delete(feed);
         for (const tag of feed.filter.tags) {
           removeFromIndex(env.byTag, tag, feed);
         }
@@ -436,6 +443,8 @@ export class EnvChangeRouter {
             addMatch(feed, record.runId);
           }
         }
+        // Unfiltered tag feeds match every record but live outside the index.
+        for (const feed of env.unfilteredTagFeeds) addMatch(feed, record.runId);
       } else {
         // Partial record (no membership data): route to every tag feed as a candidate to
         // hydrate-and-classify (rare; the publish side emits full records in practice).

apps/webapp/test/realtime/envChangeRouter.test.ts

@@ -86,6 +86,30 @@ describe("EnvChangeRouter", () => {
     reg.close();
   });
 
+  it("wakes an unfiltered tag feed (no tags) for every full record, live and via replay", async () => {
+    const rows = new Map([["r1", row("r1", { tags: ["a"] })]]);
+    const { router, src } = makeRouter(rows);
+
+    // Live path: a full record (tags defined) must reach the zero-filter feed even
+    // though it can never appear in the byTag index.
+    const reg = router.register("env_1", { kind: "tag", tags: [] }, []);
+    const wait = reg.waitForMatch(undefined, 1_000);
+    src.push("env_1", [record("r1", { tags: ["a"] })]);
+    const live = await wait;
+    expect(live.reason).toBe("notify");
+    expect(live.rows.map((m) => m.row.id)).toEqual(["r1"]);
+    reg.close();
+
+    // Replay path: the buffered record matches an unfiltered feed registered after the push.
+    const late = router.register("env_1", { kind: "tag", tags: [] }, [], {
+      replaySinceMs: Date.now() - 1_000,
+    });
+    const replayed = await late.waitForMatch(undefined, 1_000);
+    expect(replayed.reason).toBe("notify");
+    expect(replayed.rows.map((m) => m.row.id)).toEqual(["r1"]);
+    late.close();
+  });
+
   it("batch-hydrates ONCE and shares the serialized value across feeds matching the same run", async () => {
     const rows = new Map([["r1", row("r1", { tags: ["a"] })]]);
     const { router, src, hydrateSpy } = makeRouter(rows);