Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Say what caused a value, and draw what is not a node

Browse Source 
Moving a dashboard slider lit up an edge between two nodes that had done
nothing. The canvas pulsed on the message's timestamp alone, and a message
has no idea who published it — so it credited whichever node happened to
be drawn as a producer.

That was never only about dashboards. Two nodes producing one message
pulsed both their edges whichever fired, and a message produced in another
flow changed with nothing on screen to account for it at all.

Values now carry their cause: a node, a dashboard widget, another flow, an
agent or an API caller. An edge pulses only for the producer that actually
published, and the edge inspector says where a value came from when it did
not come from a node.

What is not a node in this flow is now drawn as one — a label rather than
a card, because a dashboard with twenty tiles would otherwise bury the
logic the canvas exists to show. That covers cross-flow wiring too, which
is the link in/out affordance that has been missing.

They are never part of the document. They join at render, after everything
that reads or writes the canvas nodes, so an autosave, an undo or a delete
cannot reach them — with a Playwright test that drags a node and asserts
the stored flow still holds exactly what it did.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY
This commit is contained in:
stroblme
2026-08-16 15:51:54 +02:00
co-authored by Claude Fable 5
parent 3fa9141eb9
commit 75c26ef000
25 changed files with 1072 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/flows.py
+59
View File
@@ -17,6 +17,7 @@ from sqlmodel import Session
from app.api.deps import FlowControllerDep, get_current_user, user_from_token
from app.core.db import engine
from app.flow.controller import FlowController
from app.flow.dashboards import DashboardStore
from app.flow.events import event_bus
from app.flow.messages import qualify
from app.flow.pipeline import ValidationIssue
@@ -52,6 +53,27 @@ router = APIRouter(
ws_router = APIRouter(prefix="/flows", tags=["flows"])
class Endpoint(BaseModel):
"""Something wired into this flow that is not a node in it.
A dashboard control setting one of its messages, a tile showing one, or a
node in another flow on the far side of a dotted name. The canvas draws
these so a value never appears to come from nowhere — or worse, appears to
come from whichever node happens to be drawn as a producer.
"""
#: dashboard or flow.
kind: str
#: Stable within its kind, and used as the canvas node id.
id: str
label: str
#: What sort of widget, or the node type in the other flow.
detail: str = ""
#: Messages of this flow it publishes, and ones it reads.
provides: list[str] = []
requires: list[str] = []
class FlowDetail(BaseModel):
"""A flow plus how it is currently doing.
@@ -66,6 +88,8 @@ class FlowDetail(BaseModel):
has_draft: bool = False
enabled: bool = True
paused: bool = False
#: Dashboards and other flows wired into this one.
endpoints: list[Endpoint] = []
class ValidationResult(BaseModel):
@@ -92,12 +116,45 @@ 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] = []
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"],
)
)
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,
)
)
return found
def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
name = definition.name
running = {
"enabled": controller.is_enabled(name),
"paused": controller.is_paused(name),
}
endpoints = _endpoints(controller, name)
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.
@@ -107,12 +164,14 @@ def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
nodes=preview.nodes,
issues=preview.issues,
has_draft=True,
endpoints=endpoints,
**running,
)
return FlowDetail(
definition=definition,
nodes=controller.node_statuses(name),
issues=controller.flow_issues(name),
endpoints=endpoints,
**running,
)
backend/app/api/routes/messages.py
+16 -1
View File
@@ -14,6 +14,7 @@ from pydantic import BaseModel
from app.api.deps import FlowControllerDep, get_current_user
from app.flow.messages import flow_of
from app.flow.pipeline import ValueSource
from app.flow.state import as_number
router = APIRouter(
@@ -43,6 +44,12 @@ class MessagesPublic(BaseModel):
class PublishRequest(BaseModel):
value: Any
#: Where this came from, so the canvas can show it arriving from outside
#: rather than crediting whichever node is drawn as a producer.
source_kind: str = "api"
source_id: str = ""
source_label: str = ""
source_detail: str = ""
class MessageValue(BaseModel):
@@ -73,8 +80,16 @@ async def publish_message(
Only a message some flow declares can be published to: flows own the
namespace, and a dashboard is a client of it rather than a second author.
"""
source = ValueSource(
kind=body.source_kind,
id=body.source_id,
label=body.source_label or body.source_id or "API",
detail=body.source_detail,
)
try:
await run_in_threadpool(controller.publish_message, name, body.value)
await run_in_threadpool(
controller.publish_message, name, body.value, source
)
except KeyError:
raise HTTPException(
status_code=404, detail=f"No flow declares a message named '{name}'"
backend/app/flow/controller.py
+37 -3
View File
@@ -43,7 +43,7 @@ from app.flow.nodes import (
SwitchNode,
TriggerNode,
)
from app.flow.pipeline import Pipeline, ValidationIssue
from app.flow.pipeline import Pipeline, ValidationIssue, ValueSource
from app.flow.schemas import (
FlowDef,
Health,
@@ -276,6 +276,9 @@ class FlowController:
# Without one, every trigger runs inline where it was raised.
self.execution = execution
self.alerts = alerts
# Set by the lifespan once the store exists; the canvas asks the
# controller which dashboards are wired into a flow.
self.dashboards: Any = None
self.pipeline: Pipeline | None = None
self.loaded: dict[str, LoadedNode] = {}
@@ -648,7 +651,38 @@ class FlowController:
)
return infos
def publish_message(self, name: str, value: Any) -> None:
def cross_flow_nodes(
self, flow: str
) -> list[tuple[str, str, list[str], list[str]]]:
"""Nodes in other flows on the far side of a dotted message name.
Returns ``(flow, node id, messages of `flow` it provides, ones it
requires)``. This is what link-in and link-out mean: the wiring is
already real — the canvas simply had no way to draw the other end.
"""
prefix = f"{flow}."
found = []
for other in self.store.read_all():
if other.name == 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))
return found
def publish_message(
self, name: str, value: Any, source: ValueSource | None = None
) -> None:
"""Put a value into the graph from outside. Blocking.
Refuses a name no flow declares: the flows own the namespace, and a
@@ -672,7 +706,7 @@ class FlowController:
# The same check a node's output gets; a dashboard is not looser.
spec.check(value)
self.pipeline.publish({name: value})
self.pipeline.publish({name: value}, source)
def queue_stats(self) -> dict[str, Any]:
return self.execution.stats() if self.execution is not None else {}
backend/app/flow/dashboards.py
+38
View File
@@ -28,6 +28,10 @@ DASHBOARD_DIR = "_dashboards"
#: A chart cannot ask for an unbounded series; this is the ceiling.
HISTORY_CAP = 5000
#: Resolved out here on purpose: the store has a ``list`` method, which
#: shadows the builtin for any annotation written inside the class.
Bindings = list[dict[str, Any]]
WidgetType = Literal[
# Display
"stat",
@@ -263,6 +267,40 @@ class DashboardStore:
self.flows._commit(f"Rename dashboard '{name}' to '{new_name}'")
return renamed
def bindings_for(self, flow: str) -> Bindings:
"""Every widget bound to a message of ``flow``.
What the canvas draws as an endpoint: a control that sets one of this
flow's messages, or a tile that shows one. Without this a dashboard is
an invisible participant — a value changes and nothing on the canvas
accounts for it.
"""
prefix = f"{flow}."
found: Bindings = []
for path in sorted(self.root.glob("*/dashboard.json")):
try:
defn = DashboardDef.model_validate_json(path.read_text())
except Exception:
continue
for widget in defn.widgets:
# A control produces the message; a tile consumes it.
produces = widget.target if widget.target.startswith(prefix) else ""
consumes = [m for m in widget.messages if m.startswith(prefix)]
if not produces and not consumes:
continue
found.append(
{
"dashboard": defn.name,
"dashboard_title": defn.title or defn.name,
"widget": widget.id,
"title": widget.title or widget.id,
"type": widget.type,
"provides": produces,
"requires": consumes,
}
)
return found
def history_requirements(self) -> dict[str, int]:
"""How many points to keep per message, so charts have a past to draw.
backend/app/flow/pipeline.py
+37 -1
View File
@@ -46,6 +46,29 @@ class ValidationIssue(BaseModel):
message_name: str | None = None
class ValueSource(BaseModel):
"""Who caused a message to take its current value.
The canvas draws an edge per producer, so without this it pulses every one
of them and claims a node published something it did not. It is also what
lets a value arriving from outside the flow — a dashboard control, another
flow, an agent — be shown at all, since none of those is a node here.
"""
#: node, dashboard, flow, agent or api.
kind: str = "node"
#: Node id, dashboard name, or whatever identifies the caller.
id: str = ""
#: What to call it on screen.
label: str = ""
#: The widget, for a dashboard.
detail: str = ""
def node_source(node: Node) -> ValueSource:
return ValueSource(kind="node", id=node.id, label=node.local_id)
class Pipeline:
"""Directed graph of nodes with automatic dependency resolution."""
@@ -495,6 +518,7 @@ class Pipeline:
# Append-only, so it needs no lock of its own.
state.append_history(result, ts, self.history_limits)
self._increment_message_versions(result)
origin = node_source(node)
for name, value in result.items():
self._publish(
{
@@ -503,6 +527,7 @@ class Pipeline:
"name": name,
"value": value,
"ts": ts,
"source": origin.model_dump(),
}
)
@@ -685,6 +710,7 @@ class Pipeline:
state.update({self._timestamp_key(name): ts for name in outputs})
state.append_history(outputs, ts, self.history_limits)
self._increment_message_versions(outputs)
origin = node_source(node)
for name, value in outputs.items():
self._publish(
{
@@ -693,6 +719,7 @@ class Pipeline:
"name": name,
"value": value,
"ts": ts,
"source": origin.model_dump(),
}
)
# An injecting node — an MQTT subscriber, a webhook — publishes
@@ -760,16 +787,24 @@ class Pipeline:
self.apply_outputs(node, outputs)
return self.run_downstream(node)
def publish(self, values: dict[str, Any]) -> None:
def publish(
self, values: dict[str, Any], source: ValueSource | None = None
) -> None:
"""Put values into the graph without a node having produced them.
This is what a dashboard control does: the value is real, it just came
from a person rather than a sensor. Everything consuming those names
runs, the same as if a node had published them.
``source`` says what did, so the canvas can show the value arriving
from outside instead of blaming whichever node happens to be drawn as
a producer of that message.
"""
if not values:
return
origin = source or ValueSource(kind="api", label="API")
ts = time.time()
with self._state.lock():
self._state.update(values)
@@ -784,6 +819,7 @@ class Pipeline:
"name": name,
"value": value,
"ts": ts,
"source": origin.model_dump(),
}
)
backend/app/main.py
+1
View File
@@ -86,6 +86,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
app.state.flow_controller = controller
dashboards = DashboardStore(controller.store)
app.state.dashboard_store = dashboards
controller.dashboards = dashboards
# Charts need a deeper series than the default; tell the engine
# before it starts recording.
controller.set_history_limits(dashboards.history_requirements())
backend/tests/flow/test_provenance.py
+154
View File
@@ -0,0 +1,154 @@
"""Who caused a value.
The canvas draws an edge per producer of a message. Without knowing which one
actually published, it pulses all of them — and when the cause is a dashboard
control or another flow, it pulses a node that did nothing at all.
"""
from app.flow.dashboards import (
DashboardDef,
DashboardStore,
PageDef,
SectionDef,
WidgetDef,
)
from app.flow.events import EventBus
from app.flow.messages import DType, MessageSpec
from app.flow.nodes import Node
from app.flow.pipeline import Pipeline, ValueSource
from app.flow.state import MemoryState
from app.flow.store import FlowStore
def collect(bus: EventBus) -> list[dict]:
events: list[dict] = []
bus.publish = events.append # type: ignore[method-assign]
return events
def temp_node() -> Node:
node = Node(
f=lambda params: {"temp": 21.0},
provides=[MessageSpec(name="temp", port="temp", dtype=DType.FLOAT)],
name="sensor",
)
node.assign_flow("house", "sensor")
return node
def test_a_value_a_node_produced_names_that_node():
bus = EventBus()
events = collect(bus)
node = temp_node()
pipeline = Pipeline(nodes=[node], state=MemoryState(), events=bus)
pipeline.run()
published = [e for e in events if e["type"] == "message_value"]
assert published[0]["source"] == {
"kind": "node",
"id": "house.sensor",
"label": "sensor",
"detail": "",
}
def test_a_value_a_node_injected_names_that_node():
"""An MQTT message or a webhook arrives this way rather than by executing."""
bus = EventBus()
events = collect(bus)
node = temp_node()
pipeline = Pipeline(nodes=[node], state=MemoryState(), events=bus)
pipeline.apply_outputs(node, {"house.temp": 19.0})
published = [e for e in events if e["type"] == "message_value"]
assert published[0]["source"]["id"] == "house.sensor"
def test_a_value_from_a_dashboard_says_so_rather_than_blaming_a_node():
"""The bug this exists for: a slider must not light up a node's edge."""
bus = EventBus()
events = collect(bus)
pipeline = Pipeline(nodes=[temp_node()], state=MemoryState(), events=bus)
pipeline.publish(
{"house.temp": 25.0},
ValueSource(
kind="dashboard", id="panel", label="Setpoint", detail="slider"
),
)
published = [e for e in events if e["type"] == "message_value"]
assert published[0]["source"]["kind"] == "dashboard"
assert published[0]["source"]["label"] == "Setpoint"
def test_a_value_from_nowhere_in_particular_is_still_attributed():
bus = EventBus()
events = collect(bus)
pipeline = Pipeline(nodes=[temp_node()], state=MemoryState(), events=bus)
pipeline.publish({"house.temp": 25.0})
published = [e for e in events if e["type"] == "message_value"]
assert published[0]["source"]["kind"] == "api"
# ---------------------------------------------------------------------------
# What the canvas draws for it
# ---------------------------------------------------------------------------
def test_the_widgets_wired_into_a_flow_are_reported(tmp_path):
store = DashboardStore(FlowStore(tmp_path / "flows"))
store.write(
DashboardDef(
name="panel",
title="Panel",
pages=[
PageDef(
id="main",
sections=[
SectionDef(
id="main",
widgets=[
WidgetDef(
id="setpoint",
type="slider",
title="Setpoint",
config={"target": "house.setpoint"},
),
WidgetDef(
id="reading",
type="stat",
title="Reading",
config={"message": "house.temp"},
),
# Another flow's message: not this flow's business.
WidgetDef(
id="elsewhere",
type="stat",
config={"message": "garage.temp"},
),
],
)
],
)
],
)
)
bindings = store.bindings_for("house")
assert [b["widget"] for b in bindings] == ["setpoint", "reading"]
setpoint = bindings[0]
assert setpoint["provides"] == "house.setpoint"
assert setpoint["requires"] == []
assert bindings[1]["requires"] == ["house.temp"]
def test_a_flow_nothing_points_at_has_no_endpoints(tmp_path):
store = DashboardStore(FlowStore(tmp_path / "flows"))
assert store.bindings_for("house") == []