diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index f6c4e52a4..bb7f74f6f 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -296,7 +296,6 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
     }
 }
 
-
 impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     /// Brings an arranged collection out of a nested region.
     ///
diff --git a/differential-dataflow/src/operators/mod.rs b/differential-dataflow/src/operators/mod.rs
index 4260c2b95..a2e0a32b2 100644
--- a/differential-dataflow/src/operators/mod.rs
+++ b/differential-dataflow/src/operators/mod.rs
@@ -125,6 +125,24 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
         self.replay()
     }
 
+    /// Wraps the already-built, sorted `history` for a fresh walk, WITHOUT rebuilding or re-sorting
+    /// it. Valid whenever `history` is intact — e.g. after a `replay_key` whose returned replay was
+    /// only read through [`HistoryReplay::times`] (which does not step it). Used by reduce's
+    /// `reference` tactic, whose determination reads times and whose application then walks values.
+    fn walk<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
+        self.buffer.clear();
+        HistoryReplay { replay: self }
+    }
+
+    /// A time-only, non-destructive walk over the already-built `history` (see [`TimeReplay`]). It
+    /// reads times and their precomputed meets and accumulates only *times* into the caller-supplied
+    /// `buffer`, leaving `history` intact for a later [`walk`](Self::walk) over values. The buffer is
+    /// owned by the caller so the standard value walk pays nothing for this reference-only capability.
+    fn replay_times<'history>(&'history self, buffer: &'history mut Vec<T>) -> TimeReplay<'history, T> {
+        buffer.clear();
+        TimeReplay { history: &self.history[..], buffer }
+    }
+
     /// Organizes history based on current contents of edits.
     fn replay<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
 
@@ -181,3 +199,40 @@ impl<'history, V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::differ
     }
     fn is_done(&self) -> bool { self.replay.history.is_empty() }
 }
+
+/// A time-only, non-destructive walk over an already-built [`ValueHistory`] history.
+///
+/// It mirrors the time-facing half of [`HistoryReplay`] — `time`/`meet`/`step`/`step_while_time_is`
+/// /`advance_buffer_by`/`buffer` — but it carries only *times*, and it walks a shrinking *view* of
+/// `history` rather than popping it, so the underlying history stays intact for a later value walk.
+/// `advance_buffer_by` compacts the accumulated times by joining with `meet` and deduplicating (the
+/// time-only analogue of consolidation), which is what keeps a join-closure over these times an
+/// antichain — and hence non-quadratic.
+struct TimeReplay<'history, T> {
+    history: &'history [(T, T, usize, usize)],   // shrinking view; `last()` is the least time
+    buffer: &'history mut Vec<T>,                 // accumulated (and compacted) times seen so far
+}
+
+impl<'history, T: Ord + Clone + Lattice> TimeReplay<'history, T> {
+    fn time(&self) -> Option<&T> { self.history.last().map(|entry| &entry.0) }
+    fn meet(&self) -> Option<&T> { self.history.last().map(|entry| &entry.1) }
+    fn step(&mut self) {
+        let last = self.history.len() - 1;
+        self.buffer.push(self.history[last].0.clone());
+        self.history = &self.history[..last];
+    }
+    fn step_while_time_is(&mut self, time: &T) -> bool {
+        let mut found = false;
+        while self.time() == Some(time) {
+            found = true;
+            self.step();
+        }
+        found
+    }
+    fn advance_buffer_by(&mut self, meet: &T) {
+        for time in self.buffer.iter_mut() { time.join_assign(meet); }
+        self.buffer.sort();
+        self.buffer.dedup();
+    }
+    fn buffer(&self) -> &[T] { &self.buffer[..] }
+}
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 08c99dba5..55210ace4 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -73,6 +73,11 @@ where
     reduce_with_tactic(trace, name, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
 }
 
+// The model-derived reference tactic and its entry point live in `mod reference`; re-exported here
+// (doc-hidden) as the sole public handle for its differential and oracle tests.
+#[doc(hidden)]
+pub use reference::reduce_trace_reference;
+
 /// Drives a key-wise reduction using a supplied [`ReduceTactic`].
 ///
 /// This is the general reduce operator: it does the dataflow plumbing (frontiers, capabilities, output
@@ -620,33 +625,73 @@ mod cursors {
                     // before we add the time to `synth_times`.
                     if !upper_limit.less_equal(&next_time) {
 
+                        // DETERMINATION (times only). Determine synthetic interesting times.
+                        //
+                        // Synthetic interesting times are produced differently for interesting and uninteresting
+                        // times. An uninteresting time must join with an interesting time to become interesting,
+                        // which means joins with `self.batch_history` and  `self.times_current`. I think we can
+                        // skip `self.synth_times` as we haven't gotten to them yet, but we will and they will be
+                        // joined against everything.
+
+                        // Any time, even uninteresting times, must be joined with the current accumulation of
+                        // batch times as well as the current accumulation of `times_current`.
+                        self.temporary.extend(batch_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                        self.temporary.extend(self.times_current.iter().filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+
+                        // An interesting time additionally joins with `input` and `output` history and this round's
+                        // produced output: it carries the seed, so those joins stay interesting (an uninteresting
+                        // time does not, as `input`/`output` times are not themselves seeds). We advance the buffers
+                        // by `meet` first, exactly as evaluation reads them below; by join preservation the advanced
+                        // and unadvanced times spawn the same synthetics, so this matches the pre-split behavior.
+                        if interesting {
+                            if let Some(meet) = meet.as_ref() { input_replay.advance_buffer_by(meet) };
+                            if let Some(meet) = meet.as_ref() { output_replay.advance_buffer_by(meet) };
+                            self.temporary.extend(input_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                            self.temporary.extend(output_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                            self.temporary.extend(self.output_produced.iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                        }
+                        sort_dedup(&mut self.temporary);
+
+                        // Introduce synthetic times, and re-organize if we add any.
+                        let synth_len = self.synth_times.len();
+                        for time in self.temporary.drain(..) {
+                            // We can either service `join` now, or must delay for the future.
+                            if upper_limit.less_equal(&time) {
+                                debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&time)));
+                                new_interesting.push(time);
+                            }
+                            else {
+                                self.synth_times.push(time);
+                            }
+                        }
+                        if self.synth_times.len() > synth_len {
+                            self.synth_times.sort_by(|x,y| y.cmp(x));
+                            self.synth_times.dedup();
+                        }
+
+                        // EVALUATION (values only).
                         // We should re-evaluate the computation if this is an interesting time.
                         // If the time is uninteresting (and our logic is sound) it is not possible for there to be
                         // output produced. This sounds like a good test to have for debug builds!
                         if interesting {
 
                             // Assemble the input collection at `next_time`. (`self.input_buffer` cleared just after use).
+                            // The buffers were advanced by `meet` in the determination step above.
                             debug_assert!(self.input_buffer.is_empty());
-                            if let Some(meet) = meet.as_ref() { input_replay.advance_buffer_by(meet) };
                             for ((value, time), diff) in input_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             for ((value, time), diff) in batch_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             crate::consolidation::consolidate(&mut self.input_buffer);
 
                             // Assemble the output collection at `next_time`. (`self.output_buffer` cleared just after use).
-                            if let Some(meet) = meet.as_ref() { output_replay.advance_buffer_by(meet) };
                             for ((value, time), diff) in output_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.output_buffer.push((C2::owned_val(*value), diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             for ((value, time), diff) in self.output_produced.iter() {
                                 if time.less_equal(&next_time) { self.output_buffer.push(((*value).to_owned(), diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             crate::consolidation::consolidate(&mut self.output_buffer);
 
@@ -686,37 +731,6 @@ mod cursors {
                                 }
                             }
                         }
-
-                        // Determine synthetic interesting times.
-                        //
-                        // Synthetic interesting times are produced differently for interesting and uninteresting
-                        // times. An uninteresting time must join with an interesting time to become interesting,
-                        // which means joins with `self.batch_history` and  `self.times_current`. I think we can
-                        // skip `self.synth_times` as we haven't gotten to them yet, but we will and they will be
-                        // joined against everything.
-
-                        // Any time, even uninteresting times, must be joined with the current accumulation of
-                        // batch times as well as the current accumulation of `times_current`.
-                        self.temporary.extend(batch_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
-                        self.temporary.extend(self.times_current.iter().filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
-                        sort_dedup(&mut self.temporary);
-
-                        // Introduce synthetic times, and re-organize if we add any.
-                        let synth_len = self.synth_times.len();
-                        for time in self.temporary.drain(..) {
-                            // We can either service `join` now, or must delay for the future.
-                            if upper_limit.less_equal(&time) {
-                                debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&time)));
-                                new_interesting.push(time);
-                            }
-                            else {
-                                self.synth_times.push(time);
-                            }
-                        }
-                        if self.synth_times.len() > synth_len {
-                            self.synth_times.sort_by(|x,y| y.cmp(x));
-                            self.synth_times.dedup();
-                        }
                     }
                     else if interesting {
                         // We cannot process `next_time` now, and must delay it.
@@ -760,3 +774,511 @@ mod cursors {
         }
     }
 }
+
+/// A second [`ReduceTactic`], written directly from the incremental model in
+/// `formal/Differential/Model.lean`.
+///
+/// Per key it runs two phases over one cursor walk. Phase 1 (determination) computes the
+/// interesting times as the truncated join-closure over {input, output, seeds} — the model's
+/// `Reached` — advancing by meets so the synthetic set stays bounded. Phase 2 (application) walks
+/// exactly those times in order, maintaining tight input/output accumulations by meets, and emits
+/// the corrections — the model's `emit_correct`. Determination never consults the output produced
+/// this round (it finishes first), so this tactic embodies the proven algorithm exactly and is the
+/// clean subject for differential testing against [`cursors::CursorTactic`].
+pub(crate) mod reference {
+
+    use super::*;
+    use crate::lattice::Lattice;
+    use crate::operators::ValueHistory;
+
+    /// Drives a key-wise reduction with the model-derived [`ReferenceTactic`], the analogue of the
+    /// default [`super::reduce_trace`]. Same result contract; intended for differential testing of the
+    /// two tactics against each other. Re-exported (doc-hidden) from the parent module as the sole
+    /// public handle: the reference tactic is a testing and demonstration oracle, not a stable entry
+    /// point to build on.
+    pub fn reduce_trace_reference<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
+    where
+        Tr1: TraceReader<Batch: Navigable> + 'static,
+        Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
+        Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default> + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
+    {
+        reduce_with_tactic(trace, name, ReferenceTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
+    }
+
+    /// Sorts and deduplicates, matching `cursors::sort_dedup`.
+    #[inline(never)]
+    fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
+        list.dedup();
+        list.sort();
+        list.dedup();
+    }
+
+    /// Updates an optional meet by an optional time.
+    fn update_meet<T: Lattice+Clone>(meet: &mut Option<T>, other: Option<&T>) {
+        if let Some(time) = other {
+            if let Some(meet) = meet.as_mut() { meet.meet_assign(time); }
+            else { *meet = Some(time.clone()); }
+        }
+    }
+
+    /// The model-derived [`ReduceTactic`]. Structurally a twin of [`cursors::CursorTactic`]; only the
+    /// per-key engine differs.
+    pub struct ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+    {
+        logic: L,
+        push: P,
+        pending_keys: <B1::Cursor as Cursor>::KeyContainer,
+        pending_time: <B1::Cursor as Cursor>::TimeContainer,
+        next_pending_keys: <B1::Cursor as Cursor>::KeyContainer,
+        next_pending_time: <B1::Cursor as Cursor>::TimeContainer,
+        interesting_times: Vec<B1::Time>,
+        new_interesting_times: Vec<B1::Time>,
+        output_upper: Antichain<B1::Time>,
+        output_lower: Antichain<B1::Time>,
+        _marker: PhantomData<(B2, Bu)>,
+    }
+
+    impl<B1, B2, Bu, L, P> ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+    {
+        /// Construct a tactic that applies `logic` to each key and shapes output with `push`.
+        pub fn new(logic: L, push: P) -> Self {
+            ReferenceTactic {
+                logic,
+                push,
+                pending_keys: <B1::Cursor as Cursor>::KeyContainer::with_capacity(0),
+                pending_time: <B1::Cursor as Cursor>::TimeContainer::with_capacity(0),
+                next_pending_keys: <B1::Cursor as Cursor>::KeyContainer::with_capacity(0),
+                next_pending_time: <B1::Cursor as Cursor>::TimeContainer::with_capacity(0),
+                interesting_times: Vec::new(),
+                new_interesting_times: Vec::new(),
+                output_upper: Antichain::from_elem(<B1::Time as Timestamp>::minimum()),
+                output_lower: Antichain::from_elem(<B1::Time as Timestamp>::minimum()),
+                _marker: PhantomData,
+            }
+        }
+    }
+
+    impl<B1, B2, Bu, L, P> ReduceTactic<B1, B2> for ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+        Bu: Builder<Time = B1::Time, Output = B2, Input: Default>,
+        L: FnMut(<B1::Cursor as Cursor>::Key<'_>, &[(<B1::Cursor as Cursor>::Val<'_>, <B1::Cursor as Cursor>::Diff)], &mut Vec<(<B2::Cursor as Cursor>::ValOwn, <B2::Cursor as Cursor>::Diff)>, &mut Vec<(<B2::Cursor as Cursor>::ValOwn, <B2::Cursor as Cursor>::Diff)>),
+        P: FnMut(&mut Bu::Input, <B1::Cursor as Cursor>::Key<'_>, &mut Vec<(<B2::Cursor as Cursor>::ValOwn, B1::Time, <B2::Cursor as Cursor>::Diff)>),
+    {
+        fn retire(
+            &mut self,
+            source_batches: Vec<B1>,
+            output_batches: Vec<B2>,
+            input_batches: Vec<B1>,
+            lower: &Antichain<B1::Time>,
+            upper: &Antichain<B1::Time>,
+            held: &Antichain<B1::Time>,
+        ) -> (Vec<(B1::Time, B2)>, Antichain<B1::Time>)
+        {
+            let mut produced = Vec::new();
+
+            if held.elements().iter().any(|time| !upper.less_equal(time)) {
+
+                let (mut source_cursor, ref source_storage) = cursor_list(source_batches);
+                let (mut output_cursor, ref output_storage) = cursor_list(output_batches);
+                let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches);
+
+                let mut buffers = Vec::<(B1::Time, Vec<(<B2::Cursor as Cursor>::ValOwn, B1::Time, <B2::Cursor as Cursor>::Diff)>)>::new();
+                let mut builders = Vec::new();
+                for time in held.elements().iter() {
+                    buffers.push((time.clone(), Vec::new()));
+                    builders.push(Bu::new());
+                }
+                let mut buffer = Bu::Input::default();
+
+                // Reuseable state for performing the computation.
+                let mut thinker = ReferenceThinker::new();
+
+                let mut pending_pos = 0;
+                while batch_cursor.key_valid(batch_storage) || pending_pos < self.pending_keys.len() {
+
+                    let key1 = self.pending_keys.get(pending_pos);
+                    let key2 = batch_cursor.get_key(batch_storage);
+                    let key = match (key1, key2) {
+                        (Some(key1), Some(key2)) => ::std::cmp::min(key1, key2),
+                        (Some(key1), None)       => key1,
+                        (None, Some(key2))       => key2,
+                        (None, None)             => unreachable!(),
+                    };
+
+                    let prior_pos = pending_pos;
+                    self.interesting_times.clear();
+                    while self.pending_keys.get(pending_pos) == Some(key) {
+                        let owned_time = <B1::Cursor as Cursor>::owned_time(self.pending_time.index(pending_pos));
+                        if !upper.less_equal(&owned_time) { self.interesting_times.push(owned_time); }
+                        pending_pos += 1;
+                    }
+
+                    sort_dedup(&mut self.interesting_times);
+
+                    if batch_cursor.get_key(batch_storage) == Some(key) || !self.interesting_times.is_empty() {
+
+                        thinker.compute(
+                            key,
+                            (&mut source_cursor, source_storage),
+                            (&mut output_cursor, output_storage),
+                            (&mut batch_cursor, batch_storage),
+                            &self.interesting_times,
+                            &mut self.logic,
+                            upper,
+                            &mut buffers[..],
+                            &mut self.new_interesting_times,
+                        );
+
+                        if batch_cursor.get_key(batch_storage) == Some(key) { batch_cursor.step_key(batch_storage); }
+
+                        for pos in prior_pos .. pending_pos {
+                            let owned_time = <B1::Cursor as Cursor>::owned_time(self.pending_time.index(pos));
+                            if upper.less_equal(&owned_time) { self.new_interesting_times.push(owned_time); }
+                        }
+                        sort_dedup(&mut self.new_interesting_times);
+                        for time in self.new_interesting_times.drain(..) {
+                            self.next_pending_keys.push_ref(key);
+                            self.next_pending_time.push_own(&time);
+                        }
+
+                        for index in 0 .. buffers.len() {
+                            buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0));
+                            (self.push)(&mut buffer, key, &mut buffers[index].1);
+                            buffers[index].1.clear();
+                            builders[index].push(&mut buffer);
+
+                        }
+                    }
+                    else {
+                        for pos in prior_pos .. pending_pos {
+                            self.next_pending_keys.push_ref(self.pending_keys.index(pos));
+                            self.next_pending_time.push_ref(self.pending_time.index(pos));
+                        }
+                    }
+                }
+                drop(thinker);
+
+                self.output_lower.clear();
+                self.output_lower.extend(lower.borrow().iter().cloned());
+
+                for (index, builder) in builders.drain(..).enumerate() {
+
+                    self.output_upper.clear();
+                    self.output_upper.extend(upper.borrow().iter().cloned());
+                    for time in &held.elements()[index + 1 ..] {
+                        self.output_upper.insert_ref(time);
+                    }
+
+                    if self.output_upper.borrow() != self.output_lower.borrow() {
+
+                        let description = Description::new(self.output_lower.clone(), self.output_upper.clone(), Antichain::from_elem(<B1::Time as Timestamp>::minimum()));
+                        let batch = builder.done(description);
+
+                        produced.push((held.elements()[index].clone(), batch));
+
+                        self.output_lower.clear();
+                        self.output_lower.extend(self.output_upper.borrow().iter().cloned());
+                    }
+                }
+                assert!(self.output_upper.borrow() == upper.borrow());
+
+                self.pending_keys.clear(); std::mem::swap(&mut self.next_pending_keys, &mut self.pending_keys);
+                self.pending_time.clear(); std::mem::swap(&mut self.next_pending_time, &mut self.pending_time);
+
+                let mut frontier = Antichain::<B1::Time>::new();
+                let mut owned_time = <B1::Time as Timestamp>::minimum();
+                for pos in 0 .. self.pending_time.len() {
+                    <B1::Cursor as Cursor>::clone_time_onto(self.pending_time.index(pos), &mut owned_time);
+                    frontier.insert_ref(&owned_time);
+                }
+
+                (produced, frontier)
+            }
+            else {
+                (produced, held.clone())
+            }
+        }
+    }
+
+    /// The two-phase per-key engine.
+    ///
+    /// Phase 1 (determination) reads the input/output/seed *times* and closes them into `active`
+    /// (the interesting times) and the pended set — Model.lean's `Reached`, directly. Phase 2
+    /// (application) walks the same, still-loaded histories for *values* and emits corrections.
+    pub struct ReferenceThinker<V1, V2, V, T, D1, D2> {
+        input_history: ValueHistory<V1, T, D1>,
+        output_history: ValueHistory<V2, T, D2>,
+        batch_history: ValueHistory<V1, T, D1>,
+        input_buffer: Vec<(V1, D1)>,
+        output_buffer: Vec<(V, D2)>,
+        update_buffer: Vec<(V, D2)>,
+        output_produced: Vec<((V, T), D2)>,
+        // Phase 1 (the compacted closure): synthetic reached times still to visit, the reached times
+        // in play as join partners, scratch for the joins, and suffix-meets of the supplied times.
+        synth_times: Vec<T>,
+        times_current: Vec<T>,
+        temporary: Vec<T>,
+        meets: Vec<T>,
+        // Reusable time-only buffers for phase 1's `TimeReplay` walks. Pooled here (rather than in
+        // `ValueHistory`) so the reference tactic pays for them and the standard value walk does not.
+        batch_times: Vec<T>,
+        input_times: Vec<T>,
+        output_times: Vec<T>,
+        // The interesting (in-band reached) times, handed from phase 1 to phase 2.
+        active: Vec<T>,
+    }
+
+    impl<V1, V2, V, T, D1, D2> ReferenceThinker<V1, V2, V, T, D1, D2>
+    where
+        V1: Copy + Ord,
+        V2: Copy + Ord,
+        V: Clone + Ord,
+        T: Ord + Clone + Lattice + 'static,
+        D1: Clone + crate::difference::Semigroup,
+        D2: Clone + crate::difference::Semigroup,
+    {
+        pub fn new() -> Self {
+            ReferenceThinker {
+                input_history: ValueHistory::new(),
+                output_history: ValueHistory::new(),
+                batch_history: ValueHistory::new(),
+                input_buffer: Vec::new(),
+                output_buffer: Vec::new(),
+                update_buffer: Vec::new(),
+                output_produced: Vec::new(),
+                synth_times: Vec::new(),
+                times_current: Vec::new(),
+                temporary: Vec::new(),
+                meets: Vec::new(),
+                batch_times: Vec::new(),
+                input_times: Vec::new(),
+                output_times: Vec::new(),
+                active: Vec::new(),
+            }
+        }
+
+        #[inline(never)]
+        pub fn compute<'a, K, C1, C2, C3, L>(
+            &mut self,
+            key: K,
+            (source_cursor, source_storage): (&mut C1, &'a C1::Storage),
+            (output_cursor, output_storage): (&mut C2, &'a C2::Storage),
+            (batch_cursor, batch_storage): (&mut C3, &'a C3::Storage),
+            times: &Vec<T>,
+            logic: &mut L,
+            upper_limit: &Antichain<T>,
+            outputs: &mut [(T, Vec<(V, T, D2)>)],
+            new_interesting: &mut Vec<T>)
+        where
+            C1: Cursor<Key<'a> = K, Val<'a> = V1, Time = T, Diff = D1>,
+            C2: Cursor<Key<'a> = K, Val<'a> = V2, ValOwn = V, Time = T, Diff = D2>,
+            C3: Cursor<Key<'a> = K, Val<'a> = V1, Time = T, Diff = D1>,
+            K: Copy + Ord,
+            L: FnMut(K, &[(V1, D1)], &mut Vec<(V, D2)>, &mut Vec<(V, D2)>),
+        {
+            // ================== PHASE 1 — DETERMINATION (`Reached`, compacted) ==================
+            // The interesting times are Model.lean's `Reached` — but computed the non-quadratic way.
+            // Walk the input/output/seed times in increasing order; a time is *reached* only via a
+            // seed (a batch update, a due pending time, or a synthetic join of earlier reached times);
+            // a reached in-band time joins against the live partners to spawn more reached times, and
+            // a join beyond `upper` is pended. The live partner sets are kept an antichain by
+            // `advance_buffer_by(meet)` — coincident times collapse under the running meet — so this is
+            // the closure without the all-pairs blow-up. Time-only: `TimeReplay` reads the histories
+            // without touching values or stepping the underlying `history`, so phase 2 can walk them.
+            {
+                // Suffix-meets of the supplied (due pending) times, consumed as we pass them.
+                self.meets.clear();
+                self.meets.extend(times.iter().cloned());
+                for index in (1 .. self.meets.len()).rev() {
+                    self.meets[index-1] = self.meets[index-1].meet(&self.meets[index]);
+                }
+
+                // Build each history, then read it time-only (leaving it intact for phase 2).
+                drop(self.batch_history.replay_key(batch_cursor, batch_storage, key, None));
+                drop(self.input_history.replay_key(source_cursor, source_storage, key, None));
+                drop(self.output_history.replay_key(output_cursor, output_storage, key, None));
+                let mut batch_replay = self.batch_history.replay_times(&mut self.batch_times);
+                let mut input_replay = self.input_history.replay_times(&mut self.input_times);
+                let mut output_replay = self.output_history.replay_times(&mut self.output_times);
+
+                self.synth_times.clear();
+                self.times_current.clear();
+                self.temporary.clear();
+                self.active.clear();
+
+                let mut times_slice = &times[..];
+                let mut meets_slice = &self.meets[..];
+                let mut meet: Option<T> = None;
+
+                while let Some(next_time) = [   batch_replay.time(),
+                                                times_slice.first(),
+                                                input_replay.time(),
+                                                output_replay.time(),
+                                                self.synth_times.last(),
+                                            ].into_iter().flatten().min().cloned() {
+
+                    input_replay.step_while_time_is(&next_time);
+                    output_replay.step_while_time_is(&next_time);
+
+                    // Reached via a seed: a batch update, a due pending time, or a synthetic join.
+                    // (Input/output times alone are not reached — they are only join partners.)
+                    let mut interesting = batch_replay.step_while_time_is(&next_time);
+                    if interesting { if let Some(m) = meet.as_ref() { batch_replay.advance_buffer_by(m); } }
+                    while self.synth_times.last() == Some(&next_time) {
+                        self.times_current.push(self.synth_times.pop().unwrap());
+                        interesting = true;
+                    }
+                    while times_slice.first() == Some(&next_time) {
+                        self.times_current.push(next_time.clone());
+                        times_slice = &times_slice[1..];
+                        meets_slice = &meets_slice[1..];
+                        interesting = true;
+                    }
+                    // Absorb: a time at or above a reached time is itself reached.
+                    interesting = interesting
+                        || batch_replay.buffer().iter().any(|t| t.less_equal(&next_time))
+                        || self.times_current.iter().any(|t| t.less_equal(&next_time));
+
+                    if !upper_limit.less_equal(&next_time) {
+                        // A reached in-band time is `active`; join it against the live partners —
+                        // input/output (`joinBase`) and reached-so-far (`joinActive`) — for new times.
+                        if interesting {
+                            self.active.push(next_time.clone());
+                            if let Some(m) = meet.as_ref() { input_replay.advance_buffer_by(m); output_replay.advance_buffer_by(m); }
+                            self.temporary.extend(input_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                            self.temporary.extend(output_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        }
+                        self.temporary.extend(batch_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        self.temporary.extend(self.times_current.iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        sort_dedup(&mut self.temporary);
+
+                        let synth_len = self.synth_times.len();
+                        for time in self.temporary.drain(..) {
+                            if upper_limit.less_equal(&time) { new_interesting.push(time); }  // pended
+                            else { self.synth_times.push(time); }                             // reached, later
+                        }
+                        if self.synth_times.len() > synth_len {
+                            self.synth_times.sort_by(|x,y| y.cmp(x));
+                            self.synth_times.dedup();
+                        }
+                    }
+                    else if interesting {
+                        new_interesting.push(next_time.clone());
+                    }
+
+                    // Running meet (a lower bound on every time still to visit); compact the reached
+                    // partners `times_current` with it.
+                    meet = None;
+                    update_meet(&mut meet, batch_replay.meet());
+                    update_meet(&mut meet, input_replay.meet());
+                    update_meet(&mut meet, output_replay.meet());
+                    for time in self.synth_times.iter() { update_meet(&mut meet, Some(time)); }
+                    update_meet(&mut meet, meets_slice.first());
+                    if let Some(m) = meet.as_ref() {
+                        for time in self.times_current.iter_mut() { *time = time.join(m); }
+                    }
+                    sort_dedup(&mut self.times_current);
+                }
+
+                sort_dedup(&mut self.active);
+            }
+
+            // ===================== PHASE 2 — APPLICATION (`emit_correct`) =====================
+            // Walk `self.active` in order over the SAME per-key edits (via `replay`, no cursor), keep
+            // the input/output accumulations tight by advancing to the meet of the times still to be
+            // produced, apply `logic`, and emit corrections.
+            {
+                self.meets.clear();
+                self.meets.extend(self.active.iter().cloned());
+                for index in (1 .. self.meets.len()).rev() {
+                    self.meets[index-1] = self.meets[index-1].meet(&self.meets[index]);
+                }
+
+                // Walk the histories loaded (and left intact) by phase 1 — no cursor re-read, no
+                // rebuild, no re-sort; just a fresh walk of the same sorted `history` for values.
+                let mut batch_replay = self.batch_history.walk();
+                let mut input_replay = self.input_history.walk();
+                let mut output_replay = self.output_history.walk();
+
+                self.output_produced.clear();
+
+                for index in 0 .. self.active.len() {
+                    let next_time = self.active[index].clone();
+                    let meet = self.meets[index].clone();
+
+                    // Phase 2 visits only the active times, so at each we must catch up the histories to
+                    // include every edit that will contribute to the accumulation at `next_time` (edits
+                    // at non-active times count too). `history` is sorted by the total `Ord` and `step`
+                    // pops the least, so we step the `Ord`-prefix `t <= next_time`, NOT the partial
+                    // `t.less_equal(next_time)`: the partial order interleaves with the sort, so an
+                    // `Ord`-earlier time incomparable to `next_time` would halt a `less_equal` walk early
+                    // and strand later edits that *are* `less_equal(next_time)`. Stepping the `Ord`-prefix
+                    // takes a superset; the `less_equal` filter below then selects the true `<= next_time`
+                    // edits. (Assembling with `less_equal` while stepping with `<=` is what the value-aware
+                    // cursor does via its full time-order frontier walk.)
+                    while input_replay.time().map_or(false, |t| *t <= next_time) { input_replay.step(); }
+                    while batch_replay.time().map_or(false, |t| *t <= next_time) { batch_replay.step(); }
+                    while output_replay.time().map_or(false, |t| *t <= next_time) { output_replay.step(); }
+                    input_replay.advance_buffer_by(&meet);
+                    batch_replay.advance_buffer_by(&meet);
+                    output_replay.advance_buffer_by(&meet);
+
+                    debug_assert!(self.input_buffer.is_empty());
+                    for ((value, time), diff) in input_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
+                    }
+                    for ((value, time), diff) in batch_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
+                    }
+                    crate::consolidation::consolidate(&mut self.input_buffer);
+
+                    for ((value, time), diff) in output_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.output_buffer.push((C2::owned_val(*value), diff.clone())); }
+                    }
+                    for ((value, time), diff) in self.output_produced.iter() {
+                        if time.less_equal(&next_time) { self.output_buffer.push((value.clone(), diff.clone())); }
+                    }
+                    crate::consolidation::consolidate(&mut self.output_buffer);
+
+                    if !self.input_buffer.is_empty() || !self.output_buffer.is_empty() {
+                        logic(key, &self.input_buffer[..], &mut self.output_buffer, &mut self.update_buffer);
+                        self.input_buffer.clear();
+                        self.output_buffer.clear();
+
+                        crate::consolidation::consolidate(&mut self.update_buffer);
+                        if !self.update_buffer.is_empty() {
+
+                            let idx = outputs.iter().rev().position(|(time, _)| time.less_equal(&next_time));
+                            let idx = outputs.len() - idx.expect("failed to find index") - 1;
+                            for (val, diff) in self.update_buffer.drain(..) {
+                                self.output_produced.push(((val.clone(), next_time.clone()), diff.clone()));
+                                outputs[idx].1.push((val, next_time.clone(), diff));
+                            }
+
+                            for entry in &mut self.output_produced { (entry.0).1.join_assign(&meet); }
+                            crate::consolidation::consolidate(&mut self.output_produced);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
diff --git a/differential-dataflow/tests/reduce_reference.rs b/differential-dataflow/tests/reduce_reference.rs
new file mode 100644
index 000000000..6046660ae
--- /dev/null
+++ b/differential-dataflow/tests/reduce_reference.rs
@@ -0,0 +1,409 @@
+//! Test harness for the model-derived `reference` reduce tactic. Four things:
+//!
+//! * `differential_*` — run the default (cursor) reduce and the `reference` reduce on the SAME random
+//!   input and `assert_eq` their outputs at every time. The drift detector: if either tactic changes
+//!   behavior, it fails. Inputs use `Pair<u64,u64>` (product / partial order, so joins and synthetic
+//!   times occur) with diffs ±1 (so updates cancel — the corner where the value-blind reference could
+//!   over-derive).
+//!
+//! * `oracle_*` — run one tactic and check its output against a from-scratch recomputation of
+//!   `f(accumulated input)` at many times. The correctness check: it does not rely on the two tactics
+//!   agreeing, so it catches bugs in code they share (the driver, `ValueHistory`, …). It is the
+//!   executable form of `formal/Differential/Model.lean`'s stream-correctness statement.
+//!
+//! * `bfs_*` — an iterative BFS (product time via iteration; a real computation shape) computed both
+//!   ways over a graph that grows and shrinks; another differential check.
+//!
+//! * `scc_*` — strongly connected components: two nested product-timed fixpoints (edge trimming around
+//!   iterative label propagation), so the reduce runs at three nested time coordinates. This depth is
+//!   what exposed the phase-2 catch-up walk stepping the partial-order `less_equal` set instead of the
+//!   total-`Ord` prefix (see the note at that walk in `operators/reduce.rs`); the two-coordinate shapes
+//!   above never reached it.
+
+use std::collections::BTreeMap;
+
+use rand::{Rng, SeedableRng, StdRng};
+
+use timely::PartialOrder;
+use timely::dataflow::operators::capture::{Capture, Extract};
+use timely::dataflow::operators::vec::unordered_input::UnorderedInput;
+
+use differential_dataflow::{AsCollection, Data, ExchangeData, Hashable};
+use differential_dataflow::VecCollection;
+use differential_dataflow::difference::{Abelian, Semigroup};
+use differential_dataflow::input::Input;
+use differential_dataflow::lattice::Lattice;
+use differential_dataflow::operators::*;
+use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::operators::reduce::reduce_trace_reference;
+use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};
+
+use pair::Pair;
+type Time = Pair<u64, u64>;
+
+// ---- the reduce logics under test (plain `fn`s, so they are `Copy` and can feed both tactics) ----
+
+fn count(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    output.push((input.len() as i64, 1));
+}
+fn sum(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    let s: i64 = input.iter().map(|(v, d)| **v as i64 * *d as i64).sum();
+    output.push((s, 1));
+}
+fn min(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    if let Some(m) = input.iter().map(|(v, _)| **v).min() {
+        output.push((m as i64, 1));
+    }
+}
+
+type Logic = fn(&u64, &[(&u64, isize)], &mut Vec<(i64, isize)>);
+
+// Stand-alone equivalent of the default `reduce`, driven by the model-derived reference tactic. The
+// library exposes only the low-level, doc-hidden `reduce_trace_reference`; this helper supplies the
+// same arrange / abelian-negate / as_collection glue `reduce_named` uses, so the tests compare the
+// two tactics without the reference living on the public `Collection`/`Arranged` API.
+fn reduce_reference<'scope, Ti, K, V, R, V2, R2, L>(
+    collection: VecCollection<'scope, Ti, (K, V), R>,
+    mut logic: L,
+) -> VecCollection<'scope, Ti, (K, V2), R2>
+where
+    Ti: timely::progress::Timestamp + Lattice + Ord,
+    K: ExchangeData + Hashable,
+    V: ExchangeData,
+    R: ExchangeData + Semigroup,
+    V2: Data,
+    R2: Ord + Abelian + 'static,
+    L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
+{
+    // Bind the input arrangement to an explicitly-typed local so `Tr1` is concrete going into
+    // `reduce_trace_reference` (otherwise its output-key bound is checked before the argument pins it).
+    let arranged: Arranged<'scope, TraceAgent<ValSpine<K, V, Ti, R>>> =
+        collection.arrange_by_key_named("Arrange: ReduceReference");
+
+    // The closure params are annotated to the concrete `&K` / `&V` / owned-output types (rather than
+    // left to infer the cursor's `Key<'_>` projection) so the coercion to the user's `&K` logic
+    // normalizes here, exactly as the default `reduce_named` gets for free by passing its logic through.
+    reduce_trace_reference::<TraceAgent<ValSpine<K, V, Ti, R>>, ValBuilder<K, V2, Ti, R2>, ValSpine<K, V2, Ti, R2>, _, _>(
+        arranged,
+        "ReduceReference",
+        move |key: &K, input: &[(&V, R)], output: &mut Vec<(V2, R2)>, change: &mut Vec<(V2, R2)>| {
+            if !input.is_empty() { logic(key, input, change); }
+            change.extend(output.drain(..).map(|(x, mut d)| { d.negate(); (x, d) }));
+            differential_dataflow::consolidation::consolidate(change);
+        },
+        |vec, key: &K, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
+    )
+    .as_collection(|k, v| (k.clone(), v.clone()))
+}
+
+// A random input: `((key, val), Pair(a, b), diff)`, diffs ±1 so updates cancel.
+fn random_input(seed: usize, n: usize, keys: u64, vals: u64, span: u64) -> Vec<((u64, u64), Time, isize)> {
+    let s: &[usize] = &[seed, 7, 13, 29];
+    let mut rng: StdRng = SeedableRng::from_seed(s);
+    (0..n).map(|_| {
+        let k = rng.gen_range(0, keys);
+        let v = rng.gen_range(0, vals);
+        let t = Pair::new(rng.gen_range(0, span), rng.gen_range(0, span));
+        let d = if rng.gen::<bool>() { 1 } else { -1 };
+        ((k, v), t, d)
+    }).collect()
+}
+
+// ======================= differential: cursor vs reference, all times =======================
+
+#[test] fn differential_count() { for s in 0..25 { differential(s, count); } }
+#[test] fn differential_sum()   { for s in 0..25 { differential(s, sum); } }
+#[test] fn differential_min()   { for s in 0..25 { differential(s, min); } }
+
+fn differential(seed: usize, logic: Logic) {
+    let span = 6u64;
+    let updates = random_input(seed, 400, 12, 8, span);
+    timely::execute_directly(move |worker| {
+        let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
+            let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
+            let collection = stream.as_collection();
+            let via_cursor    = collection.clone().reduce(logic);
+            let via_reference = reduce_reference(collection, logic);
+            via_cursor.assert_eq(via_reference);
+            (input, capability)
+        });
+        for u in updates { input.activate().session(&capability).give(u); }
+        drop(capability); // release the input frontier so everything finalizes
+    });
+}
+
+// ======================= oracle: a tactic vs from-scratch f(acc input) =======================
+
+#[test] fn oracle_cursor_count()    { for s in 0..15 { oracle(s, false, count); } }
+#[test] fn oracle_reference_count() { for s in 0..15 { oracle(s, true,  count); } }
+#[test] fn oracle_reference_sum()   { for s in 0..15 { oracle(s, true,  sum); } }
+#[test] fn oracle_reference_min()   { for s in 0..15 { oracle(s, true,  min); } }
+
+fn oracle(seed: usize, reference: bool, logic: Logic) {
+    let span = 6u64;
+    let updates = random_input(seed, 300, 10, 8, span);
+
+    // Capture the reduce output as `((key, out_val), time, diff)` updates.
+    let (send, recv) = std::sync::mpsc::channel();
+    let send = std::sync::Arc::new(std::sync::Mutex::new(send));
+    let drive = updates.clone();
+    timely::execute_directly(move |worker| {
+        let send = send.lock().unwrap().clone();
+        let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
+            let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
+            let collection = stream.as_collection();
+            let out = if reference { reduce_reference(collection, logic) } else { collection.reduce(logic) };
+            out.inner.capture_into(send);
+            (input, capability)
+        });
+        for u in drive { input.activate().session(&capability).give(u); }
+        drop(capability);
+    });
+    let output: Vec<((u64, i64), Time, isize)> =
+        recv.extract().into_iter().flat_map(|(_, batch)| batch).collect();
+
+    // Everything is finalized, so `acc(output, q) == f(acc(input, q))` must hold at every `q`. Check
+    // at a rich set: all input times, all output times, and their pairwise joins (where reduce's
+    // output can change).
+    let mut times: Vec<Time> = Vec::new();
+    times.extend(updates.iter().map(|(_, t, _)| t.clone()));
+    times.extend(output.iter().map(|(_, t, _)| t.clone()));
+    let base = times.clone();
+    for a in &base { for b in &base { times.push(a.join(b)); } }
+    times.sort(); times.dedup();
+
+    for q in &times {
+        // expected: per key, f applied to the accumulated input at `q`.
+        let mut expected: BTreeMap<(u64, i64), i64> = BTreeMap::new();
+        let mut per_key: BTreeMap<u64, BTreeMap<u64, isize>> = BTreeMap::new();
+        for ((k, v), t, d) in &updates {
+            if t.less_equal(q) { *per_key.entry(*k).or_default().entry(*v).or_default() += *d; }
+        }
+        for (k, vals) in &per_key {
+            let acc: Vec<(&u64, isize)> = vals.iter().filter(|(_, d)| **d != 0).map(|(v, d)| (v, *d)).collect();
+            if acc.is_empty() { continue; }
+            let mut out = Vec::new();
+            logic(k, &acc, &mut out);
+            for (ov, od) in out { *expected.entry((*k, ov)).or_default() += od as i64; }
+        }
+        expected.retain(|_, d| *d != 0);
+
+        // actual: the accumulated reduce output at `q`.
+        let mut actual: BTreeMap<(u64, i64), i64> = BTreeMap::new();
+        for ((k, ov), t, d) in &output {
+            if t.less_equal(q) { *actual.entry((*k, *ov)).or_default() += *d as i64; }
+        }
+        actual.retain(|_, d| *d != 0);
+
+        assert_eq!(expected, actual, "seed {seed} reference {reference} at time {q:?}");
+    }
+}
+
+// ======================= iterative BFS differential (a real shape) =======================
+
+#[test] fn bfs_tiny_a()   { bfs_differential(6, 10, 4, &[1, 2, 3, 4]); }
+#[test] fn bfs_tiny_b()   { bfs_differential(5, 8, 5, &[5, 6, 7, 8]); }
+#[test] fn bfs_30_60_20() { bfs_differential(30, 60, 20, &[1, 2, 3, 4]); }
+#[test] fn bfs_15_40_30() { bfs_differential(15, 40, 30, &[5, 6, 7, 8]); }
+#[test] fn bfs_sweep() {
+    for s in 0u8 .. 12 { bfs_differential(40, 120, 30, &[s as usize, (s as usize) + 7, 13, 29]); }
+    for s in 0u8 .. 6  { bfs_differential(60, 200, 25, &[3, (s as usize) + 1, (s as usize) * 5 + 2, 11]); }
+}
+
+type Node = usize;
+type Edge = (Node, Node);
+
+fn bfs_differential(nodes: usize, edges: usize, rounds: usize, seed: &[usize]) {
+    let mut rng1: StdRng = SeedableRng::from_seed(seed);
+    let mut rng2: StdRng = SeedableRng::from_seed(seed);
+    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
+    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
+    for round in 1 .. rounds {
+        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
+        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
+    }
+    let root_list = vec![(1usize, 0usize, 1isize)];
+
+    timely::execute_directly(move |worker| {
+        let mut roots_list = root_list.clone();
+        let mut edges_list = edge_list.clone();
+        let (mut roots, mut edges) = worker.dataflow::<usize, _, _>(|scope| {
+            let (root_input, roots) = scope.new_collection();
+            let (edge_input, edges) = scope.new_collection();
+            let via_cursor    = bfs(edges.clone(), roots.clone(), false);
+            let via_reference = bfs(edges, roots, true);
+            via_cursor.assert_eq(via_reference);
+            (root_input, edge_input)
+        });
+        roots_list.sort_by(|x, y| y.1.cmp(&x.1));
+        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
+        let mut round = 0;
+        while !roots_list.is_empty() || !edges_list.is_empty() {
+            while roots_list.last().map(|x| x.1) == Some(round) { let (n, _, d) = roots_list.pop().unwrap(); roots.update(n, d); }
+            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
+            round += 1;
+            roots.advance_to(round);
+            edges.advance_to(round);
+        }
+    });
+}
+
+fn bfs<'scope, T>(edges: VecCollection<'scope, T, Edge>, roots: VecCollection<'scope, T, Node>, reference: bool) -> VecCollection<'scope, T, (Node, usize)>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let nodes = roots.map(|x| (x, 0));
+    nodes.clone().iterate(|scope, inner| {
+        let edges = edges.enter(scope);
+        let nodes = nodes.enter(scope);
+        let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
+        if reference {
+            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
+        } else {
+            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
+        }
+    })
+}
+
+// ===================== iterative SCC differential (a deeper shape) =====================
+// SCC nests two Product-timed fixpoints: an outer edge-trimming loop and, inside each trim, an
+// iterative label propagation whose `reduce` is the tactic under test. The doubly-nested product
+// time produces far more synthetic interesting times than BFS, over joins / concats / filters.
+
+#[test] fn scc_tiny_a() { scc_differential(6, 12, 4, &[1, 2, 3, 4]); }
+#[test] fn scc_tiny_b() { scc_differential(8, 16, 4, &[5, 6, 7, 8]); }
+#[test] fn scc_12_24()  { scc_differential(12, 24, 5, &[2, 4, 6, 8]); }
+#[test] fn scc_sweep()  { for s in 0u8 .. 6 { scc_differential(15, 30, 5, &[s as usize, (s as usize) + 7, 13, 29]); } }
+
+fn scc_differential(nodes: usize, edges: usize, rounds: usize, seed: &[usize]) {
+    let mut rng1: StdRng = SeedableRng::from_seed(seed);
+    let mut rng2: StdRng = SeedableRng::from_seed(seed);
+    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
+    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
+    for round in 1 .. rounds {
+        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
+        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
+    }
+
+    timely::execute_directly(move |worker| {
+        let mut edges_list = edge_list.clone();
+        let mut edges = worker.dataflow::<usize, _, _>(|scope| {
+            let (edge_input, edges) = scope.new_collection();
+            let via_cursor    = scc(edges.clone(), false);
+            let via_reference = scc(edges, true);
+            via_cursor.assert_eq(via_reference);
+            edge_input
+        });
+        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
+        let mut round = 0;
+        while !edges_list.is_empty() {
+            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
+            round += 1;
+            edges.advance_to(round);
+        }
+    });
+}
+
+// The subset of edges within a strongly connected component, mirroring
+// `algorithms::graphs::scc::strongly_connected` but with the propagation `reduce` swappable.
+fn scc<'scope, T>(graph: VecCollection<'scope, T, Edge>, reference: bool) -> VecCollection<'scope, T, Edge>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let graph2 = graph.clone();
+    graph.iterate(move |scope, inner| {
+        let edges = graph2.enter(scope);
+        let trans = edges.clone().map(|(a, b)| (b, a));
+        let trimmed = trim_edges(inner, edges, reference);
+        trim_edges(trimmed, trans, reference)
+    })
+}
+
+// Keep only edges whose endpoints propagate to the same label; run twice (forward then transposed)
+// this converges on the SCC edges.
+fn trim_edges<'scope, T>(cycle: VecCollection<'scope, T, Edge>, edges: VecCollection<'scope, T, Edge>, reference: bool) -> VecCollection<'scope, T, Edge>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let nodes = edges.clone().map(|(_a, b)| (b, b));
+    let labels = propagate(cycle, nodes, reference);
+    edges.join_map(labels.clone(), |&src, &dst, &l1| (dst, (src, l1)))
+         .join_map(labels, |&dst, &(src, l1), &l2| ((src, dst), (l1, l2)))
+         .filter(|(_, (l1, l2))| l1 == l2)
+         .map(|((src, dst), _)| (dst, src))
+}
+
+// Forward min-label propagation along `edges` from per-node seed labels; the inner `reduce` (min
+// label per node) is the tactic under test.
+fn propagate<'scope, T>(edges: VecCollection<'scope, T, Edge>, nodes: VecCollection<'scope, T, (Node, Node)>, reference: bool) -> VecCollection<'scope, T, (Node, Node)>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let seed = nodes.clone();
+    nodes.filter(|_| false).iterate(move |scope, inner| {
+        let edges = edges.enter(scope);
+        let seed = seed.enter(scope);
+        let combined = inner.join_map(edges, |&_n, &l, &d| (d, l)).concat(seed);
+        if reference {
+            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
+        } else {
+            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
+        }
+    })
+}
+
+/// A minimal product-order (partially ordered) timestamp, so the harness exercises the synthetic
+/// interesting times determination is about. Copied from `examples/multitemporal.rs`.
+mod pair {
+
+    #[derive(Hash, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
+    pub struct Pair<S, T> { pub first: S, pub second: T }
+
+    impl<S, T> Pair<S, T> {
+        pub fn new(first: S, second: T) -> Self { Pair { first, second } }
+    }
+
+    use timely::order::PartialOrder;
+    impl<S: PartialOrder, T: PartialOrder> PartialOrder for Pair<S, T> {
+        fn less_equal(&self, other: &Self) -> bool {
+            self.first.less_equal(&other.first) && self.second.less_equal(&other.second)
+        }
+    }
+
+    use timely::progress::timestamp::Refines;
+    impl<S: Timestamp, T: Timestamp> Refines<()> for Pair<S, T> {
+        fn to_inner(_outer: ()) -> Self { Self::minimum() }
+        fn to_outer(self) -> () { () }
+        fn summarize(_summary: <Self>::Summary) -> () { () }
+    }
+
+    use timely::progress::PathSummary;
+    impl<S: Timestamp, T: Timestamp> PathSummary<Pair<S,T>> for () {
+        fn results_in(&self, timestamp: &Pair<S, T>) -> Option<Pair<S,T>> { Some(timestamp.clone()) }
+        fn followed_by(&self, other: &Self) -> Option<Self> { Some(other.clone()) }
+    }
+
+    use timely::progress::Timestamp;
+    impl<S: Timestamp, T: Timestamp> Timestamp for Pair<S, T> {
+        fn minimum() -> Self { Pair { first: S::minimum(), second: T::minimum() }}
+        type Summary = ();
+    }
+
+    use differential_dataflow::lattice::Lattice;
+    impl<S: Lattice, T: Lattice> Lattice for Pair<S, T> {
+        fn join(&self, other: &Self) -> Self {
+            Pair { first: self.first.join(&other.first), second: self.second.join(&other.second) }
+        }
+        fn meet(&self, other: &Self) -> Self {
+            Pair { first: self.first.meet(&other.first), second: self.second.meet(&other.second) }
+        }
+    }
+
+    use std::fmt::{Formatter, Error, Debug};
+    use serde::{Deserialize, Serialize};
+    impl<TOuter: Debug, TInner: Debug> Debug for Pair<TOuter, TInner> {
+        fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+            f.write_str(&format!("({:?}, {:?})", self.first, self.second))
+        }
+    }
+}