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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ZeGnqVsf5VHQqvz4HdUhN

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.

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.
 

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",