robonix_executor/dispatch/
builtin.rs

1// SPDX-License-Identifier: MulanPSL-2.0
2// Author: wheatfox <wheatfox17@icloud.com>
3//
4// dispatch/builtin.rs — built-in tool implementations
5
6use crate::pb::pilot::CapabilityCallResult;
7use crate::plan_runtime::PlanRuntime;
8use robonix_atlas::client::AtlasClient;
9use robonix_atlas::pb as atlas_pb;
10use serde::Deserialize;
11use std::path::{Path, PathBuf};
12
13/// Workspace root for file operations. All file paths are resolved relative to
14/// this directory, and path traversal beyond it is rejected.
15/// > `wheatfox's note: the definition of this "workspace" is where all the built-in tools
16/// > like read_file, write_file, exec command will be ran, as the linux CWD`
17fn workspace_root() -> PathBuf {
18    std::env::var("ROBONIX_WORKSPACE")
19        .map(PathBuf::from)
20        .unwrap_or_else(|_| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")))
21}
22
23/// Resolve a user-supplied path and ensure it stays within the workspace root.
24/// Returns the canonical path on success, or an error if the path escapes the
25/// allowed directory (path traversal).
26fn safe_resolve(user_path: &str) -> anyhow::Result<PathBuf> {
27    let root = workspace_root()
28        .canonicalize()
29        .unwrap_or_else(|_| workspace_root());
30    let candidate = if Path::new(user_path).is_absolute() {
31        PathBuf::from(user_path)
32    } else {
33        root.join(user_path)
34    };
35    // Canonicalize to resolve ".." components. If the file doesn't exist yet
36    // (write_file), canonicalize the parent directory instead.
37    let resolved = if candidate.exists() {
38        candidate.canonicalize()?
39    } else {
40        let parent = candidate
41            .parent()
42            .ok_or_else(|| anyhow::anyhow!("invalid path: no parent directory"))?;
43        let parent_resolved = parent.canonicalize().map_err(|_| {
44            anyhow::anyhow!("parent directory does not exist: {}", parent.display())
45        })?;
46        parent_resolved.join(
47            candidate
48                .file_name()
49                .ok_or_else(|| anyhow::anyhow!("invalid path: no file name"))?,
50        )
51    };
52    if !resolved.starts_with(&root) {
53        anyhow::bail!(
54            "path traversal denied: {} resolves outside workspace {}",
55            user_path,
56            root.display()
57        );
58    }
59    Ok(resolved)
60}
61
62use crate::pb::pilot::CapabilityCall;
63
64/// One in-process builtin (no network, runs in executor's own process).
65/// `call.contract_id`'s last segment names the operation —
66/// e.g. `robonix/system/executor/builtin/read_file` → `read_file`.
67pub async fn execute(
68    call: &CapabilityCall,
69    runtime: &PlanRuntime,
70    self_provider_id: &str,
71    atlas: &mut AtlasClient,
72    plan_id: &str,
73) -> CapabilityCallResult {
74    let op = call
75        .contract_id
76        .rsplit_once('/')
77        .map(|(_, leaf)| leaf)
78        .unwrap_or(call.contract_id.as_str());
79    if op == "cancel_plan" {
80        return runtime
81            .cancel_plan_builtin(call, self_provider_id, atlas)
82            .await;
83    }
84    if op == "stop_plan_at" {
85        return runtime.stop_plan_at_builtin(call).await;
86    }
87    if op == "get_plan_status" {
88        return runtime.get_plan_status_builtin(call).await;
89    }
90    if op == "get_all_plans" {
91        return runtime.get_all_plans_builtin(call).await;
92    }
93    if op == "read_capability_doc" {
94        return read_capability_doc(call, atlas).await;
95    }
96
97    let result = run(op, &call.args_json, runtime, plan_id).await;
98    let mut out = CapabilityCallResult {
99        call_id: call.call_id.clone(),
100        provider_id: call.provider_id.clone(),
101        contract_id: call.contract_id.clone(),
102        ..Default::default()
103    };
104    match result {
105        Ok(s) => {
106            out.success = true;
107            out.output = s;
108        }
109        Err(e) => {
110            out.success = false;
111            out.error = e.to_string();
112        }
113    }
114    out
115}
116
117async fn run(
118    op: &str,
119    args_json: &str,
120    runtime: &PlanRuntime,
121    plan_id: &str,
122) -> anyhow::Result<String> {
123    match op {
124        "read_file" => read_file(args_json),
125        "write_file" => write_file(args_json),
126        "patch_file" => patch_file(args_json),
127        "list_dir" => list_dir(args_json),
128        "run_command" => run_command(args_json, runtime, plan_id).await,
129        other => anyhow::bail!("unknown builtin: {}", other),
130    }
131}
132
133/// Static metadata for the builtin ops. Used by main.rs to declare them
134/// against atlas at startup so pilot can discover them like any other provider.
135pub struct BuiltinSpec {
136    pub op: &'static str,
137    pub description: &'static str,
138    pub input_schema_json: &'static str,
139}
140
141pub const BUILTINS: &[BuiltinSpec] = &[
142    BuiltinSpec {
143        op: "read_file",
144        description: "Read a file and return its contents",
145        input_schema_json: r#"{"type":"object","properties":{"path":{"type":"string","description":"Absolute or relative file path"}},"required":["path"]}"#,
146    },
147    BuiltinSpec {
148        op: "write_file",
149        description: "Write content to a file (creates or overwrites)",
150        input_schema_json: r#"{"type":"object","properties":{"path":{"type":"string"},"content":{"type":"string"}},"required":["path","content"]}"#,
151    },
152    BuiltinSpec {
153        op: "patch_file",
154        description: "Replace the first occurrence of a string in a file",
155        input_schema_json: r#"{"type":"object","properties":{"path":{"type":"string"},"old":{"type":"string"},"new":{"type":"string"}},"required":["path","old","new"]}"#,
156    },
157    BuiltinSpec {
158        op: "list_dir",
159        description: "List files and directories at a path",
160        input_schema_json: r#"{"type":"object","properties":{"path":{"type":"string","description":"Directory path (default: current dir)"}}}"#,
161    },
162    BuiltinSpec {
163        op: "run_command",
164        description: "Run a shell command and return stdout/stderr",
165        input_schema_json: r#"{"type":"object","properties":{"command":{"type":"string"}},"required":["command"]}"#,
166    },
167    BuiltinSpec {
168        op: "cancel_plan",
169        description: "Cancellation for an in-flight RTDL plan by plan_id",
170        input_schema_json: r#"{"type":"object","properties":{"plan_id":{"type":"string","description":"RTDL Plan.plan_id to cancel"},"wait_ms":{"type":"integer","minimum":0,"description":"Optional milliseconds to wait for the target plan to stop; default 5000"}},"required":["plan_id"]}"#,
171    },
172    BuiltinSpec {
173        op: "get_all_plans",
174        description: "List every in-flight RTDL plan with its plan_id, a short description of the task, op_count, cancelled flag, and number of armed stop points. Call this to discover which plans are currently running, then inspect one with get_plan_status before stopping it. Takes no arguments.",
175        input_schema_json: r#"{"type":"object","properties":{}}"#,
176    },
177    BuiltinSpec {
178        op: "get_plan_status",
179        description: "Inspect an in-flight RTDL plan: returns its ops, each with op_id, kind, description, current state (pending/running/succeeded/failed/canceled/timeout/paused) and any armed stop_point. Call this to find the op_id and live progress of a running plan before issuing stop_plan_at or cancel_plan. Errors if the plan is not active (stale/wrong id or already finished) — use get_all_plans to list running plans.",
180        input_schema_json: r#"{"type":"object","properties":{"plan_id":{"type":"string","description":"RTDL Plan.plan_id to inspect"}},"required":["plan_id"]}"#,
181    },
182    BuiltinSpec {
183        op: "stop_plan_at",
184        description: "Set a stop point on an in-flight RTDL plan: when execution reaches the op with the given op_id, cancel the whole plan. Use 'on_complete' (default) to stop right after that op finishes, or 'on_enter' to stop the moment it is reached, before it runs. op_ids are the per-node identifiers shown in RTDL node_state events.",
185        input_schema_json: r#"{"type":"object","properties":{"plan_id":{"type":"string","description":"RTDL Plan.plan_id to set the stop point on"},"op_id":{"type":"string","description":"Node op_id at which to stop (cancel) the plan"},"when":{"type":"string","enum":["on_enter","on_complete"],"description":"on_enter = before the op runs; on_complete = after it finishes. Default on_complete."}},"required":["plan_id","op_id"]}"#,
186    },
187    BuiltinSpec {
188        op: "read_capability_doc",
189        description: "Read a capability provider's full CAPABILITY.md manual by provider_id. Call this to learn how to use a provider before invoking it. Only providers listed as having a doc carry one; never guess or read a file path for docs.",
190        input_schema_json: r#"{"type":"object","properties":{"provider_id":{"type":"string","description":"The provider_id whose CAPABILITY.md to read"}},"required":["provider_id"]}"#,
191    },
192];
193
194#[derive(Deserialize)]
195struct DocArgs {
196    provider_id: String,
197}
198
199/// Apply the `read_capability_doc` builtin: fetch a provider's registered
200/// CAPABILITY.md *content* from atlas and return it as markdown text.
201///
202/// Unlike `read_file`, this never touches the filesystem — atlas serves the
203/// content the provider sent at registration, so it works regardless of the
204/// provider's (possibly containerised) mount layout. Errors when the provider
205/// is unknown or registered no doc; output is truncated to keep prompt size
206/// bounded.
207async fn read_capability_doc(
208    call: &CapabilityCall,
209    atlas: &mut AtlasClient,
210) -> CapabilityCallResult {
211    let mut out = CapabilityCallResult {
212        call_id: call.call_id.clone(),
213        provider_id: call.provider_id.clone(),
214        contract_id: call.contract_id.clone(),
215        ..Default::default()
216    };
217    let provider_id = match serde_json::from_str::<DocArgs>(&call.args_json) {
218        Ok(a) => a.provider_id.trim().to_string(),
219        Err(e) => {
220            out.error = format!("invalid read_capability_doc args: {e}");
221            return out;
222        }
223    };
224    if provider_id.is_empty() {
225        out.error = "read_capability_doc: provider_id is required".to_string();
226        return out;
227    }
228    match atlas
229        .query_capabilities(&provider_id, "", atlas_pb::Transport::Unspecified)
230        .await
231    {
232        Ok(providers) => match providers.iter().find(|p| p.id == provider_id) {
233            Some(p) if !p.capability_md.is_empty() => {
234                out.success = true;
235                out.output = truncate(&p.capability_md, 12000);
236            }
237            Some(_) => {
238                out.error = format!("provider '{provider_id}' registered no CAPABILITY.md");
239            }
240            None => {
241                out.error = format!("no provider '{provider_id}' registered in atlas");
242            }
243        },
244        Err(e) => {
245            out.error = format!("atlas query for '{provider_id}' failed: {e}");
246        }
247    }
248    out
249}
250
251// ── Helpers ───────────────────────────────────────────────────────────────────
252
253fn truncate(s: &str, max: usize) -> String {
254    if s.len() <= max {
255        s.to_string()
256    } else {
257        // Find a valid UTF-8 char boundary at or before `max` to avoid panic.
258        let mut end = max;
259        while end > 0 && !s.is_char_boundary(end) {
260            end -= 1;
261        }
262        format!("{}…(truncated, {} bytes total)", &s[..end], s.len())
263    }
264}
265
266#[derive(Deserialize)]
267struct ReadArgs {
268    path: String,
269}
270#[derive(Deserialize)]
271struct WriteArgs {
272    path: String,
273    content: String,
274}
275#[derive(Deserialize)]
276struct PatchArgs {
277    path: String,
278    old: String,
279    new: String,
280}
281#[derive(Deserialize)]
282struct ListArgs {
283    path: Option<String>,
284}
285#[derive(Deserialize)]
286struct CmdArgs {
287    command: String,
288}
289
290fn read_file(args: &str) -> anyhow::Result<String> {
291    let a: ReadArgs = serde_json::from_str(args)?;
292    let path = safe_resolve(&a.path)?;
293    Ok(truncate(&std::fs::read_to_string(path)?, 8000))
294}
295
296fn write_file(args: &str) -> anyhow::Result<String> {
297    let a: WriteArgs = serde_json::from_str(args)?;
298    let path = safe_resolve(&a.path)?;
299    if let Some(parent) = path.parent() {
300        std::fs::create_dir_all(parent).ok();
301    }
302    std::fs::write(&path, &a.content)?;
303    Ok(format!(
304        "wrote {} bytes to {}",
305        a.content.len(),
306        path.display()
307    ))
308}
309
310fn patch_file(args: &str) -> anyhow::Result<String> {
311    // TODO: use standard tools like sed, awk, etc.
312    let a: PatchArgs = serde_json::from_str(args)?;
313    let path = safe_resolve(&a.path)?;
314    let content = std::fs::read_to_string(&path)?;
315    if !content.contains(&a.old) {
316        anyhow::bail!("old string not found in file");
317    }
318    std::fs::write(&path, content.replacen(&a.old, &a.new, 1))?;
319    Ok(format!("patched {}", path.display()))
320}
321
322fn list_dir(args: &str) -> anyhow::Result<String> {
323    let a: ListArgs = serde_json::from_str(args)?;
324    let dir = a.path.as_deref().unwrap_or(".");
325    let resolved = safe_resolve(dir)?;
326    let mut items: Vec<String> = std::fs::read_dir(resolved)?
327        .flatten()
328        .map(|e| {
329            let ft = e
330                .file_type()
331                .map(|t| if t.is_dir() { "dir" } else { "file" })
332                .unwrap_or("?");
333            format!("{} {}", ft, e.file_name().to_string_lossy())
334        })
335        .collect();
336    items.sort();
337    Ok(items.join("\n"))
338}
339
340/// Maximum command length to prevent abuse.
341const MAX_COMMAND_LEN: usize = 8192;
342/// Command execution timeout.
343const COMMAND_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
344
345async fn run_command(args: &str, runtime: &PlanRuntime, plan_id: &str) -> anyhow::Result<String> {
346    let a: CmdArgs = serde_json::from_str(args)?;
347    if a.command.len() > MAX_COMMAND_LEN {
348        anyhow::bail!(
349            "command too long ({} bytes, max {})",
350            a.command.len(),
351            MAX_COMMAND_LEN
352        );
353    }
354    if runtime.is_cancelled(plan_id).await {
355        anyhow::bail!("command canceled before spawn");
356    }
357
358    use std::process::Stdio;
359    let mut command = tokio::process::Command::new("bash");
360    command
361        .arg("-c")
362        .arg(&a.command)
363        .stdin(Stdio::null())
364        .stdout(Stdio::piped())
365        .stderr(Stdio::piped())
366        .kill_on_drop(true)
367        .process_group(0);
368    let child = command
369        .spawn()
370        .map_err(|e| anyhow::anyhow!("failed to execute command: {e}"))?;
371    let pgid = child
372        .id()
373        .ok_or_else(|| anyhow::anyhow!("spawned command has no pid"))?;
374    let output = child.wait_with_output();
375    tokio::pin!(output);
376
377    enum StopReason {
378        Canceled,
379        Timeout,
380    }
381    let canceled = async {
382        loop {
383            if runtime.is_cancelled(plan_id).await {
384                break;
385            }
386            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
387        }
388    };
389    tokio::pin!(canceled);
390    let deadline = tokio::time::sleep(COMMAND_TIMEOUT);
391    tokio::pin!(deadline);
392
393    let stop_reason = tokio::select! {
394        result = &mut output => {
395            let out = result.map_err(|e| anyhow::anyhow!("failed to execute command: {e}"))?;
396            return format_command_output(out);
397        }
398        _ = &mut canceled => StopReason::Canceled,
399        _ = &mut deadline => StopReason::Timeout,
400    };
401
402    terminate_command_group(pgid, nix::sys::signal::Signal::SIGTERM);
403    if tokio::time::timeout(std::time::Duration::from_secs(2), &mut output)
404        .await
405        .is_err()
406    {
407        terminate_command_group(pgid, nix::sys::signal::Signal::SIGKILL);
408        let _ = output.await;
409    }
410    match stop_reason {
411        StopReason::Canceled => anyhow::bail!("command canceled"),
412        StopReason::Timeout => {
413            anyhow::bail!("command timed out after {}s", COMMAND_TIMEOUT.as_secs())
414        }
415    }
416}
417
418fn terminate_command_group(pgid: u32, signal: nix::sys::signal::Signal) {
419    use nix::sys::signal::killpg;
420    use nix::unistd::Pid;
421    match killpg(Pid::from_raw(pgid as i32), signal) {
422        Ok(()) | Err(nix::errno::Errno::ESRCH) => {}
423        Err(error) => robonix_scribe::warn!(
424            "[executor] signal {:?} command pgid {} failed: {}",
425            signal,
426            pgid,
427            error
428        ),
429    }
430}
431
432fn format_command_output(out: std::process::Output) -> anyhow::Result<String> {
433    let mut result = String::new();
434    let stdout = String::from_utf8_lossy(&out.stdout);
435    let stderr = String::from_utf8_lossy(&out.stderr);
436    if !stdout.is_empty() {
437        result.push_str(&truncate(&stdout, 4000));
438    }
439    if !stderr.is_empty() {
440        if !result.is_empty() {
441            result.push('\n');
442        }
443        result.push_str("stderr: ");
444        result.push_str(&truncate(&stderr, 2000));
445    }
446    if result.is_empty() {
447        result = format!("exit code: {}", out.status.code().unwrap_or(-1));
448    }
449    Ok(result)
450}
451
452#[cfg(test)]
453mod tests {
454    use super::*;
455
456    #[test]
457    fn path_traversal_dotdot_is_rejected() {
458        // Set workspace to a temp dir so we have a known root
459        let tmp = std::env::temp_dir().join("rbnx_test_ws");
460        std::fs::create_dir_all(&tmp).unwrap();
461        unsafe { std::env::set_var("ROBONIX_WORKSPACE", tmp.to_str().unwrap()) };
462
463        // ../../../etc/passwd must be rejected
464        let result = safe_resolve("../../../etc/passwd");
465        assert!(
466            result.is_err(),
467            "path traversal with ../../../etc/passwd should fail"
468        );
469        let err_msg = result.unwrap_err().to_string();
470        assert!(
471            err_msg.contains("path traversal denied") || err_msg.contains("does not exist"),
472            "error should mention traversal or nonexistent path, got: {err_msg}"
473        );
474    }
475
476    #[test]
477    fn path_traversal_absolute_outside_workspace_is_rejected() {
478        let tmp = std::env::temp_dir().join("rbnx_test_ws2");
479        std::fs::create_dir_all(&tmp).unwrap();
480        unsafe { std::env::set_var("ROBONIX_WORKSPACE", tmp.to_str().unwrap()) };
481
482        // /etc/hostname is a real file outside workspace
483        let result = safe_resolve("/etc/hostname");
484        assert!(
485            result.is_err(),
486            "absolute path /etc/hostname outside workspace should fail"
487        );
488        let err_msg = result.unwrap_err().to_string();
489        assert!(
490            err_msg.contains("path traversal denied"),
491            "error should mention traversal, got: {err_msg}"
492        );
493    }
494
495    #[test]
496    fn path_resolve_within_workspace_is_allowed() {
497        let tmp = std::env::temp_dir().join("rbnx_test_ws3");
498        std::fs::create_dir_all(&tmp).unwrap();
499        // Create a test file inside workspace
500        let test_file = tmp.join("allowed.txt");
501        std::fs::write(&test_file, "hello").unwrap();
502        unsafe { std::env::set_var("ROBONIX_WORKSPACE", tmp.to_str().unwrap()) };
503
504        let result = safe_resolve("allowed.txt");
505        assert!(
506            result.is_ok(),
507            "path within workspace should be allowed: {:?}",
508            result.err()
509        );
510
511        // Cleanup
512        let _ = std::fs::remove_file(&test_file);
513        let _ = std::fs::remove_dir(&tmp);
514    }
515
516    #[test]
517    fn truncate_ascii_works() {
518        let s = "hello world";
519        assert_eq!(truncate(s, 100), "hello world");
520        let t = truncate(s, 5);
521        assert!(
522            t.starts_with("hello"),
523            "should start with 'hello', got: {t}"
524        );
525        assert!(
526            t.contains("truncated"),
527            "should contain 'truncated', got: {t}"
528        );
529    }
530
531    #[test]
532    fn truncate_multibyte_utf8_does_not_panic() {
533        // Each euro sign is 3 bytes in UTF-8
534        let s = "€€€€€€€€"; // 8 chars × 3 bytes = 24 bytes
535        // Truncate at byte 7 — in the middle of the 3rd char '€'
536        let result = truncate(s, 7);
537        // Should NOT panic, and should truncate at a valid boundary
538        assert!(
539            result.contains("truncated"),
540            "should be truncated, got: {result}"
541        );
542        // The truncated prefix should be valid UTF-8 (it is, since we're returning a String)
543        assert!(
544            result.starts_with("€€"),
545            "should keep first 2 chars, got: {result}"
546        );
547    }
548
549    #[test]
550    fn truncate_emoji_boundary() {
551        // 🚀 is 4 bytes in UTF-8
552        let s = "A🚀B🚀C"; // 1 + 4 + 1 + 4 + 1 = 11 bytes
553        // Truncate at byte 3, which is in the middle of 🚀
554        let result = truncate(s, 3);
555        assert!(result.contains("truncated"), "should be truncated");
556        // Should only keep "A" since 🚀 starts at byte 1 and ends at byte 5
557        assert!(
558            result.starts_with("A"),
559            "should start with 'A', got: {result}"
560        );
561    }
562}
robonix_executor/dispatch/builtin.rs

robonix_executor/dispatch/
builtin.rs
