mecha10_behavior_runtime/actions/
instance_track_task.rs

1//! Action node that hands control to the `instance-track` node while a target is live, and cedes
2//! control back the moment there isn't one. Resolves to `NodeStatus::Running` while a target is
3//! live and `NodeStatus::Failure` when idle.
4
5use crate::actions::set_instance_track_target::{default_request_topic, InstanceTrackTargetRequest};
6use crate::{BehaviorNode, Context, NodeStatus};
7use async_trait::async_trait;
8use mecha10_core::Receiver;
9use serde_json::Value;
10
11const INSTANCE_TRACK_CAPABILITY: &str = "instance-track";
12
13const TASK_COMPLETE_TOPIC: &str = "/instance-track/task_complete";
14
15pub struct InstanceTrackTaskNode {
16    complete_receiver: Option<Receiver<Value>>,
17    target_receiver: Option<Receiver<Value>>,
18    request_topic: String,
19    has_target: bool,
20}
21
22impl std::fmt::Debug for InstanceTrackTaskNode {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_struct("InstanceTrackTaskNode")
25            .field("complete_subscribed", &self.complete_receiver.is_some())
26            .field("target_subscribed", &self.target_receiver.is_some())
27            .field("has_target", &self.has_target)
28            .finish()
29    }
30}
31
32impl InstanceTrackTaskNode {
33    /// Construct a node. `_config` is ignored - this node takes none.
34    pub fn from_json(_config: Value) -> anyhow::Result<Self> {
35        Ok(Self {
36            complete_receiver: None,
37            target_receiver: None,
38            request_topic: default_request_topic(),
39            has_target: false,
40        })
41    }
42
43    /// Construct a node with already-connected receivers, bypassing live subscription.
44    ///
45    /// Used by unit tests to exercise convergence without a real `Context`/Redis.
46    #[cfg(test)]
47    fn with_receivers(complete_receiver: Receiver<Value>, target_receiver: Receiver<Value>) -> Self {
48        Self {
49            complete_receiver: Some(complete_receiver),
50            target_receiver: Some(target_receiver),
51            request_topic: default_request_topic(),
52            has_target: false,
53        }
54    }
55
56    /// Drains every pending target request, updating `has_target` from the last one seen (if
57    /// any) - mirrors `SetInstanceTrackTargetNode::poll_latest_request`'s "only the most recent
58    /// request matters" semantics.
59    fn poll_target(&mut self) {
60        let Some(receiver) = self.target_receiver.as_mut() else {
61            return;
62        };
63
64        let mut latest = None;
65        while let Ok(value) = receiver.try_recv() {
66            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
67                Ok(request) => latest = Some(request.bbox.is_some()),
68                Err(e) => tracing::warn!("InstanceTrackTaskNode: dropping malformed target request: {}", e),
69            }
70        }
71
72        if let Some(has_target) = latest {
73            self.has_target = has_target;
74        }
75    }
76
77    /// Drain any pending completion messages - arrival of any message at all is the whole
78    /// signal. Proactively clears `has_target` so this node converges to "idle" immediately.
79    fn poll_complete(&mut self) {
80        let Some(receiver) = self.complete_receiver.as_mut() else {
81            return;
82        };
83
84        if receiver.try_recv().is_ok() {
85            self.has_target = false;
86        }
87    }
88}
89
90#[async_trait]
91impl BehaviorNode for InstanceTrackTaskNode {
92    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
93        if self.complete_receiver.is_none() {
94            self.complete_receiver = Some(ctx.subscribe_raw::<Value>(TASK_COMPLETE_TOPIC).await?);
95        }
96        if self.target_receiver.is_none() {
97            self.target_receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
98        }
99
100        self.poll_target();
101        self.poll_complete();
102
103        Ok(if self.has_target {
104            NodeStatus::Running
105        } else {
106            NodeStatus::Failure
107        })
108    }
109
110    fn name(&self) -> &str {
111        "instance_track_task"
112    }
113
114    fn active_capability(&self) -> Option<&str> {
115        Some(INSTANCE_TRACK_CAPABILITY)
116    }
117
118    fn reset_self(&mut self) {
119        self.has_target = false;
120    }
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126    use tokio::sync::mpsc;
127
128    fn node_with_channels() -> (
129        InstanceTrackTaskNode,
130        mpsc::UnboundedSender<Value>,
131        mpsc::UnboundedSender<Value>,
132    ) {
133        let (complete_tx, complete_rx) = mpsc::unbounded_channel();
134        let (target_tx, target_rx) = mpsc::unbounded_channel();
135        (
136            InstanceTrackTaskNode::with_receivers(Receiver::from(complete_rx), Receiver::from(target_rx)),
137            complete_tx,
138            target_tx,
139        )
140    }
141
142    #[test]
143    fn always_declares_the_instance_track_capability() {
144        let (node, _complete_tx, _target_tx) = node_with_channels();
145        assert_eq!(node.active_capability(), Some("instance-track"));
146    }
147
148    #[test]
149    fn no_target_ever_set_stays_without_a_target() {
150        let (mut node, _complete_tx, _target_tx) = node_with_channels();
151        node.poll_target();
152        node.poll_complete();
153        assert!(!node.has_target);
154    }
155
156    #[test]
157    fn a_live_target_request_sets_has_target() {
158        let (mut node, _complete_tx, target_tx) = node_with_channels();
159        target_tx
160            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
161            .unwrap();
162
163        node.poll_target();
164
165        assert!(node.has_target);
166    }
167
168    #[test]
169    fn a_clear_request_unsets_has_target_even_with_no_task_complete() {
170        let (mut node, _complete_tx, target_tx) = node_with_channels();
171        target_tx
172            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
173            .unwrap();
174        node.poll_target();
175        assert!(node.has_target);
176
177        target_tx.send(serde_json::json!({ "bbox": null })).unwrap();
178        node.poll_target();
179
180        assert!(!node.has_target);
181    }
182
183    #[test]
184    fn task_complete_unsets_has_target_on_the_very_next_poll() {
185        let (mut node, complete_tx, target_tx) = node_with_channels();
186        target_tx
187            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
188            .unwrap();
189        node.poll_target();
190        assert!(node.has_target);
191
192        complete_tx.send(serde_json::json!({})).unwrap();
193        node.poll_complete();
194
195        assert!(
196            !node.has_target,
197            "task_complete must converge has_target to false without waiting for a separate clear request"
198        );
199    }
200
201    #[test]
202    fn reset_self_clears_has_target_for_a_later_pass() {
203        let (mut node, _complete_tx, target_tx) = node_with_channels();
204        target_tx
205            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
206            .unwrap();
207        node.poll_target();
208        assert!(node.has_target);
209
210        node.reset_self();
211
212        assert!(!node.has_target);
213    }
214
215    #[test]
216    fn from_json_ignores_any_input() {
217        assert!(InstanceTrackTaskNode::from_json(Value::Null).is_ok());
218        assert!(InstanceTrackTaskNode::from_json(serde_json::json!({"unexpected": "field"})).is_ok());
219    }
220}
mecha10_behavior_runtime/actions/instance_track_task.rs

mecha10_behavior_runtime/actions/
instance_track_task.rs
