From 7ddf8aa8f9e7f7550dfae967b6a73d3de4b97904 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 5 Nov 2026 21:24:21 +0100 Subject: [PATCH] fluksio.arrived(): a node can tell which inputs the wave behind it wrote, so a value written again unchanged is seen --- backend/fluksio/__init__.py | 10 +++++++++ backend/fluksio/flow/nodes/base.py | 6 ++++++ backend/fluksio/flow/pipeline.py | 27 +++++++++++++++++++++--- backend/fluksio/flow/remote.py | 3 ++- backend/fluksio/flow/workers.py | 3 ++- backend/tests/flow/test_pipeline.py | 31 ++++++++++++++++++++++++++++ backend/tests/flow/test_workers.py | 18 +++++++++++++++- docs/code/nodes.md | 21 +++++++++++++++++++ worker/fluksio_worker/worker_main.py | 15 +++++++++++++- 9 files changed, 127 insertions(+), 7 deletions(-) diff --git a/backend/fluksio/__init__.py b/backend/fluksio/__init__.py index 33ed817..f01e54d 100644 --- a/backend/fluksio/__init__.py +++ b/backend/fluksio/__init__.py @@ -38,6 +38,7 @@ __all__ = [ "FLOWS", "Flow", "Port", + "arrived", "build", "emit", "load_artifact", @@ -60,6 +61,15 @@ def emit(**ports: Any) -> None: logger.debug("fluksio.emit(%s) outside a node", ", ".join(ports)) +def arrived() -> set[str] | None: + """The input ports the message that started this run wrote. + + Outside a node nothing started it, so this cannot say and returns None, + the same answer a node gets when its flow starts. + """ + return None + + def _local_store() -> Any: """The worker's own artifact code, pointed at a directory of this machine. diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index 81fef88..49c48e1 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -10,6 +10,7 @@ import asyncio import inspect import logging from collections.abc import Callable, Coroutine, Iterable, Iterator +from contextvars import ContextVar from typing import TYPE_CHECKING, Any from fluksio.flow import logs @@ -37,6 +38,11 @@ class NodeOutputError(TypeError): #: Settings the engine reads itself rather than handing to the node's function. RESERVED_SETTINGS = frozenset({"synchronous"}) +#: The input ports the wave running a node has written, for the length of that +#: run. None when the run cannot say, such as a flow starting or a replay. The +#: worker request carries it to ``fluksio.arrived()`` inside the node. +ARRIVED: ContextVar[list[str] | None] = ContextVar("arrived", default=None) + class Node: """ diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 8cd66bd..b919678 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -44,6 +44,7 @@ from fluksio.flow.namespace import ( VERSION, ) from fluksio.flow.nodes import Node +from fluksio.flow.nodes.base import ARRIVED from fluksio.flow.queue import WorkQueue from fluksio.flow.state import MemoryState, StateBackend @@ -1070,6 +1071,7 @@ class Pipeline: entry_id: str = "", overrides: dict[str, Any] | None = None, inputs: dict[str, Any] | None = None, + arrived: list[str] | None = None, ) -> dict[str, Any] | None: """Run one node and record its outputs. Never raises. @@ -1108,8 +1110,12 @@ class Pipeline: if hit: return restored - with logs.capture(collected): - result = node.execute(inputs) + token = ARRIVED.set(arrived) + try: + with logs.capture(collected): + result = node.execute(inputs) + finally: + ARRIVED.reset(token) self.publish_log(node, collected, "") if ( @@ -1483,8 +1489,23 @@ class Pipeline: if replay and entry_id and self._already_done(entry_id, n): complete(n) continue + # Which of its inputs this wave wrote, so a node can tell a + # value written again from one that only sits in state. + arrived = ( + sorted( + spec.port for name, spec in n.requires.items() if name in fresh + ) + if fresh is not None + else None + ) future = executor.submit( - self._execute_node, n, state, entry_id, overrides, ready_inputs + self._execute_node, + n, + state, + entry_id, + overrides, + ready_inputs, + arrived, ) node_futures[future] = n future.add_done_callback(finished.put) diff --git a/backend/fluksio/flow/remote.py b/backend/fluksio/flow/remote.py index 965281d..d3d4ee5 100644 --- a/backend/fluksio/flow/remote.py +++ b/backend/fluksio/flow/remote.py @@ -31,7 +31,7 @@ from collections.abc import Callable from dataclasses import dataclass from typing import Any -from fluksio.flow.nodes.base import NodeOutputError +from fluksio.flow.nodes.base import ARRIVED, NodeOutputError from fluksio.flow.resources import ResourceAccountant from fluksio.flow.workers import NodeTimeout, RemoteError, _remote_class @@ -402,6 +402,7 @@ class RemoteWorkerHub: "node": node, "source": source, "kwargs": kwargs, + "arrived": ARRIVED.get(), "run": {"id": run_id} if run_id else None, "timeout": timeout, } diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index 19dffc0..65baada 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -33,7 +33,7 @@ from fluksio_worker import worker_main as _worker_main from fluksio.flow.artifacts import is_reference from fluksio.flow.events import EventBus -from fluksio.flow.nodes.base import NodeOutputError +from fluksio.flow.nodes.base import ARRIVED, NodeOutputError logger = logging.getLogger(__name__) @@ -625,6 +625,7 @@ class PythonWorkerPool: "node": node, "source": source, "kwargs": kwargs, + "arrived": ARRIVED.get(), "run": {"id": run_id} if run_id else None, }, timeout=timeout, diff --git a/backend/tests/flow/test_pipeline.py b/backend/tests/flow/test_pipeline.py index 34ff615..ac87afd 100644 --- a/backend/tests/flow/test_pipeline.py +++ b/backend/tests/flow/test_pipeline.py @@ -348,3 +348,34 @@ def test_a_consumer_of_only_quiet_producers_still_does_not_run(): power.inject() assert ran == [] + + +def test_a_node_hears_which_of_its_inputs_arrived(): + """A value written again unchanged still arrives; one only in state does not. + + The arbiter tells a spoken "on" over a light the scene already lit from + the scene re-asserting itself this way, since the values are identical. + """ + from fluksio.flow.nodes.base import ARRIVED + + heard = [] + + def arbiter(auto, manual, params): + heard.append(ARRIVED.get()) + return {"command": manual} + + node = make_node( + "arbiter", + "lights", + arbiter, + requires=[spec("auto", DType.BOOL), spec("manual", DType.BOOL)], + provides=[spec("command", DType.BOOL)], + ) + pipeline = Pipeline(nodes=[node]) + pipeline.state["lights.auto"] = True + pipeline.state["lights.manual"] = True + + pipeline.publish({"lights.manual": True}) + pipeline.publish({"lights.auto": True}) + + assert heard == [["manual"], ["auto"]] diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 54a1775..af98175 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -12,7 +12,7 @@ from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import EmitSink, FlowController from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.nodes import Node -from fluksio.flow.nodes.base import NodeOutputError +from fluksio.flow.nodes.base import ARRIVED, NodeOutputError from fluksio.flow.pipeline import Pipeline from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.state import MemoryState @@ -87,6 +87,22 @@ def test_a_node_can_log_through_the_module_it_imports(pool, capsys): assert "tuned 7" in capsys.readouterr().out +def test_a_node_reads_which_inputs_arrived(pool): + code = ( + "import fluksio\n\n\n" + "def process(auto, manual):\n" + " arrived = fluksio.arrived()\n" + " return {'out': None if arrived is None else sorted(arrived)}\n" + ) + token = ARRIVED.set(["manual"]) + try: + assert run(pool, code, auto=True, manual=True) == {"out": ["manual"]} + finally: + ARRIVED.reset(token) + # A run nothing can account for, such as a flow starting. + assert run(pool, code, auto=True, manual=True) == {"out": None} + + def test_a_failure_keeps_its_class_and_points_at_the_node(pool): with pytest.raises(Exception) as caught: run(pool, "def process():\n raise ValueError('bad input')\n") diff --git a/docs/code/nodes.md b/docs/code/nodes.md index 5564022..91b1357 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -92,6 +92,27 @@ Pass this callback to your training framework. It publishes through the same declared ports as `yield`. Outside a node, `emit` does nothing, so ordinary tests can still call your code. +### `fluksio.arrived` + +Every input arrives as an argument, whether it was just written or only +sits in state from before. `arrived()` names the input ports that the message +which started this run wrote: + +```python +import fluksio + + +def process(auto, manual): + if "manual" in (fluksio.arrived() or ()): + return {"command": manual} + return {"command": auto} +``` + +A value written again unchanged counts as arrived. That is how a node tells +somebody repeating a command apart from a value that has not moved. When the +engine cannot say, such as when a flow starts, when a paused flow resumes, or +outside a node, it returns `None`. + ## Bytes: artifacts Save files or bytes with `save_artifact`, then return the reference through an diff --git a/worker/fluksio_worker/worker_main.py b/worker/fluksio_worker/worker_main.py index 7c85480..14cea5e 100644 --- a/worker/fluksio_worker/worker_main.py +++ b/worker/fluksio_worker/worker_main.py @@ -83,6 +83,8 @@ CHUNK = 1024 * 1024 _RPC: Any = None #: The call being served, so an event can say which one it belongs to. _CALL_ID = "" +#: The input ports the wave behind this call wrote, or None if it cannot say. +_ARRIVED: list[str] | None = None def _emit(event: dict[str, Any]) -> None: @@ -116,6 +118,16 @@ class _Reporter(ModuleType): """ _emit({"event": "emit", "outputs": dict(ports)}) + def arrived(self) -> set[str] | None: + """The input ports the message that started this run wrote. + + The other inputs still hold what they held before. A value written + again unchanged is here too, which is how a node tells somebody + repeating a command from a value that only sits in state. None when + the engine cannot say, such as when the flow starts. + """ + return None if _ARRIVED is None else set(_ARRIVED) + def save_artifact( self, source: Any, @@ -545,7 +557,7 @@ def _drain(generator: Any) -> Any: def main() -> None: - global _RPC, _CALL_ID + global _RPC, _CALL_ID, _ARRIVED _RPC = os.fdopen(os.dup(1), "w") # Everything the node writes to the real stdout now goes to the server log. @@ -558,6 +570,7 @@ def main() -> None: continue request = json.loads(line) _CALL_ID = str(request.get("call_id") or "") + _ARRIVED = request.get("arrived") captured = _Capped() # ``id`` back untouched: the engine matches it against what it sent, so # a pipe that has slipped a call is caught rather than handing one node