abxbus 2.5.74

use std::{
    collections::{BinaryHeap, HashMap, HashSet, VecDeque},
    sync::{
        atomic::{AtomicUsize, Ordering},
        mpsc as std_mpsc, Arc, OnceLock, Weak,
    },
    thread,
    time::{Duration, Instant},
};

use event_listener::Event;
use futures::executor::block_on;
use futures_timer::Delay;
use parking_lot::Mutex;
use serde::{Serialize, Serializer};
use serde_json::{json, Map, Value};

use crate::{
    base_event::{now_iso, BaseEvent, BaseEventData},
    event_handler::{EventHandler, EventHandlerCallable, EventHandlerOptions},
    event_result::{EventResult, EventResultStatus},
    id::uuid_v7_string,
    lock_manager::{LockManager, ReentrantLock},
    types::{
        EventConcurrencyMode, EventHandlerCompletionMode, EventHandlerConcurrencyMode, EventStatus,
    },
};

pub(crate) static GLOBAL_SERIAL_LOCK: OnceLock<Arc<ReentrantLock>> = OnceLock::new();
static HANDLER_EXECUTOR: OnceLock<HandlerExecutor> = OnceLock::new();
static HANDLER_CALL_EXECUTOR: OnceLock<HandlerExecutor> = OnceLock::new();
static QUEUE_JUMP_EXECUTOR: OnceLock<HandlerExecutor> = OnceLock::new();
static SLOW_WARNING_SCHEDULER: OnceLock<SlowWarningScheduler> = OnceLock::new();
static STALE_HANDLER_CONTEXTS: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
static ALL_INSTANCES: OnceLock<Mutex<Vec<Weak<EventBus>>>> = OnceLock::new();
thread_local! {
    static CURRENT_BUS_ID: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
    static CURRENT_EVENT_ID: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
    static CURRENT_HANDLER_ID: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
    static CURRENT_EVENT_CONCURRENCY: std::cell::RefCell<Option<EventConcurrencyMode>> = const { std::cell::RefCell::new(None) };
}

struct FindWaiter {
    id: u64,
    pattern: String,
    child_of_event_id: Option<String>,
    where_filter: Option<HashMap<String, Value>>,
    where_predicate: Option<FindPredicate>,
    sender: std_mpsc::Sender<Option<Arc<BaseEvent>>>,
}

#[derive(Clone)]
struct TTLDeadlineEntry {
    expires_at: f64,
    event_id: String,
}

struct BusRuntime {
    queue: Mutex<VecDeque<Arc<BaseEvent>>>,
    queue_notify: Event,
    destroyed: Mutex<bool>,
    terminal_destroyed: Mutex<bool>,
    loop_started: Mutex<bool>,
    active_event_ids: Mutex<HashSet<String>>,
    processing_event_ids: Mutex<HashSet<String>>,
    events: Mutex<HashMap<String, Arc<BaseEvent>>>,
    event_contexts: Mutex<HashMap<String, dcontext::ContextSnapshot>>,
    event_context_count: AtomicUsize,
    history_order: Mutex<VecDeque<String>>,
    max_history_size: Option<usize>,
    max_history_drop: bool,
    ttl_deadline_queue: Mutex<VecDeque<TTLDeadlineEntry>>,
    ttl_deadlines_by_event_id: Mutex<HashMap<String, f64>>,
    find_waiters: Mutex<Vec<FindWaiter>>,
    next_waiter_id: Mutex<u64>,
}

type HandlerJob = Box<dyn FnOnce() + Send + 'static>;
type SlowWarningJob = Box<dyn FnOnce() + Send + 'static>;

struct HandlerExecutor {
    sender: std_mpsc::Sender<HandlerJob>,
}

impl HandlerExecutor {
    fn new() -> Self {
        let parallelism = thread::available_parallelism()
            .map(|value| value.get())
            .unwrap_or(4);
        let worker_count = (parallelism * 4).max(32);
        let (sender, receiver) = std_mpsc::channel::<HandlerJob>();
        let receiver = Arc::new(Mutex::new(receiver));

        for _ in 0..worker_count {
            let receiver = receiver.clone();
            thread::spawn(move || loop {
                let job = receiver.lock().recv();
                match job {
                    Ok(job) => job(),
                    Err(_) => break,
                }
            });
        }

        Self { sender }
    }

    fn spawn(&self, job: impl FnOnce() + Send + 'static) {
        let _ = self.sender.send(Box::new(job));
    }
}

struct ScheduledSlowWarningJob {
    due_at: Instant,
    sequence: usize,
    job: Option<SlowWarningJob>,
}

impl PartialEq for ScheduledSlowWarningJob {
    fn eq(&self, other: &Self) -> bool {
        self.due_at == other.due_at && self.sequence == other.sequence
    }
}

impl Eq for ScheduledSlowWarningJob {}

impl PartialOrd for ScheduledSlowWarningJob {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for ScheduledSlowWarningJob {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        other
            .due_at
            .cmp(&self.due_at)
            .then_with(|| other.sequence.cmp(&self.sequence))
    }
}

struct SlowWarningScheduler {
    sender: std_mpsc::Sender<ScheduledSlowWarningJob>,
    next_sequence: AtomicUsize,
}

impl SlowWarningScheduler {
    fn new() -> Self {
        let (sender, receiver) = std_mpsc::channel::<ScheduledSlowWarningJob>();
        thread::spawn(move || {
            let mut tasks = BinaryHeap::<ScheduledSlowWarningJob>::new();
            loop {
                if tasks.is_empty() {
                    match receiver.recv() {
                        Ok(task) => tasks.push(task),
                        Err(_) => break,
                    }
                }

                let now = Instant::now();
                while tasks.peek().is_some_and(|task| task.due_at <= now) {
                    if let Some(mut task) = tasks.pop() {
                        let Some(job) = task.job.take() else {
                            continue;
                        };
                        job();
                    }
                }

                let Some(next_due) = tasks.peek().map(|task| task.due_at) else {
                    continue;
                };
                match receiver.recv_timeout(next_due.saturating_duration_since(Instant::now())) {
                    Ok(task) => tasks.push(task),
                    Err(std_mpsc::RecvTimeoutError::Timeout) => {}
                    Err(std_mpsc::RecvTimeoutError::Disconnected) => break,
                }
            }
        });
        Self {
            sender,
            next_sequence: AtomicUsize::new(0),
        }
    }

    fn schedule(&self, delay: Duration, job: impl FnOnce() + Send + 'static) {
        let sequence = self.next_sequence.fetch_add(1, Ordering::Relaxed);
        let _ = self.sender.send(ScheduledSlowWarningJob {
            due_at: Instant::now() + delay,
            sequence,
            job: Some(Box::new(job)),
        });
    }
}

#[derive(Clone)]
pub struct EventBus {
    pub name: String,
    pub id: String,
    id_ref: Arc<str>,
    pub event_concurrency: EventConcurrencyMode,
    pub event_timeout: Option<f64>,
    pub event_slow_timeout: Option<f64>,
    pub event_ttl: Option<f64>,
    pub event_result_ttl: Option<f64>,
    pub event_handler_concurrency: EventHandlerConcurrencyMode,
    pub event_handler_completion: EventHandlerCompletionMode,
    pub event_handler_slow_timeout: Option<f64>,
    pub event_handler_detect_file_paths: bool,
    pub max_handler_recursion_depth: usize,
    pub locks: Arc<LockManager>,
    handlers: Arc<Mutex<HashMap<String, Vec<EventHandler>>>>,
    runtime: Arc<BusRuntime>,
    bus_serial_lock: Arc<ReentrantLock>,
}

impl Serialize for EventBus {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        self.to_json_value().serialize(serializer)
    }
}

#[derive(Debug, Clone)]
pub struct EventBusOptions {
    pub id: Option<String>,
    pub max_history_size: Option<usize>,
    pub max_history_drop: bool,
    pub event_concurrency: EventConcurrencyMode,
    pub event_timeout: Option<f64>,
    pub event_slow_timeout: Option<f64>,
    pub event_ttl: Option<f64>,
    pub event_result_ttl: Option<f64>,
    pub event_handler_concurrency: EventHandlerConcurrencyMode,
    pub event_handler_completion: EventHandlerCompletionMode,
    pub event_handler_slow_timeout: Option<f64>,
    pub event_handler_detect_file_paths: bool,
    pub max_handler_recursion_depth: usize,
}

#[derive(Debug, Clone)]
pub struct DestroyOptions {
    pub clear: bool,
}

#[derive(Clone, Default)]
pub struct FindOptions {
    pub past: bool,
    pub past_window: Option<f64>,
    pub future: Option<f64>,
    pub child_of: Option<Arc<BaseEvent>>,
    pub where_filter: Option<HashMap<String, Value>>,
    pub where_predicate: Option<FindPredicate>,
}

#[derive(Clone)]
pub struct FilterOptions {
    pub past: bool,
    pub past_window: Option<f64>,
    pub future: Option<f64>,
    pub child_of: Option<Arc<BaseEvent>>,
    pub where_filter: Option<HashMap<String, Value>>,
    pub where_predicate: Option<FindPredicate>,
    pub limit: Option<usize>,
}

impl Default for FilterOptions {
    fn default() -> Self {
        Self {
            past: true,
            past_window: None,
            future: None,
            child_of: None,
            where_filter: None,
            where_predicate: None,
            limit: None,
        }
    }
}

pub type FindPredicate = Arc<dyn Fn(&Arc<BaseEvent>) -> bool + Send + Sync>;

impl Default for EventBusOptions {
    fn default() -> Self {
        Self {
            id: None,
            max_history_size: Some(100),
            max_history_drop: false,
            event_concurrency: EventConcurrencyMode::BusSerial,
            event_timeout: Some(60.0),
            event_slow_timeout: Some(300.0),
            event_ttl: None,
            event_result_ttl: None,
            event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
            event_handler_completion: EventHandlerCompletionMode::All,
            event_handler_slow_timeout: Some(30.0),
            event_handler_detect_file_paths: true,
            max_handler_recursion_depth: 2,
        }
    }
}

impl Default for DestroyOptions {
    fn default() -> Self {
        Self { clear: true }
    }
}

impl EventBus {
    fn handler_executor() -> &'static HandlerExecutor {
        HANDLER_EXECUTOR.get_or_init(HandlerExecutor::new)
    }

    fn handler_call_executor() -> &'static HandlerExecutor {
        HANDLER_CALL_EXECUTOR.get_or_init(HandlerExecutor::new)
    }

    fn queue_jump_executor() -> &'static HandlerExecutor {
        QUEUE_JUMP_EXECUTOR.get_or_init(HandlerExecutor::new)
    }

    fn slow_warning_scheduler() -> &'static SlowWarningScheduler {
        SLOW_WARNING_SCHEDULER.get_or_init(SlowWarningScheduler::new)
    }

    fn stale_handler_contexts() -> &'static Mutex<HashSet<String>> {
        STALE_HANDLER_CONTEXTS.get_or_init(|| Mutex::new(HashSet::new()))
    }

    fn handler_context_key(event_id: &str, handler_id: &str) -> String {
        format!("{event_id}\0{handler_id}")
    }

    fn has_completed_on_bus(&self, event: &Arc<BaseEvent>) -> bool {
        let inner = event.inner.lock();
        if inner.event_status != EventStatus::Completed {
            return false;
        }
        inner
            .event_results
            .values()
            .any(|result| result.handler.eventbus_id == self.id)
    }

    fn mark_handler_context_stale(event_id: &str, handler_id: &str) {
        Self::stale_handler_contexts()
            .lock()
            .insert(Self::handler_context_key(event_id, handler_id));
    }

    fn clear_handler_context_stale(event_id: &str, handler_id: &str) {
        Self::stale_handler_contexts()
            .lock()
            .remove(&Self::handler_context_key(event_id, handler_id));
    }

    pub fn global_serial_lock() -> Arc<ReentrantLock> {
        GLOBAL_SERIAL_LOCK
            .get_or_init(|| Arc::new(ReentrantLock::default()))
            .clone()
    }

    pub fn event_bus_serial_lock(&self) -> Arc<ReentrantLock> {
        self.bus_serial_lock.clone()
    }

    fn current_handler_context_is_stale() -> bool {
        let Some((current_event_id, current_handler_id)) = Self::current_handler_context() else {
            return false;
        };
        if Self::stale_handler_contexts()
            .lock()
            .contains(&Self::handler_context_key(
                &current_event_id,
                &current_handler_id,
            ))
        {
            return true;
        }

        let Some(current_event) = Self::live_instances()
            .into_iter()
            .find_map(|bus| bus.runtime.events.lock().get(&current_event_id).cloned())
        else {
            return false;
        };
        let Some(current_inner) = current_event.inner.try_lock() else {
            return false;
        };
        let Some(result) = current_inner.event_results.get(&current_handler_id) else {
            return false;
        };
        if result.status != EventResultStatus::Error {
            return false;
        }
        result.error.as_ref().is_some_and(|error| {
            error.starts_with("EventHandlerTimeoutError:")
                || error.starts_with("EventHandlerCancelledError:")
                || error.starts_with("EventHandlerAbortedError:")
        })
    }

    pub fn is_inside_handler_context() -> bool {
        CURRENT_HANDLER_ID.with(|handler_id| handler_id.borrow().is_some())
    }

    fn current_handler_context() -> Option<(String, String)> {
        CURRENT_EVENT_ID.with(|event_id| {
            CURRENT_HANDLER_ID
                .with(|handler_id| Some((event_id.borrow().clone()?, handler_id.borrow().clone()?)))
        })
    }

    fn handler_context_is_active_in_runtime(&self, event_id: &str, handler_id: &str) -> bool {
        let parent = { self.runtime.events.lock().get(event_id).cloned() };
        let Some(parent) = parent else {
            return false;
        };
        let status = {
            parent
                .inner
                .lock()
                .event_results
                .get(handler_id)
                .map(|result| result.status)
        };
        status.is_some_and(|status| {
            matches!(
                status,
                EventResultStatus::Pending | EventResultStatus::Started
            )
        })
    }

    fn current_handler_event_shares_event_lock_with(&self, event: &Arc<BaseEvent>) -> bool {
        let current_bus_id = CURRENT_BUS_ID.with(|bus_id| bus_id.borrow().clone());
        let current_concurrency = CURRENT_EVENT_CONCURRENCY.with(|mode| *mode.borrow());
        let (Some(current_bus_id), Some(current_concurrency)) =
            (current_bus_id, current_concurrency)
        else {
            return false;
        };
        let target_concurrency = event
            .inner
            .lock()
            .event_concurrency
            .unwrap_or(self.event_concurrency);
        match (current_concurrency, target_concurrency) {
            (EventConcurrencyMode::GlobalSerial, EventConcurrencyMode::GlobalSerial) => true,
            (EventConcurrencyMode::BusSerial, EventConcurrencyMode::BusSerial) => {
                current_bus_id == self.id
            }
            _ => false,
        }
    }

    pub(crate) fn raise_if_terminal_destroyed(&self) {
        assert!(
            !*self.runtime.terminal_destroyed.lock(),
            "{} has been destroyed and cannot be used again",
            self.label()
        );
    }

    fn live_instances() -> Vec<Arc<EventBus>> {
        let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
        instances.retain(|entry| entry.upgrade().is_some());
        instances.iter().filter_map(Weak::upgrade).collect()
    }

    pub fn live_instance_by_id(eventbus_id: &str) -> Option<Arc<EventBus>> {
        Self::live_instances()
            .into_iter()
            .find(|bus| bus.id == eventbus_id)
    }

    pub fn event_bus_for_event(event: &Arc<BaseEvent>) -> Option<Arc<EventBus>> {
        let event_id = event.inner.lock().event_id.clone();
        if let Some(current_bus_id) = CURRENT_EVENT_ID.with(|current_event_id| {
            CURRENT_BUS_ID.with(|current_bus_id| {
                (current_event_id.borrow().as_deref() == Some(event_id.as_str()))
                    .then(|| current_bus_id.borrow().clone())
                    .flatten()
            })
        }) {
            if let Some(bus) = Self::live_instance_by_id(&current_bus_id) {
                return Some(bus);
            }
        }

        event
            .runtime_eventbus_id()
            .and_then(|eventbus_id| Self::live_instance_by_id(&eventbus_id))
    }

    pub fn event_bus_for_event_id(event_id: &str) -> Option<Arc<EventBus>> {
        if event_id.is_empty() {
            return None;
        }
        if let Some(current_bus_id) = CURRENT_EVENT_ID.with(|current_event_id| {
            CURRENT_BUS_ID.with(|current_bus_id| {
                (current_event_id.borrow().as_deref() == Some(event_id))
                    .then(|| current_bus_id.borrow().clone())
                    .flatten()
            })
        }) {
            if let Some(bus) = Self::live_instance_by_id(&current_bus_id) {
                return Some(bus);
            }
        }

        Self::live_instances()
            .into_iter()
            .find(|bus| bus.runtime.events.lock().contains_key(event_id))
    }

    pub fn event_for_event_id(event_id: &str) -> Option<Arc<BaseEvent>> {
        if event_id.is_empty() {
            return None;
        }
        Self::live_instances()
            .into_iter()
            .find_map(|bus| bus.runtime.events.lock().get(event_id).cloned())
    }

    pub fn new(name: Option<String>) -> Arc<Self> {
        Self::new_with_options(name, EventBusOptions::default())
    }

    pub fn new_with_history(
        name: Option<String>,
        max_history_size: Option<usize>,
        max_history_drop: bool,
    ) -> Arc<Self> {
        Self::new_with_options(
            name,
            EventBusOptions {
                max_history_size,
                max_history_drop,
                ..EventBusOptions::default()
            },
        )
    }

    pub fn new_with_options(name: Option<String>, options: EventBusOptions) -> Arc<Self> {
        Self::new_with_options_and_loop(name, options, false, true)
    }

    fn new_with_options_and_loop(
        name: Option<String>,
        options: EventBusOptions,
        start_loop: bool,
        enforce_unique_name: bool,
    ) -> Arc<Self> {
        if let Some(timeout) = options.event_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = options.event_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_slow_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = options.event_handler_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_handler_slow_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(ttl) = options.event_ttl {
            assert!(
                ttl.is_finite() && ttl >= -1.0,
                "event_ttl must be finite and >= -1 or None"
            );
        }
        if let Some(ttl) = options.event_result_ttl {
            assert!(
                ttl.is_finite() && ttl >= -1.0,
                "event_result_ttl must be finite and >= -1 or None"
            );
        }
        let event_timeout = Some(options.event_timeout.unwrap_or(60.0));
        let event_slow_timeout = Some(options.event_slow_timeout.unwrap_or(300.0));
        let event_handler_slow_timeout = Some(options.event_handler_slow_timeout.unwrap_or(30.0));

        let id = options.id.unwrap_or_else(uuid_v7_string);
        let requested_name =
            name.unwrap_or_else(|| format!("EventBus_{}", &id[id.len().saturating_sub(8)..]));
        assert!(
            requested_name
                .chars()
                .next()
                .is_some_and(|ch| ch == '_' || ch.is_ascii_alphabetic())
                && requested_name
                    .chars()
                    .all(|ch| ch == '_' || ch.is_ascii_alphanumeric()),
            "EventBus name must be a unique identifier string, got: {requested_name}"
        );

        let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
        instances.retain(|entry| entry.upgrade().is_some());
        let resolved_name = if enforce_unique_name {
            Self::resolve_unique_name(&requested_name, &id, &instances)
        } else {
            requested_name
        };
        let bus = Arc::new(Self {
            name: resolved_name,
            id_ref: Arc::from(id.as_str()),
            id,
            event_concurrency: options.event_concurrency,
            event_timeout,
            event_slow_timeout,
            event_ttl: options.event_ttl,
            event_result_ttl: options.event_result_ttl,
            event_handler_concurrency: options.event_handler_concurrency,
            event_handler_completion: options.event_handler_completion,
            event_handler_slow_timeout,
            event_handler_detect_file_paths: options.event_handler_detect_file_paths,
            max_handler_recursion_depth: options.max_handler_recursion_depth,
            locks: Arc::new(LockManager::default()),
            handlers: Arc::new(Mutex::new(HashMap::new())),
            runtime: Arc::new(BusRuntime {
                queue: Mutex::new(VecDeque::new()),
                queue_notify: Event::new(),
                destroyed: Mutex::new(false),
                terminal_destroyed: Mutex::new(false),
                loop_started: Mutex::new(false),
                active_event_ids: Mutex::new(HashSet::new()),
                processing_event_ids: Mutex::new(HashSet::new()),
                events: Mutex::new(HashMap::new()),
                event_contexts: Mutex::new(HashMap::new()),
                event_context_count: AtomicUsize::new(0),
                history_order: Mutex::new(VecDeque::new()),
                max_history_size: options.max_history_size,
                max_history_drop: options.max_history_drop,
                ttl_deadline_queue: Mutex::new(VecDeque::new()),
                ttl_deadlines_by_event_id: Mutex::new(HashMap::new()),
                find_waiters: Mutex::new(Vec::new()),
                next_waiter_id: Mutex::new(0),
            }),
            bus_serial_lock: Arc::new(ReentrantLock::default()),
        });
        instances.push(Arc::downgrade(&bus));
        drop(instances);
        if start_loop {
            bus.ensure_loop_started();
        }
        bus
    }

    fn resolve_unique_name(requested_name: &str, id: &str, instances: &[Weak<EventBus>]) -> String {
        let has_conflict = |candidate: &str| {
            instances
                .iter()
                .filter_map(Weak::upgrade)
                .any(|existing| existing.name == candidate)
        };
        if !has_conflict(requested_name) {
            return requested_name.to_string();
        }

        let suffix = &id[id.len().saturating_sub(8)..];
        let mut candidate = format!("{requested_name}_{suffix}");
        let mut attempt = 2usize;
        while has_conflict(&candidate) {
            candidate = format!("{requested_name}_{suffix}_{attempt}");
            attempt += 1;
        }
        candidate
    }

    fn unregister_instance(&self) {
        let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
        let self_ptr = self as *const Self;
        instances.retain(|entry| {
            entry
                .upgrade()
                .is_some_and(|bus| !std::ptr::eq(Arc::as_ptr(&bus), self_ptr))
        });
    }

    pub fn all_instances_contains(bus: &Arc<Self>) -> bool {
        let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
        instances.retain(|entry| entry.upgrade().is_some());
        instances
            .iter()
            .filter_map(Weak::upgrade)
            .any(|entry| Arc::ptr_eq(&entry, bus))
    }

    pub fn all_instances_len() -> usize {
        let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
        instances.retain(|entry| entry.upgrade().is_some());
        instances.len()
    }

    fn ensure_loop_started(&self) {
        if *self.runtime.terminal_destroyed.lock() {
            return;
        }
        let mut started = self.runtime.loop_started.lock();
        if *started {
            return;
        }
        if *self.runtime.terminal_destroyed.lock() {
            return;
        }
        *self.runtime.destroyed.lock() = false;
        *started = true;
        let bus = self.clone();
        drop(started);
        Self::start_loop(bus);
    }

    fn notify_all_queues() {
        for bus in Self::live_instances() {
            bus.ensure_loop_started();
            bus.runtime.queue_notify.notify(usize::MAX);
        }
    }

    fn start_loop(bus: Self) {
        thread::spawn(move || {
            block_on(async move {
                loop {
                    bus.locks.wait_until_runloop_resumed();
                    let listener = bus.runtime.queue_notify.listen();
                    let next_event = {
                        let mut queue = bus.runtime.queue.lock();
                        if bus.locks.is_paused() {
                            drop(queue);
                            drop(listener);
                            continue;
                        }
                        let next_event = queue.pop_front();
                        if let Some(event) = &next_event {
                            let event_id = event.inner.lock().event_id.clone();
                            bus.runtime.active_event_ids.lock().insert(event_id);
                        }
                        next_event
                    };
                    if let Some(event) = next_event {
                        drop(listener);
                        let bus_for_task = bus.clone();
                        let event_id = event.inner.lock().event_id.clone();
                        let mode = event
                            .inner
                            .lock()
                            .event_concurrency
                            .unwrap_or(bus.event_concurrency);
                        match mode {
                            EventConcurrencyMode::Parallel => {
                                thread::spawn(move || {
                                    block_on(bus_for_task.process_event(event));
                                    bus_for_task
                                        .runtime
                                        .active_event_ids
                                        .lock()
                                        .remove(&event_id);
                                });
                            }
                            EventConcurrencyMode::GlobalSerial
                            | EventConcurrencyMode::BusSerial => {
                                bus.process_event(event).await;
                                bus.runtime.active_event_ids.lock().remove(&event_id);
                            }
                        }
                        continue;
                    }

                    if *bus.runtime.destroyed.lock() {
                        break;
                    }
                    listener.await;
                }
            });
        });
    }

    fn start_parallel_event_task_from_queue(&self, event: Arc<BaseEvent>) {
        let event_id = event.inner.lock().event_id.clone();
        let removed = {
            let mut queue = self.runtime.queue.lock();
            if let Some(index) = queue
                .iter()
                .position(|queued| queued.inner.lock().event_id == event_id)
            {
                queue.remove(index);
                self.runtime
                    .active_event_ids
                    .lock()
                    .insert(event_id.clone());
                true
            } else {
                false
            }
        };
        if !removed {
            return;
        }
        if self.has_completed_on_bus(&event) {
            self.runtime.active_event_ids.lock().remove(&event_id);
            return;
        }

        let bus = self.clone();
        thread::spawn(move || {
            block_on(bus.process_event(event));
            bus.runtime.active_event_ids.lock().remove(&event_id);
        });
    }

    pub fn destroy(&self) {
        self.destroy_with_options(DestroyOptions::default());
    }

    pub fn destroy_with_options(&self, options: DestroyOptions) {
        *self.runtime.destroyed.lock() = true;
        *self.runtime.terminal_destroyed.lock() = true;
        *self.runtime.loop_started.lock() = false;
        self.unregister_instance();
        self.clear_runtime_state(options.clear);
        self.runtime.queue_notify.notify(usize::MAX);
    }

    fn clear_runtime_state(&self, clear: bool) {
        let waiters = {
            let mut find_waiters = self.runtime.find_waiters.lock();
            find_waiters
                .drain(..)
                .map(|waiter| waiter.sender)
                .collect::<Vec<_>>()
        };
        for sender in waiters {
            let _ = sender.send(None);
        }
        self.runtime.queue.lock().clear();
        self.runtime.active_event_ids.lock().clear();
        self.runtime.processing_event_ids.lock().clear();
        self.runtime.event_contexts.lock().clear();
        self.runtime.event_context_count.store(0, Ordering::SeqCst);
        self.locks.clear();
        if clear {
            self.runtime.events.lock().clear();
            self.runtime.history_order.lock().clear();
            self.runtime.ttl_deadline_queue.lock().clear();
            self.runtime.ttl_deadlines_by_event_id.lock().clear();
            self.handlers.lock().clear();
        }
    }

    pub fn label(&self) -> String {
        format!(
            "{}#{}",
            self.name,
            &self.id[self.id.len().saturating_sub(4)..]
        )
    }

    pub fn max_history_size(&self) -> Option<usize> {
        self.runtime.max_history_size
    }

    pub fn max_history_drop(&self) -> bool {
        self.runtime.max_history_drop
    }

    pub fn event_history_size(&self) -> usize {
        self.runtime.history_order.lock().len()
    }

    pub fn is_idle_and_queue_empty(&self) -> bool {
        let queue_empty = self.runtime.queue.lock().is_empty();
        let no_active = self.runtime.active_event_ids.lock().is_empty();
        let no_processing = self.runtime.processing_event_ids.lock().is_empty();
        queue_empty && no_active && no_processing
    }

    pub fn is_running_for_test(&self) -> bool {
        *self.runtime.loop_started.lock() && !*self.runtime.destroyed.lock()
    }

    pub fn is_destroyed_for_test(&self) -> bool {
        *self.runtime.terminal_destroyed.lock()
    }

    pub fn runtime_payload_for_test(&self) -> HashMap<String, Arc<BaseEvent>> {
        self.runtime.events.lock().clone()
    }

    pub fn event_history_ids(&self) -> Vec<String> {
        self.runtime.history_order.lock().iter().cloned().collect()
    }

    pub fn find_waiter_count_for_test(&self) -> usize {
        self.runtime.find_waiters.lock().len()
    }

    pub fn to_json_value(&self) -> Value {
        let handlers_by_pattern = self.handlers.lock().clone();
        let mut handlers = Map::new();
        let mut handlers_by_key = Map::new();

        let mut handler_entries: Vec<_> = handlers_by_pattern
            .values()
            .flat_map(|entries| entries.iter().cloned())
            .collect();
        handler_entries.sort_by(|left, right| {
            left.handler_registered_at
                .cmp(&right.handler_registered_at)
                .then_with(|| left.id.cmp(&right.id))
        });
        for handler in handler_entries {
            handlers.insert(handler.id.clone(), handler.to_json_value());
        }

        let mut pattern_entries: Vec<_> = handlers_by_pattern.into_iter().collect();
        pattern_entries.sort_by(|left, right| {
            let left_registered_at = left
                .1
                .iter()
                .map(|handler| &handler.handler_registered_at)
                .min();
            let right_registered_at = right
                .1
                .iter()
                .map(|handler| &handler.handler_registered_at)
                .min();
            left_registered_at
                .cmp(&right_registered_at)
                .then_with(|| left.0.cmp(&right.0))
        });
        for (pattern, entries) in pattern_entries {
            let ids = entries
                .into_iter()
                .map(|handler| Value::String(handler.id))
                .collect();
            handlers_by_key.insert(pattern, Value::Array(ids));
        }

        let mut event_history = Map::new();
        let events = self.runtime.events.lock().clone();
        for event_id in self.runtime.history_order.lock().iter() {
            if let Some(event) = events.get(event_id) {
                event_history.insert(event_id.clone(), event.to_json_value());
            }
        }

        let mut pending_event_queue = Vec::new();
        for event in self.runtime.queue.lock().iter() {
            let event_id = event.inner.lock().event_id.clone();
            if !event_history.contains_key(&event_id) {
                event_history.insert(event_id.clone(), event.to_json_value());
            }
            pending_event_queue.push(Value::String(event_id));
        }

        json!({
            "id": self.id,
            "name": self.name,
            "max_history_size": self.runtime.max_history_size,
            "max_history_drop": self.runtime.max_history_drop,
            "event_concurrency": self.event_concurrency,
            "event_timeout": self.event_timeout,
            "event_slow_timeout": self.event_slow_timeout,
            "event_ttl": self.event_ttl,
            "event_result_ttl": self.event_result_ttl,
            "event_handler_concurrency": self.event_handler_concurrency,
            "event_handler_completion": self.event_handler_completion,
            "event_handler_slow_timeout": self.event_handler_slow_timeout,
            "event_handler_detect_file_paths": self.event_handler_detect_file_paths,
            "handlers": handlers,
            "handlers_by_key": handlers_by_key,
            "event_history": event_history,
            "pending_event_queue": pending_event_queue,
        })
    }

    pub fn log_tree(&self) -> String {
        let events = self.runtime.events.lock().clone();
        let history_order = self.runtime.history_order.lock().clone();
        let mut children_by_parent: HashMap<String, Vec<Arc<BaseEvent>>> = HashMap::new();
        for event in events.values() {
            let parent_id = event.inner.lock().event_parent_id.clone();
            if let Some(parent_id) = parent_id {
                if parent_id != event.inner.lock().event_id && events.contains_key(&parent_id) {
                    children_by_parent
                        .entry(parent_id)
                        .or_default()
                        .push(event.clone());
                }
            }
        }
        for children in children_by_parent.values_mut() {
            children.sort_by(|left, right| {
                left.inner
                    .lock()
                    .event_created_at
                    .cmp(&right.inner.lock().event_created_at)
            });
        }

        let mut roots = Vec::new();
        for event_id in history_order {
            let Some(event) = events.get(&event_id).cloned() else {
                continue;
            };
            let inner = event.inner.lock();
            let is_root = inner.event_parent_id.as_ref().is_none_or(|parent_id| {
                parent_id == &inner.event_id || !events.contains_key(parent_id)
            });
            drop(inner);
            if is_root {
                roots.push(event);
            }
        }
        if roots.is_empty() {
            return "(No events in history)".to_string();
        }

        let mut lines = Vec::new();
        lines.push(format!("📊 Event History Tree for {}", self.label()));
        lines.push("=".repeat(80));
        let mut visited = std::collections::HashSet::new();
        for (index, event) in roots.iter().enumerate() {
            let is_last = index == roots.len() - 1;
            self.push_log_event_lines(
                &mut lines,
                event.clone(),
                "",
                is_last,
                &children_by_parent,
                &events,
                &mut visited,
            );
        }
        lines.push("=".repeat(80));
        lines.join("\n")
    }

    #[allow(clippy::too_many_arguments)]
    fn push_log_event_lines(
        &self,
        lines: &mut Vec<String>,
        event: Arc<BaseEvent>,
        indent: &str,
        is_last: bool,
        children_by_parent: &HashMap<String, Vec<Arc<BaseEvent>>>,
        events: &HashMap<String, Arc<BaseEvent>>,
        visited: &mut std::collections::HashSet<String>,
    ) {
        let event_data = event.inner.lock().clone();
        let connector = if is_last { "└── " } else { "├── " };
        let status_icon = match event_data.event_status {
            EventStatus::Completed => "✅",
            EventStatus::Started => "🏃",
            EventStatus::Pending => "⏳",
        };
        let mut timing = format!(
            "[{}",
            Self::format_log_timestamp(&event_data.event_created_at)
        );
        if let Some(completed_at) = &event_data.event_completed_at {
            if let Some(duration) =
                Self::duration_seconds(&event_data.event_created_at, completed_at)
            {
                timing.push_str(&format!(" ({duration:.3}s)"));
            }
        }
        timing.push(']');
        lines.push(format!(
            "{indent}{connector}{status_icon} {}#{} {timing}",
            event_data.event_type,
            Self::short_suffix(&event_data.event_id)
        ));

        if !visited.insert(event_data.event_id.clone()) {
            return;
        }

        let extension = if is_last { "    " } else { "│   " };
        let child_indent = format!("{indent}{extension}");
        let mut result_items: Vec<EventResult> =
            event_data.event_results.values().cloned().collect();
        result_items.sort_by(|left, right| {
            left.handler
                .handler_registered_at
                .cmp(&right.handler.handler_registered_at)
                .then_with(|| left.handler.id.cmp(&right.handler.id))
        });

        let mut printed_child_ids = std::collections::HashSet::new();
        let children = children_by_parent
            .get(&event_data.event_id)
            .cloned()
            .unwrap_or_default();
        let extra_children: Vec<Arc<BaseEvent>> = children
            .iter()
            .filter(|child| child.inner.lock().event_emitted_by_handler_id.is_none())
            .cloned()
            .collect();

        let total = result_items.len() + extra_children.len();
        let mut item_index = 0;
        for result in result_items {
            item_index += 1;
            let is_last_item = item_index == total;
            self.push_log_result_lines(
                lines,
                &result,
                &child_indent,
                is_last_item,
                events,
                visited,
                &mut printed_child_ids,
            );
        }
        for child in extra_children {
            item_index += 1;
            if !printed_child_ids.insert(child.inner.lock().event_id.clone()) {
                continue;
            }
            self.push_log_event_lines(
                lines,
                child,
                &child_indent,
                item_index == total,
                children_by_parent,
                events,
                visited,
            );
        }
    }

    #[allow(clippy::too_many_arguments)]
    fn push_log_result_lines(
        &self,
        lines: &mut Vec<String>,
        result: &EventResult,
        indent: &str,
        is_last: bool,
        events: &HashMap<String, Arc<BaseEvent>>,
        visited: &mut std::collections::HashSet<String>,
        printed_child_ids: &mut std::collections::HashSet<String>,
    ) {
        let connector = if is_last { "└── " } else { "├── " };
        let status_icon = match result.status {
            EventResultStatus::Completed => "✅",
            EventResultStatus::Error
                if result.error.as_deref().is_some_and(|error| {
                    error.starts_with("EventHandlerCancelledError:")
                        || error.starts_with("EventHandlerAbortedError:")
                }) =>
            {
                "🚫"
            }
            EventResultStatus::Error => "❌",
            EventResultStatus::Started => "🏃",
            EventResultStatus::Pending => "⏳",
        };
        let handler_display = format!(
            "{}#{}.{}#{}",
            result.handler.eventbus_name,
            Self::short_suffix(&result.handler.eventbus_id),
            result.handler.handler_name,
            Self::short_suffix(&result.handler.id)
        );
        let mut line = format!("{indent}{connector}{status_icon} {handler_display}");
        if let Some(started_at) = &result.started_at {
            line.push_str(&format!(" [{}", Self::format_log_timestamp(started_at)));
            if let Some(completed_at) = &result.completed_at {
                if let Some(duration) = Self::duration_seconds(started_at, completed_at) {
                    line.push_str(&format!(" ({duration:.3}s)"));
                }
            }
            line.push(']');
        }
        if result.status == EventResultStatus::Error {
            if let Some(error) = &result.error {
                if let Some(message) = error.strip_prefix("EventHandlerTimeoutError: ") {
                    line.push_str(&format!(" ⏱️ Timeout: {message}"));
                } else if let Some(message) = error.strip_prefix("EventHandlerCancelledError: ") {
                    line.push_str(&format!(" Cancelled: {message}"));
                } else if let Some(message) = error.strip_prefix("EventHandlerAbortedError: ") {
                    line.push_str(&format!(" Aborted: {message}"));
                } else {
                    line.push_str(&format!(" ☠️ {}", Self::format_error_for_log(error)));
                }
            }
        } else if result.status == EventResultStatus::Completed {
            line.push_str(&format!(
                " → {}",
                Self::format_result_value(result.result.as_ref())
            ));
        }
        lines.push(line);

        let extension = if is_last { "    " } else { "│   " };
        let child_indent = format!("{indent}{extension}");
        let children: Vec<Arc<BaseEvent>> = result
            .event_children
            .iter()
            .filter_map(|child_id| events.get(child_id).cloned())
            .filter(|child| !visited.contains(&child.inner.lock().event_id))
            .collect();
        for (index, child) in children.iter().enumerate() {
            printed_child_ids.insert(child.inner.lock().event_id.clone());
            self.push_log_event_lines(
                lines,
                child.clone(),
                &child_indent,
                index == children.len() - 1,
                &HashMap::new(),
                events,
                visited,
            );
        }
    }

    fn short_suffix(value: &str) -> String {
        value[value.len().saturating_sub(4)..].to_string()
    }

    fn format_log_timestamp(value: &str) -> String {
        chrono::DateTime::parse_from_rfc3339(value)
            .map(|timestamp| {
                timestamp
                    .with_timezone(&chrono::Utc)
                    .format("%H:%M:%S%.3f")
                    .to_string()
            })
            .unwrap_or_else(|_| "N/A".to_string())
    }

    fn duration_seconds(started_at: &str, completed_at: &str) -> Option<f64> {
        let started = chrono::DateTime::parse_from_rfc3339(started_at).ok()?;
        let completed = chrono::DateTime::parse_from_rfc3339(completed_at).ok()?;
        completed
            .signed_duration_since(started)
            .num_nanoseconds()
            .map(|nanos| nanos as f64 / 1_000_000_000.0)
    }

    fn format_result_value(value: Option<&Value>) -> String {
        match value {
            None | Some(Value::Null) => "None".to_string(),
            Some(Value::String(value)) => {
                serde_json::to_string(value).unwrap_or_else(|_| "\"\"".to_string())
            }
            Some(Value::Number(value)) => value.to_string(),
            Some(Value::Bool(value)) => value.to_string(),
            Some(Value::Array(value)) => format!("list({} items)", value.len()),
            Some(Value::Object(value)) => format!("dict({} items)", value.len()),
        }
    }

    fn format_error_for_log(error: &str) -> String {
        if error.contains(": ") {
            error.to_string()
        } else {
            format!("Error: {error}")
        }
    }

    pub fn from_json_value(value: Value) -> Arc<Self> {
        let Value::Object(payload) = value else {
            panic!("EventBus.from_json_value(data) requires an object");
        };

        let options = EventBusOptions {
            id: payload
                .get("id")
                .and_then(Value::as_str)
                .map(ToString::to_string),
            max_history_size: match payload.get("max_history_size") {
                Some(Value::Null) => None,
                Some(value) => value.as_u64().map(|value| value as usize),
                None => Some(100),
            },
            max_history_drop: payload
                .get("max_history_drop")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            event_concurrency: payload
                .get("event_concurrency")
                .cloned()
                .and_then(|value| serde_json::from_value(value).ok())
                .unwrap_or(EventConcurrencyMode::BusSerial),
            event_timeout: match payload.get("event_timeout") {
                Some(Value::Null) => Some(60.0),
                Some(value) => value.as_f64(),
                None => Some(60.0),
            },
            event_slow_timeout: match payload.get("event_slow_timeout") {
                Some(Value::Null) => None,
                Some(value) => value.as_f64(),
                None => Some(300.0),
            },
            event_ttl: match payload.get("event_ttl") {
                Some(Value::Null) | None => None,
                Some(value) => value.as_f64(),
            },
            event_result_ttl: match payload.get("event_result_ttl") {
                Some(Value::Null) | None => None,
                Some(value) => value.as_f64(),
            },
            event_handler_concurrency: payload
                .get("event_handler_concurrency")
                .cloned()
                .and_then(|value| serde_json::from_value(value).ok())
                .unwrap_or(EventHandlerConcurrencyMode::Serial),
            event_handler_completion: payload
                .get("event_handler_completion")
                .cloned()
                .and_then(|value| serde_json::from_value(value).ok())
                .unwrap_or(EventHandlerCompletionMode::All),
            event_handler_slow_timeout: match payload.get("event_handler_slow_timeout") {
                Some(Value::Null) => None,
                Some(value) => value.as_f64(),
                None => Some(30.0),
            },
            event_handler_detect_file_paths: payload
                .get("event_handler_detect_file_paths")
                .and_then(Value::as_bool)
                .unwrap_or(true),
            max_handler_recursion_depth: 2,
        };
        let name = payload
            .get("name")
            .and_then(Value::as_str)
            .map(ToString::to_string);
        crate::event_handler::validate_optional_seconds_at_least_minus_one(
            "event_ttl",
            options.event_ttl,
        )
        .expect("event_ttl must be finite and >= -1 or None");
        crate::event_handler::validate_optional_seconds_at_least_minus_one(
            "event_result_ttl",
            options.event_result_ttl,
        )
        .expect("event_result_ttl must be finite and >= -1 or None");
        let bus = Self::new_with_options_and_loop(name, options, false, false);

        let mut handlers_by_id = HashMap::new();
        if let Some(Value::Object(raw_handlers)) = payload.get("handlers") {
            for (handler_id, handler_value) in raw_handlers {
                let mut handler = EventHandler::from_json_value(handler_value.clone());
                if handler.id.is_empty() {
                    handler.id = handler_id.clone();
                }
                handlers_by_id.insert(handler.id.clone(), handler);
            }
        }

        {
            let mut handlers = bus.handlers.lock();
            handlers.clear();
            if let Some(Value::Object(raw_handlers_by_key)) = payload.get("handlers_by_key") {
                for (pattern, raw_ids) in raw_handlers_by_key {
                    let Some(ids) = raw_ids.as_array() else {
                        continue;
                    };
                    for handler_id in ids.iter().filter_map(Value::as_str) {
                        if let Some(handler) = handlers_by_id.get(handler_id).cloned() {
                            handlers.entry(pattern.clone()).or_default().push(handler);
                        }
                    }
                }
            } else {
                for handler in handlers_by_id.values().cloned() {
                    handlers
                        .entry(handler.event_pattern.clone())
                        .or_default()
                        .push(handler);
                }
            }
        }

        bus.runtime.events.lock().clear();
        bus.runtime.event_contexts.lock().clear();
        bus.runtime.event_context_count.store(0, Ordering::SeqCst);
        bus.runtime.history_order.lock().clear();
        bus.runtime.ttl_deadline_queue.lock().clear();
        bus.runtime.ttl_deadlines_by_event_id.lock().clear();
        if let Some(Value::Object(raw_event_history)) = payload.get("event_history") {
            for (event_id_hint, event_value) in raw_event_history {
                let event = BaseEvent::from_json_value(event_value.clone());
                let event_id = event.inner.lock().event_id.clone();
                if event_id.is_empty() {
                    event.inner.lock().event_id = event_id_hint.clone();
                }
                for result in event.inner.lock().event_results.values() {
                    let handler = result.handler.clone();
                    let mut handlers = bus.handlers.lock();
                    let exists = handlers
                        .get(&handler.event_pattern)
                        .map(|entries| entries.iter().any(|entry| entry.id == handler.id))
                        .unwrap_or(false);
                    if !exists {
                        handlers
                            .entry(handler.event_pattern.clone())
                            .or_default()
                            .push(handler);
                    }
                }
                let event_id = event.inner.lock().event_id.clone();
                bus.runtime
                    .events
                    .lock()
                    .insert(event_id.clone(), event.clone());
                bus.runtime.history_order.lock().push_back(event_id);
                bus.track_event_ttl_deadline(&event);
            }
        }
        bus.runtime.queue.lock().clear();
        if let Some(Value::Array(raw_pending_ids)) = payload.get("pending_event_queue") {
            for event_id in raw_pending_ids.iter().filter_map(Value::as_str) {
                if let Some(event) = bus.runtime.events.lock().get(event_id).cloned() {
                    bus.runtime.queue.lock().push_back(event);
                }
            }
        }

        bus
    }

    #[track_caller]
    pub fn on_raw<F, Fut>(&self, pattern: &str, handler_name: &str, handler_fn: F) -> EventHandler
    where
        F: Fn(Arc<BaseEvent>) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<Value, String>> + Send + 'static,
    {
        self.on_raw_with_options(
            pattern,
            handler_name,
            EventHandlerOptions::default(),
            handler_fn,
        )
    }

    #[track_caller]
    pub fn on_raw_with_options<F, Fut>(
        &self,
        pattern: &str,
        handler_name: &str,
        mut options: EventHandlerOptions,
        handler_fn: F,
    ) -> EventHandler
    where
        F: Fn(Arc<BaseEvent>) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<Value, String>> + Send + 'static,
    {
        self.raise_if_terminal_destroyed();
        if let Some(timeout) = options.handler_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "handler_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = options.handler_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "handler_slow_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(ttl) = options.handler_result_ttl {
            assert!(
                ttl.is_finite() && ttl >= -1.0,
                "handler_result_ttl must be finite and >= -1 or None"
            );
        }
        let detect_handler_file_path = options
            .detect_handler_file_path
            .unwrap_or(self.event_handler_detect_file_paths);
        options.detect_handler_file_path = Some(detect_handler_file_path);
        if detect_handler_file_path && options.handler_file_path.is_none() {
            let caller = std::panic::Location::caller();
            options.handler_file_path = Some(format!("{}:{}", caller.file(), caller.line()));
        }
        let callable: EventHandlerCallable = Arc::new(move |event| Box::pin(handler_fn(event)));
        self.register_callable(pattern, handler_name, options, callable)
    }

    #[track_caller]
    pub fn on_raw_sync<F>(&self, pattern: &str, handler_name: &str, handler_fn: F) -> EventHandler
    where
        F: Fn(Arc<BaseEvent>) -> Result<Value, String> + Send + Sync + 'static,
    {
        self.on_raw_sync_with_options(
            pattern,
            handler_name,
            EventHandlerOptions::default(),
            handler_fn,
        )
    }

    #[track_caller]
    pub fn on_raw_sync_with_options<F>(
        &self,
        pattern: &str,
        handler_name: &str,
        mut options: EventHandlerOptions,
        handler_fn: F,
    ) -> EventHandler
    where
        F: Fn(Arc<BaseEvent>) -> Result<Value, String> + Send + Sync + 'static,
    {
        self.raise_if_terminal_destroyed();
        if let Some(timeout) = options.handler_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "handler_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = options.handler_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "handler_slow_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(ttl) = options.handler_result_ttl {
            assert!(
                ttl.is_finite() && ttl >= -1.0,
                "handler_result_ttl must be finite and >= -1 or None"
            );
        }
        let detect_handler_file_path = options
            .detect_handler_file_path
            .unwrap_or(self.event_handler_detect_file_paths);
        options.detect_handler_file_path = Some(detect_handler_file_path);
        if detect_handler_file_path && options.handler_file_path.is_none() {
            let caller = std::panic::Location::caller();
            options.handler_file_path = Some(format!("{}:{}", caller.file(), caller.line()));
        }
        let callable: EventHandlerCallable = Arc::new(move |event| {
            let result = handler_fn(event);
            Box::pin(async move { result })
        });
        self.register_callable(pattern, handler_name, options, callable)
    }

    fn register_callable(
        &self,
        pattern: &str,
        handler_name: &str,
        options: EventHandlerOptions,
        callable: EventHandlerCallable,
    ) -> EventHandler {
        let entry = EventHandler::from_callable_with_options(
            pattern.to_string(),
            handler_name.to_string(),
            self.name.clone(),
            self.id.clone(),
            callable,
            options,
        );
        let mut handlers = self.handlers.lock();
        let entries = handlers.entry(pattern.to_string()).or_default();
        if let Some(existing) = entries.iter_mut().find(|handler| handler.id == entry.id) {
            *existing = entry.clone();
        } else {
            entries.push(entry.clone());
        }
        entry
    }

    pub fn off(&self, pattern: &str, handler_id: Option<&str>) {
        self.raise_if_terminal_destroyed();
        let mut handlers = self.handlers.lock();
        if let Some(list) = handlers.get_mut(pattern) {
            if let Some(handler_id) = handler_id {
                list.retain(|handler| handler.id != handler_id);
            } else {
                list.clear();
            }
            if list.is_empty() {
                handlers.remove(pattern);
            }
        }
    }

    pub fn off_event<E: crate::typed::EventSpec>(&self, handler_id: Option<&str>) {
        self.off(E::event_type, handler_id);
    }

    pub fn emit_base(&self, event: Arc<BaseEvent>) -> Arc<BaseEvent> {
        self.raise_if_terminal_destroyed();
        self.enqueue_base(event)
    }

    pub fn emit_child_base(&self, event: Arc<BaseEvent>) -> Arc<BaseEvent> {
        self.raise_if_terminal_destroyed();
        self.enqueue_child_base(event)
    }

    pub(crate) fn enqueue_base(&self, event: Arc<BaseEvent>) -> Arc<BaseEvent> {
        self.enqueue_base_with_options(event, false)
    }

    pub(crate) fn enqueue_base_with_options(
        &self,
        event: Arc<BaseEvent>,
        queue_jump: bool,
    ) -> Arc<BaseEvent> {
        self.enqueue_base_with_relation(event, queue_jump, false)
    }

    pub(crate) fn enqueue_child_base(&self, event: Arc<BaseEvent>) -> Arc<BaseEvent> {
        self.enqueue_child_base_with_options(event, false)
    }

    pub(crate) fn enqueue_child_base_with_options(
        &self,
        event: Arc<BaseEvent>,
        queue_jump: bool,
    ) -> Arc<BaseEvent> {
        self.enqueue_base_with_relation(event, queue_jump, true)
    }

    fn validate_event_execution_timeout_fields(event: &BaseEvent) {
        let inner = event.inner.lock();
        if let Some(timeout) = inner.event_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = inner.event_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_slow_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = inner.event_handler_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_handler_timeout must be finite and >= 0 or None"
            );
        }
        if let Some(timeout) = inner.event_handler_slow_timeout {
            assert!(
                timeout.is_finite() && timeout >= 0.0,
                "event_handler_slow_timeout must be finite and >= 0 or None"
            );
        }
    }

    fn enqueue_base_with_relation(
        &self,
        event: Arc<BaseEvent>,
        queue_jump: bool,
        track_child: bool,
    ) -> Arc<BaseEvent> {
        self.raise_if_terminal_destroyed();
        if Self::current_handler_context_is_stale() {
            return event;
        }
        Self::validate_event_execution_timeout_fields(&event);
        let current_handler_context = Self::current_handler_context();
        let emitted_from_active_handler =
            current_handler_context
                .as_ref()
                .is_some_and(|(parent_id, handler_id)| {
                    self.handler_context_is_active_in_runtime(parent_id, handler_id)
                });
        let emitted_inside_active_handler = current_handler_context.is_some();

        let bus_label = self.label();
        let (event_id, already_in_event_path, should_skip_handler_execution) = {
            let inner = event.inner.lock();
            (
                inner.event_id.clone(),
                inner.event_path.contains(&bus_label),
                inner.event_status == EventStatus::Completed || inner.event_completed_at.is_some(),
            )
        };
        let already_in_history = self.runtime.events.lock().contains_key(&event_id);
        if (already_in_event_path || already_in_history)
            && self.completed_event_expired_for_history(&event)
        {
            self.runtime.events.lock().remove(&event_id);
            self.untrack_event_ttl_deadline(&event_id);
            self.runtime
                .history_order
                .lock()
                .retain(|id| id != &event_id);
            return event;
        }
        if already_in_event_path {
            if should_skip_handler_execution {
                self.register_in_history(event.clone());
                self.notify_find_waiters(event.clone());
                self.settle_skipped_handler_execution(&event, 0);
            }
            return event;
        }
        event.set_runtime_eventbus_id(Some(self.id_ref.clone()));

        if !self.register_in_history(event.clone()) {
            panic!(
                "EventBus history limit reached: max_history_size={} and max_history_drop=false",
                self.runtime.max_history_size.unwrap_or(0)
            );
        }

        {
            let mut inner = event.inner.lock();
            inner.event_pending_bus_count += 1;
            inner.event_path.push(bus_label);
            if track_child && inner.event_parent_id.is_none() {
                CURRENT_EVENT_ID.with(|id| {
                    let current_parent = id.borrow().clone();
                    if current_parent.as_deref() != Some(inner.event_id.as_str()) {
                        inner.event_parent_id = current_parent;
                    }
                });
            }
            if track_child
                && emitted_from_active_handler
                && inner.event_emitted_by_handler_id.is_none()
            {
                CURRENT_HANDLER_ID.with(|id| {
                    inner.event_emitted_by_handler_id = id.borrow().clone();
                });
            }
        }

        if track_child && emitted_from_active_handler {
            let emitted_child_id = event.inner.lock().event_id.clone();
            CURRENT_EVENT_ID.with(|current_event_id| {
                CURRENT_HANDLER_ID.with(|current_handler_id| {
                    let Some(parent_id) = current_event_id.borrow().clone() else {
                        return;
                    };
                    let Some(handler_id) = current_handler_id.borrow().clone() else {
                        return;
                    };
                    let Some(parent_event) = self.runtime.events.lock().get(&parent_id).cloned()
                    else {
                        return;
                    };
                    let mut parent_inner = parent_event.inner.lock();
                    if let Some(result) = parent_inner.event_results.get_mut(&handler_id) {
                        if !result.event_children.contains(&emitted_child_id) {
                            result.event_children.push(emitted_child_id.clone());
                        }
                    }
                });
            });
        }

        if should_skip_handler_execution {
            self.notify_find_waiters(event.clone());
            self.settle_skipped_handler_execution(&event, 1);
            return event;
        }

        {
            let mut queue = self.runtime.queue.lock();
            if queue_jump {
                queue.push_front(event.clone());
            } else {
                queue.push_back(event.clone());
            }
        }

        self.notify_find_waiters(event.clone());
        let event_concurrency = event
            .inner
            .lock()
            .event_concurrency
            .unwrap_or(self.event_concurrency);
        let defer_for_handler_queue_jump = emitted_from_active_handler
            || (emitted_inside_active_handler
                && self.current_handler_event_shares_event_lock_with(&event));
        if defer_for_handler_queue_jump && event_concurrency != EventConcurrencyMode::Parallel {
            self.pause_current_handler_queue_jumps();
        }
        if event_concurrency == EventConcurrencyMode::Parallel {
            self.start_parallel_event_task_from_queue(event.clone());
        }
        if !defer_for_handler_queue_jump || event_concurrency == EventConcurrencyMode::Parallel {
            self.ensure_loop_started();
            self.runtime.queue_notify.notify(1);
        }
        event
    }

    fn pause_current_handler_queue_jumps(&self) {
        let Some((current_event_id, current_handler_id)) = Self::current_handler_context() else {
            return;
        };
        for bus in Self::live_instances() {
            let Some(parent_event) = bus.runtime.events.lock().get(&current_event_id).cloned()
            else {
                continue;
            };
            let mut parent_inner = parent_event.inner.lock();
            let Some(result) = parent_inner.event_results.get_mut(&current_handler_id) else {
                continue;
            };
            result.ensure_queue_jump_pause(self.locks.clone());
            break;
        }
    }

    pub fn queue_jump_if_waited(event: Arc<BaseEvent>) {
        Self::queue_jump_if_waited_blocking(event, false);
    }

    pub fn queue_jump_if_waited_from_handler(event: Arc<BaseEvent>) {
        Self::queue_jump_if_waited_blocking(event, false);
    }

    fn queue_jump_if_waited_blocking(event: Arc<BaseEvent>, dedicated_thread: bool) {
        let (tx, rx) = std_mpsc::channel();
        let initiating_context = Self::mark_blocks_parent_completion_if_awaited(event.clone());
        let job = move || {
            block_on(Self::queue_jump_if_waited_async_with_context(
                event,
                initiating_context,
            ));
            let _ = tx.send(());
        };
        if dedicated_thread {
            thread::spawn(job);
        } else {
            Self::queue_jump_executor().spawn(job);
        }
        let _ = rx.recv();
    }

    pub async fn queue_jump_if_waited_async(event: Arc<BaseEvent>) {
        let initiating_context = Self::mark_blocks_parent_completion_if_awaited(event.clone());
        Self::queue_jump_if_waited_async_with_context(event, initiating_context).await;
    }

    pub(crate) fn spawn_queue_jump_if_waited_with_context(
        event: Arc<BaseEvent>,
        initiating_context: Option<(String, String, String)>,
    ) {
        Self::queue_jump_executor().spawn(move || {
            block_on(Self::queue_jump_if_waited_async_with_context(
                event,
                initiating_context,
            ));
        });
    }

    pub async fn queue_jump_if_waited_async_with_context(
        event: Arc<BaseEvent>,
        initiating_context: Option<(String, String, String)>,
    ) {
        let initiating_bus = initiating_context
            .as_ref()
            .and_then(|(bus_id, _, _)| Self::live_instance_by_id(bus_id))
            .or_else(|| Self::event_bus_for_event(&event));
        let initiating_event_lock = initiating_bus
            .as_ref()
            .and_then(|bus| bus.locks.get_lock_for_event(bus, &event));

        let event_path = event.inner.lock().event_path.clone();
        if event_path.is_empty() {
            return;
        }

        let live_buses: Vec<Arc<EventBus>> = {
            let mut instances = ALL_INSTANCES.get_or_init(|| Mutex::new(Vec::new())).lock();
            instances.retain(|entry| entry.upgrade().is_some());
            instances.iter().filter_map(Weak::upgrade).collect()
        };

        let mut ordered = Vec::new();
        for label in event_path {
            for bus in &live_buses {
                if bus.label() == label
                    && !ordered
                        .iter()
                        .any(|entry: &Arc<EventBus>| Arc::ptr_eq(entry, bus))
                {
                    ordered.push(bus.clone());
                }
            }
        }

        for bus in ordered {
            let should_bypass_event_lock = initiating_bus
                .as_ref()
                .is_some_and(|initiating| Arc::ptr_eq(&bus, initiating))
                || initiating_event_lock
                    .as_ref()
                    .is_some_and(|initiating_lock| {
                        bus.locks
                            .get_lock_for_event(&bus, &event)
                            .as_ref()
                            .is_some_and(|bus_lock| Arc::ptr_eq(bus_lock, initiating_lock))
                    });
            bus.process_event_immediately_if_queued(event.clone(), should_bypass_event_lock)
                .await;
        }
    }

    pub fn mark_blocks_parent_completion_if_awaited(
        event: Arc<BaseEvent>,
    ) -> Option<(String, String, String)> {
        let context = CURRENT_BUS_ID.with(|bus_id| {
            CURRENT_EVENT_ID.with(|event_id| {
                CURRENT_HANDLER_ID.with(|handler_id| {
                    Some((
                        bus_id.borrow().clone()?,
                        event_id.borrow().clone()?,
                        handler_id.borrow().clone()?,
                    ))
                })
            })
        });
        let (current_bus_id, current_event_id, current_handler_id) = context?;

        {
            let mut inner = event.inner.lock();
            if inner.event_parent_id.as_deref() == Some(current_event_id.as_str())
                && inner.event_emitted_by_handler_id.as_deref() == Some(current_handler_id.as_str())
                && inner.event_id != current_event_id
            {
                let is_linked_child = ALL_INSTANCES
                    .get_or_init(|| Mutex::new(Vec::new()))
                    .lock()
                    .iter()
                    .filter_map(Weak::upgrade)
                    .filter_map(|bus| bus.runtime.events.lock().get(&current_event_id).cloned())
                    .any(|parent| {
                        parent
                            .inner
                            .lock()
                            .event_results
                            .get(&current_handler_id)
                            .is_some_and(|result| result.event_children.contains(&inner.event_id))
                    });
                if is_linked_child {
                    inner.event_blocks_parent_completion = true;
                }
            }
        }

        Some((current_bus_id, current_event_id, current_handler_id))
    }

    async fn process_event_immediately_if_queued(
        &self,
        event: Arc<BaseEvent>,
        bypass_event_lock: bool,
    ) {
        let event_id = event.inner.lock().event_id.clone();
        if self.has_completed_on_bus(&event) {
            return;
        }
        if event
            .inner
            .lock()
            .event_results
            .values()
            .any(|result| result.handler.eventbus_id == self.id)
        {
            return;
        }

        let mut runloop_pause = None;
        let removed = {
            let mut queue = self.runtime.queue.lock();
            if let Some(index) = queue
                .iter()
                .position(|queued| queued.inner.lock().event_id == event_id)
            {
                runloop_pause = Some(self.locks.request_runloop_pause());
                queue.remove(index);
                self.runtime
                    .active_event_ids
                    .lock()
                    .insert(event_id.clone());
                true
            } else {
                false
            }
        };
        if !removed && !bypass_event_lock {
            return;
        }
        if !removed && event.should_skip_handler_execution() {
            return;
        }
        if runloop_pause.is_none() {
            runloop_pause = Some(self.locks.request_runloop_pause());
        }
        if !removed {
            self.runtime
                .active_event_ids
                .lock()
                .insert(event_id.clone());
        }
        if bypass_event_lock {
            {
                let mut processing = self.runtime.processing_event_ids.lock();
                if !processing.insert(event_id.clone()) {
                    self.runtime.active_event_ids.lock().remove(&event_id);
                    return;
                }
            }
            self.process_event_inner(event).await;
            self.runtime.processing_event_ids.lock().remove(&event_id);
        } else {
            self.process_event(event).await;
        }
        self.runtime.active_event_ids.lock().remove(&event_id);
        drop(runloop_pause);
        self.runtime.queue_notify.notify(usize::MAX);
    }

    fn register_in_history(&self, event: Arc<BaseEvent>) -> bool {
        let event_id = event.inner.lock().event_id.clone();
        self.trim_event_history(false);

        if self.runtime.events.lock().contains_key(&event_id) {
            self.runtime
                .events
                .lock()
                .insert(event_id.clone(), event.clone());
            let mut history_order = self.runtime.history_order.lock();
            let mut seen = false;
            history_order.retain(|id| {
                if id != &event_id {
                    return true;
                }
                if seen {
                    return false;
                }
                seen = true;
                true
            });
            if !seen {
                history_order.push_back(event_id);
            }
            return true;
        }

        if let Some(max_size) = self.runtime.max_history_size {
            if max_size > 0 {
                let current_size = self.runtime.history_order.lock().len();
                if current_size >= max_size && !self.runtime.max_history_drop {
                    return false;
                }
                if self.runtime.max_history_drop {
                    self.trim_history_to_capacity(max_size, true);
                }
            }
        }

        if let Some(snapshot) = Self::capture_context_snapshot() {
            let mut contexts = self.runtime.event_contexts.lock();
            if contexts.insert(event_id.clone(), snapshot).is_none() {
                self.runtime
                    .event_context_count
                    .fetch_add(1, Ordering::SeqCst);
            }
        }
        self.runtime
            .events
            .lock()
            .insert(event_id.clone(), event.clone());
        self.runtime.history_order.lock().push_back(event_id);
        self.trim_event_history(true);
        true
    }

    fn now_epoch_seconds() -> f64 {
        let now = chrono::Utc::now();
        now.timestamp() as f64 + (now.timestamp_subsec_nanos() as f64 / 1_000_000_000.0)
    }

    fn completed_at_epoch_seconds(event: &BaseEventData) -> Option<f64> {
        if event.event_status != EventStatus::Completed {
            return None;
        }
        let completed_at = event.event_completed_at.as_ref()?;
        let completed_at = chrono::DateTime::parse_from_rfc3339(&completed_at).ok()?;
        Some(
            completed_at.timestamp() as f64
                + (completed_at.timestamp_subsec_nanos() as f64 / 1_000_000_000.0),
        )
    }

    fn completed_event_age_seconds(event: &Arc<BaseEvent>) -> Option<f64> {
        let inner = event.inner.lock();
        let completed_at = Self::completed_at_epoch_seconds(&inner)?;
        Some((Self::now_epoch_seconds() - completed_at).max(0.0))
    }

    fn resolve_event_ttl(&self, event: &BaseEventData) -> Option<f64> {
        event.event_ttl.or(self.event_ttl)
    }

    fn completed_event_expired_for_history(&self, event: &Arc<BaseEvent>) -> bool {
        let inner = event.inner.lock();
        let Some(event_ttl) = self.resolve_event_ttl(&inner).filter(|ttl| *ttl >= 0.0) else {
            return false;
        };
        let Some(completed_at) = Self::completed_at_epoch_seconds(&inner) else {
            return false;
        };
        event_ttl == 0.0 || Self::now_epoch_seconds() >= completed_at + event_ttl
    }

    fn resolve_event_result_ttl(&self, event: &BaseEventData, result: &EventResult) -> Option<f64> {
        result
            .handler
            .handler_result_ttl
            .or(event.event_result_ttl)
            .or(self.event_result_ttl)
    }

    fn untrack_event_ttl_deadline(&self, event_id: &str) {
        self.runtime
            .ttl_deadlines_by_event_id
            .lock()
            .remove(event_id);
    }

    fn earliest_ttl_deadline(&self, event: &Arc<BaseEvent>) -> Option<f64> {
        let inner = event.inner.lock();
        let completed_at = Self::completed_at_epoch_seconds(&inner)?;
        let mut best_ttl = self.resolve_event_ttl(&inner).filter(|ttl| *ttl >= 0.0);
        for result in inner.event_results.values() {
            if result.handler.eventbus_id != self.id {
                continue;
            }
            let Some(result_ttl) = self
                .resolve_event_result_ttl(&inner, result)
                .filter(|ttl| *ttl >= 0.0)
            else {
                continue;
            };
            best_ttl = Some(best_ttl.map_or(result_ttl, |best| best.min(result_ttl)));
        }
        let best_ttl = best_ttl?;
        if best_ttl == 0.0 {
            return Some(0.0);
        }
        Some(completed_at + best_ttl)
    }

    fn track_event_ttl_deadline(&self, event: &Arc<BaseEvent>) {
        let event_id = event.inner.lock().event_id.clone();
        let Some(deadline) = self.earliest_ttl_deadline(event) else {
            self.untrack_event_ttl_deadline(&event_id);
            return;
        };
        {
            let mut deadlines = self.runtime.ttl_deadlines_by_event_id.lock();
            if deadlines
                .get(&event_id)
                .is_some_and(|current| *current == deadline)
            {
                return;
            }
            deadlines.insert(event_id.clone(), deadline);
        }
        let mut queue = self.runtime.ttl_deadline_queue.lock();
        let index = queue
            .iter()
            .position(|entry| entry.expires_at > deadline)
            .unwrap_or(queue.len());
        queue.insert(
            index,
            TTLDeadlineEntry {
                expires_at: deadline,
                event_id,
            },
        );
    }

    fn track_completed_event_ttl_deadlines_across_buses(&self, event: &Arc<BaseEvent>) {
        let (event_id, is_completed, event_path_len) = {
            let inner = event.inner.lock();
            (
                inner.event_id.clone(),
                inner.event_status == EventStatus::Completed,
                inner.event_path.len(),
            )
        };
        if !is_completed {
            return;
        }
        if event_path_len <= 1 {
            if self
                .runtime
                .events
                .lock()
                .get(&event_id)
                .is_some_and(|candidate| Arc::ptr_eq(candidate, event))
            {
                self.track_event_ttl_deadline(event);
            }
            return;
        }
        for bus in Self::live_instances() {
            if bus
                .runtime
                .events
                .lock()
                .get(&event_id)
                .is_some_and(|candidate| Arc::ptr_eq(candidate, event))
            {
                bus.track_event_ttl_deadline(event);
            }
        }
    }

    fn retrack_completed_history_ttl_deadlines(&self) {
        let events: Vec<Arc<BaseEvent>> = if self.has_active_ttl_backfill_policy() {
            self.runtime.events.lock().values().cloned().collect()
        } else {
            let mut events_by_id: HashMap<String, Arc<BaseEvent>> = HashMap::new();
            let event_ids: Vec<String> = self
                .runtime
                .ttl_deadlines_by_event_id
                .lock()
                .keys()
                .cloned()
                .collect();
            for event_id in event_ids {
                if let Some(event) = self.runtime.events.lock().get(&event_id).cloned() {
                    events_by_id.insert(event_id, event);
                }
            }
            // Indexed events cover normal TTL shortening; this scan catches
            // completed events whose own TTL fields changed from unset/-1 after
            // completion, before any deadline entry existed. Clone the history
            // entries first so the bus history lock is never held while taking
            // event locks; completion paths take those locks in the other order.
            let history_events: Vec<Arc<BaseEvent>> =
                self.runtime.events.lock().values().cloned().collect();
            for event in history_events {
                let inner = event.inner.lock();
                if events_by_id.contains_key(&inner.event_id)
                    || inner.event_status != EventStatus::Completed
                {
                    continue;
                }
                let has_event_ttl_override = inner.event_ttl.is_some_and(|ttl| ttl >= 0.0);
                let has_result_ttl_override = inner.event_result_ttl.is_some_and(|ttl| ttl >= 0.0);
                if has_event_ttl_override || has_result_ttl_override {
                    events_by_id.insert(inner.event_id.clone(), event.clone());
                }
            }
            events_by_id.into_values().collect()
        };
        for event in events {
            let is_completed = event.inner.lock().event_status == EventStatus::Completed;
            if is_completed {
                self.track_event_ttl_deadline(&event);
            }
        }
    }

    fn has_active_ttl_backfill_policy(&self) -> bool {
        if self.event_ttl.is_some_and(|ttl| ttl >= 0.0)
            || self.event_result_ttl.is_some_and(|ttl| ttl >= 0.0)
        {
            return true;
        }
        self.handlers.lock().values().any(|handlers| {
            handlers
                .iter()
                .any(|handler| handler.handler_result_ttl.is_some_and(|ttl| ttl >= 0.0))
        })
    }

    fn trim_event_history(&self, include_ttl: bool) {
        if let Some(max_size) = self.runtime.max_history_size {
            self.trim_history_to_capacity(max_size, false);
        }
        if !include_ttl {
            return;
        }

        // TTL cleanup is tied to normal bus cleanup points rather than a timer,
        // but it must only remove due candidates. Retracking first keeps runtime
        // TTL edits observable without storing serialized TTL state on the event.
        self.retrack_completed_history_ttl_deadlines();
        let now = Self::now_epoch_seconds();
        loop {
            let Some(entry) = ({
                let mut queue = self.runtime.ttl_deadline_queue.lock();
                if queue.front().is_none_or(|entry| entry.expires_at > now) {
                    None
                } else {
                    queue.pop_front()
                }
            }) else {
                break;
            };
            if self
                .runtime
                .ttl_deadlines_by_event_id
                .lock()
                .get(&entry.event_id)
                .is_none_or(|current| *current != entry.expires_at)
            {
                continue;
            }
            let Some(event) = self.runtime.events.lock().get(&entry.event_id).cloned() else {
                self.untrack_event_ttl_deadline(&entry.event_id);
                continue;
            };
            let Some(age_seconds) = Self::completed_event_age_seconds(&event) else {
                self.untrack_event_ttl_deadline(&entry.event_id);
                continue;
            };
            let mut inner = event.inner.lock();
            let event_id = inner.event_id.clone();
            let result_ids: Vec<String> = inner
                .event_results
                .iter()
                .filter_map(|(result_id, result)| {
                    if result.handler.eventbus_id != self.id {
                        return None;
                    }
                    let result_ttl = self.resolve_event_result_ttl(&inner, result)?;
                    (result_ttl >= 0.0 && age_seconds >= result_ttl).then(|| result_id.clone())
                })
                .collect();
            for result_id in result_ids {
                inner.event_results.remove(&result_id);
                inner.event_result_order.retain(|id| id != &result_id);
            }
            let event_ttl = self.resolve_event_ttl(&inner);
            drop(inner);
            if event_ttl.is_none_or(|ttl| ttl < 0.0 || age_seconds < ttl) {
                self.track_event_ttl_deadline(&event);
                continue;
            }
            self.runtime.events.lock().remove(&event_id);
            self.untrack_event_ttl_deadline(&event_id);
            self.runtime
                .history_order
                .lock()
                .retain(|id| id != &event_id);
            if self
                .runtime
                .event_contexts
                .lock()
                .remove(&event_id)
                .is_some()
            {
                self.runtime
                    .event_context_count
                    .fetch_sub(1, Ordering::SeqCst);
            }
        }
    }

    fn trim_history_to_capacity(&self, max_size: usize, include_equal: bool) {
        if max_size == 0 {
            return;
        }
        loop {
            let current_len = self.runtime.history_order.lock().len();
            if current_len < max_size || (!include_equal && current_len == max_size) {
                break;
            }
            let Some(oldest) = self.runtime.history_order.lock().front().cloned() else {
                break;
            };
            let candidate = self.runtime.events.lock().get(&oldest).cloned();
            let is_active = candidate
                .as_ref()
                .map(|event| {
                    let status = event.inner.lock().event_status;
                    status == EventStatus::Pending || status == EventStatus::Started
                })
                .unwrap_or(false);
            if is_active {
                break;
            }
            self.runtime.history_order.lock().pop_front();
            self.runtime.events.lock().remove(&oldest);
            self.untrack_event_ttl_deadline(&oldest);
            if self.runtime.event_contexts.lock().remove(&oldest).is_some() {
                self.runtime
                    .event_context_count
                    .fetch_sub(1, Ordering::SeqCst);
            }
        }
    }

    fn context_for_event(&self, event: &Arc<BaseEvent>) -> Option<dcontext::ContextSnapshot> {
        if self.runtime.event_context_count.load(Ordering::SeqCst) == 0 {
            return None;
        }
        let event_id = event.inner.lock().event_id.clone();
        self.runtime.event_contexts.lock().get(&event_id).cloned()
    }

    fn capture_context_snapshot() -> Option<dcontext::ContextSnapshot> {
        (!dcontext::scope_chain().is_empty()).then(dcontext::snapshot)
    }

    pub async fn find(
        &self,
        pattern: &str,
        past: bool,
        future: Option<f64>,
        child_of: Option<Arc<BaseEvent>>,
    ) -> Option<Arc<BaseEvent>> {
        self.find_with_options(
            pattern,
            FindOptions {
                past,
                past_window: None,
                future,
                child_of,
                where_filter: None,
                where_predicate: None,
            },
        )
        .await
    }

    pub async fn find_with_options(
        &self,
        pattern: &str,
        options: FindOptions,
    ) -> Option<Arc<BaseEvent>> {
        self.raise_if_terminal_destroyed();
        let child_of_event_id = options
            .child_of
            .as_ref()
            .map(|event| event.inner.lock().event_id.clone());

        let mut waiter_id: Option<u64> = None;
        let mut waiter_rx: Option<std_mpsc::Receiver<Option<Arc<BaseEvent>>>> = None;

        if options.future.is_some() {
            let (tx, rx) = std_mpsc::channel();
            let id = {
                let mut next = self.runtime.next_waiter_id.lock();
                *next += 1;
                *next
            };
            self.runtime.find_waiters.lock().push(FindWaiter {
                id,
                pattern: pattern.to_string(),
                child_of_event_id: child_of_event_id.clone(),
                where_filter: options.where_filter.clone(),
                where_predicate: options.where_predicate.clone(),
                sender: tx,
            });
            waiter_id = Some(id);
            waiter_rx = Some(rx);

            if options.past {
                if let Some(matched) = self.find_in_history(
                    pattern,
                    child_of_event_id.as_deref(),
                    options.where_filter.as_ref(),
                    options.where_predicate.as_ref(),
                    options.past_window,
                ) {
                    self.runtime
                        .find_waiters
                        .lock()
                        .retain(|waiter| waiter.id != id);
                    return Some(matched);
                }
            }
        } else if options.past {
            return self.find_in_history(
                pattern,
                child_of_event_id.as_deref(),
                options.where_filter.as_ref(),
                options.where_predicate.as_ref(),
                options.past_window,
            );
        }

        let timeout = options.future?;
        let result = waiter_rx.and_then(|rx| {
            rx.recv_timeout(Duration::from_secs_f64(timeout))
                .ok()
                .flatten()
        });

        if let Some(id) = waiter_id {
            self.runtime
                .find_waiters
                .lock()
                .retain(|waiter| waiter.id != id);
        }

        result
    }

    pub async fn filter(
        &self,
        pattern: &str,
        past: bool,
        future: Option<f64>,
        child_of: Option<Arc<BaseEvent>>,
        limit: Option<usize>,
    ) -> Vec<Arc<BaseEvent>> {
        self.filter_with_options(
            pattern,
            FilterOptions {
                past,
                future,
                child_of,
                limit,
                ..FilterOptions::default()
            },
        )
        .await
    }

    pub async fn filter_with_options(
        &self,
        pattern: &str,
        options: FilterOptions,
    ) -> Vec<Arc<BaseEvent>> {
        self.raise_if_terminal_destroyed();
        if !options.past && options.future.is_none() {
            return Vec::new();
        }
        if options.limit == Some(0) {
            return Vec::new();
        }

        let child_of_event_id = options
            .child_of
            .as_ref()
            .map(|event| event.inner.lock().event_id.clone());

        let mut results = Vec::new();
        let mut waiter_id: Option<u64> = None;
        let mut waiter_rx: Option<std_mpsc::Receiver<Option<Arc<BaseEvent>>>> = None;

        if options.future.is_some() {
            let (tx, rx) = std_mpsc::channel();
            let id = {
                let mut next = self.runtime.next_waiter_id.lock();
                *next += 1;
                *next
            };
            self.runtime.find_waiters.lock().push(FindWaiter {
                id,
                pattern: pattern.to_string(),
                child_of_event_id: child_of_event_id.clone(),
                where_filter: options.where_filter.clone(),
                where_predicate: options.where_predicate.clone(),
                sender: tx,
            });
            waiter_id = Some(id);
            waiter_rx = Some(rx);
        }

        if options.past {
            results.extend(self.filter_in_history(
                pattern,
                child_of_event_id.as_deref(),
                options.where_filter.as_ref(),
                options.where_predicate.as_ref(),
                options.past_window,
                options.limit,
            ));
            if options.limit.is_some_and(|limit| results.len() >= limit) {
                if let Some(id) = waiter_id {
                    self.runtime
                        .find_waiters
                        .lock()
                        .retain(|waiter| waiter.id != id);
                }
                return results;
            }
        }

        let Some(timeout) = options.future else {
            return results;
        };

        if let Some(future_match) = waiter_rx.and_then(|rx| {
            rx.recv_timeout(Duration::from_secs_f64(timeout))
                .ok()
                .flatten()
        }) {
            results.push(future_match);
        }

        if let Some(id) = waiter_id {
            self.runtime
                .find_waiters
                .lock()
                .retain(|waiter| waiter.id != id);
        }

        results
    }

    fn find_in_history(
        &self,
        pattern: &str,
        child_of_event_id: Option<&str>,
        where_filter: Option<&HashMap<String, Value>>,
        where_predicate: Option<&FindPredicate>,
        past_window: Option<f64>,
    ) -> Option<Arc<BaseEvent>> {
        let history = self.runtime.history_order.lock().clone();
        for event_id in history.iter().rev() {
            let Some(event) = self.runtime.events.lock().get(event_id).cloned() else {
                continue;
            };
            if !Self::is_within_past_window(&event, past_window) {
                continue;
            }
            if !self.matches_pattern(&event, pattern) {
                continue;
            }
            if let Some(parent_id) = child_of_event_id {
                let event_id = event.inner.lock().event_id.clone();
                if !self.event_is_child_of_ids(&event_id, parent_id) {
                    continue;
                }
            }
            if !Self::matches_where_filter(&event, where_filter) {
                continue;
            }
            if !Self::matches_where_predicate(&event, where_predicate) {
                continue;
            }
            return Some(event);
        }
        None
    }

    fn filter_in_history(
        &self,
        pattern: &str,
        child_of_event_id: Option<&str>,
        where_filter: Option<&HashMap<String, Value>>,
        where_predicate: Option<&FindPredicate>,
        past_window: Option<f64>,
        limit: Option<usize>,
    ) -> Vec<Arc<BaseEvent>> {
        let mut results = Vec::new();
        let history = self.runtime.history_order.lock().clone();
        for event_id in history.iter().rev() {
            let Some(event) = self.runtime.events.lock().get(event_id).cloned() else {
                continue;
            };
            if !Self::is_within_past_window(&event, past_window) {
                continue;
            }
            if !self.matches_pattern(&event, pattern) {
                continue;
            }
            if let Some(parent_id) = child_of_event_id {
                let event_id = event.inner.lock().event_id.clone();
                if !self.event_is_child_of_ids(&event_id, parent_id) {
                    continue;
                }
            }
            if !Self::matches_where_filter(&event, where_filter) {
                continue;
            }
            if !Self::matches_where_predicate(&event, where_predicate) {
                continue;
            }
            results.push(event);
            if limit.is_some_and(|limit| results.len() >= limit) {
                break;
            }
        }
        results
    }

    fn is_within_past_window(event: &Arc<BaseEvent>, past_window: Option<f64>) -> bool {
        let Some(past_window) = past_window else {
            return true;
        };
        let created_at = event.inner.lock().event_created_at.clone();
        let Ok(created_at) = chrono::DateTime::parse_from_rfc3339(&created_at) else {
            return false;
        };
        let age = chrono::Utc::now().signed_duration_since(created_at.with_timezone(&chrono::Utc));
        age.num_nanoseconds()
            .map(|nanos| (nanos as f64) / 1_000_000_000.0 <= past_window)
            .unwrap_or(false)
    }

    fn matches_pattern(&self, event: &Arc<BaseEvent>, pattern: &str) -> bool {
        if pattern == "*" {
            return true;
        }
        event.inner.lock().event_type == pattern
    }

    fn notify_find_waiters(&self, event: Arc<BaseEvent>) {
        let event_id = event.inner.lock().event_id.clone();
        let mut matched_waiter_ids = Vec::new();
        let mut matched_senders = Vec::new();

        {
            let waiters = self.runtime.find_waiters.lock();
            for waiter in waiters.iter() {
                if !self.matches_pattern(&event, &waiter.pattern) {
                    continue;
                }
                if let Some(parent_id) = waiter.child_of_event_id.as_deref() {
                    if !self.event_is_child_of_ids(&event_id, parent_id) {
                        continue;
                    }
                }
                if !Self::matches_where_filter(&event, waiter.where_filter.as_ref()) {
                    continue;
                }
                if !Self::matches_where_predicate(&event, waiter.where_predicate.as_ref()) {
                    continue;
                }
                matched_waiter_ids.push(waiter.id);
                matched_senders.push(waiter.sender.clone());
            }
        }

        if !matched_waiter_ids.is_empty() {
            self.runtime
                .find_waiters
                .lock()
                .retain(|waiter| !matched_waiter_ids.contains(&waiter.id));
            for sender in matched_senders {
                let _ = sender.send(Some(event.clone()));
            }
        }
    }

    fn matches_where_filter(
        event: &Arc<BaseEvent>,
        where_filter: Option<&HashMap<String, Value>>,
    ) -> bool {
        let Some(where_filter) = where_filter else {
            return true;
        };
        if where_filter.is_empty() {
            return true;
        }

        let Value::Object(record) = event.to_json_value() else {
            return false;
        };
        where_filter
            .iter()
            .all(|(key, expected)| record.get(key) == Some(expected))
    }

    fn matches_where_predicate(
        event: &Arc<BaseEvent>,
        where_predicate: Option<&FindPredicate>,
    ) -> bool {
        where_predicate
            .map(|predicate| predicate(event))
            .unwrap_or(true)
    }

    pub fn event_is_child_of(
        &self,
        child_event: &Arc<BaseEvent>,
        parent_event: &Arc<BaseEvent>,
    ) -> bool {
        let child_id = child_event.inner.lock().event_id.clone();
        let parent_id = parent_event.inner.lock().event_id.clone();
        self.event_is_child_of_ids(&child_id, &parent_id)
    }

    pub fn event_is_parent_of(
        &self,
        parent_event: &Arc<BaseEvent>,
        child_event: &Arc<BaseEvent>,
    ) -> bool {
        self.event_is_child_of(child_event, parent_event)
    }

    fn event_is_child_of_ids(&self, child_event_id: &str, parent_event_id: &str) -> bool {
        if child_event_id == parent_event_id {
            return false;
        }

        let mut visited = HashSet::new();
        let mut current_id = child_event_id.to_string();
        loop {
            if !visited.insert(current_id.clone()) {
                return false;
            }
            let Some(current_event) = self.runtime.events.lock().get(&current_id).cloned() else {
                return false;
            };
            let current_parent = current_event.inner.lock().event_parent_id.clone();
            let Some(current_parent_id) = current_parent else {
                return false;
            };
            if current_parent_id == parent_event_id {
                return true;
            }
            current_id = current_parent_id;
        }
    }

    pub async fn wait_until_idle(&self, timeout: Option<f64>) -> bool {
        let start = Instant::now();
        loop {
            if self.is_idle_and_queue_empty() {
                return true;
            }

            if let Some(timeout) = timeout {
                if start.elapsed() > Duration::from_secs_f64(timeout) {
                    return false;
                }
            }
            Delay::new(Duration::from_millis(5)).await;
        }
    }

    pub(crate) async fn wait_until_event_idle(event: Arc<BaseEvent>, timeout: Option<f64>) -> bool {
        let event_id = event.inner.lock().event_id.clone();
        let start = Instant::now();
        loop {
            let event_path = event.inner.lock().event_path.clone();
            let live_buses = Self::live_instances();
            let mut attached_buses = Vec::new();
            for label in &event_path {
                for bus in &live_buses {
                    if bus.label() == *label
                        && !attached_buses
                            .iter()
                            .any(|entry: &Arc<EventBus>| Arc::ptr_eq(entry, bus))
                    {
                        attached_buses.push(bus.clone());
                    }
                }
            }

            let event_is_idle = attached_buses.iter().all(|bus| {
                let queued = bus
                    .runtime
                    .queue
                    .lock()
                    .iter()
                    .any(|queued| queued.inner.lock().event_id == event_id);
                let active = bus.runtime.active_event_ids.lock().contains(&event_id);
                let processing = bus.runtime.processing_event_ids.lock().contains(&event_id);
                !queued && !active && !processing
            });
            if event_is_idle {
                return true;
            }

            if let Some(timeout) = timeout {
                if start.elapsed() > Duration::from_secs_f64(timeout) {
                    return false;
                }
            }
            Delay::new(Duration::from_millis(1)).await;
        }
    }

    async fn process_event(&self, event: Arc<BaseEvent>) {
        if self.has_completed_on_bus(&event) {
            return;
        }
        let event_id = event.inner.lock().event_id.clone();
        let mode = event
            .inner
            .lock()
            .event_concurrency
            .unwrap_or(self.event_concurrency);
        match mode {
            EventConcurrencyMode::GlobalSerial => {
                let _guard = GLOBAL_SERIAL_LOCK
                    .get_or_init(|| Arc::new(ReentrantLock::default()))
                    .lock();
                {
                    let mut processing = self.runtime.processing_event_ids.lock();
                    if !processing.insert(event_id.clone()) {
                        return;
                    }
                }
                self.process_event_inner(event).await;
            }
            EventConcurrencyMode::BusSerial => {
                let _guard = self.bus_serial_lock.lock();
                {
                    let mut processing = self.runtime.processing_event_ids.lock();
                    if !processing.insert(event_id.clone()) {
                        return;
                    }
                }
                self.process_event_inner(event).await;
            }
            EventConcurrencyMode::Parallel => {
                {
                    let mut processing = self.runtime.processing_event_ids.lock();
                    if !processing.insert(event_id.clone()) {
                        return;
                    }
                }
                self.process_event_inner(event).await;
            }
        }
        self.runtime.processing_event_ids.lock().remove(&event_id);
    }

    async fn process_event_inner(&self, event: Arc<BaseEvent>) {
        Self::validate_event_execution_timeout_fields(&event);
        event.mark_started();
        let started_at = Instant::now();
        let mut event_timeout = event.inner.lock().event_timeout.or(self.event_timeout);
        if event_timeout.is_some_and(|timeout| timeout <= 0.0) {
            event_timeout = None;
        }
        let (event_type_for_slow_warning, event_slow_timeout) = {
            let inner = event.inner.lock();
            (
                inner.event_type.clone(),
                inner.event_slow_timeout.or(self.event_slow_timeout),
            )
        };
        if let Some(event_slow_timeout) = event_slow_timeout {
            if event_slow_timeout > 0.0
                && event_timeout.is_none_or(|timeout| event_slow_timeout < timeout)
            {
                let event_for_warning = event.clone();
                let bus_name_for_warning = self.name.clone();
                Self::slow_warning_scheduler().schedule(
                    Duration::from_secs_f64(event_slow_timeout),
                    move || {
                    let still_running =
                        event_for_warning.inner.lock().event_status != EventStatus::Completed;
                    if still_running {
                        let running_handler_count = event_for_warning
                            .inner
                            .lock()
                            .event_results
                            .values()
                            .filter(|result| result.status == EventResultStatus::Started)
                            .count();
                        eprintln!(
                        "Slow event processing: {bus_name_for_warning}.on({event_type_for_slow_warning}, {running_handler_count} handlers) still running after {event_slow_timeout:.3}s"
                    );
                    }
                    },
                );
            }
        }

        let event_type = event.inner.lock().event_type.clone();
        let mut handlers = self
            .handlers
            .lock()
            .get(&event_type)
            .cloned()
            .unwrap_or_default();
        handlers.extend(self.handlers.lock().get("*").cloned().unwrap_or_default());

        let handler_concurrency = event
            .inner
            .lock()
            .event_handler_concurrency
            .unwrap_or(self.event_handler_concurrency);
        let handler_completion = self.handler_completion_for_event(&event);

        self.create_pending_handler_results(&event, &handlers, event_timeout);

        match handler_concurrency {
            EventHandlerConcurrencyMode::Serial => {
                for (index, handler) in handlers.iter().cloned().enumerate() {
                    let timed_out = self
                        .run_handler_with_context(
                            event.clone(),
                            handler,
                            started_at,
                            event_timeout,
                            false,
                        )
                        .await;
                    if timed_out {
                        self.cancel_remaining_timeout_results(&event, &handlers[index + 1..]);
                        break;
                    }
                    if self.handler_completion_for_event(&event)
                        == EventHandlerCompletionMode::First
                    {
                        let winner_id = self.winning_handler_id(&event);
                        if winner_id.is_some() {
                            self.cancel_remaining_first_mode_results(
                                &event,
                                &handlers[index + 1..],
                            );
                            break;
                        }
                    }
                }
            }
            EventHandlerConcurrencyMode::Parallel => {
                if handler_completion == EventHandlerCompletionMode::First {
                    let (tx, rx) = std_mpsc::channel();
                    let handler_count = handlers.len();
                    for handler in handlers {
                        let bus = self.clone();
                        let event_clone = event.clone();
                        let handler_id = handler.id.clone();
                        let tx = tx.clone();
                        Self::handler_executor().spawn(move || {
                            let timed_out = block_on(bus.run_handler_with_context(
                                event_clone,
                                handler,
                                started_at,
                                event_timeout,
                                true,
                            ));
                            let _ = tx.send((handler_id, timed_out));
                        });
                    }
                    drop(tx);

                    for _ in 0..handler_count {
                        let Ok((_handler_id, timed_out)) = rx.recv() else {
                            break;
                        };
                        if timed_out {
                            break;
                        }
                        if let Some(winner_id) = self.winning_handler_id(&event) {
                            self.cancel_parallel_first_mode_losers(&event, &winner_id);
                            break;
                        }
                    }
                } else {
                    let (tx, rx) = std_mpsc::channel();
                    let handler_count = handlers.len();
                    for handler in handlers {
                        let bus = self.clone();
                        let event_clone = event.clone();
                        let tx = tx.clone();
                        Self::handler_executor().spawn(move || {
                            let timed_out = block_on(bus.run_handler_with_context(
                                event_clone,
                                handler,
                                started_at,
                                event_timeout,
                                true,
                            ));
                            let _ = tx.send(timed_out);
                        });
                    }
                    drop(tx);
                    for _ in 0..handler_count {
                        let _ = rx.recv();
                    }
                }
            }
        }

        if let Some(timeout) = event_timeout {
            if started_at.elapsed() > Duration::from_secs_f64(timeout) {
                self.cancel_children(&event, &format!("parent event timed out after {timeout}s"));
            }
        }

        let should_complete = {
            let mut inner = event.inner.lock();
            inner.event_pending_bus_count = inner.event_pending_bus_count.saturating_sub(1);
            let done = inner.event_pending_bus_count == 0;
            if done && inner.event_status != EventStatus::Completed {
                inner.event_completed_at = Some(now_iso());
            }
            done
        };
        if self.runtime.max_history_size == Some(0) {
            let event_id = event.inner.lock().event_id.clone();
            self.runtime.events.lock().remove(&event_id);
            self.untrack_event_ttl_deadline(&event_id);
            if self
                .runtime
                .event_contexts
                .lock()
                .remove(&event_id)
                .is_some()
            {
                self.runtime
                    .event_context_count
                    .fetch_sub(1, Ordering::SeqCst);
            }
            self.runtime
                .history_order
                .lock()
                .retain(|id| id != &event_id);
        } else if self.runtime.max_history_size.is_some() {
            self.trim_event_history(false);
        }

        if should_complete {
            event.mark_completed();
            self.track_completed_event_ttl_deadlines_across_buses(&event);
        }
    }

    fn settle_skipped_handler_execution(
        &self,
        event: &Arc<BaseEvent>,
        pending_bus_count_decrement: usize,
    ) {
        // Events that skip handler execution still use the normal completion
        // lifecycle. The decrement is 1 only for a bus that just accepted this
        // event; already-in-path re-dispatches only refresh completion state
        // without changing pending counts.
        let should_complete = {
            let mut inner = event.inner.lock();
            inner.event_pending_bus_count = inner
                .event_pending_bus_count
                .saturating_sub(pending_bus_count_decrement);
            inner.event_pending_bus_count == 0
        };
        // No handler ran, but the event still becomes completed through the
        // same mark-completed path once every accepted bus slot has settled.
        // TTL pruning is intentionally deferred to normal trim calls so
        // completion hooks can observe the event.
        if should_complete {
            event.mark_completed();
            self.track_completed_event_ttl_deadlines_across_buses(event);
        }
    }

    fn cancel_children(&self, event: &Arc<BaseEvent>, reason: &str) {
        let results = event.inner.lock().event_results.clone();
        for result in results.values() {
            for child_id in &result.event_children {
                let child = { self.runtime.events.lock().get(child_id).cloned() };
                if let Some(child) = child {
                    if !child.inner.lock().event_blocks_parent_completion {
                        continue;
                    }
                    self.cancel_children(&child, reason);
                    for child_result in child.inner.lock().event_results.values_mut() {
                        if child_result.status == EventResultStatus::Pending {
                            child_result.status = EventResultStatus::Error;
                            child_result.error = Some(format!(
                                "EventHandlerCancelledError: Cancelled pending handler due to parent error: {reason}"
                            ));
                            child_result.completed_at = Some(now_iso());
                        } else if child_result.status == EventResultStatus::Started {
                            child_result.status = EventResultStatus::Error;
                            child_result.error = Some(format!(
                                "EventHandlerAbortedError: Aborted running handler due to parent error: {reason}"
                            ));
                            child_result.completed_at = Some(now_iso());
                        }
                    }
                    child.mark_completed();
                }
            }
        }
    }

    fn create_pending_handler_results(
        &self,
        event: &Arc<BaseEvent>,
        handlers: &[EventHandler],
        event_timeout: Option<f64>,
    ) {
        let event_id = event.inner.lock().event_id.clone();
        let pending_results: Vec<(String, EventResult)> = handlers
            .iter()
            .map(|handler| {
                (
                    handler.id.clone(),
                    EventResult::new(
                        event_id.clone(),
                        handler.clone(),
                        self.result_timeout_for_handler(event, handler, event_timeout),
                    ),
                )
            })
            .collect();
        let mut ordered_handler_ids: Vec<(String, String)> = pending_results
            .iter()
            .map(|(handler_id, result)| {
                (
                    result.handler.handler_registered_at.clone(),
                    handler_id.clone(),
                )
            })
            .collect();
        ordered_handler_ids.sort();
        let mut inner = event.inner.lock();
        let mut known_order_ids: HashSet<String> =
            inner.event_result_order.iter().cloned().collect();
        for (_, handler_id) in ordered_handler_ids {
            if known_order_ids.insert(handler_id.clone()) {
                inner.event_result_order.push(handler_id);
            }
        }
        for (handler_id, result) in pending_results {
            inner.event_results.entry(handler_id).or_insert(result);
        }
    }

    fn result_timeout_for_handler(
        &self,
        event: &Arc<BaseEvent>,
        handler: &EventHandler,
        event_timeout: Option<f64>,
    ) -> Option<f64> {
        let event_handler_timeout = event.inner.lock().event_handler_timeout;
        let mut resolved_handler_timeout = handler.handler_timeout.or(event_handler_timeout);
        if resolved_handler_timeout.is_some_and(|timeout| timeout <= 0.0) {
            resolved_handler_timeout = None;
        }
        let mut resolved_event_timeout = event_timeout;
        if resolved_event_timeout.is_some_and(|timeout| timeout <= 0.0) {
            resolved_event_timeout = None;
        }
        if resolved_handler_timeout.is_none() {
            resolved_handler_timeout = resolved_event_timeout;
        }

        match (resolved_handler_timeout, resolved_event_timeout) {
            (Some(a), Some(b)) => Some(a.min(b)),
            (Some(a), None) => Some(a),
            (None, Some(b)) => Some(b),
            (None, None) => None,
        }
    }

    fn handler_completion_for_event(&self, event: &Arc<BaseEvent>) -> EventHandlerCompletionMode {
        event
            .inner
            .lock()
            .event_handler_completion
            .unwrap_or(self.event_handler_completion)
    }

    fn winning_handler_id(&self, event: &Arc<BaseEvent>) -> Option<String> {
        let mut candidates: Vec<(String, EventResult)> = event
            .inner
            .lock()
            .event_results
            .iter()
            .filter_map(|(handler_id, result)| {
                if result.status == EventResultStatus::Completed
                    && result.error.is_none()
                    && !matches!(result.result, None | Some(Value::Null))
                    && !result
                        .result
                        .as_ref()
                        .is_some_and(BaseEvent::is_base_event_json)
                {
                    Some((handler_id.clone(), result.clone()))
                } else {
                    None
                }
            })
            .collect();
        candidates.sort_by(|left, right| {
            left.1
                .completed_at
                .cmp(&right.1.completed_at)
                .then_with(|| left.1.started_at.cmp(&right.1.started_at))
                .then_with(|| {
                    left.1
                        .handler
                        .handler_registered_at
                        .cmp(&right.1.handler.handler_registered_at)
                })
                .then_with(|| left.0.cmp(&right.0))
        });
        candidates
            .into_iter()
            .next()
            .map(|(handler_id, _)| handler_id)
    }

    fn cancel_remaining_first_mode_results(
        &self,
        event: &Arc<BaseEvent>,
        remaining_handlers: &[EventHandler],
    ) {
        let mut inner = event.inner.lock();
        for handler in remaining_handlers {
            if let Some(result) = inner.event_results.get_mut(&handler.id) {
                if result.status != EventResultStatus::Pending {
                    continue;
                }
                result.status = EventResultStatus::Error;
                result.result = None;
                result.error = Some(
                    "EventHandlerCancelledError: Cancelled: first result resolved".to_string(),
                );
                if result.started_at.is_none() {
                    result.started_at = Some(now_iso());
                }
                result.completed_at = Some(now_iso());
            }
        }
    }

    fn cancel_remaining_timeout_results(
        &self,
        event: &Arc<BaseEvent>,
        remaining_handlers: &[EventHandler],
    ) {
        let mut inner = event.inner.lock();
        for handler in remaining_handlers {
            if let Some(result) = inner.event_results.get_mut(&handler.id) {
                if result.status != EventResultStatus::Pending {
                    continue;
                }
                result.status = EventResultStatus::Error;
                result.result = None;
                result.error = Some(
                    "EventHandlerCancelledError: Cancelled pending handler due to event timeout"
                        .to_string(),
                );
                if result.started_at.is_none() {
                    result.started_at = Some(now_iso());
                }
                result.completed_at = Some(now_iso());
            }
        }
    }

    fn cancel_parallel_first_mode_losers(&self, event: &Arc<BaseEvent>, winner_id: &str) {
        let mut inner = event.inner.lock();
        let winner_completed_at = inner
            .event_results
            .get(winner_id)
            .and_then(|result| result.completed_at.clone());
        for (handler_id, result) in inner.event_results.iter_mut() {
            if handler_id == winner_id {
                continue;
            }
            if result.status == EventResultStatus::Pending {
                result.error = Some(
                    "EventHandlerCancelledError: Cancelled: first result resolved".to_string(),
                );
            } else if result.status == EventResultStatus::Started
                || (result.status == EventResultStatus::Completed
                    && winner_completed_at
                        .as_ref()
                        .zip(result.completed_at.as_ref())
                        .is_some_and(|(winner_at, result_at)| result_at >= winner_at))
            {
                result.error =
                    Some("EventHandlerAbortedError: Aborted: first result resolved".to_string());
            } else {
                continue;
            }
            result.status = EventResultStatus::Error;
            result.result = None;
            if result.started_at.is_none() {
                result.started_at = Some(now_iso());
            }
            result.completed_at = Some(now_iso());
        }
        drop(inner);
        self.cancel_children(event, "first result resolved");
    }

    fn handler_dispatched_ancestor(&self, event: &Arc<BaseEvent>, handler_id: &str) -> usize {
        Self::handler_dispatched_ancestor_inner(
            event,
            handler_id,
            &mut std::collections::HashSet::new(),
            0,
        )
    }

    fn handler_dispatched_ancestor_inner(
        event: &Arc<BaseEvent>,
        handler_id: &str,
        visited: &mut std::collections::HashSet<String>,
        depth: usize,
    ) -> usize {
        let (event_id, parent_id) = {
            let inner = event.inner.lock();
            (inner.event_id.clone(), inner.event_parent_id.clone())
        };
        if !visited.insert(event_id) {
            return depth;
        }
        let Some(parent_id) = parent_id else {
            return depth;
        };

        let parent_event = Self::live_instances()
            .into_iter()
            .find_map(|bus| bus.runtime.events.lock().get(&parent_id).cloned());
        let Some(parent_event) = parent_event else {
            return depth;
        };

        let parent_result_status = parent_event
            .inner
            .lock()
            .event_results
            .get(handler_id)
            .map(|result| result.status);
        let next_depth = if matches!(
            parent_result_status,
            Some(
                EventResultStatus::Pending
                    | EventResultStatus::Started
                    | EventResultStatus::Completed
            )
        ) {
            depth + 1
        } else {
            depth
        };

        Self::handler_dispatched_ancestor_inner(&parent_event, handler_id, visited, next_depth)
    }

    async fn run_handler_with_context(
        &self,
        event: Arc<BaseEvent>,
        handler: EventHandler,
        event_started_at: Instant,
        event_timeout: Option<f64>,
        inline_call: bool,
    ) -> bool {
        let handler_id = handler.id.clone();
        let context_snapshot = self.context_for_event(&event);
        let (event_id, event_concurrency) = {
            let inner = event.inner.lock();
            (
                inner.event_id.clone(),
                inner.event_concurrency.unwrap_or(self.event_concurrency),
            )
        };
        CURRENT_BUS_ID.with(|id| *id.borrow_mut() = Some(self.id.clone()));
        CURRENT_EVENT_ID.with(|id| *id.borrow_mut() = Some(event_id));
        CURRENT_HANDLER_ID.with(|id| *id.borrow_mut() = Some(handler_id.clone()));
        CURRENT_EVENT_CONCURRENCY.with(|mode| *mode.borrow_mut() = Some(event_concurrency));
        let timed_out = self
            .run_handler(
                event,
                handler,
                event_started_at,
                event_timeout,
                context_snapshot,
                inline_call,
            )
            .await;
        CURRENT_EVENT_CONCURRENCY.with(|mode| *mode.borrow_mut() = None);
        CURRENT_HANDLER_ID.with(|id| *id.borrow_mut() = None);
        CURRENT_EVENT_ID.with(|id| *id.borrow_mut() = None);
        CURRENT_BUS_ID.with(|id| *id.borrow_mut() = None);
        timed_out
    }

    async fn run_handler(
        &self,
        event: Arc<BaseEvent>,
        handler: EventHandler,
        _event_started_at: Instant,
        event_timeout: Option<f64>,
        context_snapshot: Option<dcontext::ContextSnapshot>,
        inline_call: bool,
    ) -> bool {
        let mut explicit_handler_timeout = handler
            .handler_timeout
            .or(event.inner.lock().event_handler_timeout);
        if explicit_handler_timeout.is_some_and(|timeout| timeout <= 0.0) {
            explicit_handler_timeout = None;
        }
        let mut resolved_event_timeout = event_timeout.or(self.event_timeout);
        if resolved_event_timeout.is_some_and(|timeout| timeout <= 0.0) {
            resolved_event_timeout = None;
        }
        let resolved_handler_timeout = explicit_handler_timeout.or(resolved_event_timeout);

        let result_timeout =
            self.result_timeout_for_handler(&event, &handler, resolved_event_timeout);

        let remaining_event_timeout = resolved_event_timeout.map(|timeout| {
            let elapsed = _event_started_at.elapsed().as_secs_f64();
            (timeout - elapsed).max(0.0)
        });

        let call_timeout = match (resolved_handler_timeout, remaining_event_timeout) {
            (Some(a), Some(b)) => Some(a.min(b)),
            (Some(a), None) => Some(a),
            (None, Some(b)) => Some(b),
            (None, None) => None,
        };
        let handler_timeout_won_on_timeout =
            explicit_handler_timeout.is_some_and(|handler_timeout| {
                remaining_event_timeout
                    .map(|event_remaining| handler_timeout <= event_remaining)
                    .unwrap_or(true)
            });

        let recursion_depth = self.handler_dispatched_ancestor(&event, &handler.id);
        if recursion_depth > self.max_handler_recursion_depth {
            let mut inner = event.inner.lock();
            let mut result = inner
                .event_results
                .get(&handler.id)
                .cloned()
                .unwrap_or_else(|| {
                    EventResult::new(inner.event_id.clone(), handler.clone(), result_timeout)
                });
            if result.status != EventResultStatus::Pending {
                return false;
            }
            result.status = EventResultStatus::Error;
            result.result = None;
            result.error = Some(format!(
                "Infinite loop detected: Handler {}.{} has recursively processed {} levels of events. Current event: {}, Handler: {}",
                handler.eventbus_name,
                handler.handler_name,
                recursion_depth,
                inner.event_id,
                handler.id
            ));
            if result.started_at.is_none() {
                result.started_at = Some(now_iso());
            }
            result.completed_at = Some(now_iso());
            inner.event_results.insert(handler.id.clone(), result);
            return false;
        }

        {
            let mut inner = event.inner.lock();
            let mut result = inner
                .event_results
                .get(&handler.id)
                .cloned()
                .unwrap_or_else(|| {
                    EventResult::new(inner.event_id.clone(), handler.clone(), result_timeout)
                });
            if result.status != EventResultStatus::Pending {
                return false;
            }
            result.status = EventResultStatus::Started;
            result.started_at = Some(now_iso());
            inner.event_results.insert(handler.id.clone(), result);
        }
        let (event_type_for_slow_warning, event_handler_slow_timeout) = {
            let inner = event.inner.lock();
            (
                inner.event_type.clone(),
                inner
                    .event_handler_slow_timeout
                    .or(self.event_handler_slow_timeout),
            )
        };
        let mut slow_timeout = if handler.handler_slow_timeout.is_some() {
            handler.handler_slow_timeout
        } else {
            event_handler_slow_timeout
        };
        if slow_timeout.is_some_and(|timeout| timeout <= 0.0) {
            slow_timeout = None;
        }
        if let Some(slow_timeout) = slow_timeout {
            if call_timeout.is_some_and(|timeout| timeout <= slow_timeout) {
                return self
                    .run_handler_call(
                        event,
                        handler,
                        call_timeout,
                        handler_timeout_won_on_timeout,
                        context_snapshot,
                        inline_call,
                    )
                    .await;
            }
            let event_for_warning = event.clone();
            let handler_id_for_warning = handler.id.clone();
            let handler_name_for_warning = handler.handler_name.clone();
            let bus_name_for_warning = self.name.clone();
            Self::slow_warning_scheduler().schedule(
                Duration::from_secs_f64(slow_timeout),
                move || {
                    let still_running = event_for_warning
                        .inner
                        .lock()
                        .event_results
                        .get(&handler_id_for_warning)
                        .is_some_and(|result| result.status == EventResultStatus::Started);
                    if still_running {
                        eprintln!(
                        "Slow event handler: {bus_name_for_warning}.on({event_type_for_slow_warning}, {handler_name_for_warning}) still running after {slow_timeout:.3}s"
                    );
                    }
                },
            );
        }

        self.run_handler_call(
            event,
            handler,
            call_timeout,
            handler_timeout_won_on_timeout,
            context_snapshot,
            inline_call,
        )
        .await
    }

    async fn run_handler_call(
        &self,
        event: Arc<BaseEvent>,
        handler: EventHandler,
        call_timeout: Option<f64>,
        handler_timeout_won_on_timeout: bool,
        context_snapshot: Option<dcontext::ContextSnapshot>,
        inline_call: bool,
    ) -> bool {
        let call = handler
            .callable
            .as_ref()
            .expect("handler callable missing")
            .clone();
        let event_clone = event.clone();
        let event_id_for_call = event.inner.lock().event_id.clone();
        let handler_id_for_call = handler.id.clone();
        let runloop_pause = self.locks.request_runloop_pause();
        let mut call_started = true;
        let call_result = if call_timeout.is_some_and(|timeout| timeout <= 0.0) {
            call_started = false;
            Err("timeout".to_string())
        } else if inline_call && call_timeout.is_none() {
            let _context_guard = context_snapshot.map(dcontext::attach);
            let response = call(event_clone).await;
            Self::clear_handler_context_stale(&event_id_for_call, &handler_id_for_call);
            Ok(response)
        } else {
            let (tx, rx) = std_mpsc::channel();
            let context_bus_id = self.id.clone();
            let context_event_id = event_id_for_call.clone();
            let context_handler_id = handler_id_for_call.clone();
            let context_event_concurrency = event
                .inner
                .lock()
                .event_concurrency
                .unwrap_or(self.event_concurrency);
            Self::handler_call_executor().spawn(move || {
                let _context_guard = context_snapshot.map(dcontext::attach);
                CURRENT_BUS_ID.with(|id| *id.borrow_mut() = Some(context_bus_id));
                CURRENT_EVENT_ID.with(|id| *id.borrow_mut() = Some(context_event_id.clone()));
                CURRENT_HANDLER_ID.with(|id| *id.borrow_mut() = Some(context_handler_id.clone()));
                CURRENT_EVENT_CONCURRENCY
                    .with(|mode| *mode.borrow_mut() = Some(context_event_concurrency));
                let response = block_on(call(event_clone));
                Self::clear_handler_context_stale(&context_event_id, &context_handler_id);
                CURRENT_EVENT_CONCURRENCY.with(|mode| *mode.borrow_mut() = None);
                CURRENT_HANDLER_ID.with(|id| *id.borrow_mut() = None);
                CURRENT_EVENT_ID.with(|id| *id.borrow_mut() = None);
                CURRENT_BUS_ID.with(|id| *id.borrow_mut() = None);
                let _ = tx.send(response);
            });

            if let Some(timeout_secs) = call_timeout {
                rx.recv_timeout(Duration::from_secs_f64(timeout_secs))
                    .map_err(|_| "timeout".to_string())
            } else {
                rx.recv().map_err(|_| "handler channel closed".to_string())
            }
        };
        if call_started && matches!(call_result, Err(ref error) if error == "timeout") {
            Self::mark_handler_context_stale(&event_id_for_call, &handler_id_for_call);
        }
        drop(runloop_pause);
        Self::notify_all_queues();

        let mut current = event
            .inner
            .lock()
            .event_results
            .get(&handler.id)
            .cloned()
            .expect("missing result row");
        if current.status != EventResultStatus::Started {
            current.release_queue_jump_pauses();
            return false;
        }
        current.release_queue_jump_pauses();

        match call_result {
            Ok(Ok(value)) => match event.validate_result_value(value) {
                Ok(value) => {
                    current.status = EventResultStatus::Completed;
                    current.result = Some(value);
                }
                Err(error) => {
                    current.status = EventResultStatus::Error;
                    current.result = None;
                    current.error = Some(error);
                }
            },
            Ok(Err(error)) => {
                current.status = EventResultStatus::Error;
                current.error = Some(error);
            }
            Err(error) => {
                current.status = EventResultStatus::Error;
                let error_type = if error == "timeout" && handler_timeout_won_on_timeout {
                    "EventHandlerTimeoutError"
                } else {
                    "EventHandlerAbortedError"
                };
                current.error = Some(format!("{error_type}: {error}"));
                current.completed_at = Some(now_iso());
                if let Some(latest) = event.inner.lock().event_results.get(&handler.id).cloned() {
                    current.event_children = latest.event_children;
                }
                event
                    .inner
                    .lock()
                    .event_results
                    .insert(handler.id.clone(), current);
                return error == "timeout" && !handler_timeout_won_on_timeout;
            }
        }

        current.completed_at = Some(now_iso());
        if let Some(latest) = event.inner.lock().event_results.get(&handler.id).cloned() {
            current.event_children = latest.event_children;
        }
        event.inner.lock().event_results.insert(handler.id, current);
        false
    }
}