mecha10_behavior_runtime/actions/
set_capability_target.rs

1//! Generic action node that relays live target requests to a capability node's control topic.
2//!
3//! Replaces the old bespoke `SetRlTargetNode`/`SetInstanceTrackTargetNode` pair. Where the old
4//! nodes hardcoded their topic names and the shape of the published command, this node takes
5//! those as config so one node type covers both (and any future capability with the same
6//! "relay a latest-wins target request onto a control topic" shape).
7
8use crate::{BehaviorNode, Context, NodeStatus, TreeStatusSnapshot};
9use async_trait::async_trait;
10use mecha10_core::Receiver;
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13
14#[derive(Debug, Serialize, Deserialize)]
15struct SetCapabilityTargetConfig {
16    control_topic: String,
17    request_topic: String,
18    action_set: String,
19    #[serde(default)]
20    action_clear: Option<String>,
21    payload_key: String,
22    #[serde(default)]
23    initial: Option<Value>,
24}
25
26pub struct SetCapabilityTargetNode {
27    control_topic: String,
28    request_topic: String,
29    action_set: String,
30    action_clear: Option<String>,
31    payload_key: String,
32    target: Option<Value>,
33    receiver: Option<Receiver<Value>>,
34    published_initial: bool,
35    last_status: Option<NodeStatus>,
36}
37
38impl std::fmt::Debug for SetCapabilityTargetNode {
39    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40        f.debug_struct("SetCapabilityTargetNode")
41            .field("control_topic", &self.control_topic)
42            .field("request_topic", &self.request_topic)
43            .field("action_set", &self.action_set)
44            .field("action_clear", &self.action_clear)
45            .field("payload_key", &self.payload_key)
46            .field("target", &self.target)
47            .field("subscribed", &self.receiver.is_some())
48            .field("published_initial", &self.published_initial)
49            .finish()
50    }
51}
52
53impl SetCapabilityTargetNode {
54    /// Create a new node from JSON configuration. `control_topic`, `request_topic`,
55    /// `action_set`, and `payload_key` are required; `action_clear` and `initial` are optional.
56    pub fn from_json(config: Value) -> anyhow::Result<Self> {
57        let cfg: SetCapabilityTargetConfig = serde_json::from_value(config)?;
58        Ok(Self {
59            control_topic: cfg.control_topic,
60            request_topic: cfg.request_topic,
61            action_set: cfg.action_set,
62            action_clear: cfg.action_clear,
63            payload_key: cfg.payload_key,
64            target: cfg.initial,
65            receiver: None,
66            published_initial: false,
67            last_status: None,
68        })
69    }
70
71    async fn publish_set(&self, ctx: &Context, value: &Value) -> anyhow::Result<()> {
72        let mut params = serde_json::Map::new();
73        params.insert(self.payload_key.clone(), value.clone());
74        let command = serde_json::json!({
75            "action": self.action_set,
76            "params": Value::Object(params),
77        });
78        ctx.publish_raw(&self.control_topic, &command).await?;
79        tracing::debug!("SetCapabilityTargetNode: published {}({:?})", self.action_set, value);
80        Ok(())
81    }
82
83    async fn publish_clear(&self, ctx: &Context) -> anyhow::Result<()> {
84        let Some(action_clear) = self.action_clear.clone() else {
85            return Ok(());
86        };
87        let command = serde_json::json!({
88            "action": action_clear,
89            "params": Value::Null,
90        });
91        ctx.publish_raw(&self.control_topic, &command).await?;
92        tracing::debug!("SetCapabilityTargetNode: published {}", action_clear);
93        Ok(())
94    }
95
96    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
97    /// the most recent request matters. A request with `{<payload_key>: null}` (or no
98    /// `payload_key` key at all) means "clear".
99    ///
100    /// A free function (not a `&self` method) so callers can borrow `self.receiver` mutably and
101    /// `self.payload_key` immutably at the same time without the borrow checker treating them as
102    /// one overlapping borrow of all of `self`.
103    fn poll_latest_request(payload_key: &str, receiver: &mut Receiver<Value>) -> Option<Option<Value>> {
104        let mut latest = None;
105        while let Ok(value) = receiver.try_recv() {
106            match value.as_object() {
107                Some(obj) => {
108                    let payload = obj.get(payload_key).cloned().filter(|v| !v.is_null());
109                    latest = Some(payload);
110                }
111                None => {
112                    tracing::warn!("SetCapabilityTargetNode: dropping malformed target request: expected a JSON object")
113                }
114            }
115        }
116        latest
117    }
118}
119
120#[async_trait]
121impl BehaviorNode for SetCapabilityTargetNode {
122    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
123        if self.receiver.is_none() {
124            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
125        }
126
127        if !self.published_initial {
128            self.published_initial = true;
129            if let Some(target) = self.target.clone() {
130                self.publish_set(ctx, &target).await?;
131            }
132        }
133
134        let latest = Self::poll_latest_request(
135            &self.payload_key,
136            self.receiver.as_mut().expect("just subscribed above"),
137        );
138        if let Some(value) = latest {
139            self.target = value.clone();
140            match value {
141                Some(value) => self.publish_set(ctx, &value).await?,
142                None => self.publish_clear(ctx).await?,
143            }
144        }
145
146        let status = if self.target.is_some() || self.action_clear.is_none() {
147            NodeStatus::Running
148        } else {
149            NodeStatus::Failure
150        };
151        self.last_status = Some(status);
152        Ok(status)
153    }
154
155    fn name(&self) -> &str {
156        "set_capability_target"
157    }
158
159    fn status_snapshot(&self) -> TreeStatusSnapshot {
160        TreeStatusSnapshot {
161            name: self.name().to_string(),
162            status: self.last_status,
163            active_capability: self.active_capability().map(String::from),
164            children: Vec::new(),
165        }
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use super::*;
172    use crate::test_helpers::create_test_context;
173
174    fn rl_config(extra: serde_json::Value) -> Value {
175        let mut base = serde_json::json!({
176            "control_topic": "/rl/control",
177            "request_topic": "/rl/target/request",
178            "action_set": "set_target_class",
179            "payload_key": "target_class"
180        });
181        merge(&mut base, extra);
182        base
183    }
184
185    fn instance_track_config(extra: serde_json::Value) -> Value {
186        let mut base = serde_json::json!({
187            "control_topic": "/instance-track/control",
188            "request_topic": "/instance-track/target/request",
189            "action_set": "set_target",
190            "action_clear": "clear",
191            "payload_key": "bbox"
192        });
193        merge(&mut base, extra);
194        base
195    }
196
197    fn merge(base: &mut Value, extra: Value) {
198        if let (Some(base), Some(extra)) = (base.as_object_mut(), extra.as_object()) {
199            for (k, v) in extra {
200                base.insert(k.clone(), v.clone());
201            }
202        }
203    }
204
205    fn sample_bbox() -> Value {
206        serde_json::json!({ "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 })
207    }
208
209    #[test]
210    fn from_json_reads_rl_shaped_config() {
211        let node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({ "initial": "cat" }))).unwrap();
212
213        assert_eq!(node.control_topic, "/rl/control");
214        assert_eq!(node.request_topic, "/rl/target/request");
215        assert_eq!(node.action_set, "set_target_class");
216        assert_eq!(node.action_clear, None);
217        assert_eq!(node.payload_key, "target_class");
218        assert_eq!(node.target, Some(serde_json::json!("cat")));
219    }
220
221    #[test]
222    fn from_json_reads_instance_track_shaped_config() {
223        let node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
224            "initial": sample_bbox()
225        })))
226        .unwrap();
227
228        assert_eq!(node.control_topic, "/instance-track/control");
229        assert_eq!(node.request_topic, "/instance-track/target/request");
230        assert_eq!(node.action_set, "set_target");
231        assert_eq!(node.action_clear, Some("clear".to_string()));
232        assert_eq!(node.payload_key, "bbox");
233        assert_eq!(node.target, Some(sample_bbox()));
234    }
235
236    #[test]
237    fn from_json_allows_no_initial_target() {
238        let node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).unwrap();
239        assert_eq!(node.target, None);
240    }
241
242    #[test]
243    fn from_json_requires_the_core_fields() {
244        assert!(SetCapabilityTargetNode::from_json(serde_json::json!({})).is_err());
245        assert!(SetCapabilityTargetNode::from_json(serde_json::json!({
246            "control_topic": "/rl/control"
247        }))
248        .is_err());
249    }
250
251    #[test]
252    fn poll_latest_request_returns_none_when_empty() {
253        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
254        let mut receiver = Receiver::from(rx);
255        assert_eq!(
256            SetCapabilityTargetNode::poll_latest_request("target_class", &mut receiver),
257            None
258        );
259    }
260
261    #[test]
262    fn poll_latest_request_keeps_only_the_most_recent_value() {
263        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
264        let mut receiver = Receiver::from(rx);
265        tx.send(serde_json::json!({ "target_class": "dog" })).unwrap();
266        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();
267
268        assert_eq!(
269            SetCapabilityTargetNode::poll_latest_request("target_class", &mut receiver),
270            Some(Some(serde_json::json!("cat")))
271        );
272    }
273
274    #[test]
275    fn poll_latest_request_accepts_a_null_value_as_a_clear_request() {
276        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
277        let mut receiver = Receiver::from(rx);
278        tx.send(serde_json::json!({ "target_class": null })).unwrap();
279
280        assert_eq!(
281            SetCapabilityTargetNode::poll_latest_request("target_class", &mut receiver),
282            Some(None)
283        );
284    }
285
286    #[test]
287    fn poll_latest_request_accepts_an_absent_key_as_a_clear_request() {
288        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
289        let mut receiver = Receiver::from(rx);
290        tx.send(serde_json::json!({})).unwrap();
291
292        assert_eq!(
293            SetCapabilityTargetNode::poll_latest_request("bbox", &mut receiver),
294            Some(None)
295        );
296    }
297
298    #[test]
299    fn poll_latest_request_keeps_the_most_recent_bbox() {
300        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
301        let mut receiver = Receiver::from(rx);
302        tx.send(serde_json::json!({ "bbox": { "x": 0.0, "y": 0.0, "width": 0.1, "height": 0.1 } }))
303            .unwrap();
304        tx.send(serde_json::json!({ "bbox": sample_bbox() })).unwrap();
305
306        assert_eq!(
307            SetCapabilityTargetNode::poll_latest_request("bbox", &mut receiver),
308            Some(Some(sample_bbox()))
309        );
310    }
311
312    #[tokio::test]
313    #[ignore]
314    async fn tick_publishes_configured_target_once_then_stays_running() {
315        let ctx = create_test_context("test_set_capability_target_rl").await;
316
317        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({ "initial": "cat" }))).unwrap();
318
319        let mut receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
320
321        let status = node.tick(&ctx).await.unwrap();
322        assert_eq!(status, NodeStatus::Running);
323
324        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
325            .await
326            .expect("timed out waiting for published command")
327            .expect("channel closed without a message");
328
329        assert_eq!(received["action"], "set_target_class");
330        assert_eq!(received["params"]["target_class"], "cat");
331
332        let status = node.tick(&ctx).await.unwrap();
333        assert_eq!(status, NodeStatus::Running);
334        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
335        assert!(timed_out.is_err(), "expected no second publish without a new request");
336    }
337
338    #[tokio::test]
339    #[ignore]
340    async fn tick_relays_a_live_request_published_on_request_topic_rl() {
341        let ctx = create_test_context("test_set_capability_target_rl_request").await;
342
343        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).unwrap();
344
345        node.tick(&ctx).await.unwrap();
346
347        let mut control_receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
348        ctx.publish_raw("/rl/target/request", &serde_json::json!({ "target_class": "dog" }))
349            .await
350            .unwrap();
351
352        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
353        node.tick(&ctx).await.unwrap();
354
355        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
356            .await
357            .expect("timed out waiting for relayed command")
358            .expect("channel closed without a message");
359
360        assert_eq!(received["action"], "set_target_class");
361        assert_eq!(received["params"]["target_class"], "dog");
362    }
363
364    #[tokio::test]
365    #[ignore]
366    async fn tick_publishes_configured_bbox_once_then_stays_running() {
367        let ctx = create_test_context("test_set_capability_target_instance_track").await;
368
369        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
370            "initial": sample_bbox()
371        })))
372        .unwrap();
373
374        let mut receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
375
376        let status = node.tick(&ctx).await.unwrap();
377        assert_eq!(status, NodeStatus::Running);
378
379        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
380            .await
381            .expect("timed out waiting for published command")
382            .expect("channel closed without a message");
383
384        assert_eq!(received["action"], "set_target");
385        assert_eq!(received["params"]["bbox"]["x"], 0.1);
386
387        let status = node.tick(&ctx).await.unwrap();
388        assert_eq!(status, NodeStatus::Running);
389        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
390        assert!(timed_out.is_err(), "expected no second publish without a new request");
391    }
392
393    #[tokio::test]
394    #[ignore]
395    async fn tick_relays_a_live_request_published_on_request_topic_instance_track() {
396        let ctx = create_test_context("test_set_capability_target_instance_track_request").await;
397
398        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();
399
400        node.tick(&ctx).await.unwrap();
401
402        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
403        ctx.publish_raw(
404            "/instance-track/target/request",
405            &serde_json::json!({ "bbox": { "x": 0.5, "y": 0.5, "width": 0.1, "height": 0.1 } }),
406        )
407        .await
408        .unwrap();
409
410        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
411        node.tick(&ctx).await.unwrap();
412
413        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
414            .await
415            .expect("timed out waiting for relayed command")
416            .expect("channel closed without a message");
417
418        assert_eq!(received["action"], "set_target");
419        assert_eq!(received["params"]["bbox"]["x"], 0.5);
420        assert_eq!(
421            node.tick(&ctx).await.unwrap(),
422            NodeStatus::Running,
423            "a live target must report Running"
424        );
425    }
426
427    #[tokio::test]
428    #[ignore = "Requires Redis infrastructure"]
429    async fn tick_returns_failure_when_no_target_has_ever_been_set_and_action_clear_is_configured() {
430        let ctx = create_test_context("test_set_capability_target_idle").await;
431
432        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();
433
434        let status = node.tick(&ctx).await.unwrap();
435        assert_eq!(
436            status,
437            NodeStatus::Failure,
438            "no target yet means cede control, not Success"
439        );
440    }
441
442    #[tokio::test]
443    #[ignore = "Requires Redis infrastructure"]
444    async fn tick_stays_running_when_no_target_has_ever_been_set_and_no_action_clear_is_configured() {
445        let ctx = create_test_context("test_set_capability_target_rl_idle").await;
446
447        let mut node = SetCapabilityTargetNode::from_json(rl_config(serde_json::json!({}))).unwrap();
448
449        let status = node.tick(&ctx).await.unwrap();
450        assert_eq!(
451            status,
452            NodeStatus::Running,
453            "rl-style config (no action_clear) always reports Running, matching the old SetRlTargetNode"
454        );
455    }
456
457    #[tokio::test]
458    #[ignore = "Requires Redis infrastructure"]
459    async fn tick_returns_failure_after_a_clear_request_even_with_a_configured_initial_target() {
460        let ctx = create_test_context("test_set_capability_target_cleared").await;
461
462        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({
463            "initial": sample_bbox()
464        })))
465        .unwrap();
466
467        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Running);
468
469        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
470            .await
471            .unwrap();
472        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
473
474        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Failure);
475    }
476
477    #[tokio::test]
478    #[ignore]
479    async fn tick_relays_a_clear_request_published_on_request_topic() {
480        let ctx = create_test_context("test_set_capability_target_clear").await;
481
482        let mut node = SetCapabilityTargetNode::from_json(instance_track_config(serde_json::json!({}))).unwrap();
483
484        node.tick(&ctx).await.unwrap();
485
486        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
487        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
488            .await
489            .unwrap();
490
491        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
492        node.tick(&ctx).await.unwrap();
493
494        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
495            .await
496            .expect("timed out waiting for relayed command")
497            .expect("channel closed without a message");
498
499        assert_eq!(received["action"], "clear");
500    }
501}
mecha10_behavior_runtime/actions/set_capability_target.rs

mecha10_behavior_runtime/actions/
set_capability_target.rs
