feat(incremental): a scheduler runs sub-circuits on a worker pool; the aggregate and grid stages placed on two units

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

context/code/improvise-incremental.md

@@ -504,3 +504,51 @@ change log, no stems. These are the readings taken.
 - **The two-second limit decided the shape of the tests** more than the design did: the corpus
   differential over a scanning source is 2.5s untrimmed, so it follows `corpus_eager`'s trimming, and
   each random test is its own binary.
+
+## Scheduling: sub-circuits on a worker pool (user ruling 2026-10-11)
+
+Multi-threading here is **scheduling**. Each sub-circuit stays synchronous and single-threaded
+(`zset.md`, "Threads"), the only thing that crosses threads is an edge (`zset.md`, "Edges"), and
+a small executor decides which thread runs which sub-circuit next. This is a ruling recorded here,
+not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` compile for wasm32
+(a `Mutex` and a `Condvar`), and `Pool`, the only thing that spawns a thread, is
+`cfg(not(target_arch = "wasm32"))`. Nothing here runs on wasm.
+
+1. **A unit is a sub-circuit with the edge ends around it** (`Unit`: `turn(budget)` and
+   `neighbours()`). Placement is manual: which nodes sit in which unit is how the graph was built,
+   and `staged::PLACEMENT` names it for the example. Two node-building functions (`aggregate_nodes`,
+   `grid_nodes`) build the same nodes into one circuit (`single_circuit`, the reference) or into
+   two.
+2. **Runnable means queued.** `Executor::kick(id)` queues a unit once. A unit is kicked by the
+   host (`Staged::submit`), and by a neighbour whose turn moved a frame (`Turn::progress`, which a
+   unit derives from its edges' `traffic()`). A kick reaches both directions, because a request
+   flows back along an edge. Progress is a count of frames, so it ends: two neighbours cannot kick
+   each other for ever.
+3. **The ownership token is the unit itself.** A claimed unit is taken out of its slot, so a second
+   worker cannot reach it. A kick that arrives while it runs sets `again`, and the owner queues it
+   when it hands the unit back (`finish`). No lock is held during a turn; one mutex guards the
+   queue and the slots, and a turn takes microseconds beside it.
+4. **Fairness is the budget and the back of the queue.** A turn handles at most `budget` frames
+   (`EdgeSource::poll(max)`), then says `more`, and `finish` queues it behind the units already
+   waiting. A full edge (`Flow::Backlogged`) is also `more`: the unit retries, and coalesces
+   meanwhile. The resumable `step(deadline)` predicate of `Rebuild` is for a sliced snapshot load,
+   and this example has none.
+5. **The executor spawns nothing.** `claim` and `finish` are what a worker calls. `Pool` is N
+   threads looping on them; the tests' scheduler calls them in an order a seed chooses, with up to N
+   units held at once and the turn of a held unit already run, so a kick can land on a running
+   unit. Both drive the same code.
+6. **Results do not depend on N or the order.** The deltas sum, and every operator in the graph is
+   a function of its input's integral. So any schedule settles where the single circuit does
+   (`placed_circuits_equal_one_circuit_for_any_workers_and_order`, 1 to 4 workers, random
+   commands, random order, commands arriving while it runs; and a real pool of 1 to 4 threads).
+   Per-tick outputs (`delay`, `differentiate`) would differ with coalescing, and the example has
+   none.
+7. **What the example is.** `Staged`: the host's end of the first edge, an aggregate stage (sum of
+   a cell's value per row and column, `aggregate`) and a grid projection (the cell each group draws,
+   a total per row). The incremental backend (`Incremental`) keeps its groups in its own shapes
+   and does not run on a `Circuit`, so the stage is the same computation built as zset nodes rather than a
+   split of `Incremental` itself. Running `Incremental` on circuits is a later step.
+8. **Not done.** A work-stealing or priority scheme (one FIFO queue); a unit with a `Rebuild` job
+   sliced by the deadline predicate; a timeout that calls `connection_lost` on the edges (the
+   edges here never lose a frame); a panic in a turn (the executor's mutex poisons and the pool
+   reports it).

context/code/zset.md

@@ -363,3 +363,5 @@ Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`.
     `Moved` correction). A pipe that blocks or notifies, because `try_send` and `try_recv` are
     polling. Key-sharded data parallelism inside a circuit: its exchange step would be this same
     edge, and nothing here forbids it.
+
+11. **Progress is counted.** `EdgeSink::traffic` and `EdgeSource::traffic` count frames sent and read and requests sent and read, so a scheduler can tell that a turn moved something (`improvise-incremental.md`, "Scheduling").

context/test-evidence/improvise-incremental.md

@@ -957,3 +957,31 @@ differs).
 
 `Pushdown::new` took a generic `Arc<Q>` that could not take a `SharedSource`; the test builds one from
 the alias and did not compile against the generic signature.
+
+### placed_circuits_equal_one_circuit_for_any_workers_and_order
+
+Random commands (twelve at most), 1 to 4 virtual workers, edge capacity 1 to 3, a budget of 1 to
+3 frames and a random order of claim and hand-back, with commands arriving between scheduler steps,
+settle to the grid of one circuit running the whole graph. Rivals, each failing it: a kick of a
+running unit queues it as well (S1, claiming a running unit panics), a kick while running is lost
+(S2), a unit that wants more goes to the front of the queue (S3, it never lets the others run), and
+progress that does not kick the neighbours (S4, the graph stalls).
+
+### real_worker_threads_equal_one_circuit
+
+The same graph on a real `Pool` of 1 to 4 threads, 400 commands, capacity 2, budget 3. Rivals S1
+and S4 fail it.
+
+### a_unit_that_wants_more_goes_to_the_back_of_the_queue
+
+Units 0 (wants two more turns), 1 and 2 run as 0, 1, 2, 0, 0. Rival S3 runs 0 again before 1.
+
+### a_kick_while_a_unit_runs_queues_it_again_and_never_claims_it_twice
+
+A claimed unit is not claimable; a kick during its turn leaves the executor not idle after it is
+handed back. Rivals S1 and S2 fail it.
+
+### at_most_one_worker_runs_a_unit_however_often_it_is_kicked
+
+Four kicking threads and four workers over one unit that asserts no two turns overlap. Rival S1
+fails it.

context/test-evidence/zset.md

@@ -689,3 +689,8 @@ with "did not panic".
 Three hundred commands on a producer thread, a consumer on another, capacity 2, the producer
 yielding at uneven points and pumping on only two of three commands. The consumer's output equals
 the combined circuit's. Rival R1 fails it.
+
+### a_poll_reads_at_most_the_frames_it_is_given
+
+Three corrections waiting, `poll(c, 2)` reads two and the next reads one, so a unit's turn is
+bounded by its budget. Rival: the bound removed from the loop fails the first assertion.

crates/improvise-incremental/src/lib.rs

@@ -12,7 +12,9 @@ mod incremental;
 mod pushdown;
 mod reading;
 mod reads;
+pub mod sched;
 mod shapes;
+pub mod staged;
 
 pub use delta::{CellDelta, DeltaValue, cell_delta, moved_records};
 pub use eager::Eager;

crates/improvise-incremental/src/sched.rs

@@ -0,0 +1,234 @@
+//! A scheduler for sub-circuits joined by edges: multi-threading as scheduling.
+//! Reasoning: `context/code/improvise-incremental.md`, "Scheduling".
+
+use std::collections::VecDeque;
+use std::sync::{Condvar, Mutex, MutexGuard};
+
+/// What one turn of a unit did.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Turn {
+    /// A frame moved on one of its edges, so its neighbours may have work.
+    pub progress: bool,
+    /// It stopped at its budget, or an edge was full: run it again.
+    pub more: bool,
+}
+
+/// A sub-circuit and the edge ends around it. One thread runs it at a time.
+pub trait Unit: Send {
+    /// Runs at most `budget` frames' worth of work, then returns.
+    fn turn(&mut self, budget: usize) -> Turn;
+    /// The units on the other end of its edges, in either direction.
+    fn neighbours(&self) -> &[usize];
+}
+
+struct Slot {
+    /// `None` while a worker owns it: this is the ownership token.
+    unit: Option<Box<dyn Unit>>,
+    queued: bool,
+    running: bool,
+    /// Kicked while running: queue it again when the owner hands it back.
+    again: bool,
+}
+
+struct State {
+    slots: Vec<Slot>,
+    queue: VecDeque<usize>,
+    running: usize,
+    shutdown: bool,
+}
+
+/// A unit a worker owns until it hands it back with `finish`.
+pub struct Claimed {
+    id: usize,
+    unit: Box<dyn Unit>,
+}
+
+impl Claimed {
+    pub fn id(&self) -> usize {
+        self.id
+    }
+
+    pub fn run(&mut self, budget: usize) -> Turn {
+        self.unit.turn(budget)
+    }
+}
+
+/// The run queue and the ownership of every unit. It spawns nothing: workers (threads, or a test
+/// scheduler choosing an order) call `claim` and `finish`.
+pub struct Executor {
+    state: Mutex<State>,
+    wake: Condvar,
+    budget: usize,
+}
+
+impl Executor {
+    /// `budget` is the frames a unit may handle per turn; zero is read as one.
+    pub fn new(units: Vec<Box<dyn Unit>>, budget: usize) -> Self {
+        Executor {
+            state: Mutex::new(State {
+                slots: units
+                    .into_iter()
+                    .map(|u| Slot {
+                        unit: Some(u),
+                        queued: false,
+                        running: false,
+                        again: false,
+                    })
+                    .collect(),
+                queue: VecDeque::new(),
+                running: 0,
+                shutdown: false,
+            }),
+            wake: Condvar::new(),
+            budget: budget.max(1),
+        }
+    }
+
+    pub fn budget(&self) -> usize {
+        self.budget
+    }
+
+    fn lock(&self) -> MutexGuard<'_, State> {
+        self.state.lock().expect("a unit panicked in a turn")
+    }
+
+    fn kick_in(st: &mut State, id: usize) {
+        let slot = &mut st.slots[id];
+        if slot.running {
+            slot.again = true;
+        } else if !slot.queued {
+            slot.queued = true;
+            st.queue.push_back(id);
+        }
+    }
+
+    /// Marks `id` runnable: it has input, or the host wants it to look.
+    pub fn kick(&self, id: usize) {
+        Self::kick_in(&mut self.lock(), id);
+        self.wake.notify_all();
+    }
+
+    pub fn kick_all(&self) {
+        let mut st = self.lock();
+        for id in 0..st.slots.len() {
+            Self::kick_in(&mut st, id);
+        }
+        drop(st);
+        self.wake.notify_all();
+    }
+
+    fn take(st: &mut State, at: usize) -> Claimed {
+        let id = st.queue.remove(at).expect("a queued unit");
+        let slot = &mut st.slots[id];
+        slot.queued = false;
+        slot.running = true;
+        st.running += 1;
+        Claimed {
+            id,
+            unit: slot.unit.take().expect("a queued unit is at rest"),
+        }
+    }
+
+    /// Takes the queued unit that `pick(queue length)` names, if any is queued.
+    pub fn claim(&self, pick: impl FnOnce(usize) -> usize) -> Option<Claimed> {
+        let mut st = self.lock();
+        let n = st.queue.len();
+        (n > 0).then(|| Self::take(&mut st, pick(n) % n))
+    }
+
+    /// Like `claim` for a worker thread: waits for work, and is `None` after `shutdown`.
+    pub fn claim_blocking(&self) -> Option<Claimed> {
+        let mut st = self.lock();
+        loop {
+            if !st.queue.is_empty() {
+                return Some(Self::take(&mut st, 0));
+            }
+            if st.shutdown {
+                return None;
+            }
+            st = self.wake.wait(st).expect("a unit panicked in a turn");
+        }
+    }
+
+    /// Hands a unit back; queues it again if it asked for more or was kicked meanwhile, and
+    /// kicks its neighbours if the turn moved a frame.
+    pub fn finish(&self, claimed: Claimed, turn: Turn) {
+        let Claimed { id, unit } = claimed;
+        let mut st = self.lock();
+        let neighbours = unit.neighbours().to_vec();
+        let slot = &mut st.slots[id];
+        slot.unit = Some(unit);
+        slot.running = false;
+        let again = std::mem::take(&mut slot.again);
+        st.running -= 1;
+        if again || turn.more {
+            Self::kick_in(&mut st, id);
+        }
+        if turn.progress {
+            for n in neighbours {
+                Self::kick_in(&mut st, n);
+            }
+        }
+        drop(st);
+        self.wake.notify_all();
+    }
+
+    /// Whether nothing is queued and nothing is running.
+    pub fn idle(&self) -> bool {
+        let st = self.lock();
+        st.queue.is_empty() && st.running == 0
+    }
+
+    /// Blocks until `idle`.
+    pub fn wait_idle(&self) {
+        let mut st = self.lock();
+        while !(st.queue.is_empty() && st.running == 0) {
+            st = self.wake.wait(st).expect("a unit panicked in a turn");
+        }
+    }
+
+    /// Tells the workers to leave once the queue is empty.
+    pub fn shutdown(&self) {
+        self.lock().shutdown = true;
+        self.wake.notify_all();
+    }
+}
+
+/// N worker threads over one executor. Native only: a wasm32 build has no threads to spawn.
+#[cfg(not(target_arch = "wasm32"))]
+pub struct Pool {
+    exec: std::sync::Arc<Executor>,
+    workers: Vec<std::thread::JoinHandle<()>>,
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+impl Pool {
+    pub fn start(exec: std::sync::Arc<Executor>, workers: usize) -> Self {
+        let workers = (0..workers.max(1))
+            .map(|_| {
+                let exec = std::sync::Arc::clone(&exec);
+                std::thread::spawn(move || {
+                    while let Some(mut claimed) = exec.claim_blocking() {
+                        let turn = claimed.run(exec.budget());
+                        exec.finish(claimed, turn);
+                    }
+                })
+            })
+            .collect();
+        Pool { exec, workers }
+    }
+
+    pub fn executor(&self) -> &Executor {
+        &self.exec
+    }
+
+    /// Waits for every unit to go quiet, stops the workers and returns the executor.
+    pub fn finish(self) -> std::sync::Arc<Executor> {
+        self.exec.wait_idle();
+        self.exec.shutdown();
+        for w in self.workers {
+            w.join().expect("a worker panicked");
+        }
+        self.exec
+    }
+}

crates/improvise-incremental/src/staged.rs

@@ -0,0 +1,216 @@
+//! One graph in two sub-circuits: the aggregate stage and the grid projection, placed on
+//! separate units of the scheduler. Reasoning: `context/code/improvise-incremental.md`, "Scheduling".
+
+use crate::sched::{Executor, Turn, Unit};
+use std::sync::{Arc, Mutex};
+use zset::{
+    ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Input, Output, Stream, ZSet,
+    channel_pipes,
+};
+
+/// A stored cell: row, column and a value; its weight counts it in or out.
+pub type Cell = (u8, u8, i64);
+/// A group of cells: its `(row, column)` and the sum of its values.
+pub type Group = ((u8, u8), i64);
+type P<K> = ChannelPipe<Frame<K>>;
+
+/// What the grid draws: each group's cell, and a total per row.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct Grid {
+    pub cells: ZSet<(u8, u8, i64)>,
+    pub rows: ZSet<(u8, i64)>,
+}
+
+/// Which unit each node of the graph runs on (manual placement): the head is the host's end of
+/// the first edge, the aggregate stage and the grid projection are circuits of their own.
+pub const PLACEMENT: [(&str, usize); 3] = [("head", 0), ("aggregate", 1), ("grid", 2)];
+
+/// The aggregate stage's nodes: the sum of the values of each `(row, column)`.
+fn aggregate_nodes(c: &mut Circuit, cells: &Stream<ZSet<Cell>>) -> Stream<ZSet<Group>> {
+    let keyed = c.map(cells, |(r, col, v)| ((*r, *col), *v));
+    c.aggregate(&keyed, |_, vals| {
+        Some(vals.iter().map(|(v, w)| v * w).sum())
+    })
+}
+
+type Drawn = (Output<(u8, u8, i64)>, Output<(u8, i64)>);
+
+/// The grid projection's nodes: the cell each group draws, and the sum of each row.
+fn grid_nodes(c: &mut Circuit, groups: &Stream<ZSet<Group>>) -> Drawn {
+    let drawn = c.map(groups, |((r, col), s)| (*r, *col, *s));
+    let by_row = c.map(groups, |((r, _), s)| (*r, *s));
+    let totals = c.aggregate(&by_row, |_, vals| {
+        Some(vals.iter().map(|(v, w)| v * w).sum())
+    });
+    let totals = c.map(&totals, |(r, t)| (*r, *t));
+    (c.output(&drawn), c.output(&totals))
+}
+
+/// The whole graph in one circuit: what the placed one must equal.
+pub fn single_circuit(commands: &[ZSet<Cell>]) -> Grid {
+    let mut c = Circuit::new();
+    let (input, cells) = c.input::<Cell>();
+    let groups = aggregate_nodes(&mut c, &cells);
+    let (drawn, totals) = grid_nodes(&mut c, &groups);
+    for d in commands {
+        c.push(&input, d.clone());
+        c.step();
+    }
+    Grid {
+        cells: c.take(&drawn),
+        rows: c.take(&totals),
+    }
+}
+
+/// The host's end of the first edge: commands come in through the inbox, each one stamped.
+struct Head {
+    sink: EdgeSink<Cell, Direct, P<Cell>>,
+    inbox: Arc<Mutex<Vec<ZSet<Cell>>>>,
+    next: [usize; 1],
+}
+
+impl Unit for Head {
+    fn turn(&mut self, _budget: usize) -> Turn {
+        let before = self.sink.traffic();
+        let commands = std::mem::take(&mut *self.inbox.lock().expect("the inbox"));
+        for d in commands {
+            self.sink.committed(d);
+        }
+        let flow = self.sink.pump();
+        Turn {
+            progress: self.sink.traffic() != before,
+            more: flow == Flow::Backlogged,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &self.next
+    }
+}
+
+struct Aggregate {
+    circuit: Circuit,
+    out: Output<Group>,
+    source: EdgeSource<Cell, Direct, P<Cell>>,
+    sink: EdgeSink<Group, Direct, P<Group>>,
+    around: [usize; 2],
+}
+
+impl Unit for Aggregate {
+    fn turn(&mut self, budget: usize) -> Turn {
+        let before = self.source.traffic() + self.sink.traffic();
+        let read = self.source.poll(&mut self.circuit, budget);
+        if read > 0 {
+            self.circuit.begin();
+            self.circuit.step();
+            self.circuit.commit();
+            self.sink.committed_from(&mut self.circuit, &self.out);
+        }
+        let flow = self.sink.pump();
+        Turn {
+            progress: self.source.traffic() + self.sink.traffic() != before,
+            more: read == budget || flow == Flow::Backlogged,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &self.around
+    }
+}
+
+struct Projection {
+    circuit: Circuit,
+    drawn: Output<(u8, u8, i64)>,
+    totals: Output<(u8, i64)>,
+    source: EdgeSource<Group, Direct, P<Group>>,
+    grid: Arc<Mutex<Grid>>,
+    back: [usize; 1],
+}
+
+impl Unit for Projection {
+    fn turn(&mut self, budget: usize) -> Turn {
+        let before = self.source.traffic();
+        let read = self.source.poll(&mut self.circuit, budget);
+        if read > 0 {
+            self.circuit.step();
+            let mut grid = self.grid.lock().expect("the grid");
+            grid.cells += self.circuit.take(&self.drawn);
+            grid.rows += self.circuit.take(&self.totals);
+        }
+        Turn {
+            progress: self.source.traffic() != before,
+            more: read == budget,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &self.back
+    }
+}
+
+/// The placed graph: three units on an executor, and the host's two handles on it.
+pub struct Staged {
+    exec: Arc<Executor>,
+    inbox: Arc<Mutex<Vec<ZSet<Cell>>>>,
+    grid: Arc<Mutex<Grid>>,
+}
+
+impl Staged {
+    /// `cap` frames per edge direction, `budget` frames per unit turn.
+    pub fn new(cap: usize, budget: usize) -> Self {
+        let inbox = Arc::new(Mutex::new(Vec::new()));
+        let grid = Arc::new(Mutex::new(Grid::default()));
+        let (head_end, agg_in) = channel_pipes::<Frame<Cell>>(cap);
+        let (agg_out, grid_in) = channel_pipes::<Frame<Group>>(cap);
+
+        let mut c = Circuit::new();
+        let (input, cells): (Input<Cell>, _) = c.input::<Cell>();
+        let groups = aggregate_nodes(&mut c, &cells);
+        let out = c.output(&groups);
+        let aggregate = Aggregate {
+            circuit: c,
+            out,
+            source: EdgeSource::new(Direct, agg_in, input),
+            sink: EdgeSink::new(Direct, agg_out, 2),
+            around: [0, 2],
+        };
+
+        let mut c = Circuit::new();
+        let (input, groups) = c.input::<Group>();
+        let (drawn, totals) = grid_nodes(&mut c, &groups);
+        let projection = Projection {
+            circuit: c,
+            drawn,
+            totals,
+            source: EdgeSource::new(Direct, grid_in, input),
+            grid: Arc::clone(&grid),
+            back: [1],
+        };
+
+        let head = Head {
+            sink: EdgeSink::new(Direct, head_end, 1),
+            inbox: Arc::clone(&inbox),
+            next: [1],
+        };
+        let units: Vec<Box<dyn Unit>> =
+            vec![Box::new(head), Box::new(aggregate), Box::new(projection)];
+        let exec = Arc::new(Executor::new(units, budget));
+        exec.kick_all();
+        Staged { exec, inbox, grid }
+    }
+
+    pub fn executor(&self) -> &Arc<Executor> {
+        &self.exec
+    }
+
+    /// One committed command from the host.
+    pub fn submit(&self, delta: ZSet<Cell>) {
+        self.inbox.lock().expect("the inbox").push(delta);
+        self.exec.kick(0);
+    }
+
+    /// What the grid has drawn so far.
+    pub fn grid(&self) -> Grid {
+        self.grid.lock().expect("the grid").clone()
+    }
+}

crates/improvise-incremental/tests/sched.proptest-regressions

@@ -0,0 +1,9 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 67b92f77a748cf7ce3b29898ed519a2926626971d2e38b295b3cd6616ea26b43 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }], workers = 3, cap = 1, budget = 1, choices = [148, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], gaps = [0]
+cc caa4c353b78a43303e8a95654896b57d676c5d95f3198ef4eb4b3d0c7531fec9 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {} }, ZSet { weights: {(0, 0, 0): -1} }, ZSet { weights: {} }], workers = 2, cap = 1, budget = 1, choices = [92, 71, 34, 29, 122, 72, 14, 0, 2, 6, 73, 132, 236, 26, 2, 20, 10], gaps = [2, 2, 4, 1, 5, 3]
+cc 4d5dba8a92cdebce288f317349a8b961e23dc0b627d27ddbeb78f728a6868b16 # shrinks to commands = [ZSet { weights: {} }, ZSet { weights: {} }], workers = 1, cap = 1, budget = 1, choices = [130, 0, 10, 0, 132, 62], gaps = [3, 3]

crates/improvise-incremental/tests/sched.rs

@@ -0,0 +1,257 @@
+//! The scheduler: sub-circuits on an executor equal one circuit, whatever the number of workers
+//! and the order they run in. Evidence: `context/test-evidence/improvise-incremental.md`.
+
+use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit};
+use improvise_incremental::staged::{Cell, Staged, single_circuit};
+use proptest::prelude::*;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use zset::ZSet;
+
+fn command() -> impl Strategy<Value = ZSet<Cell>> {
+    prop::collection::vec(((0u8..3, 0u8..3, -4i64..=4), -2i64..=2), 0..4)
+        .prop_map(|v| v.into_iter().collect())
+}
+
+/// Runs the executor with `workers` virtual workers, choosing each step from `choices`: either
+/// claim a queued unit (running its turn at once) or hand a held one back. Returns the steps.
+fn run_deterministic(
+    exec: &Executor,
+    workers: usize,
+    choices: &mut impl Iterator<Item = u8>,
+) -> usize {
+    let mut held: Vec<(Claimed, Turn)> = Vec::new();
+    let mut steps = 0;
+    loop {
+        let c = choices.next().unwrap_or(0) as usize;
+        let can_claim = held.len() < workers;
+        let mut started = false;
+        if can_claim
+            && (held.is_empty() || c.is_multiple_of(2))
+            && let Some(mut claimed) = exec.claim(|n| c / 2 % n)
+        {
+            let turn = claimed.run(exec.budget());
+            held.push((claimed, turn));
+            started = true;
+        }
+        if !started {
+            if held.is_empty() {
+                return steps;
+            }
+            let (claimed, turn) = held.remove(c / 2 % held.len());
+            exec.finish(claimed, turn);
+        }
+        steps += 1;
+        assert!(steps < 100_000, "[sched] the executor never went quiet");
+    }
+}
+
+proptest! {
+    /// The placed graph, on 1 to 4 workers in a random order with commands arriving while it
+    /// runs, ends where one circuit running the whole graph ends.
+    #[test]
+    fn placed_circuits_equal_one_circuit_for_any_workers_and_order(
+        commands in prop::collection::vec(command(), 0..12),
+        workers in 1usize..5,
+        cap in 1usize..4,
+        budget in 1usize..4,
+        choices in prop::collection::vec(any::<u8>(), 0..400),
+        gaps in prop::collection::vec(0usize..6, 0..12),
+    ) {
+        let staged = Staged::new(cap, budget);
+        let mut it = choices.into_iter();
+        for (i, d) in commands.iter().enumerate() {
+            staged.submit(d.clone());
+            // A few scheduler steps before the next command arrives.
+            let mut few = it.by_ref().take(gaps.get(i).copied().unwrap_or(0));
+            run_some(staged.executor(), workers, &mut few);
+        }
+        run_deterministic(staged.executor(), workers, &mut std::iter::repeat(0));
+        prop_assert_eq!(staged.grid(), single_circuit(&commands), "[sched] the placed graph");
+    }
+}
+
+/// A bounded number of claim-or-finish steps, leaving whatever is left in the queue.
+fn run_some(exec: &Executor, workers: usize, choices: &mut impl Iterator<Item = u8>) {
+    let mut held: Vec<(Claimed, Turn)> = Vec::new();
+    for c in choices {
+        let c = c as usize;
+        if held.len() < workers
+            && (held.is_empty() || c.is_multiple_of(2))
+            && let Some(mut claimed) = exec.claim(|n| c / 2 % n)
+        {
+            let turn = claimed.run(exec.budget());
+            held.push((claimed, turn));
+            continue;
+        }
+        if !held.is_empty() {
+            let (claimed, turn) = held.remove(c / 2 % held.len());
+            exec.finish(claimed, turn);
+        }
+    }
+    for (claimed, turn) in held {
+        exec.finish(claimed, turn);
+    }
+}
+
+#[test]
+fn real_worker_threads_equal_one_circuit() {
+    let commands: Vec<ZSet<Cell>> = (0..400u32)
+        .map(|n| {
+            [(
+                ((n % 3) as u8, (n / 3 % 3) as u8, (n % 7) as i64 - 3),
+                1 - (n % 3) as i64,
+            )]
+            .into_iter()
+            .collect()
+        })
+        .collect();
+    for workers in 1..=4 {
+        let staged = Staged::new(2, 3);
+        let pool = Pool::start(staged.executor().clone(), workers);
+        for d in &commands {
+            staged.submit(d.clone());
+        }
+        pool.finish();
+        assert_eq!(
+            staged.grid(),
+            single_circuit(&commands),
+            "[threads] {workers} workers"
+        );
+    }
+}
+
+/// A unit that records which turns ran, and asks for `more` a fixed number of times.
+struct Log {
+    id: usize,
+    more: usize,
+    ran: Arc<std::sync::Mutex<Vec<usize>>>,
+}
+
+impl Unit for Log {
+    fn turn(&mut self, _: usize) -> Turn {
+        self.ran.lock().unwrap().push(self.id);
+        let more = self.more > 0;
+        self.more = self.more.saturating_sub(1);
+        Turn {
+            progress: false,
+            more,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &[]
+    }
+}
+
+#[test]
+fn a_unit_that_wants_more_goes_to_the_back_of_the_queue() {
+    let ran = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let unit = |id, more| -> Box<dyn Unit> {
+        Box::new(Log {
+            id,
+            more,
+            ran: ran.clone(),
+        })
+    };
+    let exec = Executor::new(vec![unit(0, 2), unit(1, 0), unit(2, 0)], 1);
+    exec.kick_all();
+    while let Some(mut c) = exec.claim(|_| 0) {
+        let t = c.run(exec.budget());
+        exec.finish(c, t);
+    }
+    assert_eq!(
+        *ran.lock().unwrap(),
+        vec![0, 1, 2, 0, 0],
+        "[fair] the others run between its turns"
+    );
+    assert!(exec.idle());
+}
+
+#[test]
+fn a_kick_while_a_unit_runs_queues_it_again_and_never_claims_it_twice() {
+    let ran = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let exec = Executor::new(
+        vec![Box::new(Log {
+            id: 0,
+            more: 0,
+            ran: ran.clone(),
+        })],
+        1,
+    );
+    exec.kick(0);
+    let mut claimed = exec.claim(|_| 0).unwrap();
+    exec.kick(0);
+    assert!(
+        exec.claim(|_| 0).is_none(),
+        "[own] a running unit is not claimable"
+    );
+    let t = claimed.run(1);
+    exec.finish(claimed, t);
+    assert!(
+        !exec.idle(),
+        "[again] the kick that arrived mid-turn is kept"
+    );
+    let mut again = exec.claim(|_| 0).unwrap();
+    let t = again.run(1);
+    exec.finish(again, t);
+    assert!(exec.idle());
+}
+
+/// A unit that fails if two turns of it ever overlap.
+struct Exclusive {
+    inside: Arc<AtomicBool>,
+    turns: Arc<AtomicUsize>,
+}
+
+impl Unit for Exclusive {
+    fn turn(&mut self, _: usize) -> Turn {
+        assert!(
+            !self.inside.swap(true, Ordering::SeqCst),
+            "[own] two workers ran one unit"
+        );
+        self.turns.fetch_add(1, Ordering::SeqCst);
+        std::thread::yield_now();
+        self.inside.store(false, Ordering::SeqCst);
+        Turn {
+            progress: false,
+            more: false,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &[]
+    }
+}
+
+#[test]
+fn at_most_one_worker_runs_a_unit_however_often_it_is_kicked() {
+    let (inside, turns) = (
+        Arc::new(AtomicBool::new(false)),
+        Arc::new(AtomicUsize::new(0)),
+    );
+    let exec = Arc::new(Executor::new(
+        vec![Box::new(Exclusive {
+            inside: inside.clone(),
+            turns: turns.clone(),
+        })],
+        1,
+    ));
+    let pool = Pool::start(exec.clone(), 4);
+    let kickers: Vec<_> = (0..4)
+        .map(|_| {
+            let exec = exec.clone();
+            std::thread::spawn(move || {
+                for _ in 0..500 {
+                    exec.kick(0);
+                    std::thread::yield_now();
+                }
+            })
+        })
+        .collect();
+    for k in kickers {
+        k.join().unwrap();
+    }
+    pool.finish();
+    assert!(turns.load(Ordering::SeqCst) > 0, "[own] it ran");
+}

crates/zset/src/edge.rs

@@ -144,6 +144,7 @@ pub struct EdgeSink<K, C, P> {
     waiting: Delta<K>,
     owes_correction: bool,
     closed: bool,
+    traffic: u64,
 }
 
 impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
@@ -169,6 +170,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
             },
             owes_correction: false,
             closed: false,
+            traffic: 0,
         }
     }
 
@@ -179,6 +181,11 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
         }
     }
 
+    /// Frames sent and requests read so far: a host sees progress when it moves.
+    pub fn traffic(&self) -> u64 {
+        self.traffic
+    }
+
     /// Whether any committed command has not been sent.
     pub fn backlog(&self) -> bool {
         self.waiting.to > self.waiting.from || self.owes_correction
@@ -217,6 +224,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
         while !self.closed {
             match self.pipe.try_recv() {
                 Ok(wire) => {
+                    self.traffic += 1;
                     if let Ok(Frame::Request(_)) = self.codec.decode(wire) {
                         self.owes_correction = true;
                     }
@@ -241,6 +249,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
         };
         match self.pipe.try_send(self.codec.encode(&frame)) {
             Ok(()) => {
+                self.traffic += 1;
                 self.owes_correction = false;
                 self.waiting = Delta {
                     epoch: self.epoch,
@@ -275,6 +284,7 @@ pub struct EdgeSource<K, C, P> {
     stale: bool,
     asked: bool,
     closed: bool,
+    traffic: u64,
 }
 
 impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSource<K, C, P> {
@@ -291,6 +301,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
             stale: true,
             asked: false,
             closed: false,
+            traffic: 0,
         }
     }
 
@@ -302,6 +313,11 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
         })
     }
 
+    /// Frames read and requests sent so far: a host sees progress when it moves.
+    pub fn traffic(&self) -> u64 {
+        self.traffic
+    }
+
     /// Whether a correction is wanted and has not arrived.
     pub fn stale(&self) -> bool {
         self.stale
@@ -326,6 +342,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
             match self.pipe.try_recv() {
                 Ok(wire) => {
                     read += 1;
+                    self.traffic += 1;
                     match self.codec.decode(wire) {
                         Ok(Frame::Delta(d)) => self.on_delta(circuit, d),
                         Ok(Frame::Correction(c)) => self.on_correction(circuit, c),
@@ -396,7 +413,10 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
             .codec
             .encode(&Frame::Request(Request { have: self.stamp() }));
         match self.pipe.try_send(wire) {
-            Ok(()) => self.asked = true,
+            Ok(()) => {
+                self.asked = true;
+                self.traffic += 1;
+            }
             Err(Refused::Full(_)) => {}
             Err(Refused::Closed) => self.closed = true,
         }

crates/zset/tests/edge.rs

@@ -702,3 +702,24 @@ fn a_circuit_on_each_of_two_threads_equals_one_circuit() {
         "[threads] the consumer after the producer thread finishes"
     );
 }
+
+#[test]
+fn a_poll_reads_at_most_the_frames_it_is_given() {
+    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
+    let mut c = Circuit::new();
+    let (i, _) = c.input::<u8>();
+    let mut src = EdgeSource::new(Direct, a, i);
+    for through in 0..3 {
+        let target = Z::new();
+        let frame = Frame::Correction(Correction {
+            epoch: 1,
+            through,
+            target,
+        });
+        b.try_send(frame).unwrap();
+    }
+    assert_eq!(src.poll(&mut c, 2), 2, "[budget] two of three");
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 1 }));
+    assert_eq!(src.poll(&mut c, 2), 1, "[budget] the rest");
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
+}