Merge pull request #358 from fiddlerwoaroof/dbsp/live-source

feat(incremental): a backend over a source that reports its changes (improvise-rnak)

context/code/improvise-incremental.md

@@ -388,3 +388,46 @@ that say otherwise (step 4 items 4 and 10, step 5 item 5). Each is a user-visibl
 - **Overflow of the exact sum to infinity** follows `ExactSum::value` (round to nearest, so an overflowing
   total is infinite): correct IEEE behaviour.
 
+
+## Step 7: live sources (improvise-rnak)
+
+ADR 009 clause 2 names two kinds of live source: one that reports its changes, which feeds cell deltas
+like `DataStore`, and one that answers aggregate queries but reports none, which implements the derived
+value by pushdown. The core half of both is in `context/code/source.md` (`CellSource`, `Model::over`).
+The brief ordered this half after the evaluator hook; the hook came first because a source that reports
+changes still has its formulas evaluated over it. User ruling 2026-10-10: formulas are computed in RAM
+from the source's folds, never pushed down; `Derived` stays in this crate.
+
+### Change-reporting (`Incremental::over`)
+
+1. **`ChangeSource` is `CellSource` plus `drain_changes(&self)`, and a source's cells are those as of the
+   last drain.** A shape is built lazily, by a pass over the source, at its first read. If a source
+   showed a write before reporting it, a shape built in between would hold the write and the next
+   drain would add it again. So the contract is that nothing is visible until it is drained, which is
+   what a database does inside a read transaction and what `DataStore` does already (its log is
+   drained in the step that made the edit). `&self` because a live source is shared with the model that
+   reads it and changes behind it by interior mutability of its own.
+2. **`Incremental::over(model, feed, none_cats)`** takes a `SharedFeed`, the alias core exports (a generic
+   `Arc<S>` could not take it: `S` is `Sized`). The model is put over the feed (`Model::over`), which
+   drops the cells it came with, so none is left for `get_cell`, persistence or the harness to read. Its categories, formulas and aggregations are the definition; the
+   source holds the data. Shapes and stems are built from the feed as it stands.
+3. **`sync()` drains and maintains; `apply_changes(&[CellChange])` maintains changes the source already
+   reflects**, for a source whose changes reach the backend by another route (a trigger table, a
+   channel). `Maintained::apply` on a fed backend takes definition edits (a formula, an aggregation)
+   and drains: a cell edit is the source's to make, and `apply` refuses it with an `assert!` in every
+   build (a release build lost it silently when only `Model::set_cell`'s `debug_assert` stood in the way).
+   **A feed has one drainer**: a drain is destructive, so two backends over one feed split its changes
+   (`context/code/source.md`, `ChangeSource`).
+4. **Freshness is the source's.** ADR 003 left "polled, triggered, or stale until refreshed" open: `sync`
+   is the poll, and when it is called is the host's decision. Nothing here schedules it.
+5. **What a change costs** is what it costs for the owned store: two group updates and two exact-sum
+   updates per maintained shape, no pass over the source (`the_cost_of_a_reported_change_...` counts the
+   rows the source serves for snapshots: zero after the build). Formulas are evaluated at the stems the
+   change reaches, as in step 5.
+6. **Three routes into the store were routed through the source**: `Shape::of` (the build) reads
+   `cells(&[])`, `Stems::build` reads `keys(&[])`, and `Incremental::read` reads `value_at`. `Eager::occupancy`
+   reads `keys(&[])` too, so an `Eager` over a source would answer it.
+
+`tests/support/live.rs` holds `LiveTable`, the scripted source (an indexed store behind a lock, with
+a pending list and a counter of snapshot rows) and `Fed`, an `Incremental` over one that a differential
+can drive: a cell edit becomes a write to the table followed by `sync`.

context/code/source.md

@@ -92,6 +92,18 @@ implementation, written as a query engine with no index would answer, sharing on
 fixture (every subset of every stored key as a base, with no span, one span of each of six item
 lists, and two), at both widths and for all five functions: over 5,000 triples.
 
+## `ChangeSource`
+
+`CellSource` plus `drain_changes(&self) -> Vec<CellChange>`: a source that reports what changed in it.
+The contract that matters is that a source's cells are those as of the last drain, so a snapshot read
+now plus the changes drained later counts no change twice or not at all. Why, and how `Incremental`
+uses it: `context/code/improvise-incremental.md`, step 7. `DataStore` is not a `ChangeSource`: its log
+is `&mut` and is drained by the owner in the step that made the edit. A `SharedFeed` is the `Arc` a
+backend and a model share. **A feed has one drainer**: a drain empties the pending list, so two
+backends over one feed split the changes between them (measured in review: the same write gave one
+backend a Revenue of 128 and the other 38, and the second then disagreed with its own exact reads).
+`a_feed_has_one_drainer` pins the behaviour. Kb-style invariant: `invariant(change_source, one_drainer)`.
+
 ## Building a `Summary` from an engine's answer
 
 `Summary::from_parts(Numbers::from_parts(count, exact_sum, least, greatest)?, Agreement)?`. `Agreement`

context/test-evidence/improvise-incremental.md

@@ -801,3 +801,69 @@ User-visible changes this evidence also covers: a cycle that only flips a zero's
 same cycle over nonzero data always was; a cycle fed a NaN (`A = A + 1`, `A = A` with `A` NaN) settles on
 NaN instead of `circular`.
 
+
+### a_live_source_matches_eager_over_the_corpus_and_its_edits
+
+improvise-rnak. `Incremental` over a `LiveTable` (the model's cells moved into the table, the model's
+own store emptied) against `Eager` over the shipped documents, their views and `edits_for`
+(trimmed to three edits above 40 cells, as `corpus_eager` does). Rivals, each failing it by name: `sync`
+drains nothing; the model not put over the feed (formulas read an empty store); a shape built from an
+empty store; the stems built from the model's own store.
+
+### changes_named_by_the_caller_are_maintained_as_the_drain_would
+
+`apply_changes` with the changes the table reports, against `Eager` after each of four edits (an
+overwrite, a new record under an item the model lacks, a removal, a text), asking every probe of every
+step so far. Rivals: `apply_changes` ignores its changes (only this test and no other fails it); the
+stems from the owned store.
+
+### a_write_is_invisible_until_it_is_drained
+
+The contract of a `ChangeSource`: a read after a write and before the drain sees the old value, after
+`sync` the new one, and a second `sync` changes nothing. Rivals: the table shows a write at once (this
+test alone fails it, which is why the contract has a test of its own: the differential cannot see it,
+since every shape it asks about was built before the write); `sync` drains nothing.
+
+### the_cost_of_a_reported_change_does_not_depend_on_the_size_of_the_table
+
+One overwrite reported by the table, after the shape is built, at 216 and at 864 cells: two group
+updates and two sum updates, no build, equal meters, and zero snapshot rows served by the source
+for the change. The build itself serves exactly the table (`n * n * 6` rows), so the zero is a count
+that could have been positive. Rival: every sync rebuilds the shapes (`builds` and `built_cells` move,
+and the table serves the snapshot again).
+
+### an_overwrite_by_the_same_value_moves_nothing
+
+A table that reports an overwrite of a cell by its own value leaves the fold where it was (23). Rival:
+the model not over the feed fails it with the others.
+
+### a_live_source_matches_eager_over_random_formula_and_cell_edits
+
+`Fed` against `Eager` over random formula and cell edits on the formula fixture (no views, so the binary
+stays under two seconds). Rivals: `sync` drains nothing, the model not over the feed, a shape built from an
+empty store and the stems from the owned store all fail it.
+
+### a_backend_can_be_built_over_the_exported_feed_alias
+
+Review of #358: `over<S>(Arc<S>)` could not take the `SharedFeed` the same PR exports. The test builds
+a backend from a `SharedFeed`; it did not compile against the generic signature.
+
+### a_populated_model_put_over_a_feed_keeps_no_cells_of_its_own
+
+Review of #358: `Fed::new` emptied the store before the call, so no test passed a populated model. This
+one does, asserts the owned store is empty afterwards and that every harness probe agrees with
+`Eager`. Rival: `Model::over` keeping the cells fails it, and only it.
+
+### a_cell_edit_on_a_fed_backend_is_refused
+
+`Maintained::apply` with a `SetCell` panics with the backend's own message in every build. Rival: the
+assert removed (the test then sees the model's message, not the backend's, and fails).
+
+### a_cell_clear_on_a_fed_backend_is_refused
+
+The same for `ClearCell`.
+
+### a_feed_has_one_drainer
+
+Review of #358: two backends over one feed; one write; `a.sync(); b.sync()`. The first sees it, the
+second does not. This pins the behaviour as documented rather than as a guarantee: a drain is destructive.

crates/improvise-core/src/model/source.rs

@@ -1,7 +1,7 @@
 //! Where cell values are read from: a read-only trait, and the aggregate it answers in.
 //! Reasoning: `context/code/source.md` (ADR 003, ADR 009 clause 2; improvise-qddo).
 
-use super::cell::{CellKey, CellValue, DataStore};
+use super::cell::{CellChange, CellKey, CellValue, DataStore};
 use crate::formula::AggFunc;
 use numtower::ExactSum;
 
@@ -213,6 +213,16 @@ impl Summary {
 /// its readers (a database) does so through interior mutability of its own.
 pub type SharedSource = std::sync::Arc<dyn CellSource + Send + Sync>;
 
+/// A source that reports its changes; its cells are the state as of its last drain.
+/// A drain is destructive: a feed has one drainer (`context/code/source.md`).
+pub trait ChangeSource: CellSource {
+    /// The changes since the last drain, oldest first; reads see them from now on.
+    fn drain_changes(&self) -> Vec<CellChange>;
+}
+
+/// A change-reporting source a backend and a model share.
+pub type SharedFeed = std::sync::Arc<dyn ChangeSource + Send + Sync>;
+
 /// The read side of cell storage (ADR 003). Nothing here writes, so a source that cannot be
 /// written to is not one that fails at runtime: it has no method to fail in.
 pub trait CellSource {

crates/improvise-incremental/src/eager.rs

@@ -67,9 +67,9 @@ impl Derived for Eager {
                 held == Some(item.as_str()) || self.group_holds(cat, item, held)
             })
         };
-        let cells = self.model.data.iter_cells();
+        let cells = self.model.source().keys(&[]);
         cells
-            .filter(|(k, _)| named_by(k) && k.0.iter().all(drawn))
+            .filter(|k| named_by(k) && k.0.iter().all(drawn))
             .count()
     }
 }

crates/improvise-incremental/src/formulas.rs

@@ -240,7 +240,7 @@ impl Stems {
             order: Order::of(model),
             wide,
         };
-        let cells: Vec<CellKey> = model.data.iter_cells().map(|(k, _)| k).collect();
+        let cells: Vec<CellKey> = model.source().keys(&[]).collect();
         let presence: Vec<(&CellKey, i64)> = cells.iter().map(|k| (k, 1)).collect();
         stems.count(&presence);
         stems

crates/improvise-incremental/src/incremental.rs

@@ -9,6 +9,7 @@ use crate::{Derived, Edit, Maintained, Reading, cell_delta};
 use improvise_core::formula::ast::AggFunc;
 use improvise_core::model::category::MEASURE_CATEGORY;
 use improvise_core::model::cell::{CellChange, CellKey, CellValue};
+use improvise_core::model::source::{SharedFeed, SharedSource};
 use improvise_core::model::types::{Covers, settled};
 use improvise_core::model::{Aggregation, Model};
 use std::collections::BTreeMap;
@@ -31,6 +32,8 @@ pub struct Incremental {
     fingerprint: Vec<String>,
     evaluation: Evaluation,
     work: Meter,
+    /// The source the model is over, when its cells are not the model's own.
+    feed: Option<SharedFeed>,
 }
 
 static SUM: AggFunc = AggFunc::Sum;
@@ -38,6 +41,18 @@ static SUM: AggFunc = AggFunc::Sum;
 impl Incremental {
     pub fn new(mut model: Model, none_cats: Vec<String>) -> Self {
         model.data.record_changes();
+        Self::start(model, none_cats, None)
+    }
+
+    /// A backend over `feed`, which holds the cells and reports their changes; `model`'s own
+    /// store is not read. Reasoning: `context/code/improvise-incremental.md`, step 7.
+    pub fn over(model: Model, feed: SharedFeed, none_cats: Vec<String>) -> Self {
+        let source: SharedSource = feed.clone();
+        let model = model.over(source);
+        Self::start(model, none_cats, Some(feed))
+    }
+
+    fn start(model: Model, none_cats: Vec<String>, feed: Option<SharedFeed>) -> Self {
         let mut backend = Incremental {
             model,
             none_cats,
@@ -46,12 +61,37 @@ impl Incremental {
             fingerprint: Vec::new(),
             evaluation: Evaluation::default(),
             work: Meter::default(),
+            feed,
         };
         backend.rebuild_formulas();
         backend.work = Meter::default();
         backend
     }
 
+    /// Maintains what the feed has reported since the last drain.
+    pub fn sync(&mut self) {
+        let changes = self.drain();
+        self.absorb(None, &changes);
+    }
+
+    /// Maintains `changes`, which the source already reflects. For a source whose changes reach the
+    /// backend by another route than the drain.
+    pub fn apply_changes(&mut self, changes: &[CellChange]) {
+        self.absorb(None, changes);
+    }
+
+    fn drain(&mut self) -> Vec<CellChange> {
+        match &self.feed {
+            Some(feed) => feed.drain_changes(),
+            None => self.model.data.drain_changes(),
+        }
+    }
+
+    fn absorb(&mut self, edit: Option<&Edit>, changes: &[CellChange]) {
+        self.shapes.apply(&cell_delta(changes), changes);
+        self.refresh_formulas(edit, changes);
+    }
+
     pub fn with_evaluation(mut self, evaluation: Evaluation) -> Self {
         self.evaluation = evaluation;
         self.rebuild_formulas();
@@ -100,20 +140,24 @@ impl Incremental {
         let func = Reduce::of(self.func_for(key));
         let expansion = Expansion::of(&self.model, key, covers, over);
         match (expansion.spans.is_empty(), agg) {
-            (true, Aggregation::Exact) => self.model.data.get(key).cloned(),
+            (true, Aggregation::Exact) => self.model.source().value_at(key),
             (true, Aggregation::OverHidden) => self.fold(std::slice::from_ref(key), func),
             (false, Aggregation::Exact) => {
                 let members = expansion.members();
-                let cells = members.iter().filter_map(|m| self.model.data.get(m));
+                let cells: Vec<CellValue> = members
-                fold_groups(&[&Group::of(cells)], func)
+                    .iter()
+                    .filter_map(|m| self.model.source().value_at(m))
+                    .collect();
+                fold_groups(&[&Group::of(&cells)], func)
             }
             (false, Aggregation::OverHidden) => self.fold(&expansion.members(), func),
         }
     }
 
     fn fold(&self, keys: &[CellKey], func: Reduce) -> Option<CellValue> {
-        self.shapes
+        self.shapes.read(self.model.source(), keys, |groups| {
-            .read(&self.model.data, keys, |groups| fold_groups(groups, func))
+            fold_groups(groups, func)
+        })
     }
 
     /// The items a stored cell may hold in `cat` to be named by `item`: that item, or a member
@@ -134,7 +178,7 @@ impl Incremental {
     fn cells_at(&self, at: &CellKey) -> i64 {
         let keys = std::slice::from_ref(at);
         let cells = |groups: &[&Group]| groups.first().map_or(0, |g| g.cells());
-        self.shapes.read(&self.model.data, keys, cells)
+        self.shapes.read(self.model.source(), keys, cells)
     }
 
     /// Cells at or under `at` that draw no hidden coordinate: all of them, less those with a
@@ -190,7 +234,7 @@ impl Incremental {
     }
 
     /// Brings the formula cells up to date with an edit and the store changes it made.
-    fn refresh_formulas(&mut self, edit: &Edit, changes: &[CellChange]) {
+    fn refresh_formulas(&mut self, edit: Option<&Edit>, changes: &[CellChange]) {
         if self.fingerprint != self.fingerprint() {
             return self.rebuild_formulas();
         }
@@ -203,7 +247,7 @@ impl Incremental {
             self.clear_stem(stem);
         }
         let mut dirty = moved.dirty;
-        if let Edit::SetMeasureAgg { measure, .. } = edit {
+        if let Some(Edit::SetMeasureAgg { measure, .. }) = edit {
             dirty.extend(self.stems.carrying(measure));
             if self.stems.reads_name(measure) {
                 dirty.extend(self.stems.live().cloned());
@@ -342,9 +386,13 @@ impl Derived for Incremental {
 
 impl Maintained for Incremental {
     fn apply(&mut self, edit: &Edit) {
+        let refused = self.feed.is_some() && edit.key().is_some();
+        assert!(
+            !refused,
+            "a fed backend is read-only: cell edits go through the source"
+        );
         edit.apply(&mut self.model);
-        let changes = self.model.data.drain_changes();
+        let changes = self.drain();
-        self.shapes.apply(&cell_delta(&changes), &changes);
+        self.absorb(Some(edit), &changes);
-        self.refresh_formulas(edit, &changes);
     }
 }

crates/improvise-incremental/src/shapes.rs

@@ -3,7 +3,8 @@
 
 use crate::group::Group;
 use crate::{CellDelta, DeltaValue};
-use improvise_core::model::cell::{CellChange, CellKey, CellValue, DataStore};
+use improvise_core::model::cell::{CellChange, CellKey, CellValue};
+use improvise_core::model::source::CellSource;
 use std::cell::{Cell, RefCell};
 use std::collections::BTreeMap;
 
@@ -111,28 +112,28 @@ impl Shape {
         (updates, sums)
     }
 
-    /// The shape of `store`'s cells, one pass.
+    /// The shape of `source`'s cells, one pass.
-    fn of(store: &DataStore, cats: &[String]) -> (Shape, u64) {
+    fn of(source: &dyn CellSource, cats: &[String]) -> (Shape, u64) {
         let mut shape = Shape {
             groups: BTreeMap::new(),
         };
         let mut visited = 0;
-        for (key, value) in store.iter_cells() {
+        for (key, value) in source.cells(&[]) {
             visited += 1;
             let Some(at) = project(&key, cats) else {
                 continue;
             };
             let group = shape.groups.entry(at).or_default();
-            group.apply(&DeltaValue::new(value), 1);
+            group.apply(&DeltaValue::new(&value), 1);
             if let CellValue::Number(x) = value {
-                group.add_number(*x, 1);
+                group.add_number(x, 1);
             }
         }
         (shape, visited)
     }
 }
 
-/// Every shape a read has asked for. A shape is built from the store on its first read (a
+/// Every shape a read has asked for. A shape is built from the source on its first read (a
 /// full pass) and fed every delta after.
 #[derive(Default)]
 pub struct Shapes {
@@ -177,7 +178,7 @@ impl Shapes {
     /// shape first if no read has asked for it. A key with no group is left out.
     pub fn read<R>(
         &self,
-        store: &DataStore,
+        source: &dyn CellSource,
         keys: &[CellKey],
         read: impl FnOnce(&[&Group]) -> R,
     ) -> R {
@@ -192,7 +193,7 @@ impl Shapes {
         );
         let mut built = self.built.borrow_mut();
         let shape = built.entry(cats.clone()).or_insert_with(|| {
-            let (shape, visited) = Shape::of(store, &cats);
+            let (shape, visited) = Shape::of(source, &cats);
             self.bump(|m| {
                 m.builds += 1;
                 m.built_cells += visited;

crates/improvise-incremental/tests/live_source.rs

@@ -0,0 +1,284 @@
+//! `Incremental` over a source that reports its own changes (ADR 009 step 7, improvise-rnak).
+//! Evidence: `context/test-evidence/improvise-incremental.md`. Counts, never time.
+
+mod support;
+
+use improvise_core::formula::ast::AggFunc;
+use improvise_core::model::cell::{CellKey, CellValue};
+use improvise_core::model::source::{ChangeSource, SharedFeed};
+use improvise_core::model::{Aggregation, Model};
+use improvise_incremental::{
+    Derived, Eager, Edit, Incremental, Maintained, Meter, differential_in, edit_probes, probes,
+    same_value,
+};
+use std::sync::Arc;
+use support::live::{Fed, LiveTable};
+use support::{corpus, edits_for, hand_model, key};
+
+#[test]
+fn a_live_source_matches_eager_over_the_corpus_and_its_edits() {
+    let (mut compared, mut files_done) = (0, Vec::<String>::new());
+    for (name, model, none, views) in corpus() {
+        let file = name.split(':').next().unwrap().to_string();
+        if files_done.contains(&file) {
+            continue;
+        }
+        files_done.push(file);
+        let mut edits = edits_for(&model);
+        if model.data.iter_cells().count() > 40 {
+            let picked = [0, 2, 4].iter().filter_map(|&i| edits.get(i).cloned());
+            edits = picked.collect();
+        }
+        let candidate = Fed::new(&model, none.clone());
+        let ran = differential_in(&model, &views, &none, candidate, &edits);
+        let report = ran.unwrap_or_else(|d| panic!("{name}: {d}"));
+        assert_eq!(report.steps, edits.len() + 1, "{name}");
+        compared += report.probes;
+    }
+    assert!(compared > 10_000, "only {compared} answers were compared");
+}
+
+#[test]
+fn changes_named_by_the_caller_are_maintained_as_the_drain_would() {
+    let (model, none) = (hand_model(), vec!["Region".to_string()]);
+    let mut eager = Eager::new(model.clone(), none.clone());
+    let mut fed = Fed::new(&model, none);
+    let edits = [
+        Edit::SetCell(key("East", "A", "Revenue"), CellValue::Number(99.0)),
+        Edit::SetCell(key("North", "A", "Revenue"), CellValue::Number(1.0)),
+        Edit::ClearCell(key("West", "B", "Cost")),
+        Edit::SetCell(key("East", "A", "Note"), CellValue::Text("z".into())),
+    ];
+    let mut questions = probes(eager.model());
+    let mut asked = 0;
+    for edit in &edits {
+        eager.apply(edit);
+        let (cell, value) = match edit {
+            Edit::SetCell(cell, value) => (cell, Some(value.clone())),
+            Edit::ClearCell(cell) => (cell, None),
+            _ => unreachable!("cell edits only"),
+        };
+        fed.live.write(cell.clone(), value);
+        let changes = fed.live.drain_changes();
+        assert_eq!(changes.len(), 1);
+        fed.inc.apply_changes(&changes);
+        questions.extend(probes(eager.model()));
+        questions.extend(edit_probes(edit));
+        for probe in &questions {
+            let (want, got) = (probe.ask(&eager), probe.ask(&fed));
+            assert!(
+                same_value(&want, &got),
+                "{probe:?}: eager {want:?} != fed {got:?}"
+            );
+            asked += 1;
+        }
+    }
+    assert!(asked > 500, "only {asked} answers were compared");
+}
+
+#[test]
+fn a_write_is_invisible_until_it_is_drained() {
+    let model = hand_model();
+    let mut fed = Fed::new(&model, vec![]);
+    let cell = key("East", "A", "Revenue");
+    let before = fed.inc.value(&cell, Aggregation::Exact);
+    assert_eq!(before, Some(CellValue::Number(10.0)));
+    fed.live.write(cell.clone(), Some(CellValue::Number(50.0)));
+    assert_eq!(
+        fed.inc.value(&cell, Aggregation::Exact),
+        before,
+        "not yet drained"
+    );
+    fed.inc.sync();
+    assert_eq!(
+        fed.inc.value(&cell, Aggregation::Exact),
+        Some(CellValue::Number(50.0))
+    );
+    fed.inc.sync();
+    assert_eq!(
+        fed.inc.value(&cell, Aggregation::Exact),
+        Some(CellValue::Number(50.0))
+    );
+}
+
+fn row(r: usize, p: usize, m: usize) -> CellKey {
+    CellKey::new(vec![
+        ("Region".into(), format!("r{r}")),
+        ("Product".into(), format!("p{p}")),
+        ("Month".into(), format!("m{m}")),
+        ("_Measure".into(), "Revenue".into()),
+    ])
+}
+
+/// `n` regions by `n` products by 6 months of one measure, with no formula.
+fn table(n: usize) -> Model {
+    let mut model = Model::new("table");
+    for (cat, prefix, count) in [("Region", "r", n), ("Product", "p", n), ("Month", "m", 6)] {
+        model.add_category(cat).unwrap();
+        for i in 0..count {
+            model
+                .category_mut(cat)
+                .unwrap()
+                .add_item(format!("{prefix}{i}"));
+        }
+    }
+    model.category_mut("_Measure").unwrap().add_item("Revenue");
+    model.measure_agg.insert("Revenue".into(), AggFunc::Max);
+    let mut seed = 1u64;
+    for (r, p, m) in (0..n).flat_map(|r| (0..n).flat_map(move |p| (0..6).map(move |m| (r, p, m)))) {
+        seed = seed
+            .wrapping_mul(6364136223846793005)
+            .wrapping_add(1442695040888963407);
+        model.set_cell(
+            row(r, p, m),
+            CellValue::Number(((seed >> 33) % 1000) as f64),
+        );
+    }
+    model
+}
+
+fn since(before: Meter, after: Meter) -> Meter {
+    Meter {
+        updates: after.updates - before.updates,
+        sum_updates: after.sum_updates - before.sum_updates,
+        reads: after.reads - before.reads,
+        builds: after.builds - before.builds,
+        built_cells: after.built_cells - before.built_cells,
+        stems: after.stems - before.stems,
+        formula_cells: after.formula_cells - before.formula_cells,
+        rebuilds: after.rebuilds - before.rebuilds,
+        recomputes: after.recomputes - before.recomputes,
+    }
+}
+
+fn region_partial() -> CellKey {
+    CellKey::new(vec![
+        ("Region".into(), "r1".into()),
+        ("_Measure".into(), "Revenue".into()),
+    ])
+}
+
+/// What one reported change costs once a shape is built, and what the source served for it.
+fn change_cost(n: usize) -> (Meter, u64) {
+    let mut fed = Fed::new(&table(n), vec![]);
+    fed.inc.raw_fold(&region_partial());
+    let (before, served) = (fed.inc.meter(), fed.live.snapshot_rows());
+    assert_eq!(
+        served,
+        (n * n * 6) as u64,
+        "the build is one pass over the table"
+    );
+    fed.live
+        .write(row(1, 2, 3), Some(CellValue::Number(2000.0)));
+    fed.inc.sync();
+    let cost = since(before, fed.inc.meter());
+    assert_eq!(
+        fed.inc.raw_fold(&region_partial()),
+        Some(CellValue::Number(2000.0)),
+        "the new maximum comes from the change"
+    );
+    (cost, fed.live.snapshot_rows() - served)
+}
+
+#[test]
+fn the_cost_of_a_reported_change_does_not_depend_on_the_size_of_the_table() {
+    let (small, small_rows) = change_cost(6);
+    let (large, large_rows) = change_cost(12);
+    assert_eq!(
+        (small.updates, small.sum_updates),
+        (2, 2),
+        "retract and insert"
+    );
+    assert_eq!(small, large, "216 cells and 864 cells cost the same");
+    assert_eq!(
+        (small.builds, small.built_cells),
+        (0, 0),
+        "no shape is rebuilt"
+    );
+    assert_eq!(
+        (small_rows, large_rows),
+        (0, 0),
+        "the source serves no snapshot for a change"
+    );
+}
+
+#[test]
+fn an_overwrite_by_the_same_value_moves_nothing() {
+    let mut fed = Fed::new(&hand_model(), vec![]);
+    let cell = key("East", "A", "Revenue");
+    let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]);
+    let before = fed.inc.raw_fold(&revenue);
+    fed.live.write(cell, Some(CellValue::Number(10.0)));
+    fed.inc.sync();
+    assert_eq!(fed.inc.raw_fold(&revenue), before);
+    assert_eq!(before, Some(CellValue::Number(23.0)));
+}
+
+#[test]
+fn a_backend_can_be_built_over_the_exported_feed_alias() {
+    let model = hand_model();
+    let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone()));
+    let feed: SharedFeed = Arc::new(LiveTable::of(rows));
+    let mut inc = Incremental::over(model.clone(), feed, vec![]);
+    let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]);
+    assert_eq!(inc.raw_fold(&revenue), Some(CellValue::Number(23.0)));
+    inc.sync();
+}
+
+#[test]
+fn a_populated_model_put_over_a_feed_keeps_no_cells_of_its_own() {
+    let (model, none) = (hand_model(), vec!["Region".to_string()]);
+    assert!(
+        model.data.iter_cells().count() > 0,
+        "the input is populated"
+    );
+    let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone()));
+    let live = Arc::new(LiveTable::of(rows));
+    let inc = Incremental::over(model.clone(), live, none.clone());
+    assert_eq!(inc.model().data.iter_cells().count(), 0, "no stale copy");
+    assert_eq!(inc.model().get_cell(&key("East", "A", "Revenue")), None);
+    let eager = Eager::new(model, none);
+    for probe in probes(eager.model()) {
+        let (want, got) = (probe.ask(&eager), probe.ask(&inc));
+        assert!(
+            same_value(&want, &got),
+            "{probe:?}: eager {want:?} != fed {got:?}"
+        );
+    }
+}
+
+#[test]
+#[should_panic(expected = "cell edits go through the source")]
+fn a_cell_edit_on_a_fed_backend_is_refused() {
+    let mut fed = Fed::new(&hand_model(), vec![]);
+    let edit = Edit::SetCell(key("East", "A", "Revenue"), CellValue::Number(1.0));
+    fed.inc.apply(&edit);
+}
+
+#[test]
+#[should_panic(expected = "cell edits go through the source")]
+fn a_cell_clear_on_a_fed_backend_is_refused() {
+    let mut fed = Fed::new(&hand_model(), vec![]);
+    fed.inc.apply(&Edit::ClearCell(key("East", "A", "Revenue")));
+}
+
+#[test]
+fn a_feed_has_one_drainer() {
+    let model = hand_model();
+    let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone()));
+    let live = Arc::new(LiveTable::of(rows));
+    let mut a = Incremental::over(model.clone(), live.clone(), vec![]);
+    let mut b = Incremental::over(model, live.clone(), vec![]);
+    let revenue = CellKey::new(vec![("_Measure".into(), "Revenue".into())]);
+    let (before_a, before_b) = (a.raw_fold(&revenue), b.raw_fold(&revenue));
+    assert_eq!(before_a, before_b);
+    live.write(key("East", "A", "Revenue"), Some(CellValue::Number(100.0)));
+    a.sync();
+    b.sync();
+    let (after_a, after_b) = (a.raw_fold(&revenue), b.raw_fold(&revenue));
+    assert_ne!(after_a, before_a, "the first drainer saw the write");
+    assert_eq!(
+        after_b, before_b,
+        "the second saw nothing: a drain is destructive"
+    );
+}

crates/improvise-incremental/tests/random_live.rs

@@ -0,0 +1,22 @@
+//! `Incremental` over a live source against `Eager` over random formula and cell edits.
+//! Evidence: `context/test-evidence/improvise-incremental.md`.
+
+mod support;
+
+use improvise_incremental::differential_in;
+use proptest::prelude::*;
+use support::live::Fed;
+use support::{any_formula_edit, formula_workbook};
+
+proptest! {
+    #[test]
+    fn a_live_source_matches_eager_over_random_formula_and_cell_edits(
+        edits in prop::collection::vec(any_formula_edit(), 0..3),
+    ) {
+        let wb = formula_workbook();
+        let none = vec!["Region".to_string()];
+        let candidate = Fed::new(&wb.model, none.clone());
+        let ran = differential_in(&wb.model, &[], &none, candidate, &edits);
+        prop_assert!(ran.is_ok(), "{}", ran.unwrap_err());
+    }
+}

crates/improvise-incremental/tests/support/live.rs

@@ -0,0 +1,156 @@
+//! A live table: a scripted source that reports its changes, the way a database would.
+//! Its rows are those as of the last drain; a write is pending until then.
+#![allow(dead_code)]
+
+use improvise_core::model::cell::{CellChange, CellKey, CellValue, DataStore};
+use improvise_core::model::source::{CellSource, ChangeSource, Select, Summary};
+use improvise_core::model::{Aggregation, Model};
+use improvise_incremental::{Derived, Edit, Incremental, Maintained, Reading};
+use std::sync::{Arc, Mutex};
+
+#[derive(Default)]
+struct State {
+    /// What a read sees: the rows as of the last drain.
+    visible: DataStore,
+    /// The rows with every pending write applied, to say what a write changed.
+    latest: DataStore,
+    pending: Vec<CellChange>,
+    /// Rows handed out by a snapshot read (`cells` with no partial).
+    snapshot_rows: u64,
+}
+
+#[derive(Default)]
+pub struct LiveTable {
+    state: Mutex<State>,
+}
+
+impl LiveTable {
+    /// A table already holding `rows`, with nothing pending.
+    pub fn of(rows: impl IntoIterator<Item = (CellKey, CellValue)>) -> Self {
+        let mut state = State::default();
+        for (key, value) in rows {
+            state.visible.set(key.clone(), value.clone());
+            state.latest.set(key, value);
+        }
+        LiveTable {
+            state: Mutex::new(state),
+        }
+    }
+
+    /// Writes `value` at `key` (or removes it), as a database user would: visible after the next drain.
+    pub fn write(&self, key: CellKey, value: Option<CellValue>) {
+        let mut state = self.state.lock().unwrap();
+        let old = state.latest.get(&key).cloned();
+        match &value {
+            Some(v) => state.latest.set(key.clone(), v.clone()),
+            None if old.is_none() => return,
+            None => state.latest.remove(&key),
+        }
+        state.pending.push(CellChange {
+            key,
+            old,
+            new: value,
+        });
+    }
+
+    /// How many rows snapshot reads have handed out so far.
+    pub fn snapshot_rows(&self) -> u64 {
+        self.state.lock().unwrap().snapshot_rows
+    }
+}
+
+impl CellSource for LiveTable {
+    fn value_at(&self, key: &CellKey) -> Option<CellValue> {
+        self.state.lock().unwrap().visible.value_at(key)
+    }
+
+    fn summarize(&self, select: &Select) -> Summary {
+        self.state.lock().unwrap().visible.summarize(select)
+    }
+
+    fn keys<'a>(
+        &'a self,
+        partial: &'a [(String, String)],
+    ) -> Box<dyn Iterator<Item = CellKey> + 'a> {
+        let state = self.state.lock().unwrap();
+        let keys: Vec<CellKey> = state.visible.keys(partial).collect();
+        Box::new(keys.into_iter())
+    }
+
+    fn cells<'a>(
+        &'a self,
+        partial: &'a [(String, String)],
+    ) -> Box<dyn Iterator<Item = (CellKey, CellValue)> + 'a> {
+        let mut state = self.state.lock().unwrap();
+        let cells: Vec<_> = state.visible.cells(partial).collect();
+        if partial.is_empty() {
+            state.snapshot_rows += cells.len() as u64;
+        }
+        Box::new(cells.into_iter())
+    }
+}
+
+impl ChangeSource for LiveTable {
+    fn drain_changes(&self) -> Vec<CellChange> {
+        let mut state = self.state.lock().unwrap();
+        let changes = std::mem::take(&mut state.pending);
+        for change in &changes {
+            match &change.new {
+                Some(value) => state.visible.set(change.key.clone(), value.clone()),
+                None => state.visible.remove(&change.key),
+            }
+        }
+        changes
+    }
+}
+
+/// An `Incremental` over a live table that holds the model's cells; the model keeps no store.
+/// A cell edit is a write to the table, the way another client's would be, followed by a drain.
+pub struct Fed {
+    pub inc: Incremental,
+    pub live: Arc<LiveTable>,
+}
+
+impl Fed {
+    pub fn new(model: &Model, none_cats: Vec<String>) -> Fed {
+        let rows = model.data.iter_cells().map(|(k, v)| (k, v.clone()));
+        let live = Arc::new(LiveTable::of(rows));
+        let mut bare = model.clone();
+        bare.data = DataStore::new();
+        let inc = Incremental::over(bare, live.clone(), none_cats);
+        Fed { inc, live }
+    }
+}
+
+impl Derived for Fed {
+    fn value(&self, key: &CellKey, agg: Aggregation) -> Option<CellValue> {
+        self.inc.value(key, agg)
+    }
+
+    fn raw_fold(&self, partial: &CellKey) -> Option<CellValue> {
+        self.inc.raw_fold(partial)
+    }
+
+    fn covered(&self, key: &CellKey, reading: &Reading) -> Option<CellValue> {
+        self.inc.covered(key, reading)
+    }
+
+    fn total(&self, key: &CellKey, over: &[String], reading: &Reading) -> Option<CellValue> {
+        self.inc.total(key, over, reading)
+    }
+
+    fn occupancy(&self, partial: &CellKey, reading: &Reading) -> usize {
+        self.inc.occupancy(partial, reading)
+    }
+}
+
+impl Maintained for Fed {
+    fn apply(&mut self, edit: &Edit) {
+        match edit {
+            Edit::SetCell(key, value) => self.live.write(key.clone(), Some(value.clone())),
+            Edit::ClearCell(key) => self.live.write(key.clone(), None),
+            _ => return self.inc.apply(edit),
+        }
+        self.inc.sync();
+    }
+}

crates/improvise-incremental/tests/support/mod.rs

@@ -1,6 +1,8 @@
 //! Fixtures the integration tests share: the shipped corpus and the hand-built models.
 #![allow(dead_code)]
 
+pub mod live;
+
 use improvise_core::formula::ast::AggFunc;
 use improvise_core::formula::parse_formula;
 use improvise_core::model::Aggregation;