1use std::collections::BTreeMap;
9
10use async_trait::async_trait;
11use base64::engine::general_purpose::STANDARD as BASE64;
12use base64::Engine as _;
13use futures::stream::{self, BoxStream};
14use serde::Deserialize;
15use serde_json::json;
16
17use crate::error::{ErrorData, Result};
18use crate::providers::sandbox::{guard_for, Bounded, TimeoutReport};
19use crate::traits::{
20 Binding, CommandOutput, CreateSandboxRequest, JobPoll, JobStart, PreviewCapability,
21 ResolvedSandbox, RunCommandRequest, Sandbox, SandboxInstance, SandboxState,
22};
23use alien_core::bindings::LocalSandboxBinding;
24use alien_core::{Platform, SandboxCapabilities, SandboxCapability};
25use alien_error::{AlienError, Context, ContextError, IntoAlienError};
26
27#[derive(Debug, Deserialize)]
28#[serde(rename_all = "camelCase")]
29struct SessionBody {
30 session_id: String,
31 #[allow(dead_code)]
32 container_id: String,
33}
34
35#[derive(Debug, Deserialize)]
36#[serde(rename_all = "camelCase", tag = "stream", content = "dataBase64")]
37enum OutputFrame {
38 Stdout(String),
39 Stderr(String),
40}
41
42#[derive(Debug, Deserialize)]
43#[serde(rename_all = "camelCase")]
44struct ExecResponse {
45 output: Vec<OutputFrame>,
46 exit_code: i64,
47}
48
49#[derive(Debug, Deserialize)]
50#[serde(rename_all = "camelCase")]
51struct ReadFileResponse {
52 contents_base64: String,
53}
54
55#[derive(Debug, Deserialize)]
56#[serde(rename_all = "camelCase")]
57struct PreviewResponse {
58 endpoint: String,
59 allowed_ports: Vec<u16>,
60}
61
62#[derive(Debug)]
64pub struct LocalSandbox {
65 client: reqwest::Client,
66 base_url: String,
67 token: String,
68}
69
70impl LocalSandbox {
71 pub async fn new(binding_name: &str, binding: &LocalSandboxBinding) -> Result<Self> {
76 let base_url = binding
77 .manager_url
78 .clone()
79 .into_value(binding_name, "managerUrl")
80 .map_err(|error| {
81 AlienError::new(ErrorData::BindingConfigInvalid {
82 binding_name: binding_name.to_string(),
83 env_var: alien_core::bindings::binding_env_var_name(binding_name),
84 reason: error.to_string(),
85 })
86 })?;
87
88 let token_path = binding
89 .token_path
90 .clone()
91 .into_value(binding_name, "tokenPath")
92 .map_err(|error| {
93 AlienError::new(ErrorData::BindingConfigInvalid {
94 binding_name: binding_name.to_string(),
95 env_var: alien_core::bindings::binding_env_var_name(binding_name),
96 reason: error.to_string(),
97 })
98 })?;
99
100 let token = tokio::fs::read_to_string(&token_path)
101 .await
102 .into_alien_error()
103 .context(ErrorData::BindingConfigInvalid {
104 binding_name: binding_name.to_string(),
105 env_var: alien_core::bindings::binding_env_var_name(binding_name),
106 reason: format!("could not read the sandbox route token at '{token_path}'"),
107 })?;
108
109 Ok(Self {
110 client: reqwest::Client::new(),
111 base_url: base_url.trim_end_matches('/').to_string(),
112 token: token.trim().to_string(),
113 })
114 }
115
116 fn url(&self, path: &str) -> String {
117 format!("{}{path}", self.base_url)
118 }
119
120 async fn send<T: for<'de> Deserialize<'de>>(
121 &self,
122 request: reqwest::RequestBuilder,
123 operation: &str,
124 ) -> Result<T> {
125 let response = self
126 .request(request, operation)
127 .await?
128 .json::<T>()
129 .await
130 .into_alien_error()
131 .context(ErrorData::UnexpectedResponseFormat {
132 provider: "local-sandbox".to_string(),
133 binding_name: operation.to_string(),
134 field: "body".to_string(),
135 response_json: "the sandbox route returned a body this provider cannot parse"
136 .to_string(),
137 })?;
138
139 Ok(response)
140 }
141
142 async fn request(
143 &self,
144 request: reqwest::RequestBuilder,
145 operation: &str,
146 ) -> Result<reqwest::Response> {
147 let response = request
148 .bearer_auth(&self.token)
149 .send()
150 .await
151 .into_alien_error()
152 .context(ErrorData::OperationNotSupported {
153 operation: operation.to_string(),
154 reason: "the local sandbox route is unreachable".to_string(),
155 })?;
156
157 if response.status().is_success() {
158 return Ok(response);
159 }
160
161 let status = response.status();
162 let body = response.text().await.unwrap_or_default();
165 Err(AlienError::new(ErrorData::OperationNotSupported {
166 operation: operation.to_string(),
167 reason: format!("the local sandbox route returned {status}: {body}"),
168 }))
169 }
170
171 fn unsupported(&self, capability: &str) -> AlienError<ErrorData> {
172 AlienError::new(ErrorData::OperationNotSupported {
173 operation: capability.to_string(),
174 reason: "not supported on local".to_string(),
175 })
176 }
177
178 async fn exec_within(
188 &self,
189 sandbox_id: &str,
190 command: &[String],
191 request: &RunCommandRequest,
192 ) -> Result<ExecResponse> {
193 match tokio::time::timeout(
194 guard_for(request.timeout)?,
195 self.send(
196 self.client
197 .post(self.url(&format!("/v1/sessions/{sandbox_id}/exec")))
198 .json(&json!({ "command": command })),
199 "sandbox.runCommand",
200 ),
201 )
202 .await
203 {
204 Ok(inner) => inner,
205 Err(_) => Err(match self.terminate(sandbox_id).await {
206 Ok(()) => AlienError::new(ErrorData::SandboxCommandFailed {
210 failure: "timeoutExceeded".to_string(),
211 reason: format!(
212 "the command exceeded its {}ms timeout and the sandbox could not end it, so the sandbox was terminated",
213 request.timeout.as_millis()
214 ),
215 }),
216 Err(error) => error.context(ErrorData::SandboxOutcomeUnknown {
221 operation: "sandbox.runCommand".to_string(),
222 reason: format!(
223 "the command exceeded its {}ms timeout and the sandbox could not end it",
224 request.timeout.as_millis()
225 ),
226 }),
227 }),
228 }
229 }
230}
231
232impl Binding for LocalSandbox {}
233
234fn refuse_create_time_lifetime(request: &CreateSandboxRequest) -> Result<()> {
237 if request.timeout_ms.is_none() {
238 return Ok(());
239 }
240 Err(AlienError::new(ErrorData::OperationNotSupported {
241 operation: SandboxCapability::SandboxLifetime.as_str().to_string(),
242 reason: "local sandboxes run as Docker containers, which carry no wall-clock ceiling; \
243 terminate the sandbox when the caller is done with it"
244 .to_string(),
245 }))
246}
247
248#[async_trait]
249impl Sandbox for LocalSandbox {
250 fn capabilities(&self) -> SandboxCapabilities {
251 SandboxCapabilities::for_platform(Platform::Local).expect("Local has a sandbox backend")
252 }
253
254 async fn create(&self, request: CreateSandboxRequest) -> Result<SandboxInstance> {
255 refuse_create_time_lifetime(&request)?;
256 let sandbox_id = request
257 .sandbox_id
258 .unwrap_or_else(|| uuid::Uuid::new_v4().simple().to_string());
259
260 let created: SessionBody = self
263 .send(
264 self.client
265 .post(self.url("/v1/sessions"))
266 .json(&json!({ "sessionId": sandbox_id })),
267 "sandbox.create",
268 )
269 .await?;
270
271 Ok(SandboxInstance {
272 sandbox_id: created.session_id,
273 state: SandboxState::Running,
274 generation: 1,
275 })
276 }
277
278 async fn get(&self, sandbox_id: &str) -> Result<Option<SandboxInstance>> {
279 Ok(self
280 .list()
281 .await?
282 .into_iter()
283 .find(|sandbox| sandbox.sandbox_id == sandbox_id))
284 }
285
286 async fn get_or_create(&self, request: CreateSandboxRequest) -> Result<ResolvedSandbox> {
287 refuse_create_time_lifetime(&request)?;
290 if let Some(id) = request.sandbox_id.as_deref() {
291 if let Some(existing) = self.get(id).await? {
292 return Ok(ResolvedSandbox::found(existing));
293 }
294 }
295
296 self.create(request).await.map(ResolvedSandbox::created)
297 }
298
299 async fn list(&self) -> Result<Vec<SandboxInstance>> {
300 let sessions: Vec<SessionBody> = self
301 .send(self.client.get(self.url("/v1/sessions")), "sandbox.list")
302 .await?;
303
304 Ok(sessions
305 .into_iter()
306 .map(|session| SandboxInstance {
307 sandbox_id: session.session_id,
308 state: SandboxState::Running,
309 generation: 1,
310 })
311 .collect())
312 }
313
314 async fn run_command(
315 &self,
316 sandbox_id: &str,
317 request: RunCommandRequest,
318 ) -> Result<BoxStream<'static, Result<CommandOutput>>> {
319 if request.timeout.is_zero() {
320 return Err(AlienError::new(ErrorData::OperationNotSupported {
321 operation: "sandbox.runCommand".to_string(),
322 reason: "a command must carry a non-zero timeout".to_string(),
323 }));
324 }
325
326 let mut argv = vec![
334 "sh".to_string(),
335 "-c".to_string(),
336 TimeoutReport::bounded_program(request.timeout),
337 "sh".to_string(),
338 ];
339 argv.extend(request.argv());
340 let response = self.exec_within(sandbox_id, &argv, &request).await?;
341
342 let mut stdout = Vec::new();
346 let mut stderr = String::new();
347 for frame in response.output {
348 let (data, is_stdout) = match &frame {
349 OutputFrame::Stdout(data) => (data, true),
350 OutputFrame::Stderr(data) => (data, false),
351 };
352 let decoded = BASE64.decode(data).into_alien_error().context(
353 ErrorData::UnexpectedResponseFormat {
354 provider: "local-sandbox".to_string(),
355 binding_name: "sandbox.runCommand".to_string(),
356 field: "output".to_string(),
357 response_json: "an output frame was not valid base64".to_string(),
358 },
359 )?;
360 if is_stdout {
361 stdout.extend_from_slice(&decoded);
362 } else {
363 stderr.push_str(&String::from_utf8_lossy(&decoded));
364 }
365 }
366 let (timeout_exceeded, stderr) =
367 match TimeoutReport::read(i32::try_from(response.exit_code).ok(), &stderr) {
368 Bounded::Ran { killed, stderr } => (killed, stderr),
369 Bounded::NotRun { reason } => {
370 return Err(AlienError::new(ErrorData::SandboxCommandFailed {
371 failure: "commandNotBounded".to_string(),
372 reason,
373 }))
374 }
375 };
376
377 let mut frames: Vec<Result<CommandOutput>> = Vec::new();
378 if !stdout.is_empty() {
379 frames.push(Ok(CommandOutput::Stdout {
380 seq: 0,
381 data: stdout,
382 }));
383 }
384 if !stderr.is_empty() {
385 frames.push(Ok(CommandOutput::Stderr {
386 seq: frames.len() as u64,
387 data: stderr.into_bytes(),
388 }));
389 }
390
391 if timeout_exceeded {
392 frames.push(Err(AlienError::new(ErrorData::SandboxCommandFailed {
395 failure: "timeoutExceeded".to_string(),
396 reason: format!(
397 "the command exceeded its {}ms timeout and was killed; the sandbox is still usable",
398 request.timeout.as_millis()
399 ),
400 })));
401 } else {
402 frames.push(Ok(CommandOutput::Exit {
403 code: response.exit_code as i32,
404 truncated: false,
405 }));
406 }
407
408 Ok(Box::pin(stream::iter(frames)))
409 }
410
411 async fn read_file(&self, sandbox_id: &str, path: &str) -> Result<Vec<u8>> {
412 let response: ReadFileResponse = self
413 .send(
414 self.client
415 .get(self.url(&format!("/v1/sessions/{sandbox_id}/files")))
416 .query(&[("path", path)]),
417 "sandbox.readFile",
418 )
419 .await?;
420
421 BASE64
422 .decode(response.contents_base64)
423 .into_alien_error()
424 .context(ErrorData::UnexpectedResponseFormat {
425 provider: "local-sandbox".to_string(),
426 binding_name: "sandbox.readFile".to_string(),
427 field: "contentsBase64".to_string(),
428 response_json: "file contents were not valid base64".to_string(),
429 })
430 }
431
432 async fn write_files(&self, sandbox_id: &str, files: BTreeMap<String, Vec<u8>>) -> Result<()> {
433 for (path, contents) in files {
434 self.request(
435 self.client
436 .put(self.url(&format!("/v1/sessions/{sandbox_id}/files")))
437 .json(&json!({
438 "path": path,
439 "contentsBase64": BASE64.encode(contents),
440 })),
441 "sandbox.writeFiles",
442 )
443 .await?;
444 }
445
446 Ok(())
447 }
448
449 async fn preview(&self, sandbox_id: &str, port: u16) -> Result<PreviewCapability> {
450 let response: PreviewResponse = self
451 .send(
452 self.client
453 .get(self.url(&format!("/v1/sessions/{sandbox_id}/preview")))
454 .query(&[("port", port.to_string())]),
455 "sandbox.preview",
456 )
457 .await?;
458
459 Ok(PreviewCapability {
460 endpoint: response.endpoint,
461 headers: BTreeMap::new(),
464 allowed_ports: response.allowed_ports,
465 expires_in_seconds: 0,
466 })
467 }
468
469 async fn pause(&self, _sandbox_id: &str) -> Result<()> {
470 Err(self.unsupported("pauseResume"))
471 }
472
473 async fn resume(&self, _sandbox_id: &str) -> Result<()> {
474 Err(self.unsupported("pauseResume"))
475 }
476
477 async fn snapshot(&self, _sandbox_id: &str) -> Result<String> {
478 Err(self.unsupported("snapshot"))
479 }
480
481 async fn start_job(&self, _sandbox_id: &str, _request: RunCommandRequest) -> Result<JobStart> {
482 Err(self.unsupported("jobs"))
483 }
484
485 async fn poll_job(
486 &self,
487 _sandbox_id: &str,
488 _job_id: &str,
489 _since_seq: Option<u64>,
490 ) -> Result<JobPoll> {
491 Err(self.unsupported("jobs"))
492 }
493
494 async fn cancel_job(&self, _sandbox_id: &str, _job_id: &str) -> Result<()> {
495 Err(self.unsupported("jobs"))
496 }
497
498 async fn terminate(&self, sandbox_id: &str) -> Result<()> {
499 self.request(
500 self.client
501 .delete(self.url(&format!("/v1/sessions/{sandbox_id}"))),
502 "sandbox.terminate",
503 )
504 .await?;
505
506 Ok(())
507 }
508
509 fn as_any(&self) -> &dyn std::any::Any {
510 self
511 }
512}
513
514#[cfg(test)]
515mod tests {
516 use super::*;
517 use axum::extract::{Path, State};
518 use axum::routing::post;
519 use axum::{Json, Router};
520 use futures::StreamExt as _;
521 use std::net::SocketAddr;
522 use std::sync::{Arc, Mutex};
523
524 #[derive(Default)]
527 struct Route {
528 execs: Mutex<std::collections::VecDeque<serde_json::Value>>,
529 commands: Mutex<Vec<Vec<String>>>,
530 deleted: Mutex<Vec<String>>,
531 delete_refuses: std::sync::atomic::AtomicBool,
534 }
535
536 const TIMEOUT_PLACEHOLDER: &str = "<timeout>";
538 const SANDBOX_NONCE: &str = "a1b2c3d4a1b2c3d4a1b2c3d4a1b2c3d4";
544
545 fn as_sandbox_stderr(stderr: &str) -> String {
547 match stderr {
548 TIMEOUT_PLACEHOLDER => format!("{SANDBOX_NONCE}\npartial-err{SANDBOX_NONCE}"),
549 other => format!("{SANDBOX_NONCE}\n{other}"),
550 }
551 }
552
553 fn exec_response(exit_code: i64, stdout: &str, stderr: &str) -> serde_json::Value {
554 let mut output = Vec::new();
555 if !stdout.is_empty() {
556 output.push(json!({ "stream": "stdout", "dataBase64": BASE64.encode(stdout) }));
557 }
558 output.push(json!({
559 "stream": "stderr",
560 "dataBase64": BASE64.encode(as_sandbox_stderr(stderr)),
561 }));
562 json!({ "output": output, "exitCode": exit_code })
563 }
564
565 async fn serve(route: Arc<Route>) -> LocalSandbox {
566 async fn exec(
567 State(route): State<Arc<Route>>,
568 Path(_session): Path<String>,
569 Json(body): Json<serde_json::Value>,
570 ) -> Json<serde_json::Value> {
571 let command: Vec<String> =
572 serde_json::from_value(body["command"].clone()).expect("argv");
573 route.commands.lock().expect("commands").push(command);
574 let next = route.execs.lock().expect("execs").pop_front();
575 match next {
576 Some(response) => Json(response),
577 None => std::future::pending().await,
578 }
579 }
580 async fn delete(
581 State(route): State<Arc<Route>>,
582 Path(session): Path<String>,
583 ) -> std::result::Result<Json<serde_json::Value>, axum::http::StatusCode> {
584 if route
585 .delete_refuses
586 .load(std::sync::atomic::Ordering::SeqCst)
587 {
588 return Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR);
589 }
590 route.deleted.lock().expect("deleted").push(session);
591 Ok(Json(json!({})))
592 }
593
594 let router = Router::new()
595 .route("/v1/sessions/{session}/exec", post(exec))
596 .route("/v1/sessions/{session}", axum::routing::delete(delete))
597 .with_state(route);
598 let listener = tokio::net::TcpListener::bind::<SocketAddr>("127.0.0.1:0".parse().unwrap())
599 .await
600 .expect("bind");
601 let address = listener.local_addr().expect("address");
602 tokio::spawn(async move {
603 axum::serve(listener, router).await.expect("serve");
604 });
605
606 LocalSandbox {
607 client: reqwest::Client::new(),
608 base_url: format!("http://{address}"),
609 token: "test-token".to_string(),
610 }
611 }
612
613 fn command(timeout_secs: u64) -> RunCommandRequest {
614 RunCommandRequest {
615 command: "python".to_string(),
616 args: vec!["-u".to_string(), "main.py".to_string()],
617 cwd: None,
618 env: BTreeMap::new(),
619 timeout: std::time::Duration::from_secs(timeout_secs),
620 }
621 }
622
623 #[tokio::test]
630 async fn a_command_past_its_timeout_is_killed_in_place_and_the_sandbox_survives() {
631 let route = Arc::new(Route::default());
632 route.execs.lock().expect("execs").push_back(exec_response(
633 137,
634 "partial\n",
635 TIMEOUT_PLACEHOLDER,
636 ));
637 let sandbox = serve(route.clone()).await;
638
639 let frames: Vec<Result<CommandOutput>> = sandbox
640 .run_command("s1", command(30))
641 .await
642 .expect("the deadline is reported in the stream")
643 .collect()
644 .await;
645
646 assert!(
647 matches!(&frames[0], Ok(CommandOutput::Stdout { data, .. }) if data == b"partial\n"),
648 "{frames:?}"
649 );
650 let terminal = frames
651 .last()
652 .expect("frames")
653 .as_ref()
654 .expect_err("the stream must end in the deadline error, not an exit frame");
655 assert!(
656 terminal.to_string().contains("timeoutExceeded"),
657 "{terminal}"
658 );
659 assert!(
660 route.deleted.lock().expect("deleted").is_empty(),
661 "the sandbox survives an in-sandbox kill"
662 );
663 let sent = route.commands.lock().expect("commands").clone();
664 assert_eq!(sent.len(), 1, "one command: {sent:?}");
665 assert_eq!(sent[0][0], "sh");
666 assert_eq!(sent[0][1], "-c");
667 assert!(sent[0][2].contains("sleep 30"), "{:?}", sent[0]);
668 assert_eq!(
669 &sent[0][3..],
670 &[
671 "sh".to_string(),
672 "python".to_string(),
673 "-u".to_string(),
674 "main.py".to_string()
675 ],
676 "the program leads its arguments, in order, each passed as one argument rather than \
677 pasted into the program text"
678 );
679 }
680
681 #[tokio::test]
685 async fn a_timeout_split_across_stderr_frames_is_still_read() {
686 let route = Arc::new(Route::default());
687 route.execs.lock().expect("execs").push_back(json!({
688 "output": [
689 { "stream": "stderr", "dataBase64": BASE64.encode(format!("{SANDBOX_NONCE}\n")) },
690 { "stream": "stderr", "dataBase64": BASE64.encode("partial-err") },
691 { "stream": "stderr", "dataBase64": BASE64.encode(SANDBOX_NONCE) },
692 ],
693 "exitCode": 137,
694 }));
695 let sandbox = serve(route.clone()).await;
696
697 let frames: Vec<Result<CommandOutput>> = sandbox
698 .run_command("s1", command(30))
699 .await
700 .expect("the deadline is reported in the stream")
701 .collect()
702 .await;
703
704 assert!(
705 matches!(&frames[0], Ok(CommandOutput::Stderr { data, .. }) if data == b"partial-err"),
706 "the caller keeps its output, without the protocol bytes: {frames:?}"
707 );
708 let terminal = frames
709 .last()
710 .expect("frames")
711 .as_ref()
712 .expect_err("a killed command ends in the deadline error");
713 assert!(
714 terminal.to_string().contains("timeoutExceeded"),
715 "{terminal}"
716 );
717 }
718
719 #[tokio::test]
722 async fn a_command_exiting_124_of_its_own_accord_is_an_exit_not_a_timeout() {
723 let route = Arc::new(Route::default());
724 route
725 .execs
726 .lock()
727 .expect("execs")
728 .push_back(exec_response(124, "done\n", ""));
729 let sandbox = serve(route.clone()).await;
730
731 let frames: Vec<Result<CommandOutput>> = sandbox
732 .run_command("s1", command(300))
733 .await
734 .expect("runs")
735 .collect()
736 .await;
737
738 assert!(matches!(
739 frames.last().expect("frames"),
740 Ok(CommandOutput::Exit { code: 124, .. })
741 ));
742 }
743
744 #[tokio::test(start_paused = true)]
748 async fn a_terminate_that_fails_does_not_hide_the_unknown_outcome() {
749 let route = Arc::new(Route::default());
750 route
751 .delete_refuses
752 .store(true, std::sync::atomic::Ordering::SeqCst);
753 let sandbox = serve(route.clone()).await;
754
755 let error = sandbox
756 .run_command("s1", command(30))
757 .await
758 .err()
759 .expect("a command that outran its deadline has not succeeded");
760
761 assert_eq!(error.code, "SANDBOX_OUTCOME_UNKNOWN", "{error}");
762 assert!(
763 error.to_string().contains("could not end it"),
764 "the deadline stays the headline: {error}"
765 );
766 }
767
768 #[tokio::test(start_paused = true)]
773 async fn a_command_the_sandbox_cannot_end_takes_the_sandbox_with_it() {
774 let route = Arc::new(Route::default());
775 let sandbox = serve(route.clone()).await;
776
777 let error = sandbox
778 .run_command("s1", command(30))
779 .await
780 .err()
781 .expect("a command that outran its deadline has not succeeded");
782
783 assert_eq!(error.code, "SANDBOX_COMMAND_FAILED", "{error}");
784 assert!(
785 error.to_string().contains("timeoutExceeded"),
786 "the terminal code has to name the timeout: {error}"
787 );
788 assert!(
789 !error.retryable,
790 "the command may have run before the kill landed: {error}"
791 );
792 assert_eq!(
793 route.deleted.lock().expect("deleted").clone(),
794 vec!["s1".to_string()],
795 "the sandbox must actually be removed, not merely reported as terminated"
796 );
797 }
798}