diff --git a/hydroflow_plus/src/builder/built.rs b/hydroflow_plus/src/builder/built.rs
index 440dabe41973..ee196b3f3016 100644
--- a/hydroflow_plus/src/builder/built.rs
+++ b/hydroflow_plus/src/builder/built.rs
@@ -73,7 +73,7 @@ impl<'a> BuiltFlow<'a> {
     }
 
     pub fn with_default_optimize(self) -> BuiltFlow<'a> {
-        self.optimize_with(crate::persist_pullup::persist_pullup)
+        self.optimize_with(crate::rewrites::persist_pullup::persist_pullup)
     }
 
     fn into_deploy<D: LocalDeploy<'a>>(mut self) -> DeployFlow<'a, D> {
diff --git a/hydroflow_plus/src/lib.rs b/hydroflow_plus/src/lib.rs
index bbc38afb76d8..2e7f322ec1d8 100644
--- a/hydroflow_plus/src/lib.rs
+++ b/hydroflow_plus/src/lib.rs
@@ -34,10 +34,7 @@ pub use builder::FlowBuilder;
 
 pub mod ir;
 
-pub mod persist_pullup;
-pub mod profiler;
-
-pub mod properties;
+pub mod rewrites;
 
 mod staging_util;
 
diff --git a/hydroflow_plus/src/rewrites/mod.rs b/hydroflow_plus/src/rewrites/mod.rs
new file mode 100644
index 000000000000..20c3302f0fe3
--- /dev/null
+++ b/hydroflow_plus/src/rewrites/mod.rs
@@ -0,0 +1,3 @@
+pub mod persist_pullup;
+pub mod profiler;
+pub mod properties;
diff --git a/hydroflow_plus/src/persist_pullup.rs b/hydroflow_plus/src/rewrites/persist_pullup.rs
similarity index 100%
rename from hydroflow_plus/src/persist_pullup.rs
rename to hydroflow_plus/src/rewrites/persist_pullup.rs
diff --git a/hydroflow_plus/src/profiler.rs b/hydroflow_plus/src/rewrites/profiler.rs
similarity index 96%
rename from hydroflow_plus/src/profiler.rs
rename to hydroflow_plus/src/rewrites/profiler.rs
index 4e4ded0833cc..a2627567653c 100644
--- a/hydroflow_plus/src/profiler.rs
+++ b/hydroflow_plus/src/rewrites/profiler.rs
@@ -3,8 +3,9 @@ use std::cell::RefCell;
 use hydroflow::futures::channel::mpsc::UnboundedSender;
 use stageleft::*;
 
+use super::profiler as myself; // TODO(shadaj): stageleft does not support `self::...`
 use crate::ir::*;
-use crate::{profiler as myself, RuntimeContext};
+use crate::RuntimeContext;
 
 pub fn increment_counter(count: &mut u64) {
     *count += 1;
diff --git a/hydroflow_plus/src/properties.rs b/hydroflow_plus/src/rewrites/properties.rs
similarity index 100%
rename from hydroflow_plus/src/properties.rs
rename to hydroflow_plus/src/rewrites/properties.rs
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee-2.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee-2.snap
similarity index 59%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee-2.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee-2.snap
index 15f9404ce942..7ab17a3d0ed5 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee-2.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee-2.snap
@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: optimized.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Persist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Tee {
                     inner: <tee>: Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,
@@ -22,14 +22,14 @@ expression: optimized.ir()
         ),
     },
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Persist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Tee {
                     inner: <tee>: Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee.snap
similarity index 75%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee.snap
index 441cdcbfca74..b0cdf8700bfb 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee.snap
@@ -1,21 +1,21 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Persist(
                 Map {
-                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                     input: Tee {
                         inner: <tee>: Persist(
                             Unpersist(
                                 Persist(
                                     Source {
                                         source: Iter(
-                                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                                         ),
                                         location_kind: Process(
                                             0,
@@ -30,18 +30,18 @@ expression: built.ir()
         ),
     },
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Persist(
                 Map {
-                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                    f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                     input: Tee {
                         inner: <tee>: Persist(
                             Unpersist(
                                 Persist(
                                     Source {
                                         source: Iter(
-                                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                                         ),
                                         location_kind: Process(
                                             0,
diff --git a/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap
new file mode 100644
index 000000000000..14beff31a5de
--- /dev/null
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap
@@ -0,0 +1,20 @@
+---
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
+expression: graph.surface_syntax_string()
+---
+1v1 = source_iter ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 });
+2v1 = tee ();
+3v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+4v1 = persist :: < 'static > ();
+5v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+6v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+7v1 = persist :: < 'static > ();
+8v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+
+1v1 -> 2v1;
+2v1 -> 3v1;
+3v1 -> 4v1;
+4v1 -> 5v1;
+2v1 -> 6v1;
+6v1 -> 7v1;
+7v1 -> 8v1;
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map-2.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map-2.snap
similarity index 51%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map-2.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map-2.snap
index e9cad525e761..9811741ab041 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map-2.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map-2.snap
@@ -1,15 +1,15 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: optimized.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Map {
-            f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+            f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
             input: Source {
                 source: Iter(
-                    { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                    { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                 ),
                 location_kind: Process(
                     0,
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map.snap
similarity index 58%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map.snap
index 0feab1293dd6..a079c2bfaaaf 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map.snap
@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/persist_pullup.rs
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }),
                 input: Persist(
                     Source {
                         source: Iter(
-                            { use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,
diff --git a/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap
new file mode 100644
index 000000000000..3c16bf5d01c4
--- /dev/null
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap
@@ -0,0 +1,10 @@
+---
+source: hydroflow_plus/src/rewrites/persist_pullup.rs
+expression: graph.surface_syntax_string()
+---
+1v1 = source_iter ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; 0 .. 10 });
+2v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | v | v + 1 }));
+3v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
+
+1v1 -> 2v1;
+2v1 -> 3v1;
diff --git a/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators-2.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators-2.snap
new file mode 100644
index 000000000000..627f53dd38ed
--- /dev/null
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators-2.snap
@@ -0,0 +1,26 @@
+---
+source: hydroflow_plus/src/rewrites/profiler.rs
+expression: "&pushed_down.ir()"
+---
+[
+    ForEach {
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
+        input: Inspect {
+            f: { use crate :: __staged :: rewrites :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 0u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
+            input: Map {
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | v | v + 1 }),
+                input: Inspect {
+                    f: { use crate :: __staged :: rewrites :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 1u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
+                    input: Source {
+                        source: Iter(
+                            { use crate :: __staged :: rewrites :: profiler :: tests :: * ; 0 .. 10 },
+                        ),
+                        location_kind: Process(
+                            0,
+                        ),
+                    },
+                },
+            },
+        },
+    },
+]
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators.snap
similarity index 59%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators.snap
index c68abed47839..2d59233459b0 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__profiler__tests__profiler_wrapping_all_operators.snap
@@ -1,17 +1,17 @@
 ---
-source: hydroflow_plus/src/profiler.rs
+source: hydroflow_plus/src/rewrites/profiler.rs
 expression: "&built.ir()"
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
         input: Unpersist(
             Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: profiler :: tests :: * ; | v | v + 1 }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: rewrites :: profiler :: tests :: * ; | v | v + 1 }),
                 input: Persist(
                     Source {
                         source: Iter(
-                            { use crate :: __staged :: profiler :: tests :: * ; 0 .. 10 },
+                            { use crate :: __staged :: rewrites :: profiler :: tests :: * ; 0 .. 10 },
                         ),
                         location_kind: Process(
                             0,
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__properties__tests__property_optimized.snap b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__properties__tests__property_optimized.snap
similarity index 54%
rename from hydroflow_plus/src/snapshots/hydroflow_plus__properties__tests__property_optimized.snap
rename to hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__properties__tests__property_optimized.snap
index 2975d05aa318..d793961df8ce 100644
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__properties__tests__property_optimized.snap
+++ b/hydroflow_plus/src/rewrites/snapshots/hydroflow_plus__rewrites__properties__tests__property_optimized.snap
@@ -1,18 +1,18 @@
 ---
-source: hydroflow_plus/src/properties.rs
+source: hydroflow_plus/src/rewrites/properties.rs
 expression: built.ir()
 ---
 [
     ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < (std :: string :: String , i32) , () > ({ use crate :: __staged :: properties :: tests :: * ; | (string , count) | println ! ("{}: {}" , string , count) }),
+        f: stageleft :: runtime_support :: fn1_type_hint :: < (std :: string :: String , i32) , () > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | (string , count) | println ! ("{}: {}" , string , count) }),
         input: FoldKeyed {
-            init: stageleft :: runtime_support :: fn0_type_hint :: < i32 > ({ use crate :: __staged :: properties :: tests :: * ; | | 0 }),
-            acc: stageleft :: runtime_support :: fn2_borrow_mut_type_hint :: < i32 , () , () > ({ use crate :: __staged :: properties :: tests :: * ; | count : & mut i32 , _ | * count += 1 }),
+            init: stageleft :: runtime_support :: fn0_type_hint :: < i32 > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | | 0 }),
+            acc: stageleft :: runtime_support :: fn2_borrow_mut_type_hint :: < i32 , () , () > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | count : & mut i32 , _ | * count += 1 }),
             input: Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < std :: string :: String , (std :: string :: String , ()) > ({ use crate :: __staged :: properties :: tests :: * ; | string : String | (string , ()) }),
+                f: stageleft :: runtime_support :: fn1_type_hint :: < std :: string :: String , (std :: string :: String , ()) > ({ use crate :: __staged :: rewrites :: properties :: tests :: * ; | string : String | (string , ()) }),
                 input: Source {
                     source: Iter(
-                        { use crate :: __staged :: properties :: tests :: * ; vec ! [] },
+                        { use crate :: __staged :: rewrites :: properties :: tests :: * ; vec ! [] },
                     ),
                     location_kind: Process(
                         0,
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap b/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap
deleted file mode 100644
index 6e05ca49f735..000000000000
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_behind_tee@surface_graph_0.snap
+++ /dev/null
@@ -1,20 +0,0 @@
----
-source: hydroflow_plus/src/persist_pullup.rs
-expression: graph.surface_syntax_string()
----
-1v1 = source_iter ({ use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 });
-2v1 = tee ();
-3v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }));
-4v1 = persist :: < 'static > ();
-5v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
-6v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }));
-7v1 = persist :: < 'static > ();
-8v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
-
-1v1 -> 2v1;
-2v1 -> 3v1;
-3v1 -> 4v1;
-4v1 -> 5v1;
-2v1 -> 6v1;
-6v1 -> 7v1;
-7v1 -> 8v1;
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap b/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap
deleted file mode 100644
index 3974a6828245..000000000000
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__persist_pullup__tests__persist_pullup_through_map@surface_graph_0.snap
+++ /dev/null
@@ -1,10 +0,0 @@
----
-source: hydroflow_plus/src/persist_pullup.rs
-expression: graph.surface_syntax_string()
----
-1v1 = source_iter ({ use crate :: __staged :: persist_pullup :: tests :: * ; 0 .. 10 });
-2v1 = map (stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | v | v + 1 }));
-3v1 = for_each (stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: persist_pullup :: tests :: * ; | n | println ! ("{}" , n) }));
-
-1v1 -> 2v1;
-2v1 -> 3v1;
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map-2.snap b/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map-2.snap
deleted file mode 100644
index c2d399c162f4..000000000000
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map-2.snap
+++ /dev/null
@@ -1,29 +0,0 @@
----
-source: hydroflow_plus/src/profiler.rs
-expression: "&pushed_down.ir"
----
-[
-    ForEach {
-        f: { use crate :: __staged :: profiler :: tests :: * ; | n | println ! ("{}" , n) },
-        input: Map {
-            f: { use crate :: __staged :: profiler :: * ; let counter_queue = Fake ; let my_id_copy1 = 0u32 ; let counters_copy1 = Fake ; let my_id_copy2 = 0u32 ; let counters_copy2 = Fake ; let my_id_copy3 = 0u32 ; let context = & context ; let my_id_copy4 = 0u32 ; let counters = Fake ; let my_id = 0u32 ; { counter_queue . borrow () . unbounded_send ((my_id_copy1 as usize , counters_copy1 . borrow () [my_id_copy2 as usize])) . unwrap () ; counters_copy2 . borrow_mut () [my_id_copy3 as usize] = 0 ; move | v | { hydroflow_plus :: profiler :: increment_counter (context . current_tick () , my_id_copy4 , & mut counters . borrow_mut () [my_id as usize] ,) ; v } } },
-            input: Map {
-                f: { use crate :: __staged :: profiler :: tests :: * ; | v | v + 1 },
-                input: Map {
-                    f: { use crate :: __staged :: profiler :: * ; let counter_queue = Fake ; let my_id_copy1 = 1u32 ; let counters_copy1 = Fake ; let my_id_copy2 = 1u32 ; let counters_copy2 = Fake ; let my_id_copy3 = 1u32 ; let context = & context ; let my_id_copy4 = 1u32 ; let counters = Fake ; let my_id = 1u32 ; { counter_queue . borrow () . unbounded_send ((my_id_copy1 as usize , counters_copy1 . borrow () [my_id_copy2 as usize])) . unwrap () ; counters_copy2 . borrow_mut () [my_id_copy3 as usize] = 0 ; move | v | { hydroflow_plus :: profiler :: increment_counter (context . current_tick () , my_id_copy4 , & mut counters . borrow_mut () [my_id as usize] ,) ; v } } },
-                    input: Persist(
-                        Map {
-                            f: { use crate :: __staged :: profiler :: * ; let counter_queue = Fake ; let my_id_copy1 = 2u32 ; let counters_copy1 = Fake ; let my_id_copy2 = 2u32 ; let counters_copy2 = Fake ; let my_id_copy3 = 2u32 ; let context = & context ; let my_id_copy4 = 2u32 ; let counters = Fake ; let my_id = 2u32 ; { counter_queue . borrow () . unbounded_send ((my_id_copy1 as usize , counters_copy1 . borrow () [my_id_copy2 as usize])) . unwrap () ; counters_copy2 . borrow_mut () [my_id_copy3 as usize] = 0 ; move | v | { hydroflow_plus :: profiler :: increment_counter (context . current_tick () , my_id_copy4 , & mut counters . borrow_mut () [my_id as usize] ,) ; v } } },
-                            input: Source {
-                                source: Iter(
-                                    { use crate :: __staged :: profiler :: tests :: * ; 0 .. 10 },
-                                ),
-                                location_id: 0,
-                            },
-                        },
-                    ),
-                },
-            },
-        },
-    },
-]
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map.snap b/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map.snap
deleted file mode 100644
index 9fdb9d06e7ff..000000000000
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__predicate_pushdown_through_map.snap
+++ /dev/null
@@ -1,20 +0,0 @@
----
-source: hydroflow_plus/src/profiler.rs
-expression: "&built.ir"
----
-[
-    ForEach {
-        f: { use crate :: __staged :: profiler :: tests :: * ; | n | println ! ("{}" , n) },
-        input: Map {
-            f: { use crate :: __staged :: profiler :: tests :: * ; | v | v + 1 },
-            input: Persist(
-                Source {
-                    source: Iter(
-                        { use crate :: __staged :: profiler :: tests :: * ; 0 .. 10 },
-                    ),
-                    location_id: 0,
-                },
-            ),
-        },
-    },
-]
diff --git a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators-2.snap b/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators-2.snap
deleted file mode 100644
index b2a4cfe8e7a0..000000000000
--- a/hydroflow_plus/src/snapshots/hydroflow_plus__profiler__tests__profiler_wrapping_all_operators-2.snap
+++ /dev/null
@@ -1,26 +0,0 @@
----
-source: hydroflow_plus/src/profiler.rs
-expression: "&pushed_down.ir()"
----
-[
-    ForEach {
-        f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , () > ({ use crate :: __staged :: profiler :: tests :: * ; | n | println ! ("{}" , n) }),
-        input: Inspect {
-            f: { use crate :: __staged :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 0u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
-            input: Map {
-                f: stageleft :: runtime_support :: fn1_type_hint :: < i32 , i32 > ({ use crate :: __staged :: profiler :: tests :: * ; | v | v + 1 }),
-                input: Inspect {
-                    f: { use crate :: __staged :: profiler :: * ; let counter_queue = Fake ; let counters = Fake ; let my_id = 1u32 ; { counter_queue . borrow () . unbounded_send ((my_id as usize , counters . borrow () [my_id as usize])) . unwrap () ; counters . borrow_mut () [my_id as usize] = 0 ; move | _ | { myself :: increment_counter (& mut counters . borrow_mut () [my_id as usize]) ; } } },
-                    input: Source {
-                        source: Iter(
-                            { use crate :: __staged :: profiler :: tests :: * ; 0 .. 10 },
-                        ),
-                        location_kind: Process(
-                            0,
-                        ),
-                    },
-                },
-            },
-        },
-    },
-]