mecha10_behavior_runtime/actions/
set_instance_track_target.rs

1//! Action node that relays live target requests to the `instance-track` node's control topic.
2//!
3//! Mirrors `set_rl_target.rs` exactly - the only difference is the relayed command shape:
4//! `instance-track`'s `set_target` carries a `bbox` (the operator's click/drag-box on the
5//! dashboard's live camera feed) rather than a `target_class` string, and a `clear` request (no
6//! params) stops tracking instead of falling back to "no class filter".
7
8use crate::{BehaviorNode, Context, NodeStatus};
9use async_trait::async_trait;
10use mecha10_core::messages::Message;
11use mecha10_core::topics::Topic;
12use mecha10_core::Receiver;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
17pub(crate) struct BoundingBox {
18    x: f32,
19    y: f32,
20    width: f32,
21    height: f32,
22}
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
25struct InstanceTrackControlCommand {
26    action: String,
27    #[serde(skip_serializing_if = "Option::is_none")]
28    params: Option<Value>,
29}
30
31impl Message for InstanceTrackControlCommand {}
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub(crate) struct InstanceTrackTargetRequest {
35    #[serde(default)]
36    pub(crate) bbox: Option<BoundingBox>,
37}
38
39impl Message for InstanceTrackTargetRequest {}
40
41fn default_topic() -> String {
42    "/instance-track/control".to_string()
43}
44
45/// The topic live target requests (clicks/drag-boxes on the dashboard's camera feed) arrive on.
46/// `pub(crate)` so `InstanceTrackTaskNode` can subscribe to the same topic to track whether a
47/// target is currently live, without hardcoding the string a second time.
48pub(crate) fn default_request_topic() -> String {
49    "/instance-track/target/request".to_string()
50}
51
52pub struct SetInstanceTrackTargetNode {
53    topic: String,
54    request_topic: String,
55    target: Option<BoundingBox>,
56    receiver: Option<Receiver<Value>>,
57    published_initial: bool,
58}
59
60impl std::fmt::Debug for SetInstanceTrackTargetNode {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        f.debug_struct("SetInstanceTrackTargetNode")
63            .field("topic", &self.topic)
64            .field("request_topic", &self.request_topic)
65            .field("target", &self.target)
66            .field("subscribed", &self.receiver.is_some())
67            .field("published_initial", &self.published_initial)
68            .finish()
69    }
70}
71
72#[derive(Debug, Serialize, Deserialize)]
73struct SetInstanceTrackTargetConfig {
74    #[serde(default = "default_topic")]
75    topic: String,
76    #[serde(default = "default_request_topic")]
77    request_topic: String,
78    #[serde(default)]
79    bbox: Option<BoundingBox>,
80}
81
82impl SetInstanceTrackTargetNode {
83    /// Create a new SetInstanceTrackTargetNode from JSON configuration. See `SetRlTargetNode::from_json`'s
84    /// doc comment for why `null` is normalized to `{}` here.
85    pub fn from_json(config: Value) -> anyhow::Result<Self> {
86        let config = if config.is_null() {
87            serde_json::json!({})
88        } else {
89            config
90        };
91        let cfg: SetInstanceTrackTargetConfig = serde_json::from_value(config)?;
92        Ok(Self {
93            topic: cfg.topic,
94            request_topic: cfg.request_topic,
95            target: cfg.bbox,
96            receiver: None,
97            published_initial: false,
98        })
99    }
100
101    async fn publish_set_target(&self, ctx: &Context, bbox: BoundingBox) -> anyhow::Result<()> {
102        let command = InstanceTrackControlCommand {
103            action: "set_target".to_string(),
104            params: Some(serde_json::json!({ "bbox": bbox })),
105        };
106
107        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
108        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);
109
110        ctx.publish_to(topic, &command).await?;
111        tracing::debug!("SetInstanceTrackTargetNode: published set_target({:?})", bbox);
112        Ok(())
113    }
114
115    async fn publish_clear(&self, ctx: &Context) -> anyhow::Result<()> {
116        let command = InstanceTrackControlCommand {
117            action: "clear".to_string(),
118            params: None,
119        };
120
121        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
122        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);
123
124        ctx.publish_to(topic, &command).await?;
125        tracing::debug!("SetInstanceTrackTargetNode: published clear");
126        Ok(())
127    }
128
129    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
130    /// the most recent request matters, same as `SetRlTargetNode::poll_latest_request`. A
131    /// request with `bbox: null` (or no `bbox` key at all) means "clear".
132    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<BoundingBox>> {
133        let mut latest = None;
134        while let Ok(value) = receiver.try_recv() {
135            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
136                Ok(request) => latest = Some(request.bbox),
137                Err(e) => tracing::warn!("SetInstanceTrackTargetNode: dropping malformed target request: {}", e),
138            }
139        }
140        latest
141    }
142}
143
144#[async_trait]
145impl BehaviorNode for SetInstanceTrackTargetNode {
146    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
147        if self.receiver.is_none() {
148            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
149        }
150
151        if !self.published_initial {
152            self.published_initial = true;
153            if let Some(bbox) = self.target {
154                self.publish_set_target(ctx, bbox).await?;
155            }
156        }
157
158        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
159        if let Some(bbox) = latest {
160            self.target = bbox;
161            match bbox {
162                Some(bbox) => self.publish_set_target(ctx, bbox).await?,
163                None => self.publish_clear(ctx).await?,
164            }
165        }
166
167        // `Failure` (never `Success`) when there's no live target - this is the "cede control"
168        // signal a `Watch`-wrapped interrupt layer relies on. See `instance_track_task.rs`'s
169        // module doc for the full reasoning on why `Success` here would be wrong.
170        Ok(if self.target.is_some() {
171            NodeStatus::Running
172        } else {
173            NodeStatus::Failure
174        })
175    }
176
177    fn name(&self) -> &str {
178        "SetInstanceTrackTargetNode"
179    }
180}
181
182#[cfg(test)]
183mod tests {
184    use super::*;
185    use crate::test_helpers::create_test_context;
186
187    fn sample_bbox() -> BoundingBox {
188        BoundingBox {
189            x: 0.1,
190            y: 0.2,
191            width: 0.3,
192            height: 0.4,
193        }
194    }
195
196    #[test]
197    fn from_json_reads_topic_request_topic_and_bbox() {
198        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
199            "topic": "/instance-track/control",
200            "request_topic": "/instance-track/target/request",
201            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
202        }))
203        .unwrap();
204
205        assert_eq!(node.topic, "/instance-track/control");
206        assert_eq!(node.request_topic, "/instance-track/target/request");
207        assert_eq!(node.target, Some(sample_bbox()));
208    }
209
210    #[test]
211    fn from_json_defaults_topics_and_allows_no_target() {
212        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({})).unwrap();
213
214        assert_eq!(node.topic, "/instance-track/control");
215        assert_eq!(node.request_topic, "/instance-track/target/request");
216        assert_eq!(node.target, None);
217    }
218
219    #[test]
220    fn from_json_accepts_null_the_same_as_an_empty_object() {
221        let node = SetInstanceTrackTargetNode::from_json(Value::Null).unwrap();
222
223        assert_eq!(node.topic, "/instance-track/control");
224        assert_eq!(node.request_topic, "/instance-track/target/request");
225        assert_eq!(node.target, None);
226    }
227
228    #[test]
229    fn poll_latest_request_returns_none_when_empty() {
230        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
231        let mut receiver = Receiver::from(rx);
232        assert_eq!(SetInstanceTrackTargetNode::poll_latest_request(&mut receiver), None);
233    }
234
235    #[test]
236    fn poll_latest_request_keeps_only_the_most_recent_value() {
237        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
238        let mut receiver = Receiver::from(rx);
239        tx.send(serde_json::json!({ "bbox": { "x": 0.0, "y": 0.0, "width": 0.1, "height": 0.1 } }))
240            .unwrap();
241        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
242            .unwrap();
243
244        assert_eq!(
245            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
246            Some(Some(sample_bbox()))
247        );
248    }
249
250    #[test]
251    fn poll_latest_request_accepts_a_null_bbox_as_a_clear_request() {
252        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
253        let mut receiver = Receiver::from(rx);
254        tx.send(serde_json::json!({ "bbox": null })).unwrap();
255
256        assert_eq!(
257            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
258            Some(None)
259        );
260    }
261
262    #[test]
263    fn poll_latest_request_accepts_an_absent_bbox_key_as_a_clear_request() {
264        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
265        let mut receiver = Receiver::from(rx);
266        tx.send(serde_json::json!({})).unwrap();
267
268        assert_eq!(
269            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
270            Some(None)
271        );
272    }
273
274    #[test]
275    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
276        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
277        let mut receiver = Receiver::from(rx);
278        tx.send(serde_json::json!({ "bbox": "not-a-bbox" })).unwrap();
279        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
280            .unwrap();
281
282        assert_eq!(
283            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
284            Some(Some(sample_bbox()))
285        );
286    }
287
288    #[tokio::test]
289    #[ignore]
290    async fn tick_publishes_configured_target_once_then_stays_running() {
291        let ctx = create_test_context("test_set_instance_track_target").await;
292
293        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
294            "topic": "/instance-track/control",
295            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
296        }))
297        .unwrap();
298
299        let mut receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
300
301        let status = node.tick(&ctx).await.unwrap();
302        assert_eq!(status, NodeStatus::Running);
303
304        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
305            .await
306            .expect("timed out waiting for published command")
307            .expect("channel closed without a message");
308
309        assert_eq!(received["action"], "set_target");
310        assert_eq!(received["params"]["bbox"]["x"], 0.1);
311
312        let status = node.tick(&ctx).await.unwrap();
313        assert_eq!(status, NodeStatus::Running);
314        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
315        assert!(timed_out.is_err(), "expected no second publish without a new request");
316    }
317
318    #[tokio::test]
319    #[ignore]
320    async fn tick_relays_a_live_request_published_on_request_topic() {
321        let ctx = create_test_context("test_set_instance_track_target_request").await;
322
323        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
324            "topic": "/instance-track/control",
325            "request_topic": "/instance-track/target/request"
326        }))
327        .unwrap();
328
329        node.tick(&ctx).await.unwrap();
330
331        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
332        ctx.publish_raw(
333            "/instance-track/target/request",
334            &serde_json::json!({ "bbox": { "x": 0.5, "y": 0.5, "width": 0.1, "height": 0.1 } }),
335        )
336        .await
337        .unwrap();
338
339        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
340        node.tick(&ctx).await.unwrap();
341
342        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
343            .await
344            .expect("timed out waiting for relayed command")
345            .expect("channel closed without a message");
346
347        assert_eq!(received["action"], "set_target");
348        assert_eq!(received["params"]["bbox"]["x"], 0.5);
349        assert_eq!(
350            node.tick(&ctx).await.unwrap(),
351            NodeStatus::Running,
352            "a live target must report Running"
353        );
354    }
355
356    #[tokio::test]
357    #[ignore = "Requires Redis infrastructure"]
358    async fn tick_returns_failure_when_no_target_has_ever_been_set() {
359        let ctx = create_test_context("test_set_instance_track_target_idle").await;
360
361        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
362            "request_topic": "/instance-track/target/request"
363        }))
364        .unwrap();
365
366        let status = node.tick(&ctx).await.unwrap();
367        assert_eq!(
368            status,
369            NodeStatus::Failure,
370            "no target yet means cede control, not Success"
371        );
372    }
373
374    #[tokio::test]
375    #[ignore = "Requires Redis infrastructure"]
376    async fn tick_returns_failure_after_a_clear_request_even_with_a_configured_initial_target() {
377        let ctx = create_test_context("test_set_instance_track_target_cleared").await;
378
379        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
380            "request_topic": "/instance-track/target/request",
381            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
382        }))
383        .unwrap();
384
385        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Running);
386
387        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
388            .await
389            .unwrap();
390        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
391
392        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Failure);
393    }
394
395    #[tokio::test]
396    #[ignore]
397    async fn tick_relays_a_clear_request_published_on_request_topic() {
398        let ctx = create_test_context("test_set_instance_track_target_clear").await;
399
400        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
401            "topic": "/instance-track/control",
402            "request_topic": "/instance-track/target/request"
403        }))
404        .unwrap();
405
406        node.tick(&ctx).await.unwrap();
407
408        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
409        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
410            .await
411            .unwrap();
412
413        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
414        node.tick(&ctx).await.unwrap();
415
416        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
417            .await
418            .expect("timed out waiting for relayed command")
419            .expect("channel closed without a message");
420
421        assert_eq!(received["action"], "clear");
422    }
423}
mecha10_behavior_runtime/actions/set_instance_track_target.rs

mecha10_behavior_runtime/actions/
set_instance_track_target.rs
