From 148d50f2bd28a50d45762b3ea048b3b6c91b581b Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 23 Aug 2026 06:35:28 +0200 Subject: [PATCH] Grow a node with its ports, stop it flickering, draw what it reaches out to - A node's height follows the ports on its busiest side. It is a function of the document, so `layoutGraph` reserves exactly what is drawn and nothing measured is fed back into the layout. - The three status controls now sit in slots that are there whether the control is or not. A node running many times a second mounted and unmounted the stop button on every execution, resizing the card each time. - A port bound to another flow's message is drawn as a label, naming the node at the far end and its type. Only the opposite direction was answered before. The scan behind both is now cached on the store's commit counter rather than reading every flow per request. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016ZeGnqVsf5VHQqvz4HdUhN --- backend/fluksio/api/routes/flows.py | 79 +++++--- backend/fluksio/flow/controller.py | 81 ++++++-- backend/fluksio/flow/store.py | 10 + backend/tests/flow/test_endpoints.py | 111 +++++++++++ frontend/src/components/Flow/FlowEditor.tsx | 22 ++- frontend/src/components/Flow/FlowNode.tsx | 185 +++++++++++-------- frontend/src/components/Flow/layout.check.ts | 50 +++++ frontend/src/components/Flow/layout.ts | 47 ++++- 8 files changed, 454 insertions(+), 131 deletions(-) create mode 100644 backend/tests/flow/test_endpoints.py create mode 100644 frontend/src/components/Flow/layout.check.ts diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 325ec52..3060626 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -30,7 +30,7 @@ from fluksio.core.db import engine from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardStore from fluksio.flow.events import event_bus -from fluksio.flow.messages import qualify +from fluksio.flow.messages import flow_of, qualify from fluksio.flow.panels import messages_for from fluksio.flow.pipeline import ValidationIssue from fluksio.flow.runs import RunRejected @@ -130,36 +130,65 @@ class TriggerRequest(BaseModel): values: dict[str, Any] = {} -def _endpoints(controller: FlowController, flow: str) -> list[Endpoint]: - """Everything wired into ``flow`` from outside it.""" - found: list[Endpoint] = [] +def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]: + """Everything wired into this flow from outside it.""" + flow = definition.name + found: dict[str, Endpoint] = {} dashboards: DashboardStore | None = getattr(controller, "dashboards", None) if dashboards is not None: for binding in dashboards.bindings_for(flow): - found.append( - Endpoint( - kind="dashboard", - id=f"dashboard:{binding['dashboard']}:{binding['widget']}", - label=binding["title"], - detail=binding["type"], - provides=[binding["provides"]] if binding["provides"] else [], - requires=binding["requires"], - ) + found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint( + kind="dashboard", + id=f"dashboard:{binding['dashboard']}:{binding['widget']}", + label=binding["title"], + detail=binding["type"], + provides=[binding["provides"]] if binding["provides"] else [], + requires=binding["requires"], ) - for other, node_id, provides, requires in controller.cross_flow_nodes(flow): - found.append( - Endpoint( - kind="flow", - id=f"flow:{other}.{node_id}", - label=f"{other}.{node_id}", - detail="flow", - provides=provides, - requires=requires, - ) + def _flow_endpoint(key: str, detail: str) -> Endpoint: + """One label per node on the far side, however many names reach it.""" + return found.setdefault( + f"flow:{key}", + Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail), ) - return found + + # An outsider reaching into this flow. + for other, node_id, provides, requires in controller.cross_flow_nodes(flow): + endpoint = _flow_endpoint(f"{other}.{node_id}", "flow") + endpoint.provides += provides + endpoint.requires += requires + + # And this flow reaching out: its own ports bound to a message of another + # flow. Read from the working document rather than from the store, so a + # name just typed is drawn before it has been published — the same reason + # the canvas draws its own boundary from the document. + for node in definition.nodes: + for spec, ours_publishes in [ + *((spec, False) for spec in node.requires), + *((spec, True) for spec in node.provides), + ]: + message = qualify(flow, spec.name or "") + if not message or flow_of(message) == flow: + continue + # The node at the other end, so the label reads like a dashboard's: + # what it is on the first line, what sort of thing it is on the + # second. A message no published flow declares yet has no other end + # to name, so it is drawn as the message it is. + far = controller.message_node( + message, published=not ours_publishes, exclude=flow + ) + endpoint = _flow_endpoint( + f"{far.flow}.{far.node}" if far else message, + far.type if far else "flow", + ) + # An endpoint publishing into this flow is what this flow reads. + side = endpoint.requires if ours_publishes else endpoint.provides + if message not in side: + side.append(message) + + return list(found.values()) def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: @@ -168,7 +197,7 @@ def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: "enabled": controller.is_enabled(name), "paused": controller.is_paused(name), } - endpoints = _endpoints(controller, name) + endpoints = _endpoints(controller, definition) if controller.store.has_draft(name): # Report the draft the editor is showing, not the version running # underneath it — otherwise a node the author just broke looks fine. diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index d8d8b56..1eddce6 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -113,6 +113,17 @@ class RunContext: run_id: str +@dataclass(frozen=True, slots=True) +class Wiring: + """One published node and the qualified messages it is bound to.""" + + flow: str + node: str + type: str + provides: list[str] + requires: list[str] + + class EmitSink: """Turns a worker's mid-call frames back into the node's own outputs. @@ -315,6 +326,9 @@ class FlowController: self.history_limits: dict[str, int] = {} self._lock = asyncio.Lock() self._failures: asyncio.Task[None] | None = None + #: Every published node's ports, with the store revision it was read + #: at — see `_wiring`. + self._wiring_cache: tuple[int, list[Wiring]] | None = None # ------------------------------------------------------------------------- # Lifecycle @@ -843,6 +857,34 @@ class FlowController: ) return infos + def _wiring(self) -> list[Wiring]: + """Every published node and the qualified messages it is bound to. + + Opening one flow asks what every other flow is wired to, in both + directions, which otherwise reads and parses every flow on the disk per + request. Rebuilt when the store commits — see `FlowStore.revision` — + rather than cached with a lifetime, so a save is visible immediately. + """ + revision = self.store.revision + if self._wiring_cache is not None and self._wiring_cache[0] == revision: + return self._wiring_cache[1] + + index: list[Wiring] = [] + for other in self.store.read_all(): + for node in other.nodes: + provides = sorted( + qualify(other.name, spec.name) for spec in _bound(node.provides) + ) + requires = sorted( + qualify(other.name, spec.name) for spec in _bound(node.requires) + ) + if provides or requires: + index.append( + Wiring(other.name, node.id, node.type, provides, requires) + ) + self._wiring_cache = (revision, index) + return index + def cross_flow_nodes( self, flow: str ) -> list[tuple[str, str, list[str], list[str]]]: @@ -854,24 +896,33 @@ class FlowController: """ prefix = f"{flow}." found = [] - for other in self.store.read_all(): - if other.name == flow: + for entry in self._wiring(): + if entry.flow == flow: continue - for node in other.nodes: - provides = sorted( - qualify(other.name, spec.name) - for spec in _bound(node.provides) - if qualify(other.name, spec.name).startswith(prefix) - ) - requires = sorted( - qualify(other.name, spec.name) - for spec in _bound(node.requires) - if qualify(other.name, spec.name).startswith(prefix) - ) - if provides or requires: - found.append((other.name, node.id, provides, requires)) + provides = [name for name in entry.provides if name.startswith(prefix)] + requires = [name for name in entry.requires if name.startswith(prefix)] + if provides or requires: + found.append((entry.flow, entry.node, provides, requires)) return found + def message_node( + self, message: str, published: bool, exclude: str = "" + ) -> Wiring | None: + """The published node on the far side of ``message``. + + ``published`` asks for the one that publishes it, otherwise for one + that reads it, and ``exclude`` names the flow doing the asking — whose + own nodes are never the far side of anything. A message may have + several readers; the first stands for them, which is all a label naming + the other end has to say. + """ + for entry in self._wiring(): + if entry.flow == exclude: + continue + if message in (entry.provides if published else entry.requires): + return entry + return None + def brain_graph(self) -> BrainGraph: """Every published flow as one graph, merged on what its nodes talk to. diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index fe2ad3f..3d6cfa2 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -103,6 +103,12 @@ class FlowStore: # Draft writes are check-and-set, so two clients saving at once must not # interleave between reading the current version and writing the next. self._write_lock = threading.Lock() + #: Bumped on every commit, so something derived from every flow at once + #: can tell whether it is still current without reading them all again. + #: In memory rather than `head()`, which is a git subprocess per call — + #: it therefore counts from zero per process and misses an edit made on + #: disk behind the API, which no writer here does. + self.revision = 0 if not (self.root / ".git").exists(): self._git("init", "-q") self._commit("Initialise flow store", allow_empty=True) @@ -135,6 +141,10 @@ class FlowStore: ) def _commit(self, message: str, allow_empty: bool = False) -> None: + # Before the commit rather than after it: what a reader has to notice + # is that the files changed, which is already true whether or not git + # is there to record it. + self.revision += 1 self._git("add", "-A") result = self._git( "-c", diff --git a/backend/tests/flow/test_endpoints.py b/backend/tests/flow/test_endpoints.py new file mode 100644 index 0000000..f448e81 --- /dev/null +++ b/backend/tests/flow/test_endpoints.py @@ -0,0 +1,111 @@ +"""What a flow's canvas draws for the wiring that leaves it. + +Both directions matter and only one used to be answered: an outsider reaching +into this flow was drawn, and this flow reaching out was not — so a port bound +to another flow's message had nothing on the canvas to account for it. +""" + +from pathlib import Path + +import pytest + +from fluksio.api.routes.flows import _endpoints +from fluksio.flow.controller import FlowController +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.schemas import FlowDef, NodeDef +from fluksio.flow.store import FlowStore + + +def spec(name: str) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT) + + +@pytest.fixture +def controller(tmp_path: Path) -> FlowController: + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="house", + nodes=[ + NodeDef(id="sensor", type="mqtt", provides=[spec("temp")]), + # Reaches into `garage`, which is the direction that was drawn. + NodeDef(id="relay", type="change", provides=[spec("garage.open")]), + ], + ) + ) + store.write_flow( + FlowDef( + name="garage", + nodes=[ + NodeDef(id="door", type="python", requires=[spec("garage.open")]), + # Reaches out into `house`, which was drawn nowhere. + NodeDef(id="watch", type="python", requires=[spec("house.temp")]), + ], + ) + ) + return FlowController(store) + + +def labels( + controller: FlowController, definition: FlowDef +) -> dict[str, tuple[str, str]]: + return { + endpoint.id: (endpoint.label, endpoint.detail) + for endpoint in _endpoints(controller, definition) + } + + +def test_a_port_bound_to_another_flow_names_the_node_at_the_far_end(controller): + endpoints = labels(controller, controller.store.read_flow("garage")) + + # `house.relay` publishes `garage.open`, so it reaches in — drawn before. + # `house.sensor` publishes `house.temp`, which `garage.watch` reads, so + # this flow reaches out — and the label names the node and its type, the + # way a dashboard's label names the widget and its type. + assert endpoints == { + "flow:house.relay": ("house.relay", "flow"), + "flow:house.sensor": ("house.sensor", "mqtt"), + } + + +def test_a_node_reaching_into_and_out_of_one_flow_is_one_label(controller): + endpoints = _endpoints(controller, controller.store.read_flow("house")) + + # `garage.door` reads what `house.relay` publishes, and `garage.watch` + # reads `house.temp` — two directions, two nodes, two labels, no collision. + assert {endpoint.id for endpoint in endpoints} == { + "flow:garage.watch", + "flow:garage.door", + } + door = next(one for one in endpoints if one.id == "flow:garage.door") + assert door.requires == ["garage.open"] + + +def test_a_name_no_published_flow_declares_is_drawn_as_the_name(controller): + definition = controller.store.read_flow("house") + definition.nodes.append( + NodeDef(id="guess", type="python", requires=[spec("shed.humidity")]) + ) + + endpoints = labels(controller, definition) + + assert endpoints["flow:shed.humidity"] == ("shed.humidity", "flow") + + +def test_the_wiring_index_is_read_once_per_store_revision(controller): + reads = 0 + original = controller.store.read_all + + def counted(): + nonlocal reads + reads += 1 + return original() + + controller.store.read_all = counted # type: ignore[method-assign] + _endpoints(controller, controller.store.read_flow("house")) + _endpoints(controller, controller.store.read_flow("garage")) + assert reads == 1 + + controller.store.write_flow(FlowDef(name="shed", nodes=[])) + _endpoints(controller, controller.store.read_flow("house")) + assert reads == 2 diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index 5cf9544..28d0f69 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -61,7 +61,7 @@ import { FIT_VIEW, FIT_VIEW_PANEL, FlowDock } from "./FlowDock" import { FlowNode, type FlowNodeData } from "./FlowNode" import { FlowPanel } from "./FlowPanel" import { LiveEdge } from "./LiveEdge" -import { type Direction, layoutGraph } from "./layout" +import { type Direction, layoutGraph, nodeHeight } from "./layout" import { NodePanel } from "./NodePanel" import { RunDialog } from "./RunDialog" import "./flow.css" @@ -512,9 +512,27 @@ function FlowEditorInner({ ...external.nodes.map((node) => node.id), ] const shapeKey = `${direction}|${key}|${ids.join(",")}` + // How tall each node's ports make it, which `FlowNode` draws to the same + // number. A function of the document — the bindings key above already covers + // every port, so this changes exactly when the layout has to run again, and + // nothing measured is ever fed back into it. + const heights = + direction === "LR" + ? new Map( + definitions.map((node) => [ + node.id, + nodeHeight( + Math.max( + (node.requires ?? []).length, + (node.provides ?? []).length, + ), + ), + ]), + ) + : new Map() // biome-ignore lint/correctness/useExhaustiveDependencies: the shape key is the dependency; the arrays are rebuilt every render. const positions = useMemo( - () => layoutGraph(ids, edges, direction), + () => layoutGraph(ids, edges, direction, heights), [shapeKey, edges], ) diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index 576a31f..c87029c 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -32,6 +32,7 @@ import { useIsMobile } from "@/hooks/useMobile" import { duration } from "@/lib/motion" import { cn } from "@/lib/utils" import { portOf } from "./deriveEdges" +import { nodeHeight, PORT_SPAN } from "./layout" import { liveStore, useNodeEmits, @@ -79,7 +80,9 @@ export type FlowNodeData = { /** Spread handles along the node's edge so several ports stay reachable. */ function handleOffset(index: number, total: number): string { if (total <= 1) return "50%" - const span = 60 + // The same fraction `nodeHeight` sizes the node for, so a node is always + // tall enough for the ports this spreads down it. + const span = PORT_SPAN * 100 return `${50 - span / 2 + (span / (total - 1)) * index}%` } @@ -128,6 +131,14 @@ function FlowNodeComponent({ data, selected }: NodeProps) { // The graph runs top to bottom on a phone, so the ports have to face that // way too — see DESIGN-GUIDELINES.md → Responsive. const vertical = useIsMobile() + // Ports run down the sides only while the graph runs across, and that is the + // only direction in which their number decides the height: running downwards + // they spread along the node's width, which is fixed. Read from the document + // rather than measured, so `layoutGraph` can reserve exactly this much. + const ports = Math.max( + (definition.requires ?? []).length, + (definition.provides ?? []).length, + ) // Whether a pulse is playing right now; the ring is mounted only while it is. const [firing, setFiring] = useState(false) // The count outlives this component: the store is module-level, and a @@ -186,8 +197,9 @@ function FlowNodeComponent({ data, selected }: NodeProps) { /> ) : null}
- {running ? ( - - - - - Stop this node - - ) : null} + {/* Each control sits in a slot that is there whether the control + is or not. They come and go with what the node is doing — and a + node running many times a second comes and goes that often — so + in the flex row itself they would resize the card on every + execution, which reads as a flickering shape. */} + + {running ? ( + + + + + Stop this node + + ) : null} + - {(status === "error" || failedEarlier) && onShowLogs ? ( - - - - - - {failedEarlier - ? `Failed at ${failedAt} — show the traceback` - : "Show the traceback"} - - - ) : null} + + {(status === "error" || failedEarlier) && onShowLogs ? ( + + + + + + {failedEarlier + ? `Failed at ${failedAt} — show the traceback` + : "Show the traceback"} + + + ) : null} + - {style ? ( - - - - - - {problem || style.label} - - - ) : null} + + {style ? ( + + + + + + {problem || style.label} + + + ) : null} +
56) + +for (let ports = 2; ports <= 12; ports += 1) { + const height = nodeHeight(ports) + // `handleOffset` spreads the ports over `PORT_SPAN` of the edge, so that is + // the room they actually get. Two handles must not touch. + const between = (PORT_SPAN * height) / (ports - 1) + assert.ok( + between > HANDLE, + `${ports} ports on a ${height}px node leave ${between}px between handles`, + ) + // Growing, never shrinking: one more port can only need more room. + assert.ok(height >= nodeHeight(ports - 1)) +} + +// The layout places top-left corners, so a taller node has to be lifted by its +// own half-height rather than by the default box's. +const tall = nodeHeight(8) +const placed = layoutGraph( + ["a", "b"], + [{ source: "a", target: "b" }], + "LR", + new Map([["b", tall]]), +) +const a = placed.get("a") +const b = placed.get("b") +assert.ok(a && b) +// Dagre centres the two on one rank line, so the taller one starts higher up +// by exactly the difference in half-heights. +assert.equal(Math.round(a.y - b.y), Math.round((tall - 56) / 2)) + +console.log("layout: ok") diff --git a/frontend/src/components/Flow/layout.ts b/frontend/src/components/Flow/layout.ts index ecce134..b3681ac 100644 --- a/frontend/src/components/Flow/layout.ts +++ b/frontend/src/components/Flow/layout.ts @@ -15,8 +15,33 @@ export type Direction = "LR" | "TB" /** `FlowNode` is `min-w-[168px] max-w-[220px]`; an endpoint is narrower. */ const NODE_W = 220 -/** Icon row plus two text lines, as measured. */ +/** Icon row plus two text lines, as measured. The floor, not the height. */ const NODE_H = 56 +/** + * How much of a node's edge the ports are spread over, as a fraction. + * + * Shared with `FlowNode`'s `handleOffset`, which does the spreading: the + * height below is chosen so that span has room for the ports, so the two + * cannot be allowed to drift apart. + */ +export const PORT_SPAN = 0.6 +/** + * Centre to centre between two handles. They are 12px across (`flow.css`), so + * this leaves 6px of rim between them. + */ +const PORT_PITCH = 18 + +/** + * How tall a node with this many ports on one side has to be. + * + * A pure function of the document — the ports are declared, so the height is + * known before anything is drawn. That is what keeps it out of the measuring + * problem below: nothing is fed back, so there is nothing to oscillate. + */ +export function nodeHeight(ports: number): number { + return Math.max(NODE_H, Math.ceil(((ports - 1) * PORT_PITCH) / PORT_SPAN)) +} + /** * Room for the live value an edge carries (`LiveEdge`'s chip is * `max-w-[140px]`). Reserved on the edge itself, so dagre routes nodes around @@ -58,6 +83,7 @@ function build( edges: { source: string; target: string }[], direction: Direction, wrap: Wrap[], + heights: Map, ) { const graph = new dagre.graphlib.Graph() graph.setDefaultEdgeLabel(() => ({})) @@ -74,7 +100,7 @@ function build( // Insertion order is what makes the result deterministic, so it follows the // document rather than whatever order the edges happen to mention nodes in. for (const id of ids) { - graph.setNode(id, { width: NODE_W, height: NODE_H }) + graph.setNode(id, { width: NODE_W, height: heights.get(id) ?? NODE_H }) } for (const edge of edges) { if (!graph.hasNode(edge.source) || !graph.hasNode(edge.target)) continue @@ -120,16 +146,21 @@ function wrapWideRanks(graph: ReturnType, ids: string[]): Wrap[] { /** * Lay the graph out and return each node's top-left corner. * - * ponytail: every node is treated as 220×56 rather than measured. Measuring - * would feed the result back into the layout and oscillate; if nodes ever grow - * past that box, take the sizes from `node.measured` once they have settled. + * `heights` is what a node's ports make it, from {@link nodeHeight}; anything + * left out is the plain box. It is read from the document rather than from the + * canvas on purpose — see that function. + * + * ponytail: every node is still treated as 220 wide rather than measured. + * Measuring would feed the result back into the layout and oscillate; if nodes + * ever grow past that, take the width from `node.measured` once it has settled. */ export function layoutGraph( ids: string[], edges: { source: string; target: string }[], direction: Direction, + heights: Map = new Map(), ): Map { - let graph = build(ids, edges, direction, []) + let graph = build(ids, edges, direction, [], heights) // Running downwards, a rank wider than the screen is the one thing the // layout can still do something about. Wrapping one rank pushes whatever was @@ -142,7 +173,7 @@ export function layoutGraph( const more = wrapWideRanks(graph, ids) if (!more.length) break wrap.push(...more) - graph = build(ids, edges, direction, wrap) + graph = build(ids, edges, direction, wrap, heights) } } @@ -153,7 +184,7 @@ export function layoutGraph( return [ id, node - ? { x: node.x - NODE_W / 2, y: node.y - NODE_H / 2 } + ? { x: node.x - NODE_W / 2, y: node.y - node.height / 2 } : { x: 0, y: 0 }, ] }),