diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 29f6fc9fd..e31a24406 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -91,6 +91,12 @@ jobs:
         run: |
           python3 interactive/bench/ldbc/authored/check.py --server target/release/ddir_server \
             --fixture --workers 4 --reps 1 --warmup 0 --output /tmp/ldbc-authored-check
+      # Programs installed late read every import as of the time they start. A late install
+      # that reads an import at its stored times can hang (no progress for 120 s) or disagree.
+      - name: Late-install difftest sweep
+        run: |
+          DIFFTEST_SEED=40000000 DIFFTEST_CASES=500 DIFFTEST_ORACLES=split,bind,reinstall \
+            cargo test --release -p interactive --test difftest -- --ignored --nocapture
       # The explanation suite's heavier sweeps (soundness regression, fuzz)
       # are #[ignore]d for local debug runs but must gate merges. The metric
       # report is skipped (it prints, it does not assert).
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 12f39cbf1..fdca2291c 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -477,7 +477,7 @@ fn start_tail(
     let tag = reqid.to_string();
     let mut probe = ProbeHandle::new();
     let shutdown = interactive::server::host_dataflow(worker, |_, scope| {
-        let (containers, shutdown) = trace.import_containers(scope.clone(), "TailImport");
+        let (containers, shutdown) = trace.import_containers(scope, "TailImport", server.import_as_of());
         let pending = Arc::new(AtomicUsize::new(0));
         let (mut upper, mut dropped) = (None::<OuterTime>, false);
         containers
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 0d797aa44..854f2206e 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -10,7 +10,7 @@
 use timely::dataflow::Scope;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::generic::Operator;
-use timely::progress::Antichain;
+use timely::progress::{Antichain, Timestamp};
 
 use differential_dataflow::AsCollection;
 use differential_dataflow::Collection;
@@ -34,7 +34,7 @@ use crate::runtime::half_join::{CorgiHalfJoinBackend, HalfJoinBatcher};
 use crate::runtime::join::CorgiJoinBackend;
 use differential_dogs3::operators::half_join::half_join_with_tactic;
 use crate::runtime::reduce::CorgiReduceBackend;
-use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
+use differential_dataflow::operators::int_proxy::{BackendFrontier, ProxyJoinTactic, ProxyReduceTactic};
 use differential_dogs3::operators::half_join::proxy::ProxyHalfJoinTactic;
 use crate::lambda::{compile_flatmap, compile_join_function, compile_predicate, compile_projection, compile_scalar};
 use corgi::{Graph, NumOp, Shape};
@@ -230,7 +230,29 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &[Plan]) -> CC {
 
 /// The corgi backend's arrangement: one a program built, or a published export it imported.
 /// Both are the same trace type at the same time, so every operator reads either in place.
-pub(crate) type Arr<'scope> = Arranged<'scope, TraceAgent<CTrace>>;
+///
+/// An import is read as of `since`, the time its reader started: a published trace has compacted
+/// to that time, from which on its accumulations are correct, and before which its batches can
+/// hold counts no time had. Every reader advances the times it draws to `since`: a collection as
+/// it flattens the chunks, a join or reduce through DD's `BackendFrontier`. An arrangement the
+/// program built is read as it is, which `since` at the minimum says.
+///
+/// One time, not a frontier: the server compacts every published trace by the one-element
+/// frontier `{(epoch - 1, [])}` and a reader starts at that same element, so the frontier a
+/// reader reads as of is `{since}`, which is beyond the frontier its trace compacted by, as DD's
+/// `BackendFrontier` requires.
+#[derive(Clone)]
+pub(crate) struct Arr<'scope> {
+    pub(crate) arranged: Arranged<'scope, TraceAgent<CTrace>>,
+    pub(crate) since: Time,
+}
+
+impl<'scope> Arr<'scope> {
+    /// An arrangement the program built, read as it is.
+    fn built(arranged: Arranged<'scope, TraceAgent<CTrace>>) -> Self {
+        Arr { arranged, since: Time::minimum() }
+    }
+}
 
 /// The time each coordinate of `t` less one, at least zero: the half join's compaction of its
 /// arrangement. Every later stream time is at or beyond some such frontier element, and advancing
@@ -241,16 +263,20 @@ fn step_back(t: &Time, out: &mut Antichain<Time>) {
     out.insert(timely::order::Product::new(t.outer.saturating_sub(1), inner));
 }
 
-/// The updates of a stream of columnar batches, as corgi containers.
+/// The updates of a stream of columnar batches, as corgi containers, read as of `since`.
 ///
 /// Each chunk already IS a columnar container: its key/val columns clone by Arc bump,
 /// so a chunk becomes a `CorgiContainer` for the price of copying its time lanes and
 /// its diffs. One container per chunk — no concatenation, no gather, no
-/// columns→rows→columns round-trip.
+/// columns→rows→columns round-trip. The copied time lanes are advanced to `since` in
+/// place, one lane-wise max; a container's rows may then coincide, which the arrange
+/// that reads them consolidates.
 fn chunk_containers<'s, T: ColTime>(
     stream: timely::dataflow::Stream<'s, T, Vec<Span<T, std::rc::Rc<ChunkBatch<CorgiChunk<T, Diff>>>>>>,
     name: &str,
+    since: T,
 ) -> Collection<'s, T, CorgiContainer<T, Diff>> {
+    let since = (since != T::minimum()).then(|| Antichain::from_elem(since));
     stream
         .unary(Pipeline, name, move |_, _| {
             move |input, output| {
@@ -259,12 +285,16 @@ fn chunk_containers<'s, T: ColTime>(
                     for batch in data.iter() {
                         let Some(payload) = batch.inner.clone() else { continue };
                         for ch in payload.chunks.iter().filter(|c| c.len() > 0) {
+                            let mut times = ch.times().clone();
+                            if let Some(since) = &since {
+                                times.advance_by(since.borrow(), times.len());
+                            }
                             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().clone(),
+                                times,
                                 diffs: ch.diffs().to_vec(),
                             };
                             session.give_container(&mut c);
@@ -316,16 +346,16 @@ impl Backend for CorgiBackend {
         // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s
         // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher +
         // builder. No columns→rows→columns round-trip at the arrangement boundary.
-        arrange_core::<_, CC, _, CTrace>(
+        Arr::built(arrange_core::<_, CC, _, CTrace>(
             c.inner,
             CorgiPact,
             "CorgiArrange",
             ChunkBatcher::<CorgiChunker<Time, Diff>, _>::new,
-        )
+        ))
     }
 
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, CC> {
-        chunk_containers(a.stream, "CorgiAsCollection")
+        chunk_containers(a.arranged.stream, "CorgiAsCollection", a.since)
     }
 
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Lambda, ((k, v0), (_, v1)): (st::RowShape, st::RowShape)) -> Collection<'s, Time, CC> {
@@ -333,7 +363,15 @@ impl Backend for CorgiBackend {
         // compiles the projection once, now, at both inputs' shapes, and emits corgi columns
         // directly as `CorgiContainer`s — column-native, no row round-trip.
         let backend = CorgiJoinBackend::new(projection, [k, v0, v1]);
-        join_with_tactic::<_, _, _, CC>(l, r, "Join", ProxyJoinTactic::new(backend)).as_collection()
+        // An import is read as of its time. Reading one input as of a time `s` or both gives the
+        // same output when the other's times are at or beyond `s` (`(t0 ∨ s) ∨ t1 = t0 ∨ t1`
+        // then), as every program time is; two imports share one time. So the join reads both
+        // inputs as of the join of the two times, a one-element frontier as `Arr` says.
+        // Arrangements the program built have `since` at the minimum, where the wrapper returns
+        // after one comparison per unit, so every join is wrapped and compiles to one operator.
+        let since = differential_dataflow::lattice::Lattice::join(&l.since, &r.since);
+        let backend = BackendFrontier::new(backend, Antichain::from_elem(since));
+        join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(backend)).as_collection()
     }
 
     fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Lambda, level: usize, ((k, v0), (_, v1)): (st::RowShape, st::RowShape)) -> Collection<'s, Time, CC> {
@@ -342,16 +380,29 @@ impl Backend for CorgiBackend {
         // (`CorgiHalfJoinBackend`); dogs3's proxy half-join tactic decides which pairs meet.
         // `CorgiPact` places each update on the worker holding its key, as `arrange` placed the rows.
         // The stream's value is `(val, time)`, as `lift` makes it; the projection reads the val.
+        //
+        // An imported arrangement is read as it is stored, even one read as of a later time. The
+        // stream's updates are the program's, at times at or beyond that time (its inputs are at
+        // its epochs, and its imports are read as of it), and a published trace's times are
+        // `(epoch, [])`. So an arranged row before the as-of time precedes every stream update in
+        // the total order on times, strictly, as it would advanced to it; its data time joins the
+        // stream's, which is at or beyond the as-of time, to the same time either way; and the
+        // rows met by an update sum to the trace's accumulation at the update's epoch, which
+        // compaction preserves.
         let Shape::Prod(mut v0) = v0 else { panic!("half_join: a stream value of shape {v0}, not (val, time)") };
         let backend = CorgiHalfJoinBackend::new(projection, level + 1, [k, v0.swap_remove(0), v1]);
         let tactic = ProxyHalfJoinTactic::new(backend, strict);
         let yielding = |_timer: std::time::Instant, count: usize| count > 1_000_000;
-        half_join_with_tactic(stream.inner, arrangement, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic).as_collection()
+        half_join_with_tactic(stream.inner, arrangement.arranged, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic).as_collection()
     }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer, shape: st::RowShape) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
-        let tactic = ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone(), shape)).with_key_batch_size(256);
-        reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", tactic)
+        let backend = CorgiReduceBackend::new(reducer.clone(), shape);
+        // An import is read as of its time; the output is the program's own, read as it is.
+        // As for the join, the wrapper at the minimum is one comparison per unit.
+        let Arr { arranged, since } = a;
+        let backend = BackendFrontier::new(backend, Antichain::from_elem(since));
+        Arr::built(reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(backend).with_key_batch_size(256)))
     }
 
     fn inspect<'s>(c: Collection<'s, Time, CC>, label: String) -> Collection<'s, Time, CC> {
@@ -418,11 +469,12 @@ impl Backend for CorgiBackend {
         stream.as_collection()
     }
     fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: Scope<'r, Time>) -> Self::Arr<'r> {
-        a.enter_region(region)
+        Arr { arranged: a.arranged.enter_region(region), since: a.since }
     }
 }
 
-/// A root import for [`render_program`]: corgi containers, or a published export.
+/// A root import for [`render_program`]: corgi containers, or a published export read as of the
+/// time the program starts.
 #[allow(clippy::large_enum_variant)] // One per import, consumed as the program renders.
 pub(crate) enum CorgiImport<'s> {
     Columns(Collection<'s, Time, CC>),
@@ -451,7 +503,9 @@ pub(crate) fn render_program<'s>(
     let logger = scope.worker().logger_for::<crate::backend::LabelEventBuilder>("ddir/labels");
     let arranged = rendered.iter().zip(&s.exports).map(|(r, export)| {
         let first = scope.worker().peek_identifier();
-        let arranged = r.arrange();
+        // A re-exported import is the producer's arrangement; its own readers import it as of
+        // the times they start.
+        let arranged = r.arrange().arranged;
         if let Some(logger) = &logger {
             let end = scope.worker().peek_identifier();
             logger.log(crate::backend::LabelEvent { first, end, label: format!("export {}", export.name) });
@@ -467,7 +521,7 @@ pub(crate) type ExportTrace = TraceAgent<CTrace>;
 
 /// Arrange a collection as a published trace.
 pub(crate) fn arrange_export<'s>(c: Collection<'s, Time, CC>) -> Arranged<'s, ExportTrace> {
-    CorgiBackend::arrange(c)
+    CorgiBackend::arrange(c).arranged
 }
 
 /// The batches an export's arrangement has produced on this worker, not yet taken.
@@ -496,7 +550,8 @@ pub(crate) fn tap_export<'s>(a: &Arranged<'s, ExportTrace>, tap: ExportTap) -> t
     })
 }
 
-/// The updates of an imported columnar export, as corgi containers.
-pub(crate) fn export_containers<'s>(a: Arranged<'s, ExportTrace>) -> Collection<'s, Time, CC> {
-    chunk_containers(a.stream, "ExportContainers")
+/// The updates of an imported columnar export, as corgi containers, read as of `since` (see
+/// [`Arr`]).
+pub(crate) fn export_containers<'s>(a: Arranged<'s, ExportTrace>, since: Time) -> Collection<'s, Time, CC> {
+    chunk_containers(a.stream, "ExportContainers", since)
 }
diff --git a/interactive/src/server/bind.rs b/interactive/src/server/bind.rs
index b90663776..2706255b1 100644
--- a/interactive/src/server/bind.rs
+++ b/interactive/src/server/bind.rs
@@ -97,11 +97,12 @@ impl Server {
         // Capture the trace's containers, never forming rows.
         let buffer: Captured = Rc::new(RefCell::new(Vec::new()));
         let buffer_in = buffer.clone();
-        let Published(mut trace) = self.published(&source).expect("checked above");
+        let mut published = self.published(&source).expect("checked above");
+        let as_of = self.import_as_of();
         let shutdown = host_dataflow(worker, |_, scope| {
             use timely::dataflow::operators::{Inspect, Probe};
-            let (arranged, shutdown) = trace.import_core(scope.clone(), "BindImport");
-            crate::backend::corgi::export_containers(arranged)
+            let (containers, shutdown) = published.import_containers(scope, "BindImport", as_of);
+            containers
                 .inner
                 .inspect_core(move |event| {
                     if let Ok((_, container)) = event {
diff --git a/interactive/src/server/install.rs b/interactive/src/server/install.rs
index 2dc2e74c3..547ce4d09 100644
--- a/interactive/src/server/install.rs
+++ b/interactive/src/server/install.rs
@@ -123,6 +123,7 @@ impl Server {
         let root = &prog.root;
         let arranged_imports = self.arranged_imports;
         let tapped = self.export_taps;
+        let as_of = self.import_as_of();
         let traces = &mut self.traces;
 
         // The id this dataflow will get; captured so `drop` can remove it.
@@ -137,7 +138,7 @@ impl Server {
                 use timely::dataflow::operators::core::Input as _;
                 use timely::dataflow::operators::Probe;
                 use differential_dataflow::AsCollection;
-                use crate::backend::corgi::{export_containers, CorgiImport};
+                use crate::backend::corgi::{export_containers, Arr, CorgiImport};
                 let mut inputs: Vec<(usize, ServerInput)> = Vec::new();
 
                 // One value per root import, rendered without rows. An input converts rows to
@@ -154,13 +155,16 @@ impl Server {
                             CorgiImport::Columns(stream.as_collection().enter(scope))
                         }
                         st::Source::Trace(t) => {
-                            // The first binding point: resolve a named trace by importing it.
+                            // The first binding point: resolve a named trace by importing it, as
+                            // of the program's start.
                             let key = canonical_source_name(t);
-                            let arranged = traces.get_mut(&key).expect("validated above").import(scope);
+                            let trace = traces.get_mut(&key).expect("validated above");
+                            debug_assert!(trace.get_logical_compaction().less_equal(&as_of), "an import as of {as_of:?}, before its trace's compaction frontier");
+                            let arranged = trace.import(scope);
                             if arranged_imports {
-                                CorgiImport::Export(arranged)
+                                CorgiImport::Export(Arr { arranged, since: as_of.clone() })
                             } else {
-                                CorgiImport::Columns(export_containers(arranged))
+                                CorgiImport::Columns(export_containers(arranged, as_of.clone()))
                             }
                         }
                         st::Source::Parent(_) => unreachable!("root import from a parent scope"),
diff --git a/interactive/src/server/mod.rs b/interactive/src/server/mod.rs
index 258f91c16..a34465c1f 100644
--- a/interactive/src/server/mod.rs
+++ b/interactive/src/server/mod.rs
@@ -98,18 +98,21 @@ use self::sources::{canonical_source_name, clock_rows, generated_shape, InputCon
 pub struct Published(pub crate::backend::corgi::ExportTrace);
 
 impl Published {
-    /// Import into `scope` as containers, one per chunk of each batch, with the import's shutdown
-    /// button. Each worker reads its own share.
+    /// Import into `scope` as containers, one per chunk of each batch, read as of `as_of` (see
+    /// [`Server::import_as_of`]), with the import's shutdown button. Each worker reads its own
+    /// share.
     pub fn import_containers<'s>(
         &mut self,
         scope: timely::dataflow::Scope<'s, HostTime>,
         name: &str,
+        as_of: HostTime,
     ) -> (
         differential_dataflow::Collection<'s, HostTime, HostContainer>,
         ShutdownButton<CapabilitySet<HostTime>>,
     ) {
+        debug_assert!(self.0.get_logical_compaction().less_equal(&as_of), "an import as of {as_of:?}, before its trace's compaction frontier");
         let (arranged, shutdown) = self.0.import_core(scope, name);
-        (crate::backend::corgi::export_containers(arranged), shutdown)
+        (crate::backend::corgi::export_containers(arranged, as_of), shutdown)
     }
 }
 
@@ -278,6 +281,17 @@ impl Server {
         self.epoch
     }
 
+    /// The time a program installed now, or a binding, snapshot or tail started now, reads the
+    /// published traces it imports as of: the epoch before the current one, as a program time.
+    ///
+    /// [`Server::tick`] compacts every published trace to it, so their accumulations are correct
+    /// from it on; before it a trace's batches can hold counts no time had (a retraction advanced
+    /// to it beside the insertion it retracts), on which an iterative scope need not converge.
+    /// Every import advances the times it reads to it, as DD's `import_frontier` reads a trace.
+    pub fn import_as_of(&self) -> HostTime {
+        host_time(self.epoch.saturating_sub(1))
+    }
+
     /// Whether `name` (canonical) is published.
     fn is_published(&self, name: &str) -> bool {
         self.traces.contains_key(name)
diff --git a/interactive/src/server/read.rs b/interactive/src/server/read.rs
index ec525af68..060c0be83 100644
--- a/interactive/src/server/read.rs
+++ b/interactive/src/server/read.rs
@@ -42,7 +42,8 @@ impl Server {
 
         let name = canonical_source_name(name);
         let epoch = self.epoch;
-        let Published(mut trace) = self.published(&name).ok_or_else(|| format!("no trace {:?}", name))?;
+        let mut published = self.published(&name).ok_or_else(|| format!("no trace {:?}", name))?;
+        let as_of = self.import_as_of();
         let shape = self.shapes[&name].clone();
         let parts: Rc<RefCell<Vec<Updates>>> = Rc::new(RefCell::new(Vec::new()));
         let parts_in = parts.clone();
@@ -50,8 +51,8 @@ impl Server {
         let id = worker.next_dataflow_index();
         host_dataflow(worker, |_, scope| {
             // The dataflow is dropped once drained, which releases the import.
-            let (arranged, _shutdown) = trace.import_core(scope.clone(), "SnapshotImport");
-            crate::backend::corgi::export_containers(arranged)
+            let (containers, _shutdown) = published.import_containers(scope, "SnapshotImport", as_of);
+            containers
                 .inner
                 .unary::<timely::container::CapacityContainerBuilder<HostContainer>, _, _, _>(crate::runtime::exchange::GatherPact, "Snapshot", move |_, _| {
                     move |input, _output| {
diff --git a/interactive/tests/difftest.rs b/interactive/tests/difftest.rs
index 5977650fe..968abce84 100644
--- a/interactive/tests/difftest.rs
+++ b/interactive/tests/difftest.rs
@@ -21,8 +21,8 @@
 //! - `bind`: the consumer's inputs bound to the producer's published traces instead.
 //! - `reinstall`: the consumer dropped, and installed again later.
 //!
-//! Until imports read their traces as of their install, a consumer is installed with its producer,
-//! and `reinstall` does not run: see [`LATE_INSTALLS`].
+//! A consumer may be installed after its producer has run some epochs, reading compacted history
+//! as of its install: see [`LATE_INSTALLS`].
 //! - `explain`: every row of the first export is regenerated by its explanation alone (see
 //!   `Case::explainable` for the cases it skips).
 //!
@@ -77,9 +77,9 @@ enum Sh { Int, Unit, Tup(Vec<Sh>), List(Box<Sh>), Sum, Float }
 
 /// Whether `split` and `bind` may install their consumer after the producer has run some epochs,
 /// and whether `reinstall` runs. A consumer installed late imports traces whose history has been
-/// compacted, and does not converge until each import reads its trace as of the time it starts
-/// (#982). Turning this on changes no case: the late epoch is drawn either way.
-const LATE_INSTALLS: bool = false;
+/// compacted, which it reads as of the time it starts (`Server::import_as_of`). Turning this off
+/// changes no case: the late epoch is drawn either way.
+const LATE_INSTALLS: bool = true;
 
 /// The one declared sum, `S`: its lanes' payload shapes are `int`, `(int, int)` and `()`.
 const SUM_DECL: &str = "type S = A int | B (int, int) | C;";
@@ -1420,6 +1420,22 @@ fn now_secs() -> u64 { std::time::SystemTime::now().duration_since(std::time::UN
 /// Note progress, for the watchdog.
 fn tick() { LAST.store(now_secs(), std::sync::atomic::Ordering::Relaxed); }
 
+/// Seeds whose consumer, installed after its producer had run some epochs, once failed to
+/// converge: it read its imports' compacted history at the stored times, where a retraction
+/// advanced to the compaction frontier sits beside the insertion it retracts. Each now reads its
+/// imports as of its install.
+#[test]
+fn late_install_regressions() {
+    let oracles = [Oracle::Split, Oracle::Bind, Oracle::Reinstall];
+    for seed in [40000075, 40000160, 40000519] {
+        match check(&generate(seed), &oracles) {
+            Outcome::Pass => {}
+            Outcome::Refused(e) => panic!("seed {seed} refused: {e}"),
+            Outcome::Fail(oracle, why) => panic!("seed {seed}, {} oracle: {why}", oracle.name()),
+        }
+    }
+}
+
 #[test]
 #[ignore = "differential sweep; run with --release -- --ignored --nocapture"]
 fn difftest() {
diff --git a/interactive/tests/server_late_import.rs b/interactive/tests/server_late_import.rs
new file mode 100644
index 000000000..0ea2b95f0
--- /dev/null
+++ b/interactive/tests/server_late_import.rs
@@ -0,0 +1,151 @@
+//! A program installed late reads the published traces it imports as of the time it starts
+//! (`Server::import_as_of`), the traces' compaction frontier: every update's time advanced to it.
+//! Before that frontier a compacted trace's batches can hold counts no time ever had (a retraction
+//! advanced past the frontier, the insertion it retracts left behind), and an iterative scope
+//! reading them there need not converge. Imports read as containers, as rows of arrangements, and
+//! as arrangements read directly by a join and a reduce, all from the frontier on. (Found by
+//! `difftest`.)
+
+use std::sync::mpsc;
+use std::time::Duration;
+
+use interactive::server::Server;
+use interactive::{lower, parse};
+
+const PRODUCER: &str = r#"
+    export "seeds" = input 0 : ((int, int), ());
+    export "edges" = input 1 : ((int, int), ());
+"#;
+
+/// A row `(a, b)` of `x` meets an equal row of `edges` and steps to `(b % 5, b)`: `(5, 5)` steps
+/// to `(0, 5)`, which steps to itself while `edges` holds `(0, 5)`.
+const CONSUMER: &str = r#"
+    let seeds = import "seeds";
+    let edges = import "edges";
+    s: {
+        let step = x | join(edges, ((_, b), (), ()) -> (((b, 5) rem, b), ()));
+        var x = (seeds + step) | distinct;
+    }
+    export "result" = s::x;
+"#;
+
+/// The feeds, per epoch, as `(input, rows)`: `(0, 5)` is added at epoch 2 and retracted at 14, and
+/// enough else changes that the edges trace merges its batches unevenly.
+const FEEDS: &[(u64, usize, &str)] = &[
+    (0, 1, "0,0\n0,3 diff=2"),
+    (1, 1, "-1,4 diff=2"),
+    (2, 1, "1,2\n0,5"),
+    (3, 1, "4,3\n3,0\n0,4"),
+    (4, 1, "4,3 diff=2\n1,5"),
+    (5, 0, "5,5"),
+    (7, 1, "1,2\n1,5 diff=-1\n4,3 diff=-1"),
+    (8, 1, "1,2 diff=-1\n4,3 diff=-1"),
+    (10, 1, "5,3\n0,3 diff=-1"),
+    (11, 1, "2,3\n4,4\n-1,-1\n1,2 diff=-1"),
+    (14, 1, "5,5\n0,5 diff=-1"),
+    (21, 0, "-1,2"),
+];
+
+/// The consumer's rows after the last epoch, installed before epoch `late`, with imports as
+/// containers or as arrangements.
+fn run(late: u64, arranged: bool) -> Vec<(String, String, i64)> {
+    timely::execute_directly(move |worker| {
+        let program = |src: &str| { let mut p = lower::lower_tree(parse::pipe::parse(src)); p.optimize(); p };
+        let mut server = Server::new();
+        server.set_arranged_imports(arranged);
+        server.install(worker, "producer", &program(PRODUCER)).unwrap();
+        for epoch in 0..22 {
+            if epoch == late { server.install(worker, "consumer", &program(CONSUMER)).unwrap(); }
+            for (_, input, rows) in FEEDS.iter().filter(|(e, _, _)| *e == epoch) {
+                server.feed("producer", *input, rows, None).unwrap();
+            }
+            server.tick(worker);
+        }
+        server.snapshot(worker, "result").unwrap().rows()
+    })
+}
+
+#[test]
+fn late_importers_converge() {
+    // Without the fix, the late cases never finish their first tick: fail rather than hang.
+    let (send, recv) = mpsc::channel();
+    std::thread::spawn(move || {
+        let early = run(0, false);
+        let late: Vec<_> = [(17, false), (21, false), (17, true), (21, true)].into_iter().map(|(at, arranged)| run(at, arranged)).collect();
+        send.send((early, late)).unwrap();
+    });
+    let (early, late) = recv.recv_timeout(Duration::from_secs(60)).expect("a late importer did not converge");
+    assert_eq!(early.len(), 3, "(5, 5), (0, 5) and (-1, 2): {early:?}");
+    for rows in late {
+        assert_eq!(rows, early);
+    }
+}
+
+/// As above, with imports read as arrangements and operators reading them directly. `r` joins the
+/// imported `pairs` with itself: read from before the compaction frontier, the join met the two
+/// histories at their stored times and gave `r` rows there, and the loop then joined it with the
+/// imported `weights` where that history holds counts no time had. `w` reduces an import directly.
+const ARRANGED_PRODUCER: &str = r#"
+    export "pairs" = input 0 : ((int, int), ());
+    export "weights" = input 5 : (int, float);
+"#;
+
+const ARRANGED_CONSUMER: &str = r#"
+    let pairs = import "pairs";
+    let weights = import "weights";
+    let r = pairs | join(pairs, ((a, b), (), _) -> ((b, (a, b) min) eq, 0)) | distinct;
+    s: {
+        let step = x | join(weights, (k, v, _) -> (k, v));
+        var x = (r + step) | min;
+    }
+    export "result" = s::x;
+    export "least" = weights | min;
+"#;
+
+const ARRANGED_FEEDS: &[(u64, usize, &str)] = &[
+    (0, 0, "1,0 diff=1"),
+    (0, 5, "-1 val=-0.0 diff=2\n5 val=-0.0 diff=1\n3 val=-1.5 diff=1\n4 val=-1.5 diff=1\n3 val=1.5 diff=2"),
+    (2, 5, "1 val=1e300 diff=1\n0 val=-1.5 diff=1\n4 val=-0.0 diff=1\n3 val=4.0 diff=1"),
+    (4, 5, "5 val=2.5 diff=1"),
+    (5, 5, "3 val=1e300 diff=1\n-1 val=-1e-300 diff=1\n3 val=0.1 diff=1"),
+    (6, 5, "1 val=4.0 diff=2\n0 val=-1e-300 diff=2\n-1 val=1.5 diff=1\n5 val=0.1 diff=1"),
+    (7, 5, "1 val=-1e-300 diff=2\n4 val=0.0 diff=1\n2 val=0.0 diff=1\n0 val=0.0 diff=1\n1 val=4.0 diff=-1"),
+    (9, 5, "4 val=2.5 diff=1\n5 val=2.0 diff=1\n4 val=0.0 diff=1"),
+    (12, 5, "1 val=1e300 diff=1\n0 val=-1e-300 diff=-1"),
+    (14, 5, "2 val=1.5 diff=1\n-1 val=1.5 diff=-1\n4 val=0.0 diff=-1"),
+    (16, 5, "4 val=0.1 diff=1\n4 val=-1e-300 diff=1"),
+    (18, 5, "2 val=0.1 diff=1\n3 val=-3.25 diff=2\n-1 val=0.0 diff=1\n3 val=0.0 diff=1\n0 val=1.5 diff=1"),
+    (20, 5, "3 val=1.5 diff=1\n3 val=0.0 diff=1\n0 val=1.5 diff=1"),
+    (22, 5, "2 val=0.1 diff=1\n2 val=-3.25 diff=2\n1 val=2.0 diff=1"),
+    (23, 5, "5 val=-0.0 diff=2\n0 val=0.1 diff=1\n4 val=0.1 diff=1\n1 val=-1e-300 diff=-1\n5 val=0.1 diff=-1"),
+    (25, 5, "1 val=2.0 diff=1"),
+];
+
+/// The arranged consumer's exports after the last epoch, installed before epoch `late`.
+fn run_arranged(late: u64) -> (Vec<(String, String, i64)>, Vec<(String, String, i64)>) {
+    timely::execute_directly(move |worker| {
+        let program = |src: &str| { let mut p = lower::lower_tree(parse::pipe::parse(src)); p.optimize(); p };
+        let mut server = Server::new();
+        server.set_arranged_imports(true);
+        server.install(worker, "producer", &program(ARRANGED_PRODUCER)).unwrap();
+        for epoch in 0..27 {
+            if epoch == late { server.install(worker, "consumer", &program(ARRANGED_CONSUMER)).unwrap(); }
+            for (_, input, rows) in ARRANGED_FEEDS.iter().filter(|(e, _, _)| *e == epoch) {
+                server.feed("producer", *input, rows, None).unwrap();
+            }
+            server.tick(worker);
+        }
+        (server.snapshot(worker, "result").unwrap().rows(), server.snapshot(worker, "least").unwrap().rows())
+    })
+}
+
+#[test]
+fn late_arranged_importers_converge() {
+    let (send, recv) = mpsc::channel();
+    std::thread::spawn(move || {
+        send.send((run_arranged(0), run_arranged(25))).unwrap();
+    });
+    let (early, late) = recv.recv_timeout(Duration::from_secs(30)).expect("the late importer did not converge");
+    assert_eq!(early.0.len(), 1, "key 1, at the least value 0: {:?}", early.0);
+    assert_eq!(late, early);
+}