feat(hydro_lang): add Stream::drop_consistency() and FlowBuilder::cluster_with_consistency()

- Stream::drop_consistency(): safely forgets a consistency guarantee
  (e.g. EventualConsistency → NoConsistency). Purely type-level, no IR
  node inserted. Useful when replication strategies need to return streams
  on NoConsistency clusters after using broadcast_closed internally.

- FlowBuilder::cluster_with_consistency<C, Con>(): creates a cluster with
  an explicit consistency guarantee (e.g. EventualConsistency) rather than
  the default NoConsistency.

Author: Joe Hellerstein

hydro_lang/src/compile/builder.rs

@@ -228,6 +228,19 @@ impl<'a> FlowBuilder<'a> {
         }
     }
 
+    /// Create a cluster with a specific consistency guarantee.
+    pub fn cluster_with_consistency<C, Con: crate::location::cluster::Consistency>(
+        &mut self,
+    ) -> Cluster<'a, C, Con> {
+        let key = self.locations.insert(LocationType::Cluster);
+        self.location_names.insert(key, type_name::<C>().to_owned());
+        Cluster {
+            key,
+            flow_state: self.flow_state().clone(),
+            _phantom: PhantomData,
+        }
+    }
+
     pub fn external<E>(&mut self) -> External<'a, E> {
         let key = self.locations.insert(LocationType::External);
         self.location_names.insert(key, type_name::<E>().to_owned());

hydro_lang/src/live_collections/stream/mod.rs

@@ -521,6 +521,21 @@ where
         }
     }
 
+    /// Drop the consistency guarantee from this stream's location.
+    ///
+    /// Weakens the consistency guarantee of this stream (e.g., going from
+    /// `EventualConsistency` to `NoConsistency`). No IR node is inserted —
+    /// enforcing the direction is safe. No IR node is inserted.
+    pub fn weaken_consistency(self) -> Stream<T, L::DropConsistency, B, O, R>
+    where
+        L: Location<'a>,
+    {
+        Stream::new(
+            self.location.drop_consistency(),
+            self.ir_node.replace(HydroNode::Placeholder),
+        )
+    }
+
     pub(crate) fn collection_kind() -> CollectionKind {
         CollectionKind::Stream {
             bound: B::BOUND_KIND,