Let a flow keep state in the messages it already has

Logic nodes are pure functions with no state handle, but real automations
count things and remember the last reading. The shape for that is a
message a node both reads and writes: the graph already declines to make
a node depend on itself, so this worked by accident. It is now defined,
tested, and checked — a node that is the only writer of a message it
reads is told at edit time that it needs a starting value, rather than
silently never running.

Feeding a value back between two nodes was still a cycle, and rejected.
An input can now be marked non-triggering: read when the node runs, never
the reason it runs, and no dependency either way. That is what a back
edge actually means.

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

backend/app/flow/messages.py

@@ -46,6 +46,10 @@ class MessageSpec(BaseModel):
         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
         delivery is skipped.
+    :param trigger: Whether arriving values wake the node. An input with this
+        off is read when the node runs for some other reason, but never causes
+        a run and never makes the node wait — which is how a node reads a
+        message it also produces without depending on itself.
     """
 
     model_config = ConfigDict(frozen=True)
@@ -54,6 +58,7 @@ class MessageSpec(BaseModel):
     port: str = ""
     dtype: DType = DType.FLOAT
     interval: float = Field(default=0, ge=0)
+    trigger: bool = True
 
     @model_validator(mode="after")
     def _default_port(self) -> MessageSpec:

backend/app/flow/pipeline.py

@@ -35,6 +35,7 @@ class ValidationIssue(BaseModel):
         "missing_initial_value",
         "node_error",
         "unauthenticated_hook",
+        "self_loop_needs_initial",
     ]
     message: str
     flow: str = ""
@@ -91,10 +92,14 @@ class Pipeline:
             for msg in node.provides:
                 self.produces.setdefault(msg, []).append(node)
 
+        # A node never depends on itself: reading a message it also provides is
+        # how state is carried between runs, not a cycle. An input marked
+        # non-triggering is the same idea across two nodes.
         self.dependencies: dict[Node, frozenset[Node]] = {
             node: frozenset(
                 producer
-                for msg in node.requires
+                for msg, spec in node.requires.items()
+                if spec.trigger
                 for producer in self.produces.get(msg, ())
                 if producer is not node
             )
@@ -211,6 +216,29 @@ class Pipeline:
         for node in self._nodes:
             for msg_name, spec in node.requires.items():
                 if msg_name in self.produces:
+                    # A message a node both reads and writes carries state
+                    # between its runs. If the node is the only one writing it,
+                    # the first run has nothing to read unless the flow declares
+                    # a starting value.
+                    if (
+                        spec.trigger
+                        and self.produces[msg_name] == [node]
+                        and not declared.get(msg_name, False)
+                    ):
+                        issues.append(
+                            ValidationIssue(
+                                code="self_loop_needs_initial",
+                                message=(
+                                    f"'{node.local_id}' reads '{msg_name}' and is "
+                                    "the only node writing it, so it needs a "
+                                    "starting value to ever run."
+                                ),
+                                flow=node.flow,
+                                node=node.id,
+                                port=spec.port,
+                                message_name=msg_name,
+                            )
+                        )
                     continue
 
                 if msg_name not in declared:
@@ -290,18 +318,23 @@ class Pipeline:
             self._state.increment(self._version_key(msg_name))
 
     def _check_synchronous_ready(self, node: Node) -> tuple[bool, dict[str, int]]:
-        """A synchronous node runs once every input is newer than last time."""
-        if not node.requires:
+        """A synchronous node runs once every input is newer than last time.
+
+        Only triggering inputs count: waiting for a value the node itself
+        writes would mean waiting for a run that can never start.
+        """
+        waited_on = [msg for msg, spec in node.requires.items() if spec.trigger]
+        if not waited_on:
             return True, {}
 
-        version_keys = [self._version_key(msg) for msg in node.requires]
-        last_seen_keys = [self._last_seen_key(node.id, msg) for msg in node.requires]
+        version_keys = [self._version_key(msg) for msg in waited_on]
+        last_seen_keys = [self._last_seen_key(node.id, msg) for msg in waited_on]
         values = self._state.get_multi(version_keys + last_seen_keys)
 
         current_versions = {}
         all_newer = True
 
-        for msg_name in node.requires:
+        for msg_name in waited_on:
             current = values.get(self._version_key(msg_name)) or 0
             last_seen = values.get(self._last_seen_key(node.id, msg_name)) or 0
             current_versions[msg_name] = current
@@ -359,8 +392,11 @@ class Pipeline:
 
     def _is_node_ready(self, node: Node, state: StateBackend) -> bool:
         with state.lock():
-            for msg_name in node.requires:
-                if msg_name not in state:
+            for msg_name, spec in node.requires.items():
+                # A non-triggering input is read if it happens to be there;
+                # waiting for it would make an accumulator's first run
+                # impossible, since it is what the node is about to write.
+                if spec.trigger and msg_name not in state:
                     return False
 
         if not self._input_is_due(node):

backend/tests/flow/test_state_edges.py

@@ -0,0 +1,155 @@
+"""State carried between runs, without a state API.
+
+Logic nodes are pure functions of their inputs, so an accumulator keeps its
+running value in a message it both reads and writes. That is the sanctioned
+shape, and these tests pin what it means.
+"""
+
+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
+
+
+def counter() -> Node:
+    """Adds each reading to a total it keeps in its own output message."""
+
+    def process(reading, params, total=0.0):
+        return {"total": total + reading}
+
+    node = Node(
+        f=process,
+        requires=[
+            MessageSpec(name="reading", port="reading", dtype=DType.FLOAT),
+            MessageSpec(name="total", port="total", dtype=DType.FLOAT),
+        ],
+        provides=[MessageSpec(name="total", port="total", dtype=DType.FLOAT)],
+        name="accumulate",
+    )
+    node.assign_flow("f", "accumulate")
+    return node
+
+
+def source() -> Node:
+    def process(params):
+        return {}
+
+    node = Node(
+        f=process,
+        provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
+        name="sensor",
+    )
+    node.assign_flow("f", "sensor")
+    return node
+
+
+def test_a_node_reading_what_it_writes_does_not_depend_on_itself():
+    node = counter()
+    pipeline = Pipeline(nodes=[node])
+
+    assert pipeline.dependencies[node] == frozenset()
+    # Nor is it downstream of itself, so publishing does not re-run it.
+    assert node not in pipeline.edges[node]
+    assert pipeline.validate({"f.reading": True, "f.total": True}) == []
+
+
+def test_the_running_total_survives_between_runs():
+    state = MemoryState()
+    node = counter()
+    pipeline = Pipeline(
+        nodes=[node],
+        state=state,
+        initial_values={"f.reading": 0.0, "f.total": 0.0},
+    )
+
+    for reading in (2.0, 3.0, 5.0):
+        state["f.reading"] = reading
+        pipeline.run()
+
+    assert state["f.total"] == 10.0
+
+
+def test_a_self_loop_without_a_starting_value_is_reported():
+    """It could never run: the value it waits for is the one it writes."""
+    node = counter()
+    pipeline = Pipeline(nodes=[node, source()])
+
+    issues = pipeline.validate({"f.reading": True})
+
+    assert [i.code for i in issues] == ["self_loop_needs_initial"]
+    assert issues[0].message_name == "f.total"
+
+    # Declaring it as a flow input with a value settles it.
+    assert pipeline.validate({"f.reading": True, "f.total": True}) == []
+
+
+def test_a_non_triggering_input_makes_a_two_node_loop_legal():
+    """A→B→A is a cycle only while both edges wake their consumer.
+
+    Marking the back edge non-triggering says what is actually meant: B's
+    result is state A reads on its next run, not something that runs A.
+    """
+
+    def forward(params, value=0.0):
+        return {"echo": value}
+
+    def back(echo, params):
+        return {"value": echo + 1.0}
+
+    a = Node(
+        f=forward,
+        # The back edge: read when a runs, never the reason it runs.
+        requires=[
+            MessageSpec(name="value", port="value", dtype=DType.FLOAT, trigger=False)
+        ],
+        provides=[MessageSpec(name="echo", port="echo", dtype=DType.FLOAT)],
+        name="a",
+    )
+    b = Node(
+        f=back,
+        requires=[MessageSpec(name="echo", port="echo", dtype=DType.FLOAT)],
+        provides=[MessageSpec(name="value", port="value", dtype=DType.FLOAT)],
+        name="b",
+    )
+    a.assign_flow("f", "a")
+    b.assign_flow("f", "b")
+
+    state = MemoryState()
+    pipeline = Pipeline(nodes=[a, b], state=state, initial_values={"f.value": 1.0})
+
+    assert pipeline.dependencies[a] == frozenset()
+    assert pipeline.dependencies[b] == frozenset({a})
+    # Kahn sees no loop, so nothing is reported as cyclic.
+    assert [i.code for i in pipeline.validate({"f.value": True})] == []
+
+    pipeline.run()
+    assert state["f.echo"] == 1.0
+    assert state["f.value"] == 2.0
+
+    # The next run reads what b wrote, which is the point of the back edge.
+    pipeline.run()
+    assert state["f.echo"] == 2.0
+    assert state["f.value"] == 3.0
+
+
+def test_a_non_triggering_input_never_holds_a_node_back():
+    """Absent, it is simply left out of the call rather than awaited."""
+
+    def process(params, seen=None):
+        return {"out": 1.0 if seen is None else 2.0}
+
+    node = Node(
+        f=process,
+        requires=[
+            MessageSpec(name="seen", port="seen", dtype=DType.FLOAT, trigger=False)
+        ],
+        provides=[MessageSpec(name="out", port="out", dtype=DType.FLOAT)],
+        name="n",
+    )
+    node.assign_flow("f", "n")
+
+    state = MemoryState()
+    pipeline = Pipeline(nodes=[node], state=state)
+
+    pipeline.run()
+    assert state["f.out"] == 1.0