solari/
commands.rs

1//! `commands` namespace: run / start / stdin / kill over the control WS, plus
2//! the one-shot REST `/exec` fast path. Mirrors the `commands` surface + the
3//! `runCommand`/`startCommand` internals of `handle.ts`.
4
5use std::sync::Arc;
6
7use base64::Engine;
8use serde_json::{Map, Value};
9use tokio::sync::mpsc;
10
11use crate::channel::{AsyncFrame, ControlChannel};
12use crate::client::Sandbox;
13use crate::error::{is_route_unavailable, SolariError};
14use crate::http::encode_uri_component;
15use crate::types::CommandResult;
16
17/// A streamed-output callback (`stdout`/`stderr`). `Fn(&str)`.
18pub type OutputCallback = Arc<dyn Fn(&str) + Send + Sync>;
19
20/// Options for `commands.run` / `commands.start`.
21#[derive(Default, Clone)]
22pub struct RunOptions {
23    /// argv tail (the guest runs `cmd` with these, NOT via a shell).
24    pub args: Option<Vec<String>>,
25    pub cwd: Option<String>,
26    pub env: Option<std::collections::HashMap<String, String>>,
27    pub user: Option<String>,
28    pub timeout_ms: Option<u64>,
29    /// Run detached: returns immediately, output still streams via callbacks.
30    pub background: bool,
31    pub on_stdout: Option<OutputCallback>,
32    pub on_stderr: Option<OutputCallback>,
33}
34
35impl RunOptions {
36    pub fn new() -> Self {
37        Self::default()
38    }
39    pub fn args<I, S>(mut self, args: I) -> Self
40    where
41        I: IntoIterator<Item = S>,
42        S: Into<String>,
43    {
44        self.args = Some(args.into_iter().map(Into::into).collect());
45        self
46    }
47    pub fn cwd(mut self, cwd: impl Into<String>) -> Self {
48        self.cwd = Some(cwd.into());
49        self
50    }
51}
52
53/// A started command handle (`commands.start`).
54pub struct CommandHandle {
55    pub cmd_id: String,
56    channel: Arc<ControlChannel>,
57    rx: mpsc::UnboundedReceiver<AsyncFrame>,
58    on_stdout: Option<OutputCallback>,
59    on_stderr: Option<OutputCallback>,
60}
61
62impl CommandHandle {
63    /// Write bytes to the command's stdin.
64    pub async fn stdin(&self, data: impl AsRef<[u8]>) -> Result<(), SolariError> {
65        let b64 = base64::engine::general_purpose::STANDARD.encode(data.as_ref());
66        let mut p = Map::new();
67        p.insert("cmdId".into(), Value::String(self.cmd_id.clone()));
68        p.insert("base64".into(), Value::String(b64));
69        self.channel.call("cmd.stdin", Value::Object(p)).await.map(|_| ())
70    }
71
72    /// Send a signal (default SIGTERM guest-side).
73    pub async fn kill(&self, signal: Option<i64>) -> Result<(), SolariError> {
74        let mut p = Map::new();
75        p.insert("cmdId".into(), Value::String(self.cmd_id.clone()));
76        if let Some(s) = signal {
77            p.insert("signal".into(), Value::from(s));
78        }
79        self.channel.call("cmd.kill", Value::Object(p)).await.map(|_| ())
80    }
81
82    /// Wait for the command to exit, accumulating stdout/stderr.
83    pub async fn wait(mut self) -> Result<CommandResult, SolariError> {
84        let mut stdout = String::new();
85        let mut stderr = String::new();
86        while let Some(frame) = self.rx.recv().await {
87            match frame.typ.as_str() {
88                "cmd.data" => {
89                    let text = frame
90                        .base64
91                        .as_ref()
92                        .map(|b| decode_text(b))
93                        .unwrap_or_default();
94                    let is_err = frame.stream.as_deref() == Some("stderr");
95                    if is_err {
96                        stderr.push_str(&text);
97                        if let Some(cb) = &self.on_stderr {
98                            cb(&text);
99                        }
100                    } else {
101                        stdout.push_str(&text);
102                        if let Some(cb) = &self.on_stdout {
103                            cb(&text);
104                        }
105                    }
106                }
107                "cmd.exit" => {
108                    let exit_code = frame.exit_code.unwrap_or(0);
109                    self.channel.unregister_command(&self.cmd_id);
110                    return Ok(CommandResult { exit_code, stdout, stderr });
111                }
112                _ => {}
113            }
114        }
115        // Channel dropped before `cmd.exit`.
116        Err(SolariError::connection("Control channel closed"))
117    }
118}
119
120fn decode_text(b64: &str) -> String {
121    match base64::engine::general_purpose::STANDARD.decode(b64) {
122        Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
123        Err(_) => String::new(),
124    }
125}
126
127/// Build a JSON params object, inserting only present (`Some`) fields — mirrors
128/// how JSON.stringify omits `undefined`.
129fn cmd_start_params(cmd: &str, opts: &RunOptions) -> Value {
130    let mut p = Map::new();
131    p.insert("cmd".into(), Value::String(cmd.to_string()));
132    if let Some(args) = &opts.args {
133        p.insert("args".into(), Value::from(args.clone()));
134    }
135    if let Some(cwd) = &opts.cwd {
136        p.insert("cwd".into(), Value::String(cwd.clone()));
137    }
138    if let Some(env) = &opts.env {
139        p.insert("env".into(), serde_json::to_value(env).unwrap());
140    }
141    if let Some(user) = &opts.user {
142        p.insert("user".into(), Value::String(user.clone()));
143    }
144    Value::Object(p)
145}
146
147/// `commands.start`: start a process and return a handle immediately.
148pub(crate) async fn start_command(
149    sb: &Sandbox,
150    cmd: &str,
151    opts: &RunOptions,
152) -> Result<CommandHandle, SolariError> {
153    sb.channel().connect().await?;
154    let result = sb
155        .channel()
156        .call("cmd.start", cmd_start_params(cmd, opts))
157        .await?;
158    let cmd_id = result
159        .get("cmdId")
160        .and_then(Value::as_str)
161        .ok_or_else(|| SolariError::Other("cmd.start reply missing cmdId".into()))?
162        .to_string();
163    let rx = sb.channel().register_command(&cmd_id);
164    Ok(CommandHandle {
165        cmd_id,
166        channel: sb.channel_arc(),
167        rx,
168        on_stdout: opts.on_stdout.clone(),
169        on_stderr: opts.on_stderr.clone(),
170    })
171}
172
173/// `commands.run`: run a command to completion.
174pub(crate) async fn run_command(
175    sb: &Sandbox,
176    cmd: &str,
177    opts: &RunOptions,
178) -> Result<CommandResult, SolariError> {
179    // One-shot HTTP fast path: a plain run-to-completion (no streaming
180    // callbacks, not backgrounded, no per-command env/user) over the warm REST
181    // connection, ONLY when the control channel isn't already open.
182    if !opts.background
183        && opts.on_stdout.is_none()
184        && opts.on_stderr.is_none()
185        && opts.env.is_none()
186        && opts.user.is_none()
187        && !sb.channel().connected()
188    {
189        match exec_oneshot(sb, cmd, opts).await {
190            Ok(r) => return Ok(r),
191            // Fall back to the WS path ONLY when `/exec` is unavailable. Any
192            // other failure propagates (never silently double-execute).
193            Err(err) => {
194                if !is_route_unavailable(&err) {
195                    return Err(err);
196                }
197            }
198        }
199    }
200
201    let handle = start_command(sb, cmd, opts).await?;
202    if opts.background {
203        return Ok(CommandResult {
204            exit_code: 0,
205            stdout: String::new(),
206            stderr: String::new(),
207        });
208    }
209    handle.wait().await
210}
211
212/// POST `/sandboxes/:id/exec` — the one-shot fast path.
213async fn exec_oneshot(
214    sb: &Sandbox,
215    cmd: &str,
216    opts: &RunOptions,
217) -> Result<CommandResult, SolariError> {
218    let mut body = Map::new();
219    body.insert("cmd".into(), Value::String(cmd.to_string()));
220    if let Some(args) = &opts.args {
221        body.insert("args".into(), Value::from(args.clone()));
222    }
223    if let Some(cwd) = &opts.cwd {
224        body.insert("cwd".into(), Value::String(cwd.clone()));
225    }
226    if let Some(t) = opts.timeout_ms {
227        body.insert("timeoutMs".into(), Value::from(t));
228    }
229    let path = format!("/sandboxes/{}/exec", encode_uri_component(sb.id()));
230    sb.http()
231        .request::<CommandResult>("POST", &path, Some(Value::Object(body)), None)
232        .await
233}
234
235/// The ergonomic `commands` accessor returned by `Sandbox::commands()`.
236pub struct Commands<'a> {
237    pub(crate) sb: &'a Sandbox,
238}
239
240impl<'a> Commands<'a> {
241    /// Run a command to completion.
242    pub async fn run(&self, cmd: &str, opts: RunOptions) -> Result<CommandResult, SolariError> {
243        run_command(self.sb, cmd, &opts).await
244    }
245
246    /// Start a command and return a handle immediately.
247    pub async fn start(&self, cmd: &str, opts: RunOptions) -> Result<CommandHandle, SolariError> {
248        start_command(self.sb, cmd, &opts).await
249    }
250}
solari/commands.rs

solari/
commands.rs
