mecha10-behavior-runtime 4.2.9

//! Generic action node that relays live target requests to a capability node's control topic.
//!
//! Replaces the old bespoke `SetRlTargetNode`/`SetInstanceTrackTargetNode` pair. Where the old
//! nodes hardcoded their topic names and the shape of the published command, this node takes
//! those as config so one node type covers both (and any future capability with the same
//! "relay a latest-wins target request onto a control topic" shape).

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

#[derive(Debug, Serialize, Deserialize)]
struct SetCapabilityTargetConfig {
    control_topic: String,
    request_topic: String,
    action_set: String,
    #[serde(default)]
    action_clear: Option<String>,
    payload_key: String,
    #[serde(default)]
    initial: Option<Value>,
}

pub struct SetCapabilityTargetNode {
    control_topic: String,
    request_topic: String,
    action_set: String,
    action_clear: Option<String>,
    payload_key: String,
    target: Option<Value>,
    receiver: Option<Receiver<Value>>,
    published_initial: bool,
    last_status: Option<NodeStatus>,
}

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

impl SetCapabilityTargetNode {
    /// Create a new node from JSON configuration. `control_topic`, `request_topic`,
    /// `action_set`, and `payload_key` are required; `action_clear` and `initial` are optional.
    pub fn from_json(config: Value) -> anyhow::Result<Self> {
        let cfg: SetCapabilityTargetConfig = serde_json::from_value(config)?;
        Ok(Self {
            control_topic: cfg.control_topic,
            request_topic: cfg.request_topic,
            action_set: cfg.action_set,
            action_clear: cfg.action_clear,
            payload_key: cfg.payload_key,
            target: cfg.initial,
            receiver: None,
            published_initial: false,
            last_status: None,
        })
    }

    async fn publish_set(&self, ctx: &Context, value: &Value) -> anyhow::Result<()> {
        let mut params = serde_json::Map::new();
        params.insert(self.payload_key.clone(), value.clone());
        let command = serde_json::json!({
            "action": self.action_set,
            "params": Value::Object(params),
        });
        ctx.publish_raw(&self.control_topic, &command).await?;
        tracing::debug!("SetCapabilityTargetNode: published {}({:?})", self.action_set, value);
        Ok(())
    }

    async fn publish_clear(&self, ctx: &Context) -> anyhow::Result<()> {
        let Some(action_clear) = self.action_clear.clone() else {
            return Ok(());
        };
        let command = serde_json::json!({
            "action": action_clear,
            "params": Value::Null,
        });
        ctx.publish_raw(&self.control_topic, &command).await?;
        tracing::debug!("SetCapabilityTargetNode: published {}", action_clear);
        Ok(())
    }

    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
    /// the most recent request matters. A request with `{<payload_key>: null}` (or no
    /// `payload_key` key at all) means "clear".
    ///
    /// A free function (not a `&self` method) so callers can borrow `self.receiver` mutably and
    /// `self.payload_key` immutably at the same time without the borrow checker treating them as
    /// one overlapping borrow of all of `self`.
    fn poll_latest_request(payload_key: &str, receiver: &mut Receiver<Value>) -> Option<Option<Value>> {
        let mut latest = None;
        while let Ok(value) = receiver.try_recv() {
            match value.as_object() {
                Some(obj) => {
                    let payload = obj.get(payload_key).cloned().filter(|v| !v.is_null());
                    latest = Some(payload);
                }
                None => {
                    tracing::warn!("SetCapabilityTargetNode: dropping malformed target request: expected a JSON object")
                }
            }
        }
        latest
    }
}

#[async_trait]
impl BehaviorNode for SetCapabilityTargetNode {
    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 let Some(target) = self.target.clone() {
                self.publish_set(ctx, &target).await?;
            }
        }

        let latest = Self::poll_latest_request(
            &self.payload_key,
            self.receiver.as_mut().expect("just subscribed above"),
        );
        if let Some(value) = latest {
            self.target = value.clone();
            match value {
                Some(value) => self.publish_set(ctx, &value).await?,
                None => self.publish_clear(ctx).await?,
            }
        }

        let status = if self.target.is_some() || self.action_clear.is_none() {
            NodeStatus::Running
        } else {
            NodeStatus::Failure
        };
        self.last_status = Some(status);
        Ok(status)
    }

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

    fn status_snapshot(&self) -> TreeStatusSnapshot {
        TreeStatusSnapshot {
            name: self.name().to_string(),
            status: self.last_status,
            active_capability: self.active_capability().map(String::from),
            children: Vec::new(),
        }
    }
}

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

    fn rl_config(extra: serde_json::Value) -> Value {
        let mut base = serde_json::json!({
            "control_topic": "/rl/control",
            "request_topic": "/rl/target/request",
            "action_set": "set_target_class",
            "payload_key": "target_class"
        });
        merge(&mut base, extra);
        base
    }

    fn instance_track_config(extra: serde_json::Value) -> Value {
        let mut base = serde_json::json!({
            "control_topic": "/instance-track/control",
            "request_topic": "/instance-track/target/request",
            "action_set": "set_target",
            "action_clear": "clear",
            "payload_key": "bbox"
        });
        merge(&mut base, extra);
        base
    }

    fn merge(base: &mut Value, extra: Value) {
        if let (Some(base), Some(extra)) = (base.as_object_mut(), extra.as_object()) {
            for (k, v) in extra {
                base.insert(k.clone(), v.clone());
            }
        }
    }

    fn sample_bbox() -> Value {
        serde_json::json!({ "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 })
    }

    #[test]
    fn from_json_reads_rl_shaped_config() {
        let node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({ "initial": "cat" }))).unwrap();

        assert_eq!(node.control_topic, "/rl/control");
        assert_eq!(node.request_topic, "/rl/target/request");
        assert_eq!(node.action_set, "set_target_class");
        assert_eq!(node.action_clear, None);
        assert_eq!(node.payload_key, "target_class");
        assert_eq!(node.target, Some(serde_json::json!("cat")));
    }

    #[test]
    fn from_json_reads_instance_track_shaped_config() {
        let node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
            "initial": sample_bbox()
        })))
        .unwrap();

        assert_eq!(node.control_topic, "/instance-track/control");
        assert_eq!(node.request_topic, "/instance-track/target/request");
        assert_eq!(node.action_set, "set_target");
        assert_eq!(node.action_clear, Some("clear".to_string()));
        assert_eq!(node.payload_key, "bbox");
        assert_eq!(node.target, Some(sample_bbox()));
    }

    #[test]
    fn from_json_allows_no_initial_target() {
        let node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).unwrap();
        assert_eq!(node.target, None);
    }

    #[test]
    fn from_json_requires_the_core_fields() {
        assert!(SetCapabilityTargetNode::from_json(serde_json::json!({})).is_err());
        assert!(SetCapabilityTargetNode::from_json(serde_json::json!({
            "control_topic": "/rl/control"
        }))
        .is_err());
    }

    #[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!(
            SetCapabilityTargetNode::poll_latest_request("target_class", &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!(
            SetCapabilityTargetNode::poll_latest_request("target_class", &mut receiver),
            Some(Some(serde_json::json!("cat")))
        );
    }

    #[test]
    fn poll_latest_request_accepts_a_null_value_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!(
            SetCapabilityTargetNode::poll_latest_request("target_class", &mut receiver),
            Some(None)
        );
    }

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

        assert_eq!(
            SetCapabilityTargetNode::poll_latest_request("bbox", &mut receiver),
            Some(None)
        );
    }

    #[test]
    fn poll_latest_request_keeps_the_most_recent_bbox() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "bbox": { "x": 0.0, "y": 0.0, "width": 0.1, "height": 0.1 } }))
            .unwrap();
        tx.send(serde_json::json!({ "bbox": sample_bbox() })).unwrap();

        assert_eq!(
            SetCapabilityTargetNode::poll_latest_request("bbox", &mut receiver),
            Some(Some(sample_bbox()))
        );
    }

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

        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({ "initial": "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_rl() {
        let ctx = create_test_context("test_set_capability_target_rl_request").await;

        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).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");
    }

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

        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
            "initial": sample_bbox()
        })))
        .unwrap();

        let mut receiver = ctx.subscribe_raw::<Value>("/instance-track/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");
        assert_eq!(received["params"]["bbox"]["x"], 0.1);

        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_instance_track() {
        let ctx = create_test_context("test_set_capability_target_instance_track_request").await;

        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();

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

        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
        ctx.publish_raw(
            "/instance-track/target/request",
            &serde_json::json!({ "bbox": { "x": 0.5, "y": 0.5, "width": 0.1, "height": 0.1 } }),
        )
        .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");
        assert_eq!(received["params"]["bbox"]["x"], 0.5);
        assert_eq!(
            node.tick(&ctx).await.unwrap(),
            NodeStatus::Running,
            "a live target must report Running"
        );
    }

    #[tokio::test]
    #[ignore = "Requires Redis infrastructure"]
    async fn tick_returns_failure_when_no_target_has_ever_been_set_and_action_clear_is_configured() {
        let ctx = create_test_context("test_set_capability_target_idle").await;

        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(
            status,
            NodeStatus::Failure,
            "no target yet means cede control, not Success"
        );
    }

    #[tokio::test]
    #[ignore = "Requires Redis infrastructure"]
    async fn tick_stays_running_when_no_target_has_ever_been_set_and_no_action_clear_is_configured() {
        let ctx = create_test_context("test_set_capability_target_rl_idle").await;

        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).unwrap();

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(
            status,
            NodeStatus::Running,
            "rl-style config (no action_clear) always reports Running, matching the old SetRlTargetNode"
        );
    }

    #[tokio::test]
    #[ignore = "Requires Redis infrastructure"]
    async fn tick_returns_failure_after_a_clear_request_even_with_a_configured_initial_target() {
        let ctx = create_test_context("test_set_capability_target_cleared").await;

        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
            "initial": sample_bbox()
        })))
        .unwrap();

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

        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Failure);
    }

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

        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();

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

        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
            .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"], "clear");
    }
}