Let a node's failure outlive the run that followed it

A node's error cleared the moment it ran again, so a failure that genuinely
fired an alert could leave no trace on the canvas by the time anyone looked.
The engine records it now — on the node's status, so it survives a reload and
every client agrees — and reading the traceback is what clears it. The seam is
the event bus, which is where every failing path already meets: a queued live
run, an explicit run, a preview, and a single triggered node all publish
`node_error`, while the controller's own observer would have seen only one of
them.

That was half the confusion. The other half: clicking a failed neuron on Home
often landed on a flow where everything looked fine. Nodes merge into one
neuron by instance key — every InfluxDB node pointing at the same bucket is one
neuron — and the click went to whichever flow contributed a member first, not
the one that failed. It now goes to the failing member and selects it, and the
canvas marks a failing node rather than leaving it to the dot alone.

The inject node emitted one payload to every port it declared, whatever their
types, so an inject on a bool port carrying the text "true" raised at publish
time. Each port gets its own field now, typed and parsed by that port's dtype,
and remembers what it last sent. A port that is renamed carries its value with
it; one that is removed takes its value with it. An inject written before this
keeps emitting exactly what it did.

The derived-cron chip also appeared on the delay node, where `interval` is a
rate limit and a schedule derived from it means nothing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Uq8mtNb97A7praJLyeEYgs

backend/app/api/routes/flows.py

@@ -278,6 +278,7 @@ def read_flows(controller: FlowControllerDep) -> Any:
                 enabled=controller.is_enabled(name),
                 paused=controller.is_paused(name),
                 quarantined=controller.is_quarantined(name),
+                version=definition.version,
             )
         )
     return FlowsPublic(data=summaries, count=len(summaries))
@@ -720,6 +721,25 @@ def cancel_node(name: str, node_id: str, controller: FlowControllerDep) -> Any:
     return Message(message=f"'{node_id}' was not running")
 
 
+@router.post("/{name}/nodes/{node_id}/acknowledge", response_model=Message)
+def acknowledge_node_error(
+    name: str, node_id: str, controller: FlowControllerDep
+) -> Any:
+    """Dismiss what a node last failed with, so the canvas stops marking it.
+
+    A failure outlives the next good run on purpose — otherwise one that fired
+    an alert leaves no trace by the time anyone looks. Reading the traceback is
+    what says it has been seen.
+    """
+    try:
+        controller.acknowledge_error(f"{name}.{node_id}")
+    except KeyError:
+        raise HTTPException(
+            status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
+        ) from None
+    return Message(message=f"Cleared the failure on '{node_id}'")
+
+
 @router.get("/{name}/state", response_model=FlowStatePublic)
 def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
     """The last value seen on every message of this flow."""
@@ -815,8 +835,12 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
     """Whether a panel's socket should carry this event.
 
     The same bound as the snapshot above, applied to the stream that follows
-    it: a value the panel draws, and nothing else on the bus.
+    it: a value the panel draws, and nothing else on the bus — save for a
+    dashboard being published, which is how a screen hears that the document
+    it is drawing, or the set of them it was given, has moved.
     """
+    if event.get("type") == "dashboard_changed":
+        return True
     return event.get("type") == "message_value" and str(event.get("name") or "") in only
 
 
@@ -855,6 +879,16 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
                     sender.cancel()
                     break
                 event = sender.result()
+                if only is not None and event.get("type") == "dashboard_changed":
+                    # The scope was resolved once, at the handshake. A panel
+                    # pointed at another dashboard would otherwise fetch the
+                    # new document and then draw tiles nothing ever updates.
+                    # ``or set()`` because a panel that was deleted resolves to
+                    # None, the same as a person's token — and that would widen
+                    # this socket to everything on the bus.
+                    only = panel_scope(token, websocket.app) or set()
+                    if controller is not None:
+                        await websocket.send_json(snapshot_payload(controller, only))
                 if only is not None and not event_for_panel(event, only):
                     continue
                 await websocket.send_json(event)

backend/app/flow/controller.py

@@ -10,8 +10,10 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import time
 import traceback
 from collections.abc import Callable
+from contextlib import suppress
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, cast
@@ -91,6 +93,11 @@ class LoadedNode:
     error: str | None = None
     health: Health = "ok"
     health_detail: str | None = None
+    #: The last time this node raised while running, and what it said. Kept
+    #: after it has run again — a failure nobody saw is the one worth keeping —
+    #: so only an acknowledgement clears it, not a good run and not a rebuild.
+    last_error: str = ""
+    last_error_ts: float | None = None
 
 
 @dataclass
@@ -307,12 +314,18 @@ class FlowController:
         self.supervisor = Supervisor(events)
         self.history_limits: dict[str, int] = {}
         self._lock = asyncio.Lock()
+        self._failures: asyncio.Task[None] | None = None
 
     # -------------------------------------------------------------------------
     # Lifecycle
     # -------------------------------------------------------------------------
 
     async def start(self) -> None:
+        # Subscribed before anything runs, so no failure falls between the
+        # first build and someone watching for one.
+        self._failures = asyncio.create_task(
+            self._watch_failures(), name="node-failures"
+        )
         # Build first: the consumer must have a pipeline to execute against
         # before it claims anything, or work waiting from the last run would be
         # taken and dropped — which is the very case the queue exists for.
@@ -321,10 +334,50 @@ class FlowController:
             self.execution.start()
 
     async def stop(self) -> None:
+        watcher, self._failures = self._failures, None
+        if watcher is not None:
+            watcher.cancel()
+            with suppress(asyncio.CancelledError):
+                await watcher
         await self._teardown()
         if self.execution is not None:
             await run_in_threadpool(self.execution.stop)
 
+    async def _watch_failures(self) -> None:
+        """Record every node failure the engine reports, from the event bus.
+
+        A node can be run by the live pipeline, by a run's own, or by hand, and
+        each builds its own graph — but they all report through
+        ``Pipeline.publish_error``, so the bus is the one place they meet.
+        Which is the point: what is recorded here is the node's, not any
+        particular pipeline's, and it outlives both the good run that follows
+        it and the rebuild after that.
+        """
+        if self.events is None:
+            return
+        async with self.events.subscribe() as queue:
+            while True:
+                event = await queue.get()
+                if event.get("type") != "node_error":
+                    continue
+                entry = self.loaded.get(str(event.get("node") or ""))
+                if entry is None:
+                    continue
+                entry.last_error = str(event.get("error") or "")
+                entry.last_error_ts = float(event.get("ts") or time.time())
+
+    def acknowledge_error(self, node_id: str) -> None:
+        """Forget what a node's last failure was. The only thing that clears it.
+
+        Raises ``KeyError`` for a node the engine does not have, so a route can
+        answer that the way every other node call does.
+        """
+        entry = self.loaded.get(node_id)
+        if entry is None:
+            raise KeyError(node_id)
+        entry.last_error = ""
+        entry.last_error_ts = None
+
     async def set_enabled(self, flow: str, enabled: bool) -> None:
         """Stop or start one flow. Rebuilding is what applies it."""
         await run_in_threadpool(self.store.write_enabled, flow, enabled)
@@ -360,6 +413,15 @@ class FlowController:
                 self._build_flows, [(flow, False) for flow in published]
             )
 
+            # A rebuild is a fresh set of nodes, but not a fresh history: every
+            # publish rebuilds every flow, so dropping the failures here would
+            # wipe them constantly. They are the operator's to dismiss.
+            for node_id, entry in loaded.items():
+                previous = self.loaded.get(node_id)
+                if previous is not None and previous.last_error:
+                    entry.last_error = previous.last_error
+                    entry.last_error_ts = previous.last_error_ts
+
             self.loaded = loaded
             self.pipeline = Pipeline(
                 nodes=nodes,
@@ -600,6 +662,8 @@ class FlowController:
                 error=entry.error,
                 health=entry.health,
                 health_detail=entry.health_detail,
+                last_error=entry.last_error,
+                last_error_ts=entry.last_error_ts,
             )
             for entry in self.loaded.values()
             if flow is None or entry.flow == flow

backend/app/flow/nodes/inject.py

@@ -28,7 +28,8 @@ class InjectNode(Node):
     """Emit a value: on request, every n seconds, on a schedule, or at startup.
 
     The value is whatever ``payload`` says, or the current time when it says
-    nothing — a timestamp is what most schedules actually want.
+    nothing — a timestamp is what most schedules actually want. ``payloads``
+    overrides that per output port, for a node that starts more than one thing.
     """
 
     class Params(BaseModel):
@@ -38,6 +39,13 @@ class InjectNode(Node):
             default=None,
             description="What to emit. Empty emits the current time.",
         )
+        payloads: dict[str, Any] = Field(
+            default_factory=dict,
+            description=(
+                "What to emit on each output port, keyed by port name. A port not "
+                "named here falls back to `payload`."
+            ),
+        )
         interval: float = Field(
             default=0,
             ge=0,
@@ -76,13 +84,19 @@ class InjectNode(Node):
             name=name or "inject",
         )
 
-    def _payload(self) -> Any:
-        return time.time() if self.cfg.payload is None else self.cfg.payload
+    def _values(self) -> dict[str, Any]:
+        """One value per output port: its own, or the node-wide payload."""
+        # Read once, so an emission that falls back carries a single timestamp
+        # across every port rather than one per port.
+        fallback = time.time() if self.cfg.payload is None else self.cfg.payload
+        return {
+            spec.port: self.cfg.payloads.get(spec.port, fallback)
+            for spec in self.output_ports
+        }
 
     def _emit(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
-        """Every output carries the same value; that is what injecting means."""
-        value = self._payload()
-        return {spec.port: value for spec in self.output_ports} or None
+        """Each output carries what its port says to emit."""
+        return self._values() or None
 
     # -------------------------------------------------------------------------
     # Its own schedule
@@ -149,8 +163,7 @@ class InjectNode(Node):
             pass
 
     async def _fire(self) -> None:
-        value = self._payload()
-        outputs = {spec.port: value for spec in self.output_ports}
+        outputs = self._values()
         if outputs:
             # inject runs the graph, which is blocking work.
             await asyncio.to_thread(self.inject, outputs)

backend/tests/flow/test_node_types.py

@@ -43,7 +43,7 @@ FIXTURES: dict[str, dict] = {
         "provides": [MessageSpec(name="score", dtype=DType.FLOAT)],
     },
     "inject": {
-        "params": {"payload": 1.0, "interval": 60},
+        "params": {"payload": 1.0, "payloads": {"tick": 2.0}, "interval": 60},
         "requires": [],
         "provides": [MessageSpec(name="tick", dtype=DType.FLOAT)],
     },
@@ -124,6 +124,29 @@ def test_mqtt_routes_topics_by_port():
     assert node._topic_to_ports == {"house/temp": ["temp"], "house/hum": ["humidity"]}
 
 
+def test_inject_emits_per_port_and_falls_back_to_one_payload():
+    """A port named in `payloads` gets its own value; the rest share `payload`."""
+    # Each port is published as its own declared type, which `check` enforces.
+    typed = NODE_TYPES["inject"].cls(
+        provides=[
+            MessageSpec(name="flag", dtype=DType.BOOL),
+            MessageSpec(name="count", dtype=DType.INT),
+        ],
+        params={"payloads": {"flag": True, "count": 3}},
+    )
+    assert typed.execute({}) == {"flag": True, "count": 3}
+
+    # No `payloads` at all is what every flow written so far carries.
+    shared = NODE_TYPES["inject"].cls(
+        provides=[
+            MessageSpec(name="left", dtype=DType.FLOAT),
+            MessageSpec(name="right", dtype=DType.FLOAT),
+        ],
+        params={"payload": 1.0},
+    )
+    assert shared.execute({}) == {"left": 1.0, "right": 1.0}
+
+
 def test_every_offered_type_has_a_fixture():
     # A new built-in without a fixture here would ship untested. Connectors are
     # separate packages and carry their own tests, so they are not this suite's.