Supervisor

Monitors child agents and supervisors. Applies restart strategies on failure.

See Supervision for a full guide.

---

civitas.supervisor.Supervisor(name, children=None, strategy='ONE_FOR_ONE', max_restarts=3, restart_window=60.0, backoff='CONSTANT', backoff_base=1.0, backoff_max=60.0)

Bases: AgentProcess

Manages child processes with restart strategies.

When a child crashes, the supervisor applies the configured restart strategy. If max_restarts is exceeded within restart_window, the supervisor escalates to its parent or stops permanently.

v0.9.0 E4 (D6, design supervision-endgame.md §6): a Supervisor is now itself an actor — addressable, registered (SUPERVISOR_CAPABILITY), with its own mailbox and message loop. Phase A (this constructor + the start/stop ordering below) is a zero-behavior-change skeleton: crash events still flow through the pre-E4 queue/drain-task mechanism until Phase B swaps the control plane onto the mailbox. Public constructor signature is unchanged — the actorization is purely internal.

Source code in civitas/supervisor.py

def __init__(
    self,
    name: str,
    children: list[AgentProcess | Supervisor] | None = None,
    strategy: str = "ONE_FOR_ONE",
    max_restarts: int = 3,
    restart_window: float = 60.0,
    backoff: str = "CONSTANT",
    backoff_base: float = 1.0,
    backoff_max: float = 60.0,
) -> None:
    super().__init__(name)
    # D-E4-2: supervisors' priority queue is UNBOUNDED (0) — crash
    # self-messages are enqueued from a sync task-done callback (Phase B)
    # that cannot await a bounded put(); a bounded put_nowait() would
    # reintroduce the crash-drop bug class H2 removed. Volume is bounded
    # by child count in practice (the same judgment call H2 made for the
    # queue this replaces).
    self._mailbox = Mailbox(maxsize=1000, priority_maxsize=0)
    self.children: list[AgentProcess | Supervisor] = children or []
    self.strategy = RestartStrategy(strategy)
    # E1 (v0.9.0): budgets/window/backoff live in ONE engine shared with
    # DynamicSupervisor's per-child engines (design supervision-endgame §3).
    # The five knobs remain public attributes via properties below.
    self._engine = RestartEngine(
        max_restarts=max_restarts,
        restart_window=restart_window,
        backoff=BackoffPolicy(backoff),
        backoff_base=backoff_base,
        backoff_max=backoff_max,
    )

    # Internal state
    # Lifetime crash counters — OBSERVABILITY ONLY since B3 (logs/spans);
    # never a backoff input (backoff derives from window occupancy).
    self._restart_counts: dict[str, int] = {}
    self._child_tasks: dict[str, asyncio.Task[None]] = {}
    self._children_by_name: dict[str, AgentProcess | Supervisor] = {  # F03-11: O(1) lookup
        c.name: c for c in self.children
    }
    # H2 (#30) / v0.9.0 E4 Phase B (D-E4-1, D6): crash events are processed
    # strictly sequentially through the Supervisor's OWN mailbox/message-loop
    # (this supervisor is now an AgentProcess) — OTP supervisors handle EXIT
    # signals one at a time. Message.payload is JSON-primitives-only, but a
    # crash event carries a real Exception object (add_crash_callback's public
    # contract) and an asyncio.Task (the stale-incarnation marker); neither can
    # ride the mailbox directly. Resolution: the mailbox carries only an
    # event-id trigger (_agency.child_crashed); the real objects live here,
    # keyed by event-id. Items: (child_name, exception, task-at-crash-time |
    # None). The task is the child's incarnation marker — a queued event whose
    # task is no longer the child's current task is stale (the child was
    # already restarted by an earlier cycle) and is skipped, mirroring OTP's
    # EXIT-pid matching.
    self._pending_crash_events: dict[str, tuple[str, Exception, asyncio.Task[None] | None]] = {}
    self._running = False
    self._parent: Supervisor | None = None
    self._crash_callbacks: list[Callable[[str, Exception], Awaitable[None]]] = []

    # D1a (v0.9.0): re-invokable wiring for fresh incarnations. Runtime
    # registers this at start (ComponentSet.inject + credentials); a fresh
    # instance must be FULLY wired before its task starts (constraint 1).
    self._wire_child: Callable[[AgentProcess], None] | None = None
    # Notified with (name, new_agent) after a child object is replaced —
    # Runtime updates its O(1) map and TopologyServer references here (Q1:
    # user-held references go stale by design; route by name).
    self._child_replaced_callbacks: list[Callable[[str, AgentProcess], None]] = []
    # _bus/_registry/_tracer already initialized to None by AgentProcess.__init__
    # above; Runtime injects the real values (agent path) or the dedicated
    # supervisor-wiring block (Runtime.start) before the tree starts.

    # Heartbeat monitoring for remote agents
    self._remote_children: set[str] = set()
    self._heartbeat_task: asyncio.Task[None] | None = None
    self._missed_heartbeats: dict[str, int] = {}
    self._remote_child_config: dict[str, dict[str, float | int]] = {}  # F03-3: per-child config

start() async

Start all children and begin monitoring them.

Source code in civitas/supervisor.py

async def start(self) -> None:
    """Start all children and begin monitoring them."""
    self._running = True

    # D-E4-3 (v0.9.0, Phase A): start the supervisor's OWN message loop
    # first — it must be live before any child can crash-report through it
    # (Phase B/D6: crash events self-trigger onto this loop). Callable again
    # after stop() — same pattern H1 subtree-restart already uses.
    await self._start()

    # Set parent references for child supervisors
    for child in self.children:
        if isinstance(child, Supervisor):
            child._parent = self

    # Start children bottom-up (supervisors first start their children)
    for child in self.children:
        if isinstance(child, Supervisor):
            await child.start()
        else:
            await self._start_child(child)

    # Start heartbeat monitoring for remote children
    await self._start_heartbeat_monitor()

stop() async

Stop all children gracefully.

v0.9.0 E4 (D-E4-6, found during Phase A implementation): this INTENTIONALLY shadows AgentProcess.stop(name, drain, timeout) (the soft-stop-a-dynamic-child API). The two are unrelated operations that happen to share a name now that Supervisor is an AgentProcess. This is safe: the inherited method requires self._dynamic_supervisor_name to be wired, and Runtime's _wire_dyn_sup never sets it on a Supervisor node (only recurses through its children) — so the shadowed method could only ever have raised SpawnError on a Supervisor instance. Pre-existing public API (sup.stop(), no args) takes precedence over the newly-inherited one; renaming either public method would be the breaking change, not keeping this override.

D-E4-8 (v0.9.0 E4 Phase B, correcting D-E4-3): the own loop stops FIRST here, not last. Crash-triggered restarts (including the backoff asyncio.sleep) now run on this same loop; only cancelling it — which self._stop()'s own timeout-then-cancel fallback does — can abort a restart already asleep in backoff. Stopping it last (as Phase A did, safely, while the mailbox was inert) would let a crash's backoff complete and resurrect a child mid-teardown, once cumulative child-stop time exceeds the backoff delay. This restores exact parity with the pre-E4 "cancel crash-drain before touching children" guarantee, via the mechanism every other AgentProcess already gets.

Source code in civitas/supervisor.py

async def stop(self) -> None:  # type: ignore[override]  # D-E4-6: see docstring
    """Stop all children gracefully.

    v0.9.0 E4 (D-E4-6, found during Phase A implementation): this
    INTENTIONALLY shadows ``AgentProcess.stop(name, drain, timeout)`` (the
    soft-stop-a-dynamic-child API). The two are unrelated operations that
    happen to share a name now that Supervisor is an AgentProcess. This is
    safe: the inherited method requires ``self._dynamic_supervisor_name``
    to be wired, and Runtime's ``_wire_dyn_sup`` never sets it on a
    Supervisor node (only recurses through its children) — so the
    shadowed method could only ever have raised ``SpawnError`` on a
    Supervisor instance. Pre-existing public API (``sup.stop()``, no args)
    takes precedence over the newly-inherited one; renaming either public
    method would be the breaking change, not keeping this override.

    D-E4-8 (v0.9.0 E4 Phase B, correcting D-E4-3): the own loop stops
    FIRST here, not last. Crash-triggered restarts (including the backoff
    ``asyncio.sleep``) now run on this same loop; only cancelling it —
    which ``self._stop()``'s own timeout-then-cancel fallback does — can
    abort a restart already asleep in backoff. Stopping it last (as Phase
    A did, safely, while the mailbox was inert) would let a crash's
    backoff complete and resurrect a child mid-teardown, once cumulative
    child-stop time exceeds the backoff delay. This restores exact parity
    with the pre-E4 "cancel crash-drain before touching children"
    guarantee, via the mechanism every other AgentProcess already gets.
    """
    self._running = False

    await self._stop()
    await self._stop_heartbeat_monitor()
    for child in reversed(self.children):
        if isinstance(child, Supervisor):
            await child.stop()
        else:
            await child._stop()

add_remote_child(name, heartbeat_interval=5.0, heartbeat_timeout=2.0, missed_heartbeats_threshold=3)

Register a remote child for heartbeat-based monitoring.

Remote children are agents running in a Worker process. They are monitored via periodic heartbeat pings instead of task callbacks.

Source code in civitas/supervisor.py

def add_remote_child(
    self,
    name: str,
    heartbeat_interval: float = 5.0,
    heartbeat_timeout: float = 2.0,
    missed_heartbeats_threshold: int = 3,
) -> None:
    """Register a remote child for heartbeat-based monitoring.

    Remote children are agents running in a Worker process. They are
    monitored via periodic heartbeat pings instead of task callbacks.
    """
    self._remote_children.add(name)
    self._missed_heartbeats[name] = 0
    # F03-3: per-child config stored in dict, not shared scalars
    self._remote_child_config[name] = {
        "interval": heartbeat_interval,
        "timeout": heartbeat_timeout,
        "threshold": missed_heartbeats_threshold,
    }

all_agents()

Recursively collect all AgentProcess instances in the tree.

Source code in civitas/supervisor.py

def all_agents(self) -> list[AgentProcess]:
    """Recursively collect all AgentProcess instances in the tree."""
    agents: list[AgentProcess] = []
    for child in self.children:
        if isinstance(child, Supervisor):
            agents.extend(child.all_agents())
        else:
            agents.append(child)
    return agents

all_supervisors()

Recursively collect all Supervisor instances (including self).

Source code in civitas/supervisor.py

def all_supervisors(self) -> list[Supervisor]:
    """Recursively collect all Supervisor instances (including self)."""
    supervisors: list[Supervisor] = [self]
    for child in self.children:
        if isinstance(child, Supervisor):
            supervisors.extend(child.all_supervisors())
    return supervisors

---

civitas.supervisor.RestartStrategy

Bases: Enum

Strategy used by a Supervisor when a child process crashes.

---

civitas.supervisor.BackoffPolicy

Bases: Enum

Delay strategy applied between successive restart attempts.

---

DynamicSupervisor

Starts empty. Children are added and removed at runtime via self.spawn() / self.despawn(). Always uses ONE_FOR_ONE. See Dynamic supervision for a full guide.

---

civitas.supervisor.DynamicSupervisor(name, max_children=None, max_total_spawns=None, restart='transient', max_restarts=3, restart_window=60.0, spawner_allowlist=None, max_children_per_spawner=None, max_total_spawns_per_spawner=None, **kwargs)

Bases: AgentProcess

Dynamic supervisor — starts empty, children added at runtime via spawn().

Declared as a static child in topology YAML under type: dynamic_supervisor. Only its children change at runtime. Enforces ONE_FOR_ONE restart semantics — no escalation to parent on restart exhaustion; fires on_child_terminated instead.

Agents call self.spawn() / self.despawn() / self.stop() to manage children. All requests travel as bus messages (civitas.dynamic.*) so the same API works in-process (v0.4) and cross-process (v0.5).

spawner_allowlist (optional) restricts who may spawn children here: when a set is given, a spawn whose spawner is not in it is rejected before the on_spawn_requested hook runs. Default None keeps the open behavior. It is the built-in authorization control for cross-tree spawn_into (D8).

max_children_per_spawner / max_total_spawns_per_spawner (optional, R5) cap a single spawner's concurrent and lifetime spawns, in addition to the supervisor-wide max_children / max_total_spawns. Default None is unbounded per spawner.

Source code in civitas/supervisor.py

def __init__(
    self,
    name: str,
    max_children: int | None = None,
    max_total_spawns: int | None = None,
    restart: str = "transient",
    max_restarts: int = 3,
    restart_window: float = 60.0,
    spawner_allowlist: set[str] | None = None,
    max_children_per_spawner: int | None = None,
    max_total_spawns_per_spawner: int | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(name, **kwargs)
    self.max_children = max_children
    self.max_total_spawns = max_total_spawns
    self._restart_mode = RestartMode(restart)
    self._ds_max_restarts = max_restarts
    self._ds_restart_window = restart_window
    self.spawner_allowlist = spawner_allowlist
    self.max_children_per_spawner = max_children_per_spawner
    self.max_total_spawns_per_spawner = max_total_spawns_per_spawner
    self._current_spawner: str | None = None

    # Live child tracking
    self._dynamic_children: dict[str, _ChildRec] = {}
    self._child_tasks: dict[str, asyncio.Task[None]] = {}
    self._spawner_names: dict[str, str] = {}
    self._child_restart_counts: dict[str, int] = {}
    # E1: one RestartEngine per dynamic child (per-child budgets — the
    # pre-existing DynSup semantics; backoff=None — DynSup never delayed).
    self._child_engines: dict[str, RestartEngine] = {}
    self._total_spawns: int = 0
    self._spawner_total_counts: dict[str, int] = {}
    self._pending_child_tasks: set[asyncio.Task[None]] = set()
    # Monotonic incarnation counter stamped on each child's cluster-wide
    # announcement so peers can reject stale/reordered register/deregister (D13).
    self._spawn_epoch: int = 0

on_spawn_requested(agent_class, name, config) async

Governance veto hook. Return False to deny the spawn request.

Default implementation approves all requests. Subclass to enforce allowlists, rate limits, or policy checks. Read :attr:current_spawner inside this hook to authorize by the requesting agent's name.

Source code in civitas/supervisor.py

async def on_spawn_requested(
    self, agent_class: type, name: str, config: dict[str, Any]
) -> bool:
    """Governance veto hook. Return False to deny the spawn request.

    Default implementation approves all requests. Subclass to enforce
    allowlists, rate limits, or policy checks. Read :attr:`current_spawner`
    inside this hook to authorize by the requesting agent's name.
    """
    return True

all_dynamic_agents()

Return the currently live dynamic children.

Source code in civitas/supervisor.py

def all_dynamic_agents(self) -> list[AgentProcess]:
    """Return the currently live dynamic children."""
    return [rec.agent for rec in self._dynamic_children.values()]

on_stop() async

Cancel all dynamic children on shutdown.

Source code in civitas/supervisor.py

async def on_stop(self) -> None:
    """Cancel all dynamic children on shutdown."""
    for name, _rec in list(self._dynamic_children.items()):
        task = self._child_tasks.get(name)
        if task is not None and not task.done():
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

    for t in list(self._pending_child_tasks):
        t.cancel()
    if self._pending_child_tasks:
        await asyncio.gather(*self._pending_child_tasks, return_exceptions=True)

---

civitas.supervisor.RestartMode

Bases: Enum

Restart policy for dynamic children.