agentos_client/
language_execution.rs

1//! First-class JavaScript, TypeScript, Python, and shared execution lifecycle.
2
3use std::collections::HashMap;
4
5use agentos_sidecar_client::wire;
6use tokio::sync::{broadcast, watch};
7
8use crate::agent_os::AgentOs;
9use crate::agent_os::ProcessEntry;
10use crate::error::{ClientError, ClientResult};
11use crate::process::{ProcessOutput, ProcessRegistryReservation, ProcessStream};
12
13#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
14pub struct ContextDescriptor {
15    pub context_id: String,
16    pub state: String,
17    pub language: Option<String>,
18    pub created_at_ms: u64,
19    pub last_started_at_ms: Option<u64>,
20    pub last_completed_at_ms: Option<u64>,
21}
22
23#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
24pub struct ProcessDescriptor {
25    pub pid: u32,
26    pub state: String,
27    pub language: Option<String>,
28    pub started_at_ms: u64,
29}
30
31pub type CodeExecutionResult = wire::ExecutionCompletedResponse;
32pub type ExecutionOutputEvent = wire::ExecutionOutputEvent;
33pub type ExecutionCompletedEvent = wire::ExecutionCompletedEvent;
34pub type TypeScriptDiagnostic = wire::TypeScriptDiagnostic;
35
36#[derive(Debug, Clone, Default)]
37pub struct LanguageExecutionOptions {
38    pub context_id: Option<String>,
39    pub cwd: Option<String>,
40    pub env: HashMap<String, String>,
41    pub args: Vec<String>,
42    pub stdin: Option<Vec<u8>>,
43    pub timeout_ms: Option<u64>,
44    pub pty: Option<ExecutionPtyOptions>,
45    pub output: ExecutionOutputOptions,
46}
47
48#[derive(Debug, Clone, Default)]
49pub struct LanguageSpawnOptions {
50    pub cwd: Option<String>,
51    pub env: HashMap<String, String>,
52    pub args: Vec<String>,
53    pub stdin: Option<Vec<u8>>,
54    pub timeout_ms: Option<u64>,
55    pub pty: Option<ExecutionPtyOptions>,
56    pub retain_events: bool,
57}
58
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub enum OutputCapture {
61    #[default]
62    None,
63    Stderr,
64    All,
65}
66
67#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
68pub struct ExecutionOutputOptions {
69    pub capture: OutputCapture,
70    pub retain_events: bool,
71}
72
73#[derive(Debug, Clone, Copy, Default)]
74pub struct ExecutionPtyOptions {
75    pub cols: Option<u16>,
76    pub rows: Option<u16>,
77}
78
79#[derive(Debug, Clone, Default)]
80pub struct InlineExecutionOptions {
81    pub process: LanguageExecutionOptions,
82    pub inputs: Option<serde_json::Map<String, serde_json::Value>>,
83}
84
85#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
86pub enum JavaScriptModuleFormat {
87    /// Evaluate each call as an independent root ES module. A retained context
88    /// preserves `globalThis`, not the module's lexical scope.
89    #[default]
90    Module,
91    /// Use script/CommonJS semantics, including REPL-style top-level bindings
92    /// that remain visible to later calls in a retained context.
93    CommonJs,
94}
95
96#[derive(Debug, Clone, Default)]
97pub struct JavaScriptExecutionOptions {
98    pub inline: InlineExecutionOptions,
99    pub format: JavaScriptModuleFormat,
100    pub file_path: Option<String>,
101}
102
103#[derive(Debug, Clone, Default)]
104pub struct TypeScriptExecutionOptions {
105    pub inline: InlineExecutionOptions,
106    pub file_path: Option<String>,
107    pub tsconfig_path: Option<String>,
108    pub compiler_options: Option<serde_json::Map<String, serde_json::Value>>,
109}
110
111#[derive(Debug, Clone, Default)]
112pub struct TypeScriptCheckOptions {
113    pub context_id: Option<String>,
114    pub cwd: Option<String>,
115    pub file_path: Option<String>,
116    pub tsconfig_path: Option<String>,
117    pub compiler_options: Option<serde_json::Map<String, serde_json::Value>>,
118    pub timeout_ms: Option<u64>,
119    pub output: ExecutionOutputOptions,
120}
121
122#[derive(Debug, Clone, Default)]
123pub struct NpmProjectInstallOptions {
124    pub context_id: Option<String>,
125    pub cwd: Option<String>,
126    pub env: HashMap<String, String>,
127    pub timeout_ms: Option<u64>,
128    pub frozen: Option<bool>,
129    pub output: ExecutionOutputOptions,
130}
131
132#[derive(Debug, Clone, Default)]
133pub struct NpmPackageInstallOptions {
134    pub context_id: Option<String>,
135    pub cwd: Option<String>,
136    pub env: HashMap<String, String>,
137    pub timeout_ms: Option<u64>,
138    pub dev: Option<bool>,
139    pub global: Option<bool>,
140    pub output: ExecutionOutputOptions,
141}
142
143#[derive(Debug, Clone, Default)]
144pub struct PythonInstallOptions {
145    pub context_id: Option<String>,
146    pub cwd: Option<String>,
147    pub env: HashMap<String, String>,
148    pub timeout_ms: Option<u64>,
149    pub upgrade: Option<bool>,
150    pub requirements_file: Option<String>,
151    pub index_url: Option<String>,
152    pub extra_index_urls: Vec<String>,
153    pub output: ExecutionOutputOptions,
154}
155
156#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
157pub struct TypeScriptCheckResult {
158    pub result: CodeExecutionResult,
159    pub has_errors: Option<bool>,
160    pub diagnostics: Vec<TypeScriptDiagnostic>,
161}
162
163#[allow(clippy::large_enum_variant)] // This private enum avoids an allocation on the synchronous completion path.
164#[derive(Debug, Clone)]
165enum ExecutionSubmission {
166    Completed(CodeExecutionResult),
167    Background(wire::ExecutionDescriptor),
168}
169
170#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
171pub struct CodeEvaluationResult {
172    pub result: CodeExecutionResult,
173    pub value: Option<serde_json::Value>,
174}
175
176fn identity(options: &LanguageExecutionOptions) -> wire::ExecutionIdentityOptions {
177    wire::ExecutionIdentityOptions {
178        context_id: options.context_id.clone(),
179    }
180}
181
182fn process(options: &LanguageExecutionOptions) -> wire::ProcessExecutionOptions {
183    wire::ProcessExecutionOptions {
184        identity: identity(options),
185        output: output(options.output),
186        operation_id: None,
187        background: Some(false),
188        cwd: options.cwd.clone(),
189        env: (!options.env.is_empty()).then(|| options.env.clone()),
190        args: options.args.clone(),
191        stdin: options.stdin.clone(),
192        timeout_ms: options.timeout_ms,
193        pty: options.pty.map(|pty| wire::ExecutionPtyOptions {
194            cols: pty.cols,
195            rows: pty.rows,
196        }),
197    }
198}
199
200fn background_process(
201    options: &LanguageSpawnOptions,
202    operation_id: String,
203) -> wire::ProcessExecutionOptions {
204    wire::ProcessExecutionOptions {
205        identity: wire::ExecutionIdentityOptions { context_id: None },
206        output: wire::ExecutionOutputOptions {
207            capture: Some(wire::ExecutionOutputCapture::None),
208            retain_events: Some(options.retain_events),
209        },
210        operation_id: Some(operation_id),
211        background: Some(true),
212        cwd: options.cwd.clone(),
213        env: (!options.env.is_empty()).then(|| options.env.clone()),
214        args: options.args.clone(),
215        stdin: options.stdin.clone(),
216        timeout_ms: options.timeout_ms,
217        pty: options.pty.map(|pty| wire::ExecutionPtyOptions {
218            cols: pty.cols,
219            rows: pty.rows,
220        }),
221    }
222}
223
224fn output(options: ExecutionOutputOptions) -> wire::ExecutionOutputOptions {
225    wire::ExecutionOutputOptions {
226        capture: Some(match options.capture {
227            OutputCapture::None => wire::ExecutionOutputCapture::None,
228            OutputCapture::Stderr => wire::ExecutionOutputCapture::Stderr,
229            OutputCapture::All => wire::ExecutionOutputCapture::All,
230        }),
231        retain_events: Some(options.retain_events),
232    }
233}
234
235fn json_inputs(options: &InlineExecutionOptions) -> ClientResult<Option<String>> {
236    options
237        .inputs
238        .as_ref()
239        .map(serde_json::to_string)
240        .transpose()
241        .map_err(|error| ClientError::Sidecar(format!("failed to serialize inputs: {error}")))
242}
243
244fn context_descriptor(descriptor: wire::ExecutionDescriptor) -> ContextDescriptor {
245    ContextDescriptor {
246        context_id: descriptor.execution_id,
247        state: match descriptor.state {
248            wire::ExecutionState::Creating | wire::ExecutionState::Idle => "idle",
249            wire::ExecutionState::Running => "running",
250            wire::ExecutionState::Resetting => "resetting",
251            wire::ExecutionState::Deleting => "deleting",
252            wire::ExecutionState::Failed => "failed",
253        }
254        .to_owned(),
255        language: descriptor.retained_language.map(|language| match language {
256            wire::RetainedExecutionLanguage::JavaScript => String::from("javascript"),
257            wire::RetainedExecutionLanguage::Python => String::from("python"),
258        }),
259        created_at_ms: descriptor.created_at_ms,
260        last_started_at_ms: descriptor.last_started_at_ms,
261        last_completed_at_ms: descriptor.last_completed_at_ms,
262    }
263}
264
265fn execution_request_options(
266    payload: &wire::RequestPayload,
267) -> Option<(
268    &wire::ExecutionIdentityOptions,
269    &wire::ExecutionOutputOptions,
270)> {
271    Some(match payload {
272        wire::RequestPayload::ShellExecutionRequest(request) => {
273            (&request.process.identity, &request.process.output)
274        }
275        wire::RequestPayload::ArgvExecutionRequest(request) => {
276            (&request.process.identity, &request.process.output)
277        }
278        wire::RequestPayload::JavaScriptExecutionRequest(request) => {
279            (&request.process.identity, &request.process.output)
280        }
281        wire::RequestPayload::JavaScriptEvaluationRequest(request) => {
282            (&request.process.identity, &request.process.output)
283        }
284        wire::RequestPayload::JavaScriptFileExecutionRequest(request) => {
285            (&request.process.identity, &request.process.output)
286        }
287        wire::RequestPayload::TypeScriptExecutionRequest(request) => {
288            (&request.process.identity, &request.process.output)
289        }
290        wire::RequestPayload::TypeScriptEvaluationRequest(request) => {
291            (&request.process.identity, &request.process.output)
292        }
293        wire::RequestPayload::TypeScriptFileExecutionRequest(request) => {
294            (&request.process.identity, &request.process.output)
295        }
296        wire::RequestPayload::TypeScriptCheckRequest(request) => {
297            (&request.identity, &request.output)
298        }
299        wire::RequestPayload::TypeScriptProjectCheckRequest(request) => {
300            (&request.identity, &request.output)
301        }
302        wire::RequestPayload::NpmProjectInstallRequest(request) => {
303            (&request.identity, &request.output)
304        }
305        wire::RequestPayload::NpmPackageInstallRequest(request) => {
306            (&request.identity, &request.output)
307        }
308        wire::RequestPayload::NpmScriptExecutionRequest(request) => {
309            (&request.process.identity, &request.process.output)
310        }
311        wire::RequestPayload::NpmPackageExecutionRequest(request) => {
312            (&request.process.identity, &request.process.output)
313        }
314        wire::RequestPayload::PythonExecutionRequest(request) => {
315            (&request.process.identity, &request.process.output)
316        }
317        wire::RequestPayload::PythonEvaluationRequest(request) => {
318            (&request.process.identity, &request.process.output)
319        }
320        wire::RequestPayload::PythonFileExecutionRequest(request) => {
321            (&request.process.identity, &request.process.output)
322        }
323        wire::RequestPayload::PythonModuleExecutionRequest(request) => {
324            (&request.process.identity, &request.process.output)
325        }
326        wire::RequestPayload::PythonInstallRequest(request) => (&request.identity, &request.output),
327        _ => return None,
328    })
329}
330
331impl AgentOs {
332    fn execution_ownership(&self) -> wire::OwnershipScope {
333        let inner = self.inner();
334        wire::OwnershipScope::VmOwnership(wire::VmOwnership {
335            connection_id: inner.connection_id.clone(),
336            session_id: inner.session_id.clone(),
337            vm_id: inner.vm_id.clone(),
338        })
339    }
340
341    async fn submit_execution(
342        &self,
343        payload: wire::RequestPayload,
344        background: bool,
345    ) -> ClientResult<ExecutionSubmission> {
346        if let Some((identity, output)) = execution_request_options(&payload) {
347            if background && identity.context_id.is_some() {
348                return Err(ClientError::Sidecar(String::from(
349                    "spawned language processes cannot use context_id",
350                )));
351            }
352            if output.retain_events == Some(true) && identity.context_id.is_none() && !background {
353                return Err(ClientError::Sidecar(String::from(
354                    "retain_events requires a context or spawned language process",
355                )));
356            }
357        }
358        let mut events = self.transport().subscribe_wire_events();
359        let accepted = match self
360            .transport()
361            .request_wire(self.execution_ownership(), payload)
362            .await?
363        {
364            wire::ResponsePayload::ExecutionAcceptedResponse(response) => response,
365            wire::ResponsePayload::RejectedResponse(rejected) => {
366                return Err(ClientError::from_rejection(rejected))
367            }
368            response => {
369                return Err(ClientError::Sidecar(format!(
370                    "unexpected execution response: {response:?}"
371                )))
372            }
373        };
374        if background {
375            return Ok(ExecutionSubmission::Background(
376                accepted.execution.ok_or_else(|| {
377                    ClientError::Sidecar(String::from(
378                        "spawned process admission returned no descriptor",
379                    ))
380                })?,
381            ));
382        }
383        wait_for_completion_event(&mut events, &accepted.operation_id).await?;
384        Ok(ExecutionSubmission::Completed(
385            self.wait_execution(&accepted.operation_id).await?,
386        ))
387    }
388
389    pub async fn exec(
390        &self,
391        command: impl Into<String>,
392        options: LanguageExecutionOptions,
393    ) -> ClientResult<CodeExecutionResult> {
394        completed_submission(
395            self.submit_execution(
396                wire::RequestPayload::ShellExecutionRequest(wire::ShellExecutionRequest {
397                    process: process(&options),
398                    command: command.into(),
399                }),
400                false,
401            )
402            .await?,
403        )
404    }
405
406    pub async fn exec_argv(
407        &self,
408        command: impl Into<String>,
409        args: Vec<String>,
410        mut options: LanguageExecutionOptions,
411    ) -> ClientResult<CodeExecutionResult> {
412        options.args = args;
413        completed_submission(
414            self.submit_execution(
415                wire::RequestPayload::ArgvExecutionRequest(wire::ArgvExecutionRequest {
416                    process: process(&options),
417                    command: command.into(),
418                }),
419                false,
420            )
421            .await?,
422        )
423    }
424
425    pub async fn execute_javascript(
426        &self,
427        source: impl Into<String>,
428        options: JavaScriptExecutionOptions,
429    ) -> ClientResult<CodeExecutionResult> {
430        let inputs = json_inputs(&options.inline)?;
431        completed_submission(
432            self.submit_execution(
433                wire::RequestPayload::JavaScriptExecutionRequest(
434                    wire::JavaScriptExecutionRequest {
435                        process: process(&options.inline.process),
436                        source: source.into(),
437                        format: Some(match options.format {
438                            JavaScriptModuleFormat::Module => wire::JavaScriptModuleFormat::Module,
439                            JavaScriptModuleFormat::CommonJs => {
440                                wire::JavaScriptModuleFormat::CommonJs
441                            }
442                        }),
443                        file_path: options.file_path,
444                        inputs,
445                    },
446                ),
447                false,
448            )
449            .await?,
450        )
451    }
452
453    pub async fn evaluate_javascript(
454        &self,
455        expression: impl Into<String>,
456        options: JavaScriptExecutionOptions,
457    ) -> ClientResult<CodeEvaluationResult> {
458        let inputs = json_inputs(&options.inline)?;
459        let submission = self
460            .submit_execution(
461                wire::RequestPayload::JavaScriptEvaluationRequest(
462                    wire::JavaScriptEvaluationRequest {
463                        process: process(&options.inline.process),
464                        expression: expression.into(),
465                        format: Some(match options.format {
466                            JavaScriptModuleFormat::Module => wire::JavaScriptModuleFormat::Module,
467                            JavaScriptModuleFormat::CommonJs => {
468                                wire::JavaScriptModuleFormat::CommonJs
469                            }
470                        }),
471                        file_path: options.file_path,
472                        inputs,
473                    },
474                ),
475                false,
476            )
477            .await?;
478        evaluation_result(submission)
479    }
480
481    pub async fn execute_javascript_file(
482        &self,
483        path: impl Into<String>,
484        options: LanguageExecutionOptions,
485    ) -> ClientResult<CodeExecutionResult> {
486        completed_submission(
487            self.submit_execution(
488                wire::RequestPayload::JavaScriptFileExecutionRequest(
489                    wire::JavaScriptFileExecutionRequest {
490                        process: process(&options),
491                        path: path.into(),
492                    },
493                ),
494                false,
495            )
496            .await?,
497        )
498    }
499
500    pub async fn spawn_javascript(
501        &self,
502        source: impl Into<String>,
503        options: LanguageSpawnOptions,
504    ) -> ClientResult<ProcessDescriptor> {
505        let retain_events = options.retain_events;
506        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
507        let reservation = self.reserve_process_registry_slot()?;
508        background_submission(
509            self,
510            self.transport().subscribe_wire_events(),
511            self.submit_execution(
512                wire::RequestPayload::JavaScriptExecutionRequest(
513                    wire::JavaScriptExecutionRequest {
514                        process: background_process(&options, operation_id),
515                        source: source.into(),
516                        format: Some(wire::JavaScriptModuleFormat::Module),
517                        file_path: None,
518                        inputs: None,
519                    },
520                ),
521                true,
522            )
523            .await?,
524            "javascript",
525            retain_events,
526            reservation,
527        )
528    }
529
530    pub async fn spawn_javascript_file(
531        &self,
532        path: impl Into<String>,
533        options: LanguageSpawnOptions,
534    ) -> ClientResult<ProcessDescriptor> {
535        let retain_events = options.retain_events;
536        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
537        let reservation = self.reserve_process_registry_slot()?;
538        background_submission(
539            self,
540            self.transport().subscribe_wire_events(),
541            self.submit_execution(
542                wire::RequestPayload::JavaScriptFileExecutionRequest(
543                    wire::JavaScriptFileExecutionRequest {
544                        process: background_process(&options, operation_id),
545                        path: path.into(),
546                    },
547                ),
548                true,
549            )
550            .await?,
551            "javascript",
552            retain_events,
553            reservation,
554        )
555    }
556
557    pub async fn execute_typescript(
558        &self,
559        source: impl Into<String>,
560        options: TypeScriptExecutionOptions,
561    ) -> ClientResult<CodeExecutionResult> {
562        let inputs = json_inputs(&options.inline)?;
563        completed_submission(
564            self.submit_execution(
565                wire::RequestPayload::TypeScriptExecutionRequest(
566                    wire::TypeScriptExecutionRequest {
567                        process: process(&options.inline.process),
568                        source: source.into(),
569                        file_path: options.file_path,
570                        tsconfig_path: options.tsconfig_path,
571                        compiler_options: options
572                            .compiler_options
573                            .as_ref()
574                            .map(serde_json::to_string)
575                            .transpose()
576                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
577                        inputs,
578                    },
579                ),
580                false,
581            )
582            .await?,
583        )
584    }
585
586    pub async fn evaluate_typescript(
587        &self,
588        expression: impl Into<String>,
589        options: TypeScriptExecutionOptions,
590    ) -> ClientResult<CodeEvaluationResult> {
591        let inputs = json_inputs(&options.inline)?;
592        let submission = self
593            .submit_execution(
594                wire::RequestPayload::TypeScriptEvaluationRequest(
595                    wire::TypeScriptEvaluationRequest {
596                        process: process(&options.inline.process),
597                        expression: expression.into(),
598                        file_path: options.file_path,
599                        tsconfig_path: options.tsconfig_path,
600                        compiler_options: options
601                            .compiler_options
602                            .as_ref()
603                            .map(serde_json::to_string)
604                            .transpose()
605                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
606                        inputs,
607                    },
608                ),
609                false,
610            )
611            .await?;
612        evaluation_result(submission)
613    }
614
615    pub async fn execute_typescript_file(
616        &self,
617        path: impl Into<String>,
618        options: TypeScriptExecutionOptions,
619    ) -> ClientResult<CodeExecutionResult> {
620        completed_submission(
621            self.submit_execution(
622                wire::RequestPayload::TypeScriptFileExecutionRequest(
623                    wire::TypeScriptFileExecutionRequest {
624                        process: process(&options.inline.process),
625                        path: path.into(),
626                        tsconfig_path: options.tsconfig_path,
627                        compiler_options: options
628                            .compiler_options
629                            .as_ref()
630                            .map(serde_json::to_string)
631                            .transpose()
632                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
633                    },
634                ),
635                false,
636            )
637            .await?,
638        )
639    }
640
641    pub async fn spawn_typescript(
642        &self,
643        source: impl Into<String>,
644        options: LanguageSpawnOptions,
645    ) -> ClientResult<ProcessDescriptor> {
646        let retain_events = options.retain_events;
647        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
648        let reservation = self.reserve_process_registry_slot()?;
649        background_submission(
650            self,
651            self.transport().subscribe_wire_events(),
652            self.submit_execution(
653                wire::RequestPayload::TypeScriptExecutionRequest(
654                    wire::TypeScriptExecutionRequest {
655                        process: background_process(&options, operation_id),
656                        source: source.into(),
657                        file_path: None,
658                        tsconfig_path: None,
659                        compiler_options: None,
660                        inputs: None,
661                    },
662                ),
663                true,
664            )
665            .await?,
666            "javascript",
667            retain_events,
668            reservation,
669        )
670    }
671
672    pub async fn spawn_typescript_file(
673        &self,
674        path: impl Into<String>,
675        options: LanguageSpawnOptions,
676    ) -> ClientResult<ProcessDescriptor> {
677        let retain_events = options.retain_events;
678        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
679        let reservation = self.reserve_process_registry_slot()?;
680        background_submission(
681            self,
682            self.transport().subscribe_wire_events(),
683            self.submit_execution(
684                wire::RequestPayload::TypeScriptFileExecutionRequest(
685                    wire::TypeScriptFileExecutionRequest {
686                        process: background_process(&options, operation_id),
687                        path: path.into(),
688                        tsconfig_path: None,
689                        compiler_options: None,
690                    },
691                ),
692                true,
693            )
694            .await?,
695            "javascript",
696            retain_events,
697            reservation,
698        )
699    }
700
701    pub async fn check_typescript(
702        &self,
703        source: impl Into<String>,
704        options: TypeScriptCheckOptions,
705    ) -> ClientResult<TypeScriptCheckResult> {
706        let compiler_options = options
707            .compiler_options
708            .as_ref()
709            .map(serde_json::to_string)
710            .transpose()
711            .map_err(|error| ClientError::Sidecar(error.to_string()))?;
712        let submission = self
713            .submit_execution(
714                wire::RequestPayload::TypeScriptCheckRequest(wire::TypeScriptCheckRequest {
715                    identity: wire::ExecutionIdentityOptions {
716                        context_id: options.context_id,
717                    },
718                    output: output(options.output),
719                    source: source.into(),
720                    cwd: options.cwd,
721                    file_path: options.file_path,
722                    tsconfig_path: options.tsconfig_path,
723                    compiler_options,
724                    timeout_ms: options.timeout_ms,
725                }),
726                false,
727            )
728            .await?;
729        let result = completed_submission(submission)?;
730        typescript_check_result(result)
731    }
732
733    pub async fn check_typescript_project(
734        &self,
735        options: TypeScriptCheckOptions,
736    ) -> ClientResult<TypeScriptCheckResult> {
737        let submission = self
738            .submit_execution(
739                wire::RequestPayload::TypeScriptProjectCheckRequest(
740                    wire::TypeScriptProjectCheckRequest {
741                        identity: wire::ExecutionIdentityOptions {
742                            context_id: options.context_id,
743                        },
744                        output: output(options.output),
745                        cwd: options.cwd,
746                        tsconfig_path: options.tsconfig_path,
747                        timeout_ms: options.timeout_ms,
748                    },
749                ),
750                false,
751            )
752            .await?;
753        let result = completed_submission(submission)?;
754        typescript_check_result(result)
755    }
756
757    pub async fn install_npm_project(
758        &self,
759        options: NpmProjectInstallOptions,
760    ) -> ClientResult<CodeExecutionResult> {
761        completed_submission(
762            self.submit_execution(
763                wire::RequestPayload::NpmProjectInstallRequest(wire::NpmProjectInstallRequest {
764                    identity: wire::ExecutionIdentityOptions {
765                        context_id: options.context_id,
766                    },
767                    output: output(options.output),
768                    cwd: options.cwd,
769                    env: (!options.env.is_empty()).then_some(options.env),
770                    timeout_ms: options.timeout_ms,
771                    frozen: options.frozen,
772                }),
773                false,
774            )
775            .await?,
776        )
777    }
778
779    pub async fn install_npm_packages(
780        &self,
781        packages: Vec<String>,
782        options: NpmPackageInstallOptions,
783    ) -> ClientResult<CodeExecutionResult> {
784        completed_submission(
785            self.submit_execution(
786                wire::RequestPayload::NpmPackageInstallRequest(wire::NpmPackageInstallRequest {
787                    identity: wire::ExecutionIdentityOptions {
788                        context_id: options.context_id,
789                    },
790                    output: output(options.output),
791                    cwd: options.cwd,
792                    env: (!options.env.is_empty()).then_some(options.env),
793                    timeout_ms: options.timeout_ms,
794                    packages,
795                    dev: options.dev,
796                    global: options.global,
797                }),
798                false,
799            )
800            .await?,
801        )
802    }
803
804    pub async fn execute_npm_script(
805        &self,
806        script: impl Into<String>,
807        options: LanguageExecutionOptions,
808    ) -> ClientResult<CodeExecutionResult> {
809        completed_submission(
810            self.submit_execution(
811                wire::RequestPayload::NpmScriptExecutionRequest(wire::NpmScriptExecutionRequest {
812                    process: process(&options),
813                    script: script.into(),
814                }),
815                false,
816            )
817            .await?,
818        )
819    }
820
821    pub async fn execute_npm_package(
822        &self,
823        package_spec: impl Into<String>,
824        binary: Option<String>,
825        options: LanguageExecutionOptions,
826    ) -> ClientResult<CodeExecutionResult> {
827        completed_submission(
828            self.submit_execution(
829                wire::RequestPayload::NpmPackageExecutionRequest(
830                    wire::NpmPackageExecutionRequest {
831                        process: process(&options),
832                        package_spec: package_spec.into(),
833                        binary,
834                    },
835                ),
836                false,
837            )
838            .await?,
839        )
840    }
841
842    pub async fn execute_python(
843        &self,
844        source: impl Into<String>,
845        options: InlineExecutionOptions,
846    ) -> ClientResult<CodeExecutionResult> {
847        let inputs = json_inputs(&options)?;
848        completed_submission(
849            self.submit_execution(
850                wire::RequestPayload::PythonExecutionRequest(wire::PythonExecutionRequest {
851                    process: process(&options.process),
852                    source: source.into(),
853                    inputs,
854                }),
855                false,
856            )
857            .await?,
858        )
859    }
860
861    pub async fn evaluate_python(
862        &self,
863        expression: impl Into<String>,
864        options: InlineExecutionOptions,
865    ) -> ClientResult<CodeEvaluationResult> {
866        let inputs = json_inputs(&options)?;
867        let submission = self
868            .submit_execution(
869                wire::RequestPayload::PythonEvaluationRequest(wire::PythonEvaluationRequest {
870                    process: process(&options.process),
871                    expression: expression.into(),
872                    inputs,
873                }),
874                false,
875            )
876            .await?;
877        evaluation_result(submission)
878    }
879
880    pub async fn execute_python_file(
881        &self,
882        path: impl Into<String>,
883        options: LanguageExecutionOptions,
884    ) -> ClientResult<CodeExecutionResult> {
885        completed_submission(
886            self.submit_execution(
887                wire::RequestPayload::PythonFileExecutionRequest(
888                    wire::PythonFileExecutionRequest {
889                        process: process(&options),
890                        path: path.into(),
891                    },
892                ),
893                false,
894            )
895            .await?,
896        )
897    }
898
899    pub async fn execute_python_module(
900        &self,
901        module: impl Into<String>,
902        options: LanguageExecutionOptions,
903    ) -> ClientResult<CodeExecutionResult> {
904        completed_submission(
905            self.submit_execution(
906                wire::RequestPayload::PythonModuleExecutionRequest(
907                    wire::PythonModuleExecutionRequest {
908                        process: process(&options),
909                        module: module.into(),
910                    },
911                ),
912                false,
913            )
914            .await?,
915        )
916    }
917
918    pub async fn spawn_python(
919        &self,
920        source: impl Into<String>,
921        options: LanguageSpawnOptions,
922    ) -> ClientResult<ProcessDescriptor> {
923        let retain_events = options.retain_events;
924        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
925        let reservation = self.reserve_process_registry_slot()?;
926        background_submission(
927            self,
928            self.transport().subscribe_wire_events(),
929            self.submit_execution(
930                wire::RequestPayload::PythonExecutionRequest(wire::PythonExecutionRequest {
931                    process: background_process(&options, operation_id),
932                    source: source.into(),
933                    inputs: None,
934                }),
935                true,
936            )
937            .await?,
938            "python",
939            retain_events,
940            reservation,
941        )
942    }
943
944    pub async fn spawn_python_file(
945        &self,
946        path: impl Into<String>,
947        options: LanguageSpawnOptions,
948    ) -> ClientResult<ProcessDescriptor> {
949        let retain_events = options.retain_events;
950        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
951        let reservation = self.reserve_process_registry_slot()?;
952        background_submission(
953            self,
954            self.transport().subscribe_wire_events(),
955            self.submit_execution(
956                wire::RequestPayload::PythonFileExecutionRequest(
957                    wire::PythonFileExecutionRequest {
958                        process: background_process(&options, operation_id),
959                        path: path.into(),
960                    },
961                ),
962                true,
963            )
964            .await?,
965            "python",
966            retain_events,
967            reservation,
968        )
969    }
970
971    pub async fn spawn_python_module(
972        &self,
973        module: impl Into<String>,
974        options: LanguageSpawnOptions,
975    ) -> ClientResult<ProcessDescriptor> {
976        let retain_events = options.retain_events;
977        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
978        let reservation = self.reserve_process_registry_slot()?;
979        background_submission(
980            self,
981            self.transport().subscribe_wire_events(),
982            self.submit_execution(
983                wire::RequestPayload::PythonModuleExecutionRequest(
984                    wire::PythonModuleExecutionRequest {
985                        process: background_process(&options, operation_id),
986                        module: module.into(),
987                    },
988                ),
989                true,
990            )
991            .await?,
992            "python",
993            retain_events,
994            reservation,
995        )
996    }
997
998    pub async fn install_python_packages(
999        &self,
1000        packages: Vec<String>,
1001        options: PythonInstallOptions,
1002    ) -> ClientResult<CodeExecutionResult> {
1003        if !packages.is_empty() && options.requirements_file.is_some() {
1004            return Err(ClientError::Sidecar(String::from(
1005                "install_python_packages cannot combine packages with requirements_file",
1006            )));
1007        }
1008        completed_submission(
1009            self.submit_execution(
1010                wire::RequestPayload::PythonInstallRequest(wire::PythonInstallRequest {
1011                    identity: wire::ExecutionIdentityOptions {
1012                        context_id: options.context_id,
1013                    },
1014                    output: output(options.output),
1015                    cwd: options.cwd,
1016                    env: (!options.env.is_empty()).then_some(options.env),
1017                    timeout_ms: options.timeout_ms,
1018                    packages,
1019                    upgrade: options.upgrade,
1020                    requirements_file: options.requirements_file,
1021                    index_url: options.index_url,
1022                    extra_index_urls: options.extra_index_urls,
1023                }),
1024                false,
1025            )
1026            .await?,
1027        )
1028    }
1029
1030    pub async fn create_context(&self, context_id: &str) -> ClientResult<ContextDescriptor> {
1031        match self
1032            .transport()
1033            .request_wire(
1034                self.execution_ownership(),
1035                wire::RequestPayload::CreateContextRequest(wire::CreateContextRequest {
1036                    context_id: context_id.to_owned(),
1037                }),
1038            )
1039            .await?
1040        {
1041            wire::ResponsePayload::ExecutionDescriptorResponse(response) => {
1042                Ok(context_descriptor(response.execution))
1043            }
1044            wire::ResponsePayload::RejectedResponse(rejected) => {
1045                Err(ClientError::from_rejection(rejected))
1046            }
1047            response => Err(ClientError::Sidecar(format!(
1048                "unexpected create_context response: {response:?}"
1049            ))),
1050        }
1051    }
1052
1053    pub async fn get_context(&self, context_id: &str) -> ClientResult<ContextDescriptor> {
1054        match self
1055            .transport()
1056            .request_wire(
1057                self.execution_ownership(),
1058                wire::RequestPayload::GetExecutionRequest(wire::GetExecutionRequest {
1059                    execution_id: context_id.to_owned(),
1060                }),
1061            )
1062            .await?
1063        {
1064            wire::ResponsePayload::ExecutionDescriptorResponse(response) => {
1065                Ok(context_descriptor(response.execution))
1066            }
1067            wire::ResponsePayload::RejectedResponse(rejected) => {
1068                Err(ClientError::from_rejection(rejected))
1069            }
1070            response => Err(ClientError::Sidecar(format!(
1071                "unexpected get_context response: {response:?}"
1072            ))),
1073        }
1074    }
1075
1076    pub async fn list_contexts(&self) -> ClientResult<Vec<ContextDescriptor>> {
1077        match self
1078            .transport()
1079            .request_wire(
1080                self.execution_ownership(),
1081                wire::RequestPayload::ListExecutionsRequest,
1082            )
1083            .await?
1084        {
1085            wire::ResponsePayload::ExecutionListResponse(response) => Ok(response
1086                .executions
1087                .into_iter()
1088                .map(context_descriptor)
1089                .collect()),
1090            wire::ResponsePayload::RejectedResponse(rejected) => {
1091                Err(ClientError::from_rejection(rejected))
1092            }
1093            response => Err(ClientError::Sidecar(format!(
1094                "unexpected list_contexts response: {response:?}"
1095            ))),
1096        }
1097    }
1098
1099    async fn wait_execution(&self, execution_id: &str) -> ClientResult<CodeExecutionResult> {
1100        let mut events = self.transport().subscribe_wire_events();
1101        let first = self
1102            .transport()
1103            .request_wire(
1104                self.execution_ownership(),
1105                wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest {
1106                    execution_id: execution_id.to_owned(),
1107                }),
1108            )
1109            .await?;
1110        let response = match first {
1111            wire::ResponsePayload::RejectedResponse(rejected)
1112                if rejected.code == "execution_busy" =>
1113            {
1114                wait_for_completion_event(&mut events, execution_id).await?;
1115                self.transport()
1116                    .request_wire(
1117                        self.execution_ownership(),
1118                        wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest {
1119                            execution_id: execution_id.to_owned(),
1120                        }),
1121                    )
1122                    .await?
1123            }
1124            response => response,
1125        };
1126        match response {
1127            wire::ResponsePayload::ExecutionCompletedResponse(response) => Ok(response),
1128            wire::ResponsePayload::RejectedResponse(rejected) => {
1129                Err(ClientError::from_rejection(rejected))
1130            }
1131            response => Err(ClientError::Sidecar(format!(
1132                "unexpected wait_execution response: {response:?}"
1133            ))),
1134        }
1135    }
1136
1137    pub async fn reset_context(&self, context_id: &str) -> ClientResult<()> {
1138        self.execution_descriptor_request(wire::RequestPayload::ResetExecutionRequest(
1139            wire::ResetExecutionRequest {
1140                execution_id: context_id.to_owned(),
1141            },
1142        ))
1143        .await?;
1144        Ok(())
1145    }
1146
1147    async fn execution_descriptor_request(
1148        &self,
1149        payload: wire::RequestPayload,
1150    ) -> ClientResult<wire::ExecutionDescriptor> {
1151        match self
1152            .transport()
1153            .request_wire(self.execution_ownership(), payload)
1154            .await?
1155        {
1156            wire::ResponsePayload::ExecutionDescriptorResponse(response) => Ok(response.execution),
1157            wire::ResponsePayload::RejectedResponse(rejected) => {
1158                Err(ClientError::from_rejection(rejected))
1159            }
1160            response => Err(ClientError::Sidecar(format!(
1161                "unexpected execution lifecycle response: {response:?}"
1162            ))),
1163        }
1164    }
1165
1166    pub async fn delete_context(&self, context_id: &str) -> ClientResult<()> {
1167        match self
1168            .transport()
1169            .request_wire(
1170                self.execution_ownership(),
1171                wire::RequestPayload::DeleteExecutionRequest(wire::DeleteExecutionRequest {
1172                    execution_id: context_id.to_owned(),
1173                }),
1174            )
1175            .await?
1176        {
1177            wire::ResponsePayload::ExecutionDeletedResponse(_) => Ok(()),
1178            wire::ResponsePayload::RejectedResponse(rejected) => {
1179                Err(ClientError::from_rejection(rejected))
1180            }
1181            response => Err(ClientError::Sidecar(format!(
1182                "unexpected delete_context response: {response:?}"
1183            ))),
1184        }
1185    }
1186}
1187
1188async fn wait_for_completion_event(
1189    events: &mut broadcast::Receiver<(wire::OwnershipScope, wire::EventPayload)>,
1190    execution_id: &str,
1191) -> ClientResult<()> {
1192    loop {
1193        match events.recv().await {
1194            Ok((_, wire::EventPayload::ExecutionCompletedEvent(event)))
1195                if event.execution_id == execution_id =>
1196            {
1197                return Ok(())
1198            }
1199            Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}
1200            Err(broadcast::error::RecvError::Closed) => {
1201                return Err(crate::process::sidecar_exited_before_exit(execution_id))
1202            }
1203        }
1204    }
1205}
1206
1207fn evaluation_result(submission: ExecutionSubmission) -> ClientResult<CodeEvaluationResult> {
1208    let ExecutionSubmission::Completed(result) = submission else {
1209        return Err(ClientError::Sidecar(String::from(
1210            "evaluation unexpectedly returned a background process",
1211        )));
1212    };
1213    let value = result
1214        .evaluation_value
1215        .as_deref()
1216        .map(serde_json::from_str::<serde_json::Value>)
1217        .transpose()
1218        .map_err(|error| {
1219            ClientError::Sidecar(format!("failed to decode evaluation result: {error}"))
1220        })?;
1221    Ok(CodeEvaluationResult { result, value })
1222}
1223
1224fn typescript_check_result(result: CodeExecutionResult) -> ClientResult<TypeScriptCheckResult> {
1225    if result.outcome != wire::ExecutionOutcome::Succeeded {
1226        return Ok(TypeScriptCheckResult {
1227            result,
1228            has_errors: None,
1229            diagnostics: Vec::new(),
1230        });
1231    }
1232    let data: serde_json::Value = result
1233        .type_script_check_result
1234        .as_deref()
1235        .map(serde_json::from_str)
1236        .transpose()
1237        .map_err(|error| {
1238            ClientError::Sidecar(format!(
1239                "failed to decode TypeScript checker result: {error}"
1240            ))
1241        })?
1242        .ok_or_else(|| {
1243            ClientError::Sidecar(String::from(
1244                "TypeScript checker returned no diagnostic result",
1245            ))
1246        })?;
1247    let data = data.as_object().ok_or_else(|| {
1248        ClientError::Sidecar(String::from(
1249            "TypeScript checker returned an invalid diagnostic result",
1250        ))
1251    })?;
1252    let has_errors = data
1253        .get("hasErrors")
1254        .and_then(serde_json::Value::as_bool)
1255        .ok_or_else(|| {
1256            ClientError::Sidecar(String::from(
1257                "TypeScript checker returned an invalid hasErrors value",
1258            ))
1259        })?;
1260    let diagnostics = data
1261        .get("diagnostics")
1262        .and_then(serde_json::Value::as_array)
1263        .ok_or_else(|| {
1264            ClientError::Sidecar(String::from(
1265                "TypeScript checker returned invalid diagnostics",
1266            ))
1267        })?
1268        .iter()
1269        .map(|diagnostic| {
1270            let code = diagnostic
1271                .get("code")
1272                .and_then(serde_json::Value::as_u64)
1273                .and_then(|code| u32::try_from(code).ok())
1274                .ok_or_else(|| {
1275                    ClientError::Sidecar(String::from(
1276                        "TypeScript checker returned an invalid diagnostic code",
1277                    ))
1278                })?;
1279            let category = diagnostic
1280                .get("category")
1281                .and_then(serde_json::Value::as_str)
1282                .ok_or_else(|| {
1283                    ClientError::Sidecar(String::from(
1284                        "TypeScript checker returned an invalid diagnostic category",
1285                    ))
1286                })?
1287                .to_owned();
1288            let message = diagnostic
1289                .get("message")
1290                .and_then(serde_json::Value::as_str)
1291                .ok_or_else(|| {
1292                    ClientError::Sidecar(String::from(
1293                        "TypeScript checker returned an invalid diagnostic message",
1294                    ))
1295                })?
1296                .to_owned();
1297            Ok(wire::TypeScriptDiagnostic {
1298                code,
1299                category,
1300                message,
1301                file_path: diagnostic
1302                    .get("filePath")
1303                    .and_then(serde_json::Value::as_str)
1304                    .map(str::to_owned),
1305                line: diagnostic
1306                    .get("line")
1307                    .and_then(serde_json::Value::as_u64)
1308                    .and_then(|line| u32::try_from(line).ok()),
1309                column: diagnostic
1310                    .get("column")
1311                    .and_then(serde_json::Value::as_u64)
1312                    .and_then(|column| u32::try_from(column).ok()),
1313            })
1314        })
1315        .collect::<ClientResult<Vec<_>>>()?;
1316    Ok(TypeScriptCheckResult {
1317        result,
1318        has_errors: Some(has_errors),
1319        diagnostics,
1320    })
1321}
1322
1323fn completed_submission(submission: ExecutionSubmission) -> ClientResult<CodeExecutionResult> {
1324    match submission {
1325        ExecutionSubmission::Completed(result) => Ok(result),
1326        ExecutionSubmission::Background(_) => Err(ClientError::Sidecar(String::from(
1327            "attached operation unexpectedly returned a background process",
1328        ))),
1329    }
1330}
1331
1332// Callers create the receiver before awaiting submission. Output and completion
1333// can arrive before the admission response, including for an immediately exiting
1334// script; subscribing here would permanently lose those events.
1335fn background_submission(
1336    client: &AgentOs,
1337    mut events: broadcast::Receiver<(wire::OwnershipScope, wire::EventPayload)>,
1338    submission: ExecutionSubmission,
1339    language: &str,
1340    retain_events: bool,
1341    reservation: ProcessRegistryReservation,
1342) -> ClientResult<ProcessDescriptor> {
1343    match submission {
1344        ExecutionSubmission::Background(descriptor) => {
1345            let pid = descriptor.pid.ok_or_else(|| {
1346                ClientError::Sidecar(String::from(
1347                    "spawned process admission returned no numeric pid",
1348                ))
1349            })?;
1350            let process_id = descriptor.process_id.clone().ok_or_else(|| {
1351                ClientError::Sidecar(String::from(
1352                    "spawned process admission returned no process routing id",
1353                ))
1354            })?;
1355            let (stdout_tx, _) = broadcast::channel::<Vec<u8>>(1024);
1356            let (stderr_tx, _) = broadcast::channel::<Vec<u8>>(1024);
1357            let (output_tx, _) = broadcast::channel::<ProcessOutput>(1024);
1358            let (exit_tx, _) = watch::channel(crate::process::ProcessOutcome::Pending);
1359            let (kernel_pid_tx, _) = watch::channel(Some(pid));
1360            let operation_id = descriptor.execution_id.clone();
1361            let entry = ProcessEntry {
1362                command: format!("{language} source"),
1363                args: Vec::new(),
1364                stdout_tx: stdout_tx.clone(),
1365                stderr_tx: stderr_tx.clone(),
1366                output_tx: output_tx.clone(),
1367                exit_tx: exit_tx.clone(),
1368                process_id: process_id.clone(),
1369                kernel_pid: kernel_pid_tx,
1370                output_tasks: Vec::new(),
1371                retain_output: retain_events,
1372                execution_id: Some(operation_id.clone()),
1373                execution_generation: Some(descriptor.generation),
1374                started_at: descriptor.created_at_ms as i64,
1375            };
1376            reservation.commit(pid, entry)?;
1377            let ownership = client.vm_scope();
1378            tokio::spawn(async move {
1379                loop {
1380                    let event = match events.recv().await {
1381                        Ok((scope, event)) if scope == ownership => event,
1382                        Ok(_) => continue,
1383                        Err(error) => {
1384                            let reason =
1385                                format!("background execution event stream failed: {error}");
1386                            tracing::error!(%operation_id, %reason, "background exit observation failed");
1387                            let closed = matches!(error, broadcast::error::RecvError::Closed);
1388                            let error = if closed {
1389                                crate::process::sidecar_exited_before_exit(&process_id)
1390                            } else {
1391                                ClientError::TerminationFailed {
1392                                    process_id: process_id.clone(),
1393                                    reason,
1394                                }
1395                            };
1396                            exit_tx.send_replace(crate::process::ProcessOutcome::Failed {
1397                                error,
1398                                rejected: false,
1399                            });
1400                            if closed {
1401                                break;
1402                            }
1403                            continue;
1404                        }
1405                    };
1406                    match event {
1407                        wire::EventPayload::ExecutionOutputEvent(output)
1408                            if output.execution_id == operation_id =>
1409                        {
1410                            let (stream, tx) = match output.channel {
1411                                wire::ExecutionStreamChannel::Stdout => {
1412                                    (ProcessStream::Stdout, &stdout_tx)
1413                                }
1414                                wire::ExecutionStreamChannel::Stderr
1415                                | wire::ExecutionStreamChannel::Pty => {
1416                                    (ProcessStream::Stderr, &stderr_tx)
1417                                }
1418                            };
1419                            let _ = tx.send(output.chunk.clone());
1420                            let _ = output_tx.send(ProcessOutput {
1421                                pid,
1422                                stream,
1423                                data: output.chunk,
1424                                sequence: Some(output.sequence),
1425                                timestamp_ms: Some(output.timestamp_ms.min(i64::MAX as u64) as i64),
1426                            });
1427                        }
1428                        wire::EventPayload::ExecutionCompletedEvent(completed)
1429                            if completed.execution_id == operation_id =>
1430                        {
1431                            if completed.exit_code.is_none() {
1432                                tracing::error!(%operation_id, "background completion has no exit status");
1433                            }
1434                            exit_tx.send_replace(crate::process::ProcessOutcome::completion(
1435                                &process_id,
1436                                completed.exit_code,
1437                            ));
1438                            break;
1439                        }
1440                        _ => {}
1441                    }
1442                }
1443            });
1444            Ok(ProcessDescriptor {
1445                pid,
1446                state: String::from("running"),
1447                language: Some(language.to_owned()),
1448                started_at_ms: descriptor.created_at_ms,
1449            })
1450        }
1451        ExecutionSubmission::Completed(_) => Err(ClientError::Sidecar(String::from(
1452            "spawned process unexpectedly returned an attached result",
1453        ))),
1454    }
1455}
agentos_client/language_execution.rs

agentos_client/
language_execution.rs
