Carry per-node emit counts in the live snapshot

The brain graph counts node_executed events client-side and the websocket is
torn down on every shell change, so anything a flow published during the
navigation gap was lost. The bus now keeps a session tally per qualified node
and the snapshot hands it back, letting a reconnecting client catch up.

Both the route and the tunnel connector build that snapshot from one helper
so the portal cannot drift from the direct connection.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HTsT1isxUjw5gtkJk8WhuA

backend/app/api/routes/flows.py

@@ -730,6 +730,25 @@ def read_message_history(
 # -----------------------------------------------------------------------------
 
 
+def snapshot_payload(controller: FlowController) -> dict[str, Any]:
+    """Everything a client needs to catch up, sent the moment it connects.
+
+    Also sent by the tunnel connector, which serves this websocket inline: the
+    two have to agree, so they build the message here rather than each their
+    own. ``emits`` is what the bus counted while nobody was listening — a
+    client that reconnects between two pages would otherwise start from zero.
+    """
+    return {
+        "type": "snapshot",
+        "values": controller.values(),
+        "nodes": [s.model_dump() for s in controller.node_statuses()],
+        "issues": [i.model_dump() for i in controller.issues],
+        "paused": controller.paused_flows(),
+        "logs": list(event_bus.recent_logs),
+        "emits": dict(event_bus.emits),
+    }
+
+
 @ws_router.websocket("/ws")
 async def flow_events(websocket: WebSocket, token: str = "") -> None:
     """Stream values, node status and execution events as they happen.
@@ -749,16 +768,7 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
         websocket.app.state, "flow_controller", None
     )
     if controller is not None:
-        await websocket.send_json(
-            {
-                "type": "snapshot",
-                "values": controller.values(),
-                "nodes": [s.model_dump() for s in controller.node_statuses()],
-                "issues": [i.model_dump() for i in controller.issues],
-                "paused": controller.paused_flows(),
-                "logs": list(event_bus.recent_logs),
-            }
-        )
+        await websocket.send_json(snapshot_payload(controller))
 
     async with event_bus.subscribe() as queue:
         receiver = asyncio.create_task(websocket.receive_text())

backend/app/cloud/connector.py

@@ -282,11 +282,13 @@ class CloudConnector:
         The flows websocket is subscribed to directly rather than dialled:
         httpx cannot speak websocket, and everything that endpoint does — check
         the token, send a snapshot, forward the bus — is a few lines against
-        objects this process already holds.
+        objects this process already holds. The snapshot is built by the route
+        itself, so the two cannot drift apart behind the portal's back.
         """
         from sqlmodel import Session
 
         from app.api.deps import user_from_token
+        from app.api.routes.flows import snapshot_payload
         from app.core.db import engine
         from app.flow.events import event_bus
 
@@ -314,18 +316,7 @@ class CloudConnector:
                     {
                         "op": "ws_msg",
                         "id": stream_id,
-                        "text": _dump(
-                            {
-                                "type": "snapshot",
-                                "values": controller.values(),
-                                "nodes": [
-                                    s.model_dump() for s in controller.node_statuses()
-                                ],
-                                "issues": [i.model_dump() for i in controller.issues],
-                                "paused": controller.paused_flows(),
-                                "logs": list(event_bus.recent_logs),
-                            }
-                        ),
+                        "text": _dump(snapshot_payload(controller)),
                     }
                 )
             )

backend/app/flow/events.py

@@ -29,6 +29,11 @@ class EventBus:
         # Kept whether or not anyone is listening, so opening the log panel
         # shows what just happened rather than an empty box.
         self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY)
+        # How often each qualified node has emitted, for the same reason: a
+        # client that reconnects between two pages would otherwise start from
+        # zero and draw a busy graph as idle. A session count, not a metric —
+        # the rollups in `metrics.py` are what survives a restart.
+        self.emits: dict[str, int] = {}
 
     def bind(self, loop: asyncio.AbstractEventLoop) -> None:
         """Attach the bus to the running event loop (called once at startup)."""
@@ -36,9 +41,16 @@ class EventBus:
 
     def publish(self, event: dict[str, Any]) -> None:
         """Publish an event from any thread."""
-        if event.get("type") == "node_log":
+        kind = event.get("type")
+        if kind == "node_log":
             # deque.append is atomic, so worker threads need no lock here.
             self.recent_logs.append(event)
+        elif kind == "node_executed" and event.get("outputs"):
+            # The condition the canvas animates on, so both ends agree on what
+            # counts as an emission. Two worker threads racing here can lose an
+            # increment, which is a count nobody is going to miss.
+            node = str(event.get("node") or "")
+            self.emits[node] = self.emits.get(node, 0) + 1
         loop = self._loop
         if loop is None or not self._subscribers:
             return

backend/tests/flow/test_emit_counts.py

@@ -0,0 +1,60 @@
+"""What each node has emitted, kept so a reconnecting client is not reset.
+
+The brain graph pulses a neuron per emission and tallies them in the browser.
+Moving between the shell and the canvas tears the websocket down, and the bus
+has no replay — so the count is kept here and handed back in the snapshot.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from app.api.routes.flows import snapshot_payload
+from app.flow.controller import FlowController
+from app.flow.events import EventBus, event_bus
+from app.flow.messages import DType, MessageSpec
+from app.flow.nodes import Node
+from app.flow.pipeline import Pipeline
+from app.flow.store import FlowStore
+
+
+def make_node(node_id: str, f, provides=()) -> Node:
+    node = Node(f=f, provides=list(provides), name=node_id)
+    node.assign_flow("house", node_id)
+    return node
+
+
+def test_a_node_that_publishes_advances_its_own_count():
+    bus = EventBus()
+    node = make_node(
+        "sensor",
+        lambda params: {"temp": 21.0},
+        provides=[MessageSpec(name="temp", dtype=DType.FLOAT)],
+    )
+    pipeline = Pipeline(nodes=[node], events=bus)
+
+    pipeline.run()
+    pipeline.run()
+
+    # Keyed the way `brain_graph` names its members, which is what the graph
+    # looks a count up by.
+    assert bus.emits == {"house.sensor": 2}
+
+
+def test_a_node_that_publishes_nothing_is_not_counted():
+    """Ran, but emitted nothing — the same thing the canvas declines to pulse."""
+    bus = EventBus()
+    pipeline = Pipeline(nodes=[make_node("quiet", lambda params: None)], events=bus)
+
+    pipeline.run()
+
+    assert bus.emits == {}
+
+
+def test_the_snapshot_carries_what_every_node_has_emitted(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    monkeypatch.setattr(event_bus, "emits", {"house.sensor": 3})
+    controller = FlowController(FlowStore(tmp_path / "flows"))
+
+    assert snapshot_payload(controller)["emits"] == {"house.sensor": 3}