mecha10-behavior-runtime 0.8.1

//! Action node that hands control to the `rl` node and waits for it to report task completion.

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

const RL_CAPABILITY: &str = "rl";

const TASK_COMPLETE_TOPIC: &str = "/rl/task_complete";

pub struct RlTaskNode {
    receiver: Option<Receiver<Value>>,
    complete: bool,
}

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

impl RlTaskNode {
    /// Construct a node. `_config` is ignored - see the module doc, this node takes none.
    pub fn from_json(_config: Value) -> anyhow::Result<Self> {
        Ok(Self {
            receiver: None,
            complete: false,
        })
    }

    /// Construct a node with an already-connected receiver, bypassing live subscription.
    ///
    /// Used by unit tests to exercise completion-detection without a real `Context`/Redis.
    #[cfg(test)]
    fn with_receiver(receiver: Receiver<Value>) -> Self {
        Self {
            receiver: Some(receiver),
            complete: false,
        }
    }

    /// Drain any pending completion messages - arrival of any message at all is the whole
    /// signal, so the payload itself is never inspected.
    fn poll_complete(&mut self) {
        if self.complete {
            return;
        }
        if let Some(receiver) = self.receiver.as_mut() {
            if receiver.try_recv().is_ok() {
                self.complete = true;
            }
        }
    }
}

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

        self.poll_complete();

        Ok(if self.complete {
            NodeStatus::Success
        } else {
            NodeStatus::Running
        })
    }

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

    fn active_capability(&self) -> Option<&str> {
        Some(RL_CAPABILITY)
    }

    fn reset_self(&mut self) {
        self.complete = false;
    }
}

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

    fn node_with_channel() -> (RlTaskNode, mpsc::UnboundedSender<Value>) {
        let (tx, rx) = mpsc::unbounded_channel();
        (RlTaskNode::with_receiver(Receiver::from(rx)), tx)
    }

    #[test]
    fn always_declares_the_rl_capability() {
        let (node, _tx) = node_with_channel();
        assert_eq!(node.active_capability(), Some("rl"));
    }

    #[test]
    fn stays_incomplete_with_no_message() {
        let (mut node, _tx) = node_with_channel();
        node.poll_complete();
        assert!(!node.complete);
    }

    #[test]
    fn completes_once_any_message_arrives() {
        let (mut node, tx) = node_with_channel();
        tx.send(serde_json::json!({})).unwrap();

        node.poll_complete();

        assert!(node.complete);
    }

    #[test]
    fn reset_self_clears_completion_for_a_later_pass() {
        let (mut node, tx) = node_with_channel();
        tx.send(serde_json::json!({})).unwrap();
        node.poll_complete();
        assert!(node.complete);

        node.reset_self();

        assert!(!node.complete);
    }

    #[test]
    fn from_json_ignores_any_input() {
        assert!(RlTaskNode::from_json(Value::Null).is_ok());
        assert!(RlTaskNode::from_json(serde_json::json!({"unexpected": "field"})).is_ok());
    }
}