From c3b04ff46af00953f50dd79d891766fe3e449393 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 16 Aug 2026 22:00:53 -0400 Subject: [PATCH 1/3] Port live DDIR server to current interpreter Port of prior work on a "server" for DDIR, in which various dataflows go through a create-run-delete lifecycle, concurrently. Rebased over master-next (#838-#849); the only conflict was Cargo.toml against #835's removal of the legacy server crates. Folds in three fixes that make the port work when scripted: - Reach the documented process exit on operator 'exit': execute_directly runs a trailing 'while has_dataflows { step_or_park }' loop, and the worker always still holds live dataflows (installed programs, generated sources, the diagnostics capture), so it parked forever and main's joins never reached their own process::exit. Exit where the shutdown decision lands, after a beat for session writers to flush the final 'ok bye'. - Skip '#' comment lines between commands, so annotated command scripts pipe straight to stdin. - README: mark auto-tick as demo furniture, not a design commitment (it makes observation advance time; a watcher-must-drive-time design may be better once the server has real tenants). --- Cargo.toml | 1 + interactive/server/Cargo.toml | 21 + interactive/server/README.md | 49 +++ interactive/server/console.html | 687 ++++++++++++++++++++++++++++++++ interactive/server/src/cmd.rs | 511 ++++++++++++++++++++++++ interactive/server/src/loop_.rs | 369 +++++++++++++++++ interactive/server/src/main.rs | 382 ++++++++++++++++++ interactive/src/server.rs | 433 ++++++++++++++++---- 8 files changed, 2373 insertions(+), 80 deletions(-) create mode 100644 interactive/server/Cargo.toml create mode 100644 interactive/server/README.md create mode 100644 interactive/server/console.html create mode 100644 interactive/server/src/cmd.rs create mode 100644 interactive/server/src/loop_.rs create mode 100644 interactive/server/src/main.rs 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/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..0190fd6f3 --- /dev/null +++ b/interactive/server/README.md @@ -0,0 +1,49 @@ +# 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 annotated +command scripts can be piped straight to stdin. + +The useful commands are `load`, `drop`, `list`, `peek`, `tail`, `stop`, `tick`, +and `exit`. `load` accepts an inline pipe-syntax program: + + load graph begin + let edges = import "random:nodes=8,edges=12,seed=1,churn=1"; + export "graph.edges" = edges; + graph end-load + tail graph.edges + +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. + +`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..49edc1d33 --- /dev/null +++ b/interactive/server/console.html @@ -0,0 +1,687 @@ + + + + +ddir_server console + + + + +
+ ddir_server console + + +
disconnected
+
+ +
+ + + +
+
+

Load a dataflow

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

Protocol log

+ 

+    

+  

+
+

+
+

+  
+  
+

+
+
+
+
+
diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs
new file mode 100644
index 000000000..8f432db57
--- /dev/null
+++ b/interactive/server/src/cmd.rs
@@ -0,0 +1,511 @@
+//! Command and response shapes for the line-oriented protocol.
+//!
+//! Each request line is `  [args...]`.
+//! Each response line starts with the same `` followed by one of:
+//!   - `ok [body...]` — terminal success line
+//!   - `err [body...]` — terminal error line
+//!   - `data ` — one streamed body line (peek/tail batches)
+//!   - `end` — terminator after a stream of `data` lines
+//!
+//! Multi-line bodies (a DDIR program) come via a two-phase upload:
+//! ` load   begin` opens; subsequent lines are
+//! literal program text terminated by ` end-load`.
+
+use std::collections::BTreeMap;
+
+pub type ReqId = String;
+
+/// One client session's outbound stream. Cloned into each `Request` so
+/// dispatch can route responses back to the originating client (and so
+/// long-lived subscriptions like `tail` capture the right sender).
+pub type RespSender = std::sync::mpsc::Sender;
+
+/// Per-session identity. Lets the worker tear down long-lived
+/// subscriptions when a connection disappears without an explicit stop.
+pub type ConnectionId = u64;
+
+#[derive(Debug)]
+pub enum Cmd {
+    /// Install a dataflow.
+    /// `id_hint` — a client-chosen name; the server may keep it or assign
+    /// a fresh id (echo'd in the response).
+    /// `bindings` — `import-name -> binding`, where the binding is either a
+    /// registered trace name or a builtin call (`random(...)`).
+    /// `program` — DDIR text.
+    /// `explain` — 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 },
+    /// 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,
+}
+
+/// Tokens that introduce a command. If a line begins with one of these
+/// instead of an explicit reqid, the parser synthesizes a reqid.
+const COMMAND_KEYWORDS: &[&str] = &[
+    "load", "drop", "list", "peek", "tail", "stop", "tick", "query", "exit",
+];
+
+struct PendingLoad {
+    reqid: ReqId,
+    id_hint: String,
+    bindings: BTreeMap,
+    explain: bool,
+    body: String,
+}
+
+impl LineParser {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Feed one input line; return `Some((reqid, parsed))` if the line
+    /// completes a command (single-line or end of a multi-line body),
+    /// else `None` to indicate more input is required. The caller pairs
+    /// the result with a per-connection response sender to form a
+    /// `Request`.
+    pub fn feed(&mut self, line: &str) -> Option<(ReqId, Result)> {
+        // Inside a pending load body: every line is literal program text
+        // until ` end-load` or ` end-load`. The id_hint
+        // form is the friendly default when the load was auto-reqid'd
+        // (so the user can type `gen end-load` after `load gen … begin`
+        // without having to know the minted reqid).
+        if let Some(ref mut pl) = self.pending_load {
+            let trimmed = line.trim_end_matches(['\r', '\n']);
+            let mut parts = trimmed.split_whitespace();
+            if let (Some(tok0), Some(tok1), None) = (parts.next(), parts.next(), parts.next()) {
+                if (tok0 == pl.reqid || tok0 == pl.id_hint) && tok1 == "end-load" {
+                    let done = self.pending_load.take().unwrap();
+                    return Some((
+                        done.reqid,
+                        Ok(Cmd::Load {
+                            id_hint: done.id_hint,
+                            bindings: done.bindings,
+                            program: done.body,
+                            explain: done.explain,
+                        }),
+                    ));
+                }
+            }
+            pl.body.push_str(trimmed);
+            pl.body.push('\n');
+            return None;
+        }
+
+        let trimmed = line.trim();
+        // Blank lines and `#` comments are skipped between commands (inside
+        // a load body every line is literal program text, handled above), so
+        // annotated command scripts can be piped straight to stdin.
+        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(),
+                });
+                None
+            }
+        }
+    }
+
+    /// True if waiting for a ` end-load`. WS transport uses this
+    /// to forward blank-line program body content verbatim.
+    pub fn awaiting_body(&self) -> bool {
+        self.pending_load.is_some()
+    }
+}
+
+enum ParseOutcome {
+    Cmd(Cmd),
+    Err(String),
+    BeginLoad {
+        id_hint: String,
+        bindings: BTreeMap,
+        explain: bool,
+    },
+}
+
+fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
+    match cmd {
+        "load" => {
+            // Syntax: `load  [--explain] [name=binding ...] begin`
+            // The trailing `begin` switches the parser into body-collection
+            // mode; subsequent input lines are program text terminated by
+            // ` end-load`.
+            if args.is_empty() {
+                return ParseOutcome::Err(
+                    "load: expected ` [--explain] [name=binding ...] begin`".into(),
+                );
+            }
+            if args.last() != Some(&"begin") {
+                return ParseOutcome::Err(
+                    "load: must end with `begin` (multi-line body required)".into(),
+                );
+            }
+            let id_hint = args[0].to_string();
+            let middle = &args[1..args.len() - 1];
+            let mut bindings = BTreeMap::new();
+            let mut explain = false;
+            for tok in middle {
+                if *tok == "--explain" {
+                    explain = true;
+                    continue;
+                }
+                let Some((k, v)) = tok.split_once('=') else {
+                    return ParseOutcome::Err(format!(
+                        "load: argument {:?} must be `--explain` or `name=binding`",
+                        tok
+                    ));
+                };
+                if bindings.insert(k.to_string(), v.to_string()).is_some() {
+                    return ParseOutcome::Err(format!("load: duplicate binding for {:?}", k));
+                }
+            }
+            ParseOutcome::BeginLoad {
+                id_hint,
+                bindings,
+                explain,
+            }
+        }
+        "query" => {
+            // Syntax: `query  add|del  ; `
+            // Where k/v-fields are comma-separated i64. Empty side allowed
+            // (write nothing before/after the `;`).
+            if args.len() < 3 {
+                return ParseOutcome::Err(
+                    "query: expected ` add|del  ; `".into(),
+                );
+            }
+            let target = match args[0].parse::() {
+                Ok(n) => DataflowRef::Id(n),
+                Err(_) => DataflowRef::Name(args[0].to_string()),
+            };
+            let kind = match args[1] {
+                "add" => QueryKind::Add,
+                "del" => QueryKind::Del,
+                other => {
+                    return ParseOutcome::Err(format!(
+                        "query: kind must be add|del, got {:?}",
+                        other
+                    ))
+                }
+            };
+            // Find the `;` separator among the remaining tokens.
+            let rest = &args[2..];
+            let sep = rest.iter().position(|t| *t == ";");
+            let (k_toks, v_toks): (&[&str], &[&str]) = match sep {
+                Some(i) => (&rest[..i], &rest[i + 1..]),
+                None => (rest, &[]),
+            };
+            fn parse_fields(toks: &[&str]) -> Result, String> {
+                let mut out = Vec::new();
+                for t in toks {
+                    for piece in t.split(',') {
+                        if piece.is_empty() {
+                            continue;
+                        }
+                        out.push(piece.parse().map_err(|_| format!("bad i64 {:?}", piece))?);
+                    }
+                }
+                Ok(out)
+            }
+            let key = match parse_fields(k_toks) {
+                Ok(v) => v,
+                Err(e) => return ParseOutcome::Err(format!("query key: {}", e)),
+            };
+            let val = match parse_fields(v_toks) {
+                Ok(v) => v,
+                Err(e) => return ParseOutcome::Err(format!("query val: {}", e)),
+            };
+            ParseOutcome::Cmd(Cmd::Query {
+                target,
+                kind,
+                key,
+                val,
+            })
+        }
+        "drop" => match args {
+            [tok] => {
+                let target = match tok.parse::() {
+                    Ok(n) => DataflowRef::Id(n),
+                    Err(_) => DataflowRef::Name((*tok).to_string()),
+                };
+                ParseOutcome::Cmd(Cmd::Drop { target })
+            }
+            _ => ParseOutcome::Err("drop: expected ``".into()),
+        },
+        "list" => match args {
+            [] => ParseOutcome::Cmd(Cmd::List),
+            _ => ParseOutcome::Err("list: takes no arguments".into()),
+        },
+        "peek" => match args {
+            [name] => ParseOutcome::Cmd(Cmd::Peek {
+                name: (*name).to_string(),
+            }),
+            _ => ParseOutcome::Err("peek: expected ``".into()),
+        },
+        "tail" => match args {
+            [name] => ParseOutcome::Cmd(Cmd::Tail {
+                name: (*name).to_string(),
+            }),
+            _ => ParseOutcome::Err("tail: expected ``".into()),
+        },
+        "stop" => match args {
+            [rid] => ParseOutcome::Cmd(Cmd::Stop {
+                tail_reqid: (*rid).to_string(),
+            }),
+            _ => ParseOutcome::Err("stop: expected ``".into()),
+        },
+        "tick" => match args {
+            [] => ParseOutcome::Cmd(Cmd::Tick { n: 1 }),
+            [n] => match n.parse::() {
+                Ok(n) => ParseOutcome::Cmd(Cmd::Tick { n }),
+                Err(_) => ParseOutcome::Err(format!("tick: bad count {:?}", n)),
+            },
+            _ => ParseOutcome::Err("tick: expected `[n]`".into()),
+        },
+        "exit" => ParseOutcome::Cmd(Cmd::Exit),
+        other => ParseOutcome::Err(format!("unknown command {:?}", other)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn feed_all(p: &mut LineParser, lines: &[&str]) -> Vec<(ReqId, Result)> {
+        let mut out = Vec::new();
+        for l in lines {
+            if let Some(r) = p.feed(l) {
+                out.push(r);
+            }
+        }
+        out
+    }
+
+    #[test]
+    fn simple_commands() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &["r0 list", "r1 tick 5", "r2 drop 3", "r3 peek foo"],
+        );
+        assert_eq!(got.len(), 4);
+        assert!(matches!(got[0].1, Ok(Cmd::List)));
+        assert!(matches!(got[1].1, Ok(Cmd::Tick { n: 5 })));
+        assert!(matches!(
+            got[2].1,
+            Ok(Cmd::Drop {
+                target: DataflowRef::Id(3)
+            })
+        ));
+        assert!(matches!(got[3].1, Ok(Cmd::Peek { ref name }) if name == "foo"));
+    }
+
+    #[test]
+    fn multiline_load() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &[
+                "r0 load gen edges=random(seed=1) begin",
+                "let edges = import \"edges/v1\";",
+                "export \"reach\" = edges;",
+                "r0 end-load",
+            ],
+        );
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Load {
+                id_hint,
+                bindings,
+                program,
+                explain,
+            }) => {
+                assert_eq!(id_hint, "gen");
+                assert_eq!(
+                    bindings.get("edges").map(String::as_str),
+                    Some("random(seed=1)")
+                );
+                assert!(program.contains("import \"edges/v1\""));
+                assert!(program.contains("export \"reach\""));
+                assert!(!*explain);
+            }
+            _ => panic!("expected Load, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn load_explain() {
+        let mut p = LineParser::new();
+        let got = feed_all(&mut p, &[
+            "rE load reach --explain edges=random(seed=1,arity=2,range=10,count=2,churn=0) begin",
+            "export \"reach_out\" = import \"edges\";",
+            "rE end-load",
+        ]);
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Load { explain, .. }) => assert!(*explain),
+            _ => panic!("expected explain Load, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn query_cmd() {
+        let mut p = LineParser::new();
+        let got = feed_all(&mut p, &["rQ query 3 add 1,2 ; 99"]);
+        assert_eq!(got.len(), 1);
+        match &got[0].1 {
+            Ok(Cmd::Query {
+                target,
+                kind,
+                key,
+                val,
+            }) => {
+                assert!(matches!(target, DataflowRef::Id(3)));
+                assert!(matches!(kind, QueryKind::Add));
+                assert_eq!(key, &vec![1, 2]);
+                assert_eq!(val, &vec![99]);
+            }
+            _ => panic!("expected Query, got {:?}", got[0].1),
+        }
+    }
+
+    #[test]
+    fn auto_reqid_for_bare_command() {
+        let mut p = LineParser::new();
+        let got = feed_all(&mut p, &["list", "tick 3", "rA peek foo", "exit"]);
+        assert_eq!(got.len(), 4);
+        // Bare commands get auto-minted reqids; mixed-in explicit reqids
+        // pass through unchanged.
+        assert_eq!(got[0].0, "_1"); // list
+        assert!(matches!(got[0].1, Ok(Cmd::List)));
+        assert_eq!(got[1].0, "_2"); // tick 3
+        assert!(matches!(got[1].1, Ok(Cmd::Tick { n: 3 })));
+        assert_eq!(got[2].0, "rA"); // explicit reqid preserved
+        assert!(matches!(got[2].1, Ok(Cmd::Peek { ref name }) if name == "foo"));
+        assert_eq!(got[3].0, "_3"); // exit
+        assert!(matches!(got[3].1, Ok(Cmd::Exit)));
+    }
+
+    #[test]
+    fn drop_by_id_or_name() {
+        let mut p = LineParser::new();
+        let got = feed_all(
+            &mut p,
+            &[
+                "r0 drop 3",
+                "r1 drop my_reach",
+                "r2 drop", // missing arg → err
+            ],
+        );
+        assert_eq!(got.len(), 3);
+        assert!(matches!(
+            got[0].1,
+            Ok(Cmd::Drop {
+                target: DataflowRef::Id(3)
+            })
+        ));
+        assert!(
+            matches!(got[1].1, Ok(Cmd::Drop { target: DataflowRef::Name(ref n) }) if n == "my_reach")
+        );
+        assert!(matches!(got[2].1, Err(_)));
+    }
+}
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
new file mode 100644
index 000000000..7a2d70ddc
--- /dev/null
+++ b/interactive/server/src/loop_.rs
@@ -0,0 +1,369 @@
+//! 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::List) => {
+            for program in server.program_info() {
+                send(
+                    &resp,
+                    &reqid,
+                    "data",
+                    format!(
+                        "program name={:?} origin={} inputs={:?} imports={:?} exports={:?}",
+                        program.name,
+                        program.origin,
+                        program.inputs,
+                        program.imports,
+                        program.exports
+                    ),
+                );
+            }
+            for (name, importers) in server.trace_info() {
+                send(
+                    &resp,
+                    &reqid,
+                    "data",
+                    format!("trace name={:?} importers={}", name, importers),
+                );
+            }
+            Ok(format!("t={}", server.epoch()))
+        }
+        Ok(Cmd::Peek { name }) => match server.snapshot(worker, &name) {
+            Ok(rows) => {
+                for (key, val, diff) in rows {
+                    send(
+                        &resp,
+                        &reqid,
+                        "data",
+                        format!("diff={} key={:?} val={:?}", diff, key, val),
+                    );
+                }
+                Ok(format!("t={}", server.epoch()))
+            }
+            Err(e) => Err(e),
+        },
+        Ok(Cmd::Tail { name }) => start_tail(
+            connection_id,
+            &reqid,
+            &name,
+            resp.clone(),
+            server,
+            tails,
+            worker,
+        )
+        .map(|()| format!("tailing {:?} from t={}", name, server.epoch())),
+        Ok(Cmd::Stop { tail_reqid }) => {
+            let key = (connection_id, tail_reqid.clone());
+            match tails.remove(&key) {
+                Some(tail) => {
+                    worker.drop_dataflow(tail.dataflow_id);
+                    send(&resp, &tail_reqid, "end", String::new());
+                    Ok(format!("stopped {}", tail_reqid))
+                }
+                None => Err(format!("no tail {:?} in this session", tail_reqid)),
+            }
+        }
+        Ok(Cmd::Tick { n }) => {
+            for _ in 0..n {
+                tick(server, tails, worker);
+            }
+            Ok(format!("t={}", server.epoch()))
+        }
+        Ok(Cmd::Query { .. }) => Err(
+            "query 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..dbdb8e071 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,19 @@ use std::cell::RefCell;
 use std::collections::HashMap;
 use std::rc::Rc;
 
-use timely::worker::Worker;
-use timely::dataflow::ProbeHandle;
-use timely::progress::Antichain;
-use differential_dataflow::VecCollection;
+use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::input::{Input, InputSession};
 use differential_dataflow::operators::arrange::TraceAgent;
-use differential_dataflow::trace::TraceReader;
 use differential_dataflow::trace::implementations::ValSpine;
-use differential_dataflow::dynamic::pointstamp::PointStamp;
+use differential_dataflow::trace::TraceReader;
+use differential_dataflow::VecCollection;
+use timely::dataflow::ProbeHandle;
+use timely::progress::Antichain;
+use timely::worker::Worker;
 
-use crate::ir::{Value, Diff};
-use crate::scope_ir as st;
 use crate::backend::vec::render_tree;
+use crate::ir::{Diff, Value};
+use crate::scope_ir as st;
 
 /// The host (outer) timestamp shared across all installed programs.
 pub type OuterTime = u64;
@@ -73,9 +71,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 +92,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 +101,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 +125,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 +161,24 @@ impl Recipe {
 /// Where an installed entry came from. Only `Program` is writable by `feed`;
 /// `Clock` additionally has its single row advanced each `tick`.
 #[derive(Clone, Copy, PartialEq)]
-enum Origin { Program, Generated, Clock }
+enum Origin {
+    Program,
+    Generated,
+    Clock,
+}
 
 /// The single `clock` row for epoch `t`: `(Tuple[t] ; ())`.
-fn clock_row(t: OuterTime) -> Value { Value::Tuple(vec![Value::Int(t as i64)]) }
+fn clock_row(t: OuterTime) -> Value {
+    Value::Tuple(vec![Value::Int(t as i64)])
+}
 
 /// Map a source name to its canonical form: a recipe canonicalizes, any other
 /// name is returned unchanged. Used everywhere a source is looked up, so
 /// generated sources are shared by content regardless of how they're spelled.
 fn canonical_source_name(name: &str) -> String {
-    Recipe::parse(name).map(|r| r.canonical()).unwrap_or_else(|| name.to_string())
+    Recipe::parse(name)
+        .map(|r| r.canonical())
+        .unwrap_or_else(|| name.to_string())
 }
 
 /// A unit of server work, already parsed/lowered/validated on the intake side.
@@ -155,7 +191,14 @@ pub enum Command {
     Install { name: String, program: st::Program },
     /// Update positional `input` of `prog`: add `(key, val)` with `diff` at
     /// `time` (default the current epoch when `None`).
-    Feed { prog: String, input: usize, key: Value, val: Value, time: Option, diff: Diff },
+    Feed {
+        prog: String,
+        input: usize,
+        key: Value,
+        val: Value,
+        time: Option,
+        diff: Diff,
+    },
     /// Close the current epoch and run to quiescence.
     Tick,
     /// Drop the named program.
@@ -188,6 +231,18 @@ struct Installed {
     /// [`Origin`]. Generated/clock entries advance and drop like any program but
     /// are not writable by `feed`.
     origin: Origin,
+    /// Generator recipe and next row to retract, for changing random sources.
+    generator: Option<(Recipe, u64)>,
+}
+
+/// A stable, transport-friendly description of one installed dataflow.
+#[derive(Clone, Debug)]
+pub struct ProgramInfo {
+    pub name: String,
+    pub inputs: Vec,
+    pub imports: Vec,
+    pub exports: Vec,
+    pub origin: &'static str,
 }
 
 /// A live registry of installed programs and the traces they publish.
@@ -214,10 +269,54 @@ impl Server {
     }
 
     /// The current epoch (the open host time).
-    pub fn epoch(&self) -> OuterTime { self.epoch }
+    pub fn epoch(&self) -> OuterTime {
+        self.epoch
+    }
 
     /// Whether a trace is registered under `name`.
-    pub fn has_trace(&self, name: &str) -> bool { self.traces.contains_key(name) }
+    pub fn has_trace(&self, name: &str) -> bool {
+        self.traces.contains_key(name)
+    }
+
+    /// Clone a trace reader for a transient peek or subscription dataflow.
+    pub fn trace(&self, name: &str) -> Option {
+        self.traces.get(&canonical_source_name(name)).cloned()
+    }
+
+    /// Return registry state without coupling a caller to stdout formatting.
+    pub fn program_info(&self) -> Vec {
+        let mut result: Vec<_> = self
+            .programs
+            .iter()
+            .map(|(name, installed)| {
+                let mut inputs: Vec<_> = installed.inputs.keys().copied().collect();
+                inputs.sort();
+                ProgramInfo {
+                    name: name.clone(),
+                    inputs,
+                    imports: installed.imports.clone(),
+                    exports: installed.exports.clone(),
+                    origin: match installed.origin {
+                        Origin::Program => "program",
+                        Origin::Generated => "generated",
+                        Origin::Clock => "clock",
+                    },
+                }
+            })
+            .collect();
+        result.sort_by(|a, b| a.name.cmp(&b.name));
+        result
+    }
+
+    pub fn trace_info(&self) -> Vec<(String, usize)> {
+        let mut result: Vec<_> = self
+            .traces
+            .keys()
+            .map(|name| (name.clone(), self.importers.get(name).copied().unwrap_or(0)))
+            .collect();
+        result.sort_by(|a, b| a.0.cmp(&b.0));
+        result
+    }
 
     /// Install `prog` under `name`: build its dataflow in `worker`, wiring each
     /// root `Source::Trace` to a registered trace and registering each export's
@@ -229,7 +328,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 +349,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 +363,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 +386,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 +398,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 +449,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 +493,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 +531,18 @@ impl Server {
         self.traces.insert("clock".to_string(), trace);
         let mut inputs = HashMap::new();
         inputs.insert(0usize, input);
-        self.programs.insert("clock".to_string(), Installed {
-            inputs,
-            imports: Vec::new(),
-            exports: vec!["clock".to_string()],
-            dataflow_id,
-            probe,
-            origin: Origin::Clock,
-        });
+        self.programs.insert(
+            "clock".to_string(),
+            Installed {
+                inputs,
+                imports: Vec::new(),
+                exports: vec!["clock".to_string()],
+                dataflow_id,
+                probe,
+                origin: Origin::Clock,
+                generator: None,
+            },
+        );
     }
 
     /// Stage an update to positional input `input` of installed program `prog`:
@@ -407,18 +550,42 @@ impl Server {
     /// epoch). The time must be at or after the current epoch — you cannot
     /// insert into the closed past. Takes effect once `tick` advances the input
     /// frontier past `time`.
-    pub fn feed(&mut self, prog: &str, input: usize, key: Value, val: Value, time: Option, diff: Diff) -> Result<(), String> {
+    pub fn feed(
+        &mut self,
+        prog: &str,
+        input: usize,
+        key: Value,
+        val: Value,
+        time: Option,
+        diff: Diff,
+    ) -> Result<(), String> {
         let t = time.unwrap_or(self.epoch);
         if t < self.epoch {
-            return Err(format!("cannot feed at time {} < current epoch {}", t, self.epoch));
+            return Err(format!(
+                "cannot feed at time {} < current epoch {}",
+                t, self.epoch
+            ));
         }
         let prog = canonical_source_name(prog);
-        let installed = self.programs.get_mut(&prog).ok_or_else(|| format!("no program {:?}", prog))?;
+        let installed = self
+            .programs
+            .get_mut(&prog)
+            .ok_or_else(|| format!("no program {:?}", prog))?;
         if installed.origin != Origin::Program {
-            let kind = if installed.origin == Origin::Clock { "clock" } else { "generated" };
-            return Err(format!("{:?} is a {} source and is not writable", prog, kind));
+            let kind = if installed.origin == Origin::Clock {
+                "clock"
+            } else {
+                "generated"
+            };
+            return Err(format!(
+                "{:?} is a {} source and is not writable",
+                prog, kind
+            ));
         }
-        let handle = installed.inputs.get_mut(&input).ok_or_else(|| format!("program {:?} has no input {}", prog, input))?;
+        let handle = installed
+            .inputs
+            .get_mut(&input)
+            .ok_or_else(|| format!("program {:?} has no input {}", prog, input))?;
         handle.update_at((key, val), t, diff);
         Ok(())
     }
@@ -430,7 +597,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 +631,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 +647,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 +661,54 @@ impl Server {
         Ok(())
     }
 
+    /// Return the consolidated closed-past contents of a trace on worker 0.
+    /// This is the structured counterpart to [`Server::peek`] for protocols.
+    pub fn snapshot(
+        &mut self,
+        worker: &mut Worker,
+        name: &str,
+    ) -> Result, String> {
+        use timely::dataflow::operators::{Exchange, Inspect, Probe};
+
+        let name = canonical_source_name(name);
+        let epoch = self.epoch;
+        let mut trace = self
+            .trace(&name)
+            .ok_or_else(|| format!("no trace {:?}", name))?;
+        let acc: Rc>> = Rc::new(RefCell::new(HashMap::new()));
+        let acc_in = acc.clone();
+        let mut probe = ProbeHandle::new();
+        let id = worker.next_dataflow_index();
+        worker.dataflow::(|scope| {
+            trace
+                .import(scope.clone())
+                .as_collection(|k, v| (k.clone(), v.clone()))
+                .inner
+                .exchange(|_| 0u64)
+                .inspect(move |((k, v), t, d)| {
+                    if *t < epoch {
+                        *acc_in
+                            .borrow_mut()
+                            .entry((k.clone(), v.clone()))
+                            .or_insert(0) += *d;
+                    }
+                })
+                .probe_with(&mut probe);
+        });
+        while probe.less_than(&epoch) {
+            worker.step();
+        }
+        worker.drop_dataflow(id);
+        let mut rows: Vec<_> = acc
+            .borrow()
+            .iter()
+            .filter(|(_, d)| **d != 0)
+            .map(|((k, v), d)| (k.clone(), v.clone(), *d))
+            .collect();
+        rows.sort_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1)));
+        Ok(rows)
+    }
+
     /// Drop installed program `name`, releasing its dataflow immediately.
     ///
     /// Refuses (changing nothing) if any trace the program publishes still has a
@@ -495,11 +719,17 @@ impl Server {
     pub fn drop_program(&mut self, worker: &mut Worker, name: &str) -> Result<(), String> {
         let canon = canonical_source_name(name);
         let name = canon.as_str();
-        let installed = self.programs.get(name).ok_or_else(|| format!("no program {:?}", name))?;
+        let installed = self
+            .programs
+            .get(name)
+            .ok_or_else(|| format!("no program {:?}", name))?;
         for ex in &installed.exports {
             let live = self.importers.get(ex).copied().unwrap_or(0);
             if live > 0 {
-                return Err(format!("cannot drop {:?}: its trace {:?} has {} live importer(s); drop them first", name, ex, live));
+                return Err(format!(
+                    "cannot drop {:?}: its trace {:?} has {} live importer(s); drop them first",
+                    name, ex, live
+                ));
             }
         }
 
@@ -517,6 +747,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 +779,26 @@ impl Server {
                     h.update_at((clock_row(next), Value::unit()), next, 1);
                 }
             }
+            // A random source denotes an infinite deterministic row stream.
+            // Each tick replaces `churn` members of its fixed-size window.
+            if let Some((recipe, cursor)) = &mut installed.generator {
+                let recipe = *recipe;
+                if let Recipe::Random { edges, churn, .. } = recipe {
+                    if let Some(h) = installed.inputs.get_mut(&0) {
+                        for _ in 0..churn {
+                            let old = *cursor;
+                            let new = edges + *cursor;
+                            if (old as usize) % worker.peers() == worker.index() {
+                                h.update(recipe.row(old), -1);
+                            }
+                            if (new as usize) % worker.peers() == worker.index() {
+                                h.update(recipe.row(new), 1);
+                            }
+                            *cursor += 1;
+                        }
+                    }
+                }
+            }
             for handle in installed.inputs.values_mut() {
                 handle.advance_to(next);
                 handle.flush();
@@ -577,7 +841,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 +859,16 @@ impl Server {
                 Origin::Generated => " [generated]",
                 Origin::Clock => " [clock]",
             };
-            println!("  {}{} (inputs: {:?}, imports: {:?}, exports: {:?})", p, tag, ins, installed.imports, installed.exports);
+            println!(
+                "  {}{} (inputs: {:?}, imports: {:?}, exports: {:?})",
+                p, tag, ins, installed.imports, installed.exports
+            );
         }
     }
 }
 
 impl Default for Server {
-    fn default() -> Self { Server::new() }
+    fn default() -> Self {
+        Server::new()
+    }
 }

From bddb90085ad2cccf49e9a6c304460c9eb4097781 Mon Sep 17 00:00:00 2001
From: Frank McSherry 
Date: Sat, 29 Aug 2026 17:02:39 -0400
Subject: [PATCH 2/3] ddir-server: a write path -- feed, a size gate, and demos

feed    [val=] [time=] [diff=] pushes one update into a
loaded program's positional input, matching the ddir_server example's
value grammar, parsed panic-safe on the session thread. The stance on
contention: writes are open; policy lives in the dataflow. Cooperating
clients include their id and an ordering epoch in the data, and programs
resolve races over those facts deterministically. Identity is convention,
not enforcement -- adversarial hardening is deliberately deferred.

One intake gate: DDIR_MAX_PROGRAM_BYTES caps the load body before parsing
(oversized bodies are swallowed and reported once; the parser stays in
sync). Transport self-defense, not semantics.

Demos: claims.txt (first-claim-wins as min over (epoch, id));
txn.txt (transaction processing in the data plane, after the 2025-04-27
blog post: intents as blind appends carrying the reads under which writes
evaporate, a recursive non-monotone view resolving commits, and episodic
OUTPUT-INVARIANT maintenance -- the demo retracts 13 of 15 intent rows and
committed state is unchanged); two_sessions.py (TCP smoke: racing
convention-stamped claims settle deterministically; the size gate fires;
clean shutdown).

Co-Authored-By: Claude Fable 5 
---
 interactive/server/README.md            |  43 ++++-
 interactive/server/console.html         |   2 +-
 interactive/server/demo/claims.txt      |  23 +++
 interactive/server/demo/two_sessions.py | 139 +++++++++++++++
 interactive/server/demo/txn.txt         | 113 ++++++++++++
 interactive/server/src/cmd.rs           | 225 +++++++++++++++++++++++-
 interactive/server/src/loop_.rs         |  10 ++
 7 files changed, 544 insertions(+), 11 deletions(-)
 create mode 100644 interactive/server/demo/claims.txt
 create mode 100644 interactive/server/demo/two_sessions.py
 create mode 100644 interactive/server/demo/txn.txt

diff --git a/interactive/server/README.md b/interactive/server/README.md
index 0190fd6f3..446d61a4f 100644
--- a/interactive/server/README.md
+++ b/interactive/server/README.md
@@ -15,11 +15,11 @@ 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 annotated
-command scripts can be piped straight to stdin.
+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`, `peek`, `tail`, `stop`, `tick`,
-and `exit`. `load` accepts an inline pipe-syntax program:
+The useful commands are `load`, `drop`, `list`, `feed`, `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";
@@ -43,6 +43,41 @@ 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.
+
+## 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/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
diff --git a/interactive/server/console.html b/interactive/server/console.html
index 49edc1d33..5f2bdbbbf 100644
--- a/interactive/server/console.html
+++ b/interactive/server/console.html
@@ -258,7 +258,7 @@ 
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/two_sessions.py b/interactive/server/demo/two_sessions.py
new file mode 100644
index 000000000..f91692639
--- /dev/null
+++ b/interactive/server/demo/two_sessions.py
@@ -0,0 +1,139 @@
+#!/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),
+        )
+
+        # 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
index 8f432db57..da995d18c 100644
--- a/interactive/server/src/cmd.rs
+++ b/interactive/server/src/cmd.rs
@@ -12,6 +12,10 @@
 //! 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;
 
@@ -52,6 +56,19 @@ pub enum Cmd {
     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,
+    },
     /// Push a row into the query input of a `--explain` dataflow
     /// (reserved; unimplemented). Sign is `+1` for `add`, `-1` for `del`.
     #[allow(dead_code)]
@@ -106,25 +123,53 @@ pub struct Request {
 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",
+    "load", "drop", "list", "peek", "tail", "stop", "tick", "query", "exit", "feed",
 ];
 
+/// 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 {
-        Self::default()
+        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
@@ -144,6 +189,9 @@ impl LineParser {
             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 {
@@ -155,15 +203,23 @@ impl LineParser {
                     ));
                 }
             }
-            pl.body.push_str(trimmed);
-            pl.body.push('\n');
+            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), so
-        // annotated command scripts can be piped straight to stdin.
+        // a load body every line is literal program text, handled above).
         if trimmed.is_empty() || trimmed.starts_with('#') {
             return None;
         }
@@ -199,6 +255,7 @@ impl LineParser {
                     bindings,
                     explain,
                     body: String::new(),
+                    poisoned: None,
                 });
                 None
             }
@@ -222,6 +279,41 @@ enum ParseOutcome {
     },
 }
 
+/// 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" => {
@@ -321,6 +413,72 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
                 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,
+            })
+        }
         "drop" => match args {
             [tok] => {
                 let target = match tok.parse::() {
@@ -485,6 +643,61 @@ mod tests {
         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 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();
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 7a2d70ddc..2249a719a 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -122,6 +122,16 @@ fn dispatch(
                 }
             }
         },
+        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::List) => {
             for program in server.program_info() {
                 send(

From d1498b6573adaf646e470ac5f287ce83097b2a30 Mon Sep 17 00:00:00 2001
From: Frank McSherry 
Date: Sat, 29 Aug 2026 17:04:21 -0400
Subject: [PATCH 3/3] ddir-server: bind -- export->input feedback, one epoch
 per tick

bind    delivers the trace's changes into the input at
each tick, so the input mirrors the trace one epoch delayed. This is the
write path for installed programs: a dataflow can act on the world -- or
on itself -- with no client in the loop, one well-founded recursion step
per tick. Rationale: the language is deliberately timeless (programs
cannot observe the outer timestamp, and exports erase inner iteration
coordinates), so all temporality lives in the server, and there are
exactly two temporal operations: advance (tick) and unit delay (bind).
Cross-program cycles cannot iterate within an epoch through the trace
interface; the delay is also what keeps cycles among independently
authored programs well-founded without joint monotonicity.

Implemented in interactive::server::Server so the multi-worker example
server gains bind/unbind too (each worker's tap sees its shard; the union
delivers the delta exactly once). Lifecycle: a bound source holds an
importer and cannot be dropped; nor can the bound target (unbind first);
tick waits on tap probes before draining, so buffers hold every change
through the just-closed epoch.

The state-machine idiom (demo/counter.txt, tests/server_bind.rs): give a
program a seed input and a feedback input, 'let state = seed + feedback',
and bind 'f(state) + (seed | negate)' into the feedback input; then
state(t) = f(state(t-1)) while later seed feeds inject as perturbations.
Deliberately NOT for compaction/maintenance: a continuously-bound
compactor oscillates (deleting a row un-derives its own deadness);
maintenance is episodic peek+feed, per demo/txn.txt.

Co-Authored-By: Claude Fable 5 
---
 interactive/examples/ddir_server.rs     |  22 +++
 interactive/server/README.md            |  25 ++-
 interactive/server/console.html         |   2 +-
 interactive/server/demo/counter.txt     |  22 +++
 interactive/server/demo/two_sessions.py |   8 +
 interactive/server/src/cmd.rs           |  48 +++++-
 interactive/server/src/loop_.rs         |  14 ++
 interactive/src/server.rs               | 213 +++++++++++++++++++++++-
 interactive/tests/server_bind.rs        |  96 +++++++++++
 9 files changed, 443 insertions(+), 7 deletions(-)
 create mode 100644 interactive/server/demo/counter.txt
 create mode 100644 interactive/tests/server_bind.rs

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/README.md b/interactive/server/README.md
index 446d61a4f..1f0e9f1c0 100644
--- a/interactive/server/README.md
+++ b/interactive/server/README.md
@@ -18,8 +18,9 @@ generates one. Responses are ` data ...`, followed by ` ok ...` or
 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`, `peek`, `tail`,
-`stop`, `tick`, and `exit`. `load` accepts an inline pipe-syntax program:
+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";
@@ -61,6 +62,25 @@ 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:
@@ -74,6 +94,7 @@ if a deployment ever needs one.
 
     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
diff --git a/interactive/server/console.html b/interactive/server/console.html
index 5f2bdbbbf..e06e53646 100644
--- a/interactive/server/console.html
+++ b/interactive/server/console.html
@@ -258,7 +258,7 @@ 
Protocol log 
 
 
 

-  
+  
   
 

 
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
index f91692639..48100ef27 100644
--- a/interactive/server/demo/two_sessions.py
+++ b/interactive/server/demo/two_sessions.py
@@ -112,6 +112,14 @@ def main():
             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)
diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs
index da995d18c..d13298521 100644
--- a/interactive/server/src/cmd.rs
+++ b/interactive/server/src/cmd.rs
@@ -69,6 +69,19 @@ pub enum Cmd {
         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)]
@@ -129,7 +142,8 @@ pub struct LineParser {
 /// 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",
+    "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,
@@ -479,6 +493,22 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome {
                 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::() {
@@ -682,6 +712,22 @@ mod tests {
         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);
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 2249a719a..d8d7f2329 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -132,6 +132,12 @@ fn dispatch(
         }) => 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(
@@ -156,6 +162,14 @@ fn dispatch(
                     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) {
diff --git a/interactive/src/server.rs b/interactive/src/server.rs
index dbdb8e071..84e086102 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -45,10 +45,11 @@ use std::rc::Rc;
 
 use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::input::{Input, InputSession};
-use differential_dataflow::operators::arrange::TraceAgent;
+use differential_dataflow::operators::arrange::{ShutdownButton, TraceAgent};
 use differential_dataflow::trace::implementations::ValSpine;
 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;
@@ -205,6 +206,18 @@ pub enum Command {
     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).
@@ -245,6 +258,40 @@ pub struct ProgramInfo {
     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.
 pub struct Server {
     /// Published export name -> shareable trace.
@@ -252,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,
 }
@@ -264,6 +314,7 @@ impl Server {
             traces: HashMap::new(),
             programs: HashMap::new(),
             importers: HashMap::new(),
+            bindings: Vec::new(),
             epoch: 0,
         }
     }
@@ -590,6 +641,120 @@ impl Server {
         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
@@ -717,6 +882,12 @@ 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
@@ -807,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
@@ -864,6 +1065,12 @@ impl Server {
                 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);
+            }
+        }
     }
 }
 
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();
+    });
+}