diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml
index fd6bed9d8..fa2d72a96 100644
--- a/interactive/Cargo.toml
+++ b/interactive/Cargo.toml
@@ -14,7 +14,7 @@ workspace = true
 [dependencies]
 columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
-corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "13f3ab8f84c7735c44ee9acca650e126495f7a11", features = ["serde"] }
+corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "be003988f17e3998a8abba3e168a2520a46d10e4", features = ["serde"] }
 differential-dataflow = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"
diff --git a/interactive/server/tests/multiworker.rs b/interactive/server/tests/multiworker.rs
index 0c23dfefa..6c3b7fb57 100644
--- a/interactive/server/tests/multiworker.rs
+++ b/interactive/server/tests/multiworker.rs
@@ -215,6 +215,37 @@ fn corgi_commands_replay_on_four_workers_without_duplicating_input() {
     assert_backend("corgi");
 }
 
+fn assert_typed_sources(backend: &str) {
+    let (server, mut writer, mut reader) = start_server(backend, 4);
+    request(&mut writer, &mut reader, "g", "g load graph begin\nlet rows = input 0 : ((int, List(int), Option(List(int))) ; ());\nexport \"typed.rows\" = rows | arrange;\ng end-load\n");
+    request(&mut writer, &mut reader, "q", "q load copy begin\nlet rows = import \"typed.rows\" : ((int, List(int), Option(List(int))) ; ());\nexport \"typed.copy\" = rows;\nq end-load\n");
+    // Neither a first empty list nor a never-populated sum lane can supply
+    // its encoding by example. Both the producer and imported trace use the
+    // declared shape, without sentinel rows or altered data.
+    request(&mut writer, &mut reader, "f", "f feed graph 0 tuple(1,list(),inject(0,tuple()))\n");
+    request(&mut writer, &mut reader, "t", "t tick\n");
+    assert_eq!(request(&mut writer, &mut reader, "p", "p peek typed.copy\n"),
+        vec!["diff=1 key=Tuple([Int(1), List([]), Variant(0, Tuple([]))]) val=Tuple([])"]);
+    request(&mut writer, &mut reader, "f2", "f2 feed graph 0 tuple(2,list(97),inject(1,list()))\n");
+    request(&mut writer, &mut reader, "f3", "f3 feed graph 0 tuple(1,list(),inject(0,tuple())) diff=-1\n");
+    request(&mut writer, &mut reader, "t2", "t2 tick\n");
+    assert_eq!(request(&mut writer, &mut reader, "p2", "p2 peek typed.copy\n"),
+        vec!["diff=1 key=Tuple([Int(2), List([Int(97)]), Variant(1, List([]))]) val=Tuple([])"]);
+    drop(reader);
+    drop(writer);
+    server.stop();
+}
+
+#[test]
+fn vec_typed_empty_sources_and_imports_on_four_workers() {
+    assert_typed_sources("vec");
+}
+
+#[test]
+fn corgi_typed_empty_sources_and_imports_on_four_workers() {
+    assert_typed_sources("corgi");
+}
+
 /// Two consumers join request rows against the same named graph through TCP,
 /// while both the requests and graph change.
 fn assert_shared_import_requests(backend: &str, workers: usize) {
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index e149c4211..d8d7718b4 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -403,14 +403,16 @@ pub fn render_tree_rows<'s>(
     depth: usize,
     imports: Vec<crate::backend::vec::Col<'s>>,
 ) -> Vec<crate::backend::vec::Col<'s>> {
-    let corgi_imports: Vec<Collection<'s, Time, CC>> = imports
+    let corgi_imports: Vec<Collection<'s, Time, CC>> = crate::backend::vec::check_import_shapes(s, imports)
         .into_iter()
-        .map(|c| {
+        .zip(&s.imports)
+        .map(|(c, import)| {
+            let shape = import.shape.clone();
             c.inner
-                .unary(Pipeline, "ToCorgi", |_, _| {
-                    // The collection's shape, pinned from its first row: every later batch
-                    // transcodes against it (a misfit row is a panic in `transcode`).
-                    let mut pinned: Option<(Shape, Shape)> = None;
+                .unary(Pipeline, "ToCorgi", move |_, _| {
+                    // Ascriptions describe empty lists and inactive sum lanes;
+                    // unannotated imports retain first-row inference.
+                    let mut pinned = shape;
                     move |input, output| {
                         input.for_each(|cap, data| {
                             let rows = std::mem::take(data);
diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs
index c01cd52b7..efced8eb1 100644
--- a/interactive/src/backend/vec.rs
+++ b/interactive/src/backend/vec.rs
@@ -26,6 +26,24 @@ use crate::ir::{LinearOp, Diff, Time, Value, eval};
 pub type Row = Value;
 /// A rendered collection at the renderer's (inner, dynamic) time.
 pub type Col<'scope> = VecCollection<'scope, Time, (Row, Row), Diff>;
+
+/// Validate explicit encoding contracts at the row boundary on either backend.
+/// A mismatch panics inside a dataflow operator, which can take down the shared
+/// server on either backend. This is not transactional feed admission or
+/// per-program failure isolation.
+pub(crate) fn check_import_shapes<'s>(s: &st::Scope, imports: Vec<Col<'s>>) -> Vec<Col<'s>> {
+    use differential_dataflow::AsCollection;
+    use timely::dataflow::operators::core::Map;
+    assert_eq!(s.imports.len(), imports.len());
+    imports.into_iter().zip(&s.imports).map(|(c, import)| {
+        if let Some((key, val)) = import.shape.clone() {
+            c.inner.map(move |row| {
+                assert!(row.0.0.has_shape(&key) && row.0.1.has_shape(&val), "input does not match its shape ascription");
+                row
+            }).as_collection()
+        } else { c }
+    }).collect()
+}
 type Arr<'scope> = Arranged<'scope, TraceAgent<ValSpine<Row, Row, Time, Diff>>>;
 
 /// Append the user-iter coordinate to a value: extend a `Tuple` in place, or
@@ -167,5 +185,5 @@ pub fn render_tree<'s>(
     depth: usize,
     imports: Vec<Col<'s>>,
 ) -> Vec<Col<'s>> {
-    crate::backend::render_tree::<VecBackend>(s, scope, depth, imports)
+    crate::backend::render_tree::<VecBackend>(s, scope, depth, check_import_shapes(s, imports))
 }
diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs
index 3b54e17ac..0afbd4290 100644
--- a/interactive/src/corgi/logic.rs
+++ b/interactive/src/corgi/logic.rs
@@ -203,7 +203,11 @@ pub fn compilable(t: &Term) -> bool {
         Term::Binary(_, l, r) => compilable(l) && compilable(r),
         Term::If { cond, then, els } => compilable(cond) && compilable(then) && compilable(els),
         Term::Fold { list, init, step } => compilable(list) && compilable(init) && compilable(step),
-        Term::Unary(op, inner) => matches!(op, UnOp::Neg | UnOp::Not | UnOp::Len | UnOp::IsTag(_)) && compilable(inner),
+        // Keep this exhaustive so a new unary operator needs an explicit
+        // decision about whether it supports shape-free lowering.
+        Term::Unary(op, inner) => match op {
+            UnOp::Neg | UnOp::ToF64 | UnOp::F64Neg | UnOp::Not | UnOp::Len | UnOp::IsTag(_) => compilable(inner),
+        },
         // A literal tag into a declared type knows its whole sum; the built-ins and a data-driven
         // tag need the payload's shape.
         Term::Inject { tag, payload, sum } => {
@@ -332,6 +336,29 @@ pub fn compile(
                 BinOp::Add => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Add, Kind::U, 64), vec![p]) }
                 BinOp::Sub => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Sub, Kind::U, 64), vec![p]) }
                 BinOp::Mul => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Mul, Kind::U, 64), vec![p]) }
+                BinOp::Div => {
+                    let l = b.add(ArithOp::ToSigned, vec![lid]);
+                    let r = b.add(ArithOp::ToSigned, vec![rid]);
+                    let p = pair(b, l, r);
+                    let q = b.add(ArithOp::Bin(CBinOp::Div, Kind::I, 64), vec![p]);
+                    b.add(ArithOp::ToSigned, vec![q])
+                }
+                BinOp::Append => { let p = pair(b, lid, rid); b.add(Op::Append, vec![p]) }
+                BinOp::F64Add | BinOp::F64Sub | BinOp::F64Mul | BinOp::F64Div => {
+                    let expected = Shape::Sum(vec![Shape::Prim(64)]);
+                    if shape_of_term(l, env_shapes, None)? != expected || shape_of_term(r, env_shapes, None)? != expected {
+                        return Err("floating arithmetic expects two F64 newtypes; use float(int)".into());
+                    }
+                    let l = b.add(Op::Unwrap, vec![lid]);
+                    let r = b.add(Op::Unwrap, vec![rid]);
+                    let l = b.add(ArithOp::ToSigned, vec![l]);
+                    let r = b.add(ArithOp::ToSigned, vec![r]);
+                    let p = pair(b, l, r);
+                    let op = match op { BinOp::F64Add => CBinOp::Add, BinOp::F64Sub => CBinOp::Sub, BinOp::F64Mul => CBinOp::Mul, _ => CBinOp::Div };
+                    let f = b.add(ArithOp::Bin(op, Kind::F, 64), vec![p]);
+                    let payload = b.add(ArithOp::ToSigned, vec![f]);
+                    b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload])
+                }
                 BinOp::Eq | BinOp::Ne => {
                     // Cross-shape structural compare folds to a constant (Eq→0, Ne→1) over `anchor`;
                     // same-shape emits a real corgi `Rel`.
@@ -347,10 +374,16 @@ pub fn compile(
                 // Ordered compares go through `ToSigned` (XOR the sign bit: the order-preserving
                 // signed encoding), so they agree with `ir::eval`'s signed semantics for negative
                 // ints too. `Eq`/`Ne` are bit-equality — sign-safe as raw bits.
-                BinOp::Lt => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, ls, rs); b.add(CmpOp::Rel(Pred::Lt), vec![p]) }
-                BinOp::Le => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, ls, rs); b.add(CmpOp::Rel(Pred::Le), vec![p]) }
-                BinOp::Gt => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, rs, ls); b.add(CmpOp::Rel(Pred::Lt), vec![p]) }
-                BinOp::Ge => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, rs, ls); b.add(CmpOp::Rel(Pred::Le), vec![p]) }
+                BinOp::Lt | BinOp::Le | BinOp::Gt | BinOp::Ge => {
+                    let float_shape = Shape::Sum(vec![Shape::Prim(64)]);
+                    let (lid, rid) = if shape_of_term(l, env_shapes, None)? == float_shape && shape_of_term(r, env_shapes, None)? == float_shape {
+                        (b.add(Op::Unwrap, vec![lid]), b.add(Op::Unwrap, vec![rid]))
+                    } else { (lid, rid) };
+                    let ls = b.add(ArithOp::ToSigned, vec![lid]);
+                    let rs = b.add(ArithOp::ToSigned, vec![rid]);
+                    let p = if matches!(op, BinOp::Gt | BinOp::Ge) { pair(b, rs, ls) } else { pair(b, ls, rs) };
+                    b.add(CmpOp::Rel(if matches!(op, BinOp::Le | BinOp::Ge) { Pred::Le } else { Pred::Lt }), vec![p])
+                }
                 BinOp::And => { let p = pair(b, lid, rid); b.add(CmpOp::Min, vec![p]) }
                 BinOp::Or => { let p = pair(b, lid, rid); b.add(CmpOp::Max, vec![p]) }
             })
@@ -404,8 +437,14 @@ pub fn compile(
         // payload does not fill are built empty). A data-driven tag is a demux (`Branch`), which
         // needs every lane to share the payload's shape.
         Term::Inject { tag, payload, sum } => {
-            let pid = compile(payload, b, env, env_shapes, anchor, None)?;
-            let pshape = shape_of_term(payload, env_shapes, None)?;
+            // A declared constructor is also a type annotation for an empty
+            // payload (notably List<T>, whose T cannot come from runtime rows).
+            let payload_expected = match (sum, &**tag) {
+                (SumTy::Declared(lanes), Term::Int(t)) => usize::try_from(*t).ok().and_then(|t| lanes.get(t)),
+                _ => None,
+            };
+            let pid = compile(payload, b, env, env_shapes, anchor, payload_expected)?;
+            let pshape = shape_of_term(payload, env_shapes, payload_expected)?;
             match &**tag {
                 Term::Int(t) => {
                     let t = usize::try_from(*t).map_err(|_| format!("constructor tag {t} is negative"))?;
@@ -486,6 +525,27 @@ pub fn compile(
             Ok(match op {
                 // Wrapping negate on the raw two's-complement bits — exactly `-as_int()`.
                 UnOp::Neg => b.add(ArithOp::Neg(Kind::U, 64), vec![id]),
+                UnOp::ToF64 => {
+                    if shape != Shape::Prim(64) { return Err("float expects an Int".into()); }
+                    let sign = b.add(ArithOp::Shr(63), vec![id]);
+                    let negative = b.add(ArithOp::Neg(Kind::U, 64), vec![id]);
+                    let choices = b.tuple(vec![sign, negative, id]);
+                    let magnitude = b.add(Op::Select, vec![choices]);
+                    let positive = b.add(ArithOp::ToFloat(64), vec![magnitude]);
+                    let negative = b.add(ArithOp::Neg(Kind::F, 64), vec![positive]);
+                    let choices = b.tuple(vec![sign, negative, positive]);
+                    let f = b.add(Op::Select, vec![choices]);
+                    let payload = b.add(ArithOp::ToSigned, vec![f]);
+                    b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload])
+                }
+                UnOp::F64Neg => {
+                    if shape != Shape::Sum(vec![Shape::Prim(64)]) { return Err("fneg expects an F64 newtype".into()); }
+                    let payload = b.add(Op::Unwrap, vec![id]);
+                    let f = b.add(ArithOp::ToSigned, vec![payload]);
+                    let negative = b.add(ArithOp::Neg(Kind::F, 64), vec![f]);
+                    let payload = b.add(ArithOp::ToSigned, vec![negative]);
+                    b.add(Op::Inject(0, vec![Shape::Prim(64)]), vec![payload])
+                }
                 // `truthy` is "nonzero Int": scalars compare against zero; non-`Int` values
                 // are never truthy, so their `not` folds to the constant 1 (the cross-shape
                 // `Eq` fold's precedent).
@@ -545,7 +605,11 @@ pub fn compile(
         // list-intro kernel is corgi's call if this composition ever profiles hot.
         Term::List(fields) => {
             if fields.is_empty() {
-                return Err("an empty list literal has no element shape".into());
+                let Some(Shape::List(element)) = expected else {
+                    return Err("an empty list literal has no element shape".into());
+                };
+                let value = CValue::List(corgi::Bounds::Stride(0, 1), Box::new(CValue::empty(element)));
+                return Ok(b.add(Op::Lit(value), vec![anchor]));
             }
             let mut lanes = Vec::with_capacity(fields.len());
             for f in fields {
@@ -701,6 +765,45 @@ mod tests {
     fn u64s() -> Shape { Shape::Prim(64) }
     fn sum(lanes: Vec<Shape>) -> Shape { Shape::Sum(lanes) }
 
+    fn agrees_with_rows(term: &Term, shapes: &[Shape], rows: &[Vec<V>]) {
+        let mut b = Builder::<NumOp>::default();
+        let inp = b.input();
+        let env: Vec<_> = (0..shapes.len()).map(|i| b.add(Op::Field(i), vec![inp])).collect();
+        let out = compile(term, &mut b, &env, shapes, inp, None).unwrap();
+        let g = b.finish(out);
+        let os = corgi::shape_of(&g, &Shape::Prod(shapes.to_vec())).unwrap();
+        let cols = shapes.iter().enumerate().map(|(i, s)| {
+            transcode(&rows.iter().map(|r| r[i].clone()).collect::<Vec<_>>(), s)
+        }).collect();
+        let actual = untranscode(corgi::eval_graph(&g, CValue::Prod(cols)), &os);
+        let expected: Vec<_> = rows.iter().map(|r| crate::ir::eval(term, &mut r.clone())).collect();
+        assert_eq!(actual, expected);
+    }
+
+    #[test]
+    fn explicit_numeric_and_list_operations_agree() {
+        let rows: Vec<_> = [i64::MIN, -100, -1, 0, 1, 100, i64::MAX].into_iter()
+            .flat_map(|a| [-7, -1, 0, 1, 7].into_iter().map(move |b| vec![V::Int(a), V::Int(b)]))
+            .collect();
+        for source in ["idiv($0, $1)", "float($0)", "fneg(float($0))",
+            "fadd(float($0), float($1))", "fsub(float($0), float($1))",
+            "fmul(float($0), float($1))", "fdiv(float($0), float($1))",
+            "float($0) < float($1)", "float($0) >= float($1)",
+            "append(list($0), list($1, $0))"] {
+            let term = crate::parse::pipe::parse_term(source);
+            agrees_with_rows(&term, &[u64s(), u64s()], &rows);
+        }
+    }
+
+    #[test]
+    fn declared_constructor_types_an_empty_list() {
+        let term = Term::Inject {
+            tag: Box::new(Term::Int(0)), payload: Box::new(Term::List(vec![])),
+            sum: SumTy::Declared(vec![Shape::List(Box::new(Shape::List(Box::new(u64s()))))]),
+        };
+        agrees_with_rows(&term, &[u64s()], &[vec![V::Int(0)], vec![V::Int(1)]]);
+    }
+
     /// The pin on DDIR's `hash`: `ir::structural_hash` is a row-at-a-time transcription of
     /// `corgi::hash`, and the two backends compute the SAME program value, so they must agree
     /// bit for bit on every shape the transcode layer covers. If corgi's salts or fold change,
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 63e5281b2..6620570ed 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -14,9 +14,9 @@
 //!
 //! Transcode-free: the real keys/values never leave corgi columns. Ids are resolved to rows by
 //! integer index (`key_index`/`val_index` → offsets into the concatenated `key_blocks`/`val_blocks`
-//! pools), not by carrying `DValue`s. Min/Collect use corgi's one-pass segmented structural sort.
-//! DDIR integers are signed, so Min builds an order-only columnar view with each integer leaf's
-//! sign bit swizzled before sorting; the winning row is still gathered from the original columns.
+//! pools), not by carrying `DValue`s. Min/Collect use a segmented structural sort over an
+//! order-only columnar view: signed integer leaves are swizzled, and lists become lexicographic
+//! ranks. The winning rows are still gathered from the original columns.
 //!
 //! The changed-key restriction is honored by presenting only the changed keys: novel batches are
 //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes
@@ -44,9 +44,9 @@ use crate::parse::Reducer;
 
 type CBatch<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
 
-/// Build a sortable view whose integer leaves have signed `i64` order.
+/// Build a sortable view matching DDIR's signed leaves and lexicographic lists.
 ///
-/// DDIR's only scalar is `Int`, transcoded into a Corgi primitive as its raw
+/// DDIR's leaf scalar is `Int`, transcoded into a Corgi primitive as its raw
 /// bits. Corgi's radix sort is unsigned, so XORing each payload leaf's sign bit
 /// turns signed order into unsigned order. Sum discriminants remain untouched;
 /// only their payload lanes recurse. This consumes freshly gathered candidate
@@ -61,13 +61,57 @@ fn signed_order_view(value: CValue) -> CValue {
             // the lane assignment is untouched — only the payload lanes are swizzled.
             CValue::Sum(tags, variants.into_iter().map(signed_order_view).collect())
         }
-        CValue::List(bounds, values) => {
-            CValue::List(bounds, Box::new(signed_order_view(*values)))
-        }
+        CValue::List(bounds, values) => lexicographic_list_ranks(bounds, signed_order_view(*values)),
         CValue::Unit(len) => CValue::Unit(len),
     }
 }
 
+/// An order-only integer rank for each list, using DDIR's lexicographic order.
+/// Corgi's general structural order is intentionally length-first. Preserve
+/// that contract and adapt here: rank the element columns, then refine tied
+/// list prefixes in column batches, with end-of-list preceding every element.
+/// No DDIR rows or per-comparison interpreter calls are materialized.
+///
+/// This is a correctness adapter, not a performance-neutral view: it eagerly
+/// ranks all elements, including unused tails, and each prefix round scans all
+/// lists and allocates fresh scratch even when most prefixes are resolved.
+/// Cost therefore grows with total element count and unresolved prefix depth.
+/// Only list-valued subcolumns pay this ranking cost (including strings encoded
+/// as lists); physical arrangement-key ordering is unchanged.
+fn lexicographic_list_ranks(bounds: Bounds, ordered_elements: CValue) -> CValue {
+    let ends = bounds.to_vec();
+    let rows = ends.len();
+    let (element_perm, element_labels) = sort_blocks(&vec![0; ordered_elements.len()], &ordered_elements);
+    let mut element_ranks = vec![0; element_perm.len()];
+    for (i, &row) in element_perm.iter().enumerate() { element_ranks[row] = element_labels[i]; }
+    let starts: Vec<_> = std::iter::once(0).chain(ends.iter().copied()).take(rows).collect();
+    let mut perm: Vec<_> = (0..rows).collect();
+    let mut labels = vec![0; rows];
+    let mut position = 0;
+    loop {
+        let mut present = vec![0; rows];
+        let mut keys = vec![0; rows];
+        let mut active = false;
+        for i in 0..rows {
+            let tied = (i > 0 && labels[i] == labels[i-1]) || (i+1 < rows && labels[i] == labels[i+1]);
+            let row = perm[i];
+            if tied && position < ends[row]-starts[row] {
+                present[i] = 1;
+                keys[i] = element_ranks[starts[row]+position];
+                active = true;
+            }
+        }
+        if !active { break; }
+        let (order, refined) = sort_blocks(&labels, &CValue::Prod(vec![CValue::u64(present), CValue::u64(keys)]));
+        perm = order.into_iter().map(|i| perm[i]).collect();
+        labels = refined;
+        position += 1;
+    }
+    let mut ranks = vec![0; rows];
+    for (i, &row) in perm.iter().enumerate() { ranks[row] = labels[i]; }
+    CValue::u64(ranks)
+}
+
 /// An identity `Hasher` for the id-index maps: their keys are already well-distributed 64-bit
 /// content hashes (`hash_rows`), so passing the id straight through avoids re-hashing it (siphash
 /// on `register_keys`/lookups was ~7% of the reduce in profiling). Only `write_u64` is used.
@@ -491,7 +535,7 @@ where
                 self.register_vals(col, &out_ids);
             }
             Reducer::Collect => {
-                // One row per bracket: the values sorted in corgi structural order,
+                // One row per bracket: the values sorted in DDIR observable order,
                 // each repeated by its diff, as a `List`. One `sort_blocks` orders every bracket's
                 // entries at once; element rows are then taken columnar. A bracket emits iff some
                 // value has NON-ZERO net (as Distinct/Min: DD invokes the reducer only for a key
@@ -520,7 +564,7 @@ where
                 let perm = if entry_reps.is_empty() {
                     Vec::new()
                 } else {
-                    sort_blocks(&labels, &gather(&self.in_vals, &entry_reps)).0
+                    sort_blocks(&labels, &signed_order_view(gather(&self.in_vals, &entry_reps))).0
                 };
                 // Expand each bracket's sorted entries by their diff (max(0, ·) copies).
                 let mut elem_reps: Vec<usize> = Vec::new();
@@ -708,6 +752,27 @@ where
 mod tests {
     use super::*;
 
+    #[test]
+    fn order_view_matches_ddir_for_ragged_lists_and_signed_products() {
+        use crate::ir::Value as V;
+        use crate::corgi::logic::{transcode, shape_of_row};
+        let mut rows = Vec::new();
+        for n in [-3, 0, 7] {
+            for bytes in [vec![], vec![0], vec![1], vec![1, -1], vec![1, 0], vec![2], vec![2, -2, 0]] {
+                rows.push(V::Tuple(vec![V::Int(n), V::List(bytes.into_iter().map(V::Int).collect())]));
+            }
+        }
+        rows.reverse();
+        // Infer from a non-empty representative; the first row need not carry
+        // data for every nested-list position once the schema is declared.
+        let shape = shape_of_row(&rows[0]).unwrap();
+        let columns = transcode(&rows, &shape);
+        let (perm, _) = sort_blocks(&vec![0; rows.len()], &signed_order_view(columns));
+        let actual: Vec<_> = perm.into_iter().map(|i| rows[i].clone()).collect();
+        rows.sort();
+        assert_eq!(actual, rows);
+    }
+
     fn compound_keys(hashes: Vec<u64>, real: Vec<u64>) -> CValue {
         CValue::Prod(vec![CValue::u64(hashes), CValue::Prod(vec![CValue::u64(real), CValue::u64(vec![0, 0])])])
     }
diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs
index 1c1bba3ac..283a37831 100644
--- a/interactive/src/explain/mod.rs
+++ b/interactive/src/explain/mod.rs
@@ -119,6 +119,7 @@ fn clone_rec(orig: &Scope, out: &mut Scope, import_map: &[Ref], path: &[usize])
                 // side refs mapped into the output parent.
                 let cloned_imports: Vec<Import> = child.imports.iter().map(|imp| Import {
                     name: imp.name.clone(),
+                    shape: imp.shape.clone(),
                     from: match &imp.from {
                         Source::Parent(r) => Source::Parent(map_ref(r, &locals, &subs, import_map, var_base)),
                         other => panic!("clone: nested scope with external source {:?}", other),
@@ -399,7 +400,7 @@ impl Sb {
     }
     fn import(&mut self, name: String, from: Source) -> Ref {
         let k = self.s.imports.len();
-        self.s.imports.push(Import { name, from });
+        self.s.imports.push(Import { name, from, shape: None });
         Ref::Import(k)
     }
     fn export(&mut self, name: String, value: Ref) -> usize {
diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs
index 78c82aaeb..d26e75c12 100644
--- a/interactive/src/ir.rs
+++ b/interactive/src/ir.rs
@@ -26,6 +26,33 @@ pub enum Value {
 }
 
 impl Value {
+    /// Whether a boundary value satisfies an explicitly ascribed column shape.
+    pub fn has_shape(&self, shape: &corgi::Shape) -> bool {
+        use corgi::Shape;
+        match (self, shape) {
+            (Self::Int(_), Shape::Prim(64)) => true,
+            (Self::Tuple(xs), Shape::Unit) => xs.is_empty(),
+            (Self::Tuple(xs), Shape::Prod(fs)) => xs.len() == fs.len() && xs.iter().zip(fs).all(|(x, f)| x.has_shape(f)),
+            (Self::List(xs), Shape::List(f)) => xs.iter().all(|x| x.has_shape(f)),
+            (Self::Variant(tag, value), Shape::Sum(fs)) => fs.get(*tag as usize).is_some_and(|f| value.has_shape(f)),
+            _ => false,
+        }
+    }
+    /// F64 is an explicit one-variant newtype, not an implicit second meaning
+    /// for Int arithmetic. Its payload is the signed-order form of Corgi's
+    /// total-order float encoding, so existing structural hash/Ord and the
+    /// columnar SUM representation apply without row/column type erasure.
+    /// Like other DDIR sum types, the nominal type name is erased at runtime.
+    pub fn f64_value(value: f64) -> Self {
+        let bits = value.to_bits();
+        let ordered = if bits >> 63 == 1 { !bits } else { bits ^ (1 << 63) };
+        Self::Variant(0, Box::new(Self::Int((ordered ^ (1 << 63)) as i64)))
+    }
+    pub fn as_f64(&self) -> f64 {
+        let Self::Variant(0, payload) = self else { panic!("expected F64 newtype, got {self:?}") };
+        let ordered = payload.as_int() as u64 ^ (1 << 63);
+        f64::from_bits(if ordered >> 63 == 1 { ordered ^ (1 << 63) } else { !ordered })
+    }
     /// The empty tuple — the conventional "unit"/empty value.
     pub fn unit() -> Value { Value::Tuple(Vec::new()) }
     /// Truthiness: a nonzero `Int` is true; everything else is false.
@@ -221,6 +248,8 @@ fn build_seq(fields: &[Term], env: &mut Vec<Value>) -> Vec<Value> {
 fn eval_unary(op: UnOp, v: Value) -> Value {
     match op {
         UnOp::Neg => Value::Int(-v.as_int()),
+        UnOp::ToF64 => Value::f64_value(v.as_int() as f64),
+        UnOp::F64Neg => Value::f64_value(-v.as_f64()),
         UnOp::Not => Value::Int((!v.truthy()) as i64),
         UnOp::IsTag(t) => Value::Int(matches!(&v, Value::Variant(tag, _) if *tag == t) as i64),
         UnOp::Len => match v {
@@ -236,6 +265,15 @@ fn eval_binary(op: BinOp, l: Value, r: Value) -> Value {
         BinOp::Add => Value::Int(l.as_int() + r.as_int()),
         BinOp::Sub => Value::Int(l.as_int() - r.as_int()),
         BinOp::Mul => Value::Int(l.as_int() * r.as_int()),
+        BinOp::Div => Value::Int(if r.as_int() == 0 { 0 } else { l.as_int().wrapping_div(r.as_int()) }),
+        BinOp::Append => match (l, r) {
+            (Value::List(mut a), Value::List(b)) => { a.extend(b); Value::List(a) }
+            other => panic!("append expects two lists, got {other:?}"),
+        },
+        BinOp::F64Add => Value::f64_value(l.as_f64() + r.as_f64()),
+        BinOp::F64Sub => Value::f64_value(l.as_f64() - r.as_f64()),
+        BinOp::F64Mul => Value::f64_value(l.as_f64() * r.as_f64()),
+        BinOp::F64Div => Value::f64_value(l.as_f64() / r.as_f64()),
         // Comparisons are structural, using the derived `Ord`/`Eq` on `Value`.
         BinOp::Eq => b(l == r),
         BinOp::Ne => b(l != r),
diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs
index ff4818c2d..93641d522 100644
--- a/interactive/src/lower.rs
+++ b/interactive/src/lower.rs
@@ -47,7 +47,7 @@ fn expr_free_names<'a>(expr: &'a Expr, out: &mut BTreeSet<&'a str>) {
         Expr::Input(_) | Expr::Import(_) => {},
         Expr::Name(n) => { out.insert(n.as_str()); },
         Expr::Qualified(scope, _) => { out.insert(scope.as_str()); },
-        Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _)
+        Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _)
             | Expr::Negate(e) | Expr::EnterAt(e, _) | Expr::LiftIter(e)
             | Expr::FlatMap(e, _)
             | Expr::Inspect(e, _) | Expr::Arrange(e) => expr_free_names(e, out),
@@ -115,7 +115,7 @@ impl ScopeLower {
         assert!(self.is_root, "`input {}` used outside the root scope (not yet supported)", n);
         if let Some(&i) = self.input_import.get(&n) { return st::Ref::Import(i); }
         let i = self.imports.len();
-        self.imports.push(st::Import { name: format!("input{}", n), from: st::Source::Input(n) });
+        self.imports.push(st::Import { name: format!("input{}", n), from: st::Source::Input(n), shape: None });
         self.input_import.insert(n, i);
         st::Ref::Import(i)
     }
@@ -123,7 +123,7 @@ impl ScopeLower {
         assert!(self.is_root, "`import {:?}` used outside the root scope (not yet supported)", name);
         if let Some(&i) = self.trace_import.get(name) { return st::Ref::Import(i); }
         let i = self.imports.len();
-        self.imports.push(st::Import { name: name.to_string(), from: st::Source::Trace(name.to_string()) });
+        self.imports.push(st::Import { name: name.to_string(), from: st::Source::Trace(name.to_string()), shape: None });
         self.trace_import.insert(name.to_string(), i);
         st::Ref::Import(i)
     }
@@ -137,6 +137,17 @@ impl ScopeLower {
         match e {
             Expr::Input(n) => self.input_ref(*n),
             Expr::Import(name) => self.trace_ref(name),
+            Expr::TypedSource(source, key, val) => {
+                let r = self.lower_expr(source);
+                let st::Ref::Import(i) = r else { panic!("a shape ascription requires an external source") };
+                assert!(self.is_root, "external shape ascriptions belong at the root");
+                let shape = (key.clone(), val.clone());
+                if let Some(previous) = &self.imports[i].shape {
+                    assert_eq!(previous, &shape, "conflicting source shape ascriptions");
+                }
+                self.imports[i].shape = Some(shape);
+                r
+            }
             Expr::Name(name) => self.env.get(name).cloned()
                 .unwrap_or_else(|| panic!("unresolved name `{}`", name)),
             Expr::Qualified(s, f) => self.qualified_ref(s, f),
@@ -194,7 +205,7 @@ fn lower_scope_tree(
             let from = parent_env.get(name).cloned()
                 .unwrap_or_else(|| panic!("scope references unknown outer name `{}`", name));
             let i = s.imports.len();
-            s.imports.push(st::Import { name: name.to_string(), from: st::Source::Parent(from) });
+            s.imports.push(st::Import { name: name.to_string(), from: st::Source::Parent(from), shape: None });
             s.env.insert(name.to_string(), st::Ref::Import(i));
         }
     }
@@ -300,7 +311,7 @@ fn qualified_fields(body: &[Stmt], scope: &str) -> Vec<String> {
 fn collect_qualified(e: &Expr, scope: &str, out: &mut Vec<String>) {
     match e {
         Expr::Qualified(s, f) => if s == scope && !out.contains(f) { out.push(f.clone()); },
-        Expr::Map(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::EnterAt(e, _)
+        Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::EnterAt(e, _)
         | Expr::FlatMap(e, _)
         | Expr::LiftIter(e) | Expr::Reduce(e, _) | Expr::Inspect(e, _) | Expr::Arrange(e) => collect_qualified(e, scope, out),
         Expr::Join(l, r, _) => { collect_qualified(l, scope, out); collect_qualified(r, scope, out); },
diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs
index 4fb5846a5..7e07401f6 100644
--- a/interactive/src/parse/mod.rs
+++ b/interactive/src/parse/mod.rs
@@ -92,11 +92,20 @@ pub enum UnOp {
     IsTag(u32),
     /// Number of elements in a `Tuple` or `List`, as an `Int`.
     Len,
+    /// Explicit signed Int -> F64 newtype conversion (see Value::f64_value).
+    ToF64,
+    /// Floating-point negation; does not reinterpret integer arithmetic.
+    F64Neg,
 }
 
 #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
 pub enum BinOp {
     Add, Sub, Mul,
+    /// Truncating signed division; zero divisor returns zero, MIN / -1 wraps.
+    Div,
+    /// Concatenation of two lists with the same element type.
+    Append,
+    F64Add, F64Sub, F64Mul, F64Div,
     Eq, Ne, Lt, Le, Gt, Ge,
     And, Or,
 }
@@ -119,6 +128,9 @@ pub enum Reducer {
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum Expr {
     Input(usize),
+    /// Shape ascription on an external source: `input N : (key_shape ; val_shape)`.
+    /// The same syntax follows `import "name"` for independently installed consumers.
+    TypedSource(Box<Expr>, corgi::Shape, corgi::Shape),
     /// Named external trace resolved at install time. Carries only the name;
     /// shape comes from the registry the program is installed against.
     Import(String),
@@ -175,7 +187,21 @@ pub(crate) fn build_builtin(name: &str, args: &mut Vec<Term>) -> Term {
         "len" => { assert_eq!(args.len(), 1, "len(value)"); Term::Unary(UnOp::Len, Box::new(args.remove(0))) }
         "istag" => { assert_eq!(args.len(), 2, "istag(tag, value)"); let tag = int_arg(&args[0]) as u32; Term::Unary(UnOp::IsTag(tag), Box::new(args.remove(1))) }
         "not" => { assert_eq!(args.len(), 1, "not(value)"); Term::Unary(UnOp::Not, Box::new(args.remove(0))) }
+        "float" | "fneg" => {
+            assert_eq!(args.len(), 1, "{name}(value)");
+            Term::Unary(if name == "float" { UnOp::ToF64 } else { UnOp::F64Neg }, Box::new(args.remove(0)))
+        }
+        "fadd" | "fsub" | "fmul" | "fdiv" => {
+            assert_eq!(args.len(), 2, "{name}(a, b)");
+            let b = Box::new(args.remove(1)); let a = Box::new(args.remove(0));
+            Term::Binary(match name { "fadd" => BinOp::F64Add, "fsub" => BinOp::F64Sub, "fmul" => BinOp::F64Mul, _ => BinOp::F64Div }, a, b)
+        }
         "or" => { assert_eq!(args.len(), 2, "or(a, b)"); let b = Box::new(args.remove(1)); let a = Box::new(args.remove(0)); Term::Binary(BinOp::Or, a, b) }
+        "idiv" | "append" => {
+            assert_eq!(args.len(), 2, "{name}(a, b)");
+            let b = Box::new(args.remove(1)); let a = Box::new(args.remove(0));
+            Term::Binary(if name == "idiv" { BinOp::Div } else { BinOp::Append }, a, b)
+        }
         "if" => { assert_eq!(args.len(), 3, "if(cond, then, els)"); let els = Box::new(args.remove(2)); let then = Box::new(args.remove(1)); let cond = Box::new(args.remove(0)); Term::If { cond, then, els } }
         "hash" => { assert!(args.len() >= 2, "hash(bound, key, ...)"); Term::Hash(std::mem::take(args)) }
         other => panic!("Unknown scalar builtin: {}", other),
diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs
index adb4bf301..77b6ec0e1 100644
--- a/interactive/src/parse/pipe.rs
+++ b/interactive/src/parse/pipe.rs
@@ -9,9 +9,18 @@
 //!
 //! Sources: `input N` (positional input), `import "name"` (named trace),
 //! `name` (a `let`/`var` in scope), `scope::field` (a child scope's export).
+//! External sources may append `: (key_shape ; val_shape)`, using the shape
+//! syntax below. This supplies the column encoding even when the first row
+//! contains empty lists or inactive sum lanes. Ascriptions are asserted during
+//! execution, not checked by the server's feed-admission acknowledgement. On
+//! either backend a mismatched row can panic a dataflow worker and take down
+//! the shared server, disconnecting other clients; there is no per-program
+//! failure isolation. Use these contracts only with trusted, shape-correct data.
 //! Operators chain with `|`:
 //!
 //! - `| key(k… ; v…)` — reshape to `(key ; val)`; `map` is an alias.
+//! - `| value(term)` — replace the value without wrapping it in a tuple;
+//!   useful for collecting a list of scalars or lists instead of singleton tuples.
 //! - `| join(other, (k… ; v…))` — equijoin on the key.
 //! - `| min` / `| distinct` / `| count` / `| collect` — reduce; `collect` is
 //!   NEST (gather a key's values into a `List`).
@@ -36,8 +45,22 @@
 //!   fields; `$n[i]` selects field `i`; chains as `$n[i][j]`.
 //! - Arithmetic / compare / logic: `+ - *`, `== != < <= > >=`, `&&`,
 //!   `or(a, b)`, `not(x)`, unary `-x`.
+//! - Integer division: `idiv(a, b)` truncates toward zero, returns zero for a
+//!   zero divisor, and wraps `i64::MIN / -1` to `i64::MIN`.
+//! - Explicit floating point: `float(int)`, `fneg(x)`, and
+//!   `fadd(a, b)` / `fsub(a, b)` / `fmul(a, b)` / `fdiv(a, b)` use IEEE f64.
+//!   Values are a one-variant SUM carrying an order-encoded integer payload,
+//!   not ordinary integers or an implicit numeric coercion. Generic ordering
+//!   is IEEE total order (including distinct signed zeros and NaN payloads).
+//!   Nominal type names are erased: `fneg` and binary floating operators also
+//!   accept a user's single-variant integer newtype, treating its payload as
+//!   encoded f64 bits.
 //! - Products: `tuple(a, …)`; index with `v[i]` or `proj(v, i)`; `len(v)`.
-//! - Lists: `list(a, …)`; eliminated by `flatmap` / `collect` / `fold`.
+//! - Lists: `list(a, …)`, `append(a, b)` (concatenation); eliminated by
+//!   `flatmap` / `collect` / `fold`. A declared constructor supplies the element
+//!   shape for an otherwise ambiguous empty `list()`.
+//!   Corgi's shape inference does not propagate between `append` arguments:
+//!   `append(list(), xs)` is rejected even when `xs` has a known element shape.
 //! - Sums: every sum is a declared type, `type Size = Small u64 | Big (u64, u64)
 //!   | Empty;` — tags are positions, scoped to the type; a payload shape is
 //!   `u64`/`int`, `()` (the default when omitted), `(a, b, …)`, `List(a)`,
@@ -372,8 +395,8 @@ impl Parser {
 
     fn parse_atom(&mut self) -> Expr {
         match self.peek().clone() {
-            Token::Input => { self.next(); match self.next() { Token::Int(n) => Expr::Input(n as usize), o => panic!("Expected int, got {:?}", o) } },
-            Token::Import => { self.next(); match self.next() { Token::Str(s) => Expr::Import(s), o => panic!("Expected string literal after `import`, got {:?}", o) } },
+            Token::Input => { self.next(); let source = match self.next() { Token::Int(n) => Expr::Input(n as usize), o => panic!("Expected int, got {:?}", o) }; self.source_shape(source) },
+            Token::Import => { self.next(); let source = match self.next() { Token::Str(s) => Expr::Import(s), o => panic!("Expected string literal after `import`, got {:?}", o) }; self.source_shape(source) },
             Token::Ident(_) => { let n = self.parse_ident(); if *self.peek() == Token::ColonColon { self.next(); let f = self.parse_ident(); Expr::Qualified(n, f) } else { Expr::Name(n) } },
             Token::LParen => { self.next(); let e = self.parse_pipe_expr(); self.expect(&Token::RParen); e },
             other => panic!("Unexpected token in atom: {:?}", other),
@@ -386,10 +409,26 @@ impl Parser {
         expr
     }
 
+    fn source_shape(&mut self, source: Expr) -> Expr {
+        if *self.peek() != Token::Colon { return source; }
+        self.next();
+        self.expect(&Token::LParen);
+        let key = self.parse_shape();
+        self.expect(&Token::Semi);
+        let val = self.parse_shape();
+        self.expect(&Token::RParen);
+        Expr::TypedSource(Box::new(source), key, val)
+    }
+
     fn parse_pipe_op(&mut self, lhs: Expr) -> Expr {
         match self.peek().clone() {
             Token::Key => { self.next(); let p = self.parse_projection(); Expr::Map(Box::new(lhs), p) },
             Token::Map => { self.next(); let p = self.parse_projection(); Expr::Map(Box::new(lhs), p) },
+            Token::Ident(name) if name == "value" => {
+                self.next(); self.expect(&Token::LParen);
+                let val = self.parse_term(); self.expect(&Token::RParen);
+                Expr::Map(Box::new(lhs), Projection { key: Term::Var(0), val })
+            },
             Token::Join => { self.next(); self.expect(&Token::LParen); let r = self.parse_join_arg(); self.expect(&Token::Comma); let p = self.parse_projection(); self.expect(&Token::RParen); Expr::Join(Box::new(lhs), Box::new(r), p) },
             Token::Min => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Min) },
             Token::Distinct => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Distinct) },
diff --git a/interactive/src/scope_ir.rs b/interactive/src/scope_ir.rs
index 291986637..3f1c40179 100644
--- a/interactive/src/scope_ir.rs
+++ b/interactive/src/scope_ir.rs
@@ -76,6 +76,10 @@ pub enum Source {
 pub struct Import {
     pub name: String,
     pub from: Source,
+    /// Optional external encoding contract; unlike derived intermediate shapes,
+    /// this can describe empty lists and sum lanes absent from the input rows.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub shape: Option<(corgi::Shape, corgi::Shape)>,
 }
 
 /// A value surrendered up — `leave_region` (+ `leave_dynamic` if iterating).
@@ -414,8 +418,8 @@ mod tests {
         let inner = Scope {
             name: "reach".into(),
             imports: vec![
-                Import { name: "edges".into(), from: Source::Parent(Ref::Import(0)) },
-                Import { name: "roots".into(), from: Source::Parent(Ref::Import(1)) },
+                Import { name: "edges".into(), from: Source::Parent(Ref::Import(0)), shape: None },
+                Import { name: "roots".into(), from: Source::Parent(Ref::Import(1)), shape: None },
             ],
             vars: vec![Var { name: "reach".into() }],
             items: vec![
@@ -430,8 +434,8 @@ mod tests {
         let root = Scope {
             name: "root".into(),
             imports: vec![
-                Import { name: "in0".into(), from: Source::Input(0) },
-                Import { name: "in1".into(), from: Source::Input(1) },
+                Import { name: "in0".into(), from: Source::Input(0), shape: None },
+                Import { name: "in1".into(), from: Source::Input(1), shape: None },
             ],
             items: vec![Item::Sub(inner)], // item 0 = the reach sub-scope
             exports: vec![Export { name: "result".into(), value: Ref::ChildExport(0, 0) }],
diff --git a/interactive/src/server.rs b/interactive/src/server.rs
index 73784927e..dfd736ee7 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -10,9 +10,12 @@
 //! The server executes a [`Command`] — already parsed, lowered, and validated.
 //! Programs are parsed *off the worker threads* (on the intake side) and shipped
 //! here as `scope_ir::Program`s; a malformed program is rejected before it ever
-//! reaches a worker, so bad input can't panic the computation. [`Command`] is
-//! serializable so worker 0 can broadcast one ordered command stream to the
-//! whole worker group.
+//! reaches a worker. This does not make execution safe for arbitrary data:
+//! violating an input/import shape ascription panics inside a dataflow operator
+//! on either backend and can take down the shared server, disconnecting other
+//! clients. Feed acknowledgements do not check these contracts, and there is
+//! no per-program failure isolation. [`Command`] is serializable so worker 0
+//! can broadcast one ordered command stream to the whole worker group.
 //!
 //! # The two binding points
 //!