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

@@ -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.
 

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:
     """

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)

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,
         }

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,

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"]]

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")

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

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