diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs
index f8873617e..2c54f2ce9 100644
--- a/interactive/src/parse/pipe.rs
+++ b/interactive/src/parse/pipe.rs
@@ -62,6 +62,32 @@
 //!   and each stage applies to everything to its left: `a | map(f) + b | distinct` is
 //!   `((a | map(f)) + b) | distinct`. To add a pipeline, parenthesize it or put it first.
 //!
+//! Join fragments are an experimental surface form, not part of the plan language or a stable
+//! interface. The parser expands each one into plain DDP as it reads it, so no syntax tree,
+//! later stage or emitted plan contains one. They may change or be removed.
+//! A fragment is a pipeline source: `a(x, y) * b(y, z) | map(((x, y, z), ()) -> (x, z))`.
+//! Atoms name existing bindings (including scope paths); they always have key and value arguments.
+//! Every later key must mention at least one variable bound by an earlier atom; it may combine
+//! earlier variables with literals, tuples, unit and constructors, but not wildcards or new variables.
+//! Open built-in constructors (`Ok(x)`, `Err(e)`, `None`) cannot be built as later keys, and
+//! untyped empty lists (including `""`) cannot appear in them.
+//! A constant key would move every incoming row to one key; write a cross join as plain DDP.
+//! Atoms keep their existing arrangements: one new arrangement per `*`, of the rows so far;
+//! none at the first `*` when the first two keys are the same unrestricted variables.
+//! The first key and every value accept variables, `_`, tuples, integer/list/string literals,
+//! and constructors with payload patterns, such as `Some(x)` or `Type::Variant(x)`.
+//! A bare constructor such as `None` means its empty payload; a constructor needing a payload
+//! must supply one. A name unknown as a constructor is a variable.
+//! An empty list pattern `[]` tests emptiness; inside a list literal its element shape must
+//! come from sibling constants, otherwise DDIR rejects the literal with a shape error.
+//! Reusing a variable tests equality. Plain filters run as soon as their fields are bound,
+//! before the next rekey; a direct first join precedes its first atom's tests. Variants bind
+//! through typed payload lists in `flatmap`, which can also emit the next lookup key.
+//! Rows contain each variable once, in first-appearance order, keyed by their tuple (`(x)` for
+//! one variable, `()` for none), with unit value. A test-free final layout composes into
+//! the last join's projection. Weights multiply; nothing implies distinct.
+//! Even the parsed view shows the expansion.
+//!
 //! Statements: `let x = …;`, `set x = …;` (a feedback variable in a named
 //! scope), `name: { … }` (a nested scope), `output "name" = …;` (a
 //! program output, root scope only). An optional `var x;` or
@@ -117,6 +143,8 @@
 use super::*;
 use crate::ir::Lambda;
 
+mod fragment;
+
 #[derive(Debug, Clone, PartialEq)]
 enum Token {
     Let, Var, Set, Output, Type, Enum,
@@ -126,7 +154,7 @@ enum Token {
     Lambda { pat: String, body: String },
     LParen, RParen, LBrace, RBrace,
     Comma, Semi, Colon, ColonColon, Eq,
-    Pipe, Plus, Minus, Eof,
+    Pipe, Plus, Minus, Star, LBracket, RBracket, Eof,
 }
 
 /// A shape before names in type declarations have been resolved.
@@ -266,6 +294,9 @@ fn lex(input: &str) -> Vec<Token> {
             ';' => { chars.next(); tokens.push(Token::Semi); },
             '|' => { chars.next(); tokens.push(Token::Pipe); },
             '+' => { chars.next(); tokens.push(Token::Plus); },
+            '*' => { chars.next(); tokens.push(Token::Star); },
+            '[' => { chars.next(); tokens.push(Token::LBracket); },
+            ']' => { chars.next(); tokens.push(Token::RBracket); },
             '=' => { chars.next(); tokens.push(Token::Eq); },
             '"' => {
                 // String literal: the quoted name in `import "..."` / `output "..." = ...`.
@@ -546,13 +577,13 @@ impl Parser {
     /// An atom followed by stages, applied left to right: `| op`, `+ atom` and `- atom`. A run of
     /// `+`/`-` stages is one `Concat`.
     fn parse_pipe_expr(&mut self) -> Expr {
-        let mut expr = self.parse_atom();
+        let mut expr = self.parse_product();
         let mut concat = false;
         loop {
             let part = match self.peek() {
                 Token::Pipe => { self.next(); expr = self.parse_pipe_op(expr); concat = false; continue },
-                Token::Plus => { self.next(); self.parse_atom() },
-                Token::Minus => { self.next(); Expr::Negate(Box::new(self.parse_atom())) },
+                Token::Plus => { self.next(); self.parse_product() },
+                Token::Minus => { self.next(); Expr::Negate(Box::new(self.parse_product())) },
                 _ => break,
             };
             expr = match expr {
@@ -564,6 +595,11 @@ impl Parser {
         expr
     }
 
+    fn parse_product(&mut self) -> Expr {
+        let source = self.parse_atom();
+        if *self.peek() == Token::LParen { self.fragment(source) } else { source }
+    }
+
     fn parse_atom(&mut self) -> Expr {
         match self.peek().clone() {
             Token::Input => {
diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs
new file mode 100644
index 000000000..117c61bfb
--- /dev/null
+++ b/interactive/src/parse/pipe/fragment.rs
@@ -0,0 +1,464 @@
+//! Deliberately ordered, arrangement-preserving source-to-source expansion.
+//! Tests and payload bindings run at the first legal checkpoint, before the next rekey.
+
+use super::*;
+use std::collections::BTreeMap;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum Pattern {
+    Var(String), Wild, Tuple(Vec<Pattern>), Literal(String),
+    Variant { name: String, tag: usize, payload: Box<Pattern> },
+}
+
+impl Pattern {
+    fn shape_pattern(&self) -> String {
+        match self {
+            Self::Tuple(ps) => tuple(&ps.iter().map(Self::shape_pattern).collect::<Vec<_>>()),
+            Self::Variant { name, .. } => name.split("::").next().unwrap().to_string(),
+            _ => "_".into(),
+        }
+    }
+    fn has_variant(&self) -> bool {
+        match self {
+            Self::Variant { .. } => true,
+            Self::Tuple(ps) => ps.iter().any(Self::has_variant),
+            _ => false,
+        }
+    }
+    /// Distinct variables in tuples: a key that restricts nothing, so joining on it needs no test first.
+    fn unrestricted(&self) -> bool {
+        fn walk<'a>(p: &'a Pattern, seen: &mut Vec<&'a str>) -> bool {
+            match p {
+                Pattern::Var(n) if !seen.contains(&n.as_str()) => { seen.push(n); true }
+                Pattern::Tuple(ps) => ps.iter().all(|p| walk(p, seen)),
+                _ => false,
+            }
+        }
+        walk(self, &mut Vec::new())
+    }
+    fn constant(&self) -> bool {
+        match self { Self::Literal(_) => true, Self::Tuple(ps) => ps.iter().all(Self::constant), _ => false }
+    }
+    fn has_empty_list(&self) -> bool {
+        match self {
+            Self::Literal(s) => s == "[]",
+            Self::Tuple(ps) => ps.iter().any(Self::has_empty_list),
+            Self::Variant { payload, .. } => payload.has_empty_list(),
+            _ => false,
+        }
+    }
+    fn check_lists(&self) {
+        match self {
+            // Check the whole literal so sibling constants can supply an empty list's shape.
+            Self::Literal(s) if s != "[]" && s.contains("[]") => {
+                corgi::parse_ml(s).unwrap_or_else(|e| panic!("DDIR fragment list literal `{s}` needs a complete element shape; use plain DDP with an explicit list shape: {e}"));
+            }
+            Self::Tuple(ps) => for p in ps { p.check_lists(); },
+            Self::Variant { payload, .. } => payload.check_lists(),
+            _ => {},
+        }
+    }
+    fn text(&self) -> String {
+        match self {
+            Self::Literal(n) => n.clone(),
+            Self::Tuple(ps) => tuple(&ps.iter().map(Self::text).collect::<Vec<_>>()),
+            _ => unreachable!("only constants in list literals need text"),
+        }
+    }
+}
+
+struct Atom { source: Expr, key: Pattern, val: Pattern }
+
+/// A fragment's atoms, expanded as soon as they are parsed: no syntax tree holds a fragment.
+struct Fragment { atoms: Vec<Atom> }
+
+impl Parser {
+    pub(super) fn fragment(&mut self, source: Expr) -> Expr {
+        let mut atoms = Vec::new();
+        let mut source = source;
+        loop {
+            assert!(matches!(source, Expr::Name(_) | Expr::Path(..)), "a fragment atom names an existing binding");
+            self.expect(&Token::LParen);
+            let key = self.fragment_pattern();
+            self.expect(&Token::Comma);
+            let val = self.fragment_pattern();
+            if atoms.is_empty() { key.check_lists(); }
+            val.check_lists();
+            self.expect(&Token::RParen);
+            atoms.push(Atom { source, key, val });
+            if *self.peek() != Token::Star { break; }
+            self.next();
+            source = self.parse_atom();
+        }
+        Fragment { atoms }.expansion(self.lambda_scope())
+    }
+
+    fn fragment_pattern(&mut self) -> Pattern {
+        match self.next() {
+            Token::LParen => {
+                let mut ps = Vec::new();
+                if *self.peek() != Token::RParen {
+                    loop {
+                        ps.push(self.fragment_pattern());
+                        if *self.peek() != Token::Comma { break; }
+                        self.next();
+                    }
+                }
+                self.expect(&Token::RParen);
+                Pattern::Tuple(ps)
+            }
+            Token::Ident(n) if n == "_" => Pattern::Wild,
+            Token::Ident(mut n) => {
+                if *self.peek() == Token::ColonColon { self.next(); n += &format!("::{}", self.parse_ident()); }
+                let scope = self.lambda_scope();
+                let constructor = scope.variant(&n).map(|(tag, ty)| {
+                    let empty = matches!(scope.variants(ty).unwrap()[tag].1.as_ref(), Some(corgi::Shape::Unit));
+                    (tag, ty.to_string(), empty)
+                });
+                // An ambiguous declared constructor must not silently become a variable either.
+                let known = constructor.is_ok() || n.contains("::") || self.types.values().any(|vs| vs.iter().any(|(v, _)| v == &n));
+                if *self.peek() == Token::LParen || known {
+                    let (tag, ty, empty) = constructor.unwrap_or_else(|e| panic!("fragment constructor `{n}`: {e}"));
+                    let payload = if *self.peek() == Token::LParen {
+                        self.next();
+                        let p = if *self.peek() == Token::RParen { Pattern::Tuple(vec![]) } else { self.fragment_pattern() };
+                        self.expect(&Token::RParen);
+                        p
+                    } else {
+                        assert!(empty, "fragment constructor `{n}` requires a payload; write `{n}(pattern)`");
+                        Pattern::Tuple(vec![])
+                    };
+                    Pattern::Variant { name: format!("{ty}::{}", n.rsplit("::").next().unwrap()), tag, payload: Box::new(payload) }
+                } else {
+                    Pattern::Var(n)
+                }
+            }
+            Token::Int(n) => Pattern::Literal(n.to_string()),
+            Token::Minus => match self.next() { Token::Int(n) => Pattern::Literal(format!("-{n}")), t => panic!("expected integer after -, got {t:?}") },
+            Token::Str(s) => Pattern::Literal(format!("[{}]", s.bytes().map(|b| b.to_string()).collect::<Vec<_>>().join(", "))),
+            Token::LBracket => {
+                let mut xs = Vec::new();
+                if *self.peek() != Token::RBracket {
+                    loop {
+                        let x = self.fragment_pattern();
+                        assert!(x.constant(), "a list pattern is a literal");
+                        xs.push(x.text());
+                        if *self.peek() != Token::Comma { break; }
+                        self.next();
+                    }
+                }
+                self.expect(&Token::RBracket);
+                Pattern::Literal(format!("[{}]", xs.join(", ")))
+            }
+            t => panic!("expected fragment pattern, got {t:?}"),
+        }
+    }
+}
+
+fn tuple(xs: &[String]) -> String { format!("({})", xs.join(", ")) }
+// Emit a join's next key directly, as the design specifies. This is syntax
+// composition, with no demand analysis or choice of index.
+fn project(input: Expr, map: Lambda) -> Expr {
+    match input {
+        Expr::Join(left, right, projection) => Expr::Join(left, right,
+            Lambda::compose("row", &[(&projection, "row".into()), (&map, "_call0".into())], |r| r[1].clone())),
+        other => Expr::Map(Box::new(other), map),
+    }
+}
+
+/// Logical fields stay fixed while checkpoints change their physical locations.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+enum Access {
+    Slot(usize), Field(Box<Access>, usize), Payload(Box<Access>, usize),
+}
+
+/// A lookup key's fields are read from k, including constructed variant payloads.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+enum KeyRead {
+    Value(String), Field(Box<KeyRead>, usize), Payload(Box<KeyRead>, usize),
+}
+impl KeyRead {
+    fn scalar(&self) -> Option<String> {
+        match self {
+            Self::Value(s) => Some(s.clone()),
+            Self::Field(a, i) => Some(format!("{}.{i}", a.scalar()?)),
+            Self::Payload(..) => None,
+        }
+    }
+    fn bind(&self, bindings: &mut Vec<(String, String)>, names: &mut BTreeMap<KeyRead, String>) -> String {
+        if let Some(name) = names.get(self) { return name.clone(); }
+        match self {
+            Self::Value(s) => s.clone(),
+            Self::Field(a, i) => format!("{}.{i}", a.bind(bindings, names)),
+            Self::Payload(a, tag) => {
+                let source = a.bind(bindings, names);
+                let name = format!("key_payload{}", bindings.len());
+                bindings.push((format!("{source} enlist unweave .{}", tag + 1), name.clone()));
+                names.insert(self.clone(), name.clone());
+                name
+            }
+        }
+    }
+}
+
+enum Test { Equal(Access, Access), Literal(Access, String) }
+#[derive(Clone)]
+struct Variant { at: Access, tag: usize, shape: String }
+#[derive(Clone, Copy)]
+enum Target<'a> { Key(&'a Pattern), Keep(&'a Pattern), Current, Output }
+struct State<'a> {
+    scope: &'a corgi::Scope,
+    next: usize,
+    names: Vec<String>,
+    bound: BTreeMap<String, Access>,
+    locations: BTreeMap<Access, String>,
+    key: BTreeMap<Access, KeyRead>,
+    variants: Vec<Variant>,
+    tests: Vec<Test>,
+}
+impl State<'_> {
+    fn lambda(&self, pat: &str, body: String) -> Lambda { Lambda::new(format!("let {pat} = input in {body}"), self.scope) }
+    fn read(&self, at: &Access) -> String {
+        if let Some(expr) = self.locations.get(at) { return expr.clone(); }
+        match at {
+            Access::Field(a, i) => format!("let field = {} in field.{i}", self.read(a)),
+            _ => panic!("fragment payload or field is not bound: {at:?}"),
+        }
+    }
+    fn ready(&self, at: &Access) -> bool {
+        self.locations.contains_key(at) || matches!(at, Access::Field(a, _) if self.ready(a))
+    }
+    fn test(&self, test: &Test) -> String {
+        match test {
+            Test::Equal(a, b) => format!("({}, {}) eq", self.read(a), self.read(b)),
+            Test::Literal(a, lit) if lit == "[]" => format!("({} len, 0) eq", self.read(a)),
+            Test::Literal(a, lit) => format!("({}, {lit}) eq", self.read(a)),
+        }
+    }
+    fn pattern(&mut self, p: &Pattern, at: Access) {
+        match p {
+            Pattern::Wild => {},
+            Pattern::Var(n) => {
+                if let Some(old) = self.bound.get(n) { self.tests.push(Test::Equal(old.clone(), at)); }
+                else { self.names.push(n.clone()); self.bound.insert(n.clone(), at); }
+            }
+            Pattern::Literal(lit) => self.tests.push(Test::Literal(at, lit.clone())),
+            Pattern::Tuple(ps) => for (i, p) in ps.iter().enumerate() {
+                self.pattern(p, Access::Field(Box::new(at.clone()), i));
+            },
+            Pattern::Variant { tag, payload, .. } => {
+                self.variants.push(Variant { at: at.clone(), tag: *tag, shape: payload.shape_pattern() });
+                self.pattern(payload, Access::Payload(Box::new(at), *tag));
+            }
+        }
+    }
+    fn ingest(&mut self, p: &Pattern, expr: String) {
+        if let Pattern::Tuple(ps) = p {
+            for (i, p) in ps.iter().enumerate() { self.ingest(p, format!("{expr}.{i}")); }
+        } else if !matches!(p, Pattern::Wild) {
+            let at = Access::Slot(self.next);
+            self.next += 1;
+            self.locations.insert(at.clone(), expr);
+            self.pattern(p, at);
+        }
+    }
+    fn validate_key(&self, p: &Pattern) {
+        fn mentions_bound(s: &State<'_>, p: &Pattern) -> bool {
+            match p {
+                Pattern::Var(n) => {
+                    assert!(s.bound.contains_key(n), "fragment key variable `{n}` is not bound by an earlier atom");
+                    true
+                }
+                Pattern::Tuple(ps) => ps.iter().fold(false, |found, p| mentions_bound(s, p) || found),
+                Pattern::Variant { payload, .. } => mentions_bound(s, payload),
+                Pattern::Literal(_) => false,
+                Pattern::Wild => panic!("a later fragment key cannot contain a wildcard; use a variable bound by an earlier atom"),
+            }
+        }
+        assert!(mentions_bound(self, p), "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP");
+        assert!(!p.has_empty_list(), "DDIR fragment key literal `[]` needs a complete element shape; use plain DDP with an explicit list shape");
+        p.check_lists();
+    }
+    fn key_expr(&self, p: &Pattern) -> String {
+        match p {
+            Pattern::Var(n) => self.read(&self.bound[n]),
+            Pattern::Tuple(ps) => tuple(&ps.iter().map(|p| self.key_expr(p)).collect::<Vec<_>>()),
+            Pattern::Variant { name, payload, .. } => format!("{} inject {name}", self.key_expr(payload)),
+            Pattern::Literal(s) => s.clone(),
+            _ => unreachable!("lookup keys were validated"),
+        }
+    }
+    fn key_locations(&self, p: &Pattern, expr: KeyRead, out: &mut BTreeMap<Access, KeyRead>) {
+        match p {
+            Pattern::Var(n) => { out.entry(self.bound[n].clone()).or_insert(expr); },
+            Pattern::Tuple(ps) => for (i, p) in ps.iter().enumerate() {
+                self.key_locations(p, KeyRead::Field(Box::new(expr.clone()), i), out);
+            },
+            Pattern::Variant { tag, payload, .. } => self.key_locations(payload, KeyRead::Payload(Box::new(expr), *tag), out),
+            Pattern::Literal(_) => {},
+            _ => unreachable!("lookup keys were validated"),
+        }
+    }
+    /// Keep named fields and pending tests, retaining an enclosing sum only until it is bound.
+    fn required(&self) -> std::collections::BTreeSet<Access> {
+        fn resident(s: &State<'_>, at: &Access) -> Access {
+            if s.ready(at) || s.key.contains_key(at) { return at.clone(); }
+            match at {
+                Access::Field(a, _) | Access::Payload(a, _) => resident(s, a),
+                _ => panic!("missing fragment field {at:?}"),
+            }
+        }
+        self.inputs(&self.variants).into_iter().map(|at| resident(self, &at)).collect()
+    }
+    fn inputs(&self, variants: &[Variant]) -> Vec<Access> {
+        let mut inputs = self.bound.values().cloned().collect::<Vec<_>>();
+        for test in &self.tests { match test {
+            Test::Equal(a, b) => { inputs.push(a.clone()); inputs.push(b.clone()); }
+            Test::Literal(a, _) => inputs.push(a.clone()),
+        } }
+        inputs.extend(variants.iter().map(|v| v.at.clone()));
+        inputs
+    }
+    fn row(&mut self, key_expr: String, key: BTreeMap<Access, KeyRead>) -> String {
+        let carried = self.required().into_iter().filter(|a| !key.contains_key(a)).collect::<Vec<_>>();
+        let values = carried.iter().map(|a| self.read(a)).collect::<Vec<_>>();
+        let mut locations = key.iter().filter_map(|(a, r)| r.scalar().map(|r| (a.clone(), r))).collect::<BTreeMap<_, _>>();
+        for (i, at) in carried.into_iter().enumerate() { locations.insert(at, format!("v.{i}")); }
+        self.locations = locations;
+        self.key = key;
+        format!("({key_expr}, {})", tuple(&values))
+    }
+    fn layout(&mut self, target: Target<'_>) -> String {
+        match target {
+            Target::Output => {
+                let values = self.names.iter().map(|n| self.read(&self.bound[n])).collect::<Vec<_>>();
+                let key = tuple(&values);
+                format!("({key}, ())")
+            }
+            Target::Key(p) | Target::Keep(p) => {
+                self.validate_key(p);
+                let expr = if matches!(target, Target::Keep(_)) { "k".into() } else { self.key_expr(p) };
+                let mut key = BTreeMap::new();
+                self.key_locations(p, KeyRead::Value("k".into()), &mut key);
+                self.row(expr, key)
+            }
+            // Decoded key payloads remain scalar while a following plain filter uses them.
+            Target::Current => self.row("k".into(), self.key.iter().filter(|(_, r)| r.scalar().is_some()).map(|(a, r)| (a.clone(), r.clone())).collect()),
+        }
+    }
+    fn filters(&mut self, mut expr: Expr, pat: &str) -> Expr {
+        let mut ready = Vec::new();
+        let tests = std::mem::take(&mut self.tests);
+        for t in tests {
+            let available = match &t {
+                Test::Equal(a, b) => self.ready(a) && self.ready(b),
+                Test::Literal(a, _) => self.ready(a),
+            };
+            if available { ready.push(self.test(&t)); } else { self.tests.push(t); }
+        }
+        if !ready.is_empty() {
+            let test = if ready.len() == 1 { ready.pop().unwrap() } else { format!("{} all", tuple(&ready)) };
+            expr = Expr::Filter(Box::new(expr), self.lambda(pat, test));
+        }
+        expr
+    }
+    fn capture(&mut self, variants: &[Variant], payload: &Access, prefix: &str, preserve_key: bool) -> (String, String) {
+        fn nearest(s: &State<'_>, at: Access) -> Access {
+            if s.locations.contains_key(&at) { return at; }
+            match at {
+                Access::Field(a, _) | Access::Payload(a, _) => nearest(s, *a),
+                _ => panic!("missing fragment capture {at:?}"),
+            }
+        }
+        let mut inputs = self.inputs(variants).into_iter().map(|a| nearest(self, a)).collect::<std::collections::BTreeSet<_>>();
+        inputs.remove(payload);
+        let mut values = inputs.iter().map(|a| self.read(a)).collect::<Vec<_>>();
+        let mut names = inputs.into_iter().enumerate().map(|(i, a)| {
+            let name = format!("{prefix}_{i}");
+            self.locations.insert(a, name.clone());
+            name
+        }).collect::<Vec<_>>();
+        if preserve_key { values.insert(0, "k".into()); names.insert(0, "k".into()); }
+        (tuple(&values), tuple(&names))
+    }
+    fn payload_lists(&mut self, variants: &[Variant], i: usize, target: Target<'_>) -> String {
+        let v = &variants[i];
+        let list = format!("{} enlist unweave .{}", self.read(&v.at), v.tag + 1);
+        let payload = Access::Payload(Box::new(v.at.clone()), v.tag);
+        let name = format!("payload{i}");
+        self.locations.insert(payload.clone(), name.clone());
+        let (values, pat) = self.capture(&variants[i + 1..], &payload, &format!("capture{i}"), matches!(target, Target::Current | Target::Keep(_)));
+        let last = i + 1 == variants.len();
+        let body = if last { self.layout(target) } else { self.payload_lists(variants, i + 1, target) };
+        let binder = if v.shape == "_" { name } else { format!("{name} : {}", v.shape) };
+        let mapped = format!("({values}, {list}) cap_list map (({pat}, {binder}) -> {body})");
+        if last { mapped } else { format!("{mapped} flatten .1") }
+    }
+    /// Finish one atom, before the next arrangement. A payload list can emit that next key.
+    fn checkpoint(&mut self, expr: Expr, target: Target<'_>, pat: &str) -> Expr {
+        if let Target::Key(p) | Target::Keep(p) = target { self.validate_key(p); }
+        let mut expr = self.filters(expr, pat);
+        let mut bindings = Vec::new();
+        let mut names = BTreeMap::new();
+        for (at, read) in &self.key { self.locations.insert(at.clone(), read.bind(&mut bindings, &mut names)); }
+        let variants = std::mem::take(&mut self.variants);
+        if variants.is_empty() && bindings.is_empty() {
+            let body = self.layout(target);
+            return project(expr, self.lambda(pat, body));
+        }
+        let defer = !self.tests.is_empty();
+        // A plain key remains readable by the following filter. Final layouts and
+        // constructed payload keys wait until that filter has consumed its fields.
+        let output = if defer && !matches!(target, Target::Key(p) if !p.has_variant()) { Target::Current } else { target };
+        let mut list = !variants.is_empty();
+        let mut body = if list { self.payload_lists(&variants, 0, output) } else { self.layout(output) };
+        for (i, (payloads, name)) in bindings.iter().enumerate().rev() {
+            let mut context = vec!["k".into(), "v".into()];
+            context.extend(bindings[..i].iter().map(|(_, n)| n.clone()));
+            let context = tuple(&context);
+            body = format!("({context}, {payloads}) cap_list map (({context}, {name}) -> {body}){}", if list { " flatten .1" } else { "" });
+            list = true;
+        }
+        expr = Expr::FlatMap(Box::new(expr), self.lambda(pat, body));
+        if defer {
+            expr = self.filters(expr, "(k, v)");
+            assert!(self.tests.is_empty(), "fragment tests must be bound at their checkpoint");
+            let body = self.layout(target);
+            expr = project(expr, self.lambda("(k, v)", body));
+        }
+        expr
+    }
+}
+
+impl Fragment {
+    fn expansion(&self, scope: &corgi::Scope) -> Expr {
+        let mut s = State { scope, next: 0, names: vec![], bound: BTreeMap::new(),
+            locations: BTreeMap::new(), key: BTreeMap::new(), variants: vec![], tests: vec![] };
+        let first = &self.atoms[0];
+        s.ingest(&first.key, "k".into());
+        s.key = s.locations.iter().map(|(a, expr)| (a.clone(), KeyRead::Value(expr.clone()))).collect();
+        s.ingest(&first.val, "v".into());
+        // A restricted first key is tested before a new arrangement, so rows it rejects are never paired.
+        let same_key = self.atoms.get(1).is_some_and(|next| first.key == next.key);
+        let direct = first.key.unrestricted() && same_key;
+        let mut expr = first.source.clone();
+        if !direct {
+            // Tests may restrict the key, but an identical lookup must retain its original value.
+            let target = self.atoms.get(1).map_or(Target::Output, |a| if same_key { Target::Keep(&a.key) } else { Target::Key(&a.key) });
+            expr = s.checkpoint(expr, target, &format!("(k : {}, v : {})", first.key.shape_pattern(), first.val.shape_pattern()));
+        }
+        for i in 1..self.atoms.len() {
+            let atom = &self.atoms[i];
+            s.validate_key(&atom.key);
+            s.ingest(&atom.val, "r".into());
+            let body = s.row("k".into(), s.key.clone());
+            let pat = if i == 1 && direct {
+                format!("(k : {}, v : {}, r : {})", first.key.shape_pattern(), first.val.shape_pattern(), atom.val.shape_pattern())
+            } else { format!("(k, v, r : {})", atom.val.shape_pattern()) };
+            expr = Expr::Join(Box::new(expr), Box::new(atom.source.clone()), s.lambda(&pat, body));
+            let target = self.atoms.get(i + 1).map_or(Target::Output, |a| Target::Key(&a.key));
+            expr = s.checkpoint(expr, target, "(k, v)");
+        }
+        expr
+    }
+}
diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs
new file mode 100644
index 000000000..318d44df3
--- /dev/null
+++ b/interactive/tests/fragments.rs
@@ -0,0 +1,502 @@
+//! Weighted fragment semantics, restrictions, placement, and emitted plans.
+use interactive::{lower, parse, plan, passes::Config, server::Server};
+
+fn program(src: &str) -> interactive::scope_ir::Program { lower::lower_tree(parse::pipe::parse(src)) }
+fn rejects(src: &str, expected: &str) {
+    let panic = std::panic::catch_unwind(|| program(src)).expect_err("expected a fragment error");
+    let message = panic.downcast_ref::<String>().map(String::as_str)
+        .or_else(|| panic.downcast_ref::<&str>().copied()).expect("a string panic");
+    assert!(message.contains(expected), "expected {expected:?}, got {message:?} for {src}");
+}
+fn rows(src: &str, feeds: Vec<(usize, String)>) -> Vec<(String, String, i64)> {
+    let p = program(src);
+    timely::execute_directly(move |worker| {
+        let mut server = Server::new();
+        server.set_passes(Config::parse("all").unwrap());
+        server.install(worker, "p", &p).unwrap();
+        for (i, text) in feeds { server.feed("p", i, &text, None).unwrap(); }
+        server.tick(worker);
+        server.snapshot(worker, "p::out").unwrap().rows()
+    })
+}
+
+#[test]
+fn joins_multiply_weights_and_handle_retractions() {
+    let src = r#"
+        let a = input 0 : (int, int);
+        let b = input 1 : (int, int);
+        let c = input 2 : ((int, int), ());
+        output "out" = a(x, y) * b(y, z) * c((x, z), ())
+            | map(((x, y, z), ()) -> ((x, y, z), ()));
+    "#;
+    let p = program(src);
+    timely::execute_directly(move |worker| {
+        let mut s = Server::new();
+        s.set_passes(Config::parse("all").unwrap());
+        s.install(worker, "p", &p).unwrap();
+        for (i, text) in [(0,"1 val=2 diff=2"),(1,"2 val=3 diff=3"),(2,"1,3 diff=4")] { s.feed("p",i,text,None).unwrap(); }
+        s.tick(worker);
+        assert_eq!(s.snapshot(worker,"p::out").unwrap().rows(), vec![("tuple(1,2,3)".into(),"tuple()".into(),24)]);
+        s.feed("p",0,"1 val=2 diff=-1",None).unwrap(); s.tick(worker);
+        assert_eq!(s.snapshot(worker,"p::out").unwrap().rows()[0].2,12);
+        s.feed("p",1,"2 val=3 diff=-3",None).unwrap(); s.tick(worker);
+        assert!(s.snapshot(worker,"p::out").unwrap().is_empty());
+    });
+}
+
+#[test]
+fn patterns_tests_and_output_order() {
+    let src = r#"
+        let a = input 0 : ((int, int), (List(int), int));
+        let b = input 1 : (int, int);
+        output "out" = a((x, x), ("hi", y)) * b(y, -7);
+    "#;
+    assert_eq!(rows(src,vec![(0,"1,1 val=tuple(list(104,105),2)\n1,9 val=tuple(list(104,105),2)\n3,3 val=tuple(list(104),2)".into()),(1,"2 val=-7 diff=3\n2 val=7".into())]),vec![("tuple(1,2)".into(),"tuple()".into(),3)]);
+    assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(_, _);",vec![(0,"1 val=2 diff=4".into())]),vec![("tuple()".into(),"tuple()".into(),4)]);
+    assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(x, _);",vec![(0,"1 val=2".into())]),vec![("tuple(1)".into(),"tuple()".into(),1)]);
+}
+
+#[test]
+fn empty_list_pattern_does_not_choose_an_element_shape() {
+    assert_eq!(rows("let a = input 0 : (int,List((int,int))); output \"out\" = a(x, []);",
+        vec![(0,"1 val=list()\n2 val=list(tuple(3,4))".into())]),
+        vec![("tuple(1)".into(),"tuple()".into(),1)]);
+}
+
+#[test]
+fn nested_open_variants_and_early_key_extraction() {
+    let src = r#"
+        enum A = Shared int | Empty;
+        enum B = Shared int | Other;
+        let a = input 0 : (int, Option(Result((int, List(int)), ())));
+        let b = input 1 : (int, A);
+        let c = input 2 : (A, int);
+        output "out" = a(_, Some(Ok((x, text)))) * b(x, A::Shared(y)) * c(A::Shared(y), z);
+    "#;
+    let got = rows(src,vec![
+        (0,"1 val=inject(1,inject(0,tuple(2,list(65))))\n2 val=inject(0,tuple())\n3 val=inject(1,inject(1,tuple()))".into()),
+        (1,"2 val=inject(0,3)\n2 val=inject(1,tuple())".into()),
+        (2,"inject(0,3) val=4\ninject(1,tuple()) val=9".into()),
+    ]);
+    assert_eq!(got,vec![("tuple(2,list(65),3,4)".into(),"tuple()".into(),1)]);
+    let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap();
+    assert!(report.stages.last().unwrap().problems.is_empty());
+    let text = report.emit_ddp().unwrap().text;
+    for tail in text.lines().filter(|line| line.contains("| flatmap(")) {
+        assert!(!tail.contains(" head"), "{tail}");
+    }
+    let nested_key = r#"
+        enum Inner = Leaf int | Other;
+        enum Key = Wrap Inner | Empty;
+        let a = input 0 : (int, ());
+        let b = input 1 : (Key, int);
+        output "out" = a(y, ()) * b(Key::Wrap(Inner::Leaf(y)), z);
+    "#;
+    assert_eq!(rows(nested_key, vec![(0, "1".into()),
+        (1, "inject(0,inject(0,1)) val=9".into())]),
+        vec![("tuple(1,9)".into(), "tuple()".into(), 1)]);
+    let report = plan::report(nested_key, &|_| None, &Config::parse("all").unwrap()).unwrap();
+    let text = report.emit_ddp().unwrap().text;
+    assert!(!text.lines().filter(|line| line.contains("| flatmap(")).any(|line| line.contains(" head")), "{text}");
+}
+
+#[test]
+fn paths_feedback_pipeline_union_and_difference() {
+    let src = r#"
+        let a = input 0 : (int, int);
+        looped: { var x : (int,int); set x = a; }
+        let b = ::looped::x;
+        output "out" = (looped::x(x,y) * b(x,z) | map(((x,y,z),()) -> (y,z)))
+            + (a(x,y) | map(((x,y),()) -> (x,y))) - a;
+    "#;
+    assert_eq!(rows(src,vec![(0,"1 val=2".into())]),vec![("2".into(),"2".into(),1)]);
+}
+
+#[test]
+fn key_restrictions_and_arity() {
+    for (atom, error) in [
+        ("b(_,z)", "a later fragment key cannot contain a wildcard"),
+        ("b((x,_),z)", "a later fragment key cannot contain a wildcard"),
+        ("b(missing,z)", "fragment key variable `missing` is not bound by an earlier atom"),
+        ("b((x,missing),z)", "fragment key variable `missing` is not bound by an earlier atom"),
+        ("b(Some(_),z)", "a later fragment key cannot contain a wildcard"),
+    ] {
+        let src=format!("let a = input 0 : (int,int); let b = a; output \"out\" = a(x,y) * {atom};");
+        rejects(&src, error);
+    }
+    for (bad, error) in [("a(x)", "Expected Comma, got RParen"),
+        ("a(x,y,z)", "Expected RParen, got Comma"), ("a(x,[y])", "a list pattern is a literal")] {
+        rejects(&format!("let a=input 0 : (int,int); output \"out\"={bad};"), error);
+    }
+}
+
+#[test]
+fn constant_later_keys_explain_cross_joins() {
+    for key in ["1", "-7", "\"ok\"", "[1,2]", "[]", "[[]]", "()", "(1,())", "Empty", "Empty()", "Group(7)", "(Empty(),[1,2])"] {
+        let src = format!("enum K=Group int | Empty; let a=input 0 : (int,int); let b=a; output \"out\"=a(x,y)*b({key},z);");
+        rejects(&src, "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP");
+    }
+}
+
+#[test]
+fn later_keys_combine_bound_variables_with_constants() {
+    let src = r#"
+        enum K = Group (int,int) | Empty;
+        let a = input 0 : (int,());
+        let b = input 1 : ((int,int),int);
+        let c = input 2 : (K,int);
+        let d = input 3 : ((int,K,()),int);
+        output "out" = a(rid,()) * b((rid,7),v) * c(Group((rid,v)),w) * d((w,Empty,()),z);
+    "#;
+    assert_eq!(rows(src, vec![(0,"1\n2".into()), (1,"1,7 val=10\n2,8 val=20".into()),
+        (2,"inject(0,tuple(1,10)) val=99".into()), (3,"tuple(99,inject(1,tuple()),tuple()) val=5".into())]),
+        vec![("tuple(1,10,99,5)".into(), "tuple()".into(), 1)]);
+}
+
+#[test]
+fn bare_constructors_match_empty_payloads_and_do_not_bind_variables() {
+    assert_eq!(rows("let a=input 0 : (int,Option(int)); output \"out\"=a(x,None);",
+        vec![(0,"1 val=inject(0,tuple()) diff=3\n2 val=inject(1,7)".into())]),
+        vec![("tuple(1)".into(), "tuple()".into(), 3)]);
+    let src = "let a=input 0 : (int,(Option(int),Option(int))); output \"out\"=a(x,(None,None));";
+    assert_eq!(rows(src, vec![(0,"1 val=tuple(inject(0,tuple()),inject(0,tuple()))\n2 val=tuple(inject(0,tuple()),inject(1,7))\n3 val=tuple(inject(1,7),inject(1,7))".into())]),
+        vec![("tuple(1)".into(), "tuple()".into(), 1)]);
+    let text = plan::report(src, &|_| None, &Config::all()).unwrap().emit_ddp().unwrap().text;
+    assert!(!text.contains("| filter("), "repeated None must not become equality: {text}");
+    assert_eq!(rows("enum K=Empty | Full int; let a=input 0 : (K,int); output \"out\"=a(K::Empty,v);",
+        vec![(0,"inject(0,tuple()) val=9\ninject(1,7) val=8".into())]),
+        vec![("tuple(9)".into(), "tuple()".into(), 1)]);
+    assert_eq!(rows("let a=input 0 : (int,int); output \"out\"=a(x,Fresh);", vec![(0,"1 val=2".into())]),
+        vec![("tuple(1,2)".into(), "tuple()".into(), 1)]);
+}
+
+#[test]
+fn bare_constructors_requiring_payloads_report_the_constructor() {
+    for name in ["Some", "Group", "K::Group", "Ok"] {
+        let src = format!("enum K=Group int | Empty; let a=input 0 : (int,K); output \"out\"=a(x,{name});");
+        rejects(&src, &format!("fragment constructor `{name}` requires a payload; write `{name}(pattern)`"));
+    }
+    rejects("enum A=Shared; enum B=Shared; let a=input 0 : (int,A); output \"out\"=a(x,Shared);",
+        "fragment constructor `Shared`: 'Shared' is a variant of");
+}
+
+#[test]
+fn nested_empty_list_literals_report_a_ddir_error() {
+    for literal in ["[[]]", "[[[]]]", "[([], 7)]"] {
+        let src = format!("let a=input 0 : (int,List(List(int))); output \"out\"=a(x,{literal});");
+        rejects(&src, &format!("DDIR fragment list literal `{literal}` needs a complete element shape; use plain DDP with an explicit list shape"));
+    }
+    for (literal, value) in [("[[1],[2]]", "list(list(1),list(2))"),
+        ("[[1],[]]", "list(list(1),list())"), ("[[],[1]]", "list(list(),list(1))")] {
+        let src = format!("let a=input 0 : (int,List(List(int))); output \"out\"=a(x,{literal});");
+        assert_eq!(rows(&src, vec![(0,format!("1 val={value}\n2 val=list()"))]),
+            vec![("tuple(1)".into(), "tuple()".into(), 1)]);
+    }
+}
+
+#[test]
+fn fragments_read_their_own_feedback_variable() {
+    let src = r#"
+        let edges = input 0 : (int,int);
+        let roots = input 1 : (int,());
+        s: {
+            set x = roots + (x(src,()) * edges(src,dst) | map(((src,dst),()) -> (dst,()))) | distinct;
+        }
+        output "out" = s::x;
+    "#;
+    assert_eq!(rows(src, vec![(0,"1 val=2\n2 val=3\n3 val=2\n4 val=5\n9 val=10".into()), (1,"1\n4".into())]),
+        (1..=5).map(|n| (n.to_string(), "tuple()".into(), 1)).collect::<Vec<_>>());
+}
+
+#[test]
+fn fragments_are_absent_from_every_emitted_plan_stage() {
+    use plan::ddp::DdpStage;
+    let src = "let a=input 0 : (int,int); let b=a; output \"out\"=a(x,7)*b(x,y);";
+    let report = plan::report(src, &|_| None, &Config::all()).unwrap();
+    for stage in [DdpStage::Parsed, DdpStage::Lowered, DdpStage::Optimized, DdpStage::Projected, DdpStage::AsOf] {
+        let text = report.emit_ddp_at(stage).unwrap().text;
+        assert!(!text.contains('*'), "{stage:?}: {text}");
+        assert!(text.contains("| join("), "{stage:?}: {text}");
+    }
+    assert!(!report.emit_ddp().unwrap().text.contains('*'));
+}
+
+#[test]
+fn parsed_shows_the_expansion_and_direct_first_tests_follow_the_first_join() {
+    use plan::ddp::DdpStage;
+    let src="let a = input 0 : (int,int); let b = a; let c = a; output \"out\" = a(x,7) * b(x,y) * c(y,9);";
+    let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap();
+    let parsed=report.emit_ddp_at(DdpStage::Parsed).unwrap().text;
+    assert!(!parsed.contains('*') && parsed.contains("| join("));
+    let lowered=report.emit_ddp_at(DdpStage::Lowered).unwrap().text;
+    assert!(!lowered.contains('*'));
+    assert_eq!(lowered.matches("| join(").count(),2);
+    assert_eq!(lowered.matches("| filter(").count(), 2);
+    assert!(!lowered.contains("| flatmap("));
+    assert!(lowered.find("| join(").unwrap() < lowered.find("| filter(").unwrap());
+    assert!(lowered.find("| filter(").unwrap() < lowered.rfind("| join(").unwrap());
+    let again=plan::report(&parsed,&|_|None,&Config::parse("all").unwrap()).unwrap();
+    assert!(report.compare_final(&again).is_empty());
+}
+
+#[test]
+fn open_result_constructor_key_exposes_corgis_missing_lane_shape() {
+    let src = "let a=input 0 : (int,()); let b=input 1 : (Result(int,()),int); output \"out\"=a(x,()) * b(Ok(x),y);";
+    let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap();
+    assert!(report.stages[0].problems.iter().any(|s|s.contains("declares no payload shape")));
+}
+
+#[test]
+fn direct_first_join_reads_the_key_and_carries_it_when_rekeyed() {
+    use interactive::scope_ir::{Item, Node, Ref};
+    let src = "let a=input 0 : (int,int); let b=input 1 : (int,int); let c=input 2 : (int,int); output \"out\"=a(x,y) * b(x,z) * c(y,w);";
+    let report = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap();
+    let root = &report.stages[0].program.root;
+    let (left, projection) = root.items.iter().find_map(|item| match item {
+        Item::Op(Node::Join { left, projection, .. }) => Some((left, projection)), _ => None,
+    }).unwrap();
+    let Ref::Local(i) = left else { panic!("expected arrangement") };
+    assert!(matches!(&root.items[*i], Item::Op(Node::Arrange(Ref::Import(0)))));
+    let source = format!("{projection:?}");
+    let graph = corgi::parse_ml_in(&source[1..source.len()-1], &corgi::Scope::default()).unwrap();
+    let got = corgi::eval_graph(&graph, corgi::Value::Prod(vec![
+        corgi::Value::int([10]), corgi::Value::int([20]), corgi::Value::int([30]),
+    ]));
+    // y becomes the key. x leaves the key and must survive in the value with z.
+    assert_eq!(got, corgi::Value::Prod(vec![corgi::Value::int([20]),
+        corgi::Value::Prod(vec![corgi::Value::int([10]), corgi::Value::int([30])])]));
+    assert_eq!(rows(src, vec![(0,"10 val=20".into()),(1,"10 val=30".into()),(2,"20 val=40".into())]),
+        vec![("tuple(10,20,30,40)".into(),"tuple()".into(),1)]);
+}
+
+#[test]
+fn first_atom_tests_do_not_prevent_direct_arrangement_reuse() {
+    use interactive::scope_ir::{Item, Node, Ref};
+    let src = r#"
+        let a=input 0 : (int,(Option(int),int,int));
+        let b=input 1 : (int,int);
+        let c=input 2 : (int,int);
+        output "out"=a(x,(Some(y),x,7)) * b(x,z) * c(z,w);
+    "#;
+    let report = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap();
+    let root = &report.stages[0].program.root;
+    let left = root.items.iter().find_map(|item| match item {
+        Item::Op(Node::Join { left, .. }) => Some(left), _ => None,
+    }).unwrap();
+    let Ref::Local(i) = left else { panic!("expected arrangement") };
+    assert!(matches!(&root.items[*i], Item::Op(Node::Arrange(Ref::Import(0)))));
+    let lowered = report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text;
+    assert_eq!(lowered.matches("| filter(").count(), 1);
+    assert_eq!(lowered.matches("| flatmap(").count(), 1);
+    assert!(lowered.find("| join(").unwrap() < lowered.find("| filter(").unwrap());
+    assert!(lowered.find("| filter(").unwrap() < lowered.find("| flatmap(").unwrap());
+    assert!(lowered.find("| flatmap(").unwrap() < lowered.rfind("| join(").unwrap());
+    assert_eq!(rows(src, vec![
+        (0,"1 val=tuple(inject(1,9),1,7)\n1 val=tuple(inject(0,tuple()),1,7)\n1 val=tuple(inject(1,8),2,7)\n1 val=tuple(inject(1,8),1,6)".into()),
+        (1,"1 val=2 diff=3".into()),(2,"2 val=4 diff=2".into()),
+    ]), vec![("tuple(1,9,2,4)".into(),"tuple()".into(),6)]);
+}
+
+#[test]
+fn first_key_restrictions_are_tested_before_a_new_arrangement() {
+    use interactive::scope_ir::{Item, Node, Ref};
+    let src = r#"
+        enum Key = Group (int,int) | User (int,int);
+        let a=input 0 : (Key,int);
+        let b=input 1 : (Key,int);
+        output "out"=a(Group((x,x)),y) * b(Group((x,x)),z);
+    "#;
+    let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap();
+    let root=&report.stages[0].program.root;
+    let left=root.items.iter().find_map(|item|match item {
+        Item::Op(Node::Join{left,..})=>Some(left),_=>None,
+    }).unwrap();
+    let Ref::Local(i)=left else {panic!("expected arrangement")};
+    assert!(!matches!(&root.items[*i],Item::Op(Node::Arrange(Ref::Import(0)))));
+    let lowered=report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text;
+    // One flatmap binds the payload before the arrangement; the other reads the preserved key after the join.
+    assert_eq!(lowered.matches("| filter(").count(), 1);
+    assert_eq!(lowered.matches("| flatmap(").count(), 2);
+    assert!(lowered.find("| flatmap(").unwrap() < lowered.find("| join(").unwrap());
+    assert!(lowered.find("| filter(").unwrap() < lowered.find("| join(").unwrap());
+    assert_eq!(rows(src,vec![
+        (0,"inject(0,tuple(1,1)) val=8 diff=2\ninject(0,tuple(1,2)) val=7\ninject(1,tuple(2,2)) val=6".into()),
+        (1,"inject(0,tuple(1,1)) val=9 diff=3\ninject(0,tuple(1,2)) val=5\ninject(1,tuple(2,2)) val=4".into()),
+    ]),vec![("tuple(1,8,9)".into(),"tuple()".into(),6)]);
+}
+
+#[test]
+fn nested_variant_lists_and_equalities_build_only_survivors() {
+    let src = r#"
+        let a = input 0 : (int, Option(Result((int,int),())));
+        output "out" = a(doc, Some(Ok((u,u))));
+    "#;
+    assert_eq!(rows(src, vec![(0, "1 val=inject(1,inject(0,tuple(2,2))) diff=3\n1 val=inject(1,inject(0,tuple(2,9)))\n2 val=inject(0,tuple())\n3 val=inject(1,inject(1,tuple()))".into())]),
+        vec![("tuple(1,2)".into(), "tuple()".into(), 3)]);
+    let report = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap();
+    let text = report.emit_ddp().unwrap().text;
+    assert_eq!(text.matches("| flatmap(").count(), 1, "{text}");
+    assert!(!text.contains(" head") && !text.contains("[]") && !text.contains(" match ("), "{text}");
+    let again = plan::report(&text, &|_| None, &Config::parse("all").unwrap()).unwrap();
+    assert!(report.compare_final(&again).is_empty());
+}
+
+#[test]
+fn one_atom_variant_demux() {
+    let src = r#"
+        enum Member = User int | Group int;
+        let grants = input 0 : (int, (Member, int));
+        output "out" = grants(doc, (User(u), level)) | map(((doc, u, level), ()) -> (u, (doc, level)));
+    "#;
+    let feed = "100 val=tuple(inject(0,1),3)\n100 val=tuple(inject(1,10),2) diff=5\n200 val=tuple(inject(0,2),1) diff=2";
+    assert_eq!(rows(src, vec![(0, feed.into())]),
+        vec![("1".into(), "tuple(100,3)".into(), 1), ("2".into(), "tuple(200,1)".into(), 2)]);
+}
+
+#[test]
+fn first_literal_precedes_the_first_rekey_and_arrangement() {
+    use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}};
+    let src = r#"
+        let message = input 0 : (int, (int,int,int));
+        output "out" = message(_, (1, who, parent)) * message(parent, (_, to, _));
+    "#;
+    let p = program(src);
+    let items = &p.root.items;
+    let filter = items.iter().position(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::Filter(_)]))).unwrap();
+    let join = items.iter().position(|i| matches!(i, Item::Op(Node::Join { .. }))).unwrap();
+    let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[join] else { panic!() };
+    let Item::Op(Node::Arrange(Ref::Local(map))) = &items[*arrange] else { panic!() };
+    let Item::Op(Node::Linear { input: Ref::Local(input), ops }) = &items[*map] else { panic!() };
+    assert_eq!(*input, filter);
+    assert!(filter < *map && *map < *arrange && *arrange < join);
+    let LinearOp::Project(lambda) = &ops[0] else { panic!() };
+    assert!(!format!("{lambda:?}").contains("v.0"), "tested kind must not reach the arrangement");
+    assert_eq!(rows(src, vec![(0, "10 val=tuple(1,2,20)\n11 val=tuple(0,3,20)\n20 val=tuple(0,4,0)".into())]),
+        vec![("tuple(2,20,4)".into(), "tuple()".into(), 1)]);
+}
+
+#[test]
+fn later_literals_follow_their_join_and_leave_before_the_next_arrangement() {
+    use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}};
+    let src = r#"
+        let asked = input 0 : (int,(int,int));
+        let by_name = input 1 : (int,int);
+        let place = input 2 : (int,(int,int));
+        output "out" = asked(xname,(rid,yname)) * by_name(xname,x)
+            * place(x,(7,_)) * by_name(yname,y) * place(y,(7,_));
+    "#;
+    let p = program(src);
+    let items = &p.root.items;
+    let joins: Vec<_> = items.iter().enumerate().filter_map(|(i,n)| matches!(n, Item::Op(Node::Join { .. })).then_some(i)).collect();
+    let filters: Vec<_> = items.iter().enumerate().filter_map(|(i,n)| matches!(n, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::Filter(_)] )).then_some(i)).collect();
+    assert_eq!(filters.len(), 2);
+    assert!(joins[1] < filters[0] && filters[0] < joins[2] && joins[3] < filters[1]);
+    let Item::Op(Node::Linear { input, .. }) = &items[filters[0]] else { panic!() };
+    assert_eq!(input, &Ref::Local(joins[1]));
+    let report = plan::report(src, &|_| None, &Config::all()).unwrap();
+    let shapes = &report.stages[0].shapes;
+    let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[joins[2]] else { panic!() };
+    let before = shapes.of(&Ref::Local(filters[0])).unwrap();
+    let after = shapes.of(&Ref::Local(*arrange)).unwrap();
+    let (corgi::Shape::Prod(before), corgi::Shape::Prod(after)) = (&before.val, &after.val) else { panic!() };
+    assert_eq!(before.len(), after.len() + 1, "discard the tested kind before arranging");
+}
+
+#[test]
+fn test_free_chain_has_no_flatmap_or_layout_stage_after_the_last_join() {
+    use interactive::{ir::LinearOp, scope_ir::{Item, Node}};
+    let src = "let a=input 0 : (int,int); let b=a; let c=a; output \"out\"=a(x,y)*b(y,z)*c(z,w);";
+    let p = program(src);
+    assert!(!p.root.items.iter().any(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if ops.iter().any(|op| matches!(op, LinearOp::FlatMap(_))))));
+    assert!(matches!(p.root.items.last(), Some(Item::Op(Node::Join { .. }))));
+    assert_eq!(rows(src, vec![(0, "1 val=2\n2 val=3\n3 val=4".into())]),
+        vec![("tuple(1,2,3,4)".into(), "tuple()".into(), 1)]);
+}
+
+#[test]
+fn variant_payload_rekeys_before_its_join_without_head_or_tag_columns() {
+    use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}};
+    let src = r#"
+        enum Member = User int | Group int;
+        let members = input 0 : (int, Member);
+        let groups = input 1 : (int, Member);
+        output "out" = members(doc, Group(g)) * groups(g, m);
+    "#;
+    let p = program(src);
+    let items = &p.root.items;
+    let join = items.iter().position(|i| matches!(i, Item::Op(Node::Join { .. }))).unwrap();
+    let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[join] else { panic!() };
+    let Item::Op(Node::Arrange(Ref::Local(flatmap))) = &items[*arrange] else { panic!() };
+    let Item::Op(Node::Linear { ops, .. }) = &items[*flatmap] else { panic!() };
+    let [LinearOp::FlatMap(lambda)] = ops.as_slice() else { panic!() };
+    let text = format!("{lambda:?}");
+    assert!(text.contains("unweave .2"));
+    assert!(!text.contains(" head") && !text.contains(" match ("), "{text}");
+    assert!(!items.iter().any(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if ops.iter().any(|op| matches!(op, LinearOp::Filter(_))))));
+    assert_eq!(rows(src, vec![(0, "100 val=inject(1,10)\n200 val=inject(0,20)".into()), (1, "10 val=inject(0,3)".into())]),
+        vec![("tuple(100,10,inject(0,3))".into(), "tuple()".into(), 1)]);
+}
+
+#[test]
+fn wildcard_payloads_still_check_their_tuple_shape() {
+    for (val, error) in [("Some((_,_))", "wants a product of 2 fields, not Int"),
+        ("Some((_))", "wants a product of 1 field, not Int")] {
+        let src = format!("let a=input 0 : (int,Option(int)); output \"out\"=a(_, {val});");
+        let report = plan::report(&src, &|_| None, &Config::all()).unwrap();
+        assert!(report.stages[0].problems.iter().any(|e| e.contains(error)), "{src}: {:?}", report.stages[0].problems);
+    }
+}
+
+#[test]
+fn variants_rekey_before_pending_plain_payload_tests() {
+    use interactive::{ir::LinearOp, scope_ir::{Export, Item, Node, Ref}};
+    let src = r#"
+        enum Member = User int | Group (int,int);
+        let members = input 0 : (int,Member);
+        let groups = input 1 : (int,int);
+        output "out" = members(doc,Group((g,7))) * groups(g,m);
+    "#;
+    let mut p = program(src);
+    let flatmap = p.root.items.iter().position(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::FlatMap(_)]))).unwrap();
+    // Observe the variant stage before its plain filter: its key is already g.
+    p.root.exports.push(Export { name: "keyed".into(), value: Ref::Local(flatmap), shape: None });
+    timely::execute_directly(move |worker| {
+        let mut s = Server::new();
+        s.set_passes(Config::all());
+        s.install(worker, "p", &p).unwrap();
+        s.feed("p", 0, "100 val=inject(1,tuple(10,7))\n200 val=inject(1,tuple(20,8))\n300 val=inject(0,30)", None).unwrap();
+        s.feed("p", 1, "10 val=3\n20 val=4", None).unwrap();
+        s.tick(worker);
+        assert_eq!(s.snapshot(worker, "p::keyed").unwrap().rows(), vec![
+            ("10".into(), "tuple(100,7)".into(), 1),
+            ("20".into(), "tuple(200,8)".into(), 1),
+        ]);
+        assert_eq!(s.snapshot(worker, "p::out").unwrap().rows(),
+            vec![("tuple(100,10,3)".into(), "tuple()".into(), 1)]);
+    });
+}
+
+#[test]
+fn identical_restricted_first_keys_keep_their_original_values() {
+    for (shape, pattern, kept, dropped) in [
+        ("Result(int,int)", "Ok(x)", "inject(0,1)", "inject(1,2)"),
+        ("Result(int,int)", "Err(x)", "inject(1,1)", "inject(0,2)"),
+        ("(int,Option(int))", "(x,None)", "tuple(1,inject(0,tuple()))", "tuple(2,inject(1,7))"),
+    ] {
+        let src = format!("let a=input 0 : ({shape},int); let b=input 1 : ({shape},int); output \"out\"=a({pattern},v)*b({pattern},w);");
+        assert_eq!(rows(&src, vec![(0,format!("{kept} val=8 diff=2\n{dropped} val=7")),
+            (1,format!("{kept} val=9 diff=3\n{dropped} val=5"))]),
+            vec![("tuple(1,8,9)".into(), "tuple()".into(), 6)]);
+        let report = plan::report(&src, &|_| None, &Config::all()).unwrap();
+        let text = report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text;
+        assert!(!text.contains(" inject "), "{text}");
+    }
+}
+
+#[test]
+fn empty_lists_and_strings_in_later_keys_report_a_ddir_error() {
+    for literal in ["[]", "\"\"", "(7,[])", "Some((x,[]))"] {
+        let src = format!("let a=input 0 : (int,int); let b=input 1 : ((int,List(int)),int); output \"out\"=a(x,y)*b((x,{literal}),z);");
+        rejects(&src, "DDIR fragment key literal `[]` needs a complete element shape; use plain DDP with an explicit list shape");
+    }
+}