rustsim_core/

event_queue.rs

//! Continuous-time event-queue model.
//!
//! [`EventQueueModel`] mirrors Julia Agents.jl `EventQueueABM`. Instead of
//! fixed-tick stepping, agents schedule events at arbitrary future times.
//! A deterministic priority queue processes events in chronological order,
//! with tie-breaking by insertion sequence number.
//!
//! # Event processing
//!
//! 1. Pop the earliest event from the queue.
//! 2. Advance model time to the event's timestamp.
//! 3. Look up the target agent; skip if removed.
//! 4. Call the action function indexed by `event_idx`.
//! 5. Apply any deferred add/remove actions.
//!
//! Events for removed agents are silently skipped (not an error).
//!
//! # Determinism
//!
//! Events with identical timestamps are ordered by a monotonic sequence
//! number assigned at insertion time.
//!
//! Replay is deterministic for a fixed seed when all of the following are held
//! constant:
//! - the same initial event insertion order
//! - the same action functions
//! - the same event rescheduling behavior inside actions
//! - the same agent add/remove behavior
//!
//! Equal-time events are therefore reproducible by construction, but changing
//! the order in which they are inserted will intentionally change execution
//! order.

#![allow(clippy::type_complexity)]

use crate::{
    agent::Agent,
    interaction::{InteractionError, PositionedAgent, SpaceInteraction},
    model::Model,
    space::Space,
    step_context::DeferredAction,
    store::AgentStore,
    types::{AgentId, Time},
};
use rand::RngCore;
use std::cell::{Ref, RefMut};
use std::cmp::Ordering;
use std::collections::BinaryHeap;
use tracing::{debug, trace};

/// A scheduled event in the queue.
#[derive(Debug, Clone)]
pub struct Event {
    /// Absolute time at which this event fires.
    pub time: f64,
    /// Target agent for this event.
    pub agent_id: AgentId,
    /// Index into the model's `actions` vector selecting which function to call.
    pub event_idx: usize,
    /// Monotonic sequence number for deterministic tie-breaking.
    sequence: u64,
}

impl<S, A, Store, Props, R> EventQueueModel<S, A, Store, Props, R>
where
    A: PositionedAgent,
    S: SpaceInteraction<A>,
    Store: AgentStore<A>,
    R: RngCore,
{
    /// Insert a positioned agent into both the store and the space atomically.
    pub fn insert_positioned_agent(&mut self, agent: A) -> Result<(), InteractionError<S::Error>> {
        let id = agent.id();
        if self.agents.contains(id) {
            return Err(InteractionError::DuplicateId(id));
        }
        self.space
            .add_agent(&agent)
            .map_err(InteractionError::Space)?;
        self.agents.insert(agent);
        if id > self.max_id {
            self.max_id = id;
        }
        Ok(())
    }

    /// Remove a positioned agent from both the store and the space atomically.
    pub fn remove_positioned_agent(
        &mut self,
        id: AgentId,
    ) -> Result<Option<A>, InteractionError<S::Error>> {
        let Some(agent_ref) = self.agents.get(id) else {
            return Ok(None);
        };
        self.space
            .remove_agent(&*agent_ref)
            .map_err(InteractionError::Space)?;
        drop(agent_ref);
        Ok(self.agents.remove(id))
    }

    /// Move a positioned agent, updating both the agent value and spatial index.
    pub fn move_positioned_agent(
        &mut self,
        id: AgentId,
        new_position: A::Position,
    ) -> Result<(), InteractionError<S::Error>> {
        let mut agent_ref = self
            .agents
            .get_mut(id)
            .ok_or(InteractionError::AgentNotFound(id))?;
        let old_position = agent_ref.position().clone();

        self.space
            .remove_agent(&*agent_ref)
            .map_err(InteractionError::Space)?;
        agent_ref.set_position(new_position);

        if let Err(source) = self.space.add_agent(&*agent_ref) {
            agent_ref.set_position(old_position);
            if let Err(rollback) = self.space.add_agent(&*agent_ref) {
                return Err(InteractionError::RollbackFailed {
                    operation: "move_positioned_agent",
                    source,
                    rollback,
                });
            }
            return Err(InteractionError::Space(source));
        }
        Ok(())
    }

    /// Validate that all stored agents are represented by the spatial index.
    pub fn validate_space_index(&self) -> Result<(), InteractionError<S::Error>> {
        for id in self.agents.iter_ids() {
            let Some(agent) = self.agents.get(id) else {
                continue;
            };
            let matches = self
                .space
                .nearby_ids(agent.position(), 0)
                .into_iter()
                .filter(|candidate| *candidate == id)
                .count();
            match matches {
                0 => return Err(InteractionError::SpaceIndexMissing(id)),
                1 => {}
                _ => return Err(InteractionError::SpaceIndexDuplicate(id)),
            }
        }
        Ok(())
    }

    /// Process the next event and apply deferred add/remove actions to both the
    /// agent store and spatial index.
    pub fn step_event_spatial(&mut self) -> Result<bool, InteractionError<S::Error>> {
        let timed = match self.queue.pop() {
            Some(te) => te,
            None => {
                trace!("step_event_spatial: queue empty");
                return Ok(false);
            }
        };

        let event = timed.0;
        self.time = event.time;

        if !self.agents.contains(event.agent_id) {
            trace!(
                agent_id = event.agent_id,
                time = event.time,
                "skipping event for removed agent"
            );
            return Ok(true);
        }

        if event.event_idx < self.actions.len() {
            let action = self.actions[event.event_idx];

            let Some(mut agent_ref) = self.agents.get_mut(event.agent_id) else {
                return Ok(true);
            };

            let mut rng = self.rng.borrow_mut();
            let mut deferred: Vec<DeferredAction<A>> = Vec::new();

            {
                let mut ctx = EventContext {
                    space: &mut self.space,
                    properties: &mut self.properties,
                    rng: &mut *rng,
                    queue: &mut self.queue,
                    sequence: &mut self.sequence,
                    time: self.time,
                    deferred: &mut deferred,
                };

                action(&mut *agent_ref, &mut ctx);
            }

            drop(agent_ref);
            drop(rng);

            self.apply_deferred_actions_spatial(deferred)?;
        }

        Ok(true)
    }

    /// Process all events up to and including time `t_end` with spatially
    /// consistent deferred lifecycle updates.
    pub fn step_until_spatial(&mut self, t_end: f64) -> Result<(), InteractionError<S::Error>> {
        loop {
            match self.queue.peek() {
                Some(te) if te.0.time <= t_end => {}
                _ => {
                    self.time = t_end;
                    debug!(
                        time = t_end,
                        queue_len = self.queue.len(),
                        "step_until_spatial reached boundary"
                    );
                    return Ok(());
                }
            }
            self.step_event_spatial()?;
        }
    }

    /// Process up to `n` events with spatially consistent lifecycle updates.
    pub fn run_events_spatial(&mut self, n: usize) -> Result<(), InteractionError<S::Error>> {
        for _ in 0..n {
            if !self.step_event_spatial()? {
                break;
            }
        }
        Ok(())
    }

    fn apply_deferred_actions_spatial(
        &mut self,
        deferred: Vec<DeferredAction<A>>,
    ) -> Result<(), InteractionError<S::Error>> {
        for action in deferred {
            match action {
                DeferredAction::RemoveAgent(id) => {
                    self.remove_positioned_agent(id)?;
                }
                DeferredAction::InsertAgent(agent) => {
                    self.insert_positioned_agent(agent)?;
                }
            }
        }
        Ok(())
    }
}

/// Internal wrapper that implements `Ord` for the min-heap.
#[derive(Debug, Clone)]
pub(crate) struct TimedEvent(Event);

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

impl Eq for TimedEvent {}

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

impl Ord for TimedEvent {
    fn cmp(&self, other: &Self) -> Ordering {
        other
            .0
            .time
            .partial_cmp(&self.0.time)
            .unwrap_or(Ordering::Equal)
            .then_with(|| other.0.sequence.cmp(&self.0.sequence))
    }
}

/// Context passed to event-queue action callbacks during [`EventQueueModel::step_event`].
///
/// Provides safe mutable access to space, properties, RNG, and the event queue
/// without aliasing the currently-borrowed agent.
pub struct EventContext<'a, S, A, Props, R>
where
    A: Agent,
{
    pub(crate) space: &'a mut S,
    pub(crate) properties: &'a mut Props,
    pub(crate) rng: &'a mut R,
    pub(crate) queue: &'a mut BinaryHeap<TimedEvent>,
    pub(crate) sequence: &'a mut u64,
    pub(crate) time: f64,
    pub(crate) deferred: &'a mut Vec<DeferredAction<A>>,
}

impl<'a, S, A, Props, R> EventContext<'a, S, A, Props, R>
where
    A: Agent,
{
    /// Immutable reference to the simulation space.
    pub fn space(&self) -> &S {
        self.space
    }

    /// Mutable reference to the simulation space.
    pub fn space_mut(&mut self) -> &mut S {
        self.space
    }

    /// Immutable reference to user-defined model properties.
    pub fn properties(&self) -> &Props {
        self.properties
    }

    /// Mutable reference to user-defined model properties.
    pub fn properties_mut(&mut self) -> &mut Props {
        self.properties
    }

    /// Mutable reference to the model's RNG.
    pub fn rng(&mut self) -> &mut R {
        self.rng
    }

    /// Current model time (the timestamp of the event being processed).
    pub fn time(&self) -> f64 {
        self.time
    }

    /// Schedule a new event relative to the current time.
    ///
    /// # Panics
    ///
    /// Panics if `dt` is negative, NaN, or infinite.
    pub fn add_event(&mut self, agent_id: AgentId, event_idx: usize, dt: f64) {
        assert!(
            dt.is_finite() && dt >= 0.0,
            "event dt must be finite and non-negative, got {dt}"
        );
        *self.sequence += 1;
        let event = Event {
            time: self.time + dt,
            agent_id,
            event_idx,
            sequence: *self.sequence,
        };
        self.queue.push(TimedEvent(event));
    }

    /// Schedule an agent for removal after the current event completes.
    pub fn defer_remove_agent(&mut self, id: AgentId) {
        self.deferred.push(DeferredAction::RemoveAgent(id));
    }

    /// Schedule an agent for insertion after the current event completes.
    pub fn defer_insert_agent(&mut self, agent: A) {
        self.deferred.push(DeferredAction::InsertAgent(agent));
    }
}

/// Continuous-time agent-based model driven by an event queue.
///
/// # Type Parameters
///
/// - `S` - space type
/// - `A` - agent type implementing [`Agent`]
/// - `Store` - agent container implementing [`AgentStore<A>`]
/// - `Props` - user-defined model properties (use `()` if unused)
/// - `R` - RNG type
///
/// # Example
///
/// ```ignore
/// let actions: Vec<fn(&mut MyAgent, &mut EventContext<...>)> = vec![tick_action];
/// let mut model = EventQueueModel::new(store, space, (), rng, actions);
/// model.add_event(agent_id, 0, 1.0); // fire action 0 at t+1.0
/// model.step_until(100.0);
/// ```
///
/// [`Agent`]: crate::agent::Agent
/// [`AgentStore`]: crate::store::AgentStore
pub struct EventQueueModel<S, A, Store, Props, R>
where
    A: Agent,
    S: Space,
    Store: AgentStore<A>,
    R: RngCore,
{
    pub(crate) agents: Store,
    pub(crate) space: S,
    pub(crate) properties: Props,
    pub(crate) rng: std::cell::RefCell<R>,
    pub(crate) time: f64,
    pub(crate) max_id: AgentId,
    pub(crate) queue: BinaryHeap<TimedEvent>,
    pub(crate) sequence: u64,
    pub(crate) actions: Vec<fn(&mut A, &mut EventContext<'_, S, A, Props, R>)>,
    pub(crate) _agent: std::marker::PhantomData<A>,
}

impl<S, A, Store, Props, R> EventQueueModel<S, A, Store, Props, R>
where
    A: Agent,
    S: Space,
    Store: AgentStore<A>,
    R: RngCore,
{
    /// Create a new `EventQueueModel`.
    ///
    /// # Arguments
    ///
    /// - `agents` - pre-populated agent store.
    /// - `space` - the simulation space.
    /// - `properties` - user-defined model properties.
    /// - `rng` - seeded random number generator.
    /// - `actions` - vector of event-action functions, indexed by `event_idx`.
    pub fn new(
        agents: Store,
        space: S,
        properties: Props,
        rng: R,
        actions: Vec<fn(&mut A, &mut EventContext<'_, S, A, Props, R>)>,
    ) -> Self {
        let max_id = agents.iter_ids().into_iter().max().unwrap_or(0);
        Self {
            agents,
            space,
            properties,
            rng: std::cell::RefCell::new(rng),
            time: 0.0,
            max_id,
            queue: BinaryHeap::new(),
            sequence: 0,
            actions,
            _agent: std::marker::PhantomData,
        }
    }

    /// Current model time as `f64`.
    pub fn time_f64(&self) -> f64 {
        self.time
    }

    /// Mutable access to the model's RNG (via `RefCell`).
    pub fn rng_mut(&self) -> std::cell::RefMut<'_, R> {
        self.rng.borrow_mut()
    }

    /// Immutable reference to the simulation space.
    pub fn space(&self) -> &S {
        &self.space
    }

    /// Mutable reference to the simulation space.
    pub fn space_mut(&mut self) -> &mut S {
        &mut self.space
    }

    /// Immutable reference to user-defined properties.
    pub fn properties(&self) -> &Props {
        &self.properties
    }

    /// Mutable reference to user-defined properties.
    pub fn properties_mut(&mut self) -> &mut Props {
        &mut self.properties
    }

    /// Borrow an agent immutably by ID.
    pub fn agent(&self, id: AgentId) -> Option<Ref<'_, A>> {
        self.agents.get(id)
    }

    /// Borrow an agent mutably by ID.
    pub fn agent_mut(&self, id: AgentId) -> Option<RefMut<'_, A>> {
        self.agents.get_mut(id)
    }

    /// Insert an agent into the store.
    ///
    /// Returns `Err(agent)` if an agent with the same ID already exists.
    pub fn insert_agent(&mut self, agent: A) -> Result<(), A> {
        let id = agent.id();
        if self.agents.get(id).is_some() {
            return Err(agent);
        }
        self.agents.insert(agent);
        if id > self.max_id {
            self.max_id = id;
        }
        Ok(())
    }

    /// Remove an agent by ID, returning it if found.
    ///
    /// Events targeting this agent will be silently skipped when they fire.
    pub fn remove_agent(&mut self, id: AgentId) -> Option<A> {
        self.agents.remove(id)
    }

    /// Generate the next unused agent ID (monotonically increasing).
    pub fn next_id(&mut self) -> AgentId {
        self.max_id += 1;
        self.max_id
    }

    /// Schedule a new event at `self.time + dt`.
    ///
    /// # Panics
    ///
    /// Panics if `dt` is negative, NaN, or infinite.
    pub fn add_event(&mut self, agent_id: AgentId, event_idx: usize, dt: f64) {
        assert!(
            dt.is_finite() && dt >= 0.0,
            "event dt must be finite and non-negative, got {dt}"
        );
        self.sequence += 1;
        let event = Event {
            time: self.time + dt,
            agent_id,
            event_idx,
            sequence: self.sequence,
        };
        self.queue.push(TimedEvent(event));
    }

    /// Number of events currently in the queue.
    pub fn queue_len(&self) -> usize {
        self.queue.len()
    }

    /// Returns `true` if the event queue is empty.
    pub fn queue_is_empty(&self) -> bool {
        self.queue.is_empty()
    }

    /// Peek at the timestamp of the next (earliest) event, if any.
    pub fn peek_time(&self) -> Option<f64> {
        self.queue.peek().map(|te| te.0.time)
    }

    /// Process the next event in the queue.
    ///
    /// Returns `true` if an event was popped (even if the target agent was
    /// removed), `false` if the queue was empty.
    pub fn step_event(&mut self) -> bool {
        let timed = match self.queue.pop() {
            Some(te) => te,
            None => {
                trace!("step_event: queue empty");
                return false;
            }
        };

        let event = timed.0;
        self.time = event.time;

        if !self.agents.contains(event.agent_id) {
            trace!(
                agent_id = event.agent_id,
                time = event.time,
                "skipping event for removed agent"
            );
            return true;
        }

        if event.event_idx < self.actions.len() {
            let action = self.actions[event.event_idx];

            let Some(mut agent_ref) = self.agents.get_mut(event.agent_id) else {
                return true;
            };

            let mut rng = self.rng.borrow_mut();
            let mut deferred: Vec<DeferredAction<A>> = Vec::new();

            {
                let mut ctx = EventContext {
                    space: &mut self.space,
                    properties: &mut self.properties,
                    rng: &mut *rng,
                    queue: &mut self.queue,
                    sequence: &mut self.sequence,
                    time: self.time,
                    deferred: &mut deferred,
                };

                action(&mut *agent_ref, &mut ctx);
            }

            drop(agent_ref);
            drop(rng);

            for action in deferred {
                match action {
                    DeferredAction::RemoveAgent(id) => {
                        self.remove_agent(id);
                    }
                    DeferredAction::InsertAgent(agent) => {
                        let _ = self.insert_agent(agent);
                    }
                }
            }
        }

        true
    }

    /// Process all events up to and including time `t_end`.
    ///
    /// After this call, `self.time_f64() == t_end`. Events scheduled after
    /// `t_end` remain in the queue.
    pub fn step_until(&mut self, t_end: f64) {
        loop {
            match self.queue.peek() {
                Some(te) if te.0.time <= t_end => {}
                _ => {
                    self.time = t_end;
                    debug!(
                        time = t_end,
                        queue_len = self.queue.len(),
                        "step_until reached boundary"
                    );
                    return;
                }
            }
            self.step_event();
        }
    }

    /// Process up to `n` events from the queue.
    ///
    /// Stops early if the queue is exhausted.
    pub fn run_events(&mut self, n: usize) {
        for _ in 0..n {
            if !self.step_event() {
                break;
            }
        }
    }
}

impl<S, A, Store, Props, R> Model for EventQueueModel<S, A, Store, Props, R>
where
    A: Agent,
    S: Space,
    Store: AgentStore<A>,
    R: RngCore,
{
    type Agent = A;
    type Space = S;
    type Properties = Props;
    type Rng = R;

    // GATs implementation
    type AgentRef<'a>
        = Ref<'a, A>
    where
        Self: 'a;
    type AgentRefMut<'a>
        = RefMut<'a, A>
    where
        Self: 'a;

    fn time(&self) -> Time {
        Time::Continuous(self.time)
    }

    fn rng_mut(&self) -> impl std::ops::DerefMut<Target = Self::Rng> + '_ {
        self.rng.borrow_mut()
    }

    fn space(&self) -> &Self::Space {
        &self.space
    }

    fn properties(&self) -> &Self::Properties {
        &self.properties
    }

    fn properties_mut(&mut self) -> &mut Self::Properties {
        &mut self.properties
    }

    fn agent(&self, id: AgentId) -> Option<Self::AgentRef<'_>> {
        self.agents.get(id)
    }

    fn agent_mut(&self, id: AgentId) -> Option<Self::AgentRefMut<'_>> {
        self.agents.get_mut(id)
    }
}