diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml
index adb8ea20a..c598919ff 100644
--- a/interactive/Cargo.toml
+++ b/interactive/Cargo.toml
@@ -16,7 +16,6 @@ columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
 corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "de0f2ac91d31ac63641035e5a5bb1b5640fe9424", features = ["serde"] }
 differential-dataflow = { workspace = true }
-mimalloc = "0.1.48"
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"
 timely = { workspace = true }
diff --git a/interactive/README.md b/interactive/README.md
index c72c07436..b29d57025 100644
--- a/interactive/README.md
+++ b/interactive/README.md
@@ -73,9 +73,18 @@ The flow moves through four steps:
 4. The `examples/` directory contains back-ends that execute programs.
 
 The `examples/programs/` directory contains example programs, intentionally simple at the moment.
-You can run any of them with one of the example harnesses, for example
+The one executable is the server (the `ddir-server` crate in `server/`, documented in
+`server/README.md`; the session scripts in `examples/server/` show it at work); a program runs by
+loading it, feeding its inputs, and closing epochs. For example, on a random graph of 100 nodes and
+200 edges, 10 of which change each epoch, for 100 epochs, on four workers of the Corgi backend:
 ```
-cargo run --release --example ddir_vec -- ./examples/programs/reach.ddp 2 100 200 1 100
+cd interactive
+printf 'load reach from examples/programs/reach.ddp
+feed reach 0 from random:nodes=100,edges=200,churn=10
+feed reach 1 0
+tick 100
+exit
+' | DDIR_BACKEND=corgi DDIR_WORKERS=4 cargo run --release -p ddir-server
 ```
 
 More generally, you can run
diff --git a/interactive/examples/aoc2023/README.md b/interactive/examples/aoc2023/README.md
index 652f26c91..26d9d32ab 100644
--- a/interactive/examples/aoc2023/README.md
+++ b/interactive/examples/aoc2023/README.md
@@ -17,17 +17,17 @@ mixed-arity inputs.
 
 ## Run
 
-    cargo build --release --example ddir
-    ./run.sh    # every part on the vec backend vs expected.txt; nonzero exit on any mismatch
+    cargo build --release -p ddir-server
+    ./run.sh          # every part on the vec backend vs expected.txt; nonzero exit on any mismatch
+    ./run.sh corgi    # the same through the Corgi columnar backend
 
 `run.sh` first runs `transcribe.py` (python3) to regenerate `gen/`, then
-runs each part as `EDGES_FILE=gen/<input> ddir --backend=vec
-dayNN/partN.ddp <arity> 10 0 1 0`; the answer is the `Int` on the
-`[partN]` inspect line.
+runs each part as one server session piped into `ddir_server` — `load` the
+program from its file, `feed … from` the fact file into input 0, `tick` — and
+reads the answer off the `[partN]` inspect line.
 
-`run.sh` asserts the **vec** backend only; corgi currently disagrees or
-crashes on 4 of these parts (known open issues, minimized separately) and
-is not asserted yet.
+Both backends pass all 33 parts. Corgi needs day05's arity-padded inputs
+(`run.sh corgi` transcribes with `--pad`).
 
 ## Verdicts
 
diff --git a/interactive/examples/aoc2023/run.sh b/interactive/examples/aoc2023/run.sh
index ade1193b1..bf253f55c 100755
--- a/interactive/examples/aoc2023/run.sh
+++ b/interactive/examples/aoc2023/run.sh
@@ -1,17 +1,23 @@
 #!/bin/sh
-# Run every AoC 2023 program on the vec backend and check the answers.
-# Usage: ./run.sh [path/to/ddir]   (build with: cargo build --release --example ddir)
+# Run every AoC 2023 program through the DDIR server and check the answers.
+# Usage: ./run.sh [vec|corgi] [path/to/ddir_server]
+#   (build with: cargo build --release -p ddir-server)
 cd "$(dirname "$0")" || exit 1
-DDIR=${1:-../../../target/release/examples/ddir}
-python3 transcribe.py || exit 1   # dense dayNN/input.txt -> gen/dayNN/ fact files
+BACKEND=${1:-vec}
+SERVER=${2:-../../../target/release/ddir_server}
+PAD=; [ "$BACKEND" = corgi ] && PAD=--pad   # corgi needs day05's uniform-arity copies
+python3 transcribe.py $PAD || exit 1   # dense dayNN/input.txt -> gen/dayNN/ fact files
 fail=0
 while read -r day part expected; do
   case "$day" in ''|'#'*) continue;; esac
   dir=day$day
   inp=gen/$dir/input.txt
   [ -f "gen/$dir/input$part.txt" ] && inp=gen/$dir/input$part.txt   # day05/day15: per-part inputs
-  arity=$(head -1 "$inp" | awk '{print NF}')
-  got=$(EDGES_FILE=$inp "$DDIR" --backend=vec "$dir/part$part.ddp" "$arity" 10 0 1 0 2>&1 \
+  [ -n "$PAD" ] && [ -f "gen/$dir/input${part}p.txt" ] && inp=gen/$dir/input${part}p.txt
+  # One session per part: load the program, feed its input from the fact
+  # file, close the epoch. The answer is the `Int` on the `[partN]` inspect line.
+  got=$(printf 'load p from %s\nfeed p 0 from %s\ntick\nexit\n' "$dir/part$part.ddp" "$inp" \
+        | DDIR_BACKEND="$BACKEND" "$SERVER" 2>&1 \
         | sed -n "s/.*\\[part$part\\].*Int(\\(-\\{0,1\\}[0-9]*\\)).*/\\1/p")
   if [ "$got" = "$expected" ]; then
     echo "day$day part$part: ok ($got)"
diff --git a/interactive/examples/ddir.rs b/interactive/examples/ddir.rs
deleted file mode 100644
index d262f1d00..000000000
--- a/interactive/examples/ddir.rs
+++ /dev/null
@@ -1,248 +0,0 @@
-//! The DDIR driver: parse, lower, render on the chosen backend, execute.
-//!
-//! Usage: `ddir [flags] <program> <arity> <nodes> <edges> [batch] [rounds] [timely args]`
-//!
-//! Inputs: with `EDGES_FILE` set, rows come from that file — one row per line,
-//! whitespace-separated `i64` fields, assigned round-robin to the program's
-//! inputs (`line % n_inputs`). Otherwise rows are synthesized by `gen_row`.
-//!
-//! Flags:
-//! - `--explain`: apply the explanation rewrite after lowering; the last input
-//!   becomes the query input (the demand envelope `(key ; val ++ q)`).
-//! - `--query=K:V[,q]`: seed the query input with one row (requires --explain).
-//! - `--debug-demand`: tap every demand collection with an Inspect.
-//! - `--diag`: serve timely/DD diagnostics on port 51371.
-//! - `--backend=vec|corgi`: rendering substrate (default `vec`). The corgi
-//!   backend exchanges by key hash, so both take `-w`.
-//! - `--sync=K`: await completion only every K rounds (default 1), letting K
-//!   timestamps retire with whatever inter-timestamp concurrency the system
-//!   finds — the open(er)-loop regime DD adapts into under load.
-
-use mimalloc::MiMalloc;
-
-#[global_allocator]
-static GLOBAL: MiMalloc = MiMalloc;
-
-use differential_dataflow::dynamic::pointstamp::PointStamp;
-use differential_dataflow::input::Input;
-
-use interactive::parse;
-use interactive::lower;
-use interactive::scope_ir as st;
-use interactive::ir::{Diff, Value};
-use interactive::backend::vec::{render_tree, Row};
-use interactive::backend::corgi::render_tree_rows;
-
-#[derive(Clone, Default)]
-struct Flags {
-    explain: bool,
-    query: Option<String>,
-    debug_demand: bool,
-    diag: bool,
-    corgi: bool,
-    sync: u64,
-}
-
-fn run(
-    name: &str,
-    stmts: Vec<parse::Stmt>,
-    n_inputs: usize,
-    nodes: u64,
-    edges: u64,
-    arity: usize,
-    batch: u64,
-    rounds: Option<u64>,
-    flags: Flags,
-    timely_args: Vec<String>,
-) {
-    let explain = flags.explain;
-    let mut tree = lower::lower_tree(stmts);
-    // --explain: rewrite for self-explanation before optimization (the rules
-    // assume single-op Linears). Sources are the root's positional inputs; the
-    // query arrives as one extra input appended after them.
-    if explain {
-        let source_shapes: Vec<(usize, usize)> = tree.root.imports.iter().map(|imp| match &imp.from {
-            st::Source::Input(_) => (arity, 0usize),
-            other => panic!("ddir --explain: unsupported source {:?}", other),
-        }).collect();
-        let shape = interactive::explain::export_shape(&tree, &source_shapes);
-        eprintln!("explain: first export shape (k={}, v={}); query is (key[{}] ; val[{}] ++ q)", shape.0, shape.1, shape.0, shape.1);
-        let options = interactive::explain::Options { debug_inspects: flags.debug_demand };
-        tree = interactive::explain::explain_with(&tree, &source_shapes, options);
-    }
-    let ops_before = tree.op_count();
-    tree.optimize();
-    let tree_export_idx = tree.root.exports.iter().position(|e| e.name == "result").unwrap_or(0);
-    println!("{}: {} ops before optimize, {} after; driving export {:?}",
-        name, ops_before, tree.op_count(), tree.root.exports[tree_export_idx].name);
-    let name = name.to_string();
-    let edges_file = std::env::var("EDGES_FILE").ok();
-    let total_inputs = if explain { n_inputs + 1 } else { n_inputs };
-    let query_input_idx = if explain { Some(n_inputs) } else { None };
-
-    timely::execute_from_args(timely_args.into_iter(), move |worker| {
-
-        // --diag registers timely/DD logging and serves the diagnostics
-        // WebSocket on worker 0 (port 51371) — see diagnostics/README.md.
-        let _diag = if flags.diag {
-            let state = diagnostics::logging::register(worker, false);
-            if worker.index() == 0 {
-                Some(diagnostics::server::Server::start(51371, state.sink))
-            } else {
-                drop(state.sink);
-                None
-            }
-        } else { None };
-
-        let (mut inputs, probe) = worker.dataflow::<u64, _, _>(|scope| {
-            let mut handles = Vec::new();
-            let mut collections = Vec::new();
-            for _ in 0..total_inputs {
-                let (h, c) = scope.new_collection::<(Row, Row), Diff>();
-                handles.push(h); collections.push(c);
-            }
-            let mut probe = timely::dataflow::ProbeHandle::new();
-            let output = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
-                let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect();
-                let root_imports: Vec<_> = tree.root.imports.iter().map(|imp| match &imp.from {
-                    st::Source::Input(n) => entered[*n].clone(),
-                    st::Source::Trace(name) => panic!("ddir: Import {:?} not supported in this harness (no trace registry).", name),
-                    st::Source::Parent(_) => unreachable!("root scope cannot import from a parent"),
-                }).collect();
-                let exports = if flags.corgi {
-                    render_tree_rows(&tree.root, inner, 0, root_imports)
-                } else {
-                    render_tree(&tree.root, inner, 0, root_imports)
-                };
-                exports[tree_export_idx].clone().leave(scope)
-            });
-            output.probe_with(&mut probe);
-            (handles, probe)
-        });
-
-        let index = worker.index();
-        let peers = worker.peers();
-
-        let timer = std::time::Instant::now();
-        let timer_load = std::time::Instant::now();
-        // With `EDGES_FILE` set, rows come from the file (one row per line,
-        // whitespace-separated i64 fields), assigned round-robin to inputs.
-        // Otherwise `gen_row` synthesizes `edges` rows.
-        if let Some(path) = &edges_file {
-            let text = std::fs::read_to_string(path).unwrap_or_else(|e| panic!("Cannot read {}: {}", path, e));
-            for (e, line) in text.lines().filter(|l| !l.trim().is_empty()).enumerate() {
-                if e % peers == index {
-                    let input_idx = e % n_inputs;
-                    let fields: Vec<Value> = line.split_whitespace().map(|t| Value::Int(t.parse::<i64>().unwrap())).collect();
-                    inputs[input_idx].update((Value::Tuple(fields), Value::unit()), 1);
-                }
-            }
-        } else {
-            for e in 0..edges {
-                if (e as usize) % peers == index {
-                    let input_idx = (e as usize) % n_inputs;
-                    inputs[input_idx].update(interactive::gen_row(e, nodes, arity), 1);
-                }
-            }
-        }
-        // Seed the query input (worker 0 only) from --query="k:v[,q]". The row
-        // is the flat demand envelope `(key[k] ; val[v] ++ [q])` — at depth 0
-        // there is no chain, so the value is just V's fields then the query id.
-        if let Some(q_idx) = query_input_idx {
-            if index == 0 {
-                if let Some(qstr) = flags.query.clone() {
-                    let parse_fields = |s: &str| -> Vec<Value> {
-                        if s.is_empty() { vec![] } else { s.split(',').map(|t| Value::Int(t.trim().parse::<i64>().unwrap())).collect() }
-                    };
-                    let (k_str, vq_str) = qstr.split_once(':').unwrap_or((qstr.as_str(), ""));
-                    let q_key = Value::Tuple(parse_fields(k_str));
-                    let mut vq = parse_fields(vq_str);
-                    if vq.is_empty() { vq.push(Value::Int(0)); } // a bare query id
-                    let q_val = Value::Tuple(vq);
-                    eprintln!("seeding query: key={:?} val_with_q={:?}", q_key, q_val);
-                    inputs[q_idx].update((q_key, q_val), 1);
-                }
-            }
-        }
-        for i in inputs.iter_mut() { i.advance_to(1); i.flush(); }
-        while probe.less_than(&1u64) { worker.step(); }
-        println!("worker {}: {} loaded ({} edges, total {:.2?}, load {:.2?})", index, name, edges, timer.elapsed(), timer_load.elapsed());
-
-        let mut cursor = 0u64;
-        let mut round = 0u64;
-        let limit = rounds.unwrap_or(u64::MAX);
-        while round < limit {
-            let timer_round = std::time::Instant::now();
-            let time = (round + 2) as u64;
-            for _ in 0..batch {
-                let remove_idx = cursor;
-                let add_idx = edges + cursor;
-                if (remove_idx as usize) % peers == index {
-                    let input_idx = (remove_idx as usize) % n_inputs;
-                    inputs[input_idx].update(interactive::gen_row(remove_idx, nodes, arity), -1);
-                }
-                if (add_idx as usize) % peers == index {
-                    let input_idx = (add_idx as usize) % n_inputs;
-                    inputs[input_idx].update(interactive::gen_row(add_idx, nodes, arity), 1);
-                }
-                cursor += 1;
-            }
-            for i in inputs.iter_mut() { i.advance_to(time); i.flush(); }
-            let sync = flags.sync.max(1);
-            if (round + 1) % sync == 0 || round + 1 == limit {
-                while probe.less_than(&time) { worker.step(); }
-            }
-
-            round += 1;
-            if round % 100 == 0 {
-                println!("worker {}: {} round {} (total {:.2?}, round {:.2?})", index, name, round, timer.elapsed(), timer_round.elapsed());
-            }
-        }
-        println!("worker {}: {} done ({} rounds, batch {}, total {:.2?})", index, name, round, batch, timer.elapsed());
-    }).unwrap();
-}
-
-fn main() {
-    // Strip our flags; everything else stays positional (and the tail is
-    // forwarded to timely).
-    let raw_args: Vec<String> = std::env::args().collect();
-    let (flags, args): (Flags, Vec<String>) = {
-        let mut it = raw_args.into_iter();
-        let prog = it.next().unwrap();
-        let mut flags = Flags::default();
-        let mut rest: Vec<String> = Vec::new();
-        for a in it {
-            if a == "--explain" { flags.explain = true; }
-            else if let Some(q) = a.strip_prefix("--query=") { flags.query = Some(q.to_string()); }
-            else if a == "--debug-demand" { flags.debug_demand = true; }
-            else if a == "--diag" { flags.diag = true; }
-            else if let Some(k) = a.strip_prefix("--sync=") { flags.sync = k.parse().expect("--sync=K"); }
-            else if let Some(b) = a.strip_prefix("--backend=") {
-                flags.corgi = match b { "corgi" => true, "vec" => false, other => panic!("unknown backend {other:?} (vec|corgi)") };
-            }
-            else { rest.push(a); }
-        }
-        let mut out = vec![prog]; out.extend(rest);
-        (flags, out)
-    };
-    let program = args.get(1).cloned().unwrap_or_else(|| { std::process::exit(0); });
-    let arity: usize = args.get(2).cloned().unwrap_or("2".into()).parse().unwrap();
-    let nodes: u64 = args.get(3).cloned().unwrap_or("10".into()).parse().unwrap();
-    let edges: u64 = args.get(4).cloned().unwrap_or_else(|| (2 * nodes).to_string()).parse().unwrap();
-    let batch: u64 = args.get(5).cloned().unwrap_or("1".into()).parse().unwrap();
-    let rounds: Option<u64> = args.get(6).map(|s| s.parse().unwrap());
-
-    let source = interactive::load_program(&program);
-    let stmts = if program.ends_with(".ddp") {
-        parse::pipe::parse(&source)
-    } else {
-        parse::applicative::parse(&source)
-    };
-    let (n_inputs, imports) = interactive::survey_sources(&stmts);
-    if !imports.is_empty() {
-        panic!("ddir: program references imports {:?} but this harness has no trace registry.", imports);
-    }
-    let name = std::path::Path::new(&program).file_stem().map(|s| s.to_string_lossy().into_owned()).unwrap_or(program.clone());
-    let timely_args: Vec<String> = args.iter().skip(4).cloned().collect();
-    run(&name, stmts, n_inputs, nodes, edges, arity, batch, rounds, flags, timely_args);
-}
diff --git a/interactive/examples/ddir_server.rs b/interactive/examples/ddir_server.rs
deleted file mode 100644
index 67bf91d89..000000000
--- a/interactive/examples/ddir_server.rs
+++ /dev/null
@@ -1,323 +0,0 @@
-//! A live DDIR server: install interpreted dataflows into one running timely
-//! computation and let them share results by name. Multi-worker capable.
-//!
-//! Usage (see `examples/server/` for session scripts and a README):
-//!   cargo run --release --example ddir_server                      # stdin, 1 worker
-//!   cargo run --release --example ddir_server -- <script>          # script, 1 worker
-//!   cargo run --release --example ddir_server -- <script> -w4      # script, 4 workers
-//!   cargo run --release --example ddir_server -- -w4               # stdin, 4 workers
-//! e.g. cargo run --example ddir_server -- interactive/examples/server/sessions/shared_trace.txt
-//!
-//! # Where parsing happens
-//!
-//! Commands are parsed, lowered, and validated **on the main (intake) thread**,
-//! not on the workers. A malformed program or value is reported to the prompt
-//! and the server keeps running — bad input can never panic a worker. Only a
-//! well-typed [`Command`] is handed to worker 0, which injects it into a timely
-//! `Sequencer`; the resulting total order is replayed on every worker, so
-//! install/tick/drop stay collective while `feed` is applied on worker 0 only
-//! (the arrangement's exchange pact routes data to key owners).
-//!
-//! Commands (one per line; `#` or `--` starts a comment):
-//!   install <name> <file>
-//!   feed <prog> <in#> <value> [val=<value>] [time=<t>] [diff=<int>]
-//!   tick
-//!   drop <name>
-//!   peek <trace> [key]
-//!   list
-//!   help
-//!   exit | quit
-//!
-//! A `<value>` is either a comma-separated row of integers (becoming a
-//! `Tuple` of `Int`s; `_`/empty = unit), or — for structured/ADT data such as
-//! ASTs — any closed scalar term: `tuple(1,2)`, `list(1,2,3)`,
-//! `inject(2,tuple(3,4))`, etc. A value is one whitespace-delimited token, so
-//! write terms without spaces. `feed`'s value defaults to unit, `time` to the
-//! current epoch (use a future `time=` to schedule ahead), and `diff` to +1.
-
-use mimalloc::MiMalloc;
-
-#[global_allocator]
-static GLOBAL: MiMalloc = MiMalloc;
-
-use std::any::Any;
-use std::panic::{catch_unwind, AssertUnwindSafe};
-use std::sync::{Arc, Mutex};
-use std::sync::mpsc::{channel, TryRecvError};
-use std::time::{Duration, Instant};
-
-use timely::synchronization::Sequencer;
-use timely::worker::Worker;
-
-use interactive::parse;
-use interactive::lower;
-use interactive::scope_ir as st;
-use interactive::ir::{eval, Value};
-use interactive::server::{Command, Server};
-
-/// Parse, lower, and optimize a program file (`.ddp` = pipe syntax, else applicative).
-fn load(path: &str) -> st::Program {
-    let src = interactive::load_program(path);
-    let stmts = if path.ends_with(".ddp") {
-        parse::pipe::parse(&src)
-    } else {
-        parse::applicative::parse(&src)
-    };
-    let mut prog = lower::lower_tree(stmts);
-    prog.optimize();
-    prog
-}
-
-/// A `<value>`: a comma-separated integer row -> `Tuple`, or a closed scalar
-/// term evaluated to a constant `Value` (for ADT-shaped input).
-fn parse_value(s: &str) -> Value {
-    let s = s.trim();
-    if s.is_empty() || s == "_" {
-        return Value::unit();
-    }
-    // Fast path: a plain `a,b,c` integer row.
-    if s.chars().all(|c| c.is_ascii_digit() || c == ',' || c == '-' || c.is_whitespace()) {
-        if let Ok(ints) = s.split(',').map(|t| t.trim().parse::<i64>()).collect::<Result<Vec<_>, _>>() {
-            return Value::Tuple(ints.into_iter().map(Value::Int).collect());
-        }
-    }
-    // Otherwise: a closed scalar term (no `$n` input references).
-    eval(&parse::pipe::parse_term(s), &mut Vec::new())
-}
-
-/// Render a caught panic payload as a message.
-fn panic_msg(e: Box<dyn Any + Send>) -> String {
-    if let Some(s) = e.downcast_ref::<&str>() { (*s).to_string() }
-    else if let Some(s) = e.downcast_ref::<String>() { s.clone() }
-    else { "panic".to_string() }
-}
-
-/// Parse one command line into a typed `Command`, on the intake thread.
-/// Parsing/lowering that would `panic!` (malformed program or value) is caught
-/// and returned as an error, so the prompt survives bad input.
-fn parse_command(line: &str) -> Result<Command, String> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    match toks[0] {
-        "install" if toks.len() == 3 => {
-            let name = toks[1].to_string();
-            let file = toks[2].to_string();
-            let program = catch_unwind(AssertUnwindSafe(|| load(&file))).map_err(panic_msg)?;
-            Ok(Command::Install { name, program })
-        }
-        "feed" if toks.len() >= 4 => {
-            let prog = toks[1].to_string();
-            let input: usize = toks[2].parse().map_err(|_| format!("feed: <in#> must be a number, got {:?}", toks[2]))?;
-            let key = catch_unwind(AssertUnwindSafe(|| parse_value(toks[3]))).map_err(panic_msg)?;
-            let mut val = Value::unit();
-            let mut time = None;
-            let mut diff = 1i64;
-            for t in &toks[4..] {
-                if let Some(v) = t.strip_prefix("val=") {
-                    val = catch_unwind(AssertUnwindSafe(|| parse_value(v))).map_err(panic_msg)?;
-                } else if let Some(s) = t.strip_prefix("time=") {
-                    time = Some(s.parse().map_err(|_| format!("feed: time= must be a number, got {:?}", s))?);
-                } else if let Some(s) = t.strip_prefix("diff=") {
-                    diff = s.parse().map_err(|_| format!("feed: diff= must be an integer, got {:?}", s))?;
-                } else {
-                    return Err(format!("feed: unrecognized argument {:?}", t));
-                }
-            }
-            Ok(Command::Feed { prog, input, key, val, time, diff })
-        }
-        "tick" if toks.len() == 1 => Ok(Command::Tick { n: 1 }),
-        "bind" | "unbind" if toks.len() == 4 => {
-            let trace = toks[1].to_string();
-            let prog = toks[2].to_string();
-            let input: usize = toks[3].parse().map_err(|_| format!("{}: <in#> must be a number, got {:?}", toks[0], toks[3]))?;
-            if toks[0] == "bind" {
-                Ok(Command::Bind { trace, prog, input })
-            } else {
-                Ok(Command::Unbind { trace, prog, input })
-            }
-        }
-        "drop" if toks.len() == 2 => Ok(Command::Drop { name: toks[1].to_string() }),
-        "peek" if toks.len() == 2 || toks.len() == 3 => {
-            let trace = toks[1].to_string();
-            let key = match toks.get(2) {
-                Some(k) => Some(catch_unwind(AssertUnwindSafe(|| parse_value(k))).map_err(panic_msg)?),
-                None => None,
-            };
-            Ok(Command::Peek { trace, key })
-        }
-        "list" => Ok(Command::List),
-        "exit" | "quit" => Ok(Command::Exit),
-        other => Err(format!("unknown or malformed command {:?} (try `help`)", other)),
-    }
-}
-
-fn print_help() {
-    println!("commands:");
-    println!("  install <name> <file>");
-    println!("  feed <prog> <in#> <value> [val=<value>] [time=<t>] [diff=<int>]");
-    println!("  tick");
-    println!("  bind <trace> <prog> <in#>    (feed the trace's changes back in, each tick)");
-    println!("  unbind <trace> <prog> <in#>");
-    println!("  drop <name>");
-    println!("  peek <trace> [key]");
-    println!("  list");
-    println!("  exit");
-}
-
-/// Execute one sequenced command on this worker. Collective commands
-/// (`install`/`tick`/`drop`) run on every worker; `feed` and all printing
-/// happen on worker 0. Returns `false` for `exit`.
-fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool {
-    let w0 = worker.index() == 0;
-    match cmd {
-        Command::Install { name, program } => match server.install(worker, name, program) {
-            Ok(()) => if w0 { println!("installed {:?}", name); },
-            Err(e) => if w0 { println!("error: {}", e); },
-        },
-        Command::Feed { prog, input, key, val, time, diff } => {
-            if w0 {
-                if let Err(e) = server.feed(prog, *input, key.clone(), val.clone(), *time, *diff) {
-                    println!("error: {}", e);
-                }
-            }
-        }
-        Command::FeedBatch { prog, input, updates } => {
-            if w0 {
-                if let Err(e) = server.feed_batch(prog, *input, updates.clone()) {
-                    println!("error: {}", e);
-                }
-            }
-        }
-        Command::Tick { n } => {
-            for _ in 0..*n {
-                server.tick(worker);
-            }
-            if w0 { println!("tick -> epoch {}", server.epoch()); }
-        }
-        Command::Drop { name } => match server.drop_program(worker, name) {
-            Ok(()) => if w0 { println!("dropped {:?}", name); },
-            Err(e) => if w0 { println!("error: {}", e); },
-        },
-        // Collective: the tap dataflow is built on every worker (each sees
-        // its shard, so the union delivers the delta exactly once).
-        Command::Bind { trace, prog, input } => match server.bind(worker, trace, prog, *input) {
-            Ok(()) => if w0 { println!("bound {:?} -> {:?} input {}", trace, prog, input); },
-            Err(e) => if w0 { println!("error: {}", e); },
-        },
-        Command::Unbind { trace, prog, input } => match server.unbind(worker, trace, prog, *input) {
-            Ok(()) => if w0 { println!("unbound {:?} -> {:?} input {}", trace, prog, input); },
-            Err(e) => if w0 { println!("error: {}", e); },
-        },
-        // Collective: every worker imports its shard; `peek` gathers to worker 0
-        // (which prints) and reports an error there if the trace is unknown.
-        Command::Peek { trace, key } => {
-            if let Err(e) = server.peek(worker, trace, key.clone()) {
-                if w0 { println!("error: {}", e); }
-            }
-        }
-        Command::List => if w0 { server.list(); },
-        Command::Exit => return false,
-    }
-    true
-}
-
-fn main() {
-    // Quiet the default panic printer on the main (intake) thread, where parser
-    // panics are caught and reported as clean `error:` lines; worker-thread
-    // panics still print normally.
-    let default_hook = std::panic::take_hook();
-    std::panic::set_hook(Box::new(move |info| {
-        if std::thread::current().name() != Some("main") {
-            default_hook(info);
-        }
-    }));
-
-    // First positional arg (if it isn't a flag) is the script; the rest go to
-    // timely. We always pass a leading dummy so getopts has an argv[0] to skip.
-    let mut it = std::env::args();
-    let _bin = it.next();
-    let mut script: Option<String> = None;
-    let mut timely_args: Vec<String> = vec!["ddir_server".to_string()];
-    let mut saw_positional = false;
-    for a in it {
-        if !saw_positional && !a.starts_with('-') {
-            script = Some(a);
-            saw_positional = true;
-        } else {
-            saw_positional = true;
-            timely_args.push(a);
-        }
-    }
-
-    // Main thread -> worker 0 command channel, carrying typed commands.
-    let (send, recv) = channel::<Command>();
-    let recv = Arc::new(Mutex::new(recv));
-
-    let guards = timely::execute_from_args(timely_args.into_iter(), move |worker| {
-        let recv = recv.clone();
-        let me_zero = worker.index() == 0;
-        let mut server = Server::new();
-        let mut sequencer: Sequencer<Command> = Sequencer::new(worker, Instant::now());
-
-        let mut done = false;
-        let mut sealed = false; // worker 0 has observed the channel closing
-        while !done {
-            if me_zero && !sealed {
-                match recv.lock().expect("command channel poisoned").try_recv() {
-                    Ok(cmd) => sequencer.push(cmd),
-                    Err(TryRecvError::Empty) => {}
-                    Err(TryRecvError::Disconnected) => {
-                        sequencer.push(Command::Exit); // stop every worker together
-                        sealed = true;
-                    }
-                }
-            }
-
-            let mut worked = false;
-            while let Some(cmd) = sequencer.next() {
-                worked = true;
-                if !dispatch(&cmd, &mut server, worker) {
-                    done = true;
-                }
-            }
-
-            worker.step();
-            if !worked {
-                std::thread::sleep(Duration::from_millis(1));
-            }
-        }
-    }).expect("timely execute failed");
-
-    // Main thread: parse command lines and stream typed commands to worker 0.
-    let lines: Box<dyn Iterator<Item = String>> = match script {
-        Some(path) => match std::fs::read_to_string(&path) {
-            Ok(body) => Box::new(body.lines().map(|s| s.to_string()).collect::<Vec<_>>().into_iter()),
-            Err(e) => { eprintln!("cannot read script {:?}: {}", path, e); Box::new(std::iter::empty()) }
-        },
-        None => {
-            use std::io::BufRead;
-            println!("ddir server — type `help`, or `exit` to quit.");
-            Box::new(std::io::stdin().lock().lines().map(|l| l.unwrap()))
-        }
-    };
-    for line in lines {
-        let line = line.trim();
-        if line.is_empty() || line.starts_with('#') || line.starts_with("--") {
-            continue;
-        }
-        if line == "help" {
-            print_help();
-            continue;
-        }
-        match parse_command(line) {
-            Ok(cmd) => {
-                let is_exit = matches!(cmd, Command::Exit);
-                if send.send(cmd).is_err() { break; } // workers gone
-                if is_exit { break; }
-            }
-            Err(e) => println!("error: {}", e), // the prompt survives bad input
-        }
-    }
-    // Closing the channel makes worker 0 broadcast `Exit`.
-    drop(send);
-    guards.join();
-}
diff --git a/interactive/examples/server/README.md b/interactive/examples/server/README.md
index f1c37959e..88ce2bbe7 100644
--- a/interactive/examples/server/README.md
+++ b/interactive/examples/server/README.md
@@ -7,50 +7,87 @@ consumer maintains a computation incrementally over a producer's output. It is
 the interpreter-driven successor to the old `dd_server` crate (which hot-loaded
 compiled `.so`s); here "install" means render a DDIR program, not `dlopen`.
 
-The implementation is `interactive::server` (the registry + lifecycle) driven by
-the `ddir_server` example (a command loop). It is multi-worker: commands are
-totally ordered across workers by a timely `Sequencer`.
+The implementation is `interactive::server` (the registry + lifecycle), driven
+by the `ddir-server` crate in `interactive/server/` — a long-running process
+that takes commands over stdin, TCP and WebSocket (its README has the protocol
+and the configuration). The session scripts here are piped into its stdin.
 
 ## Two kinds of file (don't mix them up)
 
-- **`programs/*.ddp`** — DDIR *programs*: dataflow definitions you `install`.
-  The ones here are server-oriented (they use `import`/`export`), so unlike the
-  programs in `../programs/` they are not runnable by the batch `ddir_vec`
-  harness.
+- **`programs/*.ddp`** — DDIR *programs*: dataflow definitions you `load`.
+  The ones here are server-oriented (they use `import`/`export`); the programs
+  in `../programs/` read positional `input`s instead, which you fill with
+  `feed` (see "Running a program in batch" below).
 - **`sessions/*.txt`** — *command scripts*: a stream of server commands
-  (`install`/`feed`/`tick`/…) you hand to the server. You do **not** `install` a
+  (`load`/`feed`/`tick`/…) you hand to the server. You do **not** `load` a
   session; you run the server *on* it.
 
 ## Running
 
 ```
-# Run a session script (add -w4 for four workers):
-cargo run --release --example ddir_server -- interactive/examples/server/sessions/shared_trace.txt
+cd interactive
+# Run a session script (paths inside the scripts are relative to here):
+cargo run --release -p ddir-server < examples/server/sessions/shared_trace.txt
 
-# Or interactively (no script arg): type `help`, or `exit`.
-cargo run --release --example ddir_server
+# Four workers, the Corgi columnar backend (default: one worker, vec):
+DDIR_WORKERS=4 DDIR_BACKEND=corgi cargo run --release -p ddir-server < examples/server/sessions/shared_trace.txt
+
+# Or interactively: type commands; `exit` (or end of input) stops the server.
+cargo run --release -p ddir-server
 ```
 
-Paths inside the session scripts are relative to the `interactive/` crate
-directory, so run from there (as the examples above assume `cargo` is invoked at
-the repo root with `--example`; adjust if you `cd interactive` first).
+Every response line starts with the request's id (`_7 ok …`, `_7 data …`,
+`_7 err …`); the server numbers requests that do not carry one.
 
 ## Commands
 
 | command | effect |
 |---|---|
-| `install <name> <file>` | parse + lower + install a program under `<name>` |
+| `load <name> from <file> [explain=<arity>[,debug]]` | parse + lower + install a program file under `<name>`; optionally after the explanation rewrite (`load <name> begin … end-load` takes the program inline) |
 | `feed <prog> <in#> <value> [val=<value>] [time=<t>] [diff=<int>]` | stage an input update |
-| `tick` | close the epoch and run to quiescence |
+| `feed <prog> <in#> from <recipe-or-file>` | fill an input from a recipe (`random:…`, `iota:N`) or a file of integer rows, sharded across workers |
+| `tick [n]` | close `n` epochs (default 1), running to quiescence after each; reports the epoch and the wall-clock time |
+| `bind <trace> <prog> <in#>` / `unbind …` | feed a trace's changes back into an input at every tick (one-epoch-delayed feedback) |
 | `drop <name>` | evict a program (refused if a live program still imports its trace) |
-| `peek <trace> [key]` | print a trace's current contents (consolidated across workers) |
-| `list` | show traces (+ importer counts) and installed programs |
+| `peek <trace> [key]` | print a trace's current contents (consolidated across workers), one `data` line per row |
+| `tail <trace>` / `stop <reqid>` | follow a trace's changes at every tick, until stopped |
+| `list` | show programs, traces (+ importer counts) and bindings |
 
 A `<value>` is a comma-separated integer row (`1,2` → a tuple; `_`/empty → unit)
 or any **closed scalar term, written without spaces** (`inject(2,tuple(3,4))`,
 `list(1,2,3)`) for ADT-shaped data such as ASTs. `feed` defaults to value=unit,
 `time`=the current epoch (use a future `time=` to schedule ahead), `diff`=+1.
 
+## Running a program in batch
+
+The programs in `../programs/` (reachability, SCC, stable matching, …) read
+positional inputs. A session fills them in bulk and closes epochs; that is the
+whole of the old single-program harness, so there is no separate binary:
+
+```
+load scc from examples/programs/scc.ddp
+feed scc 0 from random:nodes=100000,edges=200000,churn=100
+tick          # the initial load: reported as one epoch's time
+tick 100      # 100 epochs of 100 replaced edges each, timed together
+peek result
+exit
+```
+
+`feed … from` deals the rows across the workers (each feeds its shard, and
+the exchange places every row on its key's owner). A `random:` recipe with
+`churn=C` keeps the input changing: every later `tick` retracts the next `C`
+rows of its window and adds `C` fresh ones, which is the standing-change regime
+programs are benchmarked under. A file source is one row per line of
+whitespace-separated integers, each becoming `(Tuple[ints] ; ())`; the
+`aoc2023/run.sh` suite drives every AoC program this way. A row `feed` still
+serves the small inputs (roots, queries).
+
+`load … explain=<arity>` applies the explanation rewrite before
+optimization, treating every source as `arity` key fields with no value; the
+rewritten program has one extra input after its own — the query input, fed as
+`(key ; val ++ q)` — and exports one `demand:inputN` trace per source, which
+you `peek`. `,debug` taps every demand collection with an inspect.
+
 ## Generated (named) sources
 
 A program can `import` a *recipe* name instead of another program's export:
@@ -123,7 +160,7 @@ log-work form is left for later.)
 - **`peek.txt`** — reading results back out: `peek` a whole trace and a single
   key, and the clean error for an unknown trace.
 - **`generated.txt`** — a random graph imported by recipe, installed on demand
-  and shared by two programs (`importers: 2`, one source), then GC'd on drop.
+  and shared by two programs (`importers=2`, one source), then GC'd on drop.
 - **`derived.txt`** — a random graph derived in-language from `iota` via `hash`.
 - **`clock.txt`** — the `clock` source advancing one step per tick (O(1) deltas).
 - **`arrange_idem.txt`** — the e-graph rewrite rule firing **incrementally**:
diff --git a/interactive/examples/server/sessions/arrange_idem.txt b/interactive/examples/server/sessions/arrange_idem.txt
index a0c6642b5..e8d9d00f6 100644
--- a/interactive/examples/server/sessions/arrange_idem.txt
+++ b/interactive/examples/server/sessions/arrange_idem.txt
@@ -6,7 +6,7 @@
 # 1 = Arrange, 2 = Reduce (both produce an arrangement). Run from interactive/:
 #   cargo run --release --example ddir_server -- examples/server/sessions/arrange_idem.txt
 
-install eg examples/server/programs/arrange_idem.ddp
+load eg from examples/server/programs/arrange_idem.ddp
 
 # Base: node 0 (a source), 1 = Arrange(0), 2 = Reduce(1).
 feed eg 0 0 val=tuple(5,list())
diff --git a/interactive/examples/server/sessions/clock.txt b/interactive/examples/server/sessions/clock.txt
index 65be1a12e..6be4635fc 100644
--- a/interactive/examples/server/sessions/clock.txt
+++ b/interactive/examples/server/sessions/clock.txt
@@ -1,7 +1,7 @@
 # The `clock` ticking source: a single row that advances by one each tick.
 # Watch the inspect output — each tick is an O(1) change (retract old, add new),
 # not a regeneration.
-install watch examples/server/programs/clock_watch.ddp
+load watch from examples/server/programs/clock_watch.ddp
 tick
 tick
 tick
diff --git a/interactive/examples/server/sessions/derived.txt b/interactive/examples/server/sessions/derived.txt
index 4e5f0f01c..49434a97d 100644
--- a/interactive/examples/server/sessions/derived.txt
+++ b/interactive/examples/server/sessions/derived.txt
@@ -1,6 +1,6 @@
 # A random graph derived in-language from `iota` via `hash` — the composable
 # form of `random:`, computed once. No baked-in generator; just iota + hash.
-install reach examples/server/programs/rand_reach.ddp
+load reach from examples/server/programs/rand_reach.ddp
 feed reach 0 0
 tick
 # The derived edge set (note iota:12 was installed on demand to back it):
diff --git a/interactive/examples/server/sessions/drop.txt b/interactive/examples/server/sessions/drop.txt
index 2730874be..419725fde 100644
--- a/interactive/examples/server/sessions/drop.txt
+++ b/interactive/examples/server/sessions/drop.txt
@@ -1,7 +1,7 @@
 # Teardown / drop-gate demo.
 # `reach` imports the "edges" trace published by `prod`.
-install prod examples/server/programs/producer.ddp
-install reach examples/server/programs/reach_import.ddp
+load prod from examples/server/programs/producer.ddp
+load reach from examples/server/programs/reach_import.ddp
 feed prod 0 1,2
 feed prod 0 2,3
 feed reach 0 1
@@ -17,7 +17,7 @@ drop prod
 list
 
 # Re-install a fresh producer under the now-free name to show the registry freed it.
-install prod examples/server/programs/producer.ddp
+load prod from examples/server/programs/producer.ddp
 feed prod 0 7,8
 tick
 list
diff --git a/interactive/examples/server/sessions/generated.txt b/interactive/examples/server/sessions/generated.txt
index 3cd977397..a7ba95e93 100644
--- a/interactive/examples/server/sessions/generated.txt
+++ b/interactive/examples/server/sessions/generated.txt
@@ -3,9 +3,9 @@
 
 # Importing "random:nodes=8,edges=12" auto-installs the generator — no producer
 # to set up first.
-install reach examples/server/programs/reach_gen.ddp
+load reach from examples/server/programs/reach_gen.ddp
 # A second program imports the SAME recipe; it shares the one source.
-install counter examples/server/programs/count_gen.ddp
+load counter from examples/server/programs/count_gen.ddp
 feed reach 0 0
 tick
 
diff --git a/interactive/examples/server/sessions/peek.txt b/interactive/examples/server/sessions/peek.txt
index 5b0b1fd78..ab7d24468 100644
--- a/interactive/examples/server/sessions/peek.txt
+++ b/interactive/examples/server/sessions/peek.txt
@@ -1,6 +1,6 @@
 # Reading results back out with `peek`.
-install prod examples/server/programs/producer.ddp
-install reach examples/server/programs/reach_import.ddp
+load prod from examples/server/programs/producer.ddp
+load reach from examples/server/programs/reach_import.ddp
 
 # graph 1 -> 2 -> 3, root 1
 feed prod 0 1,2
diff --git a/interactive/examples/server/sessions/shared_trace.txt b/interactive/examples/server/sessions/shared_trace.txt
index 211acf750..843860335 100644
--- a/interactive/examples/server/sessions/shared_trace.txt
+++ b/interactive/examples/server/sessions/shared_trace.txt
@@ -3,8 +3,8 @@
 # tracks the nodes reachable from its roots. Run from the interactive/ dir:
 #   cargo run --release --example ddir_server -- examples/server/sessions/shared_trace.txt
 
-install prod examples/server/programs/producer.ddp
-install reach examples/server/programs/reach_import.ddp
+load prod from examples/server/programs/producer.ddp
+load reach from examples/server/programs/reach_import.ddp
 list
 
 # graph: 1 -> 2 -> 3 ; root: 1   (reach should become {1, 2, 3})
diff --git a/interactive/examples/server/sessions/values.txt b/interactive/examples/server/sessions/values.txt
index cf549d4cc..c92353706 100644
--- a/interactive/examples/server/sessions/values.txt
+++ b/interactive/examples/server/sessions/values.txt
@@ -1,5 +1,5 @@
 # Structured (ADT) values + time-scheduled input.
-install echo examples/server/programs/echo.ddp
+load echo from examples/server/programs/echo.ddp
 
 # A Variant payload at the current epoch (0): a tagged tuple, the shape an AST node takes.
 # (values are one token — write terms without spaces)
diff --git a/interactive/server/Cargo.toml b/interactive/server/Cargo.toml
index d4b617152..4c44eb3e4 100644
--- a/interactive/server/Cargo.toml
+++ b/interactive/server/Cargo.toml
@@ -17,6 +17,7 @@ workspace = true
 differential-dataflow = { workspace = true }
 timely = { workspace = true }
 interactive = { path = ".." }
+mimalloc = "0.1.48"
 diagnostics = { path = "../../diagnostics" }
 serde = { version = "1.0", features = ["derive"] }
 tungstenite = "0.26"
diff --git a/interactive/server/README.md b/interactive/server/README.md
index 6d552e850..e3dcc9881 100644
--- a/interactive/server/README.md
+++ b/interactive/server/README.md
@@ -42,8 +42,18 @@ Between commands, blank lines and `#` comment lines are skipped, so command
 scripts can be piped to stdin (see `demo/`).
 
 The useful commands are `load`, `drop`, `list`, `feed`, `bind`, `unbind`,
-`peek`, `tail`, `stop`, `tick`, and `exit`. `load` accepts an inline
-pipe-syntax program:
+`peek`, `tail`, `stop`, `tick`, and `exit`. `load <name> from <path>` installs
+a program file (`.ddp` pipe syntax, anything else applicative); `load <name>
+from <path> explain=<arity>[,debug]` applies the explanation rewrite first,
+every source taken to have `arity` key fields and no value, after which the
+query input is the one after the program's own and the demand sets are
+`peek`able exports. `feed <prog> <in#> from <source>` fills an input from a
+source the server reads itself — a recipe such as `random:nodes=N,edges=E,churn=C`
+or `iota:N`, or a file of integer rows — with each worker taking its shard, so
+no row crosses the wire; a `churn=C` recipe then replaces `C` rows on every
+`tick`. `peek <trace> [key]` snapshots a trace or one key of it, and `tick [n]`
+reports the epoch reached and the wall-clock time it took. `load` also accepts
+an inline pipe-syntax program:
 
     load graph begin
     let edges = import "random:nodes=8,edges=12,seed=1,churn=1";
@@ -70,7 +80,7 @@ state, not elapsed wall-clock time.
     feed <prog> <in#> <key> [val=<v>] [time=<t>] [diff=<d>]
 
 pushes one update into a loaded program's positional input, exactly as in the
-`ddir_server` example (`1,2` → a tuple; `_` → unit; a closed scalar term such
+`interactive::server` tests (`1,2` → a tuple; `_` → unit; a closed scalar term such
 as `inject(2,tuple(3,4))` for ADT-shaped rows).
 
 For many updates to one target at the current epoch, frame them as one feed:
diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs
index df8a06ec6..b4a14df1f 100644
--- a/interactive/server/src/cmd.rs
+++ b/interactive/server/src/cmd.rs
@@ -37,14 +37,18 @@ pub enum Cmd {
     /// a fresh id (echo'd in the response).
     /// `bindings` — `import-name -> binding`, where the binding is either a
     /// registered trace name or a builtin call (`random(...)`).
-    /// `program` — DDIR text.
-    /// `explain` — request the explain rewrite (reserved; the server
-    /// reports an error rather than improvising a meaning).
+    /// `program` — DDIR text, inline (`load … begin` … `end-load`) or read
+    /// from a file (`load <name> from <path>`).
+    /// `explain` — `explain=<arity>[,debug]`: apply the explanation rewrite,
+    /// every source taken to have `arity` key fields and no value; the query
+    /// input is the one after the program's own, the demand sets its exports.
+    /// `applicative` — the text is in the applicative syntax (a `.ddir` file).
     Load {
         id_hint: String,
         bindings: BTreeMap<String, String>,
         program: String,
-        explain: bool,
+        explain: Option<(usize, bool)>,
+        applicative: bool,
     },
     /// Drop the dataflow named by id or by `id_hint`. Fails if any
     /// export of this dataflow is still imported by another live
@@ -52,8 +56,8 @@ pub enum Cmd {
     Drop { target: DataflowRef },
     /// List held names.
     List,
-    /// One-shot snapshot of a named trace.
-    Peek { name: String },
+    /// One-shot snapshot of a named trace, optionally of one key.
+    Peek { name: String, key: Option<Value> },
     /// Persistent subscription to a named trace.
     Tail { name: String },
     /// Cancel a previous `tail` (matched by its reqid).
@@ -77,6 +81,14 @@ pub enum Cmd {
         input: usize,
         updates: Vec<InputUpdate>,
     },
+    /// Fill one program input from a source the server reads itself — a recipe
+    /// (`random:…`, `iota:N`) or a file of integer rows — each worker taking its
+    /// shard, so no row crosses the wire. `feed <prog> <in#> from <source>`.
+    Source {
+        prog: String,
+        input: usize,
+        source: String,
+    },
     /// Bind a trace's changes into `prog`'s positional `input`, delivered at
     /// each tick one epoch delayed — the write path for installed programs.
     Bind {
@@ -143,18 +155,25 @@ pub fn prepare(command: Cmd) -> Result<PreparedCommand, String> {
             bindings,
             program,
             explain,
+            applicative,
         } => {
-            if explain {
-                return Err(
-                    "load --explain is reserved; explanation is not implemented here yet".into(),
-                );
-            }
             let mut program = catch_unwind(AssertUnwindSafe(|| {
-                let statements = interactive::parse::pipe::parse(&program);
+                let statements = if applicative {
+                    interactive::parse::applicative::parse(&program)
+                } else {
+                    interactive::parse::pipe::parse(&program)
+                };
                 interactive::lower::lower_tree(statements)
             }))
             .map_err(panic_message)?;
             apply_bindings(&mut program, &bindings)?;
+            // The rewrite precedes optimization (its rules assume single-op Linears).
+            if let Some((arity, debug)) = explain {
+                let shapes: Vec<(usize, usize)> = program.root.imports.iter().map(|_| (arity, 0)).collect();
+                let options = interactive::explain::Options { debug_inspects: debug };
+                program = catch_unwind(AssertUnwindSafe(|| interactive::explain::explain_with(&program, &shapes, options)))
+                    .map_err(panic_message)?;
+            }
             program.optimize();
             ServerCommand::Install {
                 name: id_hint,
@@ -173,10 +192,7 @@ pub fn prepare(command: Cmd) -> Result<PreparedCommand, String> {
             },
         },
         Cmd::List => ServerCommand::List,
-        Cmd::Peek { name } => ServerCommand::Peek {
-            trace: name,
-            key: None,
-        },
+        Cmd::Peek { name, key } => ServerCommand::Peek { trace: name, key },
         Cmd::Tail { name } => return Ok(PreparedCommand::Tail { name }),
         Cmd::Stop { tail_reqid } => return Ok(PreparedCommand::Stop { tail_reqid }),
         Cmd::Feed {
@@ -203,6 +219,7 @@ pub fn prepare(command: Cmd) -> Result<PreparedCommand, String> {
             input,
             updates,
         },
+        Cmd::Source { prog, input, source } => ServerCommand::Load { prog, input, source },
         Cmd::Bind { trace, prog, input } => ServerCommand::Bind { trace, prog, input },
         Cmd::Unbind { trace, prog, input } => ServerCommand::Unbind { trace, prog, input },
         Cmd::Query { .. } => {
@@ -328,7 +345,7 @@ struct PendingLoad {
     reqid: ReqId,
     id_hint: String,
     bindings: BTreeMap<String, String>,
-    explain: bool,
+    explain: Option<(usize, bool)>,
     body: String,
     /// Set once the body exceeds the size gate; the rest of the body is
     /// swallowed (not stored) and `end-load` reports this error, so an
@@ -442,6 +459,7 @@ impl LineParser {
                             bindings: done.bindings,
                             program: done.body,
                             explain: done.explain,
+                            applicative: false,
                         }),
                     ));
                 }
@@ -529,7 +547,7 @@ enum ParseOutcome {
     BeginLoad {
         id_hint: String,
         bindings: BTreeMap<String, String>,
-        explain: bool,
+        explain: Option<(usize, bool)>,
     },
     BeginFeed {
         prog: String,
@@ -537,6 +555,20 @@ enum ParseOutcome {
     },
 }
 
+/// `explain=<arity>[,debug]`: the sources' key arity, and whether to tap every
+/// demand collection with an inspect.
+fn parse_explain(spec: &str) -> Result<(usize, bool), String> {
+    let (arity, debug) = match spec.split_once(',') {
+        Some((arity, "debug")) => (arity, true),
+        Some(_) => return Err(format!("load: explain=<arity>[,debug], got {spec:?}")),
+        None => (spec, false),
+    };
+    let arity = arity
+        .parse()
+        .map_err(|_| format!("load: explain= arity must be a number, got {arity:?}"))?;
+    Ok((arity, debug))
+}
+
 /// A `<value>` for `feed`: a comma-separated integer row → `Tuple`, `_` or
 /// empty → unit, else a closed scalar term (one whitespace-free token, e.g.
 /// `inject(2,tuple(3,4))`) evaluated to a constant. Term-parser panics are
@@ -601,32 +633,38 @@ fn parse_input_update(line: &str) -> Result<InputUpdate, String> {
 fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
     match cmd {
         "load" => {
-            // Syntax: `load <id-hint> [--explain] [name=binding ...] begin`
-            // The trailing `begin` switches the parser into body-collection
-            // mode; subsequent input lines are program text terminated by
-            // `<reqid> end-load`.
-            if args.is_empty() {
-                return ParseOutcome::Err(
-                    "load: expected `<id-hint> [--explain] [name=binding ...] begin`".into(),
-                );
-            }
-            if args.last() != Some(&"begin") {
-                return ParseOutcome::Err(
-                    "load: must end with `begin` (multi-line body required)".into(),
-                );
-            }
-            let id_hint = args[0].to_string();
-            let middle = &args[1..args.len() - 1];
+            // Two forms. `load <name> from <path> [options]` installs a program file
+            // (`.ddp` = pipe syntax, else applicative), read here on the session
+            // thread. `load <name> [options] begin` switches the parser into
+            // body-collection mode; subsequent lines are program text terminated by
+            // `<reqid> end-load`. Options: `explain=<arity>[,debug]`, `name=binding`.
+            const USAGE: &str = "load: expected `<name> from <path> [options]` or `<name> [options] begin`";
+            let Some((id_hint, rest)) = args.split_first() else {
+                return ParseOutcome::Err(USAGE.into());
+            };
+            let (path, options): (Option<&str>, &[&str]) = match rest {
+                ["from", path, options @ ..] => (Some(path), options),
+                [options @ .., "begin"] => (None, options),
+                _ => return ParseOutcome::Err(USAGE.into()),
+            };
             let mut bindings = BTreeMap::new();
-            let mut explain = false;
-            for tok in middle {
-                if *tok == "--explain" {
-                    explain = true;
+            let mut explain = None;
+            for tok in options {
+                if let Some(spec) = tok.strip_prefix("explain=") {
+                    explain = match parse_explain(spec) {
+                        Ok(explain) => Some(explain),
+                        Err(error) => return ParseOutcome::Err(error),
+                    };
                     continue;
                 }
+                if *tok == "--explain" {
+                    return ParseOutcome::Err(
+                        "load: `--explain` needs the sources' arity: `explain=<arity>[,debug]`".into(),
+                    );
+                }
                 let Some((k, v)) = tok.split_once('=') else {
                     return ParseOutcome::Err(format!(
-                        "load: argument {:?} must be `--explain` or `name=binding`",
+                        "load: argument {:?} must be `explain=<arity>[,debug]` or `name=binding`",
                         tok
                     ));
                 };
@@ -634,10 +672,25 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
                     return ParseOutcome::Err(format!("load: duplicate binding for {:?}", k));
                 }
             }
-            ParseOutcome::BeginLoad {
-                id_hint,
-                bindings,
-                explain,
+            let id_hint = id_hint.to_string();
+            match path {
+                // A file the server reads itself is not an upload, so the program-size
+                // gate (an intake defense) does not apply to it.
+                Some(path) => match std::fs::read_to_string(path) {
+                    Ok(program) => ParseOutcome::Cmd(Cmd::Load {
+                        id_hint,
+                        bindings,
+                        program,
+                        explain,
+                        applicative: !path.ends_with(".ddp"),
+                    }),
+                    Err(error) => ParseOutcome::Err(format!("load: cannot read {path:?}: {error}")),
+                },
+                None => ParseOutcome::BeginLoad {
+                    id_hint,
+                    bindings,
+                    explain,
+                },
             }
         }
         "query" => {
@@ -712,10 +765,21 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
                     input,
                 };
             }
+            // Syntax: `feed <prog> <in#> from <recipe-or-file>` — the server sources the rows.
+            if let [prog, input, "from", source] = args {
+                return match input.parse() {
+                    Ok(input) => ParseOutcome::Cmd(Cmd::Source {
+                        prog: (*prog).to_string(),
+                        input,
+                        source: (*source).to_string(),
+                    }),
+                    Err(_) => ParseOutcome::Err(format!("feed: <in#> must be a number, got {input:?}")),
+                };
+            }
             // Syntax: `feed <prog> <in#> <key> [val=<v>] [time=<t>] [diff=<d>]`
             // A `<v>`/`<key>` is a comma-separated integer row (`1,2` → tuple;
             // `_`/empty → unit) or a closed scalar term written without
-            // spaces (`inject(2,tuple(3,4))`), as in the ddir_server example.
+            // spaces (`inject(2,tuple(3,4))`).
             if args.len() < 3 {
                 return ParseOutcome::Err(
                     "feed: expected `<prog> <in#> <key> [val=<v>] [time=<t>] [diff=<d>]`".into(),
@@ -810,8 +874,16 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
         "peek" => match args {
             [name] => ParseOutcome::Cmd(Cmd::Peek {
                 name: (*name).to_string(),
+                key: None,
             }),
-            _ => ParseOutcome::Err("peek: expected `<name>`".into()),
+            [name, key] => match parse_value(key) {
+                Ok(key) => ParseOutcome::Cmd(Cmd::Peek {
+                    name: (*name).to_string(),
+                    key: Some(key),
+                }),
+                Err(error) => ParseOutcome::Err(format!("peek key: {}", error)),
+            },
+            _ => ParseOutcome::Err("peek: expected `<name> [key]`".into()),
         },
         "tail" => match args {
             [name] => ParseOutcome::Cmd(Cmd::Tail {
@@ -852,6 +924,21 @@ mod tests {
         out
     }
 
+    #[test]
+    fn feed_from_sources_rows_server_side() {
+        let mut p = LineParser::default();
+        let out = feed_all(&mut p, &["r1 feed world 0 from iota:3\n"]);
+        match &out[..] {
+            [(reqid, Ok(Cmd::Source { prog, input, source }))] => {
+                assert_eq!(reqid.as_str(), "r1");
+                assert_eq!((prog.as_str(), *input, source.as_str()), ("world", 0, "iota:3"));
+            }
+            other => panic!("unexpected parse: {other:?}"),
+        }
+        let out = feed_all(&mut p, &["r2 feed world x from iota:3\n"]);
+        assert!(matches!(&out[..], [(_, Err(_))]), "a bad input index must be rejected: {out:?}");
+    }
+
     #[test]
     fn simple_commands() {
         let mut p = LineParser::new();
@@ -868,7 +955,7 @@ mod tests {
                 target: DataflowRef::Id(3)
             })
         ));
-        assert!(matches!(got[3].1, Ok(Cmd::Peek { ref name }) if name == "foo"));
+        assert!(matches!(got[3].1, Ok(Cmd::Peek { ref name, key: None }) if name == "foo"));
     }
 
     #[test]
@@ -890,6 +977,7 @@ mod tests {
                 bindings,
                 program,
                 explain,
+                applicative,
             }) => {
                 assert_eq!(id_hint, "gen");
                 assert_eq!(
@@ -898,7 +986,8 @@ mod tests {
                 );
                 assert!(program.contains("import \"edges/v1\""));
                 assert!(program.contains("export \"reach\""));
-                assert!(!*explain);
+                assert!(explain.is_none());
+                assert!(!applicative);
             }
             _ => panic!("expected Load, got {:?}", got[0].1),
         }
@@ -923,18 +1012,43 @@ mod tests {
     }
 
     #[test]
-    fn load_explain() {
-        let mut p = LineParser::new();
+    fn load_from_file_reads_the_program_here() {
+        let dir = std::env::temp_dir().join(format!("ddir-load-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let ddp = dir.join("p.ddp");
+        std::fs::write(&ddp, "export \"r\" = input 0;\n").unwrap();
+        let ddir = dir.join("q.ddir");
+        std::fs::write(&ddir, "export \"r\" = INPUT 0;\n").unwrap();
+        let mut p = LineParser::default();
         let got = feed_all(&mut p, &[
-            "rE load reach --explain edges=random(seed=1,arity=2,range=10,count=2,churn=0) begin",
-            "export \"reach_out\" = import \"edges\";",
-            "rE end-load",
+            &format!("r1 load p from {} explain=2,debug e=random(seed=1,arity=2,range=8,count=12)\n", ddp.display()),
+            &format!("r2 load q from {}\n", ddir.display()),
+            "r3 load p from /nonexistent/p.ddp\n",
+            "r4 load p --explain begin\n",
         ]);
-        assert_eq!(got.len(), 1);
         match &got[0].1 {
-            Ok(Cmd::Load { explain, .. }) => assert!(*explain),
-            _ => panic!("expected explain Load, got {:?}", got[0].1),
+            Ok(Cmd::Load { id_hint, bindings, program, explain, applicative }) => {
+                assert_eq!(id_hint, "p");
+                assert_eq!(program, "export \"r\" = input 0;\n");
+                assert_eq!(*explain, Some((2, true)));
+                assert!(!applicative);
+                assert_eq!(bindings.get("e").map(String::as_str), Some("random(seed=1,arity=2,range=8,count=12)"));
+            }
+            other => panic!("unexpected {other:?}"),
         }
+        assert!(matches!(&got[1].1, Ok(Cmd::Load { applicative: true, explain: None, .. })));
+        assert!(matches!(&got[2].1, Err(e) if e.starts_with("load: cannot read")));
+        assert!(matches!(&got[3].1, Err(e) if e.contains("explain=<arity>")));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    #[test]
+    fn peek_takes_an_optional_key() {
+        let mut p = LineParser::default();
+        let got = feed_all(&mut p, &["r1 peek t 2,3\n", "r2 peek t\n", "r3 peek t oops(\n"]);
+        assert!(matches!(&got[0].1, Ok(Cmd::Peek { key: Some(_), .. })));
+        assert!(matches!(&got[1].1, Ok(Cmd::Peek { key: None, .. })));
+        assert!(matches!(&got[2].1, Err(_)));
     }
 
     #[test]
@@ -970,7 +1084,7 @@ mod tests {
         assert_eq!(got[1].0, "_2"); // tick 3
         assert!(matches!(got[1].1, Ok(Cmd::Tick { n: 3 })));
         assert_eq!(got[2].0, "rA"); // explicit reqid preserved
-        assert!(matches!(got[2].1, Ok(Cmd::Peek { ref name }) if name == "foo"));
+        assert!(matches!(got[2].1, Ok(Cmd::Peek { ref name, key: None }) if name == "foo"));
         assert_eq!(got[3].0, "_3"); // exit
         assert!(matches!(got[3].1, Ok(Cmd::Exit)));
     }
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 231848dfe..2d6af882c 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -7,6 +7,7 @@ use std::cell::RefCell;
 use std::collections::{HashMap, VecDeque};
 use std::rc::Rc;
 use std::sync::mpsc::{Receiver, Sender, TryRecvError};
+use std::time::Instant;
 
 use differential_dataflow::operators::arrange::ShutdownButton;
 use interactive::server::{Command as ServerCommand, OuterTime, RenderBackend, Server};
@@ -278,7 +279,7 @@ fn dispatch(
         Ok(PreparedCommand::Server(command)) => match command {
             ServerCommand::Install { name, program } => server
                 .install(worker, &name, &program)
-                .map(|()| format!("installed {:?}", name)),
+                .map(|()| format!("installed {:?} ({} ops)", name, program.op_count())),
             ServerCommand::Drop { name } => {
                 if tails.values().any(|tail| {
                     server
@@ -331,6 +332,10 @@ fn dispatch(
                     server.epoch()
                 )
             }),
+            // Collective: every worker sources its own shard.
+            ServerCommand::Load { prog, input, source } => server
+                .load(worker, &prog, input, &source)
+                .map(|rows| format!("loaded {rows} rows into {prog:?} input {input}")),
             ServerCommand::Bind { trace, prog, input } => server
                 .bind(worker, &trace, &prog, input)
                 .map(|()| format!("bound {:?} -> {:?} input {}", trace, prog, input)),
@@ -395,10 +400,11 @@ fn dispatch(
                 Err(error) => Err(error),
             },
             ServerCommand::Tick { n } => {
+                let timer = Instant::now();
                 for _ in 0..n {
                     tick(server, tails, worker);
                 }
-                Ok(format!("t={}", server.epoch()))
+                Ok(format!("t={} elapsed={:.2?}", server.epoch(), timer.elapsed()))
             }
             ServerCommand::Exit => {
                 *shutdown = connection_id == 0;
diff --git a/interactive/server/src/main.rs b/interactive/server/src/main.rs
index 268807321..d568e8099 100644
--- a/interactive/server/src/main.rs
+++ b/interactive/server/src/main.rs
@@ -25,6 +25,12 @@ mod cmd;
 #[path = "loop_.rs"]
 mod control_loop;
 
+use mimalloc::MiMalloc;
+
+/// The allocator the retired example driver ran on; the arrangement-heavy paths lean on it.
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
 use std::io::{BufRead, BufReader, Write};
 use std::net::{TcpListener, TcpStream};
 use std::sync::atomic::{AtomicU64, Ordering};
diff --git a/interactive/server/tests/multiworker.rs b/interactive/server/tests/multiworker.rs
index 539ff94ea..3ff80f4eb 100644
--- a/interactive/server/tests/multiworker.rs
+++ b/interactive/server/tests/multiworker.rs
@@ -174,6 +174,27 @@ fn assert_backend(backend: &str) {
         Some("time=1 diff=1 key=Tuple([Int(8)]) val=Tuple([Int(10)])")
     );
 
+    // Server-side sourcing: every worker feeds its shard of the recipe, so the
+    // union is the source exactly once however many workers there are. Its own
+    // program, because a recipe's rows carry a unit value and `world`'s carry
+    // an integer, and one input holds one shape.
+    request(&mut writer, &mut reader, "r9", "r9 stop r6\n");
+    request(
+        &mut writer,
+        &mut reader,
+        "r10",
+        "r10 load counted begin\nlet rows = input 0;\nexport \"counted\" = rows;\nr10 end-load\n",
+    );
+    request(&mut writer, &mut reader, "r11", "r11 feed counted 0 from iota:5\n");
+    request(&mut writer, &mut reader, "r12", "r12 tick\n");
+    let rows = request(&mut writer, &mut reader, "r13", "r13 peek counted\n");
+    assert_eq!(
+        rows,
+        (0..5)
+            .map(|n| format!("diff=1 key=Tuple([Int({n})]) val=Tuple([])"))
+            .collect::<Vec<_>>()
+    );
+
     drop(reader);
     drop(writer);
     server.stop();
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 0b48a19ef..e149c4211 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -447,129 +447,3 @@ pub fn render_tree_rows<'s>(
         })
         .collect()
 }
-
-/// Evaluate `program` on explicit inputs via the **corgi** backend (mirrors [`crate::backend::vec::evaluate`]).
-///
-/// Inputs/exports cross the iterative-scope boundary as Vec rows (which support refinement
-/// enter/leave); corgi containers exist only INSIDE the dynamic scope, where `Enter`/`Leave` are the
-/// same-Time identity. The `ToCorgi`/`FromCorgi` unaries are the only row↔corgi conversions.
-pub fn evaluate(
-    program: &st::Program,
-    inputs: &[Vec<(Row, Row)>],
-) -> std::collections::BTreeMap<String, Vec<((Row, Row), Diff)>> {
-    evaluate_with_workers(program, inputs, 1)
-}
-
-/// [`evaluate`] on `workers` worker threads in one process.
-///
-/// The answer must not depend on `workers`: the exchange places each key on one worker, every
-/// operator is key-local from there, and each worker's captured output is a disjoint share of the
-/// same collection, so summing the shares reproduces the single-worker result exactly. That
-/// invariant is the correctness statement for [`CorgiPact`], and the backend gate checks it by
-/// running each program at several worker counts.
-///
-/// Input rows are dealt round-robin across workers (`row index % peers`) so each row is introduced
-/// exactly once globally — where they enter is irrelevant, since `arrange` re-places them.
-pub fn evaluate_with_workers(
-    program: &st::Program,
-    inputs: &[Vec<(Row, Row)>],
-    workers: usize,
-) -> std::collections::BTreeMap<String, Vec<((Row, Row), Diff)>> {
-    evaluate_with_config(program, inputs, timely::Config::process(workers))
-}
-
-/// [`evaluate_with_workers`] with the timely configuration named outright.
-///
-/// The configuration decides which half of the exchange gets exercised.
-/// `Config::process(n)` hands containers between worker threads as typed values — the
-/// [`CorgiDistributor`](crate::corgi::exchange::CorgiDistributor)'s partition runs, but no bytes
-/// are produced. `CommunicationConfig::ProcessBinary(n)` puts the same threads behind serializing
-/// channels, so every container makes the round trip through
-/// [the wire format](crate::corgi::bytes) — the multi-*process* path, without the processes.
-pub fn evaluate_with_config(
-    program: &st::Program,
-    inputs: &[Vec<(Row, Row)>],
-    config: timely::Config,
-) -> std::collections::BTreeMap<String, Vec<((Row, Row), Diff)>> {
-    use std::collections::BTreeMap;
-    use std::sync::{Arc, Mutex, mpsc::channel};
-    use timely::dataflow::operators::core::capture::{Capture, Event};
-    use differential_dataflow::input::Input;
-    use differential_dataflow::dynamic::pointstamp::PointStamp;
-
-    let names: Vec<String> = program.root.exports.iter().map(|e| e.name.clone()).collect();
-    let mut txs = Vec::new();
-    let mut rxs = Vec::new();
-    for _ in &names {
-        let (tx, rx) = channel::<Event<u64, Vec<((Row, Row), u64, Diff)>>>();
-        txs.push(tx);
-        rxs.push(rx);
-    }
-    // Every worker captures into the same per-export channel. `Sender` is `Send` but not `Sync`,
-    // and timely's worker closure must be both, so the senders travel behind a mutex and each
-    // worker clones its own out of it once, at construction.
-    let txs = Arc::new(Mutex::new(txs));
-
-    let program = program.clone();
-    let inputs: Vec<Vec<(Row, Row)>> = inputs.to_vec();
-    let guards = timely::execute(config, move |worker| {
-        let (index, peers) = (worker.index(), worker.peers());
-        let txs: Vec<_> = txs.lock().expect("capture senders").iter().cloned().collect();
-        let mut handles = worker.dataflow::<u64, _, _>(|scope| {
-            let mut handles = Vec::new();
-            let mut collections = Vec::new();
-            for _ in 0..inputs.len() {
-                let (h, c) = scope.new_collection::<(Row, Row), Diff>();
-                handles.push(h);
-                collections.push(c);
-            }
-            let exports = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
-                // Enter row collections (refinement); rows convert to corgi containers and
-                // back inside `render_tree_rows`.
-                let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect();
-                let root_imports: Vec<_> = program
-                    .root
-                    .imports
-                    .iter()
-                    .map(|imp| match &imp.from {
-                        st::Source::Input(n) => entered[*n].clone(),
-                        other => panic!("corgi evaluate: unsupported source {other:?}"),
-                    })
-                    .collect();
-                render_tree_rows(&program.root, inner.clone(), 0, root_imports)
-                    .into_iter()
-                    .map(|rows| rows.leave(scope))
-                    .collect::<Vec<_>>()
-            });
-            for (col, tx) in exports.into_iter().zip(txs) {
-                col.inner.capture_into(tx);
-            }
-            handles
-        });
-        for (i, rows) in inputs.iter().enumerate() {
-            for r in rows.iter().skip(index).step_by(peers) {
-                handles[i].update(r.clone(), 1);
-            }
-        }
-    })
-    .expect("corgi evaluate: worker startup");
-    // Join the workers before draining, so every capture has been sent and every cloned sender
-    // dropped; otherwise the receive loop below would block on a channel that is still open.
-    guards.join();
-
-    names
-        .into_iter()
-        .zip(rxs)
-        .map(|(name, rx)| {
-            let mut acc: BTreeMap<(Row, Row), Diff> = BTreeMap::new();
-            for event in rx {
-                if let Event::Messages(_, data) = event {
-                    for ((k, v), _, d) in data {
-                        *acc.entry((k, v)).or_insert(0) += d;
-                    }
-                }
-            }
-            (name, acc.into_iter().filter(|(_, d)| *d != 0).collect())
-        })
-        .collect()
-}
diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs
index 2a5001cea..c01cd52b7 100644
--- a/interactive/src/backend/vec.rs
+++ b/interactive/src/backend/vec.rs
@@ -169,80 +169,3 @@ pub fn render_tree<'s>(
 ) -> Vec<Col<'s>> {
     crate::backend::render_tree::<VecBackend>(s, scope, depth, imports)
 }
-
-/// Evaluate `program` on explicit inputs: single worker, in-process, every
-/// export returned as its consolidated final contents (rows with non-zero
-/// net multiplicity).
-///
-/// `inputs[i]` provides the rows of positional input `i`, each with
-/// multiplicity +1; all rows are introduced at time 0 and the computation
-/// runs to quiescence. This is the data-in/data-out entry point that tests
-/// and tools build on — e.g. feeding an explanation's demand-set back
-/// through the original program to check the queried output reproduces.
-pub fn evaluate(
-    program: &st::Program,
-    inputs: &[Vec<(Row, Row)>],
-) -> std::collections::BTreeMap<String, Vec<((Row, Row), Diff)>> {
-    use std::sync::mpsc::channel;
-    use timely::dataflow::operators::core::capture::{Capture, Event};
-    use differential_dataflow::input::Input;
-
-    let names: Vec<String> = program.root.exports.iter().map(|e| e.name.clone()).collect();
-    let mut txs = Vec::with_capacity(names.len());
-    let mut rxs = Vec::with_capacity(names.len());
-    for _ in &names {
-        let (tx, rx) = channel::<Event<u64, Vec<((Row, Row), u64, Diff)>>>();
-        txs.push(tx);
-        rxs.push(rx);
-    }
-
-    let program = program.clone();
-    let inputs: Vec<Vec<(Row, Row)>> = inputs.to_vec();
-    timely::execute_directly(move |worker| {
-        let mut handles = worker.dataflow::<u64, _, _>(|scope| {
-            let mut handles = Vec::new();
-            let mut collections = Vec::new();
-            for _ in 0..inputs.len() {
-                let (h, c) = scope.new_collection::<(Row, Row), Diff>();
-                handles.push(h);
-                collections.push(c);
-            }
-            let exports = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
-                let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect();
-                let root_imports: Vec<_> = program.root.imports.iter().map(|imp| match &imp.from {
-                    st::Source::Input(n) => entered[*n].clone(),
-                    other => panic!("evaluate: unsupported source {:?}", other),
-                }).collect();
-                let exports = render_tree(&program.root, inner.clone(), 0, root_imports);
-                exports.into_iter().map(|c| c.leave(scope)).collect::<Vec<_>>()
-            });
-            for (col, tx) in exports.into_iter().zip(txs) {
-                col.inner.capture_into(tx);
-            }
-            handles
-        });
-        for (i, rows) in inputs.iter().enumerate() {
-            for r in rows {
-                handles[i].update(r.clone(), 1);
-            }
-        }
-        // Dropping the handles closes the inputs; `execute_directly` then
-        // steps the worker until the dataflow completes.
-    });
-
-    names
-        .into_iter()
-        .zip(rxs)
-        .map(|(name, rx)| {
-            let mut acc: std::collections::BTreeMap<(Row, Row), Diff> = std::collections::BTreeMap::new();
-            for event in rx {
-                if let Event::Messages(_, data) = event {
-                    for ((k, v), _, d) in data {
-                        *acc.entry((k, v)).or_insert(0) += d;
-                    }
-                }
-            }
-            (name, acc.into_iter().filter(|(_, d)| *d != 0).collect())
-        })
-        .collect()
-}
diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs
index 434539b71..3b54e17ac 100644
--- a/interactive/src/corgi/logic.rs
+++ b/interactive/src/corgi/logic.rs
@@ -356,6 +356,12 @@ pub fn compile(
             })
         }
         Term::If { cond, then, els } => {
+            // A literal condition takes one branch, so only that branch needs a shape. This is
+            // the `if(1, $1, 0)` idiom: it carries a whole value as ONE field where a bare `$1`
+            // would splice, and its dead branch need not agree in shape.
+            if let Term::Int(c) = **cond {
+                return compile(if c != 0 { then } else { els }, b, env, env_shapes, anchor, expected);
+            }
             // `Select` blends per row and is shape-generic; the branches must share one shape,
             // and a branch that cannot fix its own (a bare `None`) takes the other's.
             let (ts, es) = branch_shapes(then, els, env_shapes, expected)?;
diff --git a/interactive/src/lib.rs b/interactive/src/lib.rs
index ba3b3a95b..cd530d93f 100644
--- a/interactive/src/lib.rs
+++ b/interactive/src/lib.rs
@@ -10,45 +10,6 @@ pub mod server;
 // the Value reverse model implements `time_le`/`strip` directly over the nested
 // chain tuple, so explain no longer needs it.
 
-use std::collections::BTreeSet;
-
-use parse::{Stmt, Expr};
-
-/// Survey a program's external sources: the count of positional inputs (one
-/// more than the largest `input N` index, zero if none appear) and the set of
-/// names referenced by `import "name"`. Two kinds because `import` does not yet
-/// subsume `input` — see `ir::Node::Import`; this returns one number when that
-/// cutover happens.
-pub fn survey_sources(stmts: &[Stmt]) -> (usize, BTreeSet<String>) {
-    let mut positional = 0usize;
-    let mut imports = BTreeSet::new();
-    walk_stmts(stmts, &mut positional, &mut imports);
-    (positional, imports)
-}
-
-fn walk_stmts(stmts: &[Stmt], positional: &mut usize, imports: &mut BTreeSet<String>) {
-    for stmt in stmts {
-        match stmt {
-            Stmt::Let(_, expr) | Stmt::Var(_, expr) | Stmt::Export(_, expr) => walk_expr(expr, positional, imports),
-            Stmt::Scope(_, body) => walk_stmts(body, positional, imports),
-        }
-    }
-}
-
-fn walk_expr(expr: &Expr, positional: &mut usize, imports: &mut BTreeSet<String>) {
-    match expr {
-        Expr::Input(n) => { *positional = (*positional).max(n + 1); },
-        Expr::Import(name) => { imports.insert(name.clone()); },
-        Expr::Map(e, _) | Expr::Negate(e) | Expr::Arrange(e)
-            | Expr::EnterAt(e, _) | Expr::LiftIter(e) | Expr::Filter(e, _)
-            | Expr::FlatMap(e, _)
-            | Expr::Reduce(e, _) | Expr::Inspect(e, _) => walk_expr(e, positional, imports),
-        Expr::Join(l, r, _) => { walk_expr(l, positional, imports); walk_expr(r, positional, imports); },
-        Expr::Concat(es) => { for e in es { walk_expr(e, positional, imports); } },
-        Expr::Name(_) | Expr::Qualified(_, _) => {},
-    }
-}
-
 /// Load a program source file.
 pub fn load_program(path: &str) -> String {
     std::fs::read_to_string(path).unwrap_or_else(|e| panic!("Cannot read {}: {}", path, e))
diff --git a/interactive/src/server.rs b/interactive/src/server.rs
index 4ccdb9b98..73784927e 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -32,6 +32,9 @@
 //!   for teardown.
 //! - **feed** stages an input update at a chosen time (default: the current
 //!   epoch) via `update_at`, so inputs can be scheduled into the future.
+//! - **load** fills an input in bulk from a recipe or a file, each worker
+//!   feeding its own shard; a churning recipe then changes the input on every
+//!   tick, which is how a program is run under standing change.
 //! - **tick** advances all inputs to the next epoch, runs to quiescence, then
 //!   lets every trace compact (an importer's own handle holds the shared
 //!   `TraceBox` back to what it still needs).
@@ -242,6 +245,14 @@ pub enum Command {
         input: usize,
         updates: Vec<InputUpdate>,
     },
+    /// Fill positional `input` of `prog` from `source` — a recipe name or a
+    /// file path — at the current epoch. Collective: every worker feeds its
+    /// own shard of the rows (see [`Server::load`]).
+    Load {
+        prog: String,
+        input: usize,
+        source: String,
+    },
     /// Close `n` epochs, running to quiescence after each one.
     Tick { n: u64 },
     /// Drop the named program.
@@ -284,8 +295,10 @@ struct Installed {
     /// [`Origin`]. Generated/clock entries advance and drop like any program but
     /// are not writable by `feed`.
     origin: Origin,
-    /// Generator recipe and next row to retract, for changing random sources.
-    generator: Option<(Recipe, u64)>,
+    /// Per input: the recipe whose rows it holds and the next row to retract,
+    /// for inputs that churn each `tick` (a generated `random:` source's own
+    /// input, or a program input bulk-loaded from such a recipe).
+    generators: HashMap<usize, (Recipe, u64)>,
 }
 
 /// A stable, transport-friendly description of one installed dataflow.
@@ -563,7 +576,7 @@ impl Server {
                 dataflow_id,
                 probe,
                 origin: Origin::Program,
-                generator: None,
+                generators: HashMap::new(),
             },
         );
         Ok(())
@@ -607,7 +620,7 @@ impl Server {
                 dataflow_id,
                 probe,
                 origin: Origin::Generated,
-                generator: Some((recipe, 0)),
+                generators: HashMap::from([(0usize, (recipe, 0u64))]),
             },
         );
     }
@@ -645,7 +658,7 @@ impl Server {
                 dataflow_id,
                 probe,
                 origin: Origin::Clock,
-                generator: None,
+                generators: HashMap::new(),
             },
         );
     }
@@ -696,6 +709,78 @@ impl Server {
         Ok(())
     }
 
+    /// Bulk-load rows into positional `input` of `prog` at the current epoch.
+    ///
+    /// `source` is either a recipe (`random:…`, `iota:N` — the same names an
+    /// `import` accepts) or the path of a text file with one row per line of
+    /// whitespace-separated integers (`(Tuple[ints] ; ())`). Collective: every
+    /// worker must call this, and each feeds only its shard (`row % peers ==
+    /// index`) through its own handle, so the union is the source exactly once
+    /// and the exchange places each row on its key's owner.
+    ///
+    /// A `random:` recipe with `churn=C` keeps churning: every later `tick`
+    /// retracts the next `C` rows of the window and adds `C` fresh ones, the
+    /// standing-change regime a program is benchmarked under. Returns the
+    /// number of rows in the source (across all workers).
+    ///
+    /// A file is validated in full on every worker before any of its rows is
+    /// applied: a malformed line fails the load on every worker, so a source
+    /// is fed whole or not at all, and the workers agree because they read the
+    /// same file (the source must be visible, and identical, to all of them).
+    pub fn load(
+        &mut self,
+        worker: &Worker,
+        prog: &str,
+        input: usize,
+        source: &str,
+    ) -> Result<u64, String> {
+        let time = self.epoch;
+        self.validate_feed(prog, input, time)?;
+        let (index, peers) = (worker.index(), worker.peers());
+        let mine = |e: u64| (e as usize) % peers == index;
+        let recipe = Recipe::parse(source);
+        let (total, rows): (u64, Vec<(Value, Value)>) = match recipe {
+            Some(recipe) => (
+                recipe.rows_len(),
+                (0..recipe.rows_len()).filter(|e| mine(*e)).map(|e| recipe.row(e)).collect(),
+            ),
+            None => {
+                let text = std::fs::read_to_string(source)
+                    .map_err(|e| format!("load: cannot read {:?}: {}", source, e))?;
+                let mut total = 0;
+                let mut rows = Vec::new();
+                // Every line is parsed, whichever worker's shard it falls in: a
+                // malformed line must fail the load everywhere, or the other
+                // workers would apply their shards and the client would see
+                // "loaded" over partial data.
+                for (e, line) in text.lines().filter(|l| !l.trim().is_empty()).enumerate() {
+                    total += 1;
+                    let fields = line
+                        .split_whitespace()
+                        .map(|t| t.parse::<i64>().map(Value::Int))
+                        .collect::<Result<Vec<_>, _>>()
+                        .map_err(|_| format!("load: line {} of {:?} is not a row of integers: {:?}", e + 1, source, line))?;
+                    if mine(e as u64) {
+                        rows.push((Value::Tuple(fields), Value::unit()));
+                    }
+                }
+                (total, rows)
+            }
+        };
+        let installed = self
+            .programs
+            .get_mut(&canonical_source_name(prog))
+            .expect("load target was prevalidated");
+        if let Some(recipe @ Recipe::Random { churn: 1.., .. }) = recipe {
+            installed.generators.insert(input, (recipe, 0));
+        }
+        let handle = installed.inputs.get_mut(&input).expect("load input was prevalidated");
+        for row in rows {
+            handle.update_at(row, time, 1);
+        }
+        Ok(total)
+    }
+
     /// Check everything about an input target that can fail without changing
     /// its handle. Both singular and batched feeds validate before applying.
     fn validate_feed(&self, prog: &str, input: usize, time: OuterTime) -> Result<(), String> {
@@ -1058,10 +1143,10 @@ impl Server {
             }
             // A random source denotes an infinite deterministic row stream.
             // Each tick replaces `churn` members of its fixed-size window.
-            if let Some((recipe, cursor)) = &mut installed.generator {
+            for (input, (recipe, cursor)) in installed.generators.iter_mut() {
                 let recipe = *recipe;
                 if let Recipe::Random { edges, churn, .. } = recipe {
-                    if let Some(h) = installed.inputs.get_mut(&0) {
+                    if let Some(h) = installed.inputs.get_mut(input) {
                         for _ in 0..churn {
                             let old = *cursor;
                             let new = edges + *cursor;
@@ -1185,3 +1270,57 @@ impl Default for Server {
         Server::new()
     }
 }
+
+/// Evaluate `program` on explicit inputs through a throwaway server: every
+/// export's consolidated final contents (rows with non-zero net multiplicity),
+/// by name.
+///
+/// This is the data-in/data-out entry point the test suites build on, and it
+/// takes the same path a live install does — `install`, one `feed` of each
+/// positional input (`inputs[i]` holds input `i`'s rows, each at multiplicity
+/// +1, dealt round-robin across the workers), one `tick`, then a `snapshot` of
+/// each export. `config` picks the worker group: `Config::process(n)` hands
+/// exchanged containers between threads as typed values, while
+/// `CommunicationConfig::ProcessBinary(n)` sends every one through the wire
+/// format. The answer must not depend on either choice.
+pub fn evaluate(
+    backend: RenderBackend,
+    config: timely::Config,
+    program: &st::Program,
+    inputs: &[Vec<(Value, Value)>],
+) -> std::collections::BTreeMap<String, Vec<((Value, Value), Diff)>> {
+    let program = program.clone();
+    let inputs = inputs.to_vec();
+    let guards = timely::execute(config, move |worker| {
+        let mut server = Server::with_backend(backend);
+        server.install(worker, "evaluate", &program).expect("evaluate: install");
+        let has_input = |i: usize| program.root.imports.iter().any(|imp| matches!(imp.from, st::Source::Input(n) if n == i));
+        for (input, rows) in inputs.iter().enumerate().filter(|(i, _)| has_input(*i)) {
+            let shard = rows
+                .iter()
+                .skip(worker.index())
+                .step_by(worker.peers())
+                .map(|(key, val)| InputUpdate { key: key.clone(), val: val.clone(), diff: 1 })
+                .collect();
+            server.feed_batch("evaluate", input, shard).expect("evaluate: feed");
+        }
+        server.tick(worker);
+        // `snapshot` gathers to worker 0; the other workers' results are empty.
+        program
+            .root
+            .exports
+            .iter()
+            .map(|e| {
+                let rows = server.snapshot(worker, &e.name).expect("evaluate: snapshot");
+                (e.name.clone(), rows.into_iter().map(|(k, v, d)| ((k, v), d)).collect())
+            })
+            .collect()
+    })
+    .expect("evaluate: worker startup");
+    guards
+        .join()
+        .into_iter()
+        .next()
+        .expect("evaluate: worker 0")
+        .expect("evaluate: worker 0 returned")
+}
diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs
index b3ca27b78..914a78923 100644
--- a/interactive/tests/corgi_backend.rs
+++ b/interactive/tests/corgi_backend.rs
@@ -1,8 +1,11 @@
 //! The corgi backend's correctness gate: each canonical `.ddp` program must evaluate
 //! identically through the corgi backend and the reference vec backend.
+//!
+//! Every evaluation goes through the server (`server::evaluate`): install, feed, tick,
+//! snapshot — the same path a live install takes.
 
-use interactive::backend::{corgi, vec};
 use interactive::ir::Value;
+use interactive::server::{evaluate, RenderBackend};
 use interactive::{lower, parse};
 
 fn tup(fields: &[i64]) -> Value {
@@ -42,6 +45,7 @@ fn inputs_for(prog: &str) -> Vec<Vec<(Value, Value)>> {
         // sum_skew: any keyed pairs — the skew is in the program, not the data.
         "sum_skew" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21], &[3, 30]])],
         "case_ops" => vec![rows(&[&[1, 10], &[2, 20], &[3, 14], &[3, 30]])],
+        "if_literal" => vec![rows(&[&[1, 10], &[1, 20], &[2, 30]])],
         // pair_keys: composite keys with overlap, fanout, and one-sided keys on both sides.
         "pair_keys" => vec![
             rows(&[&[1, 1, 10], &[1, 2, 20], &[2, 1, 30], &[2, 1, 31], &[9, 9, 90]]),
@@ -77,14 +81,14 @@ fn assert_backends_agree(prog: &str) {
     let mut tree = lower::lower_tree(parse::pipe::parse(&src));
     tree.optimize();
     let inputs = inputs_for(prog);
-    let want = vec::evaluate(&tree, &inputs);
+    let want = evaluate(RenderBackend::Vec, timely::Config::process(1), &tree, &inputs);
     // At every worker count: the exchange places each key on one worker and every operator is
     // key-local from there, so the answer must not depend on how many workers ran it. 3 is in the
     // list on purpose — it is not a power of two, so it takes the modulus path rather than the
     // mask, and it cannot divide these inputs evenly.
     for workers in [1, 2, 3, 4] {
         assert_eq!(
-            corgi::evaluate_with_workers(&tree, &inputs, workers),
+            evaluate(RenderBackend::Corgi, timely::Config::process(workers), &tree, &inputs),
             want,
             "corgi backend at {workers} worker(s) disagrees with the vec backend on {prog}",
         );
@@ -93,7 +97,7 @@ fn assert_backends_agree(prog: &str) {
     // round trip through the wire format. This is the multi-process path: `Config::process` above
     // hands containers between threads as typed values and never encodes a byte.
     assert_eq!(
-        corgi::evaluate_with_config(&tree, &inputs, serializing(3)),
+        evaluate(RenderBackend::Corgi, serializing(3), &tree, &inputs),
         want,
         "corgi backend over serializing channels disagrees with the vec backend on {prog}",
     );
@@ -120,6 +124,7 @@ fn serializing(n: usize) -> timely::Config {
 #[test] fn empty_batch() { assert_backends_agree("empty_batch"); }
 #[test] fn sum_skew() { assert_backends_agree("sum_skew"); }
 #[test] fn case_ops() { assert_backends_agree("case_ops"); }
+#[test] fn if_literal() { assert_backends_agree("if_literal"); }
 #[test] fn tour() { assert_backends_agree("tour"); }
 #[test] fn pair_keys() { assert_backends_agree("pair_keys"); }
 #[test] fn signed_min() { assert_backends_agree("signed_min"); }
diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs
index cf55f1ade..226466635 100644
--- a/interactive/tests/explain.rs
+++ b/interactive/tests/explain.rs
@@ -1,5 +1,5 @@
 //! End-to-end semantic tests for the explanation rewrite, built on
-//! `backend::vec::evaluate` (explicit inputs in, every export out). The
+//! `server::evaluate` (explicit inputs in, every export out). The
 //! sufficiency properties use that row execution to evaluate the rewrite; a
 //! final section cross-checks the behavior shared by the row and corgi
 //! implementations for these programs.
@@ -11,12 +11,12 @@
 //! regenerate the queried output row. Tests marked `#[ignore]` are heavier sweeps meant for
 //! `cargo test --release -- --ignored`.
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 
-use interactive::backend::corgi::evaluate as corgi_evaluate;
-use interactive::backend::vec::{evaluate, Row};
+use interactive::backend::vec::Row;
 use interactive::ir::Value;
 use interactive::scope_ir::Program;
+use interactive::server::{self, RenderBackend};
 use interactive::{explain, lower, parse};
 
 /// SCC with the scc edge-set itself as the result, so individual output
@@ -101,6 +101,12 @@ fn row(fields: &[i64]) -> Row {
     Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect())
 }
 
+/// Run `p` on `inputs` through the server on the vec backend: every export's
+/// consolidated rows, by name.
+fn evaluate(p: &Program, inputs: &[Vec<(Row, Row)>]) -> BTreeMap<String, Vec<((Row, Row), i64)>> {
+    server::evaluate(RenderBackend::Vec, timely::Config::process(1), p, inputs)
+}
+
 /// Run `p` on `inputs` and return one export's rows (asserting positive
 /// multiplicities — a set-like result).
 fn export_rows(p: &Program, inputs: &[Vec<(Row, Row)>], export: &str) -> BTreeSet<(Row, Row)> {
@@ -508,7 +514,7 @@ fn assert_explained_backends_agree(
     ex_inputs.push(query_rows(queries));
 
     let by_vec = evaluate(&ex, &ex_inputs);
-    let by_corgi = corgi_evaluate(&ex, &ex_inputs);
+    let by_corgi = server::evaluate(RenderBackend::Corgi, timely::Config::process(1), &ex, &ex_inputs);
     assert_eq!(
         by_vec.keys().collect::<Vec<_>>(),
         by_corgi.keys().collect::<Vec<_>>(),
diff --git a/interactive/tests/programs/if_literal.ddp b/interactive/tests/programs/if_literal.ddp
new file mode 100644
index 000000000..eebaa524c
--- /dev/null
+++ b/interactive/tests/programs/if_literal.ddp
@@ -0,0 +1,5 @@
+-- Pins the literal-condition `if`. `if(1, $1, 0)` carries a collected List as
+-- ONE field (a bare `$1` would splice its elements), and corgi compiles only
+-- the live branch, so the dead branch need not agree in shape.
+let lists = input 0 | key($0[0] ; $0[1]) | collect;
+export "result" = lists | map($0[0] ; if(1, $1, 0), if(0, 7, $0[0]));
diff --git a/interactive/tests/server_load.rs b/interactive/tests/server_load.rs
new file mode 100644
index 000000000..b4eabd671
--- /dev/null
+++ b/interactive/tests/server_load.rs
@@ -0,0 +1,68 @@
+//! `Server::load` from a file is all-or-nothing across the worker group: every
+//! worker validates the whole file before applying its shard, so a malformed
+//! line fails the load on every worker and no row of the file is fed anywhere.
+//! (With per-shard validation, two workers and the lines `1, bad, 3, 4` gave
+//! "loaded 4 rows" on worker 0 with only `1` and `3` present.)
+
+use std::collections::BTreeSet;
+
+use interactive::ir::Value;
+use interactive::server::Server;
+use interactive::{lower, parse};
+
+fn tup(fields: &[i64]) -> Value {
+    Value::Tuple(fields.iter().copied().map(Value::Int).collect())
+}
+
+/// Run `load` from a file with `body` on `workers` workers; return each
+/// worker's result and the rows worker 0 snapshots afterwards.
+fn load_on(workers: usize, body: &str) -> (Vec<Result<u64, String>>, Vec<Value>) {
+    let dir = std::env::temp_dir().join(format!("ddir-server-load-{}-{}", std::process::id(), workers));
+    std::fs::create_dir_all(&dir).unwrap();
+    let path = dir.join(format!("{}.txt", body.len()));
+    std::fs::write(&path, body).unwrap();
+    let path = path.to_string_lossy().into_owned();
+
+    let guards = timely::execute(timely::Config::process(workers), move |worker| {
+        let mut program = lower::lower_tree(parse::pipe::parse(
+            "let rows = input 0; export \"rows\" = rows;",
+        ));
+        program.optimize();
+        let mut server = Server::new();
+        server.install(worker, "world", &program).unwrap();
+        let outcome = server.load(worker, "world", 0, &path);
+        server.tick(worker);
+        let rows = server
+            .snapshot(worker, "rows")
+            .unwrap()
+            .into_iter()
+            .map(|(key, _, _)| key)
+            .collect::<Vec<_>>();
+        (outcome, rows)
+    })
+    .unwrap();
+    let mut results = guards.join().into_iter().map(Result::unwrap);
+    let (first_outcome, rows) = results.next().unwrap();
+    let mut outcomes = vec![first_outcome];
+    outcomes.extend(results.map(|(outcome, _)| outcome));
+    (outcomes, rows)
+}
+
+#[test]
+fn a_malformed_line_fails_the_load_on_every_worker_and_feeds_nothing() {
+    let (outcomes, rows) = load_on(2, "1\nbad\n3\n4\n");
+    for outcome in &outcomes {
+        let error = outcome.as_ref().expect_err("a malformed line must fail the load");
+        assert!(error.contains("line 2"), "{error}");
+    }
+    assert!(rows.is_empty(), "no row of a rejected file may be fed: {rows:?}");
+}
+
+#[test]
+fn a_well_formed_file_is_fed_exactly_once_across_workers() {
+    let (outcomes, rows) = load_on(3, "1\n\n2 20\n3\n4\n");
+    assert!(outcomes.iter().all(|o| *o == Ok(4)), "{outcomes:?}");
+    let want: BTreeSet<Value> = [tup(&[1]), tup(&[2, 20]), tup(&[3]), tup(&[4])].into_iter().collect();
+    assert_eq!(rows.iter().cloned().collect::<BTreeSet<_>>(), want);
+    assert_eq!(rows.len(), 4, "each row once: {rows:?}");
+}