diff --git a/console/doc/design/20260902_sync_engine.md b/console/doc/design/20260902_sync_engine.md
index 88213bba869b0..3188fe703be68 100644
--- a/console/doc/design/20260902_sync_engine.md
+++ b/console/doc/design/20260902_sync_engine.md
@@ -69,7 +69,7 @@ count stays the same, and a collection can even be fed from an existing atom
 | `SubscribeManager` | `api/materialize/SubscribeManager.ts` | The SUBSCRIBE protocol: buffers rows per timestamp, folds the upsert stream into the current row set, exposes snapshots via `onChange`/`getSnapshot`. Holds its last snapshot through a resubscribe so the UI does not flash empty. |
 | `WebsocketConnectionManager` | `api/materialize/WebsocketConnectionManager.ts` | Connection lifecycle: subscribes to environment health and the current region on the jotai store, connects when healthy, reconnects with backoff, reconnects when the region's address changes, and tears down a socket left pointing at a departed region. |
 | Subscribe sessions | `api/materialize/subscribeSession.ts` | Composition: one constructor wires the manager, the connection manager, the sink, the region reset, and cache hydration, in one defined order. |
-| Hooks | `api/materialize/useSubscribe.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. |
+| Hooks | `api/materialize/useSubscribe.ts`, `useScopedSubscribeCollection.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. |
 | Collection bridge | `api/materialize/subscribeCollection.ts` | `createSubscribeCollection`: turns manager snapshots into incremental insert/update/delete writes on a TanStack DB collection, owns the scoped localStorage cache, and mirrors status (error, snapshotComplete) into a companion atom. |
 | Cache scope | `store/syncEngineCache.ts` | Derives the cache scope `organizationId\|regionId` as an atom, so hydration can be driven from the store. |
 
@@ -176,11 +176,13 @@ What the engine is not:
   sinks are transport-agnostic, which is the seam a shared upstream
   (cross-tab or cross-client dedup) can plug into without touching surfaces,
   but the engine alone does not provide that.
-- **Not for polled telemetry.** Expensive introspection queries
-  (utilization history, arrangement sizes) are react-query territory, with
-  polling intervals and its persistence tooling. The engine serves
-  catalog-shaped data that a SUBSCRIBE can maintain.
-- **Not for page-scoped, request-driven subscribes.** Those use
+- **Not for polled telemetry.** Introspection data no SUBSCRIBE maintains,
+  such as utilization beyond 24h or on environments that predate the indexed
+  views, is react-query territory, with polling intervals and its persistence
+  tooling. The engine serves data that a SUBSCRIBE can maintain, including
+  per-key page data (see per-key scoped collections below).
+- **Not for one-off, request-driven subscribes.** Subscribes whose rows
+  nothing reads after the component unmounts use
   `useSubscribe`/`useSubscribeManager`, which tie a manager to a component
   and expose state via `useSyncExternalStore`.
 
@@ -194,9 +196,9 @@ queries over the objects collection (fed from the app-wide `allObjects` atom)
 and the namespaces collection (its own subscribe), and renders from cache on
 revisit before the live snapshot lands.
 
-Declare configuration once at module scope. The hooks require a referentially
-stable options object, and a new object identity restarts the session, which
-makes accidental instability visible instead of masked.
+Declare configuration once at module scope, or memoize it. The hooks require
+a referentially stable options object, and a new object identity restarts the
+session, which makes accidental instability visible instead of masked.
 
 Atom-backed data set:
 
@@ -278,6 +280,33 @@ component keyed by a path pattern survives param-only navigation, including
 region switches, which is why region handling lives in the session rather
 than in mount/unmount.
 
+## Per-key scoped collections
+
+Page data keyed by an id, such as a cluster's utilization or a replica's
+arrangement sizes, lives in `createScopedSubscribeCollections`: one in-memory
+collection per tenant scope and key, dropped when the scope changes and never
+persisted. `useScopedSubscribeCollection` resolves the scope, picks the key's
+collection and runs the subscribe into it while mounted. The rows outlive the
+component, so a return to the page serves them at once, with `isRefreshing`
+set until the new subscribe's snapshot lands, and a view derived from them,
+such as a chart window, can change without resubscribing.
+
+```ts
+const samplesByCluster = createScopedSubscribeCollections<Sample>({
+  id: "samples",
+  getKey: (sample) => sample.id,
+});
+
+export function useLiveSamples(clusterId: string) {
+  // A new builder restarts the subscribe, so memoize it.
+  const subscribe = React.useMemo(
+    () => buildSubscribeQuery(buildSamplesQuery(clusterId), { upsertKey: "id" }),
+    [clusterId],
+  );
+  return useScopedSubscribeCollection(samplesByCluster, clusterId, subscribe);
+}
+```
+
 ## Testing
 
 Sessions and the collection bridge are plain objects, so their behavior is
diff --git a/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts b/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts
index ddf182a0b88fe..750379f4938ea 100644
--- a/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts
+++ b/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts
@@ -89,20 +89,6 @@ describe("buildLargestClusterReplicaQuery", () => {
     expect(result.size).toBe("scale=1,workers=1,mem=8GiB");
     expect(result.heapLimit).toBe("8589934592");
 
-    const {
-      rows: [named],
-    } = await executeSqlHttp(
-      buildLargestClusterReplicaQuery(cluster.id, "r1").compile(),
-      {
-        sessionVariables: {
-          search_path: mockedSearchPath,
-          cluster: QUICKSTART_CLUSTER,
-        },
-      },
-    );
-    expect(named.name).toBe("r1");
-    expect(named.heapLimit).toBe("4294967296");
-
     // Object sizes cover every process of a replica, so its heap limit does
     // too. Four 4GiB processes outrank one 8GiB process.
     await client.query(`
diff --git a/console/src/api/materialize/cluster/largestClusterReplica.ts b/console/src/api/materialize/cluster/largestClusterReplica.ts
index 17e398fd9cd22..b47c8b80d15d1 100644
--- a/console/src/api/materialize/cluster/largestClusterReplica.ts
+++ b/console/src/api/materialize/cluster/largestClusterReplica.ts
@@ -17,57 +17,52 @@ import { buildClusterReplicaHeapMetricsTable } from "~/api/materialize/expressio
 // process of the replica.
 const replicaHeapLimit = sql`(crhm.heap_limit * COALESCE(crs.processes, 1))`;
 
-export function buildLargestClusterReplicaQuery(
-  clusterId: string,
-  replicaName?: string,
-) {
-  const query = queryBuilder
-    .selectFrom("mz_cluster_replicas as cr")
-    .leftJoin(
-      buildClusterReplicaHeapMetricsTable().as("crhm"),
-      "crhm.replica_id",
-      "cr.id",
-    )
-    // Prefer a fully-hydrated replica, whose reported sizes are stable.
-    // Joining mz_compute_hydration_times before aggregating lets its
-    // replica_id index bound the scan to this cluster's replicas, unlike an
-    // aggregation over the whole environment's mz_hydration_statuses.
-    .leftJoin("mz_compute_hydration_times as ht", (join) =>
-      join
-        .onRef("ht.replica_id", "=", "cr.id")
-        // System introspection indexes may never report hydrated.
-        .on("ht.object_id", "not like", "s%"),
-    )
-    .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size")
-    .select([
-      "cr.name",
-      "cr.size",
-      sql<string | null>`${replicaHeapLimit}::text`.as("heapLimit"),
-    ])
-    .where("cr.cluster_id", "=", clusterId)
-    .groupBy([
-      "cr.id",
-      "cr.name",
-      "cr.size",
-      "crhm.heap_limit",
-      "crs.processes",
-    ])
-    // Prefer a replica whose compute objects are all hydrated, so its
-    // reported sizes are stable. time_ns is NULL until an object hydrates. A
-    // replica with no hydration rows, such as one just added, has bool_and
-    // over an all-NULL row and so sorts last. Then order by heap limit.
-    .orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`)
-    .orderBy(sql`${replicaHeapLimit} DESC NULLS LAST`)
-    .limit(1);
-  return replicaName === undefined
-    ? query
-    : query.where("cr.name", "=", replicaName);
+export function buildLargestClusterReplicaQuery(clusterId: string) {
+  return (
+    queryBuilder
+      .selectFrom("mz_cluster_replicas as cr")
+      .leftJoin(
+        buildClusterReplicaHeapMetricsTable().as("crhm"),
+        "crhm.replica_id",
+        "cr.id",
+      )
+      // Prefer a fully-hydrated replica, whose reported sizes are stable.
+      // Joining mz_compute_hydration_times before aggregating lets its
+      // replica_id index bound the scan to this cluster's replicas, unlike an
+      // aggregation over the whole environment's mz_hydration_statuses.
+      .leftJoin("mz_compute_hydration_times as ht", (join) =>
+        join
+          .onRef("ht.replica_id", "=", "cr.id")
+          // System introspection indexes may never report hydrated.
+          .on("ht.object_id", "not like", "s%"),
+      )
+      .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size")
+      .select([
+        "cr.id",
+        "cr.name",
+        "cr.size",
+        sql<string | null>`${replicaHeapLimit}::text`.as("heapLimit"),
+      ])
+      .where("cr.cluster_id", "=", clusterId)
+      .groupBy([
+        "cr.id",
+        "cr.name",
+        "cr.size",
+        "crhm.heap_limit",
+        "crs.processes",
+      ])
+      // Prefer a replica whose compute objects are all hydrated, so its
+      // reported sizes are stable. time_ns is NULL until an object hydrates. A
+      // replica with no hydration rows, such as one just added, has bool_and
+      // over an all-NULL row and so sorts last. Then order by heap limit.
+      .orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`)
+      .orderBy(sql`${replicaHeapLimit} DESC NULLS LAST`)
+      .limit(1)
+  );
 }
 
 export type LargestClusterReplicaParams = {
   clusterId: string;
-  /** Reads this replica, with the same heap limit, instead of the largest. */
-  replicaName?: string;
 };
 
 /**
@@ -84,7 +79,6 @@ export async function fetchLargestClusterReplica({
 }) {
   const compiledQuery = buildLargestClusterReplicaQuery(
     params.clusterId,
-    params.replicaName,
   ).compile();
   return executeSqlV2({
     queries: compiledQuery,
diff --git a/console/src/api/materialize/cluster/largestMaintainedQueries.ts b/console/src/api/materialize/cluster/largestMaintainedQueries.ts
index 81d2ae3cd0db8..7af2fff0661c6 100644
--- a/console/src/api/materialize/cluster/largestMaintainedQueries.ts
+++ b/console/src/api/materialize/cluster/largestMaintainedQueries.ts
@@ -15,6 +15,7 @@ import {
   executeSqlV2,
   queryBuilder,
 } from "~/api/materialize";
+import { buildSubscribeQuery } from "~/api/materialize/buildSubscribeQuery";
 
 export type LargestMaintainedQueriesParams = {
   replicaHeapLimit: number | null;
@@ -164,6 +165,31 @@ export function buildLargestMaintainedObjectSizesQuery({
   );
 }
 
+/** One object's arrangement size on a replica, in bytes. */
+export interface ReplicaObjectSize {
+  objectId: string;
+  size: number | null;
+}
+
+/**
+ * Live arrangement sizes of every object on a replica. Filtering on the
+ * literal replica id makes this a lookup on `mz_object_arrangement_sizes_ind`.
+ */
+export function buildReplicaObjectSizesSubscribe(replicaId: string) {
+  return buildSubscribeQuery<ReplicaObjectSize>(
+    queryBuilder
+      .selectFrom("mz_object_arrangement_sizes")
+      .select([
+        "object_id as objectId",
+        // NOTE: The cast is required. An int8 decodes to a BigInt, which the
+        // collection bridge's JSON.stringify row signature throws on.
+        sql<number | null>`size::float8`.as("size"),
+      ])
+      .where("replica_id", "=", replicaId),
+    { upsertKey: "objectId" },
+  );
+}
+
 export async function fetchLargestMaintainedObjectSizes({
   params,
   queryKey,
diff --git a/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts b/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts
new file mode 100644
index 0000000000000..0e3da82f21a6c
--- /dev/null
+++ b/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts
@@ -0,0 +1,90 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { CATALOG_SERVER_CLUSTER, SEARCH_PATH } from "~/api/materialize";
+import {
+  executeSqlHttp,
+  getMaterializeClient,
+  QUICKSTART_CLUSTER,
+} from "~/test/sql/materializeSqlClient";
+import { testdrive } from "~/test/sql/mzcompose";
+
+import { buildReplicaHeapLimitsQuery } from "./replicaHeapLimits";
+
+const MOCKED_SEARCH_PATH = "internal_test, " + SEARCH_PATH;
+
+describe("buildReplicaHeapLimitsQuery", () => {
+  it("reads each process's heap limit and use for the given replicas", async () => {
+    await testdrive(`
+      > CREATE SCHEMA internal_test;
+      > SET schema = internal_test;
+      > CREATE TABLE mz_cluster_replica_metrics (
+          replica_id text NOT NULL,
+          process_id uint8 NOT NULL,
+          cpu_nano_cores uint8,
+          memory_bytes uint8,
+          disk_bytes uint8,
+          heap_bytes uint8,
+          heap_limit uint8
+        );
+      > INSERT INTO mz_cluster_replica_metrics VALUES
+          ('u1', 0, 100, 1000, 0, 400, 2000),
+          ('u1', 1, NULL, NULL, NULL, NULL, NULL),
+          ('u2', 0, 100, 1000, 0, 500, 2000);
+    `);
+
+    const { rows } = await executeSqlHttp(
+      buildReplicaHeapLimitsQuery({ replicaIds: ["u1"] }).compile(),
+      {
+        sessionVariables: {
+          search_path: MOCKED_SEARCH_PATH,
+          cluster: QUICKSTART_CLUSTER,
+        },
+      },
+    );
+
+    // A process whose metrics couldn't be read keeps its row, with nulls.
+    expect(
+      [...rows].sort((a, b) => Number(a.processId) - Number(b.processId)),
+    ).toEqual([
+      {
+        replicaId: "u1",
+        processId: BigInt(0),
+        heapLimit: BigInt(2000),
+        memoryBytes: BigInt(1000),
+        heapBytes: BigInt(400),
+      },
+      {
+        replicaId: "u1",
+        processId: BigInt(1),
+        heapLimit: null,
+        memoryBytes: null,
+        heapBytes: null,
+      },
+    ]);
+  });
+
+  it("is a fast-path index lookup on the catalog server", async () => {
+    const client = await getMaterializeClient();
+    await client.query(`SET cluster = ${CATALOG_SERVER_CLUSTER}`);
+    const { sql, parameters } = buildReplicaHeapLimitsQuery({
+      replicaIds: ["u1", "u2"],
+    }).compile();
+
+    const {
+      rows: [plan],
+    } = await client.query(`EXPLAIN ${sql}`, [...parameters]);
+
+    const explained = String(Object.values(plan)[0]);
+    expect(explained).toContain("Explained Query (fast path)");
+    expect(explained).toContain(
+      "Index Lookup on mz_internal.mz_cluster_replica_metrics",
+    );
+  });
+});
diff --git a/console/src/api/materialize/cluster/replicaHeapLimits.ts b/console/src/api/materialize/cluster/replicaHeapLimits.ts
index 0ead185194e3d..44d4de554891e 100644
--- a/console/src/api/materialize/cluster/replicaHeapLimits.ts
+++ b/console/src/api/materialize/cluster/replicaHeapLimits.ts
@@ -12,9 +12,9 @@ import { QueryKey } from "@tanstack/react-query";
 import { executeSqlV2, queryBuilder } from "~/api/materialize";
 
 /**
- * Each process's heap limit (RAM plus swap) for the given replicas, from their
- * latest metrics. The literal id list makes this a lookup on
- * `mz_cluster_replica_metrics_ind`.
+ * Each process's heap limit (RAM plus swap) and heap use for the given
+ * replicas, from their latest metrics. The literal id list makes this a lookup
+ * on `mz_cluster_replica_metrics_ind`.
  *
  * NOTE: callers fold the per-process rows, because a GROUP BY here turns the
  * fast-path lookup into a dataflow.
@@ -31,6 +31,7 @@ export function buildReplicaHeapLimitsQuery({
       "process_id as processId",
       "heap_limit as heapLimit",
       "memory_bytes as memoryBytes",
+      "heap_bytes as heapBytes",
     ])
     .where("replica_id", "in", replicaIds);
 }
diff --git a/console/src/api/materialize/cluster/replicaUtilizationBinning.ts b/console/src/api/materialize/cluster/replicaUtilizationBinning.ts
index 18332809610de..c373852aa8c5a 100644
--- a/console/src/api/materialize/cluster/replicaUtilizationBinning.ts
+++ b/console/src/api/materialize/cluster/replicaUtilizationBinning.ts
@@ -477,3 +477,17 @@ export const hasUtilizationReading = (point: DataPoint) =>
 export const hasNoUtilizationMetrics = (graphData: ReplicaData[]) =>
   graphData.length > 0 &&
   graphData.every(({ data }) => !data.some(hasUtilizationReading));
+
+/**
+ * Server-binned rows inside a window that starts at `startDate`, in time order.
+ * Held rows from a wider window would otherwise stretch the chart past the
+ * selected range, and ENVELOPE UPSERT yields them unordered.
+ */
+export function clipBinnedRowsToWindow<T extends { bucketStart: Date }>(
+  rows: T[],
+  startDate: Date,
+): T[] {
+  return rows
+    .filter((row) => row.bucketStart.getTime() >= startDate.getTime())
+    .sort((a, b) => a.bucketStart.getTime() - b.bucketStart.getTime());
+}
diff --git a/console/src/api/materialize/subscribeCollection.test.ts b/console/src/api/materialize/subscribeCollection.test.ts
index 0372c7894146d..d5f24c160140a 100644
--- a/console/src/api/materialize/subscribeCollection.test.ts
+++ b/console/src/api/materialize/subscribeCollection.test.ts
@@ -10,6 +10,7 @@
 import { getStore } from "~/jotai";
 
 import {
+  createScopedSubscribeCollections,
   createSubscribeCollection,
   syncEngineCacheKey,
 } from "./subscribeCollection";
@@ -257,6 +258,39 @@ describe("createSubscribeCollection", () => {
     expect(getStore().get(statusAtom).snapshotComplete).toBe(false);
   });
 
+  it("marks held rows as refreshing until a new subscribe's snapshot lands", () => {
+    const { applySnapshot, statusAtom } = freshCollection();
+    applySnapshot(state([{ id: "1", name: "a" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+
+    // A new subscribe opens with an empty pre-snapshot.
+    applySnapshot(state([], false));
+    expect(getStore().get(statusAtom)).toMatchObject({
+      snapshotComplete: true,
+      isRefreshing: true,
+    });
+
+    applySnapshot(state([{ id: "1", name: "b" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+  });
+
+  it("marks the rows a reopened socket holds as refreshing", async () => {
+    const { collection, applySnapshot, statusAtom } = freshCollection();
+    applySnapshot(state([{ id: "1", name: "a" }]));
+
+    // A reopened socket re-emits its held rows until the new snapshot lands.
+    applySnapshot({ ...state([{ id: "1", name: "a" }]), resubscribing: true });
+    await flush();
+    expect(collection.get("1")?.name).toBe("a");
+    expect(getStore().get(statusAtom)).toMatchObject({
+      snapshotComplete: true,
+      isRefreshing: true,
+    });
+
+    applySnapshot(state([{ id: "1", name: "b" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+  });
+
   it("surfaces error and snapshotComplete on the status atom", () => {
     const { applySnapshot, statusAtom } = freshCollection();
 
@@ -302,3 +336,18 @@ describe("createSubscribeCollection", () => {
     }
   });
 });
+
+describe("createScopedSubscribeCollections", () => {
+  it("keeps one collection per scope and key, dropping other scopes' on a scope change", () => {
+    const collections = createScopedSubscribeCollections<Row>({
+      id: "scoped-test",
+      getKey: (row) => row.id,
+    });
+    const first = collections.get("org|regionA", "u1");
+    expect(collections.get("org|regionA", "u1")).toBe(first);
+    expect(collections.get("org|regionA", "u2")).not.toBe(first);
+
+    expect(collections.get("org|regionB", "u1")).not.toBe(first);
+    expect(collections.get("org|regionA", "u1")).not.toBe(first);
+  });
+});
diff --git a/console/src/api/materialize/subscribeCollection.ts b/console/src/api/materialize/subscribeCollection.ts
index c8c79691cfe24..a54816b99077c 100644
--- a/console/src/api/materialize/subscribeCollection.ts
+++ b/console/src/api/materialize/subscribeCollection.ts
@@ -57,8 +57,21 @@ function writePersistedRows<T>(key: string, rows: T[]) {
 export interface SubscribeCollectionStatus {
   error: SubscribeError | undefined;
   snapshotComplete: boolean;
+  /** Rows from an earlier snapshot are served, and no running subscribe has
+   * delivered a newer one yet. */
+  isRefreshing: boolean;
 }
 
+const IDLE_STATUS: SubscribeCollectionStatus = {
+  error: undefined,
+  snapshotComplete: false,
+  isRefreshing: false,
+};
+
+/** The status of a collection that nothing has fed, for a consumer whose
+ * collection is not known yet. */
+export const idleCollectionStatusAtom = atom(() => IDLE_STATUS);
+
 export interface SubscribeCollection<T extends object> {
   collection: Collection<T, string>;
   statusAtom: PrimitiveAtom<SubscribeCollectionStatus>;
@@ -103,10 +116,7 @@ export function createSubscribeCollection<T extends object>(options: {
   // `hydrate`, then kept current by `applySnapshot`.
   let desired = new Map<string, T>();
 
-  const statusAtom = atom<SubscribeCollectionStatus>({
-    error: undefined,
-    snapshotComplete: false,
-  });
+  const statusAtom = atom<SubscribeCollectionStatus>(IDLE_STATUS);
 
   // Captured when the collection starts syncing; cleared when it pauses.
   type WriteApi = Parameters<
@@ -214,7 +224,8 @@ export function createSubscribeCollection<T extends object>(options: {
     const current = store.get(statusAtom);
     if (
       current.error !== next.error ||
-      current.snapshotComplete !== next.snapshotComplete
+      current.snapshotComplete !== next.snapshotComplete ||
+      current.isRefreshing !== next.isRefreshing
     ) {
       store.set(statusAtom, next);
     }
@@ -227,7 +238,11 @@ export function createSubscribeCollection<T extends object>(options: {
     const isEmptyPreload =
       !state.snapshotComplete && state.data.length === 0 && !state.error;
     if (isEmptyPreload) {
-      writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete });
+      writeStatus({
+        error: undefined,
+        snapshotComplete: lastSnapshotComplete,
+        isRefreshing: lastSnapshotComplete,
+      });
       return;
     }
 
@@ -240,6 +255,7 @@ export function createSubscribeCollection<T extends object>(options: {
     writeStatus({
       error: state.error,
       snapshotComplete: state.snapshotComplete,
+      isRefreshing: Boolean(state.resubscribing),
     });
   };
 
@@ -296,7 +312,12 @@ export function createSubscribeCollection<T extends object>(options: {
     // On a scope change this also flushes the emptied set, so the collection
     // stops serving the previous scope's rows while the new snapshot loads.
     materialize();
-    writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete });
+    // Cached rows stand in until the live snapshot arrives.
+    writeStatus({
+      error: undefined,
+      snapshotComplete: lastSnapshotComplete,
+      isRefreshing: lastSnapshotComplete,
+    });
   };
 
   const clearPendingPersist = () => {
@@ -312,7 +333,11 @@ export function createSubscribeCollection<T extends object>(options: {
     lastSnapshotComplete = false;
     liveSnapshotApplied = false;
     materialize();
-    writeStatus({ error: undefined, snapshotComplete: false });
+    writeStatus({
+      error: undefined,
+      snapshotComplete: false,
+      isRefreshing: false,
+    });
   };
 
   const suspendPersistence = () => {
@@ -332,3 +357,46 @@ export function createSubscribeCollection<T extends object>(options: {
     suspendPersistence,
   };
 }
+
+export interface ScopedSubscribeCollections<T extends object> {
+  /** The collection for `key` under `scope`, created on first use. */
+  get: (scope: string, key: string) => SubscribeCollection<T>;
+  getKey: (row: T) => string;
+}
+
+/**
+ * In-memory collections, one per tenant scope (`organizationId|regionId`) and
+ * key, such as a cluster id. Keys repeat across environments, so a collection
+ * never serves another scope's rows. Rows stay in memory until the scope
+ * changes, which makes returning to a page instant. They are not persisted,
+ * since an entry per key would have no bound in localStorage.
+ */
+export function createScopedSubscribeCollections<T extends object>(options: {
+  /** Prefix of each collection's id. */
+  id: string;
+  getKey: (row: T) => string;
+}): ScopedSubscribeCollections<T> {
+  const { id, getKey } = options;
+  const byKey = new Map<string, SubscribeCollection<T>>();
+  let currentScope: string | undefined;
+  return {
+    get: (scope, key) => {
+      if (scope !== currentScope) {
+        // A return to an earlier scope gets fresh collections, so the dropped
+        // ones are never served again.
+        byKey.clear();
+        currentScope = scope;
+      }
+      let collection = byKey.get(key);
+      if (!collection) {
+        collection = createSubscribeCollection<T>({
+          id: `${id}|${scope}|${key}`,
+          getKey,
+        });
+        byKey.set(key, collection);
+      }
+      return collection;
+    },
+    getKey,
+  };
+}
diff --git a/console/src/api/materialize/subscribeSession.ts b/console/src/api/materialize/subscribeSession.ts
index 6481746a2ebc6..ef790d9d6b34f 100644
--- a/console/src/api/materialize/subscribeSession.ts
+++ b/console/src/api/materialize/subscribeSession.ts
@@ -185,6 +185,14 @@ export function createCollectionSubscribeSession<
       unsubscribeScope?.();
       keepAlive.unsubscribe();
       session.destroy();
+      // No subscribe backs the held rows any more, so they read as refreshing
+      // until a later session's snapshot lands, as behind a new subscribe's
+      // empty pre-snapshot.
+      target.applySnapshot({
+        data: [],
+        snapshotComplete: false,
+        error: undefined,
+      });
     },
   };
 }
diff --git a/console/src/api/materialize/useScopedSubscribeCollection.ts b/console/src/api/materialize/useScopedSubscribeCollection.ts
new file mode 100644
index 0000000000000..e05d05d313a12
--- /dev/null
+++ b/console/src/api/materialize/useScopedSubscribeCollection.ts
@@ -0,0 +1,58 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { useAtomValue } from "jotai";
+import { RawBuilder } from "kysely";
+import React from "react";
+
+import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache";
+
+import {
+  idleCollectionStatusAtom,
+  ScopedSubscribeCollections,
+} from "./subscribeCollection";
+import { SubscribeRow } from "./SubscribeManager";
+import { useGlobalSubscribeCollection } from "./useSubscribe";
+
+// NOTE: kept out of useSubscribe.ts. Importing ~/store/syncEngineCache there
+// closes an import cycle, observed to break UsageRoutes.test.tsx.
+/**
+ * Runs `subscribe` into the collection for `key` under the current tenant
+ * scope, for as long as the calling component is mounted. While the scope
+ * resolves, `collection` is undefined and no subscribe runs.
+ *
+ * `subscribe` must be referentially stable, since a new builder restarts the
+ * subscribe session. Its upsert key must identify a row as
+ * `collections.getKey` does.
+ */
+export function useScopedSubscribeCollection<T extends object>(
+  collections: ScopedSubscribeCollections<T>,
+  key: string,
+  subscribe: RawBuilder<T> | undefined,
+) {
+  const scope = useAtomValue(syncEngineCacheScopeLoadableAtom);
+  const collection =
+    scope.state === "hasData" && scope.data
+      ? collections.get(scope.data, key)
+      : undefined;
+  const options = React.useMemo(
+    () => ({
+      target: collection,
+      subscribe,
+      select: (row: SubscribeRow<T>) => row.data,
+      upsertKey: (row: SubscribeRow<T>) => collections.getKey(row.data),
+    }),
+    [collections, collection, subscribe],
+  );
+  useGlobalSubscribeCollection(options);
+  const status = useAtomValue(
+    collection?.statusAtom ?? idleCollectionStatusAtom,
+  );
+  return { collection, status, isScopeError: scope.state === "hasError" };
+}
diff --git a/console/src/api/materialize/useSubscribe.test.tsx b/console/src/api/materialize/useSubscribe.test.tsx
index cbd432a011e02..e3c9eb0c4ab84 100644
--- a/console/src/api/materialize/useSubscribe.test.tsx
+++ b/console/src/api/materialize/useSubscribe.test.tsx
@@ -159,6 +159,34 @@ describe("subscribeSession", () => {
     }
   });
 
+  it("marks a collection's held rows as refreshing once its session ends", async () => {
+    const store = setTwoRegions();
+    const target = createSubscribeCollection<Row>({
+      id: "session-test-destroy",
+      getKey: (row) => row.id,
+    });
+    const session = createCollectionSubscribeSession<Row, Row>({
+      store,
+      request,
+      upsertKey,
+      select: (row) => row.data,
+      target,
+    });
+    target.applySnapshot({
+      data: [{ id: "1" }],
+      snapshotComplete: true,
+      error: undefined,
+    });
+    session.destroy();
+
+    await flush();
+    expect(target.collection.size).toBe(1);
+    expect(store.get(target.statusAtom)).toMatchObject({
+      snapshotComplete: true,
+      isRefreshing: true,
+    });
+  });
+
   it("clears an atom-fed collection when the region switches", async () => {
     const store = setTwoRegions();
     const sourceAtom = atom<SubscribeState<Row>>(loadingState);
diff --git a/console/src/api/materialize/useSubscribe.ts b/console/src/api/materialize/useSubscribe.ts
index 50f87b319d92f..a766e9e0a1554 100644
--- a/console/src/api/materialize/useSubscribe.ts
+++ b/console/src/api/materialize/useSubscribe.ts
@@ -141,10 +141,11 @@ export function useGlobalUpsertSubscribe<T extends object, R = SubscribeRow<T>>(
 /**
  * Runs an app-session subscribe that feeds a TanStack DB collection, for as
  * long as the calling component is mounted. Mirrors useGlobalUpsertSubscribe
- * with createCollectionSubscribeSession owning the wiring.
+ * with createCollectionSubscribeSession owning the wiring. No subscribe runs
+ * while `subscribe` or `target` is undefined.
  *
- * `options` must be referentially stable (declare it at module scope); a new
- * object restarts the subscribe session.
+ * `options` must be referentially stable (declare it at module scope, or
+ * memoize it); a new object restarts the subscribe session.
  *
  * Note that the subscribe statement must have WITH (PROGRESS) and ENVELOPE UPSERT.
  */
@@ -154,18 +155,19 @@ export function useGlobalSubscribeCollection<
 >(
   options: UseSubscribeOptions<T, R> & {
     upsertKey: UpsertKeyFunction<T>;
-    target: SubscribeCollection<R>;
+    target: SubscribeCollection<R> | undefined;
     scopeAtom?: ScopeLoadableAtom;
   },
 ) {
   const store = useStore();
   const request = useSubscribeRequest(options.subscribe);
   React.useEffect(() => {
-    if (!request) return;
+    const { target } = options;
+    if (!request || !target) return;
     const session = createCollectionSubscribeSession<T, R>({
       store,
       request,
-      target: options.target,
+      target,
       scopeAtom: options.scopeAtom,
       upsertKey: options.upsertKey,
       select: options.select,
diff --git a/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx b/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx
index cc4d8204da1a2..b707111fdda03 100644
--- a/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx
+++ b/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx
@@ -78,11 +78,7 @@ const ClusterMetricsContent = ({ cluster }: { cluster: Cluster }) => {
           timePeriodMinutes={timePeriodMinutes}
           selectedReplicaId={selectedReplica?.id}
         />
-        <MemoryByObject
-          clusterId={cluster.id}
-          clusterName={cluster.name}
-          replicaName={selectedReplica?.name}
-        />
+        <MemoryByObject cluster={cluster} replica={selectedReplica} />
       </VStack>
     </MainContentContainer>
   );
diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx
index 91fa33d36a4e1..ab486a5cac537 100644
--- a/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx
+++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx
@@ -7,11 +7,15 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-import { screen, waitFor, within } from "@testing-library/react";
+import { fireEvent, screen, waitFor, within } from "@testing-library/react";
+import { http } from "msw";
 import React from "react";
 
+import { Cluster, Replica } from "~/api/materialize/cluster/clusterList";
+import { DatabaseObject } from "~/api/materialize/objects";
 import { ErrorCode, MzDataType } from "~/api/materialize/types";
 import {
+  buildColumn,
   buildColumns,
   buildSqlQueryHandlerV2,
   mapKyselyToTabular,
@@ -19,17 +23,22 @@ import {
 } from "~/api/mocks/buildSqlQueryHandler";
 import server from "~/api/mocks/server";
 import { getStore } from "~/jotai";
+import { buildClusterServerResponse } from "~/platform/clusters/clustersTestUtils";
 import { clusterQueryKeys } from "~/platform/clusters/queries";
+import { allObjectsCollection } from "~/store/allObjectsCollection";
+import { syncEngineCacheScopeAtom } from "~/store/syncEngineCache";
+import { mockSubscribeSocket } from "~/test/mockSubscribeSocket";
 import {
-  defaultRegionId,
   healthyEnvironment,
   renderComponent,
   setFakeEnvironment,
 } from "~/test/utils";
+import { assert } from "~/util";
 import { parseDbVersion } from "~/version/api";
 
 import { MemoryByObject } from "./MemoryByObject";
-import { segmentWidths } from "./memoryByObjectModel";
+import { otherMemory, segmentWidths } from "./memoryByObjectModel";
+import { sizesByReplica } from "./useLiveLargestObjects";
 
 const flags = vi.hoisted(() => ({ values: {} as Record<string, boolean> }));
 
@@ -37,19 +46,142 @@ vi.mock("~/hooks/useFlags", () => ({ useFlags: () => flags.values }));
 
 const HEAP_LIMIT = 4069523456;
 
+const buildReplica = (id: string, name: string): Replica => ({
+  id,
+  name,
+  size: "25cc",
+  disk: true,
+  statuses: [],
+});
+
+const buildCluster = (replicas: Replica[]) =>
+  buildClusterServerResponse({ id: "u1", name: "quickstart" }, replicas);
+
+/**
+ * Answers the page's heap reads: each process's latest heap limit and use, by
+ * replica id, and the size's process count. Build after setting the
+ * environment, since the query keys embed its version.
+ */
+const heapReadsHandlers = (
+  cluster: Cluster,
+  {
+    processes = 1,
+    heapLimit = String(HEAP_LIMIT),
+    heapBytes = {},
+  }: {
+    processes?: number;
+    heapLimit?: string | null;
+    /** Each process's heap use, by replica id. Unlisted replicas report none. */
+    heapBytes?: Record<string, string>;
+  } = {},
+) => [
+  buildSqlQueryHandlerV2({
+    queryKey: clusterQueryKeys.replicaHeapLimits({
+      replicaIds: cluster.replicas.map(({ id }) => id).sort(),
+    }),
+    results: mapKyselyToTabular({
+      columns: buildColumns([
+        "replicaId",
+        "processId",
+        "heapLimit",
+        "memoryBytes",
+        "heapBytes",
+      ]),
+      rows: cluster.replicas.flatMap(({ id }) =>
+        Array.from({ length: processes }, (_, processId) => ({
+          replicaId: id,
+          processId: String(processId),
+          heapLimit,
+          memoryBytes: "1",
+          heapBytes: heapBytes[id] ?? null,
+        })),
+      ),
+    }),
+  }),
+  buildSqlQueryHandlerV2({
+    queryKey: clusterQueryKeys.replicaSizeAllocations({ sizes: ["25cc"] }),
+    results: mapKyselyToTabular({
+      columns: [
+        buildColumn({ name: "size" }),
+        ...["processes", "cpuNanoCores", "memoryBytes", "diskBytes"].map(
+          (name) => buildColumn({ name, type_oid: MzDataType.float8 }),
+        ),
+      ],
+      rows: [
+        {
+          size: "25cc",
+          processes: String(processes),
+          cpuNanoCores: "1",
+          memoryBytes: "1",
+          diskBytes: "1",
+        },
+      ],
+    }),
+  }),
+];
+
+/** Counts largest replica queries, leaving their answers to other handlers. */
+const countLargestReplicaQueries = () => {
+  const requests = { count: 0 };
+  const handler = http.post("*/api/sql", ({ request }) => {
+    const queryKey = JSON.parse(
+      new URL(request.url).searchParams.get("query_key") ?? "[]",
+    );
+    if (queryKey[1]?.scope === "largestClusterReplica") requests.count += 1;
+    return undefined;
+  });
+  return { requests, handler };
+};
+
+/** The card under a replica select, as the Metrics page lays them out. */
+const ReplicaSwitcher = ({
+  cluster,
+  initialReplicaName,
+}: {
+  cluster: Cluster;
+  initialReplicaName?: string;
+}) => {
+  const [replicaName, setReplicaName] = React.useState(initialReplicaName);
+  return (
+    <>
+      <select
+        aria-label="Replica"
+        value={replicaName ?? ""}
+        onChange={(event) => setReplicaName(event.target.value || undefined)}
+      >
+        <option value="">All replicas</option>
+        {cluster.replicas.map(({ id, name }) => (
+          <option key={id} value={name}>
+            {name}
+          </option>
+        ))}
+      </select>
+      <MemoryByObject
+        cluster={cluster}
+        replica={cluster.replicas.find(({ name }) => name === replicaName)}
+      />
+    </>
+  );
+};
+
+/** The select's value for all replicas. */
+const ALL_REPLICAS = "";
+
+const selectReplica = (name: string) =>
+  fireEvent.change(screen.getByRole("combobox", { name: "Replica" }), {
+    target: { value: name },
+  });
+
 const largestReplicaHandler = (
   rows: Array<Record<string, string | null>>,
   options?: SqlHandlerOptions,
-  replicaName?: string,
 ) =>
   buildSqlQueryHandlerV2(
     {
-      queryKey: clusterQueryKeys.largestClusterReplica({
-        clusterId: "u1",
-        replicaName,
-      }),
+      queryKey: clusterQueryKeys.largestClusterReplica({ clusterId: "u1" }),
       results: mapKyselyToTabular({
         columns: buildColumns([
+          "id",
           "name",
           "size",
           { type_oid: MzDataType.numeric, name: "heapLimit" },
@@ -117,8 +249,12 @@ const largestObjectsHandler = (
     }),
   });
 
-const renderMemoryByObject = () =>
-  renderComponent(<MemoryByObject clusterId="u1" clusterName="quickstart" />);
+const R1 = buildReplica("u678", "r1");
+
+const renderMemoryByObject = (cluster = buildCluster([R1])) => {
+  server.use(...heapReadsHandlers(cluster));
+  return renderComponent(<MemoryByObject cluster={cluster} />);
+};
 
 describe("MemoryByObject", () => {
   beforeEach(() => {
@@ -148,25 +284,15 @@ describe("MemoryByObject", () => {
     expect(within(objects).getByText("5.09 GB (31.8%)")).toBeVisible();
   });
 
-  it("shows the named replica instead of the largest", async () => {
-    server.use(
-      largestReplicaHandler(
-        [{ name: "r1", size: "25cc", heapLimit: String(HEAP_LIMIT) }],
-        undefined,
-        "r1",
-      ),
-      largestObjectsHandler(),
-    );
-    await renderComponent(
-      <MemoryByObject
-        clusterId="u1"
-        clusterName="quickstart"
-        replicaName="r1"
-      />,
-    );
+  it("reads a named replica from the page instead of querying the largest", async () => {
+    const cluster = buildCluster([R1]);
+    const { requests, handler } = countLargestReplicaQueries();
+    server.use(handler, ...heapReadsHandlers(cluster), largestObjectsHandler());
+    await renderComponent(<MemoryByObject cluster={cluster} replica={R1} />);
 
     expect(await screen.findByText("% of heap limit on r1")).toBeVisible();
     expect(screen.getByText("5.09 GB (31.8%)")).toBeVisible();
+    expect(requests.count).toEqual(0);
   });
 
   it("links to the workflow graph while the maintained objects page is off", async () => {
@@ -259,7 +385,7 @@ describe("MemoryByObject", () => {
 
   it("shows nothing for a cluster without replicas", async () => {
     server.use(largestReplicaHandler([]));
-    await renderMemoryByObject();
+    await renderMemoryByObject(buildCluster([]));
 
     await waitFor(() =>
       expect(screen.queryByTestId("loading-spinner")).not.toBeInTheDocument(),
@@ -292,91 +418,315 @@ describe("MemoryByObject", () => {
       screen.getByRole("heading", { name: "Objects by Memory" }),
     ).toBeVisible();
   });
+});
 
-  it("reads sizes from mz_object_arrangement_sizes on Materialize >= v26.35", async () => {
-    const environment = {
-      ...healthyEnvironment,
-      status: {
-        ...healthyEnvironment.status,
-        version: parseDbVersion("v26.35.0 (ea0d129f)"),
-      },
-    };
-    // Query keys embed the environment version, so set it before building
-    // the handlers that match on them.
-    await setFakeEnvironment(getStore().set, defaultRegionId, environment);
-    server.use(
-      largestReplicaHandler([
-        { name: "r1", size: "25cc", heapLimit: String(HEAP_LIMIT) },
-      ]),
-      buildSqlQueryHandlerV2({
-        queryKey: clusterQueryKeys.largestMaintainedQueries({
-          clusterId: "u1",
-          clusterName: "quickstart",
-          limit: 10,
-          replicaName: "r1",
-          replicaHeapLimit: HEAP_LIMIT,
-          unifiedSizes: true,
-        }),
-        results: mapKyselyToTabular({
-          columns: buildColumns([
-            "object_id",
-            "size",
-            { type_oid: MzDataType.numeric, name: "memoryPercentage" },
-          ]),
-          rows: [
-            {
-              object_id: "u188",
-              size: "5469140917",
-              memoryPercentage: "31.8345902256",
-            },
-            // Missing from the names below: a dropped object's dataflow.
-            {
-              object_id: "u999",
-              size: "424919434",
-              memoryPercentage: "11.2686157226",
-            },
-          ],
-        }),
-      }),
-      buildSqlQueryHandlerV2({
-        queryKey: clusterQueryKeys.maintainedObjectNames(["u188", "u999"]),
-        results: mapKyselyToTabular({
-          columns: buildColumns([
-            "id",
-            "name",
-            "type",
-            "schemaName",
-            "databaseName",
-          ]),
-          rows: [
-            {
-              id: "u188",
-              name: "customer_view",
-              type: "materialized-view",
-              schemaName: "public",
-              databaseName: "materialize",
-            },
-          ],
-        }),
-      }),
-    );
-    await renderComponent(
-      <MemoryByObject clusterId="u1" clusterName="quickstart" />,
-      {
-        initializeState: ({ set }) =>
-          setFakeEnvironment(set, defaultRegionId, environment),
-      },
-    );
+// Sizes stream from mz_object_arrangement_sizes on 26.35 and later.
+const LIVE_SIZES_ENVIRONMENT = {
+  ...healthyEnvironment,
+  status: {
+    ...healthyEnvironment.status,
+    version: parseDbVersion("v26.45.0 (0000000000)"),
+  },
+};
+const GIB = 2 ** 30;
+
+const catalogObject = (id: string, name: string): DatabaseObject => ({
+  databaseName: "materialize",
+  databaseId: "u1",
+  name,
+  schemaName: "public",
+  schemaId: "u2",
+  id,
+  objectType: "materialized-view",
+  sourceType: null,
+  isWebhookTable: null,
+  clusterId: "u1",
+  clusterName: "quickstart",
+});
+
+const sizesSnapshot = (rows: Array<[string, number]>) => ({
+  columns: [
+    buildColumn({ name: "objectId" }),
+    buildColumn({ name: "size", type_oid: MzDataType.float8 }),
+  ],
+  rows: rows.map(([objectId, size]) => [objectId, String(size)]),
+});
+
+const setLiveSizesEnvironment = () =>
+  setFakeEnvironment(getStore().set, "aws/us-east-1", LIVE_SIZES_ENVIRONMENT);
+
+const renderLive = (element: React.ReactElement) =>
+  renderComponent(element, {
+    initializeState: ({ set }) =>
+      setFakeEnvironment(set, "aws/us-east-1", LIVE_SIZES_ENVIRONMENT),
+  });
+
+// Query keys embed the environment version, so set the environment before
+// building the handlers that match on them. Each test streams replicas of its
+// own: the sizes collections outlive a test.
+const renderLiveMemoryByObject = async ({
+  replicaId,
+  heapLimit = String(HEAP_LIMIT),
+}: {
+  replicaId: string;
+  heapLimit?: string | null;
+}) => {
+  await setLiveSizesEnvironment();
+  const cluster = buildCluster([buildReplica(replicaId, "r1")]);
+  server.use(
+    ...heapReadsHandlers(cluster),
+    largestReplicaHandler([
+      { id: replicaId, name: "r1", size: "25cc", heapLimit },
+    ]),
+    mockSubscribeSocket((sql, params) =>
+      sql.includes("mz_object_arrangement_sizes") && params.includes(replicaId)
+        ? sizesSnapshot([
+            ["u188", 2 * GIB],
+            ["u404", GIB],
+          ])
+        : undefined,
+    ),
+  );
+  return renderLive(<MemoryByObject cluster={cluster} />);
+};
+
+describe("MemoryByObject with streamed sizes", () => {
+  beforeEach(() => {
+    flags.values = {};
+    allObjectsCollection.applySnapshot({
+      data: [catalogObject("u188", "customer_view")],
+      snapshotComplete: true,
+      error: undefined,
+    });
+  });
+
+  it("names streamed objects from the catalog", async () => {
+    await renderLiveMemoryByObject({ replicaId: "u5" });
 
     const objects = await screen.findByRole("list", { name: "Objects" });
     expect(
       within(objects).getByRole("link", { name: "customer_view" }),
     ).toBeVisible();
-    expect(within(objects).getByText("5.09 GB (31.8%)")).toBeVisible();
-    expect(within(objects).getByText("u999")).toBeVisible();
+    // Not in the catalog, so a dropped object's dataflow.
+    expect(within(objects).getByText("u404")).toBeVisible();
+    expect(
+      within(objects).queryByRole("link", { name: "u404" }),
+    ).not.toBeInTheDocument();
+  });
+
+  it("lists sizes without a bar when the replica reports no heap limit", async () => {
+    await renderLiveMemoryByObject({ replicaId: "u6", heapLimit: null });
+
+    const objects = await screen.findByRole("list", { name: "Objects" });
+    expect(within(objects).getByText("customer_view")).toBeVisible();
+    expect(within(objects).getByText("2.00 GB")).toBeVisible();
+    expect(screen.getByText("Size on r1")).toBeVisible();
     expect(
-      within(objects).queryByRole("link", { name: "u999" }),
+      screen.queryByRole("img", { name: "Memory by object" }),
     ).not.toBeInTheDocument();
+    expect(screen.queryByText(/Infinity|NaN/)).not.toBeInTheDocument();
+  });
+
+  it("waits for the catalog before naming objects", async () => {
+    allObjectsCollection.reset();
+    await renderLiveMemoryByObject({ replicaId: "u7" });
+
+    const scope = await getStore().get(syncEngineCacheScopeAtom);
+    assert(scope);
+    const sizes = sizesByReplica.get(scope, "u7");
+    await waitFor(() =>
+      expect(getStore().get(sizes.statusAtom).snapshotComplete).toBe(true),
+    );
+    expect(screen.getByTestId("loading-spinner")).toBeVisible();
+    expect(screen.queryByText("u188")).not.toBeInTheDocument();
+    allObjectsCollection.applySnapshot({
+      data: [catalogObject("u188", "customer_view")],
+      snapshotComplete: true,
+      error: undefined,
+    });
+    expect(await screen.findByText("customer_view")).toBeVisible();
+  });
+
+  it("takes a named replica's heap limit across its processes", async () => {
+    await setLiveSizesEnvironment();
+    const replica = buildReplica("u20", "r1");
+    const cluster = buildCluster([replica]);
+    const { requests, handler } = countLargestReplicaQueries();
+    server.use(
+      handler,
+      ...heapReadsHandlers(cluster, {
+        processes: 2,
+        heapLimit: String(4 * GIB),
+      }),
+      mockSubscribeSocket((sql, params) =>
+        params.includes("u20") ? sizesSnapshot([["u188", 2 * GIB]]) : undefined,
+      ),
+    );
+    await renderLive(<MemoryByObject cluster={cluster} replica={replica} />);
+
+    const objects = await screen.findByRole("list", { name: "Objects" });
+    expect(within(objects).getByText("2.00 GB (25.0%)")).toBeVisible();
+    expect(requests.count).toEqual(0);
+  });
+
+  it("keeps the previous replica's objects on screen until the next replica's snapshot", async () => {
+    await setLiveSizesEnvironment();
+    const cluster = buildCluster([
+      buildReplica("u21", "r1"),
+      buildReplica("u22", "r2"),
+    ]);
+    const { requests, handler } = countLargestReplicaQueries();
+    let releaseR2 = () => {};
+    const r2Released = new Promise<void>((resolve) => {
+      releaseR2 = resolve;
+    });
+    server.use(
+      handler,
+      ...heapReadsHandlers(cluster),
+      mockSubscribeSocket(async (sql, params) => {
+        if (params.includes("u21")) return sizesSnapshot([["u188", GIB]]);
+        if (!params.includes("u22")) return undefined;
+        await r2Released;
+        return sizesSnapshot([["u404", 2 * GIB]]);
+      }),
+    );
+    await renderLive(
+      <ReplicaSwitcher cluster={cluster} initialReplicaName="r1" />,
+    );
+    expect(await screen.findByText("customer_view")).toBeVisible();
+
+    selectReplica("r2");
+    expect(await screen.findByTestId("refreshing-spinner")).toBeVisible();
+    expect(screen.getByText("% of heap limit on r1")).toBeVisible();
+    expect(screen.getByText("customer_view")).toBeVisible();
+    expect(screen.queryByTestId("loading-spinner")).not.toBeInTheDocument();
+
+    releaseR2();
+    expect(await screen.findByText("% of heap limit on r2")).toBeVisible();
+    expect(screen.getByText("u404")).toBeVisible();
+    expect(screen.queryByText("customer_view")).not.toBeInTheDocument();
+    expect(screen.queryByTestId("refreshing-spinner")).not.toBeInTheDocument();
+    expect(requests.count).toEqual(0);
+  });
+
+  it("queries the largest replica once all replicas are selected", async () => {
+    await setLiveSizesEnvironment();
+    const cluster = buildCluster([
+      buildReplica("u23", "r1"),
+      buildReplica("u24", "r2"),
+    ]);
+    const { requests, handler } = countLargestReplicaQueries();
+    server.use(
+      handler,
+      ...heapReadsHandlers(cluster),
+      largestReplicaHandler([
+        { id: "u24", name: "r2", size: "25cc", heapLimit: String(HEAP_LIMIT) },
+      ]),
+      mockSubscribeSocket((sql, params) =>
+        params.includes("u23") || params.includes("u24")
+          ? sizesSnapshot([["u188", GIB]])
+          : undefined,
+      ),
+    );
+    await renderLive(
+      <ReplicaSwitcher cluster={cluster} initialReplicaName="r1" />,
+    );
+    expect(await screen.findByText("% of heap limit on r1")).toBeVisible();
+    expect(requests.count).toEqual(0);
+
+    selectReplica(ALL_REPLICAS);
+    expect(await screen.findByText("% of heap limit on r2")).toBeVisible();
+    expect(requests.count).toEqual(1);
+  });
+
+  describe("other memory", () => {
+    const renderOtherMemory = async ({
+      replicaId,
+      heapLimit = String(4 * GIB),
+      heapBytes,
+    }: {
+      replicaId: string;
+      heapLimit?: string | null;
+      heapBytes?: string;
+    }) => {
+      await setLiveSizesEnvironment();
+      const replica = buildReplica(replicaId, "r1");
+      const cluster = buildCluster([replica]);
+      server.use(
+        ...heapReadsHandlers(cluster, {
+          processes: 2,
+          heapLimit,
+          heapBytes: heapBytes ? { [replicaId]: heapBytes } : {},
+        }),
+        mockSubscribeSocket((sql, params) =>
+          params.includes(replicaId)
+            ? sizesSnapshot([
+                ["u188", 2 * GIB],
+                ["u404", GIB],
+              ])
+            : undefined,
+        ),
+      );
+      await renderLive(<MemoryByObject cluster={cluster} replica={replica} />);
+      return screen.findByRole("list", { name: "Objects" });
+    };
+
+    it("sizes it by the heap in use beyond the listed objects", async () => {
+      // 6 GiB in use over two processes, 3 GiB of it listed, of 8 GiB.
+      const objects = await renderOtherMemory({
+        replicaId: "u30",
+        heapBytes: String(3 * GIB),
+      });
+
+      const other = within(objects).getByText("Other memory").closest("li");
+      assert(other);
+      expect(within(other).getByText("3.00 GB (37.5%)")).toBeVisible();
+      expect(
+        within(other).getByLabelText(
+          "Heap in use that isn't attributed to the objects listed. Object sizes come from arrangement sizes, which undercount.",
+        ),
+      ).toBeInTheDocument();
+    });
+
+    it("leaves it out while the replica reports no heap use", async () => {
+      const objects = await renderOtherMemory({ replicaId: "u31" });
+
+      expect(within(objects).getByText("customer_view")).toBeVisible();
+      expect(
+        within(objects).queryByText("Other memory"),
+      ).not.toBeInTheDocument();
+    });
+
+    it("leaves it out while the replica reports no heap limit", async () => {
+      const objects = await renderOtherMemory({
+        replicaId: "u32",
+        heapLimit: null,
+        heapBytes: String(3 * GIB),
+      });
+
+      expect(within(objects).getByText("2.00 GB")).toBeVisible();
+      expect(
+        within(objects).queryByText("Other memory"),
+      ).not.toBeInTheDocument();
+    });
+  });
+});
+
+describe("otherMemory", () => {
+  it("is the heap in use beyond the listed sizes, as a share of the limit", () => {
+    expect(
+      otherMemory(600, 1000, [
+        { size: 200 },
+        { size: BigInt(100) },
+        { size: null },
+      ]),
+    ).toEqual({ size: 300, memoryPercentage: 30 });
+  });
+
+  it("is undefined when the listed sizes cover the heap in use", () => {
+    expect(otherMemory(300, 1000, [{ size: 200 }, { size: 100 }])).toBe(
+      undefined,
+    );
   });
 });
 
diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx
index dcfdded570930..aa6891b2dc7aa 100644
--- a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx
+++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx
@@ -12,6 +12,7 @@ import {
   Card,
   Flex,
   HStack,
+  Spinner,
   Text,
   Tooltip,
   useTheme,
@@ -20,80 +21,214 @@ import {
 import React from "react";
 import { Link as RouterLink } from "react-router-dom";
 
+import { Cluster, Replica } from "~/api/materialize/cluster/clusterList";
+import Alert from "~/components/Alert";
 import { TooltipColorSwatch } from "~/components/graphComponents";
+import { LoadingContainer } from "~/components/LoadingContainer";
 import TextLink from "~/components/TextLink";
 import { generateRainbowPalette } from "~/components/ThresholdLineGraph/thresholdLineGraphHelpers";
 import { useFlags } from "~/hooks/useFlags";
-import { LargestReplicaBoundary } from "~/platform/clusters/LargestMaintainedQueries";
-import { useLargestMaintainedQueries } from "~/platform/clusters/queries";
+import { InfoIcon } from "~/icons";
+import {
+  LargestReplica,
+  LargestReplicaBoundary,
+} from "~/platform/clusters/LargestMaintainedQueries";
+import {
+  UNIFIED_OBJECT_SIZES_MIN_VERSION,
+  useLargestMaintainedQueries,
+  useReplicaHeapBytes,
+  useReplicaHeapLimits,
+  useReplicaSizeAllocations,
+} from "~/platform/clusters/queries";
 import {
   absoluteMaintainedObjectPath,
   useBuildWorkflowGraphPath,
 } from "~/platform/routeHelpers";
-import { useRegionSlug } from "~/store/environments";
+import { useEnvironmentGate, useRegionSlug } from "~/store/environments";
 import { MaterializeTheme } from "~/theme";
 import { formatMemoryUsage } from "~/utils/format";
 
-import { segmentWidths, TOP_OBJECT_COUNT } from "./memoryByObjectModel";
+import {
+  LargestObject,
+  otherMemory,
+  segmentWidths,
+  TOP_OBJECT_COUNT,
+} from "./memoryByObjectModel";
 import { PERCENT_TICKS } from "./ResourceUsage/resourceUsageStyles";
+import { useLiveLargestObjects } from "./useLiveLargestObjects";
 
 export interface MemoryByObjectProps {
-  clusterId: string;
-  clusterName: string;
+  cluster: Cluster;
+  /** The replica to show, else the cluster's largest. */
+  replica?: Replica;
 }
 
-type LargestObject = NonNullable<
-  ReturnType<typeof useLargestMaintainedQueries>["data"]
->[number];
+const OTHER_MEMORY_DESCRIPTION =
+  "Heap in use that isn't attributed to the objects listed. Object sizes come from arrangement sizes, which undercount.";
+
+/**
+ * Heap limits and use of the cluster's current replicas, from the reads the
+ * resource usage card makes with the same keys.
+ */
+const useReplicaHeaps = (cluster: Cluster) => {
+  const replicaIds = cluster.replicas.map(({ id }) => id).sort();
+  const sizes = [
+    ...new Set(cluster.replicas.flatMap(({ size }) => (size ? [size] : []))),
+  ].sort();
+  const heapLimits = useReplicaHeapLimits(replicaIds);
+  const allocations = useReplicaSizeAllocations(sizes);
+  const heapBytes = useReplicaHeapBytes(replicaIds);
+  return {
+    isLoading:
+      heapLimits.isLoading || allocations.isLoading || heapBytes.isLoading,
+    /**
+     * The replica's heap limit across its processes, as
+     * `buildLargestClusterReplicaQuery` computes it. Null if the process counts
+     * failed to load, since a per-process limit would inflate every share.
+     */
+    heapLimitOf: ({ id, size }: Replica) => {
+      const perProcess = heapLimits.data?.get(id);
+      if (perProcess === undefined || allocations.isError) return null;
+      const processes = size
+        ? allocations.data?.get(size)?.processes
+        : undefined;
+      return perProcess * (processes ?? 1);
+    },
+    heapBytesOf: (replicaId: string) => heapBytes.data?.get(replicaId),
+  };
+};
 
 /**
  * The largest objects on a replica, as shares of its heap limit: the named
- * replica, else the cluster's largest.
+ * replica, else the cluster's largest. A named replica is read from the page's
+ * clusters and heap reads, so switching to one waits only on its sizes.
  */
-export const MemoryByObject = ({
-  clusterId,
-  clusterName,
-  replicaName,
-}: MemoryByObjectProps & { replicaName?: string }) => (
-  <LargestReplicaBoundary
-    clusterId={clusterId}
-    replicaName={replicaName}
-    wrapFallback={(fallback) => (
-      <MemoryByObjectFrame>{fallback}</MemoryByObjectFrame>
-    )}
-  >
-    {(replica) => (
-      <PolledMemoryByObject
-        clusterId={clusterId}
-        clusterName={clusterName}
-        replicaName={replica.name}
-        replicaHeapLimit={replica.heapLimit}
-      />
-    )}
-  </LargestReplicaBoundary>
-);
+export const MemoryByObject = ({ cluster, replica }: MemoryByObjectProps) => {
+  const hasLiveSizes =
+    useEnvironmentGate(UNIFIED_OBJECT_SIZES_MIN_VERSION) === true;
+  const heaps = useReplicaHeaps(cluster);
+  return (
+    <LargestReplicaBoundary
+      clusterId={cluster.id}
+      replica={
+        replica && {
+          id: replica.id,
+          name: replica.name,
+          heapLimit: heaps.heapLimitOf(replica),
+        }
+      }
+      wrapFallback={(fallback) => (
+        <MemoryByObjectFrame>{fallback}</MemoryByObjectFrame>
+      )}
+    >
+      {(shown) =>
+        hasLiveSizes ? (
+          <LiveMemoryByObject
+            replica={shown}
+            heapUsedBytes={heaps.heapBytesOf(shown.id)}
+            isHeapLoading={heaps.isLoading}
+          />
+        ) : heaps.isLoading ? (
+          // A named replica's polled sizes are keyed on its heap limit, so
+          // loading them before it would load them twice.
+          <MemoryByObjectFrame>
+            <LoadingContainer />
+          </MemoryByObjectFrame>
+        ) : (
+          <PolledMemoryByObject
+            clusterId={cluster.id}
+            clusterName={cluster.name}
+            replica={shown}
+            heapUsedBytes={heaps.heapBytesOf(shown.id)}
+          />
+        )
+      }
+    </LargestReplicaBoundary>
+  );
+};
 
 const PolledMemoryByObject = ({
   clusterId,
   clusterName,
-  replicaName,
-  replicaHeapLimit,
-}: MemoryByObjectProps & {
-  replicaName: string;
-  replicaHeapLimit: number | null;
+  replica,
+  heapUsedBytes,
+}: {
+  clusterId: string;
+  clusterName: string;
+  replica: LargestReplica;
+  heapUsedBytes: number | undefined;
 }) => {
   const { data: objects } = useLargestMaintainedQueries({
     clusterId,
     clusterName,
     limit: TOP_OBJECT_COUNT,
-    replicaName,
-    replicaHeapLimit,
+    replicaName: replica.name,
+    replicaHeapLimit: replica.heapLimit,
   });
   return (
     <MemoryByObjectCard
       objects={objects ?? []}
-      replicaName={replicaName}
-      hasHeapLimit={replicaHeapLimit !== null}
+      replicaName={replica.name}
+      heapLimit={replica.heapLimit}
+      heapUsedBytes={heapUsedBytes}
+    />
+  );
+};
+
+/** `value`, or while it's undefined, the last value it had in a commit. */
+function useLastDefined<T>(value: T | undefined) {
+  const last = React.useRef(value);
+  React.useEffect(() => {
+    if (value !== undefined) last.current = value;
+  });
+  return value ?? last.current;
+}
+
+const LiveMemoryByObject = ({
+  replica,
+  heapUsedBytes,
+  isHeapLoading,
+}: {
+  replica: LargestReplica;
+  heapUsedBytes: number | undefined;
+  isHeapLoading: boolean;
+}) => {
+  const { objects, snapshotComplete, isRefreshing, isError } =
+    useLiveLargestObjects({
+      replicaId: replica.id,
+      heapLimit: replica.heapLimit,
+    });
+  const content: MemoryByObjectContent | undefined =
+    snapshotComplete && !isHeapLoading
+      ? {
+          objects,
+          replicaName: replica.name,
+          heapLimit: replica.heapLimit,
+          heapUsedBytes,
+        }
+      : undefined;
+  // A switch to another replica keeps the previous one's content on screen
+  // until the new one's first snapshot and heap reads are in.
+  const shown = useLastDefined(content);
+  if (isError) {
+    return (
+      <MemoryByObjectFrame>
+        <Alert variant="error" message="Object memory is unavailable." />
+      </MemoryByObjectFrame>
+    );
+  }
+  if (!shown) {
+    return (
+      <MemoryByObjectFrame>
+        <LoadingContainer />
+      </MemoryByObjectFrame>
+    );
+  }
+  return (
+    <MemoryByObjectCard
+      {...shown}
+      isRefreshing={isRefreshing || !content}
+      isStale={!content}
     />
   );
 };
@@ -102,10 +237,15 @@ const PolledMemoryByObject = ({
 const MemoryByObjectFrame = ({
   title = "Objects by Memory",
   subtitle,
+  isRefreshing = false,
+  isStale = false,
   children,
 }: {
   title?: string;
   subtitle?: string;
+  isRefreshing?: boolean;
+  /** Dims content kept on screen while the selected replica's loads. */
+  isStale?: boolean;
   children: React.ReactNode;
 }) => {
   const { colors } = useTheme<MaterializeTheme>();
@@ -116,6 +256,7 @@ const MemoryByObjectFrame = ({
       borderRadius="md"
       borderWidth="1px"
       borderColor={colors.border.primary}
+      aria-busy={isRefreshing}
     >
       <VStack spacing={4} alignItems="stretch">
         <HStack spacing={2} alignItems="baseline" flexWrap="wrap">
@@ -123,26 +264,64 @@ const MemoryByObjectFrame = ({
             {title}
           </Text>
           {subtitle && (
-            <Text textStyle="text-small" color={colors.foreground.secondary}>
+            <Text
+              textStyle="text-small"
+              color={colors.foreground.secondary}
+              opacity={isStale ? 0.5 : undefined}
+            >
               {subtitle}
             </Text>
           )}
+          {isRefreshing && (
+            <Spinner
+              size="sm"
+              alignSelf="center"
+              ml="auto"
+              data-testid="refreshing-spinner"
+            />
+          )}
         </HStack>
-        {children}
+        <VStack
+          spacing={4}
+          alignItems="stretch"
+          opacity={isStale ? 0.5 : undefined}
+        >
+          {children}
+        </VStack>
       </VStack>
     </Card>
   );
 };
 
-const MemoryByObjectCard = ({
-  objects,
-  replicaName,
-  hasHeapLimit,
-}: {
+interface Segment {
+  key: string;
+  label: string;
+  usage: string;
+  color: string;
+  widthPercent: number;
+  path?: string;
+  description?: string;
+}
+
+interface MemoryByObjectContent {
   objects: LargestObject[];
   replicaName: string;
   /** Without a heap limit there are no shares, so the card lists sizes alone. */
-  hasHeapLimit: boolean;
+  heapLimit: number | null;
+  /** The replica's heap in use, which sizes the other memory segment. */
+  heapUsedBytes: number | undefined;
+}
+
+const MemoryByObjectCard = ({
+  objects,
+  replicaName,
+  heapLimit,
+  heapUsedBytes,
+  isRefreshing,
+  isStale,
+}: MemoryByObjectContent & {
+  isRefreshing?: boolean;
+  isStale?: boolean;
 }) => {
   const { colors } = useTheme<MaterializeTheme>();
   // The maintained objects page only has routes while its flag is on.
@@ -152,9 +331,10 @@ const MemoryByObjectCard = ({
   const regionSlug = useRegionSlug();
   const workflowGraphPath = useBuildWorkflowGraphPath();
 
+  const hasHeapLimit = heapLimit !== null;
   if (objects.length === 0) {
     return (
-      <MemoryByObjectFrame>
+      <MemoryByObjectFrame isRefreshing={isRefreshing} isStale={isStale}>
         <Text textStyle="text-base" color={colors.foreground.secondary}>
           No objects are using memory on {replicaName}.
         </Text>
@@ -165,11 +345,18 @@ const MemoryByObjectCard = ({
   // Colored by rank like the Freshness page's lines, so the largest objects
   // are furthest apart in hue.
   const palette = generateRainbowPalette(objects.length);
+  const other =
+    heapLimit && heapUsedBytes !== undefined
+      ? otherMemory(heapUsedBytes, heapLimit, objects)
+      : undefined;
+  // Other memory goes last, so the end of the bar clips it before any object.
   const widths = segmentWidths(
-    objects.map((object) => object.memoryPercentage),
+    [...objects, ...(other ? [other] : [])].map(
+      ({ memoryPercentage }) => memoryPercentage,
+    ),
   );
-  const segments = objects.map((object, index) => {
-    const { id, name, databaseName, schemaName } = object;
+  const segments: Segment[] = objects.map((object, index) => {
+    const { id, name, databaseName, schemaName, type } = object;
     return {
       key: id ?? object.dataflowId ?? String(index),
       label: name ?? object.dataflowName ?? "-",
@@ -177,16 +364,26 @@ const MemoryByObjectCard = ({
       color: palette[index],
       widthPercent: widths[index],
       path:
-        object.isOrphanedDataflow || !id || !name || !schemaName
+        object.isOrphanedDataflow || !id || !name || !schemaName || !type
           ? undefined
           : hasMaintainedObjectsPage
             ? absoluteMaintainedObjectPath(regionSlug, id)
             : workflowGraphPath({
                 databaseObject: { id, name, databaseName, schemaName },
-                type: object.type,
+                type,
               }),
     };
   });
+  if (other) {
+    segments.push({
+      key: "other-memory",
+      label: "Other memory",
+      usage: formatMemoryUsage(other),
+      color: colors.foreground.tertiary,
+      widthPercent: widths[objects.length],
+      description: OTHER_MEMORY_DESCRIPTION,
+    });
+  }
 
   return (
     <MemoryByObjectFrame
@@ -200,6 +397,8 @@ const MemoryByObjectCard = ({
           ? `% of heap limit on ${replicaName}`
           : `Size on ${replicaName}`
       }
+      isRefreshing={isRefreshing}
+      isStale={isStale}
     >
       {hasHeapLimit && (
         <Box>
@@ -216,7 +415,14 @@ const MemoryByObjectCard = ({
                 segment.widthPercent > 0 && (
                   <Tooltip
                     key={segment.key}
-                    label={`${segment.label} · ${segment.usage}`}
+                    label={
+                      <>
+                        <Text>{`${segment.label} · ${segment.usage}`}</Text>
+                        {segment.description && (
+                          <Text>{segment.description}</Text>
+                        )}
+                      </>
+                    }
                   >
                     <Box
                       width={`${segment.widthPercent}%`}
@@ -279,6 +485,11 @@ const MemoryByObjectCard = ({
             >
               {segment.usage}
             </Text>
+            {segment.description && (
+              <Tooltip label={segment.description}>
+                <InfoIcon aria-label={segment.description} />
+              </Tooltip>
+            )}
           </HStack>
         ))}
       </Flex>
diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx
index 884e5c23c9e2d..87942cb419b81 100644
--- a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx
+++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx
@@ -923,6 +923,61 @@ describe("ResourceUsage", () => {
     ).not.toBeInTheDocument();
   });
 
+  it("draws a window of 3h or less from the live subscribe", async () => {
+    window.localStorage.setItem("mz-cluster-graph-time-period", "60");
+    // The lineage key embeds the environment version.
+    await setFakeEnvironment(getStore().set, "aws/us-east-1", NEW_ENVIRONMENT);
+    server.use(
+      buildSqlQueryHandlerV2({
+        queryKey: clusterQueryKeys.deploymentLineage({ clusterIdsKey: "u9" }),
+        results: mapKyselyToTabular({
+          columns: buildColumns([
+            "clusterId",
+            "currentDeploymentClusterId",
+            "clusterName",
+          ]),
+          rows: [],
+        }),
+      }),
+      mockSubscribeSocket((sql) =>
+        sql.includes("utilization_overview_3h")
+          ? {
+              columns: [
+                buildColumn({ name: "replicaId" }),
+                buildColumn({ name: "name" }),
+                buildColumn({ name: "occurredAt" }),
+                buildColumn({
+                  name: "cpuPercent",
+                  type_oid: MzDataType.float8,
+                }),
+                buildColumn({
+                  name: "heapPercent",
+                  type_oid: MzDataType.float8,
+                }),
+              ],
+              rows: [
+                [
+                  "u1",
+                  "r1",
+                  new Date(NOW_MS - 30 * 60_000).toISOString(),
+                  "0.45",
+                  "0.6",
+                ],
+              ],
+            }
+          : undefined,
+      ),
+    );
+    await renderResourceUsage({ withSwap: true });
+
+    expect(
+      await screen.findByRole("img", { name: "CPU usage" }),
+    ).toBeInTheDocument();
+    expect(
+      screen.queryByText("No resource usage in this time period."),
+    ).not.toBeInTheDocument();
+  });
+
   it("says replicas ran unmeasured when the window has no readings", async () => {
     await renderResourceUsage({ withSwap: true, utilizationRowCount: 0 });
 
diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts
new file mode 100644
index 0000000000000..28a4b78ad1d43
--- /dev/null
+++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts
@@ -0,0 +1,337 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { renderHook, waitFor } from "@testing-library/react";
+import { millisecondsInMinute } from "date-fns/constants";
+import { delay, http, ws } from "msw";
+import React from "react";
+
+import { BinnedSubscribeRow } from "~/api/materialize/cluster/replicaUtilizationBinning";
+import { MzDataType } from "~/api/materialize/types";
+import {
+  buildColumn,
+  buildColumns,
+  buildSqlQueryHandlerV2,
+  mapKyselyToTabular,
+} from "~/api/mocks/buildSqlQueryHandler";
+import server from "~/api/mocks/server";
+import { clusterQueryKeys } from "~/platform/clusters/queries";
+import { buildQueryClient } from "~/queryClient";
+import {
+  mockSubscribeSocket,
+  SubscribeSnapshot,
+} from "~/test/mockSubscribeSocket";
+import { createProviderWrapper } from "~/test/utils";
+
+import {
+  StoredSample,
+  transformLiveUtilization,
+  useLiveClusterUtilization,
+} from "./useLiveClusterUtilization";
+const END_MS = Date.UTC(2026, 9, 2, 12, 0);
+
+const sample = (minutesAgo: number, cpuPercent: number): StoredSample => ({
+  replicaId: "u1",
+  clusterId: "u9",
+  size: "100cc",
+  name: "r1",
+  occurredAt: new Date(
+    END_MS - minutesAgo * millisecondsInMinute,
+  ).toISOString(),
+  cpuPercent,
+  memoryPercent: 0.4,
+  diskPercent: null,
+  heapPercent: 0.5,
+  memoryAndDiskPercent: null,
+});
+
+const bucket = (minutesAgo: number, cpuPercent: number): BinnedSubscribeRow => {
+  const start = new Date(
+    END_MS - minutesAgo * millisecondsInMinute,
+  ).toISOString();
+  return {
+    bucketStart: start,
+    bucketEnd: new Date(
+      END_MS - (minutesAgo - 5) * millisecondsInMinute,
+    ).toISOString(),
+    replicaId: "u1",
+    clusterId: "u9",
+    size: "100cc",
+    name: "r1",
+    maxMemoryPercent: 0.4,
+    maxMemoryAt: start,
+    maxDiskPercent: null,
+    maxDiskAt: start,
+    maxCpuPercent: cpuPercent,
+    maxCpuAt: start,
+    maxHeapPercent: 0.5,
+    maxHeapAt: start,
+    maxMemoryAndDiskPercent: null,
+    maxMemoryAndDiskMemoryPercent: null,
+    maxMemoryAndDiskDiskPercent: null,
+    maxMemoryAndDiskAt: start,
+    offlineEvents: null,
+  };
+};
+
+const cpuSeries = (result: ReturnType<typeof transformLiveUtilization>) =>
+  result.graphData[0].data.map((point) => point.cpuPercent);
+
+describe("transformLiveUtilization", () => {
+  it("bins 3h samples into the window's buckets, keeping each bucket's peak", () => {
+    const result = transformLiveUtilization({
+      rows: {
+        tier: "unbinned3h",
+        // The 3h subscribe holds samples from before a 1h window too.
+        samples: [sample(90, 0.9), sample(30, 0.2), sample(29.5, 0.6)],
+      },
+      endDate: new Date(END_MS),
+      timePeriodMinutes: 60,
+      bucketSizeMs: millisecondsInMinute,
+    });
+    expect(cpuSeries(result)).toEqual([60]);
+  });
+
+  it("clips 24h buckets to the window and puts them in time order", () => {
+    const result = transformLiveUtilization({
+      rows: {
+        tier: "binned24h",
+        buckets: [bucket(60, 0.3), bucket(600, 0.9), bucket(120, 0.1)],
+      },
+      endDate: new Date(END_MS),
+      timePeriodMinutes: 6 * 60,
+      bucketSizeMs: 5 * millisecondsInMinute,
+    });
+    expect(cpuSeries(result)).toEqual([10, 30]);
+  });
+});
+
+describe("useLiveClusterUtilization", () => {
+  type LiveUtilization = ReturnType<typeof useLiveClusterUtilization>;
+
+  const lineageHandler = (clusterId: string) =>
+    buildSqlQueryHandlerV2({
+      queryKey: clusterQueryKeys.deploymentLineage({
+        clusterIdsKey: clusterId,
+      }),
+      results: mapKyselyToTabular({
+        columns: buildColumns([
+          "clusterId",
+          "currentDeploymentClusterId",
+          "clusterName",
+        ]),
+        rows: [],
+      }),
+    });
+
+  const minutesAgo = (minutes: number) =>
+    new Date(Date.now() - minutes * millisecondsInMinute);
+
+  const sampleSnapshot = (occurredAt: Date): SubscribeSnapshot => ({
+    columns: [
+      buildColumn({ name: "replicaId" }),
+      buildColumn({ name: "occurredAt" }),
+      buildColumn({
+        name: "cpuPercent",
+        type_oid: MzDataType.float8,
+      }),
+    ],
+    rows: [["u1", occurredAt.toISOString(), "0.5"]],
+  });
+
+  // Serves a 3h snapshot of one sample from 70 minutes ago, outside a 1h
+  // window, while `subscribe.serve` is set.
+  const serveHeldSample = (clusterId: string) => {
+    const subscribe = { serve: true, requests: 0 };
+    const occurredAt = minutesAgo(70);
+    server.use(
+      lineageHandler(clusterId),
+      mockSubscribeSocket((sql) => {
+        if (!sql.includes("utilization_overview_3h")) return undefined;
+        subscribe.requests += 1;
+        return subscribe.serve ? sampleSnapshot(occurredAt) : undefined;
+      }),
+    );
+    return subscribe;
+  };
+
+  // Answers each cluster's 3h subscribe with `snapshotFor(clusterId)`, and
+  // returns the clusters subscribed to, in order.
+  const serveSubscribes = (
+    clusterIds: string[],
+    snapshotFor: (clusterId: string) => SubscribeSnapshot | undefined,
+  ) => {
+    const requests: string[] = [];
+    server.use(
+      ...clusterIds.map(lineageHandler),
+      mockSubscribeSocket((sql, params) => {
+        const clusterId = clusterIds.find((id) => params.includes(id));
+        if (!sql.includes("utilization_overview_3h") || !clusterId) {
+          return undefined;
+        }
+        requests.push(clusterId);
+        return snapshotFor(clusterId);
+      }),
+    );
+    return requests;
+  };
+
+  type WindowProps = { clusterId: string; timePeriodMinutes: number };
+
+  const renderWindow = (
+    initialProps: WindowProps,
+    wrapper: React.ComponentType<React.PropsWithChildren>,
+  ) => {
+    const renders: LiveUtilization[] = [];
+    const rendered = renderHook(
+      ({ clusterId, timePeriodMinutes }: WindowProps) => {
+        const result = useLiveClusterUtilization({
+          clusterId,
+          timePeriodMinutes,
+          bucketSizeMs: millisecondsInMinute,
+          tier: "unbinned3h",
+          includeMemoryBreakdown: false,
+        });
+        renders.push(result);
+        return result;
+      },
+      { wrapper, initialProps },
+    );
+    return { ...rendered, renders };
+  };
+
+  const renderOneHourWindow = (
+    clusterId: string,
+    wrapper: React.ComponentType<React.PropsWithChildren>,
+  ) => renderWindow({ clusterId, timePeriodMinutes: 60 }, wrapper);
+
+  // Held rows outside the window must never read as a window with no usage.
+  const expectEveryRenderAwaitingRows = (renders: LiveUtilization[]) => {
+    expect(renders.length).toBeGreaterThan(0);
+    for (const render of renders) {
+      expect(render).toMatchObject({
+        data: undefined,
+        isLoading: true,
+        isRefreshing: true,
+      });
+    }
+  };
+
+  it("waits for a new snapshot when held rows have left the window", async () => {
+    const subscribe = serveHeldSample("u9");
+    const ProviderWrapper = await createProviderWrapper();
+
+    const first = renderOneHourWindow("u9", ProviderWrapper);
+    await waitFor(() => expect(first.result.current.data).toBeDefined());
+    expect(first.result.current.data?.graphData).toEqual([]);
+    first.unmount();
+
+    // Coming back, the held sample is still outside the window and the new
+    // subscribe hasn't delivered yet.
+    subscribe.serve = false;
+    const second = renderOneHourWindow("u9", ProviderWrapper);
+    await waitFor(() => expect(subscribe.requests).toEqual(2));
+    expectEveryRenderAwaitingRows(second.renders);
+  });
+
+  it("holds rows as refreshing while an evicted lineage is refetched", async () => {
+    serveHeldSample("u10");
+    const first = renderOneHourWindow("u10", await createProviderWrapper());
+    await waitFor(() => expect(first.result.current.data).toBeDefined());
+    first.unmount();
+
+    // A fresh query cache has no lineage, and its refetch never returns, so
+    // no subscribe starts.
+    let lineageRequested = false;
+    server.use(
+      http.post("*/api/sql", async () => {
+        lineageRequested = true;
+        await delay("infinite");
+      }),
+    );
+    const second = renderOneHourWindow(
+      "u10",
+      await createProviderWrapper({ queryClient: buildQueryClient() }),
+    );
+    await waitFor(() => expect(lineageRequested).toBe(true));
+    expectEveryRenderAwaitingRows(second.renders);
+  });
+
+  it("keeps the subscribe when the window changes within a tier", async () => {
+    const requests = serveSubscribes(["u11"], () =>
+      sampleSnapshot(minutesAgo(30)),
+    );
+    const rendered = renderWindow(
+      { clusterId: "u11", timePeriodMinutes: 15 },
+      await createProviderWrapper(),
+    );
+    await waitFor(() => expect(rendered.result.current.data).toBeDefined());
+    expect(rendered.result.current.data?.graphData).toEqual([]);
+
+    const rendersBefore = rendered.renders.length;
+    rendered.rerender({ clusterId: "u11", timePeriodMinutes: 60 });
+    expect(rendered.result.current.data?.graphData).toHaveLength(1);
+    // Room for a restarted session to send its subscribe.
+    await new Promise((resolve) => setTimeout(resolve, 250));
+    expect(requests).toEqual(["u11"]);
+    for (const render of rendered.renders.slice(rendersBefore)) {
+      expect(render.isRefreshing).toBe(false);
+    }
+  });
+
+  it("serves no rows of the previous cluster after a cluster switch", async () => {
+    const requests = serveSubscribes(["u12", "u13"], (clusterId) =>
+      clusterId === "u12" ? sampleSnapshot(minutesAgo(30)) : undefined,
+    );
+    const rendered = renderOneHourWindow("u12", await createProviderWrapper());
+    await waitFor(() =>
+      expect(rendered.result.current.data?.graphData).toHaveLength(1),
+    );
+
+    const rendersBefore = rendered.renders.length;
+    rendered.rerender({ clusterId: "u13", timePeriodMinutes: 60 });
+    await waitFor(() => expect(requests).toEqual(["u12", "u13"]));
+    const afterSwitch = rendered.renders.slice(rendersBefore);
+    expect(afterSwitch.length).toBeGreaterThan(0);
+    for (const render of afterSwitch) {
+      expect(render).toMatchObject({ data: undefined, isLoading: true });
+    }
+  });
+
+  it("holds rows as refreshing while a dropped socket resubscribes", async () => {
+    let serve = true;
+    const requests = serveSubscribes(["u14"], () =>
+      serve ? sampleSnapshot(minutesAgo(30)) : undefined,
+    );
+    const sockets: Array<{ close: () => void }> = [];
+    server.use(
+      ws
+        .link(/^wss?:\/\//)
+        .addEventListener("connection", ({ client }) => sockets.push(client)),
+    );
+    const rendered = renderOneHourWindow("u14", await createProviderWrapper());
+    await waitFor(() =>
+      expect(rendered.result.current.data?.graphData).toHaveLength(1),
+    );
+    expect(rendered.result.current.isRefreshing).toBe(false);
+
+    serve = false;
+    sockets.forEach((socket) => socket.close());
+    // The socket reopens after the connection manager's backoff.
+    await waitFor(() => {
+      expect(requests).toEqual(["u14", "u14"]);
+      expect(rendered.result.current.isRefreshing).toBe(true);
+    });
+    expect(rendered.result.current).toMatchObject({
+      isLoading: false,
+      isError: false,
+    });
+    expect(rendered.result.current.data?.graphData).toHaveLength(1);
+  });
+});
diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts
new file mode 100644
index 0000000000000..f5d73b79649e4
--- /dev/null
+++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts
@@ -0,0 +1,195 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { useLiveQuery } from "@tanstack/react-db";
+import { subMinutes } from "date-fns";
+import React from "react";
+
+import {
+  BinnedSubscribeRow,
+  bucketRowsToBucketsByReplicaId,
+  clipBinnedRowsToWindow,
+  parseBinnedSubscribeRow,
+  rebucketUtilizationSamples,
+  toReplicaUtilizationGraphData,
+  UtilizationSample,
+} from "~/api/materialize/cluster/replicaUtilizationBinning";
+import {
+  buildConsoleClusterUtilizationOverview24hSubscribe,
+  buildConsoleClusterUtilizationUnbinned3hSubscribe,
+} from "~/api/materialize/cluster/replicaUtilizationHistory";
+import { createScopedSubscribeCollections } from "~/api/materialize/subscribeCollection";
+import { useScopedSubscribeCollection } from "~/api/materialize/useScopedSubscribeCollection";
+import {
+  LIVE_TIER_RETENTION_MINUTES,
+  useClusterLineageIds,
+} from "~/platform/clusters/queries";
+
+/** A 3h-view sample as streamed, with its timestamp still a string. */
+export type StoredSample = Omit<UtilizationSample, "occurredAt"> & {
+  occurredAt: string;
+};
+
+/**
+ * Shapes a live collection's rows for a chart window ending at `endDate`. The
+ * 3h tier holds raw samples, binned here. The 24h tier holds the whole day's
+ * 5-minute buckets, clipped here to the window.
+ */
+export function transformLiveUtilization({
+  rows,
+  endDate,
+  timePeriodMinutes,
+  bucketSizeMs,
+}: {
+  rows:
+    | { tier: "unbinned3h"; samples: StoredSample[] }
+    | { tier: "binned24h"; buckets: BinnedSubscribeRow[] };
+  endDate: Date;
+  timePeriodMinutes: number;
+  bucketSizeMs: number;
+}) {
+  const startDate = subMinutes(endDate, timePeriodMinutes);
+  const bucketRows =
+    rows.tier === "unbinned3h"
+      ? rebucketUtilizationSamples(
+          rows.samples.map((sample) => ({
+            ...sample,
+            occurredAt: new Date(sample.occurredAt),
+          })),
+          bucketSizeMs,
+          startDate.getTime(),
+        )
+      : clipBinnedRowsToWindow(
+          rows.buckets.map(parseBinnedSubscribeRow),
+          startDate,
+        );
+  return toReplicaUtilizationGraphData(
+    bucketRowsToBucketsByReplicaId(bucketRows),
+    startDate,
+    endDate,
+  );
+}
+
+const sampleKey = (sample: StoredSample) =>
+  `${sample.replicaId}|${sample.occurredAt}`;
+const bucketKey = (row: BinnedSubscribeRow) =>
+  `${row.replicaId}|${row.bucketStart}`;
+
+const samplesByCluster = createScopedSubscribeCollections<StoredSample>({
+  id: "cluster-utilization-3h",
+  getKey: sampleKey,
+});
+const bucketsByCluster = createScopedSubscribeCollections<BinnedSubscribeRow>({
+  id: "cluster-utilization-24h",
+  getKey: bucketKey,
+});
+
+/**
+ * Live utilization for a cluster over windows of 24h or less, streamed from the
+ * indexed 3h or 24h view into a per-cluster TanStack DB collection. Each
+ * subscribe covers its view's whole retention, so switching between windows
+ * that one view serves reuses the rows instead of resubscribing. Returns the
+ * same shape as `useReplicaUtilizationHistory`.
+ */
+export function useLiveClusterUtilization({
+  clusterId,
+  timePeriodMinutes,
+  bucketSizeMs,
+  tier,
+  includeMemoryBreakdown,
+}: {
+  clusterId: string;
+  timePeriodMinutes: number;
+  bucketSizeMs: number;
+  tier: "unbinned3h" | "binned24h" | undefined;
+  includeMemoryBreakdown: boolean;
+}) {
+  const lineage = useClusterLineageIds([clusterId], tier !== undefined);
+  const lineageIdsKey = lineage.data?.join(",") ?? "";
+
+  const samplesSubscribe = React.useMemo(() => {
+    const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : [];
+    return tier === "unbinned3h" && clusterIds.length > 0
+      ? buildConsoleClusterUtilizationUnbinned3hSubscribe<StoredSample>(
+          clusterIds,
+          subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.unbinned3h),
+          includeMemoryBreakdown,
+        )
+      : undefined;
+  }, [tier, lineageIdsKey, includeMemoryBreakdown]);
+  const samples = useScopedSubscribeCollection(
+    samplesByCluster,
+    clusterId,
+    samplesSubscribe,
+  );
+
+  const bucketsSubscribe = React.useMemo(() => {
+    const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : [];
+    return tier === "binned24h" && clusterIds.length > 0
+      ? buildConsoleClusterUtilizationOverview24hSubscribe<BinnedSubscribeRow>(
+          clusterIds,
+          subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.binned24h),
+          includeMemoryBreakdown,
+        )
+      : undefined;
+  }, [tier, lineageIdsKey, includeMemoryBreakdown]);
+  const buckets = useScopedSubscribeCollection(
+    bucketsByCluster,
+    clusterId,
+    bucketsSubscribe,
+  );
+
+  const active =
+    tier === "unbinned3h"
+      ? samples
+      : tier === "binned24h"
+        ? buckets
+        : undefined;
+  const collection = active?.collection;
+  const { data: rows } = useLiveQuery(
+    (q) => (collection ? q.from({ row: collection.collection }) : undefined),
+    [collection],
+  );
+  const snapshotComplete = active?.status.snapshotComplete ?? false;
+  const isRefreshing = active?.status.isRefreshing ?? false;
+
+  const data = React.useMemo(() => {
+    if (!collection || !snapshotComplete || !rows) return undefined;
+    return transformLiveUtilization({
+      rows:
+        tier === "unbinned3h"
+          ? { tier, samples: rows as StoredSample[] }
+          : { tier: "binned24h", buckets: rows as BinnedSubscribeRow[] },
+      endDate: new Date(),
+      timePeriodMinutes,
+      bucketSizeMs,
+    });
+  }, [
+    collection,
+    snapshotComplete,
+    rows,
+    tier,
+    timePeriodMinutes,
+    bucketSizeMs,
+  ]);
+
+  // Held rows can have aged out of the window entirely. Until the new
+  // snapshot lands that would read as a window with no usage.
+  const isAwaitingRows = isRefreshing && data?.graphData.length === 0;
+  return {
+    data: isAwaitingRows ? undefined : data,
+    isLoading:
+      collection !== undefined && (!snapshotComplete || isAwaitingRows),
+    isError:
+      Boolean(active?.status.error) ||
+      lineage.isLoadingError ||
+      samples.isScopeError,
+    isRefreshing,
+  };
+}
diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts
index 20a963c5dac54..050bf7b9f9e92 100644
--- a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts
+++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts
@@ -20,11 +20,12 @@ import {
   replicaStatusesUpdatedAt,
   useObjectCreationTimes,
   useObjectHydrationStretches,
+  usePolledReplicaUtilizationHistory,
   useReplicaHeapLimits,
   useReplicaHydrationEpisodes,
   useReplicaStatusHistory,
-  useReplicaUtilizationHistory,
   useUnhydratedObjectCounts,
+  useUtilizationTier,
 } from "~/platform/clusters/queries";
 import { useClusterObjectsLive } from "~/store/allObjectsCollection";
 import { useEnvironmentGate } from "~/store/environments";
@@ -42,6 +43,7 @@ import {
   transformDdlEvents,
   transformReplicaTimeline,
 } from "./resourceUsageModel";
+import { useLiveClusterUtilization } from "./useLiveClusterUtilization";
 
 const DDL_OBJECT_TYPES = ["index", "materialized-view"];
 const EMPTY_SERIES: ReplicaData[] = [];
@@ -92,17 +94,36 @@ export function useResourceUsageData({
   timePeriodMinutes: number;
 }) {
   const { colors } = useTheme<MaterializeTheme>();
-  // The status timeline reads offline stretches from status history.
-  const history = useReplicaUtilizationHistory(
+  const bucketSizeMs = Math.max(timePeriodMinutes * 1000, MIN_BUCKET_SIZE_MS);
+  const tier = useUtilizationTier(timePeriodMinutes);
+  const hasMemoryBreakdown =
+    useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true;
+  const live = useLiveClusterUtilization({
+    clusterId: cluster.id,
+    timePeriodMinutes,
+    bucketSizeMs,
+    tier: tier === "poll" ? undefined : tier,
+    includeMemoryBreakdown: hasMemoryBreakdown,
+  });
+  const polled = usePolledReplicaUtilizationHistory(
     {
-      bucketSizeMs: Math.max(timePeriodMinutes * 1000, MIN_BUCKET_SIZE_MS),
+      bucketSizeMs,
       timePeriodMinutes,
       clusterIds: [cluster.id],
       replicaId: undefined,
       includeMemoryBreakdown: true,
     },
-    { includeOfflineEvents: false },
+    { enabled: tier === "poll" },
   );
+  const history =
+    tier === "poll"
+      ? {
+          data: polled.data,
+          isLoading: polled.isLoading,
+          isError: polled.isError,
+          isRefreshing: false,
+        }
+      : live;
   const graphData = history.data?.graphData;
 
   // Replicas that share a name, such as a scheduled cluster's refreshes, share
@@ -199,10 +220,8 @@ export function useResourceUsageData({
   // A sample without a heap limit has `heap_percent` as a share of the memory
   // allocation instead, and no RAM limit share. Only rows on 26.44 and later
   // carry that share, so older environments keep the heap labels.
-  const rowsCarryHeapLimit =
-    useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true;
   const hasHeapLimit =
-    !rowsCarryHeapLimit ||
+    !hasMemoryBreakdown ||
     chartData.some(({ data }) =>
       data.some((point) => point.ramLimitPercent !== null),
     );
diff --git a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts
index 7e178aa276e00..0827b515d7e01 100644
--- a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts
+++ b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts
@@ -12,6 +12,39 @@ import { clamp } from "~/util";
 /** How many of the largest objects the card shows. */
 export const TOP_OBJECT_COUNT = 10;
 
+/** One object on the card, from the polled or the streamed sizes. */
+export interface LargestObject {
+  id: string | null;
+  name: string | null;
+  /** Bytes: a bigint when polled, a number when streamed. */
+  size: bigint | number | null;
+  memoryPercentage: number | null;
+  type: string | null;
+  schemaName: string | null;
+  databaseName: string | null;
+  dataflowId: string | null;
+  dataflowName: string | null;
+  isOrphanedDataflow: boolean;
+}
+
+/**
+ * Heap in use beyond the listed objects' sizes, with its share of the heap
+ * limit. Undefined if their sizes add up to at least `heapUsedBytes`.
+ */
+export const otherMemory = (
+  heapUsedBytes: number,
+  heapLimit: number,
+  objects: Array<Pick<LargestObject, "size">>,
+) => {
+  const size = objects.reduce(
+    (rest, object) => rest - Number(object.size ?? 0),
+    heapUsedBytes,
+  );
+  return size > 0
+    ? { size, memoryPercentage: (size / heapLimit) * 100 }
+    : undefined;
+};
+
 /**
  * Bar segment widths for objects' shares of the heap limit, in order. The bar
  * stops at the heap limit even if the shares add up to more.
diff --git a/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts
new file mode 100644
index 0000000000000..4a8379a229985
--- /dev/null
+++ b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts
@@ -0,0 +1,97 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { renderHook, waitFor } from "@testing-library/react";
+
+import { ReplicaObjectSize } from "~/api/materialize/cluster/largestMaintainedQueries";
+import { DatabaseObject } from "~/api/materialize/objects";
+import { createSubscribeCollection } from "~/api/materialize/subscribeCollection";
+import { allObjectsCollection } from "~/store/allObjectsCollection";
+import { createProviderWrapper } from "~/test/utils";
+
+import { useLargestObjects } from "./useLiveLargestObjects";
+
+const MIB = 2 ** 20;
+const HEAP_LIMIT = 1000 * MIB;
+
+const catalogObject = (id: string, name: string): DatabaseObject => ({
+  databaseName: "materialize",
+  databaseId: "u1",
+  name,
+  schemaName: "public",
+  schemaId: "u2",
+  id,
+  objectType: "index",
+  sourceType: null,
+  isWebhookTable: null,
+  clusterId: "u9",
+  clusterName: "prod",
+});
+
+const renderLargestObjects = async (sizes: ReplicaObjectSize[]) => {
+  const collection = createSubscribeCollection<ReplicaObjectSize>({
+    id: `test-sizes-${Math.random()}`,
+    getKey: (row) => row.objectId,
+  });
+  collection.applySnapshot({
+    data: sizes,
+    snapshotComplete: true,
+    error: undefined,
+  });
+  const ProviderWrapper = await createProviderWrapper();
+  return renderHook(() => useLargestObjects(collection, HEAP_LIMIT), {
+    wrapper: ProviderWrapper,
+  });
+};
+
+describe("useLargestObjects", () => {
+  beforeEach(() => {
+    allObjectsCollection.applySnapshot({
+      data: [
+        catalogObject("u1", "orders_idx"),
+        catalogObject("u2", "items_idx"),
+      ],
+      snapshotComplete: true,
+      error: undefined,
+    });
+  });
+
+  it("names the largest objects from the catalog, largest first", async () => {
+    const { result } = await renderLargestObjects([
+      { objectId: "u1", size: 100 * MIB },
+      { objectId: "u2", size: 300 * MIB },
+      { objectId: "u3", size: null },
+    ]);
+
+    await waitFor(() =>
+      expect(
+        result.current.objects.map(({ name, memoryPercentage }) => [
+          name,
+          memoryPercentage,
+        ]),
+      ).toEqual([
+        ["items_idx", 30],
+        ["orders_idx", 10],
+        ["u3", null],
+      ]),
+    );
+  });
+
+  it("keeps only the ten largest", async () => {
+    const { result } = await renderLargestObjects(
+      Array.from({ length: 12 }, (_, index) => ({
+        objectId: `u${100 + index}`,
+        size: (index + 1) * MIB,
+      })),
+    );
+
+    await waitFor(() => expect(result.current.objects).toHaveLength(10));
+    expect(result.current.objects.at(-1)?.id).toEqual("u102");
+  });
+});
diff --git a/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts
new file mode 100644
index 0000000000000..ad8a31e61543b
--- /dev/null
+++ b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts
@@ -0,0 +1,107 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+import { eq, useLiveQuery } from "@tanstack/react-db";
+import { useAtomValue } from "jotai";
+import React from "react";
+
+import {
+  buildReplicaObjectSizesSubscribe,
+  ReplicaObjectSize,
+} from "~/api/materialize/cluster/largestMaintainedQueries";
+import {
+  createScopedSubscribeCollections,
+  idleCollectionStatusAtom,
+  SubscribeCollection,
+} from "~/api/materialize/subscribeCollection";
+import { useScopedSubscribeCollection } from "~/api/materialize/useScopedSubscribeCollection";
+import { allObjectsCollection } from "~/store/allObjectsCollection";
+
+import { LargestObject, TOP_OBJECT_COUNT } from "./memoryByObjectModel";
+
+/** Each replica's streamed object sizes. */
+export const sizesByReplica =
+  createScopedSubscribeCollections<ReplicaObjectSize>({
+    id: "replica-object-sizes",
+    getKey: (row) => row.objectId,
+  });
+
+/**
+ * The largest objects in `sizes`, named from `allObjectsCollection`. An object
+ * missing from the catalog is an orphaned dataflow, named by its id.
+ */
+export function useLargestObjects(
+  sizes: SubscribeCollection<ReplicaObjectSize> | undefined,
+  heapLimit: number | null,
+) {
+  const { data: rows } = useLiveQuery(
+    (q) =>
+      sizes
+        ? q
+            .from({ size: sizes.collection })
+            .leftJoin(
+              { object: allObjectsCollection.collection },
+              ({ size, object }) => eq(size.objectId, object.id),
+            )
+            .orderBy(({ size }) => size.size, {
+              direction: "desc",
+              nulls: "last",
+            })
+            // Sizes are 10 MiB-quantized, so ties are common.
+            .orderBy(({ size }) => size.objectId)
+            .limit(TOP_OBJECT_COUNT)
+        : undefined,
+    [sizes],
+  );
+  const status = useAtomValue(sizes?.statusAtom ?? idleCollectionStatusAtom);
+  const objectsStatus = useAtomValue(allObjectsCollection.statusAtom);
+  const objects = (rows ?? []).map(({ size, object }): LargestObject => ({
+    id: size.objectId,
+    name: object?.name ?? size.objectId,
+    size: size.size,
+    // A replica that reports no heap limit has no share to show.
+    memoryPercentage:
+      size.size === null || !heapLimit ? null : (size.size / heapLimit) * 100,
+    type: object?.objectType ?? null,
+    schemaName: object?.schemaName ?? null,
+    databaseName: object?.databaseName ?? null,
+    dataflowId: null,
+    dataflowName: null,
+    isOrphanedDataflow: !object,
+  }));
+  return {
+    objects,
+    // Names come from the catalog, so without it every object would read as
+    // an orphaned dataflow.
+    snapshotComplete: status.snapshotComplete && objectsStatus.snapshotComplete,
+    isRefreshing: status.isRefreshing,
+    isError: Boolean(status.error) || Boolean(objectsStatus.error),
+  };
+}
+
+/** The largest objects on a replica, streamed into a per-replica collection. */
+export function useLiveLargestObjects({
+  replicaId,
+  heapLimit,
+}: {
+  replicaId: string;
+  heapLimit: number | null;
+}) {
+  const subscribe = React.useMemo(
+    () => buildReplicaObjectSizesSubscribe(replicaId),
+    [replicaId],
+  );
+  const { collection, isScopeError } = useScopedSubscribeCollection(
+    sizesByReplica,
+    replicaId,
+    subscribe,
+  );
+  const largest = useLargestObjects(collection, heapLimit);
+  return { ...largest, isError: largest.isError || isScopeError };
+}
diff --git a/console/src/platform/clusters/LargestMaintainedQueries.tsx b/console/src/platform/clusters/LargestMaintainedQueries.tsx
index 993e67ad1486f..ee7d30811b94b 100644
--- a/console/src/platform/clusters/LargestMaintainedQueries.tsx
+++ b/console/src/platform/clusters/LargestMaintainedQueries.tsx
@@ -119,6 +119,7 @@ const LargestMaintainedQueriesError = ({
 };
 
 export interface LargestReplica {
+  id: string;
   name: string;
   /** Null while the replica reports none, as always under the process orchestrator. */
   heapLimit: number | null;
@@ -126,8 +127,8 @@ export interface LargestReplica {
 
 export interface LargestReplicaBoundaryProps {
   clusterId: string;
-  /** The replica to load instead of the cluster's largest. */
-  replicaName?: string;
+  /** A replica to show instead of the cluster's largest, which then isn't queried. */
+  replica?: LargestReplica;
   children: (replica: LargestReplica) => React.ReactNode;
   /** Wraps the loading and error states, so a caller can draw them in its card. */
   wrapFallback?: (fallback: React.ReactNode) => React.ReactNode;
@@ -136,13 +137,13 @@ export interface LargestReplicaBoundaryProps {
 const identity = (fallback: React.ReactNode) => fallback;
 
 /**
- * Loads the cluster's largest replica and renders `children` with it, behind
- * the loading and error states of the per-object memory insights. Renders
- * nothing for a cluster without replicas.
+ * Loads the cluster's largest replica, unless given one, and renders
+ * `children` with it, behind the loading and error states of the per-object
+ * memory insights. Renders nothing for a cluster without replicas.
  */
 export const LargestReplicaBoundary = ({
   clusterId,
-  replicaName,
+  replica,
   children,
   wrapFallback = identity,
 }: LargestReplicaBoundaryProps) => {
@@ -155,7 +156,7 @@ export const LargestReplicaBoundary = ({
       <React.Suspense fallback={wrapFallback(<LoadingContainer />)}>
         <LargestReplicaLoader
           clusterId={clusterId}
-          replicaName={replicaName}
+          replica={replica}
           wrapFallback={wrapFallback}
         >
           {children}
@@ -167,37 +168,52 @@ export const LargestReplicaBoundary = ({
 
 const LargestReplicaLoader = ({
   clusterId,
-  replicaName,
+  replica,
   children,
   wrapFallback,
-}: Omit<Required<LargestReplicaBoundaryProps>, "replicaName"> &
-  Pick<LargestReplicaBoundaryProps, "replicaName">) => {
-  const { data: largestReplica } = useLargestClusterReplica({
-    clusterId,
-    replicaName,
-  });
+}: Omit<Required<LargestReplicaBoundaryProps>, "replica"> &
+  Pick<LargestReplicaBoundaryProps, "replica">) => {
+  const { data: largestReplica, error } = useLargestClusterReplica(
+    { clusterId },
+    { enabled: replica === undefined },
+  );
+  const shown =
+    replica ??
+    (largestReplica && {
+      id: largestReplica.id,
+      name: largestReplica.name,
+      heapLimit:
+        largestReplica.heapLimit === null
+          ? null
+          : Number(largestReplica.heapLimit),
+    });
 
-  // Don't show anything if the cluster has no replicas
-  if (!largestReplica) return null;
+  if (!shown) {
+    // Don't show anything if the cluster has no replicas
+    if (largestReplica === null) return null;
+    return wrapFallback(
+      error ? (
+        <LargestMaintainedQueriesError error={error} />
+      ) : (
+        <LoadingContainer />
+      ),
+    );
+  }
 
+  // NOTE: `children` renders at the same place whether or not `replica` is
+  // given, so a switch between a given replica and the largest keeps its state.
   return (
     <AppErrorBoundary
-      renderFallback={({ error }) =>
+      renderFallback={({ error: childError }) =>
         wrapFallback(
           <LargestMaintainedQueriesError
-            error={error}
-            replicaName={largestReplica.name}
+            error={childError}
+            replicaName={shown.name}
           />,
         )
       }
     >
-      {children({
-        name: largestReplica.name,
-        heapLimit:
-          largestReplica.heapLimit === null
-            ? null
-            : Number(largestReplica.heapLimit),
-      })}
+      {children(shown)}
     </AppErrorBoundary>
   );
 };
diff --git a/console/src/platform/clusters/queries.test.ts b/console/src/platform/clusters/queries.test.ts
index 8f6e6fd39870d..7893250957d2e 100644
--- a/console/src/platform/clusters/queries.test.ts
+++ b/console/src/platform/clusters/queries.test.ts
@@ -44,6 +44,7 @@ import {
   useFreshnessObjects,
   useObjectHydrationStretches,
   useOwners,
+  useReplicaHeapBytes,
   useReplicaHeapLimits,
   useReplicaOfflineEvents,
   useReplicaStatusHistory,
@@ -396,6 +397,95 @@ describe("useReplicaHeapLimits", () => {
   });
 });
 
+describe("useReplicaHeapBytes", () => {
+  let requests: number;
+
+  const answerHeapMetrics = (
+    rows: Array<{
+      replicaId: string;
+      processId: string;
+      heapBytes: string | null;
+    }>,
+  ) => {
+    requests = 0;
+    server.use(
+      http.post("*/api/sql", ({ request }) => {
+        const queryKey = JSON.parse(
+          new URL(request.url).searchParams.get("query_key") ?? "[]",
+        );
+        if (queryKey[1]?.scope !== "replicaHeapLimits") return undefined;
+        requests += 1;
+        return HttpResponse.json({
+          results: [
+            mapKyselyToTabular({
+              rows: rows.map((row) => ({
+                ...row,
+                heapLimit: "1073741824",
+                memoryBytes: "1073741824",
+              })),
+              columns: buildColumns([
+                "replicaId",
+                "processId",
+                "heapLimit",
+                "memoryBytes",
+                "heapBytes",
+              ]),
+            }),
+          ],
+        });
+      }),
+    );
+  };
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("sums each replica's processes, leaving out a replica with a process that reports none", async () => {
+    answerHeapMetrics([
+      { replicaId: "u1", processId: "0", heapBytes: "400" },
+      { replicaId: "u1", processId: "1", heapBytes: null },
+      { replicaId: "u2", processId: "0", heapBytes: "500" },
+      { replicaId: "u2", processId: "1", heapBytes: "600" },
+    ]);
+    const ProviderWrapper = await createProviderWrapper();
+    const { result } = renderHook(() => useReplicaHeapBytes(["u1", "u2"]), {
+      wrapper: ProviderWrapper,
+    });
+
+    await waitFor(() =>
+      expect(result.current.data).toEqual(new Map([["u2", 1100]])),
+    );
+  });
+
+  it("keeps polling once every heap limit is in", async () => {
+    answerHeapMetrics([{ replicaId: "u1", processId: "0", heapBytes: "400" }]);
+    vi.useFakeTimers({
+      toFake: [
+        "Date",
+        "setTimeout",
+        "clearTimeout",
+        "setInterval",
+        "clearInterval",
+      ],
+    });
+    const ProviderWrapper = await createProviderWrapper();
+    renderHook(
+      () => {
+        useReplicaHeapLimits(["u1"]);
+        useReplicaHeapBytes(["u1"]);
+      },
+      { wrapper: ProviderWrapper },
+    );
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(requests).toBe(1);
+
+    await vi.advanceTimersByTimeAsync(5 * millisecondsInMinute);
+
+    expect(requests).toBeGreaterThan(1);
+  });
+});
+
 describe("replicaStatusesUpdatedAt", () => {
   it("reads the latest change of any replica's process, system replicas included", () => {
     const status = (
diff --git a/console/src/platform/clusters/queries.ts b/console/src/platform/clusters/queries.ts
index 8a42dfab946a4..0887557b3e791 100644
--- a/console/src/platform/clusters/queries.ts
+++ b/console/src/platform/clusters/queries.ts
@@ -96,6 +96,7 @@ import {
   attachOfflineEvents,
   BinnedSubscribeRow,
   bucketRowsToBucketsByReplicaId,
+  clipBinnedRowsToWindow,
   parseBinnedSubscribeRow,
   rebucketUtilizationSamples,
   toReplicaUtilizationGraphData,
@@ -474,8 +475,12 @@ export function useIndexesList(filters: ListFilters) {
   });
 }
 
-export function useLargestClusterReplica(params: LargestClusterReplicaParams) {
-  return useSuspenseQuery({
+export function useLargestClusterReplica(
+  params: LargestClusterReplicaParams,
+  { enabled = true }: { enabled?: boolean } = {},
+) {
+  return useQuery({
+    enabled,
     refetchInterval: 60_000,
     queryKey: clusterQueryKeys.largestClusterReplica(params),
     queryFn: async ({ queryKey, signal }) => {
@@ -493,6 +498,8 @@ export function useLargestClusterReplica(params: LargestClusterReplicaParams) {
 
 const STRIP_DATAFLOW_PREFIX = /^Dataflow: /;
 
+export const UNIFIED_OBJECT_SIZES_MIN_VERSION = "26.35.0-dev";
+
 export type UseLargestMaintainedQueriesParams = {
   clusterId: string;
   clusterName: string;
@@ -506,7 +513,8 @@ export function useLargestMaintainedQueries(
 ) {
   // mz_object_arrangement_sizes reports sizes reliably from v26.35, where
   // they no longer go stale across replica restarts.
-  const unifiedSizes = useEnvironmentGate("26.35.0-dev") ?? false;
+  const unifiedSizes =
+    useEnvironmentGate(UNIFIED_OBJECT_SIZES_MIN_VERSION) ?? false;
   const queryClient = useQueryClient();
   // queryClient is a stable singleton, not query input.
   // eslint-disable-next-line @tanstack/query/exhaustive-deps
@@ -746,6 +754,11 @@ export function useMaterializationLag(params: MaterializationLagParams) {
 const SUBSCRIBE_UNBINNED_MAX_MINUTES = 180; // 3h: live un-binned base, client-binned
 const SUBSCRIBE_BINNED_MAX_MINUTES = 1440; // 24h: live 5-min binned view
 export const OVERVIEW_MAX_MINUTES = 20160; // 14d: polled overview view; beyond, ad-hoc
+/** How far back each live tier's subscribe reaches. */
+export const LIVE_TIER_RETENTION_MINUTES = {
+  unbinned3h: SUBSCRIBE_UNBINNED_MAX_MINUTES,
+  binned24h: SUBSCRIBE_BINNED_MAX_MINUTES,
+};
 
 // A blue-green cutover changes a cluster's lineage. Refetching on this interval
 // bounds how long that takes to reach an open chart.
@@ -864,7 +877,6 @@ export function useUtilizationTier(timePeriodMinutes: number) {
 function useReplicaUtilizationHistorySubscribe(
   params: ReplicaUtilizationHistoryFilters,
   enabled: boolean,
-  includeOfflineEvents: boolean,
 ) {
   const { replicaId, timePeriodMinutes, bucketSizeMs, includeMemoryBreakdown } =
     params;
@@ -901,7 +913,7 @@ function useReplicaUtilizationHistorySubscribe(
   const offlineEvents = useReplicaOfflineEvents(
     data,
     timePeriodMinutes,
-    enabled && includeOfflineEvents,
+    enabled,
   );
 
   const result = useMemo(() => {
@@ -969,15 +981,10 @@ function useReplicaUtilizationHistoryBinnedSubscribe(
     const endDate = new Date();
     const startDate = subMinutes(endDate, timePeriodMinutes);
 
-    // Clip to the window like the SQL does. Held rows from a previous wider
-    // window would otherwise stretch the chart domain past the selected range.
-    const rows = data.filter(
-      (row) =>
-        row.bucketStart.getTime() >= startDate.getTime() &&
-        (!replicaId || row.replicaId === replicaId),
+    const rows = clipBinnedRowsToWindow(
+      replicaId ? data.filter((row) => row.replicaId === replicaId) : data,
+      startDate,
     );
-    // ENVELOPE UPSERT yields an unordered keyed set; the chart needs time order.
-    rows.sort((a, b) => a.bucketStart.getTime() - b.bucketStart.getTime());
 
     return toReplicaUtilizationGraphData(
       bucketRowsToBucketsByReplicaId(rows),
@@ -994,45 +1001,31 @@ function useReplicaUtilizationHistoryBinnedSubscribe(
   };
 }
 
-export function useReplicaUtilizationHistory(
-  params: ReplicaUtilizationHistoryFilters,
-  queryOptions?: {
-    enabled?: boolean;
-    // `false` skips the <=3h windows' offline-events poll, for callers that
-    // read replica status from elsewhere. Their buckets carry no events then.
-    includeOfflineEvents?: boolean;
-  },
-) {
-  const enabled = queryOptions?.enabled ?? true;
-  const tier = useUtilizationTier(params.timePeriodMinutes);
+function useUtilizationFilters(params: ReplicaUtilizationHistoryFilters) {
   const hasMemoryBreakdown =
     useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true;
-  const filters = {
+  return {
     ...params,
     // `false` would change the query key for callers that don't pass this.
     includeMemoryBreakdown:
       (params.includeMemoryBreakdown && hasMemoryBreakdown) || undefined,
   };
+}
 
-  const useUnbinnedSubscribe = tier === "unbinned3h";
-  const useBinnedSubscribe = tier === "binned24h";
-
-  const unbinnedResult = useReplicaUtilizationHistorySubscribe(
-    filters,
-    enabled && useUnbinnedSubscribe,
-    queryOptions?.includeOfflineEvents ?? true,
-  );
-  const binnedResult = useReplicaUtilizationHistoryBinnedSubscribe(
-    filters,
-    enabled && useBinnedSubscribe,
-  );
-
-  const queryResult = useQuery({
+/**
+ * Polled utilization, for windows no subscribe serves. Unlike
+ * `useReplicaUtilizationHistory` it mounts no subscribe hooks, so it opens no
+ * WebSocket (see the NOTE on `useReplicaUtilizationHistorySubscribe`).
+ */
+export function usePolledReplicaUtilizationHistory(
+  params: ReplicaUtilizationHistoryFilters,
+  { enabled }: { enabled: boolean },
+) {
+  const filters = useUtilizationFilters(params);
+  return useQuery({
     queryKey: clusterQueryKeys.replicaUtilizationHistory(filters),
     refetchInterval: 20_000,
-    // Poll whatever a subscribe isn't serving: >24h on new mz, and every window
-    // on old mz (where the subscribes are gated off).
-    enabled: enabled && !useUnbinnedSubscribe && !useBinnedSubscribe,
+    enabled,
     queryFn: async ({ queryKey, signal }) => {
       const [, queryKeyParams] = queryKey;
 
@@ -1056,6 +1049,35 @@ export function useReplicaUtilizationHistory(
       return toReplicaUtilizationGraphData(data, startDate, endDate);
     },
   });
+}
+
+export function useReplicaUtilizationHistory(
+  params: ReplicaUtilizationHistoryFilters,
+  queryOptions?: {
+    enabled?: boolean;
+  },
+) {
+  const enabled = queryOptions?.enabled ?? true;
+  const tier = useUtilizationTier(params.timePeriodMinutes);
+  const filters = useUtilizationFilters(params);
+
+  const useUnbinnedSubscribe = tier === "unbinned3h";
+  const useBinnedSubscribe = tier === "binned24h";
+
+  const unbinnedResult = useReplicaUtilizationHistorySubscribe(
+    filters,
+    enabled && useUnbinnedSubscribe,
+  );
+  const binnedResult = useReplicaUtilizationHistoryBinnedSubscribe(
+    filters,
+    enabled && useBinnedSubscribe,
+  );
+
+  const queryResult = usePolledReplicaUtilizationHistory(params, {
+    // Poll whatever a subscribe isn't serving: >24h on new mz, and every window
+    // on old mz (where the subscribes are gated off).
+    enabled: enabled && !useUnbinnedSubscribe && !useBinnedSubscribe,
+  });
 
   const active = useUnbinnedSubscribe
     ? unbinnedResult
@@ -1353,14 +1375,32 @@ export function useRollingWindowStartMs(windowMinutes: number | undefined) {
   return start?.windowMs === windowMs ? start?.startMs : undefined;
 }
 
+/** The replicas' latest per-process heap metrics, one read for heap limits and use. */
+const replicaHeapMetricsQuery = (replicaIds: string[]) => ({
+  queryKey: clusterQueryKeys.replicaHeapLimits({ replicaIds }),
+  enabled: replicaIds.length > 0,
+  queryFn: ({
+    queryKey,
+    signal,
+  }: QueryFunctionContext<
+    ReturnType<typeof clusterQueryKeys.replicaHeapLimits>
+  >) => {
+    const [, queryKeyParams] = queryKey;
+    return fetchReplicaHeapLimits({
+      params: queryKeyParams,
+      queryKey,
+      requestOptions: { signal },
+    });
+  },
+});
+
 /**
  * Each replica's per-process heap limit in bytes, for replicas reporting one.
  * Idle while `replicaIds` is empty.
  */
 export function useReplicaHeapLimits(replicaIds: string[]) {
   return useQuery({
-    queryKey: clusterQueryKeys.replicaHeapLimits({ replicaIds }),
-    enabled: replicaIds.length > 0,
+    ...replicaHeapMetricsQuery(replicaIds),
     // A replica's heap limit is fixed by its size, so polling stops once each
     // replica has a row with one, or with memory but none (an orchestrator
     // without heap limits, or a failed usage fetch on Kubernetes). A failed
@@ -1374,18 +1414,38 @@ export function useReplicaHeapLimits(replicaIds: string[]) {
       );
       return replicaIds.every((id) => settled.has(id)) ? false : 60_000;
     },
-    queryFn: ({ queryKey, signal }) => {
-      const [, queryKeyParams] = queryKey;
-      return fetchReplicaHeapLimits({
-        params: queryKeyParams,
-        queryKey,
-        requestOptions: { signal },
-      });
-    },
     select: toHeapLimitBytesByReplicaId,
   });
 }
 
+const toHeapBytesByReplicaId = (
+  rows: Awaited<ReturnType<typeof fetchReplicaHeapLimits>>,
+) => {
+  const bytes = new Map<string, number>();
+  const unreported = new Set<string>();
+  for (const { replicaId, heapBytes } of rows) {
+    if (heapBytes === null) unreported.add(replicaId);
+    else bytes.set(replicaId, (bytes.get(replicaId) ?? 0) + Number(heapBytes));
+  }
+  for (const replicaId of unreported) bytes.delete(replicaId);
+  return bytes;
+};
+
+/**
+ * Each replica's heap use in bytes, summed over its processes, for replicas
+ * whose every process reports one. Idle while `replicaIds` is empty.
+ *
+ * NOTE: shares its read with `useReplicaHeapLimits`, so while this is mounted
+ * that read keeps polling after every heap limit is in.
+ */
+export function useReplicaHeapBytes(replicaIds: string[]) {
+  return useQuery({
+    ...replicaHeapMetricsQuery(replicaIds),
+    refetchInterval: 60_000,
+    select: toHeapBytesByReplicaId,
+  });
+}
+
 /** Creation times of the given objects over the window. Idle while `objectIds` is empty. */
 export function useObjectCreationTimes(params: {
   objectIds: string[];
diff --git a/console/src/test/mockSubscribeSocket.ts b/console/src/test/mockSubscribeSocket.ts
index 2ac4a1646f724..71be64c0668bd 100644
--- a/console/src/test/mockSubscribeSocket.ts
+++ b/console/src/test/mockSubscribeSocket.ts
@@ -27,19 +27,20 @@ const METADATA_COLUMNS = [
 /**
  * Answers each SUBSCRIBE sent over the WebSocket with one snapshot of `upsert`
  * rows and the progress row that completes it, then stays silent.
- * `snapshotFor` picks a query's rows from its SQL and parameters. A query it
- * returns undefined for never completes, like a subscribe still loading.
+ * `snapshotFor` picks a query's rows from its SQL and parameters, or returns a
+ * promise of them to hold the snapshot back. A query it returns undefined for
+ * never completes, like a subscribe still loading.
  */
 export function mockSubscribeSocket(
   snapshotFor: (
     sql: string,
     params: unknown[],
-  ) => SubscribeSnapshot | undefined,
+  ) => SubscribeSnapshot | undefined | Promise<SubscribeSnapshot | undefined>,
 ) {
   return ws.link(/^wss?:\/\//).addEventListener("connection", ({ client }) => {
     let hasOptions = false;
     const send = (message: unknown) => client.send(JSON.stringify(message));
-    client.addEventListener("message", (event) => {
+    client.addEventListener("message", async (event) => {
       // The first message carries the session options.
       if (!hasOptions) {
         hasOptions = true;
@@ -50,7 +51,10 @@ export function mockSubscribeSocket(
         queries: Array<{ query: string; params?: unknown[] }>;
       };
       const [query] = request.queries;
-      const snapshot = snapshotFor(query?.query ?? "", query?.params ?? []);
+      const snapshot = await snapshotFor(
+        query?.query ?? "",
+        query?.params ?? [],
+      );
       if (!snapshot) return;
       const progressRow = (timestamp: string) => [
         timestamp,