diff --git a/src/server/feed_hub.rs b/src/server/feed_hub.rs
new file mode 100644
index 000000000..0d2773e35
--- /dev/null
+++ b/src/server/feed_hub.rs
@@ -0,0 +1,277 @@
+//! The change feed's shared reader (docs/design/event-push.md §2).
+//!
+//! Before the hub, every `/changes/stream` connection woke on every commit,
+//! crossed the writer barrier and read its own page: N followers cost N store
+//! reads per transaction. The hub reads each new transaction ONCE, unfiltered,
+//! into a bounded in-memory ring, and every subscriber builds its own page from
+//! the ring in memory: its graph filter, its own `since`, the records renumbered
+//! exactly as a filtered store read numbers them. A subscriber does no store
+//! read while it is served from the ring, and an idle one costs nothing.
+//!
+//! The ring covers every transaction in `(floor, head]`. A subscriber behind
+//! the floor (a new one resuming from an old cursor, or one that fell behind a
+//! trim) is answered `CatchUp` and pages the store itself, exactly as before the
+//! hub, until it reaches the ring. When the store's identity or log epoch
+//! changes (a freeze or thaw re-mints it), the ring is emptied, so nobody is
+//! served a page of the old history from memory.
+
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+use std::time::Duration;
+
+use serde_json::{Value, json};
+use tokio::sync::watch;
+
+use crate::{SharedStore, StoreHandle, base::blocking};
+
+/// Transactions read per store page.
+const PAGE_TXS: usize = 64;
+/// Bytes of records the ring holds before it trims its oldest transactions.
+const RING_BYTES: usize = 64 * 1024 * 1024;
+
+/// One transaction as the hub read it: every record, in every graph.
+struct Entry {
+    tx: i64,
+    records: Arc<Vec<Value>>,
+    watermark_tx: i64,
+    watermark_timestamp: Option<String>,
+    bytes: usize,
+}
+
+struct Ring {
+    /// False until the reader's first read: every subscriber catches up.
+    ready: bool,
+    floor: i64,
+    head: i64,
+    source: Value,
+    entries: VecDeque<Entry>,
+    bytes: usize,
+}
+
+/// What the hub can do for a subscriber at its cursor.
+pub(crate) enum Serve {
+    /// The next page, built from the ring.
+    Page(Value),
+    /// Nothing in scope after the cursor yet; wait for the head to move.
+    Wait,
+    /// The cursor is behind the ring: page the store directly.
+    CatchUp,
+}
+
+pub(crate) struct Hub {
+    ring: Mutex<Ring>,
+    head: watch::Sender<i64>,
+    /// Change pages read by the hub, for the one-read-per-transaction control.
+    pub(crate) reads: AtomicU64,
+    /// Store pages read by attached subscribers catching up on their own.
+    pub(crate) catchup_reads: AtomicU64,
+}
+
+impl Hub {
+    /// The store's hub, started on first use.
+    pub(crate) fn of(store: &SharedStore) -> Arc<Self> {
+        store
+            .changes_hub
+            .get_or_init(|| {
+                let hub = Arc::new(Self {
+                    ring: Mutex::new(Ring {
+                        ready: false,
+                        floor: 0,
+                        head: 0,
+                        source: Value::Null,
+                        entries: VecDeque::new(),
+                        bytes: 0,
+                    }),
+                    head: watch::channel(0).0,
+                    reads: AtomicU64::new(0),
+                    catchup_reads: AtomicU64::new(0),
+                });
+                tokio::spawn(read_loop(Arc::downgrade(store), hub.clone()));
+                hub
+            })
+            .clone()
+    }
+
+    /// A receiver that changes whenever the ring's head moves.
+    pub(crate) fn watch(&self) -> watch::Receiver<i64> {
+        self.head.subscribe()
+    }
+
+    /// The next page for a subscriber at `cursor` with an optional graph IRI.
+    /// `scanned` is how far this subscriber has already searched the ring for
+    /// its graph, so transactions of other graphs never push it into catch-up.
+    pub(crate) fn serve(&self, cursor: i64, graph: Option<&str>, scanned: &mut i64) -> Serve {
+        let Ok(ring) = self.ring.lock() else {
+            return Serve::CatchUp;
+        };
+        let from = cursor.max(*scanned);
+        if !ring.ready || from < ring.floor {
+            return Serve::CatchUp;
+        }
+        for entry in ring.entries.iter().filter(|e| e.tx > from) {
+            let records: Vec<Value> = match graph {
+                None => entry.records.as_ref().clone(),
+                Some(iri) => entry
+                    .records
+                    .iter()
+                    .filter(|r| r["graph"].as_str() == Some(iri))
+                    .enumerate()
+                    .map(|(sequence, r)| {
+                        let mut r = r.clone();
+                        r["sequence"] = json!(sequence);
+                        r
+                    })
+                    .collect(),
+            };
+            if records.is_empty() {
+                continue;
+            }
+            return Serve::Page(json!({
+                "records": records,
+                "since": cursor,
+                "next_tx": entry.tx,
+                "watermark_tx": entry.watermark_tx,
+                "watermark_timestamp": entry.watermark_timestamp,
+                "capture": "old_and_new_values",
+                "source": ring.source,
+            }));
+        }
+        *scanned = ring.head;
+        Serve::Wait
+    }
+}
+
+/// The hub's one reader: read every new transaction after the head, then
+/// wait for the next commit. Ends when the store is gone.
+async fn read_loop(store: Weak<StoreHandle>, hub: Arc<Hub>) {
+    let Some(mut hints) = store.upgrade().map(|s| s.commit_wake.subscribe()) else {
+        return;
+    };
+    hints.borrow_and_update();
+    loop {
+        let Some(shared) = store.upgrade() else {
+            return;
+        };
+        let reader = hub.clone();
+        let read = blocking(move || {
+            // Let the hinted commit finish before reading it.
+            drop(shared.writer.lock());
+            read_new(&shared, &reader);
+            Ok(())
+        })
+        .await;
+        if read.is_err() {
+            // Unreadable: empty the ring so subscribers page the store, which
+            // reports the failure to each of them.
+            if let Ok(mut ring) = hub.ring.lock() {
+                ring.ready = false;
+                ring.entries.clear();
+                ring.bytes = 0;
+            }
+        }
+        if tokio::time::timeout(Duration::from_secs(3600), hints.changed())
+            .await
+            .is_ok_and(|r| r.is_err())
+        {
+            return;
+        }
+    }
+}
+
+/// Read every transaction after the ring's head into the ring.
+fn read_new(store: &StoreHandle, hub: &Hub) {
+    let read = store.read();
+    let Ok(source) = read.store_id().and_then(|id| Ok((id, read.log_epoch()?))) else {
+        return;
+    };
+    let source = json!({"store_id": source.0, "log_epoch": source.1});
+    let mut moved = false;
+    {
+        let Ok(mut ring) = hub.ring.lock() else {
+            return;
+        };
+        if !ring.ready || ring.source != source {
+            // First read, or a different history: start empty at the head.
+            let Ok(head) = read.latest_tx_id() else {
+                return;
+            };
+            ring.entries.clear();
+            ring.bytes = 0;
+            ring.floor = head;
+            ring.head = head;
+            ring.source = source.clone();
+            ring.ready = true;
+            moved = true;
+        }
+    }
+    // A wake with nothing new (a hint for a commit already read, or one that
+    // wrote no facts) costs one id lookup, never a page read.
+    let Ok(latest) = read.latest_tx_id() else {
+        return;
+    };
+    loop {
+        let head = match hub.ring.lock() {
+            Ok(ring) => ring.head,
+            Err(_) => return,
+        };
+        if latest <= head {
+            break;
+        }
+        hub.reads.fetch_add(1, Ordering::Relaxed);
+        let Ok(page) = read.changes_after(
+            head,
+            PAGE_TXS,
+            quipu::store::changes::Capture::OldAndNewValues,
+            None,
+        ) else {
+            return;
+        };
+        if page.records.is_empty() {
+            break;
+        }
+        // A short page reached the head: no second read to learn that.
+        let short = page.next_tx == page.watermark_tx;
+        let mut by_tx: std::collections::BTreeMap<i64, Vec<Value>> = Default::default();
+        for record in page.records {
+            by_tx
+                .entry(record["tx"].as_i64().unwrap_or_default())
+                .or_default()
+                .push(record);
+        }
+        let short = short || by_tx.len() < PAGE_TXS;
+        let Ok(mut ring) = hub.ring.lock() else {
+            return;
+        };
+        for (tx, records) in by_tx {
+            let bytes = records.iter().map(|r| r.to_string().len()).sum();
+            ring.entries.push_back(Entry {
+                tx,
+                records: Arc::new(records),
+                watermark_tx: page.watermark_tx,
+                watermark_timestamp: page.watermark_timestamp.clone(),
+                bytes,
+            });
+            ring.bytes += bytes;
+            ring.head = tx;
+        }
+        while ring.bytes > RING_BYTES {
+            let Some(old) = ring.entries.pop_front() else {
+                break;
+            };
+            ring.bytes -= old.bytes;
+            ring.floor = old.tx;
+        }
+        moved = true;
+        if short {
+            break;
+        }
+    }
+    if moved && let Ok(ring) = hub.ring.lock() {
+        hub.head.send_replace(ring.head);
+    }
+}
+
+#[cfg(test)]
+#[path = "feed_hub_tests.rs"]
+mod tests;
diff --git a/src/server/feed_hub_tests.rs b/src/server/feed_hub_tests.rs
new file mode 100644
index 000000000..ebc053031
--- /dev/null
+++ b/src/server/feed_hub_tests.rs
@@ -0,0 +1,282 @@
+//! The change feed's shared reader: one store read per committed transaction
+//! whatever the number of subscribers, pages identical to a filtered store
+//! read, catch-up behind the ring, and an emptied ring on a new log epoch.
+use std::sync::Arc;
+use std::time::Duration;
+
+use quipu::store::changes::Capture;
+use quipu::store::{Datum, Store};
+use quipu::types::{Op, Value};
+use serde_json::Value as Json;
+
+use super::{Hub, Serve};
+use crate::{SharedStore, StoreHandle};
+
+const T: &str = "2026-10-24T00:00:00Z";
+const BOARD: &str = "urn:graph:board";
+const OTHER: &str = "urn:graph:other";
+
+fn write(shared: &SharedStore, graph: &str, entity: &str) {
+    let mut store = shared.lock();
+    let datum = Datum {
+        entity: store.intern(entity).unwrap(),
+        attribute: store.intern("urn:p:title").unwrap(),
+        value: Value::Str(format!("title of {entity}")),
+        valid_from: T.into(),
+        valid_to: None,
+        op: Op::Assert,
+    };
+    // Reuse a registered graph: graph_create commits (and hints) on its own,
+    // which would make one write two commits.
+    let g = match store.registered_graph_id(graph).unwrap() {
+        Some(g) => g,
+        None => store.graph_create(graph).unwrap(),
+    };
+    store.transact_to_graph(&[datum], T, None, None, g).unwrap();
+}
+
+fn shared() -> SharedStore {
+    let shared: SharedStore = Arc::new(StoreHandle::writer_only(Store::open_in_memory().unwrap()));
+    write(&shared, BOARD, "urn:seed");
+    write(&shared, OTHER, "urn:x:seed");
+    shared
+}
+
+/// Wait until the hub's ring head reaches the store's latest transaction.
+async fn settled(shared: &SharedStore, hub: &Hub) {
+    let latest = shared.read().latest_tx_id().unwrap();
+    let mut head = hub.watch();
+    for _ in 0..200 {
+        if *head.borrow_and_update() >= latest {
+            return;
+        }
+        let _ = tokio::time::timeout(Duration::from_millis(50), head.changed()).await;
+    }
+    panic!("hub head never reached {latest}");
+}
+
+/// Wait until the hub's reader has gone quiet: no read for 100 ms. A write can
+/// commit more than one transaction, and a late hint wakes one more (empty)
+/// read; measuring before it ends would charge it to the wrong step.
+async fn quiet(hub: &Hub) {
+    let reads = || hub.reads.load(std::sync::atomic::Ordering::Relaxed);
+    let mut last = reads();
+    loop {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        let now = reads();
+        if now == last {
+            return;
+        }
+        last = now;
+    }
+}
+
+/// The page fields that do not depend on when the page was read.
+fn stable(page: &Json) -> Json {
+    serde_json::json!({
+        "records": page["records"], "since": page["since"], "next_tx": page["next_tx"],
+        "capture": page["capture"], "source": page["source"],
+    })
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn ten_subscribers_cost_one_store_read_per_commit() {
+    let shared = shared();
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    quiet(&hub).await;
+    let mut cursors = [shared.read().latest_tx_id().unwrap(); 10];
+    let mut scanned = [0; 10];
+    let reads = || hub.reads.load(std::sync::atomic::Ordering::Relaxed);
+    for i in 0..5 {
+        let before = reads();
+        write(&shared, BOARD, &format!("urn:item:{i}"));
+        settled(&shared, &hub).await;
+        quiet(&hub).await;
+        assert_eq!(
+            reads() - before,
+            1,
+            "commit {i}: one store read, whatever the number of subscribers"
+        );
+        // Serving all ten subscribers reads nothing.
+        for (cursor, scanned) in cursors.iter_mut().zip(scanned.iter_mut()) {
+            let Serve::Page(page) = hub.serve(*cursor, Some(BOARD), scanned) else {
+                panic!("subscriber not served from the ring");
+            };
+            *cursor = page["next_tx"].as_i64().unwrap();
+        }
+        assert_eq!(reads() - before, 1);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_ring_page_equals_the_filtered_store_page() {
+    let shared = shared();
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    let start = shared.read().latest_tx_id().unwrap();
+    write(&shared, OTHER, "urn:x:1");
+    write(&shared, BOARD, "urn:item:1");
+    write(&shared, OTHER, "urn:x:2");
+    settled(&shared, &hub).await;
+    for graph in [Some(BOARD), Some(OTHER), None] {
+        let mut cursor = start;
+        let mut scanned = 0;
+        loop {
+            let served = hub.serve(cursor, graph, &mut scanned);
+            let store = shared.read();
+            let g = graph.map(|iri| store.lookup(iri).unwrap().unwrap());
+            let want = store
+                .changes_after(cursor, 1, Capture::OldAndNewValues, g)
+                .unwrap()
+                .to_json();
+            match served {
+                Serve::Page(page) => {
+                    assert_eq!(stable(&page), stable(&want), "graph {graph:?} at {cursor}");
+                    cursor = page["next_tx"].as_i64().unwrap();
+                }
+                Serve::Wait => {
+                    assert!(want["records"].as_array().unwrap().is_empty());
+                    break;
+                }
+                Serve::CatchUp => panic!("a cursor inside the ring must not catch up"),
+            }
+        }
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_cursor_behind_the_ring_catches_up_from_the_store() {
+    let shared = shared();
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    let mut scanned = 0;
+    // The seed transaction predates the hub: the ring starts after it.
+    assert!(matches!(
+        hub.serve(0, Some(BOARD), &mut scanned),
+        Serve::CatchUp
+    ));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn other_graphs_never_push_a_subscriber_into_catch_up() {
+    let shared = shared();
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    let cursor = shared.read().latest_tx_id().unwrap();
+    let mut scanned = 0;
+    for i in 0..3 {
+        write(&shared, OTHER, &format!("urn:x:{i}"));
+        settled(&shared, &hub).await;
+        assert!(matches!(
+            hub.serve(cursor, Some(BOARD), &mut scanned),
+            Serve::Wait
+        ));
+    }
+    assert_eq!(scanned, shared.read().latest_tx_id().unwrap());
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_new_log_epoch_empties_the_ring() {
+    let shared = shared();
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    let cursor = shared.read().latest_tx_id().unwrap();
+    write(&shared, BOARD, "urn:item:1");
+    settled(&shared, &hub).await;
+    let mut scanned = 0;
+    assert!(matches!(
+        hub.serve(cursor, Some(BOARD), &mut scanned),
+        Serve::Page(_)
+    ));
+    shared
+        .lock()
+        .mint_log_epoch("test: history removed")
+        .unwrap();
+    write(&shared, BOARD, "urn:item:2");
+    settled(&shared, &hub).await;
+    let mut scanned = 0;
+    assert!(
+        matches!(hub.serve(cursor, Some(BOARD), &mut scanned), Serve::CatchUp),
+        "a page of the old epoch was served from memory"
+    );
+}
+
+/// A real `/changes/stream` subscriber; its frames, as `(event, id)`.
+fn subscribe(
+    shared: &SharedStore,
+    since: i64,
+    graph: &str,
+) -> tokio::sync::mpsc::Receiver<(String, String)> {
+    use futures::StreamExt;
+    let params =
+        serde_json::from_value(serde_json::json!({"since": since, "graph": graph})).unwrap();
+    let (frames, received) = tokio::sync::mpsc::channel(64);
+    let shared = shared.clone();
+    tokio::spawn(async move {
+        let response = super::super::changes_stream(
+            axum::extract::State(shared),
+            axum::http::HeaderMap::new(),
+            axum::extract::Query(params),
+        )
+        .await;
+        let mut body = response.into_body().into_data_stream();
+        let mut buffer = String::new();
+        while let Some(Ok(chunk)) = body.next().await {
+            buffer.push_str(std::str::from_utf8(&chunk).unwrap());
+            while let Some(end) = buffer.find("\n\n") {
+                let block: String = buffer.drain(..end + 2).collect();
+                let field = |name: &str| {
+                    block
+                        .lines()
+                        .find_map(|l| l.strip_prefix(name))
+                        .unwrap_or_default()
+                        .to_string()
+                };
+                let event = field("event: ");
+                if !event.is_empty() && frames.send((event, field("id: "))).await.is_err() {
+                    return;
+                }
+            }
+        }
+    });
+    received
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_quiet_graph_behind_the_ring_floor_stops_reading_the_store() {
+    // BOARD's last transaction predates the hub; OTHER moves the store past
+    // it, so the ring's floor is above the subscriber's cursor.
+    let shared = shared();
+    let board_last = shared.read().latest_tx_id().unwrap() - 1;
+    write(&shared, OTHER, "urn:x:before");
+    let hub = Hub::of(&shared);
+    settled(&shared, &hub).await;
+    let mut frames = subscribe(&shared, board_last, BOARD);
+    quiet(&hub).await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    let catchup = || hub.catchup_reads.load(std::sync::atomic::Ordering::Relaxed);
+    let before = catchup();
+
+    // Commits in other graphs only: served from the ring, no store read.
+    for i in 0..3 {
+        write(&shared, OTHER, &format!("urn:x:{i}"));
+        settled(&shared, &hub).await;
+    }
+    quiet(&hub).await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert_eq!(
+        catchup(),
+        before,
+        "unrelated commits sent the subscriber to the store"
+    );
+
+    // And nothing in scope is lost: the next BOARD commit arrives.
+    write(&shared, BOARD, "urn:item:after");
+    let board_tx = shared.read().latest_tx_id().unwrap();
+    let frame = tokio::time::timeout(Duration::from_secs(5), frames.recv())
+        .await
+        .expect("the BOARD commit never arrived")
+        .unwrap();
+    assert_eq!(frame, ("quipu.changes".to_string(), board_tx.to_string()));
+}
diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs
index fa39bd4b2..8972bd2e3 100644
--- a/src/server/feed_stream.rs
+++ b/src/server/feed_stream.rs
@@ -26,6 +26,9 @@ use super::{
     feed,
 };
 
+#[path = "feed_hub.rs"]
+pub(crate) mod feed_hub;
+
 #[derive(serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct Params {
@@ -45,6 +48,12 @@ struct Cursor {
     terminal: bool,
     /// Frames of one chunked transaction still to send, in order.
     pending: VecDeque<FeedFrame>,
+    /// A change stream is served from the shared reader (`feed_hub`) once
+    /// its first page has been read from the store (which validates the
+    /// graph); it then waits on the hub's head instead of its own hint.
+    hub: Option<(Arc<feed_hub::Hub>, watch::Receiver<i64>)>,
+    /// How far the hub's ring has been searched for this stream's graph.
+    scanned: i64,
     _slot: OwnedSemaphorePermit,
 }
 
@@ -106,6 +115,8 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool
         wait: false,
         terminal: false,
         pending: VecDeque::new(),
+        hub: None,
+        scanned: 0,
         _slot: slot,
     };
     let output = unfold(cursor, |mut cursor| async move {
@@ -116,7 +127,13 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool
             return None;
         }
         loop {
-            if cursor.wait {
+            if cursor.wait
+                && let Some((_, head)) = cursor.hub.as_mut()
+            {
+                // The hub crosses the writer barrier and reads once for every
+                // stream; this one only waits for the ring's head to move.
+                let _ = tokio::time::timeout(Duration::from_secs(3600), head.changed()).await;
+            } else if cursor.wait {
                 // Hourly metadata reconcile covers external writers that did
                 // not install this serving connection's hook. Heartbeats do
                 // not query the store. No per-connection polling loop.
@@ -145,7 +162,27 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool
                     cursor,
                 ));
             }
-            let page = if cursor.changes {
+            if let Some((_, head)) = cursor.hub.as_mut() {
+                head.borrow_and_update();
+            }
+            let served = match &cursor.hub {
+                Some((hub, _)) if cursor.changes => hub.serve(
+                    cursor.offset,
+                    cursor.params.graph.as_deref(),
+                    &mut cursor.scanned,
+                ),
+                _ => feed_hub::Serve::CatchUp,
+            };
+            let page = if let feed_hub::Serve::Page(body) = served {
+                Ok(axum::Json(body))
+            } else if matches!(served, feed_hub::Serve::Wait) {
+                cursor.wait = true;
+                continue;
+            } else if cursor.changes {
+                if let Some((hub, _)) = &cursor.hub {
+                    hub.catchup_reads
+                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                }
                 let params = serde_json::from_value(
                     json!({"since": cursor.offset, "limit": 1, "capture": "old_and_new_values", "graph": cursor.params.graph}),
                 );
@@ -199,6 +236,14 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool
                     ));
                 }
             };
+            // The first store page validated this stream's graph: from here
+            // on it is served from the shared reader.
+            let attached = cursor.changes && cursor.hub.is_none();
+            if attached {
+                let hub = feed_hub::Hub::of(&cursor.store);
+                let head = hub.watch();
+                cursor.hub = Some((hub, head));
+            }
             let key = if cursor.changes {
                 "next_tx"
             } else {
@@ -209,7 +254,19 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool
                 .and_then(Value::as_i64)
                 .unwrap_or(cursor.offset);
             if next <= cursor.offset {
-                cursor.wait = true;
+                // An empty store page proves no in-scope transaction up to its
+                // watermark: the stream need not search below it again, so a
+                // quiet graph behind the ring floor reaches the ring instead of
+                // re-reading the store on every unrelated commit. The cursor
+                // itself does not move.
+                if cursor.changes
+                    && let Some(watermark) = body.get("watermark_tx").and_then(Value::as_i64)
+                {
+                    cursor.scanned = cursor.scanned.max(watermark);
+                }
+                // A stream that just attached asks the hub before it waits, so
+                // a commit the hub read meanwhile is not slept through.
+                cursor.wait = !attached;
                 continue;
             }
             let data = body.to_string();
diff --git a/src/server/handle.rs b/src/server/handle.rs
index 6639aad01..c16071c8f 100644
--- a/src/server/handle.rs
+++ b/src/server/handle.rs
@@ -32,6 +32,8 @@ pub(crate) type SharedStore = Arc<StoreHandle>;
 pub(crate) struct StoreHandle {
     /// Coalesced hints only; durable rows remain in the existing SQLite logs.
     pub(crate) commit_wake: tokio::sync::watch::Sender<()>,
+    /// The change feed's shared reader, started by the first change stream.
+    pub(crate) changes_hub: std::sync::OnceLock<Arc<super::feed_stream::feed_hub::Hub>>,
     pub(crate) graph_metrics: super::graph_metrics::GraphMetrics,
     pub(crate) writer: FairMutex<quipu::Store>,
     pub(crate) readers: ReadPool,
@@ -154,6 +156,7 @@ impl StoreHandle {
         let commit_wake = Self::commit_wake_for(&store);
         Self {
             commit_wake,
+            changes_hub: std::sync::OnceLock::new(),
             graph_metrics: super::graph_metrics::GraphMetrics::new(db_path),
             vector_reads_pooled: store.has_sqlite_vector_backend(),
             embedding_provider: store.embedding_provider(),
@@ -195,6 +198,7 @@ impl StoreHandle {
         let commit_wake = Self::commit_wake_for(&store);
         Self {
             commit_wake,
+            changes_hub: std::sync::OnceLock::new(),
             graph_metrics: super::graph_metrics::GraphMetrics::new(":memory:"),
             vector_reads_pooled: store.has_sqlite_vector_backend(),
             embedding_provider: store.embedding_provider(),
diff --git a/src/server/tests.rs b/src/server/tests.rs
index 7f00df7de..495388f3d 100644
--- a/src/server/tests.rs
+++ b/src/server/tests.rs
@@ -1039,6 +1039,7 @@ pub(super) fn pooled_handle_with_provider(
     }
     let handle = super::StoreHandle {
         commit_wake: super::StoreHandle::commit_wake_for(&store),
+        changes_hub: std::sync::OnceLock::new(),
         graph_metrics: super::graph_metrics::GraphMetrics::new(&path),
         embedding_provider: store.embedding_provider(),
         search_config: store.search_config().clone(),