diff --git a/hydroflow_plus/src/location/external_process.rs b/hydroflow_plus/src/location/external_process.rs
index de347154722b..0735083fd328 100644
--- a/hydroflow_plus/src/location/external_process.rs
+++ b/hydroflow_plus/src/location/external_process.rs
@@ -62,7 +62,7 @@ impl<'a, P> ExternalProcess<'a, P> {
     pub fn source_external_bytes<L: Location<'a> + NoTick>(
         &self,
         to: &L,
-    ) -> (ExternalBytesPort, Stream<Bytes, Unbounded, L>) {
+    ) -> (ExternalBytesPort, Stream<Bytes, L, Unbounded>) {
         let next_external_port_id = {
             let mut flow_state = self.flow_state.borrow_mut();
             let id = flow_state.next_external_out;
@@ -97,7 +97,7 @@ impl<'a, P> ExternalProcess<'a, P> {
     pub fn source_external_bincode<L: Location<'a> + NoTick, T: Serialize + DeserializeOwned>(
         &self,
         to: &L,
-    ) -> (ExternalBincodeSink<T>, Stream<T, Unbounded, L>) {
+    ) -> (ExternalBincodeSink<T>, Stream<T, L, Unbounded>) {
         let next_external_port_id = {
             let mut flow_state = self.flow_state.borrow_mut();
             let id = flow_state.next_external_out;
diff --git a/hydroflow_plus/src/location/mod.rs b/hydroflow_plus/src/location/mod.rs
index d8dd16750e8d..8908f5827647 100644
--- a/hydroflow_plus/src/location/mod.rs
+++ b/hydroflow_plus/src/location/mod.rs
@@ -78,7 +78,7 @@ pub trait Location<'a>: Clone {
         }
     }
 
-    fn spin(&self) -> Stream<(), Unbounded, Self>
+    fn spin(&self) -> Stream<(), Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
@@ -94,7 +94,7 @@ pub trait Location<'a>: Clone {
     fn source_stream<T, E: FuturesStream<Item = T> + Unpin>(
         &self,
         e: impl Quoted<'a, E>,
-    ) -> Stream<T, Unbounded, Self>
+    ) -> Stream<T, Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
@@ -112,7 +112,7 @@ pub trait Location<'a>: Clone {
     fn source_iter<T, E: IntoIterator<Item = T>>(
         &self,
         e: impl Quoted<'a, E>,
-    ) -> Stream<T, Unbounded, Self>
+    ) -> Stream<T, Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
@@ -129,7 +129,7 @@ pub trait Location<'a>: Clone {
         )
     }
 
-    fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Unbounded, Self>
+    fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
@@ -156,7 +156,7 @@ pub trait Location<'a>: Clone {
     fn source_interval(
         &self,
         interval: impl Quoted<'a, Duration> + Copy + 'a,
-    ) -> Stream<tokio::time::Instant, Unbounded, Self>
+    ) -> Stream<tokio::time::Instant, Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
@@ -169,7 +169,7 @@ pub trait Location<'a>: Clone {
         &self,
         delay: impl Quoted<'a, Duration> + Copy + 'a,
         interval: impl Quoted<'a, Duration> + Copy + 'a,
-    ) -> Stream<tokio::time::Instant, Unbounded, Self>
+    ) -> Stream<tokio::time::Instant, Self, Unbounded>
     where
         Self: Sized + NoTick,
     {
diff --git a/hydroflow_plus/src/location/tick.rs b/hydroflow_plus/src/location/tick.rs
index 895ae45559ad..d974eb14acca 100644
--- a/hydroflow_plus/src/location/tick.rs
+++ b/hydroflow_plus/src/location/tick.rs
@@ -45,7 +45,7 @@ impl<'a, L: Location<'a>> Tick<L> {
     pub fn spin_batch(
         &self,
         batch_size: impl Quoted<'a, usize> + Copy + 'a,
-    ) -> Stream<(), Bounded, Self>
+    ) -> Stream<(), Self, Bounded>
     where
         L: NoTick,
     {
@@ -56,7 +56,7 @@ impl<'a, L: Location<'a>> Tick<L> {
             .tick_batch(self)
     }
 
-    pub fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Bounded, Self>
+    pub fn singleton<T: Clone>(&self, e: impl Quoted<'a, T>) -> Singleton<T, Self, Bounded>
     where
         L: NoTick,
     {
@@ -66,7 +66,7 @@ impl<'a, L: Location<'a>> Tick<L> {
     pub fn singleton_first_tick<T: Clone>(
         &self,
         e: impl Quoted<'a, T>,
-    ) -> Optional<T, Bounded, Self>
+    ) -> Optional<T, Self, Bounded>
     where
         L: NoTick,
     {
diff --git a/hydroflow_plus/src/optional.rs b/hydroflow_plus/src/optional.rs
index 7ca84bf6be1f..1494f8a667f7 100644
--- a/hydroflow_plus/src/optional.rs
+++ b/hydroflow_plus/src/optional.rs
@@ -12,15 +12,15 @@ use crate::ir::{HfPlusLeaf, HfPlusNode, HfPlusSource, TeeNode};
 use crate::location::{check_matching_location, LocationId, NoTick};
 use crate::{Bounded, Location, Singleton, Stream, Tick, Unbounded};
 
-pub struct Optional<T, W, N> {
-    pub(crate) location: N,
+pub struct Optional<T, L, B> {
+    pub(crate) location: L,
     pub(crate) ir_node: RefCell<HfPlusNode>,
 
-    _phantom: PhantomData<(T, N, W)>,
+    _phantom: PhantomData<(T, L, B)>,
 }
 
-impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
-    pub(crate) fn new(location: N, ir_node: HfPlusNode) -> Self {
+impl<'a, T, L: Location<'a>, B> Optional<T, L, B> {
+    pub(crate) fn new(location: L, ir_node: HfPlusNode) -> Self {
         Optional {
             location,
             ir_node: RefCell::new(ir_node),
@@ -28,7 +28,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         }
     }
 
-    pub fn some(singleton: Singleton<T, W, N>) -> Self {
+    pub fn some(singleton: Singleton<T, L, B>) -> Self {
         Optional::new(singleton.location, singleton.ir_node.into_inner())
     }
 
@@ -37,16 +37,16 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> DeferTick for Optional<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> DeferTick for Optional<T, Tick<L>, Bounded> {
     fn defer_tick(self) -> Self {
         Optional::defer_tick(self)
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleCollection<'a, TickCycle> for Optional<T, Bounded, Tick<N>> {
-    type Location = Tick<N>;
+impl<'a, T, L: Location<'a>> CycleCollection<'a, TickCycle> for Optional<T, Tick<L>, Bounded> {
+    type Location = Tick<L>;
 
-    fn create_source(ident: syn::Ident, location: Tick<N>) -> Self {
+    fn create_source(ident: syn::Ident, location: Tick<L>) -> Self {
         let location_id = location.id();
         Optional::new(
             location,
@@ -58,7 +58,7 @@ impl<'a, T, N: Location<'a>> CycleCollection<'a, TickCycle> for Optional<T, Boun
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Optional<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> CycleComplete<'a, TickCycle> for Optional<T, Tick<L>, Bounded> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -74,10 +74,10 @@ impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Optional<T, Bounde
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleCollection<'a, ForwardRef> for Optional<T, Bounded, Tick<N>> {
-    type Location = Tick<N>;
+impl<'a, T, L: Location<'a>> CycleCollection<'a, ForwardRef> for Optional<T, Tick<L>, Bounded> {
+    type Location = Tick<L>;
 
-    fn create_source(ident: syn::Ident, location: Tick<N>) -> Self {
+    fn create_source(ident: syn::Ident, location: Tick<L>) -> Self {
         let location_id = location.id();
         Optional::new(
             location,
@@ -89,7 +89,7 @@ impl<'a, T, N: Location<'a>> CycleCollection<'a, ForwardRef> for Optional<T, Bou
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleComplete<'a, ForwardRef> for Optional<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> CycleComplete<'a, ForwardRef> for Optional<T, Tick<L>, Bounded> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -105,10 +105,10 @@ impl<'a, T, N: Location<'a>> CycleComplete<'a, ForwardRef> for Optional<T, Bound
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> CycleCollection<'a, ForwardRef> for Optional<T, W, N> {
-    type Location = N;
+impl<'a, T, L: Location<'a> + NoTick, B> CycleCollection<'a, ForwardRef> for Optional<T, L, B> {
+    type Location = L;
 
-    fn create_source(ident: syn::Ident, location: N) -> Self {
+    fn create_source(ident: syn::Ident, location: L) -> Self {
         let location_id = location.id();
         Optional::new(
             location,
@@ -120,7 +120,7 @@ impl<'a, T, W, N: Location<'a> + NoTick> CycleCollection<'a, ForwardRef> for Opt
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> CycleComplete<'a, ForwardRef> for Optional<T, W, N> {
+impl<'a, T, L: Location<'a> + NoTick, B> CycleComplete<'a, ForwardRef> for Optional<T, L, B> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -136,13 +136,19 @@ impl<'a, T, W, N: Location<'a> + NoTick> CycleComplete<'a, ForwardRef> for Optio
     }
 }
 
-impl<'a, T, W, N: Location<'a>> From<Singleton<T, W, N>> for Optional<T, W, N> {
-    fn from(singleton: Singleton<T, W, N>) -> Self {
+impl<'a, T, L: Location<'a>> From<Optional<T, L, Bounded>> for Optional<T, L, Unbounded> {
+    fn from(singleton: Optional<T, L, Bounded>) -> Self {
+        Optional::new(singleton.location, singleton.ir_node.into_inner())
+    }
+}
+
+impl<'a, T, L: Location<'a>, B> From<Singleton<T, L, B>> for Optional<T, L, B> {
+    fn from(singleton: Singleton<T, L, B>) -> Self {
         Optional::some(singleton)
     }
 }
 
-impl<'a, T: Clone, W, N: Location<'a>> Clone for Optional<T, W, N> {
+impl<'a, T: Clone, L: Location<'a>, B> Clone for Optional<T, L, B> {
     fn clone(&self) -> Self {
         if !matches!(self.ir_node.borrow().deref(), HfPlusNode::Tee { .. }) {
             let orig_ir_node = self.ir_node.replace(HfPlusNode::Placeholder);
@@ -166,17 +172,17 @@ impl<'a, T: Clone, W, N: Location<'a>> Clone for Optional<T, W, N> {
     }
 }
 
-impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
+impl<'a, T, L: Location<'a>, B> Optional<T, L, B> {
     // TODO(shadaj): this is technically incorrect; we should only return the first element of the stream
-    pub fn into_stream(self) -> Stream<T, W, N> {
-        if N::is_top_level() {
+    pub fn into_stream(self) -> Stream<T, L, B> {
+        if L::is_top_level() {
             panic!("Converting an optional to a stream is not yet supported at the top level");
         }
 
         Stream::new(self.location, self.ir_node.into_inner())
     }
 
-    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<U, W, N> {
+    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<U, L, B> {
         Optional::new(
             self.location,
             HfPlusNode::Map {
@@ -189,7 +195,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
     pub fn flat_map<U, I: IntoIterator<Item = U>, F: Fn(T) -> I + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<U, W, N> {
+    ) -> Stream<U, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::FlatMap {
@@ -199,7 +205,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         )
     }
 
-    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<T, W, N> {
+    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<T, L, B> {
         Optional::new(
             self.location,
             HfPlusNode::Filter {
@@ -212,7 +218,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
     pub fn filter_map<U, F: Fn(T) -> Option<U> + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Optional<U, W, N> {
+    ) -> Optional<U, L, B> {
         Optional::new(
             self.location,
             HfPlusNode::FilterMap {
@@ -222,10 +228,10 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         )
     }
 
-    pub fn union(self, other: Optional<T, W, N>) -> Optional<T, W, N> {
+    pub fn union(self, other: Optional<T, L, B>) -> Optional<T, L, B> {
         check_matching_location(&self.location, &other.location);
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             Optional::new(
                 self.location,
                 HfPlusNode::Persist(Box::new(HfPlusNode::Union(
@@ -244,14 +250,14 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         }
     }
 
-    pub fn zip<O>(self, other: impl Into<Optional<O, W, N>>) -> Optional<(T, O), W, N>
+    pub fn zip<O>(self, other: impl Into<Optional<O, L, B>>) -> Optional<(T, O), L, B>
     where
         O: Clone,
     {
-        let other: Optional<O, W, N> = other.into();
+        let other: Optional<O, L, B> = other.into();
         check_matching_location(&self.location, &other.location);
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             Optional::new(
                 self.location,
                 HfPlusNode::Persist(Box::new(HfPlusNode::CrossSingleton(
@@ -270,10 +276,10 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         }
     }
 
-    pub fn unwrap_or(self, other: Singleton<T, W, N>) -> Singleton<T, W, N> {
+    pub fn unwrap_or(self, other: Singleton<T, L, B>) -> Singleton<T, L, B> {
         check_matching_location(&self.location, &other.location);
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             Singleton::new(
                 self.location,
                 HfPlusNode::Persist(Box::new(HfPlusNode::Union(
@@ -292,7 +298,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
         }
     }
 
-    pub fn into_singleton(self) -> Singleton<Option<T>, W, N>
+    pub fn into_singleton(self) -> Singleton<Option<T>, L, B>
     where
         T: Clone,
     {
@@ -302,7 +308,7 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
             location_kind: self.location.id().root().clone(),
         }));
 
-        let none_singleton = if N::is_top_level() {
+        let none_singleton = if L::is_top_level() {
             Singleton::new(
                 self.location.clone(),
                 HfPlusNode::Persist(Box::new(core_ir)),
@@ -315,29 +321,29 @@ impl<'a, T, W, N: Location<'a>> Optional<T, W, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> Optional<T, Bounded, N> {
-    pub fn continue_if<U>(self, signal: Optional<U, Bounded, N>) -> Optional<T, Bounded, N> {
+impl<'a, T, L: Location<'a>> Optional<T, L, Bounded> {
+    pub fn continue_if<U>(self, signal: Optional<U, L, Bounded>) -> Optional<T, L, Bounded> {
         self.zip(signal.map(q!(|_u| ()))).map(q!(|(d, _signal)| d))
     }
 
-    pub fn continue_unless<U>(self, other: Optional<U, Bounded, N>) -> Optional<T, Bounded, N> {
+    pub fn continue_unless<U>(self, other: Optional<U, L, Bounded>) -> Optional<T, L, Bounded> {
         self.continue_if(other.into_stream().count().filter(q!(|c| *c == 0)))
     }
 
-    pub fn then<U>(self, value: Singleton<U, Bounded, N>) -> Optional<U, Bounded, N> {
+    pub fn then<U>(self, value: Singleton<U, L, Bounded>) -> Optional<U, L, Bounded> {
         value.continue_if(self)
     }
 }
 
-impl<'a, T, B, N: Location<'a> + NoTick> Optional<T, B, N> {
-    pub fn latest_tick(self, tick: &Tick<N>) -> Optional<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a> + NoTick, B> Optional<T, L, B> {
+    pub fn latest_tick(self, tick: &Tick<L>) -> Optional<T, Tick<L>, Bounded> {
         Optional::new(
             tick.clone(),
             HfPlusNode::Unpersist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn tick_samples(self) -> Stream<T, Unbounded, N> {
+    pub fn tick_samples(self) -> Stream<T, L, Unbounded> {
         let tick = self.location.tick();
         self.latest_tick(&tick).all_ticks()
     }
@@ -345,7 +351,7 @@ impl<'a, T, B, N: Location<'a> + NoTick> Optional<T, B, N> {
     pub fn sample_every(
         self,
         interval: impl Quoted<'a, std::time::Duration> + Copy + 'a,
-    ) -> Stream<T, Unbounded, N> {
+    ) -> Stream<T, L, Unbounded> {
         let samples = self.location.source_interval(interval);
         let tick = self.location.tick();
 
@@ -355,36 +361,36 @@ impl<'a, T, B, N: Location<'a> + NoTick> Optional<T, B, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> Optional<T, Bounded, Tick<N>> {
-    pub fn all_ticks(self) -> Stream<T, Unbounded, N> {
+impl<'a, T, L: Location<'a>> Optional<T, Tick<L>, Bounded> {
+    pub fn all_ticks(self) -> Stream<T, L, Unbounded> {
         Stream::new(
             self.location.outer().clone(),
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn latest(self) -> Optional<T, Unbounded, N> {
+    pub fn latest(self) -> Optional<T, L, Unbounded> {
         Optional::new(
             self.location.outer().clone(),
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn defer_tick(self) -> Optional<T, Bounded, Tick<N>> {
+    pub fn defer_tick(self) -> Optional<T, Tick<L>, Bounded> {
         Optional::new(
             self.location,
             HfPlusNode::DeferTick(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn persist(self) -> Stream<T, Bounded, Tick<N>> {
+    pub fn persist(self) -> Stream<T, Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn delta(self) -> Optional<T, Bounded, Tick<N>> {
+    pub fn delta(self) -> Optional<T, Tick<L>, Bounded> {
         Optional::new(
             self.location,
             HfPlusNode::Delta(Box::new(self.ir_node.into_inner())),
diff --git a/hydroflow_plus/src/singleton.rs b/hydroflow_plus/src/singleton.rs
index b153a0947152..394aae8c0fa2 100644
--- a/hydroflow_plus/src/singleton.rs
+++ b/hydroflow_plus/src/singleton.rs
@@ -14,15 +14,15 @@ use crate::location::{check_matching_location, Location, LocationId, NoTick, Tic
 use crate::stream::{Bounded, Unbounded};
 use crate::{Optional, Stream};
 
-pub struct Singleton<T, W, N> {
-    pub(crate) location: N,
+pub struct Singleton<T, L, B> {
+    pub(crate) location: L,
     pub(crate) ir_node: RefCell<HfPlusNode>,
 
-    _phantom: PhantomData<(T, N, W)>,
+    _phantom: PhantomData<(T, L, B)>,
 }
 
-impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
-    pub(crate) fn new(location: N, ir_node: HfPlusNode) -> Self {
+impl<'a, T, L: Location<'a>, B> Singleton<T, L, B> {
+    pub(crate) fn new(location: L, ir_node: HfPlusNode) -> Self {
         Singleton {
             location,
             ir_node: RefCell::new(ir_node),
@@ -35,24 +35,24 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> From<Singleton<T, Bounded, N>> for Singleton<T, Unbounded, N> {
-    fn from(singleton: Singleton<T, Bounded, N>) -> Self {
+impl<'a, T, L: Location<'a>> From<Singleton<T, L, Bounded>> for Singleton<T, L, Unbounded> {
+    fn from(singleton: Singleton<T, L, Bounded>) -> Self {
         Singleton::new(singleton.location, singleton.ir_node.into_inner())
     }
 }
 
-impl<'a, T, N: Location<'a>> DeferTick for Singleton<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> DeferTick for Singleton<T, Tick<L>, Bounded> {
     fn defer_tick(self) -> Self {
         Singleton::defer_tick(self)
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleCollectionWithInitial<'a, TickCycle>
-    for Singleton<T, Bounded, Tick<N>>
+impl<'a, T, L: Location<'a>> CycleCollectionWithInitial<'a, TickCycle>
+    for Singleton<T, Tick<L>, Bounded>
 {
-    type Location = Tick<N>;
+    type Location = Tick<L>;
 
-    fn create_source(ident: syn::Ident, initial: Self, location: Tick<N>) -> Self {
+    fn create_source(ident: syn::Ident, initial: Self, location: Tick<L>) -> Self {
         let location_id = location.id();
         Singleton::new(
             location,
@@ -67,7 +67,7 @@ impl<'a, T, N: Location<'a>> CycleCollectionWithInitial<'a, TickCycle>
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Singleton<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> CycleComplete<'a, TickCycle> for Singleton<T, Tick<L>, Bounded> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -83,10 +83,10 @@ impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Singleton<T, Bound
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleCollection<'a, ForwardRef> for Singleton<T, Bounded, Tick<N>> {
-    type Location = Tick<N>;
+impl<'a, T, L: Location<'a>> CycleCollection<'a, ForwardRef> for Singleton<T, Tick<L>, Bounded> {
+    type Location = Tick<L>;
 
-    fn create_source(ident: syn::Ident, location: Tick<N>) -> Self {
+    fn create_source(ident: syn::Ident, location: Tick<L>) -> Self {
         let location_id = location.id();
         Singleton::new(
             location,
@@ -98,7 +98,7 @@ impl<'a, T, N: Location<'a>> CycleCollection<'a, ForwardRef> for Singleton<T, Bo
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleComplete<'a, ForwardRef> for Singleton<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> CycleComplete<'a, ForwardRef> for Singleton<T, Tick<L>, Bounded> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -114,7 +114,7 @@ impl<'a, T, N: Location<'a>> CycleComplete<'a, ForwardRef> for Singleton<T, Boun
     }
 }
 
-impl<'a, T: Clone, W, N: Location<'a>> Clone for Singleton<T, W, N> {
+impl<'a, T: Clone, L: Location<'a>, B> Clone for Singleton<T, L, B> {
     fn clone(&self) -> Self {
         if !matches!(self.ir_node.borrow().deref(), HfPlusNode::Tee { .. }) {
             let orig_ir_node = self.ir_node.replace(HfPlusNode::Placeholder);
@@ -138,13 +138,13 @@ impl<'a, T: Clone, W, N: Location<'a>> Clone for Singleton<T, W, N> {
     }
 }
 
-impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
+impl<'a, T, L: Location<'a>, B> Singleton<T, L, B> {
     // TODO(shadaj): this is technically incorrect; we should only return the first element of the stream
-    pub fn into_stream(self) -> Stream<T, Bounded, N> {
+    pub fn into_stream(self) -> Stream<T, L, Bounded> {
         Stream::new(self.location, self.ir_node.into_inner())
     }
 
-    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Singleton<U, W, N> {
+    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Singleton<U, L, B> {
         Singleton::new(
             self.location,
             HfPlusNode::Map {
@@ -157,7 +157,7 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
     pub fn flat_map<U, I: IntoIterator<Item = U>, F: Fn(T) -> I + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<U, W, N> {
+    ) -> Stream<U, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::FlatMap {
@@ -167,7 +167,7 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
         )
     }
 
-    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<T, W, N> {
+    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Optional<T, L, B> {
         Optional::new(
             self.location,
             HfPlusNode::Filter {
@@ -180,7 +180,7 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
     pub fn filter_map<U, F: Fn(T) -> Option<U> + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Optional<U, W, N> {
+    ) -> Optional<U, L, B> {
         Optional::new(
             self.location,
             HfPlusNode::FilterMap {
@@ -192,11 +192,11 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
 
     pub fn zip<Other>(self, other: Other) -> <Self as ZipResult<'a, Other>>::Out
     where
-        Self: ZipResult<'a, Other, Location = N>,
+        Self: ZipResult<'a, Other, Location = L>,
     {
         check_matching_location(&self.location, &Self::other_location(&other));
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             Self::make(
                 self.location,
                 HfPlusNode::Persist(Box::new(HfPlusNode::CrossSingleton(
@@ -216,25 +216,25 @@ impl<'a, T, W, N: Location<'a>> Singleton<T, W, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> Singleton<T, Bounded, N> {
-    pub fn continue_if<U>(self, signal: Optional<U, Bounded, N>) -> Optional<T, Bounded, N> {
+impl<'a, T, L: Location<'a>> Singleton<T, L, Bounded> {
+    pub fn continue_if<U>(self, signal: Optional<U, L, Bounded>) -> Optional<T, L, Bounded> {
         self.zip(signal.map(q!(|_u| ()))).map(q!(|(d, _signal)| d))
     }
 
-    pub fn continue_unless<U>(self, other: Optional<U, Bounded, N>) -> Optional<T, Bounded, N> {
+    pub fn continue_unless<U>(self, other: Optional<U, L, Bounded>) -> Optional<T, L, Bounded> {
         self.continue_if(other.into_stream().count().filter(q!(|c| *c == 0)))
     }
 }
 
-impl<'a, T, B, N: Location<'a> + NoTick> Singleton<T, B, N> {
-    pub fn latest_tick(self, tick: &Tick<N>) -> Singleton<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a> + NoTick, B> Singleton<T, L, B> {
+    pub fn latest_tick(self, tick: &Tick<L>) -> Singleton<T, Tick<L>, Bounded> {
         Singleton::new(
             tick.clone(),
             HfPlusNode::Unpersist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn tick_samples(self) -> Stream<T, Unbounded, N> {
+    pub fn tick_samples(self) -> Stream<T, L, Unbounded> {
         let tick = self.location.tick();
         self.latest_tick(&tick).all_ticks()
     }
@@ -242,7 +242,7 @@ impl<'a, T, B, N: Location<'a> + NoTick> Singleton<T, B, N> {
     pub fn sample_every(
         self,
         interval: impl Quoted<'a, std::time::Duration> + Copy + 'a,
-    ) -> Stream<T, Unbounded, N> {
+    ) -> Stream<T, L, Unbounded> {
         let samples = self.location.source_interval(interval);
         let tick = self.location.tick();
 
@@ -252,36 +252,36 @@ impl<'a, T, B, N: Location<'a> + NoTick> Singleton<T, B, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> Singleton<T, Bounded, Tick<N>> {
-    pub fn all_ticks(self) -> Stream<T, Unbounded, N> {
+impl<'a, T, L: Location<'a>> Singleton<T, Tick<L>, Bounded> {
+    pub fn all_ticks(self) -> Stream<T, L, Unbounded> {
         Stream::new(
             self.location.outer().clone(),
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn latest(self) -> Singleton<T, Unbounded, N> {
+    pub fn latest(self) -> Singleton<T, L, Unbounded> {
         Singleton::new(
             self.location.outer().clone(),
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn defer_tick(self) -> Singleton<T, Bounded, Tick<N>> {
+    pub fn defer_tick(self) -> Singleton<T, Tick<L>, Bounded> {
         Singleton::new(
             self.location,
             HfPlusNode::DeferTick(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn persist(self) -> Stream<T, Bounded, Tick<N>> {
+    pub fn persist(self) -> Stream<T, Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn delta(self) -> Optional<T, Bounded, Tick<N>> {
+    pub fn delta(self) -> Optional<T, Tick<L>, Bounded> {
         Optional::new(
             self.location,
             HfPlusNode::Delta(Box::new(self.ir_node.into_inner())),
@@ -299,36 +299,36 @@ pub trait ZipResult<'a, Other> {
     fn make(location: Self::Location, ir_node: HfPlusNode) -> Self::Out;
 }
 
-impl<'a, T, U: Clone, W, N: Location<'a>> ZipResult<'a, Singleton<U, W, N>> for Singleton<T, W, N> {
-    type Out = Singleton<(T, U), W, N>;
-    type Location = N;
+impl<'a, T, U: Clone, L: Location<'a>, B> ZipResult<'a, Singleton<U, L, B>> for Singleton<T, L, B> {
+    type Out = Singleton<(T, U), L, B>;
+    type Location = L;
 
-    fn other_location(other: &Singleton<U, W, N>) -> N {
+    fn other_location(other: &Singleton<U, L, B>) -> L {
         other.location.clone()
     }
 
-    fn other_ir_node(other: Singleton<U, W, N>) -> HfPlusNode {
+    fn other_ir_node(other: Singleton<U, L, B>) -> HfPlusNode {
         other.ir_node.into_inner()
     }
 
-    fn make(location: N, ir_node: HfPlusNode) -> Self::Out {
+    fn make(location: L, ir_node: HfPlusNode) -> Self::Out {
         Singleton::new(location, ir_node)
     }
 }
 
-impl<'a, T, U: Clone, W, N: Location<'a>> ZipResult<'a, Optional<U, W, N>> for Singleton<T, W, N> {
-    type Out = Optional<(T, U), W, N>;
-    type Location = N;
+impl<'a, T, U: Clone, L: Location<'a>, B> ZipResult<'a, Optional<U, L, B>> for Singleton<T, L, B> {
+    type Out = Optional<(T, U), L, B>;
+    type Location = L;
 
-    fn other_location(other: &Optional<U, W, N>) -> N {
+    fn other_location(other: &Optional<U, L, B>) -> L {
         other.location.clone()
     }
 
-    fn other_ir_node(other: Optional<U, W, N>) -> HfPlusNode {
+    fn other_ir_node(other: Optional<U, L, B>) -> HfPlusNode {
         other.ir_node.into_inner()
     }
 
-    fn make(location: N, ir_node: HfPlusNode) -> Self::Out {
+    fn make(location: L, ir_node: HfPlusNode) -> Self::Out {
         Optional::new(location, ir_node)
     }
 }
diff --git a/hydroflow_plus/src/stream.rs b/hydroflow_plus/src/stream.rs
index 3f1e1803cc94..f85eb32aa2dc 100644
--- a/hydroflow_plus/src/stream.rs
+++ b/hydroflow_plus/src/stream.rs
@@ -31,36 +31,36 @@ pub enum Unbounded {}
 /// to be complete in finite time.
 pub enum Bounded {}
 
-/// An infinite stream of elements of type `T`.
+/// An ordered sequence stream of elements of type `T`.
 ///
 /// Type Parameters:
 /// - `T`: the type of elements in the stream
+/// - `L`: the location where the stream is being materialized
 /// - `B`: the boundedness of the stream, which is either [`Bounded`]
 ///    or [`Unbounded`]
-/// - `N`: the type of the node that the stream is materialized on
-pub struct Stream<T, B, N> {
-    location: N,
+pub struct Stream<T, L, B> {
+    location: L,
     pub(crate) ir_node: RefCell<HfPlusNode>,
 
-    _phantom: PhantomData<(T, B, N)>,
+    _phantom: PhantomData<(T, L, B)>,
 }
 
-impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
+impl<'a, T, L: Location<'a>, B> Stream<T, L, B> {
     fn location_kind(&self) -> LocationId {
         self.location.id()
     }
 }
 
-impl<'a, T, N: Location<'a>> DeferTick for Stream<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> DeferTick for Stream<T, Tick<L>, Bounded> {
     fn defer_tick(self) -> Self {
         Stream::defer_tick(self)
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleCollection<'a, TickCycle> for Stream<T, Bounded, Tick<N>> {
-    type Location = Tick<N>;
+impl<'a, T, L: Location<'a>> CycleCollection<'a, TickCycle> for Stream<T, Tick<L>, Bounded> {
+    type Location = Tick<L>;
 
-    fn create_source(ident: syn::Ident, location: Tick<N>) -> Self {
+    fn create_source(ident: syn::Ident, location: Tick<L>) -> Self {
         let location_id = location.id();
         Stream::new(
             location,
@@ -72,7 +72,7 @@ impl<'a, T, N: Location<'a>> CycleCollection<'a, TickCycle> for Stream<T, Bounde
     }
 }
 
-impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Stream<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a>> CycleComplete<'a, TickCycle> for Stream<T, Tick<L>, Bounded> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -88,10 +88,10 @@ impl<'a, T, N: Location<'a>> CycleComplete<'a, TickCycle> for Stream<T, Bounded,
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> CycleCollection<'a, ForwardRef> for Stream<T, W, N> {
-    type Location = N;
+impl<'a, T, L: Location<'a> + NoTick, B> CycleCollection<'a, ForwardRef> for Stream<T, L, B> {
+    type Location = L;
 
-    fn create_source(ident: syn::Ident, location: N) -> Self {
+    fn create_source(ident: syn::Ident, location: L) -> Self {
         let location_id = location.id();
         Stream::new(
             location,
@@ -103,7 +103,7 @@ impl<'a, T, W, N: Location<'a> + NoTick> CycleCollection<'a, ForwardRef> for Str
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> CycleComplete<'a, ForwardRef> for Stream<T, W, N> {
+impl<'a, T, L: Location<'a> + NoTick, B> CycleComplete<'a, ForwardRef> for Stream<T, L, B> {
     fn complete(self, ident: syn::Ident) {
         self.location
             .flow_state()
@@ -119,8 +119,8 @@ impl<'a, T, W, N: Location<'a> + NoTick> CycleComplete<'a, ForwardRef> for Strea
     }
 }
 
-impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
-    pub(crate) fn new(location: N, ir_node: HfPlusNode) -> Self {
+impl<'a, T, L: Location<'a>, B> Stream<T, L, B> {
+    pub(crate) fn new(location: L, ir_node: HfPlusNode) -> Self {
         Stream {
             location,
             ir_node: RefCell::new(ir_node),
@@ -129,7 +129,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
     }
 }
 
-impl<'a, T: Clone, W, N: Location<'a>> Clone for Stream<T, W, N> {
+impl<'a, T: Clone, L: Location<'a>, B> Clone for Stream<T, L, B> {
     fn clone(&self) -> Self {
         if !matches!(self.ir_node.borrow().deref(), HfPlusNode::Tee { .. }) {
             let orig_ir_node = self.ir_node.replace(HfPlusNode::Placeholder);
@@ -153,8 +153,8 @@ impl<'a, T: Clone, W, N: Location<'a>> Clone for Stream<T, W, N> {
     }
 }
 
-impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
-    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<U, W, N> {
+impl<'a, T, L: Location<'a>, B> Stream<T, L, B> {
+    pub fn map<U, F: Fn(T) -> U + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<U, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::Map {
@@ -164,7 +164,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn cloned(self) -> Stream<T, W, N>
+    pub fn cloned(self) -> Stream<T, L, B>
     where
         T: Clone,
     {
@@ -174,7 +174,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
     pub fn flat_map<U, I: IntoIterator<Item = U>, F: Fn(T) -> I + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<U, W, N> {
+    ) -> Stream<U, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::FlatMap {
@@ -184,14 +184,14 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn flatten<U>(self) -> Stream<U, W, N>
+    pub fn flatten<U>(self) -> Stream<U, L, B>
     where
         T: IntoIterator<Item = U>,
     {
         self.flat_map(q!(|d| d))
     }
 
-    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<T, W, N> {
+    pub fn filter<F: Fn(&T) -> bool + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<T, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::Filter {
@@ -204,7 +204,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
     pub fn filter_map<U, F: Fn(T) -> Option<U> + 'a>(
         self,
         f: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<U, W, N> {
+    ) -> Stream<U, L, B> {
         Stream::new(
             self.location,
             HfPlusNode::FilterMap {
@@ -216,12 +216,12 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
 
     pub fn cross_singleton<O>(
         self,
-        other: impl Into<Optional<O, Bounded, N>>,
-    ) -> Stream<(T, O), W, N>
+        other: impl Into<Optional<O, L, Bounded>>,
+    ) -> Stream<(T, O), L, B>
     where
         O: Clone,
     {
-        let other: Optional<O, Bounded, N> = other.into();
+        let other: Optional<O, L, Bounded> = other.into();
         check_matching_location(&self.location, &other.location);
 
         Stream::new(
@@ -234,17 +234,17 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
     }
 
     /// Allow this stream through if the other stream has elements, otherwise the output is empty.
-    pub fn continue_if<U>(self, signal: Optional<U, Bounded, N>) -> Stream<T, W, N> {
+    pub fn continue_if<U>(self, signal: Optional<U, L, Bounded>) -> Stream<T, L, B> {
         self.cross_singleton(signal.map(q!(|_u| ())))
             .map(q!(|(d, _signal)| d))
     }
 
     /// Allow this stream through if the other stream is empty, otherwise the output is empty.
-    pub fn continue_unless<U>(self, other: Optional<U, Bounded, N>) -> Stream<T, W, N> {
+    pub fn continue_unless<U>(self, other: Optional<U, L, Bounded>) -> Stream<T, L, B> {
         self.continue_if(other.into_stream().count().filter(q!(|c| *c == 0)))
     }
 
-    pub fn cross_product<O>(self, other: Stream<O, W, N>) -> Stream<(T, O), W, N>
+    pub fn cross_product<O>(self, other: Stream<O, L, B>) -> Stream<(T, O), L, B>
     where
         T: Clone,
         O: Clone,
@@ -260,7 +260,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn union(self, other: Stream<T, W, N>) -> Stream<T, W, N> {
+    pub fn union(self, other: Stream<T, L, B>) -> Stream<T, L, B> {
         check_matching_location(&self.location, &other.location);
 
         Stream::new(
@@ -272,14 +272,14 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn enumerate(self) -> Stream<(usize, T), W, N> {
+    pub fn enumerate(self) -> Stream<(usize, T), L, B> {
         Stream::new(
             self.location,
             HfPlusNode::Enumerate(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn unique(self) -> Stream<T, W, N>
+    pub fn unique(self) -> Stream<T, L, B>
     where
         T: Eq + Hash,
     {
@@ -289,7 +289,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn filter_not_in(self, other: Stream<T, Bounded, N>) -> Stream<T, Bounded, N>
+    pub fn filter_not_in(self, other: Stream<T, L, Bounded>) -> Stream<T, L, Bounded>
     where
         T: Eq + Hash,
     {
@@ -304,12 +304,12 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         )
     }
 
-    pub fn first(self) -> Optional<T, Bounded, N> {
+    pub fn first(self) -> Optional<T, L, Bounded> {
         Optional::new(self.location, self.ir_node.into_inner())
     }
 
-    pub fn inspect<F: Fn(&T) + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<T, W, N> {
-        if N::is_top_level() {
+    pub fn inspect<F: Fn(&T) + 'a>(self, f: impl IntoQuotedMut<'a, F>) -> Stream<T, L, B> {
+        if L::is_top_level() {
             Stream::new(
                 self.location,
                 HfPlusNode::Persist(Box::new(HfPlusNode::Inspect {
@@ -332,14 +332,14 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         self,
         init: impl IntoQuotedMut<'a, I>,
         comb: impl IntoQuotedMut<'a, F>,
-    ) -> Singleton<A, W, N> {
+    ) -> Singleton<A, L, B> {
         let mut core = HfPlusNode::Fold {
             init: init.splice_fn0().into(),
             acc: comb.splice_fn2_borrow_mut().into(),
             input: Box::new(self.ir_node.into_inner()),
         };
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             // top-level (possibly unbounded) singletons are represented as
             // a stream which produces all values from all ticks every tick,
             // so Unpersist will always give the lastest aggregation
@@ -352,20 +352,20 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
     pub fn reduce<F: Fn(&mut T, T) + 'a>(
         self,
         comb: impl IntoQuotedMut<'a, F>,
-    ) -> Optional<T, W, N> {
+    ) -> Optional<T, L, B> {
         let mut core = HfPlusNode::Reduce {
             f: comb.splice_fn2_borrow_mut().into(),
             input: Box::new(self.ir_node.into_inner()),
         };
 
-        if N::is_top_level() {
+        if L::is_top_level() {
             core = HfPlusNode::Persist(Box::new(core));
         }
 
         Optional::new(self.location, core)
     }
 
-    pub fn max(self) -> Optional<T, W, N>
+    pub fn max(self) -> Optional<T, L, B>
     where
         T: Ord,
     {
@@ -376,7 +376,7 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         }))
     }
 
-    pub fn min(self) -> Optional<T, W, N>
+    pub fn min(self) -> Optional<T, L, B>
     where
         T: Ord,
     {
@@ -387,13 +387,13 @@ impl<'a, T, W, N: Location<'a>> Stream<T, W, N> {
         }))
     }
 
-    pub fn count(self) -> Singleton<usize, W, N> {
+    pub fn count(self) -> Singleton<usize, L, B> {
         self.fold(q!(|| 0usize), q!(|count, _| *count += 1))
     }
 }
 
-impl<'a, T, N: Location<'a>> Stream<T, Bounded, N> {
-    pub fn sort(self) -> Stream<T, Bounded, N>
+impl<'a, T, L: Location<'a>> Stream<T, L, Bounded> {
+    pub fn sort(self) -> Stream<T, L, Bounded>
     where
         T: Ord,
     {
@@ -404,8 +404,8 @@ impl<'a, T, N: Location<'a>> Stream<T, Bounded, N> {
     }
 }
 
-impl<'a, K, V1, W, N: Location<'a>> Stream<(K, V1), W, N> {
-    pub fn join<V2>(self, n: Stream<(K, V2), W, N>) -> Stream<(K, (V1, V2)), W, N>
+impl<'a, K, V1, L: Location<'a>, B> Stream<(K, V1), L, B> {
+    pub fn join<V2>(self, n: Stream<(K, V2), L, B>) -> Stream<(K, (V1, V2)), L, B>
     where
         K: Eq + Hash,
     {
@@ -420,7 +420,7 @@ impl<'a, K, V1, W, N: Location<'a>> Stream<(K, V1), W, N> {
         )
     }
 
-    pub fn anti_join(self, n: Stream<K, Bounded, N>) -> Stream<(K, V1), W, N>
+    pub fn anti_join(self, n: Stream<K, L, Bounded>) -> Stream<(K, V1), L, B>
     where
         K: Eq + Hash,
     {
@@ -436,12 +436,12 @@ impl<'a, K, V1, W, N: Location<'a>> Stream<(K, V1), W, N> {
     }
 }
 
-impl<'a, K: Eq + Hash, V, N: Location<'a>> Stream<(K, V), Bounded, Tick<N>> {
+impl<'a, K: Eq + Hash, V, L: Location<'a>> Stream<(K, V), Tick<L>, Bounded> {
     pub fn fold_keyed<A, I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a>(
         self,
         init: impl IntoQuotedMut<'a, I>,
         comb: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<(K, A), Bounded, Tick<N>> {
+    ) -> Stream<(K, A), Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::FoldKeyed {
@@ -455,7 +455,7 @@ impl<'a, K: Eq + Hash, V, N: Location<'a>> Stream<(K, V), Bounded, Tick<N>> {
     pub fn reduce_keyed<F: Fn(&mut V, V) + 'a>(
         self,
         comb: impl IntoQuotedMut<'a, F>,
-    ) -> Stream<(K, V), Bounded, Tick<N>> {
+    ) -> Stream<(K, V), Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::ReduceKeyed {
@@ -466,15 +466,15 @@ impl<'a, K: Eq + Hash, V, N: Location<'a>> Stream<(K, V), Bounded, Tick<N>> {
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
-    pub fn tick_batch(self, tick: &Tick<N>) -> Stream<T, Bounded, Tick<N>> {
+impl<'a, T, L: Location<'a> + NoTick, B> Stream<T, L, B> {
+    pub fn tick_batch(self, tick: &Tick<L>) -> Stream<T, Tick<L>, Bounded> {
         Stream::new(
             tick.clone(),
             HfPlusNode::Unpersist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn tick_prefix(self, tick: &Tick<N>) -> Stream<T, Bounded, Tick<N>>
+    pub fn tick_prefix(self, tick: &Tick<L>) -> Stream<T, Tick<L>, Bounded>
     where
         T: Clone,
     {
@@ -484,7 +484,7 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     pub fn sample_every(
         self,
         interval: impl Quoted<'a, std::time::Duration> + Copy + 'a,
-    ) -> Stream<T, Unbounded, N> {
+    ) -> Stream<T, L, Unbounded> {
         let samples = self.location.source_interval(interval);
         let tick = self.location.tick();
         self.tick_batch(&tick)
@@ -519,15 +519,15 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     }
 }
 
-impl<'a, T, N: Location<'a>> Stream<T, Bounded, Tick<N>> {
-    pub fn all_ticks(self) -> Stream<T, Unbounded, N> {
+impl<'a, T, L: Location<'a>> Stream<T, Tick<L>, Bounded> {
+    pub fn all_ticks(self) -> Stream<T, L, Unbounded> {
         Stream::new(
             self.location.outer().clone(),
             HfPlusNode::Persist(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn persist(self) -> Stream<T, Bounded, Tick<N>>
+    pub fn persist(self) -> Stream<T, Tick<L>, Bounded>
     where
         T: Clone,
     {
@@ -537,14 +537,14 @@ impl<'a, T, N: Location<'a>> Stream<T, Bounded, Tick<N>> {
         )
     }
 
-    pub fn defer_tick(self) -> Stream<T, Bounded, Tick<N>> {
+    pub fn defer_tick(self) -> Stream<T, Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::DeferTick(Box::new(self.ir_node.into_inner())),
         )
     }
 
-    pub fn delta(self) -> Stream<T, Bounded, Tick<N>> {
+    pub fn delta(self) -> Stream<T, Tick<L>, Bounded> {
         Stream::new(
             self.location,
             HfPlusNode::Delta(Box::new(self.ir_node.into_inner())),
@@ -593,13 +593,13 @@ pub(super) fn deserialize_bincode<T: DeserializeOwned>(tagged: Option<syn::Type>
     }
 }
 
-impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
+impl<'a, T, L: Location<'a> + NoTick, B> Stream<T, L, B> {
     pub fn decouple_process<P2>(
         self,
         other: &Process<'a, P2>,
-    ) -> Stream<T, Unbounded, Process<'a, P2>>
+    ) -> Stream<T, Process<'a, P2>, Unbounded>
     where
-        N: CanSend<'a, Process<'a, P2>, In<T> = T, Out<T> = T>,
+        L: CanSend<'a, Process<'a, P2>, In<T> = T, Out<T> = T>,
         T: Clone + Serialize + DeserializeOwned,
     {
         self.send_bincode::<Process<'a, P2>, T>(other)
@@ -608,9 +608,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     pub fn decouple_cluster<C2, Tag>(
         self,
         other: &Cluster<'a, C2>,
-    ) -> Stream<T, Unbounded, Cluster<'a, C2>>
+    ) -> Stream<T, Cluster<'a, C2>, Unbounded>
     where
-        N: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)>,
+        L: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)>,
         T: Clone + Serialize + DeserializeOwned,
     {
         let self_node_id = match self.location_kind() {
@@ -625,17 +625,17 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
             .send_bincode_interleaved(other)
     }
 
-    pub fn send_bincode<N2: Location<'a>, CoreType>(
+    pub fn send_bincode<L2: Location<'a>, CoreType>(
         self,
-        other: &N2,
-    ) -> Stream<N::Out<CoreType>, Unbounded, N2>
+        other: &L2,
+    ) -> Stream<L::Out<CoreType>, L2, Unbounded>
     where
-        N: CanSend<'a, N2, In<CoreType> = T>,
+        L: CanSend<'a, L2, In<CoreType> = T>,
         CoreType: Serialize + DeserializeOwned,
     {
-        let serialize_pipeline = Some(serialize_bincode::<CoreType>(N::is_demux()));
+        let serialize_pipeline = Some(serialize_bincode::<CoreType>(L::is_demux()));
 
-        let deserialize_pipeline = Some(deserialize_bincode::<CoreType>(N::tagged_type()));
+        let deserialize_pipeline = Some(deserialize_bincode::<CoreType>(L::tagged_type()));
 
         Stream::new(
             other.clone(),
@@ -652,16 +652,16 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
         )
     }
 
-    pub fn send_bincode_external<N2: 'a, CoreType>(
+    pub fn send_bincode_external<L2: 'a, CoreType>(
         self,
-        other: &ExternalProcess<N2>,
-    ) -> ExternalBincodeStream<N::Out<CoreType>>
+        other: &ExternalProcess<L2>,
+    ) -> ExternalBincodeStream<L::Out<CoreType>>
     where
-        N: CanSend<'a, ExternalProcess<'a, N2>, In<CoreType> = T, Out<CoreType> = CoreType>,
+        L: CanSend<'a, ExternalProcess<'a, L2>, In<CoreType> = T, Out<CoreType> = CoreType>,
         CoreType: Serialize + DeserializeOwned,
         // for now, we restirct Out<CoreType> to be CoreType, which means no tagged cluster -> external
     {
-        let serialize_pipeline = Some(serialize_bincode::<CoreType>(N::is_demux()));
+        let serialize_pipeline = Some(serialize_bincode::<CoreType>(L::is_demux()));
 
         let mut flow_state_borrow = self.location.flow_state().borrow_mut();
 
@@ -693,9 +693,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
         }
     }
 
-    pub fn send_bytes<N2: Location<'a>>(self, other: &N2) -> Stream<N::Out<Bytes>, Unbounded, N2>
+    pub fn send_bytes<L2: Location<'a>>(self, other: &L2) -> Stream<L::Out<Bytes>, L2, Unbounded>
     where
-        N: CanSend<'a, N2, In<Bytes> = T>,
+        L: CanSend<'a, L2, In<Bytes> = T>,
     {
         let root = get_this_crate();
         Stream::new(
@@ -707,7 +707,7 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
                 to_key: None,
                 serialize_pipeline: None,
                 instantiate_fn: DebugInstantiate::Building(),
-                deserialize_pipeline: if let Some(c_type) = N::tagged_type() {
+                deserialize_pipeline: if let Some(c_type) = L::tagged_type() {
                     Some(
                         parse_quote!(map(|(id, b)| (#root::ClusterId<#c_type>::from_raw(id), b.unwrap().freeze()))),
                     )
@@ -719,9 +719,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
         )
     }
 
-    pub fn send_bytes_external<N2: 'a>(self, other: &ExternalProcess<N2>) -> ExternalBytesPort
+    pub fn send_bytes_external<L2: 'a>(self, other: &ExternalProcess<L2>) -> ExternalBytesPort
     where
-        N: CanSend<'a, ExternalProcess<'a, N2>, In<Bytes> = T, Out<Bytes> = Bytes>,
+        L: CanSend<'a, ExternalProcess<'a, L2>, In<Bytes> = T, Out<Bytes> = Bytes>,
     {
         let mut flow_state_borrow = self.location.flow_state().borrow_mut();
         let external_key = flow_state_borrow.next_external_out;
@@ -751,33 +751,33 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
         }
     }
 
-    pub fn send_bincode_interleaved<N2: Location<'a>, Tag, CoreType>(
+    pub fn send_bincode_interleaved<L2: Location<'a>, Tag, CoreType>(
         self,
-        other: &N2,
-    ) -> Stream<CoreType, Unbounded, N2>
+        other: &L2,
+    ) -> Stream<CoreType, L2, Unbounded>
     where
-        N: CanSend<'a, N2, In<CoreType> = T, Out<CoreType> = (Tag, CoreType)>,
+        L: CanSend<'a, L2, In<CoreType> = T, Out<CoreType> = (Tag, CoreType)>,
         CoreType: Serialize + DeserializeOwned,
     {
-        self.send_bincode::<N2, CoreType>(other).map(q!(|(_, b)| b))
+        self.send_bincode::<L2, CoreType>(other).map(q!(|(_, b)| b))
     }
 
-    pub fn send_bytes_interleaved<N2: Location<'a>, Tag>(
+    pub fn send_bytes_interleaved<L2: Location<'a>, Tag>(
         self,
-        other: &N2,
-    ) -> Stream<Bytes, Unbounded, N2>
+        other: &L2,
+    ) -> Stream<Bytes, L2, Unbounded>
     where
-        N: CanSend<'a, N2, In<Bytes> = T, Out<Bytes> = (Tag, Bytes)>,
+        L: CanSend<'a, L2, In<Bytes> = T, Out<Bytes> = (Tag, Bytes)>,
     {
-        self.send_bytes::<N2>(other).map(q!(|(_, b)| b))
+        self.send_bytes::<L2>(other).map(q!(|(_, b)| b))
     }
 
     pub fn broadcast_bincode<C2>(
         self,
         other: &Cluster<'a, C2>,
-    ) -> Stream<N::Out<T>, Unbounded, Cluster<'a, C2>>
+    ) -> Stream<L::Out<T>, Cluster<'a, C2>, Unbounded>
     where
-        N: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T)>,
+        L: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T)>,
         T: Clone + Serialize + DeserializeOwned,
     {
         let ids = other.members();
@@ -792,9 +792,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     pub fn broadcast_bincode_interleaved<C2, Tag>(
         self,
         other: &Cluster<'a, C2>,
-    ) -> Stream<T, Unbounded, Cluster<'a, C2>>
+    ) -> Stream<T, Cluster<'a, C2>, Unbounded>
     where
-        N: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)> + 'a,
+        L: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)> + 'a,
         T: Clone + Serialize + DeserializeOwned,
     {
         self.broadcast_bincode(other).map(q!(|(_, b)| b))
@@ -803,9 +803,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     pub fn broadcast_bytes<C2>(
         self,
         other: &Cluster<'a, C2>,
-    ) -> Stream<N::Out<Bytes>, Unbounded, Cluster<'a, C2>>
+    ) -> Stream<L::Out<Bytes>, Cluster<'a, C2>, Unbounded>
     where
-        N: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T)> + 'a,
+        L: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T)> + 'a,
         T: Clone,
     {
         let ids = other.members();
@@ -820,9 +820,9 @@ impl<'a, T, W, N: Location<'a> + NoTick> Stream<T, W, N> {
     pub fn broadcast_bytes_interleaved<C2, Tag>(
         self,
         other: &Cluster<'a, C2>,
-    ) -> Stream<Bytes, Unbounded, Cluster<'a, C2>>
+    ) -> Stream<Bytes, Cluster<'a, C2>, Unbounded>
     where
-        N: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T), Out<Bytes> = (Tag, Bytes)>
+        L: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T), Out<Bytes> = (Tag, Bytes)>
             + 'a,
         T: Clone,
     {
diff --git a/hydroflow_plus_test/src/cluster/paxos.rs b/hydroflow_plus_test/src/cluster/paxos.rs
index a5d14f794063..85b9e4c06940 100644
--- a/hydroflow_plus_test/src/cluster/paxos.rs
+++ b/hydroflow_plus_test/src/cluster/paxos.rs
@@ -62,15 +62,15 @@ struct P2b<P> {
 pub fn paxos_core<'a, P: PaxosPayload, R>(
     proposers: &Cluster<'a, Proposer>,
     acceptors: &Cluster<'a, Acceptor>,
-    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Unbounded, Cluster<'a, Acceptor>>,
-    c_to_proposers: Stream<P, Unbounded, Cluster<'a, Proposer>>,
+    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Cluster<'a, Acceptor>, Unbounded>,
+    c_to_proposers: Stream<P, Cluster<'a, Proposer>, Unbounded>,
     f: usize,
     i_am_leader_send_timeout: u64,
     i_am_leader_check_timeout: u64,
     i_am_leader_check_timeout_delay_multiplier: usize,
 ) -> (
-    Stream<(), Unbounded, Cluster<'a, Proposer>>,
-    Stream<(usize, Option<P>), Unbounded, Cluster<'a, Proposer>>,
+    Stream<(), Cluster<'a, Proposer>, Unbounded>,
+    Stream<(usize, Option<P>), Cluster<'a, Proposer>, Unbounded>,
 ) {
     proposers
         .source_iter(q!(["Proposers say hello"]))
@@ -154,13 +154,13 @@ fn leader_election<'a, L: Clone + Debug + Serialize + DeserializeOwned>(
     i_am_leader_send_timeout: u64,
     i_am_leader_check_timeout: u64,
     i_am_leader_check_timeout_delay_multiplier: usize,
-    p_received_p2b_ballots: Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-    a_log: Singleton<L, Bounded, Tick<Cluster<'a, Acceptor>>>,
+    p_received_p2b_ballots: Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+    a_log: Singleton<L, Tick<Cluster<'a, Acceptor>>, Bounded>,
 ) -> (
-    Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Stream<P1b<L>, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Singleton<Ballot, Bounded, Tick<Cluster<'a, Acceptor>>>,
+    Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Stream<P1b<L>, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Singleton<Ballot, Tick<Cluster<'a, Acceptor>>, Bounded>,
 ) {
     let (a_to_proposers_p1b_complete_cycle, a_to_proposers_p1b_forward_ref) =
         proposers.forward_ref::<Stream<P1b<L>, _, _>>();
@@ -223,10 +223,10 @@ fn leader_election<'a, L: Clone + Debug + Serialize + DeserializeOwned>(
 // Proposer logic to calculate the largest ballot received so far.
 fn p_max_ballot<'a>(
     proposers: &Cluster<'a, Proposer>,
-    p_received_p1b_ballots: Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-    p_received_p2b_ballots: Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-    p_to_proposers_i_am_leader: Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-) -> Singleton<Ballot, Unbounded, Cluster<'a, Proposer>> {
+    p_received_p1b_ballots: Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+    p_received_p2b_ballots: Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+    p_to_proposers_i_am_leader: Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+) -> Singleton<Ballot, Cluster<'a, Proposer>, Unbounded> {
     p_received_p1b_ballots
         .union(p_received_p2b_ballots)
         .union(p_to_proposers_i_am_leader)
@@ -242,10 +242,10 @@ fn p_max_ballot<'a>(
 fn p_ballot_calc<'a>(
     proposers: &Cluster<'a, Proposer>,
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    p_received_max_ballot: Singleton<Ballot, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_received_max_ballot: Singleton<Ballot, Tick<Cluster<'a, Proposer>>, Bounded>,
 ) -> (
-    Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Optional<(Ballot, u32), Bounded, Tick<Cluster<'a, Proposer>>>,
+    Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Optional<(Ballot, u32), Tick<Cluster<'a, Proposer>>, Bounded>,
 ) {
     let p_id = proposers.self_id();
     let (p_ballot_num_complete_cycle, p_ballot_num) =
@@ -285,10 +285,10 @@ fn p_ballot_calc<'a>(
 
 fn p_leader_expired<'a>(
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    p_to_proposers_i_am_leader: Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-    p_is_leader: Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_to_proposers_i_am_leader: Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+    p_is_leader: Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
     i_am_leader_check_timeout: u64, // How often to check if heartbeat expired
-) -> Optional<Option<Instant>, Bounded, Tick<Cluster<'a, Proposer>>> {
+) -> Optional<Option<Instant>, Tick<Cluster<'a, Proposer>>, Bounded> {
     let p_latest_received_i_am_leader = p_to_proposers_i_am_leader.clone().fold(
         q!(|| None),
         q!(|latest, _| {
@@ -314,14 +314,14 @@ fn p_leader_expired<'a>(
 fn p_leader_heartbeat<'a>(
     proposers: &Cluster<'a, Proposer>,
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    p_is_leader: Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_is_leader: Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
     i_am_leader_send_timeout: u64,  // How often to heartbeat
     i_am_leader_check_timeout: u64, // How often to check if heartbeat expired
     i_am_leader_check_timeout_delay_multiplier: usize, /* Initial delay, multiplied by proposer pid, to stagger proposers checking for timeouts */
 ) -> (
-    Stream<Ballot, Unbounded, Cluster<'a, Proposer>>,
-    Optional<Option<Instant>, Bounded, Tick<Cluster<'a, Proposer>>>,
+    Stream<Ballot, Cluster<'a, Proposer>, Unbounded>,
+    Optional<Option<Instant>, Tick<Cluster<'a, Proposer>>, Bounded>,
 ) {
     let p_id = proposers.self_id();
     let p_to_proposers_i_am_leader = p_is_leader
@@ -359,11 +359,11 @@ fn p_leader_heartbeat<'a>(
 
 // Proposer logic to send "I am leader" messages periodically to other proposers, or send p1a to acceptors if other leaders expired.
 fn p_p1a<'a>(
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_trigger_election: Optional<Option<Instant>, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_trigger_election: Optional<Option<Instant>, Tick<Cluster<'a, Proposer>>, Bounded>,
     proposers: &Cluster<'a, Proposer>,
     acceptors: &Cluster<'a, Acceptor>,
-) -> Stream<P1a, Unbounded, Cluster<'a, Acceptor>> {
+) -> Stream<P1a, Cluster<'a, Acceptor>, Unbounded> {
     let p_id = proposers.self_id();
 
     p_trigger_election
@@ -382,12 +382,12 @@ fn p_p1a<'a>(
 #[expect(clippy::type_complexity, reason = "internal paxos code // TODO")]
 fn acceptor_p1<'a, L: Serialize + DeserializeOwned + Clone>(
     acceptor_tick: &Tick<Cluster<'a, Acceptor>>,
-    p_to_acceptors_p1a: Stream<P1a, Unbounded, Cluster<'a, Acceptor>>,
-    a_log: Singleton<L, Bounded, Tick<Cluster<'a, Acceptor>>>,
+    p_to_acceptors_p1a: Stream<P1a, Cluster<'a, Acceptor>, Unbounded>,
+    a_log: Singleton<L, Tick<Cluster<'a, Acceptor>>, Bounded>,
     proposers: &Cluster<'a, Proposer>,
 ) -> (
-    Singleton<Ballot, Bounded, Tick<Cluster<'a, Acceptor>>>,
-    Stream<P1b<L>, Unbounded, Cluster<'a, Proposer>>,
+    Singleton<Ballot, Tick<Cluster<'a, Acceptor>>, Bounded>,
+    Stream<P1b<L>, Cluster<'a, Proposer>, Unbounded>,
 ) {
     let p_to_acceptors_p1a = p_to_acceptors_p1a.tick_batch(acceptor_tick);
     let a_max_ballot = p_to_acceptors_p1a
@@ -424,13 +424,13 @@ fn acceptor_p1<'a, L: Serialize + DeserializeOwned + Clone>(
 fn p_p1b<'a, P: Clone + Serialize + DeserializeOwned>(
     proposers: &Cluster<'a, Proposer>,
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    a_to_proposers_p1b: Stream<P1b<P>, Unbounded, Cluster<'a, Proposer>>,
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_has_largest_ballot: Optional<(Ballot, u32), Bounded, Tick<Cluster<'a, Proposer>>>,
+    a_to_proposers_p1b: Stream<P1b<P>, Cluster<'a, Proposer>, Unbounded>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_has_largest_ballot: Optional<(Ballot, u32), Tick<Cluster<'a, Proposer>>, Bounded>,
     f: usize,
 ) -> (
-    Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Stream<P1b<P>, Bounded, Tick<Cluster<'a, Proposer>>>,
+    Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Stream<P1b<P>, Tick<Cluster<'a, Proposer>>, Bounded>,
 ) {
     let p_id = proposers.self_id();
     let p_relevant_p1bs = a_to_proposers_p1b
@@ -457,13 +457,13 @@ fn p_p1b<'a, P: Clone + Serialize + DeserializeOwned>(
 #[expect(clippy::type_complexity, reason = "internal paxos code // TODO")]
 fn recommit_after_leader_election<'a, P: PaxosPayload>(
     proposers: &Cluster<'a, Proposer>,
-    p_relevant_p1bs: Stream<P1b<HashMap<usize, LogValue<P>>>, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_relevant_p1bs: Stream<P1b<HashMap<usize, LogValue<P>>>, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
     f: usize,
 ) -> (
-    Stream<P2a<P>, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Optional<usize, Bounded, Tick<Cluster<'a, Proposer>>>,
-    Stream<P2a<P>, Bounded, Tick<Cluster<'a, Proposer>>>,
+    Stream<P2a<P>, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Optional<usize, Tick<Cluster<'a, Proposer>>, Bounded>,
+    Stream<P2a<P>, Tick<Cluster<'a, Proposer>>, Bounded>,
 ) {
     let p_id = proposers.self_id();
 
@@ -541,21 +541,21 @@ fn sequence_payload<'a, P: PaxosPayload, R>(
     acceptors: &Cluster<'a, Acceptor>,
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
     acceptor_tick: &Tick<Cluster<'a, Acceptor>>,
-    c_to_proposers: Stream<P, Unbounded, Cluster<'a, Proposer>>,
-    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Unbounded, Cluster<'a, Acceptor>>,
+    c_to_proposers: Stream<P, Cluster<'a, Proposer>, Unbounded>,
+    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Cluster<'a, Acceptor>, Unbounded>,
 
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_is_leader: Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_max_slot: Optional<usize, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_is_leader: Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_max_slot: Optional<usize, Tick<Cluster<'a, Proposer>>, Bounded>,
 
-    p_log_to_recommit: Stream<P2a<P>, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_log_to_recommit: Stream<P2a<P>, Tick<Cluster<'a, Proposer>>, Bounded>,
     f: usize,
 
-    a_max_ballot: Singleton<Ballot, Bounded, Tick<Cluster<'a, Acceptor>>>,
+    a_max_ballot: Singleton<Ballot, Tick<Cluster<'a, Acceptor>>, Bounded>,
 ) -> (
-    Stream<(usize, Option<P>), Unbounded, Cluster<'a, Proposer>>,
-    Singleton<(Option<usize>, HashMap<usize, LogValue<P>>), Bounded, Tick<Cluster<'a, Acceptor>>>,
-    Stream<P2b<P>, Unbounded, Cluster<'a, Proposer>>,
+    Stream<(usize, Option<P>), Cluster<'a, Proposer>, Unbounded>,
+    Singleton<(Option<usize>, HashMap<usize, LogValue<P>>), Tick<Cluster<'a, Acceptor>>, Bounded>,
+    Stream<P2b<P>, Cluster<'a, Proposer>, Unbounded>,
 ) {
     let p_to_acceptors_p2a = p_p2a(
         proposers,
@@ -595,13 +595,13 @@ enum CheckpointOrP2a<P> {
 fn p_p2a<'a, P: PaxosPayload>(
     proposers: &Cluster<'a, Proposer>,
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    p_max_slot: Optional<usize, Bounded, Tick<Cluster<'a, Proposer>>>,
-    c_to_proposers: Stream<P, Unbounded, Cluster<'a, Proposer>>,
-    p_ballot_num: Singleton<u32, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_log_to_recommit: Stream<P2a<P>, Bounded, Tick<Cluster<'a, Proposer>>>,
-    p_is_leader: Optional<bool, Bounded, Tick<Cluster<'a, Proposer>>>,
+    p_max_slot: Optional<usize, Tick<Cluster<'a, Proposer>>, Bounded>,
+    c_to_proposers: Stream<P, Cluster<'a, Proposer>, Unbounded>,
+    p_ballot_num: Singleton<u32, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_log_to_recommit: Stream<P2a<P>, Tick<Cluster<'a, Proposer>>, Bounded>,
+    p_is_leader: Optional<bool, Tick<Cluster<'a, Proposer>>, Bounded>,
     acceptors: &Cluster<'a, Acceptor>,
-) -> Stream<P2a<P>, Unbounded, Cluster<'a, Acceptor>> {
+) -> Stream<P2a<P>, Cluster<'a, Acceptor>, Unbounded> {
     let p_id = proposers.self_id();
     let (p_next_slot_complete_cycle, p_next_slot) = proposer_tick.cycle::<Optional<usize, _, _>>();
     let p_next_slot_after_reconciling_p1bs = p_max_slot
@@ -649,14 +649,14 @@ fn p_p2a<'a, P: PaxosPayload>(
 #[expect(clippy::type_complexity, reason = "internal paxos code // TODO")]
 fn acceptor_p2<'a, P: PaxosPayload, R>(
     acceptor_tick: &Tick<Cluster<'a, Acceptor>>,
-    a_max_ballot: Singleton<Ballot, Bounded, Tick<Cluster<'a, Acceptor>>>,
-    p_to_acceptors_p2a: Stream<P2a<P>, Unbounded, Cluster<'a, Acceptor>>,
-    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Unbounded, Cluster<'a, Acceptor>>,
+    a_max_ballot: Singleton<Ballot, Tick<Cluster<'a, Acceptor>>, Bounded>,
+    p_to_acceptors_p2a: Stream<P2a<P>, Cluster<'a, Acceptor>, Unbounded>,
+    r_to_acceptors_checkpoint: Stream<(ClusterId<R>, usize), Cluster<'a, Acceptor>, Unbounded>,
     proposers: &Cluster<'a, Proposer>,
     f: usize,
 ) -> (
-    Singleton<(Option<usize>, HashMap<usize, LogValue<P>>), Bounded, Tick<Cluster<'a, Acceptor>>>,
-    Stream<P2b<P>, Unbounded, Cluster<'a, Proposer>>,
+    Singleton<(Option<usize>, HashMap<usize, LogValue<P>>), Tick<Cluster<'a, Acceptor>>, Bounded>,
+    Stream<P2b<P>, Cluster<'a, Proposer>, Unbounded>,
 ) {
     let p_to_acceptors_p2a_batch = p_to_acceptors_p2a.tick_batch(acceptor_tick);
 
@@ -747,9 +747,9 @@ fn acceptor_p2<'a, P: PaxosPayload, R>(
 
 fn p_p2b<'a, P: PaxosPayload>(
     proposer_tick: &Tick<Cluster<'a, Proposer>>,
-    a_to_proposers_p2b: Stream<P2b<P>, Unbounded, Cluster<'a, Proposer>>,
+    a_to_proposers_p2b: Stream<P2b<P>, Cluster<'a, Proposer>, Unbounded>,
     f: usize,
-) -> Stream<(usize, Option<P>), Unbounded, Cluster<'a, Proposer>> {
+) -> Stream<(usize, Option<P>), Cluster<'a, Proposer>, Unbounded> {
     let (p_broadcasted_p2b_slots_complete_cycle, p_broadcasted_p2b_slots) = proposer_tick.cycle();
     let (p_persisted_p2bs_complete_cycle, p_persisted_p2bs) = proposer_tick.cycle();
     let p_p2b = a_to_proposers_p2b
diff --git a/hydroflow_plus_test/src/cluster/paxos_bench.rs b/hydroflow_plus_test/src/cluster/paxos_bench.rs
index e9c59925b274..e3db1458a809 100644
--- a/hydroflow_plus_test/src/cluster/paxos_bench.rs
+++ b/hydroflow_plus_test/src/cluster/paxos_bench.rs
@@ -72,17 +72,17 @@ pub fn paxos_bench<'a>(
 // Clients. All relations for clients will be prefixed with c. All ClientPayloads will contain the virtual client number as key and the client's machine ID (to string) as value. Expects p_to_clients_leader_elected containing Ballots whenever the leader is elected, and r_to_clients_payload_applied containing ReplicaPayloads whenever a payload is committed. Outputs (leader address, ClientPayload) when a new leader is elected or when the previous payload is committed.
 fn bench_client<'a>(
     clients: &Cluster<'a, Client>,
-    p_to_clients_leader_elected: Stream<ClusterId<Proposer>, Unbounded, Cluster<'a, Client>>,
+    p_to_clients_leader_elected: Stream<ClusterId<Proposer>, Cluster<'a, Client>, Unbounded>,
     transaction_cycle: impl FnOnce(
         Stream<
             (ClusterId<Proposer>, KvPayload<u32, ClusterId<Client>>),
-            Unbounded,
             Cluster<'a, Client>,
+            Unbounded,
         >,
     ) -> Stream<
         (ClusterId<Replica>, KvPayload<u32, ClusterId<Client>>),
-        Unbounded,
         Cluster<'a, Client>,
+        Unbounded,
     >,
     num_clients_per_node: usize,
     median_latency_window_size: usize,
diff --git a/hydroflow_plus_test/src/cluster/paxos_kv.rs b/hydroflow_plus_test/src/cluster/paxos_kv.rs
index 840e30cc9af7..9aeef0d9b7c4 100644
--- a/hydroflow_plus_test/src/cluster/paxos_kv.rs
+++ b/hydroflow_plus_test/src/cluster/paxos_kv.rs
@@ -51,15 +51,15 @@ pub fn paxos_kv<'a, K: KvKey, V: KvValue>(
     proposers: &Cluster<'a, Proposer>,
     acceptors: &Cluster<'a, Acceptor>,
     replicas: &Cluster<'a, Replica>,
-    c_to_proposers: Stream<KvPayload<K, V>, Unbounded, Cluster<'a, Proposer>>,
+    c_to_proposers: Stream<KvPayload<K, V>, Cluster<'a, Proposer>, Unbounded>,
     f: usize,
     i_am_leader_send_timeout: u64,
     i_am_leader_check_timeout: u64,
     i_am_leader_check_timeout_delay_multiplier: usize,
     checkpoint_frequency: usize,
 ) -> (
-    Stream<(), Unbounded, Cluster<'a, Proposer>>,
-    Stream<KvPayload<K, V>, Unbounded, Cluster<'a, Replica>>,
+    Stream<(), Cluster<'a, Proposer>, Unbounded>,
+    Stream<KvPayload<K, V>, Cluster<'a, Replica>, Unbounded>,
 ) {
     let (r_to_acceptors_checkpoint_complete_cycle, r_to_acceptors_checkpoint) =
         replicas.forward_ref::<Stream<_, _, _>>();
@@ -92,11 +92,11 @@ pub fn paxos_kv<'a, K: KvKey, V: KvValue>(
 #[expect(clippy::type_complexity, reason = "internal paxos code // TODO")]
 pub fn replica<'a, K: KvKey, V: KvValue>(
     replicas: &Cluster<'a, Replica>,
-    p_to_replicas: Stream<SequencedKv<K, V>, Unbounded, Cluster<'a, Replica>>,
+    p_to_replicas: Stream<SequencedKv<K, V>, Cluster<'a, Replica>, Unbounded>,
     checkpoint_frequency: usize,
 ) -> (
-    Stream<usize, Unbounded, Cluster<'a, Replica>>,
-    Stream<KvPayload<K, V>, Unbounded, Cluster<'a, Replica>>,
+    Stream<usize, Cluster<'a, Replica>, Unbounded>,
+    Stream<KvPayload<K, V>, Cluster<'a, Replica>, Unbounded>,
 ) {
     let replica_tick = replicas.tick();