D-STO-1 (resident half): owner-stamped batches over resident lanes

.claude/board/LATEST_STATE.md

@@ -85,6 +85,34 @@
 - Quack became consumable by git with lance-graph #1457
   (`ISS-LG-MASKRISC-PATH-NDARRAY`, resolved).
 
+## Store, resident half (D-STO-1)
+
+- `hubspo-store::table::Table`: one dataset (tenant × object type) as
+  resident lanes (`U32` / `I32`), a live plane (plane 0) and one validity
+  plane per nullable lane. `planes()` borrows them for mask-risc; no read
+  copies a lane.
+- Writes are `Batch`es (inserts, updates, deletes), validated whole and
+  applied as exactly one version or not at all, and only for the table's
+  owner (`MailboxId`, write on behalf). Refused: wrong owner, wrong arity,
+  wrong kind, missing required value, unknown lane, a deleted or absent
+  row, and a row written twice in one batch (update + update of one lane,
+  update + delete, delete + delete; order inside a batch means nothing).
+- `apply()` returns `Applied { version, first_insert, changed }`; `changed`
+  is the plane `list::refresh` takes. Ordinals are never reused: inserts
+  append, deletes clear the live bit and leave the row inert.
+- Tests: 60 random batches match a plain model (every lane and plane, 200+
+  updates, 30+ deletes); 12 refusals leave version, lanes and planes
+  untouched (each refused batch also carries a valid insert, so a partial
+  apply would show); `hubspo-query/tests/store_list.rs` runs catalog →
+  filter → `list::refresh` on the store's own `changed` plane over 50
+  batches and equals a full recomputation.
+  Disable runs 2026-10-10, all red: owner unchecked, deletes missing from
+  `changed` (red in the store and in the list test), delete keeps `live`,
+  a cleared value keeps validity, version frozen, conflict check off,
+  updates to deleted rows allowed, lane kind unchecked.
+- Not yet: persistence (one Lance version per published batch, P-SPARSE-1
+  base + delta) — the next half of D-STO-1.
+
 ## Board view (D-QRY-3, board half)
 
 - `hubspo-query::board::Board`: per (pipeline, stage), record count and

.claude/board/STATUS_BOARD.md

@@ -35,7 +35,7 @@
 
 | D-id | Deliverable | Size | Status | Evidence |
 |---|---|---|---|---|
-| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | Queued |  |
+| D-STO-1 | `hubspo-store` write lane: batch image → one Lance version per publish; owner-stamped (`mailbox_owner`) | L | In progress | resident half in PR: `hubspo-store::table` (owner-stamped atomic batches, one version each, `changed` plane); Lance persistence next |
 | D-STO-2 | KV value store for variable-length properties; zero-copy lens reads | M | Queued |  |
 | D-STO-3 | Edge lanes + association labels (H-ADR-4); cardinality invariants as Quack `GroupReduce` | L | Queued |  |
 | D-STO-4 | Tenant layout + P-TEN-1 probe | M | Queued | P-TEN-1 passed 2026-10-10 (7.8 ms p99 at 10k tenants, local disk) |
@@ -48,7 +48,7 @@
 |---|---|---|---|---|
 | D-QRY-1 | Filter compiler: HubSpot filter groups (AND within group, OR across groups; operators EQ/NEQ/LT/…/HAS_PROPERTY/CONTAINS_TOKEN/IN/BETWEEN) → Quack program | L | Shipped | `hubspo-store::catalog` + `hubspo-query::filter`; 600 random filter sets equal a plain-loop oracle on mask-risc; #19 |
 | D-QRY-2 | Lists: active (incremental on version delta) + static (stored mask); membership history | L | Shipped | `hubspo-query::list` (refresh, static add/remove/members, added/removed); tests vs full recomputation. P-LIST-1 2.6 ms at 10M; P-SPARSE-1 1.93 ms, 257 KiB flat at 10M. Storage of list masks lands with D-STO-1 |
-| D-QRY-3 | Saved views + board view (report pivot over pipeline × stage) | M | In progress | board half in PR: `hubspo-query::board` (count + amount per stage as two `GroupReduce` programs); saved views + sorting open |
+| D-QRY-3 | Saved views + board view (report pivot over pipeline × stage) | M | In progress | board half shipped (#21): `hubspo-query::board` (count + amount per stage as two `GroupReduce` programs); saved views + sorting open |
 | D-QRY-4 | Quack lowering of callcenter's `ParsedQuery` (the upstream DataFusion `parsed_query_to_plan` is an unimplemented stub) | M | Queued |  |
 | D-QRY-5 | Association-traversing filters ("contacts whose company has an open deal"): Quack `Semijoin` over edge lanes for one hop; multi-hop needs cypher-quack shapes that do not exist yet → upstream request first | L | Queued |  |
 | D-QRY-6 | Reports: single-object, cross-object, funnel, time-series, attribution skeleton on `lance-graph-report` | XL | Queued |  |

crates/hubspo-query/tests/store_list.rs

@@ -0,0 +1,179 @@
+//! End to end on a real `Table`: batches applied to the store feed the
+//! filter compiler and the active-list refresh, and the refreshed list
+//! equals a full recomputation after every batch.
+
+use hubspo_query::filter::{FilterGroup, Operator, PropertyFilter, compile};
+use hubspo_query::list;
+use hubspo_store::catalog::{Catalog, Codebook, FieldSpec};
+use hubspo_store::table::{Batch, LaneKind, LaneSpec, Table, Value};
+use lance_graph_mask_risc::{
+    Foreign, Out, Planes, Scratch, execute_into, materialize_rows, words_for,
+};
+use lance_graph_quack::bind::{FieldKind, create_table};
+use lance_graph_quack::{Agg, Col, Filter, Mask, Query, lower};
+
+const OWNER: u32 = 3;
+const STAGES: &[&str] = &["lead", "opportunity", "customer"];
+
+struct Rng(u64);
+impl Rng {
+    fn next(&mut self) -> u64 {
+        self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
+        let mut z = self.0;
+        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
+        z ^ (z >> 31)
+    }
+    fn below(&mut self, n: u64) -> u64 {
+        self.next() % n
+    }
+}
+
+/// Run `filter` over the table's planes plus two extra planes (stored list,
+/// changed rows) appended after the table's own.
+fn run(t: &Table, filter: &Filter, extra: [&[u64]; 2]) -> Vec<usize> {
+    let (mut masks, lanes) = t.planes();
+    masks.extend(extra);
+    let planes = Planes {
+        n_rows: t.n_rows(),
+        masks: &masks,
+        lanes: &lanes,
+    };
+    let program = lower(&Query {
+        filter: filter.clone(),
+        agg: Agg::Rows,
+    })
+    .expect("lower");
+    let mut out = vec![0u64; words_for(t.n_rows())];
+    let mut scratch = Scratch::for_program(&program, t.n_rows()).expect("scratch");
+    execute_into(
+        &program,
+        &planes,
+        &Foreign::NONE,
+        &mut scratch,
+        Out::Mask(&mut out),
+    )
+    .expect("execute");
+    materialize_rows(&out, t.n_rows())
+}
+
+fn bits(n: usize, rows: &[usize]) -> Vec<u64> {
+    let mut w = vec![0u64; words_for(n)];
+    for &r in rows {
+        w[r / 64] |= 1 << (r % 64);
+    }
+    w
+}
+
+#[test]
+fn a_list_on_the_store_follows_its_batches() {
+    let mut t = Table::new(
+        OWNER,
+        &[
+            LaneSpec {
+                kind: LaneKind::U32,
+                nullable: false,
+            },
+            LaneSpec {
+                kind: LaneKind::I32,
+                nullable: true,
+            },
+        ],
+    )
+    .expect("table");
+    let mut cat = Catalog::new();
+    let id = create_table("deals")
+        .width(8)
+        .register(&mut cat)
+        .expect("register")
+        .id;
+    cat.declare(
+        id,
+        FieldSpec {
+            name: "dealstage".into(),
+            col: Col(0),
+            kind: FieldKind::Code,
+            validity: t.validity(Col(0)),
+            codebook: Some(Codebook::new(STAGES.iter().copied()).expect("codebook")),
+        },
+    )
+    .expect("stage");
+    cat.declare(
+        id,
+        FieldSpec {
+            name: "amount".into(),
+            col: Col(1),
+            kind: FieldKind::I32,
+            validity: t.validity(Col(1)),
+            codebook: None,
+        },
+    )
+    .expect("amount");
+    // The table owns planes 0 (live) and 1 (amount validity); the list and
+    // the changed rows go after them.
+    let (prev, changed) = (Mask(2), Mask(3));
+    let pred = compile(
+        "deals",
+        &[FilterGroup {
+            filters: vec![
+                PropertyFilter::new("dealstage", Operator::Neq, "lead"),
+                PropertyFilter::new("amount", Operator::Gt, "5000"),
+            ],
+        }],
+        &cat,
+    )
+    .expect("compile");
+
+    let mut r = Rng(0x0005_7011);
+    let mut stored: Vec<usize> = Vec::new();
+    let (mut joined, mut left) = (0usize, 0usize);
+    for round in 0..50 {
+        let mut b = Batch {
+            owner: OWNER,
+            ..Batch::default()
+        };
+        for _ in 0..r.below(30) {
+            b.inserts.push(vec![
+                Some(Value::U32(r.below(3) as u32)),
+                (r.below(4) > 0).then(|| Value::I32(r.below(10_000) as i32)),
+            ]);
+        }
+        let mut touched = std::collections::BTreeSet::new();
+        for _ in 0..r.below(25) {
+            if t.n_rows() == 0 {
+                break;
+            }
+            let row = r.below(t.n_rows() as u64) as u32;
+            if !t.is_live(row) || !touched.insert(row) {
+                continue;
+            }
+            match r.below(4) {
+                0 => b.deletes.push(row),
+                1 => b
+                    .updates
+                    .push((row, Col(0), Some(Value::U32(r.below(3) as u32)))),
+                _ => b.updates.push((
+                    row,
+                    Col(1),
+                    (r.below(4) > 0).then(|| Value::I32(r.below(10_000) as i32)),
+                )),
+            }
+        }
+        let applied = t.apply(&b).expect("apply");
+        let n = t.n_rows();
+        let prev_bits = bits(n, &stored);
+        let next = run(
+            &t,
+            &list::refresh(&pred, prev, changed),
+            [&prev_bits, &applied.changed],
+        );
+        let full = run(&t, &pred, [&prev_bits, &applied.changed]);
+        assert_eq!(next, full, "round {round}");
+        let before: std::collections::BTreeSet<_> = stored.iter().copied().collect();
+        let after: std::collections::BTreeSet<_> = next.iter().copied().collect();
+        joined += after.difference(&before).count();
+        left += before.difference(&after).count();
+        stored = next;
+    }
+    assert!(joined > 50 && left > 20, "joined {joined}, left {left}");
+}

crates/hubspo-store/Cargo.toml

@@ -13,3 +13,5 @@ workspace = true
 
 [dependencies]
 lance-graph-quack.workspace = true
+lance-graph-mask-risc.workspace = true
+lance-graph-contract.workspace = true

crates/hubspo-store/src/lib.rs

@@ -4,7 +4,10 @@
 //!
 //! - Zone: zone 2 (no serde, no wire formats).
 //! - Deliverables: D-STO-1..6 (`.claude/plans/hubspo-integration-v1.md`).
-//! - First module: [`catalog`], the table and field catalog every Quack read
-//!   binds through (H-ADR-6). Storage lands with D-STO-1.
+//! - [`catalog`]: the table and field catalog every Quack read binds
+//!   through (H-ADR-6).
+//! - [`table`] (D-STO-1, resident half): one dataset's resident lanes, changed
+//!   only by whole owner-stamped batches, one version each.
 
 pub mod catalog;
+pub mod table;