Transport

The pluggable message delivery layer. Swap transports by changing one line in your topology YAML — agent code is unchanged.

See Transports for a full guide with architecture diagrams.

---

Protocol

civitas.transport.Transport

Bases: Protocol

Protocol that all transports implement.

Five methods. A new transport plugin implements these five methods and the entire Civitas runtime works on it.

start() async

Initialize connections, bind sockets.

stop() async

Gracefully close connections, flush pending messages.

subscribe(address, handler) async

Register a handler for messages arriving at this address.

unsubscribe(address) async

Remove the handler for an address. No-op if not subscribed.

publish(address, data) async

Send a message to an address (fire-and-forget).

request(address, data, timeout) async

Send a message and await a reply (request-reply).

wait_ready() async

Wait for connections and subscriptions to stabilize. No-op by default.

Transports with slow-joiner problems (e.g. ZMQ PUB/SUB) should override this to sleep or poll until messages will be reliably delivered. Callers invoke this after all subscribe() calls are done but before publishing.

has_reply_address(address)

Return True if address is an active ephemeral reply endpoint.

Ephemeral reply addresses are created by transport.request() and are not registered agents. The bus uses this to route reply messages without going through the Registry.

wait_subscribed(address, timeout=2.0) async

Block until the subscription for address is effective for peers.

Transports where subscription takes effect synchronously (in-process dict insert) or broker-side (NATS) implement this as a no-op / flush. Transports with asynchronous subscription propagation to publisher sockets (ZMQ PUB/SUB) must confirm propagation — announcing a freshly-subscribed address before its subscription reaches peer PUB sockets makes peers publish into a void (#41).

set_serializer(serializer)

Replace the serializer this transport uses for its OWN internal request()/reply-address bookkeeping (v0.9.2.1 bugfix).

Every transport that implements request() holds its own private serializer reference, captured at construction, separate from the MessageBus's. request() needs to inject reply_to into the message it was given — already-serialized bytes from the bus — which means deserializing and re-serializing internally. If the bus's serializer is swapped later (e.g. Runtime.start() activating message signing) without ALSO swapping the transport's own reference, that internal round-trip keeps using the stale one: it calls Message.from_dict() directly on a signed v2 envelope dict ({"v": 2, "msg": {...}, "sig": {...}}), none of which are real Message fields, silently reconstructing a blank message (found the hard way: ask() over ZMQ with signing enabled just times out, no exception, because the resulting blank correlation_id makes the reply-routing check in AgentProcess._dispatch() silently no-op). Call this whenever the bus's serializer changes after construction.

---

Implementations

civitas.transport.inprocess.InProcessTransport(serializer, mailbox_size=1000)

Transport for single-process deployments.

Messages are delivered by putting serialized bytes into the recipient's asyncio.Queue. Despite being in-process, messages are still serialized through the configured Serializer to ensure transport-swap compatibility.

Source code in civitas/transport/inprocess.py

def __init__(self, serializer: Serializer, mailbox_size: int = 1000) -> None:
    self._serializer = serializer
    self._mailbox_size = mailbox_size
    self._handlers: dict[str, Callable[[bytes], Awaitable[None]]] = {}
    self._reply_queues: dict[str, asyncio.Queue[bytes]] = {}
    self._started = False

wait_ready() async

No-op for in-process transport — always ready after start().

Source code in civitas/transport/inprocess.py

async def wait_ready(self) -> None:
    """No-op for in-process transport — always ready after start()."""

request(address, data, timeout) async

Send a request and await a reply.

Creates a temporary reply address, injects reply_to into the message, publishes the request, and awaits the reply with a timeout.

Source code in civitas/transport/inprocess.py

async def request(self, address: str, data: bytes, timeout: float | None) -> bytes:
    """Send a request and await a reply.

    Creates a temporary reply address, injects reply_to into the message,
    publishes the request, and awaits the reply with a timeout.
    """
    reply_address = f"_reply.{_uuid7()}"
    reply_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=1)
    self._reply_queues[reply_address] = reply_queue

    try:
        # Deserialize to inject reply_to, then re-serialize
        message = self._serializer.deserialize(data)
        message.reply_to = reply_address
        data = self._serializer.serialize(message)

        # Publish the request
        handler = self._handlers.get(address)
        if handler is None:
            raise RuntimeError(f"No handler registered for address: {address}")
        await handler(data)

        # Await the reply
        async with asyncio.timeout(timeout):
            reply_data = await reply_queue.get()
        return reply_data
    finally:
        self._reply_queues.pop(reply_address, None)

has_reply_address(address)

Return True if address is an active ephemeral reply queue.

Source code in civitas/transport/inprocess.py

def has_reply_address(self, address: str) -> bool:
    """Return True if address is an active ephemeral reply queue."""
    return address in self._reply_queues

set_serializer(serializer)

Replace the serializer used by request()'s internal reply_to round-trip (v0.9.2.1 bugfix). Harmless no-op in effect for this transport specifically — message signing is deliberately skipped for in-process transport by design (D9, same OS process, no wire to protect) — but implemented for interface consistency across all three transports (Transport.set_serializer's docstring has the full story).

Source code in civitas/transport/inprocess.py

def set_serializer(self, serializer: Serializer) -> None:
    """Replace the serializer used by request()'s internal reply_to
    round-trip (v0.9.2.1 bugfix). Harmless no-op in effect for this
    transport specifically — message signing is deliberately skipped for
    in-process transport by design (D9, same OS process, no wire to
    protect) — but implemented for interface consistency across all three
    transports (Transport.set_serializer's docstring has the full story).
    """
    self._serializer = serializer

wait_subscribed(address, timeout=2.0) async

No-op — in-process subscription is a dict insert, effective immediately.

Source code in civitas/transport/inprocess.py

async def wait_subscribed(self, address: str, timeout: float = 2.0) -> None:
    """No-op — in-process subscription is a dict insert, effective immediately."""

---

civitas.transport.zmq.ZMQTransport(serializer, pub_addr='tcp://127.0.0.1:5559', sub_addr='tcp://127.0.0.1:5560', start_proxy=False, curve_config=None)

Transport for multi-process deployments using ZeroMQ.

Implements the five-method Transport protocol. Messages flow through an XSUB/XPUB proxy for PUB/SUB delivery. Request-reply uses temporary PUB/SUB topics with reply queues, identical to InProcessTransport.

Parameters:

Name	Type	Description	Default
serializer	Serializer	Serializer for message encode/decode.	required
pub_addr	str	Address of the proxy XSUB frontend (PUB connects here).	'tcp://127.0.0.1:5559'
sub_addr	str	Address of the proxy XPUB backend (SUB connects here).	'tcp://127.0.0.1:5560'
start_proxy	bool	If True, start a ZMQProxy in this process.	False

Source code in civitas/transport/zmq.py

def __init__(
    self,
    serializer: Serializer,
    pub_addr: str = "tcp://127.0.0.1:5559",
    sub_addr: str = "tcp://127.0.0.1:5560",
    start_proxy: bool = False,
    curve_config: ZmqCurveConfig | None = None,
) -> None:
    self._serializer = serializer
    self._pub_addr = pub_addr
    self._sub_addr = sub_addr
    self._start_proxy = start_proxy
    self._curve_config = curve_config

    self._context: zmq.asyncio.Context | None = None
    self._pub: zmq.asyncio.Socket | None = None
    self._sub: zmq.asyncio.Socket | None = None
    self._proxy: ZMQProxy | None = None

    self._handlers: dict[str, Callable[[bytes], Awaitable[None]]] = {}
    self._reply_queues: dict[str, asyncio.Queue[bytes]] = {}
    self._receiver_task: asyncio.Task[None] | None = None
    self._started = False
    # Stable per-transport reply-topic prefix (#41): ONE prefix subscription
    # at start() — covered by wait_ready — instead of a fresh subscription per
    # request. A per-request subscription races its own first use: the
    # replier's PUB socket drops the reply unless the ephemeral topic's
    # subscription has propagated (SUB → XPUB → XSUB → peer PUBs, 5–25 ms),
    # and replies typically arrive in ~1 ms. ZMQ prefix matching delivers
    # every _reply.<iid>.<req> topic to us with zero per-request churn.
    self._reply_prefix = f"_reply.{_uuid7()}."

start() async

Initialize sockets and connect to the proxy.

Source code in civitas/transport/zmq.py

async def start(self) -> None:
    """Initialize sockets and connect to the proxy."""
    if self._started:
        return

    if self._start_proxy:
        self._proxy = ZMQProxy(
            frontend=self._pub_addr,
            backend=self._sub_addr,
            curve_config=self._curve_config,
        )
        # Run blocking proxy start in a thread executor to avoid blocking
        # the event loop during the ready-wait.
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, self._proxy.start)

    self._context = zmq.asyncio.Context()

    # PUB connects to proxy XSUB frontend
    self._pub = self._context.socket(zmq.PUB)
    # SUB connects to proxy XPUB backend
    self._sub = self._context.socket(zmq.SUB)

    if self._curve_config is not None and self._curve_config.enabled:
        cfg = self._curve_config
        for sock in (self._pub, self._sub):
            sock.curve_serverkey = cfg.server_public_key.encode()
            sock.curve_secretkey = cfg.client_secret_key.encode()
            sock.curve_publickey = cfg.client_public_key.encode()

    self._pub.connect(self._pub_addr)
    self._sub.connect(self._sub_addr)

    # One prefix subscription covers every future reply topic (#41); its
    # propagation is absorbed by the startup wait_ready(), not by requests.
    self._sub.subscribe(self._reply_prefix.encode())

    # Start background receiver
    self._receiver_task = asyncio.create_task(self._receiver_loop())

    self._started = True

wait_ready() async

Wait for ZMQ connections and subscriptions to stabilize.

Call after all subscribe() calls are done. Mitigates the ZMQ 'slow joiner' problem where PUB/SUB needs time for the connection handshake and subscription propagation through the proxy.

Source code in civitas/transport/zmq.py

async def wait_ready(self) -> None:
    """Wait for ZMQ connections and subscriptions to stabilize.

    Call after all subscribe() calls are done. Mitigates the ZMQ
    'slow joiner' problem where PUB/SUB needs time for the connection
    handshake and subscription propagation through the proxy.
    """
    await asyncio.sleep(0.3)

stop() async

Close sockets, stop proxy, clean up.

Source code in civitas/transport/zmq.py

async def stop(self) -> None:
    """Close sockets, stop proxy, clean up."""
    self._started = False

    if self._receiver_task is not None:
        self._receiver_task.cancel()
        try:
            await self._receiver_task
        except asyncio.CancelledError:
            pass

    if self._pub is not None:
        self._pub.close(linger=0)
    if self._sub is not None:
        self._sub.close(linger=0)
    if self._context is not None:
        self._context.term()

    if self._proxy is not None:
        self._proxy.stop()

    self._handlers.clear()
    self._reply_queues.clear()

subscribe(address, handler) async

Subscribe to messages arriving at this address.

Source code in civitas/transport/zmq.py

async def subscribe(self, address: str, handler: Callable[[bytes], Awaitable[None]]) -> None:
    """Subscribe to messages arriving at this address."""
    if self._sub is None:
        raise RuntimeError("ZMQTransport not started")
    self._handlers[address] = handler
    self._sub.subscribe(address.encode() + _TOPIC_SEP)

unsubscribe(address) async

Remove a handler and best-effort unsubscribe the SUB socket.

Source code in civitas/transport/zmq.py

async def unsubscribe(self, address: str) -> None:
    """Remove a handler and best-effort unsubscribe the SUB socket."""
    self._handlers.pop(address, None)
    if self._sub is not None:
        try:
            self._sub.unsubscribe(address.encode() + _TOPIC_SEP)
        except zmq.ZMQError as exc:
            logger.warning("[ZMQTransport] unsubscribe(%r) failed: %s", address, exc)

publish(address, data) async

Send a message to an address via PUB/SUB through the proxy.

Same-process reply queues are checked first (short-circuit for local request-reply without going through the proxy).

Source code in civitas/transport/zmq.py

async def publish(self, address: str, data: bytes) -> None:
    """Send a message to an address via PUB/SUB through the proxy.

    Same-process reply queues are checked first (short-circuit for
    local request-reply without going through the proxy).
    """
    # Short-circuit for local reply queues
    if address in self._reply_queues:
        await self._reply_queues[address].put(data)
        return

    if self._pub is None:
        raise RuntimeError("ZMQTransport not started")
    topic = address.encode() + _TOPIC_SEP
    await self._pub.send_multipart([topic, data])

request(address, data, timeout) async

Send a request and await a reply over PUB/SUB.

Creates a temporary reply topic, subscribes to it, injects reply_to into the message, publishes the request, and awaits the reply.

Source code in civitas/transport/zmq.py

async def request(self, address: str, data: bytes, timeout: float | None) -> bytes:
    """Send a request and await a reply over PUB/SUB.

    Creates a temporary reply topic, subscribes to it, injects reply_to
    into the message, publishes the request, and awaits the reply.
    """
    # Rides the stable per-transport prefix subscription made in start()
    # (#41) — no per-request subscribe, so the replier's PUB already knows
    # the prefix and the reply cannot be dropped by subscription lag.
    reply_address = f"{self._reply_prefix}{_uuid7()}"
    reply_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=1)

    if self._sub is None:
        raise RuntimeError("ZMQTransport not started")
    self._reply_queues[reply_address] = reply_queue

    try:
        # Inject reply_to and re-serialize
        message = self._serializer.deserialize(data)
        message.reply_to = reply_address
        data = self._serializer.serialize(message)

        # Publish the request
        if self._pub is None:
            raise RuntimeError("ZMQTransport not started")
        topic = address.encode() + _TOPIC_SEP
        await self._pub.send_multipart([topic, data])

        # Await the reply
        async with asyncio.timeout(timeout):
            reply_data = await reply_queue.get()
        return reply_data
    finally:
        self._reply_queues.pop(reply_address, None)

has_reply_address(address)

Return True if address is an active ephemeral reply queue.

Source code in civitas/transport/zmq.py

def has_reply_address(self, address: str) -> bool:
    """Return True if address is an active ephemeral reply queue."""
    return address in self._reply_queues

set_serializer(serializer)

Replace the serializer used by request()'s internal reply_to round-trip (v0.9.2.1 bugfix — see Transport.set_serializer's docstring for the full story: this transport's own serializer reference was never updated when Runtime.start() activated message signing, silently corrupting every ask() into a blank message).

Source code in civitas/transport/zmq.py

def set_serializer(self, serializer: Serializer) -> None:
    """Replace the serializer used by request()'s internal reply_to
    round-trip (v0.9.2.1 bugfix — see Transport.set_serializer's
    docstring for the full story: this transport's own serializer
    reference was never updated when Runtime.start() activated message
    signing, silently corrupting every ask() into a blank message).
    """
    self._serializer = serializer

wait_subscribed(address, timeout=2.0) async

Block until the subscription for address has propagated to PUB peers.

ZMQ PUB sockets silently drop messages for topics no subscriber is known for, and subscription propagation (SUB → XPUB → XSUB → every PUB) is asynchronous — measured at ~5–10 ms on Linux and ~20–25 ms on macOS over IPC (#41). Announcing a freshly-subscribed address before propagation completes makes peers publish into a void.

Mechanism: subscribe a throwaway probe topic on the SAME SUB socket after address was subscribed — same pipe, FIFO, so the probe's subscription cannot overtake it — then self-publish to the probe topic until one frame loops back through the proxy. When the probe returns, the address subscription has propagated at least as far as this process's own PUB pipe; peer PUB pipes receive the same XSUB broadcast in parallel (residual window: pipe jitter, sub-millisecond — the full at-least-once route-establishment guarantee is deferred; see docs/design/cross-process-spawn.md addendum).

Raises:

Type	Description
TimeoutError	if the probe does not loop back within timeout (proxy down or SUB pipe stalled) — callers should log and degrade rather than fail the spawn.

Source code in civitas/transport/zmq.py

async def wait_subscribed(self, address: str, timeout: float = 2.0) -> None:
    """Block until the subscription for ``address`` has propagated to PUB peers.

    ZMQ PUB sockets silently drop messages for topics no subscriber is known
    for, and subscription propagation (SUB → XPUB → XSUB → every PUB) is
    asynchronous — measured at ~5–10 ms on Linux and ~20–25 ms on macOS over
    IPC (#41). Announcing a freshly-subscribed address before propagation
    completes makes peers publish into a void.

    Mechanism: subscribe a throwaway probe topic on the SAME SUB socket
    *after* ``address`` was subscribed — same pipe, FIFO, so the probe's
    subscription cannot overtake it — then self-publish to the probe topic
    until one frame loops back through the proxy. When the probe returns,
    the ``address`` subscription has propagated at least as far as this
    process's own PUB pipe; peer PUB pipes receive the same XSUB broadcast
    in parallel (residual window: pipe jitter, sub-millisecond — the full
    at-least-once route-establishment guarantee is deferred; see
    docs/design/cross-process-spawn.md addendum).

    Raises:
        TimeoutError: if the probe does not loop back within ``timeout``
            (proxy down or SUB pipe stalled) — callers should log and
            degrade rather than fail the spawn.
    """
    if self._sub is None or self._pub is None:
        raise RuntimeError("ZMQTransport not started")
    if address not in self._handlers:
        raise ValueError(f"wait_subscribed({address!r}): address is not subscribed")

    probe_topic = f"_probe.{_uuid7()}"
    probe_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=1)
    self._sub.subscribe(probe_topic.encode() + _TOPIC_SEP)
    self._reply_queues[probe_topic] = probe_queue
    try:
        async with asyncio.timeout(timeout):
            while True:
                await self._pub.send_multipart([probe_topic.encode() + _TOPIC_SEP, b""])
                try:
                    async with asyncio.timeout(0.02):
                        await probe_queue.get()
                    return
                except TimeoutError:
                    continue  # not propagated yet — probe again
    finally:
        self._reply_queues.pop(probe_topic, None)
        self._sub.unsubscribe(probe_topic.encode() + _TOPIC_SEP)

---

civitas.transport.nats.NATSTransport(serializer, servers='nats://localhost:4222', jetstream=False, stream_name='AGENCY', create_stream_if_missing=True, tls_config=None)

Transport for distributed deployments using NATS.

Implements the five-method Transport protocol. Messages flow through a NATS server. Request-reply uses temporary subscriptions with reply addresses, consistent with InProcess and ZMQ transports.

Parameters:

Name	Type	Description	Default
serializer	Serializer	Serializer for message encode/decode.	required
servers	str | list[str]	NATS server URL(s) to connect to.	'nats://localhost:4222'
jetstream	bool	If True, use JetStream durable subscriptions.	False
stream_name	str	JetStream stream name (only used if jetstream=True).	'AGENCY'

Source code in civitas/transport/nats.py

def __init__(
    self,
    serializer: Serializer,
    servers: str | list[str] = "nats://localhost:4222",
    jetstream: bool = False,
    stream_name: str = "AGENCY",
    create_stream_if_missing: bool = True,
    tls_config: NatsTlsConfig | None = None,
) -> None:
    self._serializer = serializer
    self._servers = servers if isinstance(servers, list) else [servers]
    self._use_jetstream = jetstream
    self._stream_name = stream_name
    self._create_stream_if_missing = create_stream_if_missing
    self._tls_config = tls_config

    self._nc: NATSClient | None = None
    self._js: nats.js.JetStreamContext | None = None
    self._handlers: dict[str, Callable[[bytes], Awaitable[None]]] = {}
    self._subscriptions: dict[str, Any] = {}  # Subscription or PushSubscription
    self._reply_queues: dict[str, asyncio.Queue[bytes]] = {}
    self._started = False

start() async

Connect to the NATS server.

Source code in civitas/transport/nats.py

async def start(self) -> None:
    """Connect to the NATS server."""
    if self._started:
        return

    async def _on_disconnected() -> None:
        logger.warning("[NATSTransport] disconnected from server")

    async def _on_reconnected() -> None:
        logger.info("[NATSTransport] reconnected to server")

    async def _on_error(exc: Exception) -> None:
        logger.error("[NATSTransport] error: %s", exc)

    connect_kwargs: dict[str, Any] = {
        "servers": self._servers,
        "disconnected_cb": _on_disconnected,
        "reconnected_cb": _on_reconnected,
        "error_cb": _on_error,
    }

    if self._tls_config is not None and self._tls_config.enabled:
        connect_kwargs["tls"] = self._tls_config.build_ssl_context()

    if self._tls_config is not None and self._tls_config.nkey_seed:
        try:
            import nkeys
        except ImportError as exc:
            raise ImportError(
                "NATS nkeys auth requires 'nkeys'. "
                "Install it with: pip install 'civitas[nkeys]'"
            ) from exc
        connect_kwargs["nkeys_seed"] = nkeys.from_seed(self._tls_config.nkey_seed.encode())

    self._nc = await nats.connect(**connect_kwargs)

    if self._use_jetstream:
        self._js = self._nc.jetstream()
        try:
            await self._js.find_stream_name_by_subject(f"{_SUBJECT_PREFIX}>")
        except nats.js.errors.NotFoundError as exc:
            if not self._create_stream_if_missing:
                raise RuntimeError(
                    f"JetStream stream '{self._stream_name}' not found and "
                    f"create_stream_if_missing=False"
                ) from exc
            logger.warning(
                "[NATSTransport] auto-creating JetStream stream '%s' — "
                "set create_stream_if_missing=False in production",
                self._stream_name,
            )
            await self._js.add_stream(
                name=self._stream_name,
                subjects=[f"{_SUBJECT_PREFIX}>"],
            )

    self._started = True

wait_ready() async

No-op for NATS — connection is established synchronously in start().

Source code in civitas/transport/nats.py

async def wait_ready(self) -> None:
    """No-op for NATS — connection is established synchronously in start()."""

stop() async

Disconnect from NATS, clean up subscriptions.

Source code in civitas/transport/nats.py

async def stop(self) -> None:
    """Disconnect from NATS, clean up subscriptions."""
    self._started = False

    for sub in self._subscriptions.values():
        try:
            await sub.unsubscribe()
        except Exception:  # noqa: BLE001 — best-effort cleanup during shutdown
            continue
    self._subscriptions.clear()

    if self._nc is not None and self._nc.is_connected:
        await self._nc.drain()
        self._nc = None

    self._handlers.clear()
    self._reply_queues.clear()

subscribe(address, handler) async

Subscribe to messages arriving at this address.

Source code in civitas/transport/nats.py

async def subscribe(self, address: str, handler: Callable[[bytes], Awaitable[None]]) -> None:
    """Subscribe to messages arriving at this address."""
    if self._nc is None:
        raise RuntimeError("NATSTransport not started")
    self._handlers[address] = handler
    subject = self._to_subject(address)

    async def _on_msg(msg: Msg) -> None:
        # Reply queues take priority (for request-reply responses)
        if address in self._reply_queues:
            await self._reply_queues[address].put(msg.data)
            return
        h = self._handlers.get(address)
        if h is not None:
            await h(msg.data)

    sub: Any
    if self._use_jetstream and self._js is not None:
        sub = await self._js.subscribe(
            subject,
            durable=address.replace(".", "_").replace("-", "_"),
            cb=_on_msg,
        )
    else:
        sub = await self._nc.subscribe(subject, cb=_on_msg)

    self._subscriptions[address] = sub

unsubscribe(address) async

Remove a handler and best-effort unsubscribe the NATS subscription.

Source code in civitas/transport/nats.py

async def unsubscribe(self, address: str) -> None:
    """Remove a handler and best-effort unsubscribe the NATS subscription."""
    self._handlers.pop(address, None)
    sub = self._subscriptions.pop(address, None)
    if sub is not None:
        try:
            await sub.unsubscribe()
        except Exception as exc:  # noqa: BLE001 — best-effort cleanup
            logger.warning("[NATSTransport] unsubscribe(%r) failed: %s", address, exc)

publish(address, data) async

Publish a message to an address (fire-and-forget).

Checks local reply queues first (same-process request-reply short-circuit), then publishes via NATS.

Source code in civitas/transport/nats.py

async def publish(self, address: str, data: bytes) -> None:
    """Publish a message to an address (fire-and-forget).

    Checks local reply queues first (same-process request-reply
    short-circuit), then publishes via NATS.
    """
    # Short-circuit for local reply queues
    if address in self._reply_queues:
        await self._reply_queues[address].put(data)
        return

    if self._nc is None:
        raise RuntimeError("NATSTransport not started")
    subject = self._to_subject(address)
    await self._nc.publish(subject, data)

request(address, data, timeout) async

Send a request and await a reply.

Creates a temporary reply address, subscribes to it, injects reply_to into the message, publishes the request, and awaits the reply. Mirrors the InProcess/ZMQ pattern for consistency.

Source code in civitas/transport/nats.py

async def request(self, address: str, data: bytes, timeout: float | None) -> bytes:
    """Send a request and await a reply.

    Creates a temporary reply address, subscribes to it, injects reply_to
    into the message, publishes the request, and awaits the reply.
    Mirrors the InProcess/ZMQ pattern for consistency.
    """
    if self._nc is None:
        raise RuntimeError("NATSTransport not started")

    reply_address = f"_reply.{_uuid7()}"
    reply_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=1)
    reply_subject = self._to_subject(reply_address)

    # Subscribe to the reply subject
    async def _on_reply(msg: Msg) -> None:
        if reply_address in self._reply_queues:
            await self._reply_queues[reply_address].put(msg.data)

    sub = await self._nc.subscribe(reply_subject, cb=_on_reply)
    self._reply_queues[reply_address] = reply_queue

    try:
        # Inject reply_to and re-serialize
        message = self._serializer.deserialize(data)
        message.reply_to = reply_address
        data = self._serializer.serialize(message)

        # Publish the request
        subject = self._to_subject(address)
        await self._nc.publish(subject, data)
        await self._nc.flush()

        # Await the reply
        async with asyncio.timeout(timeout):
            reply_data = await reply_queue.get()
        return reply_data
    finally:
        self._reply_queues.pop(reply_address, None)
        await sub.unsubscribe()

has_reply_address(address)

Return True if address is an active ephemeral reply queue.

Source code in civitas/transport/nats.py

def has_reply_address(self, address: str) -> bool:
    """Return True if address is an active ephemeral reply queue."""
    return address in self._reply_queues

set_serializer(serializer)

Replace the serializer used by request()'s internal reply_to round-trip (v0.9.2.1 bugfix — see Transport.set_serializer's docstring for the full story: this transport's own serializer reference was never updated when Runtime.start() activated message signing, silently corrupting every ask() into a blank message).

Source code in civitas/transport/nats.py

def set_serializer(self, serializer: Serializer) -> None:
    """Replace the serializer used by request()'s internal reply_to
    round-trip (v0.9.2.1 bugfix — see Transport.set_serializer's
    docstring for the full story: this transport's own serializer
    reference was never updated when Runtime.start() activated message
    signing, silently corrupting every ask() into a blank message).
    """
    self._serializer = serializer

wait_subscribed(address, timeout=2.0) async

Flush the connection — the broker owns routing, so once the server has processed the SUB (flush round-trips it), peers' publishes are routed.

Source code in civitas/transport/nats.py

async def wait_subscribed(self, address: str, timeout: float = 2.0) -> None:
    """Flush the connection — the broker owns routing, so once the server has
    processed the SUB (flush round-trips it), peers' publishes are routed."""
    if self._nc is not None and self._nc.is_connected:
        await self._nc.flush(timeout=max(1, int(timeout)))

---

Worker

civitas.worker.Worker(agents, transport='zmq', zmq_pub_addr='tcp://127.0.0.1:5559', zmq_sub_addr='tcp://127.0.0.1:5560', nats_servers='nats://localhost:4222', nats_jetstream=False, serializer=None, model_provider=None, tool_registry=None, state_store=None, exporters=None, max_restarts=3, components=None)

Hosts agents in a worker process, connecting to an existing broker.

Supports ZMQ (connect to proxy) and NATS (connect to server) transports.

The Worker provides: - Transport connectivity (ZMQ or NATS) - Local registry and message bus - Heartbeat auto-response (handled by AgentProcess._message_loop) - Restart command handling (_agency.restart messages)

Source code in civitas/worker.py

def __init__(
    self,
    agents: list[AgentProcess],
    transport: str = "zmq",
    zmq_pub_addr: str = "tcp://127.0.0.1:5559",
    zmq_sub_addr: str = "tcp://127.0.0.1:5560",
    nats_servers: str | list[str] = "nats://localhost:4222",
    nats_jetstream: bool = False,
    serializer: Serializer | None = None,
    model_provider: Any = None,
    tool_registry: Any = None,
    state_store: Any = None,
    exporters: list[Any] | None = None,
    max_restarts: int = 3,
    components: ComponentSet | None = None,
) -> None:
    self._transport_type = transport
    self._zmq_pub_addr = zmq_pub_addr
    self._zmq_sub_addr = zmq_sub_addr
    self._nats_servers = nats_servers
    self._nats_jetstream = nats_jetstream
    self._custom_serializer = serializer
    self._model_provider = model_provider
    self._tool_registry = tool_registry
    self._state_store = state_store
    self._exporters = exporters or []
    self._max_restarts = max_restarts
    self._components = components
    self._otel_agent_task: asyncio.Task[None] | None = None

    # O(1) agent lookup by name (F02-8)
    self._agents: dict[str, AgentProcess] = {a.name: a for a in agents}
    self._restart_counts: dict[str, int] = {a.name: 0 for a in agents}

    # D5 (v0.9.0): per-process liveness identity + channel. Answered at the
    # TRANSPORT level (no mailbox, no agent) so liveness is never conflated
    # with any agent's dispatch latency — the A6 structural fix.
    self.id: str = _uuid7()
    self._health_channel = f"_agency.worker.{self.id}.health"
    # v0.9.1 (dashboard-v2, D-DASH-3): primed ONCE, reused for every probe
    # this Worker ever answers — see try_start_process_sampler()'s
    # docstring for why a fresh handle per-probe would be a real bug, not
    # a style choice. None (and the health-ack payload simply omits
    # 'process') when psutil isn't installed — never a hard dependency.
    self._resource_sampler = try_start_process_sampler()

    # Set during start()
    self._serializer: Serializer | None = None
    self._transport: Any = None
    self._registry: Any = None
    self._bus: Any = None
    self._started = False
    self._stop_event = asyncio.Event()

start() async

Start the worker: connect to proxy, wire agents, start loops.

Source code in civitas/worker.py

async def start(self) -> None:
    """Start the worker: connect to proxy, wire agents, start loops."""
    # Workers never use in-process transport — they connect to an existing broker
    if self._components is None and self._transport_type not in ("zmq", "nats"):
        raise ConfigurationError(
            f"Unknown transport: {self._transport_type!r}. Expected 'zmq' or 'nats'."
        )

    # Build or use provided ComponentSet
    if self._components is not None:
        cs = self._components
    else:
        cs = build_component_set(
            transport_type=self._transport_type,
            serializer=self._custom_serializer,
            model_provider=self._model_provider,
            tool_registry=self._tool_registry,
            state_store=self._state_store,
            exporters=self._exporters,
            zmq_pub_addr=self._zmq_pub_addr,
            zmq_sub_addr=self._zmq_sub_addr,
            zmq_start_proxy=False,  # Workers connect to an existing proxy
            nats_servers=self._nats_servers,
            nats_jetstream=self._nats_jetstream,
        )

    # Expose on self for _on_restart_command and stop()
    self._serializer = cs.serializer
    self._transport = cs.transport
    self._registry = cs.registry
    self._bus = cs.bus
    self._components = cs  # D1a: re-invokable wiring for fresh incarnations

    # Drain span_queue via OTELAgent when exporters are configured (FD-07/FD-09)
    if cs.span_queue is not None and cs.export_backend is not None:
        self._otel_agent_task = asyncio.create_task(
            run_otel_agent(cs.span_queue, cs.export_backend)
        )

    # Start transport first — must be running before setup_agent (F02-16)
    await self._transport.start()

    # Wait for ZMQ subscriptions to propagate (slow joiner).
    # Also lets the worker's PUB socket establish its connection to the proxy
    # before we publish registration announcements below.
    if hasattr(self._transport, "wait_ready"):
        await self._transport.wait_ready()

    # Wire and subscribe agents
    for agent in self._agents.values():
        cs.inject(agent)
        self._registry.register(agent.name, capabilities=self._agent_capabilities(agent))
        await self._bus.setup_agent(agent)

    # Subscribe to restart commands and register in registry so bus.route() can find it (F03-2)
    await self._transport.subscribe("_agency.worker.restart", self._on_restart_command)
    self._registry.register("_agency.worker.restart")

    # D5: process-level health responder — inline at the transport layer
    # (the H9 sink pattern; precedent: _agency.worker.restart above).
    await self._transport.subscribe(self._health_channel, self._on_health_probe)
    self._registry.register(self._health_channel)

    # Start agent message loops
    for agent in self._agents.values():
        await agent._start()

    # Announce agents (and the restart handler) to the runtime's registry for
    # cross-process routing. Published AFTER wait_ready so the PUB socket
    # connection is stable. The brief sleep gives the runtime's receiver loop
    # time to process the announcements before worker.start() returns.
    announce_names = list(self._agents) + ["_agency.worker.restart", self._health_channel]
    for name in announce_names:
        maybe_agent: AgentProcess | None = self._agents.get(name)
        payload: dict[str, Any] = {"name": name}
        if maybe_agent is not None:
            payload["capabilities"] = self._agent_capabilities(maybe_agent) or []
            payload["capability_metadata"] = dict(maybe_agent.capability_metadata)
            # D5: peers' supervisors group their remote children by this
            # channel and probe the PROCESS once, not each agent's mailbox.
            payload["health_channel"] = self._health_channel
        await self._transport.publish(
            "_agency.register",
            self._serializer.serialize(Message(type="_agency.register", payload=payload)),
        )
    await asyncio.sleep(0.1)

    self._started = True

stop() async

Stop all agents and disconnect from the proxy.

Source code in civitas/worker.py

async def stop(self) -> None:
    """Stop all agents and disconnect from the proxy."""
    if not self._started:
        return

    for agent in reversed(list(self._agents.values())):
        await agent._stop()

    # Deregister agents from the runtime's registry before disconnecting
    if self._serializer is not None and self._transport is not None:
        for name in self._agents:
            try:
                await self._transport.publish(
                    "_agency.deregister",
                    self._serializer.serialize(
                        Message(type="_agency.deregister", payload={"name": name})
                    ),
                )
            except Exception:  # noqa: BLE001 — best-effort during shutdown
                pass

    if self._transport is not None:
        await self._transport.stop()

    if self._otel_agent_task is not None:
        self._otel_agent_task.cancel()
        await asyncio.gather(self._otel_agent_task, return_exceptions=True)
        self._otel_agent_task = None

    self._started = False
    self._stop_event.set()