Add MergeOrdered IR node with proper sim interleaving hooks

Addresses #2768

Replace the sliced!-based merge_ordered implementation with a dedicated
HydroNode::MergeOrdered IR node that has distinct codegen for production
and simulation.

Production codegen:
- Emits union() DFIR operator (both inputs pulled in same stratum)

Simulation codegen (two modes):
- Bounded/tick: MergeOrderedHook (inline) buffers both batches and
  generates a valid interleaving via boolean coin flips, preserving
  per-input order. Explores C(a+b, a) states instead of (a+b)!.
- Unbounded/top-level: TopLevelMergeOrderedHook releases one element
  at a time from the front of either input queue, allowing feedback
  cycles to deliver elements between releases.

API changes:
- Removed NoTick bound from merge_ordered — works on any Location and
  Boundedness now. When bounded, the nondet is still explored in sim.
- State space reduced from 26 to 6 instances in the existing test
  (2 elements per input: C(4,2) = 6 valid interleavings).

Tests added:
- sim_merge_ordered: ordering preservation assertion (invalid
  interleavings like [2,1,3,4] are never produced)
- sim_merge_ordered_one_empty: pass-through when one input is empty
- sim_merge_ordered_cycle_back: feedback cycle where a cycled-back
  element arrives before elements on the other input
- sim_merge_ordered_delayed: delayed element on one input
- deploy_merge_ordered_delayed: production localhost test with timing

Author: Benjscho

hydro_lang/src/compile/ir/mod.rs

@@ -439,6 +439,16 @@ pub trait DfirBuilder {
         op_meta: &HydroIrOpMetadata,
     );
 
+    fn merge_ordered(
+        &mut self,
+        location: &LocationId,
+        first_ident: syn::Ident,
+        second_ident: syn::Ident,
+        out_ident: &syn::Ident,
+        in_kind: &CollectionKind,
+        op_meta: &HydroIrOpMetadata,
+    );
+
     #[expect(clippy::too_many_arguments, reason = "TODO")]
     fn create_network(
         &mut self,
@@ -596,6 +606,27 @@ impl DfirBuilder for SecondaryMap<LocationKey, FlatGraphBuilder> {
         );
     }
 
+    fn merge_ordered(
+        &mut self,
+        location: &LocationId,
+        first_ident: syn::Ident,
+        second_ident: syn::Ident,
+        out_ident: &syn::Ident,
+        _in_kind: &CollectionKind,
+        _op_meta: &HydroIrOpMetadata,
+    ) {
+        let builder = self.get_dfir_mut(location);
+        builder.add_dfir(
+            parse_quote! {
+                #out_ident = union();
+                #first_ident -> [0]#out_ident;
+                #second_ident -> [1]#out_ident;
+            },
+            None,
+            None,
+        );
+    }
+
     fn create_network(
         &mut self,
         from: &LocationId,
@@ -2036,6 +2067,12 @@ pub enum HydroNode {
         metadata: HydroIrMetadata,
     },
 
+    MergeOrdered {
+        first: Box<HydroNode>,
+        second: Box<HydroNode>,
+        metadata: HydroIrMetadata,
+    },
+
     ChainFirst {
         first: Box<HydroNode>,
         second: Box<HydroNode>,
@@ -2312,6 +2349,11 @@ impl HydroNode {
                 transform(second.as_mut(), seen_tees);
             }
 
+            HydroNode::MergeOrdered { first, second, .. } => {
+                transform(first.as_mut(), seen_tees);
+                transform(second.as_mut(), seen_tees);
+            }
+
             HydroNode::ChainFirst { first, second, .. } => {
                 transform(first.as_mut(), seen_tees);
                 transform(second.as_mut(), seen_tees);
@@ -2464,6 +2506,15 @@ impl HydroNode {
                 second: Box::new(second.deep_clone(seen_tees)),
                 metadata: metadata.clone(),
             },
+            HydroNode::MergeOrdered {
+                first,
+                second,
+                metadata,
+            } => HydroNode::MergeOrdered {
+                first: Box::new(first.deep_clone(seen_tees)),
+                second: Box::new(second.deep_clone(seen_tees)),
+                metadata: metadata.clone(),
+            },
             HydroNode::ChainFirst {
                 first,
                 second,
@@ -3180,6 +3231,34 @@ impl HydroNode {
                         ident_stack.push(chain_ident);
                     }
 
+                    HydroNode::MergeOrdered { first, metadata, .. } => {
+                        let second_ident = ident_stack.pop().unwrap();
+                        let first_ident = ident_stack.pop().unwrap();
+
+                        let merge_ident =
+                            syn::Ident::new(&format!("stream_{}", *next_stmt_id), Span::call_site());
+
+                        match builders_or_callback {
+                            BuildersOrCallback::Builders(graph_builders) => {
+                                graph_builders.merge_ordered(
+                                    &first.metadata().location_id,
+                                    first_ident,
+                                    second_ident,
+                                    &merge_ident,
+                                    &first.metadata().collection_kind,
+                                    &metadata.op,
+                                );
+                            }
+                            BuildersOrCallback::Callback(_, node_callback) => {
+                                node_callback(node, next_stmt_id);
+                            }
+                        }
+
+                        *next_stmt_id += 1;
+
+                        ident_stack.push(merge_ident);
+                    }
+
                     HydroNode::ChainFirst { .. } => {
                         let second_ident = ident_stack.pop().unwrap();
                         let first_ident = ident_stack.pop().unwrap();
@@ -4247,6 +4326,7 @@ impl HydroNode {
             | HydroNode::EndAtomic { .. }
             | HydroNode::Batch { .. }
             | HydroNode::Chain { .. }
+            | HydroNode::MergeOrdered { .. }
             | HydroNode::ChainFirst { .. }
             | HydroNode::CrossProduct { .. }
             | HydroNode::CrossSingleton { .. }
@@ -4324,6 +4404,7 @@ impl HydroNode {
             HydroNode::EndAtomic { metadata, .. } => metadata,
             HydroNode::Batch { metadata, .. } => metadata,
             HydroNode::Chain { metadata, .. } => metadata,
+            HydroNode::MergeOrdered { metadata, .. } => metadata,
             HydroNode::ChainFirst { metadata, .. } => metadata,
             HydroNode::CrossProduct { metadata, .. } => metadata,
             HydroNode::CrossSingleton { metadata, .. } => metadata,
@@ -4378,6 +4459,7 @@ impl HydroNode {
             HydroNode::EndAtomic { metadata, .. } => metadata,
             HydroNode::Batch { metadata, .. } => metadata,
             HydroNode::Chain { metadata, .. } => metadata,
+            HydroNode::MergeOrdered { metadata, .. } => metadata,
             HydroNode::ChainFirst { metadata, .. } => metadata,
             HydroNode::CrossProduct { metadata, .. } => metadata,
             HydroNode::CrossSingleton { metadata, .. } => metadata,
@@ -4436,6 +4518,9 @@ impl HydroNode {
             HydroNode::Chain { first, second, .. } => {
                 vec![first, second]
             }
+            HydroNode::MergeOrdered { first, second, .. } => {
+                vec![first, second]
+            }
             HydroNode::ChainFirst { first, second, .. } => {
                 vec![first, second]
             }
@@ -4526,6 +4611,13 @@ impl HydroNode {
             HydroNode::Chain { first, second, .. } => {
                 format!("Chain({}, {})", first.print_root(), second.print_root())
             }
+            HydroNode::MergeOrdered { first, second, .. } => {
+                format!(
+                    "MergeOrdered({}, {})",
+                    first.print_root(),
+                    second.print_root()
+                )
+            }
             HydroNode::ChainFirst { first, second, .. } => {
                 format!(
                     "ChainFirst({}, {})",