Repository: hhenson/hgraph (language: python)

Issue:
Merged operators



Discussion / hints:
## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=hhenson_hgraph&pullRequest=49) **Quality Gate failed**  
Failed conditions  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/failed-16px.png '') [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=hhenson_hgraph&pullRequest=49) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=hhenson_hgraph&pullRequest=49)

##   
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png '') Catch issues before they fail your Quality Gate with our IDE extension ![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png '') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)



Fix patch:
diff --git a/examples/web_ui/layouts/random.json b/examples/web_ui/layouts/random.json
index faef6f84..612bb461 100644
--- a/examples/web_ui/layouts/random.json
+++ b/examples/web_ui/layouts/random.json
@@ -1,1 +1,1 @@
-{"sizes":[0.10919970082273747,0.8908002991772626],"detail":{"main":{"type":"split-area","orientation":"vertical","children":[{"type":"split-area","orientation":"horizontal","children":[{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_1"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_4"],"currentIndex":0},{"type":"split-area","orientation":"vertical","children":[{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_5"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_7"],"currentIndex":0}],"sizes":[0.5,0.5]}],"sizes":[0.3333333333333333,0.3333333333333333,0.3333333333333333]},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_2"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_3"],"currentIndex":0}],"sizes":[0.33499677581197235,0.3325016120940138,0.3325016120940138]}},"mode":"globalFilters","master":{"widgets":["PERSPECTIVE_GENERATED_ID_0"],"sizes":[1]},"viewers":{"PERSPECTIVE_GENERATED_ID_0":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":null,"group_by":["sensor"],"split_by":[],"columns":["' '"],"filter":[],"sort":[["' '","asc"]],"expressions":{"' '":"' '"},"aggregates":{"' '":"any"},"master":true,"table":"config","linked":false,"selectable":""},"PERSPECTIVE_GENERATED_ID_1":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":"values and events","group_by":[],"split_by":[],"columns":["sensor","value","events"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"data","linked":false},"PERSPECTIVE_GENERATED_ID_4":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":"CONFIG","group_by":[],"split_by":[],"columns":["sensor","initial","randomness","trend"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"config","linked":false},"PERSPECTIVE_GENERATED_ID_5":{"version":"2.10.1","plugin":"Y Bar","plugin_config":{},"columns_config":{},"settings":false,"theme":"Pro Light","title":"data rates","group_by":["minutes"],"split_by":[],"columns":["value"],"filter":[],"sort":[],"expressions":{"minutes":"bucket(\"time\", 'm')"},"aggregates":{"value":"count"},"master":false,"table":"data_history","linked":false},"PERSPECTIVE_GENERATED_ID_7":{"version":"2.10.1","plugin":"Y Line","plugin_config":{},"columns_config":{},"settings":false,"theme":"Pro Light","title":"lag","group_by":[],"split_by":[],"columns":["value"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"engine_lag_history","linked":false},"PERSPECTIVE_GENERATED_ID_2":{"version":"2.10.1","plugin":"Y Line","plugin_config":{"zoom":{"k":1,"x":0,"y":0}},"columns_config":{},"settings":false,"theme":"Pro Light","title":"value history","group_by":["time"],"split_by":["sensor"],"columns":["value"],"filter":[],"sort":[],"expressions":{},"aggregates":{"value":"last"},"master":false,"table":"data_history","linked":false},"PERSPECTIVE_GENERATED_ID_3":{"version":"2.10.1","plugin":"Y Bar","plugin_config":{"zoom":{"k":1,"x":0,"y":0}},"columns_config":{},"settings":false,"theme":"Pro Light","title":"event history","group_by":["time"],"split_by":["sensor"],"columns":["events"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"data_history","linked":false}}}
\ No newline at end of file
+{"sizes":[0,1],"detail":{"main":{"type":"split-area","orientation":"vertical","children":[{"type":"split-area","orientation":"horizontal","children":[{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_1"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_4"],"currentIndex":0},{"type":"split-area","orientation":"vertical","children":[{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_5"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_7"],"currentIndex":0}],"sizes":[0.5,0.5]}],"sizes":[0.3333333333333333,0.3333333333333333,0.3333333333333333]},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_2"],"currentIndex":0},{"type":"tab-area","widgets":["PERSPECTIVE_GENERATED_ID_3"],"currentIndex":0}],"sizes":[0.45825426944971537,0.27087286527514226,0.27087286527514226]}},"mode":"globalFilters","master":{"widgets":["PERSPECTIVE_GENERATED_ID_0"],"sizes":[1]},"viewers":{"PERSPECTIVE_GENERATED_ID_0":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":null,"group_by":["sensor"],"split_by":[],"columns":["' '"],"filter":[],"sort":[["' '","asc"]],"expressions":{"' '":"' '"},"aggregates":{"' '":"any"},"master":true,"table":"config","linked":false,"selectable":""},"PERSPECTIVE_GENERATED_ID_1":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":"values and events","group_by":[],"split_by":[],"columns":["sensor","value","events"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"data","linked":false},"PERSPECTIVE_GENERATED_ID_4":{"version":"2.10.1","plugin":"Datagrid","plugin_config":{"columns":{},"editable":false,"scroll_lock":false},"columns_config":{},"settings":false,"theme":"Pro Light","title":"CONFIG","group_by":[],"split_by":[],"columns":["sensor","initial","randomness","trend"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"config","linked":false},"PERSPECTIVE_GENERATED_ID_5":{"version":"2.10.1","plugin":"Y Bar","plugin_config":{},"columns_config":{},"settings":false,"theme":"Pro Light","title":"data rates","group_by":["minutes"],"split_by":[],"columns":["value"],"filter":[],"sort":[],"expressions":{"minutes":"bucket(\"time\", 'm')"},"aggregates":{"value":"count"},"master":false,"table":"data_history","linked":false},"PERSPECTIVE_GENERATED_ID_7":{"version":"2.10.1","plugin":"Y Line","plugin_config":{},"columns_config":{},"settings":false,"theme":"Pro Light","title":"lag","group_by":[],"split_by":[],"columns":["value"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"engine_lag_history","linked":false},"PERSPECTIVE_GENERATED_ID_2":{"version":"2.10.1","plugin":"Y Line","plugin_config":{"zoom":{"k":1,"x":0,"y":0}},"columns_config":{},"settings":false,"theme":"Pro Light","title":"value history","group_by":["time"],"split_by":["sensor"],"columns":["value"],"filter":[],"sort":[],"expressions":{},"aggregates":{"value":"last"},"master":false,"table":"data_history","linked":false},"PERSPECTIVE_GENERATED_ID_3":{"version":"2.10.1","plugin":"Y Bar","plugin_config":{"zoom":{"k":1,"x":0,"y":0}},"columns_config":{},"settings":false,"theme":"Pro Light","title":"event history","group_by":["time"],"split_by":["sensor"],"columns":["events"],"filter":[],"sort":[],"expressions":{},"aggregates":{},"master":false,"table":"data_history","linked":false}}}
\ No newline at end of file
diff --git a/examples/web_ui/web_ui.py b/examples/web_ui/web_ui.py
index ea547ad9..b4ea4ed1 100644
--- a/examples/web_ui/web_ui.py
+++ b/examples/web_ui/web_ui.py
@@ -42,6 +42,7 @@
 )
 from hgraph.adaptors.perspective._perspective import perspective_web
 from hgraph.debug import trace_controller
+from hgraph.debug._inspector_controller import inspector_controller
 
 
 class Readings(TimeSeriesSchema):
@@ -141,6 +142,7 @@ def host_web_server():
     )
 
     trace_controller()
+    inspector_controller()
 
 @compute_node
 def wall_clock_time(ts: SIGNAL) -> TS[datetime]:
diff --git a/pyproject.toml b/pyproject.toml
index f86ef8ce..7f62d5cc 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,7 +39,8 @@ packages = [
     polars = ">=1.0"
     sqlalchemy = "*"
     duckdb = "*"
-    pyarrow = "*"
+    pyarrow = ">=16.1.0"
+    multimethod = "*"
     tornado = { version = "*", optional = true }
     requests = { version = "*", optional = true }
 
diff --git a/src/hgraph/_impl/_operators/_flow_control.py b/src/hgraph/_impl/_operators/_flow_control.py
index 286fec99..783cd33e 100644
--- a/src/hgraph/_impl/_operators/_flow_control.py
+++ b/src/hgraph/_impl/_operators/_flow_control.py
@@ -49,19 +49,25 @@ class _RaceState(CompoundScalar):
     winner: REF[OUT] = None
 
 
-@compute_node(overloads=race)
+@compute_node(overloads=race, active=("tsl",))
 def race_default(
     *tsl: TSL[REF[OUT], SIZE],
+    _values: TSL[OUT, SIZE] = None,
     _state: STATE[_RaceState] = None,
     _ec: EvaluationClock = None,
     _sz: type[SIZE] = AUTO_RESOLVE,
 ) -> REF[OUT]:
 
     # Keep track of the first time each input goes valid (and invalid)
+    pending_refs = [None] * _sz.SIZE
+
     for i in range(_sz.SIZE):
         if _tsl_ref_item_valid(tsl, i):
             if i not in _state.first_valid_times:
                 _state.first_valid_times[i] = _ec.now
+        elif tsl[i].valid:  # valid reference but invalid referee
+            pending_refs[i] = tsl[i].value
+            _state.first_valid_times.pop(i, None)
         else:
             _state.first_valid_times.pop(i, None)
 
@@ -72,7 +78,15 @@ def race_default(
         winner = min(_state.first_valid_times.items(), default=None, key=lambda item: item[1])
         if winner is not None:
             _state.winner = winner[0]
+            for v in _values:  # make all values passive and disconnect now that we have a winner
+                v.make_passive()
+                PythonTimeSeriesReference().bind_input(v)
             return tsl[_state.winner].value
+        else:  # if no winner, track timeseries where we have reference but no value
+            for i, r in enumerate(pending_refs):
+                if r is not None and not _values[i].active:
+                    r.bind_input(_values[i])
+                    _values[i].make_active()
     elif tsl[_state.winner].modified:
         # Forward the winning value
         return tsl[_state.winner].delta_value
diff --git a/src/hgraph/_impl/_operators/_stream_operators.py b/src/hgraph/_impl/_operators/_stream_operators.py
index 225383c2..67f31c01 100644
--- a/src/hgraph/_impl/_operators/_stream_operators.py
+++ b/src/hgraph/_impl/_operators/_stream_operators.py
@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from datetime import timedelta, datetime
 
+from frozendict import frozendict
+
 from hgraph import (
     compute_node,
     TIME_SERIES_TYPE,
@@ -34,7 +36,7 @@
     lag,
     TSL,
     SIZE,
-    INT_OR_TIME_DELTA,
+    INT_OR_TIME_DELTA, REMOVE_IF_EXISTS,
 )
 
 __all__ = ()
@@ -126,8 +128,8 @@ def schedule_ts(
 
 
 @graph(overloads=resample)
-def resample(ts: TIME_SERIES_TYPE, delay: timedelta) -> TIME_SERIES_TYPE:
-    return sample(schedule(delay), ts)
+def resample(ts: TIME_SERIES_TYPE, period: timedelta) -> TIME_SERIES_TYPE:
+    return sample(schedule(period), ts)
 
 
 @compute_node(overloads=dedup)
@@ -135,11 +137,32 @@ def drop_dups_default(ts: TIME_SERIES_TYPE, _output: TIME_SERIES_TYPE = None) ->
     """
     Drops duplicate values from a time-series.
     """
-    if _output.valid:
-        if ts.value != _output.value:
-            return ts.delta_value
-    else:
-        return ts.value
+    from multimethod import multimethod
+    from hgraph import PythonTimeSeriesValueInput, PythonTimeSeriesValueOutput
+    from hgraph import PythonTimeSeriesDictInput, PythonTimeSeriesDictOutput
+
+    @multimethod
+    def dedup_item(input, output):
+        return {k_new: v_new for k_new, v_new in
+                ((k, dedup_item(v, output[k])) for k, v in input.modified_items())
+                if v_new is not None}
+
+    @dedup_item.register
+    def dedup_dicts(input: PythonTimeSeriesDictInput, output):
+        out = {k_new: v_new for k_new, v_new in
+                ((k, dedup_item(v, output.get_or_create(k))) for k, v in input.modified_items())
+                if v_new is not None}
+        return out | {k: REMOVE_IF_EXISTS for k in input.removed_keys()}
+
+    @dedup_item.register
+    def dedup_value(input: PythonTimeSeriesValueInput, output):
+        if output.valid:
+            if input.value != output.value:
+                return input.delta_value
+        else:
+            return input.value
+
+    return dedup_item(ts, _output)
 
 
 @compute_node(overloads=dedup)
@@ -165,18 +188,40 @@ def throttle(ts: TIME_SERIES_TYPE,
              period: TS[timedelta],
              sched: SCHEDULER = None,
              state: STATE = None) -> TIME_SERIES_TYPE:
-    if ts.modified:
-        state.queue.append(ts.value)
-    if sched.is_scheduled_now:
-        sched.schedule(period.value)
-        if state.queue:
-            return state.queue.popleft()
+    from multimethod import multimethod
+    from hgraph import PythonTimeSeriesValueInput
+    from hgraph import PythonTimeSeriesDictInput
+
+    @multimethod
+    def collect_tick(input, out):
+        return {k_new: v_new for k_new, v_new in
+                ((k, collect_tick(v, out.setdefault(k, dict()))) for k, v in input.modified_items())
+                if v_new is not None}
+
+    @collect_tick.register
+    def collect_dict(input: PythonTimeSeriesDictInput, out):
+        out |= {k_new: v_new for k_new, v_new in
+               ((k, collect_tick(v, out.setdefault(k, dict()))) for k, v in input.modified_items())
+               if v_new is not None}
+        return out | {k: REMOVE_IF_EXISTS for k in input.removed_keys()}
+
+    @collect_tick.register
+    def collect_value(input: PythonTimeSeriesValueInput, out):
+        return input.value
 
+    if ts.modified:
+        if sched.is_scheduled:
+            state.tick = collect_tick(ts, state.tick)
+        else:
+            state.tick = {}
+            sched.schedule(period.value)
+            return ts.value
 
-@throttle.start
-def throttle_start(state: STATE, sched: SCHEDULER):
-    state.queue = deque(maxlen=1)
-    sched.schedule(MIN_TD)
+    if sched.is_scheduled_now:
+        if tick := state.tick:
+            state.tick = {}
+            sched.schedule(period.value)
+            return tick
 
 
 @dataclass
diff --git a/src/hgraph/_impl/_runtime/_component_node.py b/src/hgraph/_impl/_runtime/_component_node.py
index d065cf0f..64618b79 100644
--- a/src/hgraph/_impl/_runtime/_component_node.py
+++ b/src/hgraph/_impl/_runtime/_component_node.py
@@ -74,7 +74,7 @@ def initialise(self):
         self._wire_graph()
 
     @start_guard
-    def start(self):
+    def do_start(self):
         if self._active_graph:
             self._active_graph.start()
         else:
@@ -83,7 +83,7 @@ def start(self):
                 self.graph.schedule_node(self.node_ndx, self.graph.evaluation_clock.evaluation_time)
 
     @stop_guard
-    def stop(self):
+    def do_stop(self):
         if self._active_graph:
             self._active_graph.stop()
 
@@ -101,8 +101,13 @@ def eval(self):
                 return
 
         self.mark_evaluated()
+        self._active_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
         self._active_graph.evaluate_graph()
 
+    def enum_nested_graphs(self):
+        if self._active_graph:
+            yield None, self._active_graph
+
     def recordable_id(self) -> tuple[str, bool]:
         """The id and True or no id and False if required inputs are not ready yet"""
         outer_id = self.graph.traits.get_trait_or("recordable_id")
diff --git a/src/hgraph/_impl/_runtime/_evaluation_clock.py b/src/hgraph/_impl/_runtime/_evaluation_clock.py
index b214d72b..4c70ccb3 100644
--- a/src/hgraph/_impl/_runtime/_evaluation_clock.py
+++ b/src/hgraph/_impl/_runtime/_evaluation_clock.py
@@ -3,7 +3,7 @@
 from datetime import datetime, timedelta
 from threading import Condition
 
-from hgraph._runtime._constants import MAX_DT, MIN_TD
+from hgraph._runtime._constants import MAX_DT, MIN_TD, MIN_DT
 from hgraph._runtime._evaluation_clock import EngineEvaluationClock
 
 
@@ -99,6 +99,7 @@ def __init__(self, start_time: datetime):
         self._push_node_requires_scheduling = False
         self._push_node_requires_scheduling_condition = Condition()
         self._ready_to_push: bool = False
+        self._last_time_allowed_push: datetime = MIN_DT
 
     @property
     def now(self) -> datetime:
@@ -129,19 +130,24 @@ def advance_to_next_scheduled_time(self):
         next_scheduled_time = self.next_scheduled_evaluation_time
 
         # We only let push values to be introduced when there are no feedback entries left otherwise go compute those
+        # however if we've been busy like that for more than a second, let push values in.
+        now = datetime.utcnow()
+
         self._ready_to_push = False
-        if next_scheduled_time > self.evaluation_time + MIN_TD:
+        if next_scheduled_time > self.evaluation_time + MIN_TD or now > self._last_time_allowed_push + timedelta(seconds=1):
             with self._push_node_requires_scheduling_condition:
                 self._ready_to_push = True
-                while datetime.utcnow() < next_scheduled_time and not self._push_node_requires_scheduling:
-                    sleep_time = (next_scheduled_time - datetime.utcnow()).total_seconds()
+                self._last_time_allowed_push = now
+                while now < next_scheduled_time and not self._push_node_requires_scheduling:
+                    sleep_time = (next_scheduled_time - now).total_seconds()
                     # print(f"RealTimeEvaluationClock.advance_to_next_scheduled_time: sleeping for {sleep_time}", file=sys.stderr)
                     self._push_node_requires_scheduling_condition.wait(
                         min(sleep_time, 10)
                     )  # wake up regularly so sleep_time is not 100 years
+                    now = datetime.utcnow()
             # It could be that a push node has triggered
             # print(f"RealTimeEvaluationClock.advance_to_next_scheduled_time: setting evaluation time to {next_scheduled_time}", file=sys.stderr)
-        self.evaluation_time = min(next_scheduled_time, max(self.next_cycle_evaluation_time, datetime.utcnow()))
+        self.evaluation_time = min(next_scheduled_time, max(self.next_cycle_evaluation_time, now))
 
     def reset_push_node_requires_scheduling(self):
         """
diff --git a/src/hgraph/_impl/_runtime/_graph.py b/src/hgraph/_impl/_runtime/_graph.py
index cd89bf08..b075eaab 100644
--- a/src/hgraph/_impl/_runtime/_graph.py
+++ b/src/hgraph/_impl/_runtime/_graph.py
@@ -11,6 +11,8 @@
 from hgraph._runtime._node import NodeTypeEnum, Node
 from hgraph._runtime._traits import Traits
 
+from hgraph._impl._runtime._node import _SenderReceiverState
+
 if typing.TYPE_CHECKING:
     from hgraph._builder._graph_builder import GraphBuilder
 
@@ -36,6 +38,12 @@ def __init__(self, graph_id: tuple[int, ...], nodes: Iterable[Node], parent_node
         else:
             self._traits: Traits = Traits(parent=parent_node.graph.traits)
 
+        self._receiver: _SenderReceiverState = None
+
+        self._scheduled_times = []
+        self._evaluated_times = []
+        self._evaluated_node_times = []
+
     def copy_with(self, nodes: tuple[Node, ...]) -> "Graph":
         graph = PythonGraph(self._graph_id, nodes, self._parent_node)
         graph._schedule = self._schedule
@@ -81,6 +89,13 @@ def evaluation_engine(self, value):
             raise RuntimeError("Duplicate attempt to set evaluation engine")
         self._evaluation_engine = value
 
+        if self.push_source_nodes_end > 0:
+            self._receiver = _SenderReceiverState(evaluation_clock=self.engine_evaluation_clock)
+
+    @property
+    def receiver(self) -> _SenderReceiverState:
+        return self._receiver
+
     @property
     def schedule(self) -> list[datetime, ...]:
         return self._schedule
@@ -146,6 +161,7 @@ def schedule_node(self, node_ndx, when, force_set: bool = False):
         if force_set or st <= et or st > when:
             self.schedule[node_ndx] = when
         clock.update_next_scheduled_evaluation_time(when)
+        self._scheduled_times.append((node_ndx, et, when, force_set or st <= et or st > when, clock.next_scheduled_evaluation_time))
 
     def start_subgraph(self, start: int, end: int):
         """Start the subgraph (end is exclusive), i.e. [start, end)"""
@@ -199,21 +215,30 @@ def evaluate_graph(self):
         self._evaluation_engine.notify_before_graph_evaluation(self)
 
         now = (clock := self._evaluation_engine.engine_evaluation_clock).evaluation_time
+        self._evaluated_times.append(now)
         nodes = self._nodes
         schedule = self._schedule
 
         if self.push_source_nodes_end > 0 and clock.push_node_requires_scheduling:
             clock.reset_push_node_requires_scheduling()
-            for i in range(self.push_source_nodes_end):
-                node = nodes[i]
-                node.eval()  # This is only to move nodes on, won't call the before and after node eval here
-                if node.output.modified:
-                    self._evaluation_engine.notify_after_node_evaluation(node)
+            while self.receiver:
+                i, message = self.receiver.dequeue()
+                node = self.nodes[i]
+                self._evaluation_engine.notify_before_node_evaluation(node)
+                stop = node.apply_message(message)
+                self._evaluation_engine.notify_after_node_evaluation(node)
+                if stop:
+                    with self.receiver:
+                        # stop == True means the message was not applied, put it back up the queue.
+                        self.receiver.queue.appendleft((i, message))
+                        self.engine_evaluation_clock.mark_push_node_requires_scheduling()
+                    break
 
         for i in range(self.push_source_nodes_end, len(nodes)):
             scheduled_time, node = schedule[i], nodes[i]
             if scheduled_time == now:
                 self._evaluation_engine.notify_before_node_evaluation(node)
+                self._evaluated_node_times.append((node.node_ndx, now))
                 from hgraph._types._error_type import NodeException
 
                 try:
@@ -223,6 +248,7 @@ def evaluate_graph(self):
                 except Exception as e:
                     raise NodeException.capture_error(e, node, "During evaluation") from e
                 self._evaluation_engine.notify_after_node_evaluation(node)
+                self._evaluated_node_times.append((node.node_ndx, now, schedule[i], clock.next_scheduled_evaluation_time))
             elif scheduled_time > now:
                 # If the node has a scheduled time in the future, we need to let the execution context know.
                 clock.update_next_scheduled_evaluation_time(scheduled_time)
diff --git a/src/hgraph/_impl/_runtime/_map_node.py b/src/hgraph/_impl/_runtime/_map_node.py
index 559948a7..b7d5d25a 100644
--- a/src/hgraph/_impl/_runtime/_map_node.py
+++ b/src/hgraph/_impl/_runtime/_map_node.py
@@ -31,7 +31,7 @@ def __init__(self, engine_evaluation_clock: EngineEvaluationClock, key: K, neste
     def update_next_scheduled_evaluation_time(self, next_time: datetime):
         # First we make sure the key is correctly scheduled, then we call super, which will ensure the
         # node is scheduled if required.
-        if next_time <= self._nested_node.last_evaluation_time:
+        if next_time <= self._nested_node.last_evaluation_time or self._nested_node.is_stopping:
             return
         tm = self._nested_node._scheduled_keys.get(self._key)
         if tm is None or tm > next_time:
@@ -65,7 +65,7 @@ def __init__(
         self._scheduled_keys: dict[K, datetime] = {}
         self._active_graphs: dict[K, Graph] = {}
         self._pending_keys: set[K] = set()
-        self._count = 0
+        self._count = 1
 
     def eval(self):
         self.mark_evaluated()
@@ -94,9 +94,19 @@ def eval(self):
                 self._scheduled_keys[k] = dt
                 self.graph.schedule_node(self.node_ndx, dt)
 
+    def do_stop(self):
+        for k in list(self._active_graphs.keys()):
+            self._remove_graph(k)
+        self._active_graphs.clear()
+        self._scheduled_keys.clear()
+        self._pending_keys.clear()
+
+    def enum_nested_graphs(self):
+        return self._active_graphs.items()
+
     def _create_new_graph(self, key: K):
         """Create new graph instance and wire it into the node"""
-        graph: Graph = self.nested_graph_builder.make_instance(self.node_id + (self._count,), self, str(key))
+        graph: Graph = self.nested_graph_builder.make_instance(self.node_id + (-self._count,), self, str(key))
         self._count += 1
         self._active_graphs[key] = graph
         graph.evaluation_engine = NestedEvaluationEngine(
@@ -121,6 +131,7 @@ def _remove_graph(self, key: K):
     def _evaluate_graph(self, key: K):
         """Evaluate the graph for this key"""
         graph: Graph = self._active_graphs[key]
+        graph.evaluation_clock.reset_next_scheduled_evaluation_time()
         # TODO: This can be done at start time or initialisation time (decide on behaviour) and then we don't
         # have to pay for the if.
         if self.signature.capture_exception:
diff --git a/src/hgraph/_impl/_runtime/_mesh_node.py b/src/hgraph/_impl/_runtime/_mesh_node.py
index 9fbcee1d..9a8c3f2f 100644
--- a/src/hgraph/_impl/_runtime/_mesh_node.py
+++ b/src/hgraph/_impl/_runtime/_mesh_node.py
@@ -2,25 +2,23 @@
 from datetime import datetime
 from typing import Mapping, Any, Callable, cast
 
-from hgraph import MAX_DT, PythonTsdMapNodeImpl, GlobalState, PythonTimeSeriesReference, MIN_TD, MIN_DT
+from hgraph import MAX_DT, PythonTsdMapNodeImpl, GlobalState, PythonTimeSeriesReference, MIN_TD
 from hgraph._builder._graph_builder import GraphBuilder
 from hgraph._impl._runtime._nested_evaluation_engine import (
     NestedEngineEvaluationClock,
     NestedEvaluationEngine,
-    PythonNestedNodeImpl,
 )
-from hgraph._impl._runtime._node import NodeImpl
 from hgraph._runtime._evaluation_clock import EngineEvaluationClock
 from hgraph._runtime._graph import Graph
-from hgraph._runtime._node import Node, NodeSignature
+from hgraph._runtime._node import NodeSignature
 from hgraph._types._error_type import NodeError
-from hgraph._types._time_series_types import TIME_SERIES_TYPE, K
+from hgraph._types._time_series_types import K
 from hgraph._types._ts_type import TS
-from hgraph._types._tsd_type import TSD, TSD_OUT
+from hgraph._types._tsd_type import TSD_OUT
 from hgraph._types._tss_type import TSS
 from hgraph._wiring._map import KEYS_ARG
 
-__all__ = ("PythonTsdMapNodeImpl",)
+__all__ = ("PythonMeshNodeImpl",)
 
 
 class MeshNestedEngineEvaluationClock(NestedEngineEvaluationClock):
@@ -92,19 +90,27 @@ def __init__(
         self._active_graphs_rank: dict[K, int] = {}
         self._active_graphs_dependencies: dict[K, set[K]] = defaultdict(set)
         self._re_rank_requests: list[tuple[K, K]] = []
+        self._graphs_to_remove: set[K] = set()
         self.current_eval_rank = None
         self.current_eval_graph = None
         self.max_rank = 0
 
-    def start(self):
-        super().start()
+        self._scheduled_times = defaultdict(list)
+        self._evaluated_times = defaultdict(list)
+        self._re_rank_times = defaultdict(list)
+
+    def do_start(self):
+        super().do_start()
+
         self.output["ref"].value = PythonTimeSeriesReference(self.output["out"])
         GlobalState.instance()[self._full_context_path] = self.output["ref"]
         self._output = self.output["out"]
 
-    def stop(self):
+    def do_stop(self):
         del GlobalState.instance()[self._full_context_path]
 
+        super().do_stop()
+
     def eval(self):
         self.mark_evaluated()
         # 1. All inputs should be reference, and we should only be active on the KEYS_ARG input
@@ -117,6 +123,7 @@ def eval(self):
                 if not self._active_graphs_dependencies.get(k, None):
                     self._scheduled_keys_by_rank[self._active_graphs_rank[k]].pop(k, None)
                     self._remove_graph(k)
+
         if self._pending_keys:
             for k in self._pending_keys:
                 self._create_new_graph(k, rank=0)
@@ -124,6 +131,13 @@ def eval(self):
                     self._re_rank(d, k)
             self._pending_keys = set()
 
+        if self._graphs_to_remove:
+            for k in set(self._graphs_to_remove):
+                if not self._active_graphs_dependencies[k] and k not in self._kwargs[KEYS_ARG]:
+                    # no more dependencies and this was not asked for externally (through the keys)
+                    self._remove_graph(k)
+            self._graphs_to_remove = set()
+
         # 2. or one of the nested graphs has been scheduled for evaluation.
         next_time = MAX_DT
         rank = 0
@@ -132,8 +146,10 @@ def eval(self):
             dt = self._scheduled_ranks.pop(rank, None)
             if dt == self.last_evaluation_time:
                 graphs = self._scheduled_keys_by_rank.pop(rank, {})
+                self._evaluated_times[rank].append(dt)
                 for k, dtg in graphs.items():
                     if dtg == dt:
+                        self._evaluated_times[k].append(dt)
                         self.current_eval_graph = k
                         dtg = self._evaluate_graph(k)
                         self.current_eval_graph = None
@@ -143,6 +159,7 @@ def eval(self):
                         next_time = min(next_time, dtg)
             elif dt and dt > self.last_evaluation_time:
                 self._scheduled_ranks[rank] = dt
+                next_time = min(next_time, dt)
 
             rank += 1
 
@@ -181,17 +198,28 @@ def schedule_graph(self, key, tm):
         )
         self.graph.schedule_node(self.node_ndx, tm)
 
+        if tm == self.last_evaluation_time and self.current_eval_rank is not None and rank <= self.current_eval_rank:
+            raise ValueError(
+                f"mesh {self.signature.wiring_path_name}.{self.signature.label or self.signature.name} has a"
+                f" dependency cycle {key} -> {key}"
+            )
+
+        self._scheduled_times[rank].append((self.last_evaluation_time, self._scheduled_ranks[rank]))
+        self._scheduled_times[key].append((self.last_evaluation_time, tm, rank))
+
     def _remove_graph(self, key: K):
         """Un-wire graph and schedule for removal"""
         if self.signature.capture_exception:
             # Remove the error output associated to the graph if there is one.
             cast(TSD_OUT[K, TS[NodeError]], self.error_output).pop(key, None)
-        graph: Graph = self._active_graphs.pop(key)
-        self._un_wire_graph(key, graph)
-        graph.stop()
-        self._scheduled_keys_by_rank[self._active_graphs_rank[key]].pop(key, None)
-        self._active_graphs_rank.pop(key)
-        graph.dispose()
+        graph: Graph = self._active_graphs.pop(key, None)
+        if graph is not None:  # None can happen during shutdown as shutdown order is not dependency-driven
+            self._un_wire_graph(key, graph)
+            graph.stop()
+            self._scheduled_keys_by_rank[self._active_graphs_rank[key]].pop(key, None)
+            self._active_graphs_rank.pop(key)
+            self._re_rank_requests = [(k, d) for k, d in self._re_rank_requests if k != key]
+            graph.dispose()
 
     def _add_graph_dependency(self, key: K, depends_on: K) -> bool:  # returns True if the key is available now
         self._active_graphs_dependencies[depends_on].add(key)
@@ -216,6 +244,7 @@ def _re_rank(self, key: K, depends_on: K, re_rank_stack=()):
             new_rank = below + 1
             self.max_rank = max(self.max_rank, new_rank)
             self._active_graphs_rank[key] = new_rank
+            self._re_rank_times[key].append((self.last_evaluation_time, prev_rank, new_rank))
             if schedule:
                 self.schedule_graph(key, schedule)
             for k in self._active_graphs_dependencies[key]:
@@ -231,4 +260,4 @@ def _remove_graph_dependency(self, key: K, depends_on: K):
         self._active_graphs_dependencies[depends_on].remove(key)
         if not self._active_graphs_dependencies[depends_on] and depends_on not in self._kwargs[KEYS_ARG]:
             # no more dependencies and this was not asked for externally (through the keys)
-            self._remove_graph(depends_on)
+            self._graphs_to_remove.add(depends_on)
diff --git a/src/hgraph/_impl/_runtime/_nested_evaluation_engine.py b/src/hgraph/_impl/_runtime/_nested_evaluation_engine.py
index 457ede10..df7412ab 100644
--- a/src/hgraph/_impl/_runtime/_nested_evaluation_engine.py
+++ b/src/hgraph/_impl/_runtime/_nested_evaluation_engine.py
@@ -1,7 +1,8 @@
+from abc import abstractmethod
 from datetime import datetime
 from typing import Mapping, Any, Callable
 
-from hgraph._runtime._constants import MIN_DT
+from hgraph._runtime._constants import MIN_DT, MAX_DT, MIN_TD
 from hgraph._impl._runtime._node import NodeImpl
 from hgraph._runtime._evaluation_clock import EngineEvaluationClock, EngineEvaluationClockDelegate
 from hgraph._runtime._evaluation_engine import EvaluationEngine, EvaluationEngineDelegate
@@ -20,15 +21,28 @@ class NestedEngineEvaluationClock(EngineEvaluationClockDelegate):
     def __init__(self, engine_evaluation_clock: EngineEvaluationClock, nested_node: "PythonNestedNodeImpl"):
         super().__init__(engine_evaluation_clock)
         self._nested_node = nested_node
+        self._nested_next_scheduled_evaluation_time = MAX_DT
 
     @property
     def node(self) -> "PythonNestedNodeImpl":
         return self._nested_node
 
+    @property
+    def next_scheduled_evaluation_time(self):
+        return self._nested_next_scheduled_evaluation_time
+
+    def reset_next_scheduled_evaluation_time(self):
+        self._nested_next_scheduled_evaluation_time = MAX_DT
+
     def update_next_scheduled_evaluation_time(self, next_time: datetime):
         if (let := self._nested_node.last_evaluation_time) and let >= next_time:
             return
 
+        self._nested_next_scheduled_evaluation_time = min(next_time,
+                                                          max(
+                                                              self._nested_next_scheduled_evaluation_time,
+                                                              (let or MIN_DT) + MIN_TD))
+
         self._nested_node.graph.schedule_node(self._nested_node.node_ndx, next_time)
 
 
@@ -79,3 +93,7 @@ def last_evaluation_time(self) -> datetime:
 
     def mark_evaluated(self):
         self._last_evaluation_time = self.graph.evaluation_clock.evaluation_time
+
+    @abstractmethod
+    def enum_nested_graphs(self):
+        ...
\ No newline at end of file
diff --git a/src/hgraph/_impl/_runtime/_node.py b/src/hgraph/_impl/_runtime/_node.py
index 0ed6b0a1..0c7e56e0 100644
--- a/src/hgraph/_impl/_runtime/_node.py
+++ b/src/hgraph/_impl/_runtime/_node.py
@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 from collections import deque
 from contextlib import ExitStack, nullcontext
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from typing import Optional, Mapping, TYPE_CHECKING, Callable, Any, Iterator
 
@@ -154,6 +154,7 @@ def do_eval(self): ...
 
     def eval(self):
         scheduled = False if self._scheduler is None else self._scheduler.is_scheduled_now
+        eval = True
         if self.input:
             # Perform validity check of inputs
             args = (
@@ -162,36 +163,34 @@ def eval(self):
                 else self.signature.time_series_inputs.keys()
             )
             if not all(self.input[k].valid for k in args):
-                if scheduled:
-                    self._scheduler.advance()
-                return  # We should look into caching the result of this check.
+                eval = False  # We should look into caching the result of this check.
                 # This check could perhaps be set on a separate call?
-            all_valid = self.signature.all_valid_inputs
-            if not all(self.input[k].all_valid for k in ([] if all_valid is None else all_valid)):
-                if scheduled:
-                    self._scheduler.advance()
-                return  # This really could do with some optimisation as on large collections this will be expensive!
-            if self.signature.uses_scheduler:
-                # It is possible we have scheduled and then remove the schedule,
-                # so we need to check that something has caused this to be scheduled.
-                if not scheduled and not any(self.input[k].modified for k in self.signature.time_series_inputs.keys()):
-                    return
-
-        with ExitStack() if self.signature.context_inputs else nullcontext() as stack:
-            if self.signature.context_inputs:
-                for context in self.signature.context_inputs:
-                    if self.input[context].valid:
-                        stack.enter_context(self.input[context].value)
-
-            if self.error_output:
-                try:
-                    self.do_eval()
-                except Exception as e:
-                    from hgraph._types._error_type import NodeError
-
-                    self.error_output.apply_result(NodeError.capture_error(e, self))
             else:
-                self.do_eval()
+                all_valid = self.signature.all_valid_inputs
+                if not all(self.input[k].all_valid for k in ([] if all_valid is None else all_valid)):
+                    eval = False  # This really could do with some optimisation as on large collections this will be expensive!
+                elif self.signature.uses_scheduler:
+                    # It is possible we have scheduled and then remove the schedule,
+                    # so we need to check that something has caused this to be scheduled.
+                    if not scheduled and not any(self.input[k].modified for k in self.signature.time_series_inputs.keys()):
+                        eval = False
+
+        if eval:
+            with ExitStack() if self.signature.context_inputs else nullcontext() as stack:
+                if self.signature.context_inputs:
+                    for context in self.signature.context_inputs:
+                        if self.input[context].valid:
+                            stack.enter_context(self.input[context].value)
+
+                if self.error_output:
+                    try:
+                        self.do_eval()
+                    except Exception as e:
+                        from hgraph._types._error_type import NodeError
+
+                        self.error_output.apply_result(NodeError.capture_error(e, self))
+                else:
+                    self.do_eval()
 
         if scheduled:
             self._scheduler.advance()
@@ -427,31 +426,22 @@ def __init__(self, *args, **kwargs):
     @start_guard
     def start(self):
         self._initialise_kwargs()
-        self.receiver = _SenderReceiverState(
-            lock=threading.RLock(), queue=deque(), evaluation_evaluation_clock=self.graph.engine_evaluation_clock
-        )
-        self.eval_fn(self.receiver, **self._kwargs)
+        self.receiver = self.graph.receiver
+        self.eval_fn(lambda m: self.receiver((self.node_ndx, m)), **self._kwargs)
+        self.elide = self.scalars.get("elide", False)
 
-    def eval(self):
-        with self.receiver:
-            value = self.receiver.dequeue()
-            if value is None:
-                return
-            self.output.apply_result(value)
-            if self.receiver:
-                self.graph.engine_evaluation_clock.mark_push_node_requires_scheduling()
-
-    @stop_guard
-    def stop(self):
-        self.receiver.stopped = True
-        self.receiver = None
+    def apply_message(self, message):  # return True if stop processing further messages
+        if self.elide or self.output.can_apply_result(message):
+            self.output.apply_result(message)
+            return False
+        return True
 
 
 @dataclass
 class _SenderReceiverState:
-    lock: threading.RLock
-    queue: deque
-    evaluation_evaluation_clock: EngineEvaluationClock
+    lock: threading.RLock = field(default_factory=threading.RLock)
+    queue: deque = field(default_factory=deque)
+    evaluation_clock: EngineEvaluationClock = None
     stopped: bool = False
 
     def __call__(self, value):
@@ -462,7 +452,7 @@ def enqueue(self, value):
             if self.stopped:
                 raise RuntimeError("Cannot enqueue into a stopped receiver")
             self.queue.append(value)
-            self.evaluation_evaluation_clock.mark_push_node_requires_scheduling()
+            self.evaluation_clock.mark_push_node_requires_scheduling()
 
     def dequeue(self):
         with self.lock:
diff --git a/src/hgraph/_impl/_runtime/_reduce_node.py b/src/hgraph/_impl/_runtime/_reduce_node.py
index ca09c582..13fa564b 100644
--- a/src/hgraph/_impl/_runtime/_reduce_node.py
+++ b/src/hgraph/_impl/_runtime/_reduce_node.py
@@ -88,6 +88,7 @@ def eval(self):
         # Now we can re-balance the tree if required.
         self._re_balance_nodes()
 
+        self._nested_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
         self._nested_graph.evaluate_graph()
 
         # Now we just need to detect the change in graph shape, so we can propagate it on.
@@ -96,6 +97,9 @@ def eval(self):
         if (o := self.output).value != (v := self._last_output.value):
             o.value = v
 
+    def enum_nested_graphs(self):
+        yield (None, self._nested_graph)
+
     @property
     def _last_output(self):
         sub_graph = self._get_node(self._node_count - 1)
@@ -159,6 +163,7 @@ def _re_balance_nodes(self):
 
     def _evaluate_graph(self):
         """Evaluate the graph for this key"""
+        self._nested_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
         self._nested_graph.evaluate_graph()
 
     @functools.cached_property
diff --git a/src/hgraph/_impl/_runtime/_service_node_impl.py b/src/hgraph/_impl/_runtime/_service_node_impl.py
index 88b2af0c..e24bb46b 100644
--- a/src/hgraph/_impl/_runtime/_service_node_impl.py
+++ b/src/hgraph/_impl/_runtime/_service_node_impl.py
@@ -39,6 +39,7 @@ def do_start(self):
 
     def eval(self):
         self.mark_evaluated()
+        self._active_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
         self._active_graph.evaluate_graph()
 
     def do_stop(self):
@@ -47,3 +48,6 @@ def do_stop(self):
     def dispose(self):
         self._active_graph.dispose()
         self._active_graph = None
+
+    def enum_nested_graphs(self):
+        yield (None, self._active_graph)
\ No newline at end of file
diff --git a/src/hgraph/_impl/_runtime/_switch_node.py b/src/hgraph/_impl/_runtime/_switch_node.py
index 8e689136..34376548 100644
--- a/src/hgraph/_impl/_runtime/_switch_node.py
+++ b/src/hgraph/_impl/_runtime/_switch_node.py
@@ -73,8 +73,13 @@ def eval(self):
                     raise ValueError(f"No graph defined for key {self._active_key}")
 
         if self._active_graph:
+            self._active_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
             self._active_graph.evaluate_graph()
 
+    def enum_nested_graphs(self):
+        if self._active_graph:
+            yield self._active_key, self._active_graph
+
     def _wire_graph(self, graph: Graph):
         """Connect inputs and outputs to the nodes inputs and outputs"""
         from hgraph._wiring._switch import DEFAULT
diff --git a/src/hgraph/_impl/_runtime/_try_except_node.py b/src/hgraph/_impl/_runtime/_try_except_node.py
index c854011c..05e2104a 100644
--- a/src/hgraph/_impl/_runtime/_try_except_node.py
+++ b/src/hgraph/_impl/_runtime/_try_except_node.py
@@ -55,10 +55,10 @@ def initialise(self):
         self._active_graph.initialise()
         self._wire_graph()
 
-    def start(self):
+    def do_start(self):
         self._active_graph.start()
 
-    def stop(self):
+    def do_stop(self):
         self._active_graph.stop()
 
     def dispose(self):
@@ -68,12 +68,18 @@ def dispose(self):
     def eval(self):
         try:
             self.mark_evaluated()
+            self._active_graph.evaluation_clock.reset_next_scheduled_evaluation_time()
             self._active_graph.evaluate_graph()
         except Exception as e:
             from hgraph._types._error_type import NodeError
 
+            self._active_graph.stop()
+
             err = NodeError.capture_error(e, self)
             if type(self.signature.time_series_output) is HgTSBTypeMetaData:
                 self.output.exception.value = err
             else:
                 self.output.value = err
+
+    def enum_nested_graphs(self):
+        yield None, self._active_graph
\ No newline at end of file
diff --git a/src/hgraph/_impl/_types/_input.py b/src/hgraph/_impl/_types/_input.py
index 1198c84b..1bbc551e 100644
--- a/src/hgraph/_impl/_types/_input.py
+++ b/src/hgraph/_impl/_types/_input.py
@@ -179,11 +179,11 @@ def has_peer(self) -> bool:
 
     @property
     def value(self):
-        return self._output.value if self._output else None
+        return self._output.value if self._output is not None else None
 
     @property
     def delta_value(self):
-        return self._output.delta_value if self._output else None
+        return self._output.delta_value if self._output is not None else None
 
     @property
     def modified(self) -> bool:
diff --git a/src/hgraph/_impl/_types/_ref.py b/src/hgraph/_impl/_types/_ref.py
index e7cca601..486a3578 100644
--- a/src/hgraph/_impl/_types/_ref.py
+++ b/src/hgraph/_impl/_types/_ref.py
@@ -75,12 +75,22 @@ def bind_input(self, ts_input: TimeSeriesInput):
         if self.tp and not issubclass(self.tp, TimeSeriesOutput) and not isinstance(ts_input, self.tp):
             raise TypeError(f"Cannot bind reference of type {self.tp} to {type(ts_input)}")
 
+        reactivate = False
+        if ts_input.bound and self.has_peer != ts_input.has_peer:
+            reactivate = ts_input.active
+            ts_input.un_bind_output()
+
         if self.has_peer:
             ts_input.bind_output(self.output)
         else:
             for item, r in zip(ts_input, self.items):
                 if r:
                     r.bind_input(item)
+                elif item.bound:
+                    item.un_bind_output()
+
+        if reactivate:
+            ts_input.make_active()
 
     def __str__(self) -> str:
         if self.output is not None:
@@ -126,6 +136,9 @@ def value(self, v: TimeSeriesReference):
         for observer in self._reference_observers.values():
             self._value.bind_input(observer)
 
+    def can_apply_result(self, result: SCALAR):
+        return not self.modified
+
     def apply_result(self, result: SCALAR):
         if result is None:
             return
@@ -185,7 +198,7 @@ def do_bind_output(self, output: TimeSeriesOutput) -> bool:
             return False
 
     def do_un_bind_output(self):
-        if self._output:
+        if self._output is not None:
             super().do_un_bind_output()
         if self._value is not None:
             self._value = None
diff --git a/src/hgraph/_impl/_types/_ts.py b/src/hgraph/_impl/_types/_ts.py
index a7f39927..0a099587 100644
--- a/src/hgraph/_impl/_types/_ts.py
+++ b/src/hgraph/_impl/_types/_ts.py
@@ -46,6 +46,9 @@ def value(self, v: SCALAR):
     def invalidate(self):
         self.mark_invalid()
 
+    def can_apply_result(self, result: SCALAR):
+        return not self.modified
+
     def apply_result(self, result: SCALAR):
         if result is None:
             return
diff --git a/src/hgraph/_impl/_types/_tsb.py b/src/hgraph/_impl/_types/_tsb.py
index 71f8cca8..7de16e25 100644
--- a/src/hgraph/_impl/_types/_tsb.py
+++ b/src/hgraph/_impl/_types/_tsb.py
@@ -59,6 +59,19 @@ def invalidate(self):
     def delta_value(self):
         return {k: ts.delta_value for k, ts in self.items() if ts.modified}
 
+    def can_apply_result(self, result: Mapping[str, Any] | None):
+        if result is None:
+            return True
+        else:
+            if type(result) is self.__schema__.scalar_type():
+                return self.modified
+            else:
+                for k, v_ in result.items():
+                    if v_ is not None:
+                        if not cast(TimeSeriesOutput, self[k]).can_apply_result(v_):
+                            return False
+        return True
+
     def apply_result(self, result: Mapping[str, Any] | None):
         if result is None:
             return
diff --git a/src/hgraph/_impl/_types/_tsd.py b/src/hgraph/_impl/_types/_tsd.py
index fd8c141f..48f7dddf 100644
--- a/src/hgraph/_impl/_types/_tsd.py
+++ b/src/hgraph/_impl/_types/_tsd.py
@@ -114,6 +114,9 @@ def __delitem__(self, k):
         self._ts_values_to_keys.pop(id(item))
         self._ref_ts_feature.update(k)
         cast(TimeSeriesSetOutput, self.key_set).remove(k)
+        try:
+            self._modified_items.remove((k, item))
+        except: pass
         for observer in self._key_observers:
             observer.on_key_removed(k)
 
@@ -140,11 +143,33 @@ def pop(self, key: K) -> V:
             del self[key]
         return v
 
+    def key_from_value(self, value: V) -> K:
+        return self._ts_values_to_keys[id(value)]
+
     def invalidate(self):
         for v in self.values():
             v.invalidate()
         self.mark_invalid()
 
+    def can_apply_result(self, result: Any):
+        if result is None:
+            return True
+        if not result:
+            return True
+        # Expect a mapping of some sort or an iterable of k, v pairs
+        for k, v_ in result.items() if isinstance(result, (dict, frozendict)) else result:
+            if v_ is None:
+                continue
+            if v_ is REMOVE or v_ is REMOVE_IF_EXISTS:  # Supporting numpy arrays has its costs (==)
+                if v_ is REMOVE_IF_EXISTS and k not in self._ts_values:  # is check should be faster than contains check
+                    continue
+                if self[k].modified:
+                    return False
+            else:
+                if (v := self.get(k)) and not v.can_apply_result(v_):
+                    return False
+        return True
+
     def apply_result(self, result: Any):
         if result is None:
             return
@@ -317,6 +342,9 @@ def on_key_removed(self, key: K):
             if value.active:
                 value.make_passive()
             self._removed_items[key] = value
+            try:
+                self._modified_items.remove((key, value))
+            except: pass
         else:
             self._ts_values[key] = value
             value.un_bind_output()
@@ -334,6 +362,15 @@ def modified(self) -> bool:
         else:
             return self.key_set.modified or any(v.modified for v in self._ts_values.values())
 
+    @property
+    def last_modified_time(self) -> datetime:
+        if self.has_peer:
+            return super().last_modified_time
+        elif self.active:
+            return max(self._last_notified_time, self.key_set.last_modified_time, self._sample_time)
+        else:
+            return max(self.key_set.last_modified_time, max(v.last_modified_time for v in self._ts_values.values()))
+
     @property
     def value(self):
         return frozendict((k, v.value) for k, v in self.items())
@@ -347,6 +384,9 @@ def delta_value(self):
     def __contains__(self, item):
         return item in self._ts_values
 
+    def key_from_value(self, value: V) -> K:
+        return self._ts_values_to_keys[id(value)]
+
     def added_keys(self) -> Iterable[K]:
         return self.key_set.added()
 
diff --git a/src/hgraph/_impl/_types/_tsl.py b/src/hgraph/_impl/_types/_tsl.py
index adb40200..33c8f10b 100644
--- a/src/hgraph/_impl/_types/_tsl.py
+++ b/src/hgraph/_impl/_types/_tsl.py
@@ -45,6 +45,23 @@ def value(self, v: tuple | frozendict | dict | list):
     def delta_value(self) -> Optional[dict[int, Any]]:
         return {i: ts.delta_value for i, ts in enumerate(self._ts_values) if ts.modified}
 
+    def can_apply_result(self, result: Any):
+        if result is None:
+            return True
+        elif isinstance(result, (dict, frozendict)):
+            for k, v_ in result.items():
+                if v_ is not None:
+                    if not self[k].can_apply_result(v_):
+                        return False
+        elif isinstance(result, (tuple, list)):
+            if len(result) != len(self._ts_values):
+                raise ValueError(f"Expected {len(self._ts_values)} elements, got {len(result)}")
+            for ts, v_ in zip(self._ts_values, result):
+                if v_ is not None:
+                    if not ts.can_apply_result(v_):
+                        return False
+        return True
+
     def apply_result(self, result: Any):
         if result is None:
             return
diff --git a/src/hgraph/_impl/_types/_tss.py b/src/hgraph/_impl/_types/_tss.py
index 23844aed..68e98ffb 100644
--- a/src/hgraph/_impl/_types/_tss.py
+++ b/src/hgraph/_impl/_types/_tss.py
@@ -150,6 +150,9 @@ def clear(self):
             self._is_empty_ref_output.value = True
         self.mark_modified()
 
+    def can_apply_result(self, result: Any):
+        return not self.modified
+
     def apply_result(self, result: Any):
         if result is None:
             return
@@ -192,6 +195,9 @@ def copy_from_input(self, input: "TimeSeriesInput"):
     def __contains__(self, item: SCALAR) -> bool:
         return item in self._value
 
+    def __len__(self):
+        return len(self._value)
+
     def values(self) -> Set[SCALAR]:
         return self._value
 
@@ -232,6 +238,9 @@ def _reset_prev(self):
     def __contains__(self, item: SCALAR) -> bool:
         return self.output.__contains__(item)
 
+    def __len__(self):
+        return self.output.__len__()
+
     @property
     def delta_value(self):
         return PythonSetDelta(self.added(), self.removed())
diff --git a/src/hgraph/_types/_time_series_types.py b/src/hgraph/_types/_time_series_types.py
index a5ed6bf7..2dac9815 100644
--- a/src/hgraph/_types/_time_series_types.py
+++ b/src/hgraph/_types/_time_series_types.py
@@ -169,6 +169,12 @@ def delta_value(self):
         The scalar value wrapper of the ticked_value.
         """
 
+    @abstractmethod
+    def can_apply_result(self, result: Any):
+        """
+        Return True if the result can be applies without overwriting any value modified in this engine cycle.
+        """
+
     @abstractmethod
     def apply_result(self, result: Any):
         """
diff --git a/src/hgraph/_types/_tsb_type.py b/src/hgraph/_types/_tsb_type.py
index bc86a3c7..ccfd9eeb 100644
--- a/src/hgraph/_types/_tsb_type.py
+++ b/src/hgraph/_types/_tsb_type.py
@@ -19,7 +19,7 @@
     ValuesView,
     cast,
     ClassVar,
-    _GenericAlias,
+    _GenericAlias, Iterable, Tuple,
 )
 
 from frozendict import frozendict
@@ -304,6 +304,9 @@ def as_schema(self) -> TS_SCHEMA:
         """
         return self
 
+    def __len__(self):
+        return len(self._ts_values)
+
     def __getattr__(self, item) -> TimeSeries:
         """
         The time-series value for the property associated to item in the schema
@@ -330,20 +333,39 @@ def __getitem__(self, item: Union[int, str]) -> "TimeSeries":
         else:
             return self._ts_values[item]
 
+    def key_from_value(self, value: Any) -> str:
+        return next((k for k, v in self._ts_values.items() if v is value), None)
+
     def keys(self) -> KeysView[str]:
         """The keys of the schema defining the bundle"""
         return self._ts_values.keys()
 
-    @abstractmethod
     def items(self) -> ItemsView[str, TimeSeries]:
         """The items of the bundle"""
         return self._ts_values.items()
 
-    @abstractmethod
     def values(self) -> ValuesView[TimeSeries]:
         """The values of the bundle"""
         return self._ts_values.values()
 
+    def modified_keys(self) -> Iterable[str]:
+        return (i for i in self.keys() if self._ts_values[i].modified)
+
+    def modified_values(self) -> Iterable[TimeSeries]:
+        return (v for v in self.values() if v.modified)
+
+    def modified_items(self) -> Iterable[Tuple[str, TimeSeries]]:
+        return ((i, v) for i, v in self.items() if v.modified)
+
+    def valid_keys(self) -> Iterable[str]:
+        return (i for i in self.keys() if self._ts_values[i].valid)
+
+    def valid_values(self) -> Iterable[TimeSeries]:
+        return (v for v in self.values() if v.valid)
+
+    def valid_items(self) -> Iterable[Tuple[str, TimeSeries]]:
+        return ((i, v) for i, v in self.items() if v.valid)
+
 
 class TimeSeriesBundleInput(TimeSeriesInput, TimeSeriesBundle[TS_SCHEMA], Generic[TS_SCHEMA]):
     """
diff --git a/src/hgraph/_types/_tsl_type.py b/src/hgraph/_types/_tsl_type.py
index 6607723b..f4c974dc 100644
--- a/src/hgraph/_types/_tsl_type.py
+++ b/src/hgraph/_types/_tsl_type.py
@@ -70,6 +70,9 @@ def __iter__(self) -> Iterable[TIME_SERIES_TYPE]:
         """
         return iter(self._ts_values)
 
+    def key_from_value(self, value: TIME_SERIES_TYPE) -> int:
+        return self._ts_values.index(value)
+
     def keys(self) -> Iterable[K]:
         return range(len(self._ts_values))
 
diff --git a/src/hgraph/_types/_tss_type.py b/src/hgraph/_types/_tss_type.py
index 902d751d..ab6ed57e 100644
--- a/src/hgraph/_types/_tss_type.py
+++ b/src/hgraph/_types/_tss_type.py
@@ -37,6 +37,12 @@ def __contains__(self, item: KEYABLE_SCALAR) -> bool:
         If this time series set contain the value provided.
         """
 
+    @abstractmethod
+    def __len__(self) -> int:
+        """
+        Number of items in the set
+        """
+
     @abstractmethod
     def values(self) -> Iterable[KEYABLE_SCALAR]:
         """
diff --git a/src/hgraph/_wiring/_graph_builder.py b/src/hgraph/_wiring/_graph_builder.py
index 7633614f..a3410efe 100644
--- a/src/hgraph/_wiring/_graph_builder.py
+++ b/src/hgraph/_wiring/_graph_builder.py
@@ -113,10 +113,27 @@ def toposort(
         if not ts_nodes:
             source_nodes.add(to_node)
 
+    from graphlib import TopologicalSorter, CycleError
+
+    try:
+        list(TopologicalSorter(reverse_mapping).static_order())
+    except CycleError as e:
+        cycle_nodes = e.args[1]
+
+        cycle_node_strings = []
+        for n in cycle_nodes:
+            scalars = ", ".join(
+                f"{k}={str(v)[:100]}" for k, v in n.inputs.items() if k in n.resolved_signature.scalar_inputs
+            )
+            cycle_node_strings.append(f"{n.wiring_path_name}.{n.label or n.node_signature.name}({scalars})")
+
+        cycle_print = "\n -> ".join(cycle_node_strings)
+
+        raise RuntimeError(f"Cyclic sub graph detected that involves nodes: \n{cycle_print}")
+
     # Rank nodes
     nodes_to_process.extend(source_nodes)
     max_rank = 0
-    potential_cycle_back_preassure = 1000
     while len(nodes_to_process) > 0:
         from_node = nodes_to_process.popleft()
         if not supports_push_nodes and from_node.resolved_signature.node_type is NodeTypeEnum.PUSH_SOURCE_NODE:
@@ -132,29 +149,8 @@ def toposort(
             nodes_to_process.append(to_node)
             max_rank = max(max_rank, processed_nodes[to_node])
 
-        if max_rank > len(processed_nodes) + 1000 or len(nodes_to_process) > len(processed_nodes) + potential_cycle_back_preassure:
-            from graphlib import TopologicalSorter, CycleError
-
-            try:
-                list(TopologicalSorter(reverse_mapping).static_order())
-            except CycleError as e:
-                cycle_nodes = e.args[1]
-
-                cycle_node_strings = []
-                for n in cycle_nodes:
-                    scalars = ", ".join(
-                        f"{k}={str(v)[:100]}" for k, v in n.inputs.items() if k in n.resolved_signature.scalar_inputs
-                    )
-                    cycle_node_strings.append(f"{n.wiring_path_name}.{n.label or n.node_signature.name}({scalars})")
-
-                cycle_print = "\n -> ".join(cycle_node_strings)
-
-                raise RuntimeError(f"Cyclic sub graph detected that involves nodes: \n{cycle_print}")
-            else:
-                potential_cycle_back_preassure += 1000
-                if potential_cycle_back_preassure > 2**31 - 1:
-                    # We are not seeing a problem, but this is getting very large and so we dump out here
-                	raise RuntimeError(f"Cyclic sub graph exists but could not be detected")
+        if max_rank > len(processed_nodes) + 1000:
+            raise RuntimeError(f"Cyclic sub graph exists but could not be detected")
 
     # Sort nodes by rank
     result = [node for _, node in sorted((rank, node) for node, rank in processed_nodes.items()) if not node.is_stub]
diff --git a/src/hgraph/adaptors/perspective/_perspective.py b/src/hgraph/adaptors/perspective/_perspective.py
index 085a936d..170d6c93 100644
--- a/src/hgraph/adaptors/perspective/_perspective.py
+++ b/src/hgraph/adaptors/perspective/_perspective.py
@@ -3,12 +3,14 @@
 import os
 import tempfile
 import time
+from collections import defaultdict
 from datetime import datetime
 from glob import glob
 from pathlib import Path
 from threading import Thread
 from typing import Dict, Optional, Callable, List
 
+import pyarrow
 import tornado
 from perspective import PerspectiveManager, PerspectiveTornadoHandler, Table, View
 
@@ -114,9 +116,31 @@ def _start_table(self, name):
 
     def update_table(self, name, data, removals=None):
         table = self._tables[name][0]
-        self._manager_for_table(name).call_loop(lambda: table.update(data))
+
+        if data:
+            if isinstance(data, list):
+                d0 = {}
+                d1 = defaultdict(lambda: [None]*len(data))
+                for i, r in enumerate(data):
+                    for k, v in r.items():
+                        if v is not None:
+                            d1[k][i] = v
+                            d0[k] = True
+                data = {k: v for k, v in d1.items() if k in d0}
+
+            data = pyarrow.record_batch(data)
+            sink = pyarrow.BufferOutputStream()
+
+            with pyarrow.ipc.new_stream(sink, data.schema) as writer:
+                writer.write_batch(data)
+
+            arrow = sink.getvalue().to_pybytes()
+
+            self._manager_for_table(name).call_loop(lambda: table.update(arrow))
+
         if removals:
             self._manager_for_table(name).call_loop(lambda: table.remove(removals))
+
         # table.update(data)
 
     def get_table_names(self):
@@ -247,7 +271,7 @@ def perspective_web_start(
         + ([(k, tornado.web.StaticFileHandler, v) for k, v in static.items()] if static else [])
         + (
             [(
-                r"/(.*)",
+                r"/",
                 IndexPageHandler,
                 {
                     "mgr": perspective_manager,
@@ -304,7 +328,7 @@ def initialize(self, mgr: PerspectiveManager, layouts_path: str, index_template:
         self.host = host
         self.port = port
 
-    def get(self, url):
+    def get(self, url = ""):
         tornado.log.app_log.info(f"requesting url {url} for template {self.index_template}")
         if url == "" or True:
             layouts = glob(os.path.join(self.layouts_path, f"{url or '*'}.json"))
diff --git a/src/hgraph/adaptors/perspective/_perspetive_publish.py b/src/hgraph/adaptors/perspective/_perspetive_publish.py
index 682b9c9f..7e3649f6 100644
--- a/src/hgraph/adaptors/perspective/_perspetive_publish.py
+++ b/src/hgraph/adaptors/perspective/_perspetive_publish.py
@@ -23,12 +23,12 @@
     push_queue,
     OUT,
     nothing,
-    SCHEDULER,
+    SCHEDULER, operator,
 )
 from ._perspective import PerspectiveTablesManager
 
 
-@sink_node
+@operator
 def _publish_table(
     name: str, ts: TIME_SERIES_TYPE, editable: bool = False, empty_row: bool = False, history: int = None
 ): ...
diff --git a/src/hgraph/adaptors/perspective/workspace_template.html b/src/hgraph/adaptors/perspective/workspace_template.html
index 1023f2d0..984d9583 100644
--- a/src/hgraph/adaptors/perspective/workspace_template.html
+++ b/src/hgraph/adaptors/perspective/workspace_template.html
@@ -21,6 +21,17 @@
                 right: 0;
                 bottom: 0;
             }
+
+            .hidden-column-width {
+                min-width: 1px !important;
+                max-width: 1px !important;
+
+                user-select: none;
+                overflow: hidden;
+                white-space: nowrap;
+                text-overflow: clip;
+            }
+
         </style>
     </head>
 
diff --git a/src/hgraph/adaptors/tornado/http_server_adaptor.py b/src/hgraph/adaptors/tornado/http_server_adaptor.py
index 02340882..3c0340c4 100644
--- a/src/hgraph/adaptors/tornado/http_server_adaptor.py
+++ b/src/hgraph/adaptors/tornado/http_server_adaptor.py
@@ -100,14 +100,16 @@ def add_request(self, request_id, request):
 
     def complete_request(self, request_id, response):
         self.requests[request_id].set_result(response)
-        self.queue({request_id: REMOVE_IF_EXISTS})
         print(f"Completed request {request_id} with response {response}")
 
+    def remove_request(self, request_id):
+        self.queue({request_id: REMOVE_IF_EXISTS})
+        del self.requests[request_id]
 
 class HttpHandler(tornado.web.RequestHandler):
     def initialize(self, path, mgr):
         self.path = path
-        self.mgr = mgr
+        self.mgr: HttpAdaptorManager = mgr
 
     async def get(self, *args):
         request_obj = HttpGetRequest(
@@ -130,6 +132,7 @@ async def get(self, *args):
                 self.set_header(k, v)
 
         await self.finish(response.body)
+        self.mgr.remove_request(id(request_obj))
 
     async def post(self, *args):
         request_obj = HttpPostRequest(
@@ -149,6 +152,7 @@ async def post(self, *args):
         for k, v in response.headers.items():
             self.set_header(k, v)
         await self.finish(response.body)
+        self.mgr.remove_request(id(request_obj))
 
 
 def http_server_handler(fn: Callable = None, *, url: str):
diff --git a/src/hgraph/debug/_inspector_controller.py b/src/hgraph/debug/_inspector_controller.py
new file mode 100644
index 00000000..26a036aa
--- /dev/null
+++ b/src/hgraph/debug/_inspector_controller.py
@@ -0,0 +1,614 @@
+import asyncio
+import os
+import tempfile
+import threading
+import time
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+
+from _socket import gethostname
+from frozendict import frozendict
+from perspective import Table
+import tornado.web
+import pyarrow as pa
+
+from hgraph import graph, TS, STATE, TSD, CompoundScalar, \
+    TimeSeries, PythonNestedNodeImpl, Node, Graph, PythonTimeSeriesReference, sink_node, \
+    PythonTimeSeriesReferenceOutput, TimeSeriesOutput, TimeSeriesInput
+from hgraph._impl._runtime._node import _SenderReceiverState
+from hgraph.adaptors.perspective import PerspectiveTablesManager
+from hgraph.adaptors.tornado.http_server_adaptor import HttpRequest, HttpResponse, \
+    HttpGetRequest
+from hgraph.debug._inspector_observer import InspectionObserver
+from hgraph.debug._inspector_util import node_id_from_str, str_node_id, format_value, format_timestamp, enum_items, \
+    format_type, name, inspect_item
+
+
+@graph
+def inspector_controller(port: int = 8080):
+    @dataclass
+    class GraphInspectorData(CompoundScalar):
+        id: str
+        X: str = "+"
+        name: str = None
+        type: str = None
+        value: str = None
+        timestamp: datetime = None
+        evals: int = None
+        time: float = None
+        of_graph: float = None
+        of_total: float = None
+
+
+    @dataclass
+    class InspectorState(CompoundScalar):
+        observer: InspectionObserver = None
+        manager: PerspectiveTablesManager = None
+        table: Table = None
+        total_cycle_table: Table = None
+
+        requests: _SenderReceiverState = field(default_factory=_SenderReceiverState)
+
+        row_ids: dict = field(default_factory=dict)
+
+        key_ids: dict = field(default_factory=dict)
+        id_keys: dict = field(default_factory=dict)
+
+        node_subscriptions: dict = field(default_factory=lambda: defaultdict(dict))  # node_id -> [(item path, row_id)]
+
+        value_data: list = field(default_factory=list)
+        perf_data: list = field(default_factory=list)
+        total_data_prev: list = field(default_factory=dict)
+        total_data: list = field(default_factory=lambda: defaultdict(list))
+        last_publish_time: datetime = None
+        inspector_time: float = 0.
+
+        def id_for_key(self, key):
+            i = self.key_ids.setdefault(key, len(self.key_ids))
+            self.id_keys[i] = key
+            return i
+
+        def key_for_id(self, i):
+            return self.id_keys[i]
+
+    class NODE_ITEMS(Enum):
+        INPUTS = "zzz001"
+        OUTPUT = "zzz002"
+        SUBGRAPHS = "zzz003"
+        SCALARS = "zzz004"
+
+    NODE_ITEMS_BY_VALUE = {v.value: k for k, v in NODE_ITEMS.__members__.items()}
+
+    MORE = "zzy"
+
+    def process_tick(state: STATE, node: Node):
+        start = time.perf_counter_ns()
+
+        for row_id, path in state.node_subscriptions.get(node.node_id, {}).items():
+            if row_id is not None:
+                if path:
+                    match path[0]:
+                        case NODE_ITEMS.INPUTS:
+                            v = node.input
+                        case NODE_ITEMS.OUTPUT:
+                            v = node.output
+                        case NODE_ITEMS.SUBGRAPHS:
+                            v = node
+                        case _:
+                            continue
+
+                    for item in path[1:]:
+                        try:
+                            v = inspect_item(v, item)
+                        except:
+                            v = "This value could not be retrieved"
+                else:
+                    v = node
+
+                if isinstance(v, (TimeSeries, Node)):
+                    state.value_data.append(dict(
+                        id=row_id,
+                        value=format_value(v),
+                        timestamp=format_timestamp(v),
+                    ))
+                else:
+                    state.value_data.append(dict(
+                        id=row_id,
+                        value=format_value(v),
+                    ))
+
+        state.inspector_time += (time.perf_counter_ns() - start) / 1_000_000_000
+
+    def process_graph(state: STATE, graph: Graph):
+        start = time.perf_counter_ns()
+
+        root_graph = state.observer.get_graph_info(())
+        for node_id, items in state.node_subscriptions.items():
+            if node_row := items.get(None, None):
+                gi = state.observer.get_graph_info(node_id[:-1])
+                if gi is not None:
+                    node_ndx = node_id[-1]
+                    state.perf_data.append(dict(
+                        id=node_row,
+                        evals=gi.node_eval_counts[node_ndx],
+                        time=gi.node_eval_times[node_ndx] / 1_000_000_000,
+                        of_graph=gi.node_eval_times[node_ndx] / gi.eval_time if gi.eval_time else None,
+                        of_total=gi.node_eval_times[node_ndx] / root_graph.eval_time if root_graph.eval_time else None
+                    ))
+
+        if graph.graph_id == ():
+            gi = state.observer.get_graph_info(())
+            state.total_data['time'].append(datetime.utcnow())
+            state.total_data['evaluation_time'].append(graph.evaluation_clock.evaluation_time)
+            state.total_data['cycles'].append(gi.eval_count)
+            state.total_data['graph_time'].append(gi.eval_time)
+
+            if state.last_publish_time is None or (datetime.utcnow() - state.last_publish_time).total_seconds() > 2.5:
+                state.inspector_time += (time.perf_counter_ns() - start) / 1_000_000_000
+                start = time.perf_counter_ns()
+                publish_stats(state)
+                state.inspector_time = 0.
+
+            handle_request(state)
+
+        else:
+            gi = state.observer.get_graph_info(graph.graph_id)
+            for k, row_id in state.node_subscriptions.get(graph.graph_id, {}).items():
+                if k is None:
+                    parent_time = state.observer.get_graph_info(graph.parent_node.graph.graph_id).eval_time
+                    state.perf_data.append(dict(
+                        id=row_id,
+                        timestamp=graph.parent_node.last_evaluation_time if graph.parent_node else None,
+                        evals=gi.eval_count,
+                        time=gi.eval_time / 1_000_000_000,
+                        of_graph=gi.eval_time / parent_time if parent_time else None,
+                        of_total=gi.eval_time / root_graph.eval_time if root_graph.eval_time else None
+                    ))
+
+        state.inspector_time += (time.perf_counter_ns() - start) / 1_000_000_000
+
+
+    def publish_stats(state: STATE):
+        state.manager.update_table("inspector", [i for i in state.value_data if i["id"] in state.row_ids])
+        state.value_data = []
+
+        state.manager.update_table("inspector", [i for i in state.perf_data if i["id"] in state.row_ids])
+        state.perf_data = []
+
+        data = state.total_data
+        if data["time"]:
+            total_time = (state.total_data["time"][-1] - state.total_data_prev.get("time", datetime.min)).total_seconds()
+            total_graph_time = (data["graph_time"][-1] - state.total_data_prev.get("graph_time", 0)) / 1_000_000_000
+            lags = [(data["time"][i] - data["evaluation_time"][i]).total_seconds() for i in range(len(data["time"]))]
+
+            state.total_cycle_table.update([dict(
+                time=data["time"][-1],
+                evaluation_time=data["evaluation_time"][-1],
+                cycles=(data["cycles"][-1] - state.total_data_prev.get("cycles", 0)) / total_time,
+                graph_time=total_graph_time,
+                graph_load=total_graph_time / total_time,
+                avg_lag=sum(lags) / len(data["time"]),
+                max_lag=max(lags),
+                inspection_time=state.inspector_time / total_graph_time
+            )])
+
+            state.total_data_prev = {k: v[-1] for k, v in data.items()}
+            state.total_data = defaultdict(list)
+
+        state.last_publish_time = datetime.utcnow()
+
+    @sink_node
+    def inspector(start: TS[bool] = True, port: int = 8080, _state: STATE[InspectorState] = None):
+        ...
+
+    def set_result(f, r):
+        def apply_result(fut, res):
+            try:
+                fut.set_result(res)
+            except:
+                pass
+
+        from hgraph.adaptors.tornado._tornado_web import TornadoWeb
+        TornadoWeb.get_loop().add_callback(lambda f, r: apply_result(f, r), f, r)
+
+    def handle_request(_state):
+        data = []  # to be published to the table
+        remove = set()  # to be removed from the table
+
+        observer = _state.observer
+
+        while f_r := _state.requests.dequeue():
+            f: asyncio.Future
+            r: HttpRequest
+            f, r = f_r
+
+            command = r.url_parsed_args[0]
+            id_str = r.url_parsed_args[1]
+
+            commands = deque()
+            commands.append((command, id_str))
+
+            while commands:
+                command, id_str = commands.popleft()
+
+                if id_str[-6:-3] == MORE:
+                    more = node_id_from_str(id_str[-3:])[0]
+                    id_str = id_str[:-6]
+                else:
+                    more = None
+
+                if level := NODE_ITEMS_BY_VALUE.get(id_str[-6:], None):
+                    id_str_node = id_str[:-6]
+                else:
+                    level = "graph"
+                    id_str_node = id_str
+
+                graph_id = ()  # graph that we are inspecting or the graph the inspected node belongs to, root by default
+                gi = None  # graph info of the graph we are inspecting
+                node = None  # node we are inspecting or None
+                path = None  # path to the value we are inspecting inside the node
+
+                if id_str_node == '':
+                    # root graph
+                    pass
+                elif graph_id := _state.row_ids.get(id_str_node, None):
+                    # we have expanded this graph before
+                    path = _state.node_subscriptions[graph_id][id_str]
+                else:
+                    graph_id = ()
+                    graph = observer.get_graph_info(graph_id)
+                    node_id = node_id_from_str(id_str_node)
+                    i = 0
+                    while i < len(node_id):
+                        node = graph.graph.nodes[node_id[i]]
+                        item_str = id_str_node[(i + 1) * 3:(i + 3 * 3)]
+                        if item_str == NODE_ITEMS.SUBGRAPHS.value:
+                            key = _state._value.key_for_id(node_id[i+3]),
+                            graph = inspect_item(node, key)
+                            graph_id = graph.graph_id
+                            i += 4
+                        elif item_str in NODE_ITEMS_BY_VALUE:
+                            path = [NODE_ITEMS_BY_VALUE[item_str]] + [_state._value.key_for_id(node_id[j]) for j in range(i+2, len(node_id))]
+                            break
+                        else:
+                            # this is irregular
+                            raise ValueError(f"Invalid item {item_str} in node {node.node_id}")
+
+                if (gi := observer.get_graph_info(graph_id)) is None:
+                    level = "node" if level == "graph" else level
+                    node_ndx = graph_id[-1]
+                    graph_id = graph_id[:-1]
+                    if (gi := observer.get_graph_info(graph_id)) is None:
+                        set_result(f, HttpResponse(status_code=500, body=f"Graph {graph_id} was not found"))
+                        continue
+                    node = gi.graph.nodes[node_ndx]
+
+                if path is not None:
+                    level = path[0]
+                    if len(path) == 2 and level == NODE_ITEMS.SUBGRAPHS:
+                        try:
+                            graph = inspect_item(node, path[1])
+                            graph_id = graph.graph_id
+                            gi = observer.get_graph_info(graph_id)
+                            level = "graph"
+                        except:
+                            set_result(f, HttpResponse(status_code=500, body=f"Graph {graph_id}, {path[1]} was not found"))
+                            continue
+
+                if level == NODE_ITEMS.OUTPUT:
+                    root_item = node.output
+                    level = "value"
+                elif level == NODE_ITEMS.INPUTS:
+                    root_item = node.input
+                    level = "value"
+                elif level == NODE_ITEMS.SUBGRAPHS:
+                    root_item = node
+                    level = "value"
+                elif level == NODE_ITEMS.SCALARS:
+                    root_item = node.scalars
+                    level = "value"
+                else:
+                    root_item = node
+
+                match command:
+                    case "expand":
+                        tab = "\u00A0\u00A0" * (len(id_str) // 3)
+
+                        if id_str:
+                            data.append(dict(id=id_str, X="-"))
+
+                        match level:
+                            case "graph":
+                                for i, n in enumerate(gi.graph.nodes):
+                                    row_id = id_str + str_node_id((i,))
+                                    data.append(dict(
+                                        id=row_id,
+                                        X="+",
+                                        name=tab + name(gi.graph.nodes[i]),
+                                        type=format_type(n),
+                                        value=format_value(n),
+                                        timestamp=format_timestamp(n)))
+
+                                    _state.node_subscriptions[n.node_id][row_id] = None
+                                    _state.node_subscriptions[n.node_id][None] = row_id
+                                    _state.observer.subscribe(n.node_id)
+                                    _state.row_ids[row_id] = n.node_id
+
+                            case "node":
+                                if node.input is not None:
+                                    row_id = id_str + NODE_ITEMS.INPUTS.value
+                                    data.append(dict(
+                                        id=row_id,
+                                        X="+",
+                                        name=tab + "INPUTS",
+                                        type=format_type(node.input),
+                                        value=format_value(node.input),
+                                        timestamp=format_timestamp(node.input)))
+
+                                    _state.node_subscriptions[node.node_id][row_id] = [NODE_ITEMS.INPUTS]
+                                    _state.row_ids[row_id] = node.node_id
+
+                                if node.output is not None:
+                                    row_id = id_str + NODE_ITEMS.OUTPUT.value
+                                    data.append(dict(
+                                        id=row_id,
+                                        X="+",
+                                        name=tab + "OUTPUT",
+                                        type=format_type(node.output),
+                                        value=format_value(node.output),
+                                        timestamp=format_timestamp(node.output)))
+
+                                    _state.node_subscriptions[node.node_id][row_id] = [NODE_ITEMS.OUTPUT]
+                                    _state.row_ids[row_id] = node.node_id
+
+                                if isinstance(node, PythonNestedNodeImpl):
+                                    row_id = id_str + NODE_ITEMS.SUBGRAPHS.value
+                                    data.append(dict(
+                                        id=row_id,
+                                        X="+",
+                                        name=tab + "GRAPHS",
+                                        type="",
+                                        value=format_value(node.input),
+                                        timestamp=format_timestamp(node.input)))
+
+                                    _state.row_ids[row_id] = None
+                                    _state.node_subscriptions[node.node_id][row_id] = [NODE_ITEMS.SUBGRAPHS]
+
+                                if node.scalars:
+                                    row_id = id_str + NODE_ITEMS.SCALARS.value
+                                    data.append(dict(
+                                        id=row_id,
+                                        X="+",
+                                        name=tab + "SCALARS",
+                                        type="",
+                                        value=format_value(node.scalars),
+                                        timestamp=format_timestamp(node.scalars)))
+
+                                    _state.row_ids[row_id] = None
+                                    _state.node_subscriptions[node.node_id][row_id] = [NODE_ITEMS.SCALARS]
+
+                            case "value":
+                                start = 0
+                                path = [] if path is None else path
+                                for key in path[1:]:
+                                    try:
+                                        root_item = inspect_item(root_item, key)
+                                    except:
+                                        root_item = None
+                                        set_result(f, HttpResponse(status_code=500, body=f"Item cannot be inspected"))
+                                        break
+
+                                if more is not None:
+                                    start = more
+                                    remove.add(id_str + MORE + str_node_id((start,)))
+
+                                if root_item:
+                                    for i, (k, v) in enumerate(enum_items(root_item)):
+                                        if i < start:
+                                            continue
+                                        if i >= start + 10 and not 'all' in r.query:
+                                            row_id = id_str + MORE + str_node_id((i,))
+                                            data.append(dict(
+                                                id=row_id,
+                                                X="+",
+                                                name=tab + "...",
+                                                type="",
+                                            ))
+                                            _state.row_ids[row_id] = None
+                                            break
+
+                                        row_id = id_str + str_node_id((_state._value.id_for_key(k),))
+                                        data.append(dict(
+                                            id=row_id,
+                                            X="+",
+                                            name=tab + str(k),
+                                            type=format_type(v),
+                                            value=format_value(v),
+                                            timestamp=format_timestamp(v)
+                                        ))
+
+                                        if isinstance(v, Graph):
+                                            _state.row_ids[row_id] = v.graph_id
+                                            _state.node_subscriptions[v.graph_id][None] = row_id
+                                            _state.node_subscriptions[v.graph_id][row_id] = None
+                                            _state.observer.subscribe(v.graph_id)
+                                        else:
+                                            _state.node_subscriptions[node.node_id][row_id] = path + [k]
+                                            _state.row_ids[row_id] = node.node_id
+
+                                    remove.add(id_str + MORE + str_node_id((start,)))
+
+                    case "collapse":
+                        remove |= {i for i in _state.row_ids if i.startswith(id_str)} - {id_str}
+                        for i in remove:
+                            if node_id := _state.row_ids.get(i):
+                                (subs := _state.node_subscriptions[node_id]).pop(i, None)
+                                if len(subs) == 0:
+                                    _state.observer.unsubscribe(node_id)
+
+                        data.append(dict(id=id_str, X="+"))
+
+                    case "ref":
+                        path = [] if path is None else path
+                        for key in path[1:]:
+                            try:
+                                root_item = inspect_item(root_item, key)
+                                path.append(key)
+                            except:
+                                root_item = None
+                                set_result(f, HttpResponse(status_code=500, body=f"Item cannot be inspected"))
+                                break
+
+                        if node is None:
+                            root_item = gi.graph.parent_node
+
+                        if isinstance(root_item, Node):
+                            root_item = root_item.output
+
+                        if isinstance(root_item, PythonTimeSeriesReferenceOutput):
+                            root_item = root_item.value
+
+                        if isinstance(root_item, PythonTimeSeriesReference):
+                            if root_item.output is not None:
+                                root_item = root_item.output
+                            else:
+                                set_result(f, HttpResponse(status_code=500, body="Reference has no output"))
+                                continue
+
+                        if isinstance(root_item, TimeSeriesInput):
+                            root_item = root_item.output
+
+                        if isinstance(root_item, TimeSeriesOutput):
+                            ref_node = root_item.owning_node
+                            path = []
+                            while root_item.parent_output:
+                                path.append(root_item.parent_output.key_from_value(root_item))
+                                root_item = root_item.parent_output
+
+                            ref_id = ''
+                            i = 1
+                            while i < len(ref_node.node_id):
+                                graph_id = ref_node.node_id[:i]
+                                ref_id += str_node_id((graph_id[-1],))
+                                commands.append(("expand", ref_id))
+                                if observer.get_graph_info(graph_id) is None:
+                                    ref_id += NODE_ITEMS.SUBGRAPHS.value
+                                    commands.append(("expand", ref_id))
+                                    gi = observer.get_graph_info(ref_node.node_id[:i + 1])
+                                    ref_id += str_node_id((_state._value.id_for_key(gi.graph.label),))
+                                    commands.append(("expand", ref_id))
+                                    i += 2
+                                else:
+                                    i += 1
+
+                            ref_id += str_node_id((ref_node.node_id[-1],))
+                            commands.append(("expand", ref_id))
+                            ref_id += NODE_ITEMS.OUTPUT.value
+                            commands.append(("expand", ref_id))
+
+                            for i in path:
+                                ref_id += str_node_id((_state._value.id_for_key(i),))
+                                commands.append(("expand", ref_id))
+
+                            set_result(f, HttpResponse(status_code=200, body=ref_id))
+
+                        else:
+                            set_result(f, HttpResponse(status_code=500, body="Not a reference"))
+
+            if data or remove:
+                _state.manager.update_table("inspector", data, remove)
+                [_state.row_ids.pop(i, None) for i in remove]
+
+            set_result(f, HttpResponse(status_code=200, body=id_str))
+
+    @inspector.start
+    def start_inspector(start: TS[bool], port: int, _state: STATE[InspectorState]):
+        from hgraph.adaptors.perspective import PerspectiveTablesManager
+        from hgraph.adaptors.tornado._tornado_web import TornadoWeb
+        from hgraph.adaptors.perspective._perspective import IndexPageHandler
+        from perspective import Table
+
+        _state.requests.evaluation_clock = start.owning_graph.evaluation_clock
+
+        _state.observer = InspectionObserver(
+            start.owning_graph,
+            callback_node=lambda n: process_tick(_state, n),
+            callback_graph=lambda n: process_graph(_state, n)
+        )
+        _state.observer.on_before_node_evaluation(start.owning_node)
+        start.owning_graph.evaluation_engine.add_life_cycle_observer(_state.observer)
+        _state.observer.subscribe(())
+
+        _state.manager = PerspectiveTablesManager.current()
+        _state.table = Table({"id": str, **{k: v.py_type for k, v in GraphInspectorData.__meta_data_schema__.items()}}, index="id")
+        _state.manager.add_table("inspector", _state.table)
+
+        _state.total_cycle_table = Table({
+            "time": datetime, "evaluation_time": datetime, "cycles": float, "graph_time": float, "graph_load": float, "avg_lag": float, "max_lag": float, "inspection_time": float
+        }, limit=24*3600)
+        _state.manager.add_table("graph_performance", _state.total_cycle_table)
+
+        _state.total_data_prev = dict(
+            time=datetime.utcnow(),
+            evaluation_time=start.owning_graph.evaluation_clock.evaluation_time,
+            cycles=0,
+            graph_time=0.
+        )
+
+        tempfile.gettempdir()
+        layouts_dir = os.path.join(tempfile.tempdir, "inspector_layouts")
+
+        app = TornadoWeb.instance(port)
+        app.add_handlers(
+            [
+                (
+                    r"/inspector/(.*)",
+                    IndexPageHandler,
+                    {
+                        "mgr": _state.manager,
+                        "layouts_path": layouts_dir,
+                        "index_template": os.path.join(os.path.dirname(__file__), "inspector_template.html"),
+                        "host": gethostname(),
+                        "port": port,
+                    },
+                ),
+                (
+                    r"/inspect(?:/([^/]*))?(?:/([^/]*))?(/([^/]*))?",
+                    InspectorHttpHandler,
+                    {
+                        "queue": _state.requests,
+                    }
+                )
+            ]
+        )
+
+        app.start()
+
+    inspector()
+
+class InspectorHttpHandler(tornado.web.RequestHandler):
+    def initialize(self, queue):
+        self.queue = queue
+
+    async def get(self, *args):
+        request = HttpGetRequest(
+                url=self.request.uri,
+                url_parsed_args=args,
+                headers=self.request.headers,
+                query=frozendict({k: ''.join(i.decode() for i in v) for k, v in self.request.query_arguments.items()}),
+                cookies=frozendict(self.request.cookies))
+
+        future = asyncio.Future()
+        self.queue((future, request))
+
+        response = await future
+
+        self.set_status(response.status_code)
+
+        if response.headers:
+            for k, v in response.headers.items():
+                self.set_header(k, v)
+
+        await self.finish(response.body)
diff --git a/src/hgraph/debug/_inspector_observer.py b/src/hgraph/debug/_inspector_observer.py
new file mode 100644
index 00000000..8ddcb751
--- /dev/null
+++ b/src/hgraph/debug/_inspector_observer.py
@@ -0,0 +1,127 @@
+import time
+from dataclasses import dataclass
+from typing import Union, Callable
+
+from frozendict import frozendict
+from multimethod import multimethod
+
+from hgraph.debug._inspector_util import str_node_id
+
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+from hgraph import Graph, PythonTimeSeriesValueOutput, \
+    PythonTimeSeriesListOutput, PythonTimeSeriesSetOutput, PythonTimeSeriesBundleOutput, PythonTimeSeriesDictOutput, \
+    PythonTimeSeriesReferenceOutput, Node, TimeSeriesOutput, PythonTimeSeriesReferenceInput, TimeSeriesList, \
+    TimeSeriesSet, TimeSeriesBundle, TimeSeriesDict, PythonTimeSeriesReference, PythonNestedNodeImpl
+
+__all__ = ("InspectionObserver",)
+
+from hgraph import EvaluationLifeCycleObserver
+
+
+@dataclass(kw_only=True)
+class GraphInfo:
+    graph: "Graph"
+    id: tuple
+    label: str
+    parent_graph: int
+    stopped: bool = False
+    eval_count: int
+    eval_begin_time: float
+    eval_time: float
+    node_eval_counts: []
+    node_eval_begin_times: []
+    node_eval_times: []
+
+
+class InspectionObserver(EvaluationLifeCycleObserver):
+    def __init__(self, graph: "Graph" = None, callback_node: Callable = None, callback_graph: Callable = None):
+        self.graphs = {}
+        self.graphs_by_id = {}
+        self.current_graph: GraphInfo = None
+
+        self.callback_node = callback_node
+        self.callback_graph = callback_graph
+        self.subscriptions = set()
+
+        if graph:
+            self.walk(graph)
+            self.on_before_graph_evaluation(graph)
+
+    def get_graph_info(self, graph_id: tuple) -> GraphInfo | None:
+        return self.graphs_by_id.get(graph_id)
+
+    def walk(self, graph: "Graph"):
+        from hgraph import PythonNestedNodeImpl
+
+        self.on_before_start_graph(graph)
+        for n in graph.nodes:
+            if isinstance(n, PythonNestedNodeImpl):
+                for k, v in n.enum_nested_graphs():
+                    self.walk(v)
+
+        self.on_after_start_graph(graph)
+
+    def subscribe(self, node_id: tuple[int, ...]):
+        if node_id not in self.subscriptions:
+            self.subscriptions.add(node_id)
+
+    def unsubscribe(self, node_id: tuple[int, ...]):
+        self.subscriptions.discard(node_id)
+
+    def on_before_start_graph(self, graph: "Graph"):
+        gi = GraphInfo(
+            graph=graph,
+            id=graph.graph_id,
+            label=graph.label,
+            parent_graph=id(graph.parent_node.graph) if graph.parent_node else 0,
+            eval_count=0,
+            eval_begin_time=time.perf_counter_ns(),
+            eval_time=0.,
+            node_eval_counts=[0] * len(graph.nodes),
+            node_eval_begin_times=[0.] * len(graph.nodes),
+            node_eval_times=[0.] * len(graph.nodes),
+        )
+
+        if self.current_graph:
+            assert self.graphs[gi.parent_graph] is self.current_graph
+
+        self.graphs[id(graph)] = gi
+        self.graphs_by_id[gi.id] = gi
+        self.current_graph = gi
+
+    def on_after_start_graph(self, graph: "Graph"):
+        self.current_graph = self.graphs.get(self.current_graph.parent_graph, None)
+
+    def on_before_graph_evaluation(self, graph: "Graph"):
+        self.current_graph = self.graphs[id(graph)]
+        self.current_graph.eval_begin_time = time.perf_counter_ns()
+        if len(graph.nodes) != len(self.current_graph.node_eval_counts):
+            self.current_graph.node_eval_counts = [0] * len(graph.nodes)
+            self.current_graph.node_eval_begin_times = [0.] * len(graph.nodes)
+            self.current_graph.node_eval_times = [0.] * len(graph.nodes)
+
+    def on_before_node_evaluation(self, node: "Node"):
+        self.current_graph.node_eval_begin_times[node.node_ndx] = time.perf_counter_ns()
+
+    def on_after_node_evaluation(self, node: "Node"):
+        self.current_graph.node_eval_counts[node.node_ndx] += 1
+        self.current_graph.node_eval_times[node.node_ndx] += time.perf_counter_ns() - self.current_graph.node_eval_begin_times[node.node_ndx]
+
+        if self.callback_node and node.node_id in self.subscriptions:
+            self.callback_node(node)
+
+    def on_after_graph_evaluation(self, graph: "Graph"):
+        self.current_graph.eval_count += 1
+        self.current_graph.eval_time += time.perf_counter_ns() - self.current_graph.eval_begin_time
+        self.current_graph = self.graphs.get(self.current_graph.parent_graph, None)
+
+        if self.callback_graph and graph.graph_id in self.subscriptions:
+            self.callback_graph(graph)
+
+    def on_after_stop_graph(self, graph: "Graph"):
+        if gi := self.graphs.get(id(graph)):
+            gi.stopped = True
diff --git a/src/hgraph/debug/_inspector_util.py b/src/hgraph/debug/_inspector_util.py
new file mode 100644
index 00000000..1c7ac019
--- /dev/null
+++ b/src/hgraph/debug/_inspector_util.py
@@ -0,0 +1,306 @@
+import re
+from typing import Union
+
+from frozendict import frozendict
+from multimethod import multimethod
+
+from hgraph import Node, Graph, PythonTimeSeriesValueInput, PythonTimeSeriesValueOutput, \
+    PythonTimeSeriesReferenceOutput, PythonTimeSeriesReferenceInput, PythonTimeSeriesReference, \
+    TimeSeriesList, TimeSeriesDict, TimeSeriesBundle, TimeSeriesSet, TimeSeriesInput, TimeSeriesOutput, \
+    PythonNestedNodeImpl, PythonTsdMapNodeImpl, PythonServiceNodeImpl, PythonReduceNodeImpl, PythonSwitchNodeImpl, \
+    PythonTryExceptNodeImpl, HgTSBTypeMetaData, PythonPushQueueNodeImpl
+from hgraph._impl._runtime._component_node import PythonComponentNodeImpl
+from hgraph._impl._runtime._mesh_node import PythonMeshNodeImpl
+
+STR_ID_SYMBOLS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+STR_ID_NUMBERS = [STR_ID_SYMBOLS.find(chr(i)) for i in range(128)]
+
+
+def str_node_id(id: tuple[int, ...]) -> str:
+    return ''.join(f"{STR_ID_SYMBOLS[i // 3844]}{STR_ID_SYMBOLS[i // 62 % 62]}{STR_ID_SYMBOLS[i % 62]}" for i in id)
+
+
+def node_id_from_str(s: str) -> tuple:
+    return tuple(
+        STR_ID_NUMBERS[ord(s[i])] * 3844 + STR_ID_NUMBERS[ord(s[i + 1])] * 62 + STR_ID_NUMBERS[ord(s[i + 2])] for i in
+        range(0, len(s), 3))
+
+
+# These are visitors that are used to extract information from the graph objects to display in the inspector
+
+@multimethod
+def name(value):
+    return "?"
+
+
+@name.register
+def _(graph: "Graph") -> str:
+    graph_str = []
+    while graph:
+        if graph.parent_node:
+            graph_str.append(
+                f"{(graph.parent_node.signature.label + ':') if graph.parent_node.signature.label else ''}"
+                + f"{graph.parent_node.signature.name}<{', '.join(str(i) for i in graph.graph_id)}>"
+            )
+            graph = graph.parent_node.graph
+        else:
+            graph = None
+    return f"[{'::'.join(reversed(graph_str))}]"
+
+
+@name.register
+def _(node: "Node") -> str:
+    long_name = (
+        f"{node.signature.wiring_path_name}.{node.signature.name}"
+        f"{(':' + node.signature.label) if node.signature.label else ''}"
+    )
+    
+    if node.graph.parent_node:
+        return long_name.replace(node.graph.parent_node.signature.wiring_path_name + ".", "")
+    else:
+        return long_name
+
+
+@multimethod
+def format_value(value):
+    if type(value) in (frozendict, tuple, frozenset):
+        return f"{len(value)} items"
+    else:
+        s = str(value)
+        if len(s) > 40:
+            s = s[:40] + "..."
+        return s
+
+
+@format_value.register
+def _(value: Graph):
+    return name(value)
+
+
+@format_value.register
+def _(value: Node):
+    if (out := value.output) is not None:
+        return format_value(out)
+    else:
+        return "-"
+
+
+@format_value.register
+def _(value: PythonPushQueueNodeImpl):
+    if (receiver := value.receiver) is not None:
+        with receiver:
+            return f"{len(receiver.queue)} items in the queue"
+    else:
+        return "-"
+
+
+@format_value.register
+def _(value: Union[PythonTimeSeriesValueOutput]):
+    # Very specific types here because want to check the _tp attribute to avoid potentially costly .value call
+    if value.valid:
+        if value._tp in (frozendict, frozenset):
+            return f"{len(value.value)} items"
+        else:
+            s = str(value.value)
+            if len(s) > 40:
+                s = s[:40] + "..."
+            return s
+    else:
+        return "INVALID"
+
+
+@format_value.register
+def _(value: Union[PythonTimeSeriesValueInput]):
+    if value.output is not None:
+        return format_value(value.output)
+    else:
+        return "INVALID"
+
+
+@format_value.register
+def _(value: Union[PythonTimeSeriesReferenceOutput, PythonTimeSeriesReferenceInput]):
+    if value.valid:
+        ref = value.value
+        if ref.valid:
+            if not ref.has_peer and ref.items:
+                return f"{len(ref.items)} items"
+            else:
+                return str(ref)
+    else:
+        return "INVALID"
+
+
+@format_value.register
+def _(value: PythonTimeSeriesReference):
+    if value.valid:
+        if not value.has_peer and value.items:
+            return f"{len(value.items)} items"
+        else:
+            return str(value)
+    else:
+        return "INVALID"
+
+
+@format_value.register
+def _(value: Union[TimeSeriesList, TimeSeriesSet, TimeSeriesBundle, TimeSeriesDict]):
+    if value.valid:
+        return f"{len(value)} items"
+    else:
+        return "INVALID"
+
+
+@multimethod
+def format_timestamp(value):
+    return None
+
+
+@format_timestamp.register
+def _(value: Node):
+    return value.graph._schedule[value.node_ndx]
+
+
+@format_timestamp.register
+def _(value: Graph):
+    return value.parent_node.last_evaluation_time if value.parent_node else None
+
+
+@format_timestamp.register
+def _(value: Union[TimeSeriesOutput, TimeSeriesInput]):
+    if value.valid:
+        return value.last_modified_time
+    else:
+        return None
+
+
+@multimethod
+def enum_items(value):
+    yield from ()
+
+
+@enum_items.register
+def _(value: frozendict):
+    yield from value.items()
+
+
+@enum_items.register
+def _(value: frozenset):
+    yield from ((k, k) for k in value)
+
+
+@enum_items.register
+def _(value: Union[PythonTimeSeriesReferenceOutput, PythonTimeSeriesReferenceInput]):
+    if value.valid:
+        ref = value.value
+        if ref.valid:
+            if not ref.has_peer and ref.items:
+                yield from enumerate(ref.items)
+    yield from ()
+
+
+@enum_items.register
+def _(value: Union[TimeSeriesList, TimeSeriesBundle, TimeSeriesDict]):
+    if value.valid:
+        yield from value.items()
+    yield from ()
+
+
+@enum_items.register
+def _(value: TimeSeriesSet):
+    if value.valid:
+        yield from ((k, k) for k in value.values())
+    yield from ()
+
+
+@enum_items.register
+def _(value: PythonNestedNodeImpl):
+    yield from value.enum_nested_graphs()
+
+
+@multimethod
+def format_type(value):
+    return type(value).__name__
+
+
+@format_type.register
+def _(value: Node):
+    return value.signature.node_type.name.replace("_NODE", "")
+
+
+@format_type.register
+def _(value: PythonNestedNodeImpl):
+    return {
+        PythonTsdMapNodeImpl: "MAP",
+        PythonMeshNodeImpl: "MESH",
+        PythonServiceNodeImpl: "SERVICE",
+        PythonReduceNodeImpl: "REDUCE",
+        PythonSwitchNodeImpl: "SWITCH",
+        PythonTryExceptNodeImpl: "TRY_EXCEPT",
+        PythonComponentNodeImpl: "COMPONENT",
+    }.get(type(value), "?")
+
+
+@format_type.register
+def _(value: PythonTimeSeriesReference):
+    return "REF"
+
+@format_type.register
+def _(value: TimeSeriesInput):
+    path = []
+    while value.parent_input:
+        if isinstance(value.parent_input, TimeSeriesBundle):
+            path.append(next(k for i, (k, v) in enumerate(value.parent_input._ts_values.items()) if v is value))
+        else:
+            path.append(0)
+        value = value.parent_input
+
+    if not path:
+        return "INPUTS"
+
+    tp = value.owning_node.signature.time_series_inputs[path[-1]]
+    for i in reversed(path[:-1]):
+        if isinstance(tp, HgTSBTypeMetaData):
+            tp = tp.bundle_schema_tp.meta_data_schema[i]
+        else:
+            tp = tp.value_tp
+
+    return str(tp)
+
+
+@format_type.register
+def _(value: TimeSeriesOutput):
+    path = []
+    while value.parent_output:
+        if isinstance(value.parent_output, TimeSeriesBundle):
+            path.append(next(k for i, (k, v) in enumerate(value.parent_output._ts_values.items()) if v is value))
+        else:
+            path.append(0)
+        value = value.parent_output
+
+    tp = value.owning_node.signature.time_series_output
+    for i in reversed(path):
+        if isinstance(tp, HgTSBTypeMetaData):
+            tp = tp.bundle_schema_tp.meta_data_schema[i]
+        else:
+            tp = tp.value_tp
+
+    return re.sub(r"<class '(?:\w+\.)+", r"", str(tp)).replace("'>", "")
+
+
+@multimethod
+def inspect_item(value, key):
+    return value[key]
+
+
+@inspect_item.register
+def _(value: PythonNestedNodeImpl, key):
+    return next(v for k, v in value.enum_nested_graphs() if k == key)
+
+
+@inspect_item.register
+def _(value: PythonTimeSeriesReference, key):
+    return value.items[key]
+
+
+@inspect_item.register
+def _(value: Union[PythonTimeSeriesReferenceInput | PythonTimeSeriesReferenceOutput], key):
+    return value.value.items[key]
diff --git a/src/hgraph/debug/_trace_controller.py b/src/hgraph/debug/_trace_controller.py
index 86a062aa..f9dd7c33 100644
--- a/src/hgraph/debug/_trace_controller.py
+++ b/src/hgraph/debug/_trace_controller.py
@@ -55,7 +55,7 @@ def control_trace(request: TSD[int, TS[HttpRequest]], _state: STATE = None) -> T
                 out[i] = HttpResponse(status_code=200, body=form.format(result="", **current_options))
 
             if (command := r.url_parsed_args[0]) not in ("start", "stop"):
-                return HttpResponse(status_code=400, body=form.format(result="Invalid command", **current_options))
+                out[i] = HttpResponse(status_code=400, body=form.format(result="Invalid command", **current_options))
 
             options = r.query
             options = {
diff --git a/src/hgraph/debug/inspector_template.html b/src/hgraph/debug/inspector_template.html
new file mode 100644
index 00000000..ce54555d
--- /dev/null
+++ b/src/hgraph/debug/inspector_template.html
@@ -0,0 +1,141 @@
+<!DOCTYPE html>
+<html>
+    <head>
+        <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, minimum-scale=1, user-scalable=no" />
+
+        <script type="module" src="http://{{host}}:{{port}}/node_modules/@finos/perspective-viewer/dist/cdn/perspective-viewer.js"></script>
+        <script type="module" src="http://{{host}}:{{port}}/node_modules/@finos/perspective-workspace/dist/cdn/perspective-workspace.js"></script>
+        <script type="module" src="http://{{host}}:{{port}}/node_modules/@finos/perspective-viewer-datagrid/dist/cdn/perspective-viewer-datagrid.js"></script>
+        <script type="module" src="http://{{host}}:{{port}}/node_modules/@finos/perspective-viewer-d3fc/dist/cdn/perspective-viewer-d3fc.js"></script>
+        <script type="module" src="/node_modules/perspective-viewer-datagrid-norollups/dist/cdn/perspective-viewer-datagrid-norollups.js"></script>
+        <script type="module" src="/node_modules/perspective-viewer-summary/dist/cdn/perspective-viewer-summary.js"></script>
+
+        <link rel="stylesheet" crossorigin="anonymous" href="http://{{host}}:{{port}}/node_modules/@finos/perspective-viewer/dist/css/themes.css" />
+        <link rel="stylesheet" crossorigin="anonymous" href="http://{{host}}:{{port}}/node_modules/@finos/perspective-workspace/dist/css/pro-dark.css" />
+
+        <style>
+            perspective-workspace {
+                position: absolute;
+                top: 0;
+                left: 0;
+                right: 0;
+                bottom: 0;
+            }
+
+        </style>
+    </head>
+
+    <body>
+        <perspective-workspace id="workspace"></perspective-workspace>
+
+        <script type="module">
+            import perspective from "http://{{host}}:{{port}}/node_modules/@finos/perspective/dist/cdn/perspective.js";
+
+            window.addEventListener("DOMContentLoaded", async function () {
+                const workspace = window.workspace
+
+                const websocket_ro = perspective.websocket("ws://{{host}}:{{port}}/websocket_readonly");
+                const websocket_rw = perspective.websocket("ws://{{host}}:{{port}}/websocket_editable");
+
+                //{% for table_name in mgr.get_table_names() %}
+                workspace.tables.set(
+                    "{{table_name}}",
+                    {{"websocket_ro" if not mgr.is_table_editable(table_name) else "websocket_rw"}}.open_table("{{table_name}}")
+                );
+                //{% end %}
+
+                const req = await fetch("/layout/{{url}}");
+                const json = await req.json();
+                window.workspace.restore(json);
+
+                const observe = await fetch("/inspect/expand/", {cache: "no-store"});
+
+                window.workspace.addEventListener("workspace-layout-update", async () => {
+                    await fetch("/layout/{{url}}", {method: "post", body: JSON.stringify(await window.workspace.save())});
+
+                    for (const g of document.querySelectorAll("perspective-viewer[table='inspector'] perspective-viewer-datagrid")){
+                        const table = g.shadowRoot.querySelector("regular-table")
+
+                        var sheet = new CSSStyleSheet
+                        sheet.replaceSync( `.highlight { background-color: pink }`)
+                        g.shadowRoot.adoptedStyleSheets.push(sheet)
+
+                        let SELECTED_ROW = null;
+                        table.addStyleListener(function fixOverflow() {
+                            for (const tr of table.children[0].children[1].children) {
+                                const meta = table.getMeta(tr.children[0]);
+                                if (meta.y === SELECTED_ROW) {
+                                    tr.classList.add("highlight");
+                                }else{
+                                    tr.classList.remove("highlight");
+                                }
+
+                                for (const td of tr.children) {
+                                    td.style.overflow = "clip";
+                                }
+                            }
+                        })
+
+                        table.addEventListener("click", async function observerClickEventListener(event) {
+                            SELECTED_ROW = null
+
+                            if (event.target.tagName === "TD") {
+                                const meta = table.getMeta(event.target);
+                                const stuff = await table._view_cache.view(0, meta.y, 10, meta.y + 1);
+                                const row = Object.assign(...stuff.column_headers.map((k, i) => ({[k]: stuff.metadata[i][0]})));
+                                console.log(meta)
+                                console.log(row)
+                                if (meta.column_header[0] === 'X'){
+                                    if (meta.value === '+') {
+                                        await fetch("/inspect/expand/" + row.id, {cache: "no-store"});
+                                    } else {
+                                        await fetch("/inspect/collapse/" + row.id, {cache: "no-store"});
+                                    }
+                                }
+                                if (meta.column_header[0] === 'value'){
+                                    if (event.ctrlKey) {
+                                        const new_row = await (await fetch("/inspect/ref/" + row.id, {cache: "no-store"})).text();
+                                        window.setTimeout(async () => {
+                                            const id_col = stuff.column_headers.findIndex((c) => {
+                                                return c[0] === 'id'
+                                            })
+                                            const ids = await table._view_cache.view(id_col, 0, id_col + 1, meta.y);
+                                            let row_num = ids.metadata[0].indexOf(new_row);
+                                            table.scrollToCell(0, row_num, ids.num_columns, ids.num_rows)
+                                            SELECTED_ROW = row_num;
+                                        }, 100);
+                                    }
+                                }
+                            }
+                        });
+
+                        table.addEventListener("dblclick", async function observerClickEventListener(event) {
+                            SELECTED_ROW = null
+
+                            if (event.target.tagName === "TD") {
+                                const meta = table.getMeta(event.target);
+                                const stuff = await table._view_cache.view(0, meta.y, 10, meta.y + 1);
+                                const row = Object.assign(...stuff.column_headers.map((k, i) => ({[k]: stuff.metadata[i][0]})));
+                                console.log(meta)
+                                console.log(row)
+                                if (meta.column_header[0] === 'name'){
+                                    if (row.X === '+') {
+                                        if (event.ctrlKey) {
+                                            await fetch("/inspect/expand/" + row.id + "?" + new URLSearchParams({all: 'true'}).toString(),
+                                                {cache: "no-store"});
+                                        } else {
+                                            await fetch("/inspect/expand/" + row.id, {cache: "no-store"});
+                                        }
+                                    } else {
+                                        await fetch("/inspect/collapse/" + row.id, {cache: "no-store"});
+                                    }
+                                }
+                            }
+                        });
+                    }
+
+                });
+            });
+        </script>
+    </body>
+</html>
diff --git a/src/hgraph/nodes/_mesh_util.py b/src/hgraph/nodes/_mesh_util.py
index 9ddaf425..a893bffa 100644
--- a/src/hgraph/nodes/_mesh_util.py
+++ b/src/hgraph/nodes/_mesh_util.py
@@ -33,10 +33,14 @@ def mesh_subscribe_node(
     mesh_node: PythonMeshNodeImpl = mesh.value.output.owning_node
     key = _find_mesh_key(_output.owning_node, mesh_node)
 
+    new_dependency_key = item.value
+
     if _output.valid:
-        mesh_node._remove_graph_dependency(key, state.key)
+        if state.key == new_dependency_key:
+            return None  # mesh ref cannot change
+        else:
+            mesh_node._remove_graph_dependency(key, state.key)
 
-    new_dependency_key = item.value
     available_now = mesh_node._add_graph_dependency(key, new_dependency_key)
     state.key = item.value
 


Test patch:
diff --git a/src/hgraph/test/_node_printer.py b/src/hgraph/test/_node_printer.py
index 3de5802c..26ab2d69 100644
--- a/src/hgraph/test/_node_printer.py
+++ b/src/hgraph/test/_node_printer.py
@@ -69,7 +69,7 @@ def _print_node(
             if add_input:
                 inputs = node.inputs
                 node_signature += ", ".join(
-                    f"{f'*{arg}*' if (mod_ := (in_ := inputs[arg]).modified) else arg}={(in_.delta_value if mod_ else in_.value) if in_.valid else '<UnSet>'}"
+                    f"{f'*{arg}*' if (mod_ := (in_ := inputs[arg]).modified) else arg}={(in_.delta_value if mod_ else '') if in_.valid else '<UnSet>'}"
                     for arg in node.signature.time_series_inputs.keys()
                 )
                 if node.signature.uses_scheduler:
diff --git a/tests/python/unit/hgraph/_operators/test_control_operators.py b/tests/python/unit/hgraph/_operators/test_control_operators.py
index 3066de87..f0264f3a 100644
--- a/tests/python/unit/hgraph/_operators/test_control_operators.py
+++ b/tests/python/unit/hgraph/_operators/test_control_operators.py
@@ -1,7 +1,7 @@
 import pytest
 
 from hgraph import graph, TS, all_, any_, TSB, TimeSeriesSchema, Size, TSL, SIZE, merge, REF, const, BoolResult, if_, \
-    route_by_index, race, if_true, if_then_else
+    route_by_index, race, if_true, if_then_else, nothing
 from hgraph.test import eval_node
 
 
@@ -108,6 +108,14 @@ def app(tsl: TSL[TS[int], Size[3]]) -> REF[TS[int]]:
     assert eval_node(app, [{1: 100}, {0: 200, 2: 300}, {1: 300, 2: 500}, {0: 600}]) == [100, None, 300, None]
 
 
+def test_race_scalars_invalid():
+    @graph
+    def app(ts: TS[int]) -> REF[TS[int]]:
+        return race(ts, nothing[TS[int]]())
+
+    assert eval_node(app, [None, 1, None]) == [None, 1, None]
+
+
 def test_race_tsls():
     @graph
     def app(invalidate: TS[bool]) -> TS[int]:
diff --git a/tests/python/unit/hgraph/_operators/test_stream_operators.py b/tests/python/unit/hgraph/_operators/test_stream_operators.py
index fccc8048..272915d4 100644
--- a/tests/python/unit/hgraph/_operators/test_stream_operators.py
+++ b/tests/python/unit/hgraph/_operators/test_stream_operators.py
@@ -27,7 +27,7 @@
     NodeException,
     batch,
     step,
-    slice_,
+    slice_, TSD, REMOVE,
 )
 from hgraph.test import eval_node
 
@@ -152,6 +152,22 @@ def g_float(ts: TS[float], abs_tol: float) -> TS[float]:
     ]
 
 
+def test_drop_dups_tsd():
+    @graph
+    def g(ts: TSD[int, TS[int]]) -> TSD[int, TS[int]]:
+        return dedup(ts)
+
+    assert eval_node(g, [{1: 1, 2: 2}, {1: 1, 2: 3}, {1: REMOVE, 2: 3}]) == [{1: 1, 2: 2}, {2: 3}, {1: REMOVE}]
+
+
+def test_drop_dups_tsl():
+    @graph
+    def g(ts: TSL[TS[int], Size[2]]) -> TSL[TS[int], Size[2]]:
+        return dedup(ts)
+
+    assert eval_node(g, [{0: 1}, {0: 1, 1: 2}, None, {0: 1, 1: 3}]) == [{0: 1}, {1: 2}, None, {1: 3}]
+
+
 def test_filter_int():
     @graph
     def g(condition: TS[bool], ts: TS[int]) -> TS[int]:
@@ -179,16 +195,17 @@ def test_throttle():
     def g(ts: TS[int], period: timedelta) -> TS[int]:
         return throttle(ts, period)
 
-    assert eval_node(g, [1, 1, 2, 3, 5, 2, 1], 2 * MIN_TD, __end_time__=MIN_ST + 10 * MIN_TD) == [
-        None,
-        1,
-        None,
-        3,
-        None,
-        2,
-        None,
-        1,
-    ]
+    assert eval_node(g, [1, 1, 2, 3, 5, 2, 1], 2 * MIN_TD, __end_time__=MIN_ST + 10 * MIN_TD
+                     ) == [1, None, 2, None, 5, None, 1]
+
+
+def test_throttle_tsd():
+    @graph
+    def g(ts: TSD[int, TS[int]], period: timedelta) -> TSD[int, TS[int]]:
+        return throttle(ts, period)
+
+    assert eval_node(g, [None, {1: 1}, {2: 2}, {1: 2}, None, {2: REMOVE}, {1: REMOVE}, {1: 1}], 3 * MIN_TD, __end_time__=MIN_ST + 10 * MIN_TD
+                     ) == [None, {1: 1}, None, None, {1: 2, 2: 2}, None, None, {2: REMOVE, 1: 1}]
 
 
 def test_take():
diff --git a/tests/python/unit/hgraph/_runtime/test_scheduler.py b/tests/python/unit/hgraph/_runtime/test_scheduler.py
index d123208c..0679dcd8 100644
--- a/tests/python/unit/hgraph/_runtime/test_scheduler.py
+++ b/tests/python/unit/hgraph/_runtime/test_scheduler.py
@@ -25,9 +25,9 @@ def test_map_scheduler():
     def map_scheduler(tsd: TSD[str, TS[int]]) -> TSD[str, TS[int]]:
         return map_(my_scheduler, tsd, "TEST")
 
-    assert (eval_node(map_scheduler, [{"ab1": 9, "ab2": 9}, {"ab1": 9}, {"ab2": 9}]) == [
+    assert (eval_node(map_scheduler, [{"ab1": 9, "ab2": 9}, {"ab1": 9}, {"ab2": 9}] + [None] * 7 + [{"ab1": 2}], __trace__=True) == [
         frozendict({"ab1": 9, "ab2": 9}), frozendict({"ab1": 9}), frozendict({"ab2": 9}), ] + [None] * 7 +
-            [frozendict({"ab1": -1}), frozendict({"ab2": -1})])
+            [frozendict({"ab1": 2}), frozendict({"ab2": -1}), {"ab1": -1}])
 
 
 @compute_node(valid=("ts1",))