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.
3//!
4//! For a `Watch`-wrapped instance-track interrupt layer to correctly claim control only while a
5//! target is live and cede immediately when idle, *both* `InstanceTrackTaskNode` and
6//! `SetInstanceTrackTargetNode` must independently resolve to `NodeStatus::Failure` when there's
7//! no live target and `NodeStatus::Running` while there is one. Critically, **neither must ever
8//! return `NodeStatus::Success`** for the "task complete, hand control back" case (the old
9//! behavior here) - here's why:
10//!
11//! If the hand-back signal were `Success`, the `Watch` layer's aggregate status for that tick
12//! would be `Success` too (`RequireOne`: any child succeeding wins), which `SubsumptionNode`
13//! treats as a winning/claiming status (`Success | Running => return`) exactly like `Running` -
14//! so the *whole tree's* tick for that cycle would resolve to `Success`, which the top-level
15//! executor treats as "the tree finished" and resets *everything*, including the lower-priority
16//! main layer (e.g. `patrol-simple`'s `Sequence`), wiping its progress and restarting from the
17//! beginning instead of resuming where it was.
18//!
19//! Using `Failure` instead avoids this entirely: when the instance-track layer converges to
20//! `Failure` that tick, `SubsumptionNode` immediately falls through *within that same tick* and
21//! ticks the lower-priority layer right then - if that layer is `Running` (the normal case, e.g.
22//! mid-patrol), `SubsumptionNode`'s own tick for that cycle returns `Running`, which the executor
23//! does *not* reset on. The main layer's progress is therefore preserved with zero wasted ticks.
24//!
25//! To make that convergence happen on the *same* tick the underlying node gives up (rather than
26//! racing a separate `clear` request), this node subscribes to the same target-request topic
27//! `SetInstanceTrackTargetNode` does and tracks `has_target` locally, additionally forcing
28//! `has_target = false` the instant a `task_complete` message arrives.
29
30use crate::actions::set_instance_track_target::{default_request_topic, InstanceTrackTargetRequest};
31use crate::{BehaviorNode, Context, NodeStatus};
32use async_trait::async_trait;
33use mecha10_core::Receiver;
34use serde_json::Value;
35
36const INSTANCE_TRACK_CAPABILITY: &str = "instance-track";
37
38const TASK_COMPLETE_TOPIC: &str = "/instance-track/task_complete";
39
40pub struct InstanceTrackTaskNode {
41    complete_receiver: Option<Receiver<Value>>,
42    target_receiver: Option<Receiver<Value>>,
43    request_topic: String,
44    has_target: bool,
45}
46
47impl std::fmt::Debug for InstanceTrackTaskNode {
48    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49        f.debug_struct("InstanceTrackTaskNode")
50            .field("complete_subscribed", &self.complete_receiver.is_some())
51            .field("target_subscribed", &self.target_receiver.is_some())
52            .field("has_target", &self.has_target)
53            .finish()
54    }
55}
56
57impl InstanceTrackTaskNode {
58    /// Construct a node. `_config` is ignored - this node takes none.
59    pub fn from_json(_config: Value) -> anyhow::Result<Self> {
60        Ok(Self {
61            complete_receiver: None,
62            target_receiver: None,
63            request_topic: default_request_topic(),
64            has_target: false,
65        })
66    }
67
68    /// Construct a node with already-connected receivers, bypassing live subscription.
69    ///
70    /// Used by unit tests to exercise convergence without a real `Context`/Redis.
71    #[cfg(test)]
72    fn with_receivers(complete_receiver: Receiver<Value>, target_receiver: Receiver<Value>) -> Self {
73        Self {
74            complete_receiver: Some(complete_receiver),
75            target_receiver: Some(target_receiver),
76            request_topic: default_request_topic(),
77            has_target: false,
78        }
79    }
80
81    /// Drains every pending target request, updating `has_target` from the last one seen (if
82    /// any) - mirrors `SetInstanceTrackTargetNode::poll_latest_request`'s "only the most recent
83    /// request matters" semantics.
84    fn poll_target(&mut self) {
85        let Some(receiver) = self.target_receiver.as_mut() else {
86            return;
87        };
88
89        let mut latest = None;
90        while let Ok(value) = receiver.try_recv() {
91            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
92                Ok(request) => latest = Some(request.bbox.is_some()),
93                Err(e) => tracing::warn!("InstanceTrackTaskNode: dropping malformed target request: {}", e),
94            }
95        }
96
97        if let Some(has_target) = latest {
98            self.has_target = has_target;
99        }
100    }
101
102    /// Drain any pending completion messages - arrival of any message at all is the whole
103    /// signal, same as `RlTaskNode::poll_complete`. Proactively clears `has_target` so this
104    /// node and `SetInstanceTrackTargetNode` converge to "idle" on the same tick the underlying
105    /// `instance-track` node gives up, rather than waiting for a separate `clear` request.
106    fn poll_complete(&mut self) {
107        let Some(receiver) = self.complete_receiver.as_mut() else {
108            return;
109        };
110
111        if receiver.try_recv().is_ok() {
112            self.has_target = false;
113        }
114    }
115}
116
117#[async_trait]
118impl BehaviorNode for InstanceTrackTaskNode {
119    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
120        if self.complete_receiver.is_none() {
121            self.complete_receiver = Some(ctx.subscribe_raw::<Value>(TASK_COMPLETE_TOPIC).await?);
122        }
123        if self.target_receiver.is_none() {
124            self.target_receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
125        }
126
127        self.poll_target();
128        self.poll_complete();
129
130        // `Failure` (never `Success`) when idle - see module doc for why.
131        Ok(if self.has_target {
132            NodeStatus::Running
133        } else {
134            NodeStatus::Failure
135        })
136    }
137
138    fn name(&self) -> &str {
139        "instance_track_task"
140    }
141
142    fn active_capability(&self) -> Option<&str> {
143        Some(INSTANCE_TRACK_CAPABILITY)
144    }
145
146    fn reset_self(&mut self) {
147        self.has_target = false;
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154    use tokio::sync::mpsc;
155
156    fn node_with_channels() -> (
157        InstanceTrackTaskNode,
158        mpsc::UnboundedSender<Value>,
159        mpsc::UnboundedSender<Value>,
160    ) {
161        let (complete_tx, complete_rx) = mpsc::unbounded_channel();
162        let (target_tx, target_rx) = mpsc::unbounded_channel();
163        (
164            InstanceTrackTaskNode::with_receivers(Receiver::from(complete_rx), Receiver::from(target_rx)),
165            complete_tx,
166            target_tx,
167        )
168    }
169
170    #[test]
171    fn always_declares_the_instance_track_capability() {
172        let (node, _complete_tx, _target_tx) = node_with_channels();
173        assert_eq!(node.active_capability(), Some("instance-track"));
174    }
175
176    #[test]
177    fn no_target_ever_set_stays_without_a_target() {
178        let (mut node, _complete_tx, _target_tx) = node_with_channels();
179        node.poll_target();
180        node.poll_complete();
181        assert!(!node.has_target);
182    }
183
184    #[test]
185    fn a_live_target_request_sets_has_target() {
186        let (mut node, _complete_tx, target_tx) = node_with_channels();
187        target_tx
188            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
189            .unwrap();
190
191        node.poll_target();
192
193        assert!(node.has_target);
194    }
195
196    #[test]
197    fn a_clear_request_unsets_has_target_even_with_no_task_complete() {
198        let (mut node, _complete_tx, target_tx) = node_with_channels();
199        target_tx
200            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
201            .unwrap();
202        node.poll_target();
203        assert!(node.has_target);
204
205        target_tx.send(serde_json::json!({ "bbox": null })).unwrap();
206        node.poll_target();
207
208        assert!(!node.has_target);
209    }
210
211    #[test]
212    fn task_complete_unsets_has_target_on_the_very_next_poll() {
213        let (mut node, complete_tx, target_tx) = node_with_channels();
214        target_tx
215            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
216            .unwrap();
217        node.poll_target();
218        assert!(node.has_target);
219
220        complete_tx.send(serde_json::json!({})).unwrap();
221        node.poll_complete();
222
223        assert!(
224            !node.has_target,
225            "task_complete must converge has_target to false without waiting for a separate clear request"
226        );
227    }
228
229    #[test]
230    fn reset_self_clears_has_target_for_a_later_pass() {
231        let (mut node, _complete_tx, target_tx) = node_with_channels();
232        target_tx
233            .send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
234            .unwrap();
235        node.poll_target();
236        assert!(node.has_target);
237
238        node.reset_self();
239
240        assert!(!node.has_target);
241    }
242
243    #[test]
244    fn from_json_ignores_any_input() {
245        assert!(InstanceTrackTaskNode::from_json(Value::Null).is_ok());
246        assert!(InstanceTrackTaskNode::from_json(serde_json::json!({"unexpected": "field"})).is_ok());
247    }
248}
mecha10_behavior_runtime/actions/instance_track_task.rs

mecha10_behavior_runtime/actions/
instance_track_task.rs
