MessageBus

Central message router. Routes messages from sender to recipient by name, delegates physical delivery to the Transport, and generates tracing spans.

See Architecture for the routing resolution order.

---

civitas.bus.MessageBus(transport, registry, serializer, tracer, audit_sink=None)

Central message router.

Routes messages from sender to recipient by name, delegates physical delivery to the Transport, applies serialization via the Serializer, and generates tracing spans for every send/receive.

Routing precedence in route(): 1. Registry lookup by recipient name → use RoutingEntry.address 2. Transport ephemeral reply address → publish directly (for request-reply) 3. Neither → raise MessageRoutingError

Source code in civitas/bus.py

def __init__(
    self,
    transport: Transport,
    registry: Registry,
    serializer: Serializer,
    tracer: Tracer,
    audit_sink: AuditSink | None = None,
) -> None:
    self._transport = transport
    self._registry = registry
    self._serializer = serializer
    self._tracer = tracer
    self._audit_sink = audit_sink
    self._local_agents: dict[str, AgentProcess] = {}

setup_agent(agent) async

Subscribe the transport to deliver messages to an agent's mailbox.

Source code in civitas/bus.py

async def setup_agent(self, agent: AgentProcess) -> None:
    """Subscribe the transport to deliver messages to an agent's mailbox."""

    async def _on_message_received(data: bytes) -> None:
        message = self._serializer.deserialize(data)
        span = self._tracer.start_receive_span(message)
        try:
            await agent.receive(message)
        finally:
            span.end()

    self._local_agents[agent.name] = agent
    await self._transport.subscribe(agent.name, _on_message_received)

route(message) async

Route a message to its recipient.

Validates system message types, creates a send span, serializes the message, and publishes through the transport.

Routing order: 1. Registry lookup → use RoutingEntry.address 2. Transport ephemeral reply address (has_reply_address) → publish directly 3. Neither → raise MessageRoutingError

Source code in civitas/bus.py

async def route(self, message: Message) -> None:
    """Route a message to its recipient.

    Validates system message types, creates a send span, serializes the
    message, and publishes through the transport.

    Routing order:
    1. Registry lookup → use RoutingEntry.address
    2. Transport ephemeral reply address (has_reply_address) → publish directly
    3. Neither → raise MessageRoutingError
    """
    self._validate_message_type(message)

    entry = self._registry.lookup(message.recipient)
    if entry is not None:
        address = entry.address
    elif self._transport.has_reply_address(message.recipient):
        # Ephemeral reply endpoint — same-process request-reply short-circuit
        address = message.recipient
    elif message.recipient.startswith("_reply."):
        # Cross-process reply: the runtime's transport owns this ephemeral topic.
        # Route by address directly — ZMQ/NATS delivery handles it.
        address = message.recipient
    else:
        raise MessageRoutingError(f"No agent registered with name: {message.recipient!r}")

    span = self._tracer.start_send_span(message)
    # v0.9.3 (A1): when OTEL is active, start_send_span() may have
    # replaced span.trace_id/span_id with OTEL's own REAL, authoritative
    # IDs (OTEL mints its own; civitas's original ones aren't otherwise
    # honored -- see Tracer._make_span()'s docstring comment). Sync that
    # back onto the outgoing Message before it hits the wire, so the
    # receiving side's handle_span/recv_span parent to a span OTEL
    # actually emitted, not a dangling made-up ID.
    message.trace_id = span.trace_id
    message.span_id = span.span_id
    try:
        data = self._serializer.serialize(message)
        await self._transport.publish(address, data)
    finally:
        span.end()

    if self._audit_sink is not None:
        await self._audit_sink.emit(
            AuditEvent(
                event="message.route",
                ts=datetime.now(UTC).isoformat(),
                agent=message.sender,
                signer_id=message.sender,  # verified sender == signer when signing is active
                details={
                    "sender": message.sender,
                    "recipient": message.recipient,
                    "type": message.type,
                    "correlation_id": message.correlation_id or "",
                    "message_id": message.id,
                },
            )
        )

request(message, timeout=30.0, *, fail_if_suspended=False) async

Send a request message and await a reply.

Used by ask() — delegates to transport.request() which handles correlation and reply routing.

timeout (v0.10.0): None — or any value <= 0 (canonically -1) — means wait indefinitely (HITL approvals can take hours/days; the agent stays SUSPENDED until resumed). Normalized to None here so every transport's asyncio.timeout(None) waits forever uniformly. A positive value is a bounded wait, as before.

fail_if_suspended (v0.10.0, D2): when True, raise AgentSuspendedError immediately if the recipient is SUSPENDED, instead of buffering the request until resume. Opt-in — the default path never consults suspension state (no cost, no behavior change).

Source code in civitas/bus.py

async def request(
    self,
    message: Message,
    timeout: float | None = 30.0,
    *,
    fail_if_suspended: bool = False,
) -> Message:
    """Send a request message and await a reply.

    Used by ask() — delegates to transport.request() which handles
    correlation and reply routing.

    ``timeout`` (v0.10.0): ``None`` — or any value ``<= 0`` (canonically
    ``-1``) — means **wait indefinitely** (HITL approvals can take
    hours/days; the agent stays SUSPENDED until resumed). Normalized to
    ``None`` here so every transport's ``asyncio.timeout(None)`` waits
    forever uniformly. A positive value is a bounded wait, as before.

    ``fail_if_suspended`` (v0.10.0, D2): when True, raise
    ``AgentSuspendedError`` immediately if the recipient is SUSPENDED,
    instead of buffering the request until resume. Opt-in — the default path
    never consults suspension state (no cost, no behavior change).
    """
    self._validate_message_type(message)
    if timeout is not None and timeout <= 0:
        timeout = None
    if fail_if_suspended and self._registry.is_suspended(message.recipient):
        raise AgentSuspendedError(
            f"agent {message.recipient!r} is suspended; not waiting (fail_if_suspended=True)"
        )

    entry = self._registry.lookup(message.recipient)
    if entry is None:
        raise MessageRoutingError(f"No agent registered with name: {message.recipient!r}")

    span = self._tracer.start_send_span(message)
    # v0.9.3 (A1): see the identical comment in route() above.
    message.trace_id = span.trace_id
    message.span_id = span.span_id
    try:
        data = self._serializer.serialize(message)
        reply_data = await self._transport.request(entry.address, data, timeout)
        return self._serializer.deserialize(reply_data)
    finally:
        span.end()

lookup_all(pattern)

Return all registered agents matching a glob pattern.

Source code in civitas/bus.py

def lookup_all(self, pattern: str) -> list[RoutingEntry]:
    """Return all registered agents matching a glob pattern."""
    return self._registry.lookup_all(pattern)

---

civitas.registry.LocalRegistry()

Single-node in-memory registry.

Default implementation for single-process and same-node deployments. All reads are O(1) dict lookups — no I/O, no async.

Remote agents can be registered via register_remote() so that pattern-based broadcast works across process boundaries; they are represented as RoutingEntry(is_local=False) and carry no object reference.

Capability-based lookups (find_by_capability, find_by_capabilities) work across both local and remote entries — capability tags are included in cross-process Worker announcements so every node has a complete view.

Listeners registered via add_listener() are notified after every register/deregister. Intended for external governance systems (e.g. Presidium) that need a live view of the agent population.

Source code in civitas/registry.py

def __init__(self) -> None:
    self._entries: dict[str, RoutingEntry] = {}
    self._listeners: list[RegistryListener] = []
    # Persists across deregister so a reordered late register cannot
    # resurrect a dead name (R6 · D13).
    self._remote_epochs: dict[str, int] = {}
    # v0.10.0 (hitl-polish.md D2): names currently SUSPENDED, for opt-in
    # fail-fast ask(). A set, not a RoutingEntry field (that's frozen).
    self._suspended: set[str] = set()

register(name, address=None, *, is_local=True, capabilities=None, capability_metadata=None)

Register an agent.

address defaults to name when not given, which is correct for in-process and NATS transports. Pass an explicit address for ZMQ TCP endpoints.

Raises ValueError if the name is already registered.

Source code in civitas/registry.py

def register(
    self,
    name: str,
    address: str | None = None,
    *,
    is_local: bool = True,
    capabilities: list[str] | tuple[str, ...] | None = None,
    capability_metadata: dict[str, Any] | None = None,
) -> None:
    """Register an agent.

    ``address`` defaults to ``name`` when not given, which is correct
    for in-process and NATS transports.  Pass an explicit address for
    ZMQ TCP endpoints.

    Raises ``ValueError`` if the name is already registered.
    """
    if name in self._entries:
        raise ValueError(f"Process already registered: {name!r}")
    entry = RoutingEntry(
        name=name,
        address=address if address is not None else name,
        is_local=is_local,
        capabilities=tuple(capabilities) if capabilities else (),
        capability_metadata=dict(capability_metadata) if capability_metadata else {},
    )
    self._entries[name] = entry
    self._fire_listeners(entry, "register")

register_remote(name, capabilities=None, capability_metadata=None, *, owner='', pubkey='', epoch=0, health_channel='')

Register a remote agent for cross-process pattern matching.

Idempotent for repeated announcements of the same remote agent. The cross-process arbiter for global name uniqueness (R6 · D9): a name owned locally, or by a different remote owner, or a name→pubkey conflict is rejected rather than taken over (no last-writer-wins). A register whose epoch is older than the last seen for the name is dropped so a reordered announcement cannot resurrect a dead name (D13).

Raises:

Type	Description
ValueError	if the name is already local, owned by a different remote owner, or announces a conflicting public key.

Source code in civitas/registry.py

def register_remote(
    self,
    name: str,
    capabilities: list[str] | tuple[str, ...] | None = None,
    capability_metadata: dict[str, Any] | None = None,
    *,
    owner: str = "",
    pubkey: str = "",
    epoch: int = 0,
    health_channel: str = "",
) -> None:
    """Register a remote agent for cross-process pattern matching.

    Idempotent for repeated announcements of the same remote agent. The
    cross-process arbiter for global name uniqueness (R6 · D9): a name owned
    locally, or by a *different* remote ``owner``, or a name→``pubkey``
    conflict is rejected rather than taken over (no last-writer-wins). A
    register whose ``epoch`` is older than the last seen for the name is
    dropped so a reordered announcement cannot resurrect a dead name (D13).

    Raises:
        ValueError: if the name is already local, owned by a different
            remote owner, or announces a conflicting public key.
    """
    last_epoch = self._remote_epochs.get(name)
    if last_epoch is not None and epoch < last_epoch:
        logger.warning(
            "Dropping stale remote register for %r (epoch %d < %d)", name, epoch, last_epoch
        )
        return

    existing = self._entries.get(name)
    if existing is not None:
        if existing.is_local:
            raise ValueError(f"Cannot register {name!r} as remote: already registered as local")
        if existing.owner != owner:
            raise ValueError(
                f"Cannot register {name!r} as remote: already owned by {existing.owner!r}"
            )
        if existing.pubkey and pubkey and existing.pubkey != pubkey and epoch <= existing.epoch:
            raise ValueError(f"Cannot register {name!r} as remote: public key conflict")
        if epoch <= existing.epoch:
            return  # idempotent / superseded re-announcement
    elif last_epoch is not None and epoch <= last_epoch:
        logger.warning(
            "Ignoring resurrection of %r at epoch %d (last seen %d)", name, epoch, last_epoch
        )
        return

    entry = RoutingEntry(
        name=name,
        address=name,
        is_local=False,
        capabilities=tuple(capabilities) if capabilities else (),
        capability_metadata=dict(capability_metadata) if capability_metadata else {},
        owner=owner,
        pubkey=pubkey,
        epoch=epoch,
        health_channel=health_channel,
    )
    self._entries[name] = entry
    self._remote_epochs[name] = epoch if last_epoch is None else max(epoch, last_epoch)
    self._fire_listeners(entry, "register")

deregister(name)

Remove an agent. No-op if not registered.

Source code in civitas/registry.py

def deregister(self, name: str) -> None:
    """Remove an agent. No-op if not registered."""
    entry = self._entries.pop(name, None)
    self._suspended.discard(name)  # v0.10.0: a removed name is not suspended
    if entry is not None:
        self._fire_listeners(entry, "deregister")

lookup(name)

Return the RoutingEntry for name, or None if not registered.

Source code in civitas/registry.py

def lookup(self, name: str) -> RoutingEntry | None:
    """Return the RoutingEntry for ``name``, or None if not registered."""
    return self._entries.get(name)

lookup_all(pattern)

Return all entries whose name matches a glob pattern.

Underscore-prefixed (system) names — _runtime, _agency.* — match only when the pattern itself starts with _ (C6): broadcast("*") must not deliver business payloads to internal endpoints, but explicit intent ("_agency.*") still works.

Source code in civitas/registry.py

def lookup_all(self, pattern: str) -> list[RoutingEntry]:
    """Return all entries whose name matches a glob pattern.

    Underscore-prefixed (system) names — ``_runtime``, ``_agency.*`` — match
    only when the pattern itself starts with ``_`` (C6): ``broadcast("*")``
    must not deliver business payloads to internal endpoints, but explicit
    intent (``"_agency.*"``) still works.
    """
    if pattern.startswith("_"):
        return [
            entry for name, entry in self._entries.items() if fnmatch.fnmatch(name, pattern)
        ]
    return [
        entry
        for name, entry in self._entries.items()
        if not name.startswith("_") and fnmatch.fnmatch(name, pattern)
    ]

---

civitas.registry.RoutingEntry(name, address, is_local, capabilities=(), capability_metadata=dict(), owner='', pubkey='', epoch=0, health_channel='') dataclass

Routing metadata for a registered agent.

address is the transport-level identifier used by the bus when calling transport.publish(). For in-process and NATS deployments this equals the agent name. For ZMQ point-to-point it is the endpoint string (e.g. tcp://host:5555).

is_local is True when the agent runs inside this process, False for agents registered via cross-process discovery.

capabilities is a tuple of capability tag strings declared by the agent (e.g. ("text.summarize", "text.translate")).

capability_metadata is a free-form dict passed through verbatim to registry listeners (e.g. Presidium). The runtime never interprets it.

owner is the authenticated identity (the announcing Worker-hosted supervisor's name) that announced a cross-process entry, used to reject name takeover by a different remote owner (R6 · D9). Empty for local entries and legacy name-only announcements.

pubkey is the base64 Ed25519 verify key announced for a remotely-spawned child (R6 · D3/D11); empty when signing is disabled. epoch is the monotonic incarnation counter carried by the announcement, used to reject stale/reordered register/deregister messages (R6 · D13).

health_channel (D5, v0.9.0) is the hosting Worker's process-level health topic (_agency.worker.<id>.health) carried by the announcement; empty for local entries and pre-v0.9 workers (supervisors fall back to per-agent heartbeats — the Q2 skew tolerance).