diff --git a/interactive/examples/ddir.rs b/interactive/examples/ddir.rs
index a706bd804..576c176be 100644
--- a/interactive/examples/ddir.rs
+++ b/interactive/examples/ddir.rs
@@ -14,6 +14,9 @@
 //! - `--diag`: serve timely/DD diagnostics on port 51371.
 //! - `--backend=vec|corgi`: rendering substrate (default `vec`). The corgi
 //!   backend is single-worker (its arrange does not exchange).
+//! - `--sync=K`: await completion only every K rounds (default 1), letting K
+//!   timestamps retire with whatever inter-timestamp concurrency the system
+//!   finds — the open(er)-loop regime DD adapts into under load.
 
 use mimalloc::MiMalloc;
 
@@ -37,6 +40,7 @@ struct Flags {
     debug_demand: bool,
     diag: bool,
     corgi: bool,
+    sync: u64,
 }
 
 fn run(
@@ -184,7 +188,10 @@ fn run(
                 cursor += 1;
             }
             for i in inputs.iter_mut() { i.advance_to(time); i.flush(); }
-            while probe.less_than(&time) { worker.step(); }
+            let sync = flags.sync.max(1);
+            if (round + 1) % sync == 0 || round + 1 == limit {
+                while probe.less_than(&time) { worker.step(); }
+            }
 
             round += 1;
             if round % 100 == 0 {
@@ -209,6 +216,7 @@ fn main() {
             else if let Some(q) = a.strip_prefix("--query=") { flags.query = Some(q.to_string()); }
             else if a == "--debug-demand" { flags.debug_demand = true; }
             else if a == "--diag" { flags.diag = true; }
+            else if let Some(k) = a.strip_prefix("--sync=") { flags.sync = k.parse().expect("--sync=K"); }
             else if let Some(b) = a.strip_prefix("--backend=") {
                 flags.corgi = match b { "corgi" => true, "vec" => false, other => panic!("unknown backend {other:?} (vec|corgi)") };
             }
diff --git a/interactive/examples/ddir_col.rs.disabled b/interactive/examples/ddir_col.rs.disabled
deleted file mode 100644
index de4162508..000000000
--- a/interactive/examples/ddir_col.rs.disabled
+++ /dev/null
@@ -1,138 +0,0 @@
-//! DD IR columnar driver: parse, lower, render (via `interactive::backend::col`), execute.
-
-use mimalloc::MiMalloc;
-
-#[global_allocator]
-static GLOBAL: MiMalloc = MiMalloc;
-
-use differential_dataflow::dynamic::pointstamp::PointStamp;
-
-use interactive::parse;
-use interactive::lower;
-use interactive::backend::col::{render_tree, Row, Diff};
-
-type DdirOuterUpdate = (Row, Row, u64, Diff);
-
-fn run(name: &str, stmts: Vec<parse::Stmt>, n_inputs: usize, nodes: u64, edges: u64, arity: usize, batch: u64, rounds: Option<u64>) {
-    let mut tree = lower::lower_tree(stmts);
-    let ops_before = tree.op_count();
-    tree.optimize();
-    let tree_export_idx = tree.root.exports.iter().position(|e| e.name == "result").unwrap_or(0);
-    println!("{}: {} ops before optimize, {} after; driving export {:?}",
-        name, ops_before, tree.op_count(), tree.root.exports[tree_export_idx].name);
-    let name = name.to_string();
-
-    timely::execute_from_args(std::env::args().skip(4), move |worker| {
-        use timely::dataflow::InputHandle;
-        use timely::container::PushInto;
-        use differential_dataflow::columnar::ValColBuilder;
-
-        type OuterBuilder = ValColBuilder<DdirOuterUpdate>;
-
-        let (mut inputs, probe) = worker.dataflow::<u64, _, _>(|scope| {
-            let mut handles = Vec::new();
-            let mut collections = Vec::new();
-            for _ in 0..n_inputs {
-                let mut h = <InputHandle<_, OuterBuilder>>::new_with_builder();
-                let stream = h.to_stream(scope);
-                handles.push(h);
-                collections.push(differential_dataflow::Collection::new(stream));
-            }
-            let mut probe = timely::dataflow::ProbeHandle::new();
-            let output = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
-                let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect();
-                let root_imports: Vec<_> = tree.root.imports.iter().map(|imp| match &imp.from {
-                    interactive::scope_ir::Source::Input(n) => entered[*n].clone(),
-                    interactive::scope_ir::Source::Trace(name) => panic!("ddir_col: Import {:?} not supported in this harness (no trace registry).", name),
-                    interactive::scope_ir::Source::Parent(_) => unreachable!("root scope cannot import from a parent"),
-                }).collect();
-                let exports = render_tree(&tree.root, inner, 0, root_imports);
-                exports[tree_export_idx].clone().leave(scope)
-            });
-            output.probe_with(&mut probe);
-            (handles, probe)
-        });
-
-        let index = worker.index();
-        let peers = worker.peers();
-
-        let mut builders: Vec<OuterBuilder> = (0..n_inputs).map(|_| OuterBuilder::default()).collect();
-
-        let timer = std::time::Instant::now();
-        let timer_load = std::time::Instant::now();
-        for e in 0..edges {
-            if (e as usize) % peers == index {
-                let input_idx = (e as usize) % inputs.len();
-                let (key, val) = interactive::gen_row::<Row>(e, nodes, arity);
-                let time = *inputs[input_idx].time();
-                builders[input_idx].push_into((key, val, time, 1i64));
-            }
-        }
-        for (i, h) in inputs.iter_mut().enumerate() {
-            use timely::container::ContainerBuilder;
-            while let Some(container) = builders[i].finish() { h.send_batch(container); }
-            h.advance_to(1);
-            h.flush();
-        }
-        while probe.less_than(&1u64) { worker.step(); }
-        println!("worker {}: {} loaded ({} edges, total {:.2?}, load {:.2?})", index, name, edges, timer.elapsed(), timer_load.elapsed());
-
-        let mut cursor = 0u64;
-        let mut round = 0u64;
-        let limit = rounds.unwrap_or(u64::MAX);
-        while round < limit {
-            let timer_round = std::time::Instant::now();
-            let time = (round + 2) as u64;
-            for _ in 0..batch {
-                let remove_idx = cursor;
-                let add_idx = edges + cursor;
-                if (remove_idx as usize) % peers == index {
-                    let input_idx = (remove_idx as usize) % inputs.len();
-                    let (key, val) = interactive::gen_row::<Row>(remove_idx, nodes, arity);
-                    builders[input_idx].push_into((key, val, time, -1i64));
-                }
-                if (add_idx as usize) % peers == index {
-                    let input_idx = (add_idx as usize) % inputs.len();
-                    let (key, val) = interactive::gen_row::<Row>(add_idx, nodes, arity);
-                    builders[input_idx].push_into((key, val, time, 1i64));
-                }
-                cursor += 1;
-            }
-            for (i, h) in inputs.iter_mut().enumerate() {
-                use timely::container::ContainerBuilder;
-                while let Some(container) = builders[i].finish() { h.send_batch(container); }
-                h.advance_to(time);
-                h.flush();
-            }
-            while probe.less_than(&time) { worker.step(); }
-
-            round += 1;
-            if round % 100 == 0 {
-                println!("worker {}: {} round {} (total {:.2?}, round {:.2?})", index, name, round, timer.elapsed(), timer_round.elapsed());
-            }
-        }
-        println!("worker {}: {} done ({} rounds, batch {}, total {:.2?})", index, name, round, batch, timer.elapsed());
-    }).unwrap();
-}
-
-fn main() {
-    let program = std::env::args().nth(1).unwrap_or_else(|| { std::process::exit(0); });
-    let arity: usize = std::env::args().nth(2).unwrap_or("2".into()).parse().unwrap();
-    let nodes: u64 = std::env::args().nth(3).unwrap_or("10".into()).parse().unwrap();
-    let edges: u64 = std::env::args().nth(4).unwrap_or_else(|| (2 * nodes).to_string()).parse().unwrap();
-    let batch: u64 = std::env::args().nth(5).unwrap_or("1".into()).parse().unwrap();
-    let rounds: Option<u64> = std::env::args().nth(6).map(|s| s.parse().unwrap());
-
-    let source = interactive::load_program(&program);
-    let stmts = if program.ends_with(".ddp") {
-        parse::pipe::parse(&source)
-    } else {
-        parse::applicative::parse(&source)
-    };
-    let (n_inputs, imports) = interactive::survey_sources(&stmts);
-    if !imports.is_empty() {
-        panic!("ddir_col: program references imports {:?} but this harness has no trace registry.", imports);
-    }
-    let name = std::path::Path::new(&program).file_stem().map(|s| s.to_string_lossy().into_owned()).unwrap_or(program.clone());
-    run(&name, stmts, n_inputs, nodes, edges, arity, batch, rounds);
-}
diff --git a/interactive/src/backend/col.rs b/interactive/src/backend/col.rs
deleted file mode 100644
index 1948c55d1..000000000
--- a/interactive/src/backend/col.rs
+++ /dev/null
@@ -1,259 +0,0 @@
-//! Columnar rendering substrate.
-//!
-//! Rows are a stride-encoded `Row(Vec<i64>)`; the differential container is
-//! columnar `RecordedUpdates`. Supplies the substrate leaf operators (join,
-//! reduce, arrange over columnar builders/batchers/spines); the scope-tree walk
-//! lives in [`crate::backend::render_tree`].
-
-mod types {
-    /// A row type backed by Vec but using Strides for columnar bounds.
-    /// This ensures uniform-length rows (common in the IR) get compact
-    /// stride-based offset encoding rather than per-element u64 bounds.
-    #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
-    pub struct Row(pub Vec<i64>);
-
-    impl Row {
-        pub fn new() -> Self { Row(Vec::new()) }
-        pub fn push(&mut self, v: i64) { self.0.push(v); }
-    }
-
-    impl std::ops::Deref for Row {
-        type Target = [i64];
-        fn deref(&self) -> &[i64] { &self.0 }
-    }
-
-    impl std::iter::FromIterator<i64> for Row {
-        fn from_iter<I: IntoIterator<Item = i64>>(iter: I) -> Self {
-            Row(iter.into_iter().collect())
-        }
-    }
-
-    impl<'a> IntoIterator for &'a Row {
-        type Item = &'a i64;
-        type IntoIter = std::slice::Iter<'a, i64>;
-        fn into_iter(self) -> Self::IntoIter { self.0.iter() }
-    }
-
-    impl IntoIterator for Row {
-        type Item = i64;
-        type IntoIter = std::vec::IntoIter<i64>;
-        fn into_iter(self) -> Self::IntoIter { self.0.into_iter() }
-    }
-
-    impl columnar::Columnar for Row {
-        type Container = columnar::Vecs<Vec<i64>, columnar::primitive::offsets::Strides>;
-
-        fn into_owned<'a>(other: columnar::Ref<'a, Self>) -> Self {
-            Row(other.into_iter().copied().collect())
-        }
-
-        fn copy_from<'a>(&mut self, other: columnar::Ref<'a, Self>) {
-            self.0.clear();
-            self.0.extend(other.into_iter().copied());
-        }
-    }
-
-    impl crate::ir::RowLike for Row {
-        fn new() -> Self { Row::new() }
-        fn push(&mut self, v: i64) { Row::push(self, v); }
-        fn as_slice(&self) -> &[i64] { &self.0 }
-        fn extend_from_slice(&mut self, other: &[i64]) { self.0.extend_from_slice(other); }
-    }
-
-    pub type Diff = i64;
-    pub type Time = timely::order::Product<u64, differential_dataflow::dynamic::pointstamp::PointStamp<u64>>;
-}
-
-use differential_dataflow::columnar as columnar_support;
-
-mod columnar {
-    use super::types::*;
-
-    pub use super::columnar_support::*;
-    pub use super::columnar_support::{ValSpine, ValBatcher, ValBuilder};
-
-    pub type DdirUpdate = (Row, Row, Time, Diff);
-    pub type DdirRecordedUpdates = RecordedUpdates<DdirUpdate>;
-
-    pub type ColValSpine<K, V, T, R> = ValSpine<K, V, T, R>;
-    pub type ColValBatcher<K, V, T, R> = ValBatcher<K, V, T, R>;
-    pub type ColValBuilder<K, V, T, R> = ValBuilder<K, V, T, R>;
-    pub type ColValChunker<U> = ValChunker<U>;
-}
-
-mod render {
-    use std::sync::Arc;
-    use timely::order::Product;
-    use differential_dataflow::Collection;
-    use differential_dataflow::dynamic::pointstamp::PointStamp;
-    use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
-    use columnar::Columnar;
-    use super::types::*;
-    use crate::ir::{LinearOp, RowLike, eval_fields, eval_field_into, eval_condition};
-    use crate::parse::{Projection, Reducer};
-    use crate::backend::Backend;
-
-    use super::columnar::{DdirUpdate, DdirRecordedUpdates};
-    use super::columnar::{ColValSpine, ColValBuilder};
-
-    type ConcreteTime = Product<u64, PointStamp<u64>>;
-
-    pub type Col<'scope> = Collection<'scope, ConcreteTime, DdirRecordedUpdates>;
-    type Arr<'scope> = Arranged<'scope, TraceAgent<ColValSpine<Row, Row, ConcreteTime, Diff>>>;
-
-    /// Render a Linear chain: one pass applying the ops in sequence. `level` is
-    /// the op's scope depth — it locates the iteration coord for LiftIter and
-    /// the coordinate position EnterAt's delay lands in.
-    fn render_linear<'scope>(c: Col<'scope>, ops: Vec<LinearOp>, level: usize) -> Col<'scope> {
-        super::columnar::join_function(c, move |k, v, t_in, _d| {
-            use timely::progress::Timestamp;
-            let k: Row = Columnar::into_owned(k);
-            let v: Row = Columnar::into_owned(v);
-            // Materialize input time once so LiftIter can read
-            // the iter coord at the operator's scope depth.
-            let t_owned: Time = Columnar::into_owned(t_in);
-            let iter_at_level: i64 = level
-                .checked_sub(1)
-                .and_then(|idx| t_owned.inner.get(idx).copied())
-                .unwrap_or(0) as i64;
-            let mut results: Vec<(Row, Row, Time, Diff)> = vec![(k, v, Time::minimum(), 1i64)];
-            for op in &ops {
-                let mut next = Vec::new();
-                for (k, v, t, d) in results {
-                    match op {
-                        LinearOp::Project(proj) => {
-                            let i = [k.as_slice(), v.as_slice()];
-                            next.push((eval_fields(&proj.key, &i), eval_fields(&proj.val, &i), t, d));
-                        },
-                        LinearOp::Filter(cond) => {
-                            let i = [k.as_slice(), v.as_slice()];
-                            if eval_condition(cond, &i) { next.push((k, v, t, d)); }
-                        },
-                        LinearOp::Negate => {
-                            next.push((k, v, t, -d));
-                        },
-                        LinearOp::EnterAt(field) => {
-                            let delay = {
-                                let mut r = Row::new();
-                                eval_field_into(field, &[k.as_slice(), v.as_slice()], &mut r);
-                                256 * (64 - (r.as_slice().first().copied().unwrap_or(0) as u64).leading_zeros() as u64)
-                            };
-                            let mut coords = smallvec::SmallVec::<[u64; 1]>::new();
-                            for _ in 0..level.saturating_sub(1) { coords.push(0); }
-                            coords.push(delay);
-                            next.push((k, v, Product::new(0u64, PointStamp::new(coords)), d));
-                        },
-                        LinearOp::LiftIter => {
-                            let mut new_v = v.clone();
-                            new_v.push(iter_at_level);
-                            next.push((k, new_v, t, d));
-                        },
-                    }
-                }
-                results = next;
-            }
-            results.into_iter()
-        })
-    }
-
-    fn render_join<'scope>(l: Arr<'scope>, r: Arr<'scope>, projection: &Projection) -> Col<'scope> {
-        let proj = projection.clone();
-        use differential_dataflow::operators::join::join_traces;
-        use differential_dataflow::collection::AsCollection;
-        use super::columnar::ValColBuilder;
-        let stream = join_traces::<_, _, _, _, ValColBuilder<DdirUpdate>>(l, r, move |k, v1, v2, t, d1, d2, c| {
-            use differential_dataflow::difference::Multiply;
-            let d = d1.clone().multiply(d2);
-            let i = [k.as_slice(), v1.as_slice(), v2.as_slice()];
-            let (k2, v2): (Row, Row) = (eval_fields(&proj.key, &i), eval_fields(&proj.val, &i));
-            c.give((k2, v2, t, d));
-        });
-        stream.as_collection()
-    }
-
-    fn render_reduce<'scope>(a: Arr<'scope>, reducer: &Reducer) -> Arr<'scope> {
-        let reducer = reducer.clone();
-        type ReduceFn = dyn for<'a> Fn(columnar::Ref<'a, Row>, &[(columnar::Ref<'a, Row>, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync;
-        let f: Arc<ReduceFn> = match reducer {
-            Reducer::Min => Arc::new(|_key, vals, output| {
-                if let Some(min) = vals.iter().map(|(v, _)| v.as_slice()).min() {
-                    output.push((Row(min.to_vec()), 1));
-                }
-            }),
-            Reducer::Distinct => Arc::new(|_key, _vals, output| { output.push((Row::new(), 1)); }),
-            Reducer::Count => Arc::new(|_key, vals, output| {
-                let count: Diff = vals.iter().map(|(_, d)| *d).sum();
-                if count != 0 { let mut r = Row::new(); r.push(count); output.push((r, 1)); }
-            }),
-        };
-        a.reduce_abelian::<_, ColValBuilder<_,_,_,_>, ColValSpine<_,_,_,_>, _, _>(
-            "Reduce",
-            move |k, vals, output| { f(k, vals, output); },
-            |col, key, upds| {
-                use columnar::{Clear, Push};
-                col.keys.clear();
-                col.vals.clear();
-                col.times.clear();
-                col.diffs.clear();
-                for (val, time, diff) in upds.drain(..) { col.push((key, &val, &time, &diff)); }
-                // NOTE: required because push above doesn't group by key, val.
-                *col = std::mem::take(col).consolidate();
-            },
-        )
-    }
-
-    fn render_inspect<'scope>(col: Col<'scope>, label: String) -> Col<'scope> {
-        col.inspect_container(move |event| {
-            if let Ok((_time, container)) = event {
-                for (k, v, t, d) in container.updates.view().iter() {
-                    eprintln!("  [{}] ({:?}, {:?}, {:?}, {:?})", label, <Row as Columnar>::into_owned(k), <Row as Columnar>::into_owned(v), <Time as Columnar>::into_owned(t), <Diff as Columnar>::into_owned(d));
-                }
-            }
-        })
-    }
-
-    /// The columnar rendering substrate.
-    pub enum ColBackend {}
-
-    impl Backend for ColBackend {
-        type Container = DdirRecordedUpdates;
-        type Arr<'scope> = Arr<'scope>;
-
-        fn linear<'s>(c: Collection<'s, Time, Self::Container>, ops: Vec<LinearOp>, level: usize) -> Collection<'s, Time, Self::Container> {
-            render_linear(c, ops, level)
-        }
-        fn arrange<'s>(c: Collection<'s, Time, Self::Container>) -> Self::Arr<'s> {
-            use differential_dataflow::operators::arrange::arrangement::arrange_core;
-            use super::columnar::{ColValBatcher, ColValChunker};
-            arrange_core::<_, _, ColValChunker<(Row,Row,Time,Diff)>, ColValBatcher<Row,Row,Time,Diff>, ColValBuilder<Row,Row,Time,Diff>, ColValSpine<Row,Row,Time,Diff>>(c.inner.clone(), timely::dataflow::channels::pact::Pipeline, "Arrange")
-        }
-        fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container> {
-            super::columnar_support::as_recorded_updates::<DdirUpdate>(a)
-        }
-        fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection) -> Collection<'s, Time, Self::Container> {
-            render_join(l, r, projection)
-        }
-        fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
-            render_reduce(a, reducer)
-        }
-        fn inspect<'s>(c: Collection<'s, Time, Self::Container>, label: String) -> Collection<'s, Time, Self::Container> {
-            render_inspect(c, label)
-        }
-        fn leave_dynamic<'s>(c: Collection<'s, Time, Self::Container>, depth: usize) -> Collection<'s, Time, Self::Container> {
-            super::columnar::leave_dynamic(c, depth)
-        }
-    }
-}
-
-pub use types::{Row, Diff, Time};
-pub use render::{Col, ColBackend};
-
-/// Render `s` with the columnar substrate. See [`crate::backend::render_tree`].
-pub fn render_tree<'s>(
-    s: &crate::scope_ir::Scope,
-    scope: timely::dataflow::Scope<'s, Time>,
-    depth: usize,
-    imports: Vec<Col<'s>>,
-) -> Vec<Col<'s>> {
-    crate::backend::render_tree::<ColBackend>(s, scope, depth, imports)
-}
diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs
index df67d70c9..1736a9371 100644
--- a/interactive/src/backend/mod.rs
+++ b/interactive/src/backend/mod.rs
@@ -12,9 +12,6 @@
 
 pub mod vec;
 pub mod corgi;
-// The columnar substrate is deferred on the Value-model port (it needs a
-// `Columnar`/columnar-storage story for `Value`); see `backend::vec`.
-// pub mod col;
 
 use timely::Container;
 use timely::order::Product;
diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs
index 5866d51f4..11bb882a1 100644
--- a/interactive/src/corgi/logic.rs
+++ b/interactive/src/corgi/logic.rs
@@ -14,22 +14,10 @@ use crate::parse::{BinOp, Term, UnOp};
 
 use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, Kind, NumOp, Op, Pred, Shape, Value as CValue};
 
-/// Dynamic typing: form a corgi `Shape` by observing a sample row.
-pub fn infer_shape(sample: &DValue) -> Shape {
-    match sample {
-        DValue::Int(_) => Shape::Prim(64),
-        DValue::Tuple(xs) if xs.is_empty() => Shape::Unit, // DDIR unit ↔ corgi length-carrying Unit
-        DValue::Tuple(xs) => Shape::Prod(xs.iter().map(infer_shape).collect()),
-        DValue::List(xs) => Shape::List(Box::new(infer_shape(xs.first().expect("nonempty list")))),
-        DValue::Variant(..) => panic!("variant shape inference is unimplemented"),
-    }
-}
-
 /// Dynamic typing over a whole COLUMN: infer a `Shape` by scanning every row, not just a sample.
 /// Required for sum types — a `Variant` column's shape is the union of all arms that appear, which a
-/// single sample can't reveal (it shows only one tag). Non-sum shapes match [`infer_shape`] but recurse
-/// column-wise so nested variants are covered too. (`from_updates` uses this; `infer_shape` stays for
-/// the single-sample callers that never carry variants.)
+/// single sample can't reveal (it shows only one tag), so the scan is over every row, recursing
+/// column-wise to cover nested variants too.
 pub fn infer_shape_cols(rows: &[DValue]) -> Shape {
     let Some(first) = rows.first() else { return Shape::Unit };
     match first {
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 0017d7c29..1435d0a59 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -37,7 +37,7 @@ use differential_dataflow::trace::chunk::ChunkBatch;
 use differential_dataflow::operators::int_proxy::ProxyBridge;
 use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow};
 
-use corgi::arrange::{gather, gather_lanes, sort_blocks};
+use corgi::arrange::{find_ranges, gather, gather_lanes, sort_blocks};
 use corgi::{Bounds, Shape, Value as CValue};
 
 use crate::corgi::col_times::ColTime;
@@ -179,37 +179,84 @@ fn ids(col: &CValue) -> Vec<u64> {
     }
 }
 
+/// The `changed` set as a needle column in the chunks' own key shape — possible exactly
+/// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of
+/// structural keys cannot be inverted into needles.
+fn seek_needles(sample: &CValue, changed: &[u64]) -> Option<CValue> {
+    match corgi::shape_of_value(sample) {
+        Shape::Prim(64) => Some(CValue::u64(changed.to_vec())),
+        Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => {
+            Some(CValue::Prod(vec![CValue::u64(changed.to_vec())]))
+        }
+        _ => None,
+    }
+}
+
 /// Concatenate the records of the `changed` keys across a run of chunks into parallel
 /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the
 /// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it.
 ///
-/// NB this is a full scan of the presented chunks (incl. `source_batches`, the accumulated trace),
-/// and deliberately NOT a `find_ranges` seek of the changed keys: under label-propagation-shaped
-/// workloads the changed set is broad (most keys change each retire), so a scan touches ~every row
-/// regardless and the per-chunk gallop only adds overhead. The O(history) re-presentation is
-/// inherent to broad change sets, not a seekable-few-keys case.
+/// Seek-vs-scan, decided per retire, now that the sizes are known: seeking the changed keys
+/// (`find_ranges`, O(|changed|·log rows) per chunk, no key hashing at all) wins when the
+/// changed set is narrow — the steady incremental case; the full scan (O(rows) per chunk,
+/// plus each chunk's key hashes re-derived) wins for broad churn — loads and label-cascade
+/// retires, where most keys change and a gallop per key only adds overhead. Seeking requires
+/// ids that ARE key values (single-leaf keys, `ids`' fast paths): hashed ids of structural
+/// keys cannot be inverted into needles, so those always scan.
 ///
-/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's key
-/// hashes every retire (memoize per chunk, or a stored hash column), the membership test is a
-/// per-row `binary_search`, and each hit materializes an owned time (`times().get`); kept RANGES
-/// could move via `push_range`.
+/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's
+/// key hashes every retire (memoize per chunk, or a stored hash column), and each hit
+/// materializes an owned time (`times().get`); kept RANGES could move via `push_range`.
 fn collect_present<T>(chunks: &[&CorgiChunk<T, Diff>], changed: &[u64]) -> (CValue, CValue, Vec<u64>, Vec<T>, Vec<Diff>)
 where
     T: ColTime,
 {
+    /// Seek only when the changed set is at least this many times narrower than the
+    /// presented rows: a `find_ranges` probe is a structurally-dispatched binary search
+    /// (~log(rows) compares, each far costlier than the scan's flat membership test), so
+    /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped
+    /// retires before this was widened.
+    const SEEK_ADVANTAGE: usize = 64;
+
     let key_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.keys())).collect();
     let val_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.vals())).collect();
     let (mut tags, mut offs) = (Vec::new(), Vec::new());
     let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
-    for (ci, ch) in chunks.iter().enumerate() {
-        let kh = ids(ch.keys());
-        for i in 0..kh.len() {
-            if changed.binary_search(&kh[i]).is_ok() {
-                tags.push(ci);
-                offs.push(i);
-                khs.push(kh[i]);
-                times.push(ch.times().get(i));
-                diffs.push(ch.diffs()[i]);
+    let total: usize = chunks.iter().map(|c| c.diffs().len()).sum();
+    let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total {
+        chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed))
+    } else {
+        None
+    };
+    if let Some(needles) = needles {
+        // Narrow changed set over seekable keys: gallop each chunk once per changed key.
+        // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's.
+        for (ci, ch) in chunks.iter().enumerate() {
+            if ch.diffs().is_empty() {
+                continue;
+            }
+            let (lo, hi) = find_ranges(&needles, ch.keys());
+            for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() {
+                for i in l..h {
+                    tags.push(ci);
+                    offs.push(i);
+                    khs.push(changed[j]);
+                    times.push(ch.times().get(i));
+                    diffs.push(ch.diffs()[i]);
+                }
+            }
+        }
+    } else {
+        for (ci, ch) in chunks.iter().enumerate() {
+            let kh = ids(ch.keys());
+            for i in 0..kh.len() {
+                if changed.binary_search(&kh[i]).is_ok() {
+                    tags.push(ci);
+                    offs.push(i);
+                    khs.push(kh[i]);
+                    times.push(ch.times().get(i));
+                    diffs.push(ch.diffs()[i]);
+                }
             }
         }
     }