stroblmeandClaude Opus 5
ace980b685
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
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
"""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}
|