async_nodes/

events.rs

//! The live event stream: telemetry observers for runs, and a broadcast bridge.
//!
//! [`BspInterceptor`] is the engine's observability surface — what a GUI subscribes to for
//! its live execution view (highlight the active edge, badge the loop counter, show the
//! "waiting for input" card). Events carry **no `S` payloads**, so the trait stays
//! `dyn`-safe with no type parameter; snapshots of consumer state remain the consumer's
//! recorder concern.
//!
//! Every method has a default no-op body: interceptors written against the original
//! three-method dataflow trait (`on_node_start` / `on_node_complete` / `on_node_error`)
//! keep compiling unchanged.
//!
//! Callbacks are synchronous by design; [`ChannelInterceptor`] bridges the stream to async
//! consumers over a `tokio::sync::broadcast` channel.

use std::time::Duration;

use super::{
    error::BspError,
    graph::{EdgeId, NodeId},
    run::PendingSuspension,
};

/// Telemetry observer to audit performance and trace the state transitions of a run.
///
/// All methods are no-ops by default — implement only the events you need.
pub trait BspInterceptor: Send + Sync {
    /// A superstep is about to execute the given frontier.
    fn on_superstep_start(&self, _step: u64, _frontier: &[NodeId]) {}

    /// A node began executing.
    fn on_node_start(&self, _node_id: NodeId, _name: &str) {}

    /// A node finished executing (including skip-bypassed executions, which report a
    /// zero duration).
    fn on_node_complete(&self, _node_id: NodeId, _name: &str, _duration: Duration) {}

    /// A node's failure is propagating (after any retries were exhausted or classified
    /// fatal). Also fires when the failure is routed to an error edge.
    fn on_node_error(&self, _node_id: NodeId, _name: &str, _err: &BspError) {}

    /// A node failure was classified transient; the engine will re-execute after `wait`.
    /// `attempt` is the 1-based retry number.
    fn on_node_retry(&self, _node_id: NodeId, _attempt: u32, _wait: Duration, _err: &BspError) {}

    /// An edge fired: its condition matched and a token was emitted toward `target`.
    /// This is what makes a GUI *alive* — highlight the active edge.
    fn on_edge_taken(&self, _source: NodeId, _target: NodeId, _condition_label: &str) {}

    /// A budgeted edge fired against its iteration cap — badge the loop counter.
    fn on_loop_iteration(&self, _edge: EdgeId, _iteration: u64, _max: Option<u64>) {}

    /// One or more nodes requested suspension; the run is pausing.
    fn on_suspended(&self, _suspensions: &[PendingSuspension]) {}

    /// The run completed naturally after `steps` supersteps.
    fn on_complete(&self, _steps: u64) {}
}

/// One owned event from the execution stream, as broadcast by [`ChannelInterceptor`].
#[derive(Debug, Clone)]
pub enum BspEvent {
    /// See [`BspInterceptor::on_superstep_start`].
    SuperstepStart { step: u64, frontier: Vec<NodeId> },
    /// See [`BspInterceptor::on_node_start`].
    NodeStart { node_id: NodeId, name: String },
    /// See [`BspInterceptor::on_node_complete`].
    NodeComplete {
        node_id: NodeId,
        name: String,
        duration: Duration,
    },
    /// See [`BspInterceptor::on_node_error`].
    NodeError {
        node_id: NodeId,
        name: String,
        error: BspError,
    },
    /// See [`BspInterceptor::on_node_retry`].
    NodeRetry {
        node_id: NodeId,
        attempt: u32,
        wait: Duration,
        error: BspError,
    },
    /// See [`BspInterceptor::on_edge_taken`].
    EdgeTaken {
        source: NodeId,
        target: NodeId,
        condition_label: String,
    },
    /// See [`BspInterceptor::on_loop_iteration`].
    LoopIteration {
        edge: EdgeId,
        iteration: u64,
        max: Option<u64>,
    },
    /// See [`BspInterceptor::on_suspended`].
    Suspended { suspensions: Vec<PendingSuspension> },
    /// See [`BspInterceptor::on_complete`].
    Complete { steps: u64 },
}

/// A [`BspInterceptor`] that forwards every event onto a `tokio::sync::broadcast`
/// channel — every GUI / CLI / TUI then just subscribes.
///
/// Send failures (no active receivers, or a lagging receiver) are ignored by design:
/// telemetry must never backpressure the scheduler.
#[cfg(feature = "tokio")]
pub struct ChannelInterceptor {
    tx: tokio::sync::broadcast::Sender<BspEvent>,
}

#[cfg(feature = "tokio")]
impl ChannelInterceptor {
    /// Creates an interceptor and its broadcast channel with the given capacity.
    #[must_use]
    pub fn new(capacity: usize) -> Self {
        let (tx, _) = tokio::sync::broadcast::channel(capacity);
        Self { tx }
    }

    /// Subscribes to the event stream.
    #[must_use]
    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<BspEvent> {
        self.tx.subscribe()
    }

    fn send(&self, event: BspEvent) {
        let _ = self.tx.send(event);
    }
}

#[cfg(feature = "tokio")]
impl BspInterceptor for ChannelInterceptor {
    fn on_superstep_start(&self, step: u64, frontier: &[NodeId]) {
        self.send(BspEvent::SuperstepStart {
            step,
            frontier: frontier.to_vec(),
        });
    }

    fn on_node_start(&self, node_id: NodeId, name: &str) {
        self.send(BspEvent::NodeStart {
            node_id,
            name: name.to_string(),
        });
    }

    fn on_node_complete(&self, node_id: NodeId, name: &str, duration: Duration) {
        self.send(BspEvent::NodeComplete {
            node_id,
            name: name.to_string(),
            duration,
        });
    }

    fn on_node_error(&self, node_id: NodeId, name: &str, err: &BspError) {
        self.send(BspEvent::NodeError {
            node_id,
            name: name.to_string(),
            error: err.clone(),
        });
    }

    fn on_node_retry(&self, node_id: NodeId, attempt: u32, wait: Duration, err: &BspError) {
        self.send(BspEvent::NodeRetry {
            node_id,
            attempt,
            wait,
            error: err.clone(),
        });
    }

    fn on_edge_taken(&self, source: NodeId, target: NodeId, condition_label: &str) {
        self.send(BspEvent::EdgeTaken {
            source,
            target,
            condition_label: condition_label.to_string(),
        });
    }

    fn on_loop_iteration(&self, edge: EdgeId, iteration: u64, max: Option<u64>) {
        self.send(BspEvent::LoopIteration {
            edge,
            iteration,
            max,
        });
    }

    fn on_suspended(&self, suspensions: &[PendingSuspension]) {
        self.send(BspEvent::Suspended {
            suspensions: suspensions.to_vec(),
        });
    }

    fn on_complete(&self, steps: u64) {
        self.send(BspEvent::Complete { steps });
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The original three-method dataflow interceptor shape must keep compiling.
    struct LegacyInterceptor;

    impl BspInterceptor for LegacyInterceptor {
        fn on_node_start(&self, _node_id: NodeId, _name: &str) {}
        fn on_node_complete(&self, _node_id: NodeId, _name: &str, _duration: Duration) {}
        fn on_node_error(&self, _node_id: NodeId, _name: &str, _err: &BspError) {}
    }

    #[test]
    fn test_default_methods_are_noops() {
        let interceptor = LegacyInterceptor;
        // Every extended event is callable and a no-op by default.
        interceptor.on_superstep_start(0, &[1, 2]);
        interceptor.on_node_retry(1, 1, Duration::ZERO, &BspError::Deadlock);
        interceptor.on_edge_taken(1, 2, "always");
        interceptor.on_loop_iteration(0, 1, Some(30));
        interceptor.on_suspended(&[]);
        interceptor.on_complete(3);
    }

    #[cfg(feature = "tokio")]
    #[test]
    fn test_channel_interceptor_broadcasts() {
        let interceptor = ChannelInterceptor::new(16);
        let mut rx = interceptor.subscribe();
        interceptor.on_edge_taken(1, 2, "always");
        interceptor.on_complete(7);
        let first = rx.try_recv().unwrap();
        let second = rx.try_recv().unwrap();
        assert!(matches!(
            first,
            BspEvent::EdgeTaken {
                source: 1,
                target: 2,
                ..
            }
        ));
        assert!(matches!(second, BspEvent::Complete { steps: 7 }));
    }
}