mecha10-behavior-runtime 0.8.0

//! Action node that hands control to the `instance-track` node while a target is live, and cedes
//! control back the moment there isn't one. Resolves to `NodeStatus::Running` while a target is
//! live and `NodeStatus::Failure` when idle.

use crate::actions::set_instance_track_target::{default_request_topic, InstanceTrackTargetRequest};
use crate::{BehaviorNode, Context, NodeStatus};
use async_trait::async_trait;
use mecha10_core::Receiver;
use serde_json::Value;

const INSTANCE_TRACK_CAPABILITY: &str = "instance-track";

const TASK_COMPLETE_TOPIC: &str = "/instance-track/task_complete";

pub struct InstanceTrackTaskNode {
    complete_receiver: Option<Receiver<Value>>,
    target_receiver: Option<Receiver<Value>>,
    request_topic: String,
    has_target: bool,
}

impl std::fmt::Debug for InstanceTrackTaskNode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InstanceTrackTaskNode")
            .field("complete_subscribed", &self.complete_receiver.is_some())
            .field("target_subscribed", &self.target_receiver.is_some())
            .field("has_target", &self.has_target)
            .finish()
    }
}

impl InstanceTrackTaskNode {
    /// Construct a node. `_config` is ignored - this node takes none.
    pub fn from_json(_config: Value) -> anyhow::Result<Self> {
        Ok(Self {
            complete_receiver: None,
            target_receiver: None,
            request_topic: default_request_topic(),
            has_target: false,
        })
    }

    /// Construct a node with already-connected receivers, bypassing live subscription.
    ///
    /// Used by unit tests to exercise convergence without a real `Context`/Redis.
    #[cfg(test)]
    fn with_receivers(complete_receiver: Receiver<Value>, target_receiver: Receiver<Value>) -> Self {
        Self {
            complete_receiver: Some(complete_receiver),
            target_receiver: Some(target_receiver),
            request_topic: default_request_topic(),
            has_target: false,
        }
    }

    /// Drains every pending target request, updating `has_target` from the last one seen (if
    /// any) - mirrors `SetInstanceTrackTargetNode::poll_latest_request`'s "only the most recent
    /// request matters" semantics.
    fn poll_target(&mut self) {
        let Some(receiver) = self.target_receiver.as_mut() else {
            return;
        };

        let mut latest = None;
        while let Ok(value) = receiver.try_recv() {
            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
                Ok(request) => latest = Some(request.bbox.is_some()),
                Err(e) => tracing::warn!("InstanceTrackTaskNode: dropping malformed target request: {}", e),
            }
        }

        if let Some(has_target) = latest {
            self.has_target = has_target;
        }
    }

    /// Drain any pending completion messages - arrival of any message at all is the whole
    /// signal. Proactively clears `has_target` so this node converges to "idle" immediately.
    fn poll_complete(&mut self) {
        let Some(receiver) = self.complete_receiver.as_mut() else {
            return;
        };

        if receiver.try_recv().is_ok() {
            self.has_target = false;
        }
    }
}

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

        self.poll_target();
        self.poll_complete();

        Ok(if self.has_target {
            NodeStatus::Running
        } else {
            NodeStatus::Failure
        })
    }

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

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

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

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

    fn node_with_channels() -> (
        InstanceTrackTaskNode,
        mpsc::UnboundedSender<Value>,
        mpsc::UnboundedSender<Value>,
    ) {
        let (complete_tx, complete_rx) = mpsc::unbounded_channel();
        let (target_tx, target_rx) = mpsc::unbounded_channel();
        (
            InstanceTrackTaskNode::with_receivers(Receiver::from(complete_rx), Receiver::from(target_rx)),
            complete_tx,
            target_tx,
        )
    }

    #[test]
    fn always_declares_the_instance_track_capability() {
        let (node, _complete_tx, _target_tx) = node_with_channels();
        assert_eq!(node.active_capability(), Some("instance-track"));
    }

    #[test]
    fn no_target_ever_set_stays_without_a_target() {
        let (mut node, _complete_tx, _target_tx) = node_with_channels();
        node.poll_target();
        node.poll_complete();
        assert!(!node.has_target);
    }

    #[test]
    fn a_live_target_request_sets_has_target() {
        let (mut node, _complete_tx, target_tx) = node_with_channels();
        target_tx
            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();

        node.poll_target();

        assert!(node.has_target);
    }

    #[test]
    fn a_clear_request_unsets_has_target_even_with_no_task_complete() {
        let (mut node, _complete_tx, target_tx) = node_with_channels();
        target_tx
            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();
        node.poll_target();
        assert!(node.has_target);

        target_tx.send(serde_json::json!({ "bbox": null })).unwrap();
        node.poll_target();

        assert!(!node.has_target);
    }

    #[test]
    fn task_complete_unsets_has_target_on_the_very_next_poll() {
        let (mut node, complete_tx, target_tx) = node_with_channels();
        target_tx
            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();
        node.poll_target();
        assert!(node.has_target);

        complete_tx.send(serde_json::json!({})).unwrap();
        node.poll_complete();

        assert!(
            !node.has_target,
            "task_complete must converge has_target to false without waiting for a separate clear request"
        );
    }

    #[test]
    fn reset_self_clears_has_target_for_a_later_pass() {
        let (mut node, _complete_tx, target_tx) = node_with_channels();
        target_tx
            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();
        node.poll_target();
        assert!(node.has_target);

        node.reset_self();

        assert!(!node.has_target);
    }

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