Client graph verbs: add_input and add_data_join (#30)

src/digitaltwin/runtime.py

@@ -1053,7 +1053,8 @@ def _ensure_dtype_queue(self, dtype: DataType) -> asyncio.Queue:
 
         return self.dtype_queues[dtype]
 
-    def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON):
+    def add_input(self, dtype: DataType, channel: str,
+                  codec: str = CODEC_JSON) -> Optional[asyncio.Task]:
         """Open the graph at its input edge: bind an external channel.
 
         Sensors and other producers live outside the framework.  They
@@ -1070,22 +1071,35 @@ def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON):
 
         Internal producers keep their own path: a persistent component
         publishes through `RuntimeAPI.stream`.
+
+        Returns the subscription task (`None` for an idempotent re-bind,
+        or when the twin is tearing down): a caller that needs the
+        binding live before producers publish awaits it.
         """
 
         self._check_mutable()
 
         PubSubClient.check_channel(channel)
         check_codec(codec)
 
-        binding = _InputBinding(dtype, channel, codec)
-        if binding in self.inputs:
-            return
-        self.inputs.append(binding)
+        for existing in self.inputs:
+            if (existing.dtype, existing.channel) == (dtype, channel):
+                if existing.codec == codec:
+                    return None
+                # the stream client dedupes on (channel, dtype), so a
+                # changed codec would be recorded here yet never applied
+                # -- refuse it rather than decode with the old one forever
+                raise ValueError(
+                    f"channel {channel!r} is already bound to {dtype} with"
+                    f" codec {existing.codec!r}; a binding cannot change"
+                    " its codec"
+                )
+        self.inputs.append(_InputBinding(dtype, channel, codec))
 
         # subscribe now, so nothing published before start() is lost: the
         # queue buffers it and the consumers wait for start anyway
         logger.info(f"Bind channel {channel!r} ({codec}) to dtype: {dtype}")
-        self._to_asyncio_task(
+        return self._to_asyncio_task(
             self.streamer.subscribe_to_channel,
             channel,
             dtype,

src/digitaltwin/service/client.py

@@ -27,7 +27,8 @@
 
 from radical.orbit.client import PluginClient
 
-from ..components import DataType, TypedData
+from ..components import DataType, JoinDataType, TypedData
+from ..streaming import CODEC_JSON
 from .wire import (
     Package,
     decode,
@@ -239,6 +240,34 @@ def add_agent(
             twin_id, "add_agent", package, input_dtype, output_dtype, *args, **kwargs
         )["state"]
 
+    def add_input(
+        self,
+        twin_id: str,
+        dtype: DataType,
+        channel: str,
+        codec: str = CODEC_JSON,
+    ) -> str:
+        """Bind an external channel to one of the twin's input dtypes.
+
+        The producer lives outside the framework and publishes to
+        `channel`; the twin receives the channel's traffic, decoded by
+        `codec` (`json` for plain scripts and instruments, `raw` for
+        bytes, `cloudpickle` only inside one trust domain).  Delivery is
+        the stream's contract: at most once, newest kept under pressure.
+        The call returns once the binding is live.
+        """
+
+        return self._verb(twin_id, "add_input", dtype, channel, codec)["state"]
+
+    def add_data_join(self, twin_id: str, join_dtype: JoinDataType) -> str:
+        """Register a join: one output event per complete set of inputs.
+
+        `join_dtype` names the member dtypes; components downstream
+        consume the joined dtype like any other.
+        """
+
+        return self._verb(twin_id, "add_data_join", join_dtype)["state"]
+
     def start(self, twin_id: str) -> str:
         """Start a twin.  Starting a running twin is a no-op."""
 

src/digitaltwin/service/session.py

@@ -23,9 +23,9 @@
 from radical.orbit.plugin_session_base import PluginSession
 from rhapsody.backends.execution.orbit import OrbitExecutionBackend  # type: ignore
 
-from ..components import DataType, TypedData
+from ..components import DataType, JoinDataType, TypedData
 from ..runtime import DTRuntime, RuntimeState
-from ..streaming import PubSubClient
+from ..streaming import CODEC_JSON, PubSubClient
 from .wire import Package, check_versions, decode, encode
 
 log = logging.getLogger("radical.orbit")
@@ -77,6 +77,8 @@
     "add_task",
     "add_investigator",
     "add_agent",
+    "add_input",
+    "add_data_join",
     "start",
     "stop",
     "describe",
@@ -443,6 +445,30 @@ async def _verb_add_agent(
         component = self._instantiate(package, twin)
         twin.runtime.add_agent(component, input_dtype, output_dtype, *args, **kwargs)
 
+    async def _verb_add_input(
+        self,
+        twin: TwinInstance,
+        dtype: DataType,
+        channel: str,
+        codec: str = CODEC_JSON,
+    ) -> None:
+        # data-only arguments, no Package: the producer lives outside the
+        # framework, only the binding crosses the wire.  The runtime
+        # validates channel and codec and subscribes at bind time.
+        subscribed = twin.runtime.add_input(dtype, channel, codec)
+
+        # the verb answers only once the binding is live: a producer
+        # publishing right after the call must not race the broker
+        # registration, and a subscribe failure belongs to the caller,
+        # not to a server-side log alone
+        if subscribed is not None:
+            await subscribed
+
+    async def _verb_add_data_join(
+        self, twin: TwinInstance, join_dtype: JoinDataType
+    ) -> None:
+        twin.runtime.add_data_join(join_dtype)
+
     async def _verb_start(self, twin: TwinInstance) -> None:
         # a running twin is left running: an idempotent retry, not an error
         twin.runtime.start()

src/digitaltwin/streaming.py

@@ -926,12 +926,15 @@ async def open(
         channel: str,
         codec: str = CODEC_JSON,
         config: Optional[PubSubConfig] = None,
-        timeout: Optional[float] = None,
+        timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT,
     ) -> "ChannelPublisher":
         """Connect to a broker and publish to `channel` on it.
 
         `config` defaults to the configured broker (environment, else the
         loopback defaults); its namespace, if it has one, is ignored.
+
+        The connect is bounded by `timeout` -- an external producer
+        pointed at an unreachable broker should fail fast, not hang.
         """
 
         config = config or PubSubConfig.resolve()

src/digitaltwin/streaming_orbit.py

@@ -164,8 +164,14 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime:
         runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name,
                                   role="stream")
 
+        # ORBIT's start() takes a finite deadline only; the streaming
+        # contract's "None waits forever" is approximated by a day -- a
+        # participant that cannot register within that is stuck, not
+        # waiting.
+        effective = 86400.0 if timeout is None else timeout
+
         try:
-            runtime.start(wait=True, timeout=timeout)
+            runtime.start(wait=True, timeout=effective)
 
             # start() returns *silently* when it merely timed out, so the
             # registration has to be checked explicitly -- otherwise an
@@ -174,7 +180,7 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime:
             if not runtime.wait_registered(timeout=0):
                 raise TimeoutError(
                     f"ORBIT broker at {runtime.broker_url} did not register"
-                    f" {self.name} within {timeout}s"
+                    f" {self.name} within {effective}s"
                 )
 
         except BaseException:

test/integration/test_dt_orbit_stream.py

@@ -15,10 +15,14 @@
 
 import pytest
 
-from digitaltwin.components import TRUTHY, TypedData
+from digitaltwin.components import NULL_DTYPE, TRUTHY, DataType, JoinDataType, TypedData
 from digitaltwin.config import BACKEND_ORBIT
 from digitaltwin.service import register_user_modules
-from digitaltwin.streaming import connect_stream_client
+from digitaltwin.streaming import (
+    ChannelPublisher,
+    PubSubConfig,
+    connect_stream_client,
+)
 
 import learner_components
 import twin_components
@@ -30,6 +34,7 @@
     ECHO_DTYPE,
     INFERENCE_DTYPE,
     SENSOR_DTYPE,
+    JoinSink,
     OffsetModel,
 )
 
@@ -286,3 +291,57 @@ def test_a_persistent_component_publishes_through_the_injected_client(
     assert seen == sorted(seen), seen
 
     orbit_dt.twin_close(twin_id)
+
+
+# ---------------------------------------------------------------------------
+# external channels and joins through the client verbs (#30)
+# ---------------------------------------------------------------------------
+
+def test_external_channels_join_through_the_client_verbs(orbit_dt, twin_id):
+    """The servicified demo shape: two external channels bound with
+    `add_input`, joined with `add_data_join`, consumed downstream.  The
+    producers are plain `ChannelPublisher`s -- outside the framework,
+    knowing nothing about twins."""
+
+    a = DataType("chan-a")
+    b = DataType("chan-b")
+    joined = JoinDataType([a, b])
+
+    chan_a = f"itest/{twin_id[:8]}/a"
+    chan_b = f"itest/{twin_id[:8]}/b"
+
+    orbit_dt.create_twin(twin_id)
+    orbit_dt.add_input(twin_id, a, chan_a)
+    orbit_dt.add_input(twin_id, b, chan_b)
+    orbit_dt.add_data_join(twin_id, joined)
+    orbit_dt.add_task(twin_id, orbit_dt.package(JoinSink), joined, NULL_DTYPE)
+    assert orbit_dt.start(twin_id) == "running"
+
+    async def feed_and_collect():
+        config = PubSubConfig(kind=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL)
+        collector = await connect_stream_client(
+            twin_id, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL)
+        queue: asyncio.Queue = asyncio.Queue()
+        pub_a = await ChannelPublisher.open(chan_a, config=config)
+        pub_b = await ChannelPublisher.open(chan_b, config=config)
+
+        try:
+            await collector.subscribe_to_dtype(ECHO_DTYPE, queue)
+            for value in range(3):
+                await pub_a.publish(value)
+                await pub_b.publish(value * 10)
+            return [
+                (await asyncio.wait_for(queue.get(), COLLECT_TIMEOUT)).data
+                for _ in range(3)
+            ]
+        finally:
+            await pub_a.close()
+            await pub_b.close()
+            await collector.close()
+
+    seen = asyncio.run(feed_and_collect())
+
+    # every output is one complete (a, b) pair; joins arrive in order
+    assert seen == [0, 11, 22], seen
+
+    assert orbit_dt.twin_close(twin_id) == "closed"

test/integration/twin_components.py

@@ -121,6 +121,15 @@ async def main_loop(self, runtime, in_data):
         await runtime.stream.publish(ECHO_DTYPE, in_data.data)
 
 
+class JoinSink(UtilityTask):
+    """Terminal for a joined stream: republishes the member sum, so a
+    client can watch complete join sets arrive over the twin's stream."""
+
+    async def main_loop(self, runtime, in_data):
+        await runtime.stream.publish(
+            ECHO_DTYPE, sum(item.data for item in in_data.data))
+
+
 class CrashingTask(UtilityTask):
     """Persistent component that dies -- the twin must land in `failed`
     with the reason visible in `twin_list`."""