embassy_supervisor/

lib.rs

// `no_std` for the shipped crate and the embedded build; under `cargo test` the
// crate is built for the host, where the test harness and the unit tests need `std`.
#![cfg_attr(not(test), no_std)]
#![forbid(unsafe_code)]
#![deny(missing_docs)]
//! # embassy-supervisor — a task-lifecycle supervisor for [embassy](https://embassy.dev)
//!
//! Application- and HAL-agnostic primitives for orchestrating a set of embassy
//! tasks: bringing them up in dependency order, tearing them down in reverse,
//! scaling an elastic worker pool with load, and starting/stopping/pausing/
//! resuming individual tasks at runtime while keeping the dependency graph
//! consistent.
//!
//! ## The model
//!
//!   * The graph is declared once with the [`supervisor_graph!`] macro: each
//!     managed task becomes a [`TaskNode`] `static`, and the macro bundles the node
//!     slots, dependency table, and a topological order computed **at compile time**
//!     (a dependency cycle is a compile error) into a single [`Graph`] (`GRAPH`).
//!   * [`Supervisor::new`] takes `&GRAPH` (no work, no failure) and uses the order
//!     to bring tasks up in dependency order ([`Supervisor::start`]) and tear them
//!     down in reverse ([`Supervisor::teardown`]).
//!   * Each node carries a `TaskHandle` of per-node atomic flags and
//!     single-consumer `Signal`s. Every node is single-instance — no counts, no
//!     fan-out. See [`TaskHandle`].
//!
//! ## Three lifecycles, distinguished by [`Mode`]
//!
//!   * [`Mode::Terminate`] — the task exits its loop on shutdown and is respawned
//!     on the next bring-up. Stateless services (a network listener, a logger).
//!   * [`Mode::Pause`] — the task acks the shutdown then parks on
//!     `wait_resume()`; it is resumed in place, never respawned. Tasks that
//!     retain a resource across the pause (an open peripheral handle, a socket).
//!   * [`Mode::OnDemand`] — like `Terminate`, but not started at boot and not
//!     auto-respawned; the supervisor brings it up and down at runtime to scale
//!     an elastic worker pool ([`ElasticPool`]) with load.
//!
//! ## What the supervisor does *not* do
//!
//!   * It does not model any power-state transition (sleep/wake): it reacts to
//!     "teardown" and "bring-up" requests; the application drives them.
//!   * It does not allocate, and does no work at construction: the topological
//!     sort runs at compile time (see the `supervisor_graph!` macro).
//!   * It does not observe task internals. Tasks self-report their drop state via
//!     `ack_dropped()`; a task that fails to ack within a timeout panics the
//!     supervisor with the offending node's name.
//!
//! ## Cargo features
//!
//!   * `control` *(default)* — the runtime control plane: [`ControlOp`],
//!     [`request_control`], [`Supervisor::apply_control`].
//!   * `pool` *(default)* — elastic worker pools: [`ElasticPool`],
//!     [`Supervisor::run_pools`], and the `pools` field of [`Graph`].
//!   * `defmt` — route the supervisor's logs through `defmt`; without it the log
//!     macros are no-ops.
//!
//! Build with `default-features = false` for a minimal core that only does
//! dependency-ordered bring-up/teardown (drops the control plane and pools,
//! trimming flash and a couple of statics).
//!
//! ## Example
//!
//! [`supervisor_graph!`] declares the whole graph once — it generates the node
//! `static`s and a single [`Graph`] value `GRAPH` bundling the node slots, dep
//! table, and compile-time topological order (a dependency cycle is a compile
//! error), which [`Supervisor::new`] consumes.
//!
//! ```ignore
//! use embassy_executor::Spawner;
//! use embassy_supervisor::{supervisor_graph, Supervisor, wait_control};
//!
//! // `app` depends on `net`; each `spawn:` names a task fn spawned with the node.
//! supervisor_graph! {
//!     node NET = Terminate, deps: [], spawn: net_task;
//!     node APP = Terminate, deps: [NET], spawn: app_task;
//! }
//!
//! #[embassy_executor::task]
//! async fn supervisor_task(spawner: Spawner) {
//!     let sup = Supervisor::new(&GRAPH);
//!     sup.start(spawner).expect("initial spawn"); // brings up `net`, then `app`
//!     loop {
//!         // Apply runtime start/stop/pause/resume requests in dependency order.
//!         let cmd = wait_control().await;
//!         sup.apply_control(cmd, spawner).await;
//!     }
//!     // With the `pool` feature you'd instead drive scaling and control together:
//!     // `select(sup.run_pools(spawner), wait_control())` (see the `firmware` crate).
//! }
//! ```
//!
//! The `firmware` crate in the [repository](https://github.com/cedrivard/embassy-supervisor)
//! is a complete working example (USB-net, an HTTP control plane, an elastic pool,
//! and OTA).
//!
//! Docs:
//!   - `embassy_sync::signal::Signal`: <https://docs.embassy.dev/embassy-sync>
//!   - Kahn's algorithm (topological sort):
//!     <https://en.wikipedia.org/wiki/Topological_sorting#Kahn's_algorithm>

#[macro_use]
mod fmt;

use core::sync::atomic::Ordering;

use embassy_executor::{SpawnError, Spawner};
use embassy_futures::select::{Either, select};
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
#[cfg(feature = "control")]
use embassy_sync::channel::Channel;
use embassy_sync::signal::Signal;
use embassy_time::Timer;
use portable_atomic::AtomicBool;

// ─── Scale-request signal (task → supervisor) ──────────────────────────────
//
// Elastic pool workers fire this when their busy/idle status changes; the
// supervisor's `run_pools` loop awaits it and re-runs the pool policies
// (`ElasticPool`). Single-consumer `Signal`: many tasks may `signal()`, only the
// supervisor `wait()`s. This is the *only* path by which task status reaches the
// supervisor — it never polls.
#[cfg(feature = "pool")]
static SCALE_REQ: Signal<CriticalSectionRawMutex, ()> = Signal::new();

/// Fire the scale-request signal. Called by a task on a busy/idle transition.
/// A no-op when the `pool` feature is disabled (no pools to re-evaluate).
pub fn request_scale() {
    #[cfg(feature = "pool")]
    SCALE_REQ.signal(());
}

/// Await the next scale request. The supervisor's driver loop selects this
/// against its other wake sources and runs the scaling policy on each wake.
#[cfg(feature = "pool")]
pub async fn wait_scale() {
    SCALE_REQ.wait().await;
}

// ─── Runtime control commands (app → supervisor) ───────────────────────────
//
// An application's control surface (e.g. a network endpoint) usually can't drive
// the supervisor directly: the `Supervisor` and the `Spawner` live on the
// supervisor task's stack, not in a `static`. So control is decoupled via this
// channel — the caller `request_control()`s a (node, op) pair and returns
// immediately; the supervisor's driver loop `wait_control()`s it and runs the
// dependency-honoring `apply_control`. A `Channel` (not a `Signal`) so
// back-to-back requests aren't coalesced; capacity 4 is ample for hand-driven
// control and a full channel simply drops the surplus (`try_send`).

/// Which way to drive a node. Higher-level verbs fold onto these two:
/// `start`/`resume` → `Activate`, `stop`/`pause` → `Deactivate`. The concrete
/// mechanism (respawn vs resume vs leave-to-pool) is then chosen per node `Mode`
/// by the supervisor when it applies the command ([`Supervisor::apply_control`]).
#[cfg(feature = "control")]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ControlOp {
    /// Bring the node up (start a stopped `Terminate` node, resume a `Pause` node).
    Activate,
    /// Take the node down (and its dependents, per the graph).
    Deactivate,
}

/// A runtime control request: drive `node` (and, per the dependency graph and
/// pool membership, the nodes it implies) in the `op` direction.
#[cfg(feature = "control")]
#[derive(Clone, Copy, Debug)]
pub struct ControlCommand {
    /// The node to drive.
    pub node: &'static TaskNode,
    /// The direction to drive it.
    pub op: ControlOp,
}

/// App → supervisor control mailbox. `&'static TaskNode` is `Copy + Sync`, so
/// the target rides the channel directly — no name lookup needed supervisor-side.
#[cfg(feature = "control")]
static CONTROL_REQ: Channel<CriticalSectionRawMutex, ControlCommand, 4> = Channel::new();

/// Enqueue a control request. Non-blocking; drops if the mailbox is full (4
/// outstanding), which is harmless for low-frequency manual control. Called by
/// the application's control surface.
#[cfg(feature = "control")]
pub fn request_control(node: &'static TaskNode, op: ControlOp) {
    let _ = CONTROL_REQ.try_send(ControlCommand { node, op });
}

/// Await the next control request. Selected by the supervisor's driver loop
/// against pool scaling and any other application wake sources.
#[cfg(feature = "control")]
pub async fn wait_control() -> ControlCommand {
    CONTROL_REQ.receive().await
}

/// Per-node timeout for `wait_dropped`. A task that doesn't ack within this
/// window is a bug (e.g. a missing `ack_dropped()` call) and panics the
/// supervisor with the offending node's name. 2 s comfortably exceeds a typical
/// task's poll period and peripheral settle time.
const SHUTDOWN_ACK_TIMEOUT_MS: u64 = 2_000;

// ─── Mode ────────────────────────────────────────────────────────────────

/// Lifecycle policy for a managed task: what the task does on shutdown and what
/// the supervisor does to bring it back.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Mode {
    /// Task exits its loop on shutdown. The supervisor respawns it via the
    /// node's `spawn` fn from `respawn_terminate`.
    Terminate,
    /// Task acks shutdown and parks on `wait_resume()`. The supervisor resumes
    /// it from `resume_pausable`; the task is never respawned, so it keeps any
    /// resource it holds (a peripheral handle, a socket) across the pause.
    Pause,
    /// Like `Terminate` (exits on shutdown), but **not** started at boot and
    /// **not** auto-respawned. The supervisor brings it up and down at runtime
    /// via `start_node` / `stop_node` in response to load — see [`ElasticPool`].
    /// `start()` skips it; `respawn_terminate()` leaves it down (it
    /// re-grows under demand); `teardown()` only acts on it while it is running.
    OnDemand,
}

impl Mode {
    /// Stable lower-case wire name, used both for serialization (e.g. a JSON
    /// task-state view) and for `defmt` logging — the single source of these
    /// strings.
    pub fn as_str(&self) -> &'static str {
        match self {
            Mode::Terminate => "terminate",
            Mode::Pause => "pause",
            Mode::OnDemand => "ondemand",
        }
    }
}

#[cfg(feature = "defmt")]
impl defmt::Format for Mode {
    fn format(&self, f: defmt::Formatter) {
        defmt::write!(f, "{}", self.as_str());
    }
}

// ─── TaskHandle ──────────────────────────────────────────────────────────

/// Coordination state for one task. Embedded inside [`TaskNode`].
///
/// Every node is single-instance, so each field is a per-node atomic flag or a
/// single-consumer signal — no counts, no fan-out. Written by one side (task or
/// supervisor) and read by the other:
///   * `shutdown` / `shutdown_wake` — supervisor requests exit; the task parks
///     on the signal and reads the flag.
///   * `dropped` / `dropped_wake` — the task acks its exit; the supervisor
///     parks on the signal (with a timeout) and reads the flag.
///   * `resume_wake` — supervisor resumes a parked Pause-mode task.
///   * `running` — supervisor's record that the node is spawned; `busy` — the
///     task's active/idle status. Both read by the elastic scaling policy.
///   * `disabled` — the node has been manually deactivated; see below.
pub struct TaskHandle {
    /// Set true by the supervisor when shutdown is requested.
    /// Cleared by `reset()` before the next spawn.
    shutdown: AtomicBool,
    /// Wake source for `wait_shutdown()`. Fired by `signal_shutdown()`.
    shutdown_wake: Signal<CriticalSectionRawMutex, ()>,
    /// Set true by the instance when it acks the shutdown (a bool, not a count,
    /// since every node is single-instance). Cleared by `reset()`.
    dropped: AtomicBool,
    /// Wake source for `wait_dropped()`. Fired by `ack_dropped()`.
    dropped_wake: Signal<CriticalSectionRawMutex, ()>,
    /// True while the supervisor has the node spawned and it hasn't exited.
    /// Always-on nodes are set true by `start()`; `OnDemand` nodes are set
    /// true/false by `start_node()` / `stop_node()`. `teardown()` only acts on
    /// `running` nodes, so a down `OnDemand` node doesn't stall it.
    running: AtomicBool,
    /// True while the task is actively serving (its active/idle status). Set by
    /// `mark_busy()` / `mark_idle()`; read by the scaling policy.
    busy: AtomicBool,
    /// Wake source for `wait_resume()` on Pause-mode tasks. Fired by
    /// `signal_resume()`.
    resume_wake: Signal<CriticalSectionRawMutex, ()>,
    /// True while the node has been manually deactivated (stopped/paused) via the
    /// runtime control interface (`Supervisor::deactivate`). Unlike the other
    /// flags this one is **lifecycle-spanning**: it is *not* cleared by
    /// `reset()`, so a manual stop "sticks" — the automatic bring-up paths
    /// (`start`, `respawn_terminate`, `resume_pausable`, and the elastic pool's
    /// grow) skip a node while it is set. Cleared only by `Supervisor::activate`.
    /// Because it lives in a `static`, it also survives a power-state transition
    /// that retains RAM (e.g. a warm-resume from deep sleep).
    disabled: AtomicBool,
    /// The node manages its own lifecycle and must NOT be torn down by a
    /// dependency cascade. A dependency normally means "stop me when my dep stops",
    /// but a node can declare a dep purely for *start* ordering and then intend to
    /// outlive it — e.g. a node that needs a dependency up to initialize, then stops
    /// that dependency to reclaim its resources, must survive that teardown. While
    /// set, `deactivate` will not pull the node into a transitive-dependent teardown.
    /// Self-managed, so `reset()` leaves it (the node clears it itself when done).
    detached: AtomicBool,
}

impl TaskHandle {
    const fn new(disabled_at_boot: bool) -> Self {
        Self {
            shutdown: AtomicBool::new(false),
            shutdown_wake: Signal::new(),
            dropped: AtomicBool::new(false),
            dropped_wake: Signal::new(),
            running: AtomicBool::new(false),
            busy: AtomicBool::new(false),
            resume_wake: Signal::new(),
            disabled: AtomicBool::new(disabled_at_boot),
            detached: AtomicBool::new(false),
        }
    }
}

// ─── TaskNode ────────────────────────────────────────────────────────────

/// A node in the supervisor's task graph.
///
/// Designed to live in `static` memory: every field is `Sync`, all constructors
/// are `const`. Declared by [`supervisor_graph!`], which emits one per managed
/// task along with the [`Graph`] (`GRAPH`) that [`Supervisor::new`] consumes.
pub struct TaskNode {
    /// Human-readable name. Used in defmt logs and panic messages.
    pub name: &'static str,
    /// Lifecycle policy. See [`Mode`].
    pub mode: Mode,
    /// App-provided spawn function (typically an inline closure at the node's
    /// declaration). Called once at boot from `Supervisor::start`, again from
    /// `respawn_terminate` for Terminate nodes, and at runtime from `start_node`
    /// for `OnDemand` nodes. `None` for a **parked** node the application spawns
    /// itself (e.g. a `Pause` sensor holding a peripheral handle): the supervisor
    /// tracks its lifecycle but never spawns it.
    pub spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
    handle: TaskHandle,
}

impl TaskNode {
    /// A single-instance node started at boot (`Terminate`/`Pause`) or on demand
    /// (`Mode::OnDemand`). Every node is single-instance; an elastic service is
    /// modelled as several `OnDemand` nodes of the same pooled task fn.
    ///
    /// A `TaskNode` carries only its own identity and behaviour; the graph's
    /// dependency edges live in the compile-time index table that
    /// [`supervisor_graph!`] emits and [`Supervisor::new`] consumes.
    /// `disabled_at_boot` seeds the node's disabled flag so a control-started node
    /// (e.g. an OTA task) can be declared down and started later via a control op.
    /// `spawn` is `None` for a parked node the application spawns itself.
    pub const fn new(
        name: &'static str,
        mode: Mode,
        spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
        disabled_at_boot: bool,
    ) -> Self {
        Self {
            name,
            mode,
            spawn,
            handle: TaskHandle::new(disabled_at_boot),
        }
    }

    // ── Task-side API ────────────────────────────────────────────────────
    //
    // Called from inside the `#[embassy_executor::task] async fn` body.

    /// True iff the supervisor has requested shutdown. Checked at the loop top
    /// alongside `wait_shutdown()` in a `select`.
    pub fn shutdown_requested(&self) -> bool {
        self.handle.shutdown.load(Ordering::Acquire)
    }

    /// Park until shutdown is requested. Returns immediately if shutdown has
    /// already been requested. Use this for single-instance tasks in a `select`
    /// against the task's main work future.
    pub async fn wait_shutdown(&self) {
        // Fast path — already requested. (Important because the signal is
        // edge-triggered: if `signal()` fired before we got here, the bare
        // `wait()` below would block forever.)
        if self.handle.shutdown.load(Ordering::Acquire) {
            return;
        }
        self.handle.shutdown_wake.wait().await;
    }

    /// Mark this instance as having shut down: clears the running flag and acks
    /// the teardown handshake (so the supervisor's `wait_dropped` completes).
    /// Every instance must call this exactly once on exit (Terminate/OnDemand
    /// mode) or on each pause (Pause mode). It also covers an **autonomous** exit
    /// the supervisor didn't request — e.g. a pool worker backing off — so the
    /// pool sees the instance as down and can re-grow it under later demand.
    pub fn ack_dropped(&self) {
        self.handle.running.store(false, Ordering::Release);
        self.handle.dropped.store(true, Ordering::Release);
        self.handle.dropped_wake.signal(());
    }

    /// Pause-mode only: park until the supervisor signals resume.
    pub async fn wait_resume(&self) {
        self.handle.resume_wake.wait().await;
    }

    /// Report that this task started serving a request (active). Fires the
    /// scale-request signal on a real idle→busy transition so the scaling policy
    /// can react (e.g. grow the pool); a redundant call doesn't re-signal.
    pub fn mark_busy(&self) {
        if !self.handle.busy.swap(true, Ordering::Release) {
            request_scale();
        }
    }

    /// Report that this task finished serving and is idle again. Fires the
    /// scale-request signal on a real busy→idle transition so the scaling policy
    /// can react (e.g. shrink the pool); a redundant call doesn't re-signal.
    pub fn mark_idle(&self) {
        if self.handle.busy.swap(false, Ordering::Release) {
            request_scale();
        }
    }

    /// True while this task is actively serving. Read by the scaling policy.
    pub fn is_busy(&self) -> bool {
        self.handle.busy.load(Ordering::Acquire)
    }

    /// True while the supervisor has this node spawned (and it hasn't exited).
    /// Read by the scaling policy to count live instances, and by a task-state
    /// view.
    pub fn is_running(&self) -> bool {
        self.handle.running.load(Ordering::Acquire)
    }

    /// True while the node has been manually deactivated via the control
    /// interface and not yet re-activated. Read by a task-state view and by the
    /// automatic bring-up paths (which skip a disabled node).
    pub fn is_disabled(&self) -> bool {
        self.handle.disabled.load(Ordering::Acquire)
    }

    /// Mark/clear this node as **detached**: a node that manages its own lifecycle
    /// and must not be torn down by a dependency cascade (see the field doc). Set it
    /// before stopping a dependency the node has declared but intends to outlive.
    pub fn set_detached(&self, detached: bool) {
        self.handle.detached.store(detached, Ordering::Release);
    }

    /// True while this node is detached from dependency-cascade teardown.
    pub fn is_detached(&self) -> bool {
        self.handle.detached.load(Ordering::Acquire)
    }

    // ── Supervisor-side API ──────────────────────────────────────────────
    //
    // Driven by the `Supervisor` struct. Kept `pub(crate)` so app code doesn't
    // accidentally bypass the supervisor's orchestration.

    pub(crate) fn signal_shutdown(&self) {
        self.handle.shutdown.store(true, Ordering::Release);
        self.handle.shutdown_wake.signal(());
    }

    pub(crate) fn signal_resume(&self) {
        self.handle.resume_wake.signal(());
    }

    pub(crate) fn set_running(&self, running: bool) {
        self.handle.running.store(running, Ordering::Release);
    }

    /// Set/clear the manual-deactivation flag. Set by `Supervisor::deactivate`,
    /// cleared by `Supervisor::activate`. Deliberately *not* touched by
    /// `reset()`, so a manual stop survives respawn cycles and RAM-retaining
    /// power-state transitions.
    ///
    /// Public so an application can pre-disable a `Terminate` node *before*
    /// `Supervisor::start`, making it a stopped-at-boot task that only comes up on
    /// an explicit `Activate` control (a node started by control rather than at boot).
    pub fn set_disabled(&self, disabled: bool) {
        self.handle.disabled.store(disabled, Ordering::Release);
    }

    /// Wait until the instance has called `ack_dropped()`. Single-instance, so
    /// one ack ends the wait. The fast-path flag check handles the ack landing
    /// before this await (the `dropped_wake` signal is edge-triggered).
    pub(crate) async fn wait_dropped(&self) {
        if self.handle.dropped.load(Ordering::Acquire) {
            return;
        }
        self.handle.dropped_wake.wait().await;
    }

    /// Clear the shutdown flag, dropped flag, busy flag, and the shutdown /
    /// dropped wake-signals so the next cycle starts clean. Doesn't touch
    /// `running` (managed around spawn/stop), `resume_wake` (`resume_pausable`
    /// fires that for Pause nodes), or `disabled` (lifecycle-spanning).
    pub(crate) fn reset(&self) {
        self.handle.shutdown.store(false, Ordering::Release);
        self.handle.dropped.store(false, Ordering::Release);
        self.handle.busy.store(false, Ordering::Release);
        self.handle.shutdown_wake.reset();
        self.handle.dropped_wake.reset();
    }
}

/// Manual impl: the private `TaskHandle` (Signals + atomics) has no `Debug`, and a
/// snapshot of the *live* flags is more useful than raw handle internals anyway.
/// `finish_non_exhaustive` marks the elided fields (`spawn`, the handle).
impl core::fmt::Debug for TaskNode {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("TaskNode")
            .field("name", &self.name)
            .field("mode", &self.mode)
            .field("running", &self.is_running())
            .field("busy", &self.is_busy())
            .field("disabled", &self.is_disabled())
            .field("detached", &self.is_detached())
            .finish_non_exhaustive()
    }
}

// ─── Graph ───────────────────────────────────────────────────────────────

/// The compile-time task graph produced by [`supervisor_graph!`]: the node slots,
/// the dependency-index table, the topological order, and the elastic pools — the
/// single value [`Supervisor::new`] consumes. The macro emits one `pub static GRAPH`
/// of this type. The fields are public so the application can read them directly
/// (e.g. a status endpoint iterating `GRAPH.nodes` / `GRAPH.deps`).
///
/// `N` is capped at 256 (graph indices are `u8`); the macro enforces this at
/// expansion time.
pub struct Graph<const N: usize> {
    /// Node slots, one per declared node. `None` marks a `#[cfg]`-ed-out node.
    pub nodes: &'static [Option<&'static TaskNode>; N],
    /// Per-node dependency indices into `nodes` (`deps[i]` lists node `i`'s deps).
    pub deps: &'static [&'static [u8]; N],
    /// Topologically sorted indices into `nodes` (dependencies before dependents;
    /// reverse iteration is the teardown order). A dependency cycle is a compile error.
    pub order: [u8; N],
    /// Elastic worker pools to register with the supervisor (empty when unused).
    #[cfg(feature = "pool")]
    pub pools: &'static [&'static dyn Pool],
}

// ─── Supervisor ──────────────────────────────────────────────────────────

/// Orchestrates a set of managed tasks across spawn / teardown / bring-up.
///
/// Owned by a single supervisor task. Concurrent access from other tasks goes
/// through each [`TaskNode`]'s own atomic state, not the `Supervisor` struct.
pub struct Supervisor<const N: usize> {
    /// Node slots, one per declared node. `None` marks a slot whose node was
    /// `#[cfg]`-ed out of the build (feature-gated); every method skips those.
    nodes: &'static [Option<&'static TaskNode>],
    /// Per-node dependency indices into `nodes` (`deps[i]` lists the indices of
    /// the nodes that node `i` depends on). The single runtime source of graph
    /// topology, generated alongside `order` by the `supervisor_graph!` macro.
    deps: &'static [&'static [u8]],
    /// Topologically sorted indices into `nodes`: dependencies before their
    /// dependents; reverse iteration is the teardown order. Precomputed at
    /// compile time (a cycle is a compile error), so construction does no work.
    order: [u8; N],
    /// Elastic pools, so the control interface can co-control a whole pool from
    /// any one member (`apply_control` expands the target through
    /// [`Pool::members`]) — the same registry `run_pools` drives. Taken from
    /// `GRAPH.pools` at construction (empty when no pool is declared).
    #[cfg(feature = "pool")]
    pools: &'static [&'static dyn Pool],
}

impl<const N: usize> Supervisor<N> {
    /// Build a supervisor from a precomputed [`Graph`] — the `GRAPH` that
    /// `supervisor_graph!` emits (node slots, dependency-index table, compile-time
    /// topological `order`, and the elastic pools). A dependency cycle is a
    /// *compile* error, so construction is infallible and does no work —
    /// `start` / `teardown` / `respawn_terminate` just iterate.
    pub const fn new(graph: &'static Graph<N>) -> Self {
        Self {
            nodes: graph.nodes,
            deps: graph.deps,
            order: graph.order,
            #[cfg(feature = "pool")]
            pools: graph.pools,
        }
    }

    /// Spawn every boot node in dependency order. Called once at boot.
    /// `Mode::OnDemand` nodes are skipped — they're brought up at runtime by
    /// `start_node`. A **parked** node (no `spawn` fn) is spawned externally by
    /// `main()` (with hardware handles main owns); it's still marked `running`
    /// here. Disabled nodes, and `#[cfg]`-ed-out slots, are skipped.
    pub fn start(&self, spawner: Spawner) -> Result<(), SpawnError> {
        for i in self.order.iter() {
            let Some(node) = self.nodes[*i as usize] else {
                continue;
            };
            if matches!(node.mode, Mode::OnDemand) || node.is_disabled() {
                continue;
            }
            info!("supervisor: spawning {} ({})", node.name, node.mode);
            if let Some(spawn) = node.spawn {
                spawn(spawner)?;
            }
            node.set_running(true);
        }
        Ok(())
    }

    /// Start a single node at runtime — e.g. growing an elastic pool. Resets the
    /// handle, spawns one instance via the node's `spawn` fn (which must launch
    /// exactly one), and marks it `running`. Returns `SpawnError::Busy` if the
    /// underlying embassy task pool is exhausted (the ceiling), which the caller
    /// treats as "can't grow".
    pub fn start_node(&self, node: &'static TaskNode, spawner: Spawner) -> Result<(), SpawnError> {
        node.reset();
        if let Some(spawn) = node.spawn {
            spawn(spawner)?;
        }
        node.set_running(true);
        info!("supervisor: started {}", node.name);
        Ok(())
    }

    /// Signal `node` to shut down, wait for its ack (panicking on timeout — a
    /// missing `ack_dropped()` somewhere), then clear `running`. Shared by
    /// `stop_node` and `teardown`; the caller must have checked `is_running`.
    async fn shutdown_and_wait(&self, node: &'static TaskNode) {
        node.signal_shutdown();
        if let Either::Second(()) = select(
            node.wait_dropped(),
            Timer::after_millis(SHUTDOWN_ACK_TIMEOUT_MS),
        )
        .await
        {
            panic!(
                "supervisor: task {} did not ack shutdown within {}ms",
                node.name, SHUTDOWN_ACK_TIMEOUT_MS,
            );
        }
        node.set_running(false);
    }

    /// Stop a single running node at runtime — e.g. shrinking an elastic pool.
    /// Signals shutdown, waits for the ack, clears `running`. No-op if the node
    /// isn't running. Panics if it doesn't ack within the timeout.
    pub async fn stop_node(&self, node: &'static TaskNode) {
        if !node.is_running() {
            return;
        }
        self.shutdown_and_wait(node).await;
        info!("supervisor: stopped {}", node.name);
    }

    /// Signal every **running** node to shut down in **reverse** topological
    /// order, awaiting each node's ack before moving to its dependency. Down
    /// `OnDemand` nodes are skipped (no instance to ack). Pause-mode nodes ack
    /// and park on `wait_resume()`; Terminate/OnDemand nodes exit. Panics if a
    /// running node fails to ack within `SHUTDOWN_ACK_TIMEOUT_MS`.
    pub async fn teardown(&self) {
        for i in self.order.iter().rev() {
            let Some(node) = self.nodes[*i as usize] else {
                continue;
            };
            if !node.is_running() {
                continue;
            }
            info!("supervisor: tearing down {}", node.name);
            self.shutdown_and_wait(node).await;
        }
    }

    /// Signal every Pause-mode node to resume. Cheap and synchronous — the tasks
    /// were parked on `wait_resume()` and pick up immediately. Called separately
    /// from `respawn_terminate` so the application can fire resume independently
    /// of the respawn step. Disabled (manually-paused) nodes are skipped so a
    /// manual pause sticks; there is intentionally no dependency gate here.
    pub fn resume_pausable(&self) {
        for i in self.order.iter() {
            let Some(node) = self.nodes[*i as usize] else {
                continue;
            };
            if matches!(node.mode, Mode::Pause) && !node.is_disabled() {
                node.reset();
                info!("supervisor: resuming {}", node.name);
                node.signal_resume();
                node.set_running(true);
            }
        }
    }

    /// Reset and re-spawn every Terminate-mode node in dependency order.
    /// Pause-mode nodes are untouched (use `resume_pausable`); `OnDemand` nodes
    /// are left down — they re-grow under load via `start_node`. Disabled nodes
    /// are skipped so a manual stop sticks across the bring-up. The reset happens
    /// before the spawn so newly-running tasks see a clean handle.
    pub fn respawn_terminate(&self, spawner: Spawner) -> Result<(), SpawnError> {
        for i in self.order.iter() {
            let Some(node) = self.nodes[*i as usize] else {
                continue;
            };
            if matches!(node.mode, Mode::Terminate) && !node.is_disabled() {
                node.reset();
                info!("supervisor: respawning {}", node.name);
                if let Some(spawn) = node.spawn {
                    spawn(spawner)?;
                }
                node.set_running(true);
            }
        }
        Ok(())
    }
}

// ─── Runtime control (dependency- and pool-honoring start/stop) ────────────
//
// The `apply_control` entry point drives one `ControlCommand` from the
// application's control surface. Unlike the pool's bare `start_node`/`stop_node`,
// these honor the graph: a stop cascades through dependents (so nothing is left
// running without a dependency), a start cascades through deps (so nothing comes
// up before what it needs), and either expands across a whole `ElasticPool` so
// the pool is controlled as a unit. A manual stop/pause also sets the
// lifecycle-spanning `disabled` flag, so it sticks against the elastic policy and
// the wake respawn.

#[cfg(feature = "control")]
impl<const N: usize> Supervisor<N> {
    /// Position of `node` in `self.nodes` (pointer identity — every node is a
    /// `&'static`). `None` only if the node isn't in this graph (impossible for
    /// targets sourced from `GRAPH.nodes`; treated as a no-op by callers).
    fn index_of(&self, node: &'static TaskNode) -> Option<usize> {
        self.nodes
            .iter()
            .position(|n| n.is_some_and(|x| core::ptr::eq(x, node)))
    }

    /// Whether every dependency of `node` is currently running, resolved through
    /// the graph's index table. The pool driver checks this before growing a
    /// worker, so a pool member is never spawned while one of its dependencies is
    /// down.
    #[cfg(feature = "pool")]
    pub(crate) fn deps_running(&self, node: &'static TaskNode) -> bool {
        match self.index_of(node) {
            Some(i) => self.deps[i]
                .iter()
                .all(|&di| self.nodes[di as usize].is_some_and(|n| n.is_running())),
            None => false,
        }
    }

    /// Seed a membership set with `target` plus — if `target` belongs to an
    /// elastic pool — every member of that pool, so control is applied to the
    /// whole pool atomically. Pool membership is read from `GRAPH.pools`; with no
    /// pools (the `pool` feature off, or none declared) this is just `{target}`.
    fn seed(&self, target: &'static TaskNode, set: &mut [bool; N]) {
        if let Some(i) = self.index_of(target) {
            set[i] = true;
        }
        #[cfg(feature = "pool")]
        for pool in self.pools {
            let members = pool.members();
            if members.iter().any(|m| core::ptr::eq(*m, target)) {
                for m in members {
                    if let Some(i) = self.index_of(m) {
                        set[i] = true;
                    }
                }
            }
        }
    }

    /// Apply one control command, honoring pool membership and the dependency
    /// graph. Run from the supervisor's driver loop (never concurrently with
    /// itself), so the cascade is atomic from the application's perspective.
    pub async fn apply_control(&self, cmd: ControlCommand, spawner: Spawner) {
        match cmd.op {
            ControlOp::Deactivate => self.deactivate(cmd.node).await,
            ControlOp::Activate => self.activate(cmd.node, spawner).await,
        }
    }

    /// Bring `target` (and its pool, and every transitive dependent) down, in
    /// reverse-topological order so each dependent stops before the dependency it
    /// relies on. Marks the whole set `disabled` so the stop sticks against the
    /// elastic policy and the wake respawn until a matching `activate`.
    async fn deactivate(&self, target: &'static TaskNode) {
        let mut set = [false; N];
        self.seed(target, &mut set);

        // Grow the set to include transitive dependents. `order` is
        // dependency-first, so when we reach a node its deps are already decided;
        // a node joins if any dep it declares is already in the set.
        for i in self.order.iter() {
            let j = *i as usize;
            if set[j] {
                continue;
            }
            let Some(node) = self.nodes[j] else {
                continue;
            };
            // A detached node declares its dep only for start ordering and intends
            // to outlive it, so it's never pulled into the cascade.
            if node.is_detached() {
                continue;
            }
            if self.deps[j].iter().any(|&di| set[di as usize]) {
                set[j] = true;
            }
        }

        // Tear down in reverse topo order (dependents before their deps).
        for i in self.order.iter().rev() {
            let j = *i as usize;
            if !set[j] {
                continue;
            }
            let Some(node) = self.nodes[j] else {
                continue;
            };
            node.set_disabled(true);
            if node.is_running() {
                info!("supervisor: control-stop {}", node.name);
                self.shutdown_and_wait(node).await;
            }
        }
    }

    /// Bring `target` (and its pool, and every transitive dependency) up, in
    /// topological order so each dependency starts before its dependent. Clears
    /// `disabled` across the set. `OnDemand` (pool) members are only re-enabled,
    /// not force-spawned — the elastic policy re-grows them under load, which is
    /// the whole point of the pool.
    async fn activate(&self, target: &'static TaskNode, spawner: Spawner) {
        let mut set = [false; N];
        self.seed(target, &mut set);

        // Grow the set to include transitive deps. Walk dependents-first
        // (reverse topo); when a set member is seen, pull in its direct deps.
        for i in self.order.iter().rev() {
            let j = *i as usize;
            if set[j] {
                for &di in self.deps[j] {
                    set[di as usize] = true;
                }
            }
        }

        // Bring up in topo order (deps before dependents).
        for i in self.order.iter() {
            let j = *i as usize;
            if !set[j] {
                continue;
            }
            let Some(node) = self.nodes[j] else {
                continue;
            };
            node.set_disabled(false);
            if node.is_running() {
                continue;
            }
            match node.mode {
                Mode::Terminate => {
                    info!("supervisor: control-start {}", node.name);
                    // SpawnError::Busy (pool exhausted) → can't start, skip.
                    let _ = self.start_node(node, spawner);
                }
                Mode::Pause => {
                    info!("supervisor: control-resume {}", node.name);
                    node.reset();
                    node.signal_resume();
                    node.set_running(true);
                }
                // Pool worker — leave it down; the elastic policy regrows it on
                // demand now that `disabled` is cleared.
                Mode::OnDemand => {}
            }
        }
    }
}

// ─── Topological sort (Kahn's algorithm, const) ───────────────────────────
//
// Computes the topological order at *compile time* over a per-node
// dependency-index table; a dependency cycle is a compile error.

/// Topologically sort a graph given as a per-node dependency-index table.
///
/// `deps[i]` lists the indices of the nodes that node `i` depends on; the result
/// lists node indices in dependency-first order (a dependency appears before its
/// dependents). The supervisor iterates it forward for `start` /
/// `respawn_terminate` and in reverse for `teardown`.
///
/// Evaluated at compile time by the code `supervisor_graph!` generates — a
/// dependency **cycle is a compile error** (the `panic!` fires during const
/// evaluation). `#[doc(hidden)]`: an engine for the macro, not a user-facing API.
///
/// Supports at most 256 nodes: indices are `u8`, so a larger `N` would truncate.
/// The macro rejects bigger graphs at expansion; the assert below is defense in
/// depth for a manual caller (a const-eval panic, i.e. a compile error).
#[doc(hidden)]
#[must_use]
pub const fn topo_sort_const<const N: usize>(deps: &[&'static [u8]; N]) -> [u8; N] {
    assert!(
        N <= 256,
        "supervisor graph exceeds 256 node slots (indices are u8)"
    );
    // in_degree[i] = number of deps of node i not yet resolved.
    let mut in_degree = [0u8; N];
    let mut i = 0;
    while i < N {
        in_degree[i] = deps[i].len() as u8;
        i += 1;
    }

    // Queue (fixed array, head/tail indices) seeded with the dependency-free nodes.
    let mut queue = [0u8; N];
    let mut tail = 0;
    i = 0;
    while i < N {
        if in_degree[i] == 0 {
            queue[tail] = i as u8;
            tail += 1;
        }
        i += 1;
    }

    let mut order = [0u8; N];
    let mut produced = 0;
    let mut head = 0;
    while head < tail {
        let node = queue[head] as usize;
        head += 1;
        order[produced] = node as u8;
        produced += 1;

        // Decrement the in-degree of every node that depends on `node`.
        let mut j = 0;
        while j < N {
            if in_degree[j] != 0 {
                let mut depends = false;
                let mut k = 0;
                while k < deps[j].len() {
                    if deps[j][k] as usize == node {
                        depends = true;
                    }
                    k += 1;
                }
                if depends {
                    in_degree[j] -= 1;
                    if in_degree[j] == 0 {
                        queue[tail] = j as u8;
                        tail += 1;
                    }
                }
            }
            j += 1;
        }
    }

    // A cycle leaves some nodes unproduced. During const eval this panic is a
    // compile error, so cyclic graphs are rejected at build time. `core::panic!`
    // (not the crate's defmt-shimmed `panic!`) keeps this const-evaluable.
    if produced != N {
        core::panic!("supervisor_graph!: dependency cycle");
    }
    order
}

#[cfg(feature = "pool")]
mod pool;
#[cfg(feature = "pool")]
pub use pool::*;

/// Declare a supervised task graph and compute its topological order at compile
/// time (single source of nodes, deps, pool, and order). See the
/// `embassy-supervisor-macros` crate for the surface syntax.
#[cfg(feature = "macros")]
pub use embassy_supervisor_macros::supervisor_graph;

// ─── Tests (host-only) ─────────────────────────────────────────────────────
//
// Run on the host: `cargo test -p embassy-supervisor --target x86_64-unknown-linux-gnu`
// (the workspace `.cargo/config.toml` pins the embedded target, so `--target` is
// required to override it). These exercise the compile-time `topo_sort_const`
// over index adjacency tables — exactly what `supervisor_graph!` generates.
#[cfg(test)]
mod tests {
    use super::topo_sort_const;

    /// Position of index `x` within `order`.
    fn pos<const N: usize>(order: &[u8; N], x: u8) -> usize {
        order.iter().position(|&y| y == x).expect("index present")
    }

    #[test]
    fn linear_chain_orders_deps_before_dependents() {
        // A=0, B=1 dep A, C=2 dep B.
        const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
        const ORDER: [u8; 3] = topo_sort_const(&DEPS);
        assert_eq!(ORDER, [0, 1, 2]);
    }

    #[test]
    fn diamond_puts_root_first_and_join_last() {
        // A=0; B=1 dep A; C=2 dep A; D=3 dep B,C.
        const DEPS: [&[u8]; 4] = [&[], &[0], &[0], &[1, 2]];
        const ORDER: [u8; 4] = topo_sort_const(&DEPS);
        assert_eq!(ORDER[0], 0, "root first");
        assert_eq!(ORDER[3], 3, "join last");
        assert!(pos(&ORDER, 1) < pos(&ORDER, 3), "B before D");
        assert!(pos(&ORDER, 2) < pos(&ORDER, 3), "C before D");
    }

    #[test]
    fn independent_nodes_all_present() {
        const DEPS: [&[u8]; 2] = [&[], &[]];
        const ORDER: [u8; 2] = topo_sort_const(&DEPS);
        assert!(ORDER.contains(&0) && ORDER.contains(&1));
    }

    #[test]
    fn unsorted_input_is_sorted() {
        // Declared out of dependency order: node 0 depends on 1 and 2, node 2 on 1.
        const DEPS: [&[u8]; 3] = [&[1, 2], &[], &[1]];
        const ORDER: [u8; 3] = topo_sort_const(&DEPS);
        assert!(pos(&ORDER, 1) < pos(&ORDER, 2), "1 before 2");
        assert!(pos(&ORDER, 2) < pos(&ORDER, 0), "2 before 0");
    }

    #[test]
    fn evaluates_at_compile_time() {
        // The sort runs in a `const` context, proving it is const-evaluable.
        // (A cyclic table here would be a *compile* error, not a test failure.)
        const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
        const _: () = {
            let order = topo_sort_const(&DEPS);
            assert!(order[0] == 0 && order[1] == 1 && order[2] == 2);
        };
    }

    // Uncommenting this must fail to compile ("dependency cycle"):
    //   const CYCLE: [&[u8]; 2] = [&[1], &[0]];
    //   const _BAD: [u8; 2] = topo_sort_const(&CYCLE);
}