Dashboards as documents, and messages as something to bind to

A dashboard is its own document rather than widgets placed in a flow.
Node-RED's dashboard tab is 260 nodes, about forty of them pure layout,
which is exactly what the small-graph principle exists to avoid — and
since the graph is already wired by message name, a widget can bind to a
name without belonging to any flow.

Stored beside the flows in the same repository, sharing their write lock
and commit, under a directory the flow listing ignores. No draft/publish
split: nothing executes a dashboard, so edit mode is its own staging area.

Two things it needs from the engine. A message catalog spanning every
flow, because a wall panel shows the heating next to the solar and the
flow-scoped API is the wrong shape for that. And a way to put a value in
without owning a node — a slider is a real value that happened to come
from a person — which runs whatever consumes it and applies the same type
check a node's output gets. Only a message some flow declares can be
published to; flows own the namespace.

Charts also need more past than the 120 points a sparkline wanted, so a
chart widget declares its depth and the engine keeps that message's
series that deep.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY

backend/app/api/deps.py

@@ -12,6 +12,7 @@ from app.core import security
 from app.core.config import settings
 from app.core.db import engine
 from app.flow.controller import FlowController
+from app.flow.dashboards import DashboardStore
 from app.models import TokenPayload, User
 
 reusable_oauth2 = OAuth2PasswordBearer(
@@ -102,6 +103,16 @@ def get_flow_controller(request: Request) -> FlowController:
 FlowControllerDep = Annotated[FlowController, Depends(get_flow_controller)]
 
 
+def get_dashboard_store(request: Request) -> DashboardStore:
+    store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="The flow engine is not running")
+    return store
+
+
+DashboardStoreDep = Annotated[DashboardStore, Depends(get_dashboard_store)]
+
+
 def get_current_active_superuser(current_user: CurrentUser) -> User:
     if not current_user.is_superuser:
         raise HTTPException(

backend/app/api/main.py

@@ -2,8 +2,10 @@ from fastapi import APIRouter
 
 from app.api.routes import (
     alerts,
+    dashboards,
     flows,
     login,
+    messages,
     oauth,
     private,
     secrets,
@@ -19,6 +21,8 @@ api_router.include_router(flows.router)
 api_router.include_router(flows.ws_router)
 api_router.include_router(secrets.router)
 api_router.include_router(alerts.router)
+api_router.include_router(dashboards.router)
+api_router.include_router(messages.router)
 # Always mounted so the generated SDK stays the same shape; the endpoints
 # themselves refuse to work unless MCP is switched on.
 api_router.include_router(oauth.router)

backend/app/api/routes/dashboards.py

@@ -0,0 +1,112 @@
+"""Dashboards: documents of widgets bound to message names."""
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.concurrency import run_in_threadpool
+from pydantic import BaseModel
+
+from app.api.deps import DashboardStoreDep, FlowControllerDep, get_current_user
+from app.flow.dashboards import (
+    DashboardDef,
+    DashboardExists,
+    DashboardNotFound,
+    DashboardsPublic,
+    default_dashboard,
+)
+from app.flow.store import StaleVersion
+from app.models import Message
+
+router = APIRouter(
+    prefix="/dashboards", tags=["dashboards"], dependencies=[Depends(get_current_user)]
+)
+
+
+class RenameRequest(BaseModel):
+    name: str
+
+
+def _apply_history_limits(store: Any, controller: Any) -> None:
+    """Tell the engine how much past each charted message needs kept."""
+    controller.set_history_limits(store.history_requirements())
+
+
+@router.get("/", response_model=DashboardsPublic)
+async def read_dashboards(store: DashboardStoreDep) -> Any:
+    """Every dashboard, without its contents."""
+    summaries = await run_in_threadpool(store.list)
+    return DashboardsPublic(data=summaries, count=len(summaries))
+
+
+@router.get("/{name}", response_model=DashboardDef)
+async def read_dashboard(name: str, store: DashboardStoreDep) -> Any:
+    try:
+        return await run_in_threadpool(store.read, name)
+    except DashboardNotFound:
+        raise HTTPException(status_code=404, detail=f"No dashboard named '{name}'")
+
+
+@router.post("/{name}", response_model=DashboardDef)
+async def create_dashboard(
+    name: str, store: DashboardStoreDep, controller: FlowControllerDep
+) -> Any:
+    """Start a dashboard: one page, one section, nothing on it yet."""
+    if await run_in_threadpool(store.exists, name):
+        raise HTTPException(status_code=409, detail=f"'{name}' already exists")
+    try:
+        defn = default_dashboard(name)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+    saved = await run_in_threadpool(store.write, defn, 0)
+    await run_in_threadpool(_apply_history_limits, store, controller)
+    return saved
+
+
+@router.put("/{name}", response_model=DashboardDef)
+async def save_dashboard(
+    name: str,
+    body: DashboardDef,
+    store: DashboardStoreDep,
+    controller: FlowControllerDep,
+) -> Any:
+    """Replace a dashboard, refusing a save someone else has moved past."""
+    if body.name != name:
+        raise HTTPException(status_code=422, detail="The name in the body must match")
+    try:
+        saved = await run_in_threadpool(store.write, body, body.version)
+    except StaleVersion as exc:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "Someone else saved this dashboard first",
+                "current_version": exc.current,
+            },
+        )
+    await run_in_threadpool(_apply_history_limits, store, controller)
+    return saved
+
+
+@router.delete("/{name}", response_model=Message)
+async def delete_dashboard(
+    name: str, store: DashboardStoreDep, controller: FlowControllerDep
+) -> Any:
+    try:
+        await run_in_threadpool(store.delete, name)
+    except DashboardNotFound:
+        raise HTTPException(status_code=404, detail=f"No dashboard named '{name}'")
+    await run_in_threadpool(_apply_history_limits, store, controller)
+    return Message(message=f"Deleted dashboard '{name}'")
+
+
+@router.post("/{name}/rename", response_model=DashboardDef)
+async def rename_dashboard(
+    name: str, body: RenameRequest, store: DashboardStoreDep
+) -> Any:
+    try:
+        return await run_in_threadpool(store.rename, name, body.name)
+    except DashboardNotFound:
+        raise HTTPException(status_code=404, detail=f"No dashboard named '{name}'")
+    except DashboardExists:
+        raise HTTPException(status_code=409, detail=f"'{body.name}' already exists")
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))

backend/app/api/routes/messages.py

@@ -0,0 +1,101 @@
+"""Messages: what a dashboard binds to, across every flow.
+
+The flow API is scoped to one flow, which is the wrong shape here — a wall
+panel shows the heating alongside the solar. These endpoints are the whole
+namespace at once: what exists, what it last was, and a way to put a value
+into it without owning a node.
+"""
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.concurrency import run_in_threadpool
+from pydantic import BaseModel
+
+from app.api.deps import FlowControllerDep, get_current_user
+from app.flow.messages import flow_of
+from app.flow.state import as_number
+
+router = APIRouter(
+    prefix="/messages", tags=["messages"], dependencies=[Depends(get_current_user)]
+)
+
+
+class MessageInfo(BaseModel):
+    """One message, as something to bind a widget to."""
+
+    name: str
+    flow: str
+    dtype: str
+    #: Nodes publishing it. Empty means the flow declares it as an input.
+    providers: list[str] = []
+    #: Whether a dashboard may publish to it — that is, whether it is declared.
+    writable: bool = True
+    numeric: bool = False
+    value: Any = None
+    ts: float | None = None
+
+
+class MessagesPublic(BaseModel):
+    data: list[MessageInfo]
+    count: int
+
+
+class PublishRequest(BaseModel):
+    value: Any
+
+
+class MessageValue(BaseModel):
+    name: str
+    value: Any
+    ts: float | None = None
+
+
+class MessagePoints(BaseModel):
+    message: str
+    numeric: bool
+    points: list[dict[str, float]]
+
+
+@router.get("/", response_model=MessagesPublic)
+async def read_messages(controller: FlowControllerDep) -> Any:
+    """Every message any published flow declares, with its last value."""
+    infos = await run_in_threadpool(controller.message_catalog)
+    return MessagesPublic(data=infos, count=len(infos))
+
+
+@router.post("/{name}", response_model=MessageValue)
+async def publish_message(
+    name: str, body: PublishRequest, controller: FlowControllerDep
+) -> Any:
+    """Put a value into the graph, as a dashboard control does.
+
+    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.
+    """
+    try:
+        await run_in_threadpool(controller.publish_message, name, body.value)
+    except KeyError:
+        raise HTTPException(
+            status_code=404, detail=f"No flow declares a message named '{name}'"
+        )
+    except TypeError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+
+    values = controller.values(flow_of(name))
+    current = values.get(name, {})
+    return MessageValue(name=name, value=current.get("value"), ts=current.get("ts"))
+
+
+@router.get("/{name}/history", response_model=MessagePoints)
+def read_message_history(name: str, controller: FlowControllerDep) -> Any:
+    """The series behind a chart. Numbers only — nothing else plots."""
+    if controller.pipeline is None:
+        return MessagePoints(message=name, numeric=False, points=[])
+    series = controller.state.history(name)
+    numeric = as_number(controller.state.get(name)) is not None
+    return MessagePoints(
+        message=name,
+        numeric=numeric or bool(series),
+        points=[{"ts": ts, "value": value} for ts, value in series],
+    )

backend/app/flow/controller.py

@@ -25,7 +25,7 @@ from fastapi.concurrency import run_in_threadpool
 from app.flow.alerts import AlertManager
 from app.flow.events import EventBus
 from app.flow.executor import ExecutionService
-from app.flow.messages import MessageSpec, qualify
+from app.flow.messages import MessageSpec, flow_of, qualify
 from app.flow.nodes import (
     ChangeNode,
     DelayNode,
@@ -282,6 +282,7 @@ class FlowController:
         self.issues: list[ValidationIssue] = []
         self.disabled: set[str] = set()
         self.supervisor = Supervisor(events)
+        self.history_limits: dict[str, int] = {}
         self._lock = asyncio.Lock()
 
     # -------------------------------------------------------------------------
@@ -339,6 +340,7 @@ class FlowController:
                 work_queue=self.execution.queue if self.execution else None,
                 node_pool=self.execution.node_pool if self.execution else None,
             )
+            self.pipeline.history_limits = self.history_limits
             if self.execution is not None:
                 self.execution.bind(self.pipeline)
             self.issues = _collect_issues(loaded, self.pipeline, flow_inputs)
@@ -598,6 +600,80 @@ class FlowController:
             for item in self.execution.queue.unpark(flow):
                 self.execution.queue.add(item)
 
+    def set_history_limits(self, limits: dict[str, int]) -> None:
+        """How deep to keep each charted message's series. Applies at once."""
+        self.history_limits = limits
+        if self.pipeline is not None:
+            self.pipeline.history_limits = limits
+
+    def message_catalog(self) -> list[Any]:
+        """Every message the published flows declare, with its last value.
+
+        What a dashboard picks from, so it spans flows rather than sitting
+        inside one.
+        """
+        from app.api.routes.messages import MessageInfo
+
+        specs: dict[str, MessageSpec] = {}
+        providers: dict[str, list[str]] = {}
+        for flow in self.store.read_all():
+            for node in flow.nodes:
+                for spec in _bound(node.provides):
+                    name = qualify(flow.name, spec.name)
+                    specs.setdefault(name, spec)
+                    providers.setdefault(name, []).append(f"{flow.name}.{node.id}")
+                for spec in _bound(node.requires):
+                    specs.setdefault(qualify(flow.name, spec.name), spec)
+            for declared in flow.inputs:
+                if declared.spec.name:
+                    specs.setdefault(
+                        qualify(flow.name, declared.spec.name), declared.spec
+                    )
+
+        values = self.values()
+        infos = []
+        for name, spec in sorted(specs.items()):
+            current = values.get(name, {})
+            infos.append(
+                MessageInfo(
+                    name=name,
+                    flow=flow_of(name),
+                    dtype=spec.dtype.value,
+                    providers=sorted(providers.get(name, [])),
+                    writable=True,
+                    numeric=spec.dtype.value in ("float", "int"),
+                    value=current.get("value"),
+                    ts=current.get("ts"),
+                )
+            )
+        return infos
+
+    def publish_message(self, name: str, value: Any) -> None:
+        """Put a value into the graph from outside. Blocking.
+
+        Refuses a name no flow declares: the flows own the namespace, and a
+        message nothing reads or writes would just be a key nobody sees.
+        """
+        if self.pipeline is None:
+            raise KeyError(name)
+
+        spec = None
+        for flow in self.store.read_all():
+            for node in flow.nodes:
+                for candidate in [*_bound(node.provides), *_bound(node.requires)]:
+                    if qualify(flow.name, candidate.name) == name:
+                        spec = candidate
+                        break
+            for declared in flow.inputs:
+                if qualify(flow.name, declared.spec.name) == name:
+                    spec = declared.spec
+        if spec is None:
+            raise KeyError(name)
+
+        # The same check a node's output gets; a dashboard is not looser.
+        spec.check(value)
+        self.pipeline.publish({name: value})
+
     def queue_stats(self) -> dict[str, Any]:
         return self.execution.stats() if self.execution is not None else {}
 

backend/app/flow/dashboards.py

@@ -0,0 +1,311 @@
+"""Dashboards: what a wall panel shows, and what its buttons do.
+
+A dashboard is its own document, not a set of nodes placed in a flow. Widgets
+bind to message names — the same names that wire the graph — so a dashboard
+reads across flows without being part of any of them, and a flow stays the
+logic it was.
+
+Stored beside the flows in the same git repository, under a directory the flow
+listing ignores. No draft/publish split: nothing executes a dashboard, so edit
+mode is its own staging area and the version counter is enough to stop two
+clients overwriting each other.
+"""
+
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+from app.flow.schemas import _validate_name
+from app.flow.store import FlowStore, StaleVersion
+
+#: Sibling of the shared-node library, and likewise not a flow.
+DASHBOARD_DIR = "_dashboards"
+
+#: A chart cannot ask for an unbounded series; this is the ceiling.
+HISTORY_CAP = 5000
+
+WidgetType = Literal[
+    # Display
+    "stat",
+    "gauge",
+    "chart",
+    "markdown",
+    # Input
+    "button",
+    "switch",
+    "slider",
+    "input",
+    "dropdown",
+]
+
+INPUT_WIDGETS = {"button", "switch", "slider", "input", "dropdown"}
+
+
+class Placement(BaseModel):
+    """Where a widget sits in its section's grid, in grid units."""
+
+    x: int = 0
+    y: int = 0
+    w: int = 3
+    h: int = 2
+
+
+class WidgetDef(BaseModel):
+    """One tile: what it shows or does, and where it sits.
+
+    ``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.
+    """
+
+    id: str
+    type: WidgetType
+    title: str = ""
+    #: Keyed by breakpoint (``lg``/``md``/``sm``); missing ones are derived by
+    #: the client from the widest one it has.
+    layout: dict[str, Placement] = Field(default_factory=dict)
+    config: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("id")
+    @classmethod
+    def _check_id(cls, value: str) -> str:
+        return _validate_name(value)
+
+    @property
+    def messages(self) -> list[str]:
+        """Every message name this widget reads."""
+        if self.type == "chart":
+            return [
+                str(series.get("message"))
+                for series in self.config.get("series") or []
+                if series.get("message")
+            ]
+        name = self.config.get("message")
+        return [str(name)] if name else []
+
+    @property
+    def target(self) -> str:
+        """The message this widget publishes, if it is an input."""
+        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":
+            return 0
+        points = int((self.config.get("history") or {}).get("points") or 0)
+        return min(points, HISTORY_CAP)
+
+
+class SectionDef(BaseModel):
+    """A grid of widgets under a heading."""
+
+    id: str
+    title: str = ""
+    widgets: list[WidgetDef] = Field(default_factory=list)
+
+    @field_validator("id")
+    @classmethod
+    def _check_id(cls, value: str) -> str:
+        return _validate_name(value)
+
+
+class PageDef(BaseModel):
+    """One tab of a dashboard."""
+
+    id: str
+    title: str = ""
+    #: A lucide icon name, or empty.
+    icon: str = ""
+    sections: list[SectionDef] = Field(default_factory=list)
+
+    @field_validator("id")
+    @classmethod
+    def _check_id(cls, value: str) -> str:
+        return _validate_name(value)
+
+
+class DashboardDef(BaseModel):
+    """A dashboard as stored, and as the API hands it over."""
+
+    name: str
+    title: str = ""
+    pages: list[PageDef] = Field(default_factory=list)
+    #: Bumped on every save; a save based on an older one is refused.
+    version: int = 1
+
+    @field_validator("name")
+    @classmethod
+    def _check_name(cls, value: str) -> str:
+        return _validate_name(value)
+
+    @property
+    def widgets(self) -> list[WidgetDef]:
+        return [w for p in self.pages for s in p.sections for w in s.widgets]
+
+
+class DashboardSummary(BaseModel):
+    """A dashboard in a list, without its contents."""
+
+    name: str
+    title: str = ""
+    page_count: int = 0
+    widget_count: int = 0
+
+
+class DashboardsPublic(BaseModel):
+    data: list[DashboardSummary]
+    count: int
+
+
+class DashboardNotFound(KeyError):
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name
+
+
+class DashboardExists(ValueError):
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name
+
+
+class DashboardStore:
+    """Dashboards in the flow store's repository, invisible to the flow listing.
+
+    Shares the flow store's write lock and commit, so a dashboard save and a
+    flow save cannot interleave into one confused commit.
+    """
+
+    def __init__(self, flows: FlowStore) -> None:
+        self.flows = flows
+        self.root = flows.root / DASHBOARD_DIR
+        # Read-modify-write of the version counter, same as the flow store.
+        self._lock = threading.Lock()
+
+    def _file(self, name: str) -> Path:
+        return self.root / name / "dashboard.json"
+
+    def list(self) -> list[DashboardSummary]:
+        summaries = []
+        for path in sorted(self.root.glob("*/dashboard.json")):
+            try:
+                defn = DashboardDef.model_validate_json(path.read_text())
+            except Exception:
+                continue
+            summaries.append(
+                DashboardSummary(
+                    name=defn.name,
+                    title=defn.title,
+                    page_count=len(defn.pages),
+                    widget_count=len(defn.widgets),
+                )
+            )
+        return summaries
+
+    def exists(self, name: str) -> bool:
+        return self._file(name).exists()
+
+    def read(self, name: str) -> DashboardDef:
+        path = self._file(name)
+        if not path.exists():
+            raise DashboardNotFound(name)
+        return DashboardDef.model_validate_json(path.read_text())
+
+    def write(
+        self, defn: DashboardDef, base_version: int | None = None
+    ) -> DashboardDef:
+        """Save, refusing a write based on a version someone has moved past."""
+        with self._lock, self.flows._write_lock:
+            path = self._file(defn.name)
+            current = 0
+            if path.exists():
+                current = DashboardDef.model_validate_json(path.read_text()).version
+            if base_version is not None and base_version != current:
+                raise StaleVersion(defn.name, current)
+
+            saved = defn.model_copy(update={"version": current + 1})
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(saved.model_dump_json(indent=2))
+            self.flows._commit(f"Save dashboard '{defn.name}'")
+            return saved
+
+    def delete(self, name: str) -> None:
+        path = self._file(name)
+        if not path.exists():
+            raise DashboardNotFound(name)
+        with self.flows._write_lock:
+            path.unlink()
+            try:
+                path.parent.rmdir()
+            except OSError:
+                pass
+            self.flows._commit(f"Delete dashboard '{name}'")
+
+    def rename(self, name: str, new_name: str) -> DashboardDef:
+        defn = self.read(name)
+        if self.exists(new_name):
+            raise DashboardExists(new_name)
+        with self.flows._write_lock:
+            renamed = defn.model_copy(update={"name": new_name})
+            target = self._file(new_name)
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(renamed.model_dump_json(indent=2))
+            self._file(name).unlink()
+            try:
+                self._file(name).parent.rmdir()
+            except OSError:
+                pass
+            self.flows._commit(f"Rename dashboard '{name}' to '{new_name}'")
+        return renamed
+
+    def history_requirements(self) -> dict[str, int]:
+        """How many points to keep per message, so charts have a past to draw.
+
+        The deepest chart bound to a message wins; a message no chart reads
+        keeps the default.
+        """
+        limits: dict[str, int] = {}
+        for path in self.root.glob("*/dashboard.json"):
+            try:
+                defn = DashboardDef.model_validate_json(path.read_text())
+            except Exception:
+                continue
+            for widget in defn.widgets:
+                points = widget.history_points
+                if not points:
+                    continue
+                for message in widget.messages:
+                    limits[message] = max(limits.get(message, 0), points)
+        return limits
+
+
+def default_dashboard(name: str) -> DashboardDef:
+    """A new dashboard: one page, one section, nothing in it yet."""
+    return DashboardDef(
+        name=name,
+        title=name.replace("_", " ").capitalize(),
+        pages=[PageDef(id="main", title="Overview", sections=[SectionDef(id="main")])],
+    )
+
+
+__all__ = [
+    "DASHBOARD_DIR",
+    "HISTORY_CAP",
+    "INPUT_WIDGETS",
+    "DashboardDef",
+    "DashboardExists",
+    "DashboardNotFound",
+    "DashboardStore",
+    "DashboardSummary",
+    "DashboardsPublic",
+    "PageDef",
+    "Placement",
+    "SectionDef",
+    "WidgetDef",
+    "default_dashboard",
+]

backend/app/flow/pipeline.py

@@ -64,6 +64,7 @@ class Pipeline:
         "_gate_lock",
         "_queue",
         "_node_pool",
+        "history_limits",
     )
 
     def __init__(
@@ -94,6 +95,9 @@ class Pipeline:
         self._queue = work_queue
         # A pool owned by the execution service, so a wave does not build one.
         self._node_pool = node_pool
+        # How deep to keep each message's series; a chart asking for more
+        # than the default puts its message in here. Swapped, never mutated.
+        self.history_limits: dict[str, int] = {}
 
         # A message may have several producers; every one of them is upstream
         # of the nodes consuming it.
@@ -489,7 +493,7 @@ class Pipeline:
                         {self._timestamp_key(name): ts for name in result},
                     )
                 # Append-only, so it needs no lock of its own.
-                state.append_history(result, ts)
+                state.append_history(result, ts, self.history_limits)
                 self._increment_message_versions(result)
                 for name, value in result.items():
                     self._publish(
@@ -679,7 +683,7 @@ class Pipeline:
         with state.lock():
             state.update(outputs)
             state.update({self._timestamp_key(name): ts for name in outputs})
-        state.append_history(outputs, ts)
+        state.append_history(outputs, ts, self.history_limits)
         self._increment_message_versions(outputs)
         for name, value in outputs.items():
             self._publish(
@@ -756,6 +760,42 @@ class Pipeline:
         self.apply_outputs(node, outputs)
         return self.run_downstream(node)
 
+    def publish(self, values: dict[str, Any]) -> 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.
+        """
+        if not values:
+            return
+
+        ts = time.time()
+        with self._state.lock():
+            self._state.update(values)
+            self._state.update({self._timestamp_key(name): ts for name in values})
+        self._state.append_history(values, ts, self.history_limits)
+        self._increment_message_versions(values)
+        for name, value in values.items():
+            self._publish(
+                {
+                    "type": "message_value",
+                    "flow": flow_of(name),
+                    "name": name,
+                    "value": value,
+                    "ts": ts,
+                }
+            )
+
+        targets: set[Node] = set()
+        for name in values:
+            for consumer in self._nodes:
+                if name in consumer.requires:
+                    targets.add(consumer)
+                    targets.update(self._get_downstream(consumer))
+        if targets:
+            self._execute_parallel(targets, self._state, check_ready=True)
+
     def defer(
         self,
         node: Node,

backend/app/flow/state.py

@@ -173,7 +173,9 @@ class StateBackend(ABC):
     # -------------------------------------------------------------------------
 
     @abstractmethod
-    def append_history(self, values: dict[str, Any], ts: float) -> None:
+    def append_history(
+        self, values: dict[str, Any], ts: float, limits: dict[str, int] | None = None
+    ) -> None:
         """
         Record freshly published values in each message's capped history.
 
@@ -182,6 +184,8 @@ class StateBackend(ABC):
         :param values: Message names mapped to the value just published.
         :type values: dict[str, Any]
         :param ts: When they were published.
+        :param limits: How many points to keep per message, where a chart asks
+            for more than the default.
         :type ts: float
         """
         ...
@@ -307,14 +311,21 @@ class MemoryState(StateBackend):
             self._data.update(updates)
             return True
 
-    def append_history(self, values: dict[str, Any], ts: float) -> None:
+    def append_history(
+        self, values: dict[str, Any], ts: float, limits: dict[str, int] | None = None
+    ) -> None:
         """Append the numeric values to their message's bounded series."""
         with self._lock:
             for key, value in values.items():
                 number = as_number(value)
                 if number is None:
                     continue
-                series = self._history.setdefault(key, deque(maxlen=HISTORY_LIMIT))
+                cap = (limits or {}).get(key, HISTORY_LIMIT)
+                series = self._history.get(key)
+                if series is None or series.maxlen != cap:
+                    # A widget asking for a deeper series re-caps it in place.
+                    series = deque(series or (), maxlen=cap)
+                    self._history[key] = series
                 series.append((ts, number))
 
     def history(self, key: str) -> list[tuple[float, float]]:
@@ -563,7 +574,9 @@ class RedisState(StateBackend):
     def _history_key(self, key: str) -> str:
         return self._key(f"__history__:{key}")
 
-    def append_history(self, values: dict[str, Any], ts: float) -> None:
+    def append_history(
+        self, values: dict[str, Any], ts: float, limits: dict[str, int] | None = None
+    ) -> None:
         """Push the numeric values onto their capped list, in one round-trip."""
         pipe = self._client.pipeline()
         queued = False
@@ -572,8 +585,9 @@ class RedisState(StateBackend):
             if number is None:
                 continue
             history_key = self._history_key(key)
+            cap = (limits or {}).get(key, HISTORY_LIMIT)
             pipe.lpush(history_key, json.dumps([ts, number]))
-            pipe.ltrim(history_key, 0, HISTORY_LIMIT - 1)
+            pipe.ltrim(history_key, 0, cap - 1)
             if self._ttl:
                 pipe.expire(history_key, self._ttl)
             queued = True

backend/app/main.py

@@ -16,6 +16,7 @@ from app.core.config import settings
 from app.flow import logs
 from app.flow.alerts import AlertManager
 from app.flow.controller import FlowController
+from app.flow.dashboards import DashboardStore
 from app.flow.events import event_bus
 from app.flow.executor import ExecutionService
 from app.flow.nodes.http import close_shared_client
@@ -83,6 +84,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         alerts=alerts,
     )
     app.state.flow_controller = controller
+    dashboards = DashboardStore(controller.store)
+    app.state.dashboard_store = dashboards
+    # Charts need a deeper series than the default; tell the engine
+    # before it starts recording.
+    controller.set_history_limits(dashboards.history_requirements())
     watchdog = LoopWatchdog(event_bus)
     app.state.watchdog = watchdog
     watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")

backend/tests/flow/test_dashboards.py

@@ -0,0 +1,161 @@
+"""Dashboards: documents beside the flows, and the values their widgets move."""
+
+import pytest
+
+from app.flow.dashboards import (
+    DashboardDef,
+    DashboardNotFound,
+    DashboardStore,
+    PageDef,
+    SectionDef,
+    WidgetDef,
+    default_dashboard,
+)
+from app.flow.messages import DType, MessageSpec
+from app.flow.nodes import Node
+from app.flow.pipeline import Pipeline
+from app.flow.state import MemoryState
+from app.flow.store import FlowStore, StaleVersion
+
+
+@pytest.fixture
+def store(tmp_path) -> DashboardStore:
+    return DashboardStore(FlowStore(tmp_path / "flows"))
+
+
+def chart(message: str, points: int) -> WidgetDef:
+    return WidgetDef(
+        id="temp",
+        type="chart",
+        config={"series": [{"message": message}], "history": {"points": points}},
+    )
+
+
+def test_a_dashboard_survives_a_round_trip(store: DashboardStore):
+    saved = store.write(default_dashboard("house"))
+
+    read = store.read("house")
+
+    assert read.name == "house"
+    assert [p.id for p in read.pages] == ["main"]
+    assert read.version == saved.version
+
+
+def test_dashboards_are_invisible_to_the_flow_listing(store: DashboardStore):
+    """They share the repository; they are not flows."""
+    store.write(default_dashboard("house"))
+
+    assert store.flows.list_flows() == []
+    assert [d.name for d in store.list()] == ["house"]
+
+
+def test_a_save_based_on_a_version_someone_moved_past_is_refused(
+    store: DashboardStore,
+):
+    first = store.write(default_dashboard("house"))
+    store.write(first, first.version)
+
+    with pytest.raises(StaleVersion):
+        store.write(first, first.version)
+
+
+def test_deleting_and_renaming(store: DashboardStore):
+    store.write(default_dashboard("house"))
+
+    renamed = store.rename("house", "home")
+    assert renamed.name == "home"
+    assert not store.exists("house")
+
+    store.delete("home")
+    with pytest.raises(DashboardNotFound):
+        store.read("home")
+
+
+def test_the_deepest_chart_decides_how_much_past_is_kept(store: DashboardStore):
+    store.write(
+        DashboardDef(
+            name="house",
+            pages=[
+                PageDef(
+                    id="main",
+                    sections=[
+                        SectionDef(id="a", widgets=[chart("heating.temp", 400)]),
+                        SectionDef(
+                            id="b",
+                            widgets=[
+                                chart("heating.temp", 900),
+                                chart("solar.watts", 100),
+                            ],
+                        ),
+                    ],
+                )
+            ],
+        )
+    )
+
+    assert store.history_requirements() == {"heating.temp": 900, "solar.watts": 100}
+
+
+def test_a_chart_cannot_ask_for_an_unbounded_series(store: DashboardStore):
+    store.write(
+        DashboardDef(
+            name="house",
+            pages=[
+                PageDef(
+                    id="main",
+                    sections=[
+                        SectionDef(id="a", widgets=[chart("heating.temp", 10**9)])
+                    ],
+                )
+            ],
+        )
+    )
+
+    assert store.history_requirements() == {"heating.temp": 5000}
+
+
+def test_history_is_kept_to_the_depth_a_chart_asked_for():
+    state = MemoryState()
+    limits = {"f.temp": 300}
+
+    for i in range(400):
+        state.append_history({"f.temp": float(i)}, float(i), limits)
+
+    assert len(state.history("f.temp")) == 300
+
+
+# ---------------------------------------------------------------------------
+# What an input widget does
+# ---------------------------------------------------------------------------
+
+
+def test_publishing_a_value_runs_what_consumes_it():
+    """A slider is a value arriving; the graph should not care who sent it."""
+    seen: list[float] = []
+
+    def consume(setpoint, params):
+        seen.append(setpoint)
+        return {"applied": setpoint}
+
+    node = Node(
+        f=consume,
+        requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
+        provides=[MessageSpec(name="applied", port="applied", dtype=DType.FLOAT)],
+        name="thermostat",
+    )
+    node.assign_flow("heating", "thermostat")
+    state = MemoryState()
+    pipeline = Pipeline(nodes=[node], state=state)
+
+    pipeline.publish({"heating.setpoint": 21.5})
+
+    assert seen == [21.5]
+    assert state["heating.applied"] == 21.5
+
+
+def test_publishing_nothing_does_nothing():
+    pipeline = Pipeline(nodes=[], state=MemoryState())
+
+    pipeline.publish({})
+
+    assert pipeline.values() == {}