diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml
index a6d3781f4..705fa0a8b 100644
--- a/interactive/Cargo.toml
+++ b/interactive/Cargo.toml
@@ -16,6 +16,7 @@ columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
 corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "41e873124a850772a9c4f7df88be94e3163512ef", features = ["serde"] }
 differential-dataflow = { workspace = true }
+differential-dogs3 = { path = "../dogsdogsdogs", default-features = false }
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"
 timely = { workspace = true }
diff --git a/interactive/examples/triangles_bench.rs b/interactive/examples/triangles_bench.rs
new file mode 100644
index 000000000..94b7d5596
--- /dev/null
+++ b/interactive/examples/triangles_bench.rs
@@ -0,0 +1,133 @@
+//! Triangles in a random graph, by binary joins or by a delta query of half joins. The binary
+//! joins arrange the two-paths; the delta query arranges only the edges, by source, by
+//! destination, and by themselves. Reports the time and memory to load the graph, and then the
+//! time per tick as edges are replaced.
+//!
+//! ```text
+//! cargo run --release --example triangles_bench -- [join|delta] [edges] [nodes] [churn] [ticks] [workers]
+//! ```
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Instant;
+
+/// Counts live heap bytes and their peak, to tell the heap's size from the process's.
+struct Counting;
+static LIVE: AtomicUsize = AtomicUsize::new(0);
+static PEAK: AtomicUsize = AtomicUsize::new(0);
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        let live = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size();
+        PEAK.fetch_max(live, Ordering::Relaxed);
+        unsafe { System.alloc(layout) }
+    }
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        LIVE.fetch_sub(layout.size(), Ordering::Relaxed);
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+#[global_allocator]
+static ALLOC: Counting = Counting;
+
+fn heap() -> String {
+    format!("heap {:.1} MiB (peak {:.1} MiB)", LIVE.load(Ordering::Relaxed) as f64 / 1048576.0, PEAK.load(Ordering::Relaxed) as f64 / 1048576.0)
+}
+
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+const EDGES: &str = r#"
+    let e = input 0 : ((int, int) ; ()) | key(((x, y), ()) -> (x, y));
+    let r = e | key((x, y) -> (y, x));
+    let s = e | key((x, y) -> ((x, y), ()));
+"#;
+
+/// Each triangle `(a, b, c)` has edges `(a, b)`, `(b, c)`, and `(a, c)`.
+const JOIN: &str = r#"
+    let paths = r | join(e, (b, a, c) -> ((a, c), b));
+    let tri = paths | join(s, ((a, c), b, ()) -> ((a, b, c), ()));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// One path of half joins per edge of the triangle, each meeting the earlier edges' changes at or
+/// before its own time and the later edges' strictly before.
+const DELTA: &str = r#"
+    let d1 = r | lift | half_join(e, 1, (b, a, c) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d2 = e | lift | half_join(r, 0, (b, c, a) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d3 = e | lift | half_join(e, 0, (a, c, b) -> ((b, c), a)) | half_join(s, 0, ((b, c), a, ()) -> ((a, b, c), ()));
+    let tri = (d1 + d2 + d3) | delay((_, (_, t)) -> t) | key((abc, (v, _)) -> (abc, v));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// This process's memory in MiB: on macOS its `footprint`, elsewhere its RSS.
+fn memory_mib() -> Option<f64> {
+    let out = std::process::Command::new("footprint").args(["-p", &std::process::id().to_string()]).output();
+    if let Ok(out) = out {
+        let text = String::from_utf8_lossy(&out.stdout).into_owned();
+        let mut words = text.split("Footprint:").nth(1)?.split_whitespace();
+        let n: f64 = words.next()?.parse().ok()?;
+        return match words.next()? {
+            "B" => Some(n / (1024.0 * 1024.0)),
+            "KB" => Some(n / 1024.0),
+            "MB" => Some(n),
+            "GB" => Some(n * 1024.0),
+            _ => None,
+        };
+    }
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    let kib: f64 = status.lines().find(|l| l.starts_with("VmRSS:"))?.split_whitespace().nth(1)?.parse().ok()?;
+    Some(kib / 1024.0)
+}
+
+fn mem() -> String {
+    memory_mib().map_or_else(|| "n/a".to_string(), |m| format!("{m:>7.1} MiB"))
+}
+
+fn main() {
+    let mut args = std::env::args().skip(1);
+    let method = args.next().unwrap_or("delta".into());
+    let nums: Vec<u64> = args.map(|a| a.parse().expect("integer argument")).collect();
+    let edges = nums.first().copied().unwrap_or(1_000_000);
+    let nodes = nums.get(1).copied().unwrap_or(100_000);
+    let churn = nums.get(2).copied().unwrap_or(1_000);
+    let ticks = nums.get(3).copied().unwrap_or(20);
+    let workers = nums.get(4).copied().unwrap_or(1) as usize;
+    let query = match method.as_str() {
+        "join" => JOIN,
+        "delta" => DELTA,
+        other => panic!("unknown method {other:?}: join or delta"),
+    };
+    let source = format!("{EDGES}{query}");
+
+    timely::execute(timely::Config::process(workers), move |worker| {
+        let index = worker.index();
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        let mut program = lower::lower_tree(parse::pipe::parse(&source));
+        program.optimize();
+        server.install(worker, "p", &program).unwrap();
+        server.load(worker, "p", 0, &format!("random:nodes={nodes},edges={edges},churn={churn}")).unwrap();
+        let loaded = Instant::now();
+        server.tick(worker);
+        let count = |server: &mut Server, worker: &mut timely::worker::Worker| -> i64 {
+            server.snapshot(worker, "count").unwrap().iter().map(|(_, v, d)| match v { interactive::ir::Value::Int(n) => n * d, _ => 0 }).sum()
+        };
+        // A snapshot is a dataflow, so every worker takes part; worker 0 receives the rows.
+        let elapsed = loaded.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!("{method}: {edges} edges over {nodes} nodes, {workers} workers");
+            println!("load:  {:>9.1} ms  mem {}  {}  count {n}", elapsed.as_secs_f64() * 1e3, mem(), heap());
+        }
+        let churned = Instant::now();
+        for _ in 0..ticks { server.tick(worker); }
+        let elapsed = churned.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!(
+                "churn: {:>9.2} ms/tick over {ticks} ticks ({churn} edges replaced per tick)  mem {}  count {n}",
+                elapsed.as_secs_f64() * 1e3 / ticks as f64, mem(),
+            );
+        }
+    })
+    .unwrap();
+}
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 62a527391..53d56a498 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -32,7 +32,9 @@ use crate::corgi::chunk::{recover_key, CorgiChunk, CorgiChunker};
 use crate::corgi::col_times::ColTime;
 use crate::corgi::container::CorgiContainer;
 use crate::corgi::exchange::CorgiPact;
+use crate::corgi::half_join::{CorgiHalfJoin, HalfJoinBatcher};
 use crate::corgi::join::CorgiJoinBackend;
+use differential_dogs3::operators::half_join::{half_join_with_tactic, HalfJoinTactic};
 use crate::corgi::reduce::CorgiReduceBackend;
 use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
 use crate::corgi::logic::{compile_flatmap, compile_join_function, compile_predicate, compile_projection, compile_scalar, shape_of_row};
@@ -275,6 +277,23 @@ impl<B0: Native, B1: Native, J: JoinTactic<Time, NativeBatch, NativeBatch, CC>>
     }
 }
 
+/// A half-join tactic over native batches, accepting an imported arrangement.
+struct RetimeHalfJoin<J>(J);
+impl<B: Native, M, J: HalfJoinTactic<Time, NativeBatch, M, CC>> HalfJoinTactic<Time, B, M, CC> for RetimeHalfJoin<J> {
+    fn prep(&mut self, chunks: Vec<M>, batches: Vec<B>, lower: Antichain<Time>) -> Box<dyn Iterator<Item = (CC, Time)>> {
+        self.0.prep(chunks, batches.into_iter().map(Native::native).collect(), lower)
+    }
+}
+
+/// 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
+/// arranged times to it keeps their order against those stream times, strict and not.
+fn step_back(t: &Time, out: &mut Antichain<Time>) {
+    use differential_dataflow::dynamic::pointstamp::PointStamp;
+    let inner = PointStamp::new(t.inner.iter().map(|x| x.saturating_sub(1)).collect());
+    out.insert(timely::order::Product::new(t.outer.saturating_sub(1), inner));
+}
+
 /// A reduce tactic over native batches, accepting an imported input. (Its output is its own.)
 struct RetimeReduce<R>(R);
 impl<B1: Native, R: ReduceTactic<Time, NativeBatch, NativeBatch>> ReduceTactic<Time, B1, NativeBatch> for RetimeReduce<R> {
@@ -404,6 +423,17 @@ impl Backend for CorgiBackend {
         }
     }
 
+    fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, CC> {
+        // dogs3's driver owns progress and yielding; corgi supplies when an update is unblocked
+        // (`HalfJoinBatcher`) and how released updates meet the arrangement (`CorgiHalfJoin`).
+        // `CorgiPact` places each update on the worker holding its key, as `arrange` placed the rows.
+        let tactic = || RetimeHalfJoin(CorgiHalfJoin::new(projection.key.clone(), projection.val.clone(), strict, level + 1));
+        let yielding = |_timer: std::time::Instant, count: usize| count > 1_000_000;
+        match arrangement {
+            Arr::Local(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+            Arr::Imported(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+        }
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
         let tactic = RetimeReduce(ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())).with_key_batch_size(256));
diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs
index 28eaa8351..5c69fca01 100644
--- a/interactive/src/backend/mod.rs
+++ b/interactive/src/backend/mod.rs
@@ -49,6 +49,9 @@ pub trait Backend {
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container>;
     /// `shapes` are the two inputs' row shapes, where they are known before any data arrives.
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
+    /// See `scope_ir::Node::HalfJoin`. `level` is the scope depth, which fixes the width of a time
+    /// as data; `shapes` are the stream's and the arrangement's row shapes, where known.
+    fn half_join<'s>(stream: Collection<'s, Time, Self::Container>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
     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>;
@@ -133,6 +136,7 @@ fn reads_rows(s: &st::Scope, k: usize) -> bool {
     let is_k = |r: &st::Ref| matches!(r, st::Ref::Import(i) if *i == k);
     s.items.iter().any(|item| match item {
         st::Item::Op(st::Node::Linear { input, .. } | st::Node::Inspect { input, .. }) => is_k(input),
+        st::Item::Op(st::Node::HalfJoin { stream, .. }) => is_k(stream),
         st::Item::Op(st::Node::Concat(refs)) => refs.iter().any(is_k),
         st::Item::Op(st::Node::Arrange(_) | st::Node::Join { .. } | st::Node::Reduce { .. }) => false,
         st::Item::Sub(_) => false,
@@ -189,6 +193,11 @@ pub fn render_tree<'s, B: Backend>(
                         let r = resolve(&items, &imports, &var_cols, right).arrange();
                         Rendered::Collection(B::join(l, r, projection, shape_of(left).zip(shape_of(right))))
                     },
+                    st::Node::HalfJoin { stream, arrangement, strict, projection } => {
+                        let c = resolve(&items, &imports, &var_cols, stream).collection();
+                        let a = resolve(&items, &imports, &var_cols, arrangement).arrange();
+                        Rendered::Collection(B::half_join(c, a, *strict, projection, depth, shape_of(stream).zip(shape_of(arrangement))))
+                    },
                     st::Node::Reduce { input, reducer } => {
                         let a = resolve(&items, &imports, &var_cols, input).arrange();
                         Rendered::Arrangement(B::reduce(a, reducer))
diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs
index 5dea98f7b..8a918f6b4 100644
--- a/interactive/src/backend/vec.rs
+++ b/interactive/src/backend/vec.rs
@@ -207,6 +207,9 @@ impl Backend for VecBackend {
             (Arr::Imported(l), Arr::Imported(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
         }
     }
+    fn half_join<'s>(_stream: Collection<'s, Time, Self::Container>, _arrangement: Self::Arr<'s>, _strict: bool, _projection: &Projection, _level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container> {
+        panic!("half_join runs on the corgi backend only")
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         let f: Arc<dyn Fn(&Row, &[(&Row, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync> = match reducer {
             Reducer::Min => Arc::new(|_key, vals, output| { if let Some(min) = vals.iter().map(|(v, _)| (*v).clone()).min() { output.push((min, 1)); } }),
diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs
index eb50b47d9..bb59da9e9 100644
--- a/interactive/src/corgi/chunk.rs
+++ b/interactive/src/corgi/chunk.rs
@@ -640,7 +640,7 @@ pub fn recover_key_ref(keys: &CValue) -> &CValue {
 }
 
 /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort).
-fn concat_blocks(blocks: &[CValue]) -> CValue {
+pub(crate) fn concat_blocks(blocks: &[CValue]) -> CValue {
     if blocks.len() == 1 {
         return blocks[0].clone();
     }
diff --git a/interactive/src/corgi/col_times.rs b/interactive/src/corgi/col_times.rs
index d5613c7aa..d32f0359d 100644
--- a/interactive/src/corgi/col_times.rs
+++ b/interactive/src/corgi/col_times.rs
@@ -209,6 +209,11 @@ impl<T> ColTimes<T> {
         Ordering::Equal
     }
 
+    /// Whether this column's row `i` is less than or equal to `other`'s row `j` in the partial order.
+    pub fn less_equal_cross(&self, i: usize, other: &ColTimes<T>, j: usize) -> bool {
+        (0..self.lanes.len().max(other.lanes.len())).all(|k| self.at(k, i) <= other.at(k, j))
+    }
+
     /// Whether row `i` is less than or equal to row `j` in the partial order.
     fn less_equal(&self, i: usize, j: usize) -> bool {
         self.lanes.iter().all(|lane| lane[i] <= lane[j])
diff --git a/interactive/src/corgi/half_join/batcher.rs b/interactive/src/corgi/half_join/batcher.rs
new file mode 100644
index 000000000..11c6a8454
--- /dev/null
+++ b/interactive/src/corgi/half_join/batcher.rs
@@ -0,0 +1,387 @@
+//! The half join's batcher: arrivals wait in a merge ladder of chains in time order, and what
+//! the frontier unblocks leaves the fronts of the chains as runs in identifier order.
+
+use std::borrow::Cow;
+use std::cmp::Ordering;
+
+use timely::progress::Antichain;
+use timely::progress::frontier::AntichainRef;
+
+use differential_dataflow::batcher::Batcher;
+use differential_dataflow::batcher::merge::{MergeLadder, Merger};
+
+use corgi::arrange::{gather, gather_lanes, sort_blocks, survey_groups, GroupRun};
+use corgi::Value as CValue;
+
+use crate::corgi::chunk::concat_blocks;
+use crate::corgi::col_times::{ColTimes, Lanes};
+use crate::ir::{Diff, Time};
+
+use super::{gather_rows, Part, PartTimes, Run, CC};
+
+/// The most rows in one [`Run`]. Small under test, so that units hold many runs.
+const RUN: usize = if cfg!(test) { 4 } else { 1 << 16 };
+
+/// The most rows a merge puts in one [`TimeChunk`]. Releasing part of a chunk recomputes its
+/// lower bound, which costs its length. Small under test, so that chains hold many chunks.
+const TIME_TARGET: usize = if cfg!(test) { 5 } else { 1 << 16 };
+
+/// Updates in the total order on their times: rows `(t, key, val)`, where `t` is the update's own
+/// time as a `Prod` of `Int` lanes, sorted in corgi's structural order and consolidated. The
+/// structural order on the lanes is lexicographic, which is the total order on times.
+pub struct TimeChunk {
+    /// `Prod([t, key, val])`.
+    rows: CValue,
+    diffs: Vec<Diff>,
+    /// Rows before `start` have been released.
+    start: usize,
+    /// The minimal times among the rows from `start` on, in the partial order.
+    lower: Antichain<Time>,
+}
+
+impl Default for TimeChunk {
+    fn default() -> Self { TimeChunk { rows: CValue::Unit(0), diffs: Vec::new(), start: 0, lower: Antichain::new() } }
+}
+
+impl TimeChunk {
+    fn new(rows: CValue, diffs: Vec<Diff>) -> Self {
+        let mut chunk = TimeChunk { rows, diffs, start: 0, lower: Antichain::new() };
+        chunk.set_lower();
+        chunk
+    }
+
+    /// Sorts and consolidates rows `(t, key, val)`.
+    fn form(rows: CValue, diffs: Vec<Diff>) -> Option<Self> {
+        let (perm, labels) = sort_blocks(&[], &rows);
+        let (mut keep, mut kept) = (Vec::new(), Vec::new());
+        let mut i = 0;
+        while i < perm.len() {
+            let mut j = i + 1;
+            while j < perm.len() && labels[j] == labels[i] { j += 1; }
+            let d: Diff = perm[i..j].iter().map(|&p| diffs[p]).sum();
+            if d != 0 {
+                keep.push(perm[i]);
+                kept.push(d);
+            }
+            i = j;
+        }
+        if kept.is_empty() { return None; }
+        Some(TimeChunk::new(gather(&rows, &keep), kept))
+    }
+
+    /// The number of rows not yet released.
+    fn len(&self) -> usize { self.diffs.len() - self.start }
+
+    /// The time lanes, as integers.
+    fn lanes(&self) -> Vec<Cow<'_, [i64]>> { time_lanes(&self.rows) }
+
+    /// Recompute `lower` from the rows not yet released. Equal times are adjacent, so each distinct
+    /// time is offered once.
+    fn set_lower(&mut self) {
+        let lanes = self.lanes();
+        let mut lower = Antichain::new();
+        for i in self.start..self.diffs.len() {
+            if i > self.start && lanes.iter().all(|lane| lane[i - 1] == lane[i]) { continue; }
+            lower.insert(Time::from_coordinates(lanes.iter().map(|lane| lane[i] as u64)));
+        }
+        drop(lanes);
+        self.lower = lower;
+    }
+
+    /// The end of the rows `bound` unblocks: those before the bound in the total order, and those
+    /// equal to it when `strict`. No bound unblocks every row.
+    fn unblocked(&self, bound: Option<&[i64]>, strict: bool) -> usize {
+        let Some(bound) = bound else { return self.diffs.len() };
+        let lanes = self.lanes();
+        let (mut lo, mut hi) = (self.start, self.diffs.len());
+        while lo < hi {
+            let mid = (lo + hi) / 2;
+            if unblocked_at(&lanes, mid, bound, strict) { lo = mid + 1 } else { hi = mid }
+        }
+        lo
+    }
+
+    /// Release the rows up to `end`, which no longer count towards `lower`.
+    fn release_to(&mut self, end: usize) -> TimeChunk {
+        let rows: Vec<usize> = (self.start..end).collect();
+        let released = TimeChunk { rows: gather(&self.rows, &rows), diffs: self.diffs[self.start..end].to_vec(), start: 0, lower: Antichain::new() };
+        self.start = end;
+        self.set_lower();
+        released
+    }
+}
+
+/// Staged updates as rows `(t, key, val)`, their times written as `width` lanes, and their diffs.
+fn stage_rows(stage: Vec<CC>, width: usize) -> Option<(CValue, Vec<Diff>)> {
+    if stage.is_empty() { return None; }
+    assert!(stage.iter().all(|c| c.times.width() <= width), "half_join: a time wider than its scope");
+    let keys = concat_blocks(&stage.iter().map(|c| c.keys.clone()).collect::<Vec<_>>());
+    let vals = concat_blocks(&stage.iter().map(|c| c.vals.clone()).collect::<Vec<_>>());
+    let lanes = (0..width).map(|k| CValue::i64(stage.iter().flat_map(|c| c.times.lane(k)).map(|x| x as i64).collect())).collect();
+    let diffs: Vec<Diff> = stage.iter().flat_map(|c| c.diffs.iter().copied()).collect();
+    Some((CValue::Prod(vec![CValue::Prod(lanes), keys, vals]), diffs))
+}
+
+/// The time lanes of rows `(t, key, val)`, as integers.
+fn time_lanes(rows: &CValue) -> Vec<Cow<'_, [i64]>> {
+    let CValue::Prod(fields) = rows else { panic!("half_join: rows that are not (t, key, val)") };
+    let CValue::Prod(lanes) = &fields[0] else { panic!("half_join: a time that is not a tuple") };
+    lanes.iter().map(|lane| lane.as_i64("half_join time lane").unwrap()).collect()
+}
+
+/// Whether row `i`'s time, as `lanes`, is unblocked by `bound`: before it in the total order, or
+/// equal to it when `strict`.
+fn unblocked_at(lanes: &[Cow<'_, [i64]>], i: usize, bound: &[i64], strict: bool) -> bool {
+    match lanes.iter().zip(bound).map(|(lane, b)| lane[i].cmp(b)).find(|o| o.is_ne()).unwrap_or(Ordering::Equal) {
+        Ordering::Less => true,
+        Ordering::Equal => strict,
+        Ordering::Greater => false,
+    }
+}
+
+/// Rows `start..` of each part `(rows, start, diffs)`, in order, as one column and their diffs.
+fn concat_parts(parts: &[(&CValue, usize, &[Diff])]) -> (CValue, Vec<Diff>) {
+    if let [(rows, 0, diffs)] = parts { return ((*rows).clone(), diffs.to_vec()); }
+    let sources: Vec<Option<&CValue>> = parts.iter().map(|p| Some(p.0)).collect();
+    let (mut tags, mut offs, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
+    for (tag, (_, start, part)) in parts.iter().enumerate() {
+        tags.resize(tags.len() + (part.len() - start), tag);
+        offs.extend(*start..part.len());
+        diffs.extend_from_slice(&part[*start..]);
+    }
+    (gather_lanes(&sources, &tags, &offs), diffs)
+}
+
+/// The unreleased rows of `chunks`, in order, as one column and their diffs.
+fn live_rows(chunks: &[TimeChunk]) -> (CValue, Vec<Diff>) {
+    concat_parts(&chunks.iter().map(|c| (&c.rows, c.start, &c.diffs[..])).collect::<Vec<_>>())
+}
+
+/// Merges chains of [`TimeChunk`]s, consolidating equal rows.
+#[derive(Default)]
+pub struct TimeMerger;
+
+impl Merger for TimeMerger {
+    type Chunk = TimeChunk;
+
+    fn merge(&mut self, list1: Vec<TimeChunk>, list2: Vec<TimeChunk>, output: &mut Vec<TimeChunk>, _stash: &mut Vec<TimeChunk>) {
+        let (a, ad) = live_rows(&list1);
+        let (b, bd) = live_rows(&list2);
+        let (mut tags, mut offs, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
+        for run in survey_groups(&a, &b) {
+            match run {
+                GroupRun::A(lo, hi) => {
+                    tags.resize(tags.len() + (hi - lo), 0);
+                    offs.extend(lo..hi);
+                    diffs.extend_from_slice(&ad[lo..hi]);
+                }
+                GroupRun::B(lo, hi) => {
+                    tags.resize(tags.len() + (hi - lo), 1);
+                    offs.extend(lo..hi);
+                    diffs.extend_from_slice(&bd[lo..hi]);
+                }
+                GroupRun::Both(alo, ahi, blo, bhi) => {
+                    let d: Diff = ad[alo..ahi].iter().chain(&bd[blo..bhi]).sum();
+                    if d != 0 {
+                        tags.push(0);
+                        offs.push(alo);
+                        diffs.push(d);
+                    }
+                }
+            }
+        }
+        for lo in (0..diffs.len()).step_by(TIME_TARGET) {
+            let hi = (lo + TIME_TARGET).min(diffs.len());
+            output.push(TimeChunk::new(gather_lanes(&[Some(&a), Some(&b)], &tags[lo..hi], &offs[lo..hi]), diffs[lo..hi].to_vec()));
+        }
+    }
+
+    fn len(chunk: &TimeChunk) -> usize { chunk.len() }
+}
+
+/// Stages a lifted stream's updates and releases those the arrangement's frontier has unblocked.
+///
+/// An update at `t` is unblocked once no arrangement time that precedes it in the total order can
+/// still arrive: the frontier's least element `b` bounds every time yet to come, so `t ≤ b` suffices
+/// when only strictly earlier times meet `t`, and `t < b` when equal ones do too. An empty frontier
+/// unblocks everything.
+///
+/// Updates wait in a merge ladder of chains in time order, so what a frontier releases from each
+/// chain is a prefix: whole chunks from its front, and part of at most one more. Arrivals the
+/// frontier already unblocks skip the ladder. What one extraction releases is put in identifier
+/// order, ready to meet the arrangement in its order.
+pub struct HalfJoinBatcher {
+    strict: bool,
+    /// The number of lanes in a time: the scope depth, plus one.
+    width: usize,
+    /// Arrivals not yet sorted.
+    stage: Vec<CC>,
+    ladder: MergeLadder<TimeMerger>,
+    /// The minimal times among retained updates.
+    lower: Antichain<Time>,
+}
+
+impl HalfJoinBatcher {
+    pub fn new(strict: bool, width: usize) -> Self {
+        HalfJoinBatcher { strict, width, stage: Vec::new(), ladder: MergeLadder::new(None, 0), lower: Antichain::new() }
+    }
+}
+
+impl Batcher<CC> for HalfJoinBatcher {
+    type Time = Time;
+    type Output = Vec<Run>;
+
+    fn insert(&mut self, container: &mut CC) {
+        if !container.times.is_empty() {
+            self.stage.push(std::mem::take(container));
+        }
+    }
+
+    fn extract<'a>(&'a mut self, frontier: AntichainRef<'_, Time>) -> (Option<Vec<Run>>, AntichainRef<'a, Time>) {
+        let width = self.width;
+        let least = frontier.iter().min();
+        let bound: Option<Vec<i64>> = least.map(|b| {
+            let mut lanes: Vec<i64> = b.coordinates().map(|x| x as i64).collect();
+            assert!(lanes.len() <= width, "half_join: a frontier wider than its scope");
+            lanes.resize(width, 0);
+            lanes
+        });
+        let bound_col: Option<ColTimes<Time>> = least.map(|b| std::iter::once(b.clone()).collect());
+        let strict = self.strict;
+
+        // Arrivals the bound already unblocks are released without entering the ladder.
+        let (mut staged, mut blocked) = (Vec::new(), Vec::new());
+        for c in std::mem::take(&mut self.stage) {
+            let Some(b) = &bound_col else { staged.push(c); continue };
+            let (now, later): (Vec<usize>, Vec<usize>) = (0..c.times.len()).partition(|&i| match c.times.cmp_cross(i, b, 0) {
+                Ordering::Less => true,
+                Ordering::Equal => strict,
+                Ordering::Greater => false,
+            });
+            if later.is_empty() { staged.push(c); }
+            else if now.is_empty() { blocked.push(c); }
+            else {
+                staged.push(gather_rows(&c, &now));
+                blocked.push(gather_rows(&c, &later));
+            }
+        }
+        if let Some(chunk) = stage_rows(blocked, width).and_then(|(rows, diffs)| TimeChunk::form(rows, diffs)) {
+            self.ladder.insert_chain(vec![chunk]);
+        }
+
+        let mut released = Vec::new();
+        self.ladder.retain_chains(|chain| {
+            let whole = chain.iter().take_while(|c| c.unblocked(bound.as_deref(), strict) == c.diffs.len()).count();
+            released.extend(chain.drain(..whole));
+            if let Some(front) = chain.first_mut() {
+                let end = front.unblocked(bound.as_deref(), strict);
+                if end > front.start { released.push(front.release_to(end)); }
+            }
+        });
+        self.lower = Antichain::new();
+        for chunk in self.ladder.chains().flatten() {
+            for t in chunk.lower.elements() { self.lower.insert(t.clone()); }
+        }
+        let released = (!released.is_empty() || !staged.is_empty()).then(|| into_runs(released, staged, width));
+        (released, self.lower.borrow())
+    }
+}
+
+/// Released rows, from chunks and from arrivals released on arrival, as runs in identifier order.
+/// Each source is dropped once its runs are formed.
+fn into_runs(released: Vec<TimeChunk>, staged: Vec<CC>, width: usize) -> Vec<Run> {
+    let mut runs = Vec::new();
+    for chunk in released {
+        let CValue::Prod(fields) = &chunk.rows else { panic!("half_join: a time chunk that is not (t, key, val)") };
+        Part { keys: fields[1].clone(), vals: &fields[2], times: PartTimes::Lanes(chunk.lanes()), diffs: &chunk.diffs, rows: chunk.start..chunk.diffs.len() }.runs(RUN, width, &mut runs);
+    }
+    for c in staged {
+        Part { keys: c.keys.clone(), vals: &c.vals, times: PartTimes::Col(&c.times), diffs: &c.diffs, rows: 0..c.diffs.len() }.runs(RUN, width, &mut runs);
+    }
+    runs
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use super::super::test_util::*;
+    use crate::corgi::chunk::{ident, recover_key};
+    use crate::corgi::container::CorgiContainer;
+    use crate::ir::Value as V;
+
+    /// The batcher releases what the frontier's least element unblocks, and bounds what it keeps.
+    #[test]
+    fn releases_by_the_least_frontier_element() {
+        let released = |strict: bool, frontier: &[Time]| -> (Vec<Time>, Vec<Time>) {
+            let mut batcher = HalfJoinBatcher::new(strict, 2);
+            let times = [t(1, 0), t(0, 5), t(0, 2)];
+            batcher.insert(&mut container(&[1, 2, 3], &[0, 0, 0], &[(0, 0); 3], &times, &[1, 1, 1]));
+            let frontier: Antichain<Time> = frontier.iter().cloned().collect();
+            let (out, lower) = batcher.extract(frontier.borrow());
+            let mut out: Vec<Time> = out.unwrap_or_default().iter().flat_map(|c| c.times().to_vec()).collect();
+            out.sort();
+            let mut lower = lower.to_vec();
+            lower.sort();
+            (out, lower)
+        };
+        assert_eq!(released(true, &[t(1, 0)]), (vec![t(0, 2), t(0, 5), t(1, 0)], vec![]));
+        assert_eq!(released(false, &[t(1, 0)]), (vec![t(0, 2), t(0, 5)], vec![t(1, 0)]));
+        // (0, [5]) and (1, [0]) are incomparable, so both bound what is kept.
+        assert_eq!(released(false, &[t(0, 3), t(2, 0)]), (vec![t(0, 2)], vec![t(0, 5), t(1, 0)]));
+        assert_eq!(released(true, &[]), (vec![t(0, 2), t(0, 5), t(1, 0)], vec![]));
+    }
+
+    /// Against a model that holds each update until the frontier unblocks it: over rounds of random
+    /// arrivals, some already unblocked, and an advancing frontier, each extraction releases exactly
+    /// the unblocked updates, in identifier order, and reports a bound that covers every update it
+    /// keeps and holds only times it keeps.
+    #[test]
+    fn releases_what_the_model_releases() {
+        use std::collections::BTreeMap;
+        for strict in [false, true] {
+            for seed in 0..40 {
+                let mut rng = Rng(seed);
+                let mut batcher = HalfJoinBatcher::new(strict, 2);
+                let mut pending: BTreeMap<(Time, i64, i64), Diff> = BTreeMap::new();
+                for round in 0..14u64 {
+                    for _ in 0..rng.below(4) {
+                        let n = rng.below(9) as usize;
+                        let keys: Vec<i64> = (0..n).map(|_| rng.below(4) as i64).collect();
+                        let vals: Vec<i64> = (0..n).map(|_| rng.below(3) as i64).collect();
+                        let times: Vec<Time> = (0..n).map(|_| t(round / 3 + rng.below(3), rng.below(3))).collect();
+                        let diffs: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                        for i in 0..n { *pending.entry((times[i].clone(), keys[i], vals[i])).or_default() += diffs[i]; }
+                        batcher.insert(&mut container(&keys, &vals, &vec![(0, 0); n], &times, &diffs));
+                    }
+                    let frontier: Antichain<Time> = if round == 13 { Antichain::new() } else { [t(round / 3, round % 3), t(round / 3 + 1, 0)].into_iter().collect() };
+                    let bound = frontier.iter().min().cloned();
+                    let unblocked = |time: &Time| bound.as_ref().is_none_or(|b| time < b || (strict && time == b));
+
+                    let (out, lower) = batcher.extract(frontier.borrow());
+                    let lower = lower.to_owned();
+                    let mut got = Vec::new();
+                    for chunk in out.unwrap_or_default() {
+                        assert!(ident(chunk.keys()).windows(2).all(|w| w[0] <= w[1]), "released updates are in identifier order");
+                        let c = CorgiContainer { keys: recover_key(chunk.keys()), vals: chunk.vals().clone(), times: chunk.times().clone(), diffs: chunk.diffs().to_vec() };
+                        got.extend(c.into_updates());
+                    }
+                    let released: Vec<_> = pending.keys().filter(|(time, _, _)| unblocked(time)).cloned().collect();
+                    let mut expect = Vec::new();
+                    for entry in released {
+                        let d = pending.remove(&entry).unwrap();
+                        let (time, k, v) = entry;
+                        if d != 0 { expect.push(((V::Int(k), tup(vec![V::Int(v), tup(vec![V::Int(0), V::Int(0)])])), time, d)); }
+                    }
+                    assert_eq!(consolidated(got), consolidated(expect), "seed {seed}, round {round}, strict {strict}: the released updates");
+                    for ((time, _, _), d) in pending.iter() {
+                        assert!(*d == 0 || lower.less_equal(time), "seed {seed}, round {round}: the bound covers a kept update at {time:?}");
+                    }
+                    for b in lower.elements() {
+                        assert!(pending.keys().any(|(time, _, _)| time == b), "seed {seed}, round {round}: the bound holds {b:?}, which no kept update has");
+                    }
+                }
+                assert!(pending.values().all(|&d| d == 0), "an empty frontier releases everything");
+            }
+        }
+    }
+}
diff --git a/interactive/src/corgi/half_join/mod.rs b/interactive/src/corgi/half_join/mod.rs
new file mode 100644
index 000000000..bdc471739
--- /dev/null
+++ b/interactive/src/corgi/half_join/mod.rs
@@ -0,0 +1,157 @@
+//! The corgi half join: a [`Batcher`](differential_dataflow::batcher::Batcher) that releases a
+//! lifted stream's updates once the arrangement can no longer gain rows that precede them, and a
+//! [`HalfJoinTactic`](differential_dogs3::operators::half_join::HalfJoinTactic) that meets
+//! released updates with the arrangement's chunks, column at a time. Both plug into dogs3's
+//! `half_join_with_tactic`, which owns progress, capabilities, and yielding.
+//!
+//! An update is `(key ; (val, s))` at time `t`: `s` is a time as data, a `Prod` of `Int` lanes as
+//! `lift` writes it, and `t` the update's own time. It meets each arranged row `(key, v1)` at time
+//! `t2` with `t2 < t` in the total order on times (`≤` unless strict), giving
+//! `(key' ; (val', s ∨ t2))` at time `t` with the product of the diffs, where `(key', val')` is the
+//! projection over `(key, val, v1)`. The total order is the lexicographic order of the time lanes,
+//! which is DD's `Ord` on these times.
+//!
+//! The batcher keeps arrivals in time order and releases them as runs in identifier order; the
+//! tactic merges a unit's runs and meets them with the arrangement a block at a time. A [`Run`] is
+//! what passes between them.
+
+use std::borrow::Cow;
+use std::rc::Rc;
+
+use differential_dataflow::trace::chunk::ChunkBatch;
+
+use corgi::arrange::gather;
+use corgi::Value as CValue;
+
+use crate::corgi::chunk::{ident, present_key, CorgiChunk};
+use crate::corgi::col_times::ColTimes;
+use crate::corgi::container::CorgiContainer;
+use crate::ir::{Diff, Time};
+
+mod batcher;
+mod tactic;
+
+pub use batcher::HalfJoinBatcher;
+pub use tactic::CorgiHalfJoin;
+
+type CC = CorgiContainer<Time, Diff>;
+type NativeBatch = Rc<ChunkBatch<CorgiChunk<Time, Diff>>>;
+
+/// A run of released updates, as the batcher hands them to the tactic: in identifier order, as the
+/// arrangement's chunks are, with keys as [`present_key`] presents them. One extraction releases
+/// several runs, each sorted within a window of what was released, and the tactic merges them as
+/// it works. Runs are not consolidated: what one extraction releases from different chains may
+/// repeat a row.
+#[derive(Clone)]
+pub struct Run {
+    keys: CValue,
+    vals: CValue,
+    times: ColTimes<Time>,
+    diffs: Vec<Diff>,
+}
+
+impl Run {
+    /// Updates put in identifier order, as one run.
+    pub fn new(keys: CValue, vals: CValue, times: ColTimes<Time>, diffs: Vec<Diff>) -> Self {
+        let n = diffs.len();
+        let width = times.width();
+        let part = Part { keys, vals: &vals, times: PartTimes::Col(&times), diffs: &diffs, rows: 0..n };
+        let mut runs = Vec::new();
+        part.runs(n.max(1), width, &mut runs);
+        runs.pop().unwrap_or_else(|| Run { keys: present_key(part.keys.clone()), vals: vals.clone(), times: times.clone(), diffs: Vec::new() })
+    }
+    pub fn len(&self) -> usize { self.diffs.len() }
+    pub fn keys(&self) -> &CValue { &self.keys }
+    pub fn vals(&self) -> &CValue { &self.vals }
+    pub fn times(&self) -> &ColTimes<Time> { &self.times }
+    pub fn diffs(&self) -> &[Diff] { &self.diffs }
+}
+
+/// Rows `rows` of released columns, from a time chunk or from arrivals released on arrival.
+struct Part<'a> {
+    keys: CValue,
+    vals: &'a CValue,
+    times: PartTimes<'a>,
+    diffs: &'a [Diff],
+    rows: std::ops::Range<usize>,
+}
+
+impl Part<'_> {
+    /// The part's rows as runs of at most `size` rows, their times written as `width` lanes. Each
+    /// run is sorted and gathered from one window of the part, so its reads stay within it.
+    fn runs(&self, size: usize, width: usize, out: &mut Vec<Run>) {
+        let keys = present_key(self.keys.clone());
+        let ids = ident(&keys);
+        for lo in self.rows.clone().step_by(size) {
+            let hi = (lo + size).min(self.rows.end);
+            let mut order: Vec<(i64, usize)> = (lo..hi).map(|i| (ids[i], i)).collect();
+            order.sort_unstable_by_key(|&(id, _)| id);
+            let rows: Vec<usize> = order.into_iter().map(|(_, i)| i).collect();
+            let lanes = (0..width).map(|k| rows.iter().map(|&i| self.times.at(k, i)).collect()).collect();
+            out.push(Run {
+                keys: gather(&keys, &rows),
+                vals: gather(self.vals, &rows),
+                times: ColTimes::from_raw_lanes(lanes, rows.len()),
+                diffs: rows.iter().map(|&i| self.diffs[i]).collect(),
+            });
+        }
+    }
+}
+
+/// A part's times: a time chunk's lanes, or an arrival's column.
+enum PartTimes<'a> {
+    Lanes(Vec<Cow<'a, [i64]>>),
+    Col(&'a ColTimes<Time>),
+}
+
+impl PartTimes<'_> {
+    /// Coordinate `k` of row `i`, zero beyond the times' width.
+    fn at(&self, k: usize, i: usize) -> u64 {
+        match self {
+            PartTimes::Lanes(lanes) => lanes.get(k).map_or(0, |lane| lane[i] as u64),
+            PartTimes::Col(times) => times.raw_lanes().get(k).map_or(0, |lane| lane[i]),
+        }
+    }
+}
+/// The rows of `c` at `rows`, in that order.
+fn gather_rows(c: &CC, rows: &[usize]) -> CC {
+    CorgiContainer {
+        keys: gather(&c.keys, rows),
+        vals: gather(&c.vals, rows),
+        times: c.times.gather(rows),
+        diffs: rows.iter().map(|&i| c.diffs[i]).collect(),
+    }
+}
+
+#[cfg(test)]
+mod test_util {
+    use differential_dataflow::dynamic::pointstamp::PointStamp;
+    use timely::order::Product;
+
+    use super::*;
+    use crate::ir::Value as V;
+
+    /// The time `(outer, [inner])`, at scope depth one.
+    pub(super) fn t(outer: u64, inner: u64) -> Time { Product::new(outer, PointStamp::new(smallvec::smallvec![inner])) }
+    pub(super) fn ints(xs: &[i64]) -> CValue { CValue::i64(xs.to_vec()) }
+    pub(super) fn lifted(s: &[(i64, i64)]) -> CValue { CValue::Prod(vec![ints(&s.iter().map(|p| p.0).collect::<Vec<_>>()), ints(&s.iter().map(|p| p.1).collect::<Vec<_>>())]) }
+    pub(super) fn container(keys: &[i64], vals: &[i64], s: &[(i64, i64)], times: &[Time], diffs: &[Diff]) -> CC {
+        CorgiContainer { keys: ints(keys), vals: CValue::Prod(vec![ints(vals), lifted(s)]), times: times.iter().cloned().collect(), diffs: diffs.to_vec() }
+    }
+    pub(super) fn tup(xs: Vec<V>) -> V { V::Tuple(xs) }
+
+    /// A deterministic source of small numbers.
+    pub(super) struct Rng(pub(super) u64);
+    impl Rng {
+        pub(super) fn below(&mut self, n: u64) -> u64 {
+            self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+            (self.0 >> 33) % n
+        }
+        pub(super) fn diff(&mut self) -> Diff { if self.below(3) == 0 { -1 } else { 1 } }
+    }
+
+    pub(super) fn consolidated(mut rows: Vec<((V, V), Time, Diff)>) -> Vec<((V, V), Time, Diff)> {
+        differential_dataflow::consolidation::consolidate_updates(&mut rows);
+        rows
+    }
+}
diff --git a/interactive/src/corgi/half_join/tactic.rs b/interactive/src/corgi/half_join/tactic.rs
new file mode 100644
index 000000000..15bace195
--- /dev/null
+++ b/interactive/src/corgi/half_join/tactic.rs
@@ -0,0 +1,349 @@
+//! The half join's tactic: a unit's runs, merged by identifier a block at a time, meet the
+//! arrangement's chunks, each read forward from where the last block left it.
+
+use std::borrow::Cow;
+use std::cell::RefCell;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::rc::Rc;
+
+use timely::progress::Antichain;
+
+use differential_dataflow::trace::chunk::Chunk;
+use differential_dogs3::operators::half_join::HalfJoinTactic;
+
+use corgi::arrange::{compare_at, gather_lanes};
+use corgi::{Graph, NumOp, Shape, Value as CValue};
+
+use crate::corgi::chunk::{ident, key_is_hashed, recover_key_ref, CorgiChunk};
+use crate::corgi::col_times::ColTimes;
+use crate::corgi::container::CorgiContainer;
+use crate::corgi::logic::compile_join_projection;
+use crate::corgi::search::matching_ranges;
+use crate::ir::{Diff, Term, Time};
+
+use super::{gather_rows, NativeBatch, Run, CC};
+
+/// Released updates are met with the arrangement this many at a time, so that the driver can yield
+/// between blocks. Small under test, so that tests span many blocks.
+const BLOCK: usize = if cfg!(test) { 3 } else { 1 << 14 };
+
+/// Meets released updates with an arrangement's chunks. The projection compiles once, at the
+/// shapes of the first block that meets anything, and is shared by every unit of work.
+pub struct CorgiHalfJoin {
+    key: Term,
+    val: Term,
+    strict: bool,
+    /// The number of lanes in a time as data: the scope depth, plus one.
+    width: usize,
+    compiled: Rc<RefCell<Option<([Shape; 3], Graph<NumOp>)>>>,
+}
+
+impl CorgiHalfJoin {
+    pub fn new(key: Term, val: Term, strict: bool, width: usize) -> Self {
+        CorgiHalfJoin { key, val, strict, width, compiled: Rc::new(RefCell::new(None)) }
+    }
+}
+
+impl HalfJoinTactic<Time, NativeBatch, Run, CC> for CorgiHalfJoin {
+    fn prep(&mut self, runs: Vec<Run>, batches: Vec<NativeBatch>, lower: Antichain<Time>) -> Box<dyn Iterator<Item = (CC, Time)>> {
+        let arranged: Vec<CorgiChunk<Time, Diff>> = batches.iter().flat_map(|b| b.chunks.iter().filter(|c| c.len() > 0).cloned()).collect();
+        let lower: Vec<Time> = lower.elements().to_vec();
+        debug_assert!(runs.iter().all(|r| ident(r.keys()).windows(2).all(|w| w[0] <= w[1])), "half_join: a run out of identifier order");
+        Box::new(Work {
+            starts: vec![0; runs.len()],
+            runs,
+            seek: vec![0; arranged.len()],
+            arranged,
+            lower_times: lower.iter().cloned().collect(),
+            lower,
+            key: self.key.clone(),
+            val: self.val.clone(),
+            strict: self.strict,
+            width: self.width,
+            compiled: Rc::clone(&self.compiled),
+            ready: VecDeque::new(),
+        })
+    }
+}
+
+/// One unit of half-join work: released runs, merged and met a block at a time.
+///
+/// Runs and arranged chunks are all in identifier order. Each block takes the rows of every run
+/// below one identifier horizon, so the runs are read front to back, and the blocks meet each
+/// arranged chunk at positions that only move forward: `seek` records where each arranged chunk's
+/// next search starts, and each is walked at most once per unit.
+struct Work {
+    runs: Vec<Run>,
+    /// For each run, the first row not yet met.
+    starts: Vec<usize>,
+    arranged: Vec<CorgiChunk<Time, Diff>>,
+    /// For each arranged chunk, the first row later blocks could meet.
+    seek: Vec<usize>,
+    /// The capabilities output may ship at, and the same as a column.
+    lower: Vec<Time>,
+    lower_times: ColTimes<Time>,
+    key: Term,
+    val: Term,
+    strict: bool,
+    width: usize,
+    compiled: Rc<RefCell<Option<([Shape; 3], Graph<NumOp>)>>>,
+    /// Output met but not yet handed over.
+    ready: VecDeque<(CC, Time)>,
+}
+
+impl Iterator for Work {
+    type Item = (CC, Time);
+
+    fn next(&mut self) -> Option<(CC, Time)> {
+        while self.ready.is_empty() {
+            let live = self.runs.iter().zip(&self.starts).filter(|(r, &s)| s < r.len()).count();
+            if live == 0 { return None; }
+            let ends = self.block_ends((BLOCK / live).max(1));
+            self.meet(&ends);
+            self.starts = ends;
+        }
+        self.ready.pop_front()
+    }
+}
+
+impl Work {
+    /// Where each run's next block ends: below the least identifier `budget` rows into a run with
+    /// more than that left, past that identifier's whole run of rows, so that no identifier is
+    /// split between blocks. Every run with no more than `budget` rows left runs to its end.
+    fn block_ends(&self, budget: usize) -> Vec<usize> {
+        let mut horizon: Option<i64> = None;
+        for (run, &s) in self.runs.iter().zip(&self.starts) {
+            let ids = ident(run.keys());
+            if ids.len() - s <= budget { continue; }
+            // `None` on overflow means nothing can exceed it: run to the end.
+            let Some(bound) = ids[s + budget].checked_add(1) else { continue };
+            horizon = Some(horizon.map_or(bound, |h| h.min(bound)));
+        }
+        self.runs.iter().map(|run| match horizon {
+            Some(h) => ident(run.keys()).partition_point(|&x| x < h),
+            None => run.len(),
+        }).collect()
+    }
+
+    /// Meet each run's rows from its start to `ends` with the arrangement, queueing the output by
+    /// capability.
+    fn meet(&mut self, ends: &[usize]) {
+        // The block's rows as (identifier, run, row), in identifier order; its distinct identifiers,
+        // ascending, and the rows holding each.
+        let mut rows: Vec<(i64, usize, usize)> = Vec::new();
+        for (r, (run, (&lo, &hi))) in self.runs.iter().zip(self.starts.iter().zip(ends)).enumerate() {
+            let ids = ident(run.keys());
+            rows.extend((lo..hi).map(|u| (ids[u], r, u)));
+        }
+        rows.sort_unstable_by_key(|&(id, _, _)| id);
+        let (mut needles, mut groups) = (Vec::new(), Vec::new());
+        for (pos, &(id, _, _)) in rows.iter().enumerate() {
+            if needles.last() != Some(&id) {
+                needles.push(id);
+                groups.push(pos..pos);
+            }
+            groups.last_mut().unwrap().end = pos + 1;
+        }
+        let (Some(&first), Some(&last)) = (needles.first(), needles.last()) else { return };
+        let hashed = key_is_hashed(self.runs[rows[0].1].keys());
+        let keys: Vec<&CValue> = self.runs.iter().map(|run| recover_key_ref(run.keys())).collect();
+
+        // The pairs that meet: (run, update row, arranged chunk, arranged row).
+        let (mut rs, mut us, mut cs, mut js) = (Vec::new(), Vec::new(), Vec::new(), Vec::new());
+        let (mut scratch, mut found) = (Vec::new(), Vec::new());
+        for (cid, chunk) in self.arranged.iter().enumerate() {
+            let hay = &ident(chunk.keys())[self.seek[cid]..];
+            let base = self.seek[cid];
+            // Later blocks hold identifiers at least `last`, so they need nothing before it.
+            self.seek[cid] += hay.partition_point(|&x| x < last);
+            if hay.first().is_none_or(|&x| x > last) || hay.last().is_some_and(|&x| x < first) { continue; }
+            matching_ranges(&needles, hay, &mut scratch, &mut found);
+            let chunk_key = recover_key_ref(chunk.keys());
+            for (needle, range) in found.iter() {
+                for &(_, r, u) in &rows[groups[*needle].clone()] {
+                    for j in range.start + base..range.end + base {
+                        if hashed && compare_at(keys[r], u, chunk_key, j).is_ne() { continue; }
+                        let precedes = match chunk.times().cmp_cross(j, self.runs[r].times(), u) {
+                            Ordering::Less => true,
+                            Ordering::Equal => !self.strict,
+                            Ordering::Greater => false,
+                        };
+                        if precedes {
+                            rs.push(r);
+                            us.push(u);
+                            cs.push(cid);
+                            js.push(j);
+                        }
+                    }
+                }
+            }
+        }
+        if us.is_empty() { return; }
+
+        // The projection over (key, val, arranged val).
+        let fields: Vec<&[CValue]> = self.runs.iter().map(|run| match run.vals() {
+            CValue::Prod(fields) => &fields[..],
+            _ => panic!("half_join: a stream value that is not (val, time)"),
+        }).collect();
+        let keys = gather_lanes(&keys.iter().map(|&k| Some(k)).collect::<Vec<_>>(), &rs, &us);
+        let vals = gather_lanes(&fields.iter().map(|f| Some(&f[0])).collect::<Vec<_>>(), &rs, &us);
+        let sources: Vec<Option<&CValue>> = self.arranged.iter().map(|c| Some(c.vals())).collect();
+        let arranged_vals = gather_lanes(&sources, &cs, &js);
+        let shapes = [corgi::shape_of_value(&keys), corgi::shape_of_value(&vals), corgi::shape_of_value(&arranged_vals)];
+        let mut compiled = self.compiled.borrow_mut();
+        if compiled.is_none() {
+            let g = compile_join_projection(&self.key, &self.val, &shapes[0], &shapes[1], &shapes[2])
+                .unwrap_or_else(|e| panic!("half_join: type error at shapes ({}, {}, {}): {e}", shapes[0], shapes[1], shapes[2]));
+            *compiled = Some((shapes.clone(), g));
+        }
+        let (pinned, graph) = compiled.as_ref().unwrap();
+        assert!(*pinned == shapes, "half_join: a block of shape {shapes:?} reached an operator pinned at {pinned:?}");
+        let mut out = corgi::eval_graph(graph, CValue::Prod(vec![keys, vals, arranged_vals])).into_prod("half_join projection").unwrap();
+        let out_val = out.pop().unwrap();
+        let out_key = out.pop().unwrap();
+
+        // The update's time as data, joined lane by lane with the arranged row's time.
+        let data_times: Vec<Vec<Cow<'_, [i64]>>> = fields.iter().map(|f| {
+            let CValue::Prod(lanes) = &f[1] else { panic!("half_join: a time as data that is not a tuple") };
+            assert_eq!(lanes.len(), self.width, "half_join: a time as data of the wrong width");
+            lanes.iter().map(|lane| lane.as_i64("half_join time lane").unwrap()).collect()
+        }).collect();
+        let joined: Vec<CValue> = (0..self.width).map(|l| {
+            CValue::i64(rs.iter().zip(&us).zip(cs.iter().zip(&js)).map(|((&r, &u), (&c, &j))| {
+                let t2 = self.arranged[c].times().raw_lanes().get(l).map_or(0, |lane| lane[j]) as i64;
+                data_times[r][l][u].max(t2)
+            }).collect())
+        }).collect();
+        let mut times = ColTimes::new();
+        for (&r, &u) in rs.iter().zip(&us) { times.push_ref(self.runs[r].times(), u); }
+        let met = CorgiContainer {
+            keys: out_key,
+            vals: CValue::Prod(vec![out_val, CValue::Prod(joined)]),
+            times,
+            diffs: rs.iter().zip(&us).zip(cs.iter().zip(&js)).map(|((&r, &u), (&c, &j))| self.runs[r].diffs()[u] * self.arranged[c].diffs()[j]).collect(),
+        };
+
+        // Each row ships at the first capability at or before its own time, as the driver requires.
+        let mut by_cap: Vec<Vec<usize>> = vec![Vec::new(); self.lower.len()];
+        for row in 0..met.times.len() {
+            let cap = (0..self.lower.len()).find(|&c| self.lower_times.less_equal_cross(c, &met.times, row))
+                .expect("half_join: no capability covers a released update");
+            by_cap[cap].push(row);
+        }
+        for (cap, rows) in by_cap.into_iter().enumerate() {
+            if !rows.is_empty() {
+                self.ready.push_back((gather_rows(&met, &rows), self.lower[cap].clone()));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use super::super::test_util::*;
+    use crate::corgi::chunk::{columns_to_batch, present_key};
+    use crate::ir::Value as V;
+
+    /// An update meets the arranged rows before it in the total order on times, strictly or not:
+    /// never a later one, even one beyond it in a later coordinate alone. Its data time joins the
+    /// arranged row's, coordinate by coordinate, and each output ships at the capability covering it.
+    #[test]
+    fn meets_what_precedes_in_the_total_order() {
+        let arranged = Rc::new(columns_to_batch(
+            present_key(ints(&[1, 1, 1, 1, 1])),
+            ints(&[50, 100, 200, 300, 400]),
+            [t(0, 1), t(0, 5), t(1, 0), t(2, 0), t(1, 2)].into_iter().collect(),
+            vec![1, 1, 1, 1, 1],
+        ));
+        // Updates of key 1 at (1, [0]) and at (0, [3]), each with its own time as data.
+        let updates = container(&[1, 1], &[7, 8], &[(1, 0), (0, 3)], &[t(1, 0), t(0, 3)], &[2, 1]);
+        let updates = Run::new(updates.keys, updates.vals, updates.times, updates.diffs);
+        let lower: Antichain<Time> = [t(0, 3), t(1, 0)].into_iter().collect();
+        let projection = (Term::Var(0), Term::Tuple(vec![Term::Var(1), Term::Var(2)]));
+        let rows = |strict: bool| -> Vec<(Time, ((V, V), Time, Diff))> {
+            let mut tactic = CorgiHalfJoin::new(projection.0.clone(), projection.1.clone(), strict, 2);
+            let mut rows: Vec<_> = tactic.prep(vec![updates.clone()], vec![Rc::clone(&arranged)], lower.clone())
+                .flat_map(|(c, cap)| c.into_updates().into_iter().map(move |u| (cap.clone(), u)))
+                .collect();
+            rows.sort();
+            rows
+        };
+        let row = |cap: Time, x: i64, y: i64, data: (i64, i64), time: Time, diff: Diff| {
+            (cap, ((V::Int(1), tup(vec![tup(vec![V::Int(x), V::Int(y)]), tup(vec![V::Int(data.0), V::Int(data.1)])])), time, diff))
+        };
+        let mut strict = vec![
+            row(t(1, 0), 7, 50, (1, 1), t(1, 0), 2),
+            row(t(1, 0), 7, 100, (1, 5), t(1, 0), 2),
+            row(t(0, 3), 8, 50, (0, 3), t(0, 3), 1),
+        ];
+        strict.sort();
+        assert_eq!(rows(true), strict);
+        // Not strict, the update at (1, [0]) also meets the arranged row at (1, [0]).
+        let mut loose = strict.clone();
+        loose.push(row(t(1, 0), 7, 200, (1, 0), t(1, 0), 2));
+        loose.sort();
+        assert_eq!(rows(false), loose);
+    }
+
+    /// Against the definition, over several arranged chunks and released updates in several runs
+    /// spanning many blocks, with keys that are integers and keys that are hashed: each update
+    /// meets exactly the arranged rows of its key that precede it, and ships at a capability at or
+    /// before its time.
+    #[test]
+    fn meets_what_the_definition_meets() {
+        for hashed in [false, true] {
+            for strict in [false, true] {
+                for seed in 0..40 {
+                    let mut rng = Rng(seed);
+                    let key_col = |ks: &[i64]| if hashed { CValue::Prod(vec![ints(ks), ints(&ks.iter().map(|k| k * 7).collect::<Vec<_>>())]) } else { ints(ks) };
+                    let key_val = |k: i64| if hashed { tup(vec![V::Int(k), V::Int(k * 7)]) } else { V::Int(k) };
+                    let mut arranged_rows = Vec::new();
+                    let batches: Vec<NativeBatch> = (0..3).map(|_| {
+                        let n = 1 + rng.below(20) as usize;
+                        let ks: Vec<i64> = (0..n).map(|_| rng.below(6) as i64).collect();
+                        let vs: Vec<i64> = (0..n).map(|_| rng.below(100) as i64).collect();
+                        let ts: Vec<Time> = (0..n).map(|_| t(rng.below(4), rng.below(3))).collect();
+                        let ds: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                        for i in 0..n { arranged_rows.push((ks[i], vs[i], ts[i].clone(), ds[i])); }
+                        Rc::new(columns_to_batch(present_key(key_col(&ks)), ints(&vs), ts.into_iter().collect(), ds))
+                    }).collect();
+                    let n = 1 + rng.below(40) as usize;
+                    let ks: Vec<i64> = (0..n).map(|_| rng.below(8) as i64).collect();
+                    let vs: Vec<i64> = (0..n).map(|_| rng.below(10) as i64).collect();
+                    let ss: Vec<(i64, i64)> = (0..n).map(|_| (rng.below(4) as i64, rng.below(3) as i64)).collect();
+                    let ts: Vec<Time> = (0..n).map(|_| t(rng.below(4), rng.below(3))).collect();
+                    let ds: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                    // The updates as several runs, of random lengths, as one extraction releases them.
+                    let mut runs = Vec::new();
+                    let mut lo = 0;
+                    while lo < n {
+                        let hi = (lo + 1 + rng.below(6) as usize).min(n);
+                        runs.push(Run::new(key_col(&ks[lo..hi]), CValue::Prod(vec![ints(&vs[lo..hi]), lifted(&ss[lo..hi])]), ts[lo..hi].iter().cloned().collect(), ds[lo..hi].to_vec()));
+                        lo = hi;
+                    }
+                    let lower: Antichain<Time> = ts.iter().cloned().collect();
+
+                    let mut expect = Vec::new();
+                    for i in 0..n {
+                        for (k2, v1, t2, d2) in arranged_rows.iter() {
+                            if *k2 != ks[i] || !(*t2 < ts[i] || (!strict && *t2 == ts[i])) { continue; }
+                            let s = tup(vec![V::Int(ss[i].0.max(t2.outer as i64)), V::Int(ss[i].1.max(t2.inner.first().copied().unwrap_or(0) as i64))]);
+                            expect.push(((key_val(ks[i]), tup(vec![tup(vec![V::Int(vs[i]), V::Int(*v1)]), s])), ts[i].clone(), ds[i] * d2));
+                        }
+                    }
+                    let mut tactic = CorgiHalfJoin::new(Term::Var(0), Term::Tuple(vec![Term::Var(1), Term::Var(2)]), strict, 2);
+                    let mut got = Vec::new();
+                    for (c, cap) in tactic.prep(runs, batches, lower) {
+                        for row in c.into_updates() {
+                            assert!(timely::PartialOrder::less_equal(&cap, &row.1), "a row ships at a capability at or before its time");
+                            got.push(row);
+                        }
+                    }
+                    assert_eq!(consolidated(got), consolidated(expect), "seed {seed}, strict {strict}, hashed {hashed}");
+                }
+            }
+        }
+    }
+}
diff --git a/interactive/src/corgi/mod.rs b/interactive/src/corgi/mod.rs
index 8f862e4ed..2d773fbbd 100644
--- a/interactive/src/corgi/mod.rs
+++ b/interactive/src/corgi/mod.rs
@@ -1,8 +1,8 @@
 //! The corgi rendering substrate's internals: the columnar container on dataflow
 //! edges ([`container`]) and its wire format ([`bytes`]), the columnar chunk and
 //! its arrangement plumbing ([`chunk`]), the key-hash shuffle ([`exchange`]),
-//! the DDIR-term compiler ([`logic`]), and the join/reduce tactic bindings ([`join`],
-//! [`reduce`]). The `Backend` impl wiring these into rendering is
+//! the DDIR-term compiler ([`logic`]), and the join/half-join/reduce tactic bindings ([`join`],
+//! [`half_join`], [`reduce`]). The `Backend` impl wiring these into rendering is
 //! [`crate::backend::corgi`].
 
 pub mod bytes;
@@ -10,6 +10,7 @@ pub mod chunk;
 pub mod col_times;
 pub mod container;
 pub mod exchange;
+pub mod half_join;
 pub mod join;
 pub mod logic;
 pub mod reduce;
diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs
index 0bb2e56ac..5eb9e888f 100644
--- a/interactive/src/explain/mod.rs
+++ b/interactive/src/explain/mod.rs
@@ -77,6 +77,7 @@ fn clone_node(node: &Node, m: impl Fn(&Ref) -> Ref) -> Node {
         Node::Concat(refs) => Node::Concat(refs.iter().map(&m).collect()),
         Node::Arrange(r) => Node::Arrange(m(r)),
         Node::Join { left, right, projection } => Node::Join { left: m(left), right: m(right), projection: projection.clone() },
+        Node::HalfJoin { stream, arrangement, strict, projection } => Node::HalfJoin { stream: m(stream), arrangement: m(arrangement), strict: *strict, projection: projection.clone() },
         Node::Reduce { input, reducer } => Node::Reduce { input: m(input), reducer: reducer.clone() },
         Node::Inspect { input, label } => Node::Inspect { input: m(input), label: label.clone() },
     }
@@ -914,6 +915,8 @@ impl<'a> Reverse<'a> {
                 let contrib = ex.emit_lookup_keyed(dep_this, &side, out_user_len, reducer);
                 self.contribs.entry(target).or_default().push(contrib);
             }
+            // It reads its stream's whole times, as `lift` does.
+            Node::HalfJoin { .. } => panic!("explain: half_join is not supported"),
             Node::Join { left, right, projection } => {
                 let lt = resolve(self.orig, path, left);
                 let rt = resolve(self.orig, path, right);
diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs
index 4ce009fbc..e9d295bfd 100644
--- a/interactive/src/lower.rs
+++ b/interactive/src/lower.rs
@@ -51,7 +51,7 @@ fn expr_free_names<'a>(expr: &'a Expr, out: &mut BTreeSet<&'a str>) {
             | Expr::Negate(e) | Expr::EnterAt(e, _) | Expr::LiftIter(e)
             | Expr::FlatMap(e, _) | Expr::Lift(e) | Expr::JoinFunction(e, _)
             | Expr::Inspect(e, _) | Expr::Arrange(e) => expr_free_names(e, out),
-        Expr::Join(l, r, _) => { expr_free_names(l, out); expr_free_names(r, out); },
+        Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { expr_free_names(l, out); expr_free_names(r, out); },
         Expr::Concat(es) => { for e in es { expr_free_names(e, out); } },
     }
 }
@@ -168,6 +168,12 @@ impl ScopeLower {
                 let rr = self.lower_expr(r); let ra = self.push(st::Node::Arrange(rr));
                 self.push(st::Node::Join { left: la, right: ra, projection: p.clone() })
             },
+            // A half join reads its stream as updates and only its second input arranged.
+            Expr::HalfJoin(s, r, strict, p) => {
+                let stream = self.lower_expr(s);
+                let rr = self.lower_expr(r); let ra = self.push(st::Node::Arrange(rr));
+                self.push(st::Node::HalfJoin { stream, arrangement: ra, strict: *strict, projection: p.clone() })
+            },
             Expr::Reduce(e, red)=> {
                 let r = self.lower_expr(e); let a = self.push(st::Node::Arrange(r));
                 self.push(st::Node::Reduce { input: a, reducer: red.clone() })
@@ -316,7 +322,7 @@ fn collect_qualified(e: &Expr, scope: &str, out: &mut Vec<String>) {
         Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::EnterAt(e, _)
         | Expr::FlatMap(e, _) | Expr::Lift(e) | Expr::JoinFunction(e, _)
         | Expr::LiftIter(e) | Expr::Reduce(e, _) | Expr::Inspect(e, _) | Expr::Arrange(e) => collect_qualified(e, scope, out),
-        Expr::Join(l, r, _) => { collect_qualified(l, scope, out); collect_qualified(r, scope, out); },
+        Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { collect_qualified(l, scope, out); collect_qualified(r, scope, out); },
         Expr::Concat(es) => for e in es { collect_qualified(e, scope, out); },
         Expr::Input(_) | Expr::Import(_) | Expr::Name(_) => {}
     }
diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs
index 7b7332c75..ddd60b357 100644
--- a/interactive/src/parse/mod.rs
+++ b/interactive/src/parse/mod.rs
@@ -25,6 +25,8 @@ pub enum Expr {
     Qualified(String, String),
     Map(Box<Expr>, Projection),
     Join(Box<Expr>, Box<Expr>, Projection),
+    /// `stream | half_join(arrangement, strict, projection)`: see `scope_ir::Node::HalfJoin`.
+    HalfJoin(Box<Expr>, Box<Expr>, bool, Projection),
     Reduce(Box<Expr>, Reducer),
     Filter(Box<Expr>, Term),
     Negate(Box<Expr>),
@@ -32,8 +34,7 @@ pub enum Expr {
     /// Append the current user-iter coord (at the operator's scope depth)
     /// to each row's value. Time itself is unchanged.
     ///
-    /// Discipline (post-lowering check, see `lower::validate_lift_iter`):
-    /// the result of `LiftIter` must not be referenced inside the same
+    /// Discipline, which nothing checks: the result of `LiftIter` should not be referenced inside the same
     /// scope it appears in — only from an enclosing scope, after the
     /// implicit leave. This preserves the "loop body is a time-invariant
     /// function" property; in-scope use risks defeating the fixpoint.
diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs
index 4af9fb6cf..cf69d0792 100644
--- a/interactive/src/parse/pipe.rs
+++ b/interactive/src/parse/pipe.rs
@@ -36,6 +36,10 @@
 //!   of the epoch and every scope coordinate.
 //! - `| join_function(term)` — replace each row by the `term`'s list of
 //!   `((key, val), time, diff)`, at the row's time joined with `time`, diffs multiplied.
+//! - `| half_join(other, strict, (k, v, v1) -> (k', v'))` — each update of a lifted stream,
+//!   `(k ; (v, time))`, meets `other`'s rows at its key whose times precede the update's own in the
+//!   total order on times (strictly when `strict` is 1, or also equal when 0), giving
+//!   `(k' ; (v', time ∨ t2))`. The lambda does not see the times. `other`'s changes produce nothing.
 //! - `| delay(term)` — `join_function` with the list `[((key, val), term, 1)]`: advance each
 //!   row to at least the time `term`.
 //! - `a + b` (concat), `a - b` (concat with `b` negated).
@@ -621,6 +625,19 @@ impl Parser {
             Token::EnterAt => { self.next(); self.expect(&Token::LParen); let f = self.parse_slot_term(); self.expect(&Token::RParen); Expr::EnterAt(Box::new(lhs), f) },
             Token::LiftIter => { self.next(); Expr::LiftIter(Box::new(lhs)) },
             Token::Ident(name) if name == "lift" => { self.next(); Expr::Lift(Box::new(lhs)) },
+            Token::Ident(name) if name == "half_join" => {
+                self.next(); self.expect(&Token::LParen);
+                let r = self.parse_join_arg(); self.expect(&Token::Comma);
+                let strict = match self.next() {
+                    Token::Int(0) => false,
+                    Token::Int(1) => true,
+                    other => panic!("half_join: strict is 0 or 1, not {other:?}"),
+                };
+                self.expect(&Token::Comma);
+                let p = match self.lambda(3) { Some(l) => Self::split(l), None => self.parse_projection() };
+                self.expect(&Token::RParen);
+                Expr::HalfJoin(Box::new(lhs), Box::new(r), strict, p)
+            },
             Token::Ident(name) if name == "join_function" => {
                 self.next(); self.expect(&Token::LParen);
                 let t = self.parse_slot_term(); self.expect(&Token::RParen);
diff --git a/interactive/src/scope_ir.rs b/interactive/src/scope_ir.rs
index 138393117..44e9f27d0 100644
--- a/interactive/src/scope_ir.rs
+++ b/interactive/src/scope_ir.rs
@@ -117,6 +117,13 @@ pub enum Node {
     Concat(Vec<Ref>),
     Arrange(Ref),
     Join { left: Ref, right: Ref, projection: Projection },
+    /// Each update of `stream`, `(key ; (val, time))` with `time` a time as data (as `lift` writes
+    /// it), meets the rows of `arrangement` at its key whose times are less than the update's own
+    /// in the total order on times, or less or equal unless `strict`. Each pair gives
+    /// `(key' ; (val', time ∨ t2))`: `projection` over `(key, val, arranged val)`, and the update's
+    /// data time joined with the arranged row's. The output keeps the update's own time, and its
+    /// diff is the product of the two. The arrangement's changes produce no output.
+    HalfJoin { stream: Ref, arrangement: Ref, strict: bool, projection: Projection },
     Reduce { input: Ref, reducer: Reducer },
     Inspect { input: Ref, label: String },
 }
@@ -208,6 +215,7 @@ impl Scope {
                     Node::Linear { input, .. } | Node::Arrange(input)
                     | Node::Reduce { input, .. } | Node::Inspect { input, .. } => f(input),
                     Node::Join { left, right, .. } => { f(left); f(right); },
+                    Node::HalfJoin { stream, arrangement, .. } => { f(stream); f(arrangement); },
                     Node::Concat(refs) => for r in refs { f(r); },
                 },
                 Item::Sub(child) => for imp in child.imports.iter_mut() {
@@ -239,6 +247,7 @@ impl Scope {
                     Node::Linear { input, .. } | Node::Arrange(input)
                     | Node::Reduce { input, .. } | Node::Inspect { input, .. } => tally(input),
                     Node::Join { left, right, .. } => { tally(left); tally(right); },
+                    Node::HalfJoin { stream, arrangement, .. } => { tally(stream); tally(arrangement); },
                     Node::Concat(refs) => for r in refs { tally(r); },
                 },
                 Item::Sub(child) => for imp in &child.imports {
@@ -387,6 +396,7 @@ fn dump_scope_body(s: &Scope, indent: usize) {
                     Node::Concat(refs) => refs.iter().map(fmt_ref).collect::<Vec<_>>().join(" + "),
                     Node::Arrange(r) => format!("{} | arrange", fmt_ref(r)),
                     Node::Join { left, right, .. } => format!("join({}, {})", fmt_ref(left), fmt_ref(right)),
+                    Node::HalfJoin { stream, arrangement, strict, .. } => format!("{} | half_join({}, {})", fmt_ref(stream), fmt_ref(arrangement), *strict as u8),
                     Node::Reduce { input, reducer } => format!("{} | {:?}", fmt_ref(input), reducer),
                     Node::Inspect { input, label } => format!("{} | inspect({})", fmt_ref(input), label),
                 };
diff --git a/interactive/src/shapes.rs b/interactive/src/shapes.rs
index b30dc310e..52f360eda 100644
--- a/interactive/src/shapes.rs
+++ b/interactive/src/shapes.rs
@@ -246,6 +246,18 @@ fn infer_node(node: &st::Node, shapes: &ScopeShapes, depth: usize, at: &str, pro
             let v = match shape_of_term(&projection.val, &env, None) { Ok(s) => s, Err(e) => return fail(format!("join val: {e}")) };
             Some(CollShape::new(k, v, depth))
         }
+        st::Node::HalfJoin { stream, arrangement, projection, .. } => {
+            let (s, a) = (shapes.of(stream)?, shapes.of(arrangement)?);
+            if s.key != a.key { return fail(format!("half_join keys {} and {}", s.key, a.key)); }
+            let v = match s.val {
+                Shape::Prod(fs) if fs.len() == 2 && fs[1] == time_shape(depth) => fs[0].clone(),
+                other => return fail(format!("half_join: a stream value of shape {other}, not (val, {}) as `lift` makes it", time_shape(depth))),
+            };
+            let env = [s.key, v, a.val];
+            let k = match shape_of_term(&projection.key, &env, None) { Ok(s) => s, Err(e) => return fail(format!("half_join key: {e}")) };
+            let v = match shape_of_term(&projection.val, &env, None) { Ok(s) => s, Err(e) => return fail(format!("half_join val: {e}")) };
+            Some(CollShape::new(k, Shape::Prod(vec![v, time_shape(depth)]), depth))
+        }
         st::Node::Reduce { input, reducer } => {
             let c = shapes.of(input)?;
             let v = match reducer {
diff --git a/interactive/tests/half_join.rs b/interactive/tests/half_join.rs
new file mode 100644
index 000000000..b181aa10e
--- /dev/null
+++ b/interactive/tests/half_join.rs
@@ -0,0 +1,89 @@
+//! `half_join`: a delta join, two half joins whose results are delayed to their data times, is the
+//! join, at the root and inside an iterative scope, as inputs change over several epochs.
+
+use interactive::ir::Value;
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+const PROGRAM: &str = r#"
+    let a = input 0 : ((int, int) ; ()) | key(((k, x), ()) -> (k, x));
+    let b = input 1 : ((int, int) ; ()) | key(((k, y), ()) -> (k, y));
+
+    -- Each of `a`'s updates meets `b`'s at or before it, and each of `b`'s meets `a`'s strictly
+    -- before it, so every pair meets once, at the later of its two times.
+    let ab = a | lift | half_join(b, 0, (k, x, y) -> (k, (x, y)));
+    let ba = b | lift | half_join(a, 1, (k, y, x) -> (k, (x, y)));
+    export "delta" = (ab + ba) | delay((_, (_, t)) -> t) | key((k, (xy, _)) -> (k, xy));
+    export "join" = a | join(b, (k, x, y) -> (k, (x, y)));
+
+    -- Reachability from `a`'s keys along `b`'s edges, joining by each method.
+    let roots = a | key((k, _) -> (k, ()));
+    joined: {
+        let step = reach | join(b, (_, (), dst) -> (dst, ()));
+        var reach = roots + step | distinct;
+    }
+    halved: {
+        let fwd = reach | lift | half_join(b, 0, (_, (), dst) -> (dst, ()));
+        let rev = b | lift | half_join(reach, 1, (_, dst, ()) -> (dst, ()));
+        let step = (fwd + rev) | delay((_, (_, t)) -> t) | key((n, _) -> (n, ()));
+        var reach = roots + step | distinct;
+    }
+    export "reach_join" = joined::reach;
+    export "reach_delta" = halved::reach;
+"#;
+
+const TICKS: u64 = 24;
+
+/// The rows fed at each tick, as `(input, key, value, diff)`: a few pseudo-random rows of small
+/// keys and values each tick, each added and, a few ticks later, retracted. Enough ticks that the
+/// arrangements' batches merge, and so compact.
+fn feeds(tick: u64) -> Vec<(usize, i64, i64, i64)> {
+    let row = |n: u64| {
+        let h = n.wrapping_mul(0x9E37_79B9_7F4A_7C15).rotate_left(17).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+        ((h % 2) as usize, ((h >> 8) % 6) as i64, ((h >> 16) % 6) as i64)
+    };
+    let mut rows = Vec::new();
+    for i in 0..4 {
+        let (input, k, v) = row(4 * tick + i);
+        rows.push((input, k, v, 1));
+        if tick >= 3 {
+            let (input, k, v) = row(4 * (tick - 3) + i);
+            rows.push((input, k, v, -1));
+        }
+    }
+    rows
+}
+
+/// Every export after each tick, by name.
+fn run(workers: usize) -> Vec<[Vec<(Value, Value, i64)>; 4]> {
+    let guards = timely::execute(timely::Config::process(workers), move |worker| {
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        let mut program = lower::lower_tree(parse::pipe::parse(PROGRAM));
+        program.optimize();
+        server.install(worker, "p", &program).unwrap();
+        let mut seen = Vec::new();
+        for tick in 0..TICKS {
+            if worker.index() == 0 {
+                for (input, k, v, d) in feeds(tick) {
+                    server.feed("p", input, Value::Tuple(vec![Value::Int(k), Value::Int(v)]), Value::unit(), None, d).unwrap();
+                }
+            }
+            server.tick(worker);
+            seen.push(["delta", "join", "reach_delta", "reach_join"].map(|name| server.snapshot(worker, name).unwrap()));
+        }
+        seen
+    })
+    .unwrap();
+    guards.join().into_iter().next().unwrap().unwrap()
+}
+
+#[test]
+fn delta_joins_are_joins() {
+    for workers in [1, 3] {
+        for (tick, [delta, join, reach_delta, reach_join]) in run(workers).into_iter().enumerate() {
+            assert!(!join.is_empty() && !reach_join.is_empty(), "tick {tick}: the joins have rows");
+            assert_eq!(delta, join, "tick {tick}, {workers} workers: the delta join");
+            assert_eq!(reach_delta, reach_join, "tick {tick}, {workers} workers: reachability");
+        }
+    }
+}