Runtime

Assembles and manages the full Civitas runtime. Wires transport, registry, serializer, tracer, plugins, and the supervision tree.

See Deployment and Topology & CLI for usage.

---

civitas.runtime.Runtime(supervisor=None, transport='in_process', serializer=None, model_provider=None, tool_registry=None, state_store=None, metrics=None, exporters=None, zmq_pub_addr='tcp://127.0.0.1:5559', zmq_sub_addr='tcp://127.0.0.1:5560', zmq_start_proxy=True, nats_servers='nats://localhost:4222', nats_jetstream=False, nats_stream_name='AGENCY', components=None)

Assembles and manages the full Civitas runtime.

Startup sequence (from Implementation Guide §3): 1. Read configuration 2. Create Serializer 3. Create Tracer 4. Create Transport 5. Create Registry 6. Create MessageBus 7. Create plugin instances 8. Instantiate / wire all AgentProcesses 9. Register all AgentProcesses in Registry 10. Start Transport 11. Walk supervision tree bottom-up, start each agent 12. Start all Supervisors 13. Runtime is ready

Source code in civitas/runtime.py

def __init__(
    self,
    supervisor: Supervisor | None = None,
    transport: str = "in_process",
    serializer: Serializer | None = None,
    model_provider: Any = None,
    tool_registry: Any = None,
    state_store: Any = None,
    metrics: Any = None,
    exporters: list[Any] | None = None,
    zmq_pub_addr: str = "tcp://127.0.0.1:5559",
    zmq_sub_addr: str = "tcp://127.0.0.1:5560",
    zmq_start_proxy: bool = True,
    nats_servers: str | list[str] = "nats://localhost:4222",
    nats_jetstream: bool = False,
    nats_stream_name: str = "AGENCY",
    components: ComponentSet | None = None,
) -> None:
    self._root_supervisor = supervisor
    self._transport_type = transport
    self._custom_serializer = serializer
    self._model_provider = model_provider
    self._tool_registry = tool_registry
    self._state_store = state_store
    self._metrics = metrics
    self._exporters = exporters or []
    self._components = components
    self._otel_agent_task: asyncio.Task[None] | None = None

    # ZMQ-specific config
    self._zmq_pub_addr = zmq_pub_addr
    self._zmq_sub_addr = zmq_sub_addr
    self._zmq_start_proxy = zmq_start_proxy

    # NATS-specific config
    self._nats_servers = nats_servers
    self._nats_jetstream = nats_jetstream
    self._nats_stream_name = nats_stream_name

    # MCP server configs parsed from topology YAML
    self._mcp_configs: list[Any] = []

    # Security config — populated by from_config() when a security: block is present
    self._security_config: Any = None
    self._topology_public_keys: dict[str, str] = {}

    # Signing infrastructure — built in start() when signing is enabled on a
    # distributed transport; used to verify cross-process announcements (R6 · D8).
    self._key_registry: Any = None
    self._message_signer: Any = None
    self._signing_on: bool = False

    # Per-agent credentials — populated by from_config() from credentials: blocks
    self._agent_credentials: dict[str, dict[str, str]] = {}

    # Per-agent capabilities — populated by from_config() from capabilities: blocks
    self._agent_capabilities: dict[str, tuple[list[str], dict[str, Any]]] = {}

    # Audit sink — populated by from_config() when an audit: block is present
    self._audit_sink: Any = None

    # Transport security (ZMQ CURVE / NATS TLS) — populated by from_config()
    self._transport_security: Any = None

    # Set during start() — exposed for stop(), ask()/send(), and get_agent()
    self._serializer: Serializer | None = None
    self._tracer: Any = None
    self._transport: Any = None
    self._registry: Any = None
    self._bus: Any = None
    self._agents_by_name: dict[str, AgentProcess] = {}  # F04-10: O(1) live process lookup
    self._started = False

from_config(path, agent_classes=None, *, process_filter='*') classmethod

Build a Runtime from a YAML topology file.

The agent_classes dict maps type strings (e.g. "MyAgent") to the actual Python class. If not provided, types are resolved via importlib from dotted module paths (e.g. "myapp.agents.MyAgent").

process_filter (v0.9.2.1 bugfix): a topology node (an agent, dynamic_supervisor, or flat dotted-path node) MAY carry a process: <name> tag meaning "this belongs to a different OS process, started separately." Before this fix, from_config had no awareness of that tag at all and built EVERY node into the local tree regardless — silently duplicating whatever a real Worker process elsewhere also builds for itself (confirmed: deployment/level2_multi_process/run_supervisor.py, run alone, registered worker_a/worker_b locally even though they're process: worker-tagged).

• "*" (default): no filtering — build every node regardless of any process: tag. Unchanged from every release before this one; existing callers see no behavior change.
• None: build only UNTAGGED nodes (no process: key at all) — the coordinator/supervisor role in a multi-process topology.
• Any other string: build only nodes tagged process: <that string> — matches what Worker's own construction path already does via _find_process_agents, exposed here for symmetry/completeness.

Source code in civitas/runtime.py

@classmethod
def from_config(
    cls,
    path: str | Path,
    agent_classes: dict[str, type[AgentProcess]] | None = None,
    *,
    process_filter: str | None = "*",
) -> Runtime:
    """Build a Runtime from a YAML topology file.

    The ``agent_classes`` dict maps type strings (e.g. "MyAgent") to the
    actual Python class. If not provided, types are resolved via
    ``importlib`` from dotted module paths (e.g. "myapp.agents.MyAgent").

    ``process_filter`` (v0.9.2.1 bugfix): a topology node (an ``agent``,
    ``dynamic_supervisor``, or flat dotted-path node) MAY carry a
    ``process: <name>`` tag meaning "this belongs to a different OS
    process, started separately." Before this fix, ``from_config`` had
    no awareness of that tag at all and built EVERY node into the local
    tree regardless — silently duplicating whatever a real Worker
    process elsewhere also builds for itself (confirmed:
    ``deployment/level2_multi_process/run_supervisor.py``, run alone,
    registered ``worker_a``/``worker_b`` locally even though they're
    ``process: worker``-tagged).

    - ``"*"`` (default): no filtering — build every node regardless of
      any ``process:`` tag. Unchanged from every release before this
      one; existing callers see no behavior change.
    - ``None``: build only UNTAGGED nodes (no ``process:`` key at all)
      — the coordinator/supervisor role in a multi-process topology.
    - Any other string: build only nodes tagged ``process: <that string>``
      — matches what ``Worker``'s own construction path already does via
      ``_find_process_agents``, exposed here for symmetry/completeness.
    """
    config = yaml.safe_load(Path(path).read_text())
    config = substitute_vars(config)
    return cls.from_config_dict(
        config, agent_classes=agent_classes, process_filter=process_filter
    )

start() async

Start the runtime following the canonical initialization sequence.

Source code in civitas/runtime.py

async def start(self) -> None:
    """Start the runtime following the canonical initialization sequence."""
    if self._started:
        return

    # v0.9.1 (dashboard-v2, D-DASH-4): auto-provide a MetricsCollector when a
    # TopologyServer is present and the caller didn't already attach their own
    # sink via set_metrics()/metrics= — the dashboard needs SOMETHING to read
    # via /metrics. Must happen before build_component_set() below, which
    # captures self._metrics by value; Runtime(metrics=my_sink) callers are
    # unaffected (self._metrics is already set, this block is a no-op for them).
    if (
        self._metrics is None
        and self._components is None
        and self._root_supervisor is not None
        and any(
            isinstance(a, _TopologyIntrospection) for a in self._root_supervisor.all_agents()
        )
    ):
        self._metrics = MetricsCollector()
        self._metrics.runtime_started()
        # register_agent() is required before message_handled()/message_sent()
        # will record anything for a name (MetricsCollector no-ops for an
        # unregistered agent) — matches what the old CLI dashboard.py did
        # manually. Dynamically-spawned children are NOT covered by this loop
        # (all_agents() only sees statically-declared children) — documented
        # gap, not a spawn-time hook (design dashboard-v2.md addendum).
        for agent in self._root_supervisor.all_agents():
            self._metrics.register_agent(agent.name)
        # v0.9.1 (D-DASH addendum, 2026-07-26): agent_restarted() existed on
        # MetricsSink/MetricsCollector but was NEVER called from anywhere in
        # civitas/ — the exact same class of gap FD-01 was for llm_call()
        # (Phase C). The old CLI dashboard.py was the only caller, wired
        # manually via on_crash(); reproduced here so restart_history and
        # per-agent restart counts populate for ANY TopologyServer-having
        # Runtime, not just the (now-removed, Phase F) standalone CLI path.
        metrics_for_crash_callback = self._metrics

        async def _record_restart_for_dashboard(name: str, exc: Exception) -> None:
            metrics_for_crash_callback.agent_restarted(name, type(exc).__name__)

        self.on_crash(_record_restart_for_dashboard)

    # Steps 2–6: build or use provided ComponentSet.
    # Note: if a pre-built ComponentSet is provided, its transport must support
    # being started by this call — transport.start() is always called below. (F04-11)
    if self._components is not None:
        cs = self._components
    else:
        ts = self._transport_security
        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,
            audit_sink=self._audit_sink,
            metrics=self._metrics,
            exporters=self._exporters,
            zmq_pub_addr=self._zmq_pub_addr,
            zmq_sub_addr=self._zmq_sub_addr,
            zmq_start_proxy=self._zmq_start_proxy,
            zmq_curve_config=ts.zmq if ts is not None and ts.zmq.enabled else None,
            nats_servers=self._nats_servers,
            nats_jetstream=self._nats_jetstream,
            nats_stream_name=self._nats_stream_name,
            nats_tls_config=ts.nats if ts is not None and ts.nats.enabled else None,
        )

    # Expose on self for stop(), ask(), send(), and get_agent()
    self._serializer = cs.serializer
    self._tracer = cs.tracer
    self._transport = cs.transport
    self._registry = cs.registry
    self._bus = cs.bus
    self._state_store = cs.store

    # 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)
        )

    if self._root_supervisor is None:
        self._started = True
        return

    # 8. Inject dependencies into all AgentProcesses
    all_agents = self._root_supervisor.all_agents()

    # Security: build signing infrastructure if configured for non-InProcess transports.
    # InProcess transport skips signing entirely (D9 — same OS process, no wire to protect).
    if (
        self._security_config is not None
        and self._security_config.signing.enabled
        and self._transport_type != "in_process"
    ):
        key_dir = self._security_config.identity.key_dir
        identities: dict[str, AgentIdentity] = {}
        for agent in all_agents:
            # A DynamicSupervisor needs an identity to sign the cluster-wide
            # child announcements it publishes (R6 · D8); the topology
            # introspection unit (read-only, never signs) is exempt. v0.9.5:
            # that unit is now a TopologyAgent PLUS its internally-owned
            # HTTPGateway -- exempt the gateway too (identified by its
            # topology_agent config, so a normal user http_gateway is
            # unaffected), so a signed non-auto deployment isn't forced to
            # provision a new key for the auto-created gateway.
            if isinstance(agent, _TopologyIntrospection):
                continue
            if isinstance(agent, HTTPGateway) and agent._gw_config.topology_agent is not None:
                continue
            if self._security_config.identity.mode == "auto":
                identities[agent.name] = AgentIdentity.load_or_generate(agent.name, key_dir)
            else:
                identities[agent.name] = AgentIdentity.load(agent.name, key_dir)

        registry = KeyRegistry()
        for name, identity in identities.items():
            registry.register(name, identity.verify_key)
        for name, pub_b64 in self._topology_public_keys.items():
            if name not in registry:
                registry.register_b64(name, pub_b64)

        signer = MessageSigner(identities, registry, self._security_config.signing)
        signing_ser = SigningSerializer(signer, self._security_config.signing)
        self._serializer = signing_ser
        cs.bus._serializer = signing_ser
        # v0.9.2.1 bugfix: the transport holds its OWN private serializer
        # reference (needed for request()'s internal reply_to round-trip)
        # separate from the bus's — without this, ask() over a signing-
        # enabled ZMQ/NATS transport silently corrupted every request into
        # a blank message (empty sender/correlation_id), which made the
        # reply-routing check in AgentProcess._dispatch() no-op with no
        # exception anywhere — just a plain ask() TimeoutError. See
        # Transport.set_serializer's docstring for the full root cause.
        cs.bus._transport.set_serializer(signing_ser)
        self._key_registry = registry
        self._message_signer = signer
        self._signing_on = True

    for agent in all_agents:
        cs.inject(agent)
        # Wire per-agent credentials from topology credentials: blocks
        if self._agent_credentials:
            agent._credentials = self._agent_credentials.get(agent.name, {})

    # Inject into supervisors (supervisor-specific wiring, not via ComponentSet)
    # D1a (v0.9.0): also hand every supervisor a RE-INVOKABLE wiring callback
    # (fresh incarnations must be wired exactly like startup wiring) and a
    # replaced-callback that keeps Runtime's O(1) map + TopologyServer
    # references fresh. User-held object references go stale by design (Q1:
    # route by name, never by object).
    def _wire_child(agent: AgentProcess) -> None:
        cs.inject(agent)
        if self._agent_credentials:
            agent._credentials = self._agent_credentials.get(agent.name, {})

    def _on_child_replaced(name: str, new_agent: AgentProcess) -> None:
        self._agents_by_name[name] = new_agent
        if isinstance(new_agent, _TopologyIntrospection):
            new_agent._root_supervisor = self._root_supervisor
            new_agent._agents = self._agents_by_name
            new_agent._metrics_collector = (
                self._metrics if isinstance(self._metrics, MetricsCollector) else None
            )

    for sup in self._root_supervisor.all_supervisors():
        sup._bus = cs.bus
        sup._registry = cs.registry
        sup._tracer = cs.tracer
        sup._wire_child = _wire_child
        sup.add_child_replaced_callback(_on_child_replaced)

    # Wire _dynamic_supervisor_name for all agents based on the static topology.
    # Each agent receives the name of the nearest DynamicSupervisor in its
    # ancestor-or-sibling subtree, enabling self.spawn() without explicit naming.
    def _wire_dyn_sup(
        node: Supervisor | AgentProcess,
        nearest_dyn: str | None,
    ) -> None:
        if isinstance(node, DynamicSupervisor):
            node._dynamic_supervisor_name = node.name  # spawns into itself
        elif isinstance(node, Supervisor):
            dyn_child = next(
                (c for c in node.children if isinstance(c, DynamicSupervisor)), None
            )
            new_nearest = dyn_child.name if dyn_child is not None else nearest_dyn
            for child in node.children:
                _wire_dyn_sup(child, new_nearest)
        else:
            node._dynamic_supervisor_name = nearest_dyn

    _wire_dyn_sup(self._root_supervisor, None)

    # 9. Register all AgentProcesses in Registry; build O(1) name→process map (F04-10)
    for agent in all_agents:
        yaml_caps = self._agent_capabilities.get(agent.name)
        if yaml_caps is not None:
            caps, meta = yaml_caps
        else:
            caps = list(agent.capabilities)
            meta = dict(agent.capability_metadata)
        if isinstance(agent, DynamicSupervisor) and DYNAMIC_SUPERVISOR_CAPABILITY not in caps:
            caps = [*caps, DYNAMIC_SUPERVISOR_CAPABILITY]
        self._registry.register(agent.name, capabilities=caps, capability_metadata=meta)
    self._agents_by_name = {a.name: a for a in all_agents}

    # Inject topology introspection references before supervision tree starts
    for agent in all_agents:
        if isinstance(agent, _TopologyIntrospection):
            agent._root_supervisor = self._root_supervisor
            agent._agents = self._agents_by_name
            # v0.9.1 (D-DASH-2/D-DASH-4): only a MetricsCollector has the
            # .snapshot the /metrics endpoint reads — a custom MetricsSink
            # (message_handled/message_sent/... only) leaves this None,
            # which /metrics reports explicitly rather than guessing.
            agent._metrics_collector = (
                self._metrics if isinstance(self._metrics, MetricsCollector) else None
            )

    # 10. Start Transport
    await self._transport.start()

    # Set up transport subscriptions for each agent
    for agent in all_agents:
        await self._bus.setup_agent(agent)

    # Subscribe to cross-process agent announcements from Worker processes.
    # Workers publish _agency.register on startup so this runtime's bus can
    # route messages to remote agents without a shared registry service.
    await self._transport.subscribe("_agency.register", self._on_remote_register)
    await self._transport.subscribe("_agency.deregister", self._on_remote_deregister)

    # H9 (#33): '_runtime' sink. Runtime-initiated messages carry
    # sender="_runtime", which is not an agent — an agent doing the natural
    # `self.send(message.sender, ...)` used to crash on MessageRoutingError.
    # The sink converts that into a logged drop (and a fail-fast error reply
    # for ask()). Not an AgentProcess: bare subscription, no mailbox, no
    # lifecycle, nothing to supervise. Glob broadcasts never reach it —
    # underscore names are excluded from patterns (registry C6 rule).
    self._registry.register("_runtime")
    await self._transport.subscribe("_runtime", self._on_runtime_addressed)

    # Wait for subscriptions to propagate (ZMQ slow joiner mitigation)
    if hasattr(self._transport, "wait_ready"):
        await self._transport.wait_ready()

    # Connect MCP servers declared in topology YAML to all agents
    if self._mcp_configs:
        for agent in all_agents:
            for mcp_cfg in self._mcp_configs:
                try:
                    await agent.connect_mcp(mcp_cfg)
                except Exception as exc:
                    logger.warning(
                        "Failed to connect agent '%s' to MCP server '%s': %s",
                        agent.name,
                        mcp_cfg.name,
                        exc,
                    )

    # D6 (v0.9.0 E4 Phase A): supervisors are now addressable actors too
    # (design supervision-endgame.md §6) — register + wire their transport
    # subscription before the tree starts, exactly like agents. Own-loop-
    # first ordering (D-E4-3) is enforced inside Supervisor.start() itself.
    all_supervisors = self._root_supervisor.all_supervisors()
    for sup in all_supervisors:
        self._registry.register(sup.name, capabilities=[SUPERVISOR_CAPABILITY])
        await self._bus.setup_agent(sup)

    # 11-12. Start supervision tree (supervisors start their children)
    await self._root_supervisor.start()

    # 13. Runtime is ready
    self._started = True

stop() async

Shutdown sequence: stop agents, transport, flush tracer.

Source code in civitas/runtime.py

async def stop(self) -> None:
    """Shutdown sequence: stop agents, transport, flush tracer."""
    if not self._started:
        return

    # 1. Stop supervision tree (sends shutdown, awaits on_stop for each agent)
    if self._root_supervisor is not None:
        await self._root_supervisor.stop()

    # 2. Stop Transport
    if self._transport is not None:
        await self._transport.stop()

    # 3. Close StateStore
    if self._state_store is not None and hasattr(self._state_store, "close"):
        await self._state_store.close()

    # 4. Flush Tracer
    if self._tracer is not None:
        self._tracer.flush()

    # 5. Stop OTELAgent — cancel triggers its own drain-remaining-spans logic
    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

    # 6. Close Audit sink
    if self._audit_sink is not None:
        await self._audit_sink.close()

    if self._registry is not None:
        self._registry.deregister("_runtime")
    self._agents_by_name.clear()
    self._started = False

ask(agent_name, payload, timeout=30.0, message_type='message', *, fail_if_suspended=False) async

Send a message to an agent and await a reply.

timeout (v0.10.0): None/-1/any value <= 0 waits indefinitely — the HITL case (ask an agent that suspends for approval; the reply arrives hours/days later on resume). See AgentProcess.ask.

fail_if_suspended (v0.10.0, D2): raise AgentSuspendedError immediately instead of waiting, if the target is suspended.

Source code in civitas/runtime.py

async def ask(
    self,
    agent_name: str,
    payload: dict[str, Any],
    timeout: float | None = 30.0,
    message_type: str = "message",
    *,
    fail_if_suspended: bool = False,
) -> Message:
    """Send a message to an agent and await a reply.

    ``timeout`` (v0.10.0): ``None``/``-1``/any value ``<= 0`` waits
    indefinitely — the HITL case (ask an agent that suspends for approval;
    the reply arrives hours/days later on resume). See ``AgentProcess.ask``.

    ``fail_if_suspended`` (v0.10.0, D2): raise ``AgentSuspendedError``
    immediately instead of waiting, if the target is suspended.
    """
    if self._bus is None or self._tracer is None:
        raise RuntimeError("Runtime not started")

    trace_id = self._tracer.new_trace_id()
    message = Message(
        type=message_type,
        sender="_runtime",
        recipient=agent_name,
        payload=payload,
        correlation_id=_uuid7(),
        trace_id=trace_id,
        span_id=_new_span_id(),
    )
    return cast(
        Message,
        await self._bus.request(message, timeout=timeout, fail_if_suspended=fail_if_suspended),
    )

send(agent_name, payload, message_type='message') async

Fire-and-forget: send a message to an agent.

Source code in civitas/runtime.py

async def send(
    self,
    agent_name: str,
    payload: dict[str, Any],
    message_type: str = "message",
) -> None:
    """Fire-and-forget: send a message to an agent."""
    if self._bus is None or self._tracer is None:
        raise RuntimeError("Runtime not started")

    trace_id = self._tracer.new_trace_id()
    message = Message(
        type=message_type,
        sender="_runtime",
        recipient=agent_name,
        payload=payload,
        trace_id=trace_id,
        span_id=_new_span_id(),
    )
    await self._bus.route(message)

get_agent(name)

Return the live AgentProcess instance by name, or None.

O(1) lookup via the agents-by-name dict built during start(). Use this when you need to inspect process state (e.g. status). For routing messages use runtime.send/ask instead.

Source code in civitas/runtime.py

def get_agent(self, name: str) -> AgentProcess | None:
    """Return the live AgentProcess instance by name, or None.

    O(1) lookup via the agents-by-name dict built during start().
    Use this when you need to inspect process state (e.g. status).
    For routing messages use runtime.send/ask instead.
    """
    return self._agents_by_name.get(name)

all_agents()

Return all AgentProcess instances in the supervision tree.

Source code in civitas/runtime.py

def all_agents(self) -> list[AgentProcess]:
    """Return all AgentProcess instances in the supervision tree."""
    if self._root_supervisor is None:
        return []
    return self._root_supervisor.all_agents()

print_tree()

Return an ASCII representation of the supervision tree.

Source code in civitas/runtime.py

def print_tree(self) -> str:
    """Return an ASCII representation of the supervision tree."""
    if self._root_supervisor is None:
        return "(no supervision tree)"

    lines: list[str] = []

    def _walk(node: Supervisor | AgentProcess, prefix: str, is_last: bool) -> None:
        connector = "└── " if is_last else "├── "
        if isinstance(node, Supervisor):
            label = f"[sup] {node.name} ({node.strategy.value})"
        else:
            status = node.status.value if hasattr(node, "status") else "?"
            if isinstance(node, DynamicSupervisor):
                prefix_tag = "[dyn]"
            elif isinstance(node, _TopologyIntrospection):
                prefix_tag = "[topo]"
            elif isinstance(node, EvalAgent):
                prefix_tag = "[eval]"
            elif isinstance(node, HTTPGateway):
                prefix_tag = "[http]"
            elif isinstance(node, GenServer):
                prefix_tag = "[srv]"
            else:
                prefix_tag = "[agent]"
            label = f"{prefix_tag} {node.name} ({status})"
        lines.append(f"{prefix}{connector}{label}")

        if isinstance(node, Supervisor):
            child_prefix = prefix + ("    " if is_last else "│   ")
            for i, child in enumerate(node.children):
                _walk(child, child_prefix, i == len(node.children) - 1)

    # Root
    root = self._root_supervisor
    lines.append(f"[sup] {root.name} ({root.strategy.value})")
    for i, child in enumerate(root.children):
        _walk(child, "", i == len(root.children) - 1)

    return "\n".join(lines)

spawn(supervisor_name, agent_class, name, config=None, *, wait=True) async

Spawn a dynamic agent via the named DynamicSupervisor.

Returns the agent name on success. Raises SpawnError on failure.

With wait=False the call returns as soon as the child's task exists, before on_start() completes; a later start failure is delivered to the spawner via on_child_terminated (R1 · D2).

Source code in civitas/runtime.py

async def spawn(
    self,
    supervisor_name: str,
    agent_class: type[AgentProcess],
    name: str,
    config: dict[str, Any] | None = None,
    *,
    wait: bool = True,
) -> str:
    """Spawn a dynamic agent via the named DynamicSupervisor.

    Returns the agent name on success. Raises SpawnError on failure.

    With ``wait=False`` the call returns as soon as the child's task exists,
    before ``on_start()`` completes; a later start failure is delivered to the
    spawner via ``on_child_terminated`` (R1 · D2).
    """
    class_path = f"{agent_class.__module__}.{agent_class.__qualname__}"
    reply = await self.ask(
        supervisor_name,
        {
            "class_path": class_path,
            "name": name,
            "config": config or {},
            "spawner": "_runtime",
            "wait": wait,
            "spawn_id": _uuid7(),
        },
        message_type="civitas.dynamic.spawn",
    )
    if reply.payload.get("status") != "ok":
        reason = reply.payload.get("reason") or reply.payload.get("error") or "spawn failed"
        raise SpawnError(reason)
    return name

despawn(supervisor_name, name) async

Hard-stop a dynamic child via the named DynamicSupervisor.

Source code in civitas/runtime.py

async def despawn(self, supervisor_name: str, name: str) -> None:
    """Hard-stop a dynamic child via the named DynamicSupervisor."""
    reply = await self.ask(
        supervisor_name,
        {"name": name},
        message_type="civitas.dynamic.despawn",
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "despawn failed"))

stop_agent(supervisor_name, name, drain='current', timeout=30.0) async

Soft-stop a dynamic child via the named DynamicSupervisor.

Source code in civitas/runtime.py

async def stop_agent(
    self,
    supervisor_name: str,
    name: str,
    drain: str = "current",
    timeout: float = 30.0,
) -> None:
    """Soft-stop a dynamic child via the named DynamicSupervisor."""
    reply = await self.ask(
        supervisor_name,
        {"name": name, "drain": drain, "timeout": timeout},
        message_type="civitas.dynamic.stop",
        timeout=timeout + 5.0,
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "stop failed"))

---

civitas.components.ComponentSet(transport, registry, serializer, tracer, store=None, model_provider=None, tool_registry=None, audit_sink=None, metrics=None, span_queue=None, export_backend=None) dataclass

Assembled infrastructure wiring for a single Runtime or Worker.

MessageBus is derived automatically from the other four fields in post_init — callers should not construct it separately.

Attributes:

Name	Type	Description
transport	Any	Transport layer (InProcess, ZMQ, or NATS).
registry	Registry	LocalRegistry for agent name → address mapping.
serializer	Serializer	Serializer used by transport and bus.
tracer	Tracer	Tracer instance for span emission.
store	Any	StateStore for agent checkpoint/restore. None means no persistence; callers should default to InMemoryStateStore when appropriate.
model_provider	Any	Injected into agent.llm at startup.
tool_registry	Any	Injected into agent.tools at startup.
bus	MessageBus	MessageBus built from the other four fields.

inject(agent)

Inject bus and plugin references into an agent process.

Source code in civitas/components.py

def inject(self, agent: AgentProcess) -> None:
    """Inject bus and plugin references into an agent process."""
    agent._bus = self.bus
    agent._tracer = self.tracer
    agent._registry = self.registry
    agent.llm = self.model_provider
    agent.tools = self.tool_registry
    agent.store = self.store
    agent._audit_sink = self.audit_sink
    agent._metrics = self.metrics

---

civitas.components.build_component_set(transport_type='in_process', serializer=None, model_provider=None, tool_registry=None, state_store=None, audit_sink=None, metrics=None, exporters=None, zmq_pub_addr='tcp://127.0.0.1:5559', zmq_sub_addr='tcp://127.0.0.1:5560', zmq_start_proxy=True, zmq_curve_config=None, nats_servers='nats://localhost:4222', nats_jetstream=False, nats_stream_name='AGENCY', nats_tls_config=None)

Build a ComponentSet from primitive configuration values.

Called by Runtime.start() and Worker.start() when no pre-built ComponentSet is provided. Handles serializer selection, transport construction, and store defaulting.

Source code in civitas/components.py

def build_component_set(
    transport_type: str = "in_process",
    serializer: Serializer | None = None,
    model_provider: Any = None,
    tool_registry: Any = None,
    state_store: Any = None,
    audit_sink: AuditSink | None = None,
    metrics: MetricsSink | None = None,
    exporters: list[Any] | None = None,
    zmq_pub_addr: str = "tcp://127.0.0.1:5559",
    zmq_sub_addr: str = "tcp://127.0.0.1:5560",
    zmq_start_proxy: bool = True,
    zmq_curve_config: ZmqCurveConfig | None = None,
    nats_servers: str | list[str] = "nats://localhost:4222",
    nats_jetstream: bool = False,
    nats_stream_name: str = "AGENCY",
    nats_tls_config: NatsTlsConfig | None = None,
) -> ComponentSet:
    """Build a ComponentSet from primitive configuration values.

    Called by Runtime.start() and Worker.start() when no pre-built
    ComponentSet is provided. Handles serializer selection, transport
    construction, and store defaulting.
    """
    # Serializer
    if serializer is not None:
        built_serializer = serializer
    elif settings.serializer == "json":
        built_serializer = JsonSerializer()
    else:
        built_serializer = MsgpackSerializer()

    # Tracer — when exporters are configured (plugins.exporters in topology YAML),
    # spans flow through SpanQueue -> OTELAgent -> ExportBackend instead of the
    # Tracer's own direct OTLP path, so the two never run simultaneously (FD-09).
    # Runtime/Worker are responsible for starting/stopping the OTELAgent task.
    built_span_queue: Any = None
    built_export_backend: Any = None
    if exporters:
        from civitas.observability.export_backend import FanOutBackend
        from civitas.observability.span_queue import SpanQueue

        built_span_queue = SpanQueue()
        built_export_backend = FanOutBackend(exporters) if len(exporters) > 1 else exporters[0]
    built_tracer = Tracer(span_queue=built_span_queue)

    # Transport — imports are intentionally scoped here: ZMQ and NATS are optional
    # extras (pyzmq, nats-py) that may not be installed. Importing at module level
    # would cause ImportError on every civitas import for users without those extras.
    built_transport: Transport
    if transport_type == "zmq":
        from civitas.transport.zmq import ZMQTransport

        built_transport = ZMQTransport(
            built_serializer,
            pub_addr=zmq_pub_addr,
            sub_addr=zmq_sub_addr,
            start_proxy=zmq_start_proxy,
            curve_config=zmq_curve_config,
        )
    elif transport_type == "nats":
        from civitas.transport.nats import NATSTransport

        built_transport = NATSTransport(
            built_serializer,
            servers=nats_servers,
            jetstream=nats_jetstream,
            stream_name=nats_stream_name,
            tls_config=nats_tls_config,
        )
    else:
        from civitas.transport.inprocess import InProcessTransport

        built_transport = InProcessTransport(built_serializer)

    # Registry
    built_registry = LocalRegistry()

    # State store default
    if state_store is None:
        state_store = InMemoryStateStore()

    return ComponentSet(
        transport=built_transport,
        registry=built_registry,
        serializer=built_serializer,
        tracer=built_tracer,
        store=state_store,
        model_provider=model_provider,
        tool_registry=tool_registry,
        audit_sink=audit_sink,
        metrics=metrics,
        span_queue=built_span_queue,
        export_backend=built_export_backend,
    )