fix(incremental): over takes the exported feed alias, drops the model's cells and refuses cell edits in every build (improvise-rnak)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

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

@@ -192,8 +192,8 @@ 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 after the changes it has drained,
-/// so a snapshot read now plus the changes drained later counts no change twice or not at all.
+/// 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>;

crates/improvise-incremental/src/incremental.rs

@@ -9,11 +9,10 @@ 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::{ChangeSource, SharedFeed};
+use improvise_core::model::source::{SharedFeed, SharedSource};
 use improvise_core::model::types::{Covers, settled};
 use improvise_core::model::{Aggregation, Model};
 use std::collections::BTreeMap;
-use std::sync::Arc;
 
 /// How a stem's formulas are evaluated: in dependency order where one pass is exact, or by
 /// eager's passes always. The two must agree; `Plain` is the reference.
@@ -47,12 +46,9 @@ impl Incremental {
 
     /// 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<S: ChangeSource + Send + Sync + 'static>(
-        model: Model,
-        feed: Arc<S>,
-        none_cats: Vec<String>,
-    ) -> Self {
-        let model = model.over(feed.clone());
+    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))
     }
 
@@ -390,6 +386,11 @@ 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.drain();
         self.absorb(Some(edit), &changes);

crates/improvise-incremental/tests/live_source.rs

@@ -5,12 +5,14 @@ mod support;
 
 use improvise_core::formula::ast::AggFunc;
 use improvise_core::model::cell::{CellKey, CellValue};
-use improvise_core::model::source::ChangeSource;
+use improvise_core::model::source::{ChangeSource, SharedFeed};
 use improvise_core::model::{Aggregation, Model};
 use improvise_incremental::{
-    Derived, Eager, Edit, Maintained, Meter, differential_in, edit_probes, probes, same_value,
+    Derived, Eager, Edit, Incremental, Maintained, Meter, differential_in, edit_probes, probes,
+    same_value,
 };
-use support::live::Fed;
+use std::sync::Arc;
+use support::live::{Fed, LiveTable};
 use support::{corpus, edits_for, hand_model, key};
 
 #[test]
@@ -108,7 +110,7 @@ fn row(r: usize, p: usize, m: usize) -> CellKey {
     ])
 }
 
-/// `n` regions by `n` products by 6 months of one measure, as a model with a profit-style formula.
+/// `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)] {
@@ -211,3 +213,72 @@ fn an_overwrite_by_the_same_value_moves_nothing() {
     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"
+    );
+}