From ace980b685f512de4f8632e42f4323f1c4dc384b Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 20 Aug 2026 11:32:44 +0200 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01HTsT1isxUjw5gtkJk8WhuA --- backend/app/api/routes/flows.py | 30 ++++++++----- backend/app/cloud/connector.py | 17 ++------ backend/app/flow/events.py | 14 +++++- backend/tests/flow/test_emit_counts.py | 60 ++++++++++++++++++++++++++ 4 files changed, 97 insertions(+), 24 deletions(-) create mode 100644 backend/tests/flow/test_emit_counts.py diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 8a2ab5e..916a85e 100644 --- a/backend/app/api/routes/flows.py +++ b/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()) diff --git a/backend/app/cloud/connector.py b/backend/app/cloud/connector.py index c2f7253..27c384f 100644 --- a/backend/app/cloud/connector.py +++ b/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)), } ) ) diff --git a/backend/app/flow/events.py b/backend/app/flow/events.py index 87d7f6a..80c2755 100644 --- a/backend/app/flow/events.py +++ b/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 diff --git a/backend/tests/flow/test_emit_counts.py b/backend/tests/flow/test_emit_counts.py new file mode 100644 index 0000000..20da311 --- /dev/null +++ b/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}