diff --git a/differential-dataflow/src/operators/int_proxy/history.rs b/differential-dataflow/src/operators/int_proxy/history.rs
new file mode 100644
index 000000000..c2a8febe4
--- /dev/null
+++ b/differential-dataflow/src/operators/int_proxy/history.rs
@@ -0,0 +1,77 @@
+//! Time-ordered replay of proxy update histories, with meet-advancement.
+
+use crate::lattice::Lattice;
+
+/// A value history suitable for integer proxy values.
+pub(in crate::operators) type IdHistory<T, R> = crate::operators::ValueHistory<u64, T, R>;
+
+/// A value history minus the values (and diffs).
+///
+/// This type mirrors ValueHistory, but traverses only the times and does not watch for
+/// cancelation of data (and their times). We may learn that that pattern is preferred,
+/// but for the moment the int-proxy tactic determines all interesting times before any
+/// collection manipulation (and potential cancelation) occurs.
+pub(crate) struct TimeHistory<T> {
+    /// Un-replayed `(time, meet)`, sorted descending by time so popping replays ascending;
+    /// `meet` is the meet of this time with all times later in the replay.
+    history: Vec<(T, T)>,
+    /// Stepped-in times, advanced and deduplicated, sorted ascending.
+    buffer: Vec<T>,
+}
+
+impl<T: Lattice + Clone + Ord> TimeHistory<T> {
+    pub fn new() -> Self { TimeHistory { history: Vec::new(), buffer: Vec::new() } }
+
+    /// Load `times`, advancing each by `advance_by` if supplied, and organize the replay
+    /// (sort + suffix meets).
+    pub fn load(&mut self, times: impl Iterator<Item = T>, advance_by: Option<&T>) {
+        self.history.clear();
+        self.buffer.clear();
+        for mut time in times {
+            if let Some(m) = advance_by {
+                time = time.join(m);
+            }
+            self.history.push((time.clone(), time));
+        }
+        self.history.sort_by(|x, y| y.0.cmp(&x.0));
+        self.history.iter_mut().reduce(|prev, cur| {
+            cur.1.meet_assign(&prev.1);
+            cur
+        });
+    }
+
+    /// The next (least) un-replayed time.
+    pub fn time(&self) -> Option<&T> {
+        self.history.last().map(|x| &x.0)
+    }
+    /// The meet of all un-replayed times.
+    pub fn meet(&self) -> Option<&T> {
+        self.history.last().map(|x| &x.1)
+    }
+
+    /// Step times while the next equals `time`; true iff any did.
+    pub fn step_while_time_is(&mut self, time: &T) -> bool {
+        let mut found = false;
+        while self.time() == Some(time) {
+            found = true;
+            let (t, _) = self.history.pop().unwrap();
+            self.buffer.push(t);
+        }
+        found
+    }
+
+    /// Advance buffered times by `meet` and deduplicate — the collapse that keeps replay
+    /// linear.
+    pub fn advance_buffer_by(&mut self, meet: &T) {
+        for time in self.buffer.iter_mut() {
+            *time = time.join(meet);
+        }
+        self.buffer.sort();
+        self.buffer.dedup();
+    }
+
+    /// The buffered (stepped-in, advanced) times.
+    pub fn buffer(&self) -> &[T] {
+        &self.buffer
+    }
+}
diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
new file mode 100644
index 000000000..dbda06db0
--- /dev/null
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -0,0 +1,258 @@
+//! The proxy join framework.
+//!
+//! A conventional differential join against `(u64, u64)` values, which are provided by
+//! and then interpreted by a backend, who is relieved of lattice-time reasoning.
+
+use timely::progress::{Antichain, Timestamp};
+use timely::progress::frontier::AntichainRef;
+
+use crate::difference::{Multiply, Semigroup};
+use crate::lattice::Lattice;
+use crate::trace::BatchReader;
+use super::ProxyBridge;
+use crate::operators::join::{Fresh, JoinTactic};
+
+use super::history::IdHistory;
+
+/// A unit of proxied join work, presented to the backend.
+pub struct JoinInstance<'a, B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
+    /// The first input's batches.
+    pub batches0: &'a [B0],
+    /// The second input's batches.
+    pub batches1: &'a [B1],
+    /// The compaction frontier for loading (the unit's capability time).
+    pub lower: AntichainRef<'a, B0::Time>,
+}
+
+/// A type that can interpret and retire pairs of batches, joined by key hashes.
+///
+/// The protocol invokes the `present*` methods with the instance to produce the proxy collection,
+/// and then returns with any number (including zero) calls to `cross` to produce outputs.
+pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
+    /// Diff type presented for the first input.
+    type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
+    /// Diff type presented for the second input.
+    type R1: Semigroup;
+    /// Diff type of matched records (`R0 * R1`), computed by the tactic.
+    type ROut: Semigroup;
+    /// The output container built from matched value ids.
+    type Output;
+
+    /// Prepare a proxy bridge from the instance's first input, optionally key-hash restricted.
+    ///
+    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
+    fn present0(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R0>;
+    /// Prepare a proxy bridge from the instance's second input, optionally key-hash restricted.
+    ///
+    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
+    fn present1(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R1>;
+    /// From a list of left and right identifiers, and corresponding times and diffs, the output.
+    fn cross(&mut self, instance: &JoinInstance<'_, B0, B1>, left: &[(u64, u64)], right: &[(u64, u64)], times: Vec<B0::Time>, diffs: Vec<Self::ROut>) -> Self::Output;
+}
+
+/// A proxy-space [`JoinTactic`]: matches records of the two presented runs by `key_hash`,
+pub struct ProxyJoinTactic<B0, B1, Bk> {
+    backend: Bk,
+    _marker: std::marker::PhantomData<(B0, B1)>,
+}
+
+impl<B0, B1, Bk> ProxyJoinTactic<B0, B1, Bk> {
+    /// A join tactic deferring all value semantics to `backend`.
+    pub fn new(backend: Bk) -> Self {
+        ProxyJoinTactic { backend, _marker: std::marker::PhantomData }
+    }
+}
+
+impl<B0, B1, Bk> JoinTactic<B0, B1, Bk::Output> for ProxyJoinTactic<B0, B1, Bk>
+where
+    B0: BatchReader,
+    B1: BatchReader<Time = B0::Time>,
+    Bk: ProxyJoinBackend<B0, B1>,
+    Bk::Output: 'static,
+{
+    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = Bk::Output>> {
+        Box::new(join_prep(&mut self.backend, input0, input1, fresh, meet).into_iter())
+    }
+}
+
+/// Update count threshold used to return to the backend with join output to instantiate.
+const JOIN_CHUNK: usize = 1 << 20;
+
+/// Prepare one join unit: present both sides, merge-match key runs over the sorted hashes, and
+/// cross-product matched records into a sequence of output containers — one per ~[`JOIN_CHUNK`]
+/// matches, flushed wherever the batch fills (including mid-key; parity: their concatenation is the
+/// single-container output). `meet` is a lower bound on the fresh side's times, so the accumulated
+/// side loads compacted (see [`JoinInstance`]); every output is produced under the capability at
+/// `meet`, so advancing loaded times by it leaves the output unchanged.
+fn join_prep<B0, B1, Bk>(backend: &mut Bk, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Vec<Bk::Output>
+where
+    B0: BatchReader,
+    B1: BatchReader<Time = B0::Time>,
+    Bk: ProxyJoinBackend<B0, B1>,
+{
+    let lower = Antichain::from_elem(meet);
+    let instance = JoinInstance { batches0: &input0, batches1: &input1, lower: lower.borrow() };
+
+    // Prepare work using the keys of the just-received side.
+    // FIXME: Use the smaller of the two instead.
+    let (p0, p1) = match fresh {
+        Fresh::Input0 => {
+            let p0 = backend.present0(&instance, None);
+            if p0.is_empty() { return Vec::new(); }
+            let mut keys: Vec<u64> = p0.iter().map(|r| r.0.0).collect();
+            keys.dedup();
+            let p1 = backend.present1(&instance, Some(&keys));
+            (p0, p1)
+        }
+        Fresh::Input1 => {
+            let p1 = backend.present1(&instance, None);
+            if p1.is_empty() { return Vec::new(); }
+            let mut keys: Vec<u64> = p1.iter().map(|r| r.0.0).collect();
+            keys.dedup();
+            let p0 = backend.present0(&instance, Some(&keys));
+            (p0, p1)
+        }
+    };
+    if p0.is_empty() || p1.is_empty() { return Vec::new(); }
+    super::debug_assert_sorted_bridge(&p0, "present0");
+    super::debug_assert_sorted_bridge(&p1, "present1");
+
+    // Merge-join the two presented runs on `key_hash` (both sorted) and cross matched pairs. `flush`
+    // ships the accumulated batch as a container and resets the buffers; `join_key` calls it at each
+    // `JOIN_CHUNK` boundary — including *within* a high-fanout key's wave — so no single key
+    // materializes more than a chunk of its cross product. The closure scope releases its
+    // `backend`/`out` borrow before `out` is returned.
+    let mut out: Vec<Bk::Output> = Vec::new();
+    {
+        let mut flush = |li: &mut Vec<(u64, u64)>, ri: &mut Vec<(u64, u64)>, ot: &mut Vec<B0::Time>, od: &mut Vec<Bk::ROut>| {
+            out.push(backend.cross(&instance, li.as_slice(), ri.as_slice(), std::mem::take(ot), std::mem::take(od)));
+            li.clear();
+            ri.clear();
+        };
+
+        let (mut li, mut ri) = (Vec::new(), Vec::new());
+        let (mut ot, mut od) = (Vec::new(), Vec::new());
+        // Per-key replay histories, held across the unit and reloaded per key (only the >=16/>=16
+        // replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
+        let mut h0 = IdHistory::new();
+        let mut h1 = IdHistory::new();
+        let (mut i, mut j) = (0usize, 0usize);
+        while i < p0.len() && j < p1.len() {
+            let (ki, kj) = (p0[i].0.0, p1[j].0.0);
+            if ki < kj {
+                i += 1;
+            } else if kj < ki {
+                j += 1;
+            } else {
+                let mut e0 = i;
+                while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
+                let mut e1 = j;
+                while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
+                join_key(ki, &p0, i..e0, &p1, j..e1, &mut h0, &mut h1, &mut li, &mut ri, &mut ot, &mut od, &mut flush);
+                i = e0;
+                j = e1;
+            }
+        }
+        if !li.is_empty() {
+            flush(&mut li, &mut ri, &mut ot, &mut od);
+        }
+    }
+    out
+}
+
+/// Match one key's records across the two presented runs.
+///
+/// If either history is small, this performs a simple cross product.
+/// If both histories are large, this replays the histories compacting as it goes in
+/// order to (potentially) avoid quadratic blow-up.
+#[allow(clippy::too_many_arguments)]
+fn join_key<T, R0, R1, RO, F>(
+    kh: u64,
+    p0: &ProxyBridge<T, R0>,
+    r0: std::ops::Range<usize>,
+    p1: &ProxyBridge<T, R1>,
+    r1: std::ops::Range<usize>,
+    h0: &mut IdHistory<T, R0>,
+    h1: &mut IdHistory<T, R1>,
+    li: &mut Vec<(u64, u64)>,
+    ri: &mut Vec<(u64, u64)>,
+    ot: &mut Vec<T>,
+    od: &mut Vec<RO>,
+    flush: &mut F,
+) where
+    T: Lattice + Timestamp,
+    R0: Semigroup + Multiply<R1, Output = RO> + Clone,
+    R1: Semigroup + Clone,
+    F: FnMut(&mut Vec<(u64, u64)>, &mut Vec<(u64, u64)>, &mut Vec<T>, &mut Vec<RO>),
+{
+    // `flush` ships the accumulated batch once it reaches `JOIN_CHUNK` and resets the buffers. It's
+    // checked after every produced pair — *including inside the replay wave below* — so even a key
+    // whose fanout dwarfs `JOIN_CHUNK` never materializes more than a chunk of its cross product.
+    if r0.len() < 16 || r1.len() < 16 {
+        for a in r0 {
+            for b in r1.clone() {
+                li.push((kh, p0[a].0.1));
+                ri.push((kh, p1[b].0.1));
+                ot.push(p0[a].1.join(&p1[b].1));
+                od.push(p0[a].2.clone().multiply(&p1[b].2));
+                if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
+            }
+        }
+        return;
+    }
+
+    // Reusable replay scratch, reloaded per key (`load_iter` clears + rebuilds, keeping capacity);
+    // the caller holds `h0`/`h1` across the unit so a high-fanout join allocates no per-key history.
+    h0.load_iter(r0.map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
+    h1.load_iter(r1.map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
+
+    while h0.time().is_some() && h1.time().is_some() {
+        if h0.time().unwrap() < h1.time().unwrap() {
+            h1.advance_buffer_by(h0.meet().unwrap());
+            let (v0, t0, d0) = h0.edit().unwrap();
+            for ((v1, t1), d1) in h1.buffer() {
+                li.push((kh, v0));
+                ri.push((kh, *v1));
+                ot.push(t0.join(t1));
+                od.push(d0.clone().multiply(d1));
+                if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
+            }
+            h0.step();
+        } else {
+            h0.advance_buffer_by(h1.meet().unwrap());
+            let (v1, t1, d1) = h1.edit().unwrap();
+            for ((v0, t0), d0) in h0.buffer() {
+                li.push((kh, *v0));
+                ri.push((kh, v1));
+                ot.push(t0.join(t1));
+                od.push(d0.clone().multiply(d1));
+                if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
+            }
+            h1.step();
+        }
+    }
+    while h0.time().is_some() {
+        h1.advance_buffer_by(h0.meet().unwrap());
+        let (v0, t0, d0) = h0.edit().unwrap();
+        for ((v1, t1), d1) in h1.buffer() {
+            li.push((kh, v0));
+            ri.push((kh, *v1));
+            ot.push(t0.join(t1));
+            od.push(d0.clone().multiply(d1));
+            if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
+        }
+        h0.step();
+    }
+    while h1.time().is_some() {
+        h0.advance_buffer_by(h1.meet().unwrap());
+        let (v1, t1, d1) = h1.edit().unwrap();
+        for ((v0, t0), d0) in h0.buffer() {
+            li.push((kh, *v0));
+            ri.push((kh, v1));
+            ot.push(t0.join(t1));
+            od.push(d0.clone().multiply(d1));
+            if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
+        }
+        h1.step();
+    }
+}
diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs
new file mode 100644
index 000000000..9c72ffeab
--- /dev/null
+++ b/differential-dataflow/src/operators/int_proxy/mod.rs
@@ -0,0 +1,67 @@
+//! Backend-agnostic operator tactics using integer proxies.
+//!
+//! The tactics are intended to support custom operator implementations without rebuilding
+//! the non-trivial and often non-obvious time-based logic that supports them.
+//!
+//! The tactics here run DD's operator logic over consolidated `[((u64, u64), time, diff)]`
+//! lists, the first integer a hash of the "key" and granule of independence, the second an
+//! ephemeral data identifier understood by the backend but opaque to the operator harness.
+//! The tactics first elicit proxy identifiers from the backends, perform their necessary time
+//! and difference based computations to stage integer collections, and then re-invoke the
+//! backends with those same identifiers to produce the necessary output.
+//!
+//! The backend is oblivious to the navigation of time, and the operator to the backend's
+//! implementation.
+//!
+//! # The two integers
+//!
+//! The two integers play the role of key and value to the operator, but their connection
+//! to the key and value of the intended computation is nuanced.
+//!
+//! *   `key_hash: u64` — a content hash of the intended key.
+//!     This value should be identical for each instance of identical keys.
+//!     The value is not assumed to be a well-distributed quality hash, only distinct.
+//!     There may be collisions, and the next identifier should assist with this.
+//!
+//! *   `value_id: u64` — an ephemeral intra-hash value identifier.
+//!     These values should be distinct for each distinct datum with the same key_hash.
+//!     In particular, they should account for the conventional `(key, val)` data,
+//!     rather than only the value component, to avoid errors due to colliding keys.
+//!
+//! The key hash acts as an independence marker: keys with different hashes will be isolated,
+//! and their incremental updates performed independently. By forcing the keyspace into `u64`,
+//! there is the risk of hash collision. This means that per-hash logic should be prepared for
+//! this, and should not discard key information that remains semantically important.
+//!
+//! Informally, one should mentally rewrite one's `(key, val)` into `(hash(key), (key, val))`.
+//! Retaining the `key` as data provides access to it, and allows one to certainly respond to
+//! hash collisions that can occur. Although the operator will treat `hash(key)` as the "key",
+//! the backend implementation can apply its own subseqent logic to resolve unwanted matches.
+//! As examples, join logic can follow its hash-wise cross product with key-based filtering,
+//! and reduce logic can group by keys and apply logic to slices within the per-hash list.
+//! Both are welcome to efficiently notice that there have been no collisions and optimize,
+//! or to ignore the risk entirely and live dangerously.
+
+mod history;
+
+pub mod join;
+pub mod reduce;
+
+/// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`.
+///
+/// The [`debug_assert_sorted_bridge`] method is (and can be) used to validate this property.
+pub type ProxyBridge<T, R> = Vec<((u64, u64), T, R)>;
+
+/// Debug check that a presented [`ProxyBridge`] is consolidated.
+///
+/// Operator harnesses use the test to flag backend implementations that do not uphold it.
+pub(crate) fn debug_assert_sorted_bridge<T: Ord, R>(bridge: &ProxyBridge<T, R>, who: &str) {
+    debug_assert!(
+        bridge.windows(2).all(|w| (w[0].0, &w[0].1) < (w[1].0, &w[1].1)),
+        "{}: a presented bridge must be sorted & consolidated by ((key_hash, value_id), time)",
+        who,
+    );
+}
+
+pub use join::{JoinInstance, ProxyJoinBackend, ProxyJoinTactic};
+pub use reduce::{ProxyReduceBackend, ProxyReduceTactic, ReduceInstance, ReduceWindow};
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
new file mode 100644
index 000000000..7b29d9d82
--- /dev/null
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -0,0 +1,579 @@
+//! The proxy reduce framework.
+//!
+//! 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 timely::PartialOrder;
+use timely::progress::{Antichain, Timestamp};
+use timely::progress::frontier::AntichainRef;
+
+use crate::difference::Semigroup;
+use crate::lattice::Lattice;
+use crate::trace::{BatchReader, Description};
+use super::ProxyBridge;
+use crate::operators::reduce::{ReduceTactic, sort_dedup};
+
+use super::history::{IdHistory, TimeHistory};
+
+/// A unit of proxied reduce work, presented to the backend.
+pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
+    /// The accumulated input history.
+    pub source_batches: &'a [B1],
+    /// The freshly arrived input delta.
+    pub input_batches: &'a [B1],
+    /// The accumulated output history.
+    pub output_batches: &'a [B2],
+    /// The compaction frontier for loading (the retire's lower bound).
+    pub lower: AntichainRef<'a, B1::Time>,
+}
+
+/// One window of a retire's changed keys: a bounded, hash-contiguous snip the backend sizes.
+///
+/// The window has the input (old and new) and output histories, restricted to the window's keys.
+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.
+    pub output: ProxyBridge<T, ROut>,
+}
+
+/// 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.
+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.
+    /// The computation proceeds in windows of keys, where only the backend maintains this
+    /// 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.
+    ///
+    /// 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>>;
+
+    /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
+    ///
+    /// Multiple keys are provided concurrently, for each an accumulated input and tentative output.
+    /// The backend should provide for each key the necessary output updates to bring the output in
+    /// with its desires. The `usize` integers upper bound the range for the corresponding key.
+    fn reduce_corrections(
+        &mut self,
+        keys: &[u64],
+        in_ends: &[usize],
+        input: &[(u64, Self::RIn)],
+        out_ends: &[usize],
+        output: &[(u64, Self::ROut)],
+    ) -> (Vec<(u64, Self::ROut)>, Vec<usize>);
+
+    /// Commit to a collection of updates at a specific batch in progress.
+    ///
+    /// The `tile: usize` indexes the list of descriptions provided to `begin()`, and these updates
+    /// are aimed at that batch in progress.
+    fn emit(&mut self, tile: usize, records: &[((u64, u64), B1::Time, Self::ROut)]);
+
+    /// Complete the session matching `begin`. The outputs correspond to the descriptions it was provided.
+    fn finish(&mut self) -> Vec<B2>;
+}
+
+/// A proxy-space [`ReduceTactic`]: matches input and output records by `key_hash`.
+pub struct ProxyReduceTactic<T, Bk> {
+    backend: Bk,
+    /// Pending interesting times beyond the upper frontier, keyed by key hash.
+    pending: BTreeMap<u64, Vec<T>>,
+}
+
+impl<T, Bk> ProxyReduceTactic<T, Bk> {
+    /// A tactic deferring all value semantics to `backend`.
+    pub fn new(backend: Bk) -> Self {
+        ProxyReduceTactic { backend, pending: BTreeMap::new() }
+    }
+}
+
+impl<B1, B2, Bk> ReduceTactic<B1, B2> for ProxyReduceTactic<B1::Time, Bk>
+where
+    B1: BatchReader,
+    B2: BatchReader<Time = B1::Time>,
+    Bk: ProxyReduceBackend<B1, B2>,
+{
+    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>) {
+        if held.elements().iter().all(|t| upper.less_equal(t)) {
+            return (Vec::new(), held.clone());
+        }
+
+        let instance = ReduceInstance {
+            source_batches: &source_batches,
+            input_batches: &input_batches,
+            output_batches: &output_batches,
+            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());
+        }
+        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.
+        let held_elems: Vec<B1::Time> = held.elements().to_vec();
+        let mut tile_descs: Vec<Description<B1::Time>> = Vec::new();
+        let mut tile_held: Vec<B1::Time> = Vec::new();
+        let mut tile_of: Vec<Option<usize>> = vec![None; held_elems.len()];
+        {
+            let mut out_lower = lower.clone();
+            for index in 0..held_elems.len() {
+                let mut out_upper = upper.clone();
+                for t in &held_elems[index + 1..] {
+                    out_upper.insert(t.clone());
+                }
+                if out_upper != out_lower {
+                    tile_of[index] = Some(tile_descs.len());
+                    tile_descs.push(Description::new(out_lower.clone(), out_upper.clone(), Antichain::from_elem(<B1::Time as Timestamp>::minimum())));
+                    tile_held.push(held_elems[index].clone());
+                    out_lower = out_upper;
+                }
+            }
+        }
+        self.backend.begin(&tile_descs);
+
+        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+
+        let mut cursor = 0usize;
+        let mut ns = 0usize;
+
+        // 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.
+        let mut discover_scratch: DiscoverScratch<B1::Time, Bk::RIn> = DiscoverScratch::new();
+        let mut states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
+        let mut tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
+        let mut batch_keys: Vec<u64> = Vec::new();
+        let mut in_ends: Vec<usize> = Vec::new();
+        let mut in_all: Vec<(u64, Bk::RIn)> = Vec::new();
+        let mut out_ends: Vec<usize> = Vec::new();
+        let mut out_all: Vec<(u64, Bk::ROut)> = Vec::new();
+        let mut active: Vec<(usize, B1::Time)> = Vec::new();
+        let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new();
+        let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new();
+        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;
+            let p_out = &window.output;
+            super::debug_assert_sorted_bridge(p_in, "next_window.input");
+            super::debug_assert_sorted_bridge(p_out, "next_window.output");
+
+            for deltas in tile_deltas.iter_mut() { deltas.clear(); }
+
+            // Phase 1 (determination): for every key in the window, discover its interesting times
+            // (times only — no accumulation) and stand up its per-moment replays. Peak state is
+            // O(window presentation), bounded by the window `next_window` already materialized.
+            // `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).
+            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 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 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());
+                    let out_times = (o0..o1).map(|o| p_out[o].1.clone());
+                    discover_times(
+                        KeyView { p_in, i0, i1, pending },
+                        seed_times, out_times, upper,
+                        &mut discover_scratch,
+                        &mut moments_scratch, &mut pended_scratch,
+                    );
+                }
+                if !pended_scratch.is_empty() {
+                    new_pending.insert(key, std::mem::take(&mut pended_scratch));
+                }
+                if moments_scratch.is_empty() {
+                    continue;
+                }
+
+                // Reload slot `n_states` in place (grow the buffer by one only when a window is wider
+                // than any before). `drain` moves the discovered moments in without copy or realloc.
+                if n_states == states.len() {
+                    states.push(KeyState::empty());
+                }
+                let st = &mut states[n_states];
+                st.key = key;
+                st.cursor = 0;
+                st.produced.clear();
+                st.moments.clear();
+                st.moments.extend(moments_scratch.drain(..));
+                st.meets.clear();
+                st.meets.extend(st.moments.iter().cloned());
+                for i in (1..st.meets.len()).rev() {
+                    let m = st.meets[i].clone();
+                    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.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;
+            }
+
+            // Phase 2 (application): walk all keys' moments in ROUNDS. Each round assembles every
+            // active key's one-moment-deep input and current-output accumulations and crosses them in
+            // a SINGLE `reduce_corrections` — batching across keys (a key's own moments stay
+            // sequential, each seeing its earlier corrections via `produced`). This caps the backend
+            // call count at O(max moments over keys), not O(sum of moments), with peak materialization
+            // one moment deep per key. `produced` is meet-collapsed each round, exactly like the
+            // reference — bounded, not the O(times × values) delta history.
+            loop {
+                batch_keys.clear();
+                in_ends.clear();
+                in_all.clear();
+                out_ends.clear();
+                out_all.clear();
+                active.clear();
+                let mut advanced = false;
+                for (si, st) in states[..n_states].iter_mut().enumerate() {
+                    if st.cursor >= st.moments.len() {
+                        continue;
+                    }
+                    advanced = true;
+                    let j = st.cursor;
+                    st.cursor += 1;
+                    let t = st.moments[j].clone();
+                    st.in_replay.step_through(&t);
+                    st.out_replay.step_through(&t);
+                    st.in_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);
+
+                    in_accum.clear();
+                    for ((vid, et), d) in st.in_replay.buffer().iter() {
+                        if et.less_equal(&t) {
+                            in_accum.push((*vid, d.clone()));
+                        }
+                    }
+                    crate::consolidation::consolidate(&mut in_accum);
+                    cur_out.clear();
+                    for ((vid, et), d) in st.out_replay.buffer().iter().chain(st.produced.iter()) {
+                        if et.less_equal(&t) {
+                            cur_out.push((*vid, d.clone()));
+                        }
+                    }
+                    crate::consolidation::consolidate(&mut cur_out);
+
+                    if in_accum.is_empty() && cur_out.is_empty() {
+                        continue;
+                    }
+                    batch_keys.push(st.key);
+                    in_all.extend(in_accum.drain(..));
+                    in_ends.push(in_all.len());
+                    out_all.extend(cur_out.drain(..));
+                    out_ends.push(out_all.len());
+                    active.push((si, t));
+                }
+                // Terminate only when every key is EXHAUSTED — not merely when this round produced no
+                // crossing. A round can be empty because every key's current moment is empty-gated
+                // while keys still have later (non-empty) moments; breaking here would drop them.
+                if !advanced {
+                    break;
+                }
+                if batch_keys.is_empty() {
+                    continue;
+                }
+
+                let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
+                let mut cstart = 0usize;
+                for (bi, (si, t)) in active.iter().enumerate() {
+                    let cend = corr_ends[bi];
+                    if cstart != cend {
+                        let idx = held_elems.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
+                        for (vid, d) in &corr[cstart..cend] {
+                            states[*si].produced.push(((*vid, t.clone()), d.clone()));
+                            tile_deltas[idx].push(((states[*si].key, *vid), t.clone(), d.clone()));
+                        }
+                    }
+                    cstart = cend;
+                }
+            }
+
+            for (held_index, deltas) in tile_deltas.iter_mut().enumerate() {
+                if deltas.is_empty() {
+                    continue;
+                }
+                if let Some(tile) = tile_of[held_index] {
+                    crate::consolidation::consolidate_updates(deltas);
+                    self.backend.emit(tile, &deltas[..]);
+                }
+            }
+        }
+
+        self.pending = new_pending;
+        let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
+        let mut frontier = Antichain::new();
+        for times in self.pending.values() {
+            for t in times {
+                frontier.insert_ref(t);
+            }
+        }
+        (produced, frontier)
+    }
+}
+
+/// 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),
+/// 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.
+struct KeyState<T, RIn, ROut> {
+    key: u64,
+    moments: Vec<T>,
+    meets: Vec<T>,
+    in_replay: IdHistory<T, RIn>,
+    out_replay: IdHistory<T, ROut>,
+    produced: Vec<((u64, T), ROut)>,
+    cursor: usize,
+}
+
+impl<T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<T, RIn, ROut> {
+    /// An empty slot, to be filled by [`ProxyReduceTactic`]'s phase 1 (`reload`-style). The `states`
+    /// 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 }
+    }
+}
+
+/// Reusable per-key scratch for [`discover_times`], held once per retire and threaded through every
+/// key so the replays and time buffers are cleared-and-refilled (`load`/`load_iter` reset while
+/// keeping capacity) rather than reallocated. Mirrors the reference `HistoryReplayer`'s field-held
+/// scratch; without it each of ~n keys paid ~7 fresh allocations per call — profiled at ~54% of the
+/// hash-reduce load in `malloc`, against the cursor reduce's ~10% (see DESIGN.md F7).
+struct DiscoverScratch<T, RIn> {
+    batch_replay: TimeHistory<T>,
+    input_replay: IdHistory<T, RIn>,
+    output_replay: TimeHistory<T>,
+    synth: Vec<T>,
+    times_current: Vec<T>,
+    temporary: Vec<T>,
+    meets: Vec<T>,
+}
+
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> DiscoverScratch<T, RIn> {
+    fn new() -> Self {
+        DiscoverScratch {
+            batch_replay: TimeHistory::new(),
+            input_replay: IdHistory::new(),
+            output_replay: TimeHistory::new(),
+            synth: Vec::new(),
+            times_current: Vec::new(),
+            temporary: Vec::new(),
+            meets: Vec::new(),
+        }
+    }
+}
+
+/// A one-key view into the input presentation: the read-only arguments [`discover_times`] needs
+/// about a single key — its slice `[i0, i1)` of the merged input run `p_in` and the carried
+/// `pending` times.
+struct KeyView<'a, T, RIn> {
+    p_in: &'a ProxyBridge<T, RIn>,
+    i0: usize,
+    i1: usize,
+    pending: &'a [T],
+}
+
+/// 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 {
+        match meet.as_mut() {
+            Some(m) => m.meet_assign(time),
+            None => *meet = Some(time.clone()),
+        }
+    }
+}
+
+/// Phase A: discover a key's interesting times in `[lower, upper)` (pending those at/after `upper`)
+/// **without** accumulating input brackets. It mirrors the reference's DETERMINATION step (times
+/// only, ref `reduce.rs` :671–713): it replays the key's novel (`seed_times`) and `pending` times
+/// in ascending order, marking those carrying novel/pending updates interesting and synthesizing the
+/// joins thereof — joins against the input/output histories and the reached times. Nothing
+/// materializes an input collection, so peak memory is O(times), never O(times × values); the
+/// application walk re-assembles each moment's accumulation on the fly, one moment deep — the whole
+/// point of the tactic (see DESIGN.md F8). Buffers are advanced by the meet of the times still to
+/// come and consolidated, keeping a key with many distinct times linear rather than quadratic.
+#[allow(clippy::too_many_arguments)]
+fn discover_times<T, RIn>(
+    key: KeyView<'_, T, RIn>,
+    seed_times: impl Iterator<Item = T>,
+    out_times: impl Iterator<Item = T>,
+    upper: &Antichain<T>,
+    scratch: &mut DiscoverScratch<T, RIn>,
+    moments: &mut Vec<T>,
+    pended: &mut Vec<T>,
+) where
+    T: Timestamp + Lattice,
+    RIn: Semigroup + Clone,
+{
+    // Reuse the retire's scratch: `load`/`load_iter` reset the replays (keeping capacity); the plain
+    // buffers are cleared here. `meets_slice` reborrows `meets` immutably; the rest stay disjoint.
+    let DiscoverScratch { batch_replay, input_replay, output_replay, synth, times_current, temporary, meets } = scratch;
+    synth.clear();
+    times_current.clear();
+    temporary.clear();
+
+    batch_replay.load(seed_times, None);
+
+    meets.clear();
+    meets.extend(key.pending.iter().cloned());
+    for i in (1..meets.len()).rev() {
+        let m = meets[i].clone();
+        meets[i - 1].meet_assign(&m);
+    }
+
+    let mut meet: Option<T> = None;
+    update_meet(&mut meet, meets.first());
+    update_meet(&mut meet, batch_replay.meet());
+
+    // The merged (history ⊎ novel) run — replayed for its TIMES only (join base), never
+    // accumulated. Output times likewise: base joins, never seeds.
+    input_replay.load_iter(
+        (key.i0..key.i1).map(|i| (key.p_in[i].0.1, key.p_in[i].1.clone(), key.p_in[i].2.clone())),
+        meet.as_ref(),
+    );
+    output_replay.load(out_times, meet.as_ref());
+
+    let mut times_slice = key.pending;
+    let mut meets_slice = &meets[..];
+
+    while let Some(next_time) = [batch_replay.time(), times_slice.first(), input_replay.time(), output_replay.time(), synth.last()]
+        .into_iter()
+        .flatten()
+        .min()
+        .cloned()
+    {
+        input_replay.step_while_time_is(&next_time);
+        output_replay.step_while_time_is(&next_time);
+        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 synth.last() == Some(&next_time) {
+            times_current.push(synth.pop().expect("nonempty"));
+            interesting = true;
+        }
+        while times_slice.first() == Some(&next_time) {
+            times_current.push(times_slice[0].clone());
+            times_slice = &times_slice[1..];
+            meets_slice = &meets_slice[1..];
+            interesting = true;
+        }
+        interesting = interesting || batch_replay.buffer().iter().any(|t| t.less_equal(&next_time));
+        interesting = interesting || times_current.iter().any(|t| t.less_equal(&next_time));
+
+        if !upper.less_equal(&next_time) {
+            if interesting {
+                // Synthesize joins against the input/output histories (times only — no
+                // accumulation), then record `next_time` as an interesting moment.
+                if let Some(m) = meet.as_ref() {
+                    input_replay.advance_buffer_by(m);
+                }
+                for ((_, t), _) in input_replay.buffer().iter() {
+                    if !t.less_equal(&next_time) {
+                        temporary.push(next_time.join(t));
+                    }
+                }
+                if let Some(m) = meet.as_ref() {
+                    output_replay.advance_buffer_by(m);
+                }
+                for t in output_replay.buffer().iter() {
+                    if !t.less_equal(&next_time) {
+                        temporary.push(next_time.join(t));
+                    }
+                }
+                moments.push(next_time.clone());
+            }
+            temporary.extend(batch_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+            temporary.extend(times_current.iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+            sort_dedup(temporary);
+            let synth_len = synth.len();
+            for time in temporary.drain(..) {
+                if upper.less_equal(&time) {
+                    pended.push(time);
+                } else {
+                    synth.push(time);
+                }
+            }
+            if synth.len() > synth_len {
+                synth.sort_by(|x, y| y.cmp(x));
+                synth.dedup();
+            }
+        } else if interesting {
+            pended.push(next_time.clone());
+        }
+
+        meet = None;
+        update_meet(&mut meet, batch_replay.meet());
+        update_meet(&mut meet, input_replay.meet());
+        update_meet(&mut meet, output_replay.meet());
+        for t in synth.iter() {
+            update_meet(&mut meet, Some(t));
+        }
+        update_meet(&mut meet, meets_slice.first());
+        if let Some(m) = meet.as_ref() {
+            for t in times_current.iter_mut() {
+                *t = t.join(m);
+            }
+        }
+        sort_dedup(times_current);
+    }
+    sort_dedup(pended);
+}
diff --git a/differential-dataflow/src/operators/mod.rs b/differential-dataflow/src/operators/mod.rs
index a2e0a32b2..d230a4ac3 100644
--- a/differential-dataflow/src/operators/mod.rs
+++ b/differential-dataflow/src/operators/mod.rs
@@ -9,6 +9,7 @@ pub use self::count::CountTotal;
 pub use self::threshold::ThresholdTotal;
 
 pub mod arrange;
+pub mod int_proxy;
 pub mod reduce;
 pub mod iterate;
 pub mod join;
@@ -143,9 +144,8 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
         TimeReplay { history: &self.history[..], buffer }
     }
 
-    /// Organizes history based on current contents of edits.
-    fn replay<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
-
+    /// Organizes history based on current contents of edits (sort + suffix meets).
+    fn build(&mut self) {
         self.buffer.clear();
         self.history.clear();
         for value_index in 0 .. self.edits.values.len() {
@@ -159,45 +159,86 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
 
         self.history.sort_by(|x,y| y.cmp(x));
         self.history.iter_mut().reduce(|prev, cur| { cur.1.meet_assign(&prev.1); cur });
+    }
 
+    /// Organizes history based on current contents of edits, returning a fresh replay.
+    fn replay<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
+        self.build();
         HistoryReplay { replay: self }
     }
-}
 
-struct HistoryReplay<'history, V, T, D> {
-    replay: &'history mut ValueHistory<V, T, D>,
+    /// Loads `edits` from a plain iterator (grouped by consecutive value — the presentation
+    /// order), advancing each time by `advance_by` if supplied, then organizes. This is the
+    /// cursor-free ingestion path: the `int_proxy` tactics present `(value_id, time, diff)`
+    /// runs directly rather than through a `Cursor`, and share this machinery instead of
+    /// re-implementing it. Ungrouped input is still correct, only less compact.
+    fn load_iter(&mut self, edits: impl Iterator<Item = (V, T, D)>, advance_by: Option<&T>) {
+        self.edits.clear();
+        let mut cur: Option<V> = None;
+        for (v, mut time, diff) in edits {
+            if cur != Some(v) {
+                if let Some(pv) = cur { self.edits.seal(pv); }
+                cur = Some(v);
+            }
+            if let Some(m) = advance_by { time.join_assign(m); }
+            self.edits.push(time, diff);
+        }
+        if let Some(pv) = cur { self.edits.seal(pv); }
+        self.build();
+    }
 }
 
-impl<'history, V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::difference::Semigroup> HistoryReplay<'history, V, T, D> {
-    fn time(&self) -> Option<&T> { self.replay.history.last().map(|x| &x.0) }
-    fn meet(&self) -> Option<&T> { self.replay.history.last().map(|x| &x.1) }
+impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::difference::Semigroup> ValueHistory<V, T, D> {
+    /// The next (least) un-replayed time.
+    fn time(&self) -> Option<&T> { self.history.last().map(|x| &x.0) }
+    /// The meet of all un-replayed times.
+    fn meet(&self) -> Option<&T> { self.history.last().map(|x| &x.1) }
+    /// The next un-replayed edit, as `(value, time, diff)`.
     fn edit(&self) -> Option<(V, &T, &D)> {
-        self.replay.history.last().map(|&(ref t, _, v, e)| (self.replay.edits.values[v].0, t, &self.replay.edits.edits[e].1))
-    }
-
-    fn buffer(&self) -> &[((V, T), D)] {
-        &self.replay.buffer[..]
+        self.history.last().map(|&(ref t, _, v, e)| (self.edits.values[v].0, t, &self.edits.edits[e].1))
     }
-
+    /// The buffered (stepped-in, advanced, consolidated) edits.
+    fn buffer(&self) -> &[((V, T), D)] { &self.buffer[..] }
+    /// Move the next edit into the buffer.
     fn step(&mut self) {
-        let (time, _, value_index, edit_offset) = self.replay.history.pop().unwrap();
-        self.replay.buffer.push(((self.replay.edits.values[value_index].0, time), self.replay.edits.edits[edit_offset].1.clone()));
+        let (time, _, value_index, edit_offset) = self.history.pop().unwrap();
+        self.buffer.push(((self.edits.values[value_index].0, time), self.edits.edits[edit_offset].1.clone()));
     }
+    /// Step edits while the next time equals `time`; true iff any did.
     fn step_while_time_is(&mut self, time: &T) -> bool {
         let mut found = false;
-        while self.time() == Some(time) {
-            found = true;
-            self.step();
-        }
+        while self.time() == Some(time) { found = true; self.step(); }
         found
     }
+    /// Step edits while the next time is `<= time` in the TOTAL order (a superset of the
+    /// partially-ordered downset; readers filter the buffer by `less_equal` themselves).
+    fn step_through(&mut self, time: &T) {
+        while self.time().is_some_and(|t| t <= time) { self.step(); }
+    }
+    /// Advance buffered times by `meet` and consolidate — the collapse that keeps replay linear.
     fn advance_buffer_by(&mut self, meet: &T) {
-        for element in self.replay.buffer.iter_mut() {
-            (element.0).1.join_assign(meet);
-        }
-        crate::consolidation::consolidate(&mut self.replay.buffer);
+        for element in self.buffer.iter_mut() { (element.0).1.join_assign(meet); }
+        crate::consolidation::consolidate(&mut self.buffer);
     }
-    fn is_done(&self) -> bool { self.replay.history.is_empty() }
+    fn is_done(&self) -> bool { self.history.is_empty() }
+}
+
+struct HistoryReplay<'history, V, T, D> {
+    replay: &'history mut ValueHistory<V, T, D>,
+}
+
+// A `HistoryReplay` is a thin cursor-facing handle over a `ValueHistory`; the replay
+// machinery lives on `ValueHistory` itself (shared with the `int_proxy` tactics), and
+// these forward to it.
+impl<'history, V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::difference::Semigroup> HistoryReplay<'history, V, T, D> {
+    fn time(&self) -> Option<&T> { self.replay.time() }
+    fn meet(&self) -> Option<&T> { self.replay.meet() }
+    fn edit(&self) -> Option<(V, &T, &D)> { self.replay.edit() }
+    fn buffer(&self) -> &[((V, T), D)] { self.replay.buffer() }
+    fn step(&mut self) { self.replay.step() }
+    fn step_while_time_is(&mut self, time: &T) -> bool { self.replay.step_while_time_is(time) }
+    fn advance_buffer_by(&mut self, meet: &T) { self.replay.advance_buffer_by(meet) }
+    fn is_done(&self) -> bool { self.replay.is_done() }
 }
 
 /// A time-only, non-destructive walk over an already-built [`ValueHistory`] history.
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 7029b2274..4ba14388c 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -24,6 +24,16 @@ use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Batc
 use crate::trace::cursor::cursor_list;
 use crate::trace::implementations::containers::BatchContainer;
 
+/// Sort and deduplicate a list. Shared by the cursor and reference tactics (via their
+/// `use super::*`) and the proxy tactic (`crate::operators::reduce::sort_dedup`), which
+/// each previously carried an identical copy.
+#[inline(never)]
+pub(crate) fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
+    list.dedup();
+    list.sort();
+    list.dedup();
+}
+
 /// A type that resolves a key-wise reduction over batches arriving on the input.
 ///
 /// Unlike join, reduce does not suspend: its output is at most linear in its input, so a single
@@ -486,13 +496,6 @@ mod cursors {
     }
 
 
-    #[inline(never)]
-    fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
-        list.dedup();
-        list.sort();
-        list.dedup();
-    }
-
     /// Implementation based on replaying historical and new updates together.
     mod history_replay {
 
@@ -502,7 +505,7 @@ mod cursors {
         use crate::trace::Cursor;
         use crate::operators::ValueHistory;
 
-        use super::sort_dedup;
+        use crate::operators::reduce::sort_dedup;
 
         /// The `HistoryReplayer` is a compute strategy based on moving through existing inputs, interesting times, etc in
         /// time order, maintaining consolidated representations of updates with respect to future interesting times.
@@ -849,13 +852,6 @@ pub(crate) mod reference {
         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>) {