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/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)),
                     }
                 )
             )