robonix_executor/
plan_runtime.rs

1// SPDX-License-Identifier: MulanPSL-2.0
2// Shared in-process state for Executor plan cancellation.
3
4use crate::pb::pilot::rtdl_node_state::RtdlNodeStateEnum;
5use crate::pb::pilot::{CapabilityCall, CapabilityCallResult, Plan};
6use robonix_atlas::client::AtlasClient;
7use robonix_atlas::pb as atlas_pb;
8use robonix_scribe::warn;
9use serde::Deserialize;
10use std::collections::HashMap;
11use std::sync::Arc;
12use std::time::Duration;
13use tokio::sync::Mutex;
14
15const DEFAULT_CANCEL_WAIT_MS: u64 = 5_000;
16
17#[derive(Clone, Default)]
18pub struct PlanRuntime {
19    inner: Arc<Mutex<HashMap<String, PlanRun>>>,
20}
21
22#[derive(Clone)]
23pub struct RunningAsyncCall {
24    pub call_id: String,
25    pub provider_id: String,
26    pub cancel_contract: String,
27    pub run_id: String,
28}
29
30struct PlanRun {
31    cancelled: bool,
32    running_async: HashMap<String, RunningAsyncCall>,
33    /// op_id → when to cancel the whole plan once execution reaches that op.
34    /// Set by the `stop_plan_at` builtin; read by the execution loop at each
35    /// `do` node. At most one phase per op_id (a later set overwrites).
36    stop_points: HashMap<String, StopWhen>,
37    /// Plan-level description (the root node's description = the overall task),
38    /// so `get_all_plans` can tell the LLM what each running plan is.
39    description: String,
40    /// Static op list snapshotted at register time (arena order), so
41    /// `get_plan_status` can report op_ids/descriptions the LLM can target.
42    ops: Vec<OpMeta>,
43    /// op_id → latest RTDL node state observed during execution. Absent = the
44    /// op has not started yet (reported as `pending`).
45    op_state: HashMap<String, u32>,
46}
47
48/// One op's static identity, snapshotted from the plan arena at register time.
49#[derive(Clone)]
50struct OpMeta {
51    op_id: String,
52    description: String,
53    node_kind: u32,
54    provider_id: String,
55    contract_id: String,
56}
57
58/// When a stop point fires relative to its target op. "Stop" always means
59/// cancel the entire plan — the difference is only the moment it triggers.
60#[derive(Clone, Copy, PartialEq, Eq, Debug)]
61pub enum StopWhen {
62    /// Cancel the plan the moment execution reaches the op, before its call runs.
63    OnEnter,
64    /// Cancel the plan right after the op's call finishes.
65    OnComplete,
66}
67
68impl StopWhen {
69    /// Parse the `when` arg. `on_enter`/`before` and `on_complete`/`after` are
70    /// accepted; anything else is rejected by the caller.
71    fn parse(s: &str) -> Option<Self> {
72        match s.trim() {
73            "on_enter" | "before" => Some(StopWhen::OnEnter),
74            "on_complete" | "after" => Some(StopWhen::OnComplete),
75            _ => None,
76        }
77    }
78}
79
80struct CancelSnapshot {
81    async_calls: Vec<RunningAsyncCall>,
82}
83
84struct CancelAllSnapshot {
85    async_calls: Vec<RunningAsyncCall>,
86    plan_ids: Vec<String>,
87}
88
89#[derive(Deserialize)]
90struct CancelPlanArgs {
91    plan_id: String,
92    wait_ms: Option<u64>,
93}
94
95#[derive(Deserialize)]
96struct StopPlanAtArgs {
97    plan_id: String,
98    op_id: String,
99    /// "on_enter" (before the op runs) or "on_complete" (after it finishes).
100    /// Optional; defaults to "on_complete".
101    when: Option<String>,
102}
103
104impl PlanRuntime {
105    /// Register a plan before execution starts.
106    ///
107    /// If a stale entry with the same id exists it is replaced. Plan ids are
108    /// UUIDs generated by Pilot, so replacement is only a defensive cleanup path.
109    pub async fn register_plan(&self, plan_id: &str) {
110        let mut plans = self.inner.lock().await;
111        plans.insert(
112            plan_id.to_string(),
113            PlanRun {
114                cancelled: false,
115                running_async: HashMap::new(),
116                stop_points: HashMap::new(),
117                description: String::new(),
118                ops: Vec::new(),
119                op_state: HashMap::new(),
120            },
121        );
122    }
123
124    /// Snapshot a plan's op list (arena order) so `get_plan_status` can report
125    /// the op_ids/descriptions the LLM may target. Called right after
126    /// `register_plan`. No-op if the plan is not in the active table.
127    pub async fn record_plan_ops(&self, plan: &Plan) {
128        let ops: Vec<OpMeta> = plan
129            .nodes
130            .iter()
131            .map(|n| OpMeta {
132                op_id: n.op_id.clone(),
133                description: n.description.clone(),
134                node_kind: n.node_kind,
135                provider_id: n
136                    .call
137                    .as_ref()
138                    .map(|call| call.provider_id.clone())
139                    .unwrap_or_default(),
140                contract_id: n
141                    .call
142                    .as_ref()
143                    .map(|call| call.contract_id.clone())
144                    .unwrap_or_default(),
145            })
146            .collect();
147        let description = plan
148            .nodes
149            .get(plan.root_index as usize)
150            .map(|n| n.description.clone())
151            .unwrap_or_default();
152        if let Some(run) = self.inner.lock().await.get_mut(&plan.plan_id) {
153            run.ops = ops;
154            run.description = description;
155        }
156    }
157
158    /// Record the latest observed RTDL state for an op. Called by the execution
159    /// loop / async poller alongside every node_state event it streams, so
160    /// `get_plan_status` reflects live progress. No-op for unknown plans.
161    pub async fn record_op_state(&self, plan_id: &str, op_id: &str, state: u32) {
162        if let Some(run) = self.inner.lock().await.get_mut(plan_id) {
163            run.op_state.insert(op_id.to_string(), state);
164        }
165    }
166
167    /// Apply the `get_plan_status` builtin: return the plan's ops as JSON, each
168    /// with `op_id`, `kind`, `description`, current `state`, and any armed
169    /// `stop_point`. Lets the LLM inspect a running plan before issuing a
170    /// `stop_plan_at` (inspect first, then act). Errors when the plan is not in
171    /// the active table (stale/wrong id, or already finished) — it is a query,
172    /// so "not found" is an error; use `get_all_plans` to check liveness.
173    pub async fn get_plan_status_builtin(&self, call: &CapabilityCall) -> CapabilityCallResult {
174        #[derive(Deserialize)]
175        struct Args {
176            plan_id: String,
177        }
178        let args: Args = match serde_json::from_str(&call.args_json) {
179            Ok(a) => a,
180            Err(e) => return error_result(call, format!("invalid get_plan_status args: {e}")),
181        };
182        let plans = self.inner.lock().await;
183        let Some(run) = plans.get(&args.plan_id) else {
184            // A query for a specific plan that is not in the active table is an
185            // error (the id is stale or wrong) — unlike the cancel/stop
186            // commands, which are idempotent "ensure stopped" no-ops. Use
187            // get_all_plans to check whether a plan is still running.
188            return error_result(
189                call,
190                format!(
191                    "get_plan_status: no active RTDL plan '{}' (it never existed or already finished/cancelled); call get_all_plans to list running plans",
192                    args.plan_id
193                ),
194            );
195        };
196        let ops: Vec<serde_json::Value> = run
197            .ops
198            .iter()
199            .map(|op| {
200                let state = run
201                    .op_state
202                    .get(&op.op_id)
203                    .map(|s| state_label(*s))
204                    .unwrap_or("pending");
205                serde_json::json!({
206                    "op_id": op.op_id,
207                    "kind": kind_label(op.node_kind),
208                    "description": op.description,
209                    "provider_id": op.provider_id,
210                    "contract_id": op.contract_id,
211                    "state": state,
212                    "stop_point": run.stop_points.get(&op.op_id).map(when_label),
213                })
214            })
215            .collect();
216        ok_result(
217            call,
218            serde_json::json!({
219                "plan_id": args.plan_id,
220                "running": true,
221                "cancelled": run.cancelled,
222                "ops": ops,
223            })
224            .to_string(),
225        )
226    }
227
228    /// Apply the `get_all_plans` builtin: list every active RTDL plan with its
229    /// op count, cancelled flag, and number of armed stop points. Lets the LLM
230    /// discover which plans are running before inspecting one with
231    /// `get_plan_status`. Takes no args.
232    pub async fn get_all_plans_builtin(&self, call: &CapabilityCall) -> CapabilityCallResult {
233        ok_result(call, self.active_plans_json().await)
234    }
235
236    /// Return Executor's authoritative active-plan table for control-plane
237    /// observers. Reading this snapshot does not register a query plan.
238    pub async fn active_plans_json(&self) -> String {
239        let plans = self.inner.lock().await;
240        let mut entries: Vec<(&String, &PlanRun)> = plans.iter().collect();
241        entries.sort_by_key(|(id, _)| id.parse::<u64>().unwrap_or(u64::MAX));
242        let list: Vec<serde_json::Value> = entries
243            .iter()
244            .map(|(id, run)| {
245                serde_json::json!({
246                    "plan_id": id,
247                    "description": run.description,
248                    "op_count": run.ops.len(),
249                    "cancelled": run.cancelled,
250                    "stop_points": run.stop_points.len(),
251                    "ops": run.ops.iter().map(|op| serde_json::json!({
252                        "op_id": op.op_id,
253                        "kind": kind_label(op.node_kind),
254                        "description": op.description,
255                        "provider_id": op.provider_id,
256                        "contract_id": op.contract_id,
257                        "state": run.op_state.get(&op.op_id)
258                            .map(|state| state_label(*state))
259                            .unwrap_or("pending"),
260                    })).collect::<Vec<_>>(),
261                })
262            })
263            .collect();
264        serde_json::json!({ "count": list.len(), "plans": list }).to_string()
265    }
266
267    /// Mark a plan complete, remove it from the active table, and notify waiters.
268    pub async fn complete_plan(&self, plan_id: &str) {
269        self.inner.lock().await.remove(plan_id);
270    }
271
272    /// Return whether a registered plan has been cancelled.
273    pub async fn is_cancelled(&self, plan_id: &str) -> bool {
274        self.inner
275            .lock()
276            .await
277            .get(plan_id)
278            .is_some_and(|run| run.cancelled)
279    }
280
281    /// Add one async capability call to the plan's cancellation set, or cancel
282    /// it immediately if the plan has already been marked cancelled.
283    ///
284    /// Returns `true` when the call was registered for normal polling. Returns
285    /// `false` when the runtime sent the provider cancel contract instead.
286    pub async fn register_or_cancel_async_call(
287        &self,
288        plan_id: &str,
289        call: RunningAsyncCall,
290        self_provider_id: &str,
291        atlas: &mut AtlasClient,
292    ) -> bool {
293        let cancelled_call = {
294            let mut plans = self.inner.lock().await;
295            let Some(run) = plans.get_mut(plan_id) else {
296                return true;
297            };
298            if run.cancelled {
299                Some(call)
300            } else {
301                run.running_async.insert(call.call_id.clone(), call);
302                None
303            }
304        };
305
306        if let Some(call) = cancelled_call {
307            let _ = cancel_async_call(self_provider_id, &call, atlas).await;
308            false
309        } else {
310            true
311        }
312    }
313
314    /// Remove an async call once it reaches a terminal state.
315    pub async fn unregister_async_call(&self, plan_id: &str, call_id: &str) {
316        if let Some(run) = self.inner.lock().await.get_mut(plan_id) {
317            run.running_async.remove(call_id);
318        }
319    }
320
321    /// If a cancelled plan still owns `call_id`, remove it and send its cancel
322    /// contract. Calls already claimed by `cancel_plan` become a no-op here.
323    pub async fn cancel_async_call_for_plan(
324        &self,
325        plan_id: &str,
326        call_id: &str,
327        self_provider_id: &str,
328        atlas: &mut AtlasClient,
329    ) {
330        let running = {
331            let mut plans = self.inner.lock().await;
332            plans.get_mut(plan_id).and_then(|run| {
333                if run.cancelled {
334                    run.running_async.remove(call_id)
335                } else {
336                    None
337                }
338            })
339        };
340        // If running_async contains this call, cancel it.
341        if let Some(running) = running {
342            let _ = cancel_async_call(self_provider_id, &running, atlas).await;
343        }
344    }
345
346    /// Apply the `cancel_plan` builtin.
347    ///
348    /// Cancellation is best-effort: future sequence children are skipped by the
349    /// executor loop, and currently running async capability calls receive their
350    /// required `<contract_id>/cancel` call. Synchronous calls already in
351    /// progress are allowed to return naturally.
352    pub async fn cancel_plan_builtin(
353        &self,
354        call: &CapabilityCall,
355        self_provider_id: &str,
356        atlas: &mut AtlasClient,
357    ) -> CapabilityCallResult {
358        let args: CancelPlanArgs = match serde_json::from_str(&call.args_json) {
359            Ok(args) => args,
360            Err(e) => return error_result(call, format!("invalid cancel_plan args: {e}")),
361        };
362        let (_, output) = self
363            .cancel_plan_control(
364                &args.plan_id,
365                args.wait_ms.unwrap_or(DEFAULT_CANCEL_WAIT_MS),
366                self_provider_id,
367                atlas,
368            )
369            .await;
370        CapabilityCallResult {
371            call_id: call.call_id.clone(),
372            provider_id: call.provider_id.clone(),
373            contract_id: call.contract_id.clone(),
374            success: true,
375            output,
376            error: String::new(),
377        }
378    }
379
380    /// Out-of-band target cancellation used by Pilot meta ops and the legacy
381    /// builtin wrapper. Returns whether the plan reached terminal state before
382    /// the deadline plus a stable, idempotent message.
383    pub async fn cancel_plan_control(
384        &self,
385        plan_id: &str,
386        wait_ms: u64,
387        self_provider_id: &str,
388        atlas: &mut AtlasClient,
389    ) -> (bool, String) {
390        let snapshot = match self.begin_cancel(plan_id).await {
391            Ok(snapshot) => snapshot,
392            Err(_) => {
393                return (
394                    true,
395                    format!(
396                        "RTDL plan '{plan_id}' is not running (already finished or cancelled); nothing to cancel."
397                    ),
398                );
399            }
400        };
401        let mut cancel_errors = Vec::new();
402        for running in &snapshot.async_calls {
403            if let Err(error) = cancel_async_call(self_provider_id, running, atlas).await {
404                cancel_errors.push(format!("{}: {error:#}", running.call_id));
405            }
406        }
407        let completed = self.wait_until_inactive(plan_id, wait_ms).await;
408        let mut message = format!(
409            "Cancellation was requested for RTDL plan '{plan_id}'; cancelled=true, completed={completed}, async_cancel_attempts={}.",
410            snapshot.async_calls.len()
411        );
412        if !cancel_errors.is_empty() {
413            message.push_str(" Some async cancel requests failed: ");
414            message.push_str(&cancel_errors.join("; "));
415        }
416        (completed, message)
417    }
418
419    /// Arm a semantic stop boundary without creating a control RTDL tree.
420    pub async fn stop_plan_at_control(
421        &self,
422        plan_id: &str,
423        op_id: &str,
424        when: &str,
425    ) -> Result<String, String> {
426        let op_id = op_id.trim();
427        if op_id.is_empty() {
428            return Err("stop_plan_at requires non-empty op_id".to_string());
429        }
430        let when = StopWhen::parse(if when.trim().is_empty() {
431            "on_complete"
432        } else {
433            when
434        })
435        .ok_or_else(|| format!("invalid when '{when}' (expected on_enter or on_complete)"))?;
436        let mut plans = self.inner.lock().await;
437        let Some(run) = plans.get_mut(plan_id) else {
438            return Ok(format!(
439                "RTDL plan '{plan_id}' is not running (already finished or cancelled); no stop point needed."
440            ));
441        };
442        if !run.ops.iter().any(|op| op.op_id == op_id) {
443            return Err(format!("RTDL plan '{plan_id}' has no op_id '{op_id}'"));
444        }
445        run.stop_points.insert(op_id.to_string(), when);
446        Ok(format!(
447            "Stop point set: RTDL plan '{plan_id}' will be cancelled {} op_id={op_id}.",
448            when_label(&when)
449        ))
450    }
451
452    /// Apply the `stop_plan_at` builtin: record a stop point so the plan is
453    /// cancelled when execution reaches the given op.
454    ///
455    /// `when` selects the phase (`on_enter` before the op runs, `on_complete`
456    /// after it finishes; default `on_complete`). Setting a stop point on a
457    /// plan that is no longer active is a SUCCESS no-op — the intent ("don't
458    /// run past op X") is already satisfied, and returning an error would make
459    /// the planner retry forever (same policy as `cancel_plan`). The op_id is
460    /// not validated against the plan's nodes here; an op_id that never
461    /// executes simply never fires.
462    pub async fn stop_plan_at_builtin(&self, call: &CapabilityCall) -> CapabilityCallResult {
463        let args: StopPlanAtArgs = match serde_json::from_str(&call.args_json) {
464            Ok(args) => args,
465            Err(e) => return error_result(call, format!("invalid stop_plan_at args: {e}")),
466        };
467        let op_id = args.op_id.trim();
468        if op_id.is_empty() {
469            return error_result(call, "stop_plan_at: op_id must not be empty".to_string());
470        }
471        let when_str = args.when.as_deref().unwrap_or("on_complete");
472        let Some(when) = StopWhen::parse(when_str) else {
473            return error_result(
474                call,
475                format!(
476                    "stop_plan_at: invalid when '{when_str}' (expected on_enter or on_complete)"
477                ),
478            );
479        };
480        let output = {
481            let mut plans = self.inner.lock().await;
482            match plans.get_mut(&args.plan_id) {
483                Some(run) => {
484                    run.stop_points.insert(op_id.to_string(), when);
485                    format!(
486                        "Stop point set: RTDL plan '{}' will be cancelled {} op_id={}.",
487                        args.plan_id, when_str, op_id
488                    )
489                }
490                None => format!(
491                    "RTDL plan '{}' is not running (already finished or cancelled); no stop point needed.",
492                    args.plan_id
493                ),
494            }
495        };
496        ok_result(call, output)
497    }
498
499    /// Whether the plan has a stop point on `op_id` for the given phase.
500    /// Read by the execution loop at each `do` node entry and completion.
501    pub async fn should_stop_at(&self, plan_id: &str, op_id: &str, when: StopWhen) -> bool {
502        self.inner
503            .lock()
504            .await
505            .get(plan_id)
506            .and_then(|run| run.stop_points.get(op_id))
507            .is_some_and(|w| *w == when)
508    }
509
510    /// Fire a stop point: mark the plan cancelled and best-effort cancel its
511    /// running async calls — the same teardown `cancel_plan` performs. After
512    /// this the execution loop's `is_cancelled` checks halt the rest of the
513    /// plan. No-op if the plan already left the active table.
514    pub async fn trigger_stop(
515        &self,
516        plan_id: &str,
517        self_provider_id: &str,
518        atlas: &mut AtlasClient,
519    ) {
520        let Ok(snapshot) = self.begin_cancel(plan_id).await else {
521            return;
522        };
523        for running in &snapshot.async_calls {
524            if let Err(e) = cancel_async_call(self_provider_id, running, atlas).await {
525                warn!(
526                    "[executor] stop-point async cancel failed for {}: {e:#}",
527                    running.call_id
528                );
529            }
530        }
531    }
532
533    /// Cancel every active plan and make best-effort cancel calls for running async work.
534    pub async fn cancel_all_plans(&self, self_provider_id: &str, atlas: &mut AtlasClient) -> bool {
535        let (_, completed) = self
536            .cancel_all_plans_except(self_provider_id, atlas, None, DEFAULT_CANCEL_WAIT_MS)
537            .await;
538        completed
539    }
540
541    /// Cancel every active plan except an optional control plan and wait for
542    /// the targets to leave the active table.
543    pub async fn cancel_all_plans_except(
544        &self,
545        self_provider_id: &str,
546        atlas: &mut AtlasClient,
547        except_plan_id: Option<&str>,
548        wait_ms: u64,
549    ) -> (usize, bool) {
550        let snapshot = self.begin_cancel_all(except_plan_id).await;
551        let target_count = snapshot.plan_ids.len();
552        for running in &snapshot.async_calls {
553            if let Err(e) = cancel_async_call(self_provider_id, running, atlas).await {
554                warn!(
555                    "[executor] cancel_all_plans async cancel failed for {}: {e:#}",
556                    running.call_id
557                );
558            }
559        }
560        let completed = self
561            .wait_until_all_inactive(&snapshot.plan_ids, wait_ms)
562            .await;
563        (target_count, completed)
564    }
565
566    /// Mark a plan cancelled and return the state needed by `cancel_plan`.
567    ///
568    /// This pure runtime step is separated from atlas/MCP cancel calls so tests
569    /// can cover unknown-plan and cancellation-state behavior without a live
570    /// atlas server.
571    async fn begin_cancel(&self, plan_id: &str) -> Result<CancelSnapshot, String> {
572        let mut plans = self.inner.lock().await;
573        let Some(run) = plans.get_mut(plan_id) else {
574            return Err(format!("unknown plan_id '{plan_id}'"));
575        };
576        run.cancelled = true;
577        Ok(CancelSnapshot {
578            async_calls: run
579                .running_async
580                .drain()
581                .map(|(_, running)| running)
582                .collect(),
583        })
584    }
585
586    /// Mark every active plan cancelled and drain their async cancel sets into
587    /// one snapshot. Provider cancel calls happen after the runtime lock is
588    /// released by `cancel_all_plans`.
589    async fn begin_cancel_all(&self, except_plan_id: Option<&str>) -> CancelAllSnapshot {
590        let mut plans = self.inner.lock().await;
591        let mut async_calls = Vec::new();
592        let mut plan_ids = Vec::new();
593        for (plan_id, run) in plans.iter_mut() {
594            if except_plan_id == Some(plan_id.as_str()) {
595                continue;
596            }
597            run.cancelled = true;
598            plan_ids.push(plan_id.clone());
599            async_calls.extend(run.running_async.drain().map(|(_, running)| running));
600        }
601        CancelAllSnapshot {
602            async_calls,
603            plan_ids,
604        }
605    }
606
607    async fn wait_until_inactive(&self, plan_id: &str, wait_ms: u64) -> bool {
608        self.wait_until_all_inactive(&[plan_id.to_string()], wait_ms)
609            .await
610    }
611
612    /// Poll durable active-table state instead of awaiting `Notify`, whose
613    /// notification can race ahead of waiter registration.
614    async fn wait_until_all_inactive(&self, plan_ids: &[String], wait_ms: u64) -> bool {
615        let wait = async {
616            loop {
617                let all_inactive = {
618                    let plans = self.inner.lock().await;
619                    plan_ids.iter().all(|plan_id| !plans.contains_key(plan_id))
620                };
621                if all_inactive {
622                    return;
623                }
624                tokio::time::sleep(Duration::from_millis(10)).await;
625            }
626        };
627        tokio::time::timeout(Duration::from_millis(wait_ms), wait)
628            .await
629            .is_ok()
630    }
631}
632
633/// Call a provider's async cancel contract through atlas/MCP.
634///
635/// The cancel payload mirrors the status polling payload and sends `run_id`
636/// when present. Atlas channels are disconnected after the MCP request returns.
637async fn cancel_async_call(
638    consumer_id: &str,
639    running: &RunningAsyncCall,
640    atlas: &mut AtlasClient,
641) -> anyhow::Result<()> {
642    let args = if running.run_id.is_empty() {
643        "{}".to_string()
644    } else {
645        serde_json::json!({ "run_id": running.run_id }).to_string()
646    };
647    let cancel_call = CapabilityCall {
648        call_id: format!("cancel-{}", uuid::Uuid::new_v4()),
649        provider_id: running.provider_id.clone(),
650        contract_id: running.cancel_contract.clone(),
651        args_json: args,
652    };
653    let (channel_id, endpoint, _) = atlas
654        .connect_capability(
655            consumer_id,
656            &running.provider_id,
657            &running.cancel_contract,
658            atlas_pb::Transport::Mcp,
659        )
660        .await?;
661    let result = crate::dispatch::mcp::execute(&cancel_call, &endpoint).await;
662    let _ = atlas.disconnect_capability(&channel_id).await;
663    if result.success {
664        Ok(())
665    } else {
666        anyhow::bail!("{}", result.error)
667    }
668}
669
670/// Build a failed capability result for a builtin.
671fn error_result(call: &CapabilityCall, error: String) -> CapabilityCallResult {
672    CapabilityCallResult {
673        call_id: call.call_id.clone(),
674        provider_id: call.provider_id.clone(),
675        contract_id: call.contract_id.clone(),
676        success: false,
677        output: String::new(),
678        error,
679    }
680}
681
682/// Human-readable label for an RTDL node state (for `get_plan_status` JSON).
683fn state_label(state: u32) -> &'static str {
684    match RtdlNodeStateEnum::try_from(state as i32) {
685        Ok(RtdlNodeStateEnum::Pending) => "pending",
686        Ok(RtdlNodeStateEnum::Running) => "running",
687        Ok(RtdlNodeStateEnum::Succeeded) => "succeeded",
688        Ok(RtdlNodeStateEnum::Failed) => "failed",
689        Ok(RtdlNodeStateEnum::Canceled) => "canceled",
690        Ok(RtdlNodeStateEnum::Timeout) => "timeout",
691        Ok(RtdlNodeStateEnum::Paused) => "paused",
692        _ => "unknown",
693    }
694}
695
696/// Label an RTDL node kind (0=sequence, 1=parallel, 2=do).
697fn kind_label(node_kind: u32) -> &'static str {
698    match node_kind {
699        0 => "sequence",
700        1 => "parallel",
701        2 => "do",
702        _ => "unknown",
703    }
704}
705
706/// Label a stop-point phase for `get_plan_status` JSON.
707fn when_label(when: &StopWhen) -> &'static str {
708    match when {
709        StopWhen::OnEnter => "on_enter",
710        StopWhen::OnComplete => "on_complete",
711    }
712}
713
714/// Build a successful capability result for a builtin.
715fn ok_result(call: &CapabilityCall, output: String) -> CapabilityCallResult {
716    CapabilityCallResult {
717        call_id: call.call_id.clone(),
718        provider_id: call.provider_id.clone(),
719        contract_id: call.contract_id.clone(),
720        success: true,
721        output,
722        error: String::new(),
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use super::*;
729
730    #[tokio::test]
731    async fn begin_cancel_unknown_plan_fails() {
732        let runtime = PlanRuntime::default();
733        let err = match runtime.begin_cancel("missing").await {
734            Ok(_) => panic!("missing plan should not cancel"),
735            Err(err) => err,
736        };
737        assert!(err.contains("unknown plan_id"));
738    }
739
740    #[tokio::test]
741    async fn registered_plan_starts_uncancelled_then_flips() {
742        let runtime = PlanRuntime::default();
743        runtime.register_plan("p").await;
744        assert!(!runtime.is_cancelled("p").await);
745
746        runtime.begin_cancel("p").await.unwrap();
747
748        assert!(runtime.is_cancelled("p").await);
749    }
750
751    #[tokio::test]
752    async fn begin_cancel_drains_async_calls() {
753        let runtime = PlanRuntime::default();
754        runtime.register_plan("p").await;
755        {
756            let mut plans = runtime.inner.lock().await;
757            plans
758                .get_mut("p")
759                .unwrap()
760                .running_async
761                .insert("c".into(), running_call("c"));
762        }
763
764        let snapshot = runtime.begin_cancel("p").await.unwrap();
765
766        assert_eq!(snapshot.async_calls.len(), 1);
767        let plans = runtime.inner.lock().await;
768        assert!(plans.get("p").unwrap().running_async.is_empty());
769    }
770
771    #[tokio::test]
772    async fn begin_cancel_all_marks_every_plan() {
773        let runtime = PlanRuntime::default();
774        runtime.register_plan("p1").await;
775        runtime.register_plan("p2").await;
776        let snapshot = runtime.begin_cancel_all(None).await;
777
778        assert!(snapshot.async_calls.is_empty());
779        assert!(runtime.is_cancelled("p1").await);
780        assert!(runtime.is_cancelled("p2").await);
781    }
782
783    #[tokio::test]
784    async fn begin_cancel_all_excludes_control_plan() {
785        let runtime = PlanRuntime::default();
786        runtime.register_plan("task").await;
787        runtime.register_plan("control").await;
788
789        let snapshot = runtime.begin_cancel_all(Some("control")).await;
790
791        assert_eq!(snapshot.plan_ids, vec!["task"]);
792        assert!(runtime.is_cancelled("task").await);
793        assert!(!runtime.is_cancelled("control").await);
794    }
795
796    fn running_call(call_id: &str) -> RunningAsyncCall {
797        RunningAsyncCall {
798            call_id: call_id.into(),
799            provider_id: "provider".into(),
800            cancel_contract: "robonix/test/cancel".into(),
801            run_id: "r".into(),
802        }
803    }
804
805    fn stop_call(args_json: &str) -> CapabilityCall {
806        CapabilityCall {
807            call_id: "c".into(),
808            provider_id: "executor".into(),
809            contract_id: "robonix/system/executor/builtin/stop_plan_at".into(),
810            args_json: args_json.into(),
811        }
812    }
813
814    #[tokio::test]
815    async fn stop_plan_at_records_point_and_should_stop_matches_phase() {
816        let runtime = PlanRuntime::default();
817        runtime.register_plan("p").await;
818
819        let r = runtime
820            .stop_plan_at_builtin(&stop_call(
821                r#"{"plan_id":"p","op_id":"7","when":"on_enter"}"#,
822            ))
823            .await;
824        assert!(r.success, "set stop point should succeed: {}", r.error);
825
826        assert!(runtime.should_stop_at("p", "7", StopWhen::OnEnter).await);
827        // Same op, other phase must not match.
828        assert!(!runtime.should_stop_at("p", "7", StopWhen::OnComplete).await);
829        // Different op must not match.
830        assert!(!runtime.should_stop_at("p", "9", StopWhen::OnEnter).await);
831    }
832
833    #[tokio::test]
834    async fn stop_plan_at_defaults_to_on_complete() {
835        let runtime = PlanRuntime::default();
836        runtime.register_plan("p").await;
837        runtime
838            .stop_plan_at_builtin(&stop_call(r#"{"plan_id":"p","op_id":"3"}"#))
839            .await;
840        assert!(runtime.should_stop_at("p", "3", StopWhen::OnComplete).await);
841        assert!(!runtime.should_stop_at("p", "3", StopWhen::OnEnter).await);
842    }
843
844    #[tokio::test]
845    async fn stop_plan_at_unknown_plan_is_success_noop() {
846        let runtime = PlanRuntime::default();
847        let r = runtime
848            .stop_plan_at_builtin(&stop_call(r#"{"plan_id":"missing","op_id":"1"}"#))
849            .await;
850        assert!(r.success);
851        assert!(r.output.contains("not running"));
852    }
853
854    #[tokio::test]
855    async fn stop_plan_at_rejects_bad_when_and_empty_op_id() {
856        let runtime = PlanRuntime::default();
857        runtime.register_plan("p").await;
858        let bad_when = runtime
859            .stop_plan_at_builtin(&stop_call(r#"{"plan_id":"p","op_id":"1","when":"halt"}"#))
860            .await;
861        assert!(!bad_when.success);
862        assert!(bad_when.error.contains("invalid when"));
863
864        let empty_op = runtime
865            .stop_plan_at_builtin(&stop_call(r#"{"plan_id":"p","op_id":"  "}"#))
866            .await;
867        assert!(!empty_op.success);
868        assert!(empty_op.error.contains("op_id"));
869    }
870
871    #[tokio::test]
872    async fn should_stop_at_false_without_point() {
873        let runtime = PlanRuntime::default();
874        runtime.register_plan("p").await;
875        assert!(!runtime.should_stop_at("p", "1", StopWhen::OnComplete).await);
876    }
877
878    fn status_call(args_json: &str) -> CapabilityCall {
879        CapabilityCall {
880            call_id: "c".into(),
881            provider_id: "executor".into(),
882            contract_id: "robonix/system/executor/builtin/get_plan_status".into(),
883            args_json: args_json.into(),
884        }
885    }
886
887    #[tokio::test]
888    async fn get_plan_status_reports_ops_states_and_stop_points() {
889        use crate::pb::pilot::{Plan, RtdlNode};
890        let runtime = PlanRuntime::default();
891        runtime.register_plan("p").await;
892        let plan = Plan {
893            plan_id: "p".into(),
894            session_id: "s".into(),
895            round: 0,
896            root_index: 0,
897            nodes: vec![
898                RtdlNode {
899                    node_kind: 0,
900                    children: vec![1],
901                    call: None,
902                    op_id: "1".into(),
903                    description: "sequence".into(),
904                },
905                RtdlNode {
906                    node_kind: 2,
907                    children: vec![],
908                    call: Some(CapabilityCall {
909                        call_id: "camera-1".into(),
910                        provider_id: "realsense_camera".into(),
911                        contract_id: "robonix/primitive/camera/capture".into(),
912                        args_json: "{}".into(),
913                    }),
914                    op_id: "2".into(),
915                    description: "take snapshot".into(),
916                },
917            ],
918        };
919        runtime.record_plan_ops(&plan).await;
920        runtime
921            .record_op_state("p", "2", RtdlNodeStateEnum::Running as u32)
922            .await;
923        runtime
924            .stop_plan_at_builtin(&stop_call(
925                r#"{"plan_id":"p","op_id":"2","when":"on_complete"}"#,
926            ))
927            .await;
928
929        let r = runtime
930            .get_plan_status_builtin(&status_call(r#"{"plan_id":"p"}"#))
931            .await;
932        assert!(r.success, "{}", r.error);
933        // op 2 is running with an armed stop point; op 1 hasn't started (pending).
934        assert!(r.output.contains(r#""op_id":"2""#));
935        assert!(r.output.contains(r#""state":"running""#));
936        assert!(r.output.contains(r#""state":"pending""#));
937        assert!(r.output.contains(r#""stop_point":"on_complete""#));
938        assert!(r.output.contains(r#""running":true"#));
939        assert!(r.output.contains("realsense_camera"));
940        assert!(r.output.contains("robonix/primitive/camera/capture"));
941    }
942
943    #[tokio::test]
944    async fn get_all_plans_lists_active_plans() {
945        let runtime = PlanRuntime::default();
946        let empty = runtime.get_all_plans_builtin(&status_call("{}")).await;
947        assert!(empty.success);
948        assert!(empty.output.contains(r#""count":0"#));
949
950        use crate::pb::pilot::{Plan, RtdlNode};
951        runtime.register_plan("p1").await;
952        runtime
953            .record_plan_ops(&Plan {
954                plan_id: "p1".into(),
955                session_id: "s".into(),
956                round: 0,
957                root_index: 0,
958                nodes: vec![RtdlNode {
959                    node_kind: 0,
960                    children: vec![],
961                    call: None,
962                    op_id: "1".into(),
963                    description: "drive to the kitchen".into(),
964                }],
965            })
966            .await;
967        runtime.register_plan("p2").await;
968        let r = runtime.get_all_plans_builtin(&status_call("{}")).await;
969        assert!(r.success);
970        assert!(r.output.contains(r#""count":2"#));
971        assert!(r.output.contains(r#""plan_id":"p1""#));
972        assert!(r.output.contains(r#""plan_id":"p2""#));
973        // p1's plan-level description (root node) is surfaced for the LLM.
974        assert!(r.output.contains(r#""description":"drive to the kitchen""#));
975        let snapshot = runtime.active_plans_json().await;
976        assert!(snapshot.contains(r#""plan_id":"p1""#));
977        assert!(snapshot.contains(r#""ops""#));
978    }
979
980    #[tokio::test]
981    async fn get_plan_status_unknown_plan_errors() {
982        let runtime = PlanRuntime::default();
983        let r = runtime
984            .get_plan_status_builtin(&status_call(r#"{"plan_id":"missing"}"#))
985            .await;
986        assert!(!r.success);
987        assert!(r.error.contains("no active RTDL plan"));
988    }
989}
robonix_executor/plan_runtime.rs

robonix_executor/
plan_runtime.rs
