hydro_lang/live_collections/stream/

mod.rs

//! Definitions for the [`Stream`] live collection.

use std::cell::RefCell;
use std::future::Future;
use std::hash::Hash;
use std::marker::PhantomData;
use std::ops::Deref;
use std::rc::Rc;

use stageleft::{IntoQuotedMut, QuotedWithContext, QuotedWithContextWithProps, q, quote_type};
#[cfg(feature = "tokio")]
use tokio::time::Instant;

use super::OperatorContext;
use super::boundedness::{Bounded, Boundedness, IsBounded, Unbounded};
use super::keyed_singleton::KeyedSingleton;
use super::keyed_stream::{Generate, KeyedStream};
use super::optional::Optional;
use super::singleton::Singleton;
use crate::compile::builder::{CycleId, FlowState};
use crate::compile::ir::{
    CollectionKind, HydroIrOpMetadata, HydroNode, HydroRoot, SharedNode, StreamOrder, StreamRetry,
};
#[cfg(stageleft_runtime)]
use crate::forward_handle::{CycleCollection, CycleCollectionWithInitial, ReceiverComplete};
use crate::forward_handle::{ForwardRef, TickCycle};
use crate::live_collections::batch_atomic::BatchAtomic;
use crate::live_collections::singleton::SingletonBound;
#[cfg(stageleft_runtime)]
use crate::location::dynamic::{DynLocation, LocationId};
use crate::location::tick::{Atomic, DeferTick};
use crate::location::{Location, Tick, TopLevel, check_matching_location};
use crate::manual_expr::ManualExpr;
use crate::nondet::{NonDet, nondet};
use crate::prelude::manual_proof;
use crate::properties::{
    AggFuncAlgebra, ApplyMonotoneStream, NotProved, StreamMapFuncAlgebra, ValidCommutativityFor,
    ValidIdempotenceFor, ValidMutBorrowCommutativityFor, ValidMutBorrowIdempotenceFor,
    ValidMutCommutativityFor, ValidMutIdempotenceFor,
};

pub mod networking;

/// A trait implemented by valid ordering markers ([`TotalOrder`] and [`NoOrder`]).
#[sealed::sealed]
pub trait Ordering:
    MinOrder<Self, Min = Self> + MinOrder<TotalOrder, Min = Self> + MinOrder<NoOrder, Min = NoOrder>
{
    /// The [`StreamOrder`] corresponding to this type.
    const ORDERING_KIND: StreamOrder;
}

/// Marks the stream as being totally ordered, which means that there are
/// no sources of non-determinism (other than intentional ones) that will
/// affect the order of elements.
pub enum TotalOrder {}

#[sealed::sealed]
impl Ordering for TotalOrder {
    const ORDERING_KIND: StreamOrder = StreamOrder::TotalOrder;
}

/// Marks the stream as having no order, which means that the order of
/// elements may be affected by non-determinism.
///
/// This restricts certain operators, such as `fold` and `reduce`, to only
/// be used with commutative aggregation functions.
pub enum NoOrder {}

#[sealed::sealed]
impl Ordering for NoOrder {
    const ORDERING_KIND: StreamOrder = StreamOrder::NoOrder;
}

/// Marker trait for an [`Ordering`] that is available when `Self` is a weaker guarantee than
/// `Other`, which means that a stream with `Other` guarantees can be safely converted to
/// have `Self` guarantees instead.
#[sealed::sealed]
pub trait WeakerOrderingThan<Other: ?Sized>: Ordering {}
#[sealed::sealed]
impl<O, O2: Ordering> WeakerOrderingThan<O2> for O where O: Ordering + MinOrder<O2, Min = O> {}

/// Helper trait for determining the weakest of two orderings.
#[sealed::sealed]
pub trait MinOrder<Other: ?Sized> {
    /// The weaker of the two orderings.
    type Min: Ordering;
}

#[sealed::sealed]
impl<O: Ordering> MinOrder<O> for TotalOrder {
    type Min = O;
}

#[sealed::sealed]
impl<O: Ordering> MinOrder<O> for NoOrder {
    type Min = NoOrder;
}

/// A trait implemented by valid retries markers ([`ExactlyOnce`] and [`AtLeastOnce`]).
#[sealed::sealed]
pub trait Retries:
    MinRetries<Self, Min = Self>
    + MinRetries<ExactlyOnce, Min = Self>
    + MinRetries<AtLeastOnce, Min = AtLeastOnce>
{
    /// The [`StreamRetry`] corresponding to this type.
    const RETRIES_KIND: StreamRetry;
}

/// Marks the stream as having deterministic message cardinality, with no
/// possibility of duplicates.
pub enum ExactlyOnce {}

#[sealed::sealed]
impl Retries for ExactlyOnce {
    const RETRIES_KIND: StreamRetry = StreamRetry::ExactlyOnce;
}

/// Marks the stream as having non-deterministic message cardinality, which
/// means that duplicates may occur, but messages will not be dropped.
pub enum AtLeastOnce {}

#[sealed::sealed]
impl Retries for AtLeastOnce {
    const RETRIES_KIND: StreamRetry = StreamRetry::AtLeastOnce;
}

/// Marker trait for a [`Retries`] that is available when `Self` is a weaker guarantee than
/// `Other`, which means that a stream with `Other` guarantees can be safely converted to
/// have `Self` guarantees instead.
#[sealed::sealed]
pub trait WeakerRetryThan<Other: ?Sized>: Retries {}
#[sealed::sealed]
impl<R, R2: Retries> WeakerRetryThan<R2> for R where R: Retries + MinRetries<R2, Min = R> {}

/// Helper trait for determining the weakest of two retry guarantees.
#[sealed::sealed]
pub trait MinRetries<Other: ?Sized> {
    /// The weaker of the two retry guarantees.
    type Min: Retries + WeakerRetryThan<Self> + WeakerRetryThan<Other>;
}

#[sealed::sealed]
impl<R: Retries> MinRetries<R> for ExactlyOnce {
    type Min = R;
}

#[sealed::sealed]
impl<R: Retries> MinRetries<R> for AtLeastOnce {
    type Min = AtLeastOnce;
}

#[sealed::sealed]
#[diagnostic::on_unimplemented(
    message = "the input stream must be totally ordered (`TotalOrder`), but has order `{Self}`",
    label = "required here",
    note = "strengthen the order upstream, or consider a different API",
    note = "to intentionally process the stream by observing a non-deterministic (shuffled) order of elements, use `.assume_ordering`; this introduces non-determinism, so avoid it unless necessary"
)]
/// Marker trait that is implemented for the [`TotalOrder`] ordering guarantee.
pub trait IsOrdered: Ordering {}

#[sealed::sealed]
#[diagnostic::do_not_recommend]
impl IsOrdered for TotalOrder {}

#[sealed::sealed]
#[diagnostic::on_unimplemented(
    message = "the input stream must be exactly-once (`ExactlyOnce`), but has retries `{Self}`",
    label = "required here",
    note = "strengthen the retries guarantee upstream, or consider a different API",
    note = "to intentionally process the stream by observing non-deterministic (randomly duplicated) retries, use `.assume_retries`; this introduces non-determinism, so avoid it unless necessary"
)]
/// Marker trait that is implemented for the [`ExactlyOnce`] retries guarantee.
pub trait IsExactlyOnce: Retries {}

#[sealed::sealed]
#[diagnostic::do_not_recommend]
impl IsExactlyOnce for ExactlyOnce {}

/// Streaming sequence of elements with type `Type`.
///
/// This live collection represents a growing sequence of elements, with new elements being
/// asynchronously appended to the end of the sequence. This can be used to model the arrival
/// of network input, such as API requests, or streaming ingestion.
///
/// By default, all streams have deterministic ordering and each element is materialized exactly
/// once. But streams can also capture non-determinism via the `Order` and `Retries` type
/// parameters. When the ordering / retries guarantee is relaxed, fewer APIs will be available
/// on the stream. For example, if the stream is unordered, you cannot invoke [`Stream::first`].
///
/// Type Parameters:
/// - `Type`: the type of elements in the stream
/// - `Loc`: the location where the stream is being materialized
/// - `Bound`: the boundedness of the stream, which is either [`Bounded`] or [`Unbounded`]
/// - `Order`: the ordering of the stream, which is either [`TotalOrder`] or [`NoOrder`]
///   (default is [`TotalOrder`])
/// - `Retries`: the retry guarantee of the stream, which is either [`ExactlyOnce`] or
///   [`AtLeastOnce`] (default is [`ExactlyOnce`])
pub struct Stream<
    Type,
    Loc,
    Bound: Boundedness = Unbounded,
    Order: Ordering = TotalOrder,
    Retry: Retries = ExactlyOnce,
> {
    pub(crate) location: Loc,
    pub(crate) ir_node: Rc<RefCell<HydroNode>>,
    pub(crate) flow_state: FlowState,

    _phantom: PhantomData<(Type, Loc, Bound, Order, Retry)>,
}

impl<T, L, B: Boundedness, O: Ordering, R: Retries> Drop for Stream<T, L, B, O, R> {
    fn drop(&mut self) {
        let ir_node = self.ir_node.replace(HydroNode::Placeholder);
        if !matches!(ir_node, HydroNode::Placeholder) && !ir_node.is_shared_with_others() {
            self.flow_state.borrow_mut().try_push_root(HydroRoot::Null {
                input: Box::new(ir_node),
                op_metadata: HydroIrOpMetadata::new(),
            });
        }
    }
}

impl<'a, T, L, O: Ordering, R: Retries> From<Stream<T, L, Bounded, O, R>>
    for Stream<T, L, Unbounded, O, R>
where
    L: Location<'a>,
{
    fn from(stream: Stream<T, L, Bounded, O, R>) -> Stream<T, L, Unbounded, O, R> {
        let new_meta = stream
            .location
            .new_node_metadata(Stream::<T, L, Unbounded, O, R>::collection_kind());

        let flow_state = stream.flow_state.clone();
        Stream {
            location: stream.location.clone(),
            ir_node: super::tracked_ir_node(
                &flow_state,
                HydroNode::Cast {
                    inner: Box::new(stream.ir_node.replace(HydroNode::Placeholder)),
                    metadata: new_meta,
                },
            ),
            flow_state,
            _phantom: PhantomData,
        }
    }
}

impl<'a, T, L, B: Boundedness, R: Retries> From<Stream<T, L, B, TotalOrder, R>>
    for Stream<T, L, B, NoOrder, R>
where
    L: Location<'a>,
{
    fn from(stream: Stream<T, L, B, TotalOrder, R>) -> Stream<T, L, B, NoOrder, R> {
        stream.weaken_ordering()
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering> From<Stream<T, L, B, O, ExactlyOnce>>
    for Stream<T, L, B, O, AtLeastOnce>
where
    L: Location<'a>,
{
    fn from(stream: Stream<T, L, B, O, ExactlyOnce>) -> Stream<T, L, B, O, AtLeastOnce> {
        stream.weaken_retries()
    }
}

impl<'a, T, L, O: Ordering, R: Retries> DeferTick for Stream<T, Tick<L>, Bounded, O, R>
where
    L: Location<'a>,
{
    fn defer_tick(self) -> Self {
        Stream::defer_tick(self)
    }
}

impl<'a, T, L, O: Ordering, R: Retries> CycleCollection<'a, TickCycle>
    for Stream<T, Tick<L>, Bounded, O, R>
where
    L: Location<'a>,
{
    type Location = Tick<L>;

    fn create_source(cycle_id: CycleId, location: Tick<L>) -> Self {
        Stream::new(
            location.clone(),
            HydroNode::CycleSource {
                cycle_id,
                metadata: location.new_node_metadata(Self::collection_kind()),
            },
        )
    }
}

impl<'a, T, L, O: Ordering, R: Retries> CycleCollectionWithInitial<'a, TickCycle>
    for Stream<T, Tick<L>, Bounded, O, R>
where
    L: Location<'a>,
{
    type Location = Tick<L>;

    fn location(&self) -> &Self::Location {
        self.location()
    }

    fn create_source_with_initial(cycle_id: CycleId, initial: Self, location: Tick<L>) -> Self {
        let from_previous_tick: Stream<T, Tick<L>, Bounded, O, R> = Stream::new(
            location.clone(),
            HydroNode::DeferTick {
                input: Box::new(HydroNode::CycleSource {
                    cycle_id,
                    metadata: location.new_node_metadata(Self::collection_kind()),
                }),
                metadata: location.new_node_metadata(Self::collection_kind()),
            },
        );

        from_previous_tick.chain(initial.filter_if(location.optional_first_tick(q!(())).is_some()))
    }
}

impl<'a, T, L, O: Ordering, R: Retries> ReceiverComplete<'a, TickCycle>
    for Stream<T, Tick<L>, Bounded, O, R>
where
    L: Location<'a>,
{
    fn complete(self, cycle_id: CycleId, expected_location: LocationId) {
        assert_eq!(
            Location::id(&self.location),
            expected_location,
            "locations do not match"
        );
        self.location
            .flow_state()
            .borrow_mut()
            .push_root(HydroRoot::CycleSink {
                cycle_id,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                op_metadata: HydroIrOpMetadata::new(),
            });
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> CycleCollection<'a, ForwardRef>
    for Stream<T, L, B, O, R>
where
    L: Location<'a>,
{
    type Location = L;

    fn create_source(cycle_id: CycleId, location: L) -> Self {
        Stream::new(
            location.clone(),
            HydroNode::CycleSource {
                cycle_id,
                metadata: location.new_node_metadata(Self::collection_kind()),
            },
        )
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> ReceiverComplete<'a, ForwardRef>
    for Stream<T, L, B, O, R>
where
    L: Location<'a>,
{
    fn complete(self, cycle_id: CycleId, expected_location: LocationId) {
        assert_eq!(
            Location::id(&self.location),
            expected_location,
            "locations do not match"
        );
        self.location
            .flow_state()
            .borrow_mut()
            .push_root(HydroRoot::CycleSink {
                cycle_id,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                op_metadata: HydroIrOpMetadata::new(),
            });
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Clone for Stream<T, L, B, O, R>
where
    T: Clone,
    L: Location<'a>,
{
    fn clone(&self) -> Self {
        if !matches!(self.ir_node.borrow().deref(), HydroNode::Tee { .. }) {
            let orig_ir_node = self.ir_node.replace(HydroNode::Placeholder);
            *self.ir_node.borrow_mut() = HydroNode::Tee {
                inner: SharedNode(Rc::new(RefCell::new(orig_ir_node))),
                metadata: self.location.new_node_metadata(Self::collection_kind()),
            };
        }

        let HydroNode::Tee { inner, metadata } = &*self.ir_node.borrow() else {
            unreachable!()
        };
        Stream {
            location: self.location.clone(),
            flow_state: self.flow_state.clone(),
            ir_node: super::tracked_ir_node(
                &self.flow_state,
                HydroNode::Tee {
                    inner: SharedNode(inner.0.clone()),
                    metadata: metadata.clone(),
                },
            ),
            _phantom: PhantomData,
        }
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<T, L, B, O, R>
where
    L: Location<'a>,
{
    pub(crate) fn new(location: L, ir_node: HydroNode) -> Self {
        debug_assert_eq!(ir_node.metadata().location_id, Location::id(&location));
        debug_assert_eq!(ir_node.metadata().collection_kind, Self::collection_kind());

        let flow_state = location.flow_state().clone();
        let ir_node = super::tracked_ir_node(&flow_state, ir_node);
        Stream {
            location,
            flow_state,
            ir_node,
            _phantom: PhantomData,
        }
    }

    /// Returns the [`Location`] where this stream is being materialized.
    pub fn location(&self) -> &L {
        &self.location
    }

    /// Creates a shared reference handle to this stream's handoff buffer that can be captured
    /// inside `q!()` closures. The handle resolves to `&Vec<T>` at runtime.
    ///
    /// The stream must be bounded, otherwise reading it would be non-deterministic.
    pub fn by_ref(&self) -> crate::handoff_ref::StreamRef<'a, '_, T, L, B>
    where
        B: IsBounded,
    {
        crate::handoff_ref::StreamRef::new(&self.ir_node)
    }

    /// Returns a mutable reference handle to this stream's handoff buffer that can be captured
    /// inside `q!()` closures. The handle resolves to `&mut Vec<T>` at runtime.
    pub fn by_mut(&self) -> crate::handoff_ref::StreamMut<'a, '_, T, L, B>
    where
        B: IsBounded,
    {
        crate::handoff_ref::StreamMut::new(&self.ir_node)
    }

    /// Weakens the consistency of this live collection to not guarantee any consistency across
    /// cluster members (if this collection is on a cluster).
    pub fn weaken_consistency(self) -> Stream<T, L::DropConsistency, B, O, R>
    where
        L: Location<'a>,
    {
        if L::consistency()
            .is_none_or(|c| c == crate::location::dynamic::ClusterConsistency::NoConsistency)
        {
            // already no consistency
            Stream::new(
                self.location.drop_consistency(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else {
            Stream::new(
                self.location.drop_consistency(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: self.location.drop_consistency().new_node_metadata(Stream::<
                        T,
                        L::DropConsistency,
                        B,
                        O,
                        R,
                    >::collection_kind(
                    )),
                },
            )
        }
    }

    /// Casts this live collection to have the consistency guarantees specified in the given
    /// location type parameter. The developer must ensure that the strengthened consistency
    /// is actually guaranteed, via the proof field (see [`crate::prelude::manual_proof`]).
    pub fn assert_has_consistency_of<L2: Location<'a, DropConsistency = L::DropConsistency>>(
        self,
        _proof: impl crate::properties::ConsistencyProof,
    ) -> Stream<T, L2, B, O, R>
    where
        L: Location<'a>,
    {
        if L::consistency() == L2::consistency() {
            Stream::new(
                self.location.with_consistency_of(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else {
            Stream::new(
                self.location.with_consistency_of(),
                HydroNode::AssertIsConsistent {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: false,
                    metadata: self
                        .location
                        .clone()
                        .with_consistency_of::<L2>()
                        .new_node_metadata(Stream::<T, L2, B, O, R>::collection_kind()),
                },
            )
        }
    }

    pub(crate) fn assert_has_consistency_of_trusted<
        L2: Location<'a, DropConsistency = L::DropConsistency>,
    >(
        self,
        _proof: impl crate::properties::ConsistencyProof,
    ) -> Stream<T, L2, B, O, R>
    where
        L: Location<'a>,
    {
        if L::consistency() == L2::consistency() {
            Stream::new(
                self.location.with_consistency_of(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else {
            Stream::new(
                self.location.with_consistency_of(),
                HydroNode::AssertIsConsistent {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: true,
                    metadata: self
                        .location
                        .clone()
                        .with_consistency_of::<L2>()
                        .new_node_metadata(Stream::<T, L2, B, O, R>::collection_kind()),
                },
            )
        }
    }

    pub(crate) fn collection_kind() -> CollectionKind {
        CollectionKind::Stream {
            bound: B::BOUND_KIND,
            order: O::ORDERING_KIND,
            retry: R::RETRIES_KIND,
            element_type: quote_type::<T>().into(),
        }
    }

    /// Produces a stream based on invoking `f` on each element.
    /// If you do not want to modify the stream and instead only want to view
    /// each item use [`Stream::inspect`] instead.
    ///
    /// If the input stream is unordered **and** `f` mutably captures state (such as a
    /// [`Singleton::by_mut`](crate::live_collections::singleton::Singleton::by_mut)
    /// reference), `f` must be proven **commutative**: processing any two elements in
    /// either order must leave the mutably-captured state in the same final value *and*
    /// produce the same multiset of outputs. In particular, outputs must not expose the
    /// processing order (e.g. emitting a running total is not commutative even though
    /// addition is).
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let words = process.source_iter(q!(vec!["hello", "world"]));
    /// words.map(q!(|x| x.to_uppercase()))
    /// # }, |mut stream| async move {
    /// # for w in vec!["HELLO", "WORLD"] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn map<U, F, C, I, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, I, L::SimHookScope>,
        >,
    ) -> Stream<U, L, B, O, R>
    where
        F: FnMut(T) -> U + 'a,
        C: ValidMutCommutativityFor<F, T, U, O, WAS_MUT>,
        I: ValidMutIdempotenceFor<F, T, U, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::Map {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<U, L, B, O, R>::collection_kind()),
            },
        )
    }

    /// For each item `i` in the input stream, transform `i` using `f` and then treat the
    /// result as an [`Iterator`] to produce items one by one. The implementation for [`Iterator`]
    /// for the output type `U` must produce items in a **deterministic** order.
    ///
    /// For example, `U` could be a `Vec`, but not a `HashSet`. If the order of the items in `U` is
    /// not deterministic, use [`Stream::flat_map_unordered`] instead.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec![vec![1, 2], vec![3, 4]]))
    ///     .flat_map_ordered(q!(|x| x))
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4
    /// # for w in (1..5) {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn flat_map_ordered<U, I, F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Stream<U, L, B, O, R>
    where
        I: IntoIterator<Item = U>,
        F: FnMut(T) -> I + 'a,
        C: ValidMutCommutativityFor<F, T, I, O, WAS_MUT>,
        Idemp: ValidMutIdempotenceFor<F, T, I, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::FlatMap {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<U, L, B, O, R>::collection_kind()),
            },
        )
    }

    /// Like [`Stream::flat_map_ordered`], but allows the implementation of [`Iterator`]
    /// for the output type `U` to produce items in any order.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::{prelude::*, live_collections::stream::{NoOrder, ExactlyOnce}};
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test::<_, _, _, NoOrder, ExactlyOnce>(|process| {
    /// process
    ///     .source_iter(q!(vec![
    ///         std::collections::HashSet::<i32>::from_iter(vec![1, 2]),
    ///         std::collections::HashSet::from_iter(vec![3, 4]),
    ///     ]))
    ///     .flat_map_unordered(q!(|x| x))
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4, but in no particular order
    /// # let mut results = Vec::new();
    /// # for w in (1..5) {
    /// #     results.push(stream.next().await.unwrap());
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec![1, 2, 3, 4]);
    /// # }));
    /// # }
    /// ```
    pub fn flat_map_unordered<U, I, F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Stream<U, L, B, NoOrder, R>
    where
        I: IntoIterator<Item = U>,
        F: FnMut(T) -> I + 'a,
        C: ValidMutCommutativityFor<F, T, I, O, WAS_MUT>,
        Idemp: ValidMutIdempotenceFor<F, T, I, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::FlatMap {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<U, L, B, NoOrder, R>::collection_kind()),
            },
        )
    }

    /// For each item `i` in the input stream, treat `i` as an [`Iterator`] and produce its items one by one.
    /// The implementation for [`Iterator`] for the element type `T` must produce items in a **deterministic** order.
    ///
    /// For example, `T` could be a `Vec`, but not a `HashSet`. If the order of the items in `T` is
    /// not deterministic, use [`Stream::flatten_unordered`] instead.
    ///
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec![vec![1, 2], vec![3, 4]]))
    ///     .flatten_ordered()
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4
    /// # for w in (1..5) {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn flatten_ordered<U>(self) -> Stream<U, L, B, O, R>
    where
        T: IntoIterator<Item = U>,
    {
        self.flat_map_ordered(q!(|d| d))
    }

    /// Like [`Stream::flatten_ordered`], but allows the implementation of [`Iterator`]
    /// for the element type `T` to produce items in any order.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::{prelude::*, live_collections::stream::{NoOrder, ExactlyOnce}};
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test::<_, _, _, NoOrder, ExactlyOnce>(|process| {
    /// process
    ///     .source_iter(q!(vec![
    ///         std::collections::HashSet::<i32>::from_iter(vec![1, 2]),
    ///         std::collections::HashSet::from_iter(vec![3, 4]),
    ///     ]))
    ///     .flatten_unordered()
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4, but in no particular order
    /// # let mut results = Vec::new();
    /// # for w in (1..5) {
    /// #     results.push(stream.next().await.unwrap());
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec![1, 2, 3, 4]);
    /// # }));
    /// # }
    /// ```
    pub fn flatten_unordered<U>(self) -> Stream<U, L, B, NoOrder, R>
    where
        T: IntoIterator<Item = U>,
    {
        self.flat_map_unordered(q!(|d| d))
    }

    /// For each item in the input stream, apply `f` to produce a [`futures::stream::Stream`],
    /// then emit the elements of that stream one by one. When the inner stream yields
    /// `Pending`, this operator yields as well.
    pub fn flat_map_stream_blocking<U, S, F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Stream<U, L, B, O, R>
    where
        S: futures::Stream<Item = U>,
        F: FnMut(T) -> S + 'a,
        C: ValidMutCommutativityFor<F, T, S, O, WAS_MUT>,
        Idemp: ValidMutIdempotenceFor<F, T, S, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::FlatMapStreamBlocking {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<U, L, B, O, R>::collection_kind()),
            },
        )
    }

    /// For each item in the input stream, treat it as a [`futures::stream::Stream`] and
    /// emit its elements one by one. When the inner stream yields `Pending`, this operator
    /// yields as well.
    pub fn flatten_stream_blocking<U>(self) -> Stream<U, L, B, O, R>
    where
        T: futures::Stream<Item = U>,
    {
        self.flat_map_stream_blocking(q!(|d| d))
    }

    /// Creates a stream containing only the elements of the input stream that satisfy a predicate
    /// `f`, preserving the order of the elements.
    ///
    /// The closure `f` receives a reference `&T` rather than an owned value `T` because filtering does
    /// not modify or take ownership of the values. If you need to modify the values while filtering
    /// use [`Stream::filter_map`] instead.
    ///
    /// If the input stream is unordered **and** `f` mutably captures state, `f` must be
    /// proven **commutative**: processing any two elements in either order must leave
    /// the mutably-captured state in the same final value *and* retain the same multiset
    /// of elements. In particular, the decision for each element must not depend on the
    /// processing order: a stateful predicate like a rate limiter is not commutative —
    /// its budget converges either way, but *which* element passes depends on the order.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec![1, 2, 3, 4]))
    ///     .filter(q!(|&x| x > 2))
    /// # }, |mut stream| async move {
    /// // 3, 4
    /// # for w in (3..5) {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn filter<F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Self
    where
        F: FnMut(&T) -> bool + 'a,
        C: ValidMutBorrowCommutativityFor<F, T, bool, O, WAS_MUT>,
        Idemp: ValidMutBorrowIdempotenceFor<F, T, bool, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_borrow_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::Filter {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Self::collection_kind()),
            },
        )
    }

    /// Splits the stream into two streams based on a predicate, without cloning elements.
    ///
    /// Elements for which `f` returns `true` are sent to the first output stream,
    /// and elements for which `f` returns `false` are sent to the second output stream.
    ///
    /// Unlike using `filter` twice, this only evaluates the predicate once per element
    /// and does not require `T: Clone`.
    ///
    /// The closure `f` receives a reference `&T` rather than an owned value `T` because
    /// the predicate is only used for routing; the element itself is moved to the
    /// appropriate output stream.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use hydro_lang::live_collections::stream::{NoOrder, ExactlyOnce};
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test::<_, _, _, NoOrder, ExactlyOnce>(|process| {
    /// let numbers: Stream<_, _, Unbounded> = process.source_iter(q!(vec![1, 2, 3, 4, 5, 6])).into();
    /// let (evens, odds) = numbers.partition(q!(|&x| x % 2 == 0));
    /// // evens: 2, 4, 6 tagged with true; odds: 1, 3, 5 tagged with false
    /// evens.map(q!(|x| (x, true)))
    ///     .merge_unordered(odds.map(q!(|x| (x, false))))
    /// # }, |mut stream| async move {
    /// # let mut results = Vec::new();
    /// # for _ in 0..6 {
    /// #     results.push(stream.next().await.unwrap());
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec![(1, false), (2, true), (3, false), (4, true), (5, false), (6, true)]);
    /// # }));
    /// # }
    /// ```
    pub fn partition<F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> (Stream<T, L, B, O, R>, Stream<T, L, B, O, R>)
    where
        F: FnMut(&T) -> bool + 'a,
        C: ValidMutBorrowCommutativityFor<F, T, bool, O, WAS_MUT>,
        Idemp: ValidMutBorrowIdempotenceFor<F, T, bool, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_borrow_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        let shared = Rc::new(RefCell::new(HydroNode::PartitionShared {
            input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
            f,
            metadata: self.location.new_node_metadata(Self::collection_kind()),
        }));

        let true_stream = Stream::new(
            self.location.clone(),
            HydroNode::PartitionSide {
                inner: SharedNode(Rc::clone(&shared)),
                is_true: true,
                metadata: self.location.new_node_metadata(Self::collection_kind()),
            },
        );

        let false_stream = Stream::new(
            self.location.clone(),
            HydroNode::PartitionSide {
                inner: SharedNode(shared),
                is_true: false,
                metadata: self.location.new_node_metadata(Self::collection_kind()),
            },
        );

        (true_stream, false_stream)
    }

    /// An operator that both filters and maps. It yields only the items for which the supplied closure `f` returns `Some(value)`.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec!["1", "hello", "world", "2"]))
    ///     .filter_map(q!(|s| s.parse::<usize>().ok()))
    /// # }, |mut stream| async move {
    /// // 1, 2
    /// # for w in (1..3) {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn filter_map<U, F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Stream<U, L, B, O, R>
    where
        F: FnMut(T) -> Option<U> + 'a,
        C: ValidMutCommutativityFor<F, T, Option<U>, O, WAS_MUT>,
        Idemp: ValidMutIdempotenceFor<F, T, Option<U>, R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&expr);
            expr.into()
        });
        Stream::new(
            self.location.clone(),
            HydroNode::FilterMap {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<U, L, B, O, R>::collection_kind()),
            },
        )
    }

    /// Generates a stream that maps each input element `i` to a tuple `(i, x)`,
    /// where `x` is the final value of `other`, a bounded [`Singleton`] or [`Optional`].
    /// If `other` is an empty [`Optional`], no values will be produced.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let batch = process
    ///   .source_iter(q!(vec![1, 2, 3, 4]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let count = batch.clone().count(); // `count()` returns a singleton
    /// batch.cross_singleton(count).all_ticks()
    /// # }, |mut stream| async move {
    /// // (1, 4), (2, 4), (3, 4), (4, 4)
    /// # for w in vec![(1, 4), (2, 4), (3, 4), (4, 4)] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn cross_singleton<O2>(
        self,
        other: impl Into<Optional<O2, L, Bounded>>,
    ) -> Stream<(T, O2), L, B, O, R>
    where
        O2: Clone,
    {
        let other: Optional<O2, L, Bounded> = other.into();
        check_matching_location(&self.location, &other.location);

        Stream::new(
            self.location.clone(),
            HydroNode::CrossSingleton {
                left: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                right: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<(T, O2), L, B, O, R>::collection_kind()),
            },
        )
    }

    /// Passes this stream through if the boolean signal is `true`, otherwise the output is empty.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// // ticks are lazy by default, forces the second tick to run
    /// tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
    ///
    /// let signal = tick.optional_first_tick(q!(())).is_some(); // true on tick 1, false on tick 2
    /// let batch_first_tick = process
    ///   .source_iter(q!(vec![1, 2, 3, 4]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch_second_tick = process
    ///   .source_iter(q!(vec![5, 6, 7, 8]))
    ///   .batch(&tick, nondet!(/** test */))
    ///   .defer_tick();
    /// batch_first_tick.chain(batch_second_tick)
    ///   .filter_if(signal)
    ///   .all_ticks()
    /// # }, |mut stream| async move {
    /// // [1, 2, 3, 4]
    /// # for w in vec![1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn filter_if(self, signal: Singleton<bool, L, Bounded>) -> Stream<T, L, B, O, R> {
        self.cross_singleton(signal.filter(q!(|b| *b)))
            .map(q!(|(d, _)| d))
    }

    /// Passes this stream through if the argument (a [`Bounded`] [`Optional`]) is non-null, otherwise the output is empty.
    ///
    /// Useful for gating the release of elements based on a condition, such as only processing requests if you are the
    /// leader of a cluster.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// // ticks are lazy by default, forces the second tick to run
    /// tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
    ///
    /// let batch_first_tick = process
    ///   .source_iter(q!(vec![1, 2, 3, 4]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch_second_tick = process
    ///   .source_iter(q!(vec![5, 6, 7, 8]))
    ///   .batch(&tick, nondet!(/** test */))
    ///   .defer_tick(); // appears on the second tick
    /// let some_on_first_tick = tick.optional_first_tick(q!(()));
    /// batch_first_tick.chain(batch_second_tick)
    ///   .filter_if_some(some_on_first_tick)
    ///   .all_ticks()
    /// # }, |mut stream| async move {
    /// // [1, 2, 3, 4]
    /// # for w in vec![1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    #[deprecated(note = "use `filter_if` with `Optional::is_some()` instead")]
    pub fn filter_if_some<U>(self, signal: Optional<U, L, Bounded>) -> Stream<T, L, B, O, R> {
        self.filter_if(signal.is_some())
    }

    /// Passes this stream through if the argument (a [`Bounded`] [`Optional`]) is null, otherwise the output is empty.
    ///
    /// Useful for gating the release of elements based on a condition, such as triggering a protocol if you are missing
    /// some local state.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// // ticks are lazy by default, forces the second tick to run
    /// tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
    ///
    /// let batch_first_tick = process
    ///   .source_iter(q!(vec![1, 2, 3, 4]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch_second_tick = process
    ///   .source_iter(q!(vec![5, 6, 7, 8]))
    ///   .batch(&tick, nondet!(/** test */))
    ///   .defer_tick(); // appears on the second tick
    /// let some_on_first_tick = tick.optional_first_tick(q!(()));
    /// batch_first_tick.chain(batch_second_tick)
    ///   .filter_if_none(some_on_first_tick)
    ///   .all_ticks()
    /// # }, |mut stream| async move {
    /// // [5, 6, 7, 8]
    /// # for w in vec![5, 6, 7, 8] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    #[deprecated(note = "use `filter_if` with `!Optional::is_some()` instead")]
    pub fn filter_if_none<U>(self, other: Optional<U, L, Bounded>) -> Stream<T, L, B, O, R> {
        self.filter_if(other.is_none())
    }

    /// Forms the cross-product (Cartesian product, cross-join) of the items in the 2 input streams,
    /// returning all tupled pairs.
    ///
    /// When the right side is [`Bounded`], it is accumulated first and the left side streams
    /// through, preserving the left side's ordering. When both sides are [`Unbounded`], a
    /// symmetric hash join is used and ordering is [`NoOrder`].
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use std::collections::HashSet;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let stream1 = process.source_iter(q!(vec![1, 2]));
    /// let stream2 = process.source_iter(q!(vec!['a', 'b']));
    /// stream1.cross_product(stream2)
    /// # }, |mut stream| async move {
    /// // (1, 'a'), (1, 'b'), (2, 'a'), (2, 'b') in any order
    /// # let expected = HashSet::from([(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')]);
    /// # stream.map(|i| assert!(expected.contains(&i)));
    /// # }));
    /// # }
    pub fn cross_product<T2, B2: Boundedness, O2: Ordering, R2: Retries>(
        self,
        other: Stream<T2, L, B2, O2, R2>,
    ) -> Stream<(T, T2), L, B, B2::PreserveOrderIfBounded<O>, <R as MinRetries<R2>>::Min>
    where
        T: Clone,
        T2: Clone,
        R: MinRetries<R2>,
    {
        self.map(q!(|v| ((), v)))
            .join(other.map(q!(|v| ((), v))))
            .map(q!(|((), (v1, v2))| (v1, v2)))
    }

    /// Takes one stream as input and filters out any duplicate occurrences. The output
    /// contains all unique values from the input.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// process.source_iter(q!(vec![1, 2, 3, 2, 1, 4])).unique()
    /// # }, |mut stream| async move {
    /// # for w in vec![1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn unique(self) -> Stream<T, L, B, O, ExactlyOnce>
    where
        T: Eq + Hash,
    {
        Stream::new(
            self.location.clone(),
            HydroNode::Unique {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T, L, B, O, ExactlyOnce>::collection_kind()),
            },
        )
    }

    /// Outputs everything in this stream that is *not* contained in the `other` stream.
    ///
    /// The `other` stream must be [`Bounded`], since this function will wait until
    /// all its elements are available before producing any output.
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let stream = process
    ///   .source_iter(q!(vec![ 1, 2, 3, 4 ]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch = process
    ///   .source_iter(q!(vec![1, 2]))
    ///   .batch(&tick, nondet!(/** test */));
    /// stream.filter_not_in(batch).all_ticks()
    /// # }, |mut stream| async move {
    /// # for w in vec![3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn filter_not_in<O2: Ordering, B2>(self, other: Stream<T, L, B2, O2, R>) -> Self
    where
        T: Eq + Hash,
        B2: IsBounded,
    {
        check_matching_location(&self.location, &other.location);

        Stream::new(
            self.location.clone(),
            HydroNode::Difference {
                pos: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                neg: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T, L, Bounded, O, R>::collection_kind()),
            },
        )
    }

    /// An operator which allows you to "inspect" each element of a stream without
    /// modifying it. The closure `f` is called on a reference to each item. This is
    /// mainly useful for debugging, and should not be used to generate side-effects.
    ///
    /// If the input stream is unordered **and** `f` mutably captures state, `f` must be
    /// proven **commutative**: executing it on any two elements in either order must
    /// leave the mutably-captured state in the same final value. (The elements
    /// themselves pass through unchanged.)
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let nums = process.source_iter(q!(vec![1, 2]));
    /// // prints "1 * 10 = 10" and "2 * 10 = 20"
    /// nums.inspect(q!(|x| println!("{} * 10 = {}", x, x * 10)))
    /// # }, |mut stream| async move {
    /// # for w in vec![1, 2] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn inspect<F, C, Idemp, const WAS_MUT: bool>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L::DropConsistency, B>,
            StreamMapFuncAlgebra<T, B, C, Idemp, L::SimHookScope>,
        >,
    ) -> Self
    where
        F: FnMut(&T) + 'a,
        C: ValidMutBorrowCommutativityFor<F, T, (), O, WAS_MUT>,
        Idemp: ValidMutBorrowIdempotenceFor<F, T, (), R, WAS_MUT>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (expr, proof) =
                f.splice_fnmut1_borrow_ctx_props(&OperatorContext::<L::DropConsistency, B>::new(
                    &self.location.drop_consistency(),
                ));
            proof.register_proof(&expr);
            expr.into()
        });

        Stream::new(
            self.location.clone(),
            HydroNode::Inspect {
                f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Self::collection_kind()),
            },
        )
    }

    /// Executes the provided closure for every element in this stream.
    ///
    /// If the stream is unordered or has retries, the closure must demonstrate commutativity
    /// and/or idempotence via annotations:
    /// ```rust,ignore
    /// stream.for_each(q!(
    ///     |x| *flag_mut |= x,
    ///     commutative = manual_proof!(/** boolean OR is commutative */),
    ///     idempotent = manual_proof!(/** boolean OR is idempotent */)
    /// ));
    /// ```
    ///
    /// **Commutative** here means that executing the closure on any two elements in
    /// either order must leave its side effects (e.g. mutably-captured state) in the
    /// same final value.
    ///
    /// On a `TotalOrder + ExactlyOnce` stream, no annotations are needed.
    ///
    /// The closure may capture singletons via `by_ref()` or `by_mut()`, as long as the
    /// referenced collection lives at the same location and has the same boundedness as this
    /// stream.
    pub fn for_each<F: FnMut(T) + 'a, C, I>(
        self,
        f: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            AggFuncAlgebra<T, B, C, I, NotProved, L::SimHookScope>,
        >,
    ) where
        C: ValidCommutativityFor<O>,
        I: ValidIdempotenceFor<R>,
    {
        let f = crate::handoff_ref::with_ref_capture(|| {
            let (f, proof) =
                f.splice_fnmut1_ctx_props(&OperatorContext::<L, B>::new(&self.location));
            proof.register_proof(&f);
            f.into()
        });
        self.location
            .flow_state()
            .borrow_mut()
            .push_root(HydroRoot::ForEach {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                f,
                op_metadata: HydroIrOpMetadata::new(),
            });
    }

    /// Sends all elements of this stream to a provided [`futures::Sink`], such as an external
    /// TCP socket to some other server. You should _not_ use this API for interacting with
    /// external clients, instead see [`Location::bidi_external_many_bytes`] and
    /// [`Location::bidi_external_many_bincode`]. This should be used for custom, low-level
    /// interaction with asynchronous sinks.
    pub fn dest_sink<S>(self, sink: impl QuotedWithContext<'a, S, L>)
    where
        O: IsOrdered,
        R: IsExactlyOnce,
        S: 'a + futures::Sink<T> + Unpin,
    {
        self.location
            .flow_state()
            .borrow_mut()
            .push_root(HydroRoot::DestSink {
                sink: sink.splice_typed_ctx(&self.location).into(),
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                op_metadata: HydroIrOpMetadata::new(),
            });
    }

    /// Maps each element `x` of the stream to `(i, x)`, where `i` is the index of the element.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::{prelude::*, live_collections::stream::{TotalOrder, ExactlyOnce}};
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test::<_, _, _, TotalOrder, ExactlyOnce>(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// numbers.enumerate()
    /// # }, |mut stream| async move {
    /// // (0, 1), (1, 2), (2, 3), (3, 4)
    /// # for w in vec![(0, 1), (1, 2), (2, 3), (3, 4)] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn enumerate(self) -> Stream<(usize, T), L, B, O, R>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
    {
        Stream::new(
            self.location.clone(),
            HydroNode::Enumerate {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Stream::<
                    (usize, T),
                    L,
                    B,
                    TotalOrder,
                    ExactlyOnce,
                >::collection_kind()),
            },
        )
    }

    /// Combines elements of the stream into a [`Singleton`], by starting with an intitial value,
    /// generated by the `init` closure, and then applying the `comb` closure to each element in the stream.
    /// Unlike iterators, `comb` takes the accumulator by `&mut` reference, so that it can be modified in place.
    ///
    /// Depending on the input stream guarantees, the closure may need to be commutative
    /// (for unordered streams) or idempotent (for streams with non-deterministic duplicates).
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let words = process.source_iter(q!(vec!["HELLO", "WORLD"]));
    /// words
    ///     .fold(q!(|| String::new()), q!(|acc, x| acc.push_str(x)))
    ///     .into_stream()
    /// # }, |mut stream| async move {
    /// // "HELLOWORLD"
    /// # assert_eq!(stream.next().await.unwrap(), "HELLOWORLD");
    /// # }));
    /// # }
    /// ```
    pub fn fold<A, I, F, C, Idemp, M, B2: SingletonBound>(
        self,
        init: impl IntoQuotedMut<'a, I, OperatorContext<L, B>>,
        comb: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            AggFuncAlgebra<T, B, C, Idemp, M, L::SimHookScope>,
        >,
    ) -> Singleton<A, L, B2>
    where
        I: Fn() -> A + 'a,
        F: 'a + Fn(&mut A, T),
        C: ValidCommutativityFor<O>,
        Idemp: ValidIdempotenceFor<R>,
        B: ApplyMonotoneStream<M, B2>,
    {
        let init = init
            .splice_fn0_ctx(&OperatorContext::<L, B>::new(&self.location))
            .into();
        let (comb, proof) =
            comb.splice_fn2_borrow_mut_ctx_props(&OperatorContext::<L, B>::new(&self.location));
        let ordering_hook = proof.register_proof(&comb);

        // Only assume_retries (for idempotence), not assume_ordering.
        // The fold hook in the simulator handles ordering non-determinism directly, so an
        // ordering hook on the commutativity proof binds to the fold operator itself.
        let nondet = nondet!(/** the combinator function is commutative and idempotent */);
        let retried: Stream<T, L::DropConsistency, B, O, ExactlyOnce> = self.assume_retries(nondet);

        let mut metadata = retried
            .location
            .new_node_metadata(Singleton::<A, L::DropConsistency, B2>::collection_kind());
        metadata.op.sim_hook_id = ordering_hook.map(|hook| hook.id);

        let core = HydroNode::Fold {
            init,
            acc: comb.into(),
            input: Box::new(retried.ir_node.replace(HydroNode::Placeholder)),
            metadata,
            // we do not guarantee consistency at this point because if the algebraic properties
            // do not hold in practice, replica consistency may fail to be maintained, so we
            // would like the simulator to assert consistency; in the future, this will be dynamic
            // based on the proof mechanism
        };

        Singleton::new(retried.location.clone(), core)
            .assert_has_consistency_of(manual_proof!(/** algebraic properties */))
    }

    /// Combines elements of the stream into an [`Optional`], by starting with the first element in the stream,
    /// and then applying the `comb` closure to each element in the stream. The [`Optional`] will be empty
    /// until the first element in the input arrives. Unlike iterators, `comb` takes the accumulator by `&mut`
    /// reference, so that it can be modified in place.
    ///
    /// Depending on the input stream guarantees, the closure may need to be commutative
    /// (for unordered streams) or idempotent (for streams with non-deterministic duplicates).
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let bools = process.source_iter(q!(vec![false, true, false]));
    /// bools.reduce(q!(|acc, x| *acc |= x)).into_stream()
    /// # }, |mut stream| async move {
    /// // true
    /// # assert_eq!(stream.next().await.unwrap(), true);
    /// # }));
    /// # }
    /// ```
    pub fn reduce<F, C, Idemp>(
        self,
        comb: impl IntoQuotedMut<
            'a,
            F,
            OperatorContext<L, B>,
            AggFuncAlgebra<T, B, C, Idemp, NotProved, L::SimHookScope>,
        >,
    ) -> Optional<T, L, B::AggregatedOptional>
    where
        F: Fn(&mut T, T) + 'a,
        C: ValidCommutativityFor<O>,
        Idemp: ValidIdempotenceFor<R>,
    {
        let (f, proof) =
            comb.splice_fn2_borrow_mut_ctx_props(&OperatorContext::<L, B>::new(&self.location));
        let ordering_hook = proof.register_proof(&f);

        let nondet_retries = nondet!(/** the combinator function is commutative and idempotent */);
        let ordered_etc: Stream<T, L::DropConsistency, B> =
            self.assume_retries(nondet_retries).assume_ordering(nondet!(
                /// the combinator function is commutative; the simulator still explores
                /// (or scripts, via the proof's hook) the ordering
                hook = ordering_hook
            ));

        let core = HydroNode::Reduce {
            f: f.into(),
            input: Box::new(ordered_etc.ir_node.replace(HydroNode::Placeholder)),
            metadata: ordered_etc.location.new_node_metadata(Optional::<
                T,
                L::DropConsistency,
                B::AggregatedOptional,
            >::collection_kind()),
        };

        Optional::new(ordered_etc.location.clone(), core)
            .assert_has_consistency_of(manual_proof!(/** algebraic properties */))
    }

    /// Computes the maximum element in the stream as an [`Optional`], which
    /// will be empty until the first element in the input arrives.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.max().all_ticks()
    /// # }, |mut stream| async move {
    /// // 4
    /// # assert_eq!(stream.next().await.unwrap(), 4);
    /// # }));
    /// # }
    /// ```
    pub fn max(self) -> Optional<T, L, B::AggregatedOptional>
    where
        T: Ord,
    {
        self.assume_retries_trusted::<ExactlyOnce>(nondet!(/** max is idempotent */))
            .assume_ordering_trusted_bounded::<TotalOrder>(
                nondet!(/** max is commutative, but order affects intermediates */),
            )
            .reduce(q!(|curr, new| {
                if new > *curr {
                    *curr = new;
                }
            }))
    }

    /// Computes the minimum element in the stream as an [`Optional`], which
    /// will be empty until the first element in the input arrives.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.min().all_ticks()
    /// # }, |mut stream| async move {
    /// // 1
    /// # assert_eq!(stream.next().await.unwrap(), 1);
    /// # }));
    /// # }
    /// ```
    pub fn min(self) -> Optional<T, L, B::AggregatedOptional>
    where
        T: Ord,
    {
        self.assume_retries_trusted::<ExactlyOnce>(nondet!(/** min is idempotent */))
            .assume_ordering_trusted_bounded::<TotalOrder>(
                nondet!(/** max is commutative, but order affects intermediates */),
            )
            .reduce(q!(|curr, new| {
                if new < *curr {
                    *curr = new;
                }
            }))
    }

    /// Computes the first element in the stream as an [`Optional`], which
    /// will be empty until the first element in the input arrives.
    ///
    /// This requires the stream to have a [`TotalOrder`] guarantee, otherwise
    /// re-ordering of elements may cause the first element to change.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.first().all_ticks()
    /// # }, |mut stream| async move {
    /// // 1
    /// # assert_eq!(stream.next().await.unwrap(), 1);
    /// # }));
    /// # }
    /// ```
    pub fn first(self) -> Optional<T, L, B::AggregatedOptional>
    where
        O: IsOrdered,
    {
        self.make_totally_ordered()
            .assume_retries_trusted::<ExactlyOnce>(nondet!(/** first is idempotent */))
            .generator(q!(|| ()), q!(|_, item| Generate::Return(item)))
            .reduce(q!(|_, _| {}))
    }

    /// Computes the last element in the stream as an [`Optional`], which
    /// will be empty until an element in the input arrives.
    ///
    /// This requires the stream to have a [`TotalOrder`] guarantee, otherwise
    /// re-ordering of elements may cause the last element to change.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.last().all_ticks()
    /// # }, |mut stream| async move {
    /// // 4
    /// # assert_eq!(stream.next().await.unwrap(), 4);
    /// # }));
    /// # }
    /// ```
    pub fn last(self) -> Optional<T, L, B::AggregatedOptional>
    where
        O: IsOrdered,
    {
        self.make_totally_ordered()
            .assume_retries_trusted::<ExactlyOnce>(nondet!(/** last is idempotent */))
            .reduce(q!(|curr, new| *curr = new))
    }

    /// Returns a stream containing at most the first `n` elements of the input stream,
    /// preserving the original order. Similar to `LIMIT` in SQL.
    ///
    /// This requires the stream to have a [`TotalOrder`] guarantee and [`ExactlyOnce`]
    /// retries, since the result depends on the order and cardinality of elements.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let numbers = process.source_iter(q!(vec![10, 20, 30, 40, 50]));
    /// numbers.limit(q!(3))
    /// # }, |mut stream| async move {
    /// // 10, 20, 30
    /// # for w in vec![10, 20, 30] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn limit(
        self,
        n: impl QuotedWithContext<'a, usize, OperatorContext<L, B>> + Copy + 'a,
    ) -> Stream<T, L, B, TotalOrder, ExactlyOnce>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
    {
        self.generator(
            q!(|| 0usize),
            q!(move |count, item| {
                if *count == n {
                    Generate::Break
                } else {
                    *count += 1;
                    if *count == n {
                        Generate::Return(item)
                    } else {
                        Generate::Yield(item)
                    }
                }
            }),
        )
    }

    /// Collects all the elements of this stream into a single [`Vec`] element.
    ///
    /// If the input stream is [`Unbounded`], the output [`Singleton`] will be [`Unbounded`] as
    /// well, which means that the value of the [`Vec`] will asynchronously grow as new elements
    /// are added. On such a value, you can use [`Singleton::snapshot`] to grab an instance of
    /// the vector at an arbitrary point in time.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.collect_vec().all_ticks() // emit each tick's Vec into an unbounded stream
    /// # }, |mut stream| async move {
    /// // [ vec![1, 2, 3, 4] ]
    /// # for w in vec![vec![1, 2, 3, 4]] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn collect_vec(self) -> Singleton<Vec<T>, L, B>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
    {
        self.make_totally_ordered().make_exactly_once().fold(
            q!(|| vec![]),
            q!(|acc, v| {
                acc.push(v);
            }),
        )
    }

    /// Applies a function to each element of the stream, maintaining an internal state (accumulator)
    /// and emitting each intermediate result.
    ///
    /// Unlike `fold` which only returns the final accumulated value, `scan` produces a new stream
    /// containing all intermediate accumulated values. The scan operation can also terminate early
    /// by returning `None`.
    ///
    /// The function takes a mutable reference to the accumulator and the current element, and returns
    /// an `Option<U>`. If the function returns `Some(value)`, `value` is emitted to the output stream.
    /// If the function returns `None`, the stream is terminated and no more elements are processed.
    ///
    /// The `init` and `f` closures may capture bounded singletons, optionals, or streams by
    /// reference via [`by_ref()`](crate::live_collections::Singleton::by_ref), as long as the
    /// referenced collection lives at the same location and has the same boundedness as this
    /// stream.
    ///
    /// # Examples
    ///
    /// Basic usage - running sum:
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!(vec![1, 2, 3, 4])).scan(
    ///     q!(|| 0),
    ///     q!(|acc, x| {
    ///         *acc += x;
    ///         Some(*acc)
    ///     }),
    /// )
    /// # }, |mut stream| async move {
    /// // Output: 1, 3, 6, 10
    /// # for w in vec![1, 3, 6, 10] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    ///
    /// Early termination example:
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!(vec![1, 2, 3, 4])).scan(
    ///     q!(|| 1),
    ///     q!(|state, x| {
    ///         *state = *state * x;
    ///         if *state > 6 {
    ///             None // Terminate the stream
    ///         } else {
    ///             Some(-*state)
    ///         }
    ///     }),
    /// )
    /// # }, |mut stream| async move {
    /// // Output: -1, -2, -6
    /// # for w in vec![-1, -2, -6] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn scan<A, U, I, F>(
        self,
        init: impl IntoQuotedMut<'a, I, OperatorContext<L, B>>,
        f: impl IntoQuotedMut<'a, F, OperatorContext<L, B>>,
    ) -> Stream<U, L, B, TotalOrder, ExactlyOnce>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
        I: Fn() -> A + 'a,
        F: Fn(&mut A, T) -> Option<U> + 'a,
    {
        let init = crate::handoff_ref::with_ref_capture(|| {
            init.splice_fn0_ctx(&OperatorContext::<L, B>::new(&self.location))
                .into()
        });
        let f = crate::handoff_ref::with_ref_capture(|| {
            f.splice_fn2_borrow_mut_ctx(&OperatorContext::<L, B>::new(&self.location))
                .into()
        });

        Stream::new(
            self.location.clone(),
            HydroNode::Scan {
                init,
                acc: f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(
                    Stream::<U, L, B, TotalOrder, ExactlyOnce>::collection_kind(),
                ),
            },
        )
    }

    /// Async version of [`Stream::scan`]. Applies an async function to each element of the
    /// stream, maintaining an internal state (accumulator) and emitting the values returned
    /// by the function.
    ///
    /// The closure runs synchronously (so it can mutate the accumulator), then returns a
    /// future. The future is polled to completion. If it resolves to `Some`, the value is
    /// emitted. If it resolves to `None`, the item is filtered out.
    ///
    /// The `init` and `f` closures may capture bounded singletons, optionals, or streams by
    /// reference via [`by_ref()`](crate::live_collections::Singleton::by_ref), as long as the
    /// referenced collection lives at the same location and has the same boundedness as this
    /// stream.
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec![1, 2, 3, 4]))
    ///     .scan_async_blocking(
    ///         q!(|| 0),
    ///         q!(|acc, x| {
    ///             *acc += x;
    ///             let val = *acc;
    ///             async move { Some(val) }
    ///         }),
    ///     )
    /// # }, |mut stream| async move {
    /// // Output: 1, 3, 6, 10
    /// # for w in vec![1, 3, 6, 10] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn scan_async_blocking<A, U, I, F, Fut>(
        self,
        init: impl IntoQuotedMut<'a, I, OperatorContext<L, B>>,
        f: impl IntoQuotedMut<'a, F, OperatorContext<L, B>>,
    ) -> Stream<U, L, B, TotalOrder, ExactlyOnce>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
        I: Fn() -> A + 'a,
        F: Fn(&mut A, T) -> Fut + 'a,
        Fut: Future<Output = Option<U>> + 'a,
    {
        let init = crate::handoff_ref::with_ref_capture(|| {
            init.splice_fn0_ctx(&OperatorContext::<L, B>::new(&self.location))
                .into()
        });
        let f = crate::handoff_ref::with_ref_capture(|| {
            f.splice_fn2_borrow_mut_ctx(&OperatorContext::<L, B>::new(&self.location))
                .into()
        });

        Stream::new(
            self.location.clone(),
            HydroNode::ScanAsyncBlocking {
                init,
                acc: f,
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(
                    Stream::<U, L, B, TotalOrder, ExactlyOnce>::collection_kind(),
                ),
            },
        )
    }

    /// Iteratively processes the elements of the stream using a state machine that can yield
    /// elements as it processes its inputs. This is designed to mirror the unstable generator
    /// syntax in Rust, without requiring special syntax.
    ///
    /// Like [`Stream::scan`], this function takes in an initializer that emits the initial
    /// state. The second argument defines the processing logic, taking in a mutable reference
    /// to the state and the value to be processed. It emits a [`Generate`] value, whose
    /// variants define what is emitted and whether further inputs should be processed.
    ///
    /// The `init` and `f` closures may capture bounded singletons, optionals, or streams by
    /// reference via [`by_ref()`](crate::live_collections::Singleton::by_ref), as long as the
    /// referenced collection lives at the same location and has the same boundedness as this
    /// stream.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!(vec![1, 3, 100, 10])).generator(
    ///     q!(|| 0),
    ///     q!(|acc, x| {
    ///         *acc += x;
    ///         if *acc > 100 {
    ///             hydro_lang::live_collections::keyed_stream::Generate::Return("done!".to_owned())
    ///         } else if *acc % 2 == 0 {
    ///             hydro_lang::live_collections::keyed_stream::Generate::Yield("even".to_owned())
    ///         } else {
    ///             hydro_lang::live_collections::keyed_stream::Generate::Continue
    ///         }
    ///     }),
    /// )
    /// # }, |mut stream| async move {
    /// // Output: "even", "done!"
    /// # let mut results = Vec::new();
    /// # for _ in 0..2 {
    /// #     results.push(stream.next().await.unwrap());
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec!["done!".to_owned(), "even".to_owned()]);
    /// # }));
    /// # }
    /// ```
    pub fn generator<A, U, I, F>(
        self,
        init: impl IntoQuotedMut<'a, I, OperatorContext<L, B>> + Copy,
        f: impl IntoQuotedMut<'a, F, OperatorContext<L, B>> + Copy,
    ) -> Stream<U, L, B, TotalOrder, ExactlyOnce>
    where
        O: IsOrdered,
        R: IsExactlyOnce,
        I: Fn() -> A + 'a,
        F: Fn(&mut A, T) -> Generate<U> + 'a,
    {
        let init: ManualExpr<I, _> =
            ManualExpr::new(move |ctx: &OperatorContext<L, B>| init.splice_fn0_ctx(ctx));
        let f: ManualExpr<F, _> =
            ManualExpr::new(move |ctx: &OperatorContext<L, B>| f.splice_fn2_borrow_mut_ctx(ctx));

        let this = self.make_totally_ordered().make_exactly_once();

        // State is Option<Option<A>>:
        //   None = not yet initialized
        //   Some(Some(a)) = active with state a
        //   Some(None) = terminated
        let scan_init = crate::handoff_ref::with_ref_capture(|| {
            q!(|| None)
                .splice_fn0_ctx::<Option<Option<A>>>(&this.location)
                .into()
        });
        let scan_f = crate::handoff_ref::with_ref_capture(|| {
            q!(move |state: &mut Option<Option<_>>, v| {
                if state.is_none() {
                    *state = Some(Some(init()));
                }
                match state {
                    Some(Some(state_value)) => match f(state_value, v) {
                        Generate::Yield(out) => Some(Some(out)),
                        Generate::Return(out) => {
                            *state = Some(None);
                            Some(Some(out))
                        }
                        // Unlike KeyedStream, we can terminate the scan directly on
                        // Break/Return because there is only one state (no other keys
                        // that still need processing).
                        Generate::Break => None,
                        Generate::Continue => Some(None),
                    },
                    // State is Some(None) after Return; terminate the scan.
                    _ => None,
                }
            })
            .splice_fn2_borrow_mut_ctx::<Option<Option<A>>, T, _>(&OperatorContext::<L, B>::new(
                &this.location,
            ))
            .into()
        });

        let scan_node = HydroNode::Scan {
            init: scan_init,
            acc: scan_f,
            input: Box::new(this.ir_node.replace(HydroNode::Placeholder)),
            metadata: this.location.new_node_metadata(Stream::<
                Option<U>,
                L,
                B,
                TotalOrder,
                ExactlyOnce,
            >::collection_kind()),
        };

        let flatten_f = q!(|d| d)
            .splice_fn1_ctx::<Option<U>, _>(&this.location)
            .into();
        let flatten_node = HydroNode::FlatMap {
            f: flatten_f,
            input: Box::new(scan_node),
            metadata: this
                .location
                .new_node_metadata(Stream::<U, L, B, TotalOrder, ExactlyOnce>::collection_kind()),
        };

        Stream::new(this.location.clone(), flatten_node)
    }

    /// Given a time interval, returns a stream corresponding to samples taken from the
    /// stream roughly at that interval. The output will have elements in the same order
    /// as the input, but with arbitrary elements skipped between samples. There is also
    /// no guarantee on the exact timing of the samples.
    ///
    /// # Non-Determinism
    /// The output stream is non-deterministic in which elements are sampled, since this
    /// is controlled by a clock.
    ///
    /// In simulation tests, the internal batching of elements and of clock samples can be
    /// scripted through the guard's composite hook payload, e.g.
    /// `nondet!(/** reason */ hook = (elements_hook.into(), None))`.
    #[cfg(feature = "tokio")]
    #[expect(
        clippy::type_complexity,
        reason = "composite hook payload names each internal operator's handle type"
    )]
    pub fn sample_every(
        self,
        interval: impl QuotedWithContext<'a, std::time::Duration, L> + Copy + 'a,
        mut nondet: NonDet<(
            Option<crate::sim_hooks::BatchHook<T, O, R, L::SimHookScope>>,
            Option<crate::sim_hooks::BatchHook<(), TotalOrder, ExactlyOnce, L::SimHookScope>>,
        )>,
    ) -> Stream<T, L::DropConsistency, Unbounded, O, AtLeastOnce>
    where
        L: TopLevel<'a>,
    {
        let samples = self.location.source_interval(interval);
        let (elements_hook, samples_hook) = nondet.take_hook();

        let tick = self.location.tick();
        self.batch(
            &tick,
            nondet!(
                /// which elements are batched between samples is captured by the caller's guard
                hook = elements_hook
            ),
        )
        .filter_if(
            samples
                .batch(
                    &tick,
                    nondet!(
                        /// sample timing is captured by the caller's guard
                        hook = samples_hook
                    ),
                )
                .first()
                .is_some(),
        )
        .all_ticks()
        .weaken_retries()
    }

    /// Given a timeout duration, returns an [`Optional`]  which will have a value if the
    /// stream has not emitted a value since that duration.
    ///
    /// # Non-Determinism
    /// Timeout relies on non-deterministic sampling of the stream, so depending on when
    /// samples take place, timeouts may be non-deterministically generated or missed,
    /// and the notification of the timeout may be delayed as well. There is also no
    /// guarantee on how long the [`Optional`] will have a value after the timeout is
    /// detected based on when the next sample is taken.
    #[cfg(feature = "tokio")]
    pub fn timeout(
        self,
        duration: impl QuotedWithContext<
            'a,
            std::time::Duration,
            OperatorContext<Tick<L::DropConsistency>, Bounded>,
        > + Copy
        + 'a,
        nondet: NonDet,
    ) -> Optional<(), L::DropConsistency, Unbounded>
    where
        L: TopLevel<'a>,
    {
        let tick = self.location.tick();

        let latest_received = self
            .assume_retries::<ExactlyOnce>(nondet!(
                /// duplicates only affect the timestamp of the latest element, which is
                /// captured by the caller's guard
                nondet
            ))
            .fold(
                q!(|| None),
                q!(
                    |latest, _| {
                        *latest = Some(Instant::now());
                    },
                    commutative = manual_proof!(/** TODO */)
                ),
            );

        latest_received
            .snapshot(
                &tick,
                nondet!(
                    /// sampling timing is captured by the caller's guard
                    nondet
                ),
            )
            .filter_map(q!(move |latest_received| {
                if let Some(latest_received) = latest_received {
                    if Instant::now().duration_since(latest_received) > duration {
                        Some(())
                    } else {
                        None
                    }
                } else {
                    Some(())
                }
            }))
            .latest()
    }

    /// Shifts this stream into an atomic context, which guarantees that any downstream logic
    /// will all be executed synchronously before any outputs are yielded (in [`Stream::end_atomic`]).
    ///
    /// This is useful to enforce local consistency constraints, such as ensuring that a write is
    /// processed before an acknowledgement is emitted.
    pub fn atomic(self) -> Stream<T, Atomic<L>, B, O, R>
    where
        L: TopLevel<'a>,
    {
        let out_location = Atomic {
            tick: self.location.tick(),
        };
        Stream::new(
            out_location.clone(),
            HydroNode::BeginAtomic {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: out_location
                    .new_node_metadata(Stream::<T, Atomic<L>, B, O, R>::collection_kind()),
            },
        )
    }

    /// Given a tick, returns a stream corresponding to a batch of elements segmented by
    /// that tick. These batches are guaranteed to be contiguous across ticks and preserve
    /// the order of the input. The output stream will execute in the [`Tick`] that was
    /// used to create the atomic section.
    ///
    /// # Non-Determinism
    /// The batch boundaries are non-deterministic and may change across executions.
    ///
    /// In simulation tests, the batching decisions can be scripted by attaching a
    /// [`BatchHook`](crate::sim_hooks::BatchHook) to the guard via
    /// `nondet!(/** reason */ hook = my_hook)`.
    pub fn batch<L2: Location<'a, DropConsistency = L::DropConsistency>>(
        self,
        tick: &Tick<L2>,
        mut nondet: NonDet<Option<crate::sim_hooks::BatchHook<T, O, R, L::SimHookScope>>>,
    ) -> Stream<T, Tick<L::DropConsistency>, Bounded, O, R> {
        assert_eq!(
            Location::id(tick.parent_location()),
            Location::id(&self.location)
        );

        let mut metadata =
            tick.new_node_metadata(Stream::<T, Tick<L>, Bounded, O, R>::collection_kind());
        metadata.op.sim_hook_id = nondet.take_hook().map(|h| h.id);
        Stream::new(
            tick.drop_consistency(),
            HydroNode::Batch {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata,
            },
        )
    }

    /// An operator which allows you to "name" a `HydroNode`.
    /// This is only used for testing, to correlate certain `HydroNode`s with IDs.
    pub fn ir_node_named(self, name: &str) -> Stream<T, L, B, O, R> {
        {
            let mut node = self.ir_node.borrow_mut();
            let metadata = node.metadata_mut();
            metadata.tag = Some(name.to_owned());
        }
        self
    }

    /// Turns this [`Stream`] into a [`Optional`], under the invariant assumption that there is at
    /// most one element. If this invariant is broken, the program may exhibit undefined behavior,
    /// so uses must be carefully vetted.
    pub(crate) fn cast_at_most_one_element(self) -> Optional<T, L, B>
    where
        B: IsBounded,
    {
        Optional::new(
            self.location.clone(),
            HydroNode::Cast {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Optional::<T, L, B>::collection_kind()),
            },
        )
    }

    pub(crate) fn use_ordering_type<O2: Ordering>(self) -> Stream<T, L, B, O2, R> {
        if O::ORDERING_KIND == O2::ORDERING_KIND {
            Stream::new(
                self.location.clone(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else {
            panic!(
                "Runtime ordering {:?} did not match requested cast {:?}.",
                O::ORDERING_KIND,
                O2::ORDERING_KIND
            )
        }
    }

    /// Explicitly "casts" the stream to a type with a different ordering
    /// guarantee. Useful in unsafe code where the ordering cannot be proven
    /// by the type-system.
    ///
    /// # Non-Determinism
    /// This function is used as an escape hatch, and any mistakes in the
    /// provided ordering guarantee will propagate into the guarantees
    /// for the rest of the program.
    pub fn assume_ordering<O2: Ordering>(
        self,
        mut nondet: NonDet<Option<crate::sim_hooks::OrderingHook<T, B, R, L::SimHookScope>>>,
    ) -> Stream<T, L::DropConsistency, B, O2, R> {
        if O::ORDERING_KIND == O2::ORDERING_KIND {
            self.use_ordering_type().weaken_consistency()
        } else if O2::ORDERING_KIND == StreamOrder::NoOrder {
            // We can always weaken the ordering guarantee
            let target_location = self.location().drop_consistency();
            Stream::new(
                target_location.clone(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: target_location
                        .new_node_metadata(Stream::<T, L, B, O2, R>::collection_kind()),
                },
            )
        } else {
            let target_location = self.location().drop_consistency();
            let mut metadata =
                target_location.new_node_metadata(Stream::<T, L, B, O2, R>::collection_kind());
            metadata.op.sim_hook_id = nondet.take_hook().map(|hook| hook.id);
            Stream::new(
                target_location,
                HydroNode::ObserveNonDet {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: false,
                    metadata,
                },
            )
        }
    }

    // like `assume_ordering_trusted`, but only if the input stream is bounded and therefore
    // intermediate states will not be revealed
    fn assume_ordering_trusted_bounded<O2: Ordering>(
        self,
        nondet: NonDet,
    ) -> Stream<T, L, B, O2, R> {
        if B::BOUNDED {
            self.assume_ordering_trusted(nondet)
        } else {
            let self_location = self.location.clone();
            let inner: Stream<T, L::DropConsistency, B, O2, R> = self.assume_ordering(nondet!(
                /// the unbounded stream exposes ordering non-determinism in intermediate states
                nondet
            ));
            Stream::new(self_location, inner.ir_node.replace(HydroNode::Placeholder))
        }
    }

    // only for internal APIs that have been carefully vetted to ensure that the non-determinism
    // is not observable
    pub(crate) fn assume_ordering_trusted<O2: Ordering>(
        self,
        _nondet: NonDet,
    ) -> Stream<T, L, B, O2, R> {
        if O::ORDERING_KIND == O2::ORDERING_KIND {
            self.use_ordering_type()
        } else if O2::ORDERING_KIND == StreamOrder::NoOrder {
            // We can always weaken the ordering guarantee
            Stream::new(
                self.location.clone(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: self
                        .location
                        .new_node_metadata(Stream::<T, L, B, O2, R>::collection_kind()),
                },
            )
        } else {
            Stream::new(
                self.location.clone(),
                HydroNode::ObserveNonDet {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: true,
                    metadata: self
                        .location
                        .new_node_metadata(Stream::<T, L, B, O2, R>::collection_kind()),
                },
            )
        }
    }

    #[deprecated = "use `weaken_ordering::<NoOrder>()` instead"]
    /// Weakens the ordering guarantee provided by the stream to [`NoOrder`],
    /// which is always safe because that is the weakest possible guarantee.
    pub fn weakest_ordering(self) -> Stream<T, L, B, NoOrder, R> {
        self.weaken_ordering::<NoOrder>()
    }

    /// Weakens the ordering guarantee provided by the stream to `O2`, with the type-system
    /// enforcing that `O2` is weaker than the input ordering guarantee.
    pub fn weaken_ordering<O2: WeakerOrderingThan<O>>(self) -> Stream<T, L, B, O2, R> {
        let nondet = nondet!(/** this is a weaker ordering guarantee, so it is safe to assume */);
        self.assume_ordering_trusted::<O2>(nondet)
    }

    /// Strengthens the ordering guarantee to `TotalOrder`, given that `O: IsOrdered`, which
    /// implies that `O == TotalOrder`.
    pub fn make_totally_ordered(self) -> Stream<T, L, B, TotalOrder, R>
    where
        O: IsOrdered,
    {
        self.assume_ordering_trusted(nondet!(/** no-op */))
    }

    /// Explicitly "casts" the stream to a type with a different retries
    /// guarantee. Useful in unsafe code where the lack of retries cannot
    /// be proven by the type-system.
    ///
    /// # Non-Determinism
    /// This function is used as an escape hatch, and any mistakes in the
    /// provided retries guarantee will propagate into the guarantees
    /// for the rest of the program.
    ///
    /// In simulation tests, the retry decisions (how many times each element is
    /// observed) can be scripted by attaching a
    /// [`RetriesHook`](crate::sim_hooks::RetriesHook) to the guard via
    /// `nondet!(/** reason */ hook = my_hook)`. Because every element admits arbitrarily
    /// many retries, this non-determinism cannot be explored autonomously: simulating a
    /// program containing a strengthening `assume_retries` **requires** a bound hook.
    pub fn assume_retries<R2: Retries>(
        self,
        mut nondet: NonDet<Option<crate::sim_hooks::RetriesHook<T, O, B, L::SimHookScope>>>,
    ) -> Stream<T, L::DropConsistency, B, O, R2> {
        if R::RETRIES_KIND == R2::RETRIES_KIND {
            Stream::new(
                self.location.drop_consistency(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else if R2::RETRIES_KIND == StreamRetry::AtLeastOnce {
            // We can always weaken the retries guarantee
            let target_location = self.location.drop_consistency();
            Stream::new(
                target_location.clone(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: target_location
                        .new_node_metadata(Stream::<T, L, B, O, R2>::collection_kind()),
                },
            )
        } else {
            let target_location = self.location.drop_consistency();
            let mut metadata =
                target_location.new_node_metadata(Stream::<T, L, B, O, R2>::collection_kind());
            metadata.op.sim_hook_id = nondet.take_hook().map(|hook| hook.id);
            Stream::new(
                target_location.clone(),
                HydroNode::ObserveNonDet {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: false,
                    metadata,
                },
            )
        }
    }

    // only for internal APIs that have been carefully vetted to ensure that the non-determinism
    // is not observable
    fn assume_retries_trusted<R2: Retries>(self, _nondet: NonDet) -> Stream<T, L, B, O, R2> {
        if R::RETRIES_KIND == R2::RETRIES_KIND {
            Stream::new(
                self.location.clone(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else if R2::RETRIES_KIND == StreamRetry::AtLeastOnce {
            // We can always weaken the retries guarantee
            Stream::new(
                self.location.clone(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: self
                        .location
                        .new_node_metadata(Stream::<T, L, B, O, R2>::collection_kind()),
                },
            )
        } else {
            Stream::new(
                self.location.clone(),
                HydroNode::ObserveNonDet {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    trusted: true,
                    metadata: self
                        .location
                        .new_node_metadata(Stream::<T, L, B, O, R2>::collection_kind()),
                },
            )
        }
    }

    #[deprecated = "use `weaken_retries::<AtLeastOnce>()` instead"]
    /// Weakens the retries guarantee provided by the stream to [`AtLeastOnce`],
    /// which is always safe because that is the weakest possible guarantee.
    pub fn weakest_retries(self) -> Stream<T, L, B, O, AtLeastOnce> {
        self.weaken_retries::<AtLeastOnce>()
    }

    /// Weakens the retries guarantee provided by the stream to `R2`, with the type-system
    /// enforcing that `R2` is weaker than the input retries guarantee.
    pub fn weaken_retries<R2: WeakerRetryThan<R>>(self) -> Stream<T, L, B, O, R2> {
        let nondet = nondet!(/** this is a weaker retry guarantee, so it is safe to assume */);
        self.assume_retries_trusted::<R2>(nondet)
    }

    /// Strengthens the retry guarantee to `ExactlyOnce`, given that `R: IsExactlyOnce`, which
    /// implies that `R == ExactlyOnce`.
    pub fn make_exactly_once(self) -> Stream<T, L, B, O, ExactlyOnce>
    where
        R: IsExactlyOnce,
    {
        self.assume_retries_trusted(nondet!(/** no-op */))
    }

    /// Strengthens the boundedness guarantee to `Bounded`, given that `B: IsBounded`, which
    /// implies that `B == Bounded`.
    pub fn make_bounded(self) -> Stream<T, L, Bounded, O, R>
    where
        B: IsBounded,
    {
        self.weaken_boundedness()
    }

    /// Weakens the boundedness guarantee to an arbitrary boundedness `B2`, given that `B: IsBounded`,
    /// which implies that `B == Bounded`.
    pub fn weaken_boundedness<B2: Boundedness>(self) -> Stream<T, L, B2, O, R> {
        if B::BOUNDED == B2::BOUNDED {
            Stream::new(
                self.location.clone(),
                self.ir_node.replace(HydroNode::Placeholder),
            )
        } else {
            // We can always weaken the boundedness
            Stream::new(
                self.location.clone(),
                HydroNode::Cast {
                    inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                    metadata: self
                        .location
                        .new_node_metadata(Stream::<T, L, B2, O, R>::collection_kind()),
                },
            )
        }
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<&T, L, B, O, R>
where
    L: Location<'a>,
{
    /// Clone each element of the stream; akin to `map(q!(|d| d.clone()))`.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!(&[1, 2, 3])).cloned()
    /// # }, |mut stream| async move {
    /// // 1, 2, 3
    /// # for w in vec![1, 2, 3] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn cloned(self) -> Stream<T, L, B, O, R>
    where
        T: Clone,
    {
        self.map(q!(|d| d.clone()))
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering> Stream<T, L, B, O, ExactlyOnce>
where
    L: Location<'a>,
{
    /// Computes the number of elements in the stream as a [`Singleton`].
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.count().all_ticks()
    /// # }, |mut stream| async move {
    /// // 4
    /// # assert_eq!(stream.next().await.unwrap(), 4);
    /// # }));
    /// # }
    /// ```
    pub fn count(self) -> Singleton<usize, L, B::StreamToMonotone> {
        self.assume_ordering_trusted::<TotalOrder>(nondet!(
            /// Order does not affect eventual count, and also does not affect intermediate states.
        ))
        .fold(
            q!(|| 0usize),
            q!(
                |count, _| *count += 1,
                monotone = manual_proof!(/** += 1 is monotone */)
            ),
        )
    }
}

impl<'a, T, L: Location<'a>, O: Ordering, R: Retries> Stream<T, L, Unbounded, O, R> {
    /// Produces a new stream that merges the elements of the two input streams.
    /// The result has [`NoOrder`] because the order of merging is not guaranteed.
    ///
    /// Currently, both input streams must be [`Unbounded`]. When the streams are
    /// [`Bounded`], you can use [`Stream::chain`] instead.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let numbers: Stream<i32, _, Unbounded> = // 1, 2, 3, 4
    /// # process.source_iter(q!(vec![1, 2, 3, 4])).into();
    /// numbers.clone().map(q!(|x| x + 1)).merge_unordered(numbers)
    /// # }, |mut stream| async move {
    /// // 2, 3, 4, 5, and 1, 2, 3, 4 merged in unknown order
    /// # for w in vec![2, 3, 4, 5, 1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn merge_unordered<O2: Ordering, R2: Retries>(
        self,
        other: Stream<T, L, Unbounded, O2, R2>,
    ) -> Stream<T, L, Unbounded, NoOrder, <R as MinRetries<R2>>::Min>
    where
        R: MinRetries<R2>,
    {
        Stream::new(
            self.location.clone(),
            HydroNode::Chain {
                first: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                second: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Stream::<
                    T,
                    L,
                    Unbounded,
                    NoOrder,
                    <R as MinRetries<R2>>::Min,
                >::collection_kind()),
            },
        )
    }

    /// Deprecated: use [`Stream::merge_unordered`] instead.
    #[deprecated(note = "use `merge_unordered` instead")]
    pub fn interleave<O2: Ordering, R2: Retries>(
        self,
        other: Stream<T, L, Unbounded, O2, R2>,
    ) -> Stream<T, L, Unbounded, NoOrder, <R as MinRetries<R2>>::Min>
    where
        R: MinRetries<R2>,
    {
        self.merge_unordered(other)
    }
}

impl<'a, T, L: Location<'a>, B: Boundedness, R: Retries> Stream<T, L, B, TotalOrder, R> {
    /// Produces a new stream that combines the elements of the two input streams,
    /// preserving the relative order of elements within each input.
    ///
    /// # Non-Determinism
    /// The order in which elements *across* the two streams will be interleaved is
    /// non-deterministic, so the order of elements will vary across runs. If the output
    /// order is irrelevant, use [`Stream::merge_unordered`] instead, which is deterministic
    /// but emits an unordered stream. For deterministic first-then-second ordering on
    /// bounded streams, use [`Stream::chain`].
    ///
    /// In simulation tests, the interleaving decisions can be scripted by attaching a
    /// [`MergeOrderedHook`](crate::sim_hooks::MergeOrderedHook) to the guard via
    /// `nondet!(/** reason */ hook = my_hook)`.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let numbers: Stream<i32, _, Unbounded> = // 1, 3
    /// # process.source_iter(q!(vec![1, 3])).into();
    /// numbers.clone().merge_ordered(numbers.map(q!(|x| x + 1)), nondet!(/** example */))
    /// # }, |mut stream| async move {
    /// // 1, 3 and 2, 4 in some order, preserving the original local order
    /// # for w in vec![1, 3, 2, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn merge_ordered<R2: Retries>(
        self,
        other: Stream<T, L, B, TotalOrder, R2>,
        mut nondet: NonDet<Option<crate::sim_hooks::MergeOrderedHook<T, B, L::SimHookScope>>>,
    ) -> Stream<T, L::DropConsistency, B, TotalOrder, <R as MinRetries<R2>>::Min>
    where
        R: MinRetries<R2>,
    {
        let target_location = self.location().drop_consistency();
        let mut metadata = target_location.new_node_metadata(Stream::<
            T,
            L::DropConsistency,
            B,
            TotalOrder,
            <R as MinRetries<R2>>::Min,
        >::collection_kind());
        metadata.op.sim_hook_id = nondet.take_hook().map(|hook| hook.id);
        Stream::new(
            target_location,
            HydroNode::MergeOrdered {
                first: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                second: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata,
            },
        )
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<T, L, B, O, R>
where
    L: Location<'a>,
{
    /// Produces a new stream that emits the input elements in sorted order.
    ///
    /// The input stream can have any ordering guarantee, but the output stream
    /// will have a [`TotalOrder`] guarantee. This operator will block until all
    /// elements in the input stream are available, so it requires the input stream
    /// to be [`Bounded`].
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![4, 2, 3, 1]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.sort().all_ticks()
    /// # }, |mut stream| async move {
    /// // 1, 2, 3, 4
    /// # for w in (1..5) {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn sort(self) -> Stream<T, L, Bounded, TotalOrder, R>
    where
        B: IsBounded,
        T: Ord,
    {
        let this = self.make_bounded();
        Stream::new(
            this.location.clone(),
            HydroNode::Sort {
                input: Box::new(this.ir_node.replace(HydroNode::Placeholder)),
                metadata: this
                    .location
                    .new_node_metadata(Stream::<T, L, Bounded, TotalOrder, R>::collection_kind()),
            },
        )
    }

    /// Produces a new stream that first emits the elements of the `self` stream,
    /// and then emits the elements of the `other` stream. The output stream has
    /// a [`TotalOrder`] guarantee if and only if both input streams have a
    /// [`TotalOrder`] guarantee.
    ///
    /// Currently, both input streams must be [`Bounded`]. This operator will block
    /// on the first stream until all its elements are available. In a future version,
    /// we will relax the requirement on the `other` stream.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.clone().map(q!(|x| x + 1)).chain(batch).all_ticks()
    /// # }, |mut stream| async move {
    /// // 2, 3, 4, 5, 1, 2, 3, 4
    /// # for w in vec![2, 3, 4, 5, 1, 2, 3, 4] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn chain<O2: Ordering, R2: Retries, B2: Boundedness>(
        self,
        other: Stream<T, L, B2, O2, R2>,
    ) -> Stream<T, L, B2, <O as MinOrder<O2>>::Min, <R as MinRetries<R2>>::Min>
    where
        B: IsBounded,
        O: MinOrder<O2>,
        R: MinRetries<R2>,
    {
        check_matching_location(&self.location, &other.location);

        Stream::new(
            self.location.clone(),
            HydroNode::Chain {
                first: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                second: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Stream::<
                    T,
                    L,
                    B2,
                    <O as MinOrder<O2>>::Min,
                    <R as MinRetries<R2>>::Min,
                >::collection_kind()),
            },
        )
    }

    /// Forms the cross-product (Cartesian product, cross-join) of the items in the 2 input streams.
    /// Unlike [`Stream::cross_product`], the output order is totally ordered when the inputs are
    /// because this is compiled into a nested loop.
    pub fn cross_product_nested_loop<T2, O2: Ordering + MinOrder<O>, R2: Retries>(
        self,
        other: Stream<T2, L, Bounded, O2, R2>,
    ) -> Stream<(T, T2), L, Bounded, <O2 as MinOrder<O>>::Min, <R as MinRetries<R2>>::Min>
    where
        B: IsBounded,
        T: Clone,
        T2: Clone,
        R: MinRetries<R2>,
    {
        let this = self.make_bounded();
        check_matching_location(&this.location, &other.location);

        Stream::new(
            this.location.clone(),
            HydroNode::CrossProduct {
                left: Box::new(this.ir_node.replace(HydroNode::Placeholder)),
                right: Box::new(other.ir_node.replace(HydroNode::Placeholder)),
                metadata: this.location.new_node_metadata(Stream::<
                    (T, T2),
                    L,
                    Bounded,
                    <O2 as MinOrder<O>>::Min,
                    <R as MinRetries<R2>>::Min,
                >::collection_kind()),
            },
        )
    }

    /// Creates a [`KeyedStream`] with the same set of keys as `keys`, but with the elements in
    /// `self` used as the values for *each* key.
    ///
    /// This is helpful when "broadcasting" a set of values so that all the keys have the same
    /// values. For example, it can be used to send the same set of elements to several cluster
    /// members, if the membership information is available as a [`KeyedSingleton`].
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// # let tick = process.tick();
    /// let keyed_singleton = // { 1: (), 2: () }
    /// # process
    /// #     .source_iter(q!(vec![(1, ()), (2, ())]))
    /// #     .into_keyed()
    /// #     .batch(&tick, nondet!(/** test */))
    /// #     .first();
    /// let stream = // [ "a", "b" ]
    /// # process
    /// #     .source_iter(q!(vec!["a".to_owned(), "b".to_owned()]))
    /// #     .batch(&tick, nondet!(/** test */));
    /// stream.repeat_with_keys(keyed_singleton)
    /// # .entries().all_ticks()
    /// # }, |mut stream| async move {
    /// // { 1: ["a", "b" ], 2: ["a", "b"] }
    /// # let mut results = Vec::new();
    /// # for _ in 0..4 {
    /// #     results.push(stream.next().await.unwrap());
    /// # }
    /// # results.sort();
    /// # assert_eq!(results, vec![(1, "a".to_owned()), (1, "b".to_owned()), (2, "a".to_owned()), (2, "b".to_owned())]);
    /// # }));
    /// # }
    /// ```
    pub fn repeat_with_keys<K, V2>(
        self,
        keys: KeyedSingleton<K, V2, L, Bounded>,
    ) -> KeyedStream<K, T, L, Bounded, O, R>
    where
        B: IsBounded,
        K: Clone,
        T: Clone,
    {
        keys.keys()
            .assume_ordering_trusted::<TotalOrder>(
                nondet!(/** keyed stream does not depend on ordering of keys */),
            )
            .cross_product_nested_loop(self.make_bounded())
            .into_keyed()
    }

    /// Consumes a stream of `Future<T>`, resolving each future while blocking subgraph
    /// execution until all results are available. The output order is based on when futures
    /// complete, and may be different than the input order.
    ///
    /// Unlike [`Stream::resolve_futures`], which allows the subgraph to continue executing
    /// while futures are pending, this variant blocks until the futures resolve.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use std::collections::HashSet;
    /// # use futures::StreamExt;
    /// # use hydro_lang::prelude::*;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!([2, 3, 1, 9, 6, 5, 4, 7, 8]))
    ///     .map(q!(|x| async move {
    ///         tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    ///         x
    ///     }))
    ///     .resolve_futures_blocking()
    /// #   },
    /// #   |mut stream| async move {
    /// // 1, 2, 3, 4, 5, 6, 7, 8, 9 (in any order)
    /// #       let mut output = HashSet::new();
    /// #       for _ in 1..10 {
    /// #           output.insert(stream.next().await.unwrap());
    /// #       }
    /// #       assert_eq!(
    /// #           output,
    /// #           HashSet::<i32>::from_iter(1..10)
    /// #       );
    /// #   },
    /// # ));
    /// # }
    /// ```
    pub fn resolve_futures_blocking(self) -> Stream<T::Output, L, B, NoOrder, R>
    where
        T: Future,
    {
        Stream::new(
            self.location.clone(),
            HydroNode::ResolveFuturesBlocking {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T::Output, L, B, NoOrder, R>::collection_kind()),
            },
        )
    }

    /// Returns a [`Singleton`] containing `true` if the stream has no elements, or `false` otherwise.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let empty: Stream<i32, _, Bounded> = process
    ///   .source_iter(q!(Vec::<i32>::new()))
    ///   .batch(&tick, nondet!(/** test */));
    /// empty.is_empty().all_ticks()
    /// # }, |mut stream| async move {
    /// // true
    /// # assert_eq!(stream.next().await.unwrap(), true);
    /// # }));
    /// # }
    /// ```
    #[expect(clippy::wrong_self_convention, reason = "stream function naming")]
    pub fn is_empty(self) -> Singleton<bool, L, Bounded>
    where
        B: IsBounded,
    {
        self.make_bounded()
            .assume_ordering_trusted::<TotalOrder>(
                nondet!(/** is_empty intermediates unaffected by order */),
            )
            .first()
            .is_none()
    }
}

impl<'a, K, V1, L, B: Boundedness, O: Ordering, R: Retries> Stream<(K, V1), L, B, O, R>
where
    L: Location<'a>,
{
    /// Given two streams of pairs `(K, V1)` and `(K, V2)`, produces a new stream of nested pairs `(K, (V1, V2))`
    /// by equi-joining the two streams on the key attribute `K`.
    ///
    /// When the right-hand side is [`Bounded`], the join accumulates the right side first
    /// and streams the left side through, preserving the left side's ordering. When both
    /// sides are [`Unbounded`], a symmetric hash join is used and ordering is [`NoOrder`].
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use std::collections::HashSet;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let stream1 = process.source_iter(q!(vec![(1, 'a'), (2, 'b')]));
    /// let stream2 = process.source_iter(q!(vec![(1, 'x'), (2, 'y')]));
    /// stream1.join(stream2)
    /// # }, |mut stream| async move {
    /// // (1, ('a', 'x')), (2, ('b', 'y'))
    /// # let expected = HashSet::from([(1, ('a', 'x')), (2, ('b', 'y'))]);
    /// # stream.map(|i| assert!(expected.contains(&i)));
    /// # }));
    /// # }
    pub fn join<V2, B2: Boundedness, O2: Ordering, R2: Retries>(
        self,
        n: Stream<(K, V2), L, B2, O2, R2>,
    ) -> Stream<(K, (V1, V2)), L, B, B2::PreserveOrderIfBounded<O>, <R as MinRetries<R2>>::Min>
    where
        K: Eq + Hash + Clone,
        R: MinRetries<R2>,
        V1: Clone,
        V2: Clone,
    {
        check_matching_location(&self.location, &n.location);

        let ir_node = if B2::BOUNDED {
            HydroNode::JoinHalf {
                left: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                right: Box::new(n.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Stream::<
                    (K, (V1, V2)),
                    L,
                    B,
                    B2::PreserveOrderIfBounded<O>,
                    <R as MinRetries<R2>>::Min,
                >::collection_kind()),
            }
        } else {
            HydroNode::Join {
                left: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                right: Box::new(n.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.new_node_metadata(Stream::<
                    (K, (V1, V2)),
                    L,
                    B,
                    B2::PreserveOrderIfBounded<O>,
                    <R as MinRetries<R2>>::Min,
                >::collection_kind()),
            }
        };

        Stream::new(self.location.clone(), ir_node)
    }

    /// Given a stream of pairs `(K, V1)` and a bounded stream of keys `K`,
    /// computes the anti-join of the items in the input -- i.e. returns
    /// unique items in the first input that do not have a matching key
    /// in the second input.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let stream = process
    ///   .source_iter(q!(vec![ (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') ]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch = process
    ///   .source_iter(q!(vec![1, 2]))
    ///   .batch(&tick, nondet!(/** test */));
    /// stream.anti_join(batch).all_ticks()
    /// # }, |mut stream| async move {
    /// # for w in vec![(3, 'c'), (4, 'd')] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    pub fn anti_join<O2: Ordering, R2: Retries>(
        self,
        n: Stream<K, L, Bounded, O2, R2>,
    ) -> Stream<(K, V1), L, B, O, R>
    where
        K: Eq + Hash,
    {
        check_matching_location(&self.location, &n.location);

        Stream::new(
            self.location.clone(),
            HydroNode::AntiJoin {
                pos: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                neg: Box::new(n.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<(K, V1), L, B, O, R>::collection_kind()),
            },
        )
    }
}

impl<'a, K, V, L: Location<'a>, B: Boundedness, O: Ordering, R: Retries>
    Stream<(K, V), L, B, O, R>
{
    /// Transforms this stream into a [`KeyedStream`], where the first element of each tuple
    /// is used as the key and the second element is added to the entries associated with that key.
    ///
    /// Because [`KeyedStream`] lazily groups values into buckets, this operator has zero computational
    /// cost and _does not_ require that the key type is hashable. Keyed streams are useful for
    /// performing grouped aggregations, but also for more precise ordering guarantees such as
    /// total ordering _within_ each group but no ordering _across_ groups.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process
    ///     .source_iter(q!(vec![(1, 2), (1, 3), (2, 4)]))
    ///     .into_keyed()
    /// #   .entries()
    /// # }, |mut stream| async move {
    /// // { 1: [2, 3], 2: [4] }
    /// # for w in vec![(1, 2), (1, 3), (2, 4)] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn into_keyed(self) -> KeyedStream<K, V, L, B, O, R> {
        KeyedStream::new(
            self.location.clone(),
            HydroNode::Cast {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(KeyedStream::<K, V, L, B, O, R>::collection_kind()),
            },
        )
    }
}

impl<'a, K, V, L, O: Ordering, R: Retries> Stream<(K, V), Tick<L>, Bounded, O, R>
where
    K: Eq + Hash,
    L: Location<'a>,
{
    /// Given a stream of pairs `(K, V)`, produces a new stream of unique keys `K`.
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
    /// let batch = numbers.batch(&tick, nondet!(/** test */));
    /// batch.keys().all_ticks()
    /// # }, |mut stream| async move {
    /// // 1, 2
    /// # assert_eq!(stream.next().await.unwrap(), 1);
    /// # assert_eq!(stream.next().await.unwrap(), 2);
    /// # }));
    /// # }
    /// ```
    pub fn keys(self) -> Stream<K, Tick<L>, Bounded, NoOrder, ExactlyOnce> {
        self.into_keyed()
            .fold(
                q!(|| ()),
                q!(
                    |_, _| {},
                    commutative = manual_proof!(/** values are ignored */),
                    idempotent = manual_proof!(/** values are ignored */)
                ),
            )
            .keys()
    }
}

impl<'a, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<T, Atomic<L>, B, O, R>
where
    L: Location<'a>,
{
    /// Returns a stream corresponding to the latest batch of elements being atomically
    /// processed. These batches are guaranteed to be contiguous across ticks and preserve
    /// the order of the input.
    ///
    /// # Non-Determinism
    /// The batch boundaries are non-deterministic and may change across executions.
    pub fn batch_atomic<L2: Location<'a, DropConsistency = L::DropConsistency>>(
        self,
        tick: &Tick<L2>,
        mut nondet: NonDet<Option<crate::sim_hooks::BatchHook<T, O, R, L::SimHookScope>>>,
    ) -> Stream<T, Tick<L::DropConsistency>, Bounded, O, R> {
        assert_eq!(
            Location::id(tick.parent_location()),
            Location::id(self.location.tick.parent_location())
        );

        let mut metadata =
            tick.new_node_metadata(Stream::<T, Tick<L>, Bounded, O, R>::collection_kind());

        metadata.op.sim_hook_id = nondet.take_hook().map(|h| h.id);
        Stream::new(
            tick.drop_consistency(),
            HydroNode::Batch {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata,
            },
        )
    }

    /// Yields the elements of this stream back into a top-level, asynchronous execution context.
    /// See [`Stream::atomic`] for more details.
    pub fn end_atomic(self) -> Stream<T, L, B, O, R> {
        Stream::new(
            self.location.tick.l.clone(),
            HydroNode::EndAtomic {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .tick
                    .l
                    .new_node_metadata(Stream::<T, L, B, O, R>::collection_kind()),
            },
        )
    }
}

impl<'a, F, T, L, B: Boundedness, O: Ordering, R: Retries> Stream<F, L, B, O, R>
where
    L: TopLevel<'a>,
    F: Future<Output = T>,
{
    /// Consumes a stream of `Future<T>`, produces a new stream of the resulting `T` outputs.
    /// Future outputs are produced as available, regardless of input arrival order.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use std::collections::HashSet;
    /// # use futures::StreamExt;
    /// # use hydro_lang::prelude::*;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!([2, 3, 1, 9, 6, 5, 4, 7, 8]))
    ///     .map(q!(|x| async move {
    ///         tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    ///         x
    ///     }))
    ///     .resolve_futures()
    /// #   },
    /// #   |mut stream| async move {
    /// // 1, 2, 3, 4, 5, 6, 7, 8, 9 (in any order)
    /// #       let mut output = HashSet::new();
    /// #       for _ in 1..10 {
    /// #           output.insert(stream.next().await.unwrap());
    /// #       }
    /// #       assert_eq!(
    /// #           output,
    /// #           HashSet::<i32>::from_iter(1..10)
    /// #       );
    /// #   },
    /// # ));
    /// # }
    pub fn resolve_futures(self) -> Stream<T, L, Unbounded, NoOrder, R> {
        Stream::new(
            self.location.clone(),
            HydroNode::ResolveFutures {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T, L, Unbounded, NoOrder, R>::collection_kind()),
            },
        )
    }

    /// Consumes a stream of `Future<T>`, produces a new stream of the resulting `T` outputs.
    /// Future outputs are produced in the same order as the input stream.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use std::collections::HashSet;
    /// # use futures::StreamExt;
    /// # use hydro_lang::prelude::*;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// process.source_iter(q!([2, 3, 1, 9, 6, 5, 4, 7, 8]))
    ///     .map(q!(|x| async move {
    ///         tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    ///         x
    ///     }))
    ///     .resolve_futures_ordered()
    /// #   },
    /// #   |mut stream| async move {
    /// // 2, 3, 1, 9, 6, 5, 4, 7, 8
    /// #       let mut output = Vec::new();
    /// #       for _ in 1..10 {
    /// #           output.push(stream.next().await.unwrap());
    /// #       }
    /// #       assert_eq!(
    /// #           output,
    /// #           vec![2, 3, 1, 9, 6, 5, 4, 7, 8]
    /// #       );
    /// #   },
    /// # ));
    /// # }
    pub fn resolve_futures_ordered(self) -> Stream<T, L, Unbounded, O, R> {
        Stream::new(
            self.location.clone(),
            HydroNode::ResolveFuturesOrdered {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T, L, Unbounded, O, R>::collection_kind()),
            },
        )
    }
}

impl<'a, T, L, O: Ordering, R: Retries> Stream<T, Tick<L>, Bounded, O, R>
where
    L: Location<'a>,
{
    /// Asynchronously yields this batch of elements outside the tick as an unbounded stream,
    /// which will stream all the elements across _all_ tick iterations by concatenating the batches.
    pub fn all_ticks(self) -> Stream<T, L, Unbounded, O, R> {
        Stream::new(
            self.location.parent_location().clone(),
            HydroNode::YieldConcat {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self.location.parent_location().new_node_metadata(Stream::<
                    T,
                    L,
                    Unbounded,
                    O,
                    R,
                >::collection_kind(
                )),
            },
        )
    }

    /// Synchronously yields this batch of elements outside the tick as an unbounded stream,
    /// which will stream all the elements across _all_ tick iterations by concatenating the batches.
    ///
    /// Unlike [`Stream::all_ticks`], this preserves synchronous execution, as the output stream
    /// is emitted in an [`Atomic`] context that will process elements synchronously with the input
    /// stream's [`Tick`] context.
    pub fn all_ticks_atomic(self) -> Stream<T, Atomic<L>, Unbounded, O, R> {
        let out_location = Atomic {
            tick: self.location.clone(),
        };

        Stream::new(
            out_location.clone(),
            HydroNode::YieldConcat {
                inner: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: out_location
                    .new_node_metadata(Stream::<T, Atomic<L>, Unbounded, O, R>::collection_kind()),
            },
        )
    }

    /// Transforms the stream using the given closure in "stateful" mode, where stateful operators
    /// such as `fold` retrain their memory across ticks rather than resetting across batches of
    /// input.
    ///
    /// This API is particularly useful for stateful computation on batches of data, such as
    /// maintaining an accumulated state that is up to date with the current batch.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// # // ticks are lazy by default, forces the second tick to run
    /// # tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
    /// # let batch_first_tick = process
    /// #   .source_iter(q!(vec![1, 2, 3, 4]))
    /// #  .batch(&tick, nondet!(/** test */));
    /// # let batch_second_tick = process
    /// #   .source_iter(q!(vec![5, 6, 7]))
    /// #   .batch(&tick, nondet!(/** test */))
    /// #   .defer_tick(); // appears on the second tick
    /// let input = // [1, 2, 3, 4 (first batch), 5, 6, 7 (second batch)]
    /// # batch_first_tick.chain(batch_second_tick);
    ///
    /// input.across_ticks(|s| s.count()).all_ticks()
    /// # }, |mut stream| async move {
    /// // [4, 7]
    /// assert_eq!(stream.next().await.unwrap(), 4);
    /// assert_eq!(stream.next().await.unwrap(), 7);
    /// # }));
    /// # }
    /// ```
    pub fn across_ticks<Out: BatchAtomic<'a>>(
        self,
        thunk: impl FnOnce(Stream<T, Atomic<L>, Unbounded, O, R>) -> Out,
    ) -> Out::Batched {
        thunk(self.all_ticks_atomic()).batched_atomic()
    }

    /// Shifts the elements in `self` to the **next tick**, so that the returned stream at tick `T`
    /// always has the elements of `self` at tick `T - 1`.
    ///
    /// At tick `0`, the output stream is empty, since there is no previous tick.
    ///
    /// This operator enables stateful iterative processing with ticks, by sending data from one
    /// tick to the next. For example, you can use it to compare inputs across consecutive batches.
    ///
    /// # Example
    /// ```rust
    /// # #[cfg(feature = "deploy")] {
    /// # use hydro_lang::prelude::*;
    /// # use futures::StreamExt;
    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
    /// let tick = process.tick();
    /// // ticks are lazy by default, forces the second tick to run
    /// tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
    ///
    /// let batch_first_tick = process
    ///   .source_iter(q!(vec![1, 2, 3, 4]))
    ///   .batch(&tick, nondet!(/** test */));
    /// let batch_second_tick = process
    ///   .source_iter(q!(vec![0, 3, 4, 5, 6]))
    ///   .batch(&tick, nondet!(/** test */))
    ///   .defer_tick(); // appears on the second tick
    /// let changes_across_ticks = batch_first_tick.chain(batch_second_tick);
    ///
    /// changes_across_ticks.clone().filter_not_in(
    ///     changes_across_ticks.defer_tick() // the elements from the previous tick
    /// ).all_ticks()
    /// # }, |mut stream| async move {
    /// // [1, 2, 3, 4 /* first tick */, 0, 5, 6 /* second tick */]
    /// # for w in vec![1, 2, 3, 4, 0, 5, 6] {
    /// #     assert_eq!(stream.next().await.unwrap(), w);
    /// # }
    /// # }));
    /// # }
    /// ```
    pub fn defer_tick(self) -> Stream<T, Tick<L>, Bounded, O, R> {
        Stream::new(
            self.location.clone(),
            HydroNode::DeferTick {
                input: Box::new(self.ir_node.replace(HydroNode::Placeholder)),
                metadata: self
                    .location
                    .new_node_metadata(Stream::<T, Tick<L>, Bounded, O, R>::collection_kind()),
            },
        )
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "deploy")]
    use futures::{SinkExt, StreamExt};
    #[cfg(feature = "deploy")]
    use hydro_deploy::Deployment;
    #[cfg(feature = "deploy")]
    use serde::{Deserialize, Serialize};
    #[cfg(any(feature = "deploy", feature = "sim"))]
    use stageleft::q;

    #[cfg(any(feature = "deploy", feature = "sim"))]
    use crate::compile::builder::FlowBuilder;
    #[cfg(feature = "deploy")]
    use crate::live_collections::sliced::sliced;
    #[cfg(feature = "deploy")]
    use crate::live_collections::stream::ExactlyOnce;
    #[cfg(feature = "sim")]
    use crate::live_collections::stream::NoOrder;
    #[cfg(any(feature = "deploy", feature = "sim"))]
    use crate::live_collections::stream::TotalOrder;
    #[cfg(any(feature = "deploy", feature = "sim"))]
    use crate::location::Location;
    #[cfg(feature = "sim")]
    use crate::networking::TCP;
    #[cfg(any(feature = "deploy", feature = "sim"))]
    use crate::nondet::nondet;

    mod backtrace_chained_ops;

    #[cfg(feature = "deploy")]
    struct P1 {}
    #[cfg(feature = "deploy")]
    struct P2 {}

    #[cfg(feature = "deploy")]
    #[derive(Serialize, Deserialize, Debug)]
    struct SendOverNetwork {
        n: u32,
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn first_ten_distributed() {
        use crate::networking::TCP;

        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let first_node = flow.process::<P1>();
        let second_node = flow.process::<P2>();
        let external = flow.external::<P2>();

        let numbers = first_node.source_iter(q!(0..10));
        let out_port = numbers
            .map(q!(|n| SendOverNetwork { n }))
            .send(&second_node, TCP.fail_stop().bincode())
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&first_node, deployment.Localhost())
            .with_process(&second_node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_out = nodes.connect(out_port).await;

        deployment.start().await.unwrap();

        for i in 0..10 {
            assert_eq!(external_out.next().await.unwrap().n, i);
        }
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn first_cardinality() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let node_tick = node.tick();
        let count = node_tick
            .singleton(q!([1, 2, 3]))
            .into_stream()
            .flatten_ordered()
            .first()
            .into_stream()
            .count()
            .all_ticks()
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_out = nodes.connect(count).await;

        deployment.start().await.unwrap();

        assert_eq!(external_out.next().await.unwrap(), 1);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unbounded_reduce_remembers_state() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) = node.source_external_bincode(&external);
        let out = input
            .reduce(q!(|acc, v| *acc += v))
            .sample_eager(nondet!(/** test */))
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 1);

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 3);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn top_level_bounded_cross_singleton() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);

        let out = input
            .cross_singleton(
                node.source_iter(q!(vec![1, 2, 3]))
                    .fold(q!(|| 0), q!(|acc, v| *acc += v)),
            )
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (1, 6));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (2, 6));
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn top_level_bounded_reduce_cardinality() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);

        let out = sliced! {
            let input = use::batch(input, nondet!(/** test */));
            let v = use::snapshot(node.source_iter(q!(vec![1, 2, 3])).reduce(q!(|acc, v| *acc += v)), nondet!(/** test */));
            input.cross_singleton(v.into_stream().count())
        }
        .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (1, 1));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (2, 1));
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn top_level_bounded_into_singleton_cardinality() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);

        let out = sliced! {
            let input = use::batch(input, nondet!(/** test */));
            let v = use::snapshot(node.source_iter(q!(vec![1, 2, 3])).reduce(q!(|acc, v| *acc += v)).into_singleton(), nondet!(/** test */));
            input.cross_singleton(v.into_stream().count())
        }
        .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (1, 1));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (2, 1));
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn atomic_fold_replays_each_tick() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);
        let tick = node.tick();

        let out = input
            .batch(&tick, nondet!(/** test */))
            .cross_singleton(
                node.source_iter(q!(vec![1, 2, 3]))
                    .atomic()
                    .fold(q!(|| 0), q!(|acc, v| *acc += v))
                    .snapshot_atomic(&tick, nondet!(/** test */)),
            )
            .all_ticks()
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (1, 6));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (2, 6));
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unbounded_scan_remembers_state() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) = node.source_external_bincode(&external);
        let out = input
            .scan(
                q!(|| 0),
                q!(|acc, v| {
                    *acc += v;
                    Some(*acc)
                }),
            )
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 1);

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 3);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unbounded_enumerate_remembers_state() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) = node.source_external_bincode(&external);
        let out = input.enumerate().send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (0, 1));

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), (1, 2));
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unbounded_unique_remembers_state() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_port, input) =
            node.source_external_bincode::<_, _, TotalOrder, ExactlyOnce>(&external);
        let out = input.unique().send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut external_in = nodes.connect(input_port).await;
        let mut external_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        external_in.send(1).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 1);

        external_in.send(2).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 2);

        external_in.send(1).await.unwrap();
        external_in.send(3).await.unwrap();
        assert_eq!(external_out.next().await.unwrap(), 3);
    }

    #[cfg(feature = "sim")]
    #[test]
    #[should_panic]
    fn sim_batch_nondet_size() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, TotalOrder, _>();

        let tick = node.tick();
        let out_recv = input
            .batch(&tick, nondet!(/** test */))
            .count()
            .all_ticks()
            .sim_output();

        flow.sim().exhaustive(async || {
            in_send.send(());
            in_send.send(());
            in_send.send(());

            assert_eq!(out_recv.next().await, 3); // fails with nondet batching
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_batch_preserves_order() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();

        let tick = node.tick();
        let out_recv = input
            .batch(&tick, nondet!(/** test */))
            .all_ticks()
            .sim_output();

        flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);
            in_send.send(3);

            out_recv.assert_yields_only([1, 2, 3]).await;
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    #[should_panic]
    fn sim_batch_unordered_shuffles() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let tick = node.tick();
        let batch = input.batch(&tick, nondet!(/** test */));
        let out_recv = batch
            .clone()
            .min()
            .zip(batch.max())
            .all_ticks()
            .sim_output();

        flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3]);

            assert!(
                out_recv.collect::<Vec<_>>().await != vec![(1, 3), (2, 2)],
                "saw both (1, 3) and (2, 2), so batching must have shuffled the order"
            )
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_batch_unordered_shuffles_count() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let tick = node.tick();
        let batch = input.batch(&tick, nondet!(/** test */));
        let out_recv = batch.all_ticks().sim_output();

        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3, 4]);
            out_recv.assert_yields_only_unordered([1, 2, 3, 4]).await;
        });

        assert_eq!(
            instance_count,
            75 // ∑ (k=1 to 4) S(4,k) × k! = 75
        )
    }

    #[cfg(feature = "sim")]
    #[test]
    #[should_panic]
    fn sim_observe_order_batched() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let tick = node.tick();
        let batch = input.batch(&tick, nondet!(/** test */));
        let out_recv = batch
            .assume_ordering::<TotalOrder>(nondet!(/** test */))
            .all_ticks()
            .sim_output();

        flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3, 4]);
            out_recv.assert_yields_only([1, 2, 3, 4]).await; // fails with assume_ordering
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_observe_order_batched_count() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let tick = node.tick();
        let batch = input.batch(&tick, nondet!(/** test */));
        let out_recv = batch
            .assume_ordering::<TotalOrder>(nondet!(/** test */))
            .all_ticks()
            .sim_output();

        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3, 4]);
            let _ = out_recv.collect::<Vec<_>>().await;
        });

        assert_eq!(
            instance_count,
            192 // 4! * 2^{4 - 1}
        )
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_unordered_count_instance_count() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let tick = node.tick();
        let out_recv = input
            .count()
            .snapshot(&tick, nondet!(/** test */))
            .all_ticks()
            .sim_output();

        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3, 4]);
            assert_eq!(out_recv.collect::<Vec<_>>().await.last(), Some(&4));
        });

        assert_eq!(
            instance_count,
            16 // 2^4, { 0, 1, 2, 3 } can be a snapshot and 4 is always included
        )
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_top_level_assume_ordering() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let out_recv = input
            .assume_ordering::<TotalOrder>(nondet!(/** test */))
            .sim_output();

        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([1, 2, 3]);
            let mut out = out_recv.collect::<Vec<_>>().await;
            out.sort();
            assert_eq!(out, vec![1, 2, 3]);
        });

        assert_eq!(instance_count, 6)
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_top_level_assume_ordering_cycle_back() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let node2 = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let (complete_cycle_back, cycle_back) =
            node.forward_ref::<super::Stream<_, _, _, NoOrder>>();
        let ordered = input
            .merge_unordered(cycle_back)
            .assume_ordering::<TotalOrder>(nondet!(/** test */));
        complete_cycle_back.complete(
            ordered
                .clone()
                .map(q!(|v| v + 1))
                .filter(q!(|v| v % 2 == 1))
                .send(&node2, TCP.fail_stop().bincode())
                .send(&node, TCP.fail_stop().bincode()),
        );

        let out_recv = ordered.sim_output();

        let mut saw = false;
        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([0, 2]);
            let out = out_recv.collect::<Vec<_>>().await;

            if out.starts_with(&[0, 1, 2]) {
                saw = true;
            }
        });

        assert!(saw, "did not see an instance with 0, 1, 2 in order");
        assert_eq!(instance_count, 6);
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_top_level_assume_ordering_cycle_back_tick() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let node2 = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let (complete_cycle_back, cycle_back) =
            node.forward_ref::<super::Stream<_, _, _, NoOrder>>();
        let ordered = input
            .merge_unordered(cycle_back)
            .assume_ordering::<TotalOrder>(nondet!(/** test */));
        complete_cycle_back.complete(
            ordered
                .clone()
                .batch(&node.tick(), nondet!(/** test */))
                .all_ticks()
                .map(q!(|v| v + 1))
                .filter(q!(|v| v % 2 == 1))
                .send(&node2, TCP.fail_stop().bincode())
                .send(&node, TCP.fail_stop().bincode()),
        );

        let out_recv = ordered.sim_output();

        let mut saw = false;
        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([0, 2]);
            let out = out_recv.collect::<Vec<_>>().await;

            if out.starts_with(&[0, 1, 2]) {
                saw = true;
            }
        });

        assert!(saw, "did not see an instance with 0, 1, 2 in order");
        assert_eq!(instance_count, 58);
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_top_level_assume_ordering_multiple() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let node2 = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();
        let (_, input2) = node.sim_input::<_, NoOrder, _>();

        let (complete_cycle_back, cycle_back) =
            node.forward_ref::<super::Stream<_, _, _, NoOrder>>();
        let input1_ordered = input
            .clone()
            .merge_unordered(cycle_back)
            .assume_ordering::<TotalOrder>(nondet!(/** test */));
        let foo = input1_ordered
            .clone()
            .map(q!(|v| v + 3))
            .weaken_ordering::<NoOrder>()
            .merge_unordered(input2)
            .assume_ordering::<TotalOrder>(nondet!(/** test */));

        complete_cycle_back.complete(
            foo.filter(q!(|v| *v == 3))
                .send(&node2, TCP.fail_stop().bincode())
                .send(&node, TCP.fail_stop().bincode()),
        );

        let out_recv = input1_ordered.sim_output();

        let mut saw = false;
        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([0, 1]);
            let out = out_recv.collect::<Vec<_>>().await;

            if out.starts_with(&[0, 3, 1]) {
                saw = true;
            }
        });

        assert!(saw, "did not see an instance with 0, 3, 1 in order");
        assert_eq!(instance_count, 15);
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_atomic_assume_ordering_cycle_back() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let node2 = flow.process::<()>();

        let (in_send, input) = node.sim_input::<_, NoOrder, _>();

        let (complete_cycle_back, cycle_back) =
            node.forward_ref::<super::Stream<_, _, _, NoOrder>>();
        let ordered = input
            .merge_unordered(cycle_back)
            .atomic()
            .assume_ordering::<TotalOrder>(nondet!(/** test */))
            .end_atomic();
        complete_cycle_back.complete(
            ordered
                .clone()
                .map(q!(|v| v + 1))
                .filter(q!(|v| v % 2 == 1))
                .send(&node2, TCP.fail_stop().bincode())
                .send(&node, TCP.fail_stop().bincode()),
        );

        let out_recv = ordered.sim_output();

        let instance_count = flow.sim().exhaustive(async || {
            in_send.send_many_unordered([0, 2]);
            let out = out_recv.collect::<Vec<_>>().await;
            assert_eq!(out.len(), 4);
        });
        assert_eq!(instance_count, 22);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn partition_evens_odds() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let numbers = node.source_iter(q!(vec![1i32, 2, 3, 4, 5, 6]));
        let (evens, odds) = numbers.partition(q!(|x: &i32| x % 2 == 0));
        let evens_port = evens.send_bincode_external(&external);
        let odds_port = odds.send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut evens_out = nodes.connect(evens_port).await;
        let mut odds_out = nodes.connect(odds_port).await;

        deployment.start().await.unwrap();

        let mut even_results = Vec::new();
        for _ in 0..3 {
            even_results.push(evens_out.next().await.unwrap());
        }
        even_results.sort();
        assert_eq!(even_results, vec![2, 4, 6]);

        let mut odd_results = Vec::new();
        for _ in 0..3 {
            odd_results.push(odds_out.next().await.unwrap());
        }
        odd_results.sort();
        assert_eq!(odd_results, vec![1, 3, 5]);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unconsumed_inspect_still_runs() {
        use crate::deploy::DeployCrateWrapper;

        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        // The return value of .inspect() is intentionally dropped.
        // Before the Null-root fix, this would silently do nothing.
        node.source_iter(q!(0..5))
            .inspect(q!(|x| println!("inspect: {}", x)));

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut stdout = nodes.get_process(&node).stdout();

        deployment.start().await.unwrap();

        let mut lines = Vec::new();
        for _ in 0..5 {
            lines.push(stdout.recv().await.unwrap());
        }
        lines.sort();
        assert_eq!(
            lines,
            vec![
                "inspect: 0",
                "inspect: 1",
                "inspect: 2",
                "inspect: 3",
                "inspect: 4",
            ]
        );
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn unconsumed_inspect_alive_at_deploy_still_runs() {
        use crate::deploy::DeployCrateWrapper;

        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        // The return value of .inspect() is bound to a variable that is still alive
        // when the flow is finalized by `deploy` below, so its `Drop` runs too late
        // to register a root the usual way. The FlowBuilder must yank the IR from
        // still-live collections when finalizing.
        let _inspected = node
            .source_iter(q!(0..5))
            .inspect(q!(|x| println!("inspect: {}", x)));

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut stdout = nodes.get_process(&node).stdout();

        deployment.start().await.unwrap();

        let mut lines = Vec::new();
        for _ in 0..5 {
            lines.push(stdout.recv().await.unwrap());
        }
        lines.sort();
        assert_eq!(
            lines,
            vec![
                "inspect: 0",
                "inspect: 1",
                "inspect: 2",
                "inspect: 3",
                "inspect: 4",
            ]
        );
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_limit() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();

        let out_recv = input.limit(q!(3)).sim_output();

        flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);
            in_send.send(3);
            in_send.send(4);
            in_send.send(5);

            out_recv.assert_yields_only([1, 2, 3]).await;
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_limit_zero() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();

        let out_recv = input.limit(q!(0)).sim_output();

        flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);

            out_recv.assert_yields_only::<i32, _>([]).await;
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_merge_ordered() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();
        let (in_send2, input2) = node.sim_input();

        let out_recv = input
            .merge_ordered(input2, nondet!(/** test */))
            .sim_output();

        let mut saw_out_of_order = false;
        let instances = flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);
            in_send2.send(3);
            in_send2.send(4);

            let out = out_recv.collect::<Vec<_>>().await;

            if out == [1, 3, 2, 4] {
                saw_out_of_order = true;
            }

            // Assert ordering preservation: elements from each input must
            // appear in their original relative order.
            let mut first_elements = out.iter().filter(|v| **v <= 2).copied().collect::<Vec<_>>();
            let mut second_elements = out.iter().filter(|v| **v > 2).copied().collect::<Vec<_>>();
            assert_eq!(
                first_elements,
                vec![1, 2],
                "first input order violated: {:?}",
                out
            );
            assert_eq!(
                second_elements,
                vec![3, 4],
                "second input order violated: {:?}",
                out
            );

            first_elements.append(&mut second_elements);
            first_elements.sort();
            assert_eq!(first_elements, vec![1, 2, 3, 4]);
        });

        assert!(saw_out_of_order);
        assert_eq!(instances, 6);
    }

    /// Tests that `merge_ordered` passes through elements when only one input
    /// has data.
    #[cfg(feature = "sim")]
    #[test]
    fn sim_merge_ordered_one_empty() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();
        let (_in_send2, input2) = node.sim_input();

        let out_recv = input
            .merge_ordered(input2, nondet!(/** test */))
            .sim_output();

        let instances = flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);

            let out = out_recv.collect::<Vec<_>>().await;
            assert_eq!(out, vec![1, 2]);
        });

        // Only one possible interleaving when one input is empty
        assert_eq!(instances, 1);
    }

    /// Tests that `merge_ordered` correctly handles feedback cycles.
    /// An element output from `merge_ordered` is filtered and cycled back to
    /// one of its inputs. The one-at-a-time release must allow the cycled-back
    /// element to arrive and potentially be emitted before elements still
    /// waiting on the other input.
    #[cfg(feature = "sim")]
    #[test]
    fn sim_merge_ordered_cycle_back() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();

        // Create a forward ref for the cycle back
        let (complete_cycle_back, cycle_back) =
            node.forward_ref::<super::Stream<_, _, _, TotalOrder>>();

        // merge_ordered: input (external) with cycle_back
        let merged = input.merge_ordered(cycle_back, nondet!(/** test */));

        // Cycle back: elements equal to 1 get mapped to 10 and fed back
        complete_cycle_back.complete(merged.clone().filter(q!(|v| *v == 1)).map(q!(|v| v * 10)));

        let out_recv = merged.sim_output();

        // Send 1 and 2. Element 1 should cycle back as 10.
        // Valid orderings must have 1 before 10 (since 10 depends on 1).
        let mut saw_cycle_before_second = false;
        flow.sim().exhaustive(async || {
            in_send.send(1);
            in_send.send(2);

            let out = out_recv.collect::<Vec<_>>().await;

            // 10 must always come after 1 (causal dependency)
            let pos_1 = out.iter().position(|v| *v == 1).unwrap();
            let pos_10 = out.iter().position(|v| *v == 10).unwrap();
            assert!(pos_1 < pos_10, "causal order violated: {:?}", out);

            // Check if we see [1, 10, 2] — the cycled element beats the second input
            if out == [1, 10, 2] {
                saw_cycle_before_second = true;
            }

            let mut sorted = out;
            sorted.sort();
            assert_eq!(sorted, vec![1, 2, 10]);
        });

        assert!(
            saw_cycle_before_second,
            "never saw the cycled element arrive before the second input element"
        );
    }

    /// Tests that `merge_ordered` correctly interleaves when one input has a
    /// delayed element. With a: [1, _delay_, 2] and b: [3, 4], the delayed
    /// element 2 should be able to appear after b's elements.
    #[cfg(feature = "sim")]
    #[test]
    fn sim_merge_ordered_delayed() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in_send, input) = node.sim_input();
        let (in_send2, input2) = node.sim_input();

        let out_recv = input
            .merge_ordered(input2, nondet!(/** test */))
            .sim_output();

        let mut saw_delayed_interleaving = false;
        flow.sim().exhaustive(async || {
            // Send 1 from a, and 3, 4 from b
            in_send.send(1);
            in_send2.send(3);
            in_send2.send(4);

            // Collect what's available so far
            let first_batch = out_recv.collect::<Vec<_>>().await;

            // Now send the delayed element 2 from a
            in_send.send(2);
            let second_batch = out_recv.collect::<Vec<_>>().await;

            let mut all: Vec<_> = first_batch
                .iter()
                .chain(second_batch.iter())
                .copied()
                .collect();

            // Check if we saw [1, 3, 4, 2] — the delayed interleaving
            if all == [1, 3, 4, 2] {
                saw_delayed_interleaving = true;
            }

            all.sort();
            assert_eq!(all, vec![1, 2, 3, 4]);
        });

        assert!(saw_delayed_interleaving);
    }

    /// Deploy test: `merge_ordered` with a delayed element on one input.
    /// Sends a=1, b=3, b=4, then after receiving those, sends a=2.
    /// Expects to see [1, 3, 4] first, then [2] — demonstrating that
    /// both inputs are pulled and the delayed element arrives later.
    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn deploy_merge_ordered_delayed() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let (input_a_port, input_a) = node.source_external_bincode(&external);
        let (input_b_port, input_b) = node.source_external_bincode(&external);

        let out = input_a
            .assume_ordering(nondet!(/** test */))
            .merge_ordered(
                input_b.assume_ordering(nondet!(/** test */)),
                nondet!(/** test */),
            )
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut ext_a = nodes.connect(input_a_port).await;
        let mut ext_b = nodes.connect(input_b_port).await;
        let mut ext_out = nodes.connect(out).await;

        deployment.start().await.unwrap();

        // Send a=1, b=3, b=4
        ext_a.send(1).await.unwrap();
        ext_b.send(3).await.unwrap();
        ext_b.send(4).await.unwrap();

        // Collect the first 3 elements
        let mut received = Vec::new();
        for _ in 0..3 {
            received.push(ext_out.next().await.unwrap());
        }

        // Now send the delayed a=2
        ext_a.send(2).await.unwrap();
        received.push(ext_out.next().await.unwrap());

        // All elements should be present
        received.sort();
        assert_eq!(received, vec![1, 2, 3, 4]);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn monotone_fold_threshold() {
        use crate::properties::manual_proof;

        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let in_unbounded: super::Stream<_, _> =
            node.source_iter(q!(vec![1i32, 2, 3, 4, 5, 6])).into();
        let sum = in_unbounded.fold(
            q!(|| 0),
            q!(
                |sum, v| {
                    *sum += v;
                },
                monotone = manual_proof!(/** test */)
            ),
        );

        let threshold_out = sum
            .threshold_greater_or_equal(node.singleton(q!(7)))
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut threshold_out = nodes.connect(threshold_out).await;

        deployment.start().await.unwrap();

        assert_eq!(threshold_out.next().await.unwrap(), 7);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn monotone_count_threshold() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let in_unbounded: super::Stream<_, _> =
            node.source_iter(q!(vec![1i32, 2, 3, 4, 5, 6])).into();
        let sum = in_unbounded.count();

        let threshold_out = sum
            .threshold_greater_or_equal(node.singleton(q!(3)))
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut threshold_out = nodes.connect(threshold_out).await;

        deployment.start().await.unwrap();

        assert_eq!(threshold_out.next().await.unwrap(), 3);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn monotone_map_order_preserving_threshold() {
        use crate::properties::manual_proof;

        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();
        let external = flow.external::<()>();

        let in_unbounded: super::Stream<_, _> =
            node.source_iter(q!(vec![1i32, 2, 3, 4, 5, 6])).into();
        let sum = in_unbounded.fold(
            q!(|| 0),
            q!(
                |sum, v| {
                    *sum += v;
                },
                monotone = manual_proof!(/** test */)
            ),
        );

        // map with order_preserving should preserve monotonicity
        let doubled = sum.map(q!(
            |v| v * 2,
            order_preserving = manual_proof!(/** doubling preserves order */)
        ));

        let threshold_out = doubled
            .threshold_greater_or_equal(node.singleton(q!(14)))
            .send_bincode_external(&external);

        let nodes = flow
            .with_process(&node, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut threshold_out = nodes.connect(threshold_out).await;

        deployment.start().await.unwrap();

        assert_eq!(threshold_out.next().await.unwrap(), 14);
    }

    // === Compile-time type tests for join/cross_product ordering ===

    #[cfg(any(feature = "deploy", feature = "sim"))]
    mod join_ordering_type_tests {
        use crate::live_collections::boundedness::{Bounded, Unbounded};
        use crate::live_collections::stream::{ExactlyOnce, NoOrder, Stream, TotalOrder};
        use crate::location::{Location, Process};

        #[expect(dead_code, reason = "compile-time type test")]
        fn join_unbounded_with_bounded_preserves_order<'a>(
            left: Stream<(i32, char), Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
            right: Stream<(i32, char), Process<'a>, Bounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, (char, char)), Process<'a>, Unbounded, TotalOrder, ExactlyOnce> {
            left.join(right)
        }

        #[expect(dead_code, reason = "compile-time type test")]
        fn join_unbounded_with_unbounded_is_no_order<'a>(
            left: Stream<(i32, char), Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
            right: Stream<(i32, char), Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, (char, char)), Process<'a>, Unbounded, NoOrder, ExactlyOnce> {
            left.join(right)
        }

        #[expect(dead_code, reason = "compile-time type test")]
        fn join_bounded_with_bounded_preserves_order<'a, L: Location<'a>>(
            left: Stream<(i32, char), L, Bounded, TotalOrder, ExactlyOnce>,
            right: Stream<(i32, char), L, Bounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, (char, char)), L, Bounded, TotalOrder, ExactlyOnce> {
            left.join(right)
        }

        #[expect(dead_code, reason = "compile-time type test")]
        fn join_unbounded_noorder_with_bounded<'a>(
            left: Stream<(i32, char), Process<'a>, Unbounded, NoOrder, ExactlyOnce>,
            right: Stream<(i32, char), Process<'a>, Bounded, NoOrder, ExactlyOnce>,
        ) -> Stream<(i32, (char, char)), Process<'a>, Unbounded, NoOrder, ExactlyOnce> {
            left.join(right)
        }

        // === Compile-time type tests for cross_product ordering ===

        #[expect(dead_code, reason = "compile-time type test")]
        fn cross_product_unbounded_with_bounded_preserves_order<'a>(
            left: Stream<i32, Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
            right: Stream<char, Process<'a>, Bounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, char), Process<'a>, Unbounded, TotalOrder, ExactlyOnce> {
            left.cross_product(right)
        }

        #[expect(dead_code, reason = "compile-time type test")]
        fn cross_product_bounded_with_bounded_preserves_order<'a>(
            left: Stream<i32, Process<'a>, Bounded, TotalOrder, ExactlyOnce>,
            right: Stream<char, Process<'a>, Bounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, char), Process<'a>, Bounded, TotalOrder, ExactlyOnce> {
            left.cross_product(right)
        }

        #[expect(dead_code, reason = "compile-time type test")]
        fn cross_product_unbounded_with_unbounded_is_no_order<'a>(
            left: Stream<i32, Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
            right: Stream<char, Process<'a>, Unbounded, TotalOrder, ExactlyOnce>,
        ) -> Stream<(i32, char), Process<'a>, Unbounded, NoOrder, ExactlyOnce> {
            left.cross_product(right)
        }
    } // mod join_ordering_type_tests

    // === Runtime correctness tests for bounded join/cross_product ===

    #[cfg(feature = "sim")]
    #[test]
    fn cross_product_mixed_boundedness_correctness() {
        use stageleft::q;

        use crate::compile::builder::FlowBuilder;
        use crate::nondet::nondet;

        let mut flow = FlowBuilder::new();
        let process = flow.process::<()>();
        let tick = process.tick();

        let left = process.source_iter(q!(vec![1, 2]));
        let right = process
            .source_iter(q!(vec!['a', 'b']))
            .batch(&tick, nondet!(/** test */))
            .all_ticks();

        let out = left.cross_product(right).sim_output();

        flow.sim().exhaustive(async || {
            out.assert_yields_only_unordered(vec![(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')])
                .await;
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn join_mixed_boundedness_correctness() {
        use stageleft::q;

        use crate::compile::builder::FlowBuilder;
        use crate::nondet::nondet;

        let mut flow = FlowBuilder::new();
        let process = flow.process::<()>();
        let tick = process.tick();

        let left = process.source_iter(q!(vec![(1, 'a'), (2, 'b')]));
        let right = process
            .source_iter(q!(vec![(1, 'x'), (2, 'y')]))
            .batch(&tick, nondet!(/** test */))
            .all_ticks();

        let out = left.join(right).sim_output();

        flow.sim().exhaustive(async || {
            out.assert_yields_only_unordered(vec![(1, ('a', 'x')), (2, ('b', 'y'))])
                .await;
        });
    }

    #[cfg(feature = "sim")]
    #[test]
    fn sim_merge_unordered_independent_atomics() {
        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (in1_send, input1) = node.sim_input::<_, TotalOrder, _>();
        let (in2_send, input2) = node.sim_input::<_, TotalOrder, _>();

        let out = input1
            .atomic()
            .merge_unordered(input2.atomic())
            .end_atomic()
            .sim_output();

        flow.sim().exhaustive(async || {
            in1_send.send(1);
            in2_send.send(2);

            out.assert_yields_only_unordered(vec![1, 2]).await;
        });
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn test_stream_ref() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let external = flow.external::<()>();
        let p1 = flow.process::<()>();

        // Create a bounded stream (source_iter is bounded within a tick)
        let my_stream = p1.source_iter(q!(1..=5i32));

        let stream_ref = my_stream.by_ref();

        // Use the stream ref to get the vec's length
        let out_port = p1
            .source_iter(q!([()]))
            .map(q!(|_| stream_ref.len() as i32))
            .send_bincode_external(&external);

        // Also consume the stream via pipe
        my_stream.for_each(q!(|_| {}));

        let nodes = flow
            .with_default_optimize()
            .with_process(&p1, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut out_recv = nodes.connect(out_port).await;

        deployment.start().await.unwrap();

        let result = out_recv.next().await.unwrap();
        // stream has 5 elements
        assert_eq!(result, 5);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn test_stream_ref_contents() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let external = flow.external::<()>();
        let p1 = flow.process::<()>();

        // Create a bounded stream
        let my_stream = p1.source_iter(q!(1..=3i32));

        let stream_ref = my_stream.by_ref();

        // Sum the referenced vec's contents
        let out_port = p1
            .source_iter(q!([()]))
            .map(q!(|_| stream_ref.iter().sum::<i32>()))
            .send_bincode_external(&external);

        my_stream.for_each(q!(|_| {}));

        let nodes = flow
            .with_default_optimize()
            .with_process(&p1, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut out_recv = nodes.connect(out_port).await;

        deployment.start().await.unwrap();

        let result = out_recv.next().await.unwrap();
        // sum of 1+2+3 = 6
        assert_eq!(result, 6);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn test_stream_ref_no_consumer() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let external = flow.external::<()>();
        let p1 = flow.process::<()>();

        // Create a bounded stream — no pipe consumer, only ref
        let my_stream = p1.source_iter(q!(1..=4i32));

        let stream_ref = my_stream.by_ref();

        let out_port = p1
            .source_iter(q!([()]))
            .map(q!(|_| stream_ref.len() as i32))
            .send_bincode_external(&external);

        let nodes = flow
            .with_default_optimize()
            .with_process(&p1, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut out_recv = nodes.connect(out_port).await;

        deployment.start().await.unwrap();

        let result = out_recv.next().await.unwrap();
        assert_eq!(result, 4);
    }

    #[cfg(feature = "deploy")]
    #[tokio::test]
    async fn test_stream_mut() {
        let mut deployment = Deployment::new();

        let mut flow = FlowBuilder::new();
        let external = flow.external::<()>();
        let p1 = flow.process::<()>();

        // Create a bounded stream
        let my_stream = p1.source_iter(q!(1..=5i32));

        let stream_mut = my_stream.by_mut();

        // Mutably reference the buffer to retain only items > 3
        let out_port = p1
            .source_iter(q!([()]))
            .map(q!(|_| {
                stream_mut.retain(|x| *x > 3);
                stream_mut.len() as i32
            }))
            .send_bincode_external(&external);

        my_stream.for_each(q!(|_| {}));

        let nodes = flow
            .with_default_optimize()
            .with_process(&p1, deployment.Localhost())
            .with_external(&external, deployment.Localhost())
            .deploy(&mut deployment);

        deployment.deploy().await.unwrap();

        let mut out_recv = nodes.connect(out_port).await;

        deployment.start().await.unwrap();

        let result = out_recv.next().await.unwrap();
        // After retain(> 3): [4, 5] => len = 2
        assert_eq!(result, 2);
    }

    /// A map with a mut singleton ref on an unordered input should produce > 1
    /// simulation instance because the ordering of elements through the mut closure
    /// is non-deterministic.
    #[cfg(feature = "sim")]
    #[test]
    fn sim_map_with_mut_on_unordered_explores_multiple_states() {
        use crate::live_collections::sliced::sliced;
        use crate::live_collections::stream::ExactlyOnce;
        use crate::properties::manual_proof;

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (trigger_send, trigger) = node.sim_input::<i32, TotalOrder, ExactlyOnce>();

        let out_recv = sliced! {
            let batch = use::batch(trigger, nondet!(/** test */));
            let counter = batch.location().source_iter(q!(vec![0i32]))
                .fold(q!(|| 0i32), q!(|acc, v| *acc += v));
            let counter_mut = counter.by_mut();
            let items = batch.location().source_iter(q!(vec![1i32, 2])).weaken_ordering::<NoOrder>();
            items.map(q!(
                |x| {
                    *counter_mut += x;
                    *counter_mut
                },
                commutative = manual_proof!(/** test */)
            ))
        }
        .sim_output();

        let count = flow.sim().exhaustive(async || {
            trigger_send.send(1);
            let _all: Vec<i32> = out_recv.collect_sorted().await;
        });

        assert_eq!(
            count, 2,
            "Expected 2 simulation instances due to mut on unordered input, got {}",
            count
        );
    }

    /// A `scan` closure that captures a bounded singleton by reference should compile,
    /// run correctly, and (because the input is totally ordered) explore a single
    /// simulation instance.
    #[cfg(feature = "sim")]
    #[test]
    fn sim_scan_with_ref_capture() {
        use crate::live_collections::sliced::sliced;
        use crate::live_collections::stream::ExactlyOnce;

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let (trigger_send, trigger) = node.sim_input::<i32, TotalOrder, ExactlyOnce>();

        let out_recv = sliced! {
            let batch = use::batch(trigger, nondet!(/** test */));
            let offset = batch
                .location()
                .source_iter(q!(vec![10i32]))
                .fold(q!(|| 0i32), q!(|acc, v| *acc += v));
            let offset_ref = offset.by_ref();
            batch
                .location()
                .source_iter(q!(vec![1i32, 2, 3]))
                .scan(
                    q!(|| 0i32),
                    q!(move |acc: &mut i32, x| {
                        *acc += x + *offset_ref;
                        Some(*acc)
                    }),
                )
        }
        .sim_output();

        let count = flow.sim().exhaustive(async || {
            trigger_send.send(1);
            let all: Vec<i32> = out_recv.collect().await;
            // offset = 10, running accumulator starts at 0:
            //   x=1: acc += 1 + 10 = 11 -> 11
            //   x=2: acc += 2 + 10 = 12 -> 23
            //   x=3: acc += 3 + 10 = 13 -> 36
            assert_eq!(all, vec![11, 23, 36]);
        });

        assert_eq!(
            count, 1,
            "Expected a single simulation instance for a totally-ordered scan, got {}",
            count
        );
    }

    /// A map with a mut singleton ref on a top-level unordered input should produce > 1
    /// simulation instance. Currently panics because `observe_nondet` doesn't support
    /// top-level bounded inputs yet.
    #[cfg(feature = "sim")]
    #[test]
    #[ignore = "observe_nondet not yet supported for top-level bounded inputs (https://github.com/hydro-project/hydro/issues/2950)"]
    fn sim_map_with_mut_on_unordered_top_level() {
        use crate::properties::manual_proof;

        let mut flow = FlowBuilder::new();
        let node = flow.process::<()>();

        let counter = node
            .source_iter(q!(vec![0i32]))
            .fold(q!(|| 0i32), q!(|acc, v| *acc += v));
        let counter_mut = counter.by_mut();

        let out_recv = node
            .source_iter(q!(vec![1i32, 2]))
            .weaken_ordering::<NoOrder>()
            .map(q!(
                |x| {
                    *counter_mut += x;
                    *counter_mut
                },
                commutative = manual_proof!(/** test */)
            ))
            .assume_ordering::<TotalOrder>(nondet!(/** test */))
            .sim_output();

        counter.into_stream().for_each(q!(|_| {}));

        let count = flow.sim().exhaustive(async || {
            let _all: Vec<i32> = out_recv.collect().await;
        });

        assert_eq!(
            count, 2,
            "Expected 2 simulation instances due to mut on unordered input, got {}",
            count
        );
    }
}