AgentProcess

Base class for all agents. Subclass it and implement handle().

See Core Concepts and Getting Started for usage examples.

---

civitas.process.AgentProcess(name, mailbox_size=1000, max_retries=3, shutdown_timeout=30.0, handle_timeout=None)

Base class for all agent processes in Civitas.

Developers subclass this and override lifecycle hooks: - on_start(): called once before the first message - handle(message): called for every incoming message - on_error(error, message): called when handle() raises - on_stop(): called on graceful shutdown (always — even on crash)

Messaging methods available inside hooks: - send(recipient, payload, message_type): fire-and-forget - ask(recipient, payload, message_type, timeout): request-reply - send_capable(capability, payload, message_type): route to any capable agent - broadcast(pattern, payload): send to all matching agents - reply(payload): return from handle() for request-reply

Observability helpers (call from inside handle()): - llm_span(model, attrs): context manager for LLM call spans - tool_span(tool_name, attrs): context manager for tool call spans

Capability declaration (class-level, inherited and overridable): capabilities: list[str] = ["text.summarize", "text.translate"] capability_metadata: dict[str, Any] = { "text.summarize": {"description": "...", "version": "1"} }

Source code in civitas/process.py

def __init__(
    self,
    name: str,
    mailbox_size: int = 1000,
    max_retries: int = 3,
    shutdown_timeout: float = 30.0,
    handle_timeout: float | None = None,
) -> None:
    self.name = name
    self.id: str = _uuid7()
    self.state: dict[str, Any] = {}
    self._status = ProcessStatus.INITIALIZING
    self._mailbox = Mailbox(maxsize=mailbox_size)
    self._task: asyncio.Task[None] | None = None
    # v0.9.1 (dashboard-v2, D-DASH-1): set fresh in _start_nowait() on every
    # incarnation, including restarts (D1a) — "uptime" means THIS
    # incarnation's age, matching the fresh-instance restart semantic.
    self._incarnation_started_at: float = time.monotonic()
    # v0.9.4: "session" tracking (design/dashboard-v2.md P1) — deliberately
    # incarnation-scoped, same reset semantic as _incarnation_started_at
    # above (a restart is a fresh instance, so a fresh session too — a
    # crash-and-recover genuinely interrupts whatever was happening). This
    # is NOT the same concept as an explicit, cross-restart session_id
    # with continuation relationships -- that's a real, tracked, separate
    # future idea (docs/milestones.md); this is the small, always-derived-
    # from-existing-data version scoped for v0.9.4.
    self._first_llm_call_at: float | None = None
    self._llm_call_count: int = 0
    self._max_retries = max_retries
    self._shutdown_timeout = shutdown_timeout
    # H6: opt-in per-message watchdog. None (default) = no timeout. When set,
    # a handle() exceeding it raises TimeoutError through the normal on_error
    # path (default ESCALATE → visible crash) — a hung *async* handler stops
    # being invisible to its supervisor. Limits: cancellation lands at the
    # current await point (use `async with` for resources), and blocking code
    # (time.sleep, busy loops) never yields, so it cannot be detected here.
    self._handle_timeout = handle_timeout

    # Injected by Runtime/Worker during setup
    self._bus: MessageBus | None = None
    self._tracer: Tracer | None = None
    self._registry: Registry | None = None
    self.llm: ModelProvider | None = None
    self.tools: ToolRegistry | None = None
    self.store: StateStore | None = None
    self.config: dict[str, Any] = {}

    # Per-agent credential map — populated from topology credentials: block.
    # Keys are provider names (e.g. "anthropic"); values are credential strings.
    self._credentials: dict[str, str] = {}

    # Audit sink — injected by ComponentSet, None when auditing is disabled.
    self._audit_sink: AuditSink | None = None

    # Metrics sink — injected by ComponentSet, None when no collector is attached.
    self._metrics: MetricsSink | None = None

    # Set by Runtime to the nearest DynamicSupervisor ancestor name (if any)
    self._dynamic_supervisor_name: str | None = None

    # MCP clients opened via connect_mcp() — keyed by server name
    self._mcp_clients: dict[str, Any] = {}

    # Current message context for reply/tracing
    self._current_message: Message | None = None
    self._current_handle_span: Span | None = None

    # Signalled when the message loop enters RUNNING
    self._running_event: asyncio.Event | None = None

    # _reached_loop gates restart eligibility (D8): only children that entered
    # the dispatch loop restart. _start_phase names where a start failure hit.
    self._reached_loop = False
    self._start_phase = "restore"

    # Non-blocking suspend intent (S2): set by suspend() / _agency.suspend,
    # actioned at the next message-loop boundary. Reason carried alongside.
    self._suspend_requested = False
    self._suspend_reason = ""
    # v0.9.4: category carried the same way as reason -- see suspend_category
    # property below for why the DURABLE marker, not this attribute, is the
    # authoritative read after a restore.
    self._suspend_category = SuspendCategory.OTHER
    # v0.9.6 (control-plane-writes.md D2): who initiated a suspend over the
    # control plane -- carried on the _agency.suspend message from the
    # authenticated HTTP principal, recorded in the agent.suspend AuditEvent.
    # Empty for a direct/programmatic suspend() (no HTTP principal).
    self._suspend_initiated_by = ""

    self._pending_streams: dict[str, StreamSink] = {}
    self._stream_producers: dict[str, str] = {}
    self._out_streams: dict[str, _OutStream] = {}

on_start() async

Called once before the first message. Initialize self.state here.

Runs synchronously during startup — for a dynamically spawned agent it executes inside the spawn call, so await self.spawn(...) does not return until on_start() finishes (and raises TimeoutError if it outlasts the ask timeout). Keep it fast: do slow / I/O-bound work (LLM calls, browser sessions) in handle(), kicked off by a fire-and-forget message the spawner sends right after the spawn is confirmed. Spawn-time config is available here as self.config.

Source code in civitas/process.py

async def on_start(self) -> None:
    """Called once before the first message. Initialize self.state here.

    Runs synchronously during startup — for a dynamically spawned agent it
    executes *inside* the spawn call, so ``await self.spawn(...)`` does not
    return until ``on_start()`` finishes (and raises ``TimeoutError`` if it
    outlasts the ask timeout). Keep it fast: do slow / I/O-bound work (LLM
    calls, browser sessions) in ``handle()``, kicked off by a fire-and-forget
    message the spawner sends right after the spawn is confirmed. Spawn-time
    ``config`` is available here as ``self.config``.
    """

handle(message) async

Called for every incoming message.

Return self.reply(...) for request-reply. Return None for fire-and-forget.

Source code in civitas/process.py

async def handle(self, message: Message) -> Message | None:
    """Called for every incoming message.

    Return self.reply(...) for request-reply. Return None for fire-and-forget.
    """
    return None

on_error(error, message) async

Called when handle() raises an exception.

Return an ErrorAction. Default: ESCALATE (crash, let supervisor decide).

Source code in civitas/process.py

async def on_error(self, error: Exception, message: Message) -> ErrorAction:
    """Called when handle() raises an exception.

    Return an ErrorAction. Default: ESCALATE (crash, let supervisor decide).
    """
    return ErrorAction.ESCALATE

on_stop() async

Called on graceful shutdown. Always called — even on crash.

Source code in civitas/process.py

async def on_stop(self) -> None:
    """Called on graceful shutdown. Always called — even on crash."""

on_child_terminated(name, reason) async

Called when a dynamically spawned child is permanently removed.

reason is one of: "restarts_exhausted", "despawned", "clean_exit". Default implementation logs a warning. Override to re-spawn, alert, etc.

Source code in civitas/process.py

async def on_child_terminated(self, name: str, reason: str) -> None:
    """Called when a dynamically spawned child is permanently removed.

    reason is one of: "restarts_exhausted", "despawned", "clean_exit".
    Default implementation logs a warning. Override to re-spawn, alert, etc.
    """
    logger.warning("[%s] dynamic child '%s' terminated: %s", self.name, name, reason)

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

Fire-and-forget: send a message to another agent by name.

Source code in civitas/process.py

async def send(
    self,
    recipient: str,
    payload: dict[str, Any],
    message_type: str = "message",
) -> None:
    """Fire-and-forget: send a message to another agent by name."""
    self._reject_reserved_type(message_type)
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    trace_id = ""
    parent_span_id: str | None = None
    if self._current_message is not None:
        trace_id = self._current_message.trace_id
        parent_span_id = self._current_message.span_id

    message = Message(
        type=message_type,
        sender=self.name,
        recipient=recipient,
        payload=payload,
        trace_id=trace_id,
        span_id=_new_span_id(),
        parent_span_id=parent_span_id,
    )
    await self._bus.route(message)
    if self._metrics is not None:
        self._metrics.message_sent(self.name)

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

Request-reply: send a message and await a response.

timeout (v0.10.0): a positive value is a bounded wait (default 30s, unchanged). None — or -1/any value <= 0 — means wait INDEFINITELY until the recipient replies. This is the HITL case: an ask() to an agent that suspends for approval buffers until the agent is resumed (hours/days later), then returns the real reply. The suspension itself has always been indefinite; this lets the caller wait it out.

Caveat: an indefinite ask() holds resources (the pending reply, and on ZMQ/NATS an open subscription) for the whole wait. Fine for a background worker driving a HITL flow; a request-scoped caller (e.g. an HTTP handler) must NOT block a connection for days — use send() + poll/ webhook instead.

Source code in civitas/process.py

async def ask(
    self,
    recipient: str,
    payload: dict[str, Any],
    message_type: str = "message",
    timeout: float | None = 30.0,
    *,
    fail_if_suspended: bool = False,
) -> Message:
    """Request-reply: send a message and await a response.

    ``timeout`` (v0.10.0): a positive value is a bounded wait (default 30s,
    unchanged). ``None`` — or ``-1``/any value ``<= 0`` — means **wait
    INDEFINITELY** until the recipient replies. This is the HITL case: an
    ``ask()`` to an agent that suspends for approval buffers until the agent
    is resumed (hours/days later), then returns the real reply. The
    suspension itself has always been indefinite; this lets the caller wait
    it out.

    Caveat: an indefinite ask() holds resources (the pending reply, and on
    ZMQ/NATS an open subscription) for the whole wait. Fine for a background
    worker driving a HITL flow; a request-scoped caller (e.g. an HTTP
    handler) must NOT block a connection for days — use ``send()`` + poll/
    webhook instead.
    """
    self._reject_reserved_type(message_type)
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    trace_id = ""
    parent_span_id: str | None = None
    if self._current_message is not None:
        trace_id = self._current_message.trace_id
        parent_span_id = self._current_message.span_id

    correlation_id = _uuid7()
    message = Message(
        type=message_type,
        sender=self.name,
        recipient=recipient,
        payload=payload,
        correlation_id=correlation_id,
        trace_id=trace_id,
        span_id=_new_span_id(),
        parent_span_id=parent_span_id,
    )
    if self._metrics is not None:
        self._metrics.message_sent(self.name)
    return await self._bus.request(
        message, timeout=timeout, fail_if_suspended=fail_if_suspended
    )

broadcast(pattern, payload) async

Send a message to all agents matching a glob pattern.

Source code in civitas/process.py

async def broadcast(self, pattern: str, payload: dict[str, Any]) -> None:
    """Send a message to all agents matching a glob pattern."""
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    targets = self._bus.lookup_all(pattern)
    for target in targets:
        await self.send(target.name, payload)

reply(payload)

Create a reply message. Return this from handle() for request-reply.

Source code in civitas/process.py

def reply(self, payload: dict[str, Any]) -> Message:
    """Create a reply message. Return this from handle() for request-reply."""
    if self._current_message is None:
        raise RuntimeError("reply() called outside of handle()")
    return self.reply_to(self._current_message, payload)

checkpoint() async

Save self.state to the configured StateStore.

Call this from handle() after completing a meaningful unit of work. On restart, self.state is automatically restored from the last checkpoint. Agents that never call checkpoint() incur zero overhead.

Source code in civitas/process.py

async def checkpoint(self) -> None:
    """Save self.state to the configured StateStore.

    Call this from handle() after completing a meaningful unit of work.
    On restart, self.state is automatically restored from the last checkpoint.
    Agents that never call checkpoint() incur zero overhead.
    """
    if self.store is not None:
        await self.store.set(self.name, self.state)

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

Spawn a dynamic agent via the nearest ancestor DynamicSupervisor.

The nearest-ancestor special case of :meth:spawn_into — it resolves the ancestor DynamicSupervisor name wired at startup and delegates. Sends a civitas.dynamic.spawn message and awaits confirmation. Raises SpawnError if no DynamicSupervisor ancestor exists or spawn is denied. Returns the agent name on success.

With wait=True (default) the call returns only after the child reaches RUNNING/SUSPENDED and raises SpawnError if its start fails. With wait=False it returns as soon as the child's task exists; a later start failure is delivered via on_child_terminated (R1 · D2).

Source code in civitas/process.py

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

    The nearest-ancestor special case of :meth:`spawn_into` — it resolves the
    ancestor DynamicSupervisor name wired at startup and delegates. Sends a
    civitas.dynamic.spawn message and awaits confirmation. Raises SpawnError if
    no DynamicSupervisor ancestor exists or spawn is denied. Returns the agent
    name on success.

    With ``wait=True`` (default) the call returns only after the child reaches
    RUNNING/SUSPENDED and raises SpawnError if its start fails. With
    ``wait=False`` it returns as soon as the child's task exists; a later start
    failure is delivered via ``on_child_terminated`` (R1 · D2).
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    return await self.spawn_into(
        self._dynamic_supervisor_name, agent_class, name, config, wait=wait
    )

despawn(name) async

Hard-stop a dynamic child immediately.

Cancels the agent's task. on_stop() still fires. Pending ask() callers into the agent receive SpawnError. The slot is freed immediately.

Source code in civitas/process.py

async def despawn(self, name: str) -> None:
    """Hard-stop a dynamic child immediately.

    Cancels the agent's task. on_stop() still fires. Pending ask() callers
    into the agent receive SpawnError. The slot is freed immediately.
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    reply = await self.ask(
        self._dynamic_supervisor_name,
        {"name": name},
        message_type="civitas.dynamic.despawn",
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "despawn failed"))

stop(name, drain='current', timeout=30.0) async

Soft-stop a dynamic child. Awaitable — returns when fully stopped.

drain="current" — finishes the message currently being handled, then stops. drain="all" — drains the full mailbox, then stops. timeout — fallback hard stop if drain isn't complete in time.

Source code in civitas/process.py

async def stop(
    self,
    name: str,
    drain: str = "current",
    timeout: float = 30.0,
) -> None:
    """Soft-stop a dynamic child. Awaitable — returns when fully stopped.

    drain="current" — finishes the message currently being handled, then stops.
    drain="all"     — drains the full mailbox, then stops.
    timeout         — fallback hard stop if drain isn't complete in time.
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    reply = await self.ask(
        self._dynamic_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.process.ProcessStatus

Bases: Enum

Lifecycle states for an AgentProcess.

---

civitas.process.Mailbox(maxsize=1000, priority_maxsize=100)

Bounded async queue for incoming messages with priority support.

High-priority system messages (priority > 0) are placed at the front. Normal messages follow FIFO order. Backpressure is applied when the mailbox is full — the sender awaits until space is available.

Source code in civitas/process.py

def __init__(self, maxsize: int = 1000, priority_maxsize: int = 100) -> None:
    self._queue: asyncio.Queue[Message] = asyncio.Queue(maxsize=maxsize)
    # priority_maxsize=0 means unbounded (asyncio.Queue convention) — used by
    # Supervisor (v0.9.0 E4/D-E4-2): its crash self-messages are enqueued from
    # a SYNC task-done callback that cannot await a bounded put(), and a
    # bounded put_nowait() would reintroduce the crash-drop bug class H2
    # removed. Agents keep the default bounded 100.
    self._priority_queue: asyncio.Queue[Message] = asyncio.Queue(maxsize=priority_maxsize)
    self._notify: asyncio.Event = asyncio.Event()

put(message) async

Enqueue a message. Priority messages bypass the normal queue.

Source code in civitas/process.py

async def put(self, message: Message) -> None:
    """Enqueue a message. Priority messages bypass the normal queue."""
    if message.priority > 0:
        await self._priority_queue.put(message)
        self._notify.set()
    else:
        await self._queue.put(message)
        self._notify.set()

put_nowait(message)

Synchronous enqueue for callers that cannot await (D-E4-2) — e.g. an asyncio.Task done-callback. Priority messages only; raises asyncio.QueueFull on a bounded queue at capacity (agents' normal queue backpressure is unaffected — this bypasses put() entirely).

Source code in civitas/process.py

def put_nowait(self, message: Message) -> None:
    """Synchronous enqueue for callers that cannot await (D-E4-2) — e.g. an
    ``asyncio.Task`` done-callback. Priority messages only; raises
    ``asyncio.QueueFull`` on a bounded queue at capacity (agents' normal
    queue backpressure is unaffected — this bypasses `put()` entirely).
    """
    self._priority_queue.put_nowait(message)
    self._notify.set()

get() async

Dequeue the next message. Priority messages are served first.

Messages whose ttl has elapsed are discarded with a warning instead of being returned; the search continues for the next message.

Source code in civitas/process.py

async def get(self) -> Message:
    """Dequeue the next message. Priority messages are served first.

    Messages whose ttl has elapsed are discarded with a warning instead
    of being returned; the search continues for the next message.
    """
    while True:
        message: Message | None = None
        if not self._priority_queue.empty():
            message = self._priority_queue.get_nowait()
        elif not self._queue.empty():
            message = self._queue.get_nowait()

        if message is not None:
            if self._drop_if_expired(message):
                continue
            return message

        # Wait for a notification
        self._notify.clear()
        # Double-check after clearing (avoid race)
        if not self._priority_queue.empty() or not self._queue.empty():
            continue
        await self._notify.wait()

get_priority() async

Dequeue the next priority-queue message only, leaving normal messages buffered.

Used while an agent is SUSPENDED (durable-suspension S3): control messages (priority > 0) are actioned while business messages stay in the normal queue, preserving FIFO order and backpressure for resume.

Concurrency footgun (Oracle finding #10): _notify is shared and is also set by normal-queue puts. We therefore clear it and re-check the priority queue before awaiting, so a normal put only produces a bounded spurious wakeup (no busy-loop) and a priority put racing with the clear is never lost.

Source code in civitas/process.py

async def get_priority(self) -> Message:
    """Dequeue the next priority-queue message only, leaving normal messages buffered.

    Used while an agent is SUSPENDED (durable-suspension S3): control
    messages (priority > 0) are actioned while business messages stay in
    the normal queue, preserving FIFO order and backpressure for resume.

    Concurrency footgun (Oracle finding #10): ``_notify`` is shared and is
    also set by normal-queue puts. We therefore clear it and re-check the
    priority queue *before* awaiting, so a normal put only produces a
    bounded spurious wakeup (no busy-loop) and a priority put racing with
    the clear is never lost.
    """
    while True:
        if not self._priority_queue.empty():
            message = self._priority_queue.get_nowait()
            if self._drop_if_expired(message):
                continue
            return message

        self._notify.clear()
        if not self._priority_queue.empty():
            continue
        await self._notify.wait()

empty()

Return True if both priority and normal queues are empty.

Source code in civitas/process.py

def empty(self) -> bool:
    """Return True if both priority and normal queues are empty."""
    return self._priority_queue.empty() and self._queue.empty()

depth()

Total buffered messages (both queues). Sync-safe — used by the Worker health responder's snapshot (D5, report-only).

Source code in civitas/process.py

def depth(self) -> int:
    """Total buffered messages (both queues). Sync-safe — used by the
    Worker health responder's snapshot (D5, report-only)."""
    return self._priority_queue.qsize() + self._queue.qsize()

peek()

Non-destructively snapshot all buffered messages (priority first), WITHOUT consuming them (v0.9.6, control-plane-writes.md §6 mailbox introspection). Reads asyncio.Queue's backing collections.deque (._queue) directly -- the only non-consuming way to inspect it; get()/drain() all consume. For introspection/reporting only.

Source code in civitas/process.py

def peek(self) -> list[Message]:
    """Non-destructively snapshot all buffered messages (priority first),
    WITHOUT consuming them (v0.9.6, control-plane-writes.md §6 mailbox
    introspection). Reads ``asyncio.Queue``'s backing ``collections.deque``
    (``._queue``) directly -- the only non-consuming way to inspect it;
    ``get()``/``drain()`` all consume. For introspection/reporting only.
    """
    return list(self._priority_queue._queue) + list(self._queue._queue)  # type: ignore[attr-defined]

drain()

Remove and return all buffered messages, priority queue first.

Source code in civitas/process.py

def drain(self) -> list[Message]:
    """Remove and return all buffered messages, priority queue first."""
    drained: list[Message] = []
    while not self._priority_queue.empty():
        drained.append(self._priority_queue.get_nowait())
    while not self._queue.empty():
        drained.append(self._queue.get_nowait())
    return drained