mecha10-behavior-runtime 0.8.2

//! Action node that relays live target-class requests to the `rl` node's control topic.

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

#[derive(Debug, Clone, Serialize, Deserialize)]
struct RlControlCommand {
    action: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    params: Option<Value>,
}

impl Message for RlControlCommand {}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct RlTargetRequest {
    #[serde(default)]
    target_class: Option<String>,
}

impl Message for RlTargetRequest {}

fn default_topic() -> String {
    "/rl/control".to_string()
}

fn default_request_topic() -> String {
    "/rl/target/request".to_string()
}

pub struct SetRlTargetNode {
    topic: String,
    request_topic: String,
    target: Option<String>,
    receiver: Option<Receiver<Value>>,
    published_initial: bool,
}

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

#[derive(Debug, Serialize, Deserialize)]
struct SetRlTargetConfig {
    #[serde(default = "default_topic")]
    topic: String,
    #[serde(default = "default_request_topic")]
    request_topic: String,
    #[serde(default)]
    target: Option<String>,
}

impl SetRlTargetNode {
    /// Create a new SetRlTargetNode from JSON configuration.
    ///
    /// Normalizes a `null` config to `{}` so per-field defaults can apply.
    pub fn from_json(config: Value) -> anyhow::Result<Self> {
        let config = if config.is_null() {
            serde_json::json!({})
        } else {
            config
        };
        let cfg: SetRlTargetConfig = serde_json::from_value(config)?;
        Ok(Self {
            topic: cfg.topic,
            request_topic: cfg.request_topic,
            target: cfg.target,
            receiver: None,
            published_initial: false,
        })
    }

    async fn publish(&self, ctx: &Context, target: Option<String>) -> anyhow::Result<()> {
        let command = RlControlCommand {
            action: "set_target_class".to_string(),
            params: Some(serde_json::json!({ "target_class": target })),
        };

        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
        let topic = Topic::<RlControlCommand>::new(topic_static);

        ctx.publish_to(topic, &command).await?;
        tracing::debug!("SetRlTargetNode: published set_target_class({:?})", target);
        Ok(())
    }

    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
    /// the most recent request matters, same as `RlTaskNode::poll_complete`'s "arrival is the
    /// whole signal" approach to draining a channel once per tick.
    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<String>> {
        let mut latest = None;
        while let Ok(value) = receiver.try_recv() {
            match serde_json::from_value::<RlTargetRequest>(value) {
                Ok(request) => latest = Some(request.target_class),
                Err(e) => tracing::warn!("SetRlTargetNode: dropping malformed target request: {}", e),
            }
        }
        latest
    }
}

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

        if !self.published_initial {
            self.published_initial = true;
            if self.target.is_some() {
                self.publish(ctx, self.target.clone()).await?;
            }
        }

        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
        if let Some(target) = latest {
            self.publish(ctx, target).await?;
        }

        Ok(NodeStatus::Running)
    }

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

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

    #[test]
    fn from_json_reads_topic_request_topic_and_target() {
        let node = SetRlTargetNode::from_json(serde_json::json!({
            "topic": "/rl/control",
            "request_topic": "/rl/target/request",
            "target": "cat"
        }))
        .unwrap();

        assert_eq!(node.topic, "/rl/control");
        assert_eq!(node.request_topic, "/rl/target/request");
        assert_eq!(node.target, Some("cat".to_string()));
    }

    #[test]
    fn from_json_defaults_topics_and_allows_no_target() {
        let node = SetRlTargetNode::from_json(serde_json::json!({})).unwrap();

        assert_eq!(node.topic, "/rl/control");
        assert_eq!(node.request_topic, "/rl/target/request");
        assert_eq!(node.target, None);
    }

    #[test]
    fn from_json_accepts_null_the_same_as_an_empty_object() {
        let node = SetRlTargetNode::from_json(Value::Null).unwrap();

        assert_eq!(node.topic, "/rl/control");
        assert_eq!(node.request_topic, "/rl/target/request");
        assert_eq!(node.target, None);
    }

    #[test]
    fn poll_latest_request_returns_none_when_empty() {
        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), None);
    }

    #[test]
    fn poll_latest_request_keeps_only_the_most_recent_value() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "target_class": "dog" })).unwrap();
        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();

        assert_eq!(
            SetRlTargetNode::poll_latest_request(&mut receiver),
            Some(Some("cat".to_string()))
        );
    }

    #[test]
    fn poll_latest_request_accepts_a_null_target_class_as_a_clear_request() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "target_class": null })).unwrap();

        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), Some(None));
    }

    #[test]
    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "target_class": 42 })).unwrap();
        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();

        assert_eq!(
            SetRlTargetNode::poll_latest_request(&mut receiver),
            Some(Some("cat".to_string()))
        );
    }

    #[tokio::test]
    #[ignore]
    async fn tick_publishes_configured_target_once_then_stays_running() {
        let ctx = create_test_context("test_set_rl_target").await;

        let mut node = SetRlTargetNode::from_json(serde_json::json!({
            "topic": "/rl/control",
            "target": "cat"
        }))
        .unwrap();

        let mut receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(status, NodeStatus::Running);

        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
            .await
            .expect("timed out waiting for published command")
            .expect("channel closed without a message");

        assert_eq!(received["action"], "set_target_class");
        assert_eq!(received["params"]["target_class"], "cat");

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(status, NodeStatus::Running);
        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
        assert!(timed_out.is_err(), "expected no second publish without a new request");
    }

    #[tokio::test]
    #[ignore]
    async fn tick_relays_a_live_request_published_on_request_topic() {
        let ctx = create_test_context("test_set_rl_target_request").await;

        let mut node = SetRlTargetNode::from_json(serde_json::json!({
            "topic": "/rl/control",
            "request_topic": "/rl/target/request"
        }))
        .unwrap();

        node.tick(&ctx).await.unwrap();

        let mut control_receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
        ctx.publish_raw("/rl/target/request", &serde_json::json!({ "target_class": "dog" }))
            .await
            .unwrap();

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        node.tick(&ctx).await.unwrap();

        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
            .await
            .expect("timed out waiting for relayed command")
            .expect("channel closed without a message");

        assert_eq!(received["action"], "set_target_class");
        assert_eq!(received["params"]["target_class"], "dog");
    }
}