diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs
index aea8e598b..783bd7dff 100644
--- a/differential-dataflow/src/operators/common.rs
+++ b/differential-dataflow/src/operators/common.rs
@@ -103,12 +103,13 @@ impl<T: Lattice + Clone + Ord> TimeHistory<T> {
 /// `emit` receives every produced `(id0, id1, joined time, multiplied diff)`. Both histories
 /// must be pre-loaded (`load`/`load_iter`) and are fully drained. For small histories a plain
 /// cross product is cheaper; callers should gate on size.
-pub fn bilinear_wave<V, T, R0, R1, RO>(
-    h0: &mut ValueHistory<V, T, R0>,
-    h1: &mut ValueHistory<V, T, R1>,
-    mut emit: impl FnMut(V, V, T, RO),
+pub fn bilinear_wave<V0, V1, T, R0, R1, RO>(
+    h0: &mut ValueHistory<V0, T, R0>,
+    h1: &mut ValueHistory<V1, T, R1>,
+    mut emit: impl FnMut(V0, V1, T, RO),
 ) where
-    V: Copy + Ord,
+    V0: Copy + Ord,
+    V1: Copy + Ord,
     T: Ord + Clone + Lattice,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
     R1: Semigroup + Clone,
@@ -180,11 +181,11 @@ pub fn tile_descriptions<T: Timestamp + Lattice>(
 }
 
 /// A one-key view into an input presentation: the read-only arguments [`discover_times`] needs
-/// about a single key — its slice `[i0, i1)` of the merged `(id, time, diff)` run `p_in` and
+/// about a single key — its slice `[i0, i1)` of the merged `(token, time, diff)` run `p_in` and
 /// the carried `pending` times.
-pub struct KeyView<'a, T, RIn> {
-    /// The presented `((key_hash, value_id), time, diff)` run the key's records live in.
-    pub p_in: &'a [((u64, u64), T, RIn)],
+pub struct KeyView<'a, G, I, T, RIn> {
+    /// The presented `((group, token), time, diff)` run the key's records live in.
+    pub p_in: &'a [((G, I), T, RIn)],
     /// The key's first record.
     pub i0: usize,
     /// One past the key's last record.
@@ -205,9 +206,9 @@ fn update_meet<T: Lattice + Clone>(meet: &mut Option<T>, other: Option<&T>) {
 
 /// Reusable per-key scratch for [`discover_times`]: held once and threaded through every key,
 /// so replays and time buffers are cleared and refilled rather than reallocated per key.
-pub struct DiscoverScratch<T, RIn> {
+pub struct DiscoverScratch<I, T, RIn> {
     batch_replay: TimeHistory<T>,
-    input_replay: ValueHistory<u64, T, RIn>,
+    input_replay: ValueHistory<I, T, RIn>,
     output_replay: TimeHistory<T>,
     synth: Vec<T>,
     times_current: Vec<T>,
@@ -215,7 +216,7 @@ pub struct DiscoverScratch<T, RIn> {
     meets: Vec<T>,
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> DiscoverScratch<T, RIn> {
+impl<I: Copy + Ord, T: Timestamp + Lattice, RIn: Semigroup + Clone> DiscoverScratch<I, T, RIn> {
     /// Fresh scratch; hold one per retire and thread it through every key.
     pub fn new() -> Self {
         DiscoverScratch {
@@ -230,7 +231,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> DiscoverScratch<T, RIn> {
     }
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch<T, RIn> {
+impl<I: Copy + Ord, T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch<I, T, RIn> {
     fn default() -> Self { Self::new() }
 }
 
@@ -246,15 +247,16 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch
 /// consolidated view is unsound: compaction may advance a history record onto a novel time,
 /// where consolidation cancels the novel update and its interesting time is missed.
 #[allow(clippy::too_many_arguments)]
-pub fn discover_times<T, RIn>(
-    key: KeyView<'_, T, RIn>,
+pub fn discover_times<G, I, T, RIn>(
+    key: KeyView<'_, G, I, T, RIn>,
     seed_times: impl Iterator<Item = T>,
     out_times: impl Iterator<Item = T>,
     upper: &Antichain<T>,
-    scratch: &mut DiscoverScratch<T, RIn>,
+    scratch: &mut DiscoverScratch<I, T, RIn>,
     moments: &mut Vec<T>,
     pended: &mut Vec<T>,
 ) where
+    I: Copy + Ord,
     T: Timestamp + Lattice,
     RIn: Semigroup + Clone,
 {
diff --git a/differential-dataflow/src/operators/int_proxy/history.rs b/differential-dataflow/src/operators/int_proxy/history.rs
index 905614ac0..d03eb3b3c 100644
--- a/differential-dataflow/src/operators/int_proxy/history.rs
+++ b/differential-dataflow/src/operators/int_proxy/history.rs
@@ -1,5 +1,5 @@
 //! Time-ordered replay of proxy update histories, with meet-advancement.
 
-/// 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 over proxy value tokens.
+pub(in crate::operators) type IdHistory<I, T, R> = crate::operators::ValueHistory<I, T, R>;
 
diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 7a02d1b1f..e26823d12 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -1,6 +1,6 @@
 //! The proxy join framework.
 //!
-//! A conventional differential join against `(u64, u64)` values, which are provided by
+//! A conventional differential join against `(group, token)` values, which are provided by
 //! and then interpreted by a backend, who is relieved of lattice-time reasoning.
 
 use timely::progress::{Antichain, Timestamp};
@@ -24,178 +24,373 @@ pub struct JoinInstance<'a, B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     pub lower: AntichainRef<'a, B0::Time>,
 }
 
-/// A type that can interpret and retire pairs of batches, joined by key hashes.
+/// One window of a join unit: both inputs presented for a contiguous, ascending group range.
+pub struct JoinWindow<G, I0, I1, T, R0, R1> {
+    /// Window presentation of the first input, sorted & consolidated by `((group, token), time)`.
+    pub input0: ProxyBridge<G, I0, T, R0>,
+    /// Window presentation of the second input, same ordering, over the same group range.
+    pub input1: ProxyBridge<G, I1, T, R1>,
+}
+
+/// A type that can interpret and retire pairs of batches, joined by group tokens.
+///
+/// The protocol, per unit, is `[ next_window produce* ]* produce?`: `next_window` yields a
+/// bounded presentation of both inputs over one group range, the harness matches by group and
+/// hands the matches to `produce` a batch at a time, and a trailing `produce` delivers the last
+/// partial batch once the windows run out. Windows are produced lazily as the join driver's
+/// fuel allows, so at most one window's presentations are live at a time.
 ///
-/// 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.
+/// Both methods move data in bulk. Neither is called per record.
 pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
+    /// The group token: names the granule of independence, shared by both inputs.
+    ///
+    /// Commonly a `u64` key hash; exactly the key for small `Copy` keys. `'static` because
+    /// groups are the one token that may cross invocations.
+    type Group: Copy + Ord + 'static;
+    /// The value token for the first input, scoped to one invocation.
+    type Token0: Copy + Ord;
+    /// The value token for the second input, scoped to one invocation.
+    type Token1: Copy + Ord;
     /// 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.
+    /// The output container built from matched value tokens.
     type Output;
 
-    /// Prepare a proxy bridge from the instance's first input, optionally key-hash restricted.
+    /// Per-unit state, owned by the unit and interpreted only by the backend.
+    ///
+    /// It cannot live on `&mut self`: the driver holds units across scheduler activations and
+    /// drains them under fuel, so several half-drained units — from both input queues, and from
+    /// different invocations — interleave against one shared backend. `Default` is the state of
+    /// a fresh unit; a typical backend records a position per batch per side, plus whatever it
+    /// needs to interpret its own tokens when building output.
+    ///
+    /// [`next_window`](Self::next_window) may update it. [`produce`](Self::produce) sees it
+    /// immutably, which is what stops output from accumulating between calls.
+    type UnitState: Default;
+
+    /// Matches the harness buffers per [`produce`](Self::produce) call.
+    ///
+    /// This is the dial that amortizes the boundary crossing, and it bounds output container
+    /// size, since a backend may not carry a partial container between calls. It costs
+    /// `live units * MATCH_BATCH` buffered matches.
+    const MATCH_BATCH: usize = 1024;
+
+    /// Produce the next window of the join unit, and advance `cursor`.
+    ///
+    /// Windows must cover contiguous, strictly ascending group ranges, and together must cover
+    /// every group appearing in the `fresh` side's data; groups absent from either side should
+    /// be omitted (they produce no matches, and driving the unit by the fresh side's groups is
+    /// what makes a small fresh batch against a large trace cost `O(fresh)` presentations
+    /// rather than `O(trace)`). Note: the harness is data-oblivious and *cannot check* the
+    /// coverage clause — it checks ordering, progress, and sortedness; missing a fresh group
+    /// silently loses that group's matches. Both bridges **must be sorted and consolidated**
+    /// by `((group, token), time)`. The backend sizes windows to amortize harness crossings;
+    /// both bridges are live at once, so tighter windows mean less state.
+    ///
+    /// `reuse` returns the previous, fully-processed window: reclaim its bridge capacity
+    /// (`clear()` and refill) rather than allocating fresh, so steady-state windowing does
+    /// not churn allocation proportional to data volume.
+    fn next_window(&mut self, instance: &JoinInstance<'_, B0, B1>, fresh: Fresh, unit: &mut Self::UnitState, reuse: Option<JoinWindow<Self::Group, Self::Token0, Self::Token1, B0::Time, Self::R0, Self::R1>>) -> Option<JoinWindow<Self::Group, Self::Token0, Self::Token1, B0::Time, Self::R0, Self::R1>>;
+
+    /// Build output containers from a batch of matches, appending them to `out`.
     ///
-    /// 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.
+    /// `matches` holds at most [`MATCH_BATCH`](Self::MATCH_BATCH) matched pairs — each a left
+    /// token, a right token, the joined time, and the multiplied diff — in window order.
+    /// `keys` and `ends` cut it into per-group runs: `keys[i]` owns
+    /// `matches[ends[i-1] .. ends[i]]`, reading `ends[-1]` as `0` — the same shape reduce's
+    /// [`reduce_corrections`](super::ProxyReduceBackend::reduce_corrections) uses. A group whose
+    /// fanout exceeds the batch has its run split across calls, so a call is **not** guaranteed
+    /// to hold whole groups.
     ///
-    /// 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;
+    /// The backend decides how many containers to append. Container boundaries must be
+    /// semantically invisible: the concatenation of everything appended across all of a unit's
+    /// calls must equal the single-container output. Nothing may be carried between calls —
+    /// there is no end-of-unit call to collect a remainder, and a partial container held on
+    /// `&mut self` would ship under another unit's capability.
+    fn produce(&mut self, instance: &JoinInstance<'_, B0, B1>, unit: &Self::UnitState, keys: &[Self::Group], ends: &[usize], matches: &[(Self::Token0, Self::Token1, B0::Time, Self::ROut)], out: &mut Vec<Self::Output>);
 }
 
-/// A proxy-space [`JoinTactic`]: matches records of the two presented runs by `key_hash`,
+/// A proxy-space [`JoinTactic`]: matches records of the presented windows by group token,
+/// crosses matched pairs with joined times and multiplied diffs, and defers all value
+/// semantics to the backend.
+///
+/// The backend sits behind an `Rc<RefCell<_>>` because prepared units are *lazy*: the join
+/// driver holds each unit's iterator across scheduler activations and drains it under fuel,
+/// so several half-drained units (from both input queues) can be live at once, each needing
+/// the backend when polled. Unit progress therefore lives in the unit (the `cursor` token),
+/// never on the backend.
 pub struct ProxyJoinTactic<B0, B1, Bk> {
-    backend: Bk,
+    backend: std::rc::Rc<std::cell::RefCell<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 }
+        ProxyJoinTactic { backend: std::rc::Rc::new(std::cell::RefCell::new(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>,
+    B0: BatchReader + 'static,
+    B1: BatchReader<Time = B0::Time> + 'static,
+    Bk: ProxyJoinBackend<B0, B1> + 'static,
     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())
+        Box::new(JoinUnit {
+            task: Task { input0, input1, fresh, lower: Antichain::from_elem(meet) },
+            backend: std::rc::Rc::clone(&self.backend),
+            unit: Bk::UnitState::default(),
+            phase: Phase::Fetch,
+            high: None,
+            spent: None,
+            h0: IdHistory::new(),
+            h1: IdHistory::new(),
+            keys: Vec::new(),
+            ends: Vec::new(),
+            matches: Vec::with_capacity(Bk::MATCH_BATCH),
+            produced: Vec::new(),
+            ready: std::collections::VecDeque::new(),
+        })
+    }
+}
+
+/// The bridge type a backend `Bk` presents for batches `B0`/`B1` — [`JoinWindow`] at
+/// the backend's tokens and the batches' time. Named to keep the unit's fields legible.
+type WindowFor<B0, B1, Bk> = JoinWindow<
+    <Bk as ProxyJoinBackend<B0, B1>>::Group,
+    <Bk as ProxyJoinBackend<B0, B1>>::Token0,
+    <Bk as ProxyJoinBackend<B0, B1>>::Token1,
+    <B0 as BatchReader>::Time,
+    <Bk as ProxyJoinBackend<B0, B1>>::R0,
+    <Bk as ProxyJoinBackend<B0, B1>>::R1,
+>;
+
+/// The immutable description of a unit's work: both batch lists, which side is fresh,
+/// and the capability's time (a lower bound on the fresh side's times, so the
+/// accumulated side loads compacted — see [`JoinInstance`]; every output ships under
+/// that capability, so advancing loaded times by it leaves the output unchanged).
+struct Task<B0: BatchReader, B1> {
+    input0: Vec<B0>,
+    input1: Vec<B1>,
+    fresh: Fresh,
+    lower: Antichain<B0::Time>,
+}
+
+impl<B0: BatchReader, B1: BatchReader<Time = B0::Time>> Task<B0, B1> {
+    /// The borrowed view of the task that every backend call receives.
+    fn instance(&self) -> JoinInstance<'_, B0, B1> {
+        JoinInstance { batches0: &self.input0, batches1: &self.input1, lower: self.lower.borrow() }
     }
 }
 
-/// Update count threshold used to return to the backend with join output to instantiate.
-const JOIN_CHUNK: usize = 1 << 20;
+/// Where a unit stands in its march through the backend's windows.
+enum Phase<W> {
+    /// No window in progress; ask the backend for the next one.
+    Fetch,
+    /// Merging through a window, one key per step; the `usize`s are the merge
+    /// positions into its two bridges.
+    Merge(W, usize, usize),
+    /// The backend returned `None`: windows are exhausted, trailing batch not yet produced.
+    Drained,
+    /// The trailing batch is produced; the iterator is spent and yields only `None`.
+    Spent,
+}
 
-/// 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>
+/// One lazy join unit: owns its batches and streams outputs a key at a time.
+///
+/// Each `next` first drains `ready`; when empty it advances `phase` by one step —
+/// fetching the next window, or merge-matching **one** key from the current one into
+/// the match buffer, which goes to the backend's `produce` whenever it fills (and once
+/// more at `Drained`). Peak state is one window's presentations, one batch of buffered
+/// matches, and the containers `produce` has appended but the driver has not taken.
+///
+/// The fields group by owner. The *task* is the immutable work description. The
+/// *backend* is shared by every live unit of the operator, so the state it interprets
+/// per unit — `unit` — lives here (see [`ProxyJoinBackend::UnitState`]). The harness
+/// owns the window *machine*: `phase`, the `high` watermark enforcing ascending
+/// windows, and the `spent` window awaiting return to the backend for buffer reuse.
+/// `h0`/`h1` are merge *scratch*, reloaded per key (only the >=16/>=16 wave path
+/// touches them) so high-fanout keys pay no per-key allocation. `keys`/`ends`/`matches`
+/// are the *match buffer* handed to `produce`, in its `(keys, ends, matches)` shape;
+/// `produced` is the vector it appends containers to. `ready` is the *output* queue
+/// toward the driver.
+struct JoinUnit<B0, B1, Bk>
 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;
+    // The work.
+    task: Task<B0, B1>,
+    // The shared backend, and the per-unit state it interprets.
+    backend: std::rc::Rc<std::cell::RefCell<Bk>>,
+    unit: Bk::UnitState,
+    // The harness's window machine.
+    phase: Phase<WindowFor<B0, B1, Bk>>,
+    high: Option<Bk::Group>,
+    spent: Option<WindowFor<B0, B1, Bk>>,
+    // Per-key merge scratch, reused across keys and windows.
+    h0: IdHistory<Bk::Token0, B0::Time, Bk::R0>,
+    h1: IdHistory<Bk::Token1, B0::Time, Bk::R1>,
+    // The match buffer bound for `produce`, and the containers it appends.
+    keys: Vec<Bk::Group>,
+    ends: Vec<usize>,
+    matches: Vec<(Bk::Token0, Bk::Token1, B0::Time, Bk::ROut)>,
+    produced: Vec<Bk::Output>,
+    // Outputs not yet handed to the driver.
+    ready: std::collections::VecDeque<Bk::Output>,
+}
+
+impl<B0, B1, Bk> Iterator for JoinUnit<B0, B1, Bk>
+where
+    B0: BatchReader,
+    B1: BatchReader<Time = B0::Time>,
+    Bk: ProxyJoinBackend<B0, B1>,
+{
+    type Item = Bk::Output;
+    fn next(&mut self) -> Option<Bk::Output> {
+        while self.ready.is_empty() {
+            let backend = self.backend.clone();
+            let mut backend = backend.borrow_mut();
+            let instance = self.task.instance();
+
+            match &mut self.phase {
+                Phase::Spent => return None,
+
+                // Hand over the trailing partial batch, exactly once.
+                Phase::Drained => {
+                    self.phase = Phase::Spent;
+                    if !self.matches.is_empty() {
+                        backend.produce(&instance, &self.unit, &self.keys, &self.ends, &self.matches, &mut self.produced);
+                        self.keys.clear();
+                        self.ends.clear();
+                        self.matches.clear();
+                        self.ready.extend(self.produced.drain(..));
+                    }
+                }
+
+                // Ask for the next window, returning the spent one for buffer reuse.
+                Phase::Fetch => {
+                    match backend.next_window(&instance, self.task.fresh, &mut self.unit, self.spent.take()) {
+                        None => self.phase = Phase::Drained,
+                        Some(window) => {
+                            super::debug_assert_sorted_bridge(&window.input0, "next_window.input0");
+                            super::debug_assert_sorted_bridge(&window.input1, "next_window.input1");
+                            // Progress guard: an empty window cannot advance the watermark, so a
+                            // backend emitting them repeatedly would spin this loop forever. Skip
+                            // fully-cancelled ranges internally, or return `None`.
+                            assert!(
+                                !window.input0.is_empty() || !window.input1.is_empty(),
+                                "next_window: windows must present at least one record",
+                            );
+                            let first = window.input0.first().map(|r| r.0.0).into_iter().chain(window.input1.first().map(|r| r.0.0)).min();
+                            let last = window.input0.last().map(|r| r.0.0).into_iter().chain(window.input1.last().map(|r| r.0.0)).max();
+                            super::assert_ascending_window(&mut self.high, first, last, "join");
+                            self.phase = Phase::Merge(window, 0, 0);
+                        }
+                    }
+                }
+
+                // Merge-match at most ONE key per step. Matches accumulate in the buffer
+                // and cross to the backend whenever it fills — including *within* a
+                // high-fanout key's wave, which is what bounds the buffer at
+                // `MATCH_BATCH` regardless of fanout.
+                Phase::Merge(window, i, j) => {
+                    let p0 = &window.input0;
+                    let p1 = &window.input1;
+                    let (h0, h1) = (&mut self.h0, &mut self.h1);
+                    let ready = &mut self.ready;
+                    let unit = &self.unit;
+                    let keys = &mut self.keys;
+                    let ends = &mut self.ends;
+                    let matches = &mut self.matches;
+                    let produced = &mut self.produced;
+                    let mut matched = false;
+                    while !matched && *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; }
+                            {
+                                // Close `ki`'s run and cross whenever the buffer fills; a run
+                                // continued after a crossing reopens under `ki` in the next batch.
+                                let mut emit = |i0: Bk::Token0, i1: Bk::Token1, t: B0::Time, d: Bk::ROut| {
+                                    matches.push((i0, i1, t, d));
+                                    if matches.len() >= Bk::MATCH_BATCH {
+                                        keys.push(ki);
+                                        ends.push(matches.len());
+                                        backend.produce(&instance, unit, keys, ends, matches, produced);
+                                        keys.clear();
+                                        ends.clear();
+                                        matches.clear();
+                                        ready.extend(produced.drain(..));
+                                    }
+                                };
+                                join_key(p0, *i..e0, p1, *j..e1, h0, h1, &mut emit);
+                            }
+                            // Close the tail run, if this key left anything unclosed.
+                            let closed = ends.last().copied().unwrap_or(0);
+                            if matches.len() > closed {
+                                keys.push(ki);
+                                ends.push(matches.len());
+                            }
+                            *i = e0;
+                            *j = e1;
+                            matched = true;
+                        }
+                    }
+                    if *i >= p0.len() || *j >= p1.len() {
+                        match std::mem::replace(&mut self.phase, Phase::Fetch) {
+                            Phase::Merge(window, _, _) => self.spent = Some(window),
+                            _ => unreachable!("phase was Merge above"),
+                        }
+                    }
+                }
             }
         }
-        if !li.is_empty() {
-            flush(&mut li, &mut ri, &mut ot, &mut od);
-        }
+        self.ready.pop_front()
     }
-    out
 }
 
-/// Match one key's records across the two presented runs.
+/// Match one key's records across the two presented runs, emitting each match.
 ///
 /// 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>,
+fn join_key<G, I0, I1, T, R0, R1, RO, F>(
+    p0: &ProxyBridge<G, I0, T, R0>,
     r0: std::ops::Range<usize>,
-    p1: &ProxyBridge<T, R1>,
+    p1: &ProxyBridge<G, I1, 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,
+    h0: &mut IdHistory<I0, T, R0>,
+    h1: &mut IdHistory<I1, T, R1>,
+    emit: &mut F,
 ) where
+    G: Copy + Ord,
+    I0: Copy + Ord,
+    I1: Copy + Ord,
     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>),
+    F: FnMut(I0, I1, T, 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); }
+                emit(p0[a].0.1, p1[b].0.1, p0[a].1.join(&p1[b].1), p0[a].2.clone().multiply(&p1[b].2));
             }
         }
         return;
@@ -206,13 +401,5 @@ fn join_key<T, R0, R1, RO, F>(
     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);
 
-    crate::operators::common::bilinear_wave(h0, h1, |v0, v1, t, d| {
-        li.push((kh, v0));
-        ri.push((kh, v1));
-        ot.push(t);
-        od.push(d);
-        if li.len() >= JOIN_CHUNK {
-            flush(li, ri, ot, od);
-        }
-    });
+    crate::operators::common::bilinear_wave(h0, h1, emit);
 }
diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs
index 9c72ffeab..018761d6b 100644
--- a/differential-dataflow/src/operators/int_proxy/mod.rs
+++ b/differential-dataflow/src/operators/int_proxy/mod.rs
@@ -1,37 +1,46 @@
-//! Backend-agnostic operator tactics using integer proxies.
+//! Backend-agnostic operator tactics using token 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 tactics here run DD's operator logic over consolidated `[((group, token), time, diff)]`
+//! lists: the first coordinate names the granule of independence, the second is an ephemeral
+//! value token understood by the backend but opaque to the operator harness. Both are
+//! backend-chosen `Copy + Ord` types — commonly `u64` hashes, but exact values for small
+//! data (`G = u32` node ids, tokens `(u32, u32)` edges), wider hashes for insurance, or
+//! dense ids where exactness is required.
+//! The tactics first elicit proxy tokens from the backends, perform their necessary time
+//! and difference based computations, and then hand results back in token form — both
+//! tactics re-invoke their backend with batches of tokens, never one record at a time —
+//! for the backend 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 tokens
 //!
-//! The two integers play the role of key and value to the operator, but their connection
+//! The two tokens 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.
+//! *   `Group: Copy + Ord + 'static` — the group token, naming the granule of independence.
+//!     This value should be identical for each instance of identical keys. It is the one
+//!     token that crosses invocations (pending work is remembered per group), hence
+//!     `'static`. When it is a content hash of the intended key, it need not be a
+//!     well-distributed quality hash, only distinct; there may be collisions, and the
+//!     next token should assist with this. When it is the key itself (small `Copy` keys),
+//!     collisions are impossible and the cautions below are vacuous.
 //!
-//! *   `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.
+//! *   `Token: Copy + Ord` — an ephemeral intra-group value token, scoped to one
+//!     invocation. These should be distinct for each distinct datum with the same group
+//!     token. When derived by hashing, they should account for the conventional
+//!     `(key, val)` data, rather than only the value component, to avoid errors due to
+//!     colliding groups.
 //!
-//! 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.
+//! The group token acts as an independence marker: keys with different tokens will be isolated,
+//! and their incremental updates performed independently. When the keyspace is compressed
+//! (hashed) into the token, there is the risk of collision. This means that per-group 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
@@ -47,21 +56,40 @@ mod history;
 pub mod join;
 pub mod reduce;
 
-/// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`.
+/// Token-only exchange medium: a consolidated collection of `[((group, token), 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)>;
+/// `G` is the group token (commonly `u64`, a key hash) and `I` the value token (commonly
+/// `u64`, an id or content hash) — see the backend traits for the obligations each carries.
+///
+/// The `debug_assert_sorted_bridge` check (crate-internal) validates this property.
+pub type ProxyBridge<G, I, T, R> = Vec<((G, I), 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) {
+pub(crate) fn debug_assert_sorted_bridge<G: Ord, I: Ord, T: Ord, R>(bridge: &ProxyBridge<G, I, 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)",
+        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 ((group, token), time)",
         who,
     );
 }
 
-pub use join::{JoinInstance, ProxyJoinBackend, ProxyJoinTactic};
+/// Contract helper: `first`/`last` are a window's least and greatest group; the window
+/// must sit strictly above the `high` watermark of everything already presented, which
+/// is then advanced.
+///
+/// A hard assert, not a debug assert: the checks are O(1) per window, and the failure
+/// class they catch — a backend presenting overlapping or regressed windows — corrupts
+/// output silently (dropped matches, doubled corrections) rather than crashing.
+pub(crate) fn assert_ascending_window<G: Copy + Ord>(high: &mut Option<G>, first: Option<G>, last: Option<G>, who: &str) {
+    if let (Some(h), Some(f)) = (*high, first) {
+        assert!(f > h, "{}: windows must present strictly ascending group ranges", who);
+    }
+    if last.is_some() {
+        *high = last;
+    }
+}
+
+pub use join::{JoinInstance, JoinWindow, 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
index ee206a904..688ada2d6 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -1,9 +1,9 @@
 //! The proxy reduce framework.
 //!
-//! A conventional differential reduce against `(u64, u64)`, where the backend supplies the
-//! implementation of the interpretation of the integers.
+//! A conventional differential reduce against `(group, token)` values, where the backend
+//! supplies the implementation of the interpretation of the tokens.
 
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
 
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
@@ -30,34 +30,45 @@ pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     pub lower: AntichainRef<'a, B1::Time>,
 }
 
-/// One window of a retire's changed keys: a bounded, hash-contiguous snip the backend sizes.
+/// One window of a retire's changed keys: a bounded, group-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>,
+pub struct ReduceWindow<G, I, T, RIn, ROut> {
+    /// The window's group tokens: a contiguous, ascending run of the groups needing work.
+    pub keys: Vec<G>,
+    /// Novel-batch time support for `keys`: `(group, time)` sorted by group, drawn from the
+    /// instance's `input_batches` alone.
+    ///
+    /// These seed the interesting-time discovery, and must be the novel batches' own times,
+    /// not a view consolidated with history: compaction may advance a history record onto a
+    /// novel time, where consolidation cancels the novel update and its interesting time is
+    /// missed (see `discover_times`' seeding contract).
+    pub seeds: Vec<(G, T)>,
+    /// Input presentation for `keys`, sorted & consolidated by `((group, token), time)`.
+    pub input: ProxyBridge<G, I, T, RIn>,
     /// Output-history presentation for `keys`, same ordering.
-    pub output: ProxyBridge<T, ROut>,
+    pub output: ProxyBridge<G, I, 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.
+/// The protocol for each round of invocation:
+/// `begin [ next_window reduce_corrections* emit ]* finish`
+/// Seed times arrive per window (on [`ReduceWindow`]), so nothing is staged for the whole
+/// invocation: peak state is one window's presentations.
 pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
+    /// The group token: names the granule of independence.
+    ///
+    /// Commonly a `u64` key hash; exactly the key for small `Copy` keys. `'static` because
+    /// groups are the one token that crosses invocations (pending times are held per group).
+    type Group: Copy + Ord + 'static;
+    /// The value token, scoped to one invocation; output values share the same token space.
+    type Token: Copy + Ord;
     /// Diff type presented for the input.
     type RIn: Semigroup;
     /// Diff type of the output.
     type ROut: Semigroup + 'static;
 
-    /// Hash keys and associated times in the instance's novel input batches.
-    ///
-    /// This is used (with held times) to seed the interesting times for each key.
-    fn seed_times(&self, instance: &ReduceInstance<'_, B1, B2>) -> Vec<(u64, B1::Time)>;
-
     /// Initiate a session to create batches for these descriptions, which span `[lower, upper)`.
     ///
     /// It is the backend's job to prepare output batches for each of these descriptions.
@@ -65,12 +76,23 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// work in progress, until `finish()` is called.
     fn begin(&mut self, tiles: &[Description<B1::Time>]);
 
-    /// Produce the next window, resticted to `changed[cursor..]`, and update `cursor` to track.
+    /// Produce the next window of groups needing work.
+    ///
+    /// Windows must cover contiguous, strictly ascending group ranges, and together must cover
+    /// the union of the groups appearing in the instance's novel `input_batches` and the
+    /// supplied `pending` groups (carried interesting times from earlier retires; such a group
+    /// must appear in `keys` even when it has no novel input). Unlike join, a retire's windows
+    /// all happen within one synchronous call bracketed by `begin`/`finish`, so the backend
+    /// keeps its own progress (batch positions, pending index) in its session state.
     ///
     /// 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>>;
+    ///
+    /// `reuse` returns the previous, fully-processed window: reclaim its buffer capacity
+    /// (`clear()` and refill) rather than allocating fresh, so steady-state windowing does
+    /// not churn allocation proportional to data volume.
+    fn next_window(&mut self, instance: &ReduceInstance<'_, B1, B2>, pending: &[Self::Group], reuse: Option<ReduceWindow<Self::Group, Self::Token, B1::Time, Self::RIn, Self::ROut>>) -> Option<ReduceWindow<Self::Group, Self::Token, B1::Time, Self::RIn, Self::ROut>>;
 
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
     ///
@@ -79,38 +101,38 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// with its desires. The `usize` integers upper bound the range for the corresponding key.
     fn reduce_corrections(
         &mut self,
-        keys: &[u64],
+        keys: &[Self::Group],
         in_ends: &[usize],
-        input: &[(u64, Self::RIn)],
+        input: &[(Self::Token, Self::RIn)],
         out_ends: &[usize],
-        output: &[(u64, Self::ROut)],
-    ) -> (Vec<(u64, Self::ROut)>, Vec<usize>);
+        output: &[(Self::Token, Self::ROut)],
+    ) -> (Vec<(Self::Token, 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)]);
+    fn emit(&mut self, tile: usize, records: &[((Self::Group, Self::Token), 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> {
+/// A proxy-space [`ReduceTactic`]: matches input and output records by group token.
+pub struct ProxyReduceTactic<G, T, Bk> {
     backend: Bk,
-    /// Pending interesting times beyond the upper frontier, keyed by key hash.
-    pending: BTreeMap<u64, Vec<T>>,
+    /// Pending interesting times beyond the upper frontier, keyed by group token.
+    pending: BTreeMap<G, Vec<T>>,
 }
 
-impl<T, Bk> ProxyReduceTactic<T, Bk> {
+impl<G, T, Bk> ProxyReduceTactic<G, 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>
+impl<B1, B2, Bk> ReduceTactic<B1, B2> for ProxyReduceTactic<Bk::Group, B1::Time, Bk>
 where
     B1: BatchReader,
     B2: BatchReader<Time = B1::Time>,
@@ -129,6 +151,12 @@ where
             return (Vec::new(), held.clone());
         }
 
+        // Nothing to do when there are no novel updates and no carried times.
+        if self.pending.is_empty() && input_batches.iter().all(|b| b.len() == 0) {
+            return (Vec::new(), Antichain::new());
+        }
+        let pending_keys: Vec<Bk::Group> = self.pending.keys().copied().collect();
+
         let instance = ReduceInstance {
             source_batches: &source_batches,
             input_batches: &input_batches,
@@ -136,48 +164,70 @@ where
             lower: lower.borrow(),
         };
 
-        let seeds = self.backend.seed_times(&instance);
-        debug_assert!(seeds.windows(2).all(|w| w[0].0 <= w[1].0), "seed_times must be sorted by key_hash");
-        let mut changed: BTreeSet<u64> = seeds.iter().map(|(k, _)| *k).collect();
-        changed.extend(self.pending.keys().copied());
-        if changed.is_empty() {
-            self.pending.clear();
-            return (Vec::new(), Antichain::new());
-        }
-        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 (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
         self.backend.begin(&tile_descs);
 
-        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        let mut new_pending: BTreeMap<Bk::Group, Vec<B1::Time>> = BTreeMap::new();
 
-        let mut cursor = 0usize;
-        let mut ns = 0usize;
+        // Coverage/monotonicity tracking for the debug contract checks below.
+        let mut pend_idx = 0usize;
+        let mut prev_high: Option<Bk::Group> = None;
 
         // 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 discover_scratch: DiscoverScratch<Bk::Token, B1::Time, Bk::RIn> = DiscoverScratch::new();
+        let mut states: Vec<KeyState<Bk::Group, Bk::Token, B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
+        let mut tile_deltas: Vec<Vec<((Bk::Group, Bk::Token), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
+        let mut batch_keys: Vec<Bk::Group> = Vec::new();
         let mut in_ends: Vec<usize> = Vec::new();
-        let mut in_all: Vec<(u64, Bk::RIn)> = Vec::new();
+        let mut in_all: Vec<(Bk::Token, Bk::RIn)> = Vec::new();
         let mut out_ends: Vec<usize> = Vec::new();
-        let mut out_all: Vec<(u64, Bk::ROut)> = Vec::new();
+        let mut out_all: Vec<(Bk::Token, 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 in_accum: Vec<(Bk::Token, Bk::RIn)> = Vec::new();
+        let mut cur_out: Vec<(Bk::Token, 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 mut live: Vec<usize> = Vec::new();
+        // Fast-path staging: keys evaluated directly from their slices, one batched
+        // `reduce_corrections` for all of them per window (round zero, in effect).
+        let mut fast_keys: Vec<Bk::Group> = Vec::new();
+        let mut fast_times: Vec<B1::Time> = Vec::new();
+        let mut fast_in_ends: Vec<usize> = Vec::new();
+        let mut fast_in_all: Vec<(Bk::Token, Bk::RIn)> = Vec::new();
+        let mut fast_out_ends: Vec<usize> = Vec::new();
+        let mut fast_out_all: Vec<(Bk::Token, Bk::ROut)> = Vec::new();
+
+        let mut spent: Option<ReduceWindow<Bk::Group, Bk::Token, B1::Time, Bk::RIn, Bk::ROut>> = None;
+        while let Some(window) = self.backend.next_window(&instance, &pending_keys, spent.take()) {
             let p_in = &window.input;
             let p_out = &window.output;
+            let seeds = &window.seeds;
             super::debug_assert_sorted_bridge(p_in, "next_window.input");
             super::debug_assert_sorted_bridge(p_out, "next_window.output");
+            debug_assert!(seeds.windows(2).all(|w| w[0].0 <= w[1].0), "window seeds must be sorted by group token");
+            debug_assert!(window.keys.windows(2).all(|w| w[0] < w[1]), "window keys must be strictly ascending");
+            // Progress guard: an empty window cannot advance the watermark, so a backend
+            // emitting them repeatedly would spin this loop forever inside one retire.
+            assert!(!window.keys.is_empty(), "next_window: windows must list at least one key");
+            super::assert_ascending_window(&mut prev_high, window.keys.first().copied(), window.keys.last().copied(), "reduce");
+            // Pending coverage: windows ascend, so a pending group below a presented key can
+            // never appear later; it must have been included in this or an earlier window.
+            // A hard assert, not debug: a skipped pending group's carried times would be
+            // silently dropped below, collapsing the frontier and releasing the capability
+            // that its deferred correction needs — permanent wrong output. The walk is a few
+            // integer compares per key, noise against the per-key reduction work.
+            for key in &window.keys {
+                assert!(
+                    pending_keys.get(pend_idx).map_or(true, |p| p >= key),
+                    "next_window skipped a pending group",
+                );
+                if pending_keys.get(pend_idx) == Some(key) { pend_idx += 1; }
+            }
+            let mut ns = 0usize;
 
             for deltas in tile_deltas.iter_mut() { deltas.clear(); }
 
@@ -190,19 +240,58 @@ where
             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; }
+                // Strays are hard errors: a record or seed whose group is below the current
+                // key belongs to no listed key (earlier keys consumed their runs), and the
+                // old skip-loops would have silently discarded it — wrong output with no
+                // diagnostic. Ascending keys make the check one compare per key.
+                assert!(is >= p_in.len() || p_in[is].0.0 >= key, "next_window presented input records for a group absent from keys");
                 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; }
+                assert!(os >= p_out.len() || p_out[os].0.0 >= key, "next_window presented output records for a group absent from keys");
                 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; }
+                assert!(ns >= seeds.len() || seeds[ns].0 >= key, "next_window presented seeds for a group absent from keys");
                 let n0 = ns;
                 while ns < seeds.len() && seeds[ns].0 == key { ns += 1; }
                 let n1 = ns;
 
+                // Fast path: no pending, one distinct seed time `t*`, and every presented
+                // input/output time at-or-before it. Then `discover_times` would compute
+                // moments == [t*] with nothing pended (history times before `t*` pass
+                // without becoming moments, and no join can escape `t*`), so discovery
+                // and the replay loads are skipped and the key is evaluated directly
+                // from its slices. This is the steady-state snapshot shape, where the
+                // full apparatus was measured as the dominant per-key cost.
+                if self.pending.get(&key).is_none() {
+                    if n0 == n1 {
+                        // No seeds and no pending: no interesting times; nothing to do.
+                        continue;
+                    }
+                    let t_star = seeds[n0].1.clone();
+                    if seeds[n0..n1].iter().all(|s| s.1 == t_star)
+                        && p_in[i0..i1].iter().all(|x| x.1.less_equal(&t_star))
+                        && p_out[o0..o1].iter().all(|x| x.1.less_equal(&t_star))
+                    {
+                        in_accum.clear();
+                        in_accum.extend(p_in[i0..i1].iter().map(|x| (x.0 .1, x.2.clone())));
+                        crate::consolidation::consolidate(&mut in_accum);
+                        cur_out.clear();
+                        cur_out.extend(p_out[o0..o1].iter().map(|x| (x.0 .1, x.2.clone())));
+                        crate::consolidation::consolidate(&mut cur_out);
+                        if !in_accum.is_empty() || !cur_out.is_empty() {
+                            fast_keys.push(key);
+                            fast_times.push(t_star);
+                            fast_in_all.append(&mut in_accum);
+                            fast_in_ends.push(fast_in_all.len());
+                            fast_out_all.append(&mut cur_out);
+                            fast_out_ends.push(fast_out_all.len());
+                        }
+                        continue;
+                    }
+                }
+
                 moments_scratch.clear();
                 pended_scratch.clear();
                 {
@@ -226,7 +315,7 @@ where
                 // 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());
+                    states.push(KeyState::empty(key));
                 }
                 let st = &mut states[n_states];
                 st.key = key;
@@ -244,6 +333,34 @@ where
                 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;
             }
+            // Everything presented must have belonged to a listed key (the strays-below
+            // case is asserted per key above; this is the trailing case).
+            assert_eq!(is, p_in.len(), "next_window presented trailing input records past the last key");
+            assert_eq!(os, p_out.len(), "next_window presented trailing output records past the last key");
+            assert_eq!(ns, seeds.len(), "next_window presented trailing seeds past the last key");
+
+            // Round zero: the window's fast-path keys, one batched crossing. Their
+            // corrections go straight to the tiles — a fast key has no later moments.
+            if !fast_keys.is_empty() {
+                let (corr, corr_ends) = self.backend.reduce_corrections(&fast_keys, &fast_in_ends, &fast_in_all, &fast_out_ends, &fast_out_all);
+                let mut cstart = 0usize;
+                for (fi, t) in fast_times.iter().enumerate() {
+                    let cend = corr_ends[fi];
+                    if cstart != cend {
+                        let idx = held_elems.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= fast-path time");
+                        for (vid, d) in &corr[cstart..cend] {
+                            tile_deltas[idx].push(((fast_keys[fi], *vid), t.clone(), d.clone()));
+                        }
+                    }
+                    cstart = cend;
+                }
+                fast_keys.clear();
+                fast_times.clear();
+                fast_in_ends.clear();
+                fast_in_all.clear();
+                fast_out_ends.clear();
+                fast_out_all.clear();
+            }
 
             // 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
@@ -252,6 +369,8 @@ where
             // 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.
+            live.clear();
+            live.extend(0..n_states);
             loop {
                 batch_keys.clear();
                 in_ends.clear();
@@ -259,12 +378,19 @@ where
                 out_ends.clear();
                 out_all.clear();
                 active.clear();
-                let mut advanced = false;
-                for (si, st) in states[..n_states].iter_mut().enumerate() {
+                // Sweep the live keys, retiring exhausted ones by swap-remove, so each round
+                // costs O(live keys) rather than rescanning every exhausted slot under
+                // moment-count skew (order across keys within a round does not matter —
+                // distinct keys are independent; a key's own moments stay sequential).
+                let mut idx = 0;
+                while idx < live.len() {
+                    let si = live[idx];
+                    let st = &mut states[si];
                     if st.cursor >= st.moments.len() {
+                        live.swap_remove(idx);
                         continue;
                     }
-                    advanced = true;
+                    idx += 1;
                     let j = st.cursor;
                     st.cursor += 1;
                     let t = st.moments[j].clone();
@@ -305,7 +431,7 @@ where
                 // 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 {
+                if live.is_empty() {
                     break;
                 }
                 if batch_keys.is_empty() {
@@ -336,10 +462,19 @@ where
                     self.backend.emit(tile, &deltas[..]);
                 }
             }
+            spent = Some(window);
         }
 
+        // Hard for the same reason as the per-window pending walk: an uncovered pending
+        // group's carried times would be dropped and its capability released — permanent
+        // silent wrong output in exchange for one integer compare.
+        assert_eq!(pend_idx, pending_keys.len(), "next_window must cover all pending groups before finishing");
         self.pending = new_pending;
-        let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
+        let finished = self.backend.finish();
+        // Hard for the same reason as the other contract asserts: `zip` would silently
+        // truncate a miscounting backend, shipping batches at the wrong held times.
+        assert_eq!(finished.len(), tile_held.len(), "finish() must return exactly one batch per tile description");
+        let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(finished).collect();
         let mut frontier = Antichain::new();
         for times in self.pending.values() {
             for t in times {
@@ -355,22 +490,22 @@ where
 /// 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,
+struct KeyState<G, I, T, RIn, ROut> {
+    key: G,
     moments: Vec<T>,
     meets: Vec<T>,
-    in_replay: IdHistory<T, RIn>,
-    out_replay: IdHistory<T, ROut>,
-    produced: Vec<((u64, T), ROut)>,
+    in_replay: IdHistory<I, T, RIn>,
+    out_replay: IdHistory<I, T, ROut>,
+    produced: Vec<((I, T), ROut)>,
     cursor: usize,
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<T, RIn, ROut> {
+impl<G: Copy + Ord, I: Copy + Ord, T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<G, I, 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 }
+    fn empty(key: G) -> Self {
+        KeyState { key, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
     }
 }
 
diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 413e61d0e..d2a6b5b42 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -40,6 +40,7 @@ pub trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, C> {
 /// The fresh batch's times all lie at or beyond the capability, so its side is not advanced by the
 /// capability's meet; the opposing accumulated trace is. The marker also selects which queue a unit
 /// joins, so a burst on one input cannot starve the other.
+#[derive(Clone, Copy, Debug)]
 pub enum Fresh {
     /// The first input (`B0`) contributed the fresh batch.
     Input0,
@@ -390,7 +391,7 @@ mod cursors {
 
     /// The conventional cursor-based [`JoinTactic`].
     ///
-    /// It builds a [`CursorList`] over each input batch list and plays the merge-join out at whatever rate
+    /// It builds a [`CursorList`](crate::trace::cursor::CursorList) over each input batch list and plays the merge-join out at whatever rate
     /// the driver's fuel allows. Each prepared unit joins a `B0`-side cursor against a `B1`-side cursor,
     /// emitting `(val0, val1)` to `logic` and yielding the output containers `logic` fills. `logic` is
     /// shared across all outstanding units (an `Rc<RefCell<_>>`), preserving the single mutable-state
diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs
new file mode 100644
index 000000000..04b588eca
--- /dev/null
+++ b/differential-dataflow/tests/int_proxy.rs
@@ -0,0 +1,859 @@
+//! Property tests for the proxy tactics, against a value-token edge backend.
+//!
+//! Every test here checks a general property against an oracle, not current
+//! behavior: join against a naive nested-loop join; reduce against exact
+//! per-time counting (including a randomized differential test); the pending
+//! contract against a partial-order grid; and unit isolation under interleaved
+//! draining (which pinned a real bug). The scaffold is the value-token
+//! instantiation from the seam design: `Group = src` exact, token = the edge.
+
+use std::collections::BTreeMap;
+
+use timely::progress::Antichain;
+
+use differential_dataflow::consolidation::consolidate_updates;
+use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
+use differential_dataflow::operators::join::{Fresh, JoinTactic};
+use differential_dataflow::operators::reduce::ReduceTactic;
+
+use pair::Pair;
+
+type Time = u64;
+type GraphBatch = EdgeBatch<Time>;
+
+// ------------------------------------------------------------------- scaffold
+
+
+use timely::progress::frontier::AntichainRef;
+use timely::progress::Timestamp;
+
+use differential_dataflow::consolidation::consolidate;
+use differential_dataflow::lattice::Lattice;
+use differential_dataflow::operators::int_proxy::{
+    JoinInstance, JoinWindow, ProxyJoinBackend, ProxyReduceBackend, ReduceInstance,
+    ReduceWindow,
+};
+use differential_dataflow::trace::{BatchReader, Description};
+
+pub type Edge = (u32, u32);
+pub type Diff = isize;
+
+/// A batch of edge updates, sorted and consolidated, `G = src` exact.
+#[derive(Clone)]
+pub struct EdgeBatch<T> {
+    pub updates: Vec<(Edge, T, Diff)>,
+    pub description: Description<T>,
+}
+
+impl<T: Timestamp + Lattice> BatchReader for EdgeBatch<T> {
+    type Time = T;
+    fn len(&self) -> usize { self.updates.len() }
+    fn description(&self) -> &Description<T> { &self.description }
+}
+
+/// Sort, consolidate, and wrap updates as a batch describing `[lower, upper)`.
+pub fn batch<T: Timestamp + Lattice + Ord>(updates: Vec<(Edge, T, Diff)>, lower: T, upper: T) -> EdgeBatch<T> {
+    batch_frontiers(updates, Antichain::from_elem(lower), Antichain::from_elem(upper))
+}
+
+/// As [`batch`], with antichain frontiers for partially ordered intervals.
+pub fn batch_frontiers<T: Timestamp + Lattice + Ord>(
+    mut updates: Vec<(Edge, T, Diff)>,
+    lower: Antichain<T>,
+    upper: Antichain<T>,
+) -> EdgeBatch<T> {
+    consolidate_updates(&mut updates);
+    EdgeBatch {
+        updates,
+        description: Description::new(lower, upper, Antichain::from_elem(T::minimum())),
+    }
+}
+
+/// Advance a time to the compaction frontier (join against its elements).
+pub fn advance<T: Lattice + Clone>(t: T, lower: AntichainRef<'_, T>) -> T {
+    lower.iter().fold(t, |t, l| t.join(l))
+}
+
+/// The next group at or after the per-batch positions `pos` into `batches`.
+///
+/// Test-grade: rescans every batch head per group. A real backend should k-way merge
+/// with a heap over batch cursors.
+pub fn next_group<T>(batches: &[EdgeBatch<T>], pos: &[usize]) -> Option<u32> {
+    batches
+        .iter()
+        .zip(pos.iter())
+        .filter_map(|(b, &p)| b.updates.get(p).map(|u| u.0 .0))
+        .min()
+}
+
+/// Drain every update of group `g` from `batches` (advancing `pos`) into `logic`.
+pub fn drain_group<T: Clone>(
+    batches: &[EdgeBatch<T>],
+    pos: &mut [usize],
+    g: u32,
+    mut logic: impl FnMut(Edge, T, Diff),
+) {
+    for (b, p) in batches.iter().zip(pos.iter_mut()) {
+        while *p < b.updates.len() && b.updates[*p].0 .0 == g {
+            let (edge, ref t, r) = b.updates[*p];
+            logic(edge, t.clone(), r);
+            *p += 1;
+        }
+    }
+}
+
+// ---------------------------------------------------------------- join backend
+
+/// Per-unit state: a position per batch per side. This is the state a single `usize`
+/// could not carry — the finding that motivated `type UnitState`.
+#[derive(Default)]
+pub struct EdgeJoinState {
+    pub pos0: Vec<usize>,
+    pub pos1: Vec<usize>,
+}
+
+pub struct EdgeJoinBackend {
+    /// Groups per window; tiny in tests to force many windows.
+    pub window: usize,
+    /// Matches per output container; tiny in tests to force many containers.
+    pub target: usize,
+}
+
+impl EdgeJoinBackend {
+    pub fn new(window: usize) -> Self {
+        EdgeJoinBackend { window, target: 8 }
+    }
+}
+
+impl<T: Timestamp + Lattice + Ord> ProxyJoinBackend<EdgeBatch<T>, EdgeBatch<T>> for EdgeJoinBackend {
+    type Group = u32;
+    type Token0 = Edge;
+    type Token1 = Edge;
+    type R0 = Diff;
+    type R1 = Diff;
+    type ROut = Diff;
+    /// Matched edge pairs with their joined time and multiplied diff.
+    type Output = Vec<(Edge, Edge, T, Diff)>;
+    type UnitState = EdgeJoinState;
+    /// Tiny in tests, to force many crossings and many partial batches.
+    const MATCH_BATCH: usize = 4;
+
+    fn next_window(
+        &mut self,
+        instance: &JoinInstance<'_, EdgeBatch<T>, EdgeBatch<T>>,
+        fresh: Fresh,
+        cursor: &mut EdgeJoinState,
+        reuse: Option<JoinWindow<u32, Edge, Edge, T, Diff, Diff>>,
+    ) -> Option<JoinWindow<u32, Edge, Edge, T, Diff, Diff>> {
+        cursor.pos0.resize(instance.batches0.len(), 0);
+        cursor.pos1.resize(instance.batches1.len(), 0);
+        // Drive by the fresh side's groups: the accumulated side's other groups are
+        // skipped, never presented (O(fresh) presentations; empty fresh side is free).
+        let (fresh_b, other_b) = match fresh {
+            Fresh::Input0 => (instance.batches0, instance.batches1),
+            Fresh::Input1 => (instance.batches1, instance.batches0),
+        };
+        // Reclaim the spent window's bridge capacity.
+        // Route each reclaimed vec back to the side it served, so capacities stay
+        // side-stable rather than alternating duties across round trips.
+        let (mut fresh_run, mut other_run) = match (reuse, fresh) {
+            (Some(JoinWindow { input0, input1 }), Fresh::Input0) => (input0, input1),
+            (Some(JoinWindow { input0, input1 }), Fresh::Input1) => (input1, input0),
+            (None, _) => (Vec::new(), Vec::new()),
+        };
+        loop {
+            fresh_run.clear();
+            other_run.clear();
+            let mut groups = 0;
+            while groups < self.window {
+                let (fresh_pos, other_pos) = match fresh {
+                    Fresh::Input0 => (&mut cursor.pos0, &mut cursor.pos1),
+                    Fresh::Input1 => (&mut cursor.pos1, &mut cursor.pos0),
+                };
+                let Some(g) = next_group(fresh_b, fresh_pos) else { break };
+                groups += 1;
+                drain_group(fresh_b, fresh_pos, g, |e, t, r| {
+                    fresh_run.push(((g, e), advance(t, instance.lower), r));
+                });
+                // Skip the accumulated side below `g`, then drain its matching group.
+                for (b, p) in other_b.iter().zip(other_pos.iter_mut()) {
+                    while *p < b.updates.len() && b.updates[*p].0 .0 < g {
+                        *p += 1;
+                    }
+                }
+                drain_group(other_b, other_pos, g, |e, t, r| {
+                    other_run.push(((g, e), advance(t, instance.lower), r));
+                });
+            }
+            if groups == 0 {
+                return None;
+            }
+            consolidate_updates(&mut fresh_run);
+            consolidate_updates(&mut other_run);
+            // A fully-cancelled quota presents nothing; keep going rather than emit an
+            // empty window (the harness's progress guard forbids those).
+            if fresh_run.is_empty() && other_run.is_empty() {
+                continue;
+            }
+            let (input0, input1) = match fresh {
+                Fresh::Input0 => (fresh_run, other_run),
+                Fresh::Input1 => (other_run, fresh_run),
+            };
+            return Some(JoinWindow { input0, input1 });
+        }
+    }
+
+    fn produce(
+        &mut self,
+        _instance: &JoinInstance<'_, EdgeBatch<T>, EdgeBatch<T>>,
+        _unit: &Self::UnitState,
+        keys: &[u32],
+        ends: &[usize],
+        matches: &[(Edge, Edge, T, Diff)],
+        out: &mut Vec<Self::Output>,
+    ) {
+        // The run structure must describe exactly this batch.
+        assert_eq!(keys.len(), ends.len(), "produce: one end per key run");
+        assert!(ends.windows(2).all(|w| w[0] < w[1]), "produce: run ends must ascend");
+        assert_eq!(ends.last().copied().unwrap_or(0), matches.len(), "produce: runs must cover the batch");
+        // Self-redeeming tokens: the output is built from the tokens alone. Containers are
+        // cut at `target`, and the batch's remainder becomes an undersized container of its
+        // own — nothing is carried to the next call.
+        for chunk in matches.chunks(self.target) {
+            out.push(chunk.to_vec());
+        }
+    }
+}
+
+/// Reference join: all pairs with equal source, times joined, diffs multiplied.
+pub fn naive_join<T: Timestamp + Lattice + Ord>(
+    a: &[(Edge, T, Diff)],
+    b: &[(Edge, T, Diff)],
+) -> Vec<((Edge, Edge), T, Diff)> {
+    let mut out = Vec::new();
+    for (e0, t0, r0) in a {
+        for (e1, t1, r1) in b {
+            if e0.0 == e1.0 {
+                out.push(((*e0, *e1), t0.join(t1), r0 * r1));
+            }
+        }
+    }
+    consolidate_updates(&mut out);
+    out
+}
+
+// -------------------------------------------------------------- reduce backend
+
+/// Counts edges per source: the output value for source `s` is `(s, count)`.
+pub struct EdgeReduceBackend<T> {
+    /// Groups per window; tiny in tests to force many windows.
+    pub window: usize,
+    // Session state, bracketed by `begin`/`finish`.
+    tiles: Vec<Description<T>>,
+    emitted: Vec<Vec<(Edge, T, Diff)>>,
+    pos_source: Vec<usize>,
+    pos_input: Vec<usize>,
+    pos_output: Vec<usize>,
+    pend_idx: usize,
+}
+
+impl<T> EdgeReduceBackend<T> {
+    pub fn new(window: usize) -> Self {
+        EdgeReduceBackend {
+            window,
+            tiles: Vec::new(),
+            emitted: Vec::new(),
+            pos_source: Vec::new(),
+            pos_input: Vec::new(),
+            pos_output: Vec::new(),
+            pend_idx: 0,
+        }
+    }
+}
+
+impl<T: Timestamp + Lattice + Ord> ProxyReduceBackend<EdgeBatch<T>, EdgeBatch<T>> for EdgeReduceBackend<T> {
+    type Group = u32;
+    type Token = Edge;
+    type RIn = Diff;
+    type ROut = Diff;
+
+    fn begin(&mut self, tiles: &[Description<T>]) {
+        self.tiles = tiles.to_vec();
+        self.emitted = vec![Vec::new(); tiles.len()];
+        self.pos_source.clear();
+        self.pos_input.clear();
+        self.pos_output.clear();
+        self.pend_idx = 0;
+    }
+
+    fn next_window(
+        &mut self,
+        instance: &ReduceInstance<'_, EdgeBatch<T>, EdgeBatch<T>>,
+        pending: &[u32],
+        reuse: Option<ReduceWindow<u32, Edge, T, Diff, Diff>>,
+    ) -> Option<ReduceWindow<u32, Edge, T, Diff, Diff>> {
+        self.pos_source.resize(instance.source_batches.len(), 0);
+        self.pos_input.resize(instance.input_batches.len(), 0);
+        self.pos_output.resize(instance.output_batches.len(), 0);
+        // Reclaim the spent window's buffer capacity.
+        let (mut keys, mut seeds, mut input, mut output) = match reuse {
+            Some(ReduceWindow { mut keys, mut seeds, mut input, mut output }) => {
+                keys.clear();
+                seeds.clear();
+                input.clear();
+                output.clear();
+                (keys, seeds, input, output)
+            }
+            None => (Vec::new(), Vec::new(), Vec::new(), Vec::new()),
+        };
+        while keys.len() < self.window {
+            let g = [
+                next_group(instance.source_batches, &self.pos_source),
+                next_group(instance.input_batches, &self.pos_input),
+                next_group(instance.output_batches, &self.pos_output),
+                pending.get(self.pend_idx).copied(),
+            ]
+            .into_iter()
+            .flatten()
+            .min();
+            let Some(g) = g else { break };
+            keys.push(g);
+            if pending.get(self.pend_idx) == Some(&g) {
+                self.pend_idx += 1;
+            }
+            drain_group(instance.source_batches, &mut self.pos_source, g, |e, t, r| {
+                input.push(((g, e), advance(t, instance.lower), r));
+            });
+            drain_group(instance.input_batches, &mut self.pos_input, g, |e, t, r| {
+                // Novel batches also seed interesting times, with their own (raw) times.
+                seeds.push((g, t.clone()));
+                input.push(((g, e), advance(t, instance.lower), r));
+            });
+            drain_group(instance.output_batches, &mut self.pos_output, g, |e, t, r| {
+                output.push(((g, e), advance(t, instance.lower), r));
+            });
+        }
+        if keys.is_empty() {
+            return None;
+        }
+        consolidate_updates(&mut input);
+        consolidate_updates(&mut output);
+        seeds.sort();
+        seeds.dedup();
+        Some(ReduceWindow { keys, seeds, input, output })
+    }
+
+    fn reduce_corrections(
+        &mut self,
+        keys: &[u32],
+        in_ends: &[usize],
+        input: &[(Edge, Diff)],
+        out_ends: &[usize],
+        output: &[(Edge, Diff)],
+    ) -> (Vec<(Edge, Diff)>, Vec<usize>) {
+        let mut corr = Vec::new();
+        let mut ends = Vec::new();
+        let (mut i0, mut o0) = (0, 0);
+        for (k, (&i1, &o1)) in keys.iter().zip(in_ends.iter().zip(out_ends)) {
+            let count: Diff = input[i0..i1].iter().map(|(_, d)| d).sum();
+            assert!(count >= 0, "edge count went negative");
+            // Desired output, minus the tentative output, consolidated.
+            let mut delta: Vec<(Edge, Diff)> = Vec::new();
+            if count > 0 {
+                delta.push(((*k, count as u32), 1));
+            }
+            for (v, d) in &output[o0..o1] {
+                delta.push((*v, -d));
+            }
+            consolidate(&mut delta);
+            corr.extend(delta);
+            ends.push(corr.len());
+            i0 = i1;
+            o0 = o1;
+        }
+        (corr, ends)
+    }
+
+    fn emit(&mut self, tile: usize, records: &[((u32, Edge), T, Diff)]) {
+        self.emitted[tile].extend(records.iter().map(|((_g, e), t, r)| (*e, t.clone(), *r)));
+    }
+
+    fn finish(&mut self) -> Vec<EdgeBatch<T>> {
+        self.tiles
+            .drain(..)
+            .zip(self.emitted.drain(..))
+            .map(|(desc, mut updates)| {
+                consolidate_updates(&mut updates);
+                EdgeBatch { updates, description: desc }
+            })
+            .collect()
+    }
+}
+
+/// Accumulate `updates` at times less-or-equal `t` (the partial order's `less_equal`).
+pub fn accumulate<T: Timestamp + Lattice>(
+    updates: impl Iterator<Item = (Edge, T, Diff)>,
+    t: &T,
+) -> Vec<(Edge, Diff)> {
+    let mut acc: Vec<(Edge, Diff)> = updates
+        .filter(|(_, ut, _)| ut.less_equal(t))
+        .map(|(e, _, r)| (e, r))
+        .collect();
+    consolidate(&mut acc);
+    acc
+}
+
+/// The `Pair` partially ordered timestamp (product order on two coordinates).
+pub mod pair {
+
+    #[derive(Hash, Default, Clone, Copy, 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) }
+    }
+
+    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 serde::{Deserialize, Serialize};
+    use std::fmt::{Debug, Error, Formatter};
+    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))
+        }
+    }
+}
+
+// ----------------------------------------------------------------- the tests
+
+#[test]
+fn join_wave_path_matches_naive() {
+    // `join_key` switches implementations at 16 records per side: below, a nested-loop
+    // cross product; at or above (on BOTH sides), the `bilinear_wave` time-ordered
+    // compacting replay. Every other key in this suite is small, so this test exists to
+    // run the wave path at all: one deliberately fat key (20 presented records per
+    // side — insertions at spread times plus non-cancelling retractions at later
+    // times, so consolidation keeps all of them), alongside a small key so the two
+    // paths mix within one unit, checked against the naive oracle.
+    let mut a_upds: Vec<(Edge, Time, Diff)> = Vec::new();
+    let mut b_upds: Vec<(Edge, Time, Diff)> = Vec::new();
+    for v in 0..12u32 {
+        a_upds.push(((0, v), (v % 4) as Time, 1));
+        b_upds.push(((0, 100 + v), (v % 3) as Time, 1));
+    }
+    for v in 0..8u32 {
+        // Retractions at times distinct from the insertions: two bridge records each,
+        // netting to zero late — cancellation the wave must navigate, not lose.
+        a_upds.push(((0, v), 4 + (v % 2) as Time, -1));
+        b_upds.push(((0, 100 + v), 4 + (v % 3) as Time, -1));
+    }
+    // A small key too, so nested-loop and wave paths run in the same unit.
+    a_upds.push(((1, 7), 0, 1));
+    b_upds.push(((1, 9), 2, 1));
+
+    let a = batch(a_upds.clone(), 0, 8);
+    let b = batch(b_upds.clone(), 0, 8);
+    assert!(a.updates.iter().filter(|u| u.0 .0 == 0).count() >= 16, "fat key must reach the wave threshold");
+    assert!(b.updates.iter().filter(|u| u.0 .0 == 0).count() >= 16, "fat key must reach the wave threshold");
+    let expected = naive_join(&a.updates, &b.updates);
+
+    for window in [1, 100] {
+        for fresh in [Fresh::Input0, Fresh::Input1] {
+            let mut tactic = ProxyJoinTactic::new(EdgeJoinBackend::new(window));
+            let work = tactic.prep(vec![a.clone()], vec![b.clone()], fresh, 0);
+            let mut got: Vec<((Edge, Edge), Time, Diff)> =
+                work.flatten().map(|(l, r, t, d)| ((l, r), t, d)).collect();
+            consolidate_updates(&mut got);
+            assert_eq!(got, expected, "window={window}");
+        }
+    }
+}
+
+#[test]
+fn join_matches_naive() {
+    // Edges spread over several groups, several batches per side, including a
+    // retraction; window size 1 forces one group per window.
+    let a0 = batch(vec![((0, 1), 0, 1), ((0, 2), 1, 1), ((1, 5), 0, 1), ((3, 9), 2, 1)], 0, 3);
+    let a1 = batch(vec![((0, 2), 2, -1), ((2, 4), 2, 1)], 0, 3);
+    let b0 = batch(vec![((0, 7), 1, 1), ((1, 8), 2, 1), ((2, 6), 0, 1)], 0, 3);
+    let b1 = batch(vec![((1, 8), 2, 1), ((4, 4), 0, 1)], 0, 3);
+
+    let all_a: Vec<_> = a0.updates.iter().chain(&a1.updates).cloned().collect();
+    let all_b: Vec<_> = b0.updates.iter().chain(&b1.updates).cloned().collect();
+    let expected = naive_join(&all_a, &all_b);
+
+    for window in [1, 2, 100] {
+        for fresh in [Fresh::Input0, Fresh::Input1] {
+            let mut tactic = ProxyJoinTactic::new(EdgeJoinBackend::new(window));
+            let work = tactic.prep(
+                vec![a0.clone(), a1.clone()],
+                vec![b0.clone(), b1.clone()],
+                fresh,
+                0,
+            );
+            let mut got: Vec<((Edge, Edge), Time, Diff)> = work
+                .flatten()
+                .map(|(l, r, t, d)| ((l, r), t, d))
+                .collect();
+            consolidate_updates(&mut got);
+            assert_eq!(got, expected, "window={window}");
+        }
+    }
+}
+
+#[test]
+fn interleaved_units_share_backend() {
+    // Two units prepped from one tactic and drained alternately: the scenario that
+    // motivated per-unit `Cursor` state and the shared `Rc<RefCell>` backend — the
+    // driver holds half-drained units from both queues and polls them under fuel.
+    let a = vec![
+        batch(vec![((0, 1), 0, 1), ((0, 2), 0, 1), ((0, 3), 1, 1), ((1, 5), 0, 1), ((2, 2), 1, 1)], 0, 2),
+        batch(vec![((3, 3), 1, 1)], 0, 2),
+    ];
+    let b = vec![batch(vec![((0, 7), 1, 1), ((2, 6), 0, 1), ((3, 8), 1, 1)], 0, 2)];
+    let c = vec![batch(vec![((5, 1), 0, 1), ((6, 2), 0, 1)], 0, 2)];
+    let d = vec![
+        batch(vec![((5, 9), 1, 1), ((6, 4), 0, 1)], 0, 2),
+        batch(vec![((5, 9), 1, 1)], 0, 2),
+    ];
+
+    let all = |bs: &[GraphBatch]| bs.iter().flat_map(|b| b.updates.iter().cloned()).collect::<Vec<_>>();
+    let expected1 = naive_join(&all(&a), &all(&b));
+    let expected2 = naive_join(&all(&c), &all(&d));
+
+    // target = 2 with a 3-match key (key 0): unit 1 yields a full container and PAUSES
+    // holding one staged match, then unit 2 absorbs against the same shared backend.
+    // Staging must be unit-owned (`Sink`): with a shared buffer — the bug shape this
+    // test pins — unit 1's held match ships inside unit 2's next container, under
+    // unit 2's capability. target = 1 would NOT catch this (staging empty at every
+    // yield); the straddling partial container is the point.
+    let mut backend = EdgeJoinBackend::new(1);
+    backend.target = 2;
+    let mut tactic = ProxyJoinTactic::new(backend);
+    let mut u1 = tactic.prep(a, b, Fresh::Input0, 0);
+    let mut u2 = tactic.prep(c, d, Fresh::Input1, 0);
+    let mut got1: Vec<((Edge, Edge), Time, Diff)> = Vec::new();
+    let mut got2: Vec<((Edge, Edge), Time, Diff)> = Vec::new();
+    let (mut done1, mut done2) = (false, false);
+    while !done1 || !done2 {
+        if !done1 {
+            match u1.next() {
+                Some(out) => got1.extend(out.into_iter().map(|(l, r, t, d)| ((l, r), t, d))),
+                None => done1 = true,
+            }
+        }
+        if !done2 {
+            match u2.next() {
+                Some(out) => got2.extend(out.into_iter().map(|(l, r, t, d)| ((l, r), t, d))),
+                None => done2 = true,
+            }
+        }
+    }
+    consolidate_updates(&mut got1);
+    consolidate_updates(&mut got2);
+    assert_eq!(got1, expected1);
+    assert_eq!(got2, expected2);
+}
+
+#[test]
+fn reduce_fuzz_matches_oracle() {
+    // Deterministic randomized differential test: random updates over a small key
+    // space, random times within each round's interval (so rounds mix fast-path
+    // keys — one distinct time — with slow-path keys), random retractions of live
+    // edges, several window sizes. The oracle is exact per-time counting.
+    let mut state = 0x853c49e6748fea9bu64;
+    let mut rng = move || {
+        state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+        (state >> 33) as u32
+    };
+
+    for window in [1, 3, 100] {
+        let mut tactic = ProxyReduceTactic::new(EdgeReduceBackend::new(window));
+        let mut source: Vec<GraphBatch> = Vec::new();
+        let mut outputs: Vec<GraphBatch> = Vec::new();
+        let mut live: Vec<Edge> = Vec::new();
+        const ROUNDS: u64 = 8;
+        const WIDTH: u64 = 3;
+        for round in 0..ROUNDS {
+            let lo = round * WIDTH;
+            let hi = lo + WIDTH;
+            let mut updates: Vec<(Edge, Time, Diff)> = Vec::new();
+            // Retract only edges from PRIOR rounds (any time this round is at or after
+            // their insertion, keeping every prefix accumulation non-negative); this
+            // round's insertions join `live` only at round end.
+            let mut fresh: Vec<Edge> = Vec::new();
+            for _ in 0..40 {
+                let t = lo + (rng() as u64) % WIDTH;
+                if !live.is_empty() && rng() % 4 == 0 {
+                    let e = live[(rng() as usize) % live.len()];
+                    live.retain(|x| *x != e);
+                    updates.push((e, t, -1));
+                } else {
+                    let e = (rng() % 16, rng() % 64);
+                    fresh.push(e);
+                    updates.push((e, t, 1));
+                }
+            }
+            live.append(&mut fresh);
+            let input = batch(updates, lo, hi);
+            let lower = Antichain::from_elem(lo);
+            let upper = Antichain::from_elem(hi);
+            let held = Antichain::from_elem(lo);
+            let (produced, frontier) = tactic.retire(
+                source.clone(),
+                outputs.clone(),
+                vec![input.clone()],
+                &lower,
+                &upper,
+                &held,
+            );
+            assert!(frontier.is_empty(), "total order should defer nothing");
+            for (time, b) in produced {
+                assert!(held.elements().contains(&time));
+                outputs.push(b);
+            }
+            source.push(input);
+            for t in lo..hi {
+                assert_counts(&source, &outputs, t, &format!("fuzz window={window} round={round}"));
+            }
+        }
+    }
+}
+
+/// Per-time count oracle shared by the reduce tests.
+fn assert_counts(source: &[GraphBatch], outputs: &[GraphBatch], t: Time, label: &str) {
+    let input_acc = accumulate(source.iter().flat_map(|b| b.updates.iter().cloned()), &t);
+    let mut counts: BTreeMap<u32, Diff> = BTreeMap::new();
+    for ((src, _dst), d) in &input_acc {
+        *counts.entry(*src).or_insert(0) += d;
+    }
+    let mut expected: Vec<(Edge, Diff)> = Vec::new();
+    for (src, c) in counts {
+        assert!(c >= 0);
+        if c > 0 {
+            expected.push(((src, c as u32), 1));
+        }
+    }
+    let got = accumulate(outputs.iter().flat_map(|b| b.updates.iter().cloned()), &t);
+    assert_eq!(got, expected, "{label} at time={t}");
+}
+
+#[test]
+fn reduce_multitime_rounds() {
+    // Intervals of width two, two distinct times per batch: under total order this
+    // exercises the slow (discovery) path even with the single-moment fast path in
+    // place — a key with two distinct seed times cannot take it — and mixes fast
+    // and slow keys within one window.
+    let rounds: Vec<(Vec<(Edge, Time, Diff)>, Time, Time)> = vec![
+        (vec![((0, 1), 0, 1), ((0, 2), 1, 1), ((1, 5), 0, 1), ((1, 6), 1, 1)], 0, 2),
+        (vec![((0, 3), 2, 1), ((1, 5), 3, -1), ((2, 7), 2, 1), ((2, 8), 3, 1)], 2, 4),
+    ];
+    for window in [1, 100] {
+        let mut tactic = ProxyReduceTactic::new(EdgeReduceBackend::new(window));
+        let mut source: Vec<GraphBatch> = Vec::new();
+        let mut outputs: Vec<GraphBatch> = Vec::new();
+        for (updates, lo, hi) in rounds.iter() {
+            let input = batch(updates.clone(), *lo, *hi);
+            let lower = Antichain::from_elem(*lo);
+            let upper = Antichain::from_elem(*hi);
+            let held = Antichain::from_elem(*lo);
+            let (produced, frontier) = tactic.retire(
+                source.clone(),
+                outputs.clone(),
+                vec![input.clone()],
+                &lower,
+                &upper,
+                &held,
+            );
+            assert!(frontier.is_empty(), "total order should defer nothing");
+            for (time, b) in produced {
+                assert!(held.elements().contains(&time));
+                outputs.push(b);
+            }
+            source.push(input);
+        }
+        for t in 0..4u64 {
+            assert_counts(&source, &outputs, t, &format!("multitime window={window}"));
+        }
+    }
+}
+
+#[test]
+fn reduce_counts_match_naive() {
+    // Rounds of edge updates, including deletions that drop a source's count to zero
+    // (full retraction) and change another's (correction).
+    let rounds: Vec<Vec<(Edge, Time, Diff)>> = vec![
+        vec![((0, 1), 0, 1), ((0, 2), 0, 1), ((1, 5), 0, 1)],
+        vec![((0, 3), 1, 1), ((2, 7), 1, 1), ((1, 5), 1, -1)],
+        vec![((2, 8), 2, 1), ((0, 2), 2, -1), ((3, 4), 2, 1)],
+    ];
+
+    for window in [1, 2, 100] {
+        let mut tactic = ProxyReduceTactic::new(EdgeReduceBackend::new(window));
+        let mut source: Vec<GraphBatch> = Vec::new();
+        let mut outputs: Vec<GraphBatch> = Vec::new();
+
+        for (r, updates) in rounds.iter().enumerate() {
+            let r = r as Time;
+            let input = batch(updates.clone(), r, r + 1);
+            let lower = Antichain::from_elem(r);
+            let upper = Antichain::from_elem(r + 1);
+            let held = Antichain::from_elem(r);
+            let (produced, frontier) = tactic.retire(
+                source.clone(),
+                outputs.clone(),
+                vec![input.clone()],
+                &lower,
+                &upper,
+                &held,
+            );
+            assert!(frontier.is_empty(), "total order should defer nothing");
+            for (time, b) in produced {
+                assert!(held.elements().contains(&time));
+                outputs.push(b);
+            }
+            source.push(input);
+        }
+
+        // Oracle: at each round boundary, the accumulated output is exactly the
+        // per-source counts of the accumulated input.
+        for t in 0..rounds.len() as Time {
+            let input_acc = accumulate(source.iter().flat_map(|b| b.updates.iter().cloned()), &t);
+            let mut expected: Vec<(Edge, Diff)> = Vec::new();
+            let mut counts: BTreeMap<u32, Diff> = BTreeMap::new();
+            for ((src, _dst), d) in &input_acc {
+                *counts.entry(*src).or_insert(0) += d;
+            }
+            for (src, c) in counts {
+                assert!(c >= 0);
+                if c > 0 {
+                    expected.push(((src, c as u32), 1));
+                }
+            }
+            let got = accumulate(outputs.iter().flat_map(|b| b.updates.iter().cloned()), &t);
+            assert_eq!(got, expected, "window={window} at time={t}");
+        }
+    }
+}
+
+// ---------------------------------------------------- partial order (pending)
+
+type PTime = Pair<u64, u64>;
+
+fn t(a: u64, b: u64) -> PTime {
+    Pair::new(a, b)
+}
+
+#[test]
+fn pending_times_carry_across_retires() {
+    // Key 7 receives updates at incomparable times in round A; key 3 receives novel
+    // input only in round B, so round B's windows must merge a pending-only group
+    // (7) with a novel-only group (3), in group order, at window size 1.
+    for window in [1, 2, 100] {
+        let mut tactic = ProxyReduceTactic::new(EdgeReduceBackend::new(window));
+
+        // Round A: interval [ {(0,0)}, {(1,1)} ), capability held at (0,0).
+        let input_a = batch_frontiers(
+            vec![((7, 100), t(0, 1), 1), ((7, 200), t(1, 0), 1)],
+            Antichain::from_elem(t(0, 0)),
+            Antichain::from_elem(t(1, 1)),
+        );
+        let lower = Antichain::from_elem(t(0, 0));
+        let upper = Antichain::from_elem(t(1, 1));
+        let held = Antichain::from_elem(t(0, 0));
+        let (produced_a, frontier_a) = tactic.retire(
+            Vec::new(),
+            Vec::new(),
+            vec![input_a.clone()],
+            &lower,
+            &upper,
+            &held,
+        );
+        // The joint interesting time (1,1) lies beyond upper: it must be deferred,
+        // and the returned frontier must hold it.
+        assert_eq!(frontier_a, Antichain::from_elem(t(1, 1)), "the join of incomparable updates must be pended");
+
+        let mut outputs: Vec<EdgeBatch<PTime>> = produced_a.into_iter().map(|(_, b)| b).collect();
+
+        // Round B: interval [ {(1,1)}, {(2,2)} ), capability held at (1,1). Novel
+        // input only for key 3; key 7 participates as a pending-only group.
+        let input_b = batch_frontiers(
+            vec![((3, 5), t(1, 1), 1)],
+            Antichain::from_elem(t(1, 1)),
+            Antichain::from_elem(t(2, 2)),
+        );
+        let lower = Antichain::from_elem(t(1, 1));
+        let upper = Antichain::from_elem(t(2, 2));
+        let held = Antichain::from_elem(t(1, 1));
+        let (produced_b, frontier_b) = tactic.retire(
+            vec![input_a.clone()],
+            outputs.clone(),
+            vec![input_b.clone()],
+            &lower,
+            &upper,
+            &held,
+        );
+        assert!(frontier_b.is_empty(), "nothing further to defer");
+        outputs.extend(produced_b.into_iter().map(|(_, b)| b));
+
+        // Oracle over a grid of times, using the partial order's less_equal: the
+        // accumulated output at t must be exactly the per-source counts of the
+        // accumulated input at t. At (1,1) this requires round B's correction to
+        // have collapsed the two partial counts of key 7 into one count of 2.
+        let inputs: Vec<(Edge, PTime, Diff)> = input_a
+            .updates
+            .iter()
+            .chain(&input_b.updates)
+            .cloned()
+            .collect();
+        for probe in [t(0, 0), t(0, 1), t(1, 0), t(1, 1), t(2, 1), t(2, 2)] {
+            let input_acc = accumulate(inputs.iter().cloned(), &probe);
+            let mut counts: BTreeMap<u32, Diff> = BTreeMap::new();
+            for ((src, _dst), d) in &input_acc {
+                *counts.entry(*src).or_insert(0) += d;
+            }
+            let mut expected: Vec<(Edge, Diff)> = Vec::new();
+            for (src, c) in counts {
+                if c > 0 {
+                    expected.push(((src, c as u32), 1));
+                }
+            }
+            let got = accumulate(outputs.iter().flat_map(|b| b.updates.iter().cloned()), &probe);
+            assert_eq!(got, expected, "window={window} at probe={probe:?}");
+        }
+    }
+}