diff --git a/Cargo.toml b/Cargo.toml index 2759e726d..1d1d45622 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "dogsdogsdogs", "experiments", "interactive", + "interactive/server", #"tpchlike", #"doop", "mdbook", diff --git a/interactive/examples/ddir_server.rs b/interactive/examples/ddir_server.rs index 33a180752..3ff3e7a7e 100644 --- a/interactive/examples/ddir_server.rs +++ b/interactive/examples/ddir_server.rs @@ -125,6 +125,16 @@ fn parse_command(line: &str) -> Result { Ok(Command::Feed { prog, input, key, val, time, diff }) } "tick" => Ok(Command::Tick), + "bind" | "unbind" if toks.len() == 4 => { + let trace = toks[1].to_string(); + let prog = toks[2].to_string(); + let input: usize = toks[3].parse().map_err(|_| format!("{}: must be a number, got {:?}", toks[0], toks[3]))?; + if toks[0] == "bind" { + Ok(Command::Bind { trace, prog, input }) + } else { + Ok(Command::Unbind { trace, prog, input }) + } + } "drop" if toks.len() == 2 => Ok(Command::Drop { name: toks[1].to_string() }), "peek" if toks.len() == 2 || toks.len() == 3 => { let trace = toks[1].to_string(); @@ -146,6 +156,8 @@ fn print_help() { println!(" install "); println!(" feed [val=] [time=] [diff=]"); println!(" tick"); + println!(" bind (feed the trace's changes back in, each tick)"); + println!(" unbind "); println!(" drop "); println!(" peek [key]"); println!(" list"); @@ -177,6 +189,16 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool { Ok(()) => if w0 { println!("dropped {:?}", name); }, Err(e) => if w0 { println!("error: {}", e); }, }, + // Collective: the tap dataflow is built on every worker (each sees + // its shard, so the union delivers the delta exactly once). + Command::Bind { trace, prog, input } => match server.bind(worker, trace, prog, *input) { + Ok(()) => if w0 { println!("bound {:?} -> {:?} input {}", trace, prog, input); }, + Err(e) => if w0 { println!("error: {}", e); }, + }, + Command::Unbind { trace, prog, input } => match server.unbind(worker, trace, prog, *input) { + Ok(()) => if w0 { println!("unbound {:?} -> {:?} input {}", trace, prog, input); }, + Err(e) => if w0 { println!("error: {}", e); }, + }, // Collective: every worker imports its shard; `peek` gathers to worker 0 // (which prints) and reports an error there if the trace is unknown. Command::Peek { trace, key } => { 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..1f0e9f1c0 --- /dev/null +++ b/interactive/server/README.md @@ -0,0 +1,105 @@ +# Live DDIR server + +One long-running timely worker hosts interpreted DDIR dataflows through a +load-run-drop lifecycle. Programs share results by name — each may import +collections that others export — and clients follow along over TCP, +WebSocket, or stdin. + +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. +Between commands, blank lines and `#` comment lines are skipped, so command +scripts can be piped to stdin (see `demo/`). + +The useful commands are `load`, `drop`, `list`, `feed`, `bind`, `unbind`, +`peek`, `tail`, `stop`, `tick`, and `exit`. `load` accepts an inline +pipe-syntax program: + + 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 + +A binding may also be spelled as a call, so +`edges=random(seed=1,arity=2,range=8,count=12,churn=1)` redirects the local +import named `edges` to the same content-addressed source as the +`random:...` form. 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. Treat +auto-tick as demo furniture rather than a design commitment: as specified, +observation advances time (an observer effect), and the alternative — that a +watcher must be present to move things along, by ticking or by running a +metronome client whose ticks are ordinary logged commands — may be the better +design once the server has real tenants. + +## Writes: `feed` + + feed [val=] [time=] [diff=] + +pushes one update into a loaded program's positional input, exactly as in the +`ddir_server` example (`1,2` → a tuple; `_` → unit; a closed scalar term such +as `inject(2,tuple(3,4))` for ADT-shaped rows). + +The stance on contention: **writes are open; policy lives in the dataflow**. +The server does not decide who may write what. Cooperating clients follow a +simple protocol — include your id and an ordering epoch in the data — and +programs resolve races over those facts (first-claim-wins is a `min` over +`(epoch, id)`, see `demo/claims.txt`; full optimistic transactions are a +recursive view, see `demo/txn.txt`). Racing writes settle identically on +every replay. Identity is convention, not enforcement: we are not defending +against adversarial clients yet, and server-side attribution is deliberately +deferred until a deployment needs it. + +## Feedback: `bind` + + bind unbind + +From then on, every `tick` delivers the trace's *changes* into that input at +the next epoch, so the input mirrors the trace one epoch delayed. This is the +write path for *programs*: an installed dataflow can act on the world — or on +itself — with no client in the loop, one well-founded recursion step per tick. + +The state-machine idiom (see `demo/counter.txt` and the `server_bind` tests): +give the program a seed input and a dedicated feedback input, + + let state = seed + feedback; + +and bind the export `f(state) + (seed | negate)` to the feedback input; then +`state(t) = f(state(t-1))`, while later seed feeds still inject as +perturbations. A bound source cannot be dropped (it holds an importer), nor +can the bound target (unbind first). + +## One gate + +Loads are cheap to request and costly to render, so intake is bounded: +`DDIR_MAX_PROGRAM_BYTES` (default 65536) — a larger `load` body is swallowed +and rejected with one error, before parsing. This is transport self-defense, +not semantics. There are no ownership or quota gates: sessions are trusted, +and admission policy (auth, quotas, rate limits) belongs in a fronting proxy +if a deployment ever needs one. + +## Demos + + cargo run -p ddir-server --release + # then, or piped straight to stdin: + ./target/release/ddir_server < interactive/server/demo/counter.txt + ./target/release/ddir_server < interactive/server/demo/claims.txt + ./target/release/ddir_server < interactive/server/demo/txn.txt + python3 interactive/server/demo/two_sessions.py # races + size gate over TCP + +`load --explain` and `query` are reserved but unimplemented: explanation +support belongs on the scope-tree explanation machinery, and until that lands +the server reports an error rather than giving those commands an improvised +meaning. diff --git a/interactive/server/console.html b/interactive/server/console.html new file mode 100644 index 000000000..e06e53646 --- /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/demo/claims.txt b/interactive/server/demo/claims.txt
new file mode 100644
index 000000000..e238ae0eb
--- /dev/null
+++ b/interactive/server/demo/claims.txt
@@ -0,0 +1,23 @@
+# First-claim-wins with identity by convention. Cooperating clients include
+# (their id, an ordering epoch) in the value; the program resolves races as
+# min over (epoch, id) — deterministic on every replay. The server enforces
+# nothing here: contention policy lives in the dataflow, and honest ids are
+# part of the client protocol (we are not defending against adversaries yet;
+# when we must, attribution hardening designs are on the shelf).
+#
+#   cargo run -p ddir-server < interactive/server/demo/claims.txt
+
+load world begin
+let claims = input 0;
+export "owner" = claims | map($0 ; $1[1], $1[0]) | min | map($0 ; $1[1]);
+world end-load
+
+# Client 1 and client 2 race for cell (1,1) in epoch 0: lower id wins.
+feed world 0 1,1 val=1,0
+feed world 0 1,1 val=2,0
+feed world 0 2,2 val=2,0
+tick
+
+# owner: (1,1) -> client 1 (won the race), (2,2) -> client 2 (uncontested).
+peek owner
+exit
diff --git a/interactive/server/demo/counter.txt b/interactive/server/demo/counter.txt
new file mode 100644
index 000000000..9b1d4e091
--- /dev/null
+++ b/interactive/server/demo/counter.txt
@@ -0,0 +1,22 @@
+# The feedback primitive: a counter that advances with no client in the
+# loop. `bind` mirrors the export "next" into input 1, one epoch per tick,
+# and the `f(state) + (seed | negate)` shape makes the recursion exact:
+# state(t) = seed + next(t-1) = f(state(t-1)).
+#
+#   cargo run -p ddir-server < interactive/server/demo/counter.txt
+
+load counter begin
+let seed = input 0;
+let feedback = input 1;
+let state = seed + feedback;
+export "count" = state;
+export "next" = (state | map($0[0] + 1 ;)) + (seed | negate);
+counter end-load
+
+feed counter 0 0
+bind next counter 1
+tick 5
+
+# The closed past shows state 4 after five ticks: (0) stepped once per tick.
+peek count
+exit
diff --git a/interactive/server/demo/two_sessions.py b/interactive/server/demo/two_sessions.py
new file mode 100644
index 000000000..48100ef27
--- /dev/null
+++ b/interactive/server/demo/two_sessions.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""Two-session smoke test over TCP: convention-based races and the size gate.
+
+Sessions are trusted (identity by convention, no ownership gates): two
+clients race stamped-by-convention claims on one cell and the program's
+min-policy settles it deterministically; any session may bind into or drop
+any program; the one intake gate (program size) rejects oversized loads.
+
+Run from the repo root (release binary must be built):
+  python3 interactive/server/demo/two_sessions.py
+"""
+
+import os
+import socket
+import subprocess
+import sys
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+BIN = os.path.join(HERE, "..", "..", "..", "target", "release", "ddir_server")
+PORT = 7981
+
+WORLD = """load world begin
+let claims = input 0;
+export "owner" = claims | map($0 ; $1[1], $1[0]) | min | map($0 ; $1[1]);
+world end-load
+"""
+
+
+class Client:
+    def __init__(self, port):
+        for _ in range(100):
+            try:
+                self.sock = socket.create_connection(("127.0.0.1", port), timeout=5)
+                break
+            except OSError:
+                time.sleep(0.05)
+        else:
+            raise RuntimeError("server never came up")
+        self.buf = b""
+
+    def send(self, text):
+        self.sock.sendall(text.encode() if text.endswith("\n") else (text + "\n").encode())
+
+    def expect(self, reqid):
+        """Read until the terminal ok/err for `reqid`; return (status, body, data-lines)."""
+        data = []
+        while True:
+            while b"\n" not in self.buf:
+                chunk = self.sock.recv(4096)
+                if not chunk:
+                    raise RuntimeError("connection closed while waiting")
+                self.buf += chunk
+            line, self.buf = self.buf.split(b"\n", 1)
+            toks = line.decode().split(" ", 2)
+            if toks[0] != reqid:
+                continue
+            if toks[1] in ("ok", "err"):
+                return toks[1], toks[2] if len(toks) > 2 else "", data
+            if toks[1] == "data":
+                data.append(toks[2])
+
+
+def check(label, cond, detail=""):
+    print(("PASS " if cond else "FAIL ") + label + (f"  [{detail}]" if detail and not cond else ""))
+    if not cond:
+        sys.exit(1)
+
+
+def main():
+    env = dict(
+        os.environ,
+        DDIR_BIND=f"127.0.0.1:{PORT}",
+        DDIR_WS_BIND=f"127.0.0.1:{PORT + 1}",
+        DDIR_DIAG_PORT=str(PORT + 2),
+        DDIR_TICK_MS="0",
+        DDIR_MAX_PROGRAM_BYTES="4096",
+    )
+    server = subprocess.Popen(
+        [BIN], env=env, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL, text=True,
+    )
+    try:
+        a = Client(PORT)  # calls itself client 1 by convention
+        b = Client(PORT)  # calls itself client 2
+
+        a.send("rA " + WORLD)
+        status, body, _ = a.expect("rA")
+        check("session A loads the world", status == "ok", body)
+
+        # Both race for (5,5) with convention ids in the value: (id, epoch).
+        # The program's min over (epoch, id) settles it for client 1.
+        a.send("c1 feed world 0 5,5 val=1,0")
+        a.expect("c1")
+        b.send("c2 feed world 0 5,5 val=2,0")
+        b.expect("c2")
+        b.send("c3 feed world 0 2,2 val=2,0")
+        b.expect("c3")
+        b.send("t1 tick")
+        b.expect("t1")
+        a.send("p1 peek owner")
+        _, _, rows = a.expect("p1")
+        owned = {r.split("key=")[1].split(" val=")[0]: r.split("val=")[1] for r in rows}
+        check(
+            "same-epoch race on (5,5) goes to the lower convention id",
+            owned.get("Tuple([Int(5), Int(5)])") == "Tuple([Int(1)])",
+            str(owned),
+        )
+        check(
+            "uncontested (2,2) goes to client 2",
+            owned.get("Tuple([Int(2), Int(2)])") == "Tuple([Int(2)])",
+            str(owned),
+        )
+
+        # Sessions are trusted: B may bind into (and later drop) A's program.
+        b.send("b1 bind owner world 0")
+        status, body, _ = b.expect("b1")
+        check("B binds into A's world (trusted sessions)", status == "ok", body)
+        b.send("b2 unbind owner world 0")
+        status, body, _ = b.expect("b2")
+        check("B unbinds it again", status == "ok", body)
+
+        # The one gate: an oversized program body is swallowed, one error.
+        big = "load big begin\n" + ("let x = input 0; -- pad\n" * 300) + "big end-load\n"
+        a.send("s1 " + big)
+        status, body, _ = a.expect("s1")
+        check("oversized program body is rejected", status == "err" and "exceeds" in body, body)
+
+        b.send("d1 drop world")
+        status, body, _ = b.expect("d1")
+        check("B drops A's world (trusted sessions)", status == "ok", body)
+
+        a.sock.close()
+        b.sock.close()
+    finally:
+        try:
+            server.stdin.write("exit\n")
+            server.stdin.flush()
+        except OSError:
+            pass
+        code = server.wait(timeout=10)
+        check("server exits cleanly", code == 0, str(code))
+    print("all checks passed")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/interactive/server/demo/txn.txt b/interactive/server/demo/txn.txt
new file mode 100644
index 000000000..d257f964d
--- /dev/null
+++ b/interactive/server/demo/txn.txt
@@ -0,0 +1,113 @@
+# Transaction processing in the data plane, on the live server.
+# (After frankmcsherry/blog/posts/2025-04-27.md.)
+#
+# All writers do blind writes: a transaction is just rows appended to
+# `intents` — its writes, plus the reads under which those writes are valid
+# (the conditions under which they evaporate). A recursive view resolves
+# which transactions commit; committed state is a view; and the maintenance
+# deletions at the end are OUTPUT-INVARIANT, so any client may issue them at
+# any time (episodic, off the critical path) — compaction needs no authority
+# because it changes no outcome.
+#
+# Rows: key = (txn id, is_read, account), val = (value); accounts alice=1,
+# bob=2; "expect absent" reads use the sentinel -1. Ids order transactions;
+# the epoch seal (tick) is what makes resolved prefixes final.
+#
+#   cargo run -p ddir-server --release
+#   ./target/release/ddir_server < interactive/server/demo/txn.txt
+
+load bank begin
+let intents = input 0;
+
+-- ((rid, k) ; expected) and ((wid, k) ; v)
+let reads  = intents | filter($0[1] == 1) | map($0[0], $0[2] ; $1[0]);
+let wr_all = intents | filter($0[1] == 0) | map($0[0], $0[2] ; $1[0]);
+
+res: {
+    -- writes of transactions not (currently) rolled back: anti-join on id.
+    let wr_id  = wr_all | map($0[0] ; $0[1], $1[0]);
+    let wr_bad = wr_id | join(rollback, ($0 ; $1[0], $1[1]));
+    let writes = wr_id + (wr_bad | negate);
+
+    -- each read paired with every surviving prior write to its key ...
+    let wk = writes | map($1[0] ; $0[0], $1[1]);
+    let rk = reads  | map($0[1] ; $0[0], $1[0]);
+    let pairs = rk | join(wk, ($0, $1[0], $1[1] ; $2[0], $2[1]))
+              | filter($1[0] < $0[1]);
+    -- ... and the LATEST prior write wins: min over (-wid, v).
+    let best = pairs | map($0[0], $0[1], $0[2] ; 0 - $1[0], $1[1]) | min;
+
+    -- a read fails if the latest prior value mismatches ...
+    let bad_seen = best | filter($1[1] != $0[2]) | map($0[1] ;);
+
+    -- ... or if no prior write exists and it did not expect absence (-1).
+    let seen   = best | map($0[1], $0[0] ;);
+    let rkk    = reads | map($0 ;);
+    let unseen = rkk + (seen | negate);
+    let bad_absent = unseen | join(reads, ($0[0], $2[0] ;))
+                   | filter($0[1] != -1) | map($0[0] ;);
+
+    var rollback = (bad_seen + bad_absent) | distinct;
+}
+
+let rb = res::rollback;
+let wid  = wr_all | map($0[0] ; $0[1], $1[0]);
+let wbad = wid | join(rb, ($0 ; $1[0], $1[1]));
+let committed = wid + (wbad | negate);
+
+export "rollback" = rb;
+export "state" = committed | map($1[0] ; 0 - $0[0], $1[1]) | min | map($0 ; $1[1]);
+export "log" = intents;
+bank end-load
+
+# The five transactions from the post, all as blind appends:
+# T1: blind write alice=100.
+feed bank 0 1,0,1 val=100
+# T2: bob must be absent; write bob=50.
+feed bank 0 2,1,2 val=-1
+feed bank 0 2,0,2 val=50
+# T3: move 30 alice->bob, iff alice=100 and bob=50.
+feed bank 0 3,1,1 val=100
+feed bank 0 3,1,2 val=50
+feed bank 0 3,0,1 val=70
+feed bank 0 3,0,2 val=80
+# T4: move 40, against the SAME reads -- must roll back (T3 got there first).
+feed bank 0 4,1,1 val=100
+feed bank 0 4,1,2 val=50
+feed bank 0 4,0,1 val=60
+feed bank 0 4,0,2 val=90
+# T5: move 30 more, against T3's refreshed values -- commits.
+feed bank 0 5,1,1 val=70
+feed bank 0 5,1,2 val=80
+feed bank 0 5,0,1 val=50
+feed bank 0 5,0,2 val=100
+tick
+
+# Expect: rollback = {4}; state alice=50, bob=100.
+peek rollback
+peek state
+
+# --- Asynchronous maintenance (any client, any time; output-invariant) ---
+# Remove the failed transaction, the read sets of committed transactions,
+# and overwritten writes. The ids are sealed (their epoch has ticked), so
+# these deletions cannot change what committed.
+feed bank 0 4,1,1 val=100 diff=-1
+feed bank 0 4,1,2 val=50 diff=-1
+feed bank 0 4,0,1 val=60 diff=-1
+feed bank 0 4,0,2 val=90 diff=-1
+feed bank 0 2,1,2 val=-1 diff=-1
+feed bank 0 3,1,1 val=100 diff=-1
+feed bank 0 3,1,2 val=50 diff=-1
+feed bank 0 5,1,1 val=70 diff=-1
+feed bank 0 5,1,2 val=80 diff=-1
+feed bank 0 1,0,1 val=100 diff=-1
+feed bank 0 3,0,1 val=70 diff=-1
+feed bank 0 2,0,2 val=50 diff=-1
+feed bank 0 3,0,2 val=80 diff=-1
+tick
+
+# The log is now just the two live writes -- and state is UNCHANGED.
+peek log
+peek state
+peek rollback
+exit
diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs
new file mode 100644
index 000000000..d13298521
--- /dev/null
+++ b/interactive/server/src/cmd.rs
@@ -0,0 +1,770 @@
+//! 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;
+use std::panic::{catch_unwind, AssertUnwindSafe};
+
+use interactive::ir::{eval, Diff, Value};
+use interactive::server::OuterTime;
+
+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` — request the explain rewrite (reserved; the server
+    /// reports an error rather than improvising a meaning).
+    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 },
+    /// Update positional `input` of `prog`: add `(key, val)` with `diff` at
+    /// `time` (default the current epoch). Identity and ordering of writes
+    /// are convention, not enforcement: cooperating clients include their
+    /// session name and an ordering id in the data, and programs resolve
+    /// contention over those facts (see demo/claims.txt, demo/txn.txt).
+    Feed {
+        prog: String,
+        input: usize,
+        key: Value,
+        val: Value,
+        time: Option,
+        diff: Diff,
+    },
+    /// Bind a trace's changes into `prog`'s positional `input`, delivered at
+    /// each tick one epoch delayed — the write path for installed programs.
+    Bind {
+        trace: String,
+        prog: String,
+        input: usize,
+    },
+    /// Remove a binding installed by `bind`.
+    Unbind {
+        trace: String,
+        prog: String,
+        input: usize,
+    },
+    /// Push a row into the query input of a `--explain` dataflow
+    /// (reserved; unimplemented). Sign is `+1` for `add`, `-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,
+    max_load_bytes: usize,
+}
+
+/// 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", "feed", "bind",
+    "unbind",
+];
+
+/// Program-size gate: a `load` body larger than this is rejected at intake,
+/// before parsing — installs are cheap to request and costly to render, so
+/// the cap is the first line of defense against installation as denial of
+/// service. Override with `DDIR_MAX_PROGRAM_BYTES`.
+fn max_program_bytes() -> usize {
+    std::env::var("DDIR_MAX_PROGRAM_BYTES")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(65536)
+}
+
+struct PendingLoad {
+    reqid: ReqId,
+    id_hint: String,
+    bindings: BTreeMap,
+    explain: bool,
+    body: String,
+    /// Set once the body exceeds the size gate; the rest of the body is
+    /// swallowed (not stored) and `end-load` reports this error, so an
+    /// oversized upload cannot make the parser misread body text as commands.
+    poisoned: Option,
+}
+
+impl LineParser {
+    pub fn new() -> Self {
+        LineParser {
+            max_load_bytes: max_program_bytes(),
+            ..Self::default()
+        }
+    }
+
+    /// Test hook: a parser with an explicit program-size cap.
+    #[cfg(test)]
+    fn with_cap(max_load_bytes: usize) -> Self {
+        LineParser {
+            max_load_bytes,
+            ..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();
+                    if let Some(err) = done.poisoned {
+                        return Some((done.reqid, Err(err)));
+                    }
+                    return Some((
+                        done.reqid,
+                        Ok(Cmd::Load {
+                            id_hint: done.id_hint,
+                            bindings: done.bindings,
+                            program: done.body,
+                            explain: done.explain,
+                        }),
+                    ));
+                }
+            }
+            if pl.poisoned.is_none() {
+                pl.body.push_str(trimmed);
+                pl.body.push('\n');
+                if pl.body.len() > self.max_load_bytes {
+                    pl.poisoned = Some(format!(
+                        "load: program body exceeds {} bytes (DDIR_MAX_PROGRAM_BYTES)",
+                        self.max_load_bytes
+                    ));
+                    pl.body = String::new();
+                }
+            }
+            return None;
+        }
+
+        let trimmed = line.trim();
+        // Blank lines and `#` comments are skipped between commands (inside
+        // a load body every line is literal program text, handled above).
+        if trimmed.is_empty() || trimmed.starts_with('#') {
+            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(),
+                    poisoned: None,
+                });
+                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,
+    },
+}
+
+/// A `` for `feed`: a comma-separated integer row → `Tuple`, `_` or
+/// empty → unit, else a closed scalar term (one whitespace-free token, e.g.
+/// `inject(2,tuple(3,4))`) evaluated to a constant. Term-parser panics are
+/// caught here on the session thread and become clean errors — malformed
+/// input never reaches the worker.
+fn parse_value(s: &str) -> Result {
+    let s = s.trim();
+    if s.is_empty() || s == "_" {
+        return Ok(Value::unit());
+    }
+    if s.chars()
+        .all(|c| c.is_ascii_digit() || c == ',' || c == '-')
+    {
+        if let Ok(ints) = s
+            .split(',')
+            .map(|t| t.trim().parse::())
+            .collect::, _>>()
+        {
+            return Ok(Value::Tuple(ints.into_iter().map(Value::Int).collect()));
+        }
+    }
+    catch_unwind(AssertUnwindSafe(|| {
+        eval(&interactive::parse::pipe::parse_term(s), &mut Vec::new())
+    }))
+    .map_err(|panic| {
+        if let Some(msg) = panic.downcast_ref::<&str>() {
+            (*msg).to_string()
+        } else if let Some(msg) = panic.downcast_ref::() {
+            msg.clone()
+        } else {
+            format!("malformed value {:?}", s)
+        }
+    })
+}
+
+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,
+            })
+        }
+        "feed" => {
+            // Syntax: `feed    [val=] [time=] [diff=]`
+            // A ``/`` is a comma-separated integer row (`1,2` → tuple;
+            // `_`/empty → unit) or a closed scalar term written without
+            // spaces (`inject(2,tuple(3,4))`), as in the ddir_server example.
+            if args.len() < 3 {
+                return ParseOutcome::Err(
+                    "feed: expected `   [val=] [time=] [diff=]`".into(),
+                );
+            }
+            let prog = args[0].to_string();
+            let input: usize = match args[1].parse() {
+                Ok(n) => n,
+                Err(_) => {
+                    return ParseOutcome::Err(format!(
+                        "feed:  must be a number, got {:?}",
+                        args[1]
+                    ))
+                }
+            };
+            let key = match parse_value(args[2]) {
+                Ok(v) => v,
+                Err(e) => return ParseOutcome::Err(format!("feed key: {}", e)),
+            };
+            let mut val = Value::unit();
+            let mut time = None;
+            let mut diff: Diff = 1;
+            for tok in &args[3..] {
+                if let Some(v) = tok.strip_prefix("val=") {
+                    val = match parse_value(v) {
+                        Ok(v) => v,
+                        Err(e) => return ParseOutcome::Err(format!("feed val: {}", e)),
+                    };
+                } else if let Some(t) = tok.strip_prefix("time=") {
+                    time = match t.parse() {
+                        Ok(t) => Some(t),
+                        Err(_) => {
+                            return ParseOutcome::Err(format!(
+                                "feed: time= must be a number, got {:?}",
+                                t
+                            ))
+                        }
+                    };
+                } else if let Some(d) = tok.strip_prefix("diff=") {
+                    diff = match d.parse() {
+                        Ok(d) => d,
+                        Err(_) => {
+                            return ParseOutcome::Err(format!(
+                                "feed: diff= must be an integer, got {:?}",
+                                d
+                            ))
+                        }
+                    };
+                } else {
+                    return ParseOutcome::Err(format!("feed: unrecognized argument {:?}", tok));
+                }
+            }
+            ParseOutcome::Cmd(Cmd::Feed {
+                prog,
+                input,
+                key,
+                val,
+                time,
+                diff,
+            })
+        }
+        "bind" | "unbind" => match args {
+            [trace, prog, input] => match input.parse::() {
+                Ok(input) => {
+                    let (trace, prog) = ((*trace).to_string(), (*prog).to_string());
+                    if cmd == "bind" {
+                        ParseOutcome::Cmd(Cmd::Bind { trace, prog, input })
+                    } else {
+                        ParseOutcome::Cmd(Cmd::Unbind { trace, prog, input })
+                    }
+                }
+                Err(_) => {
+                    ParseOutcome::Err(format!("{}:  must be a number, got {:?}", cmd, input))
+                }
+            },
+            _ => ParseOutcome::Err(format!("{}: expected `  `", cmd)),
+        },
+        "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 feed_cmd() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &[
+                "r0 feed world 0 1,2",
+                "r1 feed world 0 3,4 val=7,8 time=5 diff=-1",
+                "r2 feed world 0 _ val=inject(2,tuple(3,4))",
+                "r3 feed world zero 1,2",     // bad input index
+                "r4 feed world 0 1,2 wat=7",  // unknown argument
+                "r5 feed world 0 tuple(1,",   // malformed term -> caught, not a panic
+            ],
+        );
+        assert_eq!(got.len(), 6);
+        match &got[0].1 {
+            Ok(Cmd::Feed { prog, input, key, val, time, diff }) => {
+                assert_eq!(prog, "world");
+                assert_eq!(*input, 0);
+                assert_eq!(*key, Value::Tuple(vec![Value::Int(1), Value::Int(2)]));
+                assert_eq!(*val, Value::unit());
+                assert_eq!(*time, None);
+                assert_eq!(*diff, 1);
+            }
+            other => panic!("expected Feed, got {:?}", other),
+        }
+        match &got[1].1 {
+            Ok(Cmd::Feed { time, diff, .. }) => {
+                assert_eq!(*time, Some(5));
+                assert_eq!(*diff, -1);
+            }
+            other => panic!("expected Feed, got {:?}", other),
+        }
+        assert!(matches!(&got[2].1, Ok(Cmd::Feed { key, .. }) if *key == Value::unit()));
+        assert!(got[3].1.is_err());
+        assert!(got[4].1.is_err());
+        assert!(got[5].1.is_err());
+    }
+
+    #[test]
+    fn bind_cmd() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &["r0 bind next counter 1", "r1 unbind next counter 1", "r2 bind next counter one"],
+        );
+        assert_eq!(got.len(), 3);
+        assert!(matches!(
+            &got[0].1,
+            Ok(Cmd::Bind { trace, prog, input: 1 }) if trace == "next" && prog == "counter"
+        ));
+        assert!(matches!(&got[1].1, Ok(Cmd::Unbind { input: 1, .. })));
+        assert!(got[2].1.is_err());
+    }
+
+    #[test]
+    fn oversized_load_is_rejected_cleanly() {
+        let mut p = LineParser::with_cap(64);
+        assert!(p.feed("r0 load big begin").is_none());
+        // Push well past the cap; every body line is swallowed, none parse
+        // as commands, and the terminator reports one clean error.
+        for _ in 0..16 {
+            assert!(p.feed("let x = input 0; -- padding padding padding").is_none());
+        }
+        let got = p.feed("r0 end-load").expect("end-load completes the upload");
+        assert_eq!(got.0, "r0");
+        assert!(got.1.as_ref().is_err_and(|e| e.contains("exceeds 64 bytes")));
+        // The parser is usable again afterwards.
+        assert!(matches!(p.feed("r1 list"), Some((_, Ok(Cmd::List)))));
+    }
+
+    #[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..d8d7f2329
--- /dev/null
+++ b/interactive/server/src/loop_.rs
@@ -0,0 +1,393 @@
+//! 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 reserved; explanation is not implemented here yet".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::Feed {
+            prog,
+            input,
+            key,
+            val,
+            time,
+            diff,
+        }) => server
+            .feed(&prog, input, key, val, time, diff)
+            .map(|()| format!("fed {:?} input {} at t={}", prog, input, server.epoch())),
+        Ok(Cmd::Bind { trace, prog, input }) => server
+            .bind(worker, &trace, &prog, input)
+            .map(|()| format!("bound {:?} -> {:?} input {}", trace, prog, input)),
+        Ok(Cmd::Unbind { trace, prog, input }) => server
+            .unbind(worker, &trace, &prog, input)
+            .map(|()| format!("unbound {:?} -> {:?} input {}", trace, prog, input)),
+        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),
+                );
+            }
+            for (source, target, input) in server.binding_info() {
+                send(
+                    &resp,
+                    &reqid,
+                    "data",
+                    format!("binding source={:?} target={:?} input={}", source, target, input),
+                );
+            }
+            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 is reserved for --explain dataflows and is not implemented"
+                .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 call spelling of a binding (`random(...)`) into its
+/// content-addressed source name.
+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..42fd1cca2
--- /dev/null
+++ b/interactive/server/src/main.rs
@@ -0,0 +1,382 @@
+//! 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);
+                // run_worker returns only on an operator `exit`. The worker
+                // still holds live dataflows (installed programs, generated
+                // sources, the diagnostics capture), so `execute_directly`'s
+                // trailing `while has_dataflows { step_or_park }` loop would
+                // park forever — and main's joins would never reach its own
+                // process::exit. Give session writers a beat to flush the
+                // final `ok bye`, then exit here: the documented
+                // "let the OS reap the parked listeners" shutdown, actually
+                // reached.
+                std::thread::sleep(Duration::from_millis(100));
+                std::process::exit(0);
+            });
+        })
+    };
+
+    // 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..84e086102 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -2,10 +2,8 @@
 //!
 //! Hosts a long-running timely worker group into which interpreted DDIR
 //! programs are *installed* one at a time, and lets them share results by name.
-//! This is the interpreter-driven successor to the legacy `dd_server` crate,
-//! which hot-loaded compiled `.so`s via `libloading`; here "install" means
-//! parse → lower → render an [`crate::scope_ir::Program`] against a live
-//! registry — no machine code, no `dlopen`.
+//! An install parses, lowers, and renders a [`crate::scope_ir::Program`]
+//! against a live registry of shared traces.
 //!
 //! # Typed commands
 //!
@@ -45,19 +43,20 @@ 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::operators::arrange::{ShutdownButton, TraceAgent};
 use differential_dataflow::trace::implementations::ValSpine;
-use differential_dataflow::dynamic::pointstamp::PointStamp;
+use differential_dataflow::trace::TraceReader;
+use differential_dataflow::VecCollection;
+use timely::dataflow::operators::CapabilitySet;
+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 +72,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 +93,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 +102,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 +126,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 +162,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,13 +192,32 @@ 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.
     Drop { name: String },
     /// Snapshot a registered trace (optionally one key) and print it (worker 0).
     Peek { trace: String, key: Option },
+    /// Bind trace `trace`'s changes into input `input` of `prog` at each tick.
+    Bind {
+        trace: String,
+        prog: String,
+        input: usize,
+    },
+    /// Remove a binding installed by `Bind`.
+    Unbind {
+        trace: String,
+        prog: String,
+        input: usize,
+    },
     /// Print the registry (worker 0).
     List,
     /// Print the command help (worker 0).
@@ -188,6 +244,52 @@ 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 export→input binding: a persistent tap on a published trace whose
+/// buffered changes are fed into a program's positional input at each tick.
+///
+/// This is the discrete-time feedback primitive: the target input receives
+/// the source's *changes*, one epoch delayed — and since changes telescope,
+/// the input's accumulation MIRRORS the source as of the previous epoch
+/// (plus whatever else was fed to it), with no client round-trip.
+///
+/// The state-machine idiom: give the program a seed input and a dedicated
+/// feedback input, `let state = seed + feedback;`, and bind the export
+/// `f(state) + (seed | negate)` to the feedback input. Then
+/// `state(t) = seed + f(state(t-1)) - seed = f(state(t-1))` — one step of
+/// the recursion per tick, entirely inside the server, while later seed
+/// changes still inject as perturbations.
+struct Binding {
+    /// Canonical name of the tapped trace.
+    source: String,
+    /// Target program name.
+    target: String,
+    /// Target positional input.
+    input: usize,
+    /// Changes captured since the last drain, times collapsed. Filled by the
+    /// tap dataflow's inspect as the worker steps; drained by `tick`.
+    buffer: Rc>>,
+    /// The tap dataflow's id, for teardown on `unbind`.
+    dataflow_id: usize,
+    /// The tap's probe: `tick` must wait on it so the buffer holds every
+    /// change through the just-closed epoch before draining.
+    probe: ProbeHandle,
+    /// Keeps the tap's import alive; dropped (deactivating the operator)
+    /// together with the binding.
+    _shutdown: ShutdownButton>,
 }
 
 /// A live registry of installed programs and the traces they publish.
@@ -197,7 +299,10 @@ pub struct Server {
     /// Installed program name -> its handles and lifecycle bookkeeping.
     programs: HashMap,
     /// Trace name -> number of installed programs importing it (the drop gate).
+    /// Bindings count here too: a bound source cannot be dropped.
     importers: HashMap,
+    /// Live export→input bindings, drained by each `tick`.
+    bindings: Vec,
     /// The current open epoch; inputs sit here until `tick` closes it.
     epoch: OuterTime,
 }
@@ -209,15 +314,60 @@ impl Server {
             traces: HashMap::new(),
             programs: HashMap::new(),
             importers: HashMap::new(),
+            bindings: Vec::new(),
             epoch: 0,
         }
     }
 
     /// 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 +379,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 +400,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 +414,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 +437,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 +449,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 +500,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 +544,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 +582,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,22 +601,160 @@ 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(())
     }
 
+    /// Bind trace `trace` to positional `input` of program `prog`: from now
+    /// on, every tick delivers the trace's *changes* into that input at the
+    /// next epoch, so the input mirrors the trace one epoch delayed. The
+    /// write path for programs — an installed dataflow can now act on the
+    /// world (or on itself, see [`Binding`]) without any client in the loop.
+    ///
+    /// Sharding: each worker's tap sees its shard of the trace and feeds its
+    /// local input handle, so the union across workers delivers the full
+    /// delta exactly once; the input's exchange re-routes as usual.
+    ///
+    /// The bound source gains an importer (it cannot be dropped while
+    /// bound); the target cannot be dropped either (see `drop_program`).
+    /// Errors: unknown trace or program, non-writable target (generated or
+    /// clock), no such input, or the identical binding already exists.
+    pub fn bind(
+        &mut self,
+        worker: &mut Worker,
+        trace: &str,
+        prog: &str,
+        input: usize,
+    ) -> Result<(), String> {
+        let source = canonical_source_name(trace);
+        let target = prog.to_string();
+        if !self.traces.contains_key(&source) {
+            return Err(format!("no trace {:?}", source));
+        }
+        let installed = self
+            .programs
+            .get(&target)
+            .ok_or_else(|| format!("no program {:?}", target))?;
+        if installed.origin != Origin::Program {
+            return Err(format!("{:?} is not a writable program", target));
+        }
+        if !installed.inputs.contains_key(&input) {
+            return Err(format!("program {:?} has no input {}", target, input));
+        }
+        if self
+            .bindings
+            .iter()
+            .any(|b| b.source == source && b.target == target && b.input == input)
+        {
+            return Err(format!(
+                "trace {:?} is already bound to {:?} input {}",
+                source, target, input
+            ));
+        }
+
+        let buffer: Rc>> = Rc::new(RefCell::new(Vec::new()));
+        let buffer_in = buffer.clone();
+        let mut probe = ProbeHandle::new();
+        let dataflow_id = worker.next_dataflow_index();
+        let trace_handle = self.traces.get_mut(&source).expect("checked above");
+        let shutdown = worker.dataflow::(|scope| {
+            let (arranged, shutdown) = trace_handle.import_core(scope.clone(), "BindImport");
+            arranged
+                .as_collection(|k, v| (k.clone(), v.clone()))
+                .inspect(move |((key, val), _time, diff)| {
+                    buffer_in
+                        .borrow_mut()
+                        .push(((key.clone(), val.clone()), *diff));
+                })
+                .probe_with(&mut probe);
+            shutdown
+        });
+
+        *self.importers.entry(source.clone()).or_insert(0) += 1;
+        self.bindings.push(Binding {
+            source,
+            target,
+            input,
+            buffer,
+            dataflow_id,
+            probe,
+            _shutdown: shutdown,
+        });
+        Ok(())
+    }
+
+    /// Remove the binding of `trace` into `prog`'s `input`, dropping its tap
+    /// dataflow and releasing the source's importer count.
+    pub fn unbind(
+        &mut self,
+        worker: &mut Worker,
+        trace: &str,
+        prog: &str,
+        input: usize,
+    ) -> Result<(), String> {
+        let source = canonical_source_name(trace);
+        let pos = self
+            .bindings
+            .iter()
+            .position(|b| b.source == source && b.target == prog && b.input == input)
+            .ok_or_else(|| {
+                format!(
+                    "no binding of {:?} to {:?} input {}",
+                    source, prog, input
+                )
+            })?;
+        let binding = self.bindings.remove(pos);
+        if let Some(count) = self.importers.get_mut(&binding.source) {
+            *count = count.saturating_sub(1);
+        }
+        worker.drop_dataflow(binding.dataflow_id);
+        Ok(())
+    }
+
+    /// The live bindings, as `(source-trace, target-program, input)`.
+    pub fn binding_info(&self) -> Vec<(String, String, usize)> {
+        self.bindings
+            .iter()
+            .map(|b| (b.source.clone(), b.target.clone(), b.input))
+            .collect()
+    }
+
     /// Read a snapshot of a registered trace and print it (worker 0).
     ///
     /// Builds a transient dataflow that imports the trace, optionally filters to
@@ -430,7 +762,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 +796,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 +812,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 +826,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
@@ -493,13 +882,25 @@ impl Server {
     /// `worker.drop_dataflow`, which removes the operators and frees their state
     /// at once. Safe because the gate guarantees no live dataflow still reads it.
     pub fn drop_program(&mut self, worker: &mut Worker, name: &str) -> Result<(), String> {
+        if let Some(binding) = self.bindings.iter().find(|b| b.target == name) {
+            return Err(format!(
+                "cannot drop {:?}: its input {} is bound from trace {:?}; unbind first",
+                name, binding.input, binding.source
+            ));
+        }
         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 +918,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 +950,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();
@@ -543,12 +978,42 @@ impl Server {
         self.epoch = next;
 
         // Wait for every *live* program to catch up. Per-program probes mean a
-        // dropped program leaves nothing behind to wait on.
+        // dropped program leaves nothing behind to wait on. Binding taps are
+        // waited on too, so each buffer holds every change through the epoch
+        // just closed before it is drained below.
         let epoch = self.epoch;
-        while self.programs.values().any(|p| p.probe.less_than(&epoch)) {
+        while self.programs.values().any(|p| p.probe.less_than(&epoch))
+            || self.bindings.iter().any(|b| b.probe.less_than(&epoch))
+        {
             worker.step();
         }
 
+        // Feedback: deliver each binding's buffered source changes into its
+        // target input at the (new, open) epoch — they become visible when
+        // the NEXT tick closes it. One-epoch delay is what makes the loop
+        // well-founded: each tick performs exactly one step of any
+        // program-to-program (or program-to-self) recursion.
+        let bindings = &self.bindings;
+        let programs = &mut self.programs;
+        for binding in bindings {
+            let mut buffer = binding.buffer.borrow_mut();
+            if buffer.is_empty() {
+                continue;
+            }
+            let handle = programs
+                .get_mut(&binding.target)
+                .and_then(|p| p.inputs.get_mut(&binding.input));
+            if let Some(handle) = handle {
+                for ((key, val), diff) in buffer.drain(..) {
+                    handle.update_at((key, val), epoch, diff);
+                }
+            } else {
+                // The target vanished; drop_program refuses while bound, so
+                // this is unreachable — but never let the buffer grow.
+                buffer.clear();
+            }
+        }
+
         // Allow every published trace to compact up to the previous epoch. This
         // is safe even while another program is importing the trace: each
         // importer is a separate `TraceAgent` whose contribution holds the shared
@@ -577,7 +1042,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 +1060,22 @@ 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
+            );
+        }
+        if !self.bindings.is_empty() {
+            println!("bindings ({}):", self.bindings.len());
+            for b in &self.bindings {
+                println!("  {} -> {} input {}", b.source, b.target, b.input);
+            }
         }
     }
 }
 
 impl Default for Server {
-    fn default() -> Self { Server::new() }
+    fn default() -> Self {
+        Server::new()
+    }
 }
diff --git a/interactive/tests/server_bind.rs b/interactive/tests/server_bind.rs
new file mode 100644
index 000000000..28cc012a0
--- /dev/null
+++ b/interactive/tests/server_bind.rs
@@ -0,0 +1,96 @@
+//! Semantics of `Server::bind` — export→input feedback, one epoch per tick.
+//!
+//! The counter is the canonical case: a program whose next state is a pure
+//! function of its current state, advanced by the server alone. The client's
+//! only acts are one seed row and one `bind`; every subsequent step happens
+//! because `tick` drains the bound export's changes back into the feedback
+//! input. See the `Binding` docs for the `f(state) + (seed | negate)` idiom.
+
+use interactive::ir::Value;
+use interactive::server::Server;
+use interactive::{lower, parse};
+
+fn tup(fields: &[i64]) -> Value {
+    Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect())
+}
+
+fn install(server: &mut Server, worker: &mut timely::worker::Worker, name: &str, src: &str) {
+    let statements = parse::pipe::parse(src);
+    let mut program = lower::lower_tree(statements);
+    program.optimize();
+    server.install(worker, name, &program).unwrap();
+}
+
+const COUNTER: &str = r#"
+    let seed = input 0;
+    let feedback = input 1;
+    let state = seed + feedback;
+    export "count" = state;
+    export "next" = (state | map($0[0] + 1 ;)) + (seed | negate);
+"#;
+
+#[test]
+fn bind_advances_a_counter_without_a_client() {
+    timely::execute_directly(move |worker| {
+        let mut server = Server::new();
+        install(&mut server, worker, "counter", COUNTER);
+        server
+            .feed("counter", 0, tup(&[0]), Value::unit(), None, 1)
+            .unwrap();
+        server.bind(worker, "next", "counter", 1).unwrap();
+
+        // state(t) = f(state(t-1)) = state(t-1) + 1, one step per tick. After
+        // N ticks the snapshot (which reads the closed past) shows N - 1.
+        for expected in 0..5 {
+            server.tick(worker);
+            let rows = server.snapshot(worker, "count").unwrap();
+            assert_eq!(rows, vec![(tup(&[expected]), Value::unit(), 1)]);
+        }
+
+        // A later seed change injects as a perturbation: adding a second
+        // token forks the counter into two independent tracks.
+        server
+            .feed("counter", 0, tup(&[100]), Value::unit(), None, 1)
+            .unwrap();
+        server.tick(worker);
+        server.tick(worker);
+        let rows = server.snapshot(worker, "count").unwrap();
+        assert_eq!(
+            rows,
+            vec![
+                (tup(&[6]), Value::unit(), 1),
+                (tup(&[101]), Value::unit(), 1)
+            ]
+        );
+    });
+}
+
+#[test]
+fn bind_lifecycle_guards() {
+    timely::execute_directly(move |worker| {
+        let mut server = Server::new();
+        install(&mut server, worker, "counter", COUNTER);
+
+        // Unknown pieces are rejected.
+        assert!(server.bind(worker, "nope", "counter", 1).is_err());
+        assert!(server.bind(worker, "next", "nope", 1).is_err());
+        assert!(server.bind(worker, "next", "counter", 7).is_err());
+
+        server.bind(worker, "next", "counter", 1).unwrap();
+        // Identical duplicate is rejected; the binding is listed.
+        assert!(server.bind(worker, "next", "counter", 1).is_err());
+        assert_eq!(
+            server.binding_info(),
+            vec![("next".to_string(), "counter".to_string(), 1)]
+        );
+
+        // While bound, the program can be dropped from neither side: it is
+        // its own source's exporter (importer refcount) and the binding's
+        // target (explicit guard).
+        assert!(server.drop_program(worker, "counter").is_err());
+
+        server.unbind(worker, "next", "counter", 1).unwrap();
+        assert!(server.unbind(worker, "next", "counter", 1).is_err());
+        server.drop_program(worker, "counter").unwrap();
+    });
+}