From 95ce57958c4a5cb8b88ebda19ca7b97fe736fc7f Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 1 Sep 2026 22:31:40 -0400
Subject: [PATCH 1/6] leave_dynamic: a capability per element of a
 multi-element stamp
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Both the row and the corgi `leave_dynamic` read `cap.time()`, which now
panics ("expected a singleton stamp") when a message carries timestamps
from two epochs — late iterations of one alongside early ones of the next,
which any steady-state run of an iterative program produces. Hold a
capability per stamp element, each truncated exactly as the records are,
and open the session on the set.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 .../src/columnar/collection/operators.rs      | 19 ++++++++++-----
 differential-dataflow/src/dynamic/mod.rs      | 23 +++++++++++--------
 interactive/src/backend/corgi.rs              | 22 ++++++++++--------
 3 files changed, 40 insertions(+), 24 deletions(-)

diff --git a/differential-dataflow/src/columnar/collection/operators.rs b/differential-dataflow/src/columnar/collection/operators.rs
index ff7b3f21a..fe7db42e4 100644
--- a/differential-dataflow/src/columnar/collection/operators.rs
+++ b/differential-dataflow/src/columnar/collection/operators.rs
@@ -116,12 +116,19 @@ where
         move |_frontier| {
             let mut output = output.activate();
             op_input.for_each(|cap, data| {
-                // Truncate the capability's timestamp.
-                let mut new_time = cap.time().clone();
-                let mut vec = std::mem::take(&mut new_time.inner).into_inner();
-                vec.truncate(level - 1);
-                new_time.inner = PointStamp::new(vec);
-                let new_cap = cap.delayed(&new_time, 0);
+                // A message may carry several timestamps (a multi-element stamp): hold a
+                // capability for each, truncated exactly as the updates are.
+                let new_cap: timely::dataflow::operators::CapabilitySet<_> = cap
+                    .stamp()
+                    .iter()
+                    .map(|t| {
+                        let mut new_time = t.clone();
+                        let mut vec = std::mem::take(&mut new_time.inner).into_inner();
+                        vec.truncate(level - 1);
+                        new_time.inner = PointStamp::new(vec);
+                        cap.delayed(&new_time, 0)
+                    })
+                    .collect();
                 // Push updates with truncated times into the builder.
                 // The builder's form call on flush sorts and consolidates,
                 // handling the duplicate times that truncation can produce.
diff --git a/differential-dataflow/src/dynamic/mod.rs b/differential-dataflow/src/dynamic/mod.rs
index 0a615ffbf..1020b33af 100644
--- a/differential-dataflow/src/dynamic/mod.rs
+++ b/differential-dataflow/src/dynamic/mod.rs
@@ -16,6 +16,7 @@ pub mod pointstamp;
 use timely::order::Product;
 use timely::progress::Timestamp;
 use timely::dataflow::operators::generic::{OutputBuilder, builder_rc::OperatorBuilder};
+use timely::dataflow::operators::CapabilitySet;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::progress::Antichain;
 
@@ -47,17 +48,21 @@ where
         builder.build(move |_capability| move |_frontier| {
             let mut output = output.activate();
             input.for_each(|cap, data| {
-                let mut new_time = cap.time().clone();
-                let mut vec = std::mem::take(&mut new_time.inner).into_inner();
-                vec.truncate(level - 1);
-                new_time.inner = PointStamp::new(vec);
-                let new_cap = cap.delayed(&new_time, 0);
-                for (_data, time, _diff) in data.iter_mut() {
-                    let mut vec = std::mem::take(&mut time.inner).into_inner();
+                // A message may carry several timestamps (a multi-element stamp, e.g. late
+                // iterations of one epoch alongside early ones of the next): hold a capability
+                // for each, truncated exactly as the records are.
+                let truncate = |time: &Product<TOuter, PointStamp<T>>| {
+                    let mut new_time = time.clone();
+                    let mut vec = std::mem::take(&mut new_time.inner).into_inner();
                     vec.truncate(level - 1);
-                    time.inner = PointStamp::new(vec);
+                    new_time.inner = PointStamp::new(vec);
+                    new_time
+                };
+                let caps: CapabilitySet<_> = cap.stamp().iter().map(|t| cap.delayed(&truncate(t), 0)).collect();
+                for (_data, time, _diff) in data.iter_mut() {
+                    *time = truncate(time);
                 }
-                output.session(&new_cap).give_container(data);
+                output.session(&caps).give_container(data);
             });
         });
 
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 4422b235c..3a1cb3eff 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -367,17 +367,21 @@ impl Backend for CorgiBackend {
         builder.build(move |_capability| move |_frontier| {
             let mut output = output.activate();
             input.for_each(|cap, data| {
-                let mut new_time = cap.time().clone();
-                let mut v = std::mem::take(&mut new_time.inner).into_inner();
-                v.truncate(level - 1);
-                new_time.inner = PointStamp::new(v);
-                let new_cap = cap.delayed(&new_time, 0);
-                for t in data.times.iter_mut() {
-                    let mut v = std::mem::take(&mut t.inner).into_inner();
+                // A message may carry several timestamps (a multi-element stamp): hold a
+                // capability for each, truncated exactly as the records are.
+                let truncate = |t: &Time| {
+                    let mut new_time = t.clone();
+                    let mut v = std::mem::take(&mut new_time.inner).into_inner();
                     v.truncate(level - 1);
-                    t.inner = PointStamp::new(v);
+                    new_time.inner = PointStamp::new(v);
+                    new_time
+                };
+                let caps: timely::dataflow::operators::CapabilitySet<Time> =
+                    cap.stamp().iter().map(|t| cap.delayed(&truncate(t), 0)).collect();
+                for t in data.times.iter_mut() {
+                    *t = truncate(t);
                 }
-                output.session(&new_cap).give_container(data);
+                output.session(&caps).give_container(data);
             });
         });
 

From 9c50a5d4f2438d285a046d78c93e908eb0c8074c Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 1 Sep 2026 22:31:40 -0400
Subject: [PATCH 2/6] ddir driver: with EDGES_FILE, the rounds edit the file's
 rows

The round loop churned synthetic rows from `gen_row`, which with a file-fed
input meant retracting rows the dataflow never saw and inserting random
ones (duplicate ids, cycles). Each update now retracts one of the file's
rows and re-inserts it with its last field rewired below its first, so a
round is an edit to data the dataflow holds.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 interactive/examples/ddir.rs | 27 ++++++++++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

diff --git a/interactive/examples/ddir.rs b/interactive/examples/ddir.rs
index d262f1d00..8e38e7f0e 100644
--- a/interactive/examples/ddir.rs
+++ b/interactive/examples/ddir.rs
@@ -128,15 +128,23 @@ fn run(
         // With `EDGES_FILE` set, rows come from the file (one row per line,
         // whitespace-separated i64 fields), assigned round-robin to inputs.
         // Otherwise `gen_row` synthesizes `edges` rows.
+        // With a file, the rounds EDIT its rows (retract one, re-insert it with its last field
+        // rewired to a value below its first) rather than churning synthetic rows, so an update is
+        // a change to data the dataflow has, not a phantom retraction of a row it never saw.
+        let mut file_rows: Option<Vec<(usize, Vec<i64>)>> = None;
         if let Some(path) = &edges_file {
             let text = std::fs::read_to_string(path).unwrap_or_else(|e| panic!("Cannot read {}: {}", path, e));
+            let mut mine = Vec::new();
             for (e, line) in text.lines().filter(|l| !l.trim().is_empty()).enumerate() {
                 if e % peers == index {
                     let input_idx = e % n_inputs;
-                    let fields: Vec<Value> = line.split_whitespace().map(|t| Value::Int(t.parse::<i64>().unwrap())).collect();
+                    let ints: Vec<i64> = line.split_whitespace().map(|t| t.parse::<i64>().unwrap()).collect();
+                    let fields: Vec<Value> = ints.iter().map(|&n| Value::Int(n)).collect();
                     inputs[input_idx].update((Value::Tuple(fields), Value::unit()), 1);
+                    mine.push((input_idx, ints));
                 }
             }
+            file_rows = Some(mine);
         } else {
             for e in 0..edges {
                 if (e as usize) % peers == index {
@@ -175,6 +183,23 @@ fn run(
             let timer_round = std::time::Instant::now();
             let time = (round + 2) as u64;
             for _ in 0..batch {
+                if let Some(rows) = file_rows.as_mut() {
+                    if !rows.is_empty() {
+                        let idx = (cursor as usize) % rows.len();
+                        let (input_idx, old) = rows[idx].clone();
+                        let mut new = old.clone();
+                        let last = new.len() - 1;
+                        let mut h = cursor.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ 0xD1B5_4A32_D192_ED03;
+                        h ^= h >> 29; h = h.wrapping_mul(0xBF58_476D_1CE4_E5B9); h ^= h >> 32;
+                        new[last] = if old[0] > 0 { (h % old[0] as u64) as i64 } else { 0 };
+                        let row = |f: &Vec<i64>| (Value::Tuple(f.iter().map(|&n| Value::Int(n)).collect()), Value::unit());
+                        inputs[input_idx].update(row(&old), -1);
+                        inputs[input_idx].update(row(&new), 1);
+                        rows[idx] = (input_idx, new);
+                    }
+                    cursor += 1;
+                    continue;
+                }
                 let remove_idx = cursor;
                 let add_idx = edges + cursor;
                 if (remove_idx as usize) % peers == index {

From 4a42154a8852468803429bc7f47fcace14ba67e6 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 1 Sep 2026 22:33:17 -0400
Subject: [PATCH 3/6] DDIR: pin corgi at the sum-compare merge

compare_idx and gather read sum discriminants in place (WIP master
3c4b0d6f), which is what took the e-graph's 200k-node saturation from 43s
to 23s on the corgi backend.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 interactive/Cargo.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml
index 1673d1440..5cb661b6a 100644
--- a/interactive/Cargo.toml
+++ b/interactive/Cargo.toml
@@ -14,7 +14,7 @@ workspace = true
 [dependencies]
 columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
-corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "578ef3fe79fdaa813ba0c721c08f60323f22d420", features = ["serde"] }
+corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "3c4b0d6f90393dc25105ebcc4c869ca7d429411c", features = ["serde"] }
 differential-dataflow = { workspace = true }
 mimalloc = "0.1.48"
 serde = { version = "1.0", features = ["derive"] }

From 50c19886dd75cc68f0f65e644f0c2bae61e986c4 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 2 Sep 2026 07:03:42 -0400
Subject: [PATCH 4/6] DDIR: consolidate feedback before setting a var; DD's
 identifiers as a program
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An iteration quiesces when its feedback carries no records, so a var whose
step's additions and retractions cancel only logically — a `negate` under
a `concat` — never quiesced: the vec backend ran forever on a program
whose fixpoint was reached in one round. Every var is now set to its
consolidated step (`Backend::consolidate`, through the arrangement by
default). The suite's timings are unchanged; the programs there all fed
their vars through a reduce, which consolidated by nature.

tests/identifiers.rs writes DD's `identifiers` algorithm in DDIR: each
record proposes hash(round, record); among the proposers of an id the
least (round, record) wins and the losers try the next round; with hash,
min, negate and a var, over 64 slots for 32 records, every record gets
its own id (25 settle at round 0, the rest by round 4). It is the
primitive an e-graph rule needs to mint a node for a new term inside the
program.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 interactive/src/backend/mod.rs   | 10 ++++-
 interactive/tests/identifiers.rs | 64 ++++++++++++++++++++++++++++++++
 2 files changed, 73 insertions(+), 1 deletion(-)
 create mode 100644 interactive/tests/identifiers.rs

diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs
index 1736a9371..8b9de26e8 100644
--- a/interactive/src/backend/mod.rs
+++ b/interactive/src/backend/mod.rs
@@ -48,6 +48,14 @@ pub trait Backend {
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s>;
     fn inspect<'s>(c: Collection<'s, Time, Self::Container>, label: String) -> Collection<'s, Time, Self::Container>;
     fn leave_dynamic<'s>(c: Collection<'s, Time, Self::Container>, depth: usize) -> Collection<'s, Time, Self::Container>;
+    /// The same collection with its changes consolidated: rows that cancel are gone. A feedback
+    /// variable is set to this — an iteration quiesces when its feedback carries no *records*,
+    /// so a step whose additions and retractions cancel only logically (a `negate` under a
+    /// `concat`, say) would never quiesce. Through the arrangement by default; a backend with a
+    /// cheaper consolidation can override.
+    fn consolidate<'s>(c: Collection<'s, Time, Self::Container>) -> Collection<'s, Time, Self::Container> {
+        Self::as_collection(Self::arrange(c))
+    }
 }
 
 /// A rendered item's value: a collection, or an arrangement.
@@ -170,7 +178,7 @@ pub fn render_tree<'s, B: Backend>(
 
     for bind in &s.binds {
         let c = resolve(&items, &imports, &var_cols, &bind.value).collection();
-        var_handles[bind.var].take().expect("bind: variable already bound").set(c);
+        var_handles[bind.var].take().expect("bind: variable already bound").set(B::consolidate(c));
     }
 
     s.exports.iter().map(|e| resolve(&items, &imports, &var_cols, &e.value).collection()).collect()
diff --git a/interactive/tests/identifiers.rs b/interactive/tests/identifiers.rs
new file mode 100644
index 000000000..2f693f6e1
--- /dev/null
+++ b/interactive/tests/identifiers.rs
@@ -0,0 +1,64 @@
+//! DD's `identifiers` algorithm as a DDIR program: every record proposes `hash(round, record)`
+//! as its id; among records proposing one id the least `(round, record)` wins and the losers
+//! propose again at the next round. Written with `hash`, `min`, `negate`, and a `var`, over a
+//! hash space small enough to force collisions, it still gives every record its own id — the
+//! primitive an e-graph rule needs to mint a node for a new term inside the program.
+
+use interactive::backend::vec;
+use interactive::ir::Value;
+use interactive::{lower, parse};
+
+const PROGRAM_SRC: &str = "\
+-- input 0: records (r) ;
+let init = input 0 | key(0, $0[0] ; );                                   -- (round, record) ;
+ids: {
+    let all   = init + diff;                                              -- (round, record) ;
+    let keyed = all | key(hash(HASHBOUND, $0[0], $0[1]) ; $0[0], $0[1]);         -- id ; (round, record)
+    let win   = keyed | min;                                              -- id ; the least (round, record)
+    let lose  = keyed + (win | negate);                                   -- id ; the rest
+    -- a winner past round 0 has moved from its round-0 proposal; a loser moves on to the next round
+    let moved = (win | filter($1[0] > 0) | map($1[0], $1[1] ; )) + (lose | map($1[0] + 1, $1[1] ; ));
+    let orig  = moved | map(0, $0[1] ; ) | negate;                        -- less their round-0 proposals
+    var diff  = moved + orig;
+}
+export \"ids\" = (init + ids::diff) | key($0[1] ; hash(HASHBOUND, $0[0], $0[1])) | arrange;   -- record ; id
+export \"rounds\" = (init + ids::diff) | key($0[0] ; ) | count | arrange;               -- round ; how many settled there
+";
+
+fn int(v: &Value) -> i64 {
+    match v {
+        Value::Tuple(xs) => xs[0].as_int(),
+        other => other.as_int(),
+    }
+}
+
+fn run(bound: i64, n: i64) -> std::collections::BTreeMap<String, Vec<((Value, Value), i64)>> {
+    let src = PROGRAM_SRC.replace("HASHBOUND", &bound.to_string());
+    let tree = lower::lower_tree(parse::pipe::parse(&src));
+    let records: Vec<(Value, Value)> = (1..=n).map(|n| (Value::Tuple(vec![Value::Int(n)]), Value::unit())).collect();
+    vec::evaluate(&tree, &[records])
+}
+
+#[test]
+fn identifiers_without_collisions() {
+    let out = run(1 << 40, 8);
+    let ids: Vec<(i64, i64)> = out["ids"].iter().filter(|(_, d)| *d > 0).map(|((k, v), _)| (int(k), int(v))).collect();
+    println!("ids: {ids:?}");
+    assert_eq!(ids.len(), 8);
+}
+
+#[test]
+fn identifiers_are_unique_despite_collisions() {
+    let out = run(64, 32);
+    let ids: Vec<(i64, i64)> = out["ids"]
+        .iter()
+        .filter(|(_, d)| *d > 0)
+        .map(|((k, v), _)| (int(k), int(v)))
+        .collect();
+    assert_eq!(ids.len(), 32, "every record has an id: {ids:?}");
+    let distinct: std::collections::BTreeSet<i64> = ids.iter().map(|(_, id)| *id).collect();
+    assert_eq!(distinct.len(), 32, "ids are unique: {ids:?}");
+    let rounds: Vec<(i64, i64)> = out["rounds"].iter().filter(|(_, d)| *d > 0).map(|((k, v), _)| (int(k), int(v))).collect();
+    println!("records per round: {rounds:?}");
+    assert!(rounds.iter().any(|(round, _)| *round > 0), "the hash space was wide enough that nothing collided; narrow it");
+}

From f6d0854c9408456134481159ad3b65960c88c5da Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 2 Sep 2026 08:01:57 -0400
Subject: [PATCH 5/6] leave_dynamic: a test with a message over two epochs, and
 where such messages come from
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Instrumenting the batch-shipping operators on the run that panicked (a
DDIR program at 20k nodes, four workers, epochs fed without waiting) shows
the multi-element stamps being minted: an arrange retiring (1, [18]) and
(2, []) — epoch 1 at round 18 alongside epoch 2 just entered — in one
batch, then a join forwarding its input batch's set, then a reduce
retiring the same pair, and finally an arrange whose batch, forwarded by
as_collection under the retiring set and delayed element-wise by the
feedback, reached the scope's exit as (1, [19]), (2, [1]).

Which retirements coincide depends on scheduling, so the test makes the
message directly: an operator that ships its input under a capability for
epoch e round 3 and one for epoch e + 1 round 0, into leave_dynamic. It
panics with "expected a singleton stamp" before this branch's fix and
passes with it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 differential-dataflow/tests/dynamic.rs | 82 ++++++++++++++++++++++++++
 1 file changed, 82 insertions(+)
 create mode 100644 differential-dataflow/tests/dynamic.rs

diff --git a/differential-dataflow/tests/dynamic.rs b/differential-dataflow/tests/dynamic.rs
new file mode 100644
index 000000000..1ee5bd2c1
--- /dev/null
+++ b/differential-dataflow/tests/dynamic.rs
@@ -0,0 +1,82 @@
+//! Leaving a dynamic scope on a message that carries two timestamps.
+//!
+//! Since timely's stamps became multisets, a message can carry several timestamps, and DD's
+//! batch-shipping operators make such messages: `arrange` ships a batch under the set of
+//! capabilities it retires, `reduce` likewise, and `join` forwards its input batch's set. In an
+//! iterative scope with two epochs in flight, an arrange holding `(1, [18])` (epoch 1, round 18)
+//! and `(2, [])` (epoch 2, just entered) retires both in one batch when its input frontier passes
+//! both at once. `as_collection` then forwards the batch's records under that same set, and the
+//! feedback delays it element-wise, so the set reaches whatever reads the *message's* time
+//! rather than the records' — which `leave_dynamic` did, to truncate it, and so panicked on
+//! "expected a singleton stamp". (Observed on a DDIR program at 20k nodes with four workers;
+//! the first multi-element stamp came from an arrange, then a join, then a reduce, then the
+//! arrange whose batch reached the scope's exit.)
+//!
+//! Which retirements coincide depends on scheduling, so this test makes the message directly: an
+//! operator that ships its input under a capability for epoch `e` round 3 and one for epoch
+//! `e + 1` round 0, into `leave_dynamic`.
+
+use differential_dataflow::collection::AsCollection;
+use differential_dataflow::dynamic::pointstamp::PointStamp;
+use differential_dataflow::input::Input;
+use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::operators::generic::{builder_rc::OperatorBuilder, OutputBuilder};
+use timely::dataflow::operators::CapabilitySet;
+use timely::dataflow::Scope;
+use timely::order::Product;
+
+type Time = Product<u64, PointStamp<u64>>;
+
+#[test]
+fn leave_dynamic_on_a_message_over_two_epochs() {
+    let received = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
+    let seen = received.clone();
+    timely::execute_directly(move |worker| {
+        let mut probe = timely::dataflow::ProbeHandle::new();
+        let mut input = worker.dataflow::<u64, _, _>(|scope| {
+            let (input, data) = scope.new_collection::<u64, isize>();
+            let left = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
+                let entered = data.enter(inner);
+                let mut builder = OperatorBuilder::new("TwoEpochs".to_string(), inner.clone());
+                let (output, stream) = builder.new_output();
+                let mut output = OutputBuilder::from(output);
+                let mut input = builder.new_input(entered.inner, Pipeline);
+                builder.build(move |mut caps| {
+                    // the initial capability, at (0, []), can be delayed to any later time; it is
+                    // dropped once the input is done, so the computation can finish
+                    let mut root = caps.pop();
+                    move |frontier| {
+                        let mut output = output.activate();
+                        input.for_each(|cap, data| {
+                            let Some(root) = root.as_ref() else { return };
+                            let epoch = cap.time().outer;
+                            let t1 = Product::new(epoch, PointStamp::new([3].into_iter().collect()));
+                            let t2 = Product::new(epoch + 1, PointStamp::new([0].into_iter().collect()));
+                            let caps: CapabilitySet<Time> = [root.delayed(&t1), root.delayed(&t2)].into_iter().collect();
+                            for (i, (_datum, time, _diff)) in data.iter_mut().enumerate() {
+                                *time = if i % 2 == 0 { t1.clone() } else { t2.clone() };
+                            }
+                            output.session(&caps).give_container(data);
+                        });
+                        if frontier[0].frontier().is_empty() {
+                            root = None;
+                        }
+                    }
+                });
+                stream.as_collection().leave_dynamic(1).leave(scope)
+            });
+            left.inspect(move |(datum, time, diff)| seen.lock().unwrap().push((*datum, *time, *diff))).probe_with(&mut probe);
+            input
+        });
+        for x in 0..8u64 {
+            input.insert(x);
+        }
+        input.close();
+        worker.step_while(|| !probe.done());
+    });
+    let mut got = received.lock().unwrap().clone();
+    got.sort();
+    // even records at epoch 0 (round 3, truncated away), odd ones at epoch 1 (round 0, likewise)
+    let want: Vec<(u64, u64, isize)> = (0..8u64).map(|x| (x, x % 2, 1)).collect();
+    assert_eq!(got, want);
+}

From 8cebac378b5a5056e99d04a0ed7e7533b5610b12 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 2 Sep 2026 08:16:23 -0400
Subject: [PATCH 6/6] Records of a batch go out under one capability each:
 as_collection, count, threshold route by time

A batch is shipped under the set of capabilities it retires, so a message
of batches carries several timestamps when two epochs retire together.
The operators that turn batches back into records forwarded the records
under that same set, which is how a multi-element stamp reached
`leave_dynamic` and anything else that reads a message's time as one.

`StampRouter` (collection.rs) gives one capability per element of a
message's stamp and routes each record to the first element at or before
its time (a batch's times are at or beyond one of the capabilities it
retired under, so every record has one). It is applied at every seam
that makes records from batches: the vec `as_collection`
(`flat_map_batches`), `count`, `threshold`, the columnar
`as_recorded_updates`, and DDIR's corgi `as_collection` (a chunk under one
capability goes whole, as before; under several it is split by row). With
that, a collection's messages carry one timestamp each; only streams of
batches carry sets.

Not routed: `join_with_tactic`'s output, whose tactic builds containers the
operator cannot split by time (the fix belongs in the tactics, a builder
per element of the fresh batch's set), and `as_container`, which forwards
opaque containers. `leave_dynamic` keeps holding a capability per element,
so it is safe either way.

tests/dynamic.rs: a hand-built batch under capabilities for epoch 0 round
3 and epoch 1 round 0, through `as_collection` into an operator that reads
each message's time as one, lands its records in their two epochs.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK
---
 differential-dataflow/src/collection.rs       | 36 ++++++++
 .../src/columnar/collection/operators.rs      | 36 +++++---
 .../src/operators/arrange/arrangement.rs      | 16 +++-
 differential-dataflow/src/operators/count.rs  | 18 +++-
 .../src/operators/threshold.rs                | 16 +++-
 differential-dataflow/tests/dynamic.rs        | 82 ++++++++++++++++++-
 interactive/src/backend/corgi.rs              | 37 ++++++---
 7 files changed, 207 insertions(+), 34 deletions(-)

diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index 356f1346f..3500aa9cb 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -1252,6 +1252,42 @@ pub mod vec {
     }
 }
 
+/// Routes the records of one message to singleton capabilities when the message's stamp has
+/// several elements.
+///
+/// A batch is shipped under the set of capabilities it retires, so a message of batches can
+/// carry several timestamps. Records each have one time, and an operator over records reads a
+/// message's time as one (`InputCapability::time`); an operator that turns batches into records
+/// keeps that true by sending each record under the first element of the stamp at or before its
+/// time. The elements cover every record: a batch's times are at or beyond one of the
+/// capabilities it retired under.
+pub struct StampRouter<T: Timestamp> {
+    caps: Vec<timely::dataflow::operators::Capability<T>>,
+}
+
+impl<T: Timestamp> StampRouter<T> {
+    /// One capability per element of the message's stamp, for output `port`.
+    pub fn new(cap: &timely::dataflow::operators::InputCapability<T>, port: usize) -> Self {
+        Self { caps: cap.stamp().iter().map(|t| cap.delayed(t, port)).collect() }
+    }
+    /// One capability per element of a set.
+    pub fn from_set(set: &timely::dataflow::operators::CapabilitySet<T>) -> Self {
+        Self { caps: set.iter().cloned().collect() }
+    }
+    /// The capabilities, in the order `index` names them.
+    pub fn capabilities(&self) -> &[timely::dataflow::operators::Capability<T>] {
+        &self.caps
+    }
+    /// Which capability a record at `time` goes under.
+    pub fn index(&self, time: &T) -> usize {
+        self.caps.iter().position(|c| c.time().less_equal(time)).expect("a record's time is at or beyond an element of its message's stamp")
+    }
+    /// Empty buffers, one per capability, to route records into and then give under each.
+    pub fn buffers<D>(&self) -> Vec<Vec<D>> {
+        (0..self.caps.len()).map(|_| Vec::new()).collect()
+    }
+}
+
 /// Conversion to a differential dataflow Collection.
 pub trait AsCollection<'scope, T: Timestamp, C> {
     /// Converts the type to a differential dataflow collection.
diff --git a/differential-dataflow/src/columnar/collection/operators.rs b/differential-dataflow/src/columnar/collection/operators.rs
index fe7db42e4..57880ed8a 100644
--- a/differential-dataflow/src/columnar/collection/operators.rs
+++ b/differential-dataflow/src/columnar/collection/operators.rs
@@ -174,20 +174,30 @@ where
         .unary::<Builder<U>, _, _, _>(Pipeline, "AsRecordedUpdates", |_, _| {
             move |input, output| {
                 input.for_each(|time, batches| {
-                    let mut session = output.session_with_builder(&time);
-                    for batch in batches.drain(..) {
-                        let Some(batch) = batch.inner else { continue };
-                        let mut cursor = batch.cursor();
-                        while cursor.key_valid(&batch) {
-                            while cursor.val_valid(&batch) {
-                                let key = cursor.key(&batch);
-                                let val = cursor.val(&batch);
-                                cursor.map_times(&batch, |time, diff| {
-                                    session.give((key, val, time, diff));
-                                });
-                                cursor.step_val(&batch);
+                    // A message of batches may carry several timestamps; each record goes out
+                    // under the one at or before its time (a pass per element, the rare case),
+                    // so the collection's messages carry one each.
+                    let router = crate::collection::StampRouter::new(&time, 0);
+                    let batches: Vec<_> = batches.drain(..).filter_map(|b| b.inner).collect();
+                    let mut owned_time = U::Time::default();
+                    for (index, cap) in router.capabilities().iter().enumerate() {
+                        let mut session = output.session_with_builder(cap);
+                        for batch in batches.iter() {
+                            let mut cursor = batch.cursor();
+                            while cursor.key_valid(batch) {
+                                while cursor.val_valid(batch) {
+                                    let key = cursor.key(batch);
+                                    let val = cursor.val(batch);
+                                    cursor.map_times(batch, |time, diff| {
+                                        columnar::Columnar::copy_from(&mut owned_time, time);
+                                        if router.index(&owned_time) == index {
+                                            session.give((key, val, time, diff));
+                                        }
+                                    });
+                                    cursor.step_val(batch);
+                                }
+                                cursor.step_key(batch);
                             }
-                            cursor.step_key(&batch);
                         }
                     }
                 });
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 75eba8177..a7d0d817e 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -179,15 +179,20 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     {
         stream.unary(Pipeline, "AsCollection", move |_,_| move |input, output| {
             input.for_each(|time, data| {
-                let mut session = output.session(&time);
+                // A message of batches may carry several timestamps; each record goes out under
+                // the one at or before its time, so the collection's messages carry one each.
+                let router = crate::collection::StampRouter::new(&time, 0);
+                let mut buffers = router.buffers();
                 for wrapper in data.iter() {
                     let Some(batch) = wrapper.inner.as_ref() else { continue };
                     let mut cursor = batch.cursor();
                     while let Some(key) = cursor.get_key(batch) {
                         while let Some(val) = cursor.get_val(batch) {
                             for datum in logic(key, val) {
-                                cursor.map_times(batch, |time, diff| {
-                                    session.give((datum.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
+                                cursor.map_times(batch, |t, diff| {
+                                    let t = <BatchCursor<Tr> as Cursor>::owned_time(t);
+                                    let index = router.index(&t);
+                                    buffers[index].push((datum.clone(), t, <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
                                 });
                             }
                             cursor.step_val(batch);
@@ -195,6 +200,11 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
                         cursor.step_key(batch);
                     }
                 }
+                for (cap, mut buffer) in router.capabilities().iter().zip(buffers) {
+                    if !buffer.is_empty() {
+                        output.session(cap).give_container(&mut buffer);
+                    }
+                }
             });
         })
         .as_collection()
diff --git a/differential-dataflow/src/operators/count.rs b/differential-dataflow/src/operators/count.rs
index 1f707c3b5..a3c1d1727 100644
--- a/differential-dataflow/src/operators/count.rs
+++ b/differential-dataflow/src/operators/count.rs
@@ -94,7 +94,14 @@ where
 
                 if !caps.is_empty() {
 
-                    let mut session = output.session(&caps);
+                    // The batches' capabilities may span several timestamps; each record goes out
+                    // under the one at or before its time, so the collection's messages carry one.
+                    let router = crate::collection::StampRouter::from_set(&caps);
+                    let mut buffers = router.buffers();
+                    let mut give = |record: (_, Tr::Time, _)| {
+                        let index = router.index(&record.1);
+                        buffers[index].push(record);
+                    };
 
                     let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage.into_iter().filter_map(|b| b.inner).collect());
                     let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap());
@@ -114,20 +121,25 @@ where
 
                             if let Some(count) = count.as_ref() {
                                 if !count.is_zero() {
-                                    session.give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(-1i8)));
+                                    give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(-1i8)));
                                 }
                             }
                             count.as_mut().map(|c| c.plus_equals(&diff));
                             if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
                             if let Some(count) = count.as_ref() {
                                 if !count.is_zero() {
-                                    session.give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(1i8)));
+                                    give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(1i8)));
                                 }
                             }
                         });
 
                         batch_cursor.step_key(&batch_storage);
                     }
+                    for (cap, mut buffer) in router.capabilities().iter().zip(buffers) {
+                        if !buffer.is_empty() {
+                            output.session(cap).give_container(&mut buffer);
+                        }
+                    }
                 }
 
                 // tidy up the shared input trace.
diff --git a/differential-dataflow/src/operators/threshold.rs b/differential-dataflow/src/operators/threshold.rs
index c9502976f..abba9fe60 100644
--- a/differential-dataflow/src/operators/threshold.rs
+++ b/differential-dataflow/src/operators/threshold.rs
@@ -144,7 +144,14 @@ where
 
                 if !caps.is_empty() {
 
-                    let mut session = output.session(&caps);
+                    // The batches' capabilities may span several timestamps; each record goes out
+                    // under the one at or before its time, so the collection's messages carry one.
+                    let router = crate::collection::StampRouter::from_set(&caps);
+                    let mut buffers = router.buffers();
+                    let mut give = |record: (_, Tr::Time, _)| {
+                        let index = router.index(&record.1);
+                        buffers[index].push(record);
+                    };
 
                     let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage.into_iter().filter_map(|b| b.inner).collect());
                     let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap());
@@ -185,13 +192,18 @@ where
 
                             if let Some(difference) = difference {
                                 if !difference.is_zero() {
-                                    session.give((key.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), difference));
+                                    give((key.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), difference));
                                 }
                             }
                         });
 
                         batch_cursor.step_key(&batch_storage);
                     }
+                    for (cap, mut buffer) in router.capabilities().iter().zip(buffers) {
+                        if !buffer.is_empty() {
+                            output.session(cap).give_container(&mut buffer);
+                        }
+                    }
                 }
 
                 // tidy up the shared input trace.
diff --git a/differential-dataflow/tests/dynamic.rs b/differential-dataflow/tests/dynamic.rs
index 1ee5bd2c1..05e896bb7 100644
--- a/differential-dataflow/tests/dynamic.rs
+++ b/differential-dataflow/tests/dynamic.rs
@@ -22,7 +22,6 @@ use differential_dataflow::input::Input;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::generic::{builder_rc::OperatorBuilder, OutputBuilder};
 use timely::dataflow::operators::CapabilitySet;
-use timely::dataflow::Scope;
 use timely::order::Product;
 
 type Time = Product<u64, PointStamp<u64>>;
@@ -30,14 +29,14 @@ type Time = Product<u64, PointStamp<u64>>;
 #[test]
 fn leave_dynamic_on_a_message_over_two_epochs() {
     let received = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
-    let seen = received.clone();
+    let seen = std::sync::Arc::clone(&received);
     timely::execute_directly(move |worker| {
         let mut probe = timely::dataflow::ProbeHandle::new();
         let mut input = worker.dataflow::<u64, _, _>(|scope| {
             let (input, data) = scope.new_collection::<u64, isize>();
             let left = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
                 let entered = data.enter(inner);
-                let mut builder = OperatorBuilder::new("TwoEpochs".to_string(), inner.clone());
+                let mut builder = OperatorBuilder::new("TwoEpochs".to_string(), inner);
                 let (output, stream) = builder.new_output();
                 let mut output = OutputBuilder::from(output);
                 let mut input = builder.new_input(entered.inner, Pipeline);
@@ -80,3 +79,80 @@ fn leave_dynamic_on_a_message_over_two_epochs() {
     let want: Vec<(u64, u64, isize)> = (0..8u64).map(|x| (x, x % 2, 1)).collect();
     assert_eq!(got, want);
 }
+
+/// The seam itself: `as_collection` on a message of batches carrying two timestamps hands each
+/// record on under the one at or before its time, so the collection's messages carry one each
+/// and an operator over records may read `cap.time()`.
+#[test]
+fn as_collection_routes_records_under_singleton_stamps() {
+    use differential_dataflow::batcher::Batcher;
+    use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+    use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
+    use differential_dataflow::trace::{Description, Span, SpanOf};
+    use timely::dataflow::operators::generic::Operator;
+    use timely::progress::Antichain;
+
+    type Tr = ValSpine<u64, u64, Time, isize>;
+
+    let received = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
+    let seen = std::sync::Arc::clone(&received);
+    timely::execute_directly(move |worker| {
+        let mut probe = timely::dataflow::ProbeHandle::new();
+        let mut input = worker.dataflow::<u64, _, _>(|scope| {
+            let (input, data) = scope.new_collection::<u64, isize>();
+            let left = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
+                let entered = data.enter(inner);
+                // on its first input, ship one batch holding an update at epoch 0 round 3 and one
+                // at epoch 1 round 0, under a capability for each: what an arrange does when it
+                // retires both at once.
+                let mut builder = OperatorBuilder::new("TwoEpochBatch".to_string(), inner);
+                let (output, stream) = builder.new_output();
+                let mut output = OutputBuilder::from(output);
+                let mut input = builder.new_input(entered.inner, Pipeline);
+                builder.build(move |mut caps| {
+                    let mut root = caps.pop();
+                    move |frontier| {
+                        let mut output = output.activate();
+                        input.for_each(|_cap, _data| {
+                            let Some(root) = root.take() else { return };
+                            let t1 = Product::new(0, PointStamp::new([3].into_iter().collect()));
+                            let t2 = Product::new(1, PointStamp::new([0].into_iter().collect()));
+                            let mut batcher = ValBatcher::<u64, u64, Time, isize>::new(None, 0);
+                            let mut updates = vec![((1u64, 10u64), t1.clone(), 1isize), ((2, 20), t2.clone(), 1)];
+                            batcher.insert(&mut updates);
+                            let upper = Antichain::from_elem(Product::new(2, PointStamp::new(Default::default())));
+                            let (batch, _retained) = Batcher::<Vec<((u64, u64), Time, isize)>>::extract(&mut batcher, upper.borrow());
+                            let lower = Antichain::from_elem(Product::new(0, PointStamp::new(Default::default())));
+                            let description = Description::new(lower.clone(), upper, lower);
+                            let mut spans: Vec<SpanOf<Tr>> = vec![Span::new(description, batch.map(Into::into))];
+                            let caps: CapabilitySet<Time> = [root.delayed(&t1), root.delayed(&t2)].into_iter().collect();
+                            output.session(&caps).give_container(&mut spans);
+                        });
+                        if frontier[0].frontier().is_empty() {
+                            root = None;
+                        }
+                    }
+                });
+                let records = Arranged::<TraceAgent<Tr>>::flat_map_batches(stream, |key, val| Some((*key, *val)));
+                // every message of records now names one time
+                let checked = records.inner.unary(Pipeline, "OneTime", |_, _| {
+                    move |input, output| {
+                        input.for_each(|time, data| {
+                            let _one = time.time();
+                            output.session(&time).give_container(data);
+                        });
+                    }
+                });
+                checked.as_collection().leave(scope)
+            });
+            left.inspect(move |(datum, time, diff)| seen.lock().unwrap().push((*datum, *time, *diff))).probe_with(&mut probe);
+            input
+        });
+        input.insert(0);
+        input.close();
+        worker.step_while(|| !probe.done());
+    });
+    let mut got = received.lock().unwrap().clone();
+    got.sort();
+    assert_eq!(got, vec![((1, 10), 0, 1), ((2, 20), 1, 1)]);
+}
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 3a1cb3eff..49134ddba 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -284,19 +284,36 @@ impl Backend for CorgiBackend {
             .unary(Pipeline, "CorgiAsCollection", |_, _| {
                 |input, output| {
                     input.for_each(|cap, data| {
-                        let mut session = output.session(&cap);
+                        // A message of batches may carry several timestamps; each row goes out
+                        // under the one at or before its time, so the collection's messages
+                        // carry one each. One capability is the usual case and takes the chunk
+                        // whole; several split it by row.
+                        let router = differential_dataflow::collection::StampRouter::new(&cap, 0);
                         for batch in data.iter() {
                             let Some(payload) = batch.inner.as_ref() else { continue };
                             for ch in payload.chunks.iter().filter(|c| c.len() > 0) {
-                                let mut c = CorgiContainer {
-                                    // Drop the arrangement's leading identifier lane: edges carry
-                                    // the key the program wrote, so `$0` indexes what it always did.
-                                    keys: recover_key(ch.keys()),
-                                    vals: ch.vals().clone(),
-                                    times: ch.times().to_vec(),
-                                    diffs: ch.diffs().to_vec(),
-                                };
-                                session.give_container(&mut c);
+                                // Drop the arrangement's leading identifier lane: edges carry
+                                // the key the program wrote, so `$0` indexes what it always did.
+                                let keys = recover_key(ch.keys());
+                                let times = ch.times().to_vec();
+                                if router.capabilities().len() == 1 {
+                                    let mut c = CorgiContainer { keys, vals: ch.vals().clone(), times, diffs: ch.diffs().to_vec() };
+                                    output.session(&router.capabilities()[0]).give_container(&mut c);
+                                    continue;
+                                }
+                                for (index, cap) in router.capabilities().iter().enumerate() {
+                                    let keep: Vec<usize> = (0..times.len()).filter(|&i| router.index(&times[i]) == index).collect();
+                                    if keep.is_empty() {
+                                        continue;
+                                    }
+                                    let mut c = CorgiContainer {
+                                        keys: gather(&keys, &keep),
+                                        vals: gather(ch.vals(), &keep),
+                                        times: keep.iter().map(|&i| times[i].clone()).collect(),
+                                        diffs: keep.iter().map(|&i| ch.diffs()[i]).collect(),
+                                    };
+                                    output.session(cap).give_container(&mut c);
+                                }
                             }
                         }
                     });