diff --git a/differential-dataflow/src/operators/int_proxy/history.rs b/differential-dataflow/src/operators/int_proxy/history.rs
index 83d8369fc..81a98c9ba 100644
--- a/differential-dataflow/src/operators/int_proxy/history.rs
+++ b/differential-dataflow/src/operators/int_proxy/history.rs
@@ -1,4 +1,4 @@
 //! 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::history::ValueHistory<u64, T, R>;
+/// A value history suitable for ordered proxy values.
+pub(in crate::operators) type IdHistory<T, R, V = u64> = crate::operators::history::ValueHistory<V, T, R>;
diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 31986ca00..eb1baee51 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -1,7 +1,7 @@
 //! The proxy join framework.
 //!
-//! A conventional differential join against `(u64, u64)` values, which are provided by
-//! and then interpreted by a backend, who is relieved of lattice-time reasoning.
+//! A conventional differential join against ordered, copyable proxies.
+//! The backend interprets proxies and is relieved of lattice-time reasoning.
 
 use std::cell::RefCell;
 use std::rc::Rc;
@@ -10,17 +10,23 @@ use timely::progress::Timestamp;
 
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
-use super::ProxyBridge;
+use super::{KeyPosition, ProxyBridge};
 use crate::operators::join::{Fresh, JoinTactic};
 use crate::operators::history::ValueHistory;
 
 use super::history::IdHistory;
 
-/// A type that can interpret and retire pairs of lists of batches, joined by key hashes.
+/// A type that can interpret and retire pairs of lists of batches, joined by keys.
 ///
 /// The harness repeatedly invokes [`advance`](Self::advance) to draw a block of the proxy collection,
 /// then [`cross`](Self::cross) to turn that block's matches into output containers, until `advance` reports the key space exhausted.
 pub trait ProxyJoinBackend<T, B0, B1> {
+    /// Independent groups, common to both inputs.
+    type Key: Copy + Ord;
+    /// First input value proxies, valid throughout a block.
+    type V0: Copy + Ord;
+    /// Second input value proxies, valid throughout a block.
+    type V1: Copy + Ord;
     /// Diff type of the first input.
     type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
     /// Diff type of the second input.
@@ -32,16 +38,17 @@ pub trait ProxyJoinBackend<T, B0, B1> {
 
     /// Populates the two bridges with all updates for all keys that match in a returned range.
     ///
-    /// The `from` indicates an inclusive lower bound on key hash, and should be updated by the implementor to an exclusive
-    /// upper bound for the range of keys it intends to return in this call. The `None` value indicates the keys are exhausted.
+    /// The `from` position is `Start` or an inclusive `At(key)` lower bound.
+    /// The backend writes the window's exclusive upper bound as `At(key)`, or `End` when exhausted.
+    /// Each call must strictly advance `from`.
     /// The returned bridges must contain all updates from both `instance` inputs for keys that are present in both inputs, and
     /// which are greater or equal to the initial `from`, and not greater or equal to its value when returned.
     fn advance(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
-        from: &mut Option<u64>,
-        bridge0: &mut ProxyBridge<T, Self::R0>,
-        bridge1: &mut ProxyBridge<T, Self::R1>,
+        from: &mut KeyPosition<Self::Key>,
+        bridge0: &mut ProxyBridge<T, Self::R0, Self::Key, Self::V0>,
+        bridge1: &mut ProxyBridge<T, Self::R1, Self::Key, Self::V1>,
     );
 
     /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place
@@ -51,7 +58,7 @@ pub trait ProxyJoinBackend<T, B0, B1> {
     fn cross(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
-        matches: &mut JoinMatches<T, Self::ROut>,
+        matches: &mut JoinMatches<T, Self::ROut, Self::Key, Self::V0, Self::V1>,
         output: &mut Vec<Self::Output>,
     );
 }
@@ -71,20 +78,22 @@ pub struct JoinInstance<T, B0, B1> {
 /// Presentation of discovered join matches.
 ///
 /// The arrays have common lengths, and are in key order but may not be consolidated.
-pub struct JoinMatches<T, R> {
+pub struct JoinMatches<T, R, K = u64, V0 = u64, V1 = u64> {
     /// Triples of `(key, (val0, val1))` of matches.
-    pub ids: Vec<(u64, (u64, u64))>,
+    pub ids: Vec<(K, (V0, V1))>,
     /// Times of the updates.
     pub times: Vec<T>,
     /// Diffs of the updates.
     pub diffs: Vec<R>,
 }
 
-impl<T, R> Default for JoinMatches<T, R> {
+impl<T, R, K, V0, V1> Default for JoinMatches<T, R, K, V0, V1> {
     fn default() -> Self { Self { ids: vec![], times: vec![], diffs: vec![] } }
 }
 
-/// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key_hash`.
+/// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key`.
+/// Its deferred work owns the presentations, so proxy types must be `'static`.
+/// Borrowing proxies from a backend's temporary storage requires a different presentation interface.
 pub struct ProxyJoinTactic<B0, B1, Bk> {
     backend: Rc<RefCell<Bk>>,
     _marker: std::marker::PhantomData<(B0, B1)>,
@@ -104,12 +113,15 @@ where
     B1: 'static,
     Bk: ProxyJoinBackend<T, B0, B1> + 'static,
     Bk::Output: 'static,
+    Bk::Key: 'static,
+    Bk::V0: 'static,
+    Bk::V1: 'static,
 {
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, _fresh: Fresh, meet: T) -> Box<dyn Iterator<Item = Bk::Output>> {
         Box::new(ProxyJoinIter {
             backend: Rc::clone(&self.backend),
             instance: JoinInstance { batches0: input0, batches1: input1, lower: meet },
-            from: Some(0),
+            from: KeyPosition::Start,
             p0: Vec::new(),
             p1: Vec::new(),
             h0: IdHistory::new(),
@@ -133,17 +145,16 @@ where
     backend: Rc<RefCell<Bk>>,
     /// The iterator's inputs, and the time at which they can consolidate as they load.
     instance: JoinInstance<T, B0, B1>,
-    /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
-    /// once the backend reports the iteration is complete.
-    from: Option<u64>,
+    /// Progress through the key space, from `Start` through inclusive key bounds to `End`.
+    from: KeyPosition<Bk::Key>,
     /// The current block: the two runs `advance` last drew, which one `next` consumes entirely.
-    p0: ProxyBridge<T, Bk::R0>,
-    p1: ProxyBridge<T, Bk::R1>,
+    p0: ProxyBridge<T, Bk::R0, Bk::Key, Bk::V0>,
+    p1: ProxyBridge<T, Bk::R1, Bk::Key, Bk::V1>,
     /// Per-key replay histories, held across the iterator and reloaded per key when needed.
-    h0: IdHistory<T, Bk::R0>,
-    h1: IdHistory<T, Bk::R1>,
+    h0: IdHistory<T, Bk::R0, Bk::V0>,
+    h1: IdHistory<T, Bk::R1, Bk::V1>,
     /// The block's matched records, held across blocks to keep their allocations.
-    matches: JoinMatches<T, Bk::ROut>,
+    matches: JoinMatches<T, Bk::ROut, Bk::Key, Bk::V0, Bk::V1>,
     /// The last block's containers, in reverse, served from the back one `next` at a time.
     ready: Vec<Bk::Output>,
 }
@@ -157,7 +168,7 @@ where
 
     /// Serve a ready container, else draw and cross blocks until one yields any.
     fn next(&mut self) -> Option<Bk::Output> {
-        while self.ready.is_empty() && self.from.is_some() {
+        while self.ready.is_empty() && self.from != KeyPosition::End {
             self.refill();
             self.work();
             if !self.matches.ids.is_empty() { self.cross(); }
@@ -179,20 +190,19 @@ where
         self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
         // Without progress the iterator would never retire, so this guards liveness as well as contract.
         debug_assert!(
-            self.from.is_none() || self.from > before,
+            self.from > before,
             "advance must either strictly increase `from` or report the iteration complete",
         );
         super::debug_assert_sorted_bridge(&self.p0, "advance (bridge0)");
         super::debug_assert_sorted_bridge(&self.p1, "advance (bridge1)");
-        // A key hash outside `[before, from)` is either one an earlier block already retired, or one
-        // a later block may yet report: both split a key across blocks, which silently drops the
-        // matches that would have crossed the split.
+        // A key outside `[before, from)` belongs to an earlier or later block.
+        // Both cases split a key across blocks, silently dropping matches that would cross the split.
         debug_assert!(
             {
                 let mut keys = self.p0.iter().map(|r| r.0.0).chain(self.p1.iter().map(|r| r.0.0));
-                keys.all(|k| before.is_none_or(|b| b <= k) && self.from.is_none_or(|f| k < f))
+                keys.all(|k| before <= KeyPosition::At(k) && KeyPosition::At(k) < self.from)
             },
-            "advance must report a key hash entirely within the block that first mentions it",
+            "advance must report a key entirely within the block that first mentions it",
         );
     }
 
@@ -205,7 +215,7 @@ where
         let (mut i, mut j) = (0usize, 0usize);
         while i < p0.len() && j < p1.len() {
             let ki = p0[i].0.0;
-            debug_assert_eq!(ki, p1[j].0.0, "advance must report common keys");
+            debug_assert!(ki == p1[j].0.0, "advance must report common keys");
             let mut e0 = i;
             while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
             let mut e1 = j;
@@ -237,15 +247,15 @@ where
 /// If either history is small, this performs a direct cross product.
 /// If both histories are large, this replays the histories compacting as it goes in
 /// order to (potentially) avoid quadratic blow-up.
-fn join_key<T, R0, R1, RO>(
-    kh: u64,
-    p0: &ProxyBridge<T, R0>,
+fn join_key<T, R0, R1, RO, K: Copy + Ord, V0: Copy + Ord, V1: Copy + Ord>(
+    kh: K,
+    p0: &ProxyBridge<T, R0, K, V0>,
     r0: std::ops::Range<usize>,
-    p1: &ProxyBridge<T, R1>,
+    p1: &ProxyBridge<T, R1, K, V1>,
     r1: std::ops::Range<usize>,
-    h0: &mut IdHistory<T, R0>,
-    h1: &mut IdHistory<T, R1>,
-    matches: &mut JoinMatches<T, RO>,
+    h0: &mut IdHistory<T, R0, V0>,
+    h1: &mut IdHistory<T, R1, V1>,
+    matches: &mut JoinMatches<T, RO, K, V0, V1>,
 ) where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
@@ -280,12 +290,13 @@ fn join_key<T, R0, R1, RO>(
 /// `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.
-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),
+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,
diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs
index 50764452a..15dbf0fe5 100644
--- a/differential-dataflow/src/operators/int_proxy/mod.rs
+++ b/differential-dataflow/src/operators/int_proxy/mod.rs
@@ -1,17 +1,19 @@
-//! Backend-agnostic operator tactics using integer proxies.
+//! Backend-agnostic operator tactics using ordered, copyable 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 support custom operator implementations without rebuilding the non-trivial and often non-obvious time-based logic that supports them.
+//! They run DD's operator logic over consolidated `[((key, val), time, diff)]` lists.
+//! Keys identify independent groups; values identify data within a group.
+//! Both require only `Copy + Ord` and need not be the types presented to user logic.
+//! The tactics elicit proxies, perform time and difference computations, and return proxies to the backend for interpretation.
+//! The backend is oblivious to the navigation of time, and the operator to the backend's implementation.
 //!
-//! 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 default proxy types remain `u64`, with the interpretation described below.
+//! Exact key proxies do not require collision handling.
 //!
-//! The backend is oblivious to the navigation of time, and the operator to the backend's
-//! implementation.
+//! Input and output value proxies may have different types.
+//! Standalone reduce calls can use borrowed strings whose storage outlives the call and any pending keys.
+//! Deferred join work still requires `'static` proxies.
+//! Borrowing directly from backend-owned presentation storage needs a further lifetime or container abstraction.
 //!
 //! # The two integers
 //!
@@ -49,21 +51,33 @@ pub mod reduce;
 mod pending;
 pub mod vec_backend;
 
-/// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`.
+/// A consolidated collection of `[((key, val), time, diff)]` proxies.
 ///
 /// 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)>;
+pub type ProxyBridge<T, R, K = u64, V = u64> = Vec<((K, V), 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<T: Ord, R, K: Copy + Ord, V: Copy + Ord>(bridge: &ProxyBridge<T, R, K, V>, 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)",
+        "{}: a presented bridge must be sorted & consolidated by ((key, val), time)",
         who,
     );
 }
 
 pub use join::{JoinInstance, ProxyJoinBackend, ProxyJoinTactic};
 pub use reduce::{ProxyReduceBackend, ProxyReduceTactic, ReduceInstance, ReduceWindow};
+
+/// A position between windows of an ordered key space.
+/// No minimum key or successor operation is required.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum KeyPosition<K> {
+    /// No keys have been visited.
+    Start,
+    /// The inclusive lower bound on keys still to visit.
+    At(K),
+    /// All keys have been visited.
+    End,
+}
diff --git a/differential-dataflow/src/operators/int_proxy/pending.rs b/differential-dataflow/src/operators/int_proxy/pending.rs
index 35821a55d..534badf91 100644
--- a/differential-dataflow/src/operators/int_proxy/pending.rs
+++ b/differential-dataflow/src/operators/int_proxy/pending.rs
@@ -3,20 +3,20 @@ use std::ops::Range;
 use timely::progress::{Antichain, frontier::AntichainRef, Timestamp};
 
 /// Activated associations refer to a shared time table, sorted by (key, time).
-pub(super) struct Due<T> {
+pub(super) struct Due<T, K> {
     pub times: Vec<T>,
-    pub rows: Vec<(u64, usize)>,
+    pub rows: Vec<(K, usize)>,
 }
 
-struct Run<T> {
+struct Run<T, K> {
     times: Vec<T>,
-    keys: Vec<u64>,
+    keys: Vec<K>,
     ends: Vec<usize>,
     live: Vec<usize>,
     live_keys: usize,
     frontier: Antichain<T>,
 }
-impl<T: Timestamp> Run<T> {
+impl<T: Timestamp, K: Copy + Ord> Run<T, K> {
     fn empty() -> Self {
         Self { times: vec![], keys: vec![], ends: vec![], live: vec![], live_keys: 0, frontier: Antichain::new() }
     }
@@ -24,7 +24,7 @@ impl<T: Timestamp> Run<T> {
         (if row == 0 { 0 } else { self.ends[row - 1] })..self.ends[row]
     }
     fn weight(&self) -> usize { self.live.len() + self.live_keys }
-    fn new(mut pairs: Vec<(T, u64)>) -> Self {
+    fn new(mut pairs: Vec<(T, K)>) -> Self {
         pairs.sort_unstable();
         pairs.dedup();
         let mut run = Self::empty();
@@ -83,12 +83,12 @@ impl<T: Timestamp> Run<T> {
     }
 }
 
-pub(super) struct Pending<T> { runs: Vec<Option<Run<T>>> }
-impl<T> Default for Pending<T> {
+pub(super) struct Pending<T, K = u64> { runs: Vec<Option<Run<T, K>>> }
+impl<T, K> Default for Pending<T, K> {
     fn default() -> Self { Self { runs: vec![] } }
 }
-impl<T: Timestamp> Pending<T> {
-    pub fn insert(&mut self, pairs: Vec<(T, u64)>) {
+impl<T: Timestamp, K: Copy + Ord> Pending<T, K> {
+    pub fn insert(&mut self, pairs: Vec<(T, K)>) {
         if pairs.is_empty() { return; }
         let mut run = Run::new(pairs);
         loop {
@@ -105,7 +105,7 @@ impl<T: Timestamp> Pending<T> {
         }
         frontier
     }
-    pub fn activate(&mut self, upper: AntichainRef<T>) -> Due<T> {
+    pub fn activate(&mut self, upper: AntichainRef<T>) -> Due<T, K> {
         let mut due = Due { times: vec![], rows: vec![] };
         for bin in &mut self.runs {
             let Some(run) = bin else { continue; };
@@ -133,7 +133,7 @@ impl<T: Timestamp> Pending<T> {
                 *bin = Some(old.merge(Run::empty()));
             }
         }
-        // Rank the small time table once; association sorting then compares only integers.
+        // Rank the small time table once; association sorting then compares keys and integer time ranks.
         let mut order: Vec<_> = (0..due.times.len()).collect();
         order.sort_unstable_by_key(|&row| &due.times[row]);
         let mut ranks = vec![0; order.len()];
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 2235d0aae..9107ff350 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -1,7 +1,7 @@
 //! 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 ordered, copyable proxies.
+//! The backend supplies their interpretation.
 
 use super::pending::Pending;
 
@@ -11,7 +11,7 @@ use timely::progress::frontier::AntichainRef;
 use crate::difference::Semigroup;
 use crate::lattice::Lattice;
 use crate::trace::{Span, Description};
-use super::ProxyBridge;
+use super::{KeyPosition, ProxyBridge};
 use crate::operators::reduce::{sort_dedup, ReduceTactic};
 use crate::operators::history::ValueHistory;
 
@@ -27,7 +27,7 @@ pub struct ReduceInstance<'a, T, B1, B2> {
     pub lower: AntichainRef<'a, T>,
 }
 
-/// One window of the key space: the presentations a bounded, hash-contiguous snip needs.
+/// One window of the key space: the presentations a bounded, key-contiguous snip needs.
 ///
 /// Seeds travel as times; records travel netted. The novel data's two roles are carried by two
 /// different channels: its TIME SUPPORT seeds interesting times and rides `seeds`, raw; its
@@ -37,23 +37,23 @@ pub struct ReduceInstance<'a, T, B1, B2> {
 /// interesting time is lost to netting — the invariant that once forced the runs apart.
 ///
 /// Owned by the harness and refilled by [`ProxyReduceBackend::next_window`].
-pub struct ReduceWindow<T, RIn, ROut> {
-    /// The key's full input — novel and prior merged, netted — sorted & consolidated by
-    /// `((key_hash, value_id), time)`. May be advanced to the compaction frontier.
-    pub input: ProxyBridge<T, RIn>,
-    /// The RAW novel time support: `(key_hash, time)` pairs sorted by `(key_hash, time)` and
-    /// deduplicated, recorded from the novel batches BEFORE any consolidation or advancement —
-    /// a netted-away record's time must still appear here.
-    pub seeds: Vec<(u64, T)>,
+pub struct ReduceWindow<T, RIn, ROut, K = u64, VIn = u64, VOut = u64> {
+    /// The key's full input — novel and prior merged, netted — sorted & consolidated by `((key, value_id), time)`.
+    /// May be advanced to the compaction frontier.
+    pub input: ProxyBridge<T, RIn, K, VIn>,
+    /// The RAW novel time support: `(key, time)` pairs sorted by `(key, time)` and deduplicated.
+    /// Record these from the novel batches BEFORE any consolidation or advancement.
+    /// A netted-away record's time must still appear here.
+    pub seeds: Vec<(K, T)>,
     /// Accumulated output preceding the retire's interval, same ordering as `input`.
-    pub output: ProxyBridge<T, ROut>,
+    pub output: ProxyBridge<T, ROut, K, VOut>,
 }
 
-impl<T, RIn, ROut> Default for ReduceWindow<T, RIn, ROut> {
+impl<T, RIn, ROut, K, VIn, VOut> Default for ReduceWindow<T, RIn, ROut, K, VIn, VOut> {
     fn default() -> Self { ReduceWindow { input: Vec::new(), seeds: Vec::new(), output: Vec::new() } }
 }
 
-impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
+impl<T, RIn, ROut, K, VIn, VOut> ReduceWindow<T, RIn, ROut, K, VIn, VOut> {
     /// Clear the presentations, keeping their allocations.
     pub fn clear(&mut self) {
         self.input.clear();
@@ -68,6 +68,12 @@ impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
 /// `begin [ next_window reduce_corrections* emit ]* finish`,
 /// where the window loop runs until `next_window` reports the key space exhausted.
 pub trait ProxyReduceBackend<T, B1, B2> {
+    /// Independent groups, with identities stable across retires.
+    type Key: Copy + Ord;
+    /// Input value proxies, valid throughout a window.
+    type VIn: Copy + Ord;
+    /// Output value proxies, including newly produced values, valid throughout a window.
+    type VOut: Copy + Ord;
     /// Diff type presented for the input.
     type RIn: Semigroup;
     /// Diff type of the output.
@@ -82,17 +88,14 @@ pub trait ProxyReduceBackend<T, B1, B2> {
 
     /// Present the next window of the key space, and advance `from` past it.
     ///
-    /// On entry `from` is the inclusive lower bound on key hashes still to be covered. The backend
-    /// chooses the window's exclusive upper bound and writes it back, or writes `None` to report the
-    /// key space exhausted. An implementor must advance `from`, as it is guaranteed to be non-`None`.
+    /// On entry `from` is `Start` or an inclusive `At(key)` lower bound on keys still to be covered.
+    /// The backend writes the window's exclusive upper bound as `At(key)`, or `End` when exhausted.
+    /// Each call must strictly advance `from`.
     ///
-    /// The window must present, for every key hash in `[from_before, from_after)` that either
-    /// carries an update in the instance's novel batches or appears in `changed`: that key's merged
-    /// input (novel and prior together, netted), its raw novel time support in `seeds`, and its
-    /// accumulated output. A key must be reported entirely within the window that first mentions
-    /// it: splitting one across windows drops the interaction between the halves. `changed` is
-    /// ascending; the harness reads no key outside the window's range, so a backend that keeps its
-    /// own key order need not consult the whole space.
+    /// The window must cover every key in `[from_before, from_after)` that carries a novel update or appears in `changed`.
+    /// For each key, present merged input (novel and prior together, netted), raw novel time support in `seeds`, and accumulated output.
+    /// A key must be reported entirely within the window that first mentions it: splitting one across windows drops the interaction between the halves.
+    /// `changed` is ascending; the harness reads no key outside the window's range, so a backend that keeps its own key order need not consult the whole space.
     ///
     /// `seeds` must be recorded from the novel batches before any consolidation or advancement:
     /// a novel record that nets to zero against compacted history vanishes from `input`, but its
@@ -103,9 +106,9 @@ pub trait ProxyReduceBackend<T, B1, B2> {
     fn next_window(
         &mut self,
         instance: &ReduceInstance<'_, T, B1, B2>,
-        changed: &[u64],
-        from: &mut Option<u64>,
-        window: &mut ReduceWindow<T, Self::RIn, Self::ROut>,
+        changed: &[Self::Key],
+        from: &mut KeyPosition<Self::Key>,
+        window: &mut ReduceWindow<T, Self::RIn, Self::ROut, Self::Key, Self::VIn, Self::VOut>,
     );
 
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
@@ -115,31 +118,31 @@ pub trait ProxyReduceBackend<T, B1, B2> {
     /// with its desires. The `usize` integers upper bound the range for the corresponding key.
     fn reduce_corrections(
         &mut self,
-        keys: &[u64],
+        keys: &[Self::Key],
         in_ends: &[usize],
-        input: &[(u64, Self::RIn)],
+        input: &[(Self::VIn, Self::RIn)],
         out_ends: &[usize],
-        output: &[(u64, Self::ROut)],
-    ) -> (Vec<(u64, Self::ROut)>, Vec<usize>);
+        output: &[(Self::VOut, Self::ROut)],
+    ) -> (Vec<(Self::VOut, Self::ROut)>, Vec<usize>);
 
     /// Commit a collection of updates to the batch in progress.
-    fn emit(&mut self, records: &[((u64, u64), T, Self::ROut)]);
+    fn emit(&mut self, records: &[((Self::Key, Self::VOut), T, Self::ROut)]);
 
     /// Complete the session matching `begin`, yielding the batch it described,
     /// or `None` when the span it described carries no updates.
     fn finish(&mut self) -> Option<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 `key`.
+pub struct ProxyReduceTactic<T, Bk, K = u64> {
     backend: Bk,
-    /// Maximum number of key hashes with live sweep state at once.
+    /// Maximum number of keys with live sweep state at once.
     key_batch_size: usize,
     /// Pending interesting times, shared across flat key ranges.
-    pending: Pending<T>,
+    pending: Pending<T, K>,
 }
 
-impl<T, Bk> ProxyReduceTactic<T, Bk> {
+impl<T, Bk, K> ProxyReduceTactic<T, Bk, K> {
     /// A tactic deferring all value semantics to `backend`.
     pub fn new(backend: Bk) -> Self {
         ProxyReduceTactic { backend, key_batch_size: usize::MAX, pending: Pending::default() }
@@ -147,7 +150,7 @@ impl<T, Bk> ProxyReduceTactic<T, Bk> {
 
     /// Limit simultaneous sweeps independently of the backend's presentation window.
     ///
-    /// Complete key hashes stay together, including all real keys sharing a hash.
+    /// Complete proxy keys stay together, including all real keys sharing a proxy.
     /// Sweep scratch is reused between groups within a retire. The bound does not
     /// limit a single key's size, the presentation, or the output batch. Corrections
     /// remain batched, and emission still happens once per backend window.
@@ -159,10 +162,10 @@ impl<T, Bk> ProxyReduceTactic<T, Bk> {
     }
 }
 
-impl<T, B1, B2, Bk> ReduceTactic<T, B1, B2> for ProxyReduceTactic<T, Bk>
+impl<T, B1, B2, Bk, K: Copy + Ord> ReduceTactic<T, B1, B2> for ProxyReduceTactic<T, Bk, K>
 where
     T: Timestamp + Lattice,
-    Bk: ProxyReduceBackend<T, B1, B2>,
+    Bk: ProxyReduceBackend<T, B1, B2, Key = K>,
 {
     fn retire(
         &mut self,
@@ -195,7 +198,7 @@ where
         let due = self.pending.activate(upper.borrow());
         // The keys the harness knows must be revisited. The backend adds those its novel batches
         // touch, which it discovers while reading them; neither side scans the whole key space.
-        let mut changed: Vec<u64> = due.rows.iter().map(|r| r.0).collect();
+        let mut changed: Vec<K> = due.rows.iter().map(|r| r.0).collect();
         changed.dedup();
         let mut deferred = Vec::new();
         let mut due_pos = 0;
@@ -213,27 +216,26 @@ where
         let description = Description::new(lower.clone(), upper.clone(), Antichain::from_elem(T::minimum()));
         self.backend.begin(description.clone());
 
-        // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
-        // once the backend reports the space covered.
-        let mut from = Some(0u64);
-        let mut window: ReduceWindow<T, Bk::RIn, Bk::ROut> = ReduceWindow::default();
+        // Start before every key; each window advances to an inclusive key bound or the end.
+        let mut from = KeyPosition::Start;
+        let mut window: ReduceWindow<T, Bk::RIn, Bk::ROut, K, Bk::VIn, Bk::VOut> = ReduceWindow::default();
 
         // Retire-wide reusable scratch: cleared per group, window or wave, retaining capacity. Fresh
         // per-key/per-wave `Vec`s were once the dominant cost here, which is why the slots and the
         // staging buffers are held across the whole retire rather than built where they are used.
-        let mut slots: Vec<KeySweep<T, Bk::RIn, Bk::ROut>> = Vec::new();
+        let mut slots: Vec<KeySweep<T, Bk::RIn, Bk::ROut, K, Bk::VIn, Bk::VOut>> = Vec::new();
         let mut live: Vec<usize> = Vec::new();
-        let mut deltas: Vec<((u64, u64), T, Bk::ROut)> = Vec::new();
-        let mut batch_keys: Vec<u64> = Vec::new();
+        let mut deltas: Vec<((K, Bk::VOut), T, Bk::ROut)> = Vec::new();
+        let mut batch_keys: Vec<K> = 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::VIn, 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::VOut, Bk::ROut)> = Vec::new();
         let mut active: Vec<(usize, T)> = 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::VIn, Bk::RIn)> = Vec::new();
+        let mut cur_out: Vec<(Bk::VOut, Bk::ROut)> = Vec::new();
 
-        while from.is_some() {
+        while from != KeyPosition::End {
             let before = from;
             window.clear();
             self.backend.next_window(&instance, &changed, &mut from, &mut window);
@@ -244,26 +246,26 @@ where
             super::debug_assert_sorted_bridge(p_out, "next_window.output");
             debug_assert!(
                 seeds.windows(2).all(|w| w[0] < w[1]),
-                "next_window.seeds must be sorted by (key_hash, time) and deduplicated",
+                "next_window.seeds must be sorted by (key, time) and deduplicated",
             );
             // Without progress the window loop would never retire, so this guards liveness as well
             // as contract; the range check catches a key reported outside the window that owns it,
             // which would silently drop the interaction between its halves.
             debug_assert!(
-                from.is_none() || from > before,
+                from > before,
                 "next_window must either advance `from` or report the key space exhausted",
             );
             debug_assert!(
                 {
                     let mut keys = p_in.iter().map(|r| r.0.0).chain(seeds.iter().map(|s| s.0)).chain(p_out.iter().map(|r| r.0.0));
-                    keys.all(|k| before.is_none_or(|b| b <= k) && from.is_none_or(|f| k < f))
+                    keys.all(|k| before <= KeyPosition::At(k) && KeyPosition::At(k) < from)
                 },
-                "next_window must report a key hash entirely within the window that first mentions it",
+                "next_window must report a key entirely within the window that first mentions it",
             );
 
             deltas.clear();
 
-            // The window's keys are the hashes its presentations mention: the least of the three
+            // The window's keys are the proxies its presentations mention: the least of the three
             // heads, each iteration, until all three are drained. A `changed` key that appears in
             // none of them has no records at all, so its reduction has nothing to read and nothing
             // to retract — the time its due moment would raise reaches the evaluation gate with an
@@ -279,7 +281,7 @@ where
             while is < p_in.len() || ns < seeds.len() || os < p_out.len() {
                 let mut n_slots = 0usize;
                 live.clear();
-                // Mapped to hashes before the min: the sources differ in shape.
+                // Mapped to keys before the min: the sources differ in shape.
                 while let Some(key) = [
                     p_in.get(is).map(|record| record.0.0),
                     seeds.get(ns).map(|seed| seed.0),
@@ -295,7 +297,7 @@ where
                     while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
                     let o1 = os;
 
-                    if n_slots == slots.len() { slots.push(KeySweep::empty()); }
+                    if n_slots == slots.len() { slots.push(KeySweep::empty(key)); }
                     let slot = &mut slots[n_slots];
                     slot.key = key;
                     slot.pended.clear();
@@ -414,9 +416,9 @@ where
 /// One key's slot in a window: its [`Sweep`], the time it is suspended at, and the times it has
 /// pended so far. Slots and their scratch capacity are reused across groups and windows
 /// within a retire, then dropped when the retire completes.
-struct KeySweep<T, RIn, ROut> {
-    key: u64,
-    sweep: Sweep<T, RIn, ROut>,
+struct KeySweep<T, RIn, ROut, K, VIn, VOut> {
+    key: K,
+    sweep: Sweep<T, RIn, ROut, VIn, VOut>,
     direct: Option<(std::ops::Range<usize>, std::ops::Range<usize>)>,
     /// Times at or beyond `upper` the sweep has reached; carried forward when the slot retires.
     pended: Vec<T>,
@@ -424,9 +426,9 @@ struct KeySweep<T, RIn, ROut> {
     at: Option<T>,
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> KeySweep<T, RIn, ROut> {
-    fn empty() -> Self {
-        KeySweep { key: 0, sweep: Sweep::new(), direct: None, pended: Vec::new(), at: None }
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone, K, VIn: Copy + Ord, VOut: Copy + Ord> KeySweep<T, RIn, ROut, K, VIn, VOut> {
+    fn empty(key: K) -> Self {
+        KeySweep { key, sweep: Sweep::new(), direct: None, pended: Vec::new(), at: None }
     }
 }
 
@@ -463,12 +465,12 @@ fn update_meet<T: Lattice + Clone>(meet: &mut Option<T>, other: Option<&T>) {
 /// witness duty, which is what lets them net and advance. Coverage is invariant under that move:
 /// the witness clause reads only times, and consolidation cancels only equal-time pairs whose time
 /// the seed list retains.
-struct Sweep<T, RIn, ROut> {
+struct Sweep<T, RIn, ROut, VIn, VOut> {
     /// The accumulated input (novel and prior, merged and netted) and output: join partners, and
     /// the accumulations to evaluate over. Both may be advanced freely — witness duty lives in
     /// `seeds`, not in any record.
-    input: ValueHistory<u64, T, RIn>,
-    output: ValueHistory<u64, T, ROut>,
+    input: ValueHistory<VIn, T, RIn>,
+    output: ValueHistory<VOut, T, ROut>,
     /// The key's seed times — the harness's due (warned) times merged with the raw novel time
     /// support — ascending and deduplicated, with their suffix meets; `seed_pos` consumes them.
     /// These are the ONLY source of interest: the schedule is stated over them, so they are held
@@ -486,7 +488,7 @@ struct Sweep<T, RIn, ROut> {
     temporary: Vec<T>,
     /// Corrections emitted so far this sweep, meet-collapsed; both a join partner and part of the
     /// output accumulation.
-    produced: Vec<((u64, T), ROut)>,
+    produced: Vec<((VOut, T), ROut)>,
     /// The meet of every time still to come.
     meet: Option<T>,
     /// Whether the last `next_crossing` returned a time whose step is not yet settled.
@@ -505,7 +507,7 @@ enum Tick<T> {
     Done,
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sweep<T, RIn, ROut> {
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone, VIn: Copy + Ord, VOut: Copy + Ord> Sweep<T, RIn, ROut, VIn, VOut> {
     /// An empty sweep, to be `load`ed and reused for successive keys.
     fn new() -> Self {
         Sweep {
@@ -528,8 +530,8 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sw
         &mut self,
         owed: impl Iterator<Item = T>,
         novel_times: impl Iterator<Item = T>,
-        input: impl Iterator<Item = (u64, T, RIn)>,
-        output: impl Iterator<Item = (u64, T, ROut)>,
+        input: impl Iterator<Item = (VIn, T, RIn)>,
+        output: impl Iterator<Item = (VOut, T, ROut)>,
     ) {
         // Merge the two ascending seed sources, deduplicated.
         self.seeds.clear();
@@ -696,7 +698,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sw
     }
 
     /// The input accumulation at the suspended time.
-    fn input_at(&self, at: &T, into: &mut Vec<(u64, RIn)>) {
+    fn input_at(&self, at: &T, into: &mut Vec<(VIn, RIn)>) {
         for ((id, time), diff) in self.input.buffer().iter() {
             if time.less_equal(at) { into.push((*id, diff.clone())); }
         }
@@ -704,7 +706,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sw
     }
 
     /// The tentative output accumulation at the suspended time, including this sweep's corrections.
-    fn output_at(&self, at: &T, into: &mut Vec<(u64, ROut)>) {
+    fn output_at(&self, at: &T, into: &mut Vec<(VOut, ROut)>) {
         for ((id, time), diff) in self.output.buffer().iter().chain(self.produced.iter()) {
             if time.less_equal(at) { into.push((*id, diff.clone())); }
         }
@@ -712,7 +714,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sw
     }
 
     /// Record the corrections evaluated at the suspended time, and collapse them by the meet.
-    fn commit(&mut self, at: &T, corrections: impl Iterator<Item = (u64, ROut)>) {
+    fn commit(&mut self, at: &T, corrections: impl Iterator<Item = (VOut, ROut)>) {
         let before = self.produced.len();
         for (id, diff) in corrections { self.produced.push(((id, at.clone()), diff)); }
         if self.produced.len() > before {
diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs
index 253534169..dc8d08a34 100644
--- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs
+++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs
@@ -34,7 +34,7 @@ use crate::trace::chunk::ChunkBatch;
 use crate::trace::chunk::vec::VecChunk;
 use crate::trace::Description;
 
-use super::{ProxyReduceBackend, ReduceInstance, ReduceWindow};
+use super::{KeyPosition, ProxyReduceBackend, ReduceInstance, ReduceWindow};
 
 /// The batch type of a hash-keyed [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine): updates
 /// `D` is `(K, V)` on the input side and `(K, W)` on the output side.
@@ -100,6 +100,9 @@ where
     R: Semigroup + Ord + Clone + 'static,
     L: FnMut(&K, &[(V, R)], &mut Vec<(W, R)>, &mut Vec<(W, R)>),
 {
+    type Key = u64;
+    type VIn = u64;
+    type VOut = u64;
     type RIn = R;
     type ROut = R;
 
@@ -112,10 +115,10 @@ where
         &mut self,
         instance: &ReduceInstance<'_, T, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>,
         changed: &[u64],
-        from: &mut Option<u64>,
+        from: &mut KeyPosition<u64>,
         window: &mut ReduceWindow<T, R, R>,
     ) {
-        let Some(start) = *from else { return };
+        let start = match *from { KeyPosition::Start => 0, KeyPosition::At(key) => key, KeyPosition::End => return };
 
         // Populating the novel and prior input bridges.
         self.in_pool.clear();
@@ -176,7 +179,7 @@ where
             if collides { self.collisions.push(key); }
             if budget == 0 { break; }
         }
-        *from = walk.due();
+        *from = walk.due().map_or(KeyPosition::End, KeyPosition::At);
 
         // Populating the output bridge.
         self.out_ids.clear();
diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs
index 0798076be..54f71caac 100644
--- a/interactive/src/corgi/join.rs
+++ b/interactive/src/corgi/join.rs
@@ -33,7 +33,7 @@ use std::cmp::Ordering;
 use std::marker::PhantomData;
 use std::rc::Rc;
 
-use differential_dataflow::operators::int_proxy::{JoinInstance, ProxyBridge, ProxyJoinBackend};
+use differential_dataflow::operators::int_proxy::{KeyPosition, JoinInstance, ProxyBridge, ProxyJoinBackend};
 use differential_dataflow::operators::int_proxy::join::JoinMatches;
 use differential_dataflow::trace::chunk::{Chunk, ChunkBatch};
 
@@ -73,6 +73,9 @@ impl<T: ColTime> CorgiJoinBackend<T> {
 }
 
 impl<T: ColTime> ProxyJoinBackend<T, CBatch<T>, CBatch<T>> for CorgiJoinBackend<T> {
+    type Key = u64;
+    type V0 = u64;
+    type V1 = u64;
     type R0 = Diff;
     type R1 = Diff;
     type ROut = Diff;
@@ -81,15 +84,20 @@ impl<T: ColTime> ProxyJoinBackend<T, CBatch<T>, CBatch<T>> for CorgiJoinBackend<
     fn advance(
         &mut self,
         instance: &JoinInstance<T, CBatch<T>, CBatch<T>>,
-        from: &mut Option<u64>,
+        from: &mut KeyPosition<u64>,
         bridge0: &mut ProxyBridge<T, Diff>,
         bridge1: &mut ProxyBridge<T, Diff>,
     ) {
+        let mut next = match *from {
+            KeyPosition::Start => Some(0),
+            KeyPosition::At(key) => Some(key),
+            KeyPosition::End => return,
+        };
         self.colliding.clear();
         let chunks0 = side_chunks(&instance.batches0);
         let chunks1 = side_chunks(&instance.batches1);
         if chunks0.is_empty() || chunks1.is_empty() {
-            *from = None;
+            *from = KeyPosition::End;
             return;
         }
         debug_assert_eq!(
@@ -110,11 +118,12 @@ impl<T: ColTime> ProxyJoinBackend<T, CBatch<T>, CBatch<T>> for CorgiJoinBackend<
             &chunks0,
             &chunks1,
             &instance.lower,
-            from,
+            &mut next,
             bridge0,
             bridge1,
             &mut self.colliding,
         );
+        *from = next.map_or(KeyPosition::End, KeyPosition::At);
     }
 
     fn cross(
@@ -894,7 +903,7 @@ mod tests {
                     }
                     let mut backend = backend();
                     let (mut left, mut right) = (Vec::new(), Vec::new());
-                    backend.advance(&instance, &mut Some(0), &mut left, &mut right);
+                    backend.advance(&instance, &mut KeyPosition::Start, &mut left, &mut right);
                     assert_eq!((left.len(), right.len()), (1, 1));
                     assert_eq!((left[0].1, left[0].2), (5, 2));
                     if !compound0 { assert_eq!(left[0].0.1, u64::MAX); }
@@ -909,7 +918,7 @@ mod tests {
                     assert_eq!(output[0].times.get(0), 5);
                     instance.batches0.push(make(compound0, vec![u64::MAX], vec![-2], 3));
                     left.clear(); right.clear();
-                    backend.advance(&instance, &mut Some(0), &mut left, &mut right);
+                    backend.advance(&instance, &mut KeyPosition::Start, &mut left, &mut right);
                     assert!(left.is_empty() && right.is_empty(), "suppress a fully cancelled key");
                 }
             }
@@ -943,11 +952,11 @@ mod tests {
             lower: 0,
         };
         let mut backend = backend();
-        let mut from = Some(0);
+        let mut from = KeyPosition::Start;
         let (mut left, mut right) = (Vec::new(), Vec::new());
 
         backend.advance(&instance, &mut from, &mut left, &mut right);
-        assert!(from.is_some(), "the first block must leave work for the collision block");
+        assert!(from != KeyPosition::End, "the first block must leave work for the collision block");
         assert!(backend.colliding.is_empty());
 
         left.clear();
@@ -969,7 +978,7 @@ mod tests {
             lower: 0,
         };
         let mut backend = backend();
-        let mut from = Some(0);
+        let mut from = KeyPosition::Start;
         let (mut left, mut right) = (Vec::new(), Vec::new());
 
         backend.advance(&instance, &mut from, &mut left, &mut right);
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 26a02ecd5..893202f64 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -31,7 +31,7 @@ use std::rc::Rc;
 use differential_dataflow::consolidation::consolidate_updates;
 use differential_dataflow::trace::Description;
 use differential_dataflow::trace::chunk::ChunkBatch;
-use differential_dataflow::operators::int_proxy::ProxyBridge;
+use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge};
 use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow};
 
 use corgi::arrange::{compare_at, gather, gather_lanes, sort_blocks};
@@ -573,6 +573,9 @@ impl<T> ProxyReduceBackend<T, CBatch<T>, CBatch<T>> for CorgiReduceBackend<T>
 where
     T: ColTime + Ord,
 {
+    type Key = u64;
+    type VIn = u64;
+    type VOut = u64;
     type RIn = Diff;
     type ROut = Diff;
 
@@ -584,17 +587,17 @@ where
         self.rows = (Vec::new(), Vec::new(), ColTimes::default(), Vec::new());
     }
 
-    fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch<T>, CBatch<T>>, changed: &[u64], from: &mut Option<u64>, window: &mut ReduceWindow<T, Diff, Diff>) {
+    fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch<T>, CBatch<T>>, changed: &[u64], from: &mut KeyPosition<u64>, window: &mut ReduceWindow<T, Diff, Diff>) {
         // Single window: present the WHOLE key space at once, and report it covered. This is NOT a
         // deferred refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
         // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS
         // fell only 356MB -> 340MB. Two reasons: the per-window, per-chunk seek setup is a
         // fixed cost that multiplies by the window count, and the presentation is not the
         // memory peak in the first place (the trace is).
-        if from.is_none() {
+        if *from == KeyPosition::End {
             return;
         }
-        *from = None;
+        *from = KeyPosition::End;
 
         // The window's keys: the hashes the novel batches touch, merged with the `changed` set the
         // harness supplies. The novel hashes come from the scan the presentation needs anyway — the