feat(hydro_lang): add m2e/e2m external inputs for clusters in the simulator

hydro_lang/src/compile/deploy_provider.rs

@@ -167,6 +167,49 @@ pub trait Deploy<'a> {
         shared_handle: String,
     ) -> syn::Expr;
 
+    fn e2m_source(
+        extra_stmts: &mut Vec<syn::Stmt>,
+        p1: &Self::External,
+        p1_port: &<Self::External as Node>::Port,
+        c2: &Self::Cluster,
+        c2_port: &<Self::Cluster as Node>::Port,
+        codec_type: &syn::Type,
+        shared_handle: String,
+    ) -> syn::Expr {
+        let _ = (
+            extra_stmts,
+            p1,
+            p1_port,
+            c2,
+            c2_port,
+            codec_type,
+            shared_handle,
+        );
+        todo!("e2m_source is not yet supported for this deploy backend")
+    }
+
+    fn e2m_connect(
+        p1: &Self::External,
+        p1_port: &<Self::External as Node>::Port,
+        c2: &Self::Cluster,
+        c2_port: &<Self::Cluster as Node>::Port,
+        server_hint: NetworkHint,
+    ) -> Box<dyn FnOnce()> {
+        let _ = (p1, p1_port, c2, c2_port, server_hint);
+        todo!("e2m_connect is not yet supported for this deploy backend")
+    }
+
+    fn m2e_sink(
+        c1: &Self::Cluster,
+        c1_port: &<Self::Cluster as Node>::Port,
+        p2: &Self::External,
+        p2_port: &<Self::External as Node>::Port,
+        shared_handle: String,
+    ) -> syn::Expr {
+        let _ = (c1, c1_port, p2, p2_port, shared_handle);
+        todo!("m2e_sink is not yet supported for this deploy backend")
+    }
+
     fn cluster_ids(
         of_cluster: LocationKey,
     ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a;

hydro_lang/src/compile/ir/mod.rs

@@ -815,7 +815,35 @@ impl HydroRoot {
                                         )
                                     }
                                 }
-                                LocationId::Cluster(_) => todo!("SendExternal from a cluster location is not yet supported"),
+                                LocationId::Cluster(cluster_key) => {
+                                    let from_node = clusters
+                                        .get(*cluster_key)
+                                        .unwrap_or_else(|| {
+                                            panic!("A cluster used in the graph was not instantiated: {}", cluster_key)
+                                        })
+                                        .clone();
+
+                                    let sink_port = from_node.next_port();
+                                    let source_port = to_node.next_port();
+
+                                    if *unpaired {
+                                        to_node.register(*to_port_id, source_port.clone());
+                                    }
+
+                                    (
+                                        (
+                                            D::m2e_sink(
+                                                &from_node,
+                                                &sink_port,
+                                                &to_node,
+                                                &source_port,
+                                                format!("{}_{}", *to_external_key, *to_port_id)
+                                            ),
+                                            parse_quote!(DUMMY),
+                                        ),
+                                        Box::new(|| {}) as Box<dyn FnOnce()>,
+                                    )
+                                }
                                 _ => panic!()
                             }
                         },
@@ -936,7 +964,33 @@ impl HydroRoot {
                                         D::e2o_connect(&from_node, &sink_port, &to_node, &source_port, *from_many, *port_hint),
                                     )
                                 }
-                                LocationId::Cluster(_) => todo!("ExternalInput to a cluster location is not yet supported"),
+                                LocationId::Cluster(cluster_key) => {
+                                    let to_node = clusters
+                                        .get(*cluster_key)
+                                        .unwrap_or_else(|| {
+                                            panic!("A cluster used in the graph was not instantiated: {}", cluster_key)
+                                        })
+                                        .clone();
+
+                                    let sink_port = from_node.next_port();
+                                    let source_port = to_node.next_port();
+
+                                    from_node.register(*from_port_id, sink_port.clone());
+
+                                    (
+                                        (
+                                            parse_quote!(DUMMY),
+                                            D::e2m_source(
+                                                refcell_extra_stmts.borrow_mut().entry(*cluster_key).expect("location was removed").or_default(),
+                                                &from_node, &sink_port,
+                                                &to_node, &source_port,
+                                                codec_type.0.as_ref(),
+                                                format!("{}_{}", *from_external_key, *from_port_id)
+                                            ),
+                                        ),
+                                        D::e2m_connect(&from_node, &sink_port, &to_node, &source_port, *port_hint),
+                                    )
+                                }
                                 _ => panic!()
                             }
                         },

hydro_lang/src/live_collections/stream/networking.rs

@@ -1092,6 +1092,54 @@ impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<T, Cluster<'a, L>
         }
         .demux(to, via)
     }
+
+    #[cfg(feature = "sim")]
+    /// Sends elements of this cluster stream to an external location using bincode serialization.
+    fn send_bincode_external<L2>(self, other: &External<L2>) -> ExternalBincodeStream<T, O, R>
+    where
+        T: Serialize + DeserializeOwned,
+    {
+        let serialize_pipeline = Some(serialize_bincode::<T>(false));
+
+        let mut flow_state_borrow = self.location.flow_state().borrow_mut();
+
+        let external_port_id = flow_state_borrow.next_external_port();
+
+        flow_state_borrow.push_root(HydroRoot::SendExternal {
+            to_external_key: other.key,
+            to_port_id: external_port_id,
+            to_many: false,
+            unpaired: true,
+            serialize_fn: serialize_pipeline.map(|e| e.into()),
+            instantiate_fn: DebugInstantiate::Building,
+            input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
+            op_metadata: HydroIrOpMetadata::new(),
+        });
+
+        ExternalBincodeStream {
+            process_key: other.key,
+            port_id: external_port_id,
+            _phantom: PhantomData,
+        }
+    }
+
+    #[cfg(feature = "sim")]
+    /// Sets up a simulation output port for this cluster stream, allowing test code
+    /// to receive `(member_id, T)` pairs during simulation.
+    pub fn sim_cluster_output(self) -> crate::sim::SimClusterReceiver<T, O, R>
+    where
+        T: Serialize + DeserializeOwned,
+    {
+        let external_location: External<'a, ()> = External {
+            key: LocationKey::FIRST,
+            flow_state: self.location.flow_state().clone(),
+            _phantom: PhantomData,
+        };
+
+        let external = self.send_bincode_external(&external_location);
+
+        crate::sim::SimClusterReceiver(external.port_id, PhantomData)
+    }
 }
 
 impl<'a, T, L, L2, B: Boundedness, O: Ordering, R: Retries>

hydro_lang/src/location/cluster.rs

@@ -88,6 +88,49 @@ impl<'a, C> Location<'a> for Cluster<'a, C> {
     }
 }
 
+#[cfg(feature = "sim")]
+impl<'a, C> Cluster<'a, C> {
+    /// Sets up a simulated input port on this cluster for testing.
+    ///
+    /// Returns a `SimClusterSender` that sends `(member_id, T)` messages targeting
+    /// specific cluster members, and a `Stream<T>` received by each member.
+    #[expect(clippy::type_complexity, reason = "stream markers")]
+    pub fn sim_input<T>(
+        &self,
+    ) -> (
+        crate::sim::SimClusterSender<
+            T,
+            crate::live_collections::stream::TotalOrder,
+            crate::live_collections::stream::ExactlyOnce,
+        >,
+        crate::live_collections::Stream<
+            T,
+            Self,
+            crate::live_collections::boundedness::Unbounded,
+            crate::live_collections::stream::TotalOrder,
+            crate::live_collections::stream::ExactlyOnce,
+        >,
+    )
+    where
+        T: serde::Serialize + serde::de::DeserializeOwned,
+    {
+        use crate::location::Location;
+
+        let external_location: crate::location::External<'a, ()> = crate::location::External {
+            key: LocationKey::FIRST,
+            flow_state: self.flow_state.clone(),
+            _phantom: PhantomData,
+        };
+
+        let (external, stream) = self.source_external_bincode(&external_location);
+
+        (
+            crate::sim::SimClusterSender(external.port_id, PhantomData),
+            stream,
+        )
+    }
+}
+
 /// A free variable that resolves to the list of member IDs in a cluster at runtime.
 ///
 /// When spliced into a quoted snippet, this provides access to the set of

hydro_lang/src/sim/builder.rs

@@ -1124,9 +1124,12 @@ impl DfirBuilder for SimBuilder {
         tag_id: usize,
     ) {
         let grabbed_ident = syn::Ident::new(&format!("__sink_{tag_id}"), Span::call_site());
-        self.extra_stmts_global.push(syn::parse_quote! {
-            let #grabbed_ident = #sink_expr;
-        });
+        self.add_extra_stmt_internal(
+            on,
+            syn::parse_quote! {
+                let #grabbed_ident = #sink_expr;
+            },
+        );
 
         if let Some(serialize_pipeline) = serialize {
             self.get_dfir_mut(on).add_dfir(