From bd732795ff94159fbb9836175f3245059d9f0745 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 19:52:48 +0200
Subject: [PATCH 01/11] A second data plane: the ORBIT-backed pubsub backend

The DT streams carry cloudpickled payloads over ZMQ ports that nobody
authenticates: reaching them is code execution in every subscriber (plan
risk R7).  This puts the same payloads inside ORBIT's token-authenticated
star instead, behind the backend seam M0 built, so nothing above it
notices.

`OrbitPubSubBackend` publishes with `EndpointRuntime.send_notification`
and subscribes with `register_callback`, one registration per DT topic
under a single `dt_stream` plugin namespace -- ORBIT matches topics by
exact equality, which is what DT topics already are.  It owns one
participant connection per twin (rhapsody's `OrbitExecutionBackend`
pattern, loopback wrinkle and all), or rides an injected one.  Frames
cross from the runtime's callback thread to the host loop through a
bounded drop-oldest inbox -- the broker's own discipline, and the
conflation semantics the data plane is specified with.  Oversized
payloads raise at publish, because ORBIT would drop them with nothing but
a log line.

The two backends had grown the same receive loop, subscriber registry and
closed/running state, so those move up into `PubSubBackend`.

`DT_STREAM_BACKEND` selects the transport at deployment time.  Under
'orbit' the plugin's embedded ZMQ broker is never started at all -- the
`connect_stream` seam is the only caller of `stream_addresses`, so that
is structural rather than remembered.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/digitaltwin/__init__.py        |   3 +-
 src/digitaltwin/config.py          |  57 ++++-
 src/digitaltwin/service/plugin.py  |  79 +++++--
 src/digitaltwin/service/session.py |   9 +-
 src/digitaltwin/streaming.py       | 168 ++++++++++----
 src/digitaltwin/streaming_orbit.py | 337 +++++++++++++++++++++++++++++
 test/unit/test_service.py          |   3 +-
 7 files changed, 580 insertions(+), 76 deletions(-)
 create mode 100644 src/digitaltwin/streaming_orbit.py

diff --git a/src/digitaltwin/__init__.py b/src/digitaltwin/__init__.py
index d882b6c..dc541d0 100644
--- a/src/digitaltwin/__init__.py
+++ b/src/digitaltwin/__init__.py
@@ -13,7 +13,7 @@
     WindowDataType,
     WindowedTypeData,
 )
-from .config import stream_addresses
+from .config import stream_addresses, stream_backend
 from .runtime import DTRuntime, RuntimeAPI, RuntimeState
 from .streaming import (
     CODEC_CLOUDPICKLE,
@@ -56,4 +56,5 @@
     "ZMQ_PS_Client",
     "connect_stream_client",
     "stream_addresses",
+    "stream_backend",
 ]
diff --git a/src/digitaltwin/config.py b/src/digitaltwin/config.py
index b43292a..65a674c 100644
--- a/src/digitaltwin/config.py
+++ b/src/digitaltwin/config.py
@@ -1,13 +1,22 @@
 """Deployment configuration for the digital twin framework.
 
-The single place where stream endpoint addresses are decided.  No other
-module (and no demo) may contain a hardcoded transport address.
-
-Binding policy: the framework binds to loopback unless the deployment
-explicitly configures something else.  The pubsub data plane carries
-cloudpickled payloads, so anyone who can reach the broker ports can
-execute code in every subscriber -- non-loopback binds require an
-explicit configuration and a firewalled/private network.
+The single place where the stream transport is decided: which backend
+carries the data plane, and -- for the ZMQ one -- at which addresses.  No
+other module (and no demo) may contain a hardcoded transport address.
+
+Backend choice (`DT_STREAM_BACKEND`) is a deployment-time decision at the
+same altitude as the compute backend; nothing above `PubSubBackend`
+depends on which one is in use.
+
+- `zmq` (default): the framework's own XSUB/XPUB broker.  Local and
+  two-terminal use.  Its payloads are cloudpickled and its ports carry no
+  authentication, so anyone who can reach them can execute code in every
+  subscriber (risk R7).  Hence the binding policy: loopback unless the
+  deployment explicitly configures something else, and a non-loopback
+  bind requires a firewalled/private network.
+- `orbit`: ORBIT eventing.  The same cloudpickled payloads, but inside
+  the token-authenticated WS star -- no DT-owned ports at all.  This is
+  what closes R7 for a deployment, and it is required before production.
 """
 
 import os
@@ -27,6 +36,38 @@
 ENV_PUB_ADDR = "DT_STREAM_PUB_ADDR"
 ENV_SUB_ADDR = "DT_STREAM_SUB_ADDR"
 
+# which transport carries the data plane -- see the module docstring
+BACKEND_ZMQ = "zmq"
+BACKEND_ORBIT = "orbit"
+STREAM_BACKENDS = (BACKEND_ZMQ, BACKEND_ORBIT)
+
+# zmq: the local/two-terminal default.  A deployment that wants R7 closed
+# selects 'orbit' -- deliberately, not by accident of the environment.
+DEFAULT_STREAM_BACKEND = BACKEND_ZMQ
+
+ENV_STREAM_BACKEND = "DT_STREAM_BACKEND"
+
+
+def stream_backend(name: str | None = None) -> str:
+    """Resolve which pubsub backend carries the data plane.
+
+    Precedence: explicit argument, then `DT_STREAM_BACKEND`, then `zmq`.
+    An unknown name is an error rather than a silent fallback -- a typo
+    must not quietly reopen the ZMQ ports of a deployment that asked for
+    the token-authenticated one.
+    """
+
+    chosen = (name or os.environ.get(ENV_STREAM_BACKEND)
+              or DEFAULT_STREAM_BACKEND).strip().lower()
+
+    if chosen not in STREAM_BACKENDS:
+        raise ValueError(
+            f"unknown stream backend {chosen!r};"
+            f" expected one of {', '.join(STREAM_BACKENDS)}"
+        )
+
+    return chosen
+
 
 def stream_addresses(
     pub_addr: str | None = None, sub_addr: str | None = None
diff --git a/src/digitaltwin/service/plugin.py b/src/digitaltwin/service/plugin.py
index 60146f0..b94d986 100644
--- a/src/digitaltwin/service/plugin.py
+++ b/src/digitaltwin/service/plugin.py
@@ -11,11 +11,17 @@
 - the wire: one graph verb per `twin_call`, cloudpickle-base64 payloads
   with a version stamp, and `twin_list` as the only observation path.
 
-**Binding policy (risk R7)**: the stream broker binds loopback by
-default.  Its payloads are cloudpickled, so anyone who can reach its
-ports executes code in every subscriber.  A non-loopback bind needs an
-explicit `DT_STREAM_PUB_ADDR` / `DT_STREAM_SUB_ADDR` configuration *and*
-a firewalled/private network.
+**Data plane (risk R7)**: `DT_STREAM_BACKEND` picks the transport.
+
+- `orbit` puts the twins' streams inside the same token-authenticated
+  WebSocket star as the control plane, and the embedded ZMQ broker is
+  then never started -- no DT-owned ports exist at all.  This is what
+  closes R7, and it is what a production deployment selects.
+- `zmq` (the default) runs the embedded stream broker, and *binds
+  loopback*.  Its payloads are cloudpickled, so anyone who can reach its
+  ports executes code in every subscriber.  A non-loopback bind needs an
+  explicit `DT_STREAM_PUB_ADDR` / `DT_STREAM_SUB_ADDR` configuration
+  *and* a firewalled/private network.
 """
 
 import asyncio
@@ -30,8 +36,13 @@
 from radical.orbit.plugin_base import Plugin
 from starlette.requests import Request
 
-from ..config import embedded_stream_addresses
-from ..streaming import ZMQ_BrokerProcess
+from ..config import BACKEND_ORBIT, embedded_stream_addresses, stream_backend
+from ..streaming import (
+    CLIENT_CONNECT_TIMEOUT,
+    PubSubClient,
+    ZMQ_BrokerProcess,
+    connect_stream_client,
+)
 from .client import DTClient
 from .session import VERBS, DTSession
 
@@ -82,8 +93,13 @@ def __init__(self, app: FastAPI, instance_name: str = "dt"):
 
         self.broker_url: Optional[str] = os.environ.get(ENV_BROKER_URL) or None
 
+        # which transport carries the twins' streams.  Resolved once, at
+        # plugin construction: a deployment decision, not a per-twin one.
+        self.stream_backend: str = stream_backend()
+
         # the embedded DT stream broker, shared plugin-wide and started on
-        # first need (see `stream_addresses`)
+        # first need (see `stream_addresses`).  Never started at all under
+        # the 'orbit' backend -- that is the point of it.
         self._stream_broker: Optional[ZMQ_BrokerProcess] = None
         self._stream_addrs: Optional[tuple[str, str]] = None
         self._stream_lock = asyncio.Lock()
@@ -254,13 +270,7 @@ async def admin_sessions(self, request: Request) -> dict:
                 entry["sid"] = sid
             sessions.append(entry)
 
-        return {
-            "sessions": sessions,
-            "stream_broker": {
-                "addresses": self._stream_addrs,
-                "alive": bool(self._stream_broker and self._stream_broker.is_alive()),
-            },
-        }
+        return {"sessions": sessions, "stream_broker": self.stream_summary()}
 
     # -- observability ------------------------------------------------------
 
@@ -309,6 +319,45 @@ async def on_topology_change(self, participants: dict) -> None:
                     sid, ", ".join(sorted(lost)), ", ".join(failed),
                 )
 
+    # -- the data plane -----------------------------------------------------
+
+    async def connect_stream(
+        self, namespace: str, timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT
+    ) -> PubSubClient:
+        """A connected, namespaced stream client on the selected backend.
+
+        The single place where the plugin's transport choice is applied.
+        The `orbit` branch never touches `stream_addresses`, which is what
+        keeps the embedded ZMQ broker from being started in a deployment
+        that asked for the token-authenticated data plane.
+        """
+
+        if self.stream_backend == BACKEND_ORBIT:
+            return await connect_stream_client(
+                namespace,
+                timeout=timeout,
+                backend=BACKEND_ORBIT,
+                broker_url=self.broker_url,
+            )
+
+        pub_addr, sub_addr = await self.stream_addresses()
+
+        return await connect_stream_client(namespace, pub_addr, sub_addr, timeout)
+
+    def stream_summary(self) -> dict:
+        """The data plane's entry in `admin/sessions`."""
+
+        if self.stream_backend == BACKEND_ORBIT:
+            # no addresses and nothing to supervise: the streams ride the
+            # ORBIT connection the control plane already has
+            return {"backend": BACKEND_ORBIT}
+
+        return {
+            "backend": self.stream_backend,
+            "addresses": self._stream_addrs,
+            "alive": bool(self._stream_broker and self._stream_broker.is_alive()),
+        }
+
     # -- embedded stream broker --------------------------------------------
 
     async def stream_addresses(self) -> tuple[str, str]:
diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py
index 4a69473..bf29d4f 100644
--- a/src/digitaltwin/service/session.py
+++ b/src/digitaltwin/service/session.py
@@ -23,7 +23,7 @@
 
 from ..components import DataType, TypedData
 from ..runtime import DTRuntime, RuntimeState
-from ..streaming import PubSubClient, connect_stream_client
+from ..streaming import PubSubClient
 from .wire import Package, check_versions, decode, encode
 
 log = logging.getLogger("radical.orbit")
@@ -669,9 +669,10 @@ async def _init_twin(self, twin: TwinInstance) -> None:
 
                 flow, *_ = await asyncio.gather(*map(self.engine, names))
 
-                pub_addr, sub_addr = await self._plugin.stream_addresses()
-                stream = await connect_stream_client(
-                    twin.twin_id, pub_addr, sub_addr, STREAM_CONNECT_TIMEOUT
+                # the plugin owns the transport choice (zmq / orbit); the
+                # twin only ever sees a connected, namespaced client
+                stream = await self._plugin.connect_stream(
+                    twin.twin_id, STREAM_CONNECT_TIMEOUT
                 )
                 twin.ready(DTRuntime(flow, stream), stream)
 
diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py
index 71572e2..b3a68e4 100644
--- a/src/digitaltwin/streaming.py
+++ b/src/digitaltwin/streaming.py
@@ -19,7 +19,13 @@
 from zmq.utils.monitor import recv_monitor_message
 
 from .components import DataType, TypedData
-from .config import RANDOM_PUB_ADDR, RANDOM_SUB_ADDR, stream_addresses
+from .config import (
+    BACKEND_ORBIT,
+    RANDOM_PUB_ADDR,
+    RANDOM_SUB_ADDR,
+    stream_addresses,
+    stream_backend,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -75,6 +81,15 @@ def decode_payload(payload: bytes, codec: str):
 
 
 class PubSubBackend(ABC):
+    """The transport seam: everything above this is transport-agnostic.
+
+    A backend delivers to per-topic subscriber callbacks from a single
+    receive loop it owns.  That loop, the subscriber registry and the
+    closed/running state are the same in every backend, so they live
+    here; a subclass supplies `connect`, `publish`, `subscribe`,
+    `unsubscribe`, `close` and a `_run()` body.
+    """
+
     label = "generic"
 
     # names this backend in a PubSubConfig: what has to reopen the
@@ -92,12 +107,77 @@ def __init__(self):
         # the failure mode this exists to prevent.
         self.on_error: Optional[Callable[[BaseException], None]] = None
 
+        # topic -> subscriber callbacks.  Delivery is filtered by exact
+        # topic lookup, whatever the transport matched on the wire.
+        self.topics: dict[str, list[Callable]] = {}
+
+        self._task: Optional[asyncio.Task] = None
+        self._closed = False
+        self.is_running = asyncio.Event()
+
     def _report_error(self, exc: BaseException):
         logger.error("stream backend failed: %s", exc, exc_info=exc)
 
         if self.on_error is not None:
             self.on_error(exc)
 
+    def _check_open(self):
+        if self._closed:
+            raise RuntimeError("stream client is closed")
+
+    async def _await_running(self, what: str):
+        """Callers which arrive before `connect()` finished are made to
+        wait rather than to lose their message."""
+
+        if not self.is_running.is_set():
+            logger.warning("requesting %s before connecting to broker. Waiting",
+                           what)
+            await self.is_running.wait()
+
+    def _start_receiving(self):
+        """Arm the supervised receive loop.  Called at the end of connect."""
+
+        self._task = asyncio.create_task(self._run())
+        self._task.add_done_callback(self._run_done)
+
+    async def _dispatch(self, topic: str, message):
+        """Hand one decoded message to the topic's subscribers.
+
+        One failing subscriber must not starve its siblings
+        (`CancelledError` is not an `Exception`: close() still wins).
+        """
+
+        for task in self.topics.get(topic, []):
+            try:
+                await task(message)
+            except Exception:
+                logger.exception("subscriber failed on topic %r", topic)
+
+    async def _cancel_receiving(self):
+        """Cancel and await the receive loop.  Part of every close()."""
+
+        task, self._task = self._task, None
+        if task is not None:
+            task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await task
+
+    def _run_done(self, task: asyncio.Task):
+        """The receive loop only ends on close().  Any other exit means the
+        twin stopped receiving -- report it instead of stalling silently."""
+
+        if self._closed or task.cancelled():
+            return
+
+        exc = task.exception() or RuntimeError("stream receive loop exited")
+        self._report_error(exc)
+
+    async def _run(self):
+        """Receive loop: decode frames and `_dispatch` them.  Runs until
+        cancelled by close()."""
+
+        raise NotImplementedError
+
     @abstractmethod
     async def connect(self, *args, **kwargs):
         pass
@@ -308,18 +388,10 @@ def __init__(self, pub_addr: Optional[str] = None, sub_addr: Optional[str] = Non
             self._ctx.socket(zmq.SUB) if sub_addr is not None else None
         )
 
-        # subscribe: store the callback for the topic
-        # publish: send a message to each of the callbacks.
-
-        self.topics: dict[str, list[Callable]] = {}
-
         # topics whose payload the transport must hand over untouched:
         # something above the seam owns their wire format (see the codecs)
         self.raw_topics: set[str] = set()
 
-        self._task: Optional[asyncio.Task] = None
-        self._closed = False
-        self.is_running = asyncio.Event()
 
     async def _connect_socket(self, sock, addr):
         """Connect `sock` and wait until the connection is established.
@@ -374,8 +446,7 @@ async def connect(self, timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT):
             raise
 
         if self.sub_soc is not None:
-            self._task = asyncio.create_task(self._run())
-            self._task.add_done_callback(self._run_done)
+            self._start_receiving()
 
         self.is_running.set()
 
@@ -384,9 +455,7 @@ async def publish(self, topic, message, raw=False):
         if self.pub_soc is None:
             raise ValueError("Publishing endpoint not connected")
 
-        if not (self.is_running.is_set()):
-            logger.warning("Requesting publish before connecting to broker. Waiting")
-            await self.is_running.wait()
+        await self._await_running("publish")
 
         topic_b = topic.encode("utf-8")
         message_b = message if raw else cloudpickle.dumps(message)
@@ -397,9 +466,8 @@ async def subscribe(self, topic, callback, raw=False, **backend_params):
         if self.sub_soc is None:
             raise ValueError("Subscribe endpoint not connected")
 
-        if not (self.is_running.is_set()):
-            logger.warning("Requesting subscribe before connecting to broker. Waiting")
-            await self.is_running.wait()
+        await self._await_running("subscribe")
+
         self.sub_soc.setsockopt(zmq.SUBSCRIBE, topic.encode("utf-8"))
 
         self.topics.setdefault(topic, []).append(callback)
@@ -425,12 +493,8 @@ async def close(self):
             return
         self._closed = True
 
-        task, self._task = self._task, None
         try:
-            if task is not None:
-                task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await task
+            await self._cancel_receiving()
         finally:
             # the guard above makes close() a one-shot, so the sockets and
             # the context have to go even if the await was cancelled
@@ -446,10 +510,6 @@ async def close(self):
             self._ctx.term()
             self.is_running.clear()
 
-    def _check_open(self):
-        if self._closed:
-            raise RuntimeError("stream client is closed")
-
     async def _run(self):
         """Receive loop.  A single bad payload or a raising callback must
         not take the stream down -- it is dropped and logged."""
@@ -467,23 +527,7 @@ async def _run(self):
                 logger.exception("dropping malformed message: %r", frames[:1])
                 continue
 
-            for task in self.topics.get(item, []):
-                # one failing subscriber must not starve its siblings
-                # (CancelledError is not an Exception: close() still wins)
-                try:
-                    await task(data)
-                except Exception:
-                    logger.exception("subscriber failed on topic %r", item)
-
-    def _run_done(self, task: asyncio.Task):
-        """The receive loop only ends on close().  Any other exit means the
-        twin stopped receiving -- report it instead of stalling silently."""
-
-        if self._closed or task.cancelled():
-            return
-
-        exc = task.exception() or RuntimeError("stream receive loop exited")
-        self._report_error(exc)
+            await self._dispatch(item, data)
 
 
 # The pubsub client abstracts away the specifics of the pub / sub
@@ -546,6 +590,7 @@ def config(self) -> "PubSubConfig":
             pub_addr=self._backend.pub_addr,
             sub_addr=self._backend.sub_addr,
             kind=self._backend.kind,
+            broker_url=getattr(self._backend, "broker_url", None),
         )
 
     def topic(self, dtype: DataType) -> str:
@@ -692,12 +737,17 @@ class PubSubConfig:
     `kind` names the backend which has to open the endpoint; a backend
     declares its own (`PubSubBackend.kind`).  It stays a plain string so a
     further backend needs no change here.
+
+    The address fields belong to the `zmq` kind; `broker_url` to the
+    `orbit` kind (`None` uses ORBIT's own resolution, and the auth token
+    is deliberately not part of the config -- it is resolved locally).
     """
 
     namespace: Optional[str] = None
     pub_addr: Optional[str] = None
     sub_addr: Optional[str] = None
     kind: str = ZMQ_PS_Client.kind
+    broker_url: Optional[str] = None
 
     @classmethod
     def resolve(
@@ -722,13 +772,22 @@ async def connect_backend(self, timeout: Optional[float] = None) -> PubSubBacken
         broker leaks neither sockets nor a context.
         """
 
-        if self.kind != ZMQ_PS_Client.kind:
+        if self.kind == BACKEND_ORBIT:
+            # imported here: `radical.orbit` is the optional 'service'
+            # extra, and a plain ZMQ install must not need it
+            from .streaming_orbit import OrbitPubSubBackend
+
+            backend = OrbitPubSubBackend(self.broker_url)
+
+        elif self.kind == ZMQ_PS_Client.kind:
+            backend = ZMQ_PS_Client(self.pub_addr, self.sub_addr)
+
+        else:
             raise ValueError(
                 f"cannot open a {self.kind!r} stream endpoint here:"
                 f" no backend of that kind is available"
             )
 
-        backend = ZMQ_PS_Client(self.pub_addr, self.sub_addr)
         await backend.connect(timeout)
 
         return backend
@@ -793,11 +852,26 @@ async def connect_stream_client(
     pub_addr: Optional[str] = None,
     sub_addr: Optional[str] = None,
     timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT,
+    *,
+    backend: Optional[str] = None,
+    broker_url: Optional[str] = None,
 ) -> PubSubClient:
     """Build and connect a namespaced stream client from configuration.
 
-    The connect is bounded by `timeout` -- see `ZMQ_PS_Client.connect`.
+    `backend` selects the transport (`zmq` / `orbit`); unset, it comes
+    from `DT_STREAM_BACKEND` (see `config.stream_backend`).  The address
+    arguments belong to the ZMQ backend and `broker_url` to the ORBIT one
+    -- each is ignored by the other, and both default to their own
+    resolution.
+
+    The connect is bounded by `timeout` in either case.
     """
 
-    cfg = PubSubConfig.resolve(namespace, pub_addr, sub_addr)
+    name = stream_backend(backend)
+
+    if name == BACKEND_ORBIT:
+        cfg = PubSubConfig(namespace, kind=BACKEND_ORBIT, broker_url=broker_url)
+    else:
+        cfg = PubSubConfig.resolve(namespace, pub_addr, sub_addr)
+
     return await cfg.connect(timeout)
diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py
new file mode 100644
index 0000000..301cee7
--- /dev/null
+++ b/src/digitaltwin/streaming_orbit.py
@@ -0,0 +1,337 @@
+"""The DT data plane on ORBIT eventing (DTaaS plan M3, risk R7).
+
+A second `PubSubBackend` behind the M0 seam.  Nothing above the seam
+changes: `PubSubClient` still hands topics down and gets decoded messages
+back, and a persistent component still publishes through
+`RuntimeAPI.stream`.  What changes is where the bytes travel.
+
+Why this exists
+---------------
+The ZMQ backend's payloads are cloudpickled and its XSUB/XPUB ports carry
+no authentication: anyone who can reach them executes code in every
+subscriber.  This backend moves exactly the same payloads inside ORBIT's
+token-authenticated WebSocket star, so the data plane is no weaker than
+the control plane and the deployment opens no DT-owned ports at all.
+The payloads are *still* cloudpickle -- the win is the trust boundary
+around them, which is the same one that already accepts client-shipped
+component classes (plan risk R4).
+
+How it maps
+-----------
+- **Publish** is `EndpointRuntime.send_notification(plugin, topic,
+  data)`: a plain sync, non-blocking call that emits one `event` frame.
+  The DT topic (`dt/<namespace>/dtypes/<label>`) becomes the ORBIT
+  `topic` verbatim; `plugin` is the fixed `dt_stream` namespace, which
+  keeps DT traffic out of every other subscriber's match set.  The
+  cloudpickled message rides as `bytes` in the event's `data` dict.
+- **Subscribe** is `EndpointRuntime.register_callback(plugin_name=...,
+  topic=...)`, one registration per DT topic.  ORBIT matches topics by
+  exact equality (there are no wildcards), which is exactly what DT
+  needs: the broker then filters on the wire and only matching frames
+  cross the socket.
+- **Connection ownership**: one `EndpointRuntime` per backend instance,
+  i.e. one per twin, created in `connect()` and stopped in `close()` --
+  the same pattern (and the same loopback wrinkle: the service connects
+  to the broker that hosts it) as rhapsody's `OrbitExecutionBackend`.  A
+  runtime can also be injected, in which case the backend uses it and
+  does not own its lifetime.
+
+Semantics
+---------
+At-most-once with per-subscriber bounded drop-oldest queues, broker-side
+and again on the hop into the host loop.  That is the DT conflation
+contract, so nothing above adds a second one.  Losses are observable as
+gaps in the broker-assigned `seq`; the `replay` plugin would give late
+joiners history and is deliberately not integrated in v1.
+"""
+
+import asyncio
+import contextlib
+import logging
+import uuid
+
+from typing import Callable, Optional
+
+import cloudpickle
+
+from radical.orbit import EndpointRuntime  # type: ignore
+from radical.orbit.protocol import FRAME_CAP  # type: ignore
+
+from .streaming import CLIENT_CONNECT_TIMEOUT, PubSubBackend
+
+logger = logging.getLogger(__name__)
+
+# the ORBIT `plugin` field of every DT stream event: one namespace for
+# the whole data plane.  It is a free string on the wire -- a bare
+# participant does not have to host a plugin of that name to use it.
+STREAM_PLUGIN = "dt_stream"
+
+# the cloudpickled message rides under this key of the event data dict
+PAYLOAD_KEY = "p"
+
+# headroom for the msgpack envelope (topic, ids, participant name) on top
+# of the payload.  Generous: the check exists to produce a clear error,
+# not to squeeze the last kilobyte out of a frame.
+FRAME_OVERHEAD = 64 * 1024
+
+# bounded handoff from the runtime's callback thread to the host loop.
+# Drop-oldest, matching the broker's own per-subscriber queues.
+INBOX_SIZE = 1024
+
+
+class OrbitPubSubBackend(PubSubBackend):
+    """A `PubSubBackend` over ORBIT eventing.
+
+    Args:
+        broker_url: the ORBIT broker; `None` uses ORBIT's own resolution
+            (`RADICAL_ORBIT_BROKER_URL`, then `~/.radical/orbit`), as does
+            the token.
+        runtime: an already-started `EndpointRuntime` to ride on.  The
+            backend then does *not* own it and will not stop it.
+        name: the participant name; defaults to a unique one, because two
+            participants may not register under the same name.
+    """
+
+    label = "orbit"
+
+    # names this backend in a PubSubConfig (see PubSubBackend.kind)
+    kind = "orbit"
+
+    def __init__(
+        self,
+        broker_url: Optional[str] = None,
+        runtime: Optional[EndpointRuntime] = None,
+        name: Optional[str] = None,
+    ):
+        super().__init__()
+
+        self.broker_url = broker_url
+        self.name = name or f"{STREAM_PLUGIN}.{uuid.uuid4().hex[:8]}"
+
+        self._runtime = runtime
+        # an injected runtime is somebody else's to stop
+        self._owns_runtime = runtime is None
+
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._inbox: Optional[asyncio.Queue] = None
+        self.dropped = 0
+
+    # -- lifecycle ----------------------------------------------------------
+
+    async def connect(self, timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT):
+        """Bring the participant connection up, bounded by `timeout`.
+
+        A runtime which cannot reach its broker must fail here -- the
+        service turns that into a failed twin.  It must not come up half
+        connected and deliver nothing.
+        """
+
+        self._check_open()
+
+        self._loop = asyncio.get_running_loop()
+        self._inbox = asyncio.Queue(INBOX_SIZE)
+
+        try:
+            if self._owns_runtime:
+                # start() is blocking and joins threads: off the loop
+                self._runtime = await asyncio.to_thread(self._start_runtime, timeout)
+
+        except BaseException:
+            await self.close()
+            raise
+
+        self._start_receiving()
+        self.is_running.set()
+
+    def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime:
+        """Build and register the participant.  Runs in a worker thread."""
+
+        logger.info("connecting stream participant %s", self.name)
+
+        runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name)
+
+        try:
+            runtime.start(wait=True, timeout=timeout)
+
+            # start() returns *silently* when it merely timed out, so the
+            # registration has to be checked explicitly -- otherwise an
+            # unreachable broker would surface much later, as a stream
+            # that never delivers
+            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"
+                )
+
+        except BaseException:
+            # the runtime's threads and its socket are live from start();
+            # a caller that never reaches close() would keep them alive
+            runtime.stop()
+            raise
+
+        return runtime
+
+    async def close(self):
+        """Cancel the receive loop, drop every subscription, stop the
+        participant.  Idempotent."""
+
+        if self._closed:
+            return
+        self._closed = True
+
+        try:
+            await self._cancel_receiving()
+
+        finally:
+            # the guard above makes close() a one-shot, so the connection
+            # has to go even if the await was cancelled
+            runtime, self._runtime = self._runtime, None
+
+            try:
+                for topic in list(self.topics):
+                    self._unregister(runtime, topic)
+                self.topics.clear()
+
+                if runtime is not None and self._owns_runtime:
+                    # blocking: closes the socket and joins three threads
+                    await asyncio.to_thread(runtime.stop)
+
+            finally:
+                self.is_running.clear()
+                self._inbox = None
+                self._loop = None
+
+    # -- pubsub -------------------------------------------------------------
+
+    async def publish(self, topic, message):
+        """Emit one event frame carrying the cloudpickled message.
+
+        Oversized payloads raise: ORBIT drops a frame over the cap with a
+        log line and no exception, and a data plane that silently swallows
+        every large sample is worse than one that refuses it.
+        """
+
+        self._check_open()
+        await self._await_running("publish")
+
+        payload = cloudpickle.dumps(message)
+        cap = self.payload_cap()
+
+        if len(payload) > cap:
+            raise ValueError(
+                f"stream payload for {topic!r} is {len(payload)} bytes, over"
+                f" the {cap} byte ceiling (ORBIT's {self.frame_cap()} byte"
+                f" frame cap, less envelope overhead)"
+            )
+
+        self._runtime.send_notification(STREAM_PLUGIN, topic,
+                                        {PAYLOAD_KEY: payload})
+
+    async def subscribe(self, topic, callback: Callable, **backend_params):
+        """Register `callback` for `topic`, subscribing on first use."""
+
+        self._check_open()
+        await self._await_running("subscribe")
+
+        callbacks = self.topics.setdefault(topic, [])
+        callbacks.append(callback)
+
+        # one ORBIT registration per topic, however many local subscribers
+        # it feeds -- a second one would deliver every frame twice
+        if len(callbacks) == 1:
+            self._runtime.register_callback(
+                plugin_name=STREAM_PLUGIN, topic=topic, callback=self._on_event
+            )
+
+    async def unsubscribe(self, topic):
+        if self._closed:
+            return
+
+        if self.topics.pop(topic, None) is not None:
+            self._unregister(self._runtime, topic)
+
+    def _unregister(self, runtime: Optional[EndpointRuntime], topic: str):
+        if runtime is None:
+            return
+
+        runtime.unregister_callback(
+            plugin_name=STREAM_PLUGIN, topic=topic, callback=self._on_event
+        )
+
+    # -- frame cap ----------------------------------------------------------
+
+    def frame_cap(self) -> int:
+        """The effective ORBIT frame cap of this participant.
+
+        Read off the runtime when there is one: a deployment may tune it,
+        and the module default would then be wrong in either direction.
+        """
+
+        cap = getattr(self._runtime, "_frame_cap", None)
+
+        return cap if isinstance(cap, int) and cap > 0 else FRAME_CAP
+
+    def payload_cap(self) -> int:
+        """The largest cloudpickled message that fits in one frame."""
+
+        return self.frame_cap() - FRAME_OVERHEAD
+
+    # -- receiving ----------------------------------------------------------
+
+    def _on_event(self, endpoint, plugin, topic, data):
+        """ORBIT event callback.  Runs on the runtime's callback thread.
+
+        That thread is shared by every callback of this runtime and its
+        queue is bounded, so this returns immediately: the frame is handed
+        to the host loop and decoded there.
+        """
+
+        loop = self._loop
+
+        if loop is None or loop.is_closed():
+            return
+
+        try:
+            loop.call_soon_threadsafe(self._enqueue, topic, data)
+        except RuntimeError:  # the loop closed under us
+            pass
+
+    def _enqueue(self, topic, data):
+        """Queue one frame for the receive loop.  Runs on the host loop.
+
+        Drop-oldest when the host loop falls behind: the same discipline
+        the broker applies to its own per-subscriber queues, and the
+        conflation semantics the DT data plane is specified with.
+        """
+
+        inbox = self._inbox
+        if inbox is None:
+            return
+
+        if inbox.full():
+            with contextlib.suppress(asyncio.QueueEmpty):
+                inbox.get_nowait()
+
+            self.dropped += 1
+            if self.dropped % INBOX_SIZE == 1:
+                logger.warning(
+                    "stream inbox of %s is full: %d message(s) dropped",
+                    self.name, self.dropped,
+                )
+
+        inbox.put_nowait((topic, data))
+
+    async def _run(self):
+        """Receive loop.  A single bad payload or a raising callback must
+        not take the stream down -- it is dropped and logged."""
+
+        while True:
+            topic, data = await self._inbox.get()
+
+            try:
+                message = cloudpickle.loads(data[PAYLOAD_KEY])
+            except Exception:
+                logger.exception("dropping malformed message on topic %r", topic)
+                continue
+
+            await self._dispatch(topic, message)
diff --git a/test/unit/test_service.py b/test/unit/test_service.py
index 611cbd5..9487d43 100644
--- a/test/unit/test_service.py
+++ b/test/unit/test_service.py
@@ -91,7 +91,8 @@ def test_admin_sessions_reports_policy_and_broker(plugin, client):
     assert entry["engines"] == []
 
     # nothing needed the stream broker yet, so none was started
-    assert listing["stream_broker"] == {"addresses": None, "alive": False}
+    assert listing["stream_broker"] == {
+        "backend": "zmq", "addresses": None, "alive": False}
 
 
 # ---------------------------------------------------------------------------

From 79255445d3c236c9ed745f91aabb6a5648514181 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 19:55:50 +0200
Subject: [PATCH 02/11] Unit tests for the orbit backend, against a loopback
 event router

The stand-in reproduces the three ORBIT behaviours the backend is written
against: exact-match subscriptions, fan-out that does *not* exclude the
sender, and callbacks arriving on a foreign thread.  The first of those
is load-bearing for DT -- a twin's runtime consumes the dtypes its own
persistent components publish, so a backend that never got its own events
back would deliver nothing at all.

Covers namespacing isolation on the second backend (same dtype label, two
twins, one broker), unsubscribe and re-subscribe, one wire subscription
per topic however many local subscribers, the payload ceiling, drop-
oldest under a stalled loop, and teardown: the participant is stopped, an
injected one is not, and a runtime that cannot register is not left
behind.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 test/unit/test_streaming_orbit.py | 494 ++++++++++++++++++++++++++++++
 1 file changed, 494 insertions(+)
 create mode 100644 test/unit/test_streaming_orbit.py

diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py
new file mode 100644
index 0000000..9cc0c50
--- /dev/null
+++ b/test/unit/test_streaming_orbit.py
@@ -0,0 +1,494 @@
+"""M3 -- the ORBIT-backed pubsub backend.
+
+Everything here runs against a loopback stand-in for ORBIT's event
+router: an in-process registry with the two properties the backend
+actually depends on -- events reach *every* matching subscriber
+including the sender (the DT case: a twin's runtime subscribes to the
+dtypes its own components publish), and callbacks arrive on a foreign
+thread, not on the host loop.
+
+The live-broker coverage is in `test/integration/test_dt_orbit_stream.py`.
+"""
+
+import asyncio
+import threading
+
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+pytest.importorskip("radical.orbit")
+
+from digitaltwin import DataType, PubSubClient  # noqa: E402
+from digitaltwin.streaming_orbit import (  # noqa: E402
+    FRAME_OVERHEAD,
+    INBOX_SIZE,
+    PAYLOAD_KEY,
+    STREAM_PLUGIN,
+    OrbitPubSubBackend,
+)
+
+SENSOR = DataType("sensor")
+
+# 64 KiB envelope headroom + 1 KiB of payload: enough to test the ceiling
+# without pickling megabytes
+SMALL_FRAME_CAP = 64 * 1024 + 1024
+
+
+# ---------------------------------------------------------------------------
+# the loopback stand-in
+# ---------------------------------------------------------------------------
+
+class LoopbackBroker:
+    """ORBIT's event router, in process.
+
+    Exact-match subscriptions, fan-out to every matching participant with
+    no exclusion of the sender, and delivery on a dedicated thread -- the
+    three behaviours the backend is written against.
+    """
+
+    def __init__(self):
+        self.subscriptions: list[tuple] = []  # (runtime, plugin, topic)
+        self.published: list[tuple] = []      # (src, plugin, topic, data)
+        self.registered: dict[str, "LoopbackRuntime"] = {}
+
+        # stands in for the runtime's own 'orbit-callbacks' thread
+        self._pool = ThreadPoolExecutor(1, thread_name_prefix="loopback-cb")
+
+    def runtime(self, name: str, registers: bool = True) -> "LoopbackRuntime":
+        return LoopbackRuntime(self, name, registers)
+
+    def publish(self, src, plugin, topic, data) -> None:
+        self.published.append((src, plugin, topic, data))
+
+        for runtime, sub_plugin, sub_topic in list(self.subscriptions):
+            if (sub_plugin, sub_topic) == (plugin, topic):
+                self._pool.submit(runtime.deliver, src, plugin, topic, data)
+
+    def drop(self, runtime) -> None:
+        self.subscriptions = [s for s in self.subscriptions if s[0] is not runtime]
+        self.registered.pop(runtime.name, None)
+
+    def shutdown(self) -> None:
+        self._pool.shutdown(wait=True)
+
+    def topics(self) -> list[str]:
+        return [topic for _, _, topic in self.subscriptions]
+
+
+class LoopbackRuntime:
+    """The slice of `EndpointRuntime` the backend uses."""
+
+    def __init__(self, broker: LoopbackBroker, name: str, registers: bool):
+        self._broker = broker
+        self._registers = registers
+        self._frame_cap = SMALL_FRAME_CAP
+
+        self.name = name
+        self.broker_url = "loopback://broker"
+        self.started = False
+        self.stopped = False
+        self.callbacks: dict[tuple, list] = {}
+
+    # -- lifecycle
+    def start(self, wait=True, timeout=None):
+        self.started = True
+        if self._registers:
+            self._broker.registered[self.name] = self
+
+    def wait_registered(self, timeout=0):
+        return self.name in self._broker.registered
+
+    def stop(self):
+        self.stopped = True
+        self._broker.drop(self)
+
+    # -- eventing
+    def register_callback(self, endpoint_id=None, plugin_name=None, topic=None,
+                          callback=None, with_meta=False):
+        self.callbacks.setdefault((plugin_name, topic), []).append(callback)
+        self._broker.subscriptions.append((self, plugin_name, topic))
+
+    def unregister_callback(self, endpoint_id=None, plugin_name=None,
+                            topic=None, callback=None):
+        key = (plugin_name, topic)
+        self.callbacks[key] = [
+            cb for cb in self.callbacks.get(key, []) if cb is not callback
+        ]
+        entry = (self, plugin_name, topic)
+        if entry in self._broker.subscriptions:
+            self._broker.subscriptions.remove(entry)
+
+    def send_notification(self, plugin_name, topic, data):
+        self._broker.publish(self.name, plugin_name, topic, data)
+
+    def deliver(self, src, plugin, topic, data):
+        """Runs on the broker's callback thread, never on the host loop."""
+
+        for callback in list(self.callbacks.get((plugin, topic), [])):
+            callback(src, plugin, topic, data)
+
+
+# ---------------------------------------------------------------------------
+# fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def loopback():
+    broker = LoopbackBroker()
+    try:
+        yield broker
+    finally:
+        broker.shutdown()
+
+
+@pytest.fixture
+async def orbit_backends(loopback, monkeypatch):
+    """Factory for connected `OrbitPubSubBackend`s on the loopback broker.
+
+    The runtime class is patched, so the backends take the *owned*
+    connection path -- the one a twin uses, and the one whose teardown
+    the leak assertions are about.  Everything handed out is closed with
+    the test.
+    """
+
+    monkeypatch.setattr(
+        "digitaltwin.streaming_orbit.EndpointRuntime",
+        lambda broker_url=None, name=None: loopback.runtime(name),
+    )
+
+    backends = []
+
+    async def make(**kwargs):
+        backend = OrbitPubSubBackend(**kwargs)
+        backends.append(backend)
+        await backend.connect(timeout=5)
+        return backend
+
+    try:
+        yield make
+    finally:
+        for backend in backends:
+            await backend.close()
+
+
+@pytest.fixture
+def orbit_clients(orbit_backends):
+    """Factory for namespaced `PubSubClient`s on the loopback broker."""
+
+    async def make(namespace: str):
+        return PubSubClient(await orbit_backends(), namespace)
+
+    return make
+
+
+async def drain(queue: asyncio.Queue, timeout=5.0):
+    return await asyncio.wait_for(queue.get(), timeout)
+
+
+# ---------------------------------------------------------------------------
+# delivery
+# ---------------------------------------------------------------------------
+
+async def test_a_twin_receives_what_it_publishes(orbit_clients):
+    """The DT case: one client both publishes and subscribes.
+
+    Persistent components publish through the twin's stream client and
+    the twin's own runtime consumes those dtypes -- so the backend must
+    get its own events back, off the broker.
+    """
+
+    twin = await orbit_clients("twin-a")
+    queue: asyncio.Queue = asyncio.Queue()
+
+    await twin.subscribe_to_dtype(SENSOR, queue)
+    await twin.publish(SENSOR, {"value": 42})
+
+    received = await drain(queue)
+    assert received.dtype == SENSOR
+    assert received.data == {"value": 42}
+
+
+async def test_delivery_crosses_from_the_callback_thread(orbit_clients):
+    """Callbacks arrive on ORBIT's thread; subscribers run on the loop."""
+
+    twin = await orbit_clients("twin-a")
+    seen: list = []
+
+    async def record(message):
+        seen.append((message, threading.current_thread()))
+
+    await twin._backend.subscribe(twin.topic(SENSOR), record)
+    await twin.publish(SENSOR, "hello")
+
+    for _ in range(100):
+        if seen:
+            break
+        await asyncio.sleep(0.02)
+
+    assert seen, "message never reached the host loop"
+    assert seen[0] == ("hello", threading.current_thread())
+
+
+async def test_identical_dtype_labels_do_not_cross_subscribe(orbit_clients):
+    """The multi-tenancy target, on the orbit backend: same dtype label,
+    two namespaces, one broker -- no crosstalk."""
+
+    twin_a = await orbit_clients("twin-a")
+    twin_b = await orbit_clients("twin-b")
+
+    queue_a: asyncio.Queue = asyncio.Queue()
+    queue_b: asyncio.Queue = asyncio.Queue()
+
+    await twin_a.subscribe_to_dtype(SENSOR, queue_a)
+    await twin_b.subscribe_to_dtype(SENSOR, queue_b)
+
+    for i in range(3):
+        await twin_a.publish(SENSOR, f"a-{i}")
+
+    assert (await drain(queue_a)).data == "a-0"
+
+    await asyncio.sleep(0.2)
+    assert queue_b.empty()
+
+    # ... and the other direction, so this is isolation and not silence
+    await twin_b.publish(SENSOR, "b-0")
+    assert (await drain(queue_b)).data == "b-0"
+
+
+async def test_unsubscribe_stops_delivery_and_the_subscription(
+    orbit_clients, loopback
+):
+    twin = await orbit_clients("twin-a")
+    queue: asyncio.Queue = asyncio.Queue()
+
+    await twin.subscribe_to_dtype(SENSOR, queue)
+    assert loopback.topics() == [twin.topic(SENSOR)]
+
+    await twin.publish(SENSOR, "first")
+    assert (await drain(queue)).data == "first"
+
+    await twin.unsubscribe_dtype(SENSOR)
+    assert SENSOR not in twin.subscriptions
+    assert loopback.topics() == []
+
+    await twin.publish(SENSOR, "second")
+    await asyncio.sleep(0.2)
+    assert queue.empty()
+
+    # re-subscription is possible after unsubscribe
+    await twin.subscribe_to_dtype(SENSOR, queue)
+    await twin.publish(SENSOR, "third")
+    assert (await drain(queue)).data == "third"
+
+
+async def test_one_wire_subscription_per_topic(orbit_backends, loopback):
+    """Two local subscribers on one topic must not double every frame."""
+
+    backend = await orbit_backends()
+    seen: list = []
+
+    async def record(message):
+        seen.append(message)
+
+    await backend.subscribe("dt/twin-a/dtypes/sensor", record)
+    await backend.subscribe("dt/twin-a/dtypes/sensor", record)
+
+    assert loopback.topics() == ["dt/twin-a/dtypes/sensor"]
+
+    await backend.publish("dt/twin-a/dtypes/sensor", "once")
+
+    for _ in range(100):
+        if len(seen) == 2:
+            break
+        await asyncio.sleep(0.02)
+
+    assert seen == ["once", "once"], "one frame, both subscribers, once each"
+
+
+async def test_events_use_one_plugin_namespace(orbit_backends, loopback):
+    """DT traffic stays under `dt_stream`, so no other subscriber on the
+    broker ever has to match it."""
+
+    backend = await orbit_backends()
+    await backend.publish("dt/twin-a/dtypes/sensor", "x")
+
+    _, plugin, topic, data = loopback.published[0]
+    assert plugin == STREAM_PLUGIN
+    assert topic == "dt/twin-a/dtypes/sensor"
+    assert isinstance(data[PAYLOAD_KEY], bytes)
+
+
+# ---------------------------------------------------------------------------
+# the frame cap
+# ---------------------------------------------------------------------------
+
+async def test_oversized_payload_is_refused_at_publish(orbit_backends):
+    """ORBIT drops an oversized frame with a log line and no exception,
+    so the ceiling has to be enforced here."""
+
+    backend = await orbit_backends()
+    cap = backend.payload_cap()
+    assert cap == SMALL_FRAME_CAP - FRAME_OVERHEAD
+
+    with pytest.raises(ValueError, match="over the .* byte ceiling"):
+        await backend.publish("dt/twin-a/dtypes/sensor", b"x" * (cap + 1))
+
+    # ... and the backend is still usable afterwards
+    await backend.publish("dt/twin-a/dtypes/sensor", "small")
+
+
+async def test_frame_cap_follows_the_runtime(orbit_backends, loopback):
+    """A deployment may tune the cap; the module default would then be
+    wrong in either direction."""
+
+    backend = await orbit_backends()
+    assert backend.frame_cap() == SMALL_FRAME_CAP
+
+    backend._runtime._frame_cap = 8 * 1024 * 1024
+    assert backend.frame_cap() == 8 * 1024 * 1024
+
+
+# ---------------------------------------------------------------------------
+# resilience
+# ---------------------------------------------------------------------------
+
+async def test_receive_loop_survives_bad_payloads_and_callbacks(orbit_backends):
+    backend = await orbit_backends()
+    queue: asyncio.Queue = asyncio.Queue()
+
+    async def raises(message):
+        raise ValueError("bad callback")
+
+    async def record(message):
+        await queue.put(message)
+
+    topic = "dt/twin-a/dtypes/sensor"
+    await backend.subscribe(topic, raises)
+    await backend.subscribe(topic, record)
+
+    # a frame with no payload where one belongs
+    backend._enqueue(topic, {"nothing": "here"})
+
+    # the raising callback fires on this one, the queue still gets it
+    await backend.publish(topic, "good")
+
+    assert await drain(queue) == "good"
+    assert not backend._task.done()
+
+
+async def test_inbox_drops_the_oldest_when_the_loop_falls_behind(orbit_backends):
+    """Bounded drop-oldest, like the broker's own subscriber queues -- a
+    stalled host loop must cost messages, not memory."""
+
+    backend = await orbit_backends()
+    topic = "dt/twin-a/dtypes/sensor"
+
+    for i in range(INBOX_SIZE + 10):
+        backend._enqueue(topic, {PAYLOAD_KEY: str(i).encode()})
+
+    assert backend._inbox.qsize() == INBOX_SIZE
+    assert backend.dropped == 10
+
+    # the oldest went, the newest stayed
+    _, first = backend._inbox.get_nowait()
+    assert first[PAYLOAD_KEY] == b"10"
+
+
+async def test_unexpected_receive_loop_exit_is_reported(orbit_backends):
+    backend = await orbit_backends()
+    seen: list[BaseException] = []
+
+    backend.on_error = seen.append
+
+    ended = asyncio.create_task(asyncio.sleep(0))
+    await ended
+    backend._run_done(ended)
+
+    assert len(seen) == 1
+    assert isinstance(seen[0], RuntimeError)
+
+    # ... but the loop ending as part of close() is not an error
+    await backend.close()
+    backend._run_done(ended)
+    assert len(seen) == 1
+
+
+# ---------------------------------------------------------------------------
+# connection ownership and teardown
+# ---------------------------------------------------------------------------
+
+async def test_close_stops_the_participant_and_leaks_nothing(
+    orbit_backends, loopback, no_task_leaks
+):
+    backend = await orbit_backends()
+    runtime = backend._runtime
+
+    await backend.subscribe("dt/twin-a/dtypes/sensor", lambda m: None)
+    await backend.close()
+
+    assert runtime.stopped
+    assert backend.topics == {}
+    assert loopback.subscriptions == []
+    assert backend._task is None
+    assert backend._inbox is None
+
+    # idempotent, and a closed backend refuses further use
+    await backend.close()
+    with pytest.raises(RuntimeError):
+        await backend.publish("dt/twin-a/dtypes/sensor", "nope")
+
+
+async def test_an_injected_runtime_is_not_stopped(loopback, no_task_leaks):
+    """A shared connection is the caller's to end, not the backend's."""
+
+    runtime = loopback.runtime("shared")
+    runtime.start()
+
+    backend = OrbitPubSubBackend(runtime=runtime)
+    await backend.connect(timeout=5)
+
+    await backend.subscribe("dt/twin-a/dtypes/sensor", lambda m: None)
+    await backend.close()
+
+    assert not runtime.stopped
+    # but its subscriptions are gone -- they were the backend's
+    assert loopback.subscriptions == []
+
+
+async def test_a_runtime_which_cannot_register_is_not_left_behind(
+    loopback, monkeypatch, no_task_leaks
+):
+    """An unreachable broker must fail `connect`, not come up half
+    connected and deliver nothing -- and must not leak its threads."""
+
+    made = []
+
+    def build(broker_url=None, name=None):
+        runtime = loopback.runtime(name, registers=False)
+        made.append(runtime)
+        return runtime
+
+    monkeypatch.setattr("digitaltwin.streaming_orbit.EndpointRuntime", build)
+
+    backend = OrbitPubSubBackend()
+
+    with pytest.raises(TimeoutError, match="did not register"):
+        await backend.connect(timeout=0.1)
+
+    assert made[0].stopped
+    assert backend._runtime is None
+    assert backend._task is None
+
+
+async def test_participant_names_are_unique():
+    """Two participants may not register under one name."""
+
+    first = OrbitPubSubBackend()
+    second = OrbitPubSubBackend()
+
+    assert first.name != second.name
+    assert first.name.startswith(f"{STREAM_PLUGIN}.")
+
+    # ... unless the caller insists
+    assert OrbitPubSubBackend(name="fixed").name == "fixed"

From c190403bca2bc62cab5e979e332d6debe805730f Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 20:02:40 +0200
Subject: [PATCH 03/11] A live stack whose data plane is ORBIT, and no ZMQ
 broker in it

A second deployment on the next port, started with
DT_STREAM_BACKEND=orbit -- separate because the transport is a
deployment-time choice and no client can ask for it.

The tests run whole twins on it: sensor to investigator to sink, the M2
learner retraining off the twin's input stream, two twins with the same
dtype label staying apart, inference unaffected, churn leaking no
participants.  The subscriber they attach is itself an ORBIT
participant, so even the observation path opens no port -- where the ZMQ
suite has to go and ask the service for addresses.

The R7 assertion is the direct one: the embedded stream broker is a
spawned subprocess, so "none was started" is "the plugin host has no
children", checked after a twin has actually streamed.  Plus the
oversized-payload refusal, which is the one place the 4 MiB frame cap is
visible to a user.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 test/integration/conftest.py             | 119 ++++++++--
 test/integration/test_dt_orbit_stream.py | 288 +++++++++++++++++++++++
 2 files changed, 389 insertions(+), 18 deletions(-)
 create mode 100644 test/integration/test_dt_orbit_stream.py

diff --git a/test/integration/conftest.py b/test/integration/conftest.py
index e9844fa..cef283c 100644
--- a/test/integration/conftest.py
+++ b/test/integration/conftest.py
@@ -10,6 +10,12 @@
   is where the `'exsitu'` engine sends learner tasks,
 - a consumer runtime the tests get `DTClient`s from.
 
+Plus a *second, independent* stack on the next port whose `dt` plugin
+runs with `DT_STREAM_BACKEND=orbit` (M3): same shape, but the twins'
+streams ride ORBIT eventing and no ZMQ broker is started anywhere.  It is
+a separate deployment because the backend is a deployment-time choice --
+which is exactly what the M3 tests assert.
+
 Every endpoint carries a `DT_TEST_ENDPOINT_TAG` in its environment, so a
 task can report where it ran -- which is how the dual-engine tests prove
 that learning and inference really landed on different endpoints.
@@ -46,10 +52,15 @@
 BROKER_PORT = int(os.environ.get("DT_TEST_BROKER_PORT", "8031"))
 BROKER_URL = f"https://{BROKER_HOST}:{BROKER_PORT}"
 
+# the M3 stack: a `dt` deployment whose data plane is ORBIT eventing
+ORBIT_BROKER_PORT = BROKER_PORT + 1
+ORBIT_BROKER_URL = f"https://{BROKER_HOST}:{ORBIT_BROKER_PORT}"
+
 TASK_ENDPOINT = "dt_test_task_ep"
 EXSITU_ENDPOINT = "dt_test_exsitu_ep"  # stands in for remote HPC hardware
 DOOMED_ENDPOINT = "dt_test_doomed_ep"  # started to be killed (R8)
 DT_ENDPOINT = "dt_test_dt_ep"  # endpoint-hosted `dt`, for the smoke test
+ORBIT_TASK_ENDPOINT = "dt_test_orbit_task_ep"  # on the M3 stack
 
 STARTUP_TIMEOUT = 60.0
 LOGS = Path(os.environ.get("DT_TEST_LOG_DIR", "/tmp")) / "dt-integration-logs"
@@ -75,6 +86,9 @@ def engines(**endpoints: str) -> dict:
 # same, but with an ex-situ engine on an endpoint the test will kill
 ENGINES_DOOMED = engines(task=TASK_ENDPOINT, exsitu=DOOMED_ENDPOINT)
 
+# the M3 stack's single engine
+ENGINES_ORBIT = engines(task=ORBIT_TASK_ENDPOINT)
+
 
 def _port_free(port: int) -> bool:
     with socket.socket() as sock:
@@ -86,7 +100,9 @@ def _child_env(**extra: str) -> dict:
     """Environment for a broker / endpoint child process.
 
     `src` goes first on PYTHONPATH so the children run the working tree,
-    not whatever `digitaltwin` happens to be installed.
+    not whatever `digitaltwin` happens to be installed.  Anything in
+    `extra` wins -- that is how the M3 children get their own broker URL
+    and `DT_STREAM_BACKEND`.
     """
 
     env = dict(os.environ)
@@ -168,41 +184,63 @@ class Broker(NamedTuple):
     pid: int
 
 
-@pytest.fixture(scope="session")
-def broker():
-    """An ORBIT broker on a non-default loopback port, hosting `dt`."""
+@contextlib.contextmanager
+def _dt_broker(label: str, port: int, url: str, **env: str):
+    """An ORBIT broker on a loopback port, hosting `dt`, torn down on exit."""
 
-    if not _port_free(BROKER_PORT):
-        pytest.skip(f"port {BROKER_PORT} is busy")
+    if not _port_free(port):
+        pytest.skip(f"port {port} is busy")
 
     argv = _orbit_script("radical-orbit-broker.py") + [
         "--host", BROKER_HOST,
-        "--port", str(BROKER_PORT),
+        "--port", str(port),
         "--plugins", "default,dt",
     ]
-    proc = _spawn("broker", argv)
+    proc = _spawn(label, argv, RADICAL_ORBIT_BROKER_URL=url, **env)
 
     try:
-        _await_broker(proc)
-        yield Broker(BROKER_URL, proc.pid)
+        _await_broker(label, url, proc)
+        yield Broker(url, proc.pid)
     finally:
         _terminate(proc)
 
 
-def _await_broker(proc: subprocess.Popen) -> None:
+@pytest.fixture(scope="session")
+def broker():
+    """An ORBIT broker on a non-default loopback port, hosting `dt`."""
+
+    with _dt_broker("broker", BROKER_PORT, BROKER_URL) as running:
+        yield running
+
+
+@pytest.fixture(scope="session")
+def orbit_broker():
+    """The M3 deployment: the same broker with the ORBIT data plane.
+
+    `DT_STREAM_BACKEND=orbit` is set on the broker process, because the
+    transport is a property of the deployment -- no client and no session
+    can ask for it.
+    """
+
+    with _dt_broker("orbit-broker", ORBIT_BROKER_PORT, ORBIT_BROKER_URL,
+                    DT_STREAM_BACKEND="orbit") as running:
+        yield running
+
+
+def _await_broker(label: str, url: str, proc: subprocess.Popen) -> None:
     deadline = time.time() + STARTUP_TIMEOUT
 
     while time.time() < deadline:
         if proc.poll() is not None:
-            pytest.skip(f"broker exited early -- see {LOGS / 'broker.log'}")
+            pytest.skip(f"{label} exited early -- see {LOGS / label}.log")
         try:
-            httpx.get(BROKER_URL + "/", verify=False, timeout=2)
+            httpx.get(url + "/", verify=False, timeout=2)
             return
         except Exception:
             time.sleep(0.25)
 
     _terminate(proc)
-    pytest.skip(f"broker did not come up -- see {LOGS / 'broker.log'}")
+    pytest.skip(f"{label} did not come up -- see {LOGS / label}.log")
 
 
 @contextlib.contextmanager
@@ -220,13 +258,14 @@ def _rhapsody_endpoint(name: str, broker, **env: str):
     proc = _spawn(
         name,
         argv,
+        RADICAL_ORBIT_BROKER_URL=broker.url,
         RADICAL_ORBIT_RHAPSODY_BACKEND="concurrent",
         DT_TEST_ENDPOINT_TAG=name,
         **env,
     )
 
     try:
-        _await_plugin(name, "rhapsody", proc)
+        _await_plugin(name, "rhapsody", proc, broker.url)
         yield proc
     finally:
         _terminate(proc)
@@ -283,16 +322,17 @@ def dt_endpoint(broker):
     proc = _spawn(DT_ENDPOINT, argv)
 
     try:
-        _await_plugin(DT_ENDPOINT, "dt", proc)
+        _await_plugin(DT_ENDPOINT, "dt", proc, BROKER_URL)
         yield DT_ENDPOINT
     finally:
         _terminate(proc)
 
 
-def _await_plugin(endpoint: str, plugin: str, proc: subprocess.Popen) -> None:
+def _await_plugin(endpoint: str, plugin: str, proc: subprocess.Popen,
+                  broker_url: str = BROKER_URL) -> None:
     """Wait until `endpoint` advertises `plugin` in the broker topology."""
 
-    runtime = EndpointRuntime(broker_url=BROKER_URL)
+    runtime = EndpointRuntime(broker_url=broker_url)
     runtime.start(wait=True)
 
     try:
@@ -385,3 +425,46 @@ def twin_id():
     """A fresh client-supplied twin uuid."""
 
     return str(uuid.uuid4())
+
+
+# ---------------------------------------------------------------------------
+# the M3 stack: the same shape, with the ORBIT data plane
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(scope="session")
+def orbit_task_endpoint(orbit_broker):
+    """The M3 stack's co-located rhapsody endpoint."""
+
+    with _rhapsody_endpoint(ORBIT_TASK_ENDPOINT, orbit_broker,
+                            RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW="0"):
+        yield ORBIT_TASK_ENDPOINT
+
+
+@pytest.fixture(scope="session")
+def orbit_stack(orbit_broker, orbit_task_endpoint):
+    """The full M3 stack; returns its broker URL."""
+
+    return orbit_broker.url
+
+
+@pytest.fixture
+def orbit_runtime(orbit_stack):
+    """A consumer runtime on the M3 broker -- one per test."""
+
+    rt = EndpointRuntime(broker_url=orbit_stack)
+    rt.start(wait=True)
+    try:
+        yield rt
+    finally:
+        rt.stop()
+
+
+@pytest.fixture
+def orbit_dt(orbit_runtime):
+    """A `DTClient` on a fresh session of the M3 stack."""
+
+    client = orbit_runtime.get_plugin("broker", "dt", config=ENGINES_ORBIT)
+    try:
+        yield client
+    finally:
+        _drop_session(client)
diff --git a/test/integration/test_dt_orbit_stream.py b/test/integration/test_dt_orbit_stream.py
new file mode 100644
index 0000000..46bec5a
--- /dev/null
+++ b/test/integration/test_dt_orbit_stream.py
@@ -0,0 +1,288 @@
+"""Integration tests for M3: the DT data plane on ORBIT eventing.
+
+Against a `dt` deployment started with `DT_STREAM_BACKEND=orbit`: full
+twins run, streams flow, twins tear down -- and no ZMQ broker exists
+anywhere, which is the point of the milestone (plan risk R7).  The
+subscriber the tests attach is itself an ORBIT participant, so even the
+observation path opens no port.
+"""
+
+import asyncio
+import os
+import uuid
+
+from pathlib import Path
+
+import pytest
+
+from digitaltwin.components import TRUTHY, TypedData
+from digitaltwin.config import BACKEND_ORBIT
+from digitaltwin.service import register_user_modules
+from digitaltwin.streaming import connect_stream_client
+
+import learner_components
+import twin_components
+
+from conftest import ORBIT_BROKER_URL
+from test_dt_learner import await_learned, build_learner_twin, infer
+from test_dt_service import await_state, broker_fds, build_pipeline
+from twin_components import (
+    ECHO_DTYPE,
+    INFERENCE_DTYPE,
+    SENSOR_DTYPE,
+    OffsetModel,
+)
+
+pytestmark = pytest.mark.integration
+
+register_user_modules([learner_components, twin_components])
+
+COLLECT_TIMEOUT = 60.0
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+async def collect(twin, dtype, count, timeout=COLLECT_TIMEOUT):
+    """Subscribe to one twin's stream and collect `count` messages.
+
+    The subscriber joins the ORBIT star as an ordinary participant: no
+    addresses, no ports, the same token as every other call.
+    """
+
+    client = await connect_stream_client(
+        twin, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL
+    )
+    queue: asyncio.Queue = asyncio.Queue()
+
+    try:
+        await client.subscribe_to_dtype(dtype, queue)
+        return [
+            (await asyncio.wait_for(queue.get(), timeout)).data
+            for _ in range(count)
+        ]
+    finally:
+        await client.close()
+
+
+def child_pids(pid: int) -> list:
+    """The child processes of `pid` (Linux only).
+
+    An embedded DT stream broker is a spawned subprocess of the plugin
+    host, so "no ZMQ broker was started" is literally "this process has
+    no children".
+    """
+
+    if not Path("/proc/self/status").exists():
+        pytest.skip("no /proc: cannot enumerate child processes")
+
+    children = []
+
+    for entry in os.listdir("/proc"):
+        if not entry.isdigit():
+            continue
+        try:
+            status = Path(f"/proc/{entry}/status").read_text()
+        except OSError:  # it exited under us
+            continue
+
+        for line in status.splitlines():
+            if line.startswith("PPid:") and int(line.split()[1]) == pid:
+                children.append(int(entry))
+                break
+
+    return children
+
+
+# ---------------------------------------------------------------------------
+# a full twin, no ZMQ anywhere
+# ---------------------------------------------------------------------------
+
+def test_a_full_twin_runs_on_the_orbit_data_plane(orbit_dt, twin_id):
+    """sensor -> investigator -> sink, with every stream hop an event."""
+
+    orbit_dt.create_twin(twin_id)
+    build_pipeline(orbit_dt, twin_id, offset=500)
+    assert orbit_dt.start(twin_id) == "running"
+
+    seen = asyncio.run(collect(twin_id, ECHO_DTYPE, 3))
+    assert len(seen) == 3
+    assert all(500 <= value < 500 + 100_000 for value in seen), seen
+
+    # the in-situ inference path is unaffected by the data plane choice
+    answer = orbit_dt.get_inference(twin_id, TypedData(SENSOR_DTYPE, 5),
+                                    INFERENCE_DTYPE, timeout=120)
+    assert answer.data == 505
+
+    assert orbit_dt.twin_close(twin_id) == "closed"
+    assert orbit_dt.twin_list() == []
+
+
+def test_no_zmq_stream_broker_is_started(orbit_dt, orbit_broker, twin_id):
+    """R7 closed: the deployment owns no stream ports at all.
+
+    The embedded ZMQ broker is a spawned subprocess of the plugin host,
+    so the assertion is that the host has no children -- after a twin has
+    actually run, which is when the ZMQ deployment would have started one.
+    """
+
+    orbit_dt.create_twin(twin_id)
+    build_pipeline(orbit_dt, twin_id)
+    orbit_dt.start(twin_id)
+
+    assert asyncio.run(collect(twin_id, ECHO_DTYPE, 1))
+
+    listing = orbit_dt.admin_sessions()
+    assert listing["stream_broker"] == {"backend": "orbit"}
+
+    assert child_pids(orbit_broker.pid) == [], (
+        "the orbit data plane must not spawn a stream broker")
+
+    orbit_dt.twin_close(twin_id)
+
+
+def test_identical_dtype_labels_do_not_cross_subscribe(orbit_dt):
+    """Namespacing isolation on the second backend: two twins, one label,
+    one broker -- and now one shared event stream."""
+
+    first, second = str(uuid.uuid4()), str(uuid.uuid4())
+    bands = {first: 1_000_000, second: 2_000_000}
+
+    for twin, offset in bands.items():
+        orbit_dt.create_twin(twin)
+        build_pipeline(orbit_dt, twin, offset=offset)
+        orbit_dt.start(twin)
+
+    seen_first = asyncio.run(collect(first, ECHO_DTYPE, 3))
+    seen_second = asyncio.run(collect(second, ECHO_DTYPE, 3))
+
+    def in_band(values, base):
+        return all(base <= v < base + 100_000 for v in values)
+
+    assert in_band(seen_first, bands[first]), seen_first
+    assert in_band(seen_second, bands[second]), seen_second
+
+    # closing one leaves the other producing
+    assert orbit_dt.twin_close(first) == "closed"
+    assert asyncio.run(collect(second, ECHO_DTYPE, 2))
+    assert orbit_dt.twin_close(second) == "closed"
+
+
+# ---------------------------------------------------------------------------
+# the M2 learner path, on the new data plane
+# ---------------------------------------------------------------------------
+
+def test_a_learner_twin_learns_off_the_orbit_stream(orbit_dt, twin_id):
+    """The ex-situ learner is fed from the twin's input stream, so it is
+    the component that depends most on the data plane.
+
+    One engine here: `'exsitu'` aliases `'task'` when it is not
+    configured, and which endpoint the learner tasks land on is M2's
+    question, not this one.
+    """
+
+    build_learner_twin(orbit_dt, twin_id)
+
+    bootstrap = infer(orbit_dt, twin_id)
+    learned = await_learned(orbit_dt, twin_id)
+
+    assert learned != bootstrap, (bootstrap, learned)
+    assert orbit_dt.twin_close(twin_id) == "closed"
+
+
+# ---------------------------------------------------------------------------
+# teardown
+# ---------------------------------------------------------------------------
+
+def test_twin_churn_leaks_no_participants(orbit_dt, orbit_broker):
+    """Every twin brings its own participant connection up and takes it
+    down again -- churn must not accumulate sockets or threads."""
+
+    def cycle():
+        twin = str(uuid.uuid4())
+        orbit_dt.create_twin(twin)
+        build_pipeline(orbit_dt, twin)
+        orbit_dt.start(twin)
+        await_state(orbit_dt, twin, "running")
+        orbit_dt.twin_close(twin)
+
+    cycle()  # the first cycle pays the engine startup
+    before = broker_fds(orbit_broker.pid)
+
+    for _ in range(4):
+        cycle()
+
+    assert orbit_dt.twin_list() == []
+    assert child_pids(orbit_broker.pid) == []
+
+    after = broker_fds(orbit_broker.pid)
+    assert after <= before + 8, f"broker fds {before} -> {after}"
+
+
+def test_an_oversized_payload_is_a_clear_error(orbit_dt, twin_id):
+    """The 4 MiB frame cap surfaces as a refusal, not as a stream that
+    quietly drops every large sample."""
+
+    orbit_dt.create_twin(twin_id)
+    orbit_dt.add_investigator(twin_id, orbit_dt.package(OffsetModel, offset=1),
+                              SENSOR_DTYPE, INFERENCE_DTYPE)
+    orbit_dt.start(twin_id)
+
+    async def publish_a_huge_sample():
+        client = await connect_stream_client(
+            twin_id, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL
+        )
+        try:
+            with pytest.raises(ValueError, match="byte ceiling"):
+                await client.publish(SENSOR_DTYPE,
+                                     b"x" * (client._backend.payload_cap() + 1))
+
+            # ... and a payload that fits still goes through
+            await client.publish(SENSOR_DTYPE, 1)
+        finally:
+            await client.close()
+
+    asyncio.run(publish_a_huge_sample())
+
+    # the twin survived the refusal and is still serving
+    assert orbit_dt.twin(twin_id)["state"] == "running"
+    assert orbit_dt.get_inference(twin_id, TypedData(SENSOR_DTYPE, 1),
+                                  INFERENCE_DTYPE, timeout=120).data == 2
+
+
+def test_a_twin_whose_stream_is_a_participant_reports_the_backend(orbit_dt):
+    """The admin listing names the data plane -- an operator must be able
+    to tell which one a deployment is running."""
+
+    assert orbit_dt.admin_sessions()["stream_broker"]["backend"] == BACKEND_ORBIT
+
+
+def test_the_endpoint_hosted_dt_smoke_still_holds(orbit_dt, twin_id):
+    """A twin that never streams must still come up and go away cleanly
+    on the orbit backend (the stream client is built regardless)."""
+
+    orbit_dt.create_twin(twin_id)
+    orbit_dt.add_investigator(twin_id, orbit_dt.package(OffsetModel, offset=1),
+                              SENSOR_DTYPE, INFERENCE_DTYPE)
+    assert orbit_dt.start(twin_id) == "running"
+    assert orbit_dt.twin_close(twin_id) == "closed"
+    assert orbit_dt.twin_list() == []
+
+
+def test_a_persistent_component_publishes_through_the_injected_client(
+    orbit_dt, twin_id
+):
+    """The persistent-component contract is untouched by M3: the twin's
+    sensor publishes through `RuntimeAPI.stream`, knowing nothing about
+    the transport underneath it."""
+
+    orbit_dt.create_twin(twin_id)
+    orbit_dt.add_task(twin_id, orbit_dt.package(twin_components.CountingSensor),
+                      TRUTHY, SENSOR_DTYPE, is_persistent=True)
+    orbit_dt.start(twin_id)
+
+    seen = asyncio.run(collect(twin_id, SENSOR_DTYPE, 3))
+    assert seen == sorted(seen), seen
+
+    orbit_dt.twin_close(twin_id)

From 1e013d82c5b115c1ca43b6a151109cf1f639c224 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 20:15:36 +0200
Subject: [PATCH 04/11] What the token domain costs a stream hop

`bench_streams.py` times publish -> deliver through the same
`PubSubClient` on both backends, so the difference is the transport and
nothing else.  Loopback: 0.84 ms p50 on ZMQ, 1.94 ms on ORBIT, and about
a quarter of the burst throughput.

Roughly a millisecond per hop for the security property -- noise against
the ~20 ms of a single in-situ prediction.  Numbers in the perf README,
informational, not a gate.

Neither backend acknowledges a subscription, so the harness opens with a
round trip it retries until one comes back; without that barrier the
first publishes are simply lost and the measurement never starts.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 perf/README.md        |  34 +++++++++
 perf/bench_streams.py | 168 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 202 insertions(+)
 create mode 100644 perf/bench_streams.py

diff --git a/perf/README.md b/perf/README.md
index 3469e77..67bd01d 100644
--- a/perf/README.md
+++ b/perf/README.md
@@ -29,6 +29,40 @@ Routing all user compute through the Rhapsody abstraction therefore
 costs single-digit milliseconds per sequential prediction and wins by an
 order of magnitude under concurrency.
 
+## `bench_streams.py` — stream latency, ZMQ vs ORBIT data plane
+
+One publish awaited until the subscriber's queue hands it back: the shape
+of every hop in a twin's graph.  Both rows go through the same
+`PubSubClient`, so the only difference is the backend (M3).
+
+```sh
+# the zmq row starts its own embedded broker; the orbit row needs a live one
+python perf/bench_streams.py both --broker https://127.0.0.1:8031
+```
+
+Loopback, one host, bare-int payloads (2026-08-15):
+
+| data plane                  | p50     | p99     | burst          |
+|-----------------------------|---------|---------|----------------|
+| zmq, embedded broker        | 0.84 ms | 1.14 ms | 20 400 msg/s   |
+| orbit eventing              | 1.94 ms | 2.47 ms |  5 300 msg/s   |
+
+With 64 KiB payloads (`--payload 65536`): 1.04 ms / 3.04 ms p50, and
+6 800 vs 1 100 msg/s in burst.
+
+So the ORBIT data plane costs roughly **1 ms per stream hop** and about a
+quarter of the burst throughput, in exchange for the security property of
+M3: the payloads ride the token-authenticated WebSocket star and the
+deployment opens no unauthenticated ports (risk R7).  Against the ~20 ms
+of a single in-situ prediction (the row above), that is noise.
+Informational, not a gate.
+
+The extra hop is structural: ZMQ's XSUB/XPUB proxy forwards a frame
+between two sockets, while an ORBIT event is packed, sent to the broker,
+stamped with a `seq`, fanned out, and handed across a thread boundary
+into the host loop.  Payload size hurts the ORBIT row more because the
+frame is msgpacked around the pickle.
+
 ## `streaming_learner_perf.py`, `plot_streaming_perf.py`
 
 Throughput of the streaming active learner (ROSE); unrelated to the
diff --git a/perf/bench_streams.py b/perf/bench_streams.py
new file mode 100644
index 0000000..1550988
--- /dev/null
+++ b/perf/bench_streams.py
@@ -0,0 +1,168 @@
+"""Stream latency: the ZMQ data plane vs the ORBIT one.
+
+One publish, awaited until the subscriber's queue hands it back -- the
+shape of every hop in a twin's graph (a persistent component publishes a
+dtype, the runtime consumes it).  Both rows go through the *same*
+`PubSubClient`, so the difference is the backend and nothing else.
+
+The ZMQ row starts its own embedded broker on a random loopback port,
+exactly as the service does.  The ORBIT row needs a live broker::
+
+    # against the integration stack (test/integration/conftest.py has one)
+    python perf/bench_streams.py both --broker https://127.0.0.1:8031
+
+Informational, not a gate: the ORBIT row buys the security property of
+milestone M3 (no unauthenticated ports) and pays a WebSocket round trip
+through the broker for it.
+"""
+
+import argparse
+import asyncio
+import os
+import sys
+import time
+
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from bench_insitu import report  # noqa: E402
+
+from digitaltwin import DataType, ZMQ_BrokerProcess  # noqa: E402
+from digitaltwin.config import BACKEND_ORBIT, BACKEND_ZMQ  # noqa: E402
+from digitaltwin.streaming import connect_stream_client  # noqa: E402
+
+SAMPLE = DataType("sample")
+
+N_WARM = 20
+N_MEAS = 200
+N_BURST = 200
+
+
+SETTLE_TIMEOUT = 30.0
+DELIVER_TIMEOUT = 10.0
+
+
+async def settle(client, queue, message) -> None:
+    """Publish until something comes back.
+
+    Neither backend acknowledges a subscription -- ZMQ's SUBSCRIBE and
+    ORBIT's `subscribe` frame are both fire-and-forget -- so the only
+    honest barrier is a message that made the round trip.
+    """
+
+    deadline = time.perf_counter() + SETTLE_TIMEOUT
+
+    while perf_left(deadline):
+        await client.publish(SAMPLE, message)
+        try:
+            await asyncio.wait_for(queue.get(), 0.25)
+            break
+        except TimeoutError:
+            continue
+    else:
+        raise TimeoutError(f"no message came back within {SETTLE_TIMEOUT}s")
+
+    while not queue.empty():  # drop whatever the barrier left behind
+        queue.get_nowait()
+
+
+def perf_left(deadline: float) -> bool:
+    return time.perf_counter() < deadline
+
+
+async def measure(client, label: str, n_meas: int, n_burst: int,
+                  payload: int) -> None:
+    """Publish -> deliver round trips through one stream client."""
+
+    queue: asyncio.Queue = asyncio.Queue()
+    await client.subscribe_to_dtype(SAMPLE, queue)
+
+    message = b"x" * payload if payload else 0
+
+    await settle(client, queue, message)
+
+    for _ in range(N_WARM):
+        await client.publish(SAMPLE, message)
+        await asyncio.wait_for(queue.get(), DELIVER_TIMEOUT)
+
+    latencies = []
+    for _ in range(n_meas):
+        start = time.perf_counter()
+        await client.publish(SAMPLE, message)
+        await asyncio.wait_for(queue.get(), DELIVER_TIMEOUT)
+        latencies.append(time.perf_counter() - start)
+
+    report(label, latencies)
+
+    # burst: how fast the data plane drains a producer that does not wait.
+    # Both backends drop the oldest when a queue overruns, so the count
+    # that arrives is part of the measurement.
+    start = time.perf_counter()
+    for _ in range(n_burst):
+        await client.publish(SAMPLE, message)
+
+    received = 0
+    while received < n_burst:
+        try:
+            await asyncio.wait_for(queue.get(), 5.0)
+        except TimeoutError:
+            break
+        received += 1
+
+    elapsed = time.perf_counter() - start
+    print(f"{label:28s} {received}/{n_burst} burst messages in "
+          f"{elapsed * 1000:.1f}ms ({received / elapsed:.0f} msg/s)")
+
+
+async def bench_zmq(args) -> None:
+    broker = ZMQ_BrokerProcess()
+    await broker.start()
+
+    try:
+        client = await connect_stream_client(
+            "bench-zmq", *broker.get_connection_str(), backend=BACKEND_ZMQ
+        )
+        try:
+            await measure(client, "zmq (embedded broker)", args.messages,
+                          args.burst, args.payload)
+        finally:
+            await client.close()
+    finally:
+        await broker.stop()
+
+
+async def bench_orbit(args) -> None:
+    client = await connect_stream_client(
+        "bench-orbit", backend=BACKEND_ORBIT, broker_url=args.broker
+    )
+    try:
+        await measure(client, "orbit eventing", args.messages, args.burst,
+                      args.payload)
+    finally:
+        await client.close()
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("which", nargs="?", default="both",
+                        choices=["zmq", "orbit", "both"])
+    parser.add_argument("--broker", default=os.environ.get(
+        "RADICAL_ORBIT_BROKER_URL"),
+        help="ORBIT broker URL (default: ORBIT's own resolution)")
+    parser.add_argument("--messages", type=int, default=N_MEAS)
+    parser.add_argument("--burst", type=int, default=N_BURST)
+    parser.add_argument("--payload", type=int, default=0,
+                        help="payload size in bytes (0: a bare int)")
+
+    args = parser.parse_args()
+
+    if args.which in ("zmq", "both"):
+        await bench_zmq(args)
+
+    if args.which in ("orbit", "both"):
+        await bench_orbit(args)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

From c0f3de15fc355a0db8bf7b63d94dabe467648818 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 20:16:33 +0200
Subject: [PATCH 05/11] Document the second data plane, and what it closes

The framework's own pubsub ports authenticate nobody, so the data plane
was weaker than the control plane wrapped around it: reaching those ports
was code execution in every subscriber, token-free (risk R7).
`DT_STREAM_BACKEND=orbit` puts the same payloads on the token-
authenticated star instead, and starts no ZMQ broker at all -- there is
then nothing left to firewall.

Says so plainly, including the two things it does not do (per-tenant auth
is still post-v1; the 4 MiB frame cap is new and the ZMQ backend has
none), how a reviewer checks the guarantee, and that the ZMQ mitigations
remain in force for the deployments that keep it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md | 104 ++++++++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 90 insertions(+), 14 deletions(-)

diff --git a/README.md b/README.md
index 9acc2b0..9bb12ee 100644
--- a/README.md
+++ b/README.md
@@ -5,7 +5,7 @@ Currently implemented:
 - Utility Tasks
 - Persistent Tasks
 - Callbacks
-- Simple ZMQ pubsub backend
+- Two pubsub backends: ZMQ, and ORBIT eventing
 - Graph builder
 - Convert to a Python Package
 - Several Tests / Examples
@@ -71,7 +71,57 @@ are cloudpickled, so anyone who can reach the broker ports can execute code
 in every subscriber.  A non-loopback bind needs an explicit configuration
 and a private/firewalled network.  External channels are decoded with the
 codec their binding names: `json` (the default) and `raw` are safe to
-accept from a producer you do not control, `cloudpickle` is not.
+accept from a producer you do not control, `cloudpickle` is not.  The
+demos are the reason the ZMQ backend exists; anything beyond a laptop
+should be on the ORBIT one below.
+
+## Choosing a data plane
+
+`DT_STREAM_BACKEND` picks which transport carries the twins' streams.  It
+is a **deployment-time** choice, resolved once where the framework runs;
+no client and no session can ask for a different one.  Nothing above
+`PubSubBackend` -- not `DTRuntime`, not a component, not the injected
+`RuntimeAPI.stream` client -- knows which is in use.
+
+| `DT_STREAM_BACKEND` | transport | ports it opens | use |
+|---------------------|-----------|----------------|-----|
+| `zmq` (default)     | the framework's own XSUB/XPUB broker | two, unauthenticated, loopback by default | local, demos, the two-terminal loop |
+| `orbit`             | ORBIT eventing (`radical.orbit`)     | **none** | anything shared, and everything in production |
+
+```sh
+# a service deployment with the data plane inside the token domain
+DT_STREAM_BACKEND=orbit radical-orbit-broker.py --plugins default,dt
+```
+
+An external subscriber joins the same way -- as an ORBIT participant, so
+it needs the broker URL and the token, and no addresses at all:
+
+```python
+from digitaltwin.streaming import connect_stream_client
+
+stream = await connect_stream_client(twin_id, backend='orbit')
+await stream.subscribe_to_dtype(ECHO, queue)
+```
+
+**Payload ceiling**: an ORBIT frame is capped at 4 MiB, so a single
+stream message must cloudpickle to less than that (64 KiB of the budget
+is reserved for the envelope).  Oversized payloads raise a clear
+`ValueError` at `publish` -- ORBIT itself would drop the frame with
+nothing but a log line, which on a days-long twin is indistinguishable
+from a stalled stream.  The ZMQ backend has no such ceiling; a twin meant
+to run on either should stay well under it.  Chunk large artifacts, or
+stream a reference and move the bytes with the staging plugin.
+
+**Semantics** are the same on both: at-most-once, with bounded
+drop-oldest queues (broker-side, and again on the hop into the host
+loop).  That *is* the DT conflation contract, so nothing above the
+backend adds a second one -- a slow consumer loses samples rather than
+memory, and never backpressures a producer.  Loss is visible as a gap in
+the broker-assigned sequence numbers.  ORBIT's `replay` plugin would give
+late joiners history; it is deliberately not integrated in v1.
+
+`perf/bench_streams.py` measures what the choice costs: about a
+millisecond per stream hop on loopback.
 
 ## Running it as a service (the `dt` ORBIT plugin)
 
@@ -197,17 +247,43 @@ again, so a `twin_create` after it fails immediately with `engine
 `ready` and stalling.  `unregister_session`, then build the session and
 its twins again.
 
-### Binding policy for the service (R7)
+### The data plane and its trust boundary (R7)
 
-The plugin runs its own DT stream broker, embedded, one per plugin and
-shared by every twin.  **It binds to loopback on a random port by
-default, and that default is the safe one**: its payloads are
-cloudpickled, so anyone who can reach the XSUB/XPUB ports gets code
-execution in every subscriber -- weaker than the token-authenticated
-ORBIT channel around it.
+The DT streams carry cloudpickled payloads.  That is accepted -- the
+service already executes client-shipped component classes, and both sit
+inside ORBIT's single-token trust domain (risk R4).  What was *not*
+acceptable is where those payloads used to travel: a pair of ZMQ ports
+that authenticate nobody, so anyone who could reach them got code
+execution in every subscriber, no token required.  The data plane was
+weaker than the control plane wrapped around it.
+
+**`DT_STREAM_BACKEND=orbit` closes that gap**, and a production
+deployment must use it:
+
+```sh
+DT_STREAM_BACKEND=orbit radical-orbit-broker.py --plugins default,dt
+```
 
-A non-loopback bind is possible (`DT_STREAM_PUB_ADDR` /
-`DT_STREAM_SUB_ADDR` on the service host) but requires a deliberate
-decision *and* a firewalled or private network.  Until the data plane
-moves inside ORBIT's authenticated channel, do not expose those ports --
-including in demos.
+The twins' streams become ORBIT events on the same token-authenticated
+WebSocket star as every other call, under one `dt_stream` plugin
+namespace.  The embedded ZMQ broker is then **never started** -- the
+service opens no data-plane port at all, and there is nothing left to
+firewall.  The payloads are still cloudpickle; what changed is that
+reaching them now requires the same token as calling `twin_create`.
+Reviewers can check the guarantee directly: the plugin host has no child
+processes, and `admin/sessions` reports `{"stream_broker": {"backend":
+"orbit"}}`.
+
+Two things this does *not* do.  It does not make the payloads safe to
+receive from an untrusted party -- per-tenant auth is post-v1, so
+everything inside the token domain is still mutually trusting.  And it
+does not remove the 4 MiB frame cap, which the ZMQ backend did not have
+(see "Choosing a data plane").
+
+**With the `zmq` backend the old mitigations still apply, in full.** The
+plugin runs its own DT stream broker, embedded, one per plugin and shared
+by every twin.  It binds to loopback on a random port by default, and
+that default is the safe one.  A non-loopback bind is possible
+(`DT_STREAM_PUB_ADDR` / `DT_STREAM_SUB_ADDR` on the service host) but
+requires a deliberate decision *and* a firewalled or private network.  Do
+not expose those ports -- including in demos.

From e69a7b6b265b79195d9373818a61826ccee47371 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 20:21:10 +0200
Subject: [PATCH 06/11] A waiter woken by close() gets the closed-client error

`_await_running` parks a publish or subscribe that arrives before
connect() finished.  If the client is closed while somebody waits there,
the wait ends on a backend that has already let its transport go -- so
re-check, and raise the ordinary closed-client error instead of failing
somewhere further in.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/digitaltwin/streaming.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py
index b3a68e4..d2deb2b 100644
--- a/src/digitaltwin/streaming.py
+++ b/src/digitaltwin/streaming.py
@@ -127,12 +127,18 @@ def _check_open(self):
 
     async def _await_running(self, what: str):
         """Callers which arrive before `connect()` finished are made to
-        wait rather than to lose their message."""
+        wait rather than to lose their message.
+
+        The re-check afterwards matters: a client closed while somebody
+        was waiting here must produce the ordinary closed-client error,
+        not an attribute error somewhere in a half-dismantled backend.
+        """
 
         if not self.is_running.is_set():
             logger.warning("requesting %s before connecting to broker. Waiting",
                            what)
             await self.is_running.wait()
+            self._check_open()
 
     def _start_receiving(self):
         """Arm the supervised receive loop.  Called at the end of connect."""

From 5d338ec06999dd87be6b4dba30ef66a24abb12fe Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Sat, 15 Aug 2026 20:36:10 +0200
Subject: [PATCH 07/11] Review findings: the third queue, a stranded waiter, an
 explicit backend

- The docstring claimed two bounded queues on the receive path; there are
  three.  The middle one is ORBIT's own `CallbackDispatcher`, and it is
  the odd one out -- it drops the *newest* frame, not the oldest.  Name
  it, and log its counter alongside ours at close, so a twin that lost
  samples does not have to be inferred from `seq` gaps in somebody else's
  log.
- `_await_running` parked a caller that arrived before connect.  If the
  connect was then abandoned, `close()` merely cleared the flag and the
  waiter kept waiting for a connection that would never come.  Closing
  now sets the event before clearing it: the waiters wake, hit
  `_check_open`, and get the ordinary closed-client error.  Tested for
  both verbs that can park.
- The zmq branch of `connect_stream` re-read the environment per twin,
  which could contradict the choice the plugin resolved once at
  construction.  Name the backend.
- `frame_cap()` reaches into a private `EndpointRuntime` attribute; say
  so, and that the fix belongs upstream.
- The benchmark's barrier drained with an `empty()` check, which leaves a
  barrier message still in flight to arrive mid-measurement and pair with
  the wrong publish.  Wait for quiet instead.  Numbers re-measured.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 perf/README.md                     |  8 ++---
 perf/bench_streams.py              | 25 +++++++++++-----
 src/digitaltwin/service/plugin.py  | 13 +++++++--
 src/digitaltwin/streaming.py       | 17 ++++++++++-
 src/digitaltwin/streaming_orbit.py | 47 ++++++++++++++++++++++++++----
 test/unit/test_streaming_orbit.py  | 26 +++++++++++++++++
 6 files changed, 115 insertions(+), 21 deletions(-)

diff --git a/perf/README.md b/perf/README.md
index 67bd01d..3cd0931 100644
--- a/perf/README.md
+++ b/perf/README.md
@@ -44,11 +44,11 @@ Loopback, one host, bare-int payloads (2026-08-15):
 
 | data plane                  | p50     | p99     | burst          |
 |-----------------------------|---------|---------|----------------|
-| zmq, embedded broker        | 0.84 ms | 1.14 ms | 20 400 msg/s   |
-| orbit eventing              | 1.94 ms | 2.47 ms |  5 300 msg/s   |
+| zmq, embedded broker        | 0.98 ms | 1.32 ms | 20 200 msg/s   |
+| orbit eventing              | 2.09 ms | 2.60 ms |  4 300 msg/s   |
 
-With 64 KiB payloads (`--payload 65536`): 1.04 ms / 3.04 ms p50, and
-6 800 vs 1 100 msg/s in burst.
+With 64 KiB payloads (`--payload 65536`): 1.18 ms / 3.56 ms p50, and
+6 500 vs 1 150 msg/s in burst.
 
 So the ORBIT data plane costs roughly **1 ms per stream hop** and about a
 quarter of the burst throughput, in exchange for the security property of
diff --git a/perf/bench_streams.py b/perf/bench_streams.py
index 1550988..3d98990 100644
--- a/perf/bench_streams.py
+++ b/perf/bench_streams.py
@@ -43,32 +43,41 @@
 DELIVER_TIMEOUT = 10.0
 
 
+QUIET_WAIT = 0.25
+
+
 async def settle(client, queue, message) -> None:
-    """Publish until something comes back.
+    """Publish until something comes back, then wait for quiet.
 
     Neither backend acknowledges a subscription -- ZMQ's SUBSCRIBE and
     ORBIT's `subscribe` frame are both fire-and-forget -- so the only
     honest barrier is a message that made the round trip.
+
+    Draining until the queue *stays* empty matters as much as the barrier
+    itself: a `queue.empty()` check would leave any barrier message still
+    in flight to arrive during the measurement, where it would pair with
+    the wrong publish and shift every latency after it.
     """
 
     deadline = time.perf_counter() + SETTLE_TIMEOUT
 
-    while perf_left(deadline):
+    while time.perf_counter() < deadline:
         await client.publish(SAMPLE, message)
         try:
-            await asyncio.wait_for(queue.get(), 0.25)
+            await asyncio.wait_for(queue.get(), QUIET_WAIT)
             break
         except TimeoutError:
             continue
     else:
         raise TimeoutError(f"no message came back within {SETTLE_TIMEOUT}s")
 
-    while not queue.empty():  # drop whatever the barrier left behind
-        queue.get_nowait()
-
+    while time.perf_counter() < deadline:
+        try:
+            await asyncio.wait_for(queue.get(), QUIET_WAIT)
+        except TimeoutError:  # nothing left in flight
+            return
 
-def perf_left(deadline: float) -> bool:
-    return time.perf_counter() < deadline
+    raise TimeoutError(f"the stream never went quiet within {SETTLE_TIMEOUT}s")
 
 
 async def measure(client, label: str, n_meas: int, n_burst: int,
diff --git a/src/digitaltwin/service/plugin.py b/src/digitaltwin/service/plugin.py
index b94d986..c5c31c7 100644
--- a/src/digitaltwin/service/plugin.py
+++ b/src/digitaltwin/service/plugin.py
@@ -36,7 +36,12 @@
 from radical.orbit.plugin_base import Plugin
 from starlette.requests import Request
 
-from ..config import BACKEND_ORBIT, embedded_stream_addresses, stream_backend
+from ..config import (
+    BACKEND_ORBIT,
+    BACKEND_ZMQ,
+    embedded_stream_addresses,
+    stream_backend,
+)
 from ..streaming import (
     CLIENT_CONNECT_TIMEOUT,
     PubSubClient,
@@ -342,7 +347,11 @@ async def connect_stream(
 
         pub_addr, sub_addr = await self.stream_addresses()
 
-        return await connect_stream_client(namespace, pub_addr, sub_addr, timeout)
+        # named explicitly: the choice was resolved once at construction,
+        # and re-reading the environment per twin could contradict it
+        return await connect_stream_client(
+            namespace, pub_addr, sub_addr, timeout, backend=BACKEND_ZMQ
+        )
 
     def stream_summary(self) -> dict:
         """The data plane's entry in `admin/sessions`."""
diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py
index d2deb2b..53ac3c6 100644
--- a/src/digitaltwin/streaming.py
+++ b/src/digitaltwin/streaming.py
@@ -140,6 +140,21 @@ async def _await_running(self, what: str):
             await self.is_running.wait()
             self._check_open()
 
+    def _stop_running(self):
+        """Mark the backend disconnected, waking anyone parked in
+        `_await_running` on the way.
+
+        A waiter is otherwise stranded: it parked on a connect that has
+        now been abandoned, and a plain `clear()` would leave it waiting
+        for one that will never arrive.  `set()` resolves the waiters'
+        futures immediately and the `clear()` right after does not
+        un-resolve them, so they wake into `_check_open` and get the
+        ordinary closed-client error.
+        """
+
+        self.is_running.set()
+        self.is_running.clear()
+
     def _start_receiving(self):
         """Arm the supervised receive loop.  Called at the end of connect."""
 
@@ -514,7 +529,7 @@ async def close(self):
 
             # all sockets are closed, so this returns immediately
             self._ctx.term()
-            self.is_running.clear()
+            self._stop_running()
 
     async def _run(self):
         """Receive loop.  A single bad payload or a raising callback must
diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py
index 301cee7..7324668 100644
--- a/src/digitaltwin/streaming_orbit.py
+++ b/src/digitaltwin/streaming_orbit.py
@@ -38,11 +38,22 @@
 
 Semantics
 ---------
-At-most-once with per-subscriber bounded drop-oldest queues, broker-side
-and again on the hop into the host loop.  That is the DT conflation
-contract, so nothing above adds a second one.  Losses are observable as
-gaps in the broker-assigned `seq`; the `replay` plugin would give late
-joiners history and is deliberately not integrated in v1.
+At-most-once with bounded queues at every hop, which is the DT conflation
+contract -- so nothing above adds a second one.  There are *three* such
+queues on the path, and only two of them drop the oldest:
+
+1. the broker's per-subscriber out-queue (drop-oldest),
+2. the receiving runtime's `CallbackDispatcher` (drop-*newest*: a full
+   queue refuses the arriving frame).  It is ORBIT's, not ours, and its
+   losses are counted in `runtime._cb.dropped` -- logged here at close
+   when nonzero, because otherwise they would only ever be a warning in
+   somebody else's log,
+3. the inbox this module puts between that dispatcher thread and the
+   host loop (drop-oldest).
+
+Losses are observable as gaps in the broker-assigned `seq`; the `replay`
+plugin would give late joiners history and is deliberately not integrated
+in v1.
 """
 
 import asyncio
@@ -192,15 +203,34 @@ async def close(self):
                     self._unregister(runtime, topic)
                 self.topics.clear()
 
+                self._report_losses(runtime)
+
                 if runtime is not None and self._owns_runtime:
                     # blocking: closes the socket and joins three threads
                     await asyncio.to_thread(runtime.stop)
 
             finally:
-                self.is_running.clear()
+                self._stop_running()
                 self._inbox = None
                 self._loop = None
 
+    def _report_losses(self, runtime: Optional[EndpointRuntime]) -> None:
+        """Account for both queues that dropped on the way in.
+
+        The dispatcher's counter is ORBIT's and drops the *newest* frame;
+        ours is the inbox and drops the oldest.  Neither is an error --
+        the data plane is specified to conflate -- but a twin that lost
+        samples should not have to be inferred from `seq` gaps.
+        """
+
+        cb_dropped = getattr(getattr(runtime, "_cb", None), "dropped", 0) or 0
+
+        if self.dropped or cb_dropped:
+            logger.warning(
+                "%s dropped %d message(s) at the inbox and %d at the ORBIT"
+                " callback queue", self.name, self.dropped, cb_dropped,
+            )
+
     # -- pubsub -------------------------------------------------------------
 
     async def publish(self, topic, message):
@@ -265,6 +295,11 @@ def frame_cap(self) -> int:
 
         Read off the runtime when there is one: a deployment may tune it,
         and the module default would then be wrong in either direction.
+
+        `_frame_cap` is private, and reaching into it is the wart here --
+        `EndpointRuntime` should expose it (upstream follow-up).  Until it
+        does, the `getattr` fallback keeps this correct against a runtime
+        that does not have the attribute at all.
         """
 
         cap = getattr(self._runtime, "_frame_cap", None)
diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py
index 9cc0c50..05c0fd4 100644
--- a/test/unit/test_streaming_orbit.py
+++ b/test/unit/test_streaming_orbit.py
@@ -481,6 +481,32 @@ def build(broker_url=None, name=None):
     assert backend._task is None
 
 
+@pytest.mark.parametrize("verb", ["publish", "subscribe"])
+async def test_a_waiter_parked_before_connect_is_woken_by_close(verb):
+    """A caller which arrives before `connect()` parks until the backend
+    is running.  If the connect is abandoned instead, that waiter must
+    get the ordinary closed-client error -- not wait for a connection
+    that will never come.
+    """
+
+    backend = OrbitPubSubBackend()
+    topic = "dt/twin-a/dtypes/sensor"
+
+    call = {
+        "publish": lambda: backend.publish(topic, "parked"),
+        "subscribe": lambda: backend.subscribe(topic, lambda m: None),
+    }[verb]
+
+    parked = asyncio.create_task(call())
+    await asyncio.sleep(0.05)
+    assert not parked.done(), "the caller should be waiting for connect"
+
+    await backend.close()
+
+    with pytest.raises(RuntimeError, match="closed"):
+        await asyncio.wait_for(parked, 5)
+
+
 async def test_participant_names_are_unique():
     """Two participants may not register under one name."""
 

From bc5c28207c3523ae400788f8465a9bdd5e179964 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Mon, 17 Aug 2026 16:35:43 +0200
Subject: [PATCH 08/11] test: the travelled config gained the orbit broker
 field

The M3 data plane added broker_url to PubSubConfig; the M0 round-trip
test compares the dict shape exactly and follows.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 test/unit/test_streaming.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/test/unit/test_streaming.py b/test/unit/test_streaming.py
index 5bb153d..5d3d749 100644
--- a/test/unit/test_streaming.py
+++ b/test/unit/test_streaming.py
@@ -149,6 +149,7 @@ async def test_stream_config_travels_and_reopens_the_endpoint(broker, stream_cli
         "pub_addr": config.pub_addr,
         "sub_addr": config.sub_addr,
         "kind": "zmq",
+        "broker_url": None,
     }
 
     # ... and the far end opens its own client from it, in the same

From 8d7b4d2258203c3a52c2b6140675f0c8ad53a06f Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Mon, 17 Aug 2026 23:57:49 +0200
Subject: [PATCH 09/11] The ORBIT plane carries external channels too

A channel's payload is encoded by the codec its binding names, so the
transport has to hand those bytes over untouched -- the `raw` flag the
seam already carries.  The ZMQ backend honoured it; the ORBIT one did
not, so a twin binding a channel on that data plane failed at the first
subscribe.

The set of raw topics is the same notion in both backends, so it moves
up to PubSubBackend where the rest of the shared receive machinery
already lives.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/digitaltwin/streaming.py       |  8 ++++----
 src/digitaltwin/streaming_orbit.py | 29 +++++++++++++++++++++++------
 2 files changed, 27 insertions(+), 10 deletions(-)

diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py
index 53ac3c6..c85ad10 100644
--- a/src/digitaltwin/streaming.py
+++ b/src/digitaltwin/streaming.py
@@ -111,6 +111,10 @@ def __init__(self):
         # topic lookup, whatever the transport matched on the wire.
         self.topics: dict[str, list[Callable]] = {}
 
+        # topics whose payload the transport hands over untouched:
+        # something above the seam owns their wire format (see the codecs)
+        self.raw_topics: set[str] = set()
+
         self._task: Optional[asyncio.Task] = None
         self._closed = False
         self.is_running = asyncio.Event()
@@ -409,10 +413,6 @@ def __init__(self, pub_addr: Optional[str] = None, sub_addr: Optional[str] = Non
             self._ctx.socket(zmq.SUB) if sub_addr is not None else None
         )
 
-        # topics whose payload the transport must hand over untouched:
-        # something above the seam owns their wire format (see the codecs)
-        self.raw_topics: set[str] = set()
-
 
     async def _connect_socket(self, sock, addr):
         """Connect `sock` and wait until the connection is established.
diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py
index 7324668..ceef240 100644
--- a/src/digitaltwin/streaming_orbit.py
+++ b/src/digitaltwin/streaming_orbit.py
@@ -233,8 +233,13 @@ def _report_losses(self, runtime: Optional[EndpointRuntime]) -> None:
 
     # -- pubsub -------------------------------------------------------------
 
-    async def publish(self, topic, message):
-        """Emit one event frame carrying the cloudpickled message.
+    async def publish(self, topic, message, raw=False, **backend_params):
+        """Emit one event frame carrying the message.
+
+        `raw` means the payload is already bytes whose format something
+        above the seam owns (an external channel and its codec): it goes
+        on the wire untouched.  Twin-internal traffic is cloudpickled
+        here.
 
         Oversized payloads raise: ORBIT drops a frame over the cap with a
         log line and no exception, and a data plane that silently swallows
@@ -244,7 +249,7 @@ async def publish(self, topic, message):
         self._check_open()
         await self._await_running("publish")
 
-        payload = cloudpickle.dumps(message)
+        payload = bytes(message) if raw else cloudpickle.dumps(message)
         cap = self.payload_cap()
 
         if len(payload) > cap:
@@ -257,12 +262,19 @@ async def publish(self, topic, message):
         self._runtime.send_notification(STREAM_PLUGIN, topic,
                                         {PAYLOAD_KEY: payload})
 
-    async def subscribe(self, topic, callback: Callable, **backend_params):
-        """Register `callback` for `topic`, subscribing on first use."""
+    async def subscribe(self, topic, callback: Callable, raw=False, **backend_params):
+        """Register `callback` for `topic`, subscribing on first use.
+
+        With `raw`, the topic's payloads reach their subscribers as the
+        bytes that arrived: an external channel decodes them itself.
+        """
 
         self._check_open()
         await self._await_running("subscribe")
 
+        if raw:
+            self.raw_topics.add(topic)
+
         callbacks = self.topics.setdefault(topic, [])
         callbacks.append(callback)
 
@@ -364,7 +376,12 @@ async def _run(self):
             topic, data = await self._inbox.get()
 
             try:
-                message = cloudpickle.loads(data[PAYLOAD_KEY])
+                payload = data[PAYLOAD_KEY]
+                message = (
+                    payload
+                    if topic in self.raw_topics
+                    else cloudpickle.loads(payload)
+                )
             except Exception:
                 logger.exception("dropping malformed message on topic %r", topic)
                 continue

From a1ce0fbb8628ffc9e5f4fdd55eb6f173f93694b1 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Wed, 19 Aug 2026 11:38:09 +0200
Subject: [PATCH 10/11] unsubscribe went sync on devel; this implementor missed
 the memo

The devel merge changed `PubSubBackend.unsubscribe` from `async def` to
`def`, and `PubSubClient` now calls it without `await`.  Two spots on
this branch were still on the old contract:

  * `OrbitPubSubBackend.unsubscribe` still declared `async def` -- its
    body is entirely synchronous -- so the client's plain call created a
    coroutine and dropped it, and the unsubscribe silently never
    happened ("coroutine ... was never awaited" in the test run).
  * the test awaited `unsubscribe_dtype`, which now returns None, so it
    died on a TypeError before it could even observe the first problem.

Both are the same mismatch, both fixed by dropping the keyword.

test_streaming_orbit: 17 passed (was 1 failed, 16 passed).  The two
remaining unit failures on this branch (`kind == "generic"` and the
describe() shape) are devel regressions, fixed on fix/devel-regressions;
they converge when that merges and devel flows down.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/digitaltwin/streaming_orbit.py | 2 +-
 test/unit/test_streaming_orbit.py  | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py
index ceef240..58283f4 100644
--- a/src/digitaltwin/streaming_orbit.py
+++ b/src/digitaltwin/streaming_orbit.py
@@ -285,7 +285,7 @@ async def subscribe(self, topic, callback: Callable, raw=False, **backend_params
                 plugin_name=STREAM_PLUGIN, topic=topic, callback=self._on_event
             )
 
-    async def unsubscribe(self, topic):
+    def unsubscribe(self, topic):
         if self._closed:
             return
 
diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py
index 05c0fd4..1781bd0 100644
--- a/test/unit/test_streaming_orbit.py
+++ b/test/unit/test_streaming_orbit.py
@@ -268,7 +268,7 @@ async def test_unsubscribe_stops_delivery_and_the_subscription(
     await twin.publish(SENSOR, "first")
     assert (await drain(queue)).data == "first"
 
-    await twin.unsubscribe_dtype(SENSOR)
+    twin.unsubscribe_dtype(SENSOR)
     assert SENSOR not in twin.subscriptions
     assert loopback.topics() == []
 

From f5cdbb4d0ef7b8477d87d055af300599f5a9db32 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Mon, 24 Aug 2026 12:26:30 +0200
Subject: [PATCH 11/11] ci: the learner tests need the real ROSE

M2/M3 bring integration tests importing `rose`.  PyPI's `rose` is an
unrelated version-string helper, so the real one is pinned like the
other radical deps, and the learn extra installs.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .github/workflows/ci.yml | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5f189a5..1fde91f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -32,12 +32,13 @@ jobs:
         run: >-
           pip install "radical.asyncflow==0.5.1" "radical.orbit==0.5.0"
           "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2"
+          "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
 
       # src/ layout: pytest tests the *installed* package, never the
       # working tree.  Always install before testing -- a stale install
       # will silently pass against old code.
       - name: Install digitaltwin (test + service extras)
-        run: pip install ".[test,service]" pytest-timeout
+        run: pip install ".[test,service,learn]" pytest-timeout
 
       # --continue-on-collection-errors: one broken test module (see the
       # PR description) must not hide the rest of the suite behind an
@@ -71,12 +72,13 @@ jobs:
         run: >-
           pip install "radical.asyncflow==0.5.1" "radical.orbit==0.5.0"
           "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2"
+          "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
 
       # src/ layout: pytest tests the *installed* package, never the
       # working tree.  Always install before testing -- a stale install
       # will silently pass against old code.
       - name: Install digitaltwin (test + service extras)
-        run: pip install ".[test,service]" pytest-timeout
+        run: pip install ".[test,service,learn]" pytest-timeout
 
       # The ORBIT broker needs a self-signed TLS cert/key and a shared
       # ingress token at the default ~/.radical/orbit location -- see