mecha10_behavior_runtime/actions/

sensing.rs

//! Action nodes that inspect sensor state.

use crate::{BehaviorNode, Context, NodeStatus};
use async_trait::async_trait;
use mecha10_core::Receiver;
use serde::{Deserialize, Serialize};
use serde_json::Value;

/// SensorCheckNode - Checks sensor state
///
/// Subscribes to a sensor topic and checks if a condition is met.
///
/// # Configuration
///
/// ```json
/// {
///   "type": "sensor_check",
///   "config": {
///     "topic": "/sensors/obstacle",
///     "field": "distance",
///     "operator": "less_than",
///     "threshold": 0.5
///   }
/// }
/// ```
///
/// # Behavior
///
/// - Lazily subscribes to `topic` on the first tick.
/// - Each tick, drains any pending sensor messages and keeps the most recent one. If no new
///   message has arrived since the last tick, the previously received reading is reused (a
///   publisher is not required to publish on every tick).
/// - If no sensor message has ever been received, the node returns `Running` - it is still
///   waiting for a first reading, per the documented semantics of `NodeStatus::Running`.
/// - If a message has been received but `field` is missing or not numeric, the node returns
///   `Failure` - the condition cannot be evaluated.
/// - Otherwise the node compares `field`'s value against `threshold` using `operator` and
///   returns `Success` if the comparison holds, `Failure` otherwise.
pub struct SensorCheckNode {
    topic: String,
    field: String,
    operator: ComparisonOperator,
    threshold: f64,
    receiver: Option<Receiver<Value>>,
    latest: Option<Value>,
}

impl std::fmt::Debug for SensorCheckNode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SensorCheckNode")
            .field("topic", &self.topic)
            .field("field", &self.field)
            .field("operator", &self.operator)
            .field("threshold", &self.threshold)
            .field("subscribed", &self.receiver.is_some())
            .field("latest", &self.latest)
            .finish()
    }
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum ComparisonOperator {
    LessThan,
    GreaterThan,
    Equal,
}

impl ComparisonOperator {
    /// Evaluate `actual <op> threshold`.
    fn evaluate(&self, actual: f64, threshold: f64) -> bool {
        match self {
            ComparisonOperator::LessThan => actual < threshold,
            ComparisonOperator::GreaterThan => actual > threshold,
            // Floating point sensor readings are rarely bit-for-bit identical; use a small
            // tolerance rather than exact equality.
            ComparisonOperator::Equal => (actual - threshold).abs() < f64::EPSILON,
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct SensorCheckConfig {
    topic: String,
    field: String,
    operator: ComparisonOperator,
    threshold: f64,
}

impl SensorCheckNode {
    /// Create a new SensorCheckNode from JSON configuration
    pub fn from_json(config: Value) -> anyhow::Result<Self> {
        let cfg: SensorCheckConfig = serde_json::from_value(config)?;
        Ok(Self {
            topic: cfg.topic,
            field: cfg.field,
            operator: cfg.operator,
            threshold: cfg.threshold,
            receiver: None,
            latest: None,
        })
    }

    /// Drain any pending sensor messages (keeping only the most recent) and evaluate the
    /// configured condition against the latest known reading.
    ///
    /// Pure with respect to messaging infrastructure - it only touches the already-subscribed
    /// `Receiver`, which makes it directly unit-testable without a live `Context`/Redis.
    fn evaluate(&mut self) -> NodeStatus {
        if let Some(receiver) = self.receiver.as_mut() {
            while let Ok(msg) = receiver.try_recv() {
                self.latest = Some(msg);
            }
        }

        let Some(latest) = self.latest.as_ref() else {
            tracing::debug!(topic = %self.topic, "SensorCheckNode: waiting for first sensor reading");
            return NodeStatus::Running;
        };

        let Some(actual) = latest.get(&self.field).and_then(Value::as_f64) else {
            tracing::warn!(
                topic = %self.topic,
                field = %self.field,
                message = %latest,
                "SensorCheckNode: field missing or not numeric in sensor message"
            );
            return NodeStatus::Failure;
        };

        let met = self.operator.evaluate(actual, self.threshold);
        tracing::debug!(
            "SensorCheckNode: checking {} {} {} -> {}",
            actual,
            match self.operator {
                ComparisonOperator::LessThan => "<",
                ComparisonOperator::GreaterThan => ">",
                ComparisonOperator::Equal => "==",
            },
            self.threshold,
            met
        );

        if met {
            NodeStatus::Success
        } else {
            NodeStatus::Failure
        }
    }

    /// Construct a node with an already-connected receiver, bypassing live subscription.
    ///
    /// Used by unit tests to exercise `evaluate()` without requiring a real `Context`/Redis.
    #[cfg(test)]
    fn with_receiver(field: &str, operator: ComparisonOperator, threshold: f64, receiver: Receiver<Value>) -> Self {
        Self {
            topic: "test/topic".to_string(),
            field: field.to_string(),
            operator,
            threshold,
            receiver: Some(receiver),
            latest: None,
        }
    }
}

#[async_trait]
impl BehaviorNode for SensorCheckNode {
    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
        if self.receiver.is_none() {
            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.topic).await?);
        }
        Ok(self.evaluate())
    }

    fn name(&self) -> &str {
        "SensorCheckNode"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use tokio::sync::mpsc;

    fn node_with_channel(
        field: &str,
        operator: ComparisonOperator,
        threshold: f64,
    ) -> (SensorCheckNode, mpsc::UnboundedSender<Value>) {
        let (tx, rx) = mpsc::unbounded_channel();
        let node = SensorCheckNode::with_receiver(field, operator, threshold, Receiver::from(rx));
        (node, tx)
    }

    #[test]
    fn no_message_received_returns_running() {
        let (mut node, _tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
        assert_eq!(node.evaluate(), NodeStatus::Running);
    }

    #[test]
    fn missing_field_returns_failure() {
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
        tx.send(json!({ "other_field": 1.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn non_numeric_field_returns_failure() {
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
        tx.send(json!({ "distance": "close" })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn less_than_success_when_below_threshold() {
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
        tx.send(json!({ "distance": 0.2 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Success);
    }

    #[test]
    fn less_than_failure_when_at_or_above_threshold() {
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
        tx.send(json!({ "distance": 0.5 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);

        tx.send(json!({ "distance": 1.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn greater_than_success_when_above_threshold() {
        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
        tx.send(json!({ "battery": 55.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Success);
    }

    #[test]
    fn greater_than_failure_when_at_or_below_threshold() {
        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
        tx.send(json!({ "battery": 20.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);

        tx.send(json!({ "battery": 5.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn equal_success_when_matching() {
        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
        tx.send(json!({ "state": 1.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Success);
    }

    #[test]
    fn equal_failure_when_not_matching() {
        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
        tx.send(json!({ "state": 0.0 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn stale_topic_reuses_last_known_reading() {
        // A publisher that stops sending updates shouldn't invalidate a previously received
        // reading - the node should keep evaluating against the last known value.
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);

        tx.send(json!({ "distance": 0.1 })).unwrap();
        assert_eq!(node.evaluate(), NodeStatus::Success);

        // No new message published before the next tick - still Success from cached reading.
        assert_eq!(node.evaluate(), NodeStatus::Success);
        assert_eq!(node.evaluate(), NodeStatus::Success);
    }

    #[test]
    fn later_message_overrides_earlier_one() {
        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);

        tx.send(json!({ "distance": 0.1 })).unwrap();
        tx.send(json!({ "distance": 1.0 })).unwrap();

        // Both messages are drained in a single evaluate() call; the most recent wins.
        assert_eq!(node.evaluate(), NodeStatus::Failure);
    }

    #[test]
    fn from_json_parses_all_operators() {
        for operator in ["less_than", "greater_than", "equal"] {
            let config = json!({
                "topic": "/sensors/distance",
                "field": "value",
                "operator": operator,
                "threshold": 0.5
            });
            assert!(SensorCheckNode::from_json(config).is_ok());
        }
    }

    #[test]
    fn from_json_rejects_unknown_operator() {
        let config = json!({
            "topic": "/sensors/distance",
            "field": "value",
            "operator": "not_a_real_operator",
            "threshold": 0.5
        });
        assert!(SensorCheckNode::from_json(config).is_err());
    }
}