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
diff --git a/README.md b/README.md
index 38f14b3..27618ec 100644
--- a/README.md
+++ b/README.md
@@ -5,7 +5,7 @@ Main set of features 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
@@ -69,7 +69,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)
 
@@ -195,17 +245,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.
diff --git a/perf/README.md b/perf/README.md
index 3469e77..3cd0931 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.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.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
+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..3d98990
--- /dev/null
+++ b/perf/bench_streams.py
@@ -0,0 +1,177 @@
+"""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
+
+
+QUIET_WAIT = 0.25
+
+
+async def settle(client, queue, message) -> None:
+    """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 time.perf_counter() < deadline:
+        await client.publish(SAMPLE, message)
+        try:
+            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 time.perf_counter() < deadline:
+        try:
+            await asyncio.wait_for(queue.get(), QUIET_WAIT)
+        except TimeoutError:  # nothing left in flight
+            return
+
+    raise TimeoutError(f"the stream never went quiet within {SETTLE_TIMEOUT}s")
+
+
+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())
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..c5c31c7 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,18 @@
 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,
+    BACKEND_ZMQ,
+    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 +98,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 +275,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 +324,49 @@ 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()
+
+        # 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`."""
+
+        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 86da102..b453405 100644
--- a/src/digitaltwin/streaming.py
+++ b/src/digitaltwin/streaming.py
@@ -40,7 +40,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__)
 
@@ -96,12 +102,17 @@ def decode_payload(payload: bytes, codec: str):
 
 
 class PubSubBackend(ABC):
-    """Abstract base class for publish/subscribe backends.
+    """The transport seam: everything above this is transport-agnostic.
 
-    Subclasses must implement the asynchronous ``connect`` method and the
-    core publish/subscribe operations.
+    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
     # endpoint.  Every backend declares its own.
     kind = "generic"
@@ -117,12 +128,102 @@ 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]] = {}
+
+        # 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()
+
     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.
+
+        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 _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."""
+
+        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: Any, **kwargs: Any) -> None:
         """Connect the backend to the message broker.
@@ -357,18 +458,6 @@ def __init__(
             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.
@@ -423,8 +512,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()
 
@@ -438,9 +526,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)
@@ -456,9 +542,7 @@ 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"))
 
@@ -485,12 +569,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
@@ -504,11 +584,7 @@ async def close(self):
 
             # all sockets are closed, so this returns immediately
             self._ctx.term()
-            self.is_running.clear()
-
-    def _check_open(self):
-        if self._closed:
-            raise RuntimeError("stream client is closed")
+            self._stop_running()
 
     async def _run(self):
         """Receive loop.  A single bad payload or a raising callback must
@@ -527,23 +603,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
@@ -606,6 +666,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:
@@ -767,12 +828,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(
@@ -797,13 +863,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
@@ -871,11 +946,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..58283f4
--- /dev/null
+++ b/src/digitaltwin/streaming_orbit.py
@@ -0,0 +1,389 @@
+"""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 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
+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()
+
+                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._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, 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
+        every large sample is worse than one that refuses it.
+        """
+
+        self._check_open()
+        await self._await_running("publish")
+
+        payload = bytes(message) if raw else 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, 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)
+
+        # 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
+            )
+
+    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.
+
+        `_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)
+
+        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:
+                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
+
+            await self._dispatch(topic, message)
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)
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}
 
 
 # ---------------------------------------------------------------------------
diff --git a/test/unit/test_streaming.py b/test/unit/test_streaming.py
index aec9c79..6244145 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
diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py
new file mode 100644
index 0000000..1781bd0
--- /dev/null
+++ b/test/unit/test_streaming_orbit.py
@@ -0,0 +1,520 @@
+"""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"
+
+    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
+
+
+@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."""
+
+    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"