mecha10_behavior_runtime/actions/
rl_task.rs

1//! Action node that hands control to the `rl` node and waits for it to report task completion.
2
3use crate::{BehaviorNode, Context, NodeStatus};
4use async_trait::async_trait;
5use mecha10_core::Receiver;
6use serde_json::Value;
7
8const RL_CAPABILITY: &str = "rl";
9
10const TASK_COMPLETE_TOPIC: &str = "/rl/task_complete";
11
12pub struct RlTaskNode {
13    receiver: Option<Receiver<Value>>,
14    complete: bool,
15}
16
17impl std::fmt::Debug for RlTaskNode {
18    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
19        f.debug_struct("RlTaskNode")
20            .field("subscribed", &self.receiver.is_some())
21            .field("complete", &self.complete)
22            .finish()
23    }
24}
25
26impl RlTaskNode {
27    /// Construct a node. `_config` is ignored - see the module doc, this node takes none.
28    pub fn from_json(_config: Value) -> anyhow::Result<Self> {
29        Ok(Self {
30            receiver: None,
31            complete: false,
32        })
33    }
34
35    /// Construct a node with an already-connected receiver, bypassing live subscription.
36    ///
37    /// Used by unit tests to exercise completion-detection without a real `Context`/Redis.
38    #[cfg(test)]
39    fn with_receiver(receiver: Receiver<Value>) -> Self {
40        Self {
41            receiver: Some(receiver),
42            complete: false,
43        }
44    }
45
46    /// Drain any pending completion messages - arrival of any message at all is the whole
47    /// signal, so the payload itself is never inspected.
48    fn poll_complete(&mut self) {
49        if self.complete {
50            return;
51        }
52        if let Some(receiver) = self.receiver.as_mut() {
53            if receiver.try_recv().is_ok() {
54                self.complete = true;
55            }
56        }
57    }
58}
59
60#[async_trait]
61impl BehaviorNode for RlTaskNode {
62    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
63        if self.receiver.is_none() {
64            self.receiver = Some(ctx.subscribe_raw::<Value>(TASK_COMPLETE_TOPIC).await?);
65        }
66
67        self.poll_complete();
68
69        Ok(if self.complete {
70            NodeStatus::Success
71        } else {
72            NodeStatus::Running
73        })
74    }
75
76    fn name(&self) -> &str {
77        "rl_task"
78    }
79
80    fn active_capability(&self) -> Option<&str> {
81        Some(RL_CAPABILITY)
82    }
83
84    fn reset_self(&mut self) {
85        self.complete = false;
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92    use tokio::sync::mpsc;
93
94    fn node_with_channel() -> (RlTaskNode, mpsc::UnboundedSender<Value>) {
95        let (tx, rx) = mpsc::unbounded_channel();
96        (RlTaskNode::with_receiver(Receiver::from(rx)), tx)
97    }
98
99    #[test]
100    fn always_declares_the_rl_capability() {
101        let (node, _tx) = node_with_channel();
102        assert_eq!(node.active_capability(), Some("rl"));
103    }
104
105    #[test]
106    fn stays_incomplete_with_no_message() {
107        let (mut node, _tx) = node_with_channel();
108        node.poll_complete();
109        assert!(!node.complete);
110    }
111
112    #[test]
113    fn completes_once_any_message_arrives() {
114        let (mut node, tx) = node_with_channel();
115        tx.send(serde_json::json!({})).unwrap();
116
117        node.poll_complete();
118
119        assert!(node.complete);
120    }
121
122    #[test]
123    fn reset_self_clears_completion_for_a_later_pass() {
124        let (mut node, tx) = node_with_channel();
125        tx.send(serde_json::json!({})).unwrap();
126        node.poll_complete();
127        assert!(node.complete);
128
129        node.reset_self();
130
131        assert!(!node.complete);
132    }
133
134    #[test]
135    fn from_json_ignores_any_input() {
136        assert!(RlTaskNode::from_json(Value::Null).is_ok());
137        assert!(RlTaskNode::from_json(serde_json::json!({"unexpected": "field"})).is_ok());
138    }
139}
mecha10_behavior_runtime/actions/rl_task.rs

mecha10_behavior_runtime/actions/
rl_task.rs
