flow: structured dtypes, and the widgets that read them

A series, record or list message declares its shape instead of riding
DType.JSON, so a widget binds a shape rather than some JSON and a wrong
binding is refused before anything runs. A list declares its item type,
which is what keeps list[float] expressible for a pipeline.

On top of that: an agenda over a list, a notification over a record, and
a dashboard alert channel that publishes engine faults as one — so a
panel can show what went wrong without a flow wiring it by hand.

Also: only None means a node published nothing, a falsy value of the
wrong shape is now the named error it always should have been; and the
gauge's readout says its size is viewBox geometry rather than type scale.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/app/flow/alerts.py

@@ -16,6 +16,7 @@ import asyncio
 import logging
 import time
 from collections import deque
+from collections.abc import Callable
 from typing import Any, Literal
 
 import httpx
@@ -51,9 +52,10 @@ class Channel(BaseModel):
     """Somewhere to send an alert."""
 
     name: str
-    kind: Literal["ntfy", "smtp", "webhook"]
+    kind: Literal["ntfy", "smtp", "webhook", "dashboard"]
     enabled: bool = True
     # ntfy: server + topic + optional token. smtp: to. webhook: url.
+    # dashboard: message — the record a notification widget reads.
     config: dict[str, Any] = Field(default_factory=dict)
 
 
@@ -160,6 +162,10 @@ class AlertManager:
         self._events = events
         self.config = config or AlertsConfig()
         self._now = now
+        #: How a "dashboard" channel puts its alert into the graph. Bound after
+        #: construction, because the controller that publishes does not exist
+        #: yet when the manager is built.
+        self.publish: Callable[[str, Any], None] | None = None
         self._last_sent: dict[str, float] = {}
         self._suppressed: dict[str, int] = {}
         self._health_flips: dict[str, deque[float]] = {}
@@ -294,6 +300,8 @@ class AlertManager:
                 await self._send_ntfy(config, alert)
             elif channel.kind == "webhook":
                 await self._send_webhook(config, alert)
+            elif channel.kind == "dashboard":
+                await self._send_dashboard(config, alert)
             else:
                 await self._send_email(config, alert)
         except Exception as exc:
@@ -319,6 +327,21 @@ class AlertManager:
             )
             response.raise_for_status()
 
+    async def _send_dashboard(self, config: dict[str, Any], alert: Alert) -> None:
+        """Put the alert into the graph, where a notification widget shows it.
+
+        The message has to be one a flow declares, like any other a dashboard
+        writes to — so a panel that shows engine faults says so in a flow
+        rather than appearing from nowhere.
+        """
+        message = str(config.get("message") or "")
+        if not message:
+            raise ValueError("a dashboard channel needs a message name")
+        if self.publish is None:
+            raise RuntimeError("nothing is wired up to publish this")
+        # Blocking: it reads the store to find the declared message.
+        await asyncio.to_thread(self.publish, message, alert.model_dump())
+
     async def _send_webhook(self, config: dict[str, Any], alert: Alert) -> None:
         url = config.get("url")
         if not url:

backend/app/flow/dashboards.py

@@ -40,6 +40,8 @@ WidgetType = Literal[
     "gauge",
     "chart",
     "markdown",
+    "agenda",
+    "notification",
     # Input
     "button",
     "switch",
@@ -57,9 +59,13 @@ INPUT_WIDGETS = {"button", "switch", "slider", "input", "dropdown"}
 #: (``frontend/src/components/Dashboard/widgets.tsx``).
 WIDGET_DTYPES: dict[str, set[str]] = {
     "gauge": {"float", "int"},
+    # A chart reading the engine's ring. One that queries binds a `series`
+    # answer and a `record` request instead, checked separately below.
     "chart": {"float", "int"},
     "slider": {"float", "int"},
     "switch": {"bool"},
+    "agenda": {"list"},
+    "notification": {"record"},
 }
 
 
@@ -78,6 +84,13 @@ class WidgetDef(BaseModel):
     ``config`` is per type — a chart names its series, a button names the
     message it publishes — and is validated against the type below rather than
     by a schema per class, because the whole set is small and closed.
+
+    A chart comes in two kinds. The default reads what the engine kept for a
+    message. One with ``source: "query"`` asks instead, and its config is
+    ``{source, request, request_dtype: "record", message, dtype: "series",
+    refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to
+    ``request`` exactly as a slider publishes a value, and draws the ``series``
+    a flow answers with on ``message``.
     """
 
     id: str
@@ -93,9 +106,17 @@ class WidgetDef(BaseModel):
     def _check_id(cls, value: str) -> str:
         return _validate_name(value)
 
+    @property
+    def _query_chart(self) -> bool:
+        """A chart that asks a flow for its series instead of reading the ring."""
+        return self.type == "chart" and self.config.get("source") == "query"
+
     @property
     def messages(self) -> list[str]:
         """Every message name this widget reads."""
+        if self._query_chart:
+            name = self.config.get("message")
+            return [str(name)] if name else []
         if self.type == "chart":
             return [
                 str(series.get("message"))
@@ -107,13 +128,23 @@ class WidgetDef(BaseModel):
 
     @property
     def target(self) -> str:
-        """The message this widget publishes, if it is an input."""
+        """The message this widget publishes, if it is an input.
+
+        A querying chart is one too: its request is a value it puts into the
+        graph, so the canvas draws it as an endpoint like any other control.
+        """
+        if self._query_chart:
+            return str(self.config.get("request") or "")
         return str(self.config.get("target") or "")
 
     @property
     def history_points(self) -> int:
-        """How much past this widget needs kept for it."""
-        if self.type != "chart":
+        """How much past this widget needs kept for it.
+
+        Nothing, for a chart that queries: the answer carries its own past, so
+        asking the engine to keep a ring as well would store it twice.
+        """
+        if self.type != "chart" or self._query_chart:
             return 0
         points = int((self.config.get("history") or {}).get("points") or 0)
         return min(points, HISTORY_CAP)
@@ -135,6 +166,15 @@ class WidgetDef(BaseModel):
     @model_validator(mode="after")
     def _check_binding(self) -> WidgetDef:
         """Refuse a widget wired to a message it cannot carry."""
+        if self._query_chart:
+            for key, want in (("dtype", "series"), ("request_dtype", "record")):
+                bound = str(self.config.get(key) or "")
+                if bound and bound != want:
+                    raise ValueError(
+                        f"a querying chart's {key} must be '{want}', not '{bound}'"
+                    )
+            return self
+
         allowed = WIDGET_DTYPES.get(self.type)
         if not allowed:
             return self

backend/app/flow/messages.py

@@ -20,6 +20,11 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
 class DType(str, Enum):
     """Serializable payload types.
 
+    The scalars carry what a single reading can say. The three structured ones
+    are declared shapes rather than "some JSON": a widget or a downstream node
+    knows what it is getting before anything runs, which is what lets the
+    dashboard picker offer a message and refuse a wrong binding.
+
     Binary payloads (tensors, images) will arrive later as explicitly declared
     codec fields; until then everything on the wire is JSON.
     """
@@ -29,10 +34,78 @@ class DType(str, Enum):
     STR = "str"
     BOOL = "bool"
     JSON = "json"
+    #: ``{"lines": [{"label": str, "points": [[ts, value], ...]}], ...}``.
+    #: Keys beside ``lines`` are carried through untouched — a chart's query
+    #: puts the range and interval it asked for there and reads them back.
+    SERIES = "series"
+    #: Flat named scalars: ``{"title": "Boiler", "severity": "error"}``.
+    RECORD = "record"
+    #: Ordered items of one declared shape; see :attr:`MessageSpec.item`.
+    LIST = "list"
 
 
 _JSON_TYPES = (dict, list, str, int, float, bool, type(None))
 
+#: What a record may hold. Nesting is deliberately out: a record that can
+#: contain a record is a schema language, and the shape stops being readable
+#: from the declaration alone.
+_SCALARS = (str, int, float, bool)
+
+#: Item types a list may declare. Recursion is refused for the same reason.
+_ITEM_TYPES = frozenset(
+    {DType.FLOAT, DType.INT, DType.STR, DType.BOOL, DType.JSON, DType.RECORD}
+)
+
+
+def _is_number(value: Any) -> bool:
+    """A measurement. bool is an int subclass; a flag is not a number here."""
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _is_record(value: Any) -> bool:
+    return isinstance(value, dict) and all(
+        isinstance(key, str) and (item is None or isinstance(item, _SCALARS))
+        for key, item in value.items()
+    )
+
+
+def _is_series(value: Any) -> bool:
+    """Labelled lines of ``(timestamp, value)`` pairs.
+
+    Checked all the way down. That is O(n) in the number of points, but so is
+    the JSON encoding every message already pays for.
+    """
+    if not isinstance(value, dict) or not isinstance(value.get("lines"), list):
+        return False
+    return all(
+        isinstance(line, dict)
+        and isinstance(line.get("label"), str)
+        and isinstance(line.get("points"), list)
+        and all(
+            isinstance(point, (list, tuple))
+            and len(point) == 2
+            and _is_number(point[0])
+            and _is_number(point[1])
+            for point in line["points"]
+        )
+        for line in value["lines"]
+    )
+
+
+def _matches(dtype: DType, value: Any) -> bool:
+    """Whether one value satisfies a scalar or record type."""
+    if dtype is DType.BOOL:
+        return isinstance(value, bool)
+    if dtype is DType.INT:
+        return isinstance(value, int) and not isinstance(value, bool)
+    if dtype is DType.FLOAT:
+        return _is_number(value)
+    if dtype is DType.STR:
+        return isinstance(value, str)
+    if dtype is DType.RECORD:
+        return _is_record(value)
+    return isinstance(value, _JSON_TYPES)
+
 
 class MessageSpec(BaseModel):
     """A single port of a node, and the message it is bound to.
@@ -42,6 +115,10 @@ class MessageSpec(BaseModel):
     :param port: The identifier the node function sees. Defaults to the last
         segment of ``name``, so unqualified flows read naturally.
     :param dtype: Payload type, validated on every message that passes through.
+    :param item: The type of each item of a ``list`` port, ignored otherwise.
+        Unset means ``record``, which is what the agenda and forecast widgets
+        read; ``float`` is the numeric list a pipeline passes around. A list of
+        lists, or of series, is refused — one declared level is the point.
     :param interval: Deliver at most every this many seconds; 0 is every time.
         On an output it holds back publishing, on an input it holds back waking
         the node. The value is never lost — state keeps the latest — only the
@@ -57,6 +134,7 @@ class MessageSpec(BaseModel):
     name: str = ""
     port: str = ""
     dtype: DType = DType.FLOAT
+    item: DType | None = None
     interval: float = Field(default=0, ge=0)
     trigger: bool = True
 
@@ -64,25 +142,36 @@ class MessageSpec(BaseModel):
     def _default_port(self) -> MessageSpec:
         if not self.port and self.name:
             object.__setattr__(self, "port", self.name.rsplit(".", 1)[-1])
+        if self.item is not None and self.item not in _ITEM_TYPES:
+            raise ValueError(f"a list cannot hold '{self.item.value}' items")
         return self
 
+    @property
+    def item_dtype(self) -> DType:
+        """What each item of a ``list`` port is, declared or defaulted."""
+        return self.item or DType.RECORD
+
     def check(self, value: Any) -> None:
         """Raise if ``value`` does not match this port's declared type."""
-        if self.dtype is DType.BOOL:
-            ok = isinstance(value, bool)
-        elif self.dtype is DType.INT:
-            # bool is an int subclass; a flag is not a number here.
-            ok = isinstance(value, int) and not isinstance(value, bool)
-        elif self.dtype is DType.FLOAT:
-            ok = isinstance(value, (int, float)) and not isinstance(value, bool)
-        elif self.dtype is DType.STR:
-            ok = isinstance(value, str)
+        where = self.name or self.port
+        if self.dtype is DType.SERIES:
+            ok = _is_series(value)
+        elif self.dtype is DType.LIST:
+            if not isinstance(value, list):
+                raise TypeError(f"{where}: expected list, got {type(value).__name__}")
+            item = self.item_dtype
+            for index, element in enumerate(value):
+                if not _matches(item, element):
+                    raise TypeError(
+                        f"{where}: expected list[{item.value}], got "
+                        f"{type(element).__name__} at index {index}"
+                    )
+            return
         else:
-            ok = isinstance(value, _JSON_TYPES)
+            ok = _matches(self.dtype, value)
         if not ok:
             raise TypeError(
-                f"{self.name or self.port}: expected {self.dtype.value}, "
-                f"got {type(value).__name__}"
+                f"{where}: expected {self.dtype.value}, got {type(value).__name__}"
             )
 
     def coerce(self, value: Any) -> Any:
@@ -101,6 +190,10 @@ class MessageSpec(BaseModel):
             return str(value).lower() in ("true", "1", "yes", "on")
         if self.dtype is DType.STR:
             return value if isinstance(value, str) else json.dumps(value)
+        if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST):
+            # A structured payload arriving as text is the same hint a numeric
+            # one is; the shape itself is still checked afterwards.
+            return json.loads(value) if isinstance(value, str) else value
         return value
 
     def __repr__(self) -> str:

backend/app/flow/nodes/base.py

@@ -271,13 +271,19 @@ class Node:
         return kwargs
 
     def _to_messages(self, retval: Any) -> dict[str, Any] | None:
-        """Map a function's port-keyed return value onto message names."""
-        if not retval:
+        """Map a function's port-keyed return value onto message names.
+
+        Only ``None`` means "nothing to publish". A falsy value of the wrong
+        shape — ``0``, ``""``, an empty list — is a mistake worth naming rather
+        than silence someone has to debug from an empty canvas.
+        """
+        if retval is None:
             return None
         if not isinstance(retval, dict):
             raise NodeOutputError(
                 f"'{self.local_id}' returned {type(retval).__name__}. Outputs are "
-                "keyed by port, so return a dict like {'out': value}, or None."
+                "keyed by port, so return a dict like {'out': value}, or None to "
+                "publish nothing."
             )
         by_port = {s.port: s for s in self.output_ports if s.name}
         outputs = {}

backend/app/flow/nodes/influx.py

@@ -29,6 +29,22 @@ class InfluxDbNode(Node):
     Both operations use a similar configuration pattern in params, making the
     API consistent and the input/output data simple (just values).
 
+    - **Query passthrough**: an incoming message holding a ``flux`` key is run
+      as written, and every row comes back on the first output port as
+      ``{"rows": [{ts, value, field, measurement, tags}], **echo}``.
+
+    The passthrough is what keeps a database node a database node. It holds the
+    credentials and the connection and nothing else: building a query and
+    shaping its rows are ordinary Python nodes on either side, so a dashboard
+    widget never learns which database answered it. A chart drawing an
+    InfluxDB series is::
+
+        [chart] -record-> [py: build] -record{flux, ...}-> [influx]
+        [chart] <-series- [py: parse] <-json{rows, ...}---'
+
+    Nothing rate-limits the requests here; an ``interval`` on the request port
+    is what holds back a caller asking too often.
+
     :param requires: Messages to write to InfluxDB. The actual value from each
         message is written according to the corresponding config in ``writes``.
     :type requires: MessageSpec | list[MessageSpec] | None
@@ -208,10 +224,13 @@ class InfluxDbNode(Node):
 
     def _handler(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
         """
-        Handle incoming data - write to InfluxDB and optionally query.
+        Handle incoming data - write to InfluxDB, run a query, or both.
 
         This method is called when upstream dependencies (requires) are satisfied.
-        It writes the incoming data to InfluxDB and can also perform reads.
+
+        An incoming dict carrying a ``flux`` key is a query request rather than
+        something to store: it is executed as written and answered on the first
+        output port. Everything else on the inputs is written as points.
 
         :param params: Node parameters.
         :type params: dict
@@ -220,9 +239,24 @@ class InfluxDbNode(Node):
         :returns: Query results if provides is configured, None otherwise.
         :rtype: dict | None
         """
-        # Write incoming data
-        if kwargs:
-            self._write_points(kwargs)
+        # ponytail: one request/answer pair per node — the first input holding
+        # a "flux" key is the request and the answer goes out on the first
+        # output port. A second query stream means a second node.
+        request = next(
+            (v for v in kwargs.values() if isinstance(v, dict) and "flux" in v), None
+        )
+        writes = {k: v for k, v in kwargs.items() if v is not request}
+
+        if writes:
+            self._write_points(writes)
+
+        if request is not None:
+            if not self.output_ports:
+                raise ValueError(
+                    f"Node '{self.name}' was sent a query but has no output to "
+                    "answer on"
+                )
+            return {self.output_ports[0].port: self._run_flux(request)}
 
         # If we have provides, perform queries
         if self.provides:
@@ -230,6 +264,49 @@ class InfluxDbNode(Node):
 
         return None
 
+    def _run_flux(self, request: dict[str, Any]) -> dict[str, Any]:
+        """
+        Execute a Flux query exactly as it was handed over, and answer its rows.
+
+        The node stays out of the query's business: building it is the job of
+        whatever produced the request, and shaping the rows into something a
+        widget draws is the job of whatever reads the answer. Every field of
+        the request except ``flux`` is echoed back untouched, which is how a
+        caller tells its own answer from someone else's.
+
+        :param request: The query and whatever the caller wants echoed.
+        :type request: dict
+        :returns: ``{"rows": [...], **echo}``.
+        :rtype: dict
+        """
+        from influxdb_client import InfluxDBClient
+
+        flux = str(request["flux"])
+        echo = {key: value for key, value in request.items() if key != "flux"}
+
+        logger.info("Running Flux for node '%s': %s", self.name, flux)
+        with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
+            tables = client.query_api().query(flux, org=self.org)
+
+        rows = [
+            {
+                "ts": record.get_time().timestamp() if record.get_time() else None,
+                "value": record.get_value(),
+                # Read off `values` rather than the getters: a query that keeps
+                # only what it needs drops these columns, and the getters raise.
+                "field": record.values.get("_field"),
+                "measurement": record.values.get("_measurement"),
+                "tags": {
+                    key: value
+                    for key, value in record.values.items()
+                    if not key.startswith("_") and key not in ("result", "table")
+                },
+            }
+            for table in tables
+            for record in table.records
+        ]
+        return {"rows": rows, **echo}
+
     def _write_points(self, data: dict[str, Any]) -> None:
         """
         Write data points to InfluxDB using configuration from ``writes``.

backend/app/main.py

@@ -22,6 +22,7 @@ from app.flow.events import event_bus
 from app.flow.executor import ExecutionService
 from app.flow.metrics import MetricsCollector
 from app.flow.nodes.http import close_shared_client
+from app.flow.pipeline import ValueSource
 from app.flow.plugins import load_plugins
 from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
 from app.flow.secrets import init_secrets
@@ -98,6 +99,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         workers=pool,
     )
     app.state.flow_controller = controller
+    # A "dashboard" alert channel puts its alert into the graph. Bound here
+    # rather than passed in: the manager is built before the controller is.
+    alerts.publish = lambda name, value: controller.publish_message(
+        name, value, ValueSource(kind="api", id="alerts", label="Alerts")
+    )
     dashboards = DashboardStore(controller.store)
     app.state.dashboard_store = dashboards
     controller.dashboards = dashboards