diff --git a/README.md b/README.md
index e510cbaf..a4321924 100644
--- a/README.md
+++ b/README.md
@@ -69,7 +69,7 @@ Optional integrations for different cloud providers can be installed using `plug
 
 Support for parallelisation and hyperparameter optimisation can be installed using `plugboard[ray]`.
 
-Additional optional extras: `plugboard[llm]` for LLM components, `plugboard[redis]` for Redis-based connectors, and `plugboard[websockets]` for WebSocket I/O.
+Additional optional extras: `plugboard[llm]` for LLM components, `plugboard[redis]` for Redis-based connectors, `plugboard[omq]` for the pyomq backend for ZMQ connectors, and `plugboard[websockets]` for WebSocket I/O.
 
 ## ⚡ Quickstart with AI
 
diff --git a/docs/examples/tutorials/running-in-parallel.md b/docs/examples/tutorials/running-in-parallel.md
index c8203abb..48b680f6 100644
--- a/docs/examples/tutorials/running-in-parallel.md
+++ b/docs/examples/tutorials/running-in-parallel.md
@@ -51,7 +51,7 @@ With some small changes we can make the same model run in parallel on Ray. First
 !!! info
     [`Channel`][plugboard.connector.Channel] objects are used by Plugboard to handle the communication between components. So far we have used [`AsyncioChannel`][plugboard.connector.AsyncioChannel], which is the best option for simple models that don't require parallelisation.
 
-    Plugboard provides different channel classes for use in parallel environments: [`RayChannel`][plugboard.connector.RayChannel] is suitable for single and multi-host Ray environments. [`ZMQChannel`][plugboard.connector.ZMQChannel] is faster, but currently only works on a single host.
+    Plugboard provides different channel classes for use in parallel environments: [`RayChannel`][plugboard.connector.RayChannel] is suitable for single and multi-host Ray environments. [`ZMQChannel`][plugboard.connector.ZMQChannel] is faster, but currently only works on a single host. Set `PLUGBOARD_ZMQ_BACKEND=pyomq` to use the optional pyomq backend instead of PyZMQ.
 
 ```python
 --8<-- "examples/tutorials/004_using_ray/hello_ray.py:ray"
diff --git a/plugboard/_zmq/backend.py b/plugboard/_zmq/backend.py
new file mode 100644
index 00000000..929bb44b
--- /dev/null
+++ b/plugboard/_zmq/backend.py
@@ -0,0 +1,47 @@
+"""Selects the ZeroMQ Python backend."""
+
+from __future__ import annotations
+
+import os
+import typing as _t
+
+
+ZMQ_BACKEND_ENV = "PLUGBOARD_ZMQ_BACKEND"
+ZMQ_BACKEND_PYZMQ = "pyzmq"
+ZMQ_BACKEND_PYOMQ = "pyomq"
+ZMQ_BACKENDS = frozenset({ZMQ_BACKEND_PYZMQ, ZMQ_BACKEND_PYOMQ})
+
+
+class ZMQBackendImportError(ImportError):
+    """Raised when the selected ZeroMQ backend cannot be imported."""
+
+
+def _backend_name() -> str:
+    backend = os.environ.get(ZMQ_BACKEND_ENV, ZMQ_BACKEND_PYZMQ).strip().lower()
+    if not backend:
+        return ZMQ_BACKEND_PYZMQ
+    if backend not in ZMQ_BACKENDS:
+        choices = ", ".join(sorted(ZMQ_BACKENDS))
+        raise ValueError(
+            f"Unsupported ZMQ backend {backend!r}. Set {ZMQ_BACKEND_ENV} to one of: {choices}."
+        )
+    return backend
+
+
+def _load_backend() -> tuple[str, _t.Any, _t.Any]:
+    backend = _backend_name()
+    try:
+        if backend == ZMQ_BACKEND_PYOMQ:
+            import pyomq as zmq
+            import pyomq.asyncio as zmq_asyncio
+        else:
+            import zmq
+            import zmq.asyncio as zmq_asyncio
+    except ImportError as e:
+        raise ZMQBackendImportError(
+            f"Failed to import {backend!r} ZMQ backend selected by {ZMQ_BACKEND_ENV}."
+        ) from e
+    return backend, zmq, zmq_asyncio
+
+
+zmq_backend, zmq, zmq_asyncio = _load_backend()
diff --git a/plugboard/_zmq/zmq_proxy.py b/plugboard/_zmq/zmq_proxy.py
index 8134706c..e611be6f 100644
--- a/plugboard/_zmq/zmq_proxy.py
+++ b/plugboard/_zmq/zmq_proxy.py
@@ -7,8 +7,8 @@
 import typing as _t
 
 from pydantic import BaseModel, Field, ValidationError
-import zmq
-import zmq.asyncio
+
+from plugboard._zmq.backend import zmq, zmq_asyncio
 
 
 try:
@@ -23,8 +23,8 @@
 def create_socket(
     socket_type: int,
     socket_opts: zmq_sockopts_t,
-    ctx: _t.Optional[zmq.asyncio.Context] = None,
-) -> zmq.asyncio.Socket:
+    ctx: _t.Optional[zmq_asyncio.Context] = None,
+) -> zmq_asyncio.Socket:
     """Creates a ZeroMQ socket with the given type and options.
 
     Args:
@@ -35,7 +35,7 @@ def create_socket(
     Returns:
         The created ZMQ socket.
     """
-    _ctx = ctx or zmq.asyncio.Context.instance()
+    _ctx = ctx or zmq_asyncio.Context.instance()
     socket = _ctx.socket(socket_type)
     for opt, value in socket_opts:
         socket.setsockopt(opt, value)
@@ -184,7 +184,7 @@ def _connect_socket_req_socket(self) -> None:
         """Connects the REQ socket to the REP socket in the subprocess."""
         if self._socket_rep_port is None:
             raise RuntimeError("ZMQ proxy socket REP port not set.")
-        self._socket_req_socket: zmq.asyncio.Socket = create_socket(zmq.REQ, [])
+        self._socket_req_socket: zmq_asyncio.Socket = create_socket(zmq.REQ, [])
         socket_rep_socket_address: str = f"{self._zmq_address}:{self._socket_rep_port}"
         self._socket_req_socket.connect(socket_rep_socket_address)
         self._socket_req_lock: asyncio.Lock = asyncio.Lock()
@@ -205,8 +205,8 @@ async def add_push_socket(self, topic: str, maxsize: int = 2000) -> str:
 
     async def _run(self) -> None:
         """Async multiprocessing entrypoint to run ZMQ proxy."""
-        self._push_poller: zmq.asyncio.Poller = zmq.asyncio.Poller()
-        self._push_sockets: dict[str, tuple[str, zmq.asyncio.Socket]] = {}
+        self._push_poller: zmq_asyncio.Poller = zmq_asyncio.Poller()
+        self._push_sockets: dict[str, tuple[str, zmq_asyncio.Socket]] = {}
 
         self._create_proxy_sockets()
 
@@ -285,13 +285,17 @@ def _create_push_socket(self, topic: str, maxsize: int, reuse: bool = True) -> s
     async def _poll_push_sockets(self) -> None:
         """Polls push sockets for messages and sends them to the proxy."""
         while True:
+            # Some backends return immediately when polling an empty socket set.
+            if not self._push_poller.sockets:
+                await asyncio.sleep(1)
+                continue
             # Set a timeout of 1 second to allow for new push sockets to be added
             events = dict(await self._push_poller.poll(timeout=1000))
             async with asyncio.TaskGroup() as tg:
                 for socket in events:
                     tg.create_task(self._handle_push_socket(socket))
 
-    async def _handle_push_socket(self, socket: zmq.asyncio.Socket) -> None:
+    async def _handle_push_socket(self, socket: zmq_asyncio.Socket) -> None:
         msg = await socket.recv_multipart()
         topic = msg[0].decode("utf8")
         _, push_socket = self._push_sockets[topic]
diff --git a/plugboard/connector/connector.py b/plugboard/connector/connector.py
index ceeb35c3..04aaca7e 100644
--- a/plugboard/connector/connector.py
+++ b/plugboard/connector/connector.py
@@ -17,6 +17,10 @@ class Connector(ABC, ExportMixin):
     def __init__(self, spec: ConnectorSpec, *args: _t.Any, **kwargs: _t.Any) -> None:
         self.spec: ConnectorSpec = spec
 
+    async def init(self) -> None:
+        """Acquire resources required by this connector."""
+        pass
+
     @abstractmethod
     async def connect_send(self) -> Channel:
         """Returns a `Channel` for sending messages."""
diff --git a/plugboard/connector/ray_channel.py b/plugboard/connector/ray_channel.py
index 63a653be..f04ba625 100644
--- a/plugboard/connector/ray_channel.py
+++ b/plugboard/connector/ray_channel.py
@@ -1,5 +1,6 @@
 """Provides `RayChannel` for use in cluster compute environments."""
 
+import asyncio
 import typing as _t
 
 from plugboard.connector.asyncio_channel import AsyncioChannel
@@ -34,8 +35,18 @@ def __init__(  # noqa: D417
         """
         default_options = {"num_cpus": 0}
         actor_options = actor_options or {}
-        actor_options = {**default_options, **actor_options}
-        self._actor = ray.remote(**actor_options)(_AsyncioChannelActor).remote(**kwargs)
+        self._actor_options = {**default_options, **actor_options}
+        self._channel_kwargs = kwargs
+        self._actor: _t.Any = None
+        self._init_lock = asyncio.Lock()
+
+    async def init(self) -> None:
+        """Create the channel actor when execution starts."""
+        async with self._init_lock:
+            if self._actor is None:
+                self._actor = ray.remote(**self._actor_options)(_AsyncioChannelActor).remote(
+                    **self._channel_kwargs
+                )
 
     @property
     def maxsize(self) -> int:
@@ -73,10 +84,16 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None:
             raise ValueError("RayConnector only supports `PIPELINE` type connections.")
         self._channel = RayChannel()
 
+    async def init(self) -> None:
+        """Create the remote channel actor."""
+        await self._channel.init()
+
     async def connect_send(self) -> RayChannel:
         """Returns a `RayChannel` for sending messages."""
+        await self.init()
         return self._channel
 
     async def connect_recv(self) -> RayChannel:
         """Returns a `RayChannel` for receiving messages."""
+        await self.init()
         return self._channel
diff --git a/plugboard/connector/zmq_channel.py b/plugboard/connector/zmq_channel.py
index d74ad849..9d1734fc 100644
--- a/plugboard/connector/zmq_channel.py
+++ b/plugboard/connector/zmq_channel.py
@@ -7,9 +7,8 @@
 import typing as _t
 
 from that_depends import Provide, inject
-import zmq
-import zmq.asyncio
 
+from plugboard._zmq.backend import ZMQ_BACKEND_PYOMQ, zmq, zmq_asyncio, zmq_backend
 from plugboard._zmq.zmq_proxy import ZMQ_ADDR, ZMQProxy, create_socket, zmq_sockopts_t
 from plugboard.connector.connector import Connector
 from plugboard.connector.serde_channel import SerdeChannel
@@ -19,6 +18,7 @@
 
 
 ZMQ_CONFIRM_MSG: str = "__PLUGBOARD_CHAN_CONFIRM_MSG__"
+PYOMQ_CLOSE_DRAIN_SECONDS: float = 0.1
 
 # Collection of poll tasks for ZMQ channels required to create strong refs to polling tasks
 # to avoid destroying tasks before they are done on garbage collection. Is there a better way?
@@ -32,8 +32,8 @@ class ZMQChannel(SerdeChannel):
     def __init__(  # noqa: D417
         self,
         *args: _t.Any,
-        send_socket: _t.Optional[zmq.asyncio.Socket] = None,
-        recv_socket: _t.Optional[zmq.asyncio.Socket] = None,
+        send_socket: _t.Optional[zmq_asyncio.Socket] = None,
+        recv_socket: _t.Optional[zmq_asyncio.Socket] = None,
         topic: str = "",
         maxsize: int = 2000,
         **kwargs: _t.Any,
@@ -54,8 +54,8 @@ def __init__(  # noqa: D417
             maxsize: Optional; Queue maximum item capacity, defaults to 2000.
         """
         super().__init__(*args, **kwargs)
-        self._send_socket: _t.Optional[zmq.asyncio.Socket] = send_socket
-        self._recv_socket: _t.Optional[zmq.asyncio.Socket] = recv_socket
+        self._send_socket: _t.Optional[zmq_asyncio.Socket] = send_socket
+        self._recv_socket: _t.Optional[zmq_asyncio.Socket] = recv_socket
         self._is_send_closed = send_socket is None
         self._is_recv_closed = recv_socket is None
         self._send_hwm = max(maxsize // 2, 1)
@@ -83,6 +83,10 @@ async def close(self) -> None:
         """Closes the `ZMQChannel`."""
         if self._send_socket is not None:
             await super().close()
+            if zmq_backend == ZMQ_BACKEND_PYOMQ:
+                # pyomq does not expose an awaitable socket drain; give queued PUB frames,
+                # including the close sentinel, a short window to reach the proxy.
+                await asyncio.sleep(PYOMQ_CLOSE_DRAIN_SECONDS)
             self._send_socket.close()
             self._send_socket = None
         if self._recv_socket is not None:
@@ -101,6 +105,7 @@ def __init__(
         super().__init__(*args, **kwargs)
         self._zmq_address = zmq_address
         self._maxsize = maxsize
+        self._init_lock = asyncio.Lock()
 
     @abstractmethod
     async def connect_send(self) -> ZMQChannel:
@@ -120,23 +125,31 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None:
         super().__init__(*args, **kwargs)
         self._send_channel: _t.Optional[ZMQChannel] = None
         self._recv_channel: _t.Optional[ZMQChannel] = None
-
-        # Socket to receive sender address from sender
-        self._sender_rep_socket = create_socket(zmq.REP, [])
-        self._sender_rep_socket_port = self._sender_rep_socket.bind_to_random_port("tcp://*")
-        self._sender_rep_socket_addr = f"{self._zmq_address}:{self._sender_rep_socket_port}"
-        self._sender_req_lock = asyncio.Lock()
+        self._sender_rep_socket: _t.Optional[zmq_asyncio.Socket] = None
+        self._sender_rep_socket_addr: _t.Optional[str] = None
+        self._sender_req_lock: _t.Optional[asyncio.Lock] = None
         self._sender_addr: _t.Optional[str] = None
-
-        # Socket to send sender address to receiver
-        self._receiver_rep_socket = create_socket(zmq.REP, [])
-        self._receiver_rep_socket_port = self._receiver_rep_socket.bind_to_random_port("tcp://*")
-        self._receiver_rep_socket_addr = f"{self._zmq_address}:{self._receiver_rep_socket_port}"
-        self._receiver_req_lock = asyncio.Lock()
-
-        self._exchange_addr_task = asyncio.create_task(self._exchange_address())
-        _zmq_exchange_addr_tasks.add(self._exchange_addr_task)
-        self._exchange_addr_task.add_done_callback(_zmq_exchange_addr_tasks.discard)
+        self._receiver_rep_socket: _t.Optional[zmq_asyncio.Socket] = None
+        self._receiver_rep_socket_addr: _t.Optional[str] = None
+        self._receiver_req_lock: _t.Optional[asyncio.Lock] = None
+        self._exchange_addr_task: _t.Optional[asyncio.Task[None]] = None
+
+    async def init(self) -> None:
+        """Allocate address exchange sockets when execution starts."""
+        async with self._init_lock:
+            if self._sender_rep_socket_addr is not None:
+                return
+            self._sender_rep_socket = create_socket(zmq.REP, [])
+            sender_port = self._sender_rep_socket.bind_to_random_port("tcp://*")
+            self._sender_rep_socket_addr = f"{self._zmq_address}:{sender_port}"
+            self._sender_req_lock = asyncio.Lock()
+            self._receiver_rep_socket = create_socket(zmq.REP, [])
+            receiver_port = self._receiver_rep_socket.bind_to_random_port("tcp://*")
+            self._receiver_rep_socket_addr = f"{self._zmq_address}:{receiver_port}"
+            self._receiver_req_lock = asyncio.Lock()
+            self._exchange_addr_task = asyncio.create_task(self._exchange_address())
+            _zmq_exchange_addr_tasks.add(self._exchange_addr_task)
+            self._exchange_addr_task.add_done_callback(_zmq_exchange_addr_tasks.discard)
 
     def __getstate__(self) -> dict:
         state = self.__dict__.copy()
@@ -148,6 +161,7 @@ def __getstate__(self) -> dict:
             "_exchange_addr_task",
             "_send_channel",
             "_recv_channel",
+            "_init_lock",
         ):
             if attr in state:
                 del state[attr]
@@ -157,28 +171,41 @@ def __setstate__(self, state: dict) -> None:
         self.__dict__.update(state)
         self._send_channel = None
         self._recv_channel = None
+        self._init_lock = asyncio.Lock()
 
     async def _exchange_address(self) -> None:
+        if (
+            self._sender_req_lock is None
+            or self._sender_rep_socket is None
+            or self._receiver_req_lock is None
+            or self._receiver_rep_socket is None
+        ):
+            raise ChannelSetupError("ZMQ connector is not initialized")
+        sender_req_lock = self._sender_req_lock
+        sender_rep_socket = self._sender_rep_socket
+        receiver_req_lock = self._receiver_req_lock
+        receiver_rep_socket = self._receiver_rep_socket
+
         async def _handle_sender_requests() -> None:
-            async with self._sender_req_lock:
-                sender_request = await self._sender_rep_socket.recv_json()
+            async with sender_req_lock:
+                sender_request = await sender_rep_socket.recv_json()
                 if (sender_addr := sender_request.get("sender_address")) is None:
-                    await self._sender_rep_socket.send_json({"success": False})
+                    await sender_rep_socket.send_json({"success": False})
                 else:
                     self._sender_addr = sender_addr
-                await self._sender_rep_socket.send_json({"success": True})
+                await sender_rep_socket.send_json({"success": True})
 
                 while True:
-                    await self._sender_rep_socket.recv_json()
-                    await self._sender_rep_socket.send_json({"success": False})
+                    await sender_rep_socket.recv_json()
+                    await sender_rep_socket.send_json({"success": False})
 
         async def _handle_receiver_requests() -> None:
             while self._sender_addr is None:
                 await asyncio.sleep(0.5)
             while True:
-                async with self._receiver_req_lock:
-                    await self._receiver_rep_socket.recv()
-                    await self._receiver_rep_socket.send(self._sender_addr.encode())
+                async with receiver_req_lock:
+                    await receiver_rep_socket.recv()
+                    await receiver_rep_socket.send(self._sender_addr.encode())
 
         async with asyncio.TaskGroup() as tg:
             tg.create_task(_handle_sender_requests())
@@ -186,8 +213,11 @@ async def _handle_receiver_requests() -> None:
 
     async def connect_send(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for sending messages."""
+        await self.init()
         if self._send_channel is not None:
             return self._send_channel
+        if self._sender_rep_socket_addr is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         send_socket = create_socket(zmq.PUSH, [(zmq.SNDHWM, self._maxsize)])
         send_port = send_socket.bind_to_random_port("tcp://*")
         send_addr = f"{self._zmq_address}:{send_port}"
@@ -206,8 +236,11 @@ async def connect_send(self) -> ZMQChannel:
 
     async def connect_recv(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for receiving messages."""
+        await self.init()
         if self._recv_channel is not None:
             return self._recv_channel
+        if self._receiver_rep_socket_addr is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         recv_socket = create_socket(zmq.PULL, [(zmq.RCVHWM, self._maxsize)])
 
         receiver_req_socket = create_socket(zmq.REQ, [])
@@ -228,16 +261,28 @@ class _ZMQPubsubConnector(_ZMQConnector):
     def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None:
         super().__init__(*args, **kwargs)
         self._topic = str(self.spec.source)
-        self._xsub_socket = create_socket(zmq.XSUB, [(zmq.RCVHWM, self._maxsize)])
-        self._xsub_port = self._xsub_socket.bind_to_random_port("tcp://*")
-        self._xpub_socket = create_socket(zmq.XPUB, [(zmq.SNDHWM, self._maxsize)])
-        self._xpub_port = self._xpub_socket.bind_to_random_port("tcp://*")
-        self._poller = zmq.asyncio.Poller()
-        self._poller.register(self._xsub_socket, zmq.POLLIN)
-        self._poller.register(self._xpub_socket, zmq.POLLIN)
-        self._poll_task = asyncio.create_task(self._poll())
-        _zmq_proxy_tasks.add(self._poll_task)
-        self._poll_task.add_done_callback(_zmq_proxy_tasks.discard)
+        self._xsub_port: _t.Optional[int] = None
+        self._xpub_port: _t.Optional[int] = None
+        self._poller: _t.Optional[zmq_asyncio.Poller] = None
+        self._poll_task: _t.Optional[asyncio.Task[None]] = None
+        self._xsub_socket: _t.Optional[zmq_asyncio.Socket] = None
+        self._xpub_socket: _t.Optional[zmq_asyncio.Socket] = None
+
+    async def init(self) -> None:
+        """Allocate proxy sockets when execution starts."""
+        async with self._init_lock:
+            if self._xsub_port is not None:
+                return
+            self._xsub_socket = create_socket(zmq.XSUB, [(zmq.RCVHWM, self._maxsize)])
+            self._xsub_port = self._xsub_socket.bind_to_random_port("tcp://*")
+            self._xpub_socket = create_socket(zmq.XPUB, [(zmq.SNDHWM, self._maxsize)])
+            self._xpub_port = self._xpub_socket.bind_to_random_port("tcp://*")
+            self._poller = zmq_asyncio.Poller()
+            self._poller.register(self._xsub_socket, zmq.POLLIN)
+            self._poller.register(self._xpub_socket, zmq.POLLIN)
+            self._poll_task = asyncio.create_task(self._poll())
+            _zmq_proxy_tasks.add(self._poll_task)
+            self._poll_task.add_done_callback(_zmq_proxy_tasks.discard)
 
     def __getstate__(self) -> dict:
         state = self.__dict__.copy()
@@ -248,10 +293,12 @@ def __getstate__(self) -> dict:
         return state
 
     async def _poll(self) -> None:
+        if self._poller is None or self._xpub_socket is None or self._xsub_socket is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         poll_fn, xps, xss = self._poller.poll, self._xpub_socket, self._xsub_socket
         try:
             while True:
-                events = dict(await poll_fn())
+                events = dict(await poll_fn(timeout=1000))
                 if xps in events:
                     await xss.send_multipart(await xps.recv_multipart())
                 if xss in events:
@@ -262,6 +309,9 @@ async def _poll(self) -> None:
 
     async def connect_send(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for sending pubsub messages."""
+        await self.init()
+        if self._xsub_port is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         send_socket = create_socket(zmq.PUB, [(zmq.SNDHWM, self._maxsize)])
         send_socket.connect(f"{self._zmq_address}:{self._xsub_port}")
         await asyncio.sleep(0.1)  # Ensure connections established before first send. Better way?
@@ -269,6 +319,9 @@ async def connect_send(self) -> ZMQChannel:
 
     async def connect_recv(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for receiving pubsub messages."""
+        await self.init()
+        if self._xpub_port is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         socket_opts: zmq_sockopts_t = [
             (zmq.RCVHWM, self._maxsize),
             (zmq.SUBSCRIBE, self._topic.encode("utf8")),
@@ -282,17 +335,24 @@ async def connect_recv(self) -> ZMQChannel:
 class _ZMQPubsubConnectorProxy(_ZMQConnector):
     """`_ZMQPubsubConnectorProxy` acts is a python asyncio based proxy for `ZMQChannel` messages."""
 
-    @inject
-    def __init__(
-        self, *args: _t.Any, zmq_proxy: ZMQProxy = Provide[DI.zmq_proxy], **kwargs: _t.Any
-    ) -> None:
+    def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None:
         super().__init__(*args, **kwargs)
         self._topic = str(self.spec.source)
-        self._zmq_proxy = zmq_proxy
+        self._zmq_proxy: _t.Optional[ZMQProxy] = None
 
         self._send_channel: _t.Optional[ZMQChannel] = None
         self._recv_channel: _t.Optional[ZMQChannel] = None
 
+    @inject
+    async def _resolve_proxy(self, zmq_proxy: ZMQProxy = Provide[DI.zmq_proxy]) -> ZMQProxy:
+        return zmq_proxy
+
+    async def init(self) -> None:
+        """Resolve the shared proxy when execution starts."""
+        async with self._init_lock:
+            if self._zmq_proxy is None:
+                self._zmq_proxy = await self._resolve_proxy()
+
     def __getstate__(self) -> dict:
         state = self.__dict__.copy()
         for attr in ("_send_channel", "_recv_channel"):
@@ -307,8 +367,11 @@ def __setstate__(self, state: dict) -> None:
 
     async def connect_send(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for sending pubsub messages."""
+        await self.init()
         if self._send_channel is not None:
             return self._send_channel
+        if self._zmq_proxy is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         send_socket = create_socket(zmq.PUB, [(zmq.SNDHWM, self._maxsize)])
         send_socket.connect(self._zmq_proxy.xsub_addr)
         self._send_channel = ZMQChannel(
@@ -319,6 +382,9 @@ async def connect_send(self) -> ZMQChannel:
 
     async def connect_recv(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for receiving pubsub messages."""
+        await self.init()
+        if self._zmq_proxy is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         socket_opts: zmq_sockopts_t = [
             (zmq.RCVHWM, self._maxsize),
             (zmq.SUBSCRIBE, self._topic.encode("utf8")),
@@ -343,8 +409,11 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None:
 
     async def connect_recv(self) -> ZMQChannel:
         """Returns a `ZMQChannel` for receiving messages."""
+        await self.init()
         if self._recv_channel is not None:
             return self._recv_channel
+        if self._zmq_proxy is None:
+            raise ChannelSetupError("ZMQ connector is not initialized")
         self._push_address = await self._zmq_proxy.add_push_socket(
             self._topic, maxsize=self._maxsize
         )
@@ -380,6 +449,10 @@ def __init__(
                 raise ValueError(f"Unsupported connector mode: {self.spec.mode}")
         self._zmq_conn_impl: _ZMQConnector = zmq_conn_cls(*args, **kwargs)
 
+    async def init(self) -> None:
+        """Allocate resources for the selected ZMQ implementation."""
+        await self._zmq_conn_impl.init()
+
     @property
     def zmq_address(self) -> str:
         """The ZMQ address used for communication."""
diff --git a/plugboard/library/file_io.py b/plugboard/library/file_io.py
index 6d313279..61990a34 100644
--- a/plugboard/library/file_io.py
+++ b/plugboard/library/file_io.py
@@ -1,5 +1,6 @@
 """Provides `FileReader` and `FileWriter` components to access files from Plugboard models."""
 
+import asyncio
 from collections import deque
 from pathlib import Path
 import typing as _t
@@ -106,7 +107,12 @@ def __init__(
             raise ValueError("Only CSV files support chunked writing.")
         self._storage_options = storage_options or {}
         self._header_written = False
-        self._check_file()
+
+    async def init(self) -> None:
+        """Open and truncate the destination when execution starts."""
+        await asyncio.to_thread(self._check_file)
+        self._header_written = False
+        await super().init()
 
     def _check_file(self) -> None:
         with fsspec.open(self._file_path, mode="w", **self._storage_options):
diff --git a/plugboard/process/local_process.py b/plugboard/process/local_process.py
index c673eb44..f636f585 100644
--- a/plugboard/process/local_process.py
+++ b/plugboard/process/local_process.py
@@ -57,6 +57,10 @@ async def _connect_state(self) -> None:
 
     async def init(self) -> None:
         """Performs component initialisation actions."""
+        self.validate()
+        async with asyncio.TaskGroup() as tg:
+            for connector in self.connectors.values():
+                tg.create_task(connector.init())
         async with asyncio.TaskGroup() as tg:
             await self.connect_state()
             await self._connect_components()
diff --git a/plugboard/process/process.py b/plugboard/process/process.py
index 939ae94a..0f570358 100644
--- a/plugboard/process/process.py
+++ b/plugboard/process/process.py
@@ -109,12 +109,21 @@ async def _set_status(self, status: Status, publish: bool = True) -> None:
     @abstractmethod
     async def init(self) -> None:
         """Performs component initialisation actions."""
+        self.validate()
+        self._is_initialised = True
+        await self._set_status(Status.INIT)
+
+    def validate(self) -> None:
+        """Validate the process topology without acquiring external resources."""
+        for component in self.components.values():
+            if not hasattr(component, "_state_is_connected"):
+                raise ValidationError(
+                    "Component invalid: did you forget to call super().__init__ in the constructor?"
+                )
         errors = validate_process(self.dict())
         if errors:
             msg = "Process validation failed:\n" + "\n".join(errors)
             raise ValidationError(msg)
-        self._is_initialised = True
-        await self._set_status(Status.INIT)
 
     @abstractmethod
     async def step(self) -> None:
diff --git a/plugboard/process/ray_process.py b/plugboard/process/ray_process.py
index 81a3e28f..a607b009 100644
--- a/plugboard/process/ray_process.py
+++ b/plugboard/process/ray_process.py
@@ -43,11 +43,7 @@ def __init__(
         """
         # TODO: Replace with a namespace based on the job ID or similar
         self._namespace = f"plugboard-{gen_rand_str(16)}"
-        self._component_actors = {
-            # Recreate components on remote actors
-            c.id: self._create_component_actor(c)
-            for c in components
-        }
+        self._component_actors: dict[str, _t.Any] = {}
         self._tasks: dict[str, ray.ObjectRef] = {}
 
         super().__init__(
@@ -113,6 +109,13 @@ async def _connect_state(self) -> None:
 
     async def init(self) -> None:
         """Performs component initialisation actions."""
+        self.validate()
+        self._component_actors = {
+            # Recreate components on remote actors only when execution starts.
+            component.id: self._create_component_actor(component)
+            for component in self.components.values()
+        }
+        await asyncio.gather(*(connector.init() for connector in self.connectors.values()))
         await self.connect_state()
         await self._connect_components()
         coros = [component.init.remote() for component in self._component_actors.values()]
diff --git a/plugboard/state/ray_state_backend.py b/plugboard/state/ray_state_backend.py
index 010ec02a..07390f7b 100644
--- a/plugboard/state/ray_state_backend.py
+++ b/plugboard/state/ray_state_backend.py
@@ -67,8 +67,14 @@ def __init__(
         super().__init__(*args, **kwargs)
         default_options = {"num_cpus": 0}
         actor_options = actor_options or {}
-        actor_options = {**default_options, **actor_options}
-        self._actor = ray.remote(**actor_options)(_DictionaryActor).remote()
+        self._actor_options = {**default_options, **actor_options}
+        self._actor: _t.Any = None
+
+    async def init(self) -> None:
+        """Create the state actor when process execution starts."""
+        if self._actor is None:
+            self._actor = ray.remote(**self._actor_options)(_DictionaryActor).remote()
+        await super().init()
 
     @property
     def _state(self) -> dict[str, _t.Any]:
diff --git a/pyproject.toml b/pyproject.toml
index bfc7fb00..aade645c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -49,6 +49,7 @@ llm = [
     "llama-index-core>=0.12.30,<1",
     "llama-index-llms-openai>=0.3.33,<1",
 ]
+omq = ["pyomq>=0.20.1,<1"]
 # Pinning jsonschema due to performance issues with Lark and rfc3987-syntax parser
 # https://github.com/python-jsonschema/jsonschema/issues/1392
 ray = ["ray[tune]>=2.47.1,<3", "jsonschema<4.25.0", "optuna>=3.0,<5"]
@@ -80,6 +81,7 @@ test = [
     "moto[server]>=5.0,<6",
     "openai-responses>=0.11.4,<1",
     "optuna>=3.0,<5",
+    "pyomq>=0.20.1,<1",
     "pytest>=8.3,<10",
     "pytest-asyncio>=1.4,<2",
     "pytest-benchmark>=5.1.0",
diff --git a/tests/integration/test_process_validation.py b/tests/integration/test_process_validation.py
index 8eaf9d05..8e3e3af0 100644
--- a/tests/integration/test_process_validation.py
+++ b/tests/integration/test_process_validation.py
@@ -77,7 +77,5 @@ async def step(self) -> None:
         ],
     )
 
-    with pytest.raises(ExceptionGroup) as exc_info:
+    with pytest.raises(exceptions.ValidationError, match="forget to call super"):
         await process.init()
-
-    assert exc_info.group_contains(exceptions.ValidationError), "No ValidationError raised"
diff --git a/tests/unit/test_channel.py b/tests/unit/test_channel.py
index 810a20be..408ad91c 100644
--- a/tests/unit/test_channel.py
+++ b/tests/unit/test_channel.py
@@ -50,6 +50,7 @@ async def test_channel(connector_cls: type[Connector], ray_ctx: None, job_id_ctx
     """Tests the various Channel implementations."""
     spec = ConnectorSpec(mode=ConnectorMode.PIPELINE, source="test.send", target="test.recv")
     connector = ConnectorBuilder(connector_cls=connector_cls).build(spec)
+    await connector.init()
 
     send_channel, recv_channel = await asyncio.gather(
         connector.connect_send(), connector.connect_recv()
@@ -94,6 +95,7 @@ async def test_multiprocessing_channel(
     """Tests the various Channel implementations in a multiprocess environment."""
     spec = ConnectorSpec(mode=ConnectorMode.PIPELINE, source="test.send", target="test.recv")
     connector = ConnectorBuilder(connector_cls=connector_cls_mp).build(spec)
+    await connector.init()
 
     container_ctx = container_context(
         DI, global_context={"job_id": job_id_ctx}, scope=ContextScopes.APP
diff --git a/tests/unit/test_side_effect_free_construction.py b/tests/unit/test_side_effect_free_construction.py
new file mode 100644
index 00000000..3573dae3
--- /dev/null
+++ b/tests/unit/test_side_effect_free_construction.py
@@ -0,0 +1,173 @@
+"""Tests that process inspection does not acquire external resources."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import msgspec
+import pytest
+from typer.testing import CliRunner
+
+from plugboard.cli import app
+from plugboard.component import Component, IOController as IO
+from plugboard.connector import AsyncioConnector, RayConnector, ZMQConnector
+from plugboard.diagram import MermaidDiagram
+from plugboard.exceptions import ValidationError
+from plugboard.library import FileWriter
+from plugboard.process import LocalProcess, RayProcess
+from plugboard.schemas import ConfigSpec, ConnectorMode, ConnectorSpec
+from plugboard.state import RayStateBackend
+from plugboard.utils import DI, Settings
+
+
+class Source(Component):
+    """A source used to describe a valid topology."""
+
+    io = IO(outputs=["value"])
+
+    async def step(self) -> None:
+        """Produce no values; only topology metadata is used by these tests."""
+        pass
+
+
+class Sink(Component):
+    """A sink used to describe a valid topology."""
+
+    io = IO(inputs=["value"])
+
+    async def step(self) -> None:
+        """Consume no values; only topology metadata is used by these tests."""
+        pass
+
+
+def test_file_writer_inspection_does_not_touch_destination(tmp_path: Path) -> None:
+    """Construction, validation, diagramming, and export leave output files untouched."""
+    output_path = tmp_path / "output.csv"
+    output_path.write_text("existing data\n")
+    writer = FileWriter(name="writer", path=str(output_path), field_names=["value"])
+    process = LocalProcess(
+        components=[Source(name="source"), writer],
+        connectors=[
+            AsyncioConnector(spec=ConnectorSpec(source="source.value", target="writer.value"))
+        ],
+    )
+
+    process.validate()
+    MermaidDiagram.from_process(process)
+    process.export()
+    config_path = tmp_path / "process.yaml"
+    process.dump(config_path)
+
+    runner = CliRunner()
+    for command in ("validate", "diagram"):
+        result = runner.invoke(app, ["process", command, str(config_path)])
+        assert result.exit_code == 0
+        assert output_path.read_text() == "existing data\n"
+
+    assert output_path.read_text() == "existing data\n"
+
+
+@pytest.mark.asyncio
+async def test_invalid_process_fails_before_initializing_writer_or_connector(
+    tmp_path: Path,
+) -> None:
+    """Invalid topology cannot truncate files or initialize connectors."""
+    output_path = tmp_path / "output.csv"
+    output_path.write_text("existing data\n")
+    writer = FileWriter(name="writer", path=str(output_path), field_names=["value"])
+    connector = AsyncioConnector(spec=ConnectorSpec(source="source.value", target="sink.value"))
+    connector.init = AsyncMock()
+    process = LocalProcess(
+        components=[Source(name="source"), Sink(name="sink"), writer],
+        connectors=[connector],
+    )
+
+    with pytest.raises(ValidationError):
+        await process.init()
+
+    connector.init.assert_not_awaited()
+    assert output_path.read_text() == "existing data\n"
+
+
+def test_ray_process_inspection_does_not_create_actors(tmp_path: Path) -> None:
+    """Ray process metadata can be inspected without creating any actors."""
+    with (
+        patch("plugboard.process.ray_process.ray.remote") as component_remote,
+        patch("plugboard.connector.ray_channel.ray.remote") as channel_remote,
+        patch("plugboard.state.ray_state_backend.ray.remote") as state_remote,
+    ):
+        process = RayProcess(
+            components=[Source(name="source"), Sink(name="sink")],
+            connectors=[
+                RayConnector(spec=ConnectorSpec(source="source.value", target="sink.value"))
+            ],
+            state=RayStateBackend(),
+        )
+
+        process.validate()
+        MermaidDiagram.from_process(process)
+        exported = process.export()
+        exported["connector_builder"] = {"type": "plugboard.connector.RayConnector"}
+        config_path = tmp_path / "ray-process.yaml"
+        config = ConfigSpec.model_validate({"plugboard": {"process": exported}})
+        config_path.write_bytes(msgspec.yaml.encode(config.model_dump()))
+
+        runner = CliRunner()
+        for command in ("validate", "diagram"):
+            result = runner.invoke(app, ["process", command, str(config_path)])
+            assert result.exit_code == 0
+
+    component_remote.assert_not_called()
+    channel_remote.assert_not_called()
+    state_remote.assert_not_called()
+
+
+@pytest.mark.parametrize("use_proxy", [False, True])
+def test_zmq_process_inspection_does_not_allocate_resources(use_proxy: bool) -> None:
+    """ZMQ sockets and proxy processes are deferred until initialization."""
+    settings = Settings.model_validate({"flags": {"zmq_pubsub_proxy": use_proxy}})
+    with (
+        DI.override_providers_sync({"settings": settings}),
+        patch("plugboard.connector.zmq_channel.create_socket") as create_socket,
+        patch("plugboard.utils.di.ZMQProxy") as proxy,
+    ):
+        process = LocalProcess(
+            components=[Source(name="source"), Sink(name="sink")],
+            connectors=[
+                ZMQConnector(
+                    spec=ConnectorSpec(
+                        source="source.value",
+                        target="sink.value",
+                        mode=ConnectorMode.PUBSUB,
+                    )
+                )
+            ],
+        )
+
+        process.validate()
+        MermaidDiagram.from_process(process)
+        process.export()
+
+    create_socket.assert_not_called()
+    proxy.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_invalid_ray_process_fails_before_creating_actors() -> None:
+    """Invalid topology is rejected before process, channel, or state actors start."""
+    with (
+        patch("plugboard.process.ray_process.ray.remote") as component_remote,
+        patch("plugboard.connector.ray_channel.ray.remote") as channel_remote,
+        patch("plugboard.state.ray_state_backend.ray.remote") as state_remote,
+    ):
+        process = RayProcess(
+            components=[Sink(name="sink")],
+            connectors=[],
+            state=RayStateBackend(),
+        )
+
+        with pytest.raises(ValidationError, match="unconnected inputs"):
+            await process.init()
+
+    component_remote.assert_not_called()
+    channel_remote.assert_not_called()
+    state_remote.assert_not_called()
diff --git a/tests/unit/test_zmq_backend.py b/tests/unit/test_zmq_backend.py
new file mode 100644
index 00000000..8ac05669
--- /dev/null
+++ b/tests/unit/test_zmq_backend.py
@@ -0,0 +1,177 @@
+"""Tests for ZMQ backend selection."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import subprocess
+import sys
+import textwrap
+from types import ModuleType
+
+import pytest
+
+import plugboard._zmq.backend as backend
+from plugboard._zmq.backend import ZMQ_BACKEND_ENV
+
+
+def _run_backend_probe(
+    code: str,
+    backend: str | None = None,
+    extra_env: dict[str, str] | None = None,
+) -> subprocess.CompletedProcess[str]:
+    env = os.environ.copy()
+    if backend is not None:
+        env[ZMQ_BACKEND_ENV] = backend
+    if extra_env is not None:
+        env.update(extra_env)
+    return subprocess.run(  # noqa: S603
+        [sys.executable, "-c", textwrap.dedent(code)],
+        check=False,
+        capture_output=True,
+        env=env,
+        text=True,
+    )
+
+
+def test_backend_name_normalizes_environment_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Backend names are normalized and default to PyZMQ."""
+    monkeypatch.delenv(ZMQ_BACKEND_ENV, raising=False)
+    assert backend._backend_name() == "pyzmq"
+
+    monkeypatch.setenv(ZMQ_BACKEND_ENV, "")
+    assert backend._backend_name() == "pyzmq"
+
+    monkeypatch.setenv(ZMQ_BACKEND_ENV, " PyOmQ ")
+    assert backend._backend_name() == "pyomq"
+
+
+def test_backend_name_rejects_unsupported_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Unsupported backend names fail before imports are attempted."""
+    monkeypatch.setenv(ZMQ_BACKEND_ENV, "not-a-backend")
+
+    with pytest.raises(ValueError, match="Unsupported ZMQ backend"):
+        backend._backend_name()
+
+
+def test_load_pyomq_backend(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The loader imports both modules for the pyomq backend."""
+    pyomq = ModuleType("pyomq")
+    pyomq_asyncio = ModuleType("pyomq.asyncio")
+    monkeypatch.setitem(sys.modules, "pyomq", pyomq)
+    monkeypatch.setitem(sys.modules, "pyomq.asyncio", pyomq_asyncio)
+    monkeypatch.setenv(ZMQ_BACKEND_ENV, "pyomq")
+
+    selected_backend, selected_zmq, selected_asyncio = backend._load_backend()
+
+    assert selected_backend == "pyomq"
+    assert selected_zmq is pyomq
+    assert selected_asyncio is pyomq_asyncio
+
+
+def test_load_backend_reports_missing_dependency(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The loader reports a missing selected backend clearly."""
+    monkeypatch.setitem(sys.modules, "pyomq", None)
+    monkeypatch.setenv(ZMQ_BACKEND_ENV, "pyomq")
+
+    with pytest.raises(backend.ZMQBackendImportError, match="pyomq"):
+        backend._load_backend()
+
+
+def test_default_zmq_backend_is_pyzmq() -> None:
+    """The default ZMQ backend remains PyZMQ."""
+    result = _run_backend_probe(
+        """
+        from plugboard._zmq.backend import zmq_backend, zmq
+        print(zmq_backend)
+        print(zmq.__name__)
+        """,
+        extra_env={ZMQ_BACKEND_ENV: ""},
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.splitlines() == ["pyzmq", "zmq"]
+
+
+def test_invalid_zmq_backend_fails_with_clear_error() -> None:
+    """Unsupported backend names fail during import with a clear error."""
+    result = _run_backend_probe(
+        """
+        import plugboard._zmq.backend
+        """,
+        backend="not-a-backend",
+    )
+
+    assert result.returncode != 0
+    assert "Unsupported ZMQ backend" in result.stderr
+    assert ZMQ_BACKEND_ENV in result.stderr
+
+
+@pytest.mark.skipif(importlib.util.find_spec("pyomq") is None, reason="pyomq not installed")
+def test_pyomq_backend_supports_create_socket() -> None:
+    """The pyomq backend can run the ZMQ socket helper."""
+    result = _run_backend_probe(
+        """
+        import asyncio
+
+        from plugboard._zmq.backend import zmq, zmq_backend
+        from plugboard._zmq.zmq_proxy import create_socket
+
+        async def main() -> None:
+            pull = create_socket(zmq.PULL, [(zmq.RCVHWM, 100)])
+            port = pull.bind_to_random_port("tcp://127.0.0.1")
+            push = create_socket(zmq.PUSH, [(zmq.SNDHWM, 100)])
+            push.connect(f"tcp://127.0.0.1:{port}")
+            await asyncio.sleep(0.2)
+            await push.send_multipart([b"", b"payload"])
+            got = await asyncio.wait_for(pull.recv_multipart(), timeout=1.0)
+            assert got == [b"", b"payload"]
+            push.close(linger=0)
+            pull.close(linger=0)
+            print(zmq_backend)
+
+        asyncio.run(main())
+        """,
+        backend="pyomq",
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == "pyomq"
+
+
+@pytest.mark.skipif(importlib.util.find_spec("pyomq") is None, reason="pyomq not installed")
+def test_pyomq_backend_supports_zmq_proxy() -> None:
+    """The pyomq backend can run the ZMQ proxy process."""
+    result = _run_backend_probe(
+        """
+        import asyncio
+
+        from plugboard._zmq.backend import zmq
+        from plugboard._zmq.zmq_proxy import ZMQProxy, create_socket
+
+        async def main() -> None:
+            proxy = ZMQProxy(maxsize=100)
+            try:
+                topic = b"topic"
+                sub = create_socket(
+                    zmq.SUB,
+                    [(zmq.RCVHWM, 100), (zmq.SUBSCRIBE, topic)],
+                )
+                sub.connect(proxy.xpub_addr)
+                pub = create_socket(zmq.PUB, [(zmq.SNDHWM, 100)])
+                pub.connect(proxy.xsub_addr)
+                await asyncio.sleep(0.3)
+                await pub.send_multipart([topic, b"payload"])
+                got = await asyncio.wait_for(sub.recv_multipart(), timeout=1.0)
+                assert got == [topic, b"payload"]
+                pub.close(linger=0)
+                sub.close(linger=0)
+            finally:
+                proxy.terminate(timeout=5.0)
+
+        asyncio.run(main())
+        """,
+        backend="pyomq",
+    )
+
+    assert result.returncode == 0, result.stderr
diff --git a/tests/unit/test_zmq_channel.py b/tests/unit/test_zmq_channel.py
new file mode 100644
index 00000000..91cafc64
--- /dev/null
+++ b/tests/unit/test_zmq_channel.py
@@ -0,0 +1,26 @@
+"""Tests for ZMQChannel backend-specific behavior."""
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+
+from plugboard._zmq.backend import ZMQ_BACKEND_PYOMQ
+from plugboard.connector.zmq_channel import PYOMQ_CLOSE_DRAIN_SECONDS, ZMQChannel
+
+
+@pytest.mark.asyncio
+async def test_close_drains_pyomq_send_socket() -> None:
+    """PyOMQ channels allow queued close frames to reach the proxy before closing."""
+    send_socket = Mock()
+    send_socket.send_multipart = AsyncMock()
+    channel = ZMQChannel(send_socket=send_socket)
+
+    with patch("plugboard.connector.zmq_channel.zmq_backend", ZMQ_BACKEND_PYOMQ):
+        with patch(
+            "plugboard.connector.zmq_channel.asyncio.sleep", new_callable=AsyncMock
+        ) as sleep:
+            await channel.close()
+
+    sleep.assert_awaited_once_with(PYOMQ_CLOSE_DRAIN_SECONDS)
+    send_socket.close.assert_called_once_with()
+    assert channel.is_closed
diff --git a/tests/unit/test_zmq_proxy.py b/tests/unit/test_zmq_proxy.py
index 7007bfea..b3ecc338 100644
--- a/tests/unit/test_zmq_proxy.py
+++ b/tests/unit/test_zmq_proxy.py
@@ -1,7 +1,9 @@
 """Tests for ZMQProxy class."""
 
 import asyncio
+from contextlib import suppress
 import typing as _t
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 import pytest_asyncio
@@ -11,6 +13,36 @@
 from plugboard._zmq.zmq_proxy import ZMQ_ADDR, ZMQProxy, create_socket, zmq_sockopts_t
 
 
+@pytest.mark.asyncio
+async def test_empty_push_poller_yields_and_resumes() -> None:
+    """An empty proxy yields to other tasks and handles subsequently registered sockets."""
+    proxy = ZMQProxy.__new__(ZMQProxy)
+    poller = Mock(spec=zmq.asyncio.Poller)
+    poller.sockets = []
+    # Fail immediately instead of hanging if the empty poller is called in a busy loop.
+    poller.poll = AsyncMock(side_effect=AssertionError("Polled an empty socket set"))
+    proxy._push_poller = poller
+    handler = AsyncMock()
+    proxy._handle_push_socket = handler  # type: ignore[method-assign]
+
+    task = asyncio.create_task(proxy._poll_push_sockets())
+    try:
+        await asyncio.sleep(0.01)
+        assert not task.done()
+        poller.poll.assert_not_called()
+
+        socket = Mock(spec=zmq.asyncio.Socket)
+        poller.sockets = [(socket, zmq.POLLIN)]
+        poller.poll.side_effect = [[(socket, zmq.POLLIN)], asyncio.CancelledError()]
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=3)
+        handler.assert_awaited_once_with(socket)
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
 @pytest_asyncio.fixture
 async def zmq_proxy() -> _t.AsyncGenerator[ZMQProxy, None]:
     """Fixture for ZMQProxy instance."""
diff --git a/uv.lock b/uv.lock
index a11b365c..df9700e2 100644
--- a/uv.lock
+++ b/uv.lock
@@ -3795,6 +3795,9 @@ llm = [
     { name = "llama-index-core" },
     { name = "llama-index-llms-openai" },
 ]
+omq = [
+    { name = "pyomq" },
+]
 ray = [
     { name = "jsonschema" },
     { name = "optuna" },
@@ -3829,6 +3832,7 @@ all = [
     { name = "openai-responses" },
     { name = "optuna" },
     { name = "pre-commit" },
+    { name = "pyomq" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-benchmark" },
@@ -3875,6 +3879,7 @@ test = [
     { name = "moto", extra = ["server"] },
     { name = "openai-responses" },
     { name = "optuna" },
+    { name = "pyomq" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-benchmark" },
@@ -3908,6 +3913,7 @@ requires-dist = [
     { name = "pydantic", marker = "python_full_version < '3.14'", specifier = ">=2.8.0,<3" },
     { name = "pydantic", marker = "python_full_version >= '3.14'", specifier = ">=2.13.1,<3" },
     { name = "pydantic-settings", specifier = ">=2.7.1,<3" },
+    { name = "pyomq", marker = "extra == 'omq'", specifier = ">=0.20.1,<1" },
     { name = "pyzmq", specifier = ">=26.2,<28" },
     { name = "ray", extras = ["tune"], marker = "extra == 'ray'", specifier = ">=2.47.1,<3" },
     { name = "redis", marker = "extra == 'redis'", specifier = ">=7.1,<9" },
@@ -3920,7 +3926,7 @@ requires-dist = [
     { name = "uvloop", marker = "sys_platform != 'win32'", specifier = ">=0.21.0,<1" },
     { name = "websockets", marker = "extra == 'websockets'", specifier = ">=14.2,<18" },
 ]
-provides-extras = ["aws", "azure", "gcp", "llm", "ray", "redis", "websockets"]
+provides-extras = ["aws", "azure", "gcp", "llm", "omq", "ray", "redis", "websockets"]
 
 [package.metadata.requires-dev]
 all = [
@@ -3944,6 +3950,7 @@ all = [
     { name = "openai-responses", specifier = ">=0.11.4,<1" },
     { name = "optuna", specifier = ">=3.0,<5" },
     { name = "pre-commit", specifier = ">=3.8,<5" },
+    { name = "pyomq", specifier = ">=0.20.1,<1" },
     { name = "pytest", specifier = ">=8.3,<10" },
     { name = "pytest-asyncio", specifier = ">=1.4,<2" },
     { name = "pytest-benchmark", specifier = ">=5.1.0" },
@@ -3990,6 +3997,7 @@ test = [
     { name = "moto", extras = ["server"], specifier = ">=5.0,<6" },
     { name = "openai-responses", specifier = ">=0.11.4,<1" },
     { name = "optuna", specifier = ">=3.0,<5" },
+    { name = "pyomq", specifier = ">=0.20.1,<1" },
     { name = "pytest", specifier = ">=8.3,<10" },
     { name = "pytest-asyncio", specifier = ">=1.4,<2" },
     { name = "pytest-benchmark", specifier = ">=5.1.0" },
@@ -4459,6 +4467,22 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f7/27/a2fc51a4a122dfd1015e921ae9d22fee3d20b0b8080d9a704578bf9deece/pymdown_extensions-10.21.2-py3-none-any.whl", hash = "sha256:5c0fd2a2bea14eb39af8ff284f1066d898ab2187d81b889b75d46d4348c01638", size = 268901, upload-time = "2026-03-29T15:01:53.244Z" },
 ]
 
+[[package]]
+name = "pyomq"
+version = "0.20.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/4b/f8/80ab2fc7b7ecab59bcdd4ddcae72dbd819e708ae12109a7005b473510486/pyomq-0.20.1.tar.gz", hash = "sha256:2f42c4c48666fdc4b621e9e7c9d5c94082ef3c9147f35ceb5c3eb42e6ed23819", size = 649938, upload-time = "2026-08-23T21:03:55.159Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/4c/b5/fd6a4a81754a5b5736f15f1a6215752f38e61cca41c7ff420de63a7448ae/pyomq-0.20.1-cp311-abi3-macosx_10_12_x86_64.whl", hash = "sha256:0b62acc91d0882889cba23aca60530062f6109bebc5b1c68a8aea29334f75b82", size = 1779867, upload-time = "2026-08-23T21:03:44.028Z" },
+    { url = "https://files.pythonhosted.org/packages/ba/d7/a14d7176942a9bca468584c7147cf6e6eb86574102c7e1bc4dc7fadaffdf/pyomq-0.20.1-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:9f737cabb6f3f6300c55236f2f6f4ded09a6a6d31b0e8f381965f9f4f16c436c", size = 1725287, upload-time = "2026-08-23T21:03:45.442Z" },
+    { url = "https://files.pythonhosted.org/packages/02/67/5b4a4c648f945ad3b8dc5ac1dac1495a89e22d62795a922bf1e5759e9210/pyomq-0.20.1-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7362231d5e222f30ad43ee6e66572c98b665dde282fbcb94c0952282d873d0af", size = 1806696, upload-time = "2026-08-23T21:03:46.781Z" },
+    { url = "https://files.pythonhosted.org/packages/c9/d1/7d0baccf1eaa34e25efcffd4e750e198af1dd3a703a8abda1f0cf1ba1dc3/pyomq-0.20.1-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ebe3ea509a7de86be592763ce90af4c53065701bd4b9f58c99887df2d389a269", size = 1829920, upload-time = "2026-08-23T21:03:48.24Z" },
+    { url = "https://files.pythonhosted.org/packages/d9/dc/c9e5c16b2925761dee05f0c430d68e4772b1347620a9b84111c51a8633e1/pyomq-0.20.1-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:85f47fa8c6a66a9ee0cddc8c7ec940582e9bb80dcc555c1c1fae182c15764d42", size = 1983314, upload-time = "2026-08-23T21:03:49.594Z" },
+    { url = "https://files.pythonhosted.org/packages/c5/c8/aca1a3a033f4ca1f8359eae1cb5e458a9798181d43bfbf474b4ce6a357a1/pyomq-0.20.1-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:1c57a649272d11f162b4d8ca126b0096278a38fb27a51e3319f8e4e4dd744555", size = 2041966, upload-time = "2026-08-23T21:03:51.117Z" },
+    { url = "https://files.pythonhosted.org/packages/e8/32/217f446433516bcf168d4c262e31c30006fda30eabbe8cf3e259b7b0caed/pyomq-0.20.1-cp311-abi3-win_amd64.whl", hash = "sha256:4aa0ad5b91adeb264739944c4ba505e73885db1a04ac2a7bf6d6956db14e6a9a", size = 1874489, upload-time = "2026-08-23T21:03:52.48Z" },
+    { url = "https://files.pythonhosted.org/packages/b1/61/6c5469e45eaa62325eaf1ca9c6258fe28534f5d763498dd5c58c5aa2e807/pyomq-0.20.1-cp311-abi3-win_arm64.whl", hash = "sha256:24361da804107a7431cfaba3c6bb49d89746441827f9532875d7c3b84725bd64", size = 1750463, upload-time = "2026-08-23T21:03:53.747Z" },
+]
+
 [[package]]
 name = "pyparsing"
 version = "3.3.2"