diff --git a/Cargo.toml b/Cargo.toml index 88102a7e3..1a93cf88b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "dogsdogsdogs", "experiments", "interactive", + "interactive/server", "server", "server/dataflows/degr_dist", "server/dataflows/neighborhood", diff --git a/interactive/server/Cargo.toml b/interactive/server/Cargo.toml new file mode 100644 index 000000000..6df438ef8 --- /dev/null +++ b/interactive/server/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "ddir-server" +version = "0.1.0" +edition.workspace = true +rust-version.workspace = true +description = "Interactive differential dataflow server: hold named arrangements live across DDIR program installs and drops." +publish = false + +[[bin]] +name = "ddir_server" +path = "src/main.rs" + +[lints] +workspace = true + +[dependencies] +differential-dataflow = { workspace = true } +timely = { workspace = true } +interactive = { path = ".." } +diagnostics = { path = "../../diagnostics" } +tungstenite = "0.26" diff --git a/interactive/server/README.md b/interactive/server/README.md new file mode 100644 index 000000000..fdc80072d --- /dev/null +++ b/interactive/server/README.md @@ -0,0 +1,39 @@ +# Live DDIR server + +This crate transports the earlier `ddir-server` prototype onto the current +scope-tree DDIR interpreter. It does not use dynamic loading or a Rust FFI. + +Run `cargo run -p ddir-server`, then open `interactive/server/console.html` or +connect a line-oriented client to TCP port 7777. The same protocol is available +over WebSocket on port 7778. Set `DDIR_BIND`, `DDIR_WS_BIND`, or +`DDIR_TICK_MS` to change those defaults; `DDIR_TICK_MS=0` disables automatic +progress while subscriptions are active. The current `diagnostics` crate is +connected on `DDIR_DIAG_PORT` (default 51371). + +Every request can begin with an arbitrary request id. If omitted, the server +generates one. Responses are ` data ...`, followed by ` ok ...` or +` err ...`. A `tail` remains active after its `ok` and ends when stopped. + +The useful commands are `load`, `drop`, `list`, `peek`, `tail`, `stop`, `tick`, +and `exit`. `load` accepts an inline pipe-syntax program: + + load graph begin + let edges = import "random:nodes=8,edges=12,seed=1,churn=1"; + export "graph.edges" = edges; + graph end-load + tail graph.edges + +The old binding spelling is also accepted during upload, so +`edges=random(seed=1,arity=2,range=8,count=12,churn=1)` can redirect the local +import named `edges`. It is translated to the current content-addressed source +name. Such a source is deterministic: it begins with a fixed-size window into +an infinite hash-derived sequence and replaces `churn` rows on every tick. + +Automatic ticking happens only while at least one tail is active. This makes a +live demonstration move without assigning input durability semantics to DDIR. +Explicit `tick [n]` remains available for reproducible sessions. + +The prototype's `--explain`/`query` path is intentionally not ported yet. It was +tied to the former flat IR and should instead be rebuilt around the current +scope-tree explanation work. The live server reports an error rather than +silently changing those commands' meaning. diff --git a/interactive/server/console.html b/interactive/server/console.html new file mode 100644 index 000000000..49edc1d33 --- /dev/null +++ b/interactive/server/console.html @@ -0,0 +1,687 @@ + + + + +ddir_server console + + + + +
+ ddir_server console + + +
disconnected
+
+ +
+ + + +
+
+

Load a dataflow

+
+
+ + + +
+ + + +
+ + +
+
+
+ +
+

Protocol log

+ 

+    

+  

+
+

+
+

+  
+  
+

+
+
+
+
+
diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs
new file mode 100644
index 000000000..f78df9b23
--- /dev/null
+++ b/interactive/server/src/cmd.rs
@@ -0,0 +1,510 @@
+//! Command and response shapes for the line-oriented protocol.
+//!
+//! Each request line is `  [args...]`.
+//! Each response line starts with the same `` followed by one of:
+//!   - `ok [body...]` — terminal success line
+//!   - `err [body...]` — terminal error line
+//!   - `data ` — one streamed body line (peek/tail batches)
+//!   - `end` — terminator after a stream of `data` lines
+//!
+//! Multi-line bodies (a DDIR program) come via a two-phase upload:
+//! ` load   begin` opens; subsequent lines are
+//! literal program text terminated by ` end-load`.
+
+use std::collections::BTreeMap;
+
+pub type ReqId = String;
+
+/// One client session's outbound stream. Cloned into each `Request` so
+/// dispatch can route responses back to the originating client (and so
+/// long-lived subscriptions like `tail` capture the right sender).
+pub type RespSender = std::sync::mpsc::Sender;
+
+/// Per-session identity. Lets the worker tear down long-lived
+/// subscriptions when a connection disappears without an explicit stop.
+pub type ConnectionId = u64;
+
+#[derive(Debug)]
+pub enum Cmd {
+    /// Install a dataflow.
+    /// `id_hint` — a client-chosen name; the server may keep it or assign
+    /// 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` — when true, run the explain rewrite on the parsed
+    /// program before optimize/install. The resulting dataflow has one
+    /// extra positional input (the query input) and exports
+    /// `demand:` per original data source.
+    Load {
+        id_hint: String,
+        bindings: BTreeMap,
+        program: String,
+        explain: bool,
+    },
+    /// Drop the dataflow named by id or by `id_hint`. Fails if any
+    /// export of this dataflow is still imported by another live
+    /// dataflow or held by a reader.
+    Drop { target: DataflowRef },
+    /// List held names.
+    List,
+    /// One-shot snapshot of a named trace.
+    Peek { name: String },
+    /// Persistent subscription to a named trace.
+    Tail { name: String },
+    /// Cancel a previous `tail` (matched by its reqid).
+    Stop { tail_reqid: ReqId },
+    /// Push a row into the query input of a `--explain` dataflow.
+    /// Sign is `+1` for `add` and `-1` for `del`.
+    #[allow(dead_code)]
+    Query {
+        target: DataflowRef,
+        kind: QueryKind,
+        key: Vec,
+        val: Vec,
+    },
+    /// Advance ambient time by `n` (default 1).
+    Tick { n: u64 },
+    /// End the session.
+    Exit,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum QueryKind {
+    Add,
+    Del,
+}
+
+/// A reference to a registered dataflow, used by both `drop` and
+/// `query`. Either a numeric dataflow id or a name (the load's
+/// `id_hint`). Parsed by reading the token as a `u64` first, then
+/// falling back to a string name. So `drop 5` and `drop my_reach`
+/// both work, and `drop 5_alt` (which fails to parse as u64) falls
+/// through to the name lookup.
+#[derive(Debug, Clone)]
+pub enum DataflowRef {
+    Id(u64),
+    Name(String),
+}
+
+/// A parsed request: reqid plus the command (or a parse error).
+#[derive(Debug)]
+pub struct Request {
+    pub reqid: ReqId,
+    pub kind: Result,
+    /// Where to route responses for this request (and, for `tail`, all
+    /// subsequent batches until `stop`). Cloned from the per-connection
+    /// outbound sender.
+    pub resp: RespSender,
+    /// Originating session; lets the worker tear down per-connection
+    /// state (tails) when this session ends.
+    pub connection_id: ConnectionId,
+}
+
+/// State carried between lines so the parser can splice a multi-line
+/// `load  ... begin` body together. The parser hands back either a
+/// complete `Request` or `None` (more lines required).
+#[derive(Default)]
+pub struct LineParser {
+    pending_load: Option,
+    auto_reqid_counter: u64,
+}
+
+/// Tokens that introduce a command. If a line begins with one of these
+/// instead of an explicit reqid, the parser synthesizes a reqid.
+const COMMAND_KEYWORDS: &[&str] = &[
+    "load", "drop", "list", "peek", "tail", "stop", "tick", "query", "exit",
+];
+
+struct PendingLoad {
+    reqid: ReqId,
+    id_hint: String,
+    bindings: BTreeMap,
+    explain: bool,
+    body: String,
+}
+
+impl LineParser {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Feed one input line; return `Some((reqid, parsed))` if the line
+    /// completes a command (single-line or end of a multi-line body),
+    /// else `None` to indicate more input is required. The caller pairs
+    /// the result with a per-connection response sender to form a
+    /// `Request`.
+    pub fn feed(&mut self, line: &str) -> Option<(ReqId, Result)> {
+        // Inside a pending load body: every line is literal program text
+        // until ` end-load` or ` end-load`. The id_hint
+        // form is the friendly default when the load was auto-reqid'd
+        // (so the user can type `gen end-load` after `load gen … begin`
+        // without having to know the minted reqid).
+        if let Some(ref mut pl) = self.pending_load {
+            let trimmed = line.trim_end_matches(['\r', '\n']);
+            let mut parts = trimmed.split_whitespace();
+            if let (Some(tok0), Some(tok1), None) = (parts.next(), parts.next(), parts.next()) {
+                if (tok0 == pl.reqid || tok0 == pl.id_hint) && tok1 == "end-load" {
+                    let done = self.pending_load.take().unwrap();
+                    return Some((
+                        done.reqid,
+                        Ok(Cmd::Load {
+                            id_hint: done.id_hint,
+                            bindings: done.bindings,
+                            program: done.body,
+                            explain: done.explain,
+                        }),
+                    ));
+                }
+            }
+            pl.body.push_str(trimmed);
+            pl.body.push('\n');
+            return None;
+        }
+
+        let trimmed = line.trim();
+        if trimmed.is_empty() {
+            return None;
+        }
+        let mut toks = trimmed.split_whitespace();
+        let first = toks.next()?;
+        // If the line starts with a known command, mint a reqid so the
+        // user can type bare `list`, `tick 5`, `peek foo` without a
+        // hand-rolled tag. The minted reqid is echoed in the response
+        // so it can still be used with e.g. `stop `.
+        let (reqid, cmd) = if COMMAND_KEYWORDS.contains(&first) {
+            self.auto_reqid_counter += 1;
+            (format!("_{}", self.auto_reqid_counter), first)
+        } else {
+            // Normal `  ...` form.
+            let cmd = match toks.next() {
+                Some(c) => c,
+                None => return Some((first.to_string(), Err("missing command".into()))),
+            };
+            (first.to_string(), cmd)
+        };
+        let rest: Vec<&str> = toks.collect();
+        match parse_cmd(cmd, &rest) {
+            ParseOutcome::Cmd(c) => Some((reqid, Ok(c))),
+            ParseOutcome::Err(e) => Some((reqid, Err(e))),
+            ParseOutcome::BeginLoad {
+                id_hint,
+                bindings,
+                explain,
+            } => {
+                self.pending_load = Some(PendingLoad {
+                    reqid,
+                    id_hint,
+                    bindings,
+                    explain,
+                    body: String::new(),
+                });
+                None
+            }
+        }
+    }
+
+    /// True if waiting for a ` end-load`. WS transport uses this
+    /// to forward blank-line program body content verbatim.
+    pub fn awaiting_body(&self) -> bool {
+        self.pending_load.is_some()
+    }
+}
+
+enum ParseOutcome {
+    Cmd(Cmd),
+    Err(String),
+    BeginLoad {
+        id_hint: String,
+        bindings: BTreeMap,
+        explain: bool,
+    },
+}
+
+fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
+    match cmd {
+        "load" => {
+            // Syntax: `load  [--explain] [name=binding ...] begin`
+            // The trailing `begin` switches the parser into body-collection
+            // mode; subsequent input lines are program text terminated by
+            // ` end-load`.
+            if args.is_empty() {
+                return ParseOutcome::Err(
+                    "load: expected ` [--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];
+            let mut bindings = BTreeMap::new();
+            let mut explain = false;
+            for tok in middle {
+                if *tok == "--explain" {
+                    explain = true;
+                    continue;
+                }
+                let Some((k, v)) = tok.split_once('=') else {
+                    return ParseOutcome::Err(format!(
+                        "load: argument {:?} must be `--explain` or `name=binding`",
+                        tok
+                    ));
+                };
+                if bindings.insert(k.to_string(), v.to_string()).is_some() {
+                    return ParseOutcome::Err(format!("load: duplicate binding for {:?}", k));
+                }
+            }
+            ParseOutcome::BeginLoad {
+                id_hint,
+                bindings,
+                explain,
+            }
+        }
+        "query" => {
+            // Syntax: `query  add|del  ; `
+            // Where k/v-fields are comma-separated i64. Empty side allowed
+            // (write nothing before/after the `;`).
+            if args.len() < 3 {
+                return ParseOutcome::Err(
+                    "query: expected ` add|del  ; `".into(),
+                );
+            }
+            let target = match args[0].parse::() {
+                Ok(n) => DataflowRef::Id(n),
+                Err(_) => DataflowRef::Name(args[0].to_string()),
+            };
+            let kind = match args[1] {
+                "add" => QueryKind::Add,
+                "del" => QueryKind::Del,
+                other => {
+                    return ParseOutcome::Err(format!(
+                        "query: kind must be add|del, got {:?}",
+                        other
+                    ))
+                }
+            };
+            // Find the `;` separator among the remaining tokens.
+            let rest = &args[2..];
+            let sep = rest.iter().position(|t| *t == ";");
+            let (k_toks, v_toks): (&[&str], &[&str]) = match sep {
+                Some(i) => (&rest[..i], &rest[i + 1..]),
+                None => (rest, &[]),
+            };
+            fn parse_fields(toks: &[&str]) -> Result, String> {
+                let mut out = Vec::new();
+                for t in toks {
+                    for piece in t.split(',') {
+                        if piece.is_empty() {
+                            continue;
+                        }
+                        out.push(piece.parse().map_err(|_| format!("bad i64 {:?}", piece))?);
+                    }
+                }
+                Ok(out)
+            }
+            let key = match parse_fields(k_toks) {
+                Ok(v) => v,
+                Err(e) => return ParseOutcome::Err(format!("query key: {}", e)),
+            };
+            let val = match parse_fields(v_toks) {
+                Ok(v) => v,
+                Err(e) => return ParseOutcome::Err(format!("query val: {}", e)),
+            };
+            ParseOutcome::Cmd(Cmd::Query {
+                target,
+                kind,
+                key,
+                val,
+            })
+        }
+        "drop" => match args {
+            [tok] => {
+                let target = match tok.parse::() {
+                    Ok(n) => DataflowRef::Id(n),
+                    Err(_) => DataflowRef::Name((*tok).to_string()),
+                };
+                ParseOutcome::Cmd(Cmd::Drop { target })
+            }
+            _ => ParseOutcome::Err("drop: expected ``".into()),
+        },
+        "list" => match args {
+            [] => ParseOutcome::Cmd(Cmd::List),
+            _ => ParseOutcome::Err("list: takes no arguments".into()),
+        },
+        "peek" => match args {
+            [name] => ParseOutcome::Cmd(Cmd::Peek {
+                name: (*name).to_string(),
+            }),
+            _ => ParseOutcome::Err("peek: expected ``".into()),
+        },
+        "tail" => match args {
+            [name] => ParseOutcome::Cmd(Cmd::Tail {
+                name: (*name).to_string(),
+            }),
+            _ => ParseOutcome::Err("tail: expected ``".into()),
+        },
+        "stop" => match args {
+            [rid] => ParseOutcome::Cmd(Cmd::Stop {
+                tail_reqid: (*rid).to_string(),
+            }),
+            _ => ParseOutcome::Err("stop: expected ``".into()),
+        },
+        "tick" => match args {
+            [] => ParseOutcome::Cmd(Cmd::Tick { n: 1 }),
+            [n] => match n.parse::() {
+                Ok(n) => ParseOutcome::Cmd(Cmd::Tick { n }),
+                Err(_) => ParseOutcome::Err(format!("tick: bad count {:?}", n)),
+            },
+            _ => ParseOutcome::Err("tick: expected `[n]`".into()),
+        },
+        "exit" => ParseOutcome::Cmd(Cmd::Exit),
+        other => ParseOutcome::Err(format!("unknown command {:?}", other)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn feed_all(p: &mut LineParser, lines: &[&str]) -> Vec<(ReqId, Result)> {
+        let mut out = Vec::new();
+        for l in lines {
+            if let Some(r) = p.feed(l) {
+                out.push(r);
+            }
+        }
+        out
+    }
+
+    #[test]
+    fn simple_commands() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &["r0 list", "r1 tick 5", "r2 drop 3", "r3 peek foo"],
+        );
+        assert_eq!(got.len(), 4);
+        assert!(matches!(got[0].1, Ok(Cmd::List)));
+        assert!(matches!(got[1].1, Ok(Cmd::Tick { n: 5 })));
+        assert!(matches!(
+            got[2].1,
+            Ok(Cmd::Drop {
+                target: DataflowRef::Id(3)
+            })
+        ));
+        assert!(matches!(got[3].1, Ok(Cmd::Peek { ref name }) if name == "foo"));
+    }
+
+    #[test]
+    fn multiline_load() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &[
+                "r0 load gen edges=random(seed=1) begin",
+                "let edges = import \"edges/v1\";",
+                "export \"reach\" = edges;",
+                "r0 end-load",
+            ],
+        );
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Load {
+                id_hint,
+                bindings,
+                program,
+                explain,
+            }) => {
+                assert_eq!(id_hint, "gen");
+                assert_eq!(
+                    bindings.get("edges").map(String::as_str),
+                    Some("random(seed=1)")
+                );
+                assert!(program.contains("import \"edges/v1\""));
+                assert!(program.contains("export \"reach\""));
+                assert!(!*explain);
+            }
+            _ => panic!("expected Load, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn load_explain() {
+        let mut p = LineParser::new();
+        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",
+        ]);
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Load { explain, .. }) => assert!(*explain),
+            _ => panic!("expected explain Load, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn query_cmd() {
+        let mut p = LineParser::new();
+        let got = feed_all(&mut p, &["rQ query 3 add 1,2 ; 99"]);
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Query {
+                target,
+                kind,
+                key,
+                val,
+            }) => {
+                assert!(matches!(target, DataflowRef::Id(3)));
+                assert!(matches!(kind, QueryKind::Add));
+                assert_eq!(key, &vec![1, 2]);
+                assert_eq!(val, &vec![99]);
+            }
+            _ => panic!("expected Query, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn auto_reqid_for_bare_command() {
+        let mut p = LineParser::new();
+        let got = feed_all(&mut p, &["list", "tick 3", "rA peek foo", "exit"]);
+        assert_eq!(got.len(), 4);
+        // Bare commands get auto-minted reqids; mixed-in explicit reqids
+        // pass through unchanged.
+        assert_eq!(got[0].0, "_1"); // list
+        assert!(matches!(got[0].1, Ok(Cmd::List)));
+        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_eq!(got[3].0, "_3"); // exit
+        assert!(matches!(got[3].1, Ok(Cmd::Exit)));
+    }
+
+    #[test]
+    fn drop_by_id_or_name() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &[
+                "r0 drop 3",
+                "r1 drop my_reach",
+                "r2 drop", // missing arg → err
+            ],
+        );
+        assert_eq!(got.len(), 3);
+        assert!(matches!(
+            got[0].1,
+            Ok(Cmd::Drop {
+                target: DataflowRef::Id(3)
+            })
+        ));
+        assert!(
+            matches!(got[1].1, Ok(Cmd::Drop { target: DataflowRef::Name(ref n) }) if n == "my_reach")
+        );
+        assert!(matches!(got[2].1, Err(_)));
+    }
+}
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
new file mode 100644
index 000000000..67fbd12cd
--- /dev/null
+++ b/interactive/server/src/loop_.rs
@@ -0,0 +1,368 @@
+//! Single-worker live control loop. Network sessions parse commands off-worker;
+//! this thread alone owns timely and the DDIR registry.
+
+use std::any::{type_name_of_val, Any};
+use std::collections::HashMap;
+use std::panic::{catch_unwind, AssertUnwindSafe};
+use std::sync::mpsc::{Receiver, Sender, TryRecvError};
+use std::time::{Duration, Instant};
+
+use differential_dataflow::operators::arrange::ShutdownButton;
+use interactive::scope_ir::{Program, Source};
+use interactive::server::{OuterTime, Server};
+use timely::dataflow::operators::probe::Handle as ProbeHandle;
+use timely::dataflow::operators::CapabilitySet;
+use timely::worker::Worker;
+
+use crate::cmd::{Cmd, ConnectionId, DataflowRef, Request};
+
+struct Tail {
+    dataflow_id: usize,
+    _shutdown: ShutdownButton>,
+    trace: String,
+    probe: ProbeHandle,
+}
+
+type TailKey = (ConnectionId, String);
+
+pub fn run_worker(
+    worker: &mut Worker,
+    requests: Receiver,
+    session_ends: Receiver,
+) {
+    let diagnostics_port = std::env::var("DDIR_DIAG_PORT")
+        .ok()
+        .and_then(|port| port.parse().ok())
+        .unwrap_or(51371);
+    let diagnostics = diagnostics::logging::register(worker, false);
+    let _diagnostics_server =
+        diagnostics::server::Server::start(diagnostics_port, diagnostics.sink);
+    let mut server = Server::new();
+    let mut tails: HashMap = HashMap::new();
+    let tick_ms = std::env::var("DDIR_TICK_MS")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(250u64);
+    let interval = Duration::from_millis(tick_ms);
+    let mut last_tick = Instant::now();
+    let mut shutdown = false;
+
+    while !shutdown {
+        match requests.try_recv() {
+            Ok(request) => dispatch(request, &mut server, &mut tails, worker, &mut shutdown),
+            Err(TryRecvError::Disconnected) => break,
+            Err(TryRecvError::Empty) => {
+                // Session-end notifications use a separate channel. Only
+                // consume them after all already-queued commands, so a final
+                // `stop` followed by `exit` cannot race its own cleanup.
+                while let Ok(connection) = session_ends.try_recv() {
+                    stop_connection(connection, &mut tails, worker);
+                }
+                if tick_ms > 0 && !tails.is_empty() && last_tick.elapsed() >= interval {
+                    tick(&mut server, &mut tails, worker);
+                    last_tick = Instant::now();
+                } else {
+                    worker.step();
+                    std::thread::sleep(Duration::from_millis(5));
+                }
+            }
+        }
+    }
+    for (_, tail) in tails.drain() {
+        worker.drop_dataflow(tail.dataflow_id);
+    }
+}
+
+fn dispatch(
+    request: Request,
+    server: &mut Server,
+    tails: &mut HashMap,
+    worker: &mut Worker,
+    shutdown: &mut bool,
+) {
+    let Request {
+        reqid,
+        kind,
+        resp,
+        connection_id,
+    } = request;
+    let result = match kind {
+        Err(e) => Err(e),
+        Ok(Cmd::Load {
+            id_hint,
+            bindings,
+            program,
+            explain,
+        }) => {
+            if explain {
+                Err("load --explain is not yet transported to the scope-tree server".into())
+            } else {
+                load(&id_hint, &bindings, &program, server, worker)
+                    .map(|()| format!("installed {:?}", id_hint))
+            }
+        }
+        Ok(Cmd::Drop { target }) => match name_ref(target) {
+            Err(e) => Err(e),
+            Ok(name) => {
+                if tails.values().any(|tail| {
+                    server
+                        .program_info()
+                        .iter()
+                        .find(|p| p.name == name)
+                        .is_some_and(|p| p.exports.contains(&tail.trace))
+                }) {
+                    Err(format!(
+                        "cannot drop {:?}: a tail is reading one of its exports",
+                        name
+                    ))
+                } else {
+                    server
+                        .drop_program(worker, &name)
+                        .map(|()| format!("dropped {:?}", name))
+                }
+            }
+        },
+        Ok(Cmd::List) => {
+            for program in server.program_info() {
+                send(
+                    &resp,
+                    &reqid,
+                    "data",
+                    format!(
+                        "program name={:?} origin={} inputs={:?} imports={:?} exports={:?}",
+                        program.name,
+                        program.origin,
+                        program.inputs,
+                        program.imports,
+                        program.exports
+                    ),
+                );
+            }
+            for (name, importers) in server.trace_info() {
+                send(
+                    &resp,
+                    &reqid,
+                    "data",
+                    format!("trace name={:?} importers={}", name, importers),
+                );
+            }
+            Ok(format!("t={}", server.epoch()))
+        }
+        Ok(Cmd::Peek { name }) => match server.snapshot(worker, &name) {
+            Ok(rows) => {
+                for (key, val, diff) in rows {
+                    send(
+                        &resp,
+                        &reqid,
+                        "data",
+                        format!("diff={} key={:?} val={:?}", diff, key, val),
+                    );
+                }
+                Ok(format!("t={}", server.epoch()))
+            }
+            Err(e) => Err(e),
+        },
+        Ok(Cmd::Tail { name }) => start_tail(
+            connection_id,
+            &reqid,
+            &name,
+            resp.clone(),
+            server,
+            tails,
+            worker,
+        )
+        .map(|()| format!("tailing {:?} from t={}", name, server.epoch())),
+        Ok(Cmd::Stop { tail_reqid }) => {
+            let key = (connection_id, tail_reqid.clone());
+            match tails.remove(&key) {
+                Some(tail) => {
+                    worker.drop_dataflow(tail.dataflow_id);
+                    send(&resp, &tail_reqid, "end", String::new());
+                    Ok(format!("stopped {}", tail_reqid))
+                }
+                None => Err(format!("no tail {:?} in this session", tail_reqid)),
+            }
+        }
+        Ok(Cmd::Tick { n }) => {
+            for _ in 0..n {
+                tick(server, tails, worker);
+            }
+            Ok(format!("t={}", server.epoch()))
+        }
+        Ok(Cmd::Query { .. }) => Err(
+            "query belongs to the old explanation input and is not supported; use named sources"
+                .into(),
+        ),
+        Ok(Cmd::Exit) => {
+            *shutdown = connection_id == 0;
+            Ok("bye".into())
+        }
+    };
+    match result {
+        Ok(body) => send(&resp, &reqid, "ok", body),
+        Err(body) => send(&resp, &reqid, "err", body),
+    }
+}
+
+fn load(
+    name: &str,
+    bindings: &std::collections::BTreeMap,
+    source: &str,
+    server: &mut Server,
+    worker: &mut Worker,
+) -> Result<(), String> {
+    let mut program = catch_unwind(AssertUnwindSafe(|| {
+        let statements = interactive::parse::pipe::parse(source);
+        interactive::lower::lower_tree(statements)
+    }))
+    .map_err(panic_message)?;
+    apply_bindings(&mut program, bindings)?;
+    program.optimize();
+    server.install(worker, name, &program)
+}
+
+fn apply_bindings(
+    program: &mut Program,
+    bindings: &std::collections::BTreeMap,
+) -> Result<(), String> {
+    for (local, binding) in bindings {
+        let import = program
+            .root
+            .imports
+            .iter_mut()
+            .find(|import| import.name == *local)
+            .ok_or_else(|| format!("binding names no import {:?}", local))?;
+        import.from = Source::Trace(random_binding(binding)?);
+    }
+    Ok(())
+}
+
+/// Translate the prototype spelling into the current content-addressed source.
+fn random_binding(binding: &str) -> Result {
+    let Some(body) = binding
+        .strip_prefix("random(")
+        .and_then(|s| s.strip_suffix(')'))
+    else {
+        return Ok(binding.to_string());
+    };
+    let mut values: HashMap<&str, &str> = HashMap::new();
+    for field in body.split(',') {
+        let (key, value) = field
+            .trim()
+            .split_once('=')
+            .ok_or_else(|| format!("malformed random field {:?}", field))?;
+        values.insert(key.trim(), value.trim());
+    }
+    let nodes = values.remove("range").ok_or("random requires range")?;
+    let edges = values.remove("count").ok_or("random requires count")?;
+    let arity = values.remove("arity").unwrap_or("2");
+    let seed = values.remove("seed").unwrap_or("0");
+    let churn = values.remove("churn").unwrap_or("0");
+    if !values.is_empty() {
+        return Err(format!("unknown random fields: {:?}", values.keys()));
+    }
+    Ok(format!(
+        "random:nodes={},edges={},arity={},seed={},churn={}",
+        nodes, edges, arity, seed, churn
+    ))
+}
+
+fn start_tail(
+    connection: ConnectionId,
+    reqid: &str,
+    name: &str,
+    response: Sender,
+    server: &Server,
+    tails: &mut HashMap,
+    worker: &mut Worker,
+) -> Result<(), String> {
+    let key = (connection, reqid.to_string());
+    if tails.contains_key(&key) {
+        return Err(format!("tail reqid {:?} is already active", reqid));
+    }
+    let mut trace = server
+        .trace(name)
+        .ok_or_else(|| format!("no trace {:?}", name))?;
+    let dataflow_id = worker.next_dataflow_index();
+    let tag = reqid.to_string();
+    let mut probe = ProbeHandle::new();
+    let shutdown = worker.dataflow::(|scope| {
+        let (arranged, shutdown) = trace.import_core(scope.clone(), "TailImport");
+        arranged
+            .as_collection(|k, v| (k.clone(), v.clone()))
+            .inspect(move |((key, val), time, diff)| {
+                send(
+                    &response,
+                    &tag,
+                    "data",
+                    format!("time={} diff={} key={:?} val={:?}", time, diff, key, val),
+                );
+            })
+            .probe_with(&mut probe);
+        shutdown
+    });
+    tails.insert(
+        key,
+        Tail {
+            dataflow_id,
+            _shutdown: shutdown,
+            trace: name.to_string(),
+            probe,
+        },
+    );
+    Ok(())
+}
+
+fn tick(server: &mut Server, tails: &mut HashMap, worker: &mut Worker) {
+    server.tick(worker);
+    let epoch = server.epoch();
+    while tails.values().any(|tail| tail.probe.less_than(&epoch)) {
+        worker.step();
+    }
+}
+
+fn stop_connection(
+    connection: ConnectionId,
+    tails: &mut HashMap,
+    worker: &mut Worker,
+) {
+    let keys: Vec<_> = tails
+        .keys()
+        .filter(|(id, _)| *id == connection)
+        .cloned()
+        .collect();
+    for key in keys {
+        if let Some(tail) = tails.remove(&key) {
+            worker.drop_dataflow(tail.dataflow_id);
+        }
+    }
+}
+
+fn name_ref(target: DataflowRef) -> Result {
+    match target {
+        DataflowRef::Name(name) => Ok(name),
+        DataflowRef::Id(id) => Err(format!(
+            "numeric dataflow id {} is no longer exposed; use its name",
+            id
+        )),
+    }
+}
+
+fn send(sender: &Sender, reqid: &str, kind: &str, body: String) {
+    let suffix = if body.is_empty() {
+        String::new()
+    } else {
+        format!(" {}", body)
+    };
+    let _ = sender.send(format!("{} {}{}\n", reqid, kind, suffix));
+}
+
+fn panic_message(panic: Box) -> String {
+    if let Some(s) = panic.downcast_ref::<&str>() {
+        (*s).to_string()
+    } else if let Some(s) = panic.downcast_ref::() {
+        s.clone()
+    } else {
+        format!("DDIR parser panicked ({})", type_name_of_val(&panic))
+    }
+}
diff --git a/interactive/server/src/main.rs b/interactive/server/src/main.rs
new file mode 100644
index 000000000..d46c5abd3
--- /dev/null
+++ b/interactive/server/src/main.rs
@@ -0,0 +1,371 @@
+//! ddir_server entry point.
+//!
+//! v0 is single-binary, single-worker, with three transports:
+//!   - stdin/stdout (always on, process-wide)
+//!   - raw TCP line-protocol on `DDIR_BIND` (default 127.0.0.1:7777)
+//!   - WebSocket on `DDIR_WS_BIND` (default 127.0.0.1:7778) — one WS
+//!     text message per protocol line; same protocol otherwise.
+//!
+//! Each client (stdin, TCP, WS) is its own "session" with its own
+//! outbound channel, so a `tail` issued by one client streams updates
+//! only to that client while other clients continue to operate
+//! independently.
+//!
+//! Threads:
+//!   - main: spawns the worker, the TCP listener, and the stdin session;
+//!     then waits for the worker to finish.
+//!   - worker: timely worker driving the registry + dispatch loop.
+//!   - per-session reader: parses lines into commands, tagging each with
+//!     this session's response sender, and feeds the shared cmd channel.
+//!   - per-session writer: drains the response channel onto the wire.
+
+mod cmd;
+#[path = "loop_.rs"]
+mod control_loop;
+
+use std::io::{BufRead, BufReader, Write};
+use std::net::{TcpListener, TcpStream};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::mpsc::{channel, Sender};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use cmd::{ConnectionId, LineParser, Request};
+
+/// Process-wide allocator for per-session ids. Stdin uses 0;
+/// subsequent connections get 1, 2, 3, ....
+static NEXT_CONNECTION_ID: AtomicU64 = AtomicU64::new(1);
+
+fn alloc_connection_id() -> ConnectionId {
+    NEXT_CONNECTION_ID.fetch_add(1, Ordering::Relaxed)
+}
+
+fn main() {
+    let bind_addr = std::env::var("DDIR_BIND").unwrap_or_else(|_| "127.0.0.1:7777".to_string());
+    let ws_bind = std::env::var("DDIR_WS_BIND").unwrap_or_else(|_| "127.0.0.1:7778".to_string());
+
+    let (cmd_tx, cmd_rx) = channel::();
+    // Session-end notifications: a session's writer thread emits its
+    // connection_id when the channel closes (the client went away).
+    // The worker drains this and tears down any tails for that session.
+    let (session_end_tx, session_end_rx) = channel::();
+
+    // Spawn the timely worker on its own thread; it owns the registry.
+    let cmd_rx_cell = Arc::new(Mutex::new(Some(cmd_rx)));
+    let session_end_rx_cell = Arc::new(Mutex::new(Some(session_end_rx)));
+    let worker_thread = {
+        let cmd_rx_cell = cmd_rx_cell.clone();
+        let session_end_rx_cell = session_end_rx_cell.clone();
+        std::thread::spawn(move || {
+            timely::execute_directly(move |worker| {
+                let rx = cmd_rx_cell
+                    .lock()
+                    .unwrap()
+                    .take()
+                    .expect("cmd_rx taken twice");
+                let sx = session_end_rx_cell
+                    .lock()
+                    .unwrap()
+                    .take()
+                    .expect("session_end_rx taken twice");
+                control_loop::run_worker(worker, rx, sx);
+            });
+        })
+    };
+
+    // TCP listener: each accepted connection spawns its own reader and
+    // writer pair. Failures to bind are non-fatal — the stdin transport
+    // still works.
+    let tcp_handle = {
+        let session_end_tx = session_end_tx.clone();
+        spawn_listener(&bind_addr, "tcp", cmd_tx.clone(), move |stream, cmd_tx| {
+            run_tcp_session(stream, cmd_tx, session_end_tx.clone()).map_err(|e| e.to_string())
+        })
+    };
+
+    // WebSocket listener on a separate port. Per-connection session uses
+    // a single-thread cooperative read/write loop so the WebSocket isn't
+    // shared across threads (tungstenite::WebSocket isn't easily split).
+    let ws_handle = {
+        let session_end_tx = session_end_tx.clone();
+        spawn_listener(&ws_bind, "ws", cmd_tx.clone(), move |stream, cmd_tx| {
+            run_ws_session(stream, cmd_tx, session_end_tx.clone()).map_err(|e| e.to_string())
+        })
+    };
+
+    // Stdin session: shares the same cmd channel; responses go to stdout
+    // via this session's per-connection channel.
+    let stdin_done = std::thread::spawn({
+        let cmd_tx = cmd_tx.clone();
+        let session_end_tx = session_end_tx.clone();
+        move || run_stdin_session(cmd_tx, session_end_tx)
+    });
+    drop(cmd_tx);
+    drop(session_end_tx);
+
+    let _ = stdin_done.join();
+    let _ = worker_thread.join();
+    // The listener threads are parked on accept(); without an explicit
+    // shutdown signal, the cleanest exit is to let the OS reap them.
+    let _ = tcp_handle;
+    let _ = ws_handle;
+    std::process::exit(0);
+}
+
+/// Spawn a TcpListener accept loop that hands each accepted stream to a
+/// per-connection session function. Returns `None` if the bind itself
+/// failed (logged, but non-fatal — other transports still work).
+fn spawn_listener(
+    bind: &str,
+    label: &'static str,
+    cmd_tx: Sender,
+    session: F,
+) -> Option>
+where
+    F: Fn(TcpStream, Sender) -> Result<(), String> + Send + Sync + 'static,
+{
+    match TcpListener::bind(bind) {
+        Ok(listener) => {
+            eprintln!("ddir_server: {} listening on {}", label, bind);
+            let session = Arc::new(session);
+            Some(std::thread::spawn(move || {
+                for incoming in listener.incoming() {
+                    match incoming {
+                        Ok(stream) => {
+                            let cmd_tx = cmd_tx.clone();
+                            let session = session.clone();
+                            std::thread::spawn(move || {
+                                if let Err(e) = session(stream, cmd_tx) {
+                                    eprintln!("ddir_server: {} session ended: {}", label, e);
+                                }
+                            });
+                        }
+                        Err(e) => eprintln!("ddir_server: {} accept failed: {}", label, e),
+                    }
+                }
+            }))
+        }
+        Err(e) => {
+            eprintln!(
+                "ddir_server: {} bind {} failed: {} (other transports still work)",
+                label, bind, e
+            );
+            None
+        }
+    }
+}
+
+/// Run one session against a `BufRead` source and a `Write` sink. Spawns
+/// the writer pump, then loops on lines, parses each, tags it with this
+/// session's `connection_id`, and forwards to the worker. On return,
+/// announces the session's end via `session_end_tx` so the worker can
+/// tear down any tails this session initiated.
+fn run_session(
+    input: R,
+    output: W,
+    cmd_tx: Sender,
+    session_end_tx: Sender,
+    connection_id: ConnectionId,
+    on_exit: impl FnOnce() + Send + 'static,
+) -> std::io::Result<()> {
+    let (resp_tx, resp_rx) = channel::();
+    let writer_thread = std::thread::spawn(move || {
+        let mut out = output;
+        while let Ok(line) = resp_rx.recv() {
+            if out.write_all(line.as_bytes()).is_err() {
+                break;
+            }
+            let _ = out.flush();
+        }
+        on_exit();
+    });
+
+    let mut parser = LineParser::new();
+    for line in input.lines() {
+        let Ok(line) = line else {
+            break;
+        };
+        if let Some((reqid, kind)) = parser.feed(&line) {
+            let is_exit = matches!(kind, Ok(cmd::Cmd::Exit));
+            let req = Request {
+                reqid,
+                kind,
+                resp: resp_tx.clone(),
+                connection_id,
+            };
+            if cmd_tx.send(req).is_err() {
+                break;
+            }
+            // For stdin, `exit` terminates the whole server; for TCP, it
+            // terminates only this session. Either way the reader stops.
+            if is_exit {
+                break;
+            }
+        }
+    }
+    // The reader loop is done — the client is gone (or asked to exit).
+    // Notify the worker BEFORE joining the writer thread: any live tail
+    // operator holds a clone of `resp_tx` inside its inspect closure,
+    // which would otherwise keep `resp_rx` open indefinitely. The worker
+    // sees the session_end event, auto-stops those tails, which drops
+    // the closure (and its resp_tx clone), letting the writer pump exit.
+    drop(resp_tx);
+    let _ = session_end_tx.send(connection_id);
+    let _ = writer_thread.join();
+    Ok(())
+}
+
+fn run_stdin_session(cmd_tx: Sender, session_end_tx: Sender) {
+    let stdin = std::io::stdin();
+    let stdout = std::io::stdout();
+    // stdin always gets connection_id 0.
+    let _ = run_session(stdin.lock(), stdout, cmd_tx, session_end_tx, 0, || {});
+}
+
+fn run_tcp_session(
+    stream: TcpStream,
+    cmd_tx: Sender,
+    session_end_tx: Sender,
+) -> std::io::Result<()> {
+    let connection_id = alloc_connection_id();
+    let peer = stream
+        .peer_addr()
+        .map(|a| a.to_string())
+        .unwrap_or_else(|_| "?".into());
+    eprintln!(
+        "ddir_server: tcp client {} (conn={}) connected",
+        peer, connection_id
+    );
+    let reader_stream = stream.try_clone()?;
+    let writer_stream = stream;
+    let peer_for_exit = peer.clone();
+    run_session(
+        BufReader::new(reader_stream),
+        writer_stream,
+        cmd_tx,
+        session_end_tx,
+        connection_id,
+        move || {
+            eprintln!(
+                "ddir_server: tcp client {} (conn={}) disconnected",
+                peer_for_exit, connection_id
+            )
+        },
+    )
+}
+
+/// Run one WebSocket session. Single thread per connection: a short
+/// read timeout on the underlying TCP socket lets us interleave reads
+/// and writes (draining the per-connection outbound channel between
+/// read attempts). tungstenite's `WebSocket` isn't easily split across
+/// threads, so this is the cleanest shape.
+fn run_ws_session(
+    stream: TcpStream,
+    cmd_tx: Sender,
+    session_end_tx: Sender,
+) -> Result<(), tungstenite::Error> {
+    let connection_id = alloc_connection_id();
+    let peer = stream
+        .peer_addr()
+        .map(|a| a.to_string())
+        .unwrap_or_else(|_| "?".into());
+    eprintln!(
+        "ddir_server: ws client {} (conn={}) connecting",
+        peer, connection_id
+    );
+    let mut ws = match tungstenite::accept(stream) {
+        Ok(ws) => ws,
+        Err(e) => {
+            eprintln!(
+                "ddir_server: ws client {} (conn={}) handshake failed: {}",
+                peer, connection_id, e
+            );
+            let _ = session_end_tx.send(connection_id);
+            return Ok(());
+        }
+    };
+    // Apply the read timeout to the underlying TCP socket so .read()
+    // returns WouldBlock periodically, letting us interleave writes.
+    ws.get_ref()
+        .set_read_timeout(Some(Duration::from_millis(20)))
+        .ok();
+    eprintln!(
+        "ddir_server: ws client {} (conn={}) connected",
+        peer, connection_id
+    );
+
+    let (resp_tx, resp_rx) = channel::();
+    let mut parser = LineParser::new();
+    let mut should_exit = false;
+
+    loop {
+        // Drain any pending outbound first.
+        while let Ok(line) = resp_rx.try_recv() {
+            // WS frames don't carry a trailing newline by convention;
+            // strip the one our handlers append.
+            let payload = line.trim_end_matches('\n').to_string();
+            ws.send(tungstenite::Message::Text(payload.into()))?;
+        }
+
+        match ws.read() {
+            Ok(tungstenite::Message::Text(text)) => {
+                // One WS message may carry multiple lines (e.g., a
+                // multi-line load body sent as a single message). Split
+                // on '\n' and feed each through the parser.
+                for line in text.lines() {
+                    if line.trim().is_empty() && !parser.awaiting_body() {
+                        continue;
+                    }
+                    if let Some((reqid, kind)) = parser.feed(line) {
+                        let is_exit = matches!(kind, Ok(cmd::Cmd::Exit));
+                        let req = Request {
+                            reqid,
+                            kind,
+                            resp: resp_tx.clone(),
+                            connection_id,
+                        };
+                        if cmd_tx.send(req).is_err() {
+                            should_exit = true;
+                            break;
+                        }
+                        if is_exit {
+                            should_exit = true;
+                        }
+                    }
+                }
+            }
+            Ok(tungstenite::Message::Close(_)) => break,
+            Ok(_) => {} // ping/pong/binary/frame — ignore for v0
+            Err(tungstenite::Error::Io(e))
+                if e.kind() == std::io::ErrorKind::WouldBlock
+                    || e.kind() == std::io::ErrorKind::TimedOut =>
+            {
+                // Read timed out; loop back to drain outbound.
+            }
+            Err(e) => {
+                eprintln!(
+                    "ddir_server: ws client {} (conn={}) read error: {}",
+                    peer, connection_id, e
+                );
+                break;
+            }
+        }
+
+        if should_exit {
+            break;
+        }
+    }
+
+    // Final outbound drain before close.
+    while let Ok(line) = resp_rx.try_recv() {
+        let payload = line.trim_end_matches('\n').to_string();
+        let _ = ws.send(tungstenite::Message::Text(payload.into()));
+    }
+    let _ = ws.close(None);
+    eprintln!(
+        "ddir_server: ws client {} (conn={}) disconnected",
+        peer, connection_id
+    );
+    let _ = session_end_tx.send(connection_id);
+    Ok(())
+}
diff --git a/interactive/src/server.rs b/interactive/src/server.rs
index f4ee5359e..f0ff89eb0 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -45,19 +45,19 @@ use std::cell::RefCell;
 use std::collections::HashMap;
 use std::rc::Rc;
 
-use timely::worker::Worker;
-use timely::dataflow::ProbeHandle;
-use timely::progress::Antichain;
-use differential_dataflow::VecCollection;
+use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::input::{Input, InputSession};
 use differential_dataflow::operators::arrange::TraceAgent;
-use differential_dataflow::trace::TraceReader;
 use differential_dataflow::trace::implementations::ValSpine;
-use differential_dataflow::dynamic::pointstamp::PointStamp;
+use differential_dataflow::trace::TraceReader;
+use differential_dataflow::VecCollection;
+use timely::dataflow::ProbeHandle;
+use timely::progress::Antichain;
+use timely::worker::Worker;
 
-use crate::ir::{Value, Diff};
-use crate::scope_ir as st;
 use crate::backend::vec::render_tree;
+use crate::ir::{Diff, Value};
+use crate::scope_ir as st;
 
 /// The host (outer) timestamp shared across all installed programs.
 pub type OuterTime = u64;
@@ -73,9 +73,16 @@ type ServerInput = InputSession;
 /// generator on demand, and two imports of the same recipe share one source.
 #[derive(Clone, Copy)]
 enum Recipe {
-    /// `random:nodes=N,edges=E[,arity=A][,seed=S]` — a deterministic random
-    /// graph: `E` rows of `A` fields each, every field in `0..N`.
-    Random { nodes: u64, edges: u64, arity: usize, seed: u64 },
+    /// `random:nodes=N,edges=E[,arity=A][,seed=S][,churn=C]` — a deterministic
+    /// random graph: a window of `E` rows of `A` fields, every field in `0..N`.
+    /// Each tick replaces `C` rows (default zero).
+    Random {
+        nodes: u64,
+        edges: u64,
+        arity: usize,
+        seed: u64,
+        churn: u64,
+    },
     /// `iota:N` — the rows `(0) .. (N-1)`, each a one-field `Tuple`. The minimal
     /// index source from which richer generators are derived in-language (with
     /// `hash`).
@@ -87,7 +94,8 @@ impl Recipe {
     /// trace lookup). Unknown keys, missing required keys, or non-numbers reject.
     fn parse(name: &str) -> Option {
         if let Some(params) = name.strip_prefix("random:") {
-            let (mut nodes, mut edges, mut arity, mut seed) = (None, None, 2usize, 0u64);
+            let (mut nodes, mut edges, mut arity, mut seed, mut churn) =
+                (None, None, 2usize, 0u64, 0u64);
             for kv in params.split(',') {
                 let (k, v) = kv.split_once('=')?;
                 match k.trim() {
@@ -95,12 +103,21 @@ impl Recipe {
                     "edges" => edges = Some(v.trim().parse().ok()?),
                     "arity" => arity = v.trim().parse().ok()?,
                     "seed" => seed = v.trim().parse().ok()?,
+                    "churn" => churn = v.trim().parse().ok()?,
                     _ => return None,
                 }
             }
-            Some(Recipe::Random { nodes: nodes?, edges: edges?, arity, seed })
+            Some(Recipe::Random {
+                nodes: nodes?,
+                edges: edges?,
+                arity,
+                seed,
+                churn,
+            })
         } else if let Some(n) = name.strip_prefix("iota:") {
-            Some(Recipe::Iota { n: n.trim().parse().ok()? })
+            Some(Recipe::Iota {
+                n: n.trim().parse().ok()?,
+            })
         } else {
             None
         }
@@ -110,21 +127,34 @@ impl Recipe {
     /// omitted defaults address the same source.
     fn canonical(&self) -> String {
         match self {
-            Recipe::Random { nodes, edges, arity, seed } =>
-                format!("random:nodes={},edges={},arity={},seed={}", nodes, edges, arity, seed),
+            Recipe::Random {
+                nodes,
+                edges,
+                arity,
+                seed,
+                churn,
+            } => format!(
+                "random:nodes={},edges={},arity={},seed={},churn={}",
+                nodes, edges, arity, seed, churn
+            ),
             Recipe::Iota { n } => format!("iota:{}", n),
         }
     }
 
     /// The number of rows the source contains.
     fn rows_len(&self) -> u64 {
-        match self { Recipe::Random { edges, .. } => *edges, Recipe::Iota { n } => *n }
+        match self {
+            Recipe::Random { edges, .. } => *edges,
+            Recipe::Iota { n } => *n,
+        }
     }
 
     /// The generated row at index `e`.
     fn row(&self, e: u64) -> (Value, Value) {
         match self {
-            Recipe::Random { nodes, arity, seed, .. } => crate::gen_row_seeded(*seed, e, *nodes, *arity),
+            Recipe::Random {
+                nodes, arity, seed, ..
+            } => crate::gen_row_seeded(*seed, e, *nodes, *arity),
             Recipe::Iota { .. } => (Value::Tuple(vec![Value::Int(e as i64)]), Value::unit()),
         }
     }
@@ -133,16 +163,24 @@ impl Recipe {
 /// Where an installed entry came from. Only `Program` is writable by `feed`;
 /// `Clock` additionally has its single row advanced each `tick`.
 #[derive(Clone, Copy, PartialEq)]
-enum Origin { Program, Generated, Clock }
+enum Origin {
+    Program,
+    Generated,
+    Clock,
+}
 
 /// The single `clock` row for epoch `t`: `(Tuple[t] ; ())`.
-fn clock_row(t: OuterTime) -> Value { Value::Tuple(vec![Value::Int(t as i64)]) }
+fn clock_row(t: OuterTime) -> Value {
+    Value::Tuple(vec![Value::Int(t as i64)])
+}
 
 /// Map a source name to its canonical form: a recipe canonicalizes, any other
 /// name is returned unchanged. Used everywhere a source is looked up, so
 /// generated sources are shared by content regardless of how they're spelled.
 fn canonical_source_name(name: &str) -> String {
-    Recipe::parse(name).map(|r| r.canonical()).unwrap_or_else(|| name.to_string())
+    Recipe::parse(name)
+        .map(|r| r.canonical())
+        .unwrap_or_else(|| name.to_string())
 }
 
 /// A unit of server work, already parsed/lowered/validated on the intake side.
@@ -155,7 +193,14 @@ pub enum Command {
     Install { name: String, program: st::Program },
     /// Update positional `input` of `prog`: add `(key, val)` with `diff` at
     /// `time` (default the current epoch when `None`).
-    Feed { prog: String, input: usize, key: Value, val: Value, time: Option, diff: Diff },
+    Feed {
+        prog: String,
+        input: usize,
+        key: Value,
+        val: Value,
+        time: Option,
+        diff: Diff,
+    },
     /// Close the current epoch and run to quiescence.
     Tick,
     /// Drop the named program.
@@ -188,6 +233,18 @@ 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)>,
+}
+
+/// A stable, transport-friendly description of one installed dataflow.
+#[derive(Clone, Debug)]
+pub struct ProgramInfo {
+    pub name: String,
+    pub inputs: Vec,
+    pub imports: Vec,
+    pub exports: Vec,
+    pub origin: &'static str,
 }
 
 /// A live registry of installed programs and the traces they publish.
@@ -214,10 +271,54 @@ impl Server {
     }
 
     /// The current epoch (the open host time).
-    pub fn epoch(&self) -> OuterTime { self.epoch }
+    pub fn epoch(&self) -> OuterTime {
+        self.epoch
+    }
 
     /// Whether a trace is registered under `name`.
-    pub fn has_trace(&self, name: &str) -> bool { self.traces.contains_key(name) }
+    pub fn has_trace(&self, name: &str) -> bool {
+        self.traces.contains_key(name)
+    }
+
+    /// Clone a trace reader for a transient peek or subscription dataflow.
+    pub fn trace(&self, name: &str) -> Option {
+        self.traces.get(&canonical_source_name(name)).cloned()
+    }
+
+    /// Return registry state without coupling a caller to stdout formatting.
+    pub fn program_info(&self) -> Vec {
+        let mut result: Vec<_> = self
+            .programs
+            .iter()
+            .map(|(name, installed)| {
+                let mut inputs: Vec<_> = installed.inputs.keys().copied().collect();
+                inputs.sort();
+                ProgramInfo {
+                    name: name.clone(),
+                    inputs,
+                    imports: installed.imports.clone(),
+                    exports: installed.exports.clone(),
+                    origin: match installed.origin {
+                        Origin::Program => "program",
+                        Origin::Generated => "generated",
+                        Origin::Clock => "clock",
+                    },
+                }
+            })
+            .collect();
+        result.sort_by(|a, b| a.name.cmp(&b.name));
+        result
+    }
+
+    pub fn trace_info(&self) -> Vec<(String, usize)> {
+        let mut result: Vec<_> = self
+            .traces
+            .keys()
+            .map(|name| (name.clone(), self.importers.get(name).copied().unwrap_or(0)))
+            .collect();
+        result.sort_by(|a, b| a.0.cmp(&b.0));
+        result
+    }
 
     /// Install `prog` under `name`: build its dataflow in `worker`, wiring each
     /// root `Source::Trace` to a registered trace and registering each export's
@@ -229,7 +330,12 @@ impl Server {
     /// so two importers of the same recipe share one source. Any other
     /// unregistered import errors (install its producer first). Also errors if
     /// the name is taken or it would republish an existing export name.
-    pub fn install(&mut self, worker: &mut Worker, name: &str, prog: &st::Program) -> Result<(), String> {
+    pub fn install(
+        &mut self,
+        worker: &mut Worker,
+        name: &str,
+        prog: &st::Program,
+    ) -> Result<(), String> {
         if self.programs.contains_key(name) {
             return Err(format!("a program named {:?} is already installed", name));
         }
@@ -245,7 +351,10 @@ impl Server {
                     } else if let Some(recipe) = Recipe::parse(&key) {
                         self.install_generated(worker, &key, recipe);
                     } else {
-                        return Err(format!("program {:?} imports unknown trace {:?}; install its producer first", name, t));
+                        return Err(format!(
+                            "program {:?} imports unknown trace {:?}; install its producer first",
+                            name, t
+                        ));
                     }
                 }
             }
@@ -256,8 +365,14 @@ impl Server {
             }
         }
 
-        let import_names: Vec = prog.root.imports.iter()
-            .filter_map(|imp| match &imp.from { st::Source::Trace(t) => Some(canonical_source_name(t)), _ => None })
+        let import_names: Vec = prog
+            .root
+            .imports
+            .iter()
+            .filter_map(|imp| match &imp.from {
+                st::Source::Trace(t) => Some(canonical_source_name(t)),
+                _ => None,
+            })
             .collect();
         let export_names: Vec = prog.root.exports.iter().map(|e| e.name.clone()).collect();
 
@@ -273,8 +388,10 @@ impl Server {
                 let mut inputs: Vec<(usize, ServerInput)> = Vec::new();
 
                 // One outer (host-time) collection per root import.
-                let outer_cols: Vec> =
-                    root.imports.iter().map(|imp| match &imp.from {
+                let outer_cols: Vec> = root
+                    .imports
+                    .iter()
+                    .map(|imp| match &imp.from {
                         st::Source::Input(n) => {
                             let (handle, col) = outer.new_collection::<(Value, Value), Diff>();
                             inputs.push((*n, handle));
@@ -283,24 +400,40 @@ impl Server {
                         st::Source::Trace(t) => {
                             // The first binding point: resolve a named trace by importing it.
                             let key = canonical_source_name(t);
-                            let arranged = traces.get_mut(&key).expect("validated above").import(outer.clone());
+                            let arranged = traces
+                                .get_mut(&key)
+                                .expect("validated above")
+                                .import(outer.clone());
                             arranged.as_collection(|k, v| (k.clone(), v.clone()))
                         }
                         st::Source::Parent(_) => unreachable!("root import from a parent scope"),
-                    }).collect();
+                    })
+                    .collect();
 
                 // Render the program body in its own iterative scope, then bring
                 // every export back out to the host time (mirrors `vec::evaluate`).
-                let leaved: Vec> =
-                    outer.iterative::, _, _>(|inner| {
-                        let entered: Vec<_> = outer_cols.iter().map(|c| c.clone().enter(inner)).collect();
+                let leaved: Vec> = outer
+                    .iterative::, _, _>(|inner| {
+                        let entered: Vec<_> =
+                            outer_cols.iter().map(|c| c.clone().enter(inner)).collect();
                         let exports = render_tree(root, inner.clone(), 0, entered);
-                        exports.into_iter().map(|c| c.leave(outer)).collect::>()
+                        exports
+                            .into_iter()
+                            .map(|c| c.leave(outer))
+                            .collect::>()
                     });
 
                 // The second binding point: probe and publish each export's trace.
-                let published: Vec<(String, ServerTrace)> = root.exports.iter().zip(leaved)
-                    .map(|(e, col)| (e.name.clone(), col.probe_with(&probe).arrange_by_key().trace))
+                let published: Vec<(String, ServerTrace)> = root
+                    .exports
+                    .iter()
+                    .zip(leaved)
+                    .map(|(e, col)| {
+                        (
+                            e.name.clone(),
+                            col.probe_with(&probe).arrange_by_key().trace,
+                        )
+                    })
                     .collect();
 
                 (published, inputs)
@@ -318,14 +451,18 @@ impl Server {
             handle.flush();
             by_pos.insert(pos, handle);
         }
-        self.programs.insert(name.to_string(), Installed {
-            inputs: by_pos,
-            imports: import_names,
-            exports: export_names,
-            dataflow_id,
-            probe,
-            origin: Origin::Program,
-        });
+        self.programs.insert(
+            name.to_string(),
+            Installed {
+                inputs: by_pos,
+                imports: import_names,
+                exports: export_names,
+                dataflow_id,
+                probe,
+                origin: Origin::Program,
+                generator: None,
+            },
+        );
         Ok(())
     }
 
@@ -358,14 +495,18 @@ impl Server {
         self.traces.insert(name.to_string(), trace);
         let mut inputs = HashMap::new();
         inputs.insert(0usize, input);
-        self.programs.insert(name.to_string(), Installed {
-            inputs,
-            imports: Vec::new(),
-            exports: vec![name.to_string()],
-            dataflow_id,
-            probe,
-            origin: Origin::Generated,
-        });
+        self.programs.insert(
+            name.to_string(),
+            Installed {
+                inputs,
+                imports: Vec::new(),
+                exports: vec![name.to_string()],
+                dataflow_id,
+                probe,
+                origin: Origin::Generated,
+                generator: Some((recipe, 0)),
+            },
+        );
     }
 
     /// Install the `clock` source: a single row holding the current epoch, which
@@ -392,14 +533,18 @@ impl Server {
         self.traces.insert("clock".to_string(), trace);
         let mut inputs = HashMap::new();
         inputs.insert(0usize, input);
-        self.programs.insert("clock".to_string(), Installed {
-            inputs,
-            imports: Vec::new(),
-            exports: vec!["clock".to_string()],
-            dataflow_id,
-            probe,
-            origin: Origin::Clock,
-        });
+        self.programs.insert(
+            "clock".to_string(),
+            Installed {
+                inputs,
+                imports: Vec::new(),
+                exports: vec!["clock".to_string()],
+                dataflow_id,
+                probe,
+                origin: Origin::Clock,
+                generator: None,
+            },
+        );
     }
 
     /// Stage an update to positional input `input` of installed program `prog`:
@@ -407,18 +552,42 @@ impl Server {
     /// epoch). The time must be at or after the current epoch — you cannot
     /// insert into the closed past. Takes effect once `tick` advances the input
     /// frontier past `time`.
-    pub fn feed(&mut self, prog: &str, input: usize, key: Value, val: Value, time: Option, diff: Diff) -> Result<(), String> {
+    pub fn feed(
+        &mut self,
+        prog: &str,
+        input: usize,
+        key: Value,
+        val: Value,
+        time: Option,
+        diff: Diff,
+    ) -> Result<(), String> {
         let t = time.unwrap_or(self.epoch);
         if t < self.epoch {
-            return Err(format!("cannot feed at time {} < current epoch {}", t, self.epoch));
+            return Err(format!(
+                "cannot feed at time {} < current epoch {}",
+                t, self.epoch
+            ));
         }
         let prog = canonical_source_name(prog);
-        let installed = self.programs.get_mut(&prog).ok_or_else(|| format!("no program {:?}", prog))?;
+        let installed = self
+            .programs
+            .get_mut(&prog)
+            .ok_or_else(|| format!("no program {:?}", prog))?;
         if installed.origin != Origin::Program {
-            let kind = if installed.origin == Origin::Clock { "clock" } else { "generated" };
-            return Err(format!("{:?} is a {} source and is not writable", prog, kind));
+            let kind = if installed.origin == Origin::Clock {
+                "clock"
+            } else {
+                "generated"
+            };
+            return Err(format!(
+                "{:?} is a {} source and is not writable",
+                prog, kind
+            ));
         }
-        let handle = installed.inputs.get_mut(&input).ok_or_else(|| format!("program {:?} has no input {}", prog, input))?;
+        let handle = installed
+            .inputs
+            .get_mut(&input)
+            .ok_or_else(|| format!("program {:?} has no input {}", prog, input))?;
         handle.update_at((key, val), t, diff);
         Ok(())
     }
@@ -430,7 +599,12 @@ impl Server {
     /// multiplicities as of the current epoch — so the result is the complete,
     /// consolidated contents even when the trace is sharded across workers, not
     /// each worker's slice. The dataflow is dropped as soon as it has drained.
-    pub fn peek(&mut self, worker: &mut Worker, name: &str, key: Option) -> Result<(), String> {
+    pub fn peek(
+        &mut self,
+        worker: &mut Worker,
+        name: &str,
+        key: Option,
+    ) -> Result<(), String> {
         use timely::dataflow::operators::{Exchange, Inspect, Probe};
 
         let canon = canonical_source_name(name);
@@ -459,7 +633,10 @@ impl Server {
                 .inspect(move |((k, v), t, d)| {
                     // The snapshot as of `epoch`: the closed past (t < epoch).
                     if *t < epoch {
-                        *acc_in.borrow_mut().entry((k.clone(), v.clone())).or_insert(0) += *d;
+                        *acc_in
+                            .borrow_mut()
+                            .entry((k.clone(), v.clone()))
+                            .or_insert(0) += *d;
                     }
                 })
                 .probe_with(&mut peek_probe);
@@ -472,7 +649,8 @@ impl Server {
 
         if worker.index() == 0 {
             let acc = acc.borrow();
-            let mut rows: Vec<(&(Value, Value), &Diff)> = acc.iter().filter(|(_, d)| **d != 0).collect();
+            let mut rows: Vec<(&(Value, Value), &Diff)> =
+                acc.iter().filter(|(_, d)| **d != 0).collect();
             rows.sort_by(|a, b| a.0.cmp(b.0));
             match &key {
                 Some(k) => println!("peek {:?} key={:?} ({} rows):", name, k, rows.len()),
@@ -485,6 +663,54 @@ impl Server {
         Ok(())
     }
 
+    /// Return the consolidated closed-past contents of a trace on worker 0.
+    /// This is the structured counterpart to [`Server::peek`] for protocols.
+    pub fn snapshot(
+        &mut self,
+        worker: &mut Worker,
+        name: &str,
+    ) -> Result, String> {
+        use timely::dataflow::operators::{Exchange, Inspect, Probe};
+
+        let name = canonical_source_name(name);
+        let epoch = self.epoch;
+        let mut trace = self
+            .trace(&name)
+            .ok_or_else(|| format!("no trace {:?}", name))?;
+        let acc: Rc>> = Rc::new(RefCell::new(HashMap::new()));
+        let acc_in = acc.clone();
+        let mut probe = ProbeHandle::new();
+        let id = worker.next_dataflow_index();
+        worker.dataflow::(|scope| {
+            trace
+                .import(scope.clone())
+                .as_collection(|k, v| (k.clone(), v.clone()))
+                .inner
+                .exchange(|_| 0u64)
+                .inspect(move |((k, v), t, d)| {
+                    if *t < epoch {
+                        *acc_in
+                            .borrow_mut()
+                            .entry((k.clone(), v.clone()))
+                            .or_insert(0) += *d;
+                    }
+                })
+                .probe_with(&mut probe);
+        });
+        while probe.less_than(&epoch) {
+            worker.step();
+        }
+        worker.drop_dataflow(id);
+        let mut rows: Vec<_> = acc
+            .borrow()
+            .iter()
+            .filter(|(_, d)| **d != 0)
+            .map(|((k, v), d)| (k.clone(), v.clone(), *d))
+            .collect();
+        rows.sort_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1)));
+        Ok(rows)
+    }
+
     /// Drop installed program `name`, releasing its dataflow immediately.
     ///
     /// Refuses (changing nothing) if any trace the program publishes still has a
@@ -495,11 +721,17 @@ impl Server {
     pub fn drop_program(&mut self, worker: &mut Worker, name: &str) -> Result<(), String> {
         let canon = canonical_source_name(name);
         let name = canon.as_str();
-        let installed = self.programs.get(name).ok_or_else(|| format!("no program {:?}", name))?;
+        let installed = self
+            .programs
+            .get(name)
+            .ok_or_else(|| format!("no program {:?}", name))?;
         for ex in &installed.exports {
             let live = self.importers.get(ex).copied().unwrap_or(0);
             if live > 0 {
-                return Err(format!("cannot drop {:?}: its trace {:?} has {} live importer(s); drop them first", name, ex, live));
+                return Err(format!(
+                    "cannot drop {:?}: its trace {:?} has {} live importer(s); drop them first",
+                    name, ex, live
+                ));
             }
         }
 
@@ -517,6 +749,20 @@ impl Server {
         // still exist), then remove the dataflow outright.
         drop(installed);
         worker.drop_dataflow(id);
+        // Generated sources are installed on demand and have no independent
+        // owner. Reclaim any whose last importing program was just removed.
+        let garbage: Vec<_> = self
+            .programs
+            .iter()
+            .filter(|(source, program)| {
+                program.origin != Origin::Program
+                    && self.importers.get(*source).copied().unwrap_or(0) == 0
+            })
+            .map(|(source, _)| source.clone())
+            .collect();
+        for source in garbage {
+            self.drop_program(worker, &source)?;
+        }
         Ok(())
     }
 
@@ -535,6 +781,26 @@ impl Server {
                     h.update_at((clock_row(next), Value::unit()), next, 1);
                 }
             }
+            // 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 {
+                let recipe = *recipe;
+                if let Recipe::Random { edges, churn, .. } = recipe {
+                    if let Some(h) = installed.inputs.get_mut(&0) {
+                        for _ in 0..churn {
+                            let old = *cursor;
+                            let new = edges + *cursor;
+                            if (old as usize) % worker.peers() == worker.index() {
+                                h.update(recipe.row(old), -1);
+                            }
+                            if (new as usize) % worker.peers() == worker.index() {
+                                h.update(recipe.row(new), 1);
+                            }
+                            *cursor += 1;
+                        }
+                    }
+                }
+            }
             for handle in installed.inputs.values_mut() {
                 handle.advance_to(next);
                 handle.flush();
@@ -577,7 +843,11 @@ impl Server {
         let mut names: Vec<&String> = self.traces.keys().collect();
         names.sort();
         for n in names {
-            println!("  {} (importers: {})", n, self.importers.get(n).copied().unwrap_or(0));
+            println!(
+                "  {} (importers: {})",
+                n,
+                self.importers.get(n).copied().unwrap_or(0)
+            );
         }
         println!("programs ({}):", self.programs.len());
         let mut progs: Vec<&String> = self.programs.keys().collect();
@@ -591,11 +861,16 @@ impl Server {
                 Origin::Generated => " [generated]",
                 Origin::Clock => " [clock]",
             };
-            println!("  {}{} (inputs: {:?}, imports: {:?}, exports: {:?})", p, tag, ins, installed.imports, installed.exports);
+            println!(
+                "  {}{} (inputs: {:?}, imports: {:?}, exports: {:?})",
+                p, tag, ins, installed.imports, installed.exports
+            );
         }
     }
 }
 
 impl Default for Server {
-    fn default() -> Self { Server::new() }
+    fn default() -> Self {
+        Server::new()
+    }
 }