diff --git a/backend/app/flow/messages.py b/backend/app/flow/messages.py index f0ffc08..f832e8d 100644 --- a/backend/app/flow/messages.py +++ b/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: diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index ab11c9e..1f62000 100644 --- a/backend/app/flow/pipeline.py +++ b/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): diff --git a/backend/tests/flow/test_state_edges.py b/backend/tests/flow/test_state_edges.py new file mode 100644 index 0000000..6307aca --- /dev/null +++ b/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 diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 357cdd3..5d41d39 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -507,6 +507,11 @@ export const MessageSpecSchema = { minimum: 0, title: 'Interval', default: 0 + }, + trigger: { + type: 'boolean', + title: 'Trigger', + default: true } }, type: 'object', @@ -521,7 +526,11 @@ export const MessageSpecSchema = { :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 - delivery is skipped.` + 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.` } as const; export const MessageValueSchema = { @@ -1377,7 +1386,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial'], title: 'Code' }, message: { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index e2b660c..cf3a938 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -152,12 +152,17 @@ export type MessageHistory = { * 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. */ export type MessageSpec = { name?: string; port?: string; dtype?: DType; interval?: number; + trigger?: boolean; }; /** @@ -382,7 +387,7 @@ export type ValidationError = { * A problem that keeps a flow from running correctly. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial'; message: string; flow?: string; nodes?: Array<(string)>; @@ -391,7 +396,7 @@ export type ValidationIssue = { message_name?: (string | null); }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial'; export type ValidationResult = { issues?: Array;