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, 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)]
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)]
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 operation_id = format!("process-{}", uuid::Uuid::new_v4());
506        background_submission(
507            self,
508            self.submit_execution(
509                wire::RequestPayload::JavaScriptExecutionRequest(
510                    wire::JavaScriptExecutionRequest {
511                        process: background_process(&options, operation_id),
512                        source: source.into(),
513                        format: Some(wire::JavaScriptModuleFormat::Module),
514                        file_path: None,
515                        inputs: None,
516                    },
517                ),
518                true,
519            )
520            .await?,
521            "javascript",
522        )
523    }
524
525    pub async fn spawn_javascript_file(
526        &self,
527        path: impl Into<String>,
528        options: LanguageSpawnOptions,
529    ) -> ClientResult<ProcessDescriptor> {
530        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
531        background_submission(
532            self,
533            self.submit_execution(
534                wire::RequestPayload::JavaScriptFileExecutionRequest(
535                    wire::JavaScriptFileExecutionRequest {
536                        process: background_process(&options, operation_id),
537                        path: path.into(),
538                    },
539                ),
540                true,
541            )
542            .await?,
543            "javascript",
544        )
545    }
546
547    pub async fn execute_typescript(
548        &self,
549        source: impl Into<String>,
550        options: TypeScriptExecutionOptions,
551    ) -> ClientResult<CodeExecutionResult> {
552        let inputs = json_inputs(&options.inline)?;
553        completed_submission(
554            self.submit_execution(
555                wire::RequestPayload::TypeScriptExecutionRequest(
556                    wire::TypeScriptExecutionRequest {
557                        process: process(&options.inline.process),
558                        source: source.into(),
559                        file_path: options.file_path,
560                        tsconfig_path: options.tsconfig_path,
561                        compiler_options: options
562                            .compiler_options
563                            .as_ref()
564                            .map(serde_json::to_string)
565                            .transpose()
566                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
567                        inputs,
568                    },
569                ),
570                false,
571            )
572            .await?,
573        )
574    }
575
576    pub async fn evaluate_typescript(
577        &self,
578        expression: impl Into<String>,
579        options: TypeScriptExecutionOptions,
580    ) -> ClientResult<CodeEvaluationResult> {
581        let inputs = json_inputs(&options.inline)?;
582        let submission = self
583            .submit_execution(
584                wire::RequestPayload::TypeScriptEvaluationRequest(
585                    wire::TypeScriptEvaluationRequest {
586                        process: process(&options.inline.process),
587                        expression: expression.into(),
588                        file_path: options.file_path,
589                        tsconfig_path: options.tsconfig_path,
590                        compiler_options: options
591                            .compiler_options
592                            .as_ref()
593                            .map(serde_json::to_string)
594                            .transpose()
595                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
596                        inputs,
597                    },
598                ),
599                false,
600            )
601            .await?;
602        evaluation_result(submission)
603    }
604
605    pub async fn execute_typescript_file(
606        &self,
607        path: impl Into<String>,
608        options: TypeScriptExecutionOptions,
609    ) -> ClientResult<CodeExecutionResult> {
610        completed_submission(
611            self.submit_execution(
612                wire::RequestPayload::TypeScriptFileExecutionRequest(
613                    wire::TypeScriptFileExecutionRequest {
614                        process: process(&options.inline.process),
615                        path: path.into(),
616                        tsconfig_path: options.tsconfig_path,
617                        compiler_options: options
618                            .compiler_options
619                            .as_ref()
620                            .map(serde_json::to_string)
621                            .transpose()
622                            .map_err(|error| ClientError::Sidecar(error.to_string()))?,
623                    },
624                ),
625                false,
626            )
627            .await?,
628        )
629    }
630
631    pub async fn spawn_typescript(
632        &self,
633        source: impl Into<String>,
634        options: LanguageSpawnOptions,
635    ) -> ClientResult<ProcessDescriptor> {
636        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
637        background_submission(
638            self,
639            self.submit_execution(
640                wire::RequestPayload::TypeScriptExecutionRequest(
641                    wire::TypeScriptExecutionRequest {
642                        process: background_process(&options, operation_id),
643                        source: source.into(),
644                        file_path: None,
645                        tsconfig_path: None,
646                        compiler_options: None,
647                        inputs: None,
648                    },
649                ),
650                true,
651            )
652            .await?,
653            "javascript",
654        )
655    }
656
657    pub async fn spawn_typescript_file(
658        &self,
659        path: impl Into<String>,
660        options: LanguageSpawnOptions,
661    ) -> ClientResult<ProcessDescriptor> {
662        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
663        background_submission(
664            self,
665            self.submit_execution(
666                wire::RequestPayload::TypeScriptFileExecutionRequest(
667                    wire::TypeScriptFileExecutionRequest {
668                        process: background_process(&options, operation_id),
669                        path: path.into(),
670                        tsconfig_path: None,
671                        compiler_options: None,
672                    },
673                ),
674                true,
675            )
676            .await?,
677            "javascript",
678        )
679    }
680
681    pub async fn check_typescript(
682        &self,
683        source: impl Into<String>,
684        options: TypeScriptCheckOptions,
685    ) -> ClientResult<TypeScriptCheckResult> {
686        let compiler_options = options
687            .compiler_options
688            .as_ref()
689            .map(serde_json::to_string)
690            .transpose()
691            .map_err(|error| ClientError::Sidecar(error.to_string()))?;
692        let submission = self
693            .submit_execution(
694                wire::RequestPayload::TypeScriptCheckRequest(wire::TypeScriptCheckRequest {
695                    identity: wire::ExecutionIdentityOptions {
696                        context_id: options.context_id,
697                    },
698                    output: output(options.output),
699                    source: source.into(),
700                    cwd: options.cwd,
701                    file_path: options.file_path,
702                    tsconfig_path: options.tsconfig_path,
703                    compiler_options,
704                    timeout_ms: options.timeout_ms,
705                }),
706                false,
707            )
708            .await?;
709        let result = completed_submission(submission)?;
710        typescript_check_result(result)
711    }
712
713    pub async fn check_typescript_project(
714        &self,
715        options: TypeScriptCheckOptions,
716    ) -> ClientResult<TypeScriptCheckResult> {
717        let submission = self
718            .submit_execution(
719                wire::RequestPayload::TypeScriptProjectCheckRequest(
720                    wire::TypeScriptProjectCheckRequest {
721                        identity: wire::ExecutionIdentityOptions {
722                            context_id: options.context_id,
723                        },
724                        output: output(options.output),
725                        cwd: options.cwd,
726                        tsconfig_path: options.tsconfig_path,
727                        timeout_ms: options.timeout_ms,
728                    },
729                ),
730                false,
731            )
732            .await?;
733        let result = completed_submission(submission)?;
734        typescript_check_result(result)
735    }
736
737    pub async fn install_npm_project(
738        &self,
739        options: NpmProjectInstallOptions,
740    ) -> ClientResult<CodeExecutionResult> {
741        completed_submission(
742            self.submit_execution(
743                wire::RequestPayload::NpmProjectInstallRequest(wire::NpmProjectInstallRequest {
744                    identity: wire::ExecutionIdentityOptions {
745                        context_id: options.context_id,
746                    },
747                    output: output(options.output),
748                    cwd: options.cwd,
749                    env: (!options.env.is_empty()).then_some(options.env),
750                    timeout_ms: options.timeout_ms,
751                    frozen: options.frozen,
752                }),
753                false,
754            )
755            .await?,
756        )
757    }
758
759    pub async fn install_npm_packages(
760        &self,
761        packages: Vec<String>,
762        options: NpmPackageInstallOptions,
763    ) -> ClientResult<CodeExecutionResult> {
764        completed_submission(
765            self.submit_execution(
766                wire::RequestPayload::NpmPackageInstallRequest(wire::NpmPackageInstallRequest {
767                    identity: wire::ExecutionIdentityOptions {
768                        context_id: options.context_id,
769                    },
770                    output: output(options.output),
771                    cwd: options.cwd,
772                    env: (!options.env.is_empty()).then_some(options.env),
773                    timeout_ms: options.timeout_ms,
774                    packages,
775                    dev: options.dev,
776                    global: options.global,
777                }),
778                false,
779            )
780            .await?,
781        )
782    }
783
784    pub async fn execute_npm_script(
785        &self,
786        script: impl Into<String>,
787        options: LanguageExecutionOptions,
788    ) -> ClientResult<CodeExecutionResult> {
789        completed_submission(
790            self.submit_execution(
791                wire::RequestPayload::NpmScriptExecutionRequest(wire::NpmScriptExecutionRequest {
792                    process: process(&options),
793                    script: script.into(),
794                }),
795                false,
796            )
797            .await?,
798        )
799    }
800
801    pub async fn execute_npm_package(
802        &self,
803        package_spec: impl Into<String>,
804        binary: Option<String>,
805        options: LanguageExecutionOptions,
806    ) -> ClientResult<CodeExecutionResult> {
807        completed_submission(
808            self.submit_execution(
809                wire::RequestPayload::NpmPackageExecutionRequest(
810                    wire::NpmPackageExecutionRequest {
811                        process: process(&options),
812                        package_spec: package_spec.into(),
813                        binary,
814                    },
815                ),
816                false,
817            )
818            .await?,
819        )
820    }
821
822    pub async fn execute_python(
823        &self,
824        source: impl Into<String>,
825        options: InlineExecutionOptions,
826    ) -> ClientResult<CodeExecutionResult> {
827        let inputs = json_inputs(&options)?;
828        completed_submission(
829            self.submit_execution(
830                wire::RequestPayload::PythonExecutionRequest(wire::PythonExecutionRequest {
831                    process: process(&options.process),
832                    source: source.into(),
833                    inputs,
834                }),
835                false,
836            )
837            .await?,
838        )
839    }
840
841    pub async fn evaluate_python(
842        &self,
843        expression: impl Into<String>,
844        options: InlineExecutionOptions,
845    ) -> ClientResult<CodeEvaluationResult> {
846        let inputs = json_inputs(&options)?;
847        let submission = self
848            .submit_execution(
849                wire::RequestPayload::PythonEvaluationRequest(wire::PythonEvaluationRequest {
850                    process: process(&options.process),
851                    expression: expression.into(),
852                    inputs,
853                }),
854                false,
855            )
856            .await?;
857        evaluation_result(submission)
858    }
859
860    pub async fn execute_python_file(
861        &self,
862        path: impl Into<String>,
863        options: LanguageExecutionOptions,
864    ) -> ClientResult<CodeExecutionResult> {
865        completed_submission(
866            self.submit_execution(
867                wire::RequestPayload::PythonFileExecutionRequest(
868                    wire::PythonFileExecutionRequest {
869                        process: process(&options),
870                        path: path.into(),
871                    },
872                ),
873                false,
874            )
875            .await?,
876        )
877    }
878
879    pub async fn execute_python_module(
880        &self,
881        module: impl Into<String>,
882        options: LanguageExecutionOptions,
883    ) -> ClientResult<CodeExecutionResult> {
884        completed_submission(
885            self.submit_execution(
886                wire::RequestPayload::PythonModuleExecutionRequest(
887                    wire::PythonModuleExecutionRequest {
888                        process: process(&options),
889                        module: module.into(),
890                    },
891                ),
892                false,
893            )
894            .await?,
895        )
896    }
897
898    pub async fn spawn_python(
899        &self,
900        source: impl Into<String>,
901        options: LanguageSpawnOptions,
902    ) -> ClientResult<ProcessDescriptor> {
903        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
904        background_submission(
905            self,
906            self.submit_execution(
907                wire::RequestPayload::PythonExecutionRequest(wire::PythonExecutionRequest {
908                    process: background_process(&options, operation_id),
909                    source: source.into(),
910                    inputs: None,
911                }),
912                true,
913            )
914            .await?,
915            "python",
916        )
917    }
918
919    pub async fn spawn_python_file(
920        &self,
921        path: impl Into<String>,
922        options: LanguageSpawnOptions,
923    ) -> ClientResult<ProcessDescriptor> {
924        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
925        background_submission(
926            self,
927            self.submit_execution(
928                wire::RequestPayload::PythonFileExecutionRequest(
929                    wire::PythonFileExecutionRequest {
930                        process: background_process(&options, operation_id),
931                        path: path.into(),
932                    },
933                ),
934                true,
935            )
936            .await?,
937            "python",
938        )
939    }
940
941    pub async fn spawn_python_module(
942        &self,
943        module: impl Into<String>,
944        options: LanguageSpawnOptions,
945    ) -> ClientResult<ProcessDescriptor> {
946        let operation_id = format!("process-{}", uuid::Uuid::new_v4());
947        background_submission(
948            self,
949            self.submit_execution(
950                wire::RequestPayload::PythonModuleExecutionRequest(
951                    wire::PythonModuleExecutionRequest {
952                        process: background_process(&options, operation_id),
953                        module: module.into(),
954                    },
955                ),
956                true,
957            )
958            .await?,
959            "python",
960        )
961    }
962
963    pub async fn install_python_packages(
964        &self,
965        packages: Vec<String>,
966        options: PythonInstallOptions,
967    ) -> ClientResult<CodeExecutionResult> {
968        if !packages.is_empty() && options.requirements_file.is_some() {
969            return Err(ClientError::Sidecar(String::from(
970                "install_python_packages cannot combine packages with requirements_file",
971            )));
972        }
973        completed_submission(
974            self.submit_execution(
975                wire::RequestPayload::PythonInstallRequest(wire::PythonInstallRequest {
976                    identity: wire::ExecutionIdentityOptions {
977                        context_id: options.context_id,
978                    },
979                    output: output(options.output),
980                    cwd: options.cwd,
981                    env: (!options.env.is_empty()).then_some(options.env),
982                    timeout_ms: options.timeout_ms,
983                    packages,
984                    upgrade: options.upgrade,
985                    requirements_file: options.requirements_file,
986                    index_url: options.index_url,
987                    extra_index_urls: options.extra_index_urls,
988                }),
989                false,
990            )
991            .await?,
992        )
993    }
994
995    pub async fn create_context(&self, context_id: &str) -> ClientResult<()> {
996        match self
997            .transport()
998            .request_wire(
999                self.execution_ownership(),
1000                wire::RequestPayload::CreateContextRequest(wire::CreateContextRequest {
1001                    context_id: context_id.to_owned(),
1002                }),
1003            )
1004            .await?
1005        {
1006            wire::ResponsePayload::ExecutionDescriptorResponse(_) => Ok(()),
1007            wire::ResponsePayload::RejectedResponse(rejected) => {
1008                Err(ClientError::from_rejection(rejected))
1009            }
1010            response => Err(ClientError::Sidecar(format!(
1011                "unexpected create_context response: {response:?}"
1012            ))),
1013        }
1014    }
1015
1016    pub async fn get_context(&self, context_id: &str) -> ClientResult<ContextDescriptor> {
1017        match self
1018            .transport()
1019            .request_wire(
1020                self.execution_ownership(),
1021                wire::RequestPayload::GetExecutionRequest(wire::GetExecutionRequest {
1022                    execution_id: context_id.to_owned(),
1023                }),
1024            )
1025            .await?
1026        {
1027            wire::ResponsePayload::ExecutionDescriptorResponse(response) => {
1028                Ok(context_descriptor(response.execution))
1029            }
1030            wire::ResponsePayload::RejectedResponse(rejected) => {
1031                Err(ClientError::from_rejection(rejected))
1032            }
1033            response => Err(ClientError::Sidecar(format!(
1034                "unexpected get_context response: {response:?}"
1035            ))),
1036        }
1037    }
1038
1039    pub async fn list_contexts(&self) -> ClientResult<Vec<ContextDescriptor>> {
1040        match self
1041            .transport()
1042            .request_wire(
1043                self.execution_ownership(),
1044                wire::RequestPayload::ListExecutionsRequest,
1045            )
1046            .await?
1047        {
1048            wire::ResponsePayload::ExecutionListResponse(response) => Ok(response
1049                .executions
1050                .into_iter()
1051                .map(context_descriptor)
1052                .collect()),
1053            wire::ResponsePayload::RejectedResponse(rejected) => {
1054                Err(ClientError::from_rejection(rejected))
1055            }
1056            response => Err(ClientError::Sidecar(format!(
1057                "unexpected list_contexts response: {response:?}"
1058            ))),
1059        }
1060    }
1061
1062    async fn wait_execution(&self, execution_id: &str) -> ClientResult<CodeExecutionResult> {
1063        let mut events = self.transport().subscribe_wire_events();
1064        let first = self
1065            .transport()
1066            .request_wire(
1067                self.execution_ownership(),
1068                wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest {
1069                    execution_id: execution_id.to_owned(),
1070                }),
1071            )
1072            .await?;
1073        let response = match first {
1074            wire::ResponsePayload::RejectedResponse(rejected)
1075                if rejected.code == "execution_busy" =>
1076            {
1077                wait_for_completion_event(&mut events, execution_id).await?;
1078                self.transport()
1079                    .request_wire(
1080                        self.execution_ownership(),
1081                        wire::RequestPayload::WaitExecutionRequest(wire::WaitExecutionRequest {
1082                            execution_id: execution_id.to_owned(),
1083                        }),
1084                    )
1085                    .await?
1086            }
1087            response => response,
1088        };
1089        match response {
1090            wire::ResponsePayload::ExecutionCompletedResponse(response) => Ok(response),
1091            wire::ResponsePayload::RejectedResponse(rejected) => {
1092                Err(ClientError::from_rejection(rejected))
1093            }
1094            response => Err(ClientError::Sidecar(format!(
1095                "unexpected wait_execution response: {response:?}"
1096            ))),
1097        }
1098    }
1099
1100    pub async fn reset_context(&self, context_id: &str) -> ClientResult<()> {
1101        self.execution_descriptor_request(wire::RequestPayload::ResetExecutionRequest(
1102            wire::ResetExecutionRequest {
1103                execution_id: context_id.to_owned(),
1104            },
1105        ))
1106        .await?;
1107        Ok(())
1108    }
1109
1110    async fn execution_descriptor_request(
1111        &self,
1112        payload: wire::RequestPayload,
1113    ) -> ClientResult<wire::ExecutionDescriptor> {
1114        match self
1115            .transport()
1116            .request_wire(self.execution_ownership(), payload)
1117            .await?
1118        {
1119            wire::ResponsePayload::ExecutionDescriptorResponse(response) => Ok(response.execution),
1120            wire::ResponsePayload::RejectedResponse(rejected) => {
1121                Err(ClientError::from_rejection(rejected))
1122            }
1123            response => Err(ClientError::Sidecar(format!(
1124                "unexpected execution lifecycle response: {response:?}"
1125            ))),
1126        }
1127    }
1128
1129    pub async fn delete_context(&self, context_id: &str) -> ClientResult<()> {
1130        match self
1131            .transport()
1132            .request_wire(
1133                self.execution_ownership(),
1134                wire::RequestPayload::DeleteExecutionRequest(wire::DeleteExecutionRequest {
1135                    execution_id: context_id.to_owned(),
1136                }),
1137            )
1138            .await?
1139        {
1140            wire::ResponsePayload::ExecutionDeletedResponse(_) => Ok(()),
1141            wire::ResponsePayload::RejectedResponse(rejected) => {
1142                Err(ClientError::from_rejection(rejected))
1143            }
1144            response => Err(ClientError::Sidecar(format!(
1145                "unexpected delete_context response: {response:?}"
1146            ))),
1147        }
1148    }
1149}
1150
1151async fn wait_for_completion_event(
1152    events: &mut broadcast::Receiver<(wire::OwnershipScope, wire::EventPayload)>,
1153    execution_id: &str,
1154) -> ClientResult<()> {
1155    loop {
1156        match events.recv().await {
1157            Ok((_, wire::EventPayload::ExecutionCompletedEvent(event)))
1158                if event.execution_id == execution_id =>
1159            {
1160                return Ok(())
1161            }
1162            Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}
1163            Err(broadcast::error::RecvError::Closed) => {
1164                return Err(ClientError::Sidecar(String::from(
1165                    "execution event stream closed before completion",
1166                )))
1167            }
1168        }
1169    }
1170}
1171
1172fn evaluation_result(submission: ExecutionSubmission) -> ClientResult<CodeEvaluationResult> {
1173    let ExecutionSubmission::Completed(result) = submission else {
1174        return Err(ClientError::Sidecar(String::from(
1175            "evaluation unexpectedly returned a background process",
1176        )));
1177    };
1178    let value = result
1179        .evaluation_value
1180        .as_deref()
1181        .map(serde_json::from_str::<serde_json::Value>)
1182        .transpose()
1183        .map_err(|error| {
1184            ClientError::Sidecar(format!("failed to decode evaluation result: {error}"))
1185        })?;
1186    Ok(CodeEvaluationResult { result, value })
1187}
1188
1189fn typescript_check_result(result: CodeExecutionResult) -> ClientResult<TypeScriptCheckResult> {
1190    if result.outcome != wire::ExecutionOutcome::Succeeded {
1191        return Ok(TypeScriptCheckResult {
1192            result,
1193            has_errors: None,
1194            diagnostics: Vec::new(),
1195        });
1196    }
1197    let data: serde_json::Value = result
1198        .type_script_check_result
1199        .as_deref()
1200        .map(serde_json::from_str)
1201        .transpose()
1202        .map_err(|error| {
1203            ClientError::Sidecar(format!(
1204                "failed to decode TypeScript checker result: {error}"
1205            ))
1206        })?
1207        .ok_or_else(|| {
1208            ClientError::Sidecar(String::from(
1209                "TypeScript checker returned no diagnostic result",
1210            ))
1211        })?;
1212    let data = data.as_object().ok_or_else(|| {
1213        ClientError::Sidecar(String::from(
1214            "TypeScript checker returned an invalid diagnostic result",
1215        ))
1216    })?;
1217    let has_errors = data
1218        .get("hasErrors")
1219        .and_then(serde_json::Value::as_bool)
1220        .ok_or_else(|| {
1221            ClientError::Sidecar(String::from(
1222                "TypeScript checker returned an invalid hasErrors value",
1223            ))
1224        })?;
1225    let diagnostics = data
1226        .get("diagnostics")
1227        .and_then(serde_json::Value::as_array)
1228        .ok_or_else(|| {
1229            ClientError::Sidecar(String::from(
1230                "TypeScript checker returned invalid diagnostics",
1231            ))
1232        })?
1233        .iter()
1234        .map(|diagnostic| {
1235            let code = diagnostic
1236                .get("code")
1237                .and_then(serde_json::Value::as_u64)
1238                .and_then(|code| u32::try_from(code).ok())
1239                .ok_or_else(|| {
1240                    ClientError::Sidecar(String::from(
1241                        "TypeScript checker returned an invalid diagnostic code",
1242                    ))
1243                })?;
1244            let category = diagnostic
1245                .get("category")
1246                .and_then(serde_json::Value::as_str)
1247                .ok_or_else(|| {
1248                    ClientError::Sidecar(String::from(
1249                        "TypeScript checker returned an invalid diagnostic category",
1250                    ))
1251                })?
1252                .to_owned();
1253            let message = diagnostic
1254                .get("message")
1255                .and_then(serde_json::Value::as_str)
1256                .ok_or_else(|| {
1257                    ClientError::Sidecar(String::from(
1258                        "TypeScript checker returned an invalid diagnostic message",
1259                    ))
1260                })?
1261                .to_owned();
1262            Ok(wire::TypeScriptDiagnostic {
1263                code,
1264                category,
1265                message,
1266                file_path: diagnostic
1267                    .get("filePath")
1268                    .and_then(serde_json::Value::as_str)
1269                    .map(str::to_owned),
1270                line: diagnostic
1271                    .get("line")
1272                    .and_then(serde_json::Value::as_u64)
1273                    .and_then(|line| u32::try_from(line).ok()),
1274                column: diagnostic
1275                    .get("column")
1276                    .and_then(serde_json::Value::as_u64)
1277                    .and_then(|column| u32::try_from(column).ok()),
1278            })
1279        })
1280        .collect::<ClientResult<Vec<_>>>()?;
1281    Ok(TypeScriptCheckResult {
1282        result,
1283        has_errors: Some(has_errors),
1284        diagnostics,
1285    })
1286}
1287
1288fn completed_submission(submission: ExecutionSubmission) -> ClientResult<CodeExecutionResult> {
1289    match submission {
1290        ExecutionSubmission::Completed(result) => Ok(result),
1291        ExecutionSubmission::Background(_) => Err(ClientError::Sidecar(String::from(
1292            "attached operation unexpectedly returned a background process",
1293        ))),
1294    }
1295}
1296
1297fn background_submission(
1298    client: &AgentOs,
1299    submission: ExecutionSubmission,
1300    language: &str,
1301) -> ClientResult<ProcessDescriptor> {
1302    match submission {
1303        ExecutionSubmission::Background(descriptor) => {
1304            let pid = descriptor.pid.ok_or_else(|| {
1305                ClientError::Sidecar(String::from(
1306                    "spawned process admission returned no numeric pid",
1307                ))
1308            })?;
1309            let process_id = descriptor.process_id.clone().ok_or_else(|| {
1310                ClientError::Sidecar(String::from(
1311                    "spawned process admission returned no process routing id",
1312                ))
1313            })?;
1314            let (stdout_tx, _) = broadcast::channel::<Vec<u8>>(1024);
1315            let (stderr_tx, _) = broadcast::channel::<Vec<u8>>(1024);
1316            let (output_tx, _) = broadcast::channel::<ProcessOutput>(1024);
1317            let (exit_tx, _) = watch::channel::<Option<i32>>(None);
1318            let (kernel_pid_tx, _) = watch::channel(Some(pid));
1319            let entry = ProcessEntry {
1320                command: format!("{language} source"),
1321                args: Vec::new(),
1322                stdout_tx: stdout_tx.clone(),
1323                stderr_tx: stderr_tx.clone(),
1324                output_tx: output_tx.clone(),
1325                exit_tx: exit_tx.clone(),
1326                process_id: process_id.clone(),
1327                kernel_pid: kernel_pid_tx,
1328                output_tasks: Vec::new(),
1329                started_at: descriptor.created_at_ms as i64,
1330            };
1331            let _ = client.inner().processes.insert(pid, entry);
1332            let mut events = client.transport().subscribe_wire_events();
1333            let operation_id = descriptor.execution_id.clone();
1334            tokio::spawn(async move {
1335                loop {
1336                    let Ok((_, event)) = events.recv().await else {
1337                        break;
1338                    };
1339                    match event {
1340                        wire::EventPayload::ExecutionOutputEvent(output)
1341                            if output.execution_id == operation_id =>
1342                        {
1343                            let (stream, tx) = match output.channel {
1344                                wire::ExecutionStreamChannel::Stdout => {
1345                                    (ProcessStream::Stdout, &stdout_tx)
1346                                }
1347                                wire::ExecutionStreamChannel::Stderr
1348                                | wire::ExecutionStreamChannel::Pty => {
1349                                    (ProcessStream::Stderr, &stderr_tx)
1350                                }
1351                            };
1352                            let _ = tx.send(output.chunk.clone());
1353                            let _ = output_tx.send(ProcessOutput {
1354                                pid,
1355                                stream,
1356                                data: output.chunk,
1357                            });
1358                        }
1359                        wire::EventPayload::ExecutionCompletedEvent(completed)
1360                            if completed.execution_id == operation_id =>
1361                        {
1362                            let _ = exit_tx.send(Some(completed.exit_code.unwrap_or(1)));
1363                            break;
1364                        }
1365                        _ => {}
1366                    }
1367                }
1368            });
1369            Ok(ProcessDescriptor {
1370                pid,
1371                state: String::from("running"),
1372                language: Some(language.to_owned()),
1373                started_at_ms: descriptor.created_at_ms,
1374            })
1375        }
1376        ExecutionSubmission::Completed(_) => Err(ClientError::Sidecar(String::from(
1377            "spawned process unexpectedly returned an attached result",
1378        ))),
1379    }
1380}
agentos_client/language_execution.rs

agentos_client/
language_execution.rs
