taktora_executor/

executor.rs

//! `Executor` and `ExecutorBuilder`. Run loop lives in Task 8.

// Fields consumed by the run loop (Task 8) and graph scheduler (Task 14).
#![allow(dead_code)]
// pub(crate) inside a private module — intentional, Task 8+ will use them.
#![allow(clippy::redundant_pub_crate)]

use crate::Channel;
use crate::context::Stoppable;
use crate::error::ExecutorError;
use crate::fault::{
    ExecutorFaultAtomic, ExecutorFaultReason, ExecutorFaultState, FaultAtomic, FaultReason,
    FaultState, duration_to_ms_sat, instant_to_since_ms,
};
use crate::item::ExecutableItem;
use crate::monitor::{ExecutionMonitor, NoopMonitor};
use crate::observer::{NoopObserver, Observer};
use crate::payload::Payload;
use crate::pool::Pool;
use crate::task_id::TaskId;
use crate::task_kind::TaskKind;
use crate::thread_attrs::ThreadAttributes;
use crate::trigger::{TriggerDecl, TriggerDeclarer};
use core::sync::atomic::AtomicU32;
use iceoryx2::node::Node;
use iceoryx2::port::listener::Listener as IxListener;
use iceoryx2::prelude::ipc;
use iceoryx2::prelude::*;
use iceoryx2::waitset::WaitSetRunResult;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::time::{Duration, Instant};

/// Monotonically increasing counter so multiple executors in the same process
/// each get a unique stop-event service name.
static EXEC_COUNTER: AtomicU64 = AtomicU64::new(0);

/// One registered task entry.
pub(crate) struct TaskEntry {
    /// Task identifier.
    pub(crate) id: TaskId,
    /// The kind of work this entry holds (single item or chain).
    pub(crate) kind: TaskKind,
    /// Trigger declarations recorded at `add` time.
    pub(crate) decls: Vec<TriggerDecl>,
    /// Pre-allocated dispatch closure. Built once at `add` / `add_chain`
    /// time and re-invoked on every dispatch iteration via
    /// `Pool::submit_borrowed`, avoiding the per-iteration `Box::new(closure)`
    /// that `Pool::submit<F>` requires in threaded mode. Required for
    /// `REQ_0060` (zero-alloc steady-state dispatch). `None` for
    /// `TaskKind::Graph`, which dispatches its vertices via a separate
    /// path and is handled by `REQ_0062` / `REQ_0063` follow-on work.
    pub(crate) job: Option<Box<dyn FnMut() + Send + 'static>>,

    /// Per-task budget declared via `TriggerDeclarer::budget`. `None`
    /// means no per-task check; the executor-wide iteration budget
    /// still applies. `REQ_0070`.
    pub(crate) budget: Option<Duration>,

    /// Per-task fault state. Wait-free read on the dispatch hot path.
    /// Wrapped in `Arc` so dispatch closures built at `add` time can
    /// capture an owning handle into the same atomic the `TaskEntry`
    /// holds — `Arc::clone` is refcount-only, so this stays compatible
    /// with `REQ_0060` (no per-iteration allocation). `REQ_0070`.
    pub(crate) fault: Arc<FaultAtomic>,

    /// Monotonic per-task overrun counter. Increments on EVERY budget
    /// breach, including breaches while already `Faulted`. Never reset
    /// by clearing the fault. Shared with the dispatch closure via
    /// `Arc::clone`. `REQ_0102`.
    pub(crate) overrun_count: Arc<AtomicU64>,

    /// Pre-built dispatch closure for the fault-handler item. Mirrors
    /// `job`. `None` means no handler — the task is simply skipped
    /// during fault. `REQ_0072`.
    pub(crate) handler_job: Option<Box<dyn FnMut() + Send + 'static>>,
}

/// Top-level executor. One per process is the typical case.
pub struct Executor {
    pub(crate) node: Node<ipc::Service>,
    pub(crate) pool: Arc<Pool>,
    pub(crate) tasks: Vec<TaskEntry>,
    pub(crate) running: Arc<AtomicBool>,
    pub(crate) stoppable: Stoppable,
    pub(crate) next_id: AtomicU64,
    /// Listener for the internal stop event service. Held here so it outlives
    /// the `WaitSet` guard inside `dispatch_loop`. Created at `build()` time so
    /// any `Stoppable` clone (taken before or after `run()`) carries the waker.
    pub(crate) stop_listener: Arc<IxListener<ipc::Service>>,
    /// Lifecycle observer. Defaults to a no-op.
    pub(crate) observer: Arc<dyn Observer>,
    /// Execution monitor. Defaults to a no-op.
    pub(crate) monitor: Arc<dyn ExecutionMonitor>,
    /// Per-iteration error capture slot — allocated once at build time and
    /// reset to `None` at the top of each `dispatch_loop` iteration. Pool
    /// workers obtain a refcount-only `Arc::clone` of this slot, avoiding
    /// the per-iteration heap allocation that the previous design incurred.
    /// Required for `REQ_0060`.
    pub(crate) iter_err: Arc<std::sync::Mutex<Option<ExecutorError>>>,
    /// Executor-wide iteration budget from `ExecutorBuilder::iteration_budget`.
    /// `None` means no executor-wide check.
    pub(crate) iteration_budget: Option<Duration>,
    /// Executor-wide fault state. Wrapped in `Arc` so each dispatch
    /// closure can hold an owning handle without re-borrowing through
    /// `self`. `REQ_0071`.
    pub(crate) exec_fault: Arc<ExecutorFaultAtomic>,

    /// Index of the task whose `execute()` overran when the executor
    /// transitioned to `Faulted`. Read alongside `exec_fault`.
    pub(crate) exec_fault_task_idx: Arc<AtomicU32>,

    /// Budget that was breached when the executor transitioned to
    /// `Faulted`, in ms (saturated). Read alongside `exec_fault`.
    pub(crate) exec_fault_budget_ms: Arc<AtomicU32>,

    /// Executor start time, set on first dispatch. Used to compute
    /// `since_ms` for faults relative to `Executor::run` entry. Wrapped
    /// in `Arc` so dispatch closures share the same `OnceLock` with the
    /// executor — `get_or_init` is idempotent and wait-free.
    pub(crate) start_time: Arc<OnceLock<Instant>>,
}

// SAFETY: `IxListener<ipc::Service>` is `!Send` for the same Rc-based
// `SingleThreaded` reason as `IxNotifier`. After construction, the only
// per-iteration call is `listener.try_wait_one()`, which does not mutate the
// Rc. `Executor` is never shared across threads (it requires `&mut self` for
// `run()`), so there is no aliased concurrent mutation.
#[allow(unsafe_code, clippy::non_send_fields_in_send_ty)]
unsafe impl Send for Executor {}

impl Executor {
    /// Start a new builder.
    #[must_use]
    pub fn builder() -> ExecutorBuilder {
        ExecutorBuilder::default()
    }

    /// Open or create a pub/sub channel bound to this executor's node.
    pub fn channel<T: Payload>(&mut self, name: &str) -> Result<Arc<Channel<T>>, ExecutorError> {
        Channel::open_or_create(&self.node, name)
    }

    /// Open or create a request/response service bound to this executor's node.
    pub fn service<Req, Resp>(
        &mut self,
        name: &str,
    ) -> Result<Arc<crate::Service<Req, Resp>>, ExecutorError>
    where
        Req: Payload,
        Resp: Payload,
    {
        crate::Service::open_or_create(&self.node, name)
    }

    /// Add an item to the executor with an auto-generated id.
    pub fn add(&mut self, item: impl ExecutableItem) -> Result<TaskId, ExecutorError> {
        let id = TaskId::new(format!(
            "task-{}",
            self.next_id.fetch_add(1, Ordering::SeqCst)
        ));
        self.add_with_id(id, item)
    }

    /// Add an item with a user-supplied id.
    ///
    /// The item's [`ExecutableItem::task_id`] override takes precedence over
    /// the caller-supplied `id`, which itself takes precedence over the
    /// auto-generated id assigned by [`Executor::add`].
    pub fn add_with_id(
        &mut self,
        id: impl Into<TaskId>,
        mut item: impl ExecutableItem,
    ) -> Result<TaskId, ExecutorError> {
        let id_arg: TaskId = id.into();
        // The item's `task_id()` override wins over the user-supplied id.
        let id = item.task_id().map_or(id_arg, TaskId::new);
        let mut declarer = TriggerDeclarer::new_internal();
        item.declare_triggers(&mut declarer)?;
        let budget = declarer.budget;
        let decls = declarer.into_decls();

        let mut item_box: Box<dyn ExecutableItem> = Box::new(item);
        let app_id = item_box.app_id();
        let app_inst = item_box.app_instance_id();
        // SAFETY: the raw pointer points into the heap allocation of
        // `item_box`. `Box` keeps that allocation at a stable address even
        // when the `Box` itself is moved (e.g. when `self.tasks` grows),
        // so the pointer remains valid for the lifetime of the
        // `TaskEntry`. See SendItemPtr safety doc for the rest of the
        // discipline (barrier() pairs with worker access).
        #[allow(unsafe_code)]
        let item_ptr =
            SendItemPtr::new(std::ptr::from_mut::<dyn ExecutableItem>(item_box.as_mut()));

        // Allocate the per-task atomics now so the dispatch closure
        // and the `TaskEntry` share the same `Arc` storage. The task
        // will occupy `self.tasks.len()` after the push below — capture
        // that index up front for `task_idx_u32`. Bounded workspace, so
        // the `as u32` cast is sound; explicit allow keeps clippy quiet.
        let task_fault = Arc::new(FaultAtomic::new());
        let overrun_count = Arc::new(AtomicU64::new(0));
        #[allow(clippy::cast_possible_truncation)]
        let task_idx_u32 = self.tasks.len() as u32;
        let fault_ctx = FaultDispatchCtx {
            task_budget: budget,
            task_fault: Arc::clone(&task_fault),
            overrun_count: Arc::clone(&overrun_count),
            iteration_budget: self.iteration_budget,
            exec_fault: Arc::clone(&self.exec_fault),
            exec_fault_task_idx: Arc::clone(&self.exec_fault_task_idx),
            exec_fault_budget_ms: Arc::clone(&self.exec_fault_budget_ms),
            task_idx_u32,
            exec_start: Arc::clone(&self.start_time),
            observer: Arc::clone(&self.observer),
        };

        let job = build_single_job(
            id.clone(),
            self.stoppable.clone(),
            Arc::clone(&self.observer),
            Arc::clone(&self.monitor),
            Arc::clone(&self.iter_err),
            app_id,
            app_inst,
            item_ptr,
            fault_ctx,
        );

        self.tasks.push(TaskEntry {
            id: id.clone(),
            kind: TaskKind::Single(item_box),
            decls,
            job: Some(job),
            budget,
            fault: task_fault,
            overrun_count,
            handler_job: None,
        });
        Ok(id)
    }

    /// Register an item plus a fault-handler item.
    ///
    /// The main item is registered through the canonical [`add`](Self::add)
    /// path. The handler's [`declare_triggers`](ExecutableItem::declare_triggers)
    /// is called (so handlers that internally rely on the declarer being
    /// invoked observe the call) but its returned trigger list is
    /// **ignored** — the handler dispatches on the main item's triggers
    /// while the task is in `Faulted` state and runs in place of the main
    /// item's `execute()`. The pre-built handler dispatch closure is
    /// stashed on the same task entry as the main item's `job`,
    /// satisfying `REQ_0072`.
    ///
    /// # Errors
    ///
    /// Propagates any error from registering the main item via `add`, or
    /// from the handler's `declare_triggers` call.
    ///
    /// # Panics
    ///
    /// Panics if the task entry just inserted by [`add`](Self::add) cannot
    /// be located in `self.tasks` — this is unreachable by construction
    /// and indicates a logic bug.
    pub fn add_with_fault_handler<I, H>(
        &mut self,
        main: I,
        handler: H,
    ) -> Result<TaskId, ExecutorError>
    where
        I: ExecutableItem,
        H: ExecutableItem,
    {
        let task_id = self.add(main)?;

        // Drain the handler's trigger declarations — they are ignored by
        // design (the handler runs on the main item's triggers).
        let mut handler_box: Box<dyn ExecutableItem> = Box::new(handler);
        let mut throwaway = TriggerDeclarer::new_internal();
        handler_box.declare_triggers(&mut throwaway)?;
        drop(throwaway);

        let app_id = handler_box.app_id();
        let app_inst = handler_box.app_instance_id();

        // Locate the task we just added so we can share its per-task
        // atomics with the handler's `FaultDispatchCtx`. The handler
        // runs on the same `TaskEntry`; per §4.6 invariant 5, a handler
        // breach increments `overrun_count` and keeps state `Faulted`
        // without re-firing the observer.
        let task_idx = self
            .tasks
            .iter()
            .position(|t| t.id == task_id)
            .expect("just added; must exist");
        let task = &self.tasks[task_idx];
        #[allow(clippy::cast_possible_truncation)]
        let task_idx_u32 = task_idx as u32;
        let handler_fault_ctx = FaultDispatchCtx {
            task_budget: task.budget,
            task_fault: Arc::clone(&task.fault),
            overrun_count: Arc::clone(&task.overrun_count),
            iteration_budget: self.iteration_budget,
            exec_fault: Arc::clone(&self.exec_fault),
            exec_fault_task_idx: Arc::clone(&self.exec_fault_task_idx),
            exec_fault_budget_ms: Arc::clone(&self.exec_fault_budget_ms),
            task_idx_u32,
            exec_start: Arc::clone(&self.start_time),
            observer: Arc::clone(&self.observer),
        };

        let handler_closure = build_handler_job(
            task_id.clone(),
            self.stoppable.clone(),
            Arc::clone(&self.observer),
            Arc::clone(&self.monitor),
            Arc::clone(&self.iter_err),
            app_id,
            app_inst,
            handler_box,
            handler_fault_ctx,
        );

        self.tasks[task_idx].handler_job = Some(handler_closure);

        Ok(task_id)
    }

    /// Clear a per-task fault. Returns the previous `FaultState`.
    /// Fires `Observer::on_task_clear` if the state changed from
    /// `Faulted` to `Running`. `REQ_0070`.
    ///
    /// # Errors
    ///
    /// * [`ExecutorError::TaskNotFound`] if `task` is unknown.
    /// * [`ExecutorError::TaskNotFaulted`] if `task` is already `Running`.
    pub fn clear_task_fault(&self, task: TaskId) -> Result<FaultState, ExecutorError> {
        let entry = self
            .tasks
            .iter()
            .find(|t| t.id == task)
            .ok_or_else(|| ExecutorError::TaskNotFound(task.clone()))?;
        let budget_ms = entry.budget.map_or(0_u32, crate::fault::duration_to_ms_sat);
        let prev = entry.fault.swap(FaultState::Running, budget_ms);
        match prev {
            FaultState::Running => Err(ExecutorError::TaskNotFaulted(task)),
            FaultState::Faulted { .. } => {
                self.observer.on_task_clear(task);
                Ok(prev)
            }
        }
    }

    /// Clear the executor-wide fault and cascade-clear every task whose
    /// state is `Faulted{ExecutorFaulted}`. Tasks whose state is
    /// `Faulted{BudgetExceeded}` are NOT cleared (their own contract
    /// breach is independent). Fires `Observer::on_executor_clear` and
    /// one `Observer::on_task_clear` per cascade-cleared task.
    /// `REQ_0071`.
    ///
    /// # Errors
    ///
    /// * [`ExecutorError::ExecutorNotFaulted`] if the executor is `Running`.
    pub fn clear_executor_fault(&self) -> Result<ExecutorFaultState, ExecutorError> {
        let task_idx = self.exec_fault_task_idx.load(Ordering::Acquire);
        let budget_ms = self.exec_fault_budget_ms.load(Ordering::Acquire);
        let prev = self
            .exec_fault
            .swap(ExecutorFaultState::Running, task_idx, budget_ms);
        match prev {
            ExecutorFaultState::Running => Err(ExecutorError::ExecutorNotFaulted),
            ExecutorFaultState::Faulted { .. } => {
                // Cascade-clear tasks whose reason is ExecutorFaulted.
                for entry in &self.tasks {
                    let task_budget_ms =
                        entry.budget.map_or(0_u32, crate::fault::duration_to_ms_sat);
                    if let FaultState::Faulted {
                        reason: FaultReason::ExecutorFaulted,
                        ..
                    } = entry.fault.load(task_budget_ms)
                    {
                        let _ = entry.fault.swap(FaultState::Running, task_budget_ms);
                        self.observer.on_task_clear(entry.id.clone());
                    }
                }
                self.observer.on_executor_clear();
                Ok(prev)
            }
        }
    }

    /// Return the per-task overrun counter — number of times the task's
    /// `execute()` exceeded its budget over the executor's lifetime.
    /// Monotonic; not reset by `clear_task_fault`. `REQ_0102`.
    ///
    /// # Errors
    ///
    /// * [`ExecutorError::TaskNotFound`] if `task` is unknown.
    pub fn overrun_count(&self, task: TaskId) -> Result<u64, ExecutorError> {
        self.tasks
            .iter()
            .find(|t| t.id == task)
            .map(|t| t.overrun_count.load(Ordering::Acquire))
            .ok_or_else(|| ExecutorError::TaskNotFound(task))
    }

    /// Return a snapshot of the per-task `FaultState`. `REQ_0073` (pull path).
    ///
    /// # Errors
    ///
    /// * [`ExecutorError::TaskNotFound`] if `task` is unknown.
    pub fn task_fault_state(&self, task: TaskId) -> Result<FaultState, ExecutorError> {
        self.tasks
            .iter()
            .find(|t| t.id == task)
            .map(|t| {
                let budget_ms = t.budget.map_or(0_u32, crate::fault::duration_to_ms_sat);
                t.fault.load(budget_ms)
            })
            .ok_or_else(|| ExecutorError::TaskNotFound(task))
    }

    /// Return a snapshot of the executor-wide `ExecutorFaultState`.
    /// `REQ_0073` (pull path).
    #[must_use]
    pub fn executor_fault_state(&self) -> ExecutorFaultState {
        let task_idx = self.exec_fault_task_idx.load(Ordering::Acquire);
        let budget_ms = self.exec_fault_budget_ms.load(Ordering::Acquire);
        self.exec_fault.load(task_idx, budget_ms)
    }

    /// Add a sequential chain of items. Only the head item's
    /// `declare_triggers` is consulted; non-head triggers are ignored with a
    /// tracing warn.
    pub fn add_chain<I, C>(&mut self, items: C) -> Result<TaskId, ExecutorError>
    where
        I: ExecutableItem,
        C: IntoIterator<Item = I>,
    {
        let id = TaskId::new(format!(
            "chain-{}",
            self.next_id.fetch_add(1, Ordering::SeqCst)
        ));
        let boxed: Vec<Box<dyn ExecutableItem>> = items
            .into_iter()
            .map(|i| Box::new(i) as Box<dyn ExecutableItem>)
            .collect();
        self.add_chain_with_id_boxed(id, boxed)
    }

    /// Like [`Executor::add_chain`] but with a user-supplied id.
    pub fn add_chain_with_id<I, C>(
        &mut self,
        id: impl Into<TaskId>,
        items: C,
    ) -> Result<TaskId, ExecutorError>
    where
        I: ExecutableItem,
        C: IntoIterator<Item = I>,
    {
        let boxed: Vec<Box<dyn ExecutableItem>> = items
            .into_iter()
            .map(|i| Box::new(i) as Box<dyn ExecutableItem>)
            .collect();
        self.add_chain_with_id_boxed(id.into(), boxed)
    }

    fn add_chain_with_id_boxed(
        &mut self,
        id: TaskId,
        mut items: Vec<Box<dyn ExecutableItem>>,
    ) -> Result<TaskId, ExecutorError> {
        if items.is_empty() {
            return Err(ExecutorError::Builder(
                "chain must contain at least one item".into(),
            ));
        }

        // Head item's `task_id()` override wins over the user-supplied id.
        let id = items[0].task_id().map_or(id, TaskId::new);

        // Head's triggers gate the chain.
        let mut head_declarer = TriggerDeclarer::new_internal();
        items[0].declare_triggers(&mut head_declarer)?;
        let decls = head_declarer.into_decls();

        // Warn if non-head items declared triggers (those will be ignored).
        for (i, body) in items.iter_mut().enumerate().skip(1) {
            let mut spurious = TriggerDeclarer::new_internal();
            let _ = body.declare_triggers(&mut spurious);
            if !spurious.is_empty() {
                #[cfg(feature = "tracing")]
                tracing::warn!(
                    target: "taktora-executor",
                    task = %id,
                    position = i,
                    "non-head chain item declared triggers; they will be ignored"
                );
                #[cfg(not(feature = "tracing"))]
                {
                    let _ = i;
                }
            }
        }

        let mut items = items;
        // SAFETY: pointer into the chain's `items` Vec. The Vec lives
        // inside `TaskKind::Chain` inside `TaskEntry`. The Vec's buffer
        // is stable once `add_chain` returns — `self.tasks` may grow
        // (moving the `Vec<Box<...>>` header itself), but the Vec's
        // heap buffer is referenced via the header's data pointer and
        // is unaffected by header moves. We never resize the chain Vec
        // after this point. See SendChainPtr safety doc for the rest.
        #[allow(unsafe_code)]
        let chain_ptr = SendChainPtr::new(std::ptr::from_mut::<Vec<Box<dyn ExecutableItem>>>(
            &mut items,
        ));
        // NB: the pointer above is to the local `items` Vec on the
        // stack — it's invalid after the `push` below moves items into
        // the TaskEntry. We rederive a stable pointer after the push.
        // (See the rebuild step below.)
        let _ = chain_ptr;

        // Pre-allocate the per-task atomics so the chain's dispatch
        // closure can capture clones of the same `Arc`s the `TaskEntry`
        // holds. The chain occupies `self.tasks.len()` after the push.
        let task_fault = Arc::new(FaultAtomic::new());
        let overrun_count = Arc::new(AtomicU64::new(0));
        #[allow(clippy::cast_possible_truncation)]
        let task_idx_u32 = self.tasks.len() as u32;

        self.tasks.push(TaskEntry {
            id: id.clone(),
            kind: TaskKind::Chain(items),
            decls,
            job: None, // populated in the rebuild step below
            // TODO(post-Task-10): chain budgets carried separately; for now None.
            budget: None,
            fault: Arc::clone(&task_fault),
            overrun_count: Arc::clone(&overrun_count),
            handler_job: None,
        });

        // After the push, the TaskEntry lives at a stable position in
        // `self.tasks` for the duration of this `add_chain_with_id_boxed`
        // call. Take a stable pointer to its chain Vec and build the
        // dispatch closure. If `self.tasks` later grows, the Vec header
        // inside the TaskEntry moves but the header's data pointer
        // (which addresses the chain's heap buffer) does not — and the
        // closure derefs that pointer per dispatch, so it re-reads the
        // current heap address each time. Sound under the same
        // discipline as `tasks_ptr` in dispatch_loop.
        let task_idx = self.tasks.len() - 1;
        let chain_vec_ptr: *mut Vec<Box<dyn ExecutableItem>> = match &mut self.tasks[task_idx].kind
        {
            TaskKind::Chain(v) => std::ptr::from_mut::<Vec<Box<dyn ExecutableItem>>>(v),
            // The push above used TaskKind::Chain, so this arm is
            // unreachable. Mark it explicitly to satisfy `match`.
            _ => unreachable!("just-pushed task is TaskKind::Chain"),
        };
        #[allow(unsafe_code)]
        let chain_ptr = SendChainPtr::new(chain_vec_ptr);
        let fault_ctx = FaultDispatchCtx {
            task_budget: None, // chain budgets are intentionally None for now
            task_fault,
            overrun_count,
            iteration_budget: self.iteration_budget,
            exec_fault: Arc::clone(&self.exec_fault),
            exec_fault_task_idx: Arc::clone(&self.exec_fault_task_idx),
            exec_fault_budget_ms: Arc::clone(&self.exec_fault_budget_ms),
            task_idx_u32,
            exec_start: Arc::clone(&self.start_time),
            observer: Arc::clone(&self.observer),
        };
        let job = build_chain_job(
            id.clone(),
            self.stoppable.clone(),
            Arc::clone(&self.observer),
            Arc::clone(&self.monitor),
            Arc::clone(&self.iter_err),
            chain_ptr,
            fault_ctx,
        );
        self.tasks[task_idx].job = Some(job);
        Ok(id)
    }

    /// Returns a [`Stoppable`] handle that is waker-aware from the moment the
    /// executor is built. Clone before calling `run()` — any clone taken at any
    /// time will wake the `WaitSet` when `stop()` is called.
    #[must_use]
    pub fn stoppable(&self) -> Stoppable {
        self.stoppable.clone()
    }

    /// Borrow the underlying iceoryx2 node (escape hatch for power users).
    pub const fn iceoryx_node(&self) -> &Node<ipc::Service> {
        &self.node
    }

    /// Begin building a graph. Call `.build()` on the returned builder to
    /// register the graph as a task.
    pub fn add_graph(&mut self) -> ExecutorGraphBuilder<'_> {
        ExecutorGraphBuilder {
            executor: self,
            builder: crate::graph::GraphBuilder::new(),
            custom_id: None,
        }
    }
}

/// Builder for [`Executor`].
pub struct ExecutorBuilder {
    worker_threads: Option<usize>,
    observer: Option<Arc<dyn Observer>>,
    monitor: Option<Arc<dyn ExecutionMonitor>>,
    worker_attrs: ThreadAttributes,
    /// Executor-wide iteration budget (`REQ_0071`). `None` means no
    /// executor-wide check.
    iteration_budget: Option<Duration>,
}

impl Default for ExecutorBuilder {
    fn default() -> Self {
        Self {
            worker_threads: None,
            observer: None,
            monitor: None,
            worker_attrs: ThreadAttributes::new(),
            iteration_budget: None,
        }
    }
}

impl ExecutorBuilder {
    /// Number of worker threads. `0` → inline (no pool). Default → physical
    /// cores.
    #[must_use]
    pub const fn worker_threads(mut self, n: usize) -> Self {
        self.worker_threads = Some(n);
        self
    }

    /// Attach a lifecycle observer. If not called, a no-op observer is used.
    #[must_use]
    pub fn observer(mut self, obs: Arc<dyn Observer>) -> Self {
        self.observer = Some(obs);
        self
    }

    /// Attach an execution monitor. If not called, a no-op monitor is used.
    #[must_use]
    pub fn monitor(mut self, mon: Arc<dyn ExecutionMonitor>) -> Self {
        self.monitor = Some(mon);
        self
    }

    /// Configure the executor-wide iteration budget. Any task whose
    /// `execute()` exceeds `dur` transitions the executor to `Faulted`
    /// (`REQ_0071`). Default: unset (no executor-wide check).
    #[must_use]
    pub const fn iteration_budget(mut self, dur: Duration) -> Self {
        self.iteration_budget = Some(dur);
        self
    }

    /// Set thread attributes (name prefix, CPU affinity, scheduling priority)
    /// for worker threads. Has no effect when `worker_threads` is `0` (inline
    /// mode). Requires the `thread_attrs` feature for non-default settings.
    #[must_use]
    #[allow(clippy::missing_const_for_fn)]
    pub fn worker_attrs(mut self, attrs: ThreadAttributes) -> Self {
        self.worker_attrs = attrs;
        self
    }

    /// Build the [`Executor`]. Creates a fresh iceoryx2 node and wires up the
    /// internal stop-event service so that any `Stoppable` clone (taken before
    /// or after `run()`) will wake the `WaitSet` when `stop()` is called.
    ///
    /// # Panics
    ///
    /// Panics if the internally-generated stop-event service name exceeds the
    /// iceoryx2 service name length limit (this cannot happen under normal use
    /// because the name is derived from the process id and a monotonic counter).
    #[allow(clippy::arc_with_non_send_sync)] // see SAFETY on `impl Send for Executor`
    #[track_caller]
    pub fn build(self) -> Result<Executor, ExecutorError> {
        let node = NodeBuilder::new()
            .create::<ipc::Service>()
            .map_err(ExecutorError::iceoryx2)?;

        let n_workers = self.worker_threads.unwrap_or_else(num_cpus::get_physical);
        let pool = Arc::new(Pool::new(n_workers, self.worker_attrs)?);

        // Build the internal stop event service with a unique-per-process name
        // so multiple executors in the same process don't collide.
        let exec_seq = EXEC_COUNTER.fetch_add(1, Ordering::Relaxed);
        let stop_topic = format!(
            "taktora.exec.stop.{}.{exec_seq}.__taktora_event",
            std::process::id()
        );
        let stop_event = node
            .service_builder(&stop_topic.as_str().try_into().unwrap())
            .event()
            .open_or_create()
            .map_err(ExecutorError::iceoryx2)?;

        let stop_notifier = Arc::new(
            stop_event
                .notifier_builder()
                .create()
                .map_err(ExecutorError::iceoryx2)?,
        );

        // SAFETY: see module-level note; Arc<IxListener> is held here and only
        // accessed on the executor thread.
        let stop_listener = Arc::new(
            stop_event
                .listener_builder()
                .create()
                .map_err(ExecutorError::iceoryx2)?,
        );

        // Wire the notifier into the Stoppable so every clone is waker-aware
        // from the moment the executor is built.
        let stoppable = Stoppable::with_waker(stop_notifier);

        let observer: Arc<dyn Observer> = self.observer.unwrap_or_else(|| Arc::new(NoopObserver));

        let monitor: Arc<dyn ExecutionMonitor> =
            self.monitor.unwrap_or_else(|| Arc::new(NoopMonitor));

        let exec = Executor {
            node,
            pool,
            tasks: Vec::new(),
            running: Arc::new(AtomicBool::new(false)),
            stoppable,
            next_id: AtomicU64::new(0),
            stop_listener,
            observer,
            monitor,
            iter_err: Arc::new(std::sync::Mutex::new(None)),
            iteration_budget: self.iteration_budget,
            exec_fault: Arc::new(ExecutorFaultAtomic::new()),
            exec_fault_task_idx: Arc::new(AtomicU32::new(0)),
            exec_fault_budget_ms: Arc::new(AtomicU32::new(0)),
            start_time: Arc::new(OnceLock::new()),
        };

        Ok(exec)
    }
}

// ── Run loop ──────────────────────────────────────────────────────────────────

impl Executor {
    /// Run the executor until [`Stoppable::stop`] is called or a task signals
    /// stop via [`crate::Context::stop_executor`].
    ///
    /// # Errors
    ///
    /// Returns the **first** [`ExecutorError`] surfaced during dispatch:
    ///
    /// * [`ExecutorError::Item`] if any item returns `Err` or panics.
    /// * [`ExecutorError::Iceoryx2`] if a `WaitSet` operation fails.
    /// * [`ExecutorError::AlreadyRunning`] if the executor is already running.
    ///
    /// If multiple items error in the same dispatch iteration, only the first
    /// is preserved; subsequent errors are discarded silently. To observe
    /// every error, attach an [`Observer`](crate::Observer) and read errors
    /// via [`Observer::on_app_error`](crate::Observer::on_app_error).
    pub fn run(&mut self) -> Result<(), ExecutorError> {
        self.run_inner(RunMode::Forever)
    }

    /// Run for at most `max` wall-clock duration, then return.
    ///
    /// # Errors
    ///
    /// Returns the **first** [`ExecutorError`] surfaced during dispatch:
    ///
    /// * [`ExecutorError::Item`] if any item returns `Err` or panics.
    /// * [`ExecutorError::Iceoryx2`] if a `WaitSet` operation fails.
    /// * [`ExecutorError::AlreadyRunning`] if the executor is already running.
    ///
    /// If multiple items error in the same dispatch iteration, only the first
    /// is preserved; subsequent errors are discarded silently. To observe
    /// every error, attach an [`Observer`](crate::Observer) and read errors
    /// via [`Observer::on_app_error`](crate::Observer::on_app_error).
    pub fn run_for(&mut self, max: Duration) -> Result<(), ExecutorError> {
        self.run_inner(RunMode::Until(Instant::now() + max))
    }

    /// Run until `n` full barrier-cycles (`WaitSet` wakeups) have completed.
    ///
    /// # Errors
    ///
    /// Returns the **first** [`ExecutorError`] surfaced during dispatch:
    ///
    /// * [`ExecutorError::Item`] if any item returns `Err` or panics.
    /// * [`ExecutorError::Iceoryx2`] if a `WaitSet` operation fails.
    /// * [`ExecutorError::AlreadyRunning`] if the executor is already running.
    ///
    /// If multiple items error in the same dispatch iteration, only the first
    /// is preserved; subsequent errors are discarded silently. To observe
    /// every error, attach an [`Observer`](crate::Observer) and read errors
    /// via [`Observer::on_app_error`](crate::Observer::on_app_error).
    pub fn run_n(&mut self, n: usize) -> Result<(), ExecutorError> {
        self.run_inner(RunMode::Iterations(n))
    }

    /// Run until `predicate()` returns true. Checked after each `WaitSet`
    /// wakeup.
    ///
    /// # Errors
    ///
    /// Returns the **first** [`ExecutorError`] surfaced during dispatch:
    ///
    /// * [`ExecutorError::Item`] if any item returns `Err` or panics.
    /// * [`ExecutorError::Iceoryx2`] if a `WaitSet` operation fails.
    /// * [`ExecutorError::AlreadyRunning`] if the executor is already running.
    ///
    /// If multiple items error in the same dispatch iteration, only the first
    /// is preserved; subsequent errors are discarded silently. To observe
    /// every error, attach an [`Observer`](crate::Observer) and read errors
    /// via [`Observer::on_app_error`](crate::Observer::on_app_error).
    pub fn run_until<F: FnMut() -> bool>(&mut self, mut predicate: F) -> Result<(), ExecutorError> {
        self.run_inner(RunMode::Predicate(&mut predicate))
    }
}

enum RunMode<'a> {
    Forever,
    Until(Instant),
    Iterations(usize),
    Predicate(&'a mut dyn FnMut() -> bool),
}

impl Executor {
    fn run_inner(&mut self, mut mode: RunMode<'_>) -> Result<(), ExecutorError> {
        // NOTE: Once `Stoppable::stop()` has been called, `self.stoppable.is_stopped()`
        // remains true permanently. Calling `run()` again after a stop will return
        // promptly without doing any meaningful work (it blocks until the first
        // trigger fires, then immediately exits the dispatch loop). Task 10's
        // Runner accommodates this by treating an Executor as one-shot: each
        // Runner owns the Executor and consumes it.
        if self.running.swap(true, Ordering::SeqCst) {
            return Err(ExecutorError::AlreadyRunning);
        }

        self.observer.on_executor_up();
        let result = self.dispatch_loop(&mut mode);
        match &result {
            Ok(()) => self.observer.on_executor_down(),
            Err(e) => self.observer.on_executor_error(e),
        }

        self.running.store(false, Ordering::SeqCst);
        result
    }

    #[allow(
        unsafe_code,
        clippy::too_many_lines,
        clippy::ref_as_ptr,
        clippy::borrow_as_ptr
    )]
    fn dispatch_loop(&mut self, mode: &mut RunMode<'_>) -> Result<(), ExecutorError> {
        let waitset: WaitSet<ipc::Service> = WaitSetBuilder::new()
            .create()
            .map_err(ExecutorError::iceoryx2)?;

        // Keep Arc<RawListener> alive for at least as long as the WaitSet
        // guards — the guard borrows the listener via 'attachment lifetime.
        let mut listener_storage: Vec<Arc<crate::trigger::RawListener>> = Vec::new();
        // Guards must outlive the run loop.
        let mut guards: Vec<WaitSetGuard<'_, '_, ipc::Service>> = Vec::new();
        // Maps guard index → task index.
        let mut attachment_to_task: Vec<usize> = Vec::new();

        for (task_idx, task) in self.tasks.iter().enumerate() {
            for decl in &task.decls {
                match decl {
                    TriggerDecl::Subscriber { listener } => {
                        // Clone Arc to extend listener lifetime to this scope.
                        let l = Arc::clone(listener);
                        listener_storage.push(l);
                        let l_ref = listener_storage.last().unwrap().as_ref();
                        // SAFETY: we cast the reference lifetime to match
                        // 'waitset / 'attachment; both listener_storage and
                        // waitset are stack-local and dropped together at the
                        // end of dispatch_loop.  Guards are dropped before
                        // listener_storage below.
                        let l_ref: &crate::trigger::RawListener = unsafe { &*(l_ref as *const _) };
                        let guard = waitset
                            .attach_notification(l_ref)
                            .map_err(ExecutorError::iceoryx2)?;
                        guards.push(guard);
                        attachment_to_task.push(task_idx);
                    }
                    TriggerDecl::Interval(d) => {
                        let guard = waitset
                            .attach_interval(*d)
                            .map_err(ExecutorError::iceoryx2)?;
                        guards.push(guard);
                        attachment_to_task.push(task_idx);
                    }
                    TriggerDecl::Deadline { listener, deadline } => {
                        let l = Arc::clone(listener);
                        listener_storage.push(l);
                        let l_ref = listener_storage.last().unwrap().as_ref();
                        let l_ref: &crate::trigger::RawListener = unsafe { &*(l_ref as *const _) };
                        let guard = waitset
                            .attach_deadline(l_ref, *deadline)
                            .map_err(ExecutorError::iceoryx2)?;
                        guards.push(guard);
                        attachment_to_task.push(task_idx);
                    }
                    TriggerDecl::RawListener(listener) => {
                        let l = Arc::clone(listener);
                        listener_storage.push(l);
                        let l_ref = listener_storage.last().unwrap().as_ref();
                        let l_ref: &crate::trigger::RawListener = unsafe { &*(l_ref as *const _) };
                        let guard = waitset
                            .attach_notification(l_ref)
                            .map_err(ExecutorError::iceoryx2)?;
                        guards.push(guard);
                        attachment_to_task.push(task_idx);
                    }
                }
            }
        }

        // Attach the internal stop listener so the WaitSet wakes when
        // stop() is called. We hold `self.stop_listener` (Arc) in the Executor
        // struct which is valid for the lifetime of dispatch_loop. We use the
        // same raw-pointer-cast pattern as user listeners above.
        //
        // SAFETY: `self.stop_listener` is an Arc stored on `self`, which is
        // exclusively borrowed for the duration of `run_inner` (which calls
        // `dispatch_loop`). The listener is not freed while the guard is alive
        // because the Arc keeps it alive and `self` outlives this function.
        let stop_listener_ref: &IxListener<ipc::Service> =
            unsafe { &*(self.stop_listener.as_ref() as *const _) };
        let _stop_guard = waitset
            .attach_notification(stop_listener_ref)
            .map_err(ExecutorError::iceoryx2)?;

        let iterations_done = AtomicUsize::new(0);
        let stop_flag = self.stoppable.clone();

        loop {
            // Reset the pre-allocated per-iteration error slot (REQ_0060):
            // the slot is owned by `self.iter_err`, allocated once at build
            // time. Pool worker closures obtain a refcount-only clone of
            // the `Arc`; the slot itself is reused across iterations.
            *self.iter_err.lock().unwrap() = None;

            // SAFETY: we capture &mut self.tasks via a raw pointer because
            // wait_and_process expects FnMut and Rust can't see the closure
            // outlives `self`. The discipline that makes this sound:
            //   1. The closure body on the executor thread is the *only* code that
            //      reads `tasks_ptr`. The pool jobs it submits hold borrowed
            //      `*mut dyn ExecutableItem` slices into individual TaskEntries,
            //      not into the Vec itself, so they don't race with the Vec.
            //   2. `pool.barrier()` at the end of this callback ensures every
            //      submitted pool job has completed (and dropped its raw pointer)
            //      before the callback returns. The next iteration of the WaitSet
            //      loop is therefore the sole user of `tasks_ptr` again.
            //   3. The Vec is never resized inside this loop (no `push` / `remove`
            //      after dispatch starts), so the underlying buffer addresses are
            //      stable for the lifetime of `dispatch_loop`.
            let tasks_ptr = &mut self.tasks as *mut Vec<TaskEntry>;
            let pool = &self.pool;
            // Refcount-only clone of the pre-allocated error slot. Pool jobs
            // need a `'static` handle, and an `Arc::clone` does not allocate.
            // The Single/Chain paths use the closure baked into `task.job`,
            // which already captured stable Arc clones at `add`-time; the
            // Graph path uses closures pre-built by `prepare_dispatch`. Only
            // the error-aggregation logic on the WaitSet thread still needs
            // the slot here.
            let iter_err_inner = Arc::clone(&self.iter_err);
            // Raw pointer to the stop listener for draining inside the callback.
            // SAFETY: same as stop_listener_ref above — the Arc is alive for
            // the lifetime of dispatch_loop.
            let stop_listener_ptr = self.stop_listener.as_ref() as *const IxListener<ipc::Service>;
            // Raw pointer to the executor-wide fault state. Same safety
            // discipline as `tasks_ptr`: `Executor` is alive for the
            // duration of `dispatch_loop`; the WaitSet callback is the
            // only reader. REQ_0071. `self.exec_fault` is
            // `Arc<ExecutorFaultAtomic>` — we deref once to obtain a
            // pointer to the inner `ExecutorFaultAtomic`.
            let exec_fault_ptr = &*self.exec_fault as *const ExecutorFaultAtomic;
            // Raw pointer to the executor start time. Used by the lazy
            // cascade below to compute `since_ms` on task transitions
            // triggered by an executor-wide fault.
            let exec_start_ptr = &*self.start_time as *const OnceLock<Instant>;

            let cb_result = waitset.wait_and_process_once(
                |attachment_id: WaitSetAttachmentId<ipc::Service>| {
                    // Drain stop notifications first (no dispatch — the stop_flag
                    // check after the callback returns handles termination).
                    // SAFETY: stop_listener_ptr is valid for the duration of the
                    // closure; the Arc in self.stop_listener keeps it alive.
                    let stop_l = unsafe { &*stop_listener_ptr };
                    while let Ok(Some(_)) = stop_l.try_wait_one() {}

                    for (i, guard) in guards.iter().enumerate() {
                        let fired = attachment_id.has_event_from(guard)
                            || attachment_id.has_missed_deadline(guard);
                        if !fired {
                            continue;
                        }
                        let task_idx = attachment_to_task[i];

                        // SAFETY: we are the only thread that may touch
                        // `self` during the callback. wait_and_process_once
                        // is single-threaded; we hold &mut self in
                        // dispatch_loop. The pointer is valid for the
                        // duration of this closure.
                        let task = unsafe { &mut (&mut *tasks_ptr)[task_idx] };

                        // Pre-dispatch fault check (REQ_0070, REQ_0071, REQ_0072).
                        // Only applies to Single/Chain — Graph tasks use their
                        // own per-vertex scheduling and are out of scope for
                        // FEAT_0018.
                        if matches!(task.kind, TaskKind::Single(_) | TaskKind::Chain(_)) {
                            // SAFETY: exec_fault_ptr derefs into the Executor that
                            // owns this dispatch_loop — alive for the callback's
                            // lifetime.
                            let exec_faulted = matches!(
                                unsafe { &*exec_fault_ptr }.load(0, 0),
                                ExecutorFaultState::Faulted { .. }
                            );
                            let task_budget_ms = task.budget.map_or(0_u32, duration_to_ms_sat);
                            let task_state = task.fault.load(task_budget_ms);

                            // Lazy cascade: if executor is `Faulted` and task
                            // is still `Running`, silently transition the task
                            // to `Faulted{ExecutorFaulted}`. No `on_task_fault`
                            // — Observer already heard about the executor-wide
                            // fault via `on_executor_fault` (cascade-noise
                            // invariant from FEAT_0018 §4.6).
                            let task_faulted =
                                if exec_faulted && matches!(task_state, FaultState::Running) {
                                    // SAFETY: exec_start_ptr derefs into the same
                                    // `Executor` owning this dispatch_loop. The
                                    // `OnceLock` is wait-free.
                                    let exec_start = *unsafe { &*exec_start_ptr }
                                        .get_or_init(std::time::Instant::now);
                                    let since_ms =
                                        instant_to_since_ms(std::time::Instant::now(), exec_start);
                                    let _ = task.fault.swap(
                                        FaultState::Faulted {
                                            reason: FaultReason::ExecutorFaulted,
                                            since_ms,
                                        },
                                        task_budget_ms,
                                    );
                                    true
                                } else {
                                    matches!(task_state, FaultState::Faulted { .. })
                                };
                            let route_to_handler = exec_faulted || task_faulted;

                            if route_to_handler {
                                // If a handler is registered, dispatch it.
                                // Otherwise, skip dispatch entirely this wakeup.
                                if let Some(handler_box) = task.handler_job.as_deref_mut() {
                                    let job_ptr: *mut (dyn FnMut() + Send) =
                                        handler_box as *mut (dyn FnMut() + Send);
                                    // SAFETY: same as the main-job dispatch
                                    // below — handler_job is owned by the
                                    // TaskEntry; pool.barrier() awaits its
                                    // completion before the next callback.
                                    #[allow(unsafe_code)]
                                    unsafe {
                                        pool.submit_borrowed(crate::pool::BorrowedJob::new(
                                            job_ptr,
                                        ));
                                    }
                                }
                                // No handler and not Running — skip silently.
                                continue;
                            }
                        }

                        match &mut task.kind {
                            TaskKind::Single(_) | TaskKind::Chain(_) => {
                                // The dispatch closure was pre-allocated at
                                // task-add time and stashed on `task.job`.
                                // Submit it via `submit_borrowed` — no
                                // per-iteration Box allocation. Required by
                                // REQ_0060.
                                let job_box = task
                                    .job
                                    .as_deref_mut()
                                    .expect("Single/Chain tasks carry a pre-built job");
                                let job_ptr: *mut (dyn FnMut() + Send) =
                                    job_box as *mut (dyn FnMut() + Send);
                                // SAFETY: the closure lives in
                                // `task.job` which is owned by
                                // `self.tasks[task_idx]`; `tasks_ptr` is
                                // sound for the duration of this
                                // callback. `pool.barrier()` below
                                // finishes the closure invocation before
                                // we re-enter the next iteration's
                                // callback. The WaitSet thread does not
                                // touch the closure between this submit
                                // and that barrier.
                                #[allow(unsafe_code)]
                                unsafe {
                                    pool.submit_borrowed(crate::pool::BorrowedJob::new(job_ptr));
                                }
                            }
                            TaskKind::Graph(graph) => {
                                // Outer driver runs on the WaitSet thread; vertices run on the
                                // pool. The graph holds its own pre-built per-vertex closures
                                // and SPSC ready ring (REQ_0060), so dispatch is allocation-free
                                // in steady state.
                                let outcome = graph.run_once_borrowed(pool);
                                if let Some(source) = outcome.error {
                                    let mut g = iter_err_inner.lock().unwrap();
                                    if g.is_none() {
                                        *g = Some(ExecutorError::Item {
                                            task_id: task.id.clone(),
                                            source,
                                        });
                                    }
                                }
                                let _ = outcome.stopped_chain; // chain-abort semantics: no extra bookkeeping at task level
                            }
                        }
                    }

                    // Wait for all submitted jobs to finish before leaving
                    // the callback scope (validates item_ptr safety contract).
                    pool.barrier();
                    CallbackProgression::Continue
                },
            );

            let cb_result = cb_result.map_err(ExecutorError::iceoryx2)?;

            // iceoryx2's WaitSet catches SIGINT/SIGTERM internally; honor that
            // here for a clean exit.
            if matches!(
                cb_result,
                WaitSetRunResult::Interrupt | WaitSetRunResult::TerminationRequest
            ) {
                return Ok(());
            }

            // Extract the error before dropping the MutexGuard — avoids
            // holding the lock across the return (clippy::significant_drop_in_scrutinee).
            let maybe_err = self.iter_err.lock().unwrap().take();
            if let Some(err) = maybe_err {
                return Err(err);
            }
            if stop_flag.is_stopped() {
                return Ok(());
            }

            iterations_done.fetch_add(1, Ordering::SeqCst);
            match mode {
                RunMode::Forever => {}
                RunMode::Iterations(n) => {
                    if iterations_done.load(Ordering::SeqCst) >= *n {
                        return Ok(());
                    }
                }
                RunMode::Until(deadline) => {
                    if Instant::now() >= *deadline {
                        return Ok(());
                    }
                }
                RunMode::Predicate(p) => {
                    if (p)() {
                        return Ok(());
                    }
                }
            }
        }
    }
}

/// Wraps a `*mut dyn ExecutableItem` so it can cross thread boundaries inside
/// `Pool::submit`. The send is safe because:
///   1. The executor guarantees at most one invocation of a given item at a
///      time (via `pool.barrier()` before the pointer is reused).
///   2. `ExecutableItem: Send`, so moving the pointee across threads is sound
///      when no aliasing exists.
#[allow(unsafe_code)]
struct SendItemPtr {
    ptr: *mut dyn ExecutableItem,
}

impl SendItemPtr {
    fn new(ptr: *mut dyn ExecutableItem) -> Self {
        Self { ptr }
    }

    /// Returns the raw pointer. Takes `&self` so the wrapper can be invoked
    /// repeatedly from an `FnMut` dispatch closure (`REQ_0060` requires the
    /// dispatch closure to be reusable across iterations without allocation).
    fn get(&self) -> *mut dyn ExecutableItem {
        self.ptr
    }
}

// SAFETY: see doc comment above. `Sync` is required so the FnMut dispatch
// closure can borrow `&SendItemPtr` per invocation without making the
// closure itself `!Send`.
#[allow(unsafe_code)]
unsafe impl Send for SendItemPtr {}
#[allow(unsafe_code)]
unsafe impl Sync for SendItemPtr {}

/// Wraps a `*mut Vec<Box<dyn ExecutableItem>>` so a chain dispatch
/// closure can iterate the chain's items in place without first
/// collecting them into a freshly-allocated `Vec`. The send is safe
/// for the same reason as [`SendItemPtr`] (see above): the executor
/// holds `&mut self` for the duration of `dispatch_loop`, and the
/// `pool.barrier()` at the end of each callback ensures the closure
/// has finished using this pointer before the Vec could be touched
/// from the `WaitSet` thread again. The Vec is never resized after
/// dispatch begins. Required for `REQ_0060` — chain dispatch must not
/// allocate per iteration.
#[allow(unsafe_code)]
struct SendChainPtr {
    ptr: *mut Vec<Box<dyn ExecutableItem>>,
}

impl SendChainPtr {
    fn new(ptr: *mut Vec<Box<dyn ExecutableItem>>) -> Self {
        Self { ptr }
    }

    fn get(&self) -> *mut Vec<Box<dyn ExecutableItem>> {
        self.ptr
    }
}

// SAFETY: see doc comment above. `Sync` lets the FnMut dispatch closure
// borrow `&SendChainPtr` per invocation while staying `Send`.
#[allow(unsafe_code)]
unsafe impl Send for SendChainPtr {}
#[allow(unsafe_code)]
unsafe impl Sync for SendChainPtr {}

/// Captured state needed by a dispatch closure to perform post-execute
/// fault detection. All fields are `Arc`-shared with the owning
/// `Executor` and `TaskEntry` so the closure can read/write them
/// wait-free from any pool worker thread. `REQ_0070`, `REQ_0071`,
/// `REQ_0102`.
struct FaultDispatchCtx {
    /// Per-task budget. `None` for chain / graph tasks (no per-task
    /// check) — the executor-wide iteration budget still applies.
    task_budget: Option<Duration>,
    /// Per-task fault state (shared with `TaskEntry::fault`).
    task_fault: Arc<FaultAtomic>,
    /// Per-task monotonic overrun counter (shared with
    /// `TaskEntry::overrun_count`). Increments on EVERY budget breach.
    overrun_count: Arc<AtomicU64>,
    /// Executor-wide iteration budget. `None` means no executor-wide
    /// check.
    iteration_budget: Option<Duration>,
    /// Executor-wide fault state (shared with `Executor::exec_fault`).
    exec_fault: Arc<ExecutorFaultAtomic>,
    /// Executor-wide offending-task index storage (shared with
    /// `Executor::exec_fault_task_idx`).
    exec_fault_task_idx: Arc<AtomicU32>,
    /// Executor-wide breached-budget storage (shared with
    /// `Executor::exec_fault_budget_ms`).
    exec_fault_budget_ms: Arc<AtomicU32>,
    /// Index of this task in the executor's task table.
    task_idx_u32: u32,
    /// Executor start time (shared with `Executor::start_time`).
    exec_start: Arc<OnceLock<Instant>>,
    /// Observer for `on_task_fault` / `on_executor_fault` notifications.
    observer: Arc<dyn Observer>,
}

/// Build the per-iteration dispatch closure for a `TaskKind::Single`.
///
/// The returned closure is stored on `TaskEntry::job` and invoked once
/// per dispatch via `Pool::submit_borrowed`, which (unlike `submit`)
/// performs no allocation. The closure captures Arc clones of the
/// executor's shared state — those clones are refcount-only at build
/// time and are reused on every dispatch. Required for `REQ_0060`.
#[allow(clippy::too_many_arguments)]
fn build_single_job(
    id: TaskId,
    stop: Stoppable,
    obs: Arc<dyn Observer>,
    mon: Arc<dyn ExecutionMonitor>,
    err_slot: Arc<std::sync::Mutex<Option<ExecutorError>>>,
    app_id: Option<u32>,
    app_inst: Option<u32>,
    item_ptr: SendItemPtr,
    fault_ctx: FaultDispatchCtx,
) -> Box<dyn FnMut() + Send + 'static> {
    Box::new(move || {
        let mut ctx = crate::context::Context::new(&id, &stop, obs.as_ref());
        if let Some(aid) = app_id {
            obs.on_app_start(id.clone(), aid, app_inst);
        }
        let raw = item_ptr.get();
        let started = std::time::Instant::now();
        mon.pre_execute(id.clone(), started);
        // SAFETY: barrier() pairs with this invocation; the WaitSet
        // thread does not touch the item between `submit_borrowed` and
        // the matching `barrier()`. See SendItemPtr safety doc.
        #[allow(unsafe_code)]
        let res = run_item_catch_unwind(unsafe { &mut *raw }, &mut ctx);
        let took = started.elapsed();
        mon.post_execute(id.clone(), started, took, res.is_ok());
        if let Err(ref e) = res {
            obs.on_app_error(id.clone(), e.as_ref());
        }
        if app_id.is_some() {
            obs.on_app_stop(id.clone());
        }
        post_execute_detect_fault(&id, started, took, &fault_ctx);
        record_first_err(&err_slot, &id, res);
    })
}

/// Build the per-iteration dispatch closure for a fault-handler item.
///
/// Mirrors [`build_single_job`] in every detail (same monitor /
/// observer / first-error capture wiring) but owns the
/// `Box<dyn ExecutableItem>` directly inside the closure instead of
/// dereferencing a raw [`SendItemPtr`]. The handler has no parallel
/// owner inside [`TaskEntry`] — the handler closure stored in
/// `handler_job` is the sole owner — so the simpler owning form is
/// both sound and avoids the aliasing dance the main item needs.
/// `REQ_0072`.
#[allow(clippy::too_many_arguments)]
fn build_handler_job(
    id: TaskId,
    stop: Stoppable,
    obs: Arc<dyn Observer>,
    mon: Arc<dyn ExecutionMonitor>,
    err_slot: Arc<std::sync::Mutex<Option<ExecutorError>>>,
    app_id: Option<u32>,
    app_inst: Option<u32>,
    mut handler: Box<dyn ExecutableItem>,
    fault_ctx: FaultDispatchCtx,
) -> Box<dyn FnMut() + Send + 'static> {
    Box::new(move || {
        let mut ctx = crate::context::Context::new(&id, &stop, obs.as_ref());
        if let Some(aid) = app_id {
            obs.on_app_start(id.clone(), aid, app_inst);
        }
        let started = std::time::Instant::now();
        mon.pre_execute(id.clone(), started);
        let res = run_item_catch_unwind(handler.as_mut(), &mut ctx);
        let took = started.elapsed();
        mon.post_execute(id.clone(), started, took, res.is_ok());
        if let Err(ref e) = res {
            obs.on_app_error(id.clone(), e.as_ref());
        }
        if app_id.is_some() {
            obs.on_app_stop(id.clone());
        }
        // Per §4.6 invariant 5 of FEAT_0018: a handler that ALSO breaches
        // budget keeps the task in `Faulted` (state already `Faulted`),
        // `overrun_count` increments, NO new `on_task_fault` fires —
        // the `matches!(prev, FaultState::Running)` gate inside
        // `post_execute_detect_fault` enforces that.
        post_execute_detect_fault(&id, started, took, &fault_ctx);
        record_first_err(&err_slot, &id, res);
    })
}

/// Build the per-iteration dispatch closure for a `TaskKind::Chain`.
#[allow(clippy::too_many_arguments)]
fn build_chain_job(
    id: TaskId,
    stop: Stoppable,
    obs: Arc<dyn Observer>,
    mon: Arc<dyn ExecutionMonitor>,
    err_slot: Arc<std::sync::Mutex<Option<ExecutorError>>>,
    chain_ptr: SendChainPtr,
    fault_ctx: FaultDispatchCtx,
) -> Box<dyn FnMut() + Send + 'static> {
    Box::new(move || {
        let mut ctx = crate::context::Context::new(&id, &stop, obs.as_ref());
        // SAFETY: barrier() pairs with this invocation; the chain Vec
        // and the items it owns are not touched by the WaitSet thread
        // until barrier() returns. See SendChainPtr safety doc.
        #[allow(unsafe_code)]
        let chain_items = unsafe { &mut *chain_ptr.get() };
        for item_box in chain_items.iter_mut() {
            let app_id = item_box.app_id();
            let app_inst = item_box.app_instance_id();
            if let Some(aid) = app_id {
                obs.on_app_start(id.clone(), aid, app_inst);
            }
            let raw = std::ptr::from_mut::<dyn ExecutableItem>(item_box.as_mut());
            let started = std::time::Instant::now();
            mon.pre_execute(id.clone(), started);
            #[allow(unsafe_code)]
            let res = run_item_catch_unwind(unsafe { &mut *raw }, &mut ctx);
            let took = started.elapsed();
            mon.post_execute(id.clone(), started, took, res.is_ok());
            if let Err(ref e) = res {
                obs.on_app_error(id.clone(), e.as_ref());
            }
            if app_id.is_some() {
                obs.on_app_stop(id.clone());
            }
            // Per-item post-execute fault detection. `task_budget` is
            // `None` for chains (see `add_chain_with_id_boxed`), so the
            // per-task check no-ops; the executor-wide iteration-budget
            // check still fires per item. `REQ_0071`.
            post_execute_detect_fault(&id, started, took, &fault_ctx);
            match res {
                Ok(crate::ControlFlow::Continue) => {}
                Ok(crate::ControlFlow::StopChain) => break,
                Err(_) => {
                    record_first_err(&err_slot, &id, res);
                    break;
                }
            }
        }
    })
}

#[derive(Debug)]
struct PanickedTask(String);

impl core::fmt::Display for PanickedTask {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(f, "task panicked: {}", self.0)
    }
}

impl std::error::Error for PanickedTask {}

/// Execute `item` inside `catch_unwind`, converting any panic into an `Err`.
#[allow(clippy::option_if_let_else)]
fn run_item_catch_unwind(
    item: &mut dyn ExecutableItem,
    ctx: &mut crate::context::Context<'_>,
) -> crate::ExecuteResult {
    std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| item.execute(ctx))).unwrap_or_else(
        |payload| {
            let msg = if let Some(s) = payload.downcast_ref::<&str>() {
                (*s).to_string()
            } else if let Some(s) = payload.downcast_ref::<String>() {
                s.clone()
            } else {
                "panicked task".to_string()
            };
            Err::<crate::ControlFlow, crate::ItemError>(Box::new(PanickedTask(msg)))
        },
    )
}

/// Public-within-crate wrapper so `graph.rs` can call `run_item_catch_unwind`
/// without depending on its private name.
pub(crate) fn run_item_catch_unwind_external(
    item: &mut dyn ExecutableItem,
    ctx: &mut crate::context::Context<'_>,
) -> crate::ExecuteResult {
    run_item_catch_unwind(item, ctx)
}

/// Record the first error into `slot`. Subsequent errors are silently dropped.
fn record_first_err(
    slot: &Arc<std::sync::Mutex<Option<ExecutorError>>>,
    id: &TaskId,
    res: crate::ExecuteResult,
) {
    if let Err(source) = res {
        let mut g = slot.lock().unwrap();
        if g.is_none() {
            *g = Some(ExecutorError::Item {
                task_id: id.clone(),
                source,
            });
        }
    }
}

/// Post-execute fault detection — runs on a pool worker AFTER
/// `mon.post_execute` so the full `took` is available. Implements:
///
///   * `REQ_0070` / `REQ_0102` — per-task budget overrun: increments
///     `overrun_count` on every breach, transitions
///     `Running -> Faulted{BudgetExceeded}` exactly once (subsequent
///     breaches keep the state `Faulted` and do NOT re-fire the
///     observer).
///   * `REQ_0071` — executor-wide iteration overrun: transitions
///     `Running -> Faulted{IterationBudgetExceeded}` exactly once;
///     cascade to per-task state is LAZY (see the pre-dispatch block
///     in `dispatch_loop`), so the per-task `on_task_fault` does NOT
///     fire during cascade — only `on_executor_fault` does.
fn post_execute_detect_fault(
    id: &TaskId,
    started: Instant,
    took: Duration,
    fault_ctx: &FaultDispatchCtx,
) {
    // REQ_0070 / REQ_0102 — per-task budget overrun.
    if let Some(budget) = fault_ctx.task_budget {
        if took > budget {
            fault_ctx.overrun_count.fetch_add(1, Ordering::Relaxed);
            let took_ms = duration_to_ms_sat(took);
            let budget_ms = duration_to_ms_sat(budget);
            let exec_start = *fault_ctx.exec_start.get_or_init(|| started);
            let since_ms = instant_to_since_ms(started, exec_start);
            let new_state = FaultState::Faulted {
                reason: FaultReason::BudgetExceeded { took_ms, budget_ms },
                since_ms,
            };
            let prev = fault_ctx.task_fault.swap(new_state, budget_ms);
            if matches!(prev, FaultState::Running) {
                fault_ctx.observer.on_task_fault(
                    id.clone(),
                    FaultReason::BudgetExceeded { took_ms, budget_ms },
                );
            }
        }
    }

    // REQ_0071 — executor-wide iteration overrun.
    if let Some(iter_budget) = fault_ctx.iteration_budget {
        if took > iter_budget {
            let took_ms = duration_to_ms_sat(took);
            let budget_ms = duration_to_ms_sat(iter_budget);
            let exec_start = *fault_ctx.exec_start.get_or_init(|| started);
            let since_ms = instant_to_since_ms(started, exec_start);
            fault_ctx
                .exec_fault_task_idx
                .store(fault_ctx.task_idx_u32, Ordering::Release);
            fault_ctx
                .exec_fault_budget_ms
                .store(budget_ms, Ordering::Release);
            let new_state = ExecutorFaultState::Faulted {
                reason: ExecutorFaultReason::IterationBudgetExceeded {
                    task_idx: fault_ctx.task_idx_u32,
                    took_ms,
                    budget_ms,
                },
                since_ms,
            };
            let prev = fault_ctx
                .exec_fault
                .swap(new_state, fault_ctx.task_idx_u32, budget_ms);
            if matches!(prev, ExecutorFaultState::Running) {
                fault_ctx.observer.on_executor_fault(
                    ExecutorFaultReason::IterationBudgetExceeded {
                        task_idx: fault_ctx.task_idx_u32,
                        took_ms,
                        budget_ms,
                    },
                );
                // NO eager cascade here. Cascade is lazy: the
                // pre-dispatch block in `dispatch_loop` transitions
                // each `Running` task to `Faulted{ExecutorFaulted}` on
                // the next wakeup — silently, so per-task observers
                // do not fire (see §4.6 invariant on cascade-noise).
            }
        }
    }
}

// ── ExecutorGraphBuilder ──────────────────────────────────────────────────────

/// Borrowed wrapper that finalises a [`GraphBuilder`](crate::graph::GraphBuilder)
/// into a registered task.
pub struct ExecutorGraphBuilder<'e> {
    executor: &'e mut Executor,
    builder: crate::graph::GraphBuilder,
    custom_id: Option<TaskId>,
}

impl ExecutorGraphBuilder<'_> {
    /// Add a vertex to the graph; returns its handle.
    pub fn vertex<I: ExecutableItem>(&mut self, item: I) -> crate::graph::Vertex {
        self.builder.vertex(item)
    }

    /// Add a directed edge from one vertex to another.
    pub fn edge(&mut self, from: crate::graph::Vertex, to: crate::graph::Vertex) -> &mut Self {
        self.builder.edge(from, to);
        self
    }

    /// Designate the root vertex (its triggers gate the graph).
    pub const fn root(&mut self, v: crate::graph::Vertex) -> &mut Self {
        self.builder.root(v);
        self
    }

    /// Override the auto-generated id with a custom one.
    pub fn id(&mut self, id: impl Into<TaskId>) -> &mut Self {
        self.custom_id = Some(id.into());
        self
    }

    /// Validate and register the graph. Returns the task id.
    ///
    /// The root vertex's [`ExecutableItem::task_id`] override takes precedence
    /// over any id set via [`ExecutorGraphBuilder::id`], which itself takes
    /// precedence over the auto-generated id.
    pub fn build(self) -> Result<TaskId, ExecutorError> {
        let g = self.builder.finish()?;
        // Root vertex's task_id() override wins over the custom id, which wins
        // over the auto-generated fallback.
        let auto_id = || {
            TaskId::new(format!(
                "graph-{}",
                self.executor.next_id.fetch_add(1, Ordering::SeqCst)
            ))
        };
        let id = g
            .root_task_id()
            .map(TaskId::new)
            .or(self.custom_id)
            .unwrap_or_else(auto_id);
        let decls = g.decls.clone();

        // Box the graph for address stability — per-vertex dispatch
        // closures capture `*const Graph` and must not see it move.
        let mut graph_box: Box<crate::graph::Graph> = Box::new(g);
        // Pre-build the per-vertex closures now that we know the
        // task_id and have access to the executor's shared state.
        graph_box.prepare_dispatch(
            id.clone(),
            self.executor.stoppable.clone(),
            Arc::clone(&self.executor.observer),
            Arc::clone(&self.executor.monitor),
            Arc::clone(&self.executor.iter_err),
        );

        self.executor.tasks.push(TaskEntry {
            id: id.clone(),
            kind: TaskKind::Graph(graph_box),
            decls,
            // Graph tasks dispatch their vertices via `vertex_jobs`
            // stored inside the `Graph`; the per-task `job` slot
            // is unused for graphs.
            job: None,
            // TODO(post-Task-10): graph budgets carried separately; for now None.
            budget: None,
            fault: Arc::new(FaultAtomic::new()),
            overrun_count: Arc::new(AtomicU64::new(0)),
            handler_job: None,
        });
        Ok(id)
    }
}

// ── Unit tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{ControlFlow, item};

    #[test]
    fn add_returns_unique_ids() {
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let a = exec.add(item(|_| Ok(ControlFlow::Continue))).unwrap();
        let b = exec.add(item(|_| Ok(ControlFlow::Continue))).unwrap();
        assert_ne!(a, b);
    }

    #[test]
    fn custom_id_is_preserved() {
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let id = exec
            .add_with_id("my-task", item(|_| Ok(ControlFlow::Continue)))
            .unwrap();
        assert_eq!(id.as_str(), "my-task");
    }

    #[test]
    fn add_persists_declared_budget() {
        use core::time::Duration;
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let task_id = exec
            .add(crate::item::item_with_triggers(
                |d| {
                    d.interval(Duration::from_millis(10));
                    d.budget(Duration::from_millis(5));
                    Ok(())
                },
                |_| Ok(crate::ControlFlow::Continue),
            ))
            .unwrap();
        let entry = exec
            .tasks
            .iter()
            .find(|t| t.id == task_id)
            .expect("task present");
        assert_eq!(entry.budget, Some(Duration::from_millis(5)));
    }

    #[test]
    fn add_with_fault_handler_stores_handler_job() {
        use core::time::Duration;
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let task_id = exec
            .add_with_fault_handler(
                crate::item::item_with_triggers(
                    |d| {
                        d.interval(Duration::from_millis(10));
                        d.budget(Duration::from_millis(5));
                        Ok(())
                    },
                    |_| Ok(crate::ControlFlow::Continue),
                ),
                crate::item::item_with_triggers(|_d| Ok(()), |_| Ok(crate::ControlFlow::Continue)),
            )
            .unwrap();
        let entry = exec
            .tasks
            .iter()
            .find(|t| t.id == task_id)
            .expect("task present");
        assert!(
            entry.handler_job.is_some(),
            "handler_job should be Some after add_with_fault_handler"
        );
        // Main job should still be present.
        assert!(entry.job.is_some(), "main job should still be present");
    }

    #[test]
    fn declare_triggers_called_at_add_time() {
        let called = Arc::new(AtomicBool::new(false));
        let called_d = Arc::clone(&called);

        let it = crate::item::item_with_triggers(
            move |_d| {
                called_d.store(true, Ordering::SeqCst);
                Ok(())
            },
            |_| Ok(ControlFlow::Continue),
        );

        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        exec.add(it).unwrap();
        assert!(called.load(Ordering::SeqCst));
    }

    #[test]
    fn clear_task_fault_errors_on_running_task() {
        use core::time::Duration;
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let task_id = exec
            .add(crate::item::item_with_triggers(
                |d| {
                    d.interval(Duration::from_millis(10));
                    Ok(())
                },
                |_| Ok(crate::ControlFlow::Continue),
            ))
            .unwrap();
        // Task starts in Running state — clearing should error.
        let err = exec.clear_task_fault(task_id).expect_err("not faulted");
        assert!(matches!(err, ExecutorError::TaskNotFaulted(_)));
    }

    #[test]
    fn clear_executor_fault_errors_on_running_executor() {
        let exec = Executor::builder().worker_threads(0).build().unwrap();
        let err = exec.clear_executor_fault().expect_err("not faulted");
        assert!(matches!(err, ExecutorError::ExecutorNotFaulted));
    }

    #[test]
    fn overrun_count_returns_zero_for_new_task() {
        use core::time::Duration;
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let task_id = exec
            .add(crate::item::item_with_triggers(
                |d| {
                    d.interval(Duration::from_millis(10));
                    d.budget(Duration::from_millis(5));
                    Ok(())
                },
                |_| Ok(crate::ControlFlow::Continue),
            ))
            .unwrap();
        assert_eq!(exec.overrun_count(task_id).unwrap(), 0);
    }

    #[test]
    fn overrun_count_errors_for_unknown_task() {
        let exec = Executor::builder().worker_threads(0).build().unwrap();
        let err = exec
            .overrun_count(crate::TaskId::new("nope"))
            .expect_err("unknown task");
        assert!(matches!(err, ExecutorError::TaskNotFound(_)));
    }

    #[test]
    fn task_fault_state_starts_running() {
        use core::time::Duration;
        let mut exec = Executor::builder().worker_threads(0).build().unwrap();
        let task_id = exec
            .add(crate::item::item_with_triggers(
                |d| {
                    d.interval(Duration::from_millis(10));
                    Ok(())
                },
                |_| Ok(crate::ControlFlow::Continue),
            ))
            .unwrap();
        assert_eq!(exec.task_fault_state(task_id).unwrap(), FaultState::Running);
    }

    #[test]
    fn executor_fault_state_starts_running() {
        let exec = Executor::builder().worker_threads(0).build().unwrap();
        assert_eq!(exec.executor_fault_state(), ExecutorFaultState::Running);
    }
}