boxxkite_client/
processes.rs

1//! Background processes/sessions (`docs/PROCESS-SESSIONS-DESIGN.md`):
2//! `POST /v1/sandboxes/{id}/processes` and its `/{process_id}/output`,
3//! `/{process_id}/input`, `/{process_id}/stop` siblings, plus
4//! `GET /v1/sandboxes/{id}/processes`. Mirrors `sdk-python`'s
5//! `start_process`/`list_processes`/`get_process_output`/
6//! `send_process_input`/`stop_process`.
7//!
8//! Distinct from [`crate::Client::exec`]: `exec` is one-shot request/
9//! response bounded by its own timeout; these routes track a process across
10//! multiple calls. Polling-style, not streaming -- the control plane's own
11//! route comments mark SSE/streaming process output as a separate, later
12//! phase, not part of this route set.
13
14use std::pin::Pin;
15
16use futures_util::{Stream, StreamExt};
17use reqwest::Method;
18use reqwest_eventsource::{Event, EventSource};
19use serde::{Deserialize, Serialize};
20
21use crate::client::Client;
22use crate::error::BoxxkiteError;
23
24/// `max_runtime_seconds` is required by the server
25/// (`SandboxProcessStartRequest`) with a 4h ceiling
26/// (`SANDBOX_PROCESS_MAX_RUNTIME_SECONDS_CEILING`) -- this is only this
27/// crate's client-side default when the caller doesn't set one, matching
28/// `sdk-python`/`sdk-js`/`sdk-go`'s own default of 3600. The ceiling itself
29/// is enforced server-side and deliberately not duplicated here.
30const DEFAULT_MAX_RUNTIME_SECONDS: i64 = 3600;
31
32/// Builder for `POST /v1/sandboxes/{id}/processes`'s optional fields.
33#[derive(Debug, Clone)]
34pub struct StartProcessOptions {
35    description: Option<String>,
36    max_runtime_seconds: i64,
37    expose_port: Option<i32>,
38}
39
40impl Default for StartProcessOptions {
41    fn default() -> Self {
42        Self {
43            description: None,
44            max_runtime_seconds: DEFAULT_MAX_RUNTIME_SECONDS,
45            expose_port: None,
46        }
47    }
48}
49
50impl StartProcessOptions {
51    pub fn new() -> Self {
52        Self::default()
53    }
54
55    pub fn description(mut self, description: impl Into<String>) -> Self {
56        self.description = Some(description.into());
57        self
58    }
59
60    /// Hard ceiling on how long the process may run before being
61    /// force-killed, in seconds. Defaults to 3600; capped server-side at 4h.
62    pub fn max_runtime_seconds(mut self, max_runtime_seconds: i64) -> Self {
63        self.max_runtime_seconds = max_runtime_seconds;
64        self
65    }
66
67    /// Opt-in port to expose via a signed preview URL once the process is
68    /// listening -- see `docs/NETWORK-INGRESS-DESIGN.md` and
69    /// `POST /v1/sandboxes/{id}/preview/{port}`. Leave unset for a normal,
70    /// fully network-isolated background process.
71    pub fn expose_port(mut self, expose_port: i32) -> Self {
72        self.expose_port = Some(expose_port);
73        self
74    }
75}
76
77#[derive(Serialize)]
78struct StartProcessBody<'a> {
79    command: &'a str,
80    #[serde(skip_serializing_if = "Option::is_none")]
81    description: Option<&'a str>,
82    max_runtime_seconds: i64,
83    #[serde(skip_serializing_if = "Option::is_none")]
84    expose_port: Option<i32>,
85}
86
87/// `POST /v1/sandboxes/{id}/processes`'s response.
88#[derive(Debug, Clone, Deserialize)]
89pub struct ProcessStartResult {
90    pub process_id: String,
91    pub status: String,
92    pub started_at: String,
93}
94
95/// One tracked background process (`SandboxProcessInfo`), as returned by
96/// [`Client::list_processes`].
97#[derive(Debug, Clone, Deserialize)]
98pub struct ProcessInfo {
99    pub process_id: String,
100    pub command: String,
101    pub description: Option<String>,
102    pub status: String,
103    pub started_at: String,
104    pub exit_code: Option<i32>,
105    pub expose_port: Option<i32>,
106}
107
108/// `GET /v1/sandboxes/{id}/processes`'s response.
109#[derive(Debug, Clone, Deserialize)]
110pub struct ProcessListResult {
111    pub processes: Vec<ProcessInfo>,
112}
113
114/// `GET /v1/sandboxes/{id}/processes/{process_id}/output`'s response.
115#[derive(Debug, Clone, Deserialize)]
116pub struct ProcessOutputResult {
117    pub status: String,
118    pub stdout_chunk: String,
119    pub next_offset: i64,
120    pub truncated: bool,
121    pub exit_code: Option<i32>,
122}
123
124/// One event from `GET /v1/sandboxes/{id}/processes/{process_id}/stream`: an
125/// [`Output`](ProcessStreamEvent::Output) per new stdout chunk, then a
126/// terminal [`Exit`](ProcessStreamEvent::Exit) once the process finishes.
127#[derive(Debug, Clone, Deserialize)]
128#[serde(tag = "type", rename_all = "lowercase")]
129pub enum ProcessStreamEvent {
130    Output {
131        stdout_chunk: String,
132        next_offset: i64,
133        truncated: bool,
134    },
135    Exit {
136        status: String,
137        exit_code: Option<i32>,
138    },
139}
140
141/// `POST /v1/sandboxes/{id}/processes/{process_id}/input`'s response.
142#[derive(Debug, Clone, Deserialize)]
143pub struct ProcessInputResult {
144    pub bytes_written: i64,
145}
146
147/// `POST /v1/sandboxes/{id}/processes/{process_id}/stop`'s response.
148#[derive(Debug, Clone, Deserialize)]
149pub struct ProcessStopResult {
150    pub status: String,
151    pub exit_code: Option<i32>,
152}
153
154impl Client {
155    /// `POST /v1/sandboxes/{session_id}/processes` -- start a background
156    /// process (a dev server, a test watcher, a long build, a REPL) that
157    /// keeps running after this call returns. Not reachable over the
158    /// network from any other call by default -- same per-exec network
159    /// isolation applies here -- unless `options.expose_port(..)` is set.
160    pub async fn start_process(
161        &self,
162        session_id: &str,
163        command: &str,
164        options: StartProcessOptions,
165    ) -> Result<ProcessStartResult, BoxxkiteError> {
166        let body = StartProcessBody {
167            command,
168            description: options.description.as_deref(),
169            max_runtime_seconds: options.max_runtime_seconds,
170            expose_port: options.expose_port,
171        };
172        let builder = self
173            .request(
174                Method::POST,
175                &format!("/v1/sandboxes/{session_id}/processes"),
176            )
177            .json(&body);
178        self.send(builder).await
179    }
180
181    /// `GET /v1/sandboxes/{session_id}/processes` -- every background
182    /// process currently tracked for this session (running, exited, or
183    /// stopped).
184    pub async fn list_processes(
185        &self,
186        session_id: &str,
187    ) -> Result<ProcessListResult, BoxxkiteError> {
188        let builder = self.request(
189            Method::GET,
190            &format!("/v1/sandboxes/{session_id}/processes"),
191        );
192        self.send(builder).await
193    }
194
195    /// `GET /v1/sandboxes/{session_id}/processes/{process_id}/output` --
196    /// poll a background process's output since a given byte offset.
197    /// Polling-style, not streaming. `since_offset` (from a previous call's
198    /// `next_offset`, or 0 the first time) fetches only the new output
199    /// since the last check.
200    pub async fn get_process_output(
201        &self,
202        session_id: &str,
203        process_id: &str,
204        since_offset: i64,
205    ) -> Result<ProcessOutputResult, BoxxkiteError> {
206        let builder = self
207            .request(
208                Method::GET,
209                &format!("/v1/sandboxes/{session_id}/processes/{process_id}/output"),
210            )
211            .query(&[("since_offset", since_offset.to_string())]);
212        self.send(builder).await
213    }
214
215    /// `GET /v1/sandboxes/{session_id}/processes/{process_id}/stream` -- live
216    /// SSE stream of a background process's stdout, the streaming counterpart
217    /// to [`get_process_output`](Client::get_process_output). Yields an
218    /// [`ProcessStreamEvent::Output`] per new chunk, then a terminal
219    /// [`ProcessStreamEvent::Exit`]. Bring `futures_util::StreamExt` into
220    /// scope to consume it; the stream ends when the process finishes, the
221    /// session is destroyed, or the caller drops it.
222    pub fn stream_process_output(
223        &self,
224        session_id: &str,
225        process_id: &str,
226        since_offset: i64,
227    ) -> Pin<Box<dyn Stream<Item = Result<ProcessStreamEvent, BoxxkiteError>> + Send + 'static>> {
228        let request_builder = self
229            .request(
230                Method::GET,
231                &format!("/v1/sandboxes/{session_id}/processes/{process_id}/stream"),
232            )
233            .query(&[("since_offset", since_offset.to_string())]);
234
235        Box::pin(async_stream::stream! {
236            let mut event_source = match EventSource::new(request_builder) {
237                Ok(event_source) => event_source,
238                Err(err) => {
239                    yield Err(BoxxkiteError::Config(format!("failed to build stream request: {err}")));
240                    return;
241                }
242            };
243
244            while let Some(event) = event_source.next().await {
245                match event {
246                    Ok(Event::Open) => continue,
247                    Ok(Event::Message(message)) => {
248                        yield serde_json::from_str::<ProcessStreamEvent>(&message.data).map_err(BoxxkiteError::from);
249                    }
250                    Err(reqwest_eventsource::Error::StreamEnded) => break,
251                    Err(reqwest_eventsource::Error::InvalidStatusCode(status, response)) => {
252                        let bytes = response.bytes().await.unwrap_or_default();
253                        yield Err(crate::error::api_error_from_bytes(status.as_u16(), &bytes));
254                        break;
255                    }
256                    Err(err) => {
257                        yield Err(BoxxkiteError::from(err));
258                        break;
259                    }
260                }
261            }
262            event_source.close();
263        })
264    }
265
266    /// `POST /v1/sandboxes/{session_id}/processes/{process_id}/input` --
267    /// write to a tracked background process's stdin pipe (e.g. answering
268    /// an interactive prompt).
269    pub async fn send_process_input(
270        &self,
271        session_id: &str,
272        process_id: &str,
273        data: &str,
274    ) -> Result<ProcessInputResult, BoxxkiteError> {
275        #[derive(Serialize)]
276        struct Body<'a> {
277            data: &'a str,
278        }
279        let builder = self
280            .request(
281                Method::POST,
282                &format!("/v1/sandboxes/{session_id}/processes/{process_id}/input"),
283            )
284            .json(&Body { data });
285        self.send(builder).await
286    }
287
288    /// `POST /v1/sandboxes/{session_id}/processes/{process_id}/stop` -- stop
289    /// a tracked background process: SIGTERM, a short grace period, then
290    /// SIGKILL if still alive.
291    pub async fn stop_process(
292        &self,
293        session_id: &str,
294        process_id: &str,
295    ) -> Result<ProcessStopResult, BoxxkiteError> {
296        let builder = self.request(
297            Method::POST,
298            &format!("/v1/sandboxes/{session_id}/processes/{process_id}/stop"),
299        );
300        self.send(builder).await
301    }
302}
boxxkite_client/processes.rs

boxxkite_client/
processes.rs
