hydro_lang/properties/

mod.rs

//! Types for reasoning about algebraic properties for Rust closures.

use std::marker::PhantomData;

use stageleft::properties::Property;

use crate::live_collections::boundedness::Boundedness;
use crate::live_collections::keyed_singleton::KeyedSingletonBound;
use crate::live_collections::singleton::SingletonBound;
use crate::live_collections::stream::{ExactlyOnce, Ordering, Retries, TotalOrder};
use crate::sim_hooks::{OnProcess, OrderingHook};

/// A trait for proof mechanisms that can validate commutativity.
///
/// `T` and `B` name the element type and boundedness of the stream the commutative
/// function consumes. The simulator does not trust commutativity proofs — it still
/// explores the input ordering — so a proof may carry an [`OrderingHook`] for scripting
/// that exploration, surfaced through [`Self::take_hook`]. `S` is the hook's
/// [scope](crate::sim_hooks#hook-scopes).
#[sealed::sealed]
pub trait CommutativeProof<T, B: Boundedness, S = OnProcess> {
    /// Registers the expression with the proof mechanism.
    ///
    /// This should not perform any blocking analysis; it is only intended to record the expression for later processing.
    fn register_proof(&self, expr: &syn::Expr);

    /// Takes the simulator ordering hook attached to this proof, if any.
    fn take_hook(&mut self) -> Option<OrderingHook<T, B, ExactlyOnce, S>>;
}

/// A trait for proof mechanisms that can validate idempotence.
#[sealed::sealed]
pub trait IdempotentProof {
    /// Registers the expression with the proof mechanism.
    ///
    /// This should not perform any blocking analysis; it is only intended to record the expression for later processing.
    fn register_proof(&self, expr: &syn::Expr);
}

/// A trait for proof mechanisms that can validate monotonicity.
#[sealed::sealed]
pub trait MonotoneProof {
    /// Registers the expression with the proof mechanism.
    ///
    /// This should not perform any blocking analysis; it is only intended to record the expression for later processing.
    fn register_proof(&self, expr: &syn::Expr);
}

/// A trait for proof mechanisms that can validate order-preservation (monotonicity of a map function).
#[sealed::sealed]
pub trait OrderPreservingProof {
    /// Registers the expression with the proof mechanism.
    ///
    /// This should not perform any blocking analysis; it is only intended to record the expression for later processing.
    fn register_proof(&self, expr: &syn::Expr);
}

/// A trait for proof mechanisms that can validate consistency of a collection.
#[sealed::sealed]
pub trait ConsistencyProof {}

/// A hand-written human proof of the correctness property.
///
/// To create a manual proof, use the [`manual_proof!`] macro, which takes in a doc comment
/// explaining why the property holds.
///
/// Manual proofs are not trusted by the simulator, which still explores the guarded
/// non-determinism. `H` is the simulator hook payload (like [`crate::nondet::NonDet`]) so a
/// commutativity proof can carry an ordering hook for scripting that exploration.
pub struct ManualProof<H = ()> {
    hook: H,
}

impl<H> ManualProof<H> {
    #[doc(hidden)]
    pub fn unhooked() -> Self
    where
        H: Default,
    {
        ManualProof { hook: H::default() }
    }
}

impl<T, B: Boundedness, S> ManualProof<Option<OrderingHook<T, B, ExactlyOnce, S>>> {
    #[doc(hidden)]
    pub fn hooked(hook: impl Into<Option<OrderingHook<T, B, ExactlyOnce, S>>>) -> Self {
        ManualProof { hook: hook.into() }
    }
}

#[sealed::sealed]
impl<T, B: Boundedness, S> CommutativeProof<T, B, S>
    for ManualProof<Option<OrderingHook<T, B, ExactlyOnce, S>>>
{
    fn register_proof(&self, _expr: &syn::Expr) {}

    fn take_hook(&mut self) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        self.hook.take()
    }
}

#[sealed::sealed]
impl<T, B: Boundedness, S> CommutativeProof<T, B, S> for ManualProof {
    fn register_proof(&self, _expr: &syn::Expr) {}

    fn take_hook(&mut self) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        None
    }
}
#[sealed::sealed]
impl IdempotentProof for ManualProof {
    fn register_proof(&self, _expr: &syn::Expr) {}
}
#[sealed::sealed]
impl MonotoneProof for ManualProof {
    fn register_proof(&self, _expr: &syn::Expr) {}
}
#[sealed::sealed]
impl OrderPreservingProof for ManualProof {
    fn register_proof(&self, _expr: &syn::Expr) {}
}
#[sealed::sealed]
impl ConsistencyProof for ManualProof {}

/// A machine-checked proof of **commutativity**, verified by [Verus](https://verus-lang.github.io/verus/guide/).
///
/// Created by the [`verus_proof_commutative_fold!`], [`verus_proof_commutative_map!`],
/// [`verus_proof_commutative_filter!`], and [`verus_proof_commutative_effect!`] macros,
/// one per closure shape, each generating the precise obligation that makes reordering
/// unobservable for that shape (final accumulator, output multiset, retained multiset,
/// or captured state, respectively). The obligation is generated by the macro directly
/// from the *actual closure body* — it symbolically executes the body in both orders
/// (`x` then `y`, and `y` then `x`) from equal initial states and requires the
/// observable results to be equal — so users never write (and cannot weaken) the
/// assertion. Users only declare the types involved and, optionally, provide a proof
/// *script* to help the SMT solver, which is itself checked by Verus.
///
/// When the crate is verified with `cargo verus verify`, Verus checks the obligation
/// (which also proves the closure body panic-free, e.g. no arithmetic overflow). Under
/// normal compilation, the proof is erased and this type simply marks the property as
/// proven. This type only implements [`CommutativeProof`], so it cannot be used to
/// fulfill a different property (e.g. `idempotent = ...`).
pub struct VerusCommutativeProof {
    _private: (),
}

impl VerusCommutativeProof {
    #[doc(hidden)]
    pub fn new() -> Self {
        VerusCommutativeProof { _private: () }
    }
}

impl Default for VerusCommutativeProof {
    fn default() -> Self {
        Self::new()
    }
}

#[sealed::sealed]
impl<T, B: Boundedness, S> CommutativeProof<T, B, S> for VerusCommutativeProof {
    fn register_proof(&self, _expr: &syn::Expr) {}

    fn take_hook(&mut self) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        // Verus proofs are still not trusted by the simulator, which explores the
        // input ordering on its own; no scripting hook is attached.
        None
    }
}

#[doc(inline)]
pub use crate::__manual_proof__ as manual_proof;

// Verus only sees items declared inside `verus!`, so under the Verus driver (with the
// `verus` feature) the function backing `verus_panic!` is declared there.
// `external_body`: Verus trusts the signature without checking the body (which calls
// `panic!`, whose vstd specification is `requires false`). The signature has no
// precondition and the default unwind specification (may unwind), so callers need not
// prove anything, and since the return type is `!`, nothing after the call is reachable.
#[cfg(all(verus_keep_ghost, feature = "verus"))]
vstd::prelude::verus! {
    #[doc(hidden)]
    #[verifier::external_body]
    pub fn __verus_panic() -> ! {
        panic!("verus_panic! was reached")
    }
}

#[doc(inline)]
pub use crate::__verus_panic__ as verus_panic;

#[cfg(all(verus_keep_ghost, feature = "verus"))]
#[doc(hidden)]
#[macro_export]
macro_rules! __verus_panic__ {
    // The message arguments are dropped: Verus does not support `format_args!`, and they
    // would only be evaluated on a path that panics anyway.
    ($($arg:tt)*) => {
        $crate::properties::__verus_panic()
    };
}

#[cfg(not(all(verus_keep_ghost, feature = "verus")))]
#[macro_export]
/// Panics, in a way that **Verus proofs of algebraic properties treat as an acceptable
/// outcome**. Takes the same arguments as [`panic!`].
///
/// A Verus proof (e.g. [`verus_proof_commutative_fold!`]) normally requires the closure body
/// to be panic-free for every input, which rejects closures such as `*acc += x` (which
/// panics on overflow) even though they are commutative whenever they complete. Calling
/// `verus_panic!(...)` instead tells Verus that this path ends the program: the property
/// only has to hold for executions that do **not** reach it. That matches Hydro's
/// requirement, since a panicking closure crashes the process rather than producing a
/// result.
///
/// The typical use is to guard an operation that could otherwise panic, so that Verus
/// knows it cannot fail on the paths that continue:
///
/// ```rust,ignore
/// q!(
///     |acc, x| {
///         if *acc > u32::MAX - x {
///             verus_panic!("sum overflowed");
///         }
///         *acc += x; // cannot overflow here, so Verus accepts it
///     },
///     commutative = verus_proof_commutative_fold!(acc = u32, item = u32)
/// )
/// ```
///
/// This is sound because the guard really executes: the closure panics at runtime whenever
/// the guard condition holds, so a guard that is stricter than necessary is still correct.
/// A guard that is too weak is caught by Verus, which reports the operation that may still
/// panic. Unlike `assume(...)` in a proof script, nothing is taken on faith beyond the fact
/// that `verus_panic!` never returns.
///
/// Under normal compilation, this forwards its arguments to [`panic!`]. When verifying
/// with `cargo verus verify`, the `verus` feature of `hydro_lang` must be enabled; Verus
/// then sees a call to a function that never returns (the message arguments are not
/// evaluated by the proof).
macro_rules! __verus_panic__ {
    ($($arg:tt)*) => {
        ::core::panic!($($arg)*)
    };
}

#[macro_export]
/// Fulfills a proof parameter by declaring a human-written justification for why
/// the algebraic property (e.g. commutativity, idempotence) holds.
///
/// The argument must be a doc comment explaining why the property is satisfied.
///
/// # Examples
/// ```rust
/// # #[cfg(feature = "deploy")] {
/// # use hydro_lang::prelude::*;
/// # use hydro_lang::live_collections::stream::NoOrder;
/// # use futures::StreamExt;
/// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
/// # let stream = process.source_iter(q!(vec![1, 2, 3])).weaken_ordering::<NoOrder>();
/// // stream: [1, 2, 3] (unordered)
/// stream
///     .fold(
///         q!(|| 0),
///         q!(
///             |acc, x| *acc += x,
///             commutative = manual_proof!(/** integer addition is commutative */)
///         ),
///     )
///     .into_stream()
/// # }, |mut stream| async move {
/// # assert_eq!(stream.next().await.unwrap(), 6);
/// # }));
/// # }
/// ```
/// An optional trailing `hook = ...` argument attaches a **simulator ordering hook** to a
/// commutativity proof (see `hydro_lang::sim::hooks`). The simulator does not trust manual
/// proofs — it still explores the input ordering — so the hook lets a simulation test
/// script that exploration:
///
/// ```rust,ignore
/// commutative = manual_proof!(/** set insert is commutative */ hook = my_ordering_hook)
/// ```
macro_rules! __manual_proof__ {
    (
        $(#[doc = $doc:expr])+hook =
        $hook:expr,__target =
        { $($target:tt)* } $(,__captures = [$($captures:ident),*])? $(,)?
    ) => {
        $crate::properties::ManualProof::hooked($hook)
    };
    ($(#[doc = $doc:expr])+hook = $hook:expr $(,)?) => {
        $crate::properties::ManualProof::hooked($hook)
    };
    (
        $(#[doc = $doc:expr])+,__target =
        { $($target:tt)* } $(,__captures = [$($captures:ident),*])? $(,)?
    ) => {
        $crate::properties::ManualProof::<()>::unhooked()
    };
    ($(#[doc = $doc:expr])+) => {
        $crate::properties::ManualProof::<()>::unhooked()
    };
}

#[doc(inline)]
pub use crate::__verus_proof_commutative_effect__ as verus_proof_commutative_effect;
#[doc(inline)]
pub use crate::__verus_proof_commutative_filter__ as verus_proof_commutative_filter;
#[doc(inline)]
pub use crate::__verus_proof_commutative_fold__ as verus_proof_commutative_fold;
#[doc(inline)]
pub use crate::__verus_proof_commutative_map__ as verus_proof_commutative_map;

#[macro_export]
/// Fulfills a `commutative = ...` proof parameter for an **aggregation closure**
/// (`fold` / `reduce`, shape `|acc, item|` with `acc: &mut A`) with a **Verus-checked
/// proof of commutativity**.
///
/// Users never write the proof obligation (so they cannot get it wrong): this macro
/// receives the quoted closure itself from `q!` (as a trailing `__target = { ... }`
/// argument) and generates a Verus function that symbolically executes the *actual
/// closure body* in both orders — `x` then `y`, and `y` then `x` — starting from equal
/// accumulators, and asserts that the final accumulators are equal. Verifying this
/// obligation also proves the closure body panic-free (e.g. no arithmetic overflow).
/// Users only declare the accumulator and item types:
///
/// ```rust,ignore
/// batch.reduce(q!(
///     |curr, new| { if new > *curr { *curr = new; } },
///     commutative = verus_proof_commutative_fold!(acc = u32, item = u32)
/// ));
/// ```
///
/// If the closure captures (read-only) variables from its environment, re-declare them
/// (with their runtime types) in a `captures = |...|` clause; the obligation is then
/// universally quantified over the capture values, which is sound since any particular
/// execution uses some fixed value. `q!` also passes the closure's capture list (as a
/// trailing `__captures = [...]` argument), and the macro checks at compile time that
/// every capture is declared.
///
/// An optional `proof = |state, x, y| { ... }` clause supplies a proof *script* to help
/// the SMT solver, with ghost bindings for the initial accumulator and the two items
/// (e.g. `proof = |s, x, y| { assert((s | x) | y == (s | y) | x) by (bit_vector); }`).
/// The script is checked by Verus and cannot weaken the obligation (though, as anywhere
/// in Verus, `assume` is an explicit soundness escape hatch).
///
/// The proof is gated on `cfg(verus_keep_ghost)`, so it only exists when the crate is
/// compiled by the Verus driver (`cargo verus verify`). Under normal compilation the
/// macro just produces a [`VerusCommutativeProof`] marker, so the types go through the
/// ordinary `commutative = ...` mechanism with zero overhead.
macro_rules! __verus_proof_commutative_fold__ {
    (
        acc = $accty:ty, item = $itemty:ty
        $(, captures = |$($cap:ident : $capty:ty),* $(,)?|)?
        $(, proof = |$pstate:ident, $px:ident, $py:ident| { $($proof_body:tt)* })?
        , __target = { $(move)? |$tacc:ident $(: $taccty:ty)?, $titem:ident $(: $titemty:ty)?| $target_body:expr }
        , __captures = [$($fv:ident),* $(,)?] $(,)?
    ) => {{
        /// Checks that every free variable captured by the closure is declared (with its
        /// type) in the `captures = |...|` clause: within this function, the *only* names
        /// in scope are the declared captures and the closure parameters, so an
        /// undeclared capture fails to resolve.
        #[allow(unused, reason = "only checks name resolution")]
        fn __hydro_verus_captures_declared($($($cap: $capty),*)?) {
            $(let $fv = &$fv;)*
        }
        #[cfg(verus_keep_ghost)]
        const _: () = {
            #[allow(unused_imports)]
            use ::vstd::prelude::*;

            ::vstd::prelude::verus! {
                /// Commutativity obligation, generated by `verus_proof_commutative_fold!`:
                /// applying the closure body with `x` then `y` must yield the same
                /// accumulator as applying it with `y` then `x`, from any equal starting
                /// accumulators, for all items and capture values. Verifying this also
                /// proves the body panic-free.
                fn __hydro_verus_commutative(
                    $($($cap: $capty,)*)?
                    __hydro_acc_xy: $accty,
                    __hydro_acc_yx: $accty,
                    __hydro_x_0: $itemty,
                    __hydro_x_1: $itemty,
                    __hydro_y_0: $itemty,
                    __hydro_y_1: $itemty,
                )
                    requires
                        __hydro_acc_xy == __hydro_acc_yx,
                        __hydro_x_0 == __hydro_x_1,
                        __hydro_y_0 == __hydro_y_1,
                {
                    $(
                        let ghost $pstate = __hydro_acc_xy;
                        let ghost $px = __hydro_x_0;
                        let ghost $py = __hydro_y_0;
                    )?
                    let mut __hydro_acc_xy = __hydro_acc_xy;
                    let mut __hydro_acc_yx = __hydro_acc_yx;
                    { let $tacc = &mut __hydro_acc_xy; let $titem = __hydro_x_0; let _ = { $target_body }; }
                    { let $tacc = &mut __hydro_acc_xy; let $titem = __hydro_y_0; let _ = { $target_body }; }
                    { let $tacc = &mut __hydro_acc_yx; let $titem = __hydro_y_1; let _ = { $target_body }; }
                    { let $tacc = &mut __hydro_acc_yx; let $titem = __hydro_x_1; let _ = { $target_body }; }
                    $(proof { $($proof_body)* })?
                    assert(__hydro_acc_xy == __hydro_acc_yx);
                }
            }
        };

        $crate::properties::VerusCommutativeProof::new()
    }};
    (
        $($rest:tt)*
    ) => {{
        ::core::compile_error!(
            "verus_proof_commutative_fold! must be used as a property annotation inside `q!(...)` on a `|acc, item| ...` closure, e.g. `commutative = verus_proof_commutative_fold!(acc = u32, item = u32)`"
        );
        $crate::properties::VerusCommutativeProof::new()
    }};
}

#[macro_export]
/// Fulfills a `commutative = ...` proof parameter for a **map-like closure**
/// (shape `|item| -> out`, e.g. for `map`) that may mutate a captured singleton
/// reference (from [`Singleton::by_mut`]), with a **Verus-checked proof of
/// commutativity**.
///
/// **What commutativity means for a map closure:** processing any two items in either
/// order, starting from the same captured state, must (1) leave the captured state in
/// the same final value, *and* (2) produce the same **multiset of output values** (the
/// outputs may be swapped in order, but not changed). Both conditions are required for
/// downstream determinism: outputs that depend on the processing order (e.g. emitting a
/// running total) are **not** commutative even if the state update is.
///
/// The commuting state is the mutable capture, declared (with the type behind the
/// reference) in the required `captures_mut = |state: S|` clause; additional read-only
/// captures can be declared in a `captures = |...|` clause. `q!` also passes the
/// closure's capture list (as a trailing `__captures = [...]` argument), and the macro
/// checks at compile time that every capture is declared. The `item = ...` type must be
/// written exactly as the closure receives it.
///
/// Users never write the proof obligation: this macro receives the quoted closure itself
/// from `q!` (as a trailing `__target = { ... }` argument) and generates a Verus function
/// that symbolically executes the *actual closure body* in both orders — `x` then `y`,
/// and `y` then `x` — from equal captured states, then asserts that the final states are
/// equal and that the two runs' outputs are equal as a multiset (pairwise or crossed).
/// Verifying this also proves the body panic-free.
///
/// An optional `proof = |state, x, y| { ... }` clause supplies a proof *script* to help
/// the SMT solver, with ghost bindings for the initial captured state and the two items.
/// The script is checked by Verus and cannot weaken the obligation.
///
/// The proof is gated on `cfg(verus_keep_ghost)`, so it only exists when the crate is
/// compiled by the Verus driver (`cargo verus verify`). Under normal compilation the
/// macro just produces a [`VerusCommutativeProof`] marker.
///
/// # Example
/// ```rust,ignore
/// let count_mut = my_count.by_mut();
/// stream.map(q!(
///     |x| {
///         *count_mut = count_mut.wrapping_add(x);
///         x // outputting `*count_mut` here would NOT be commutative!
///     },
///     commutative = verus_proof_commutative_map!(
///         item = i32,
///         captures_mut = |count_mut: i32|
///     )
/// ));
/// ```
///
/// [`Singleton::by_mut`]: crate::live_collections::singleton::Singleton::by_mut
macro_rules! __verus_proof_commutative_map__ {
    (
        item = $itemty:ty
        $(, captures = |$($cap:ident : $capty:ty),* $(,)?|)?
        , captures_mut = |$mcap:ident : $mcapty:ty $(,)?|
        $(, proof = |$pstate:ident, $px:ident, $py:ident| { $($proof_body:tt)* })?
        , __target = { $(move)? |$titem:ident $(: $titemty:ty)?| $target_body:expr }
        , __captures = [$($fv:ident),* $(,)?] $(,)?
    ) => {{
        /// Checks that every free variable captured by the closure is declared (with its
        /// type) in the `captures = |...|` or `captures_mut = |...|` clauses: within this
        /// function, the *only* names in scope are the declared captures, so an
        /// undeclared capture fails to resolve.
        #[allow(unused, reason = "only checks name resolution")]
        fn __hydro_verus_captures_declared($($($cap: $capty,)*)? $mcap: $mcapty) {
            $(let $fv = &$fv;)*
        }

        #[cfg(verus_keep_ghost)]
        const _: () = {
            #[allow(unused_imports)]
            use ::vstd::prelude::*;

            ::vstd::prelude::verus! {
                /// Commutativity obligation, generated by `verus_proof_commutative_map!`:
                /// from any equal starting captured states, processing `x` then `y` must
                /// yield the same final captured state as processing `y` then `x`, and
                /// the two runs must produce the same multiset of outputs (pairwise or
                /// crossed equality). Verifying this also proves the body panic-free.
                fn __hydro_verus_commutative(
                    $($($cap: $capty,)*)?
                    __hydro_state_xy: $mcapty,
                    __hydro_state_yx: $mcapty,
                    __hydro_x_0: $itemty,
                    __hydro_x_1: $itemty,
                    __hydro_y_0: $itemty,
                    __hydro_y_1: $itemty,
                )
                    requires
                        __hydro_state_xy == __hydro_state_yx,
                        __hydro_x_0 == __hydro_x_1,
                        __hydro_y_0 == __hydro_y_1,
                {
                    $(
                        let ghost $pstate = __hydro_state_xy;
                        let ghost $px = __hydro_x_0;
                        let ghost $py = __hydro_y_0;
                    )?
                    let mut __hydro_state_xy = __hydro_state_xy;
                    let mut __hydro_state_yx = __hydro_state_yx;
                    let __hydro_out_xy_x = { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_x_0; $target_body };
                    let __hydro_out_xy_y = { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_y_0; $target_body };
                    let __hydro_out_yx_y = { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_y_1; $target_body };
                    let __hydro_out_yx_x = { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_x_1; $target_body };
                    $(proof { $($proof_body)* })?
                    assert(__hydro_state_xy == __hydro_state_yx);
                    assert(
                        (__hydro_out_xy_x == __hydro_out_yx_x && __hydro_out_xy_y == __hydro_out_yx_y)
                        || (__hydro_out_xy_x == __hydro_out_yx_y && __hydro_out_xy_y == __hydro_out_yx_x)
                    );
                }
            }
        };

        $crate::properties::VerusCommutativeProof::new()
    }};
    (
        $($rest:tt)*
    ) => {{
        ::core::compile_error!(
            "verus_proof_commutative_map! must be used as a property annotation inside `q!(...)` on a `|item| ...` closure with exactly one `captures_mut` entry, e.g. `commutative = verus_proof_commutative_map!(item = u32, captures_mut = |state: u32|)`"
        );
        $crate::properties::VerusCommutativeProof::new()
    }};
}

#[macro_export]
/// Fulfills a `commutative = ...` proof parameter for a **filter predicate**
/// (shape `|item| -> bool` where `item` is received by reference) that may mutate a
/// captured singleton reference (from [`Singleton::by_mut`]), with a **Verus-checked
/// proof of commutativity**.
///
/// **What commutativity means for a filter predicate:** processing any two items in
/// either order, starting from the same captured state, must (1) leave the captured
/// state in the same final value, *and* (2) retain the same **multiset of elements**.
/// The second condition is essential for soundness: a stateful predicate like a rate
/// limiter converges to the same state either way, but *which* element passes depends on
/// the order, which is **not** commutative (unless the elements are equal).
///
/// The commuting state is the mutable capture, declared (with the type behind the
/// reference) in the required `captures_mut = |state: S|` clause; additional read-only
/// captures can be declared in a `captures = |...|` clause. `q!` also passes the
/// closure's capture list (as a trailing `__captures = [...]` argument), and the macro
/// checks at compile time that every capture is declared. The `item = ...` type must be
/// written exactly as the closure receives it (for `filter`, a reference like `&u32`).
///
/// Users never write the proof obligation: this macro receives the quoted closure itself
/// from `q!` (as a trailing `__target = { ... }` argument) and generates a Verus function
/// that symbolically executes the *actual predicate body* in both orders — `x` then `y`,
/// and `y` then `x` — from equal captured states, then asserts that the final states are
/// equal and that the retained multisets are equal: either the per-item decisions match
/// across the two orders, or the two items are equal and the number of retained copies
/// matches. Verifying this also proves the body panic-free.
///
/// An optional `proof = |state, x, y| { ... }` clause supplies a proof *script* to help
/// the SMT solver, with ghost bindings for the initial captured state and the two items.
/// The script is checked by Verus and cannot weaken the obligation.
///
/// The proof is gated on `cfg(verus_keep_ghost)`, so it only exists when the crate is
/// compiled by the Verus driver (`cargo verus verify`). Under normal compilation the
/// macro just produces a [`VerusCommutativeProof`] marker.
///
/// # Example
/// ```rust,ignore
/// let seen_mut = seen_count.by_mut();
/// stream.filter(q!(
///     |x| {
///         *seen_mut = seen_mut.wrapping_add(1);
///         *x > 1 // the decision must not depend on the mutable state!
///     },
///     commutative = verus_proof_commutative_filter!(
///         item = &u32,
///         captures_mut = |seen_mut: u32|
///     )
/// ));
/// ```
///
/// [`Singleton::by_mut`]: crate::live_collections::singleton::Singleton::by_mut
macro_rules! __verus_proof_commutative_filter__ {
    (
        item = $itemty:ty
        $(, captures = |$($cap:ident : $capty:ty),* $(,)?|)?
        , captures_mut = |$mcap:ident : $mcapty:ty $(,)?|
        $(, proof = |$pstate:ident, $px:ident, $py:ident| { $($proof_body:tt)* })?
        , __target = { $(move)? |$titem:ident $(: $titemty:ty)?| $target_body:expr }
        , __captures = [$($fv:ident),* $(,)?] $(,)?
    ) => {{
        /// Checks that every free variable captured by the closure is declared (with its
        /// type) in the `captures = |...|` or `captures_mut = |...|` clauses: within this
        /// function, the *only* names in scope are the declared captures, so an
        /// undeclared capture fails to resolve.
        #[allow(unused, reason = "only checks name resolution")]
        fn __hydro_verus_captures_declared($($($cap: $capty,)*)? $mcap: $mcapty) {
            $(let $fv = &$fv;)*
        }

        #[cfg(verus_keep_ghost)]
        const _: () = {
            #[allow(unused_imports)]
            use ::vstd::prelude::*;

            ::vstd::prelude::verus! {
                /// Commutativity obligation, generated by
                /// `verus_proof_commutative_filter!`: from any equal starting captured
                /// states, processing `x` then `y` must yield the same final captured
                /// state as processing `y` then `x`, and the retained multisets must be
                /// equal: either the per-item decisions match across the two orders, or
                /// the two items are equal and the retained counts match. Verifying this
                /// also proves the body panic-free.
                fn __hydro_verus_commutative(
                    $($($cap: $capty,)*)?
                    __hydro_state_xy: $mcapty,
                    __hydro_state_yx: $mcapty,
                    __hydro_x_0: $itemty,
                    __hydro_x_1: $itemty,
                    __hydro_y_0: $itemty,
                    __hydro_y_1: $itemty,
                )
                    requires
                        __hydro_state_xy == __hydro_state_yx,
                        __hydro_x_0 == __hydro_x_1,
                        __hydro_y_0 == __hydro_y_1,
                {
                    $(
                        let ghost $pstate = __hydro_state_xy;
                        let ghost $px = __hydro_x_0;
                        let ghost $py = __hydro_y_0;
                    )?
                    let mut __hydro_state_xy = __hydro_state_xy;
                    let mut __hydro_state_yx = __hydro_state_yx;
                    let __hydro_keep_xy_x: bool = { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_x_0; $target_body };
                    let __hydro_keep_xy_y: bool = { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_y_0; $target_body };
                    let __hydro_keep_yx_y: bool = { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_y_1; $target_body };
                    let __hydro_keep_yx_x: bool = { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_x_1; $target_body };
                    $(proof { $($proof_body)* })?
                    assert(__hydro_state_xy == __hydro_state_yx);
                    assert(
                        (__hydro_keep_xy_x == __hydro_keep_yx_x && __hydro_keep_xy_y == __hydro_keep_yx_y)
                        || (__hydro_x_0 == __hydro_y_0
                            && (__hydro_keep_xy_x as int) + (__hydro_keep_xy_y as int)
                                == (__hydro_keep_yx_x as int) + (__hydro_keep_yx_y as int))
                    );
                }
            }
        };

        $crate::properties::VerusCommutativeProof::new()
    }};
    (
        $($rest:tt)*
    ) => {{
        ::core::compile_error!(
            "verus_proof_commutative_filter! must be used as a property annotation inside `q!(...)` on a `|item| -> bool` closure with exactly one `captures_mut` entry, e.g. `commutative = verus_proof_commutative_filter!(item = &u32, captures_mut = |state: u32|)`"
        );
        $crate::properties::VerusCommutativeProof::new()
    }};
}

#[macro_export]
/// Fulfills a `commutative = ...` proof parameter for a **unit-returning, effectful
/// closure** (shape `|item| -> ()`, e.g. for `for_each` or `inspect`) that mutates a
/// captured singleton reference (from [`Singleton::by_mut`]), with a **Verus-checked
/// proof of commutativity** of the captured-state update.
///
/// **What commutativity means for an effectful closure:** processing any two items in
/// either order, starting from the same captured state, must leave the captured state in
/// the same final value. Because the closure returns `()` (enforced by this macro; use
/// [`verus_proof_commutative_map!`] or [`verus_proof_commutative_filter!`] for closures
/// whose return value is observable), the state is the only observable effect.
///
/// The commuting state is the mutable capture, declared (with the type behind the
/// reference) in the required `captures_mut = |state: S|` clause; additional read-only
/// captures can be declared in a `captures = |...|` clause. `q!` also passes the
/// closure's capture list (as a trailing `__captures = [...]` argument), and the macro
/// checks at compile time that every capture is declared. The `item = ...` type must be
/// written exactly as the closure receives it (e.g. `&u32` for `inspect`).
///
/// Users never write the proof obligation: this macro receives the quoted closure itself
/// from `q!` (as a trailing `__target = { ... }` argument) and generates a Verus function
/// that symbolically executes the *actual closure body* in both orders — `x` then `y`,
/// and `y` then `x` — from equal captured states, and asserts that the final states are
/// equal. Verifying this also proves the body panic-free.
///
/// An optional `proof = |state, x, y| { ... }` clause supplies a proof *script* to help
/// the SMT solver, with ghost bindings for the initial captured state and the two items.
/// The script is checked by Verus and cannot weaken the obligation.
///
/// The proof is gated on `cfg(verus_keep_ghost)`, so it only exists when the crate is
/// compiled by the Verus driver (`cargo verus verify`). Under normal compilation the
/// macro just produces a [`VerusCommutativeProof`] marker.
///
/// # Example
/// ```rust,ignore
/// let flags_mut = flags.by_mut();
/// stream.for_each(q!(
///     |x| { *flags_mut |= x; },
///     commutative = verus_proof_commutative_effect!(
///         item = u32,
///         captures_mut = |flags_mut: u32|,
///         proof = |s, x, y| { assert(((s | x) | y) == ((s | y) | x)) by (bit_vector); }
///     )
/// ));
/// ```
///
/// [`Singleton::by_mut`]: crate::live_collections::singleton::Singleton::by_mut
macro_rules! __verus_proof_commutative_effect__ {
    (
        item = $itemty:ty
        $(, captures = |$($cap:ident : $capty:ty),* $(,)?|)?
        , captures_mut = |$mcap:ident : $mcapty:ty $(,)?|
        $(, proof = |$pstate:ident, $px:ident, $py:ident| { $($proof_body:tt)* })?
        , __target = { $(move)? |$titem:ident $(: $titemty:ty)?| $target_body:expr }
        , __captures = [$($fv:ident),* $(,)?] $(,)?
    ) => {{
        /// Checks that every free variable captured by the closure is declared (with its
        /// type) in the `captures = |...|` or `captures_mut = |...|` clauses: within this
        /// function, the *only* names in scope are the declared captures, so an
        /// undeclared capture fails to resolve.
        #[allow(unused, reason = "only checks name resolution")]
        fn __hydro_verus_captures_declared($($($cap: $capty,)*)? $mcap: $mcapty) {
            $(let $fv = &$fv;)*
        }

        #[cfg(verus_keep_ghost)]
        const _: () = {
            #[allow(unused_imports)]
            use ::vstd::prelude::*;

            ::vstd::prelude::verus! {
                /// Commutativity obligation, generated by
                /// `verus_proof_commutative_effect!`: applying the closure body with `x`
                /// then `y` must yield the same captured state as applying it with `y`
                /// then `x`, from any equal starting states. The closure must return
                /// `()`, so the state is the only observable effect. Verifying this also
                /// proves the body panic-free.
                fn __hydro_verus_commutative(
                    $($($cap: $capty,)*)?
                    __hydro_state_xy: $mcapty,
                    __hydro_state_yx: $mcapty,
                    __hydro_x_0: $itemty,
                    __hydro_x_1: $itemty,
                    __hydro_y_0: $itemty,
                    __hydro_y_1: $itemty,
                )
                    requires
                        __hydro_state_xy == __hydro_state_yx,
                        __hydro_x_0 == __hydro_x_1,
                        __hydro_y_0 == __hydro_y_1,
                {
                    $(
                        let ghost $pstate = __hydro_state_xy;
                        let ghost $px = __hydro_x_0;
                        let ghost $py = __hydro_y_0;
                    )?
                    let mut __hydro_state_xy = __hydro_state_xy;
                    let mut __hydro_state_yx = __hydro_state_yx;
                    { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_x_0; let __hydro_out: () = { $target_body }; }
                    { let $mcap = &mut __hydro_state_xy; let $titem = __hydro_y_0; let __hydro_out: () = { $target_body }; }
                    { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_y_1; let __hydro_out: () = { $target_body }; }
                    { let $mcap = &mut __hydro_state_yx; let $titem = __hydro_x_1; let __hydro_out: () = { $target_body }; }
                    $(proof { $($proof_body)* })?
                    assert(__hydro_state_xy == __hydro_state_yx);
                }
            }
        };

        $crate::properties::VerusCommutativeProof::new()
    }};
    (
        $($rest:tt)*
    ) => {{
        ::core::compile_error!(
            "verus_proof_commutative_effect! must be used as a property annotation inside `q!(...)` on a `|item| -> ()` closure with exactly one `captures_mut` entry, e.g. `commutative = verus_proof_commutative_effect!(item = u32, captures_mut = |state: u32|)`"
        );
        $crate::properties::VerusCommutativeProof::new()
    }};
}

/// Marks that the property is not proved.
pub enum NotProved {}

/// Marks that the property is proven.
pub enum Proved {}

/// Algebraic properties for an aggregation function of type (T, &mut A) -> ().
///
/// Commutativity:
/// ```rust,ignore
/// let mut state = ???;
/// f(a, &mut state); f(b, &mut state) // produces same final state as
/// f(b, &mut state); f(a, &mut state)
/// ```
///
/// Idempotence:
/// ```rust,ignore
/// let mut state = ???;
/// f(a, &mut state);
/// let state1 = *state;
/// f(a, &mut state);
/// // state1 must be equal to state
/// ```
pub struct AggFuncAlgebra<
    T = (),
    B: Boundedness = crate::live_collections::boundedness::Unbounded,
    Commutative = NotProved,
    Idempotent = NotProved,
    Monotone = NotProved,
    S = OnProcess,
>(
    Option<Box<dyn CommutativeProof<T, B, S>>>,
    Option<Box<dyn IdempotentProof>>,
    Option<Box<dyn MonotoneProof>>,
    PhantomData<(Commutative, Idempotent, Monotone)>,
);

impl<T, B: Boundedness, C, I, M, S> AggFuncAlgebra<T, B, C, I, M, S> {
    /// Marks the function as being commutative, with the given proof mechanism.
    pub fn commutative(
        self,
        proof: impl CommutativeProof<T, B, S> + 'static,
    ) -> AggFuncAlgebra<T, B, Proved, I, M, S> {
        AggFuncAlgebra(Some(Box::new(proof)), self.1, self.2, PhantomData)
    }

    /// Marks the function as being idempotent, with the given proof mechanism.
    pub fn idempotent(
        self,
        proof: impl IdempotentProof + 'static,
    ) -> AggFuncAlgebra<T, B, C, Proved, M, S> {
        AggFuncAlgebra(self.0, Some(Box::new(proof)), self.2, PhantomData)
    }

    /// Marks the function as being monotone, with the given proof mechanism.
    pub fn monotone(
        self,
        proof: impl MonotoneProof + 'static,
    ) -> AggFuncAlgebra<T, B, C, I, Proved, S> {
        AggFuncAlgebra(self.0, self.1, Some(Box::new(proof)), PhantomData)
    }

    /// Registers the expression with the underlying proof mechanisms, and takes the
    /// simulator ordering hook attached to the commutativity proof, if any.
    pub(crate) fn register_proof(
        self,
        expr: &syn::Expr,
    ) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        let mut hook = None;
        if let Some(mut comm_proof) = self.0 {
            comm_proof.register_proof(expr);
            hook = comm_proof.take_hook();
        }

        if let Some(idem_proof) = self.1 {
            idem_proof.register_proof(expr);
        }

        if let Some(monotone_proof) = self.2 {
            monotone_proof.register_proof(expr);
        }

        hook
    }
}

impl<T, B: Boundedness, C, I, M, S> Property for AggFuncAlgebra<T, B, C, I, M, S> {
    type Root = AggFuncAlgebra<T, B, NotProved, NotProved, NotProved, S>;

    fn make_root(_target: &mut Option<Self>) -> Self::Root {
        AggFuncAlgebra(None, None, None, PhantomData)
    }
}

/// Algebraic properties for a singleton map function of type T -> U.
///
/// Order-preserving means that if the input grows monotonically, the output also grows monotonically.
pub struct SingletonMapFuncAlgebra<
    T = (),
    B: Boundedness = crate::live_collections::boundedness::Unbounded,
    OrderPreserving = NotProved,
    Commutative = NotProved,
    Idempotent = NotProved,
    S = OnProcess,
>(
    Option<Box<dyn OrderPreservingProof>>,
    Option<Box<dyn CommutativeProof<T, B, S>>>,
    Option<Box<dyn IdempotentProof>>,
    PhantomData<(OrderPreserving, Commutative, Idempotent)>,
);

impl<T, B: Boundedness, O, C, I, S> SingletonMapFuncAlgebra<T, B, O, C, I, S> {
    /// Marks the function as being order-preserving, with the given proof mechanism.
    pub fn order_preserving(
        self,
        proof: impl OrderPreservingProof + 'static,
    ) -> SingletonMapFuncAlgebra<T, B, Proved, C, I, S> {
        SingletonMapFuncAlgebra(Some(Box::new(proof)), self.1, self.2, PhantomData)
    }

    /// Marks the function as being commutative, with the given proof mechanism.
    pub fn commutative(
        self,
        proof: impl CommutativeProof<T, B, S> + 'static,
    ) -> SingletonMapFuncAlgebra<T, B, O, Proved, I, S> {
        SingletonMapFuncAlgebra(self.0, Some(Box::new(proof)), self.2, PhantomData)
    }

    /// Marks the function as being idempotent, with the given proof mechanism.
    pub fn idempotent(
        self,
        proof: impl IdempotentProof + 'static,
    ) -> SingletonMapFuncAlgebra<T, B, O, C, Proved, S> {
        SingletonMapFuncAlgebra(self.0, self.1, Some(Box::new(proof)), PhantomData)
    }

    /// Registers the expression with the underlying proof mechanisms, and takes the
    /// simulator ordering hook attached to the commutativity proof, if any.
    pub(crate) fn register_proof(
        self,
        expr: &syn::Expr,
    ) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        if let Some(proof) = self.0 {
            proof.register_proof(expr);
        }
        self.1.and_then(|mut proof| {
            proof.register_proof(expr);
            proof.take_hook()
        })
    }
}

impl<T, B: Boundedness, O, C, I, S> Property for SingletonMapFuncAlgebra<T, B, O, C, I, S> {
    type Root = SingletonMapFuncAlgebra<T, B, NotProved, NotProved, NotProved, S>;

    fn make_root(_target: &mut Option<Self>) -> Self::Root {
        SingletonMapFuncAlgebra(None, None, None, PhantomData)
    }
}

/// Algebraic properties for a stream map function of type T -> U.
pub struct StreamMapFuncAlgebra<
    T = (),
    B: Boundedness = crate::live_collections::boundedness::Unbounded,
    Commutative = NotProved,
    Idempotent = NotProved,
    S = OnProcess,
>(
    Option<Box<dyn CommutativeProof<T, B, S>>>,
    Option<Box<dyn IdempotentProof>>,
    PhantomData<(Commutative, Idempotent)>,
);

impl<T, B: Boundedness, C, I, S> StreamMapFuncAlgebra<T, B, C, I, S> {
    /// Marks the function as being commutative, with the given proof mechanism.
    pub fn commutative(
        self,
        proof: impl CommutativeProof<T, B, S> + 'static,
    ) -> StreamMapFuncAlgebra<T, B, Proved, I, S> {
        StreamMapFuncAlgebra(Some(Box::new(proof)), self.1, PhantomData)
    }

    /// Marks the function as being idempotent, with the given proof mechanism.
    pub fn idempotent(
        self,
        proof: impl IdempotentProof + 'static,
    ) -> StreamMapFuncAlgebra<T, B, C, Proved, S> {
        StreamMapFuncAlgebra(self.0, Some(Box::new(proof)), PhantomData)
    }

    /// Registers the expression with the underlying proof mechanisms, and takes the
    /// simulator ordering hook attached to the commutativity proof, if any.
    pub(crate) fn register_proof(
        self,
        expr: &syn::Expr,
    ) -> Option<OrderingHook<T, B, ExactlyOnce, S>> {
        let hook = self.0.and_then(|mut proof| {
            proof.register_proof(expr);
            proof.take_hook()
        });
        if let Some(proof) = self.1 {
            proof.register_proof(expr);
        }
        hook
    }
}

impl<T, B: Boundedness, C, I, S> Property for StreamMapFuncAlgebra<T, B, C, I, S> {
    type Root = StreamMapFuncAlgebra<T, B, NotProved, NotProved, S>;

    fn make_root(_target: &mut Option<Self>) -> Self::Root {
        StreamMapFuncAlgebra(None, None, PhantomData)
    }
}

/// Marker trait identifying that the commutativity property is valid for the given stream ordering.
///
/// **Definition (aggregations, `|acc: &mut A, item: T|`):** for any accumulator value
/// and any two items `x`, `y`, applying the closure with `x` then `y` must produce the
/// same final accumulator as applying it with `y` then `x`. This makes the final
/// aggregate independent of the (non-deterministic) arrival order; note that
/// *intermediate* accumulator values may still differ and must not be observed without
/// a non-determinism annotation.
#[diagnostic::on_unimplemented(
    message = "because the input stream has ordering `{O}`, the closure must demonstrate commutativity",
    label = "required for this call",
    note = "prove it with a `commutative = manual_proof!(/** why the order does not matter */)` annotation on the closure",
    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"
)]
#[sealed::sealed]
pub trait ValidCommutativityFor<O: Ordering> {}
#[sealed::sealed]
impl ValidCommutativityFor<TotalOrder> for NotProved {}
#[sealed::sealed]
impl<O: Ordering> ValidCommutativityFor<O> for Proved {}

/// Marker trait identifying that the idempotence property is valid for the given stream ordering.
#[diagnostic::on_unimplemented(
    message = "because the input stream has retries `{R}`, the closure must demonstrate idempotence",
    label = "required for this call",
    note = "prove it with an `idempotent = manual_proof!(/** why a repeat does not matter */)` annotation on the closure",
    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"
)]
#[sealed::sealed]
pub trait ValidIdempotenceFor<R: Retries> {}
#[sealed::sealed]
impl ValidIdempotenceFor<ExactlyOnce> for NotProved {}
#[sealed::sealed]
impl<R: Retries> ValidIdempotenceFor<R> for Proved {}

/// Marker trait identifying that the commutativity property is valid for the given stream ordering.
///
/// A proof is required when the stream is unordered **and** the closure mutably captures
/// state (`WAS_MUT`). **Definition (owned-item closures, `|item: T| -> Out`, e.g. `map`
/// / `for_each`):** processing any two items in either order must (1) leave the
/// mutably-captured state (e.g. [`Singleton::by_mut`] references) in the same final
/// value, *and* (2) produce the same **multiset of return values**. Condition (2) is
/// required whenever the return value is observable (as in `map`): a closure that
/// emits, say, a running total is **not** commutative even if its state update is. For
/// `()`-returning closures (as in `for_each`), condition (2) is trivial.
///
/// [`Singleton::by_mut`]: crate::live_collections::singleton::Singleton::by_mut
#[sealed::sealed]
#[diagnostic::on_unimplemented(
    message = "because the input stream has ordering `{O}`, the closure must demonstrate commutativity",
    label = "required for this call",
    note = "prove it with a `commutative = manual_proof!(/** why the order does not matter */)` annotation on the closure",
    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"
)]
pub trait ValidMutCommutativityFor<F: FnMut(In) -> Out, In, Out, O: Ordering, const WAS_MUT: bool> {}
#[sealed::sealed]
impl<In, Out, F: FnMut(In) -> Out> ValidMutCommutativityFor<F, In, Out, TotalOrder, true>
    for NotProved
{
}
#[sealed::sealed]
impl<In, Out, F: Fn(In) -> Out, O: Ordering> ValidMutCommutativityFor<F, In, Out, O, false>
    for NotProved
{
}
#[sealed::sealed]
impl<In, Out, F: FnMut(In) -> Out, O: Ordering> ValidMutCommutativityFor<F, In, Out, O, true>
    for Proved
{
}
#[sealed::sealed]
impl<In, Out, F: Fn(In) -> Out, O: Ordering> ValidMutCommutativityFor<F, In, Out, O, false>
    for Proved
{
}

/// Marker trait identifying that the idempotence property is valid for the given stream ordering.
#[diagnostic::on_unimplemented(
    message = "because the input stream has retries `{R}`, the closure must demonstrate idempotence",
    label = "required for this call",
    note = "prove it with an `idempotent = manual_proof!(/** why a repeat does not matter */)` annotation on the closure",
    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"
)]
#[sealed::sealed]
pub trait ValidMutIdempotenceFor<F: FnMut(In) -> Out, In, Out, R: Retries, const WAS_MUT: bool> {}
#[sealed::sealed]
impl<In, Out, F: FnMut(In) -> Out> ValidMutIdempotenceFor<F, In, Out, ExactlyOnce, true>
    for NotProved
{
}
#[sealed::sealed]
impl<In, Out, F: Fn(In) -> Out, R: Retries> ValidMutIdempotenceFor<F, In, Out, R, false>
    for NotProved
{
}
#[sealed::sealed]
impl<In, Out, F: FnMut(In) -> Out, R: Retries> ValidMutIdempotenceFor<F, In, Out, R, true>
    for Proved
{
}
#[sealed::sealed]
impl<In, Out, F: Fn(In) -> Out, R: Retries> ValidMutIdempotenceFor<F, In, Out, R, false>
    for Proved
{
}

/// Marker trait for commutativity of closures that borrow their input (`FnMut(&In) -> Out`).
///
/// A proof is required when the stream is unordered **and** the closure mutably captures
/// state (`WAS_MUT`). **Definition (borrowing closures, `|item: &T| -> Out`, e.g.
/// `filter` / `inspect`):** processing any two items in either order must (1) leave the
/// mutably-captured state (e.g. [`Singleton::by_mut`] references) in the same final
/// value, *and* (2) keep the operator's observable output identical as a multiset. For
/// `filter`, the outputs are the **retained elements**, so the predicate's decisions
/// 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. For `inspect`, the elements pass through unchanged, so only
/// condition (1) applies.
///
/// [`Singleton::by_mut`]: crate::live_collections::singleton::Singleton::by_mut
#[sealed::sealed]
#[diagnostic::on_unimplemented(
    message = "because the input stream has ordering `{O}`, the closure must demonstrate commutativity",
    label = "required for this call",
    note = "prove it with a `commutative = manual_proof!(/** why the order does not matter */)` annotation on the closure",
    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"
)]
pub trait ValidMutBorrowCommutativityFor<
    F: FnMut(&In) -> Out,
    In: ?Sized,
    Out,
    O: Ordering,
    const WAS_MUT: bool,
>
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: FnMut(&In) -> Out>
    ValidMutBorrowCommutativityFor<F, In, Out, TotalOrder, true> for NotProved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: Fn(&In) -> Out, O: Ordering>
    ValidMutBorrowCommutativityFor<F, In, Out, O, false> for NotProved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: FnMut(&In) -> Out, O: Ordering>
    ValidMutBorrowCommutativityFor<F, In, Out, O, true> for Proved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: Fn(&In) -> Out, O: Ordering>
    ValidMutBorrowCommutativityFor<F, In, Out, O, false> for Proved
{
}

/// Marker trait for idempotence of closures that borrow their input (`FnMut(&In) -> Out`).
#[diagnostic::on_unimplemented(
    message = "because the input stream has retries `{R}`, the closure must demonstrate idempotence",
    label = "required for this call",
    note = "prove it with an `idempotent = manual_proof!(/** why a repeat does not matter */)` annotation on the closure",
    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"
)]
#[sealed::sealed]
pub trait ValidMutBorrowIdempotenceFor<
    F: FnMut(&In) -> Out,
    In: ?Sized,
    Out,
    R: Retries,
    const WAS_MUT: bool,
>
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: FnMut(&In) -> Out>
    ValidMutBorrowIdempotenceFor<F, In, Out, ExactlyOnce, true> for NotProved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: Fn(&In) -> Out, R: Retries>
    ValidMutBorrowIdempotenceFor<F, In, Out, R, false> for NotProved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: FnMut(&In) -> Out, R: Retries>
    ValidMutBorrowIdempotenceFor<F, In, Out, R, true> for Proved
{
}
#[sealed::sealed]
impl<In: ?Sized, Out, F: Fn(&In) -> Out, R: Retries>
    ValidMutBorrowIdempotenceFor<F, In, Out, R, false> for Proved
{
}

/// Marker trait identifying the boundedness of a singleton given a monotonicity property of
/// an aggregation on a stream.
#[sealed::sealed]
pub trait ApplyMonotoneStream<P, B2: SingletonBound> {}

#[sealed::sealed]
impl<B: Boundedness> ApplyMonotoneStream<NotProved, B> for B {}

#[sealed::sealed]
impl<B: Boundedness> ApplyMonotoneStream<Proved, B::StreamToMonotone> for B {}

/// Marker trait identifying the boundedness of a singleton given a monotonicity property of
/// an aggregation on a keyed stream.
#[sealed::sealed]
pub trait ApplyMonotoneKeyedStream<P, B2: KeyedSingletonBound> {}

#[sealed::sealed]
impl<B: Boundedness> ApplyMonotoneKeyedStream<NotProved, B::KeyedStreamToNonMonotone> for B {}

#[sealed::sealed]
impl<B: Boundedness> ApplyMonotoneKeyedStream<Proved, B::KeyedStreamToMonotone> for B {}

/// Marker trait identifying the boundedness of a singleton after a map operation,
/// given an order-preserving property.
#[sealed::sealed]
pub trait ApplyOrderPreservingSingleton<P, B2: SingletonBound> {}

#[sealed::sealed]
impl<B: SingletonBound> ApplyOrderPreservingSingleton<NotProved, B::UnderlyingBound> for B {}

#[sealed::sealed]
impl<B: SingletonBound> ApplyOrderPreservingSingleton<Proved, B> for B {}