Record the passive-edge mask, so multi-edge ops can be emitted

crates/wingfoil-derive/src/lib.rs

@@ -2078,6 +2078,9 @@ fn expand_builder(args: &OpArgs, b: &BuilderShape<'_>) -> syn::Result<TokenStrea
     // takes no config from one whose closure the engine erased: both leave
     // `NodeInfo::src` empty. Without it an emitter cannot say which nodes are
     // ineligible, and silently prints a call with a missing argument.
+    // The op's passive-edge mask, recorded so an emitter can rebuild the
+    // original call order from the partitioned active/passive lists.
+    let passive_bits = args.passive;
     let takes_closure_cfg = {
         let cfg_name = quote! { #cfg_ty }.to_string();
         let is_param = b.type_params.iter().any(|p| *p == cfg_name);
@@ -2349,7 +2352,7 @@ fn expand_builder(args: &OpArgs, b: &BuilderShape<'_>) -> syn::Result<TokenStrea
                 // recorded as the label. An emitter walking the wired graph
                 // needs `map`, not `Map` — plus whether its config is a closure,
                 // which is what distinguishes "no config" from "erased closure".
-                self.set_node_build(#build_name, #takes_closure_cfg);
+                self.set_node_build(#build_name, #takes_closure_cfg, #passive_bits);
                 self.set_reset(::std::boxed::Box::new(move || {
                     __cs_reset.borrow_mut().1 = #state_reseed;
                     *__out_reset.borrow_mut() = #out_reseed;

crates/wingfoil/src/interp.rs

@@ -463,6 +463,17 @@ struct NodeRt {
     /// `map` whose closure the engine erased: both leave `src` empty. An
     /// emitter needs the difference to know which nodes it must refuse.
     takes_closure_cfg: bool,
+    /// The op's `#[op(passive = [..])]` mask: bit `i` set means edge `i` of the
+    /// original call is passive.
+    ///
+    /// Needed because [`active_ups`](Self::active_ups) and
+    /// [`passive_ups`](Self::passive_ups) are *partitioned* lists — each keeps
+    /// its own order, but the interleaving is gone. `sample`'s data leg is
+    /// edge 0 and passive, its trigger is edge 1 and active, so the two lists
+    /// alone cannot say whether the call was `data.sample(&trigger)` or the
+    /// reverse. With the mask an emitter walks positions `0..n` and takes from
+    /// whichever list that bit selects.
+    passive_mask: u32,
 }
 
 /// A type-free description of one wired node — what survives the interpreted
@@ -506,6 +517,42 @@ pub struct NodeInfo {
     /// not quote" — the one an emitter must refuse on. Neither field alone says
     /// it, because a config-free op also reports `src: None`.
     pub takes_closure_cfg: bool,
+    /// The op's passive-edge mask: bit `i` set means edge `i` of the original
+    /// call is passive. Use it with [`edges_in_call_order`](Self::edges_in_call_order),
+    /// which is what it exists for.
+    pub passive_mask: u32,
+}
+
+impl NodeInfo {
+    /// The node's upstream edges **in the order the original call listed
+    /// them** — receiver first, then each argument.
+    ///
+    /// [`active_ups`](Self::active_ups) and [`passive_ups`](Self::passive_ups)
+    /// are partitioned lists: each preserves its own order, but the
+    /// interleaving between them is not recoverable from the pair alone.
+    /// `sample`'s data leg is edge 0 and passive while its trigger is edge 1
+    /// and active, so `active_ups = [trigger]`, `passive_ups = [data]` — and
+    /// nothing there says the call was `data.sample(&trigger)` rather than the
+    /// reverse. [`passive_mask`](Self::passive_mask) supplies exactly that
+    /// missing bit, and this walks it.
+    pub fn edges_in_call_order(&self) -> Vec<usize> {
+        let total = self.active_ups.len() + self.passive_ups.len();
+        let (mut active, mut passive) = (self.active_ups.iter(), self.passive_ups.iter());
+        (0..total)
+            .map(|i| {
+                let from_passive = i < 32 && (self.passive_mask >> i) & 1 == 1;
+                let next = if from_passive {
+                    passive.next()
+                } else {
+                    active.next()
+                };
+                *next.expect(
+                    "invariant: passive_mask agrees with the active/passive \
+                     partition it was recorded alongside",
+                )
+            })
+            .collect()
+    }
 }
 
 /// The producer half of an [`external`](Builder::external) source: send a
@@ -1275,20 +1322,27 @@ impl Builder {
             build: None,
             cfg_src: None,
             takes_closure_cfg: false,
+            passive_mask: 0,
         });
         self.ticked.borrow_mut().push(false);
     }
 
     /// Record the op's `#[op(build = …)]` method name against the node most
     /// recently pushed. Called by the generated `Builder` method right after
     /// `push_node`, in the same style as `set_reset`.
-    pub(crate) fn set_node_build(&mut self, build: &'static str, takes_closure_cfg: bool) {
+    pub(crate) fn set_node_build(
+        &mut self,
+        build: &'static str,
+        takes_closure_cfg: bool,
+        passive_mask: u32,
+    ) {
         let node = self
             .nodes
             .last_mut()
             .expect("invariant: set_node_build called immediately after push_node");
         node.build = Some(build);
         node.takes_closure_cfg = takes_closure_cfg;
+        node.passive_mask = passive_mask;
     }
 
     /// Record the source text of a node's closure config — the quotation half
@@ -1350,6 +1404,7 @@ impl Builder {
                 build: n.build,
                 cfg_src: n.cfg_src.clone(),
                 takes_closure_cfg: n.takes_closure_cfg,
+                passive_mask: n.passive_mask,
             })
             .collect()
     }
@@ -3382,6 +3437,7 @@ impl Runner {
                 build: n.build,
                 cfg_src: n.cfg_src.clone(),
                 takes_closure_cfg: n.takes_closure_cfg,
+                passive_mask: n.passive_mask,
             })
             .collect()
     }
@@ -3675,6 +3731,7 @@ impl Runner {
             build: None,
             cfg_src: None,
             takes_closure_cfg: false,
+            passive_mask: 0,
         });
         self.ticked.borrow_mut().push(false);
         self.active_downs.push(Vec::new());

crates/wingfoil/tests/emission_spike.rs

@@ -34,11 +34,17 @@
 //! emitter must refuse. Neither field alone says it, because a config-free op
 //! like `count` also reports `src: None`.
 //!
-//! What is left: emission covers **single-edge chains only**. Multi-edge
-//! (`join`) and passive-edge (`sample`) ops are refused rather than emitted,
-//! and the refusals name node indices rather than the *call sites*
-//! `#[track_caller]` would give. Both are pinned by tests at the bottom so
-//! neither can regress into a silent partial emission.
+//! **Multi-edge and passive-edge ops emit.** `join` reconstructs as
+//! `a.join(&b, f)`, and `sample` — whose data leg is edge 0 and *passive* —
+//! comes back in the order it was written, which the partitioned
+//! active/passive lists genuinely could not express on their own. The recorded
+//! `passive_mask` supplies the missing bit.
+//!
+//! What is left: **variadic** ops (`merge_all`, `combine`) have hand-written
+//! forwarders and so no `build` name, and are refused. Refusals also name node
+//! indices rather than the *call sites* `#[track_caller]` would give. Both are
+//! pinned by tests at the bottom so neither can regress into a silent partial
+//! emission.
 
 use std::fmt::Write as _;
 use std::time::Duration;
@@ -85,20 +91,6 @@ fn emit(nodes: &[NodeInfo], fn_name: &str, out_ty: &str) -> Result<String, Vec<I
             });
             continue;
         };
-        if !n.passive_ups.is_empty() {
-            bad.push(Ineligible {
-                index: n.index,
-                label: n.label,
-                reason: format!("`{build}` has passive edges; emission order is unproven"),
-            });
-        }
-        if n.active_ups.len() > 1 {
-            bad.push(Ineligible {
-                index: n.index,
-                label: n.label,
-                reason: format!("`{build}` is multi-edge; receiver/argument split unproven"),
-            });
-        }
         // The precise statement of "erased closure": the op takes one, and the
         // wiring did not quote it. Neither field alone says this — a
         // config-free op like `count` also reports `src: None`.
@@ -137,12 +129,27 @@ fn emit(nodes: &[NodeInfo], fn_name: &str, out_ty: &str) -> Result<String, Vec<I
             (None, Some(src)) => src.to_string(),
             (None, None) => String::new(),
         };
-        match n.active_ups.first() {
+        // Edges in the order the original call listed them: receiver first,
+        // then each `&stream` argument, then the config. Reconstructed from
+        // the passive mask — the partitioned active/passive lists alone cannot
+        // say which position each edge occupied.
+        let edges = n.edges_in_call_order();
+        let args: String = match edges.split_first() {
+            None => arg,
+            Some((_, rest)) => {
+                let mut parts: Vec<String> = rest.iter().map(|u| format!("&n{u}")).collect();
+                if !arg.is_empty() {
+                    parts.push(arg);
+                }
+                parts.join(", ")
+            }
+        };
+        match edges.first() {
             None => {
-                let _ = writeln!(s, "        let n{} = g.{build}({arg});", n.index);
+                let _ = writeln!(s, "        let n{} = g.{build}({args});", n.index);
             }
-            Some(&up) => {
-                let _ = writeln!(s, "        let n{} = n{up}.{build}({arg});", n.index);
+            Some(recv) => {
+                let _ = writeln!(s, "        let n{} = n{recv}.{build}({args});", n.index);
             }
         }
     }
@@ -167,6 +174,16 @@ wingfoil::nitro! {
     }
 }
 
+wingfoil::nitro! {
+    fn joined_target(g: &GraphBuilder) -> Stream<u64> {
+        let n0 = g.ticker(::core::time::Duration::new(0u64, 1000000u32));
+        let n1 = n0.count();
+        let n2 = n1.map(|i: &u64| i * 10);
+        let n3 = n1.join(&n2, |x: &u64, y: &u64| x + y);
+        n3
+    }
+}
+
 /// The same graph, wired procedurally — what a generator's pass 1 consumes.
 fn wire_source_graph() -> (GraphBuilder, Stream<u64>) {
     let g = GraphBuilder::new();
@@ -314,28 +331,95 @@ fn config_free_ops_are_not_mistaken_for_erased_closures() {
     emit(&nodes, "target", "u64").expect("a fully quoted graph still emits");
 }
 
-/// **Gap 3: multi-edge and passive-edge ops are refused.** `active_ups`
-/// preserves receiver-first order, so `join` is *probably* recoverable as
-/// `a.join(&b, f)` — but nothing distinguishes a receiver from an argument, and
-/// `sample`'s passive leg is absent from `active_ups` entirely. Both need
-/// proving before the walker can claim general coverage; until then it fails
-/// loudly, which is the right default.
+/// **Closed (was gap 3): multi-edge ops emit.** `active_ups` keeps
+/// receiver-first order, so a `join` reconstructs as `a.join(&b, f)` — the
+/// receiver, then each `&stream` argument, then the config.
 #[test]
-fn multi_edge_ops_are_refused_with_reasons() {
+fn a_multi_edge_graph_emits_and_matches() {
     let g = GraphBuilder::new();
     let a = g.ticker(PERIOD).with_cfg(&PERIOD).count();
     let scale = func!(|i: &u64| i * 10);
     let b = a.map(scale.f).with_src(&scale);
     let combine = func!(|x: &u64, y: &u64| x + y);
-    let _joined = a.join(&b, combine.f).with_src(&combine);
-
-    // Everything else in this graph is quoted or config-free, so the join is
-    // the *only* refusal — which is what makes this a test of the multi-edge
-    // rule rather than of eligibility in general.
-    let err = emit(&g.describe(), "joined", "u64").expect_err("join must be refused");
-    assert_eq!(1, err.len(), "only the join is ineligible: {err:?}");
-    assert_eq!("Join", err[0].label);
-    assert!(err[0].reason.contains("multi-edge"), "{:?}", err[0]);
+    let joined = a.join(&b, combine.f).with_src(&combine);
+
+    let emitted = emit(&g.describe(), "joined_target", "u64").expect("join must emit");
+    let expected = "\
+wingfoil::nitro! {
+    fn joined_target(g: &GraphBuilder) -> Stream<u64> {
+        let n0 = g.ticker(::core::time::Duration::new(0u64, 1000000u32));
+        let n1 = n0.count();
+        let n2 = n1.map(|i: &u64| i * 10);
+        let n3 = n1.join(&n2, |x: &u64, y: &u64| x + y);
+        n3
+    }
+}";
+    assert_eq!(expected, emitted);
+
+    // And it computes the same thing, values and tick times.
+    let acc = joined.with_time().accumulate();
+    let mut runner = g.build();
+    runner.run(HISTORICAL, RUN).unwrap();
+    let source_graph = runner.value(acc);
+
+    let g2 = GraphBuilder::new();
+    let acc2 = joined_target::nested(&g2).with_time().accumulate();
+    let mut runner2 = g2.build();
+    runner2.run(HISTORICAL, RUN).unwrap();
+
+    assert_eq!(source_graph, runner2.value(acc2));
+}
+
+/// **The passive case, which the partitioned lists genuinely could not
+/// express.** `sample`'s data leg is edge 0 and *passive*; its trigger is edge
+/// 1 and active. So `active_ups = [ticker]` and `passive_ups = [count]`, and
+/// nothing in that pair says the call was `count.sample(&ticker)` rather than
+/// the reverse — both orderings produce the identical pair. The recorded
+/// `passive_mask` is what supplies the missing bit.
+#[test]
+fn passive_edges_reconstruct_the_original_call_order() {
+    let g = GraphBuilder::new();
+    let tick = g.ticker(PERIOD).with_cfg(&PERIOD);
+    let count = tick.count();
+    let _sampled = count.sample(&tick);
+
+    let nodes = g.describe();
+    let sample = &nodes[2];
+    assert_eq!(1, sample.passive_mask, "sample is `passive = [0]`");
+    assert_eq!(vec![0usize], sample.active_ups, "the trigger");
+    assert_eq!(vec![1usize], sample.passive_ups, "the data leg");
+    assert_eq!(
+        vec![1usize, 0],
+        sample.edges_in_call_order(),
+        "data first, trigger second — the order `count.sample(&tick)` was written in"
+    );
+
+    let emitted = emit(&nodes, "sampled", "u64").expect("sample must emit");
+    assert!(
+        emitted.contains("let n2 = n1.sample(&n0);"),
+        "receiver and argument the wrong way round:\n{emitted}"
+    );
+}
+
+/// A variadic op (`merge_all` -> `MergeN`) has hand-written forwarders and so
+/// no `#[op(build = ..)]` name. It is refused rather than emitted as an
+/// n-ary call it would not accept — the remaining loud failure, and the right
+/// default for a shape the walker cannot express.
+#[test]
+fn variadic_ops_are_still_refused() {
+    let g = GraphBuilder::new();
+    let a = g.ticker(PERIOD).with_cfg(&PERIOD).count();
+    let scale = func!(|i: &u64| i * 10);
+    let b = a.map(scale.f).with_src(&scale);
+    let other = func!(|i: &u64| i * 100);
+    let c = a.map(other.f).with_src(&other);
+    let _merged = a.merge_all(&[&b, &c]);
+
+    let err = emit(&g.describe(), "merged", "u64").expect_err("merge_all must be refused");
+    assert!(
+        err.iter().any(|e| e.reason.contains("hand-written")),
+        "{err:?}"
+    );
 }
 
 /// The unrolling property, which is the whole point: pass 1 *ran the loop*, so