diff --git a/.changeset/zero-rust-live-view.md b/.changeset/zero-rust-live-view.md
new file mode 100644
index 0000000000..3689d1052e
--- /dev/null
+++ b/.changeset/zero-rust-live-view.md
@@ -0,0 +1,5 @@
+---
+'zero-rust': minor
+---
+
+Add `LiveView`, a typed live query after the Swift client's `LiveQuery` (zero-react's `useQuery`). `Zero::live(&query, &ttl)` takes a typed `Query` or a named query's `QueryRequest` and returns a `LiveView<T>` whose `value()` is the query's rows, `one()` row, or aggregate read as its Rust type and kept current, with `result_type()`, `error()` (a `LiveError`: the server's query error, or rows that didn't read, which keep the last value), `add_listener`, `update_ttl`, and `view()` for the `MaterializedView`. An update reads only the rows whose entries changed, so an optimistic update with a live view of 100,000 rows read as structs no longer reads every row again. A live view follows the client when it replaces itself, and dropping it removes its listeners and destroys its view. `QueryValue` gains `update_from_view`, which has a default, and the new `typed::TypedQuery` trait names what a query reads as.
diff --git a/apps/zbugs-gpui/src/app.rs b/apps/zbugs-gpui/src/app.rs
index aeed8672b6..7d6511188c 100644
--- a/apps/zbugs-gpui/src/app.rs
+++ b/apps/zbugs-gpui/src/app.rs
@@ -729,10 +729,8 @@ impl ZbugsApp {
         }
         let exists = self
             .labels
-            .snapshot()
-            .value
+            .value()
             .iter()
-            .flatten()
             .any(|l| l.name.eq_ignore_ascii_case(&name));
         if exists {
             self.error = Some(format!("There's already a label called {name}."));
@@ -784,7 +782,7 @@ impl ZbugsApp {
         let login = self
             .me
             .as_ref()
-            .and_then(|me| me.snapshot().value.clone().flatten())
+            .and_then(|me| (*me.value()).clone())
             .map(|u| u.login);
         let logged_in = self.session.auth.is_some();
         let app = cx.entity().downgrade();
@@ -835,8 +833,7 @@ impl ZbugsApp {
     /// just edited) doesn't drag the list. At the top, the list stays at
     /// the top.
     fn keep_scroll_anchor(&mut self) {
-        let snapshot = self.list.snapshot();
-        let rows = snapshot.value.as_deref().unwrap_or_default();
+        let rows = self.list.value();
         let handle = self.scroll.0.borrow().base_handle.clone();
         let mut offset = handle.offset();
         if offset.y < px(0.) {
@@ -992,7 +989,7 @@ impl ZbugsApp {
     /// Near the end of what's loaded: ask for the next page. The rows
     /// loaded so far stay where they are, so the list doesn't move.
     fn grow(&mut self, cx: &mut Context<Self>) {
-        let loaded = self.list.snapshot().value.as_ref().map_or(0, Vec::len) as i64;
+        let loaded = self.list.value().len() as i64;
         if loaded < self.limit {
             return;
         }
@@ -1042,12 +1039,7 @@ impl ZbugsApp {
         if !self.select_first {
             return;
         }
-        let first = self
-            .list
-            .snapshot()
-            .value
-            .as_ref()
-            .and_then(|rows| Some(rows.first()?.id.clone()));
+        let first = self.list.value().first().map(|row| row.id.clone());
         let Some(id) = first else {
             return;
         };
@@ -1060,8 +1052,7 @@ impl ZbugsApp {
     /// Selects the issue `delta` rows from the selected one, or the first.
     fn move_selection(&mut self, delta: isize, cx: &mut Context<Self>) {
         let next = {
-            let snapshot = self.list.snapshot();
-            let rows = snapshot.value.as_deref().unwrap_or_default();
+            let rows = self.list.value();
             if rows.is_empty() {
                 return;
             }
@@ -1265,11 +1256,12 @@ impl ZbugsApp {
 
     /// Checks or unchecks the selected issue, for space on the list.
     fn toggle_selected_checked(&mut self, cx: &mut Context<Self>) {
-        let issue = self.list.snapshot().value.as_ref().and_then(|rows| {
-            rows.iter()
-                .find(|r| Some(&r.id) == self.selected.as_ref())
-                .cloned()
-        });
+        let issue = self
+            .list
+            .value()
+            .iter()
+            .find(|r| Some(&r.id) == self.selected.as_ref())
+            .cloned();
         if let Some(issue) = issue {
             self.toggle_checked(&issue, cx);
         }
@@ -1331,8 +1323,8 @@ impl ZbugsApp {
         let Some(detail) = self.detail.as_ref().filter(|d| d.is_synced()) else {
             return;
         };
-        let snapshot = detail.snapshot();
-        let Some(issue) = snapshot.value.as_ref().and_then(Option::as_ref) else {
+        let issue = detail.value();
+        let Some(issue) = issue.as_ref() else {
             return;
         };
         let comments = issue.comments.as_deref().unwrap_or_default();
@@ -1374,10 +1366,7 @@ impl ZbugsApp {
     /// "Zero Bugs → " and the shown issue's title, or the list's, as
     /// zbugs' issue and list pages title themselves.
     fn window_title(&self) -> String {
-        let issue = self
-            .detail
-            .as_ref()
-            .and_then(|d| d.snapshot().value.clone().flatten());
+        let issue = self.detail.as_ref().and_then(|d| (*d.value()).clone());
         let f = &self.filters;
         let shown = match issue {
             Some(issue) => issue.title,
@@ -1407,8 +1396,8 @@ impl ZbugsApp {
             .as_ref()
             .and(self.detail.as_ref())
             .and_then(|d| {
-                let snapshot = d.snapshot();
-                let issue = snapshot.value.as_ref()?.as_ref()?;
+                let issue = d.value();
+                let issue = issue.as_ref().as_ref()?;
                 is_unread(issue).then(|| issue.id.clone())
             });
         let Some(id) = unread else {
@@ -1442,7 +1431,7 @@ impl ZbugsApp {
                 let modified = this
                     .detail
                     .as_ref()
-                    .and_then(|d| d.snapshot().value.clone().flatten())
+                    .and_then(|d| (*d.value()).clone())
                     .filter(|issue| issue.id == id)
                     .map(|issue| issue.modified);
                 if let Some(modified) = modified {
@@ -1497,7 +1486,7 @@ impl ZbugsApp {
         let pref = self
             .skin_tone
             .as_ref()
-            .and_then(|live| live.snapshot().value.clone().flatten());
+            .and_then(|live| (*live.value()).clone());
         pref.and_then(|p| p.value.parse().ok())
             .filter(|tone| *tone < SKIN_TONES.len() as u32)
             .unwrap_or(0)
@@ -1618,10 +1607,8 @@ impl ZbugsApp {
     fn project_id(&self) -> Option<String> {
         let name = self.filters.project.to_lowercase();
         self.projects
-            .snapshot()
-            .value
+            .value()
             .iter()
-            .flatten()
             .find(|p| p.lower_case_name == name)
             .map(|p| p.id.clone())
     }
@@ -1801,7 +1788,7 @@ impl ZbugsApp {
     fn render_sidebar(&self, lights: Pixels, cx: &mut Context<Self>) -> impl IntoElement {
         let f = &self.filters;
         let current = f.project.to_lowercase();
-        let mut projects = self.projects.snapshot().value.clone().unwrap_or_default();
+        let mut projects = (*self.projects.value()).clone();
         projects.sort_by(|a, b| a.lower_case_name.cmp(&b.lower_case_name));
         let project = self.picker(
             Button::new("project")
@@ -1889,7 +1876,7 @@ impl ZbugsApp {
         let login = self
             .me
             .as_ref()
-            .and_then(|me| me.snapshot().value.clone().flatten())
+            .and_then(|me| (*me.value()).clone())
             .map(|u| u.login);
         let mine = |id: &'static str, on: bool, assigned: bool, cx: &mut Context<Self>| {
             let login = login.clone();
@@ -1937,10 +1924,8 @@ impl ZbugsApp {
         );
         let mut names = self
             .labels
-            .snapshot()
-            .value
+            .value()
             .iter()
-            .flatten()
             .map(|l| l.name.clone())
             .collect::<Vec<_>>();
         names.sort_by_key(|n| n.to_lowercase());
@@ -2035,11 +2020,13 @@ impl ZbugsApp {
     /// and log in or out.
     fn render_list_toolbar(&self, lights: Pixels, cx: &mut Context<Self>) -> impl IntoElement {
         let count = {
-            let s = self.count.snapshot();
-            match (s.value, self.list.snapshot().value.as_ref()) {
-                (Some(n), _) if self.count.is_synced() || n > 0 => format!("{n}"),
-                (_, Some(rows)) if !rows.is_empty() => format!("{}+", rows.len()),
-                _ => String::new(),
+            let (n, rows) = (*self.count.value(), self.list.value());
+            if self.count.is_synced() || n > 0 {
+                format!("{n}")
+            } else if !rows.is_empty() {
+                format!("{}+", rows.len())
+            } else {
+                String::new()
             }
         };
         let logged_in = self.session.auth.is_some();
@@ -2167,7 +2154,7 @@ impl ZbugsApp {
             value: None,
             checked: current.is_none(),
         }];
-        let mut users = users.snapshot().value.clone().unwrap_or_default();
+        let mut users = (*users.value()).clone();
         users.sort_by_key(|u| u.login.to_lowercase());
         choices.extend(users.into_iter().map(|u| Choice {
             avatar: Some(avatar_url(&u)),
@@ -2250,7 +2237,7 @@ impl ZbugsApp {
     }
 
     fn render_list(&self, cx: &mut Context<Self>) -> impl IntoElement {
-        let rows = self.list.snapshot().value.clone().unwrap_or_default();
+        let rows = (*self.list.value()).clone();
         let len = rows.len();
         let rows = Rc::new(rows);
         if len == 0 {
@@ -2624,7 +2611,7 @@ impl ZbugsApp {
         let Some(detail) = &self.detail else {
             return empty("Select an issue", cx);
         };
-        let issue = detail.snapshot().value.clone().flatten();
+        let issue = (*detail.value()).clone();
         let Some(issue) = issue else {
             return empty(
                 if detail.is_synced() {
@@ -2638,10 +2625,7 @@ impl ZbugsApp {
         // The newest comments first; shown oldest first. The page takes
         // over once it has at least what the issue brought.
         let mut comments: Vec<Comment> = issue.comments.clone().unwrap_or_default();
-        if let Some(page) = self
-            .comment_page
-            .as_ref()
-            .and_then(|p| p.snapshot().value.clone())
+        if let Some(page) = self.comment_page.as_ref().map(|p| (*p.value()).clone())
             && page.len() >= comments.len()
         {
             comments = page;
@@ -2658,7 +2642,7 @@ impl ZbugsApp {
             value: None,
             checked: assignee_id.is_none(),
         }];
-        let mut crew = self.crew.snapshot().value.clone().unwrap_or_default();
+        let mut crew = (*self.crew.value()).clone();
         crew.sort_by_key(|u| u.login.to_lowercase());
         assignee_choices.extend(crew.into_iter().map(|u| Choice {
             avatar: Some(avatar_url(&u)),
@@ -2681,7 +2665,7 @@ impl ZbugsApp {
             Rc::new(move |this, user_id, _, cx| this.set_assignee(id.clone(), user_id, cx)),
             cx,
         );
-        let mut all_labels = self.labels.snapshot().value.clone().unwrap_or_default();
+        let mut all_labels = (*self.labels.value()).clone();
         all_labels.sort_by_key(|l| l.name.to_lowercase());
         let mut label_choices: Vec<Choice> = all_labels
             .into_iter()
@@ -2991,10 +2975,11 @@ impl Render for ZbugsApp {
             window.push_notification(Notification::error(message), cx);
         }
         if self.reveal_selected {
-            let at = self.list.snapshot().value.as_ref().and_then(|rows| {
-                rows.iter()
-                    .position(|r| Some(&r.id) == self.selected.as_ref())
-            });
+            let at = self
+                .list
+                .value()
+                .iter()
+                .position(|r| Some(&r.id) == self.selected.as_ref());
             if let Some(ix) = at {
                 self.scroll.scroll_to_item(ix, ScrollStrategy::Center);
                 self.reveal_selected = false;
@@ -3013,10 +2998,7 @@ impl Render for ZbugsApp {
         } else {
             px(TRAFFIC_LIGHTS)
         };
-        let issue = self
-            .detail
-            .as_ref()
-            .and_then(|d| d.snapshot().value.clone().flatten());
+        let issue = self.detail.as_ref().and_then(|d| (*d.value()).clone());
         h_flex()
             .size_full()
             .bg(cx.theme().background)
diff --git a/apps/zbugs-gpui/src/app/e2e.rs b/apps/zbugs-gpui/src/app/e2e.rs
index eb2fac09ac..f62ad3e842 100644
--- a/apps/zbugs-gpui/src/app/e2e.rs
+++ b/apps/zbugs-gpui/src/app/e2e.rs
@@ -123,11 +123,11 @@ async fn settled(app: &Entity<ZbugsApp>, cx: &mut AsyncApp) -> Step {
 }
 
 fn rows(app: &ZbugsApp) -> Vec<Issue> {
-    app.list.snapshot().value.clone().unwrap_or_default()
+    (*app.list.value()).clone()
 }
 
 fn issue(app: &ZbugsApp) -> Option<Issue> {
-    app.detail.as_ref()?.snapshot().value.clone().flatten()
+    (*app.detail.as_ref()?.value()).clone()
 }
 
 async fn online(window: AnyWindowHandle, app: &Entity<ZbugsApp>, cx: &mut AsyncApp) -> Step {
@@ -143,7 +143,7 @@ async fn online(window: AnyWindowHandle, app: &Entity<ZbugsApp>, cx: &mut AsyncA
         app.selected.is_some() && app.selected.as_ref() == rows(app).first().map(|i| &i.id)
     })
     .await?;
-    let open = app.read_with(cx, |app, _| app.count.snapshot().value.unwrap_or(0));
+    let open = app.read_with(cx, |app, _| *app.count.value());
     shot("1-list");
 
     // Issues created and deleted again, one from the issue and two
@@ -206,8 +206,7 @@ async fn online(window: AnyWindowHandle, app: &Entity<ZbugsApp>, cx: &mut AsyncA
     })
     .await?;
     until(app, cx, "the new issue at the top of the list", |app| {
-        rows(app).first().is_some_and(|i| i.id == id)
-            && app.count.snapshot().value == Some(open + 1)
+        rows(app).first().is_some_and(|i| i.id == id) && *app.count.value() == open + 1
     })
     .await?;
 
@@ -312,21 +311,14 @@ async fn online(window: AnyWindowHandle, app: &Entity<ZbugsApp>, cx: &mut AsyncA
 
     // The bug label, and the issue assigned to the user.
     until(app, cx, "the labels", |app| {
-        app.labels
-            .snapshot()
-            .value
-            .iter()
-            .flatten()
-            .any(|l| l.name == "bug")
+        app.labels.value().iter().any(|l| l.name == "bug")
     })
     .await?;
     act(window, app, cx, |app, _, cx| {
         let bug = app
             .labels
-            .snapshot()
-            .value
+            .value()
             .iter()
-            .flatten()
             .find(|l| l.name == "bug")
             .map(|l| l.id.clone());
         let issue = issue(app).expect("the issue is shown");
@@ -397,9 +389,7 @@ async fn online(window: AnyWindowHandle, app: &Entity<ZbugsApp>, cx: &mut AsyncA
     // and All shows it.
     let login = app
         .read_with(cx, |app, _| {
-            app.me
-                .as_ref()
-                .and_then(|me| me.snapshot().value.clone().flatten())
+            app.me.as_ref().and_then(|me| (*me.value()).clone())
         })
         .map(|u| u.login)
         .ok_or("the user's row didn't sync")?;
diff --git a/apps/zbugs-gpui/src/live.rs b/apps/zbugs-gpui/src/live.rs
index 46710b20ae..aebfa7ee7d 100644
--- a/apps/zbugs-gpui/src/live.rs
+++ b/apps/zbugs-gpui/src/live.rs
@@ -1,36 +1,28 @@
-//! A typed Zero query kept current in a GPUI view: the view's rows are read
-//! when the client's view changes, and the GPUI view is notified then.
+//! A typed Zero query kept current in a GPUI view: a `LiveView`, whose
+//! updates notify the GPUI view.
 //!
 //! The client calls a view's listeners while it applies a change, which may
-//! be inside a GPUI update of the same view, so a listener only stores the
-//! rows and sends a wakeup; a foreground task notifies the view afterwards.
+//! be inside a GPUI update of the same view, so a listener only sends a
+//! wakeup; a foreground task notifies the view afterwards.
 
 use futures::StreamExt;
 use futures::channel::mpsc;
 use gpui_kit::{Context, Task};
-use std::cell::RefCell;
 use std::rc::Rc;
 use zbugs_model::Context as Auth;
-use zero_client::client::{MaterializedView, Zero};
+use zero_client::client::{Error, LiveView, Zero};
 use zero_client::define::QueryRequest;
 use zero_client::ivm::ResultType;
 use zero_client::ttl::Ttl;
 use zero_client::typed::QueryValue;
 
 pub struct Live<T> {
-    value: Rc<RefCell<Snapshot<T>>>,
-    view: MaterializedView,
-    remove: Box<dyn Fn()>,
+    /// None when the query's rows didn't read at first.
+    live: Option<LiveView<T>>,
     _notify: Task<()>,
 }
 
-pub struct Snapshot<T> {
-    pub value: Option<T>,
-    pub result_type: ResultType,
-    pub error: Option<String>,
-}
-
-impl<T: QueryValue + 'static> Live<T> {
+impl<T: QueryValue + Default + 'static> Live<T> {
     /// Materializes `query` and notifies the view `cx` is for whenever its
     /// rows change, until dropped.
     pub fn new<R: 'static, V: 'static>(
@@ -39,28 +31,23 @@ impl<T: QueryValue + 'static> Live<T> {
         ttl: &Ttl,
         cx: &mut Context<V>,
     ) -> Live<T> {
-        let value = Rc::new(RefCell::new(Snapshot {
-            value: None,
-            result_type: ResultType::Unknown,
-            error: None,
-        }));
         let (wake, mut woken) = mpsc::unbounded::<()>();
-        let view = zero
-            .materialize(&query, ttl)
-            .expect("a query of the app's schema materializes");
-        let v = value.clone();
-        let remove = view.add_listener(move |data, result_type, _| {
-            let mut s = v.borrow_mut();
-            match query.read(data) {
-                Ok(t) => {
-                    s.value = Some(t);
-                    s.error = None;
-                }
-                Err(e) => s.error = Some(e.0),
+        let live = match zero.live(&query, ttl) {
+            Ok(live) => {
+                // Removed when the live view drops.
+                let _ = live.add_listener(move |_, _| {
+                    let _ = wake.unbounded_send(());
+                });
+                Some(live)
             }
-            s.result_type = result_type;
-            let _ = wake.unbounded_send(());
-        });
+            // Rows the generated types can't read are a bug of the port:
+            // say so, and show none until they read.
+            Err(Error::Message(e)) => {
+                eprintln!("zbugs-gpui: {}: {e}", query.name());
+                None
+            }
+            Err(_) => panic!("a query of the app's schema materializes"),
+        };
         let notify = cx.spawn(async move |this, cx| {
             while woken.next().await.is_some() {
                 // Several changes at once notify once.
@@ -71,30 +58,21 @@ impl<T: QueryValue + 'static> Live<T> {
             }
         });
         Live {
-            value,
-            view,
-            remove,
+            live,
             _notify: notify,
         }
     }
 
-    pub fn snapshot(&self) -> std::cell::Ref<'_, Snapshot<T>> {
-        self.value.borrow()
+    /// The rows last read; none if they never could be.
+    pub fn value(&self) -> Rc<T> {
+        self.live.as_ref().map_or_else(Rc::default, LiveView::value)
     }
 
     /// The server has confirmed the rows, on this connection or an earlier
     /// one.
     pub fn is_synced(&self) -> bool {
-        matches!(
-            self.value.borrow().result_type,
-            ResultType::Complete | ResultType::Cached
-        )
-    }
-}
-
-impl<T> Drop for Live<T> {
-    fn drop(&mut self) {
-        (self.remove)();
-        self.view.destroy();
+        self.live
+            .as_ref()
+            .is_some_and(|l| matches!(l.result_type(), ResultType::Complete | ResultType::Cached))
     }
 }
diff --git a/apps/zbugs-rune/src/sync.rs b/apps/zbugs-rune/src/sync.rs
index e15a4d1e8b..ddeabde304 100644
--- a/apps/zbugs-rune/src/sync.rs
+++ b/apps/zbugs-rune/src/sync.rs
@@ -4,11 +4,13 @@
 //! `LocalSet`, while Rune owns the main thread and runs no executor. So the
 //! client lives on a thread of its own, with a current-thread runtime. The
 //! UI sends it commands (closures that run with the client), and it sends
-//! back each view's rows, read as the generated row types, with
-//! `AppHandle::update`, which runs them on the UI thread before the next
-//! frame. Reading a poke into IVM and the rows out of it stays off the UI
-//! thread.
+//! back each view's rows, read as the generated row types by a `LiveView`,
+//! with `AppHandle::update`, which runs them on the UI thread before the
+//! next frame. Reading a poke into IVM and the rows out of it stays off the
+//! UI thread.
 
+use std::any::Any;
+use std::cell::OnceCell;
 use std::collections::HashMap;
 use std::rc::Rc;
 use std::sync::Arc;
@@ -16,8 +18,8 @@ use std::sync::Arc;
 use rune::AppHandle;
 use tokio::sync::mpsc;
 use zero_client::client::{
-    ConnectionState, MaterializedView, MutationOutcome, Mutator, MutatorResult, Preload, Zero,
-    ZeroOptions, ZeroStorage,
+    ConnectionState, Error, LiveError, LiveView, MutationOutcome, Mutator, MutatorResult, Preload,
+    Zero, ZeroOptions, ZeroStorage,
 };
 use zero_client::define::QueryRequest;
 use zero_client::ivm::ResultType;
@@ -113,7 +115,7 @@ impl<S: 'static, C: Clone + Send + 'static> SyncThread<S, C> {
     /// Keeps `slot` showing `query`'s rows: `apply` gets them now, from the
     /// local store, and on every change. Watching a slot again replaces its
     /// query; the old rows stay in the app's state until the new ones come.
-    pub fn watch<R: 'static, T: QueryValue + Send + 'static>(
+    pub fn watch<R: 'static, T: QueryValue + Clone + Send + 'static>(
         &self,
         slot: &'static str,
         query: QueryRequest<C, R, T>,
@@ -212,23 +214,15 @@ struct Client<C: 'static> {
     unsubscribe: Box<dyn Fn()>,
 }
 
-type Remake<C> = Rc<dyn Fn(&Zero<C>) -> Option<(MaterializedView, Box<dyn Fn()>)>>;
+/// Makes a slot's live view on a client; dropping it stops it.
+type Remake<C> = Rc<dyn Fn(&Zero<C>) -> Option<Rc<dyn Any>>>;
 
-/// A slot's view, and how to make it again on a new client.
+/// A slot's live view, and how to make it again on a new client.
 struct Watch<C> {
-    view: Option<(MaterializedView, Box<dyn Fn()>)>,
+    _live: Option<Rc<dyn Any>>,
     remake: Remake<C>,
 }
 
-impl<C> Drop for Watch<C> {
-    fn drop(&mut self) {
-        if let Some((view, remove)) = self.view.take() {
-            remove();
-            view.destroy();
-        }
-    }
-}
-
 pub struct Worker<S: 'static, C: 'static> {
     ui: Ui<S>,
     setup: Setup<C>,
@@ -301,7 +295,7 @@ impl<S: 'static, C: Clone + 'static> Worker<S, C> {
         }
     }
 
-    fn watch<R: 'static, T: QueryValue + Send + 'static>(
+    fn watch<R: 'static, T: QueryValue + Clone + Send + 'static>(
         &mut self,
         slot: &'static str,
         query: QueryRequest<C, R, T>,
@@ -311,17 +305,34 @@ impl<S: 'static, C: Clone + 'static> Worker<S, C> {
         // Built for each client, with its context: logging in changes what
         // a query asks for.
         let remake: Remake<C> = Rc::new(move |zero| {
-            let view = zero.materialize(&query, &Ttl::seconds(10.0)).ok()?;
-            let (ui, query) = (ui.clone(), query.clone());
-            let remove = view.add_listener(move |data, result_type, _| {
-                // Rows the generated types can't read are a bug of the
-                // port; say so rather than show nothing.
-                match query.read(data) {
-                    Ok(rows) => ui.update(move |s, tx| apply(s, rows, result_type, tx)),
-                    Err(e) => eprintln!("zbugs-rune: {slot}: {e}"),
+            // Rows the generated types can't read are a bug of the port;
+            // say so rather than show nothing.
+            let live = match zero.live(&query, &Ttl::seconds(10.0)) {
+                Ok(live) => live,
+                Err(Error::Message(e)) => {
+                    eprintln!("zbugs-rune: {slot}: {e}");
+                    return None;
+                }
+                Err(_) => return None,
+            };
+            let cell: Rc<OnceCell<LiveView<T>>> = Rc::default();
+            let weak = Rc::downgrade(&cell);
+            let ui = ui.clone();
+            // Removed when the live view drops.
+            let _ = live.add_listener(move |rows, result_type| {
+                // Only set once this first call is over, which reads the
+                // rows `zero.live` already read.
+                let error = weak.upgrade().and_then(|c| c.get()?.error());
+                match error {
+                    Some(LiveError::Decode(e)) => eprintln!("zbugs-rune: {slot}: {e}"),
+                    _ => {
+                        let rows = rows.clone();
+                        ui.update(move |s, tx| apply(s, rows, result_type, tx));
+                    }
                 }
             });
-            Some((view, remove))
+            let _ = cell.set(live);
+            Some(cell as Rc<dyn Any>)
         });
         self.watches.remove(slot);
         self.install(slot, remake);
@@ -329,11 +340,23 @@ impl<S: 'static, C: Clone + 'static> Worker<S, C> {
 
     fn install(&mut self, slot: &'static str, remake: Remake<C>) {
         let Some(client) = &self.client else {
-            self.watches.insert(slot, Watch { view: None, remake });
+            self.watches.insert(
+                slot,
+                Watch {
+                    _live: None,
+                    remake,
+                },
+            );
             return;
         };
-        let view = remake(&client.zero);
-        self.watches.insert(slot, Watch { view, remake });
+        let live = remake(&client.zero);
+        self.watches.insert(
+            slot,
+            Watch {
+                _live: live,
+                remake,
+            },
+        );
     }
 
     fn close(&mut self) {
diff --git a/rust/BENCHMARKS.md b/rust/BENCHMARKS.md
index 99429d2faf..6c88c14b1b 100644
--- a/rust/BENCHMARKS.md
+++ b/rust/BENCHMARKS.md
@@ -26,11 +26,12 @@ The JS sides:
 | `memory`, `hydration`, `mutations`    | `pnpm --filter zero-conformance bench` (`src/bench/client.bench.ts`)              |
 | `pokes`, `first sync`, the round trip | none: the JS client applies pokes on the next frame, so a JS timing measures that |
 
-Two of Swift's aren't ported: `btree-set.bench.ts` (the memory sources'
+One of Swift's isn't ported: `btree-set.bench.ts` (the memory sources'
 B+ tree is internal to the crate; the benchmarks time it through views and
-mutations) and the optimistic update with a live query of structs (there is
-no typed live query; an app reads a view's rows as structs, which
-`hydration` times).
+mutations). The optimistic update with a live query of structs times a
+`LiveView`, and `mutations` also times the same update with the view's rows
+read as structs by hand on every change, as an app did before `LiveView`;
+`hydration` times making a `LiveView` next to materializing as structs.
 
 The benchmarks don't run in CI; numbers from shared runners vary too much to
 gate on. CI builds them with the workspace (`cargo clippy --all-targets`,
@@ -167,6 +168,18 @@ timed from the poke's frames arriving to the view being flushed.
   Materializing and memory didn't move (20.4 and 20.5 ms for 100,000
   rows; 82.8 and 81.5 MB).
 
+- **A view read as structs read every row on every change.** An app
+  kept a view's rows as structs by reading them all in its listener, so
+  an optimistic update with a view of 100,000 rows cost 13.1 ms where the
+  view itself took 18 µs. A `LiveView` reads only the rows whose entries
+  changed, as Swift's `LiveQuery` does, and lets the view change its rows
+  in place rather than copy them. In turns (three runs, `--time 1`), the
+  update with a view of 1,000, 10,000, and 100,000 rows read as structs
+  went from 82.8, 759, and 13,130 µs to 19.1, 61.6, and 521 µs; it still
+  compares every entry, about 5 ns each. Making a live view costs what
+  materializing and reading the rows did (116 µs and 119 µs at 1,000
+  rows, 1.46 and 1.45 ms at 10,000).
+
 Where Rust is still behind:
 
 - **The smallest changes.** Putting a row in its table's layout compares
diff --git a/rust/PARITY.md b/rust/PARITY.md
index 26e0546bc6..9b79396a96 100644
--- a/rust/PARITY.md
+++ b/rust/PARITY.md
@@ -194,11 +194,18 @@ registered throws in JS; in Rust both outcomes are a `ZeroError` saying so, and
 
 ### React and SolidJS
 
-No UI bindings: `useQuery` is a `MaterializedView` and its listener.
-`apps/zbugs-rune` shows how a Rune app binds views to its UI, on the
-shared zbugs model (`apps/zbugs-model`), whose queries and mutators are
-`define_query` and `define_mutator` definitions; `apps/zbugs-gpui` is a
-GPUI app on the same model.
+No UI bindings. `useQuery`'s typed equivalent is `LiveView`, after the
+Swift client's `LiveQuery`: `Zero::live` takes a typed `Query` or a named
+query's `QueryRequest` and keeps its value read as the query's Rust type,
+reading only the rows whose entries changed, with its result type and an
+error that is the server's query error or a row that didn't read (which
+keeps the last value). It follows the client when the client replaces
+itself, and has listeners a UI wakes itself from; the untyped equivalent is
+a `MaterializedView` and its listener. `apps/zbugs-rune` shows how a Rune
+app binds live views to its UI, on the shared zbugs model
+(`apps/zbugs-model`), whose queries and mutators are `define_query` and
+`define_mutator` definitions; `apps/zbugs-gpui` is a GPUI app on the same
+model.
 
 ## How some vectors are compared
 
diff --git a/rust/benchmarks/src/main.rs b/rust/benchmarks/src/main.rs
index f8b7e316db..939b0203be 100644
--- a/rust/benchmarks/src/main.rs
+++ b/rust/benchmarks/src/main.rs
@@ -16,13 +16,15 @@ mod harness;
 
 use fixtures::{Client, sum, update_user_args, user_json, users};
 use harness::{Bench, check, resident_memory};
-use std::cell::Cell;
+use std::cell::{Cell, RefCell};
 use std::hint::black_box;
+use std::rc::Rc;
 use zero_client::ast::SimpleOperator;
 use zero_client::hash::xxhash32;
 use zero_client::ivm::{ViewEntry, ViewValue};
 use zero_client::json::Value;
-use zero_client::typed::{QueryElement, QueryValue, ZeroValueError, column};
+use zero_client::ttl::Ttl;
+use zero_client::typed::{Query, QueryElement, QueryValue, ZeroValueError, column};
 
 fn main() {
     let mut filters = Vec::new();
@@ -144,6 +146,7 @@ fn hydration(bench: &mut Bench) {
         for n in [1_000, 10_000, 100_000] {
             if !bench.wants(&format!("materialize {n} rows"))
                 && !bench.wants(&format!("materialize {n} rows as structs"))
+                && !bench.wants(&format!("live view of {n} rows as structs"))
             {
                 continue;
             }
@@ -161,6 +164,13 @@ fn hydration(bench: &mut Bench) {
                 black_box(rows.iter().map(User::total).sum::<f64>());
                 view.destroy();
             });
+            let typed = Query::<User>::from_any(all.clone());
+            bench.run(&format!("live view of {n} rows as structs"), || {
+                let live = client.zero.live(&typed, &Ttl::default()).unwrap();
+                let rows = live.value();
+                check(rows.len() == n, "every row");
+                black_box(rows.iter().map(User::total).sum::<f64>());
+            });
         }
     });
 }
@@ -284,6 +294,45 @@ fn mutations(bench: &mut Bench) {
             });
             check_updated(&local_view, before, "updated");
 
+            // The same with the rows read as structs on every update, as an
+            // app reads them: by hand, and with a live view, which reads only
+            // the rows that changed.
+            let all = Query::<User>::from_any(users(local.zero.schema()));
+            let read = Client::new(n, false);
+            let read_view = read.materialize(all.any());
+            let rows = Rc::new(RefCell::new(Vec::new()));
+            let into = rows.clone();
+            let _remove = read_view.add_listener(move |data, _, _| {
+                *into.borrow_mut() = Vec::<User>::from_view(data).unwrap();
+            });
+            let before = value.get();
+            bench.run(
+                &format!("optimistic update, {n}-row view read as structs"),
+                || {
+                    value.set(value.get() + 1);
+                    let args = update_user_args(value.get() % n, value.get());
+                    let _ = read.zero.mutate("updateUser", Some(args));
+                },
+            );
+            let updated = |rows: &[User]| rows[value.get() % n].b == value.get() as f64;
+            if value.get() > before {
+                check(updated(&rows.borrow()), "read");
+            }
+            let structs = Client::new(n, false);
+            let live = structs.zero.live(&all, &Ttl::default()).unwrap();
+            let before = value.get();
+            bench.run(
+                &format!("optimistic update, {n}-row live query of structs"),
+                || {
+                    value.set(value.get() + 1);
+                    let args = update_user_args(value.get() % n, value.get());
+                    let _ = structs.zero.mutate("updateUser", Some(args));
+                },
+            );
+            if value.get() > before {
+                check(updated(&live.value()), "read");
+            }
+
             // The client's side of a round trip: the optimistic update, the
             // push, and the poke that confirms it, which rebases the pending
             // mutations.
diff --git a/rust/zero-client/src/client/live_view.rs b/rust/zero-client/src/client/live_view.rs
new file mode 100644
index 0000000000..cab6740359
--- /dev/null
+++ b/rust/zero-client/src/client/live_view.rs
@@ -0,0 +1,225 @@
+//! A view read as its query's Rust type and kept current (`LiveQuery` in
+//! the Swift client, `useQuery` in zero-react).
+
+use super::query_context::MaterializedView;
+use super::{Error, QueryError};
+use crate::ivm::{ResultType, ViewEntry, ViewValue};
+use crate::protocol::ErroredQuery;
+use crate::ttl::Ttl;
+use crate::typed::{QueryValue, ZeroValueError};
+use std::cell::{Cell, RefCell};
+use std::fmt;
+use std::rc::{Rc, Weak};
+
+/// Why a [`LiveView`]'s value may not be what the server has.
+#[derive(Clone, Debug, PartialEq)]
+pub enum LiveError {
+    /// The server could not run the query.
+    Query(ErroredQuery),
+    /// The rows could not be read as the query's result type. The value is
+    /// the last one read.
+    Decode(ZeroValueError),
+}
+
+impl fmt::Display for LiveError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            LiveError::Query(e) => QueryError::new(Some(e.clone())).fmt(f),
+            LiveError::Decode(e) => e.fmt(f),
+        }
+    }
+}
+
+impl std::error::Error for LiveError {}
+
+type Listener<T> = Rc<dyn Fn(&T, ResultType)>;
+
+/// What a [`LiveView`] last read.
+struct State<T> {
+    value: Rc<T>,
+    /// The entries `value` was read from.
+    read: ViewValue,
+    /// The view's rows when they were read, which it shares until they
+    /// change. Unlike holding them, which would make the view copy them to
+    /// change them, it lets them change in place, into another allocation.
+    seen: Weak<Vec<ViewEntry>>,
+    result_type: ResultType,
+    error: Option<LiveError>,
+}
+
+fn rows_of(data: &ViewValue) -> Weak<Vec<ViewEntry>> {
+    match data {
+        ViewValue::Many(rows) => Rc::downgrade(rows),
+        _ => Weak::new(),
+    }
+}
+
+struct Inner<T> {
+    view: MaterializedView,
+    state: RefCell<State<T>>,
+    listeners: RefCell<Vec<(u64, Listener<T>)>>,
+    next_listener_id: Cell<u64>,
+}
+
+/// A query's rows, the row of a `one()` query, or the value of an
+/// aggregate, read as its Rust type and kept current, from
+/// [`Zero::live`](super::Zero::live) (`LiveQuery` in the Swift client,
+/// `useQuery` in zero-react).
+///
+/// Each update of the view is read into the value, reading only the rows
+/// whose entries changed. When the client replaces itself, the view moves
+/// to the new client, and the live view with it. Dropping it removes its
+/// listeners and destroys the view; the server keeps syncing the query for
+/// its TTL.
+pub struct LiveView<T> {
+    inner: Rc<Inner<T>>,
+    remove: Box<dyn Fn()>,
+}
+
+impl<T: QueryValue + 'static> LiveView<T> {
+    /// Reads `view`, and follows it. Fails, destroying it, when its rows
+    /// don't read as `T`.
+    pub(crate) fn new(view: MaterializedView) -> Result<Self, Error> {
+        let read = view.data();
+        let value = match T::from_view(&read) {
+            Ok(value) => value,
+            Err(e) => {
+                view.destroy();
+                return Err(Error::Message(e.0));
+            }
+        };
+        let inner = Rc::new(Inner {
+            state: RefCell::new(State {
+                value: Rc::new(value),
+                seen: rows_of(&read),
+                read,
+                result_type: view.result_type(),
+                error: view.error().map(LiveError::Query),
+            }),
+            view,
+            listeners: RefCell::default(),
+            next_listener_id: Cell::new(0),
+        });
+        let weak: Weak<Inner<T>> = Rc::downgrade(&inner);
+        let remove = inner.view.add_listener(move |data, result_type, error| {
+            if let Some(inner) = weak.upgrade() {
+                inner.update(data, result_type, error);
+            }
+        });
+        Ok(LiveView { inner, remove })
+    }
+}
+
+impl<T: 'static> LiveView<T> {
+    /// The value last read. Holding it past the next update makes that
+    /// update read every row again rather than only those that changed.
+    pub fn value(&self) -> Rc<T> {
+        self.inner.state.borrow().value.clone()
+    }
+
+    /// Whether the server has confirmed the rows on this connection
+    /// (`Complete`) or an earlier one (`Cached`), or not yet (`Unknown`), or
+    /// the query failed (`Error`).
+    pub fn result_type(&self) -> ResultType {
+        self.inner.state.borrow().result_type
+    }
+
+    /// Why the server could not run the query, or why the rows could not be
+    /// read as `T`, which keeps the value last read.
+    pub fn error(&self) -> Option<LiveError> {
+        self.inner.state.borrow().error.clone()
+    }
+
+    /// The view the value is read from.
+    pub fn view(&self) -> &MaterializedView {
+        &self.inner.view
+    }
+
+    /// Raises the query's TTL; a lower one is ignored.
+    pub fn update_ttl(&self, ttl: &Ttl) {
+        self.inner.view.update_ttl(ttl);
+    }
+
+    /// Calls `listener` with the value and result type now, and after every
+    /// update of the view, once the value is read. After a read that
+    /// failed, it gets the value kept, and [`error`](LiveView::error) says
+    /// why. Returns the function that removes it.
+    pub fn add_listener(&self, listener: impl Fn(&T, ResultType) + 'static) -> Box<dyn Fn()> {
+        let id = self.inner.next_listener_id.get();
+        self.inner.next_listener_id.set(id + 1);
+        let listener: Listener<T> = Rc::new(listener);
+        self.inner
+            .listeners
+            .borrow_mut()
+            .push((id, listener.clone()));
+        let (value, result_type) = {
+            let s = self.inner.state.borrow();
+            (s.value.clone(), s.result_type)
+        };
+        listener(&value, result_type);
+        let weak = Rc::downgrade(&self.inner);
+        Box::new(move || {
+            if let Some(inner) = weak.upgrade() {
+                inner.listeners.borrow_mut().retain(|(i, _)| *i != id);
+            }
+        })
+    }
+}
+
+impl<T: QueryValue> Inner<T> {
+    /// Reads the view's update, and tells the listeners.
+    fn update(&self, data: &ViewValue, result_type: ResultType, error: Option<&ErroredQuery>) {
+        let value = {
+            let mut state = self.state.borrow_mut();
+            let s = &mut *state;
+            s.result_type = result_type;
+            let unchanged = match data {
+                ViewValue::Many(rows) => std::ptr::eq(s.seen.as_ptr(), Rc::as_ptr(rows)),
+                data => s.read == *data,
+            };
+            let read = if unchanged {
+                Ok(())
+            } else {
+                let read = match Rc::get_mut(&mut s.value) {
+                    Some(value) => value.update_from_view(&mut s.read, data),
+                    // A value the app still holds is left as it is.
+                    None => T::from_view(data).map(|value| {
+                        s.value = Rc::new(value);
+                        s.read = data.clone();
+                    }),
+                };
+                if read.is_ok() {
+                    s.seen = rows_of(data);
+                }
+                read
+            };
+            s.error = match read {
+                Ok(()) => error.cloned().map(LiveError::Query),
+                Err(e) => Some(LiveError::Decode(e)),
+            };
+            s.value.clone()
+        };
+        // Each listener in turn, reading the list again after every call,
+        // as `MaterializedView` does.
+        let mut last: Option<u64> = None;
+        loop {
+            let next = self
+                .listeners
+                .borrow()
+                .iter()
+                .find(|(id, _)| last.is_none_or(|l| *id > l))
+                .cloned();
+            let Some((id, listener)) = next else { break };
+            last = Some(id);
+            listener(&value, result_type);
+        }
+    }
+}
+
+impl<T> Drop for LiveView<T> {
+    fn drop(&mut self) {
+        (self.remove)();
+        self.inner.listeners.borrow_mut().clear();
+        self.inner.view.destroy();
+    }
+}
diff --git a/rust/zero-client/src/client/mod.rs b/rust/zero-client/src/client/mod.rs
index 00e207a486..072d782c8e 100644
--- a/rust/zero-client/src/client/mod.rs
+++ b/rust/zero-client/src/client/mod.rs
@@ -12,6 +12,7 @@ mod error;
 mod event_loop;
 mod inspector;
 mod ivm_branch;
+mod live_view;
 mod log;
 mod measure_push;
 mod mutation_tracker;
@@ -38,6 +39,7 @@ pub use inspector::{
     AnalyzeQueryOptions, AnalyzeQueryResult, Inspector, InspectorClient, InspectorClientGroup,
     InspectorQuery, Metrics, PlanDebugEvent, RowCountsBySource, RowsByQuery, RowsBySource,
 };
+pub use live_view::{LiveError, LiveView};
 pub use log::{LogLevel, LogSink, StderrLogSink};
 pub use query_context::{MaterializedView, Preload, RunType, ViewListener};
 pub use scheduler::*;
diff --git a/rust/zero-client/src/client/zero.rs b/rust/zero-client/src/client/zero.rs
index 149d8ceff7..9866e3ae44 100644
--- a/rust/zero-client/src/client/zero.rs
+++ b/rust/zero-client/src/client/zero.rs
@@ -5,6 +5,7 @@ use super::connection_manager::ConnectionState;
 use super::database::{ReplicaDatabase, replicache_name};
 use super::engine::{Engine, MutatorFn, MutatorResult, UpdateNeededReason, ZeroOptions};
 use super::inspector::{Inspector, InspectorHost};
+use super::live_view::LiveView;
 use super::query_context::{MaterializedView, Preload, RunType, WeakView};
 use super::storage::DroppedDatabases;
 use super::transaction::Transaction;
@@ -14,6 +15,7 @@ use crate::json::Value;
 use crate::query::{AnyQuery, ZeroQuery};
 use crate::schema::Schema;
 use crate::ttl::Ttl;
+use crate::typed::TypedQuery;
 use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
 use std::rc::Rc;
@@ -236,6 +238,19 @@ impl<C: 'static> Zero<C> {
         Ok(view)
     }
 
+    /// The query's result read as its Rust type and kept current: its rows,
+    /// the row of a `one()` query, or the value of an aggregate, from a
+    /// typed [`Query`](crate::typed::Query) or a named query's
+    /// `QueryRequest`. Fails when the query can't be materialized, or when
+    /// its rows don't read as the type (an [`Error::Message`] saying why).
+    pub fn live<Q>(&self, query: &Q, ttl: &Ttl) -> Result<LiveView<Q::Value>, Error>
+    where
+        Q: TypedQuery<C> + ?Sized,
+        Q::Value: 'static,
+    {
+        LiveView::new(self.materialize(query, ttl)?)
+    }
+
     /// The query's rows: what the replica holds now (`Unknown`), or once
     /// the server has confirmed them on this connection (`Complete`) or
     /// any (`Cached`). Await the result, or read it once settled.
diff --git a/rust/zero-client/src/define.rs b/rust/zero-client/src/define.rs
index 0c94bac87a..b10fdcb2e1 100644
--- a/rust/zero-client/src/define.rs
+++ b/rust/zero-client/src/define.rs
@@ -39,7 +39,7 @@ use crate::client::{ApplicationError, Error, MutateRequest, Mutator, Transaction
 use crate::ivm::ViewValue;
 use crate::json::{self, Value};
 use crate::query::{AnyQuery, ZeroQuery};
-use crate::typed::{Query, QueryValue, ZeroValueError};
+use crate::typed::{Query, QueryValue, TypedQuery, ZeroValueError};
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use std::borrow::Cow;
@@ -160,6 +160,10 @@ impl<C, R, T> ZeroQuery<C> for QueryRequest<C, R, T> {
     }
 }
 
+impl<C, R, T: QueryValue> TypedQuery<C> for QueryRequest<C, R, T> {
+    type Value = T;
+}
+
 /// A named query whose arguments come as JSON, as the server gets them,
 /// whatever its types: for a list of an app's queries, such as a test that
 /// checks them against the TS definitions.
diff --git a/rust/zero-client/src/typed.rs b/rust/zero-client/src/typed.rs
index e23a6054b4..99a5031bd6 100644
--- a/rust/zero-client/src/typed.rs
+++ b/rust/zero-client/src/typed.rs
@@ -36,6 +36,7 @@ use crate::client::Error;
 use crate::query::{ExpressionBuilder, JsonPathReference};
 use std::fmt;
 use std::marker::PhantomData;
+use std::rc::Rc;
 use std::sync::Arc;
 
 /// A value that doesn't fit the Rust type a row declares for it.
@@ -396,12 +397,170 @@ impl<V: ZeroValue> ColumnUpdate<V> {
 pub trait QueryValue: Sized {
     /// Reads the query's result from its view.
     fn from_view(view: &ViewValue) -> Result<Self, ZeroValueError>;
+
+    /// Reads `view` into `self`, which was read from `previous`, as
+    /// [`from_view`](QueryValue::from_view) would, keeping what it can of
+    /// `self`, and makes `previous` what was read: a
+    /// [`LiveView`](crate::client::LiveView) reads each update so. On an
+    /// error both are left as they were. By default it reads `view` anew;
+    /// rows (`Vec<E>`) read only the entries that changed.
+    fn update_from_view(
+        &mut self,
+        previous: &mut ViewValue,
+        view: &ViewValue,
+    ) -> Result<(), ZeroValueError> {
+        *self = Self::from_view(view)?;
+        *previous = view.clone();
+        Ok(())
+    }
 }
 
 impl<E: QueryElement> QueryValue for Vec<E> {
     fn from_view(view: &ViewValue) -> Result<Self, ZeroValueError> {
         view.entries().into_iter().map(E::from_entry).collect()
     }
+
+    /// Keeps the rows whose entries are equal, and reads the others (the
+    /// Swift client's `RowCache`). Most updates change a few rows in place,
+    /// or insert or remove one, so the entries are matched in order,
+    /// skipping at most one. `previous` becomes a copy of the entries
+    /// changed in the same way, so the view, which then shares its rows with
+    /// nothing, changes them in place.
+    fn update_from_view(
+        &mut self,
+        previous: &mut ViewValue,
+        view: &ViewValue,
+    ) -> Result<(), ZeroValueError> {
+        let (ViewValue::Many(old), ViewValue::Many(new)) = (&mut *previous, view) else {
+            *self = Self::from_view(view)?;
+            *previous = view.clone();
+            return Ok(());
+        };
+        if old.len() != self.len() {
+            *self = Self::from_view(view)?;
+            *previous = view.clone();
+            return Ok(());
+        }
+        // What the rows become, in order: runs of old rows kept, by index,
+        // and of rows read anew. Rows are read before anything changes, so a
+        // row that fails leaves both as they were.
+        let mut runs: Vec<Run> = Vec::new();
+        let mut fresh = Vec::new();
+        // Whether every row kept stays at its index, so the rows read anew
+        // replace the others where they are.
+        let mut in_place = old.len() == new.len();
+        let mut j = 0;
+        for (i, entry) in new.iter().enumerate() {
+            let kept = if old.get(j).is_some_and(|o| same_entry(o, entry)) {
+                Some(j)
+            } else if old.get(j + 1).is_some_and(|o| same_entry(o, entry)) {
+                Some(j + 1)
+            } else {
+                None
+            };
+            match (kept, runs.last_mut()) {
+                (Some(k), Some(Run::Keep { from, len })) if *from + *len == k => *len += 1,
+                (Some(k), _) => runs.push(Run::Keep { from: k, len: 1 }),
+                (None, Some(Run::Fresh(len))) => *len += 1,
+                (None, _) => runs.push(Run::Fresh(1)),
+            }
+            match kept {
+                Some(k) => {
+                    in_place &= k == i;
+                    j = k + 1;
+                }
+                None => fresh.push(E::from_entry(entry)?),
+            }
+        }
+        if in_place && fresh.is_empty() {
+            return Ok(());
+        }
+        let mut fresh = fresh.into_iter();
+        apply_runs(self, &runs, in_place, new.len(), |_| {
+            fresh.next().expect("a row read for each new entry")
+        });
+        apply_runs(Rc::make_mut(old), &runs, in_place, new.len(), |i| {
+            new[i].clone()
+        });
+        Ok(())
+    }
+}
+
+/// Whether two entries are equal, as `==` says, comparing the rows and the
+/// lists they share by address first: an entry a view didn't change shares
+/// them with the copy read before.
+fn same_entry(a: &ViewEntry, b: &ViewEntry) -> bool {
+    (Rc::ptr_eq(&a.row, &b.row) || a.row == b.row)
+        && a.relationships.len() == b.relationships.len()
+        && a.relationships
+            .iter()
+            .zip(&b.relationships)
+            .all(|((m, a), (n, b))| m == n && same_value(a, b))
+}
+
+fn same_value(a: &ViewValue, b: &ViewValue) -> bool {
+    match (a, b) {
+        (ViewValue::Many(a), ViewValue::Many(b)) => {
+            Rc::ptr_eq(a, b)
+                || (a.len() == b.len() && a.iter().zip(b.iter()).all(|(a, b)| same_entry(a, b)))
+        }
+        (ViewValue::One(Some(a)), ViewValue::One(Some(b))) => same_entry(a, b),
+        (a, b) => a == b,
+    }
+}
+
+/// A run of `Vec::update_from_view`'s rows.
+enum Run {
+    /// `len` old rows from `from`.
+    Keep { from: usize, len: usize },
+    /// `len` rows read anew.
+    Fresh(usize),
+}
+
+/// Makes `items` the `len` items `runs` lists: the old ones kept, and
+/// `fresh(i)` for the `i`th, read anew. `in_place`: each item kept stays
+/// where it is.
+fn apply_runs<X>(
+    items: &mut Vec<X>,
+    runs: &[Run],
+    in_place: bool,
+    len: usize,
+    mut fresh: impl FnMut(usize) -> X,
+) {
+    if in_place {
+        let mut at = 0;
+        for run in runs {
+            match *run {
+                Run::Keep { len, .. } => at += len,
+                Run::Fresh(n) => {
+                    for (i, item) in items[at..at + n].iter_mut().enumerate() {
+                        *item = fresh(at + i);
+                    }
+                    at += n;
+                }
+            }
+        }
+        return;
+    }
+    let mut out = Vec::with_capacity(len);
+    let mut old = std::mem::take(items).into_iter();
+    let mut next = 0;
+    for run in runs {
+        match *run {
+            Run::Keep { from, len } => {
+                if from > next {
+                    old.nth(from - next - 1);
+                }
+                out.extend(old.by_ref().take(len));
+                next = from + len;
+            }
+            Run::Fresh(n) => {
+                let at = out.len();
+                out.extend((at..at + n).map(&mut fresh));
+            }
+        }
+    }
+    *items = out;
 }
 
 /// What an `Option` query value holds: the row of `one()`, or the value of
@@ -1031,6 +1190,18 @@ impl<R, T: QueryValue> Query<R, T> {
     }
 }
 
+/// A query whose view reads as a [`QueryValue`]: a typed [`Query`], or a
+/// named query's `QueryRequest`. What
+/// [`Zero::live`](crate::client::Zero::live) takes.
+pub trait TypedQuery<C>: crate::query::ZeroQuery<C> {
+    /// What the query's view reads as.
+    type Value: QueryValue;
+}
+
+impl<C, R, T: QueryValue> TypedQuery<C> for Query<R, T> {
+    type Value = T;
+}
+
 fn names<R>(columns: Vec<ColumnRef<R>>) -> Vec<String> {
     columns.into_iter().map(|c| c.name.to_owned()).collect()
 }
diff --git a/rust/zero-client/tests/live_view.rs b/rust/zero-client/tests/live_view.rs
new file mode 100644
index 0000000000..5c8aba1c27
--- /dev/null
+++ b/rust/zero-client/tests/live_view.rs
@@ -0,0 +1,416 @@
+//! `Zero::live`: a view read as row structs and kept current, against a
+//! mock server.
+
+use std::cell::{Cell, RefCell};
+use std::rc::Rc;
+use std::sync::Arc;
+use zero_client::ast::System;
+use zero_client::client::{LiveError, LiveView, Zero, ZeroOptions};
+use zero_client::hash::query::hash_of_name_and_args;
+use zero_client::ivm::{ResultType, ViewEntry, ViewValue};
+use zero_client::json;
+use zero_client::query::AnyQuery;
+use zero_client::schema::Schema;
+use zero_client::testing::{ManualScheduler, MockSocket, MockSocketFactory, RecordingLogSink};
+use zero_client::ttl::Ttl;
+use zero_client::typed::{
+    OptionalQueryElement, Query, QueryElement, ZeroValueError, column, optional_row,
+};
+
+#[derive(Debug, PartialEq)]
+struct Issue {
+    id: String,
+    title: String,
+}
+
+thread_local! {
+    /// How many issues were read from entries.
+    static READS: Cell<usize> = const { Cell::new(0) };
+}
+
+fn reads() -> usize {
+    READS.with(Cell::get)
+}
+
+impl QueryElement for Issue {
+    fn from_entry(entry: &ViewEntry) -> Result<Self, ZeroValueError> {
+        READS.with(|r| r.set(r.get() + 1));
+        let title: String = column(entry, "title", 1)?;
+        // A title a newer server might send, which this app can't read.
+        if title == "unreadable" {
+            return Err(ZeroValueError("an unreadable title".into()));
+        }
+        Ok(Issue {
+            id: column(entry, "id", 0)?,
+            title,
+        })
+    }
+}
+
+impl OptionalQueryElement for Issue {
+    fn optional_from_view(view: &ViewValue) -> Result<Option<Self>, ZeroValueError> {
+        optional_row(view)
+    }
+}
+
+fn issue(id: &str, title: &str) -> String {
+    format!(r#"{{"op":"put","tableName":"issue","value":{{"id":"{id}","title":"{title}"}}}}"#)
+}
+
+fn del(id: &str) -> String {
+    format!(r#"{{"op":"del","tableName":"issue","id":{{"id":"{id}"}}}}"#)
+}
+
+struct Harness {
+    zero: Zero,
+    scheduler: Rc<ManualScheduler>,
+    factory: Rc<MockSocketFactory>,
+    schema: Arc<Schema>,
+    cookie: Cell<u32>,
+}
+
+impl Harness {
+    fn new() -> Self {
+        let schema = Arc::new(
+            Schema::from_json(
+                &json::parse(
+                    r#"{"tables": {"issue": {"name": "issue", "primaryKey": ["id"],
+                        "columns": {"id": {"type": "string"}, "title": {"type": "string"}}}}}"#,
+                )
+                .unwrap(),
+            )
+            .unwrap(),
+        );
+        let scheduler = ManualScheduler::new(1_700_000_000_000.0);
+        let factory = MockSocketFactory::new();
+        let options = ZeroOptions {
+            cache_url: Some("http://zero.test".into()),
+            user_id: Some("u1".into()),
+            log_sink: Some(Rc::new(RecordingLogSink::default())),
+            ..ZeroOptions::default()
+        };
+        let zero = Zero::new(
+            schema.clone(),
+            (),
+            vec![],
+            options,
+            scheduler.clone(),
+            Some(factory.clone()),
+        )
+        .unwrap();
+        let h = Harness {
+            zero,
+            scheduler,
+            factory,
+            schema,
+            cookie: Cell::new(0),
+        };
+        h.connect();
+        h
+    }
+
+    fn settle(&self) {
+        self.zero.dispatch(|| {});
+        self.scheduler.run_due_timers();
+    }
+
+    fn socket(&self) -> Rc<MockSocket> {
+        self.factory.sockets().last().cloned().expect("a socket")
+    }
+
+    fn connect(&self) {
+        self.settle();
+        self.socket().receive_text(r#"["connected",{"wsid":"w1"}]"#);
+        self.settle();
+    }
+
+    /// Every issue, by `id`.
+    fn issues(&self) -> Query<Issue> {
+        Query::from_any(
+            AnyQuery::new(self.schema.clone(), "issue", System::Client)
+                .order_by("id", zero_client::ast::Direction::Asc),
+        )
+    }
+
+    /// Every issue, as the named query `issues`.
+    fn named_issues(&self) -> Query<Issue> {
+        Query::from_any(self.issues().into_any().name_and_args("issues", vec![]))
+    }
+
+    fn live(&self, query: &Query<Issue>) -> LiveView<Vec<Issue>> {
+        self.zero.live(query, &Ttl::default()).unwrap()
+    }
+
+    /// A poke from the server: `rows` (row patch ops), and the queries it
+    /// has now got.
+    fn poke(&self, rows: &[String], got: &[&str]) {
+        let base = match self.cookie.get() {
+            0 => "null".to_owned(),
+            c => format!("\"{c:04}\""),
+        };
+        let cookie = self.cookie.get() + 1;
+        self.cookie.set(cookie);
+        let got: Vec<String> = got
+            .iter()
+            .map(|h| format!(r#"{{"op":"put","hash":"{h}"}}"#))
+            .collect();
+        let socket = self.socket();
+        socket.receive_text(&format!(
+            r#"["pokeStart",{{"pokeID":"p{cookie}","baseCookie":{base}}}]"#
+        ));
+        socket.receive_text(&format!(
+            r#"["pokePart",{{"pokeID":"p{cookie}","rowsPatch":[{}],"gotQueriesPatch":[{}]}}]"#,
+            rows.join(","),
+            got.join(",")
+        ));
+        socket.receive_text(&format!(
+            r#"["pokeEnd",{{"pokeID":"p{cookie}","cookie":"{cookie:04}"}}]"#
+        ));
+        self.settle();
+    }
+}
+
+fn titles(live: &LiveView<Vec<Issue>>) -> Vec<String> {
+    live.value().iter().map(|i| i.title.clone()).collect()
+}
+
+#[test]
+fn the_value_follows_pokes() {
+    let h = Harness::new();
+    let live = h.live(&h.issues());
+    assert!(live.value().is_empty());
+    assert_eq!(live.result_type(), ResultType::Unknown);
+    h.poke(&[issue("a", "Apple"), issue("b", "Banana")], &[]);
+    assert_eq!(
+        *live.value(),
+        vec![
+            Issue {
+                id: "a".into(),
+                title: "Apple".into()
+            },
+            Issue {
+                id: "b".into(),
+                title: "Banana".into()
+            },
+        ]
+    );
+    assert_eq!(live.error(), None);
+}
+
+#[test]
+fn a_listener_hears_the_value_now_and_on_every_update() {
+    let h = Harness::new();
+    let live = h.live(&h.issues());
+    let heard: Rc<RefCell<Vec<Vec<String>>>> = Rc::default();
+    let h2 = heard.clone();
+    let remove = live.add_listener(move |issues, _| {
+        h2.borrow_mut()
+            .push(issues.iter().map(|i| i.title.clone()).collect());
+    });
+    h.poke(&[issue("a", "Apple")], &[]);
+    h.poke(&[issue("a", "Apricot")], &[]);
+    assert_eq!(
+        *heard.borrow(),
+        vec![vec![], vec!["Apple".to_owned()], vec!["Apricot".to_owned()]]
+    );
+    remove();
+    h.poke(&[issue("b", "Banana")], &[]);
+    assert_eq!(heard.borrow().len(), 3, "removed");
+    assert_eq!(titles(&live), ["Apricot", "Banana"]);
+}
+
+#[test]
+fn the_result_type_goes_complete_once_the_server_has_the_query() {
+    let h = Harness::new();
+    let live = h.live(&h.named_issues());
+    let types: Rc<RefCell<Vec<ResultType>>> = Rc::default();
+    let t = types.clone();
+    let _remove = live.add_listener(move |_, ty| t.borrow_mut().push(ty));
+    assert_eq!(live.result_type(), ResultType::Unknown);
+    h.poke(
+        &[issue("a", "Apple")],
+        &[&hash_of_name_and_args("issues", &[])],
+    );
+    assert_eq!(live.result_type(), ResultType::Complete);
+    assert_eq!(titles(&live), ["Apple"]);
+    assert_eq!(types.borrow().last(), Some(&ResultType::Complete));
+}
+
+#[test]
+fn a_query_the_server_could_not_run_is_an_error() {
+    let h = Harness::new();
+    let live = h.live(&h.named_issues());
+    let hash = hash_of_name_and_args("issues", &[]);
+    h.socket().receive_text(&format!(
+        r#"["transformError",[{{"error":"app","id":"{hash}","name":"issues","message":"boom"}}]]"#
+    ));
+    h.settle();
+    assert_eq!(live.result_type(), ResultType::Error);
+    let error = live.error().expect("an error");
+    assert!(matches!(error, LiveError::Query(_)), "{error:?}");
+    assert_eq!(error.to_string(), "boom");
+}
+
+#[test]
+fn rows_that_do_not_read_keep_the_last_value_and_set_the_error() {
+    let h = Harness::new();
+    let live = h.live(&h.issues());
+    let heard = Rc::new(Cell::new(0));
+    let n = heard.clone();
+    let _remove = live.add_listener(move |_, _| n.set(n.get() + 1));
+    h.poke(&[issue("a", "Apple")], &[]);
+    h.poke(&[issue("b", "unreadable")], &[]);
+    assert_eq!(titles(&live), ["Apple"], "the last value read");
+    assert_eq!(
+        live.error(),
+        Some(LiveError::Decode(ZeroValueError(
+            "an unreadable title".into()
+        )))
+    );
+    assert_eq!(heard.get(), 3, "the listener heard the failed read");
+    // Rows that read again clear it.
+    h.poke(&[issue("b", "Banana")], &[]);
+    assert_eq!(titles(&live), ["Apple", "Banana"]);
+    assert_eq!(live.error(), None);
+}
+
+#[test]
+fn rows_that_do_not_read_at_first_fail_live() {
+    let h = Harness::new();
+    h.poke(&[issue("a", "unreadable")], &[]);
+    let error = h.zero.live(&h.issues(), &Ttl::default()).err();
+    assert_eq!(
+        error.map(|e| e.message()),
+        Some("an unreadable title".to_owned())
+    );
+}
+
+#[test]
+fn only_the_rows_that_changed_are_read_again() {
+    let h = Harness::new();
+    h.poke(
+        &["a", "b", "c", "d", "e"].map(|id| issue(id, &id.to_uppercase())),
+        &[],
+    );
+    let live = h.live(&h.issues());
+    assert_eq!(titles(&live), ["A", "B", "C", "D", "E"]);
+    let mut before = reads();
+
+    let mut read = |live: &LiveView<Vec<Issue>>, titles_now: &[&str]| {
+        assert_eq!(titles(live), titles_now);
+        let n = reads() - before;
+        before = reads();
+        n
+    };
+    h.poke(&[issue("c", "C2")], &[]);
+    assert_eq!(read(&live, &["A", "B", "C2", "D", "E"]), 1, "a change");
+    h.poke(&[issue("bb", "BB")], &[]);
+    assert_eq!(
+        read(&live, &["A", "B", "BB", "C2", "D", "E"]),
+        1,
+        "an insert"
+    );
+    h.poke(&[del("a")], &[]);
+    assert_eq!(read(&live, &["B", "BB", "C2", "D", "E"]), 0, "a removal");
+    h.poke(&[del("c"), issue("f", "F")], &[]);
+    assert_eq!(read(&live, &["B", "BB", "D", "E", "F"]), 1, "both");
+    h.poke(&[issue("b", "B2"), issue("e", "E2")], &[]);
+    assert_eq!(read(&live, &["B2", "BB", "D", "E2", "F"]), 2, "two changes");
+
+    // A value the app still holds is left alone: the update reads the rows
+    // anew.
+    let held = live.value();
+    h.poke(&[issue("d", "D2")], &[]);
+    assert_eq!(read(&live, &["B2", "BB", "D2", "E2", "F"]), 5, "held");
+    assert_eq!(held[2].title, "D");
+}
+
+#[test]
+fn the_view_moves_to_the_client_that_replaces_this_one() {
+    let h = Harness::new();
+    let live = h.live(&h.issues());
+    let heard: Rc<RefCell<Vec<Vec<String>>>> = Rc::default();
+    let h2 = heard.clone();
+    let _remove = live.add_listener(move |issues, _| {
+        h2.borrow_mut()
+            .push(issues.iter().map(|i| i.title.clone()).collect());
+    });
+    h.poke(&[issue("a", "Apple")], &[]);
+    let first = h.zero.client_id();
+    let sockets = h.factory.sockets().len();
+
+    // The server lost the client's state: the client replaces itself.
+    h.socket().receive_text(
+        r#"["error",{"kind":"ClientNotFound","message":"gone","origin":"zeroCache"}]"#,
+    );
+    h.settle();
+    assert_ne!(h.zero.client_id(), first, "replaced");
+    assert!(h.factory.sockets().len() > sockets, "a new connection");
+    h.cookie.set(0);
+    h.connect();
+    h.poke(&[issue("b", "Banana")], &[]);
+    assert_eq!(titles(&live), ["Banana"]);
+    // The new client's view, of its own client group's rows, and then its
+    // first poke.
+    assert_eq!(
+        *heard.borrow(),
+        [
+            vec![],
+            vec!["Apple".to_owned()],
+            vec![],
+            vec!["Banana".to_owned()]
+        ]
+    );
+    assert!(!live.view().is_destroyed());
+}
+
+#[test]
+fn dropping_it_removes_its_listeners_and_destroys_the_view() {
+    let h = Harness::new();
+    let live = h.live(&h.issues());
+    let heard = Rc::new(Cell::new(0));
+    let n = heard.clone();
+    let _remove = live.add_listener(move |_, _| n.set(n.get() + 1));
+    let view = live.view().clone();
+    assert_eq!(Rc::strong_count(&heard), 2);
+    drop(live);
+    assert_eq!(Rc::strong_count(&heard), 1, "the listener is gone");
+    assert!(view.is_destroyed());
+    h.poke(&[issue("a", "Apple")], &[]);
+    assert_eq!(heard.get(), 1);
+}
+
+#[test]
+fn a_one_query_reads_its_row() {
+    let h = Harness::new();
+    let first: Query<Issue, Option<Issue>> = Query::from_any(h.issues().into_any().one());
+    let first = h.zero.live(&first, &Ttl::default()).unwrap();
+    assert_eq!(*first.value(), None);
+    h.poke(&[issue("a", "Apple"), issue("b", "Banana")], &[]);
+    assert_eq!(
+        first.value().as_ref().as_ref().map(|i| i.id.as_str()),
+        Some("a")
+    );
+}
+
+#[cfg(feature = "serde")]
+#[test]
+fn a_named_query_request_reads_as_its_rows() {
+    use zero_client::define::{QueryDefinition, define_query};
+    thread_local! {
+        static SCHEMA: RefCell<Option<Arc<Schema>>> = const { RefCell::new(None) };
+    }
+    const ISSUES: QueryDefinition<(), (), Issue> = define_query("issues", |_, _| {
+        let schema = SCHEMA.with(|s| s.borrow().clone()).unwrap();
+        Query::from_any(AnyQuery::new(schema, "issue", System::Client))
+    });
+    let h = Harness::new();
+    SCHEMA.with(|s| *s.borrow_mut() = Some(h.schema.clone()));
+    let live = h.zero.live(&ISSUES.with(()), &Ttl::default()).unwrap();
+    h.poke(
+        &[issue("a", "Apple")],
+        &[&hash_of_name_and_args("issues", &[])],
+    );
+    assert_eq!(titles(&live), ["Apple"]);
+    assert_eq!(live.result_type(), ResultType::Complete);
+}