diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs
index aea8e598b..069846f12 100644
--- a/differential-dataflow/src/operators/common.rs
+++ b/differential-dataflow/src/operators/common.rs
@@ -179,9 +179,10 @@ pub fn tile_descriptions<T: Timestamp + Lattice>(
     (tile_descs, tile_held, tile_of)
 }
 
-/// A one-key view into an input presentation: the read-only arguments [`discover_times`] needs
-/// about a single key — its slice `[i0, i1)` of the merged `(id, time, diff)` run `p_in` and
-/// the carried `pending` times.
+/// A one-key view into the ACCUMULATED input presentation: the read-only arguments
+/// [`discover_times`] needs about a single key — its slice `[i0, i1)` of the `(id, time, diff)` run
+/// `p_in` and the carried `pending` times. The novel run is not here; it arrives as `seed_times`,
+/// and the two are deliberately never merged (see the note on `seed_times` below).
 pub struct KeyView<'a, T, RIn> {
     /// The presented `((key_hash, value_id), time, diff)` run the key's records live in.
     pub p_in: &'a [((u64, u64), T, RIn)],
@@ -236,15 +237,18 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch
 
 /// Determines the times in `[lower, upper)` at which a key's reduction must be re-evaluated
 /// (`moments`), and the times at or beyond `upper` to carry into the next invocation
-/// (`pended`). Replays the key's `seed_times` and `pending` times in ascending order, marking
+/// (`pended`). Replays the key's `seed_times` (the novel run) and `pending` times in ascending
+/// order, marking
 /// those that carry updates and closing the set under joins with the input and output
 /// histories' times and with each other. No input collection is materialized, so peak memory
 /// is O(times); buffers are advanced by the meet of the times still to come, keeping a key
 /// with many distinct times linear rather than quadratic.
 ///
-/// `seed_times` must be the novel batch's own time support for this key. Seeding from a
-/// consolidated view is unsound: compaction may advance a history record onto a novel time,
-/// where consolidation cancels the novel update and its interesting time is missed.
+/// `seed_times` must be the novel batch's own time support for this key, and `key.p_in` the
+/// accumulated input WITHOUT it. Seeding from a view of the two consolidated together is unsound:
+/// compaction may advance a history record onto a novel time, where consolidation cancels the novel
+/// update and its interesting time is missed. This is why the caller keeps the two runs apart and
+/// combines them only when accumulating.
 #[allow(clippy::too_many_arguments)]
 pub fn discover_times<T, RIn>(
     key: KeyView<'_, T, RIn>,
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index ee206a904..27668ae49 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -3,7 +3,7 @@
 //! A conventional differential reduce against `(u64, u64)`, where the backend supplies the
 //! implementation of the interpretation of the integers.
 
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
 
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
@@ -30,34 +30,48 @@ pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     pub lower: AntichainRef<'a, B1::Time>,
 }
 
-/// One window of a retire's changed keys: a bounded, hash-contiguous snip the backend sizes.
+/// One window of the key space: the presentations a bounded, hash-contiguous snip needs.
 ///
-/// The window has the input (old and new) and output histories, restricted to the window's keys.
+/// The two input runs are held apart because the novel and prior data have different roles:
+/// the novel data seed interesting times, and the prior data may benefit from historical rollup.
+/// Their updates are combined after we are able to see the distinction between the two.
+///
+/// Owned by the harness and refilled by [`ProxyReduceBackend::next_window`].
 pub struct ReduceWindow<T, RIn, ROut> {
-    /// The window's key hashes: a contiguous, ascending slice of the retire's `changed` keys.
-    pub keys: Vec<u64>,
-    /// Input presentation for `keys`, sorted & consolidated by `((key_hash, value_id), time)`.
-    pub input: ProxyBridge<T, RIn>,
-    /// Output-history presentation for `keys`, same ordering.
+    /// Accumulated input preceding the retire's interval, sorted & consolidated by
+    /// `((key_hash, value_id), time)`.
+    pub history: ProxyBridge<T, RIn>,
+    /// The retire's novel input delta, same ordering. Its times per key are the interesting-time
+    /// seeds, so it must be the batch's own time support — not a view netted against `history`.
+    pub novel: ProxyBridge<T, RIn>,
+    /// Accumulated output preceding the retire's interval, same ordering.
     pub output: ProxyBridge<T, ROut>,
 }
 
+impl<T, RIn, ROut> Default for ReduceWindow<T, RIn, ROut> {
+    fn default() -> Self { ReduceWindow { history: Vec::new(), novel: Vec::new(), output: Vec::new() } }
+}
+
+impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
+    /// Clear the three proxy bridges.
+    pub fn clear(&mut self) {
+        self.history.clear();
+        self.novel.clear();
+        self.output.clear();
+    }
+}
+
 /// The reduce backend: value semantics for a proxy-space reduction, driven by [`ProxyReduceTactic`].
 ///
-/// The protocol is currently (temporarily) for each round of invocation:
-/// `seed_times begin [ next_window reduce_correction* emit ]* finish`
-/// This should be improved to put the `seed_times` in the per-window loop, or remove it entirely.
+/// The protocol for each round of invocation is
+/// `begin [ next_window reduce_corrections* emit ]* finish`,
+/// where the window loop runs until `next_window` reports the key space exhausted.
 pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
     /// Diff type presented for the input.
     type RIn: Semigroup;
     /// Diff type of the output.
     type ROut: Semigroup + 'static;
 
-    /// Hash keys and associated times in the instance's novel input batches.
-    ///
-    /// This is used (with held times) to seed the interesting times for each key.
-    fn seed_times(&self, instance: &ReduceInstance<'_, B1, B2>) -> Vec<(u64, B1::Time)>;
-
     /// Initiate a session to create batches for these descriptions, which span `[lower, upper)`.
     ///
     /// It is the backend's job to prepare output batches for each of these descriptions.
@@ -65,12 +79,28 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// work in progress, until `finish()` is called.
     fn begin(&mut self, tiles: &[Description<B1::Time>]);
 
-    /// Produce the next window, resticted to `changed[cursor..]`, and update `cursor` to track.
+    /// Present the next window of the key space, and advance `from` past it.
+    ///
+    /// On entry `from` is the inclusive lower bound on key hashes still to be covered. The backend
+    /// chooses the window's exclusive upper bound and writes it back, or writes `None` to report the
+    /// key space exhausted. An implementor must advance `from`, as it is guaranteed to be non-`None`.
     ///
-    /// The size of the window is up to the backend, where the window should be large enough to
-    /// amortize the crossings between the harness and the backend. The proxy bridges for the
-    /// whole window will be active at the same time, so tighter windows reduce the required state.
-    fn next_window(&mut self, instance: &ReduceInstance<'_, B1, B2>, changed: &[u64], cursor: &mut usize) -> Option<ReduceWindow<B1::Time, Self::RIn, Self::ROut>>;
+    /// The window must present, for every key hash in `[from_before, from_after)` that either
+    /// carries an update in the instance's novel batches or appears in `changed`, that key's novel
+    /// updates, its accumulated input, and its accumulated output. A key must be reported entirely
+    /// within the window that first mentions it: splitting one across windows drops the interaction
+    /// between the halves. `changed` is ascending; the harness reads no key outside the window's
+    /// range, so a backend that keeps its own key order need not consult the whole space.
+    ///
+    /// The size of the window is up to the backend: large enough to amortize the crossings, small
+    /// enough that the three presentations are affordable, as all are live at once.
+    fn next_window(
+        &mut self,
+        instance: &ReduceInstance<'_, B1, B2>,
+        changed: &[u64],
+        from: &mut Option<u64>,
+        window: &mut ReduceWindow<B1::Time, Self::RIn, Self::ROut>,
+    );
 
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
     ///
@@ -136,15 +166,34 @@ where
             lower: lower.borrow(),
         };
 
-        let seeds = self.backend.seed_times(&instance);
-        debug_assert!(seeds.windows(2).all(|w| w[0].0 <= w[1].0), "seed_times must be sorted by key_hash");
-        let mut changed: BTreeSet<u64> = seeds.iter().map(|(k, _)| *k).collect();
-        changed.extend(self.pending.keys().copied());
-        if changed.is_empty() {
-            self.pending.clear();
-            return (Vec::new(), Antichain::new());
+        // Split the carried interesting times against `upper`.
+        // A time below it is DUE: its key must be re-evaluated this retire, so the key is `changed`.
+        // A time at or beyond it is carried forward untouched, seeding `new_pending`.
+        let mut due: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        for (key, times) in self.pending.iter() {
+            let (carried, ready): (Vec<_>, Vec<_>) = times.iter().cloned().partition(|t| upper.less_equal(t));
+            if !ready.is_empty() { due.insert(*key, ready); }
+            if !carried.is_empty() { new_pending.insert(*key, carried); }
+        }
+        // The keys the harness knows must be revisited. The backend adds those its novel batches
+        // touch, which it discovers while reading them; neither side scans the whole key space.
+        let changed: Vec<u64> = due.keys().copied().collect();
+
+        // Nothing due and nothing novel: no time in the interval can be interesting, so there is no
+        // work and no output. Return the frontier bounding the times still withheld — NOT an empty
+        // one. This is exactly where a due-only `changed` differs from the whole pending set: times
+        // beyond `upper` can remain when nothing is due, and releasing their capabilities would
+        // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
+        if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
+            let mut frontier = Antichain::new();
+            for times in self.pending.values() {
+                for time in times {
+                    frontier.insert_ref(time);
+                }
+            }
+            return (Vec::new(), frontier);
         }
-        let changed: Vec<u64> = changed.into_iter().collect();
 
         // The output tiling (identical to the Abelian tactic): one tile per held time, keeping
         // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile.
@@ -152,10 +201,10 @@ where
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
         self.backend.begin(&tile_descs);
 
-        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
-
-        let mut cursor = 0usize;
-        let mut ns = 0usize;
+        // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
+        // once the backend reports the space covered.
+        let mut from = Some(0u64);
+        let mut window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut> = ReduceWindow::default();
 
         // Retire-wide reusable scratch (cleared per window/round/moment, not reallocated). See the
         // profiling note on `DiscoverScratch`: fresh per-key/per-round `Vec`s were the dominant cost.
@@ -173,11 +222,30 @@ where
         let mut moments_scratch: Vec<B1::Time> = Vec::new();
         let mut pended_scratch: Vec<B1::Time> = Vec::new();
 
-        while let Some(window) = self.backend.next_window(&instance, &changed, &mut cursor) {
-            let p_in = &window.input;
+        while from.is_some() {
+            let before = from;
+            window.clear();
+            self.backend.next_window(&instance, &changed, &mut from, &mut window);
+            let p_in = &window.history;
+            let p_nv = &window.novel;
             let p_out = &window.output;
-            super::debug_assert_sorted_bridge(p_in, "next_window.input");
+            super::debug_assert_sorted_bridge(p_in, "next_window.history");
+            super::debug_assert_sorted_bridge(p_nv, "next_window.novel");
             super::debug_assert_sorted_bridge(p_out, "next_window.output");
+            // Without progress the window loop would never retire, so this guards liveness as well
+            // as contract; the range check catches a key reported outside the window that owns it,
+            // which would silently drop the interaction between its halves.
+            debug_assert!(
+                from.is_none() || from > before,
+                "next_window must either advance `from` or report the key space exhausted",
+            );
+            debug_assert!(
+                {
+                    let mut keys = p_in.iter().chain(p_nv.iter()).map(|r| r.0.0).chain(p_out.iter().map(|r| r.0.0));
+                    keys.all(|k| before.is_none_or(|b| b <= k) && from.is_none_or(|f| k < f))
+                },
+                "next_window must report a key hash entirely within the window that first mentions it",
+            );
 
             for deltas in tile_deltas.iter_mut() { deltas.clear(); }
 
@@ -187,27 +255,41 @@ where
             // `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's
             // `Vec`s and replays are allocated once and reused, so keys cost no per-key alloc/free.
             // `n_states` is the live prefix this window; higher slots persist (retaining capacity).
+            // The window's keys are the hashes its presentations mention: the least of the three
+            // heads, each iteration, until all three are drained. A `changed` key that appears in
+            // none of them has no records at all, so its reduction has nothing to read and nothing
+            // to retract — the moment its due time would raise reaches the evaluation gate with an
+            // empty input and an empty output, and produces nothing. Skipping it is exactly what
+            // visiting it would do. (`changed` is still the backend's instruction about which keys
+            // to present; it is just not a source of keys here.)
             let mut n_states = 0usize;
-            let (mut is, mut os) = (0usize, 0usize);
-            for &key in &window.keys {
-                while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
+            let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
+            loop {
+                // Mapped to hashes before the min: the three runs differ in their diff type.
+                let Some(key) = [
+                    p_in.get(is).map(|record| record.0.0),
+                    p_nv.get(ns).map(|record| record.0.0),
+                    p_out.get(os).map(|record| record.0.0),
+                ].into_iter().flatten().min() else { break };
                 let i0 = is;
                 while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
                 let i1 = is;
-                while os < p_out.len() && p_out[os].0.0 < key { os += 1; }
+                let n0 = ns;
+                while ns < p_nv.len() && p_nv[ns].0.0 == key { ns += 1; }
+                let n1 = ns;
                 let o0 = os;
                 while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
                 let o1 = os;
-                while ns < seeds.len() && seeds[ns].0 < key { ns += 1; }
-                let n0 = ns;
-                while ns < seeds.len() && seeds[ns].0 == key { ns += 1; }
-                let n1 = ns;
 
                 moments_scratch.clear();
                 pended_scratch.clear();
                 {
-                    let pending = self.pending.get(&key).map(|p| &p[..]).unwrap_or(&[]);
-                    let seed_times = seeds[n0..n1].iter().map(|(_, t)| t.clone());
+                    // Only the DUE times seed determination; the carried ones are already in
+                    // `new_pending` and must not be re-derived (nor re-pended) here.
+                    let pending = due.get(&key).map(|p| &p[..]).unwrap_or(&[]);
+                    // The seeds are the novel run's own time support for this key — the whole point
+                    // of presenting it apart from `history`.
+                    let seed_times = (n0..n1).map(|n| p_nv[n].1.clone());
                     let out_times = (o0..o1).map(|o| p_out[o].1.clone());
                     discover_times(
                         KeyView { p_in: &p_in[..], i0, i1, pending },
@@ -217,7 +299,10 @@ where
                     );
                 }
                 if !pended_scratch.is_empty() {
-                    new_pending.insert(key, std::mem::take(&mut pended_scratch));
+                    // The key may already carry times beyond `upper` from the split above; union.
+                    let entry = new_pending.entry(key).or_default();
+                    entry.append(&mut pended_scratch);
+                    crate::operators::reduce::sort_dedup(entry);
                 }
                 if moments_scratch.is_empty() {
                     continue;
@@ -241,6 +326,7 @@ where
                     st.meets[i - 1].meet_assign(&m);
                 }
                 st.in_replay.load_iter((i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), st.meets.first());
+                st.nv_replay.load_iter((n0..n1).map(|n| (p_nv[n].0.1, p_nv[n].1.clone(), p_nv[n].2.clone())), st.meets.first());
                 st.out_replay.load_iter((o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), st.meets.first());
                 n_states += 1;
             }
@@ -269,16 +355,19 @@ where
                     st.cursor += 1;
                     let t = st.moments[j].clone();
                     st.in_replay.step_through(&t);
+                    st.nv_replay.step_through(&t);
                     st.out_replay.step_through(&t);
                     st.in_replay.advance_buffer_by(&st.meets[j]);
+                    st.nv_replay.advance_buffer_by(&st.meets[j]);
                     st.out_replay.advance_buffer_by(&st.meets[j]);
                     for ((_, et), _) in st.produced.iter_mut() {
                         *et = et.join(&st.meets[j]);
                     }
                     crate::consolidation::consolidate(&mut st.produced);
 
+                    // The two input runs meet HERE, at the accumulation — never as presentations.
                     in_accum.clear();
-                    for ((vid, et), d) in st.in_replay.buffer().iter() {
+                    for ((vid, et), d) in st.in_replay.buffer().iter().chain(st.nv_replay.buffer().iter()) {
                         if et.less_equal(&t) {
                             in_accum.push((*vid, d.clone()));
                         }
@@ -351,7 +440,8 @@ where
 }
 
 /// Per-key application state for [`ProxyReduceTactic`]'s round-batched walk: the key's ordered
-/// interesting `moments` and their suffix `meets`, its input and output replays (meet-collapsed),
+/// interesting `moments` and their suffix `meets`, its accumulated-input, novel-input, and output
+/// replays (meet-collapsed; the two input replays combine only in the per-moment accumulation),
 /// the corrections `produced` this round so far, and a `cursor` into `moments`. Held for all of a
 /// window's keys at once so each round's crossing batches across keys — a key's own moments stay
 /// sequential (each sees its earlier corrections via `produced`), but distinct keys are independent.
@@ -360,6 +450,7 @@ struct KeyState<T, RIn, ROut> {
     moments: Vec<T>,
     meets: Vec<T>,
     in_replay: IdHistory<T, RIn>,
+    nv_replay: IdHistory<T, RIn>,
     out_replay: IdHistory<T, ROut>,
     produced: Vec<((u64, T), ROut)>,
     cursor: usize,
@@ -370,7 +461,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<T, RIn, R
     /// vector holds these across windows and reloads them in place, so a key's buffers are allocated
     /// once (per slot) and reused — never dropped per key (which was ~18% of load in `free`).
     fn empty() -> Self {
-        KeyState { key: 0, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
+        KeyState { key: 0, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), nv_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
     }
 }
 
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 68ceacd21..1accf08b2 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -288,6 +288,33 @@ impl<T> CorgiReduceBackend<T>
 where
     T: ColTime + Ord,
 {
+    /// Present one input run — the accumulated history or the novel delta — restricted to `keys`.
+    ///
+    /// Fills `bridge`, registers the run's representative keys, and extends the shared value pool
+    /// (`blocks`/`len`, concatenated into `in_vals`) with its values, so `in_index` resolves a value
+    /// id from EITHER run to a row. The two runs stay apart as presentations and meet only in the
+    /// tactic's accumulation; the pool is shared because a value id means the same thing in both.
+    fn present_input(
+        &mut self,
+        chunks: &[&CorgiChunk<T, Diff>],
+        keys: &[u64],
+        blocks: &mut Vec<CValue>,
+        len: &mut usize,
+        bridge: &mut ProxyBridge<T, Diff>,
+    ) {
+        let (p_keys, p_vals, khs, times, diffs) = collect_present(chunks, keys);
+        if khs.is_empty() {
+            return;
+        }
+        let vids = ids(&p_vals);
+        for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
+        *len += p_vals.len();
+        blocks.push(p_vals);
+        self.register_keys(p_keys, &khs);
+        bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
+        consolidate_updates(bridge);
+    }
+
     /// The one value crossing for a retire: every `(key, time)` bracket at once. Builds the output
     /// value COLUMN directly per reducer, registers it (id → row) into the val pool, and returns the
     /// proxy `(value_id, diff)` deltas with per-bracket ends. `input[k] = (rep index into the input
@@ -430,22 +457,6 @@ where
     type RIn = Diff;
     type ROut = Diff;
 
-    fn seed_times(&self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>) -> Vec<(u64, T)> {
-        // The batch's raw (key_hash, time) support — hash the novel KEY columns only, one entry per
-        // record, sorted by key_hash. Seeds may over-derive (a non-changing seed yields a zero delta),
-        // so this superset of b.support suffices; `instance.lower` is not applied (see ReduceInstance).
-        let mut out: Vec<(u64, T)> = Vec::new();
-        for ch in chunks_of(instance.input_batches) {
-            let kh = ids(ch.keys());
-            let times = ch.times();
-            for (i, h) in kh.into_iter().enumerate() {
-                out.push((h, times.get(i)));
-            }
-        }
-        out.sort_by_key(|(k, _)| *k);
-        out
-    }
-
     fn begin(&mut self, tiles: &[Description<T>]) {
         // Open a tiled output session for this retire; reset the per-retire resolution pools.
         self.reset_pools();
@@ -453,57 +464,70 @@ where
         self.tile_rows = (0..tiles.len()).map(|_| (Vec::new(), Vec::new(), Vec::new(), Vec::new())).collect();
     }
 
-    fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>, changed: &[u64], cursor: &mut usize) -> Option<ReduceWindow<T, Diff, Diff>> {
-        // Single window: present ALL remaining changed keys at once. This is NOT a deferred
-        // refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
+    fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>, changed: &[u64], from: &mut Option<u64>, window: &mut ReduceWindow<T, Diff, Diff>) {
+        // Single window: present the WHOLE key space at once, and report it covered. This is NOT a
+        // deferred refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
         // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS
         // fell only 356MB -> 340MB. Two reasons: the per-window, per-chunk seek setup is a
         // fixed cost that multiplies by the window count, and the presentation is not the
-        // memory peak in the first place (the trace is). `changed` is ascending, so
-        // `binary_search` is the changed-key filter.
-        if *cursor >= changed.len() {
-            return None;
+        // memory peak in the first place (the trace is).
+        if from.is_none() {
+            return;
         }
-        let keys: Vec<u64> = changed[*cursor..].to_vec();
-        *cursor = changed.len();
-        let present = |chunks: &[&CorgiChunk<T, Diff>]| collect_present(chunks, &keys);
-
-        // Input presentation: accumulated history ∪ novel delta, restricted to the window's keys.
-        // value_id = content hash of the value (equal values share an id → the tactic nets them);
-        // `in_index` resolves an id back to a representative in_vals row for `reduce_corrections`.
-        let mut in_chunks = chunks_of(instance.input_batches);
-        in_chunks.extend(chunks_of(instance.source_batches));
-        let (in_keys, in_vals, in_khs, in_times, in_diffs) = present(&in_chunks);
-        self.in_index = IdMap::default();
-        let input: ProxyBridge<T, Diff> = if in_khs.is_empty() {
+        *from = None;
+
+        // The window's keys: the hashes the novel batches touch, merged with the `changed` set the
+        // harness supplies. The novel hashes come from the scan the presentation needs anyway — the
+        // separate seeding pass this replaced read the delta a second time to derive them.
+        let novel_chunks = chunks_of(instance.input_batches);
+        let mut keys: Vec<u64> = Vec::new();
+        for ch in novel_chunks.iter() {
+            keys.extend(ids(ch.keys()));
+        }
+        keys.sort_unstable();
+        keys.dedup();
+        if !changed.is_empty() {
+            // Both sides ascend, so this is a merge.
+            let mut merged: Vec<u64> = Vec::with_capacity(keys.len() + changed.len());
+            let (mut a, mut b) = (0usize, 0usize);
+            while a < keys.len() || b < changed.len() {
+                let key = match (keys.get(a), changed.get(b)) {
+                    (Some(x), Some(y)) => *x.min(y),
+                    (Some(x), None) => *x,
+                    (None, Some(y)) => *y,
+                    (None, None) => unreachable!("loop condition ensures one is present"),
+                };
+                if keys.get(a) == Some(&key) { a += 1; }
+                if changed.get(b) == Some(&key) { b += 1; }
+                merged.push(key);
+            }
+            keys = merged;
+        }
+        if keys.is_empty() {
             self.in_vals = CValue::Unit(0);
-            Vec::new()
-        } else {
-            let vids = ids(&in_vals);
-            for (r, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(r); }
-            self.in_vals = in_vals;
-            self.register_keys(in_keys, &in_khs);
-            let mut b: ProxyBridge<T, Diff> =
-                (0..in_khs.len()).map(|i| ((in_khs[i], vids[i]), in_times[i].clone(), in_diffs[i])).collect();
-            consolidate_updates(&mut b);
-            b
-        };
+            self.in_index = IdMap::default();
+            return;
+        }
+
+        // The two input presentations, held apart: `history` is the accumulated input and `novel`
+        // the delta whose times seed the interesting times. `in_index` resolves a value id back to a
+        // representative row of `in_vals`, which spans BOTH runs, for `reduce_corrections`.
+        self.in_index = IdMap::default();
+        let mut in_blocks: Vec<CValue> = Vec::new();
+        let mut in_len = 0usize;
+        self.present_input(&chunks_of(instance.source_batches), &keys, &mut in_blocks, &mut in_len, &mut window.history);
+        self.present_input(&novel_chunks, &keys, &mut in_blocks, &mut in_len, &mut window.novel);
+        self.in_vals = concat_columns(&in_blocks);
 
         // Output-history presentation, same keys (register keys + values for correction resolution).
-        let (o_keys, o_vals, o_khs, o_times, o_diffs) = present(&chunks_of(instance.output_batches));
-        let output: ProxyBridge<T, Diff> = if o_khs.is_empty() {
-            Vec::new()
-        } else {
+        let (o_keys, o_vals, o_khs, o_times, o_diffs) = collect_present(&chunks_of(instance.output_batches), &keys);
+        if !o_khs.is_empty() {
             let vids = ids(&o_vals);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            let mut b: ProxyBridge<T, Diff> =
-                (0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])).collect();
-            consolidate_updates(&mut b);
-            b
-        };
-
-        Some(ReduceWindow { keys, input, output })
+            window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
+            consolidate_updates(&mut window.output);
+        }
     }
 
     fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec<usize>) {