1use crate::common::{encode_json_string, frozen_time_ms};
2use crate::javascript::{
3 CreateJavascriptContextRequest, GuestRuntimeConfig, JavascriptExecution,
4 JavascriptExecutionEngine, JavascriptExecutionError, JavascriptExecutionEvent,
5 JavascriptExecutionLimits, JavascriptSyncRpcRequest, StartJavascriptExecutionRequest,
6};
7use crate::node_import_cache::{NodeImportCache, NODE_IMPORT_CACHE_ASSET_ROOT_ENV};
8use crate::runtime_support::{
9 env_flag_enabled, file_fingerprint, resolve_execution_path, warmup_marker_path,
10 NODE_DISABLE_COMPILE_CACHE_ENV, NODE_FROZEN_TIME_ENV,
11};
12use crate::v8_runtime;
13use agentos_runtime::RuntimeContext;
14use base64::Engine as _;
15use serde::{Deserialize, Serialize};
16use serde_json::{json, Value};
17use std::collections::BTreeMap;
18use std::fmt;
19use std::fs;
20use std::io::{Read, Seek, SeekFrom};
21use std::os::unix::fs::MetadataExt;
22use std::path::{Component, Path, PathBuf};
23use std::sync::{Arc, Mutex};
24use std::time::{Duration, Instant};
25use tokio::sync::Notify;
26const NODE_ALLOW_PROCESS_BINDINGS_ENV: &str = "AGENTOS_ALLOW_PROCESS_BINDINGS";
27const NODE_GUEST_PATH_MAPPINGS_ENV: &str = "AGENTOS_GUEST_PATH_MAPPINGS";
28const NODE_SYNC_RPC_DATA_BYTES_ENV: &str = "AGENTOS_NODE_SYNC_RPC_DATA_BYTES";
29const PYODIDE_INDEX_URL_ENV: &str = "AGENTOS_PYODIDE_INDEX_URL";
30const PYODIDE_PACKAGE_BASE_URL_ENV: &str = "AGENTOS_PYODIDE_PACKAGE_BASE_URL";
31const PYODIDE_PACKAGE_CACHE_DIR_ENV: &str = "AGENTOS_PYODIDE_PACKAGE_CACHE_DIR";
32const PYODIDE_GUEST_ROOT: &str = "/__agentos_pyodide";
33const PYODIDE_CACHE_GUEST_ROOT: &str = "/__agentos_pyodide_cache";
34const PYTHON_CODE_ENV: &str = "AGENTOS_PYTHON_CODE";
35const PYTHON_FILE_ENV: &str = "AGENTOS_PYTHON_FILE";
36const PYTHON_PREWARM_ONLY_ENV: &str = "AGENTOS_PYTHON_PREWARM_ONLY";
37const PYTHON_WARMUP_DEBUG_ENV: &str = "AGENTOS_PYTHON_WARMUP_DEBUG";
38const PYTHON_WARMUP_METRICS_PREFIX: &str = "__AGENTOS_PYTHON_WARMUP_METRICS__:";
39const PYTHON_WARMUP_MARKER_VERSION: &str = "2";
40const DEFAULT_PYTHON_OUTPUT_BUFFER_MAX_BYTES: usize = 1024 * 1024;
41const DEFAULT_PYTHON_EXECUTION_TIMEOUT_MS: u64 = 5 * 60 * 1000;
42const DEFAULT_PYTHON_MAX_OLD_SPACE_MB: usize = 0;
43const DEFAULT_PYTHON_VFS_RPC_TIMEOUT_MS: u64 = 30_000;
44const PYTHON_SYNC_RPC_DATA_BYTES: usize = 20 * 1024 * 1024;
45const PYTHON_SYNC_RPC_WAIT_TIMEOUT_MS: u64 = 120_000;
46const PYTHON_PREWARM_TIMEOUT: Duration = Duration::from_secs(120);
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum PythonVfsRpcMethod {
50 Read,
51 Write,
52 Stat,
53 Lstat,
54 ReadDir,
55 Mkdir,
56 Unlink,
57 Rmdir,
58 Rename,
59 Symlink,
60 ReadLink,
61 Setattr,
62 HttpRequest,
63 DnsLookup,
64 SubprocessRun,
65 SocketConnect,
66 SocketSend,
67 SocketRecv,
68 SocketClose,
69 UdpCreate,
70 UdpSendto,
71 UdpRecvfrom,
72}
73
74impl PythonVfsRpcMethod {
75 fn from_wire(value: &str) -> Option<Self> {
76 match value {
77 "fsRead" => Some(Self::Read),
78 "fsWrite" => Some(Self::Write),
79 "fsStat" => Some(Self::Stat),
80 "fsLstat" => Some(Self::Lstat),
81 "fsReaddir" => Some(Self::ReadDir),
82 "fsMkdir" => Some(Self::Mkdir),
83 "fsUnlink" => Some(Self::Unlink),
84 "fsRmdir" => Some(Self::Rmdir),
85 "fsRename" => Some(Self::Rename),
86 "fsSymlink" => Some(Self::Symlink),
87 "fsReadlink" => Some(Self::ReadLink),
88 "fsSetattr" => Some(Self::Setattr),
89 "httpRequest" => Some(Self::HttpRequest),
90 "dnsLookup" => Some(Self::DnsLookup),
91 "subprocessRun" => Some(Self::SubprocessRun),
92 "socketConnect" => Some(Self::SocketConnect),
93 "socketSend" => Some(Self::SocketSend),
94 "socketRecv" => Some(Self::SocketRecv),
95 "socketClose" => Some(Self::SocketClose),
96 "udpCreate" => Some(Self::UdpCreate),
97 "udpSendto" => Some(Self::UdpSendto),
98 "udpRecvfrom" => Some(Self::UdpRecvfrom),
99 _ => None,
100 }
101 }
102}
103
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct PythonVfsRpcRequest {
106 pub id: u64,
107 pub method: PythonVfsRpcMethod,
108 pub path: String,
109 pub destination: Option<String>,
111 pub target: Option<String>,
113 pub mode: Option<u32>,
115 pub uid: Option<u32>,
116 pub gid: Option<u32>,
117 pub atime_ms: Option<u64>,
118 pub mtime_ms: Option<u64>,
119 pub content_base64: Option<String>,
120 pub recursive: bool,
121 pub url: Option<String>,
122 pub http_method: Option<String>,
123 pub headers: BTreeMap<String, String>,
124 pub body_base64: Option<String>,
125 pub hostname: Option<String>,
126 pub family: Option<u8>,
127 pub port: Option<u16>,
129 pub socket_id: Option<u64>,
131 pub command: Option<String>,
132 pub args: Vec<String>,
133 pub argv0: Option<String>,
135 pub cwd: Option<String>,
136 pub env: BTreeMap<String, String>,
137 pub shell: bool,
138 pub max_buffer: Option<usize>,
139 pub timeout_ms: Option<u64>,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct PythonVfsRpcStat {
146 pub mode: u32,
147 pub size: u64,
148 pub is_directory: bool,
149 pub is_symbolic_link: bool,
150}
151
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub enum PythonVfsRpcResponsePayload {
154 Empty,
155 Read {
156 content_base64: String,
157 },
158 Stat {
159 stat: PythonVfsRpcStat,
160 },
161 ReadDir {
162 entries: Vec<String>,
163 },
164 Http {
165 status: u16,
166 reason: String,
167 url: String,
168 headers: BTreeMap<String, Vec<String>>,
169 body_base64: String,
170 },
171 DnsLookup {
172 addresses: Vec<String>,
173 },
174 SubprocessRun {
175 exit_code: i32,
176 stdout: String,
177 stderr: String,
178 max_buffer_exceeded: bool,
179 },
180 SocketCreated {
181 socket_id: u64,
182 },
183 SocketSent {
184 bytes_sent: usize,
185 },
186 SocketReceived {
187 data_base64: String,
188 closed: bool,
189 timed_out: bool,
190 },
191 UdpReceived {
192 data_base64: String,
193 host: String,
194 port: u16,
195 timed_out: bool,
196 },
197 SymlinkTarget {
198 target: String,
199 },
200}
201
202#[derive(Debug, Deserialize)]
203#[serde(rename_all = "camelCase")]
204struct PythonVfsBridgeRequestWire {
205 method: String,
206 #[serde(default)]
207 path: String,
208 #[serde(default)]
209 destination: Option<String>,
210 #[serde(default)]
211 target: Option<String>,
212 #[serde(default)]
214 mode: Option<f64>,
215 #[serde(default)]
216 uid: Option<f64>,
217 #[serde(default)]
218 gid: Option<f64>,
219 #[serde(default, rename = "atimeMs")]
220 atime_ms: Option<f64>,
221 #[serde(default, rename = "mtimeMs")]
222 mtime_ms: Option<f64>,
223 #[serde(default)]
224 content_base64: Option<String>,
225 #[serde(default)]
226 recursive: bool,
227 #[serde(default)]
228 url: Option<String>,
229 #[serde(default, rename = "httpMethod")]
230 http_method: Option<String>,
231 #[serde(default)]
232 headers: BTreeMap<String, String>,
233 #[serde(default, rename = "bodyBase64")]
234 body_base64: Option<String>,
235 #[serde(default)]
236 hostname: Option<String>,
237 #[serde(default)]
238 family: Option<u8>,
239 #[serde(default)]
240 port: Option<u16>,
241 #[serde(default, rename = "socketId")]
242 socket_id: Option<u64>,
243 #[serde(default)]
244 command: Option<String>,
245 #[serde(default)]
246 args: Vec<String>,
247 #[serde(default)]
248 argv0: Option<String>,
249 #[serde(default)]
250 cwd: Option<String>,
251 #[serde(default)]
252 env: BTreeMap<String, String>,
253 #[serde(default)]
254 shell: bool,
255 #[serde(default, rename = "maxBuffer")]
256 max_buffer: Option<usize>,
257 #[serde(default, rename = "timeoutMs")]
258 timeout_ms: Option<u64>,
259}
260
261#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
262#[serde(rename_all = "camelCase")]
263struct PythonGuestPathMappingWire {
264 guest_path: String,
265 host_path: String,
266}
267
268#[derive(Debug, Clone, PartialEq, Eq)]
269pub struct CreatePythonContextRequest {
270 pub vm_id: String,
271 pub pyodide_dist_path: PathBuf,
272}
273
274#[derive(Debug, Clone, PartialEq, Eq)]
275pub struct PythonContext {
276 pub context_id: String,
277 pub vm_id: String,
278 pub pyodide_dist_path: PathBuf,
279}
280
281#[derive(Debug, Clone, Default, PartialEq, Eq)]
286pub struct PythonExecutionLimits {
287 pub output_buffer_max_bytes: Option<usize>,
289 pub execution_timeout_ms: Option<u64>,
292 pub max_old_space_mb: Option<usize>,
295 pub vfs_rpc_timeout_ms: Option<u64>,
297 pub reactor_work_quantum: Option<usize>,
299 pub bridge_call_timeout_ms: Option<u64>,
301 pub max_open_fds: Option<usize>,
305}
306
307#[derive(Debug, Clone, PartialEq, Eq)]
308pub struct StartPythonExecutionRequest {
309 pub vm_id: String,
310 pub context_id: String,
311 pub code: String,
312 pub file_path: Option<PathBuf>,
313 pub env: BTreeMap<String, String>,
314 pub cwd: PathBuf,
315 pub limits: PythonExecutionLimits,
317 pub guest_runtime: GuestRuntimeConfig,
320}
321
322#[derive(Debug, Clone, PartialEq, Eq)]
323pub enum PythonExecutionEvent {
324 Stdout(Vec<u8>),
325 Stderr(Vec<u8>),
326 JavascriptSyncRpcRequest(JavascriptSyncRpcRequest),
327 VfsRpcRequest(Box<PythonVfsRpcRequest>),
328 Exited(i32),
329}
330
331#[derive(Debug, Clone, PartialEq, Eq)]
332pub struct PythonExecutionResult {
333 pub execution_id: String,
334 pub exit_code: i32,
335 pub stdout: Vec<u8>,
336 pub stderr: Vec<u8>,
337}
338
339#[derive(Debug)]
340pub enum PythonExecutionError {
341 MissingContext(String),
342 InvalidLimit(String),
343 VmMismatch {
344 expected: String,
345 found: String,
346 },
347 RuntimeUnavailable,
350 PrepareRuntime(std::io::Error),
351 PrepareWarmPath(std::io::Error),
352 WarmupFailed {
353 exit_code: i32,
354 stderr: String,
355 },
356 Spawn(std::io::Error),
357 StdinClosed,
358 Stdin(std::io::Error),
359 Kill(std::io::Error),
360 Control(std::io::Error),
361 TimedOut(Duration),
362 PendingVfsRpcRequest(u64),
363 RpcResponse(String),
364 OutputBufferExceeded {
365 stream: &'static str,
366 limit: usize,
367 },
368 EventChannelClosed,
369}
370
371impl fmt::Display for PythonExecutionError {
372 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
373 match self {
374 Self::MissingContext(context_id) => {
375 write!(f, "unknown guest Python context: {context_id}")
376 }
377 Self::InvalidLimit(message) => write!(f, "invalid Python limit: {message}"),
378 Self::VmMismatch { expected, found } => {
379 write!(
380 f,
381 "guest Python context belongs to vm {expected}, not {found}"
382 )
383 }
384 Self::RuntimeUnavailable => write!(
385 f,
386 "guest Python execution is unavailable: this build of agentos-execution \
387 was compiled without the bundled Pyodide runtime assets"
388 ),
389 Self::PrepareRuntime(err) => {
390 write!(f, "failed to prepare guest Python runtime assets: {err}")
391 }
392 Self::PrepareWarmPath(err) => {
393 write!(f, "failed to prepare guest Python warm path: {err}")
394 }
395 Self::WarmupFailed { exit_code, stderr } => {
396 if stderr.trim().is_empty() {
397 write!(f, "guest Python warmup exited with status {exit_code}")
398 } else {
399 write!(
400 f,
401 "guest Python warmup exited with status {exit_code}: {}",
402 stderr.trim()
403 )
404 }
405 }
406 Self::Spawn(err) => write!(f, "failed to start guest Python runtime: {err}"),
407 Self::StdinClosed => f.write_str("guest Python stdin is already closed"),
408 Self::Stdin(err) => write!(f, "failed to write guest stdin: {err}"),
409 Self::Kill(err) => write!(f, "failed to kill guest Python runtime: {err}"),
410 Self::Control(err) => write!(f, "failed to control guest Python runtime: {err}"),
411 Self::TimedOut(timeout) => write!(
412 f,
413 "guest Python runtime timed out after {}ms",
414 timeout.as_millis()
415 ),
416 Self::PendingVfsRpcRequest(id) => {
417 write!(
418 f,
419 "guest Python execution requires servicing pending VFS RPC request {id}"
420 )
421 }
422 Self::RpcResponse(message) => {
423 write!(
424 f,
425 "failed to reply to guest Python VFS RPC request: {message}"
426 )
427 }
428 Self::OutputBufferExceeded { stream, limit } => {
429 write!(
430 f,
431 "guest Python {stream} exceeded the captured output limit of {limit} bytes"
432 )
433 }
434 Self::EventChannelClosed => {
435 f.write_str("guest Python event channel closed unexpectedly")
436 }
437 }
438 }
439}
440
441impl std::error::Error for PythonExecutionError {}
442
443fn ensure_pyodide_available() -> Result<(), PythonExecutionError> {
447 #[cfg(agentos_pyodide_unavailable)]
448 {
449 return Err(PythonExecutionError::RuntimeUnavailable);
450 }
451 #[cfg(not(agentos_pyodide_unavailable))]
452 {
453 Ok(())
454 }
455}
456
457#[derive(Debug)]
458pub struct PythonExecution {
459 runtime: RuntimeContext,
460 execution_id: String,
461 child_pid: u32,
462 inner: JavascriptExecution,
463 pyodide_dist_path: PathBuf,
464 managed_host_files: PythonManagedHostFiles,
465 pending_vfs_rpc: Arc<Mutex<Option<PendingVfsRpc>>>,
466 v8_session: crate::v8_host::V8SessionHandle,
467 output_buffer_max_bytes: usize,
468 execution_timeout: Option<Duration>,
469 vfs_rpc_timeout: Duration,
470}
471
472#[derive(Debug, Clone)]
479pub struct PythonVfsRpcResponder {
480 pending_vfs_rpc: Arc<Mutex<Option<PendingVfsRpc>>>,
481 v8_session: crate::v8_host::V8SessionHandle,
482}
483
484#[derive(Debug)]
485struct PendingVfsRpc {
486 state: PendingVfsRpcState,
487 timeout_abort: Option<tokio::task::AbortHandle>,
488}
489
490#[derive(Debug, Clone, Copy, PartialEq, Eq)]
491enum PendingVfsRpcState {
492 Pending(u64),
493 TimedOut(u64),
494}
495
496#[derive(Debug, Clone, Copy, PartialEq, Eq)]
497enum PendingVfsRpcResolution {
498 Pending,
499 TimedOut,
500 Missing,
501}
502
503impl PythonExecution {
504 pub fn vfs_rpc_responder(&self) -> PythonVfsRpcResponder {
505 PythonVfsRpcResponder {
506 pending_vfs_rpc: Arc::clone(&self.pending_vfs_rpc),
507 v8_session: self.v8_session.clone(),
508 }
509 }
510
511 pub fn execution_id(&self) -> &str {
512 &self.execution_id
513 }
514
515 pub fn child_pid(&self) -> u32 {
516 self.child_pid
517 }
518
519 pub fn uses_shared_v8_runtime(&self) -> bool {
520 self.inner.uses_shared_v8_runtime()
521 }
522
523 pub fn execute_retained(&mut self, source: String) -> Result<(), PythonExecutionError> {
526 let source = serde_json::to_string(&source).map_err(|error| {
527 PythonExecutionError::Control(std::io::Error::other(format!(
528 "failed to encode retained Python source: {error}"
529 )))
530 })?;
531 let runner = format!(
532 "process.exitCode = 0; (async () => {{ const pyodide = globalThis.__agentOSRetainedPyodide; if (!pyodide) throw new Error('retained Python interpreter is unavailable'); await pyodide.runPythonAsync({source}); }})()"
533 );
534 self.inner
535 .execute_retained(runner, String::from("/[agentos-python-retained]"), false)
536 .map_err(map_javascript_error)
537 }
538
539 pub fn start_prepared(&mut self) -> Result<(), PythonExecutionError> {
540 self.inner.start_prepared().map_err(map_javascript_error)
541 }
542
543 #[doc(hidden)]
544 pub fn is_prepared_for_start(&self) -> bool {
545 self.inner.is_prepared_for_start()
546 }
547
548 pub fn write_stdin(&mut self, chunk: &[u8]) -> Result<(), PythonExecutionError> {
549 self.inner
550 .write_kernel_stdin_only(chunk)
551 .map_err(map_javascript_error)
552 }
553
554 pub fn close_stdin(&mut self) -> Result<(), PythonExecutionError> {
555 self.inner.close_kernel_stdin_only();
556 Ok(())
557 }
558
559 pub fn cancel(&mut self) -> Result<(), PythonExecutionError> {
560 self.kill()
561 }
562
563 pub fn kill(&mut self) -> Result<(), PythonExecutionError> {
564 self.close_stdin()?;
565 self.inner.terminate().map_err(map_javascript_error)
566 }
567
568 pub fn pause(&self) -> Result<(), PythonExecutionError> {
569 self.inner.pause().map_err(map_javascript_error)
570 }
571
572 pub fn resume(&self) -> Result<(), PythonExecutionError> {
573 self.inner.resume().map_err(map_javascript_error)
574 }
575
576 pub fn respond_vfs_rpc_success(
577 &mut self,
578 id: u64,
579 payload: PythonVfsRpcResponsePayload,
580 ) -> Result<(), PythonExecutionError> {
581 self.vfs_rpc_responder().respond_success(id, payload)
582 }
583
584 pub fn respond_vfs_rpc_error(
585 &mut self,
586 id: u64,
587 code: impl Into<String>,
588 message: impl Into<String>,
589 ) -> Result<(), PythonExecutionError> {
590 self.vfs_rpc_responder().respond_error(id, code, message)
591 }
592
593 pub fn respond_javascript_sync_rpc_success(
594 &mut self,
595 id: u64,
596 result: Value,
597 ) -> Result<(), PythonExecutionError> {
598 self.inner
599 .respond_sync_rpc_success(id, result)
600 .map_err(map_javascript_error)
601 }
602
603 pub fn claim_javascript_sync_rpc_response(
604 &mut self,
605 id: u64,
606 ) -> Result<bool, PythonExecutionError> {
607 self.inner
608 .claim_sync_rpc_response(id)
609 .map_err(map_javascript_error)
610 }
611
612 pub fn respond_claimed_javascript_sync_rpc_success(
613 &mut self,
614 id: u64,
615 result: Value,
616 ) -> Result<(), PythonExecutionError> {
617 self.inner
618 .respond_claimed_sync_rpc_success(id, result)
619 .map_err(map_javascript_error)
620 }
621
622 pub fn respond_javascript_sync_rpc_error(
623 &mut self,
624 id: u64,
625 code: impl Into<String>,
626 message: impl Into<String>,
627 ) -> Result<(), PythonExecutionError> {
628 self.inner
629 .respond_sync_rpc_error(id, code, message)
630 .map_err(map_javascript_error)
631 }
632
633 pub fn respond_claimed_javascript_sync_rpc_error(
634 &mut self,
635 id: u64,
636 code: impl Into<String>,
637 message: impl Into<String>,
638 ) -> Result<(), PythonExecutionError> {
639 self.inner
640 .respond_claimed_sync_rpc_error(id, code, message)
641 .map_err(map_javascript_error)
642 }
643
644 pub async fn poll_event(
645 &mut self,
646 timeout: Duration,
647 ) -> Result<Option<PythonExecutionEvent>, PythonExecutionError> {
648 self.poll_event_until(Some(timeout)).await
649 }
650
651 pub fn try_poll_event(&mut self) -> Result<Option<PythonExecutionEvent>, PythonExecutionError> {
652 loop {
653 let Some(event) = self.inner.try_poll_event().map_err(map_javascript_error)? else {
654 return Ok(None);
655 };
656 if let Some(event) = self.translate_javascript_event(event)? {
657 return Ok(Some(event));
658 }
659 }
660 }
661
662 async fn poll_event_until(
663 &mut self,
664 timeout: Option<Duration>,
665 ) -> Result<Option<PythonExecutionEvent>, PythonExecutionError> {
666 let started = Instant::now();
667 loop {
668 let remaining = timeout.map(|timeout| {
669 if timeout.is_zero() {
670 Duration::ZERO
671 } else {
672 timeout.saturating_sub(started.elapsed())
673 }
674 });
675 match self
676 .inner
677 .poll_event_until(remaining)
678 .await
679 .map_err(map_javascript_error)?
680 {
681 Some(event) => {
682 if let Some(event) = self.translate_javascript_event(event)? {
683 return Ok(Some(event));
684 }
685 }
686 None => return Ok(None),
687 }
688 }
689 }
690
691 pub fn try_service_standalone_module_sync_rpc(
695 &mut self,
696 request: &JavascriptSyncRpcRequest,
697 ) -> Result<bool, PythonExecutionError> {
698 self.inner
699 .try_service_standalone_module_sync_rpc(request)
700 .map_err(map_javascript_error)
701 }
702
703 #[doc(hidden)]
706 pub fn try_service_standalone_stdin_sync_rpc(
707 &mut self,
708 request: &JavascriptSyncRpcRequest,
709 ) -> Result<bool, PythonExecutionError> {
710 self.inner
711 .handle_kernel_stdin_sync_rpc(request)
712 .map_err(map_javascript_error)
713 }
714
715 pub fn poll_event_blocking(
716 &mut self,
717 timeout: Duration,
718 ) -> Result<Option<PythonExecutionEvent>, PythonExecutionError> {
719 let deadline = Instant::now() + timeout;
720 loop {
721 let remaining = deadline.saturating_duration_since(Instant::now());
722 match self
723 .inner
724 .poll_event_blocking(remaining)
725 .map_err(map_javascript_error)?
726 {
727 Some(event) => {
728 if let Some(event) = self.translate_javascript_event(event)? {
729 return Ok(Some(event));
730 }
731 }
732 None => {
733 if Instant::now() >= deadline {
734 return Ok(None);
735 }
736 }
737 }
738 }
739 }
740
741 fn next_event_blocking(&mut self) -> Result<PythonExecutionEvent, PythonExecutionError> {
742 loop {
743 let event = self
744 .inner
745 .next_event_blocking()
746 .map_err(map_javascript_error)?;
747 if let Some(event) = self.translate_javascript_event(event)? {
748 return Ok(event);
749 }
750 }
751 }
752
753 pub fn wait(
754 mut self,
755 timeout: Option<Duration>,
756 ) -> Result<PythonExecutionResult, PythonExecutionError> {
757 self.close_stdin()?;
758
759 let mut stdout = PythonOutputBuffer::new(self.output_buffer_max_bytes);
760 let mut stderr = PythonOutputBuffer::new(self.output_buffer_max_bytes);
761 let started = Instant::now();
762 let timeout = match (timeout, self.execution_timeout) {
763 (Some(requested), Some(configured)) => Some(requested.min(configured)),
764 (Some(requested), None) => Some(requested),
765 (None, Some(configured)) => Some(configured),
766 (None, None) => None,
767 };
768 loop {
769 let poll_timeout = python_wait_remaining(timeout, started);
770 let event = match poll_timeout {
771 Some(timeout) => self.poll_event_blocking(timeout)?,
772 None => Some(self.next_event_blocking()?),
773 };
774
775 match event {
776 Some(PythonExecutionEvent::Stdout(chunk)) => stdout.extend(&chunk),
777 Some(PythonExecutionEvent::Stderr(chunk)) => stderr.extend(&chunk),
778 Some(PythonExecutionEvent::JavascriptSyncRpcRequest(request)) => {
779 if self
783 .inner
784 .try_service_standalone_module_sync_rpc(&request)
785 .map_err(map_javascript_error)?
786 {
787 continue;
788 }
789 if let Some((code, message)) = python_javascript_sync_rpc_error(&request) {
790 self.inner
791 .respond_sync_rpc_error(request.id, code, message)
792 .map_err(map_javascript_error)?;
793 continue;
794 }
795 return Err(PythonExecutionError::RpcResponse(format!(
796 "guest Python execution requires servicing pending JavaScript sync RPC request {} {} {:?}",
797 request.id, request.method, request.args
798 )));
799 }
800 Some(PythonExecutionEvent::VfsRpcRequest(request)) => {
801 return Err(PythonExecutionError::PendingVfsRpcRequest(request.id));
802 }
803 Some(PythonExecutionEvent::Exited(exit_code)) => {
804 return Ok(PythonExecutionResult {
805 execution_id: self.execution_id.clone(),
806 exit_code,
807 stdout: stdout.into_inner(),
808 stderr: stderr.into_inner(),
809 });
810 }
811 None => {}
812 }
813
814 if let Some(limit) = timeout {
815 if started.elapsed() >= limit {
816 self.kill()?;
817 return Err(PythonExecutionError::TimedOut(limit));
818 }
819 }
820 }
821 }
822
823 fn translate_javascript_event(
824 &mut self,
825 event: JavascriptExecutionEvent,
826 ) -> Result<Option<PythonExecutionEvent>, PythonExecutionError> {
827 match event {
828 JavascriptExecutionEvent::Stdout(chunk) => {
829 Ok(Some(PythonExecutionEvent::Stdout(chunk)))
830 }
831 JavascriptExecutionEvent::Stderr(chunk) => {
832 Ok(Some(PythonExecutionEvent::Stderr(chunk)))
833 }
834 JavascriptExecutionEvent::Exited(code) => Ok(Some(PythonExecutionEvent::Exited(code))),
835 JavascriptExecutionEvent::SignalState { .. } => Ok(None),
836 JavascriptExecutionEvent::SyncRpcRequest(request) => {
837 if request.method == "_pythonRpc" {
838 let request = parse_python_bridge_sync_rpc_request(&request)?;
839 set_pending_vfs_rpc_state(&self.pending_vfs_rpc, request.id)?;
840 spawn_python_vfs_rpc_timeout(
841 &self.runtime,
842 request.id,
843 self.vfs_rpc_timeout,
844 self.pending_vfs_rpc.clone(),
845 self.v8_session.clone(),
846 )?;
847 Ok(Some(PythonExecutionEvent::VfsRpcRequest(Box::new(request))))
848 } else {
849 if self.try_service_standalone_module_sync_rpc(&request)? {
850 return Ok(None);
851 }
852 if let Some(action) = python_javascript_sync_rpc_action(
853 &self.pyodide_dist_path,
854 &mut self.managed_host_files,
855 &request,
856 )? {
857 respond_python_javascript_sync_rpc_action(
858 &mut self.inner,
859 request.id,
860 action,
861 )?;
862 Ok(None)
863 } else {
864 Ok(Some(PythonExecutionEvent::JavascriptSyncRpcRequest(
865 request,
866 )))
867 }
868 }
869 }
870 }
871 }
872}
873
874fn python_wait_remaining(timeout: Option<Duration>, started: Instant) -> Option<Duration> {
875 timeout.map(|limit| limit.saturating_sub(started.elapsed()))
876}
877
878impl PythonVfsRpcResponder {
879 pub fn respond_success(
880 &self,
881 id: u64,
882 payload: PythonVfsRpcResponsePayload,
883 ) -> Result<(), PythonExecutionError> {
884 match clear_pending_vfs_rpc(&self.pending_vfs_rpc, id)? {
885 PendingVfsRpcResolution::Pending => {}
886 PendingVfsRpcResolution::TimedOut | PendingVfsRpcResolution::Missing => {
887 return Err(PythonExecutionError::RpcResponse(format!(
888 "VFS RPC request {id} is no longer pending"
889 )));
890 }
891 }
892
893 let result = match payload {
894 PythonVfsRpcResponsePayload::Empty => json!({}),
895 PythonVfsRpcResponsePayload::Read { content_base64 } => {
896 json!({ "contentBase64": content_base64 })
897 }
898 PythonVfsRpcResponsePayload::Stat { stat } => json!({
899 "stat": {
900 "mode": stat.mode,
901 "size": stat.size,
902 "isDirectory": stat.is_directory,
903 "isSymbolicLink": stat.is_symbolic_link,
904 }
905 }),
906 PythonVfsRpcResponsePayload::ReadDir { entries } => {
907 json!({ "entries": entries })
908 }
909 PythonVfsRpcResponsePayload::Http {
910 status,
911 reason,
912 url,
913 headers,
914 body_base64,
915 } => json!({
916 "status": status,
917 "reason": reason,
918 "url": url,
919 "headers": headers,
920 "bodyBase64": body_base64,
921 }),
922 PythonVfsRpcResponsePayload::DnsLookup { addresses } => {
923 json!({ "addresses": addresses })
924 }
925 PythonVfsRpcResponsePayload::SubprocessRun {
926 exit_code,
927 stdout,
928 stderr,
929 max_buffer_exceeded,
930 } => json!({
931 "exitCode": exit_code,
932 "stdout": stdout,
933 "stderr": stderr,
934 "maxBufferExceeded": max_buffer_exceeded,
935 }),
936 PythonVfsRpcResponsePayload::SocketCreated { socket_id } => json!({
937 "socketId": socket_id,
938 }),
939 PythonVfsRpcResponsePayload::SocketSent { bytes_sent } => json!({
940 "bytesSent": bytes_sent,
941 }),
942 PythonVfsRpcResponsePayload::SocketReceived {
943 data_base64,
944 closed,
945 timed_out,
946 } => json!({
947 "dataBase64": data_base64,
948 "closed": closed,
949 "timedOut": timed_out,
950 }),
951 PythonVfsRpcResponsePayload::UdpReceived {
952 data_base64,
953 host,
954 port,
955 timed_out,
956 } => json!({
957 "dataBase64": data_base64,
958 "host": host,
959 "port": port,
960 "timedOut": timed_out,
961 }),
962 PythonVfsRpcResponsePayload::SymlinkTarget { target } => json!({
963 "target": target,
964 }),
965 };
966
967 let payload = v8_runtime::json_to_cbor_payload(&result)
968 .map_err(|error| PythonExecutionError::RpcResponse(error.to_string()))?;
969 self.v8_session
970 .send_bridge_response(id, 0, payload)
971 .map_err(|error| PythonExecutionError::RpcResponse(error.to_string()))
972 }
973
974 pub fn respond_error(
975 &self,
976 id: u64,
977 code: impl Into<String>,
978 message: impl Into<String>,
979 ) -> Result<(), PythonExecutionError> {
980 match clear_pending_vfs_rpc(&self.pending_vfs_rpc, id)? {
981 PendingVfsRpcResolution::Pending => {}
982 PendingVfsRpcResolution::TimedOut | PendingVfsRpcResolution::Missing => {
983 return Err(PythonExecutionError::RpcResponse(format!(
984 "VFS RPC request {id} is no longer pending"
985 )));
986 }
987 }
988
989 let error = format!("{}: {}", code.into(), message.into());
990 self.v8_session
991 .send_bridge_response(id, 1, error.into_bytes())
992 .map_err(|error| PythonExecutionError::RpcResponse(error.to_string()))
993 }
994}
995
996fn clear_pending_vfs_rpc(
997 pending_vfs_rpc: &Arc<Mutex<Option<PendingVfsRpc>>>,
998 id: u64,
999) -> Result<PendingVfsRpcResolution, PythonExecutionError> {
1000 let mut pending = pending_vfs_rpc
1001 .lock()
1002 .map_err(|_| PythonExecutionError::EventChannelClosed)?;
1003 let resolution = match pending.as_ref().map(|rpc| rpc.state) {
1004 Some(PendingVfsRpcState::Pending(current)) if current == id => {
1005 PendingVfsRpcResolution::Pending
1006 }
1007 Some(PendingVfsRpcState::TimedOut(current)) if current == id => {
1008 PendingVfsRpcResolution::TimedOut
1009 }
1010 _ => return Ok(PendingVfsRpcResolution::Missing),
1011 };
1012 if let Some(rpc) = pending.take() {
1013 if let Some(timeout_abort) = rpc.timeout_abort {
1014 timeout_abort.abort();
1015 }
1016 }
1017 Ok(resolution)
1018}
1019
1020fn cancel_pending_vfs_rpc(pending_vfs_rpc: &Arc<Mutex<Option<PendingVfsRpc>>>) {
1021 let pending = pending_vfs_rpc
1022 .lock()
1023 .map(|mut pending| pending.take())
1024 .unwrap_or_else(|poisoned| {
1025 eprintln!("ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: cancelling pending timeout");
1026 poisoned.into_inner().take()
1027 });
1028 if let Some(timeout_abort) = pending.and_then(|rpc| rpc.timeout_abort) {
1029 timeout_abort.abort();
1030 }
1031}
1032
1033impl Drop for PythonExecution {
1034 fn drop(&mut self) {
1035 cancel_pending_vfs_rpc(&self.pending_vfs_rpc);
1036 if let Err(error) = self.close_stdin() {
1037 eprintln!("ERR_AGENTOS_PYTHON_STDIN_CLOSE: {error}");
1038 }
1039 if let Err(error) = self.inner.terminate() {
1040 eprintln!("ERR_AGENTOS_PYTHON_TERMINATE: {error}");
1041 }
1042 }
1043}
1044
1045#[derive(Debug)]
1046pub struct PythonExecutionEngine {
1047 runtime: Option<RuntimeContext>,
1048 next_context_id: usize,
1049 next_execution_id: usize,
1050 contexts: BTreeMap<String, PythonContext>,
1051 import_caches: BTreeMap<String, NodeImportCache>,
1052 javascript_context_ids: BTreeMap<String, String>,
1053 javascript_engine: JavascriptExecutionEngine,
1054}
1055
1056impl Default for PythonExecutionEngine {
1057 fn default() -> Self {
1058 let runtime = default_python_test_runtime_context();
1059 let javascript_engine = runtime
1060 .as_ref()
1061 .map_or_else(JavascriptExecutionEngine::default, |runtime| {
1062 JavascriptExecutionEngine::new(runtime.clone())
1063 });
1064 Self {
1065 runtime,
1066 next_context_id: 0,
1067 next_execution_id: 0,
1068 contexts: BTreeMap::new(),
1069 import_caches: BTreeMap::new(),
1070 javascript_context_ids: BTreeMap::new(),
1071 javascript_engine,
1072 }
1073 }
1074}
1075
1076#[cfg(test)]
1077fn default_python_test_runtime_context() -> Option<RuntimeContext> {
1078 agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default())
1079 .ok()
1080 .map(agentos_runtime::SidecarRuntime::context)
1081}
1082
1083#[cfg(not(test))]
1084fn default_python_test_runtime_context() -> Option<RuntimeContext> {
1085 None
1086}
1087
1088impl PythonExecutionEngine {
1089 pub fn new(runtime: RuntimeContext) -> Self {
1090 Self {
1091 runtime: Some(runtime.clone()),
1092 next_context_id: 0,
1093 next_execution_id: 0,
1094 contexts: BTreeMap::new(),
1095 import_caches: BTreeMap::new(),
1096 javascript_context_ids: BTreeMap::new(),
1097 javascript_engine: JavascriptExecutionEngine::new(runtime),
1098 }
1099 }
1100
1101 pub fn set_runtime_context(&mut self, runtime: RuntimeContext) {
1102 self.javascript_engine.set_runtime_context(runtime.clone());
1103 self.runtime = Some(runtime);
1104 }
1105
1106 fn runtime_context(&self) -> Result<&RuntimeContext, PythonExecutionError> {
1107 self.runtime.as_ref().ok_or_else(|| {
1108 PythonExecutionError::Spawn(std::io::Error::other(
1109 "ERR_AGENTOS_RUNTIME_NOT_INJECTED: PythonExecutionEngine requires a process RuntimeContext; construct it with PythonExecutionEngine::new(runtime)",
1110 ))
1111 })
1112 }
1113
1114 pub fn set_event_notify(&mut self, notify: Option<Arc<Notify>>) {
1115 self.javascript_engine.set_event_notify(notify);
1116 }
1117
1118 pub fn bundled_pyodide_dist_path_for_vm(
1119 &mut self,
1120 vm_id: &str,
1121 ) -> Result<PathBuf, PythonExecutionError> {
1122 ensure_pyodide_available()?;
1123 let runtime = self.runtime_context()?.clone();
1124 let import_cache = self.import_caches.entry(vm_id.to_owned()).or_default();
1125 import_cache
1126 .ensure_materialized_with_runtime(&runtime)
1127 .map_err(PythonExecutionError::PrepareRuntime)?;
1128 Ok(import_cache.pyodide_dist_path().to_path_buf())
1129 }
1130
1131 pub async fn bundled_pyodide_dist_path_for_vm_async(
1132 &mut self,
1133 vm_id: &str,
1134 runtime: &RuntimeContext,
1135 ) -> Result<PathBuf, PythonExecutionError> {
1136 ensure_pyodide_available()?;
1137 let import_cache = self.import_caches.entry(vm_id.to_owned()).or_default();
1138 import_cache
1139 .ensure_materialized_with_timeout_and_runtime_async(runtime, PYTHON_PREWARM_TIMEOUT)
1140 .await
1141 .map_err(PythonExecutionError::PrepareRuntime)?;
1142 Ok(import_cache.pyodide_dist_path().to_path_buf())
1143 }
1144
1145 pub fn create_context(&mut self, request: CreatePythonContextRequest) -> PythonContext {
1146 self.next_context_id += 1;
1147 self.import_caches.entry(request.vm_id.clone()).or_default();
1148 let javascript_context =
1149 self.javascript_engine
1150 .create_context(CreateJavascriptContextRequest {
1151 vm_id: request.vm_id.clone(),
1152 bootstrap_module: None,
1153 compile_cache_root: None,
1154 });
1155
1156 let context = PythonContext {
1157 context_id: format!("python-ctx-{}", self.next_context_id),
1158 vm_id: request.vm_id,
1159 pyodide_dist_path: request.pyodide_dist_path,
1160 };
1161 self.javascript_context_ids
1162 .insert(context.context_id.clone(), javascript_context.context_id);
1163 self.contexts
1164 .insert(context.context_id.clone(), context.clone());
1165 context
1166 }
1167
1168 pub fn dispose_context(&mut self, context_id: &str) -> bool {
1171 let removed = self.contexts.remove(context_id).is_some();
1172 if let Some(javascript_context_id) = self.javascript_context_ids.remove(context_id) {
1173 self.javascript_engine
1174 .dispose_context(&javascript_context_id);
1175 }
1176 removed
1177 }
1178
1179 #[doc(hidden)]
1180 pub fn context_count_for_test(&self) -> usize {
1181 self.contexts.len()
1182 }
1183
1184 #[doc(hidden)]
1185 pub fn javascript_context_count_for_test(&self) -> usize {
1186 self.javascript_engine.context_count_for_test()
1187 }
1188
1189 pub fn start_execution(
1190 &mut self,
1191 request: StartPythonExecutionRequest,
1192 ) -> Result<PythonExecution, PythonExecutionError> {
1193 let runtime = self.runtime_context()?.clone();
1194 self.create_execution_with_runtime(request, runtime, false)
1195 }
1196
1197 pub fn prepare_execution(
1198 &mut self,
1199 request: StartPythonExecutionRequest,
1200 ) -> Result<PythonExecution, PythonExecutionError> {
1201 let runtime = self.runtime_context()?.clone();
1202 self.create_execution_with_runtime(request, runtime, true)
1203 }
1204
1205 pub fn start_execution_with_runtime(
1206 &mut self,
1207 request: StartPythonExecutionRequest,
1208 runtime: RuntimeContext,
1209 ) -> Result<PythonExecution, PythonExecutionError> {
1210 self.create_execution_with_runtime(request, runtime, false)
1211 }
1212
1213 fn create_execution_with_runtime(
1214 &mut self,
1215 request: StartPythonExecutionRequest,
1216 runtime: RuntimeContext,
1217 defer_execute: bool,
1218 ) -> Result<PythonExecution, PythonExecutionError> {
1219 ensure_pyodide_available()?;
1220 let context = self
1221 .contexts
1222 .get(&request.context_id)
1223 .cloned()
1224 .ok_or_else(|| PythonExecutionError::MissingContext(request.context_id.clone()))?;
1225
1226 if context.vm_id != request.vm_id {
1227 return Err(PythonExecutionError::VmMismatch {
1228 expected: context.vm_id,
1229 found: request.vm_id,
1230 });
1231 }
1232
1233 let frozen_time_ms = frozen_time_ms();
1234 let javascript_context_id = self
1235 .javascript_context_ids
1236 .get(&context.context_id)
1237 .cloned()
1238 .ok_or_else(|| PythonExecutionError::MissingContext(context.context_id.clone()))?;
1239 let warmup_metrics = {
1240 let import_cache = self.import_caches.entry(context.vm_id.clone()).or_default();
1241 import_cache
1242 .ensure_materialized_with_runtime(&runtime)
1243 .map_err(PythonExecutionError::PrepareRuntime)?;
1244 prewarm_python_path(
1245 import_cache,
1246 &mut self.javascript_engine,
1247 &javascript_context_id,
1248 &context,
1249 &request,
1250 frozen_time_ms,
1251 &runtime,
1252 )?
1253 };
1254
1255 self.finish_start_execution(
1256 request,
1257 runtime,
1258 &context,
1259 javascript_context_id,
1260 frozen_time_ms,
1261 warmup_metrics,
1262 defer_execute,
1263 )
1264 }
1265
1266 pub async fn start_execution_with_runtime_async(
1269 &mut self,
1270 request: StartPythonExecutionRequest,
1271 runtime: RuntimeContext,
1272 ) -> Result<PythonExecution, PythonExecutionError> {
1273 ensure_pyodide_available()?;
1274 let context = self
1275 .contexts
1276 .get(&request.context_id)
1277 .cloned()
1278 .ok_or_else(|| PythonExecutionError::MissingContext(request.context_id.clone()))?;
1279
1280 if context.vm_id != request.vm_id {
1281 return Err(PythonExecutionError::VmMismatch {
1282 expected: context.vm_id,
1283 found: request.vm_id,
1284 });
1285 }
1286
1287 let frozen_time_ms = frozen_time_ms();
1288 let javascript_context =
1289 self.javascript_engine
1290 .create_context(CreateJavascriptContextRequest {
1291 vm_id: request.vm_id.clone(),
1292 bootstrap_module: None,
1293 compile_cache_root: None,
1294 });
1295 let javascript_context_id = javascript_context.context_id.clone();
1296 self.javascript_context_ids
1297 .insert(context.context_id.clone(), javascript_context_id.clone());
1298 let warmup_metrics = {
1299 let import_cache = self.import_caches.entry(context.vm_id.clone()).or_default();
1300 import_cache
1301 .ensure_materialized_with_timeout_and_runtime_async(
1302 &runtime,
1303 PYTHON_PREWARM_TIMEOUT,
1304 )
1305 .await
1306 .map_err(PythonExecutionError::PrepareRuntime)?;
1307 prewarm_python_path_async(
1308 import_cache,
1309 &mut self.javascript_engine,
1310 &javascript_context_id,
1311 &context,
1312 &request,
1313 frozen_time_ms,
1314 &runtime,
1315 )
1316 .await?
1317 };
1318
1319 self.finish_start_execution(
1320 request,
1321 runtime,
1322 &context,
1323 javascript_context_id,
1324 frozen_time_ms,
1325 warmup_metrics,
1326 false,
1327 )
1328 }
1329
1330 #[allow(clippy::too_many_arguments)]
1331 fn finish_start_execution(
1332 &mut self,
1333 request: StartPythonExecutionRequest,
1334 runtime: RuntimeContext,
1335 context: &PythonContext,
1336 javascript_context_id: String,
1337 frozen_time_ms: u128,
1338 warmup_metrics: Option<Vec<u8>>,
1339 defer_execute: bool,
1340 ) -> Result<PythonExecution, PythonExecutionError> {
1341 self.next_execution_id += 1;
1342 let execution_id = format!("exec-{}", self.next_execution_id);
1343 let import_cache = self
1344 .import_caches
1345 .get(&context.vm_id)
1346 .expect("vm import cache should exist after materialization");
1347 let pyodide_dist_path =
1348 resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd);
1349 let javascript_execution = start_python_javascript_execution(
1350 &mut self.javascript_engine,
1351 &runtime,
1352 import_cache,
1353 &javascript_context_id,
1354 context,
1355 &request,
1356 PythonJavascriptExecutionOptions {
1357 frozen_time_ms,
1358 prewarm_only: false,
1359 warmup_metrics: warmup_metrics.as_deref(),
1360 defer_execute,
1361 },
1362 )?;
1363 let pending_vfs_rpc = Arc::new(Mutex::new(None));
1364 let vfs_rpc_timeout = python_vfs_rpc_timeout(&request);
1365
1366 Ok(PythonExecution {
1367 runtime,
1368 execution_id,
1369 child_pid: javascript_execution.child_pid(),
1370 v8_session: javascript_execution.v8_session_handle(),
1371 inner: javascript_execution,
1372 pyodide_dist_path,
1373 managed_host_files: PythonManagedHostFiles::new(python_managed_host_file_limit(
1374 &request,
1375 )),
1376 pending_vfs_rpc,
1377 output_buffer_max_bytes: python_output_buffer_max_bytes(&request),
1378 execution_timeout: python_execution_timeout(&request),
1379 vfs_rpc_timeout,
1380 })
1381 }
1382
1383 pub fn dispose_vm(&mut self, vm_id: &str) {
1384 self.contexts.retain(|_, context| context.vm_id != vm_id);
1385 self.javascript_context_ids
1386 .retain(|python_context_id, _| self.contexts.contains_key(python_context_id));
1387 self.import_caches.remove(vm_id);
1388 self.javascript_engine.dispose_vm(vm_id);
1389 }
1390}
1391
1392fn set_pending_vfs_rpc_state(
1393 pending_vfs_rpc: &Arc<Mutex<Option<PendingVfsRpc>>>,
1394 id: u64,
1395) -> Result<(), PythonExecutionError> {
1396 let mut pending = pending_vfs_rpc
1397 .lock()
1398 .map_err(|_| PythonExecutionError::EventChannelClosed)?;
1399 if let Some(PendingVfsRpc {
1400 state: PendingVfsRpcState::Pending(current),
1401 ..
1402 }) = pending.as_ref()
1403 {
1404 return Err(PythonExecutionError::PendingVfsRpcRequest(*current));
1405 }
1406 if let Some(previous) = pending.take() {
1407 if let Some(timeout_abort) = previous.timeout_abort {
1408 timeout_abort.abort();
1409 }
1410 }
1411 *pending = Some(PendingVfsRpc {
1412 state: PendingVfsRpcState::Pending(id),
1413 timeout_abort: None,
1414 });
1415 Ok(())
1416}
1417
1418fn map_javascript_error(error: JavascriptExecutionError) -> PythonExecutionError {
1419 match error {
1420 JavascriptExecutionError::EmptyArgv => PythonExecutionError::Spawn(std::io::Error::new(
1421 std::io::ErrorKind::InvalidInput,
1422 "guest Python bootstrap requires a JavaScript entrypoint",
1423 )),
1424 JavascriptExecutionError::InvalidLimit(message) => {
1425 PythonExecutionError::InvalidLimit(message)
1426 }
1427 JavascriptExecutionError::MissingContext(context_id) => {
1428 PythonExecutionError::MissingContext(context_id)
1429 }
1430 JavascriptExecutionError::VmMismatch { expected, found } => {
1431 PythonExecutionError::VmMismatch { expected, found }
1432 }
1433 JavascriptExecutionError::PrepareImportCache(error) => {
1434 PythonExecutionError::PrepareRuntime(error)
1435 }
1436 JavascriptExecutionError::Spawn(error) => PythonExecutionError::Spawn(error),
1437 JavascriptExecutionError::PendingSyncRpcRequest(id) => {
1438 PythonExecutionError::PendingVfsRpcRequest(id)
1439 }
1440 JavascriptExecutionError::ExpiredSyncRpcRequest(id) => {
1441 PythonExecutionError::RpcResponse(format!("VFS RPC request {id} is no longer pending"))
1442 }
1443 JavascriptExecutionError::RpcResponse(message) => {
1444 PythonExecutionError::RpcResponse(message)
1445 }
1446 JavascriptExecutionError::Terminate(error) => PythonExecutionError::Kill(error),
1447 JavascriptExecutionError::Control(error) => PythonExecutionError::Control(error),
1448 JavascriptExecutionError::StdinClosed => PythonExecutionError::StdinClosed,
1449 JavascriptExecutionError::Stdin(error) => PythonExecutionError::Stdin(error),
1450 JavascriptExecutionError::OutputBufferExceeded { stream, limit } => {
1451 PythonExecutionError::OutputBufferExceeded { stream, limit }
1452 }
1453 JavascriptExecutionError::EventChannelClosed => PythonExecutionError::EventChannelClosed,
1454 }
1455}
1456
1457struct PythonJavascriptExecutionOptions<'a> {
1458 frozen_time_ms: u128,
1459 prewarm_only: bool,
1460 warmup_metrics: Option<&'a [u8]>,
1461 defer_execute: bool,
1462}
1463
1464fn start_python_javascript_execution(
1465 javascript_engine: &mut JavascriptExecutionEngine,
1466 runtime: &RuntimeContext,
1467 import_cache: &NodeImportCache,
1468 javascript_context_id: &str,
1469 context: &PythonContext,
1470 request: &StartPythonExecutionRequest,
1471 options: PythonJavascriptExecutionOptions<'_>,
1472) -> Result<JavascriptExecution, PythonExecutionError> {
1473 let internal_env = build_python_internal_env(
1474 import_cache,
1475 context,
1476 request,
1477 options.frozen_time_ms,
1478 options.prewarm_only,
1479 );
1480 let inline_code =
1481 build_python_runner_module_source(import_cache, &internal_env, options.warmup_metrics)?;
1482 let mut env = request.env.clone();
1483 env.extend(internal_env);
1484
1485 let max_old_space_mb = python_max_old_space_mb(request);
1489 let runner_limits = python_runner_javascript_limits(&request.limits, max_old_space_mb);
1490
1491 let javascript_request = StartJavascriptExecutionRequest {
1492 vm_id: request.vm_id.clone(),
1493 context_id: javascript_context_id.to_owned(),
1494 argv: vec![import_cache.python_runner_path().display().to_string()],
1495 argv0: None,
1496 env,
1497 cwd: request.cwd.clone(),
1498 limits: runner_limits,
1499 guest_runtime: request.guest_runtime.clone(),
1502 wasm_module_bytes: None,
1503 inline_code: Some(inline_code),
1504 };
1505 if options.defer_execute {
1506 javascript_engine.prepare_execution_with_runtime(javascript_request, runtime.clone())
1507 } else {
1508 javascript_engine.start_execution_with_runtime(javascript_request, runtime.clone())
1509 }
1510 .map_err(map_javascript_error)
1511}
1512
1513fn python_runner_javascript_limits(
1514 limits: &PythonExecutionLimits,
1515 max_old_space_mb: usize,
1516) -> JavascriptExecutionLimits {
1517 JavascriptExecutionLimits {
1518 v8_heap_limit_mb: (max_old_space_mb > 0).then_some(max_old_space_mb as u32),
1519 sync_rpc_wait_timeout_ms: Some(PYTHON_SYNC_RPC_WAIT_TIMEOUT_MS),
1520 reactor_work_quantum: limits.reactor_work_quantum,
1521 bridge_call_timeout_ms: limits.bridge_call_timeout_ms,
1522 ..JavascriptExecutionLimits::default()
1523 }
1524}
1525
1526fn build_python_internal_env(
1527 import_cache: &NodeImportCache,
1528 context: &PythonContext,
1529 request: &StartPythonExecutionRequest,
1530 frozen_time_ms: u128,
1531 prewarm_only: bool,
1532) -> BTreeMap<String, String> {
1533 let mut internal_env = request
1534 .env
1535 .iter()
1536 .filter(|(key, _)| key.starts_with("AGENTOS_"))
1537 .map(|(key, value)| (key.clone(), value.clone()))
1538 .collect::<BTreeMap<_, _>>();
1539 let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd);
1540
1541 add_python_guest_path_mapping(&mut internal_env, &pyodide_dist_path);
1542
1543 internal_env.insert(
1544 PYODIDE_INDEX_URL_ENV.to_string(),
1545 String::from(PYODIDE_GUEST_ROOT),
1546 );
1547 internal_env.insert(
1548 PYODIDE_PACKAGE_BASE_URL_ENV.to_string(),
1549 request
1550 .env
1551 .get(PYODIDE_PACKAGE_BASE_URL_ENV)
1552 .cloned()
1553 .unwrap_or_else(|| String::from(PYODIDE_GUEST_ROOT)),
1554 );
1555 internal_env.insert(
1556 PYODIDE_PACKAGE_CACHE_DIR_ENV.to_string(),
1557 String::from(PYODIDE_CACHE_GUEST_ROOT),
1558 );
1559 internal_env.insert(
1560 NODE_IMPORT_CACHE_ASSET_ROOT_ENV.to_string(),
1561 import_cache.asset_root().display().to_string(),
1562 );
1563 internal_env.insert(
1564 NODE_ALLOW_PROCESS_BINDINGS_ENV.to_string(),
1565 String::from("1"),
1566 );
1567 internal_env.insert(
1568 NODE_SYNC_RPC_DATA_BYTES_ENV.to_string(),
1569 PYTHON_SYNC_RPC_DATA_BYTES.to_string(),
1570 );
1571 internal_env.insert(
1572 NODE_DISABLE_COMPILE_CACHE_ENV.to_string(),
1573 String::from("1"),
1574 );
1575 internal_env.insert(PYTHON_CODE_ENV.to_string(), request.code.clone());
1580 internal_env.insert(NODE_FROZEN_TIME_ENV.to_string(), frozen_time_ms.to_string());
1581 if prewarm_only {
1582 internal_env.insert(PYTHON_PREWARM_ONLY_ENV.to_string(), String::from("1"));
1583 } else {
1584 internal_env.insert(String::from("AGENTOS_KEEP_STDIN_OPEN"), String::from("1"));
1585 internal_env.insert(
1586 String::from("AGENTOS_FORWARD_KERNEL_STDIN_RPC"),
1587 String::from("1"),
1588 );
1589 internal_env.remove(PYTHON_PREWARM_ONLY_ENV);
1590 }
1591 if let Some(file_path) = &request.file_path {
1592 internal_env.insert(PYTHON_FILE_ENV.to_string(), file_path.display().to_string());
1593 } else {
1594 internal_env.remove(PYTHON_FILE_ENV);
1595 }
1596
1597 internal_env
1598}
1599
1600fn add_python_guest_path_mapping(
1601 internal_env: &mut BTreeMap<String, String>,
1602 pyodide_dist_path: &Path,
1603) {
1604 let pyodide_cache_path = pyodide_cache_path(pyodide_dist_path);
1605 let mut mappings = internal_env
1606 .get(NODE_GUEST_PATH_MAPPINGS_ENV)
1607 .and_then(|value| serde_json::from_str::<Vec<PythonGuestPathMappingWire>>(value).ok())
1608 .unwrap_or_default();
1609
1610 mappings.retain(|mapping| {
1611 mapping.guest_path != PYODIDE_GUEST_ROOT && mapping.guest_path != PYODIDE_CACHE_GUEST_ROOT
1612 });
1613 mappings.push(PythonGuestPathMappingWire {
1614 guest_path: String::from(PYODIDE_GUEST_ROOT),
1615 host_path: pyodide_dist_path.display().to_string(),
1616 });
1617 mappings.push(PythonGuestPathMappingWire {
1618 guest_path: String::from(PYODIDE_CACHE_GUEST_ROOT),
1619 host_path: pyodide_cache_path.display().to_string(),
1620 });
1621
1622 let serialized = serde_json::to_string(&mappings).unwrap_or_else(|_| String::from("[]"));
1623 internal_env.insert(String::from(NODE_GUEST_PATH_MAPPINGS_ENV), serialized);
1624}
1625
1626fn pyodide_cache_path(pyodide_dist_path: &Path) -> PathBuf {
1627 let base = pyodide_dist_path
1628 .parent()
1629 .and_then(|parent| {
1630 if parent.file_name().is_some_and(|name| name == "assets") {
1631 parent.parent()
1632 } else {
1633 Some(parent)
1634 }
1635 })
1636 .unwrap_or(pyodide_dist_path);
1637
1638 base.join("pyodide-package-cache")
1639}
1640
1641fn build_python_runner_module_source(
1642 import_cache: &NodeImportCache,
1643 internal_env: &BTreeMap<String, String>,
1644 warmup_metrics: Option<&[u8]>,
1645) -> Result<String, PythonExecutionError> {
1646 let runner_source = fs::read_to_string(import_cache.python_runner_path())
1647 .map_err(PythonExecutionError::PrepareRuntime)?;
1648 let runner_source =
1649 format!("import * as __agentOSConstantsBinding from 'node:constants';\n{runner_source}");
1650 let bootstrap = build_python_runner_bootstrap(internal_env, warmup_metrics);
1651 Ok(insert_python_runner_bootstrap(&runner_source, &bootstrap))
1652}
1653
1654fn build_python_runner_bootstrap(
1655 internal_env: &BTreeMap<String, String>,
1656 warmup_metrics: Option<&[u8]>,
1657) -> String {
1658 let internal_env_json =
1659 serde_json::to_string(internal_env).unwrap_or_else(|_| String::from("{}"));
1660 let warmup_metrics_json = warmup_metrics.map(|bytes| {
1661 serde_json::to_string(&String::from_utf8_lossy(bytes).to_string())
1662 .unwrap_or_else(|_| String::from("\"\""))
1663 });
1664
1665 match warmup_metrics_json {
1666 Some(warmup_metrics_json) => format!(
1667 "globalThis.__agentOSPythonInternalEnv = {internal_env_json};\n\
1668if (typeof process !== 'undefined') {{\n process.env = {{ ...(process.env || {{}}), ...globalThis.__agentOSPythonInternalEnv }};\n}}\n\
1669if (typeof process?.stderr?.write === 'function') {{\n process.stderr.write({warmup_metrics_json});\n}}\n"
1670 ),
1671 None => format!(
1672 "globalThis.__agentOSPythonInternalEnv = {internal_env_json};\n\
1673if (typeof process !== 'undefined') {{\n process.env = {{ ...(process.env || {{}}), ...globalThis.__agentOSPythonInternalEnv }};\n}}\n"
1674 ),
1675 }
1676}
1677
1678fn insert_python_runner_bootstrap(source: &str, bootstrap: &str) -> String {
1679 let mut insert_at = 0usize;
1680 let mut saw_import = false;
1681 for line in source.split_inclusive('\n') {
1682 let trimmed = line.trim_start();
1683 if trimmed.starts_with("import ") || (saw_import && trimmed.is_empty()) {
1684 insert_at += line.len();
1685 saw_import = saw_import || trimmed.starts_with("import ");
1686 continue;
1687 }
1688 break;
1689 }
1690
1691 format!(
1692 "{}{}{}",
1693 &source[..insert_at],
1694 bootstrap,
1695 &source[insert_at..]
1696 )
1697}
1698
1699fn parse_python_bridge_sync_rpc_request(
1700 request: &JavascriptSyncRpcRequest,
1701) -> Result<PythonVfsRpcRequest, PythonExecutionError> {
1702 if request.method != "_pythonRpc" {
1703 return Err(PythonExecutionError::RpcResponse(format!(
1704 "unexpected JavaScript sync RPC method for guest Python runtime: {}",
1705 request.method
1706 )));
1707 }
1708
1709 let payload = request.args.first().ok_or_else(|| {
1710 PythonExecutionError::RpcResponse(String::from(
1711 "guest Python bridge call did not include a request payload",
1712 ))
1713 })?;
1714
1715 let wire: PythonVfsBridgeRequestWire =
1716 serde_json::from_value(payload.clone()).map_err(|error| {
1717 PythonExecutionError::RpcResponse(format!(
1718 "invalid guest Python bridge request payload: {error}"
1719 ))
1720 })?;
1721
1722 let method = PythonVfsRpcMethod::from_wire(&wire.method).ok_or_else(|| {
1723 PythonExecutionError::RpcResponse(format!(
1724 "unsupported agentos python rpc method {} for {}",
1725 wire.method, request.id
1726 ))
1727 })?;
1728
1729 Ok(PythonVfsRpcRequest {
1730 id: request.id,
1731 method,
1732 path: wire.path,
1733 destination: wire.destination,
1734 target: wire.target,
1735 mode: wire.mode.map(|value| value as u32),
1736 uid: wire.uid.map(|value| value as u32),
1737 gid: wire.gid.map(|value| value as u32),
1738 atime_ms: wire.atime_ms.map(|value| value as u64),
1739 mtime_ms: wire.mtime_ms.map(|value| value as u64),
1740 content_base64: wire.content_base64,
1741 recursive: wire.recursive,
1742 url: wire.url,
1743 http_method: wire.http_method,
1744 headers: wire.headers,
1745 body_base64: wire.body_base64,
1746 hostname: wire.hostname,
1747 family: wire.family,
1748 port: wire.port,
1749 socket_id: wire.socket_id,
1750 command: wire.command,
1751 args: wire.args,
1752 argv0: wire.argv0,
1753 cwd: wire.cwd,
1754 env: wire.env,
1755 shell: wire.shell,
1756 max_buffer: wire.max_buffer,
1757 timeout_ms: wire.timeout_ms,
1758 })
1759}
1760
1761#[derive(Debug)]
1762struct PythonOutputBuffer {
1763 bytes: Vec<u8>,
1764 max_bytes: usize,
1765}
1766
1767impl PythonOutputBuffer {
1768 fn new(max_bytes: usize) -> Self {
1769 Self {
1770 bytes: Vec::new(),
1771 max_bytes,
1772 }
1773 }
1774
1775 fn extend(&mut self, chunk: &[u8]) {
1776 if self.bytes.len() >= self.max_bytes {
1777 return;
1778 }
1779
1780 let remaining = self.max_bytes - self.bytes.len();
1781 let take = remaining.min(chunk.len());
1782 self.bytes.extend_from_slice(&chunk[..take]);
1783 }
1784
1785 fn into_inner(self) -> Vec<u8> {
1786 self.bytes
1787 }
1788}
1789
1790fn python_output_buffer_max_bytes(request: &StartPythonExecutionRequest) -> usize {
1791 request
1792 .limits
1793 .output_buffer_max_bytes
1794 .unwrap_or(DEFAULT_PYTHON_OUTPUT_BUFFER_MAX_BYTES)
1795}
1796
1797fn python_execution_timeout(request: &StartPythonExecutionRequest) -> Option<Duration> {
1798 match request.limits.execution_timeout_ms {
1799 Some(0) => None,
1801 Some(value) => Some(Duration::from_millis(value)),
1802 None => Some(Duration::from_millis(DEFAULT_PYTHON_EXECUTION_TIMEOUT_MS)),
1803 }
1804}
1805
1806fn python_max_old_space_mb(request: &StartPythonExecutionRequest) -> usize {
1807 request
1808 .limits
1809 .max_old_space_mb
1810 .filter(|value| *value > 0)
1811 .unwrap_or(DEFAULT_PYTHON_MAX_OLD_SPACE_MB)
1812}
1813
1814fn python_vfs_rpc_timeout(request: &StartPythonExecutionRequest) -> Duration {
1815 Duration::from_millis(
1816 request
1817 .limits
1818 .vfs_rpc_timeout_ms
1819 .filter(|value| *value > 0)
1820 .unwrap_or(DEFAULT_PYTHON_VFS_RPC_TIMEOUT_MS),
1821 )
1822}
1823
1824fn spawn_python_vfs_rpc_timeout(
1825 runtime: &RuntimeContext,
1826 id: u64,
1827 timeout: Duration,
1828 pending: Arc<Mutex<Option<PendingVfsRpc>>>,
1829 v8_session: crate::v8_host::V8SessionHandle,
1830) -> Result<(), PythonExecutionError> {
1831 let cancellation = runtime.clone();
1832 let pending_for_task = Arc::clone(&pending);
1833 let handle = runtime
1834 .spawn(agentos_runtime::TaskClass::Timer, async move {
1835 tokio::select! {
1836 _ = tokio::time::sleep(timeout) => {}
1837 _ = cancellation.admission_closed() => {
1838 let mut guard = pending_for_task.lock().unwrap_or_else(|poisoned| {
1839 eprintln!(
1840 "ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: recovering request {id} during runtime shutdown"
1841 );
1842 poisoned.into_inner()
1843 });
1844 if guard.as_ref().map(|rpc| rpc.state)
1845 == Some(PendingVfsRpcState::Pending(id))
1846 {
1847 *guard = None;
1848 }
1849 return;
1850 }
1851 }
1852
1853 let mut guard = pending_for_task.lock().unwrap_or_else(|poisoned| {
1854 eprintln!(
1855 "ERR_AGENTOS_PYTHON_VFS_RPC_STATE_POISONED: recovering request {id} while delivering its timeout"
1856 );
1857 poisoned.into_inner()
1858 });
1859 let should_timeout =
1860 if guard.as_ref().map(|rpc| rpc.state) == Some(PendingVfsRpcState::Pending(id)) {
1861 *guard = Some(PendingVfsRpc {
1862 state: PendingVfsRpcState::TimedOut(id),
1863 timeout_abort: None,
1864 });
1865 true
1866 } else {
1867 false
1868 };
1869 drop(guard);
1870
1871 if !should_timeout {
1872 return;
1873 }
1874
1875 if let Err(error) = v8_session.send_bridge_response(
1876 id,
1877 1,
1878 format!(
1879 "ERR_AGENTOS_PYTHON_VFS_RPC_TIMEOUT: guest Python VFS RPC request {id} timed out after {}ms",
1880 timeout.as_millis()
1881 )
1882 .into_bytes(),
1883 ) {
1884 eprintln!(
1885 "ERR_AGENTOS_PYTHON_VFS_RPC_TIMEOUT_DELIVERY: could not deliver timeout for request {id}: {error}"
1886 );
1887 }
1888 })
1889 .map_err(|error| {
1890 PythonExecutionError::RpcResponse(format!(
1891 "could not arm Python VFS RPC timeout for request {id}: {error}"
1892 ))
1893 })?;
1894
1895 let timeout_abort = handle.abort_handle();
1896 let mut guard = pending
1897 .lock()
1898 .map_err(|_| PythonExecutionError::EventChannelClosed)?;
1899 if let Some(rpc) = guard.as_mut() {
1900 if rpc.state == PendingVfsRpcState::Pending(id) {
1901 rpc.timeout_abort = Some(timeout_abort);
1902 return Ok(());
1903 }
1904 }
1905 timeout_abort.abort();
1906 Ok(())
1907}
1908
1909fn resolved_pyodide_dist_path(path: &Path, cwd: &Path) -> PathBuf {
1910 resolve_execution_path(path, cwd)
1911}
1912
1913struct PythonPrewarmOutput {
1914 stdout: Vec<u8>,
1915 stderr: Vec<u8>,
1916 sync_rpc_log: Vec<String>,
1917 managed_host_files: PythonManagedHostFiles,
1918}
1919
1920impl PythonPrewarmOutput {
1921 fn new(max_open_fds: usize) -> Self {
1922 Self {
1923 stdout: Vec::new(),
1924 stderr: Vec::new(),
1925 sync_rpc_log: Vec::new(),
1926 managed_host_files: PythonManagedHostFiles::new(max_open_fds),
1927 }
1928 }
1929}
1930
1931fn handle_python_prewarm_event(
1932 prewarm_execution: &mut JavascriptExecution,
1933 context: &PythonContext,
1934 request: &StartPythonExecutionRequest,
1935 event: Option<JavascriptExecutionEvent>,
1936 output: &mut PythonPrewarmOutput,
1937) -> Result<Option<PythonExecutionResult>, PythonExecutionError> {
1938 match event {
1939 Some(JavascriptExecutionEvent::Stdout(chunk)) => output.stdout.extend(chunk),
1940 Some(JavascriptExecutionEvent::Stderr(chunk)) => output.stderr.extend(chunk),
1941 Some(JavascriptExecutionEvent::Exited(exit_code)) => {
1942 return Ok(Some(PythonExecutionResult {
1943 execution_id: String::from("python-prewarm"),
1944 exit_code,
1945 stdout: std::mem::take(&mut output.stdout),
1946 stderr: std::mem::take(&mut output.stderr),
1947 }));
1948 }
1949 Some(JavascriptExecutionEvent::SignalState { .. }) => {}
1950 Some(JavascriptExecutionEvent::SyncRpcRequest(sync_request)) => {
1951 output.sync_rpc_log.push(format!(
1952 "{} {} {:?}",
1953 sync_request.id, sync_request.method, sync_request.args
1954 ));
1955 if prewarm_execution
1959 .try_service_standalone_module_sync_rpc(&sync_request)
1960 .map_err(map_javascript_error)?
1961 {
1962 output
1963 .sync_rpc_log
1964 .push(format!("responded {} (module)", sync_request.id));
1965 return Ok(None);
1966 }
1967 let pyodide_dist_path =
1968 resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd);
1969 if let Some(action) = python_javascript_sync_rpc_action(
1970 &pyodide_dist_path,
1971 &mut output.managed_host_files,
1972 &sync_request,
1973 )? {
1974 respond_python_javascript_sync_rpc_action(
1975 prewarm_execution,
1976 sync_request.id,
1977 action,
1978 )?;
1979 output
1980 .sync_rpc_log
1981 .push(format!("responded {}", sync_request.id));
1982 return Ok(None);
1983 }
1984 if let Some((code, message)) = python_javascript_sync_rpc_error(&sync_request) {
1985 prewarm_execution
1986 .respond_sync_rpc_error(sync_request.id, code, message)
1987 .map_err(map_javascript_error)?;
1988 output
1989 .sync_rpc_log
1990 .push(format!("errored {}", sync_request.id));
1991 return Ok(None);
1992 }
1993 if sync_request.method == "_pythonRpc" {
1994 let request = parse_python_bridge_sync_rpc_request(&sync_request)?;
1995 return Err(PythonExecutionError::WarmupFailed {
1996 exit_code: 1,
1997 stderr: format!(
1998 "unexpected Python prewarm VFS RPC request {} {} {:?}",
1999 request.id, request.path, request.method
2000 ),
2001 });
2002 }
2003 return Err(PythonExecutionError::WarmupFailed {
2004 exit_code: 1,
2005 stderr: format!(
2006 "unexpected Python prewarm JavaScript sync RPC request {} {} {:?}",
2007 sync_request.id, sync_request.method, sync_request.args
2008 ),
2009 });
2010 }
2011 None => {
2012 return Err(PythonExecutionError::WarmupFailed {
2013 exit_code: 1,
2014 stderr: format!(
2015 "python prewarm timed out after {}s\nstdout:\n{}\nstderr:\n{}\nsync rpc:\n{}",
2016 PYTHON_PREWARM_TIMEOUT.as_secs(),
2017 String::from_utf8_lossy(&output.stdout),
2018 String::from_utf8_lossy(&output.stderr),
2019 output.sync_rpc_log.join("\n"),
2020 ),
2021 });
2022 }
2023 }
2024 Ok(None)
2025}
2026
2027fn prewarm_python_path(
2028 import_cache: &NodeImportCache,
2029 javascript_engine: &mut JavascriptExecutionEngine,
2030 javascript_context_id: &str,
2031 context: &PythonContext,
2032 request: &StartPythonExecutionRequest,
2033 frozen_time_ms: u128,
2034 runtime: &RuntimeContext,
2035) -> Result<Option<Vec<u8>>, PythonExecutionError> {
2036 let debug_enabled = python_warmup_metrics_enabled(request);
2037 let marker_contents = warmup_marker_contents(import_cache, context, request);
2038 let marker_path = warmup_marker_path(
2039 import_cache.prewarm_marker_dir(),
2040 "python-runner-prewarm",
2041 PYTHON_WARMUP_MARKER_VERSION,
2042 &marker_contents,
2043 );
2044 let marker_exists = marker_path.exists();
2045
2046 let started = Instant::now();
2047 let mut prewarm_execution = start_python_javascript_execution(
2048 javascript_engine,
2049 runtime,
2050 import_cache,
2051 javascript_context_id,
2052 context,
2053 request,
2054 PythonJavascriptExecutionOptions {
2055 frozen_time_ms,
2056 prewarm_only: true,
2057 warmup_metrics: None,
2058 defer_execute: false,
2059 },
2060 )?;
2061 let mut output = PythonPrewarmOutput::new(python_managed_host_file_limit(request));
2062 let result = loop {
2063 let event = prewarm_execution
2064 .poll_event_blocking(PYTHON_PREWARM_TIMEOUT)
2065 .map_err(map_javascript_error)?;
2066 if let Some(result) = handle_python_prewarm_event(
2067 &mut prewarm_execution,
2068 context,
2069 request,
2070 event,
2071 &mut output,
2072 )? {
2073 break result;
2074 }
2075 };
2076 let duration_ms = started.elapsed().as_secs_f64() * 1000.0;
2077
2078 if result.exit_code != 0 {
2079 return Err(PythonExecutionError::WarmupFailed {
2080 exit_code: result.exit_code,
2081 stderr: String::from_utf8_lossy(&result.stderr).into_owned(),
2082 });
2083 }
2084
2085 if marker_exists {
2086 return Ok(warmup_metrics_line(
2087 debug_enabled,
2088 false,
2089 "cached",
2090 0.0,
2091 import_cache,
2092 context,
2093 request,
2094 ));
2095 }
2096
2097 fs::write(&marker_path, marker_contents).map_err(PythonExecutionError::PrepareWarmPath)?;
2098 Ok(warmup_metrics_line(
2099 debug_enabled,
2100 true,
2101 "executed",
2102 duration_ms,
2103 import_cache,
2104 context,
2105 request,
2106 ))
2107}
2108
2109async fn prewarm_python_path_async(
2110 import_cache: &NodeImportCache,
2111 javascript_engine: &mut JavascriptExecutionEngine,
2112 javascript_context_id: &str,
2113 context: &PythonContext,
2114 request: &StartPythonExecutionRequest,
2115 frozen_time_ms: u128,
2116 runtime: &RuntimeContext,
2117) -> Result<Option<Vec<u8>>, PythonExecutionError> {
2118 let debug_enabled = python_warmup_metrics_enabled(request);
2119 let marker_contents = warmup_marker_contents(import_cache, context, request);
2120 let marker_path = warmup_marker_path(
2121 import_cache.prewarm_marker_dir(),
2122 "python-runner-prewarm",
2123 PYTHON_WARMUP_MARKER_VERSION,
2124 &marker_contents,
2125 );
2126 let marker_exists = marker_path.exists();
2127
2128 let started = Instant::now();
2129 let mut prewarm_execution = start_python_javascript_execution(
2130 javascript_engine,
2131 runtime,
2132 import_cache,
2133 javascript_context_id,
2134 context,
2135 request,
2136 PythonJavascriptExecutionOptions {
2137 frozen_time_ms,
2138 prewarm_only: true,
2139 warmup_metrics: None,
2140 defer_execute: false,
2141 },
2142 )?;
2143 let mut output = PythonPrewarmOutput::new(python_managed_host_file_limit(request));
2144 let result = loop {
2145 let event = prewarm_execution
2146 .poll_event(PYTHON_PREWARM_TIMEOUT)
2147 .await
2148 .map_err(map_javascript_error)?;
2149 if let Some(result) = handle_python_prewarm_event(
2150 &mut prewarm_execution,
2151 context,
2152 request,
2153 event,
2154 &mut output,
2155 )? {
2156 break result;
2157 }
2158 };
2159 let duration_ms = started.elapsed().as_secs_f64() * 1000.0;
2160
2161 if result.exit_code != 0 {
2162 return Err(PythonExecutionError::WarmupFailed {
2163 exit_code: result.exit_code,
2164 stderr: String::from_utf8_lossy(&result.stderr).into_owned(),
2165 });
2166 }
2167
2168 if marker_exists {
2169 return Ok(warmup_metrics_line(
2170 debug_enabled,
2171 false,
2172 "cached",
2173 0.0,
2174 import_cache,
2175 context,
2176 request,
2177 ));
2178 }
2179
2180 fs::write(&marker_path, marker_contents).map_err(PythonExecutionError::PrepareWarmPath)?;
2181 Ok(warmup_metrics_line(
2182 debug_enabled,
2183 true,
2184 "executed",
2185 duration_ms,
2186 import_cache,
2187 context,
2188 request,
2189 ))
2190}
2191
2192#[derive(Debug)]
2193enum PythonJavascriptSyncRpcAction {
2194 Success(Value),
2195 RawSuccess(Vec<u8>),
2196 Error { code: &'static str, message: String },
2197}
2198
2199#[derive(Debug)]
2200struct PythonManagedHostFiles {
2201 next_fd: u64,
2202 max_files: usize,
2203 files: BTreeMap<u64, fs::File>,
2204}
2205
2206const DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT: usize = 256;
2210
2211impl Default for PythonManagedHostFiles {
2212 fn default() -> Self {
2213 Self::new(DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT)
2214 }
2215}
2216
2217impl PythonManagedHostFiles {
2218 fn new(max_files: usize) -> Self {
2219 Self {
2220 next_fd: 0x4000_0000,
2223 max_files,
2224 files: BTreeMap::new(),
2225 }
2226 }
2227}
2228
2229fn python_managed_host_file_limit(request: &StartPythonExecutionRequest) -> usize {
2230 request
2231 .limits
2232 .max_open_fds
2233 .unwrap_or(DEFAULT_PYTHON_MANAGED_HOST_FILE_LIMIT)
2234}
2235
2236fn python_javascript_sync_rpc_action(
2237 pyodide_dist_path: &Path,
2238 managed_host_files: &mut PythonManagedHostFiles,
2239 request: &JavascriptSyncRpcRequest,
2240) -> Result<Option<PythonJavascriptSyncRpcAction>, PythonExecutionError> {
2241 if matches!(request.method.as_str(), "fs.readSync" | "_fsReadRaw") {
2242 let Some(fd) = request.args.first().and_then(Value::as_u64) else {
2243 return Ok(None);
2244 };
2245 let Some(file) = managed_host_files.files.get_mut(&fd) else {
2246 return Ok(None);
2247 };
2248 let length = request
2249 .args
2250 .get(1)
2251 .and_then(Value::as_u64)
2252 .and_then(|length| usize::try_from(length).ok())
2253 .ok_or_else(|| {
2254 PythonExecutionError::RpcResponse(String::from(
2255 "managed fs.readSync length must fit within usize",
2256 ))
2257 })?;
2258 if length > PYTHON_SYNC_RPC_DATA_BYTES {
2259 return Err(PythonExecutionError::RpcResponse(format!(
2260 "managed fs.readSync length {length} exceeds {PYTHON_SYNC_RPC_DATA_BYTES} bytes"
2261 )));
2262 }
2263 if let Some(position) = request.args.get(2).and_then(Value::as_u64) {
2264 file.seek(SeekFrom::Start(position))
2265 .map_err(PythonExecutionError::PrepareRuntime)?;
2266 }
2267 let mut bytes = vec![0; length];
2268 let bytes_read = file
2269 .read(&mut bytes)
2270 .map_err(PythonExecutionError::PrepareRuntime)?;
2271 bytes.truncate(bytes_read);
2272 return Ok(Some(if request.raw_bytes_args.contains_key(&usize::MAX) {
2273 PythonJavascriptSyncRpcAction::RawSuccess(bytes)
2274 } else {
2275 PythonJavascriptSyncRpcAction::Success(Value::String(v8_runtime::base64_encode_pub(
2276 &bytes,
2277 )))
2278 }));
2279 }
2280
2281 if request.method == "fs.closeSync" {
2282 let Some(fd) = request.args.first().and_then(Value::as_u64) else {
2283 return Ok(None);
2284 };
2285 if managed_host_files.files.remove(&fd).is_none() {
2286 return Ok(None);
2287 }
2288 return Ok(Some(PythonJavascriptSyncRpcAction::Success(Value::Null)));
2289 }
2290
2291 if request.method == "fs.fstatSync" {
2292 let Some(fd) = request.args.first().and_then(Value::as_u64) else {
2293 return Ok(None);
2294 };
2295 let Some(file) = managed_host_files.files.get(&fd) else {
2296 return Ok(None);
2297 };
2298 let metadata = file
2299 .metadata()
2300 .map_err(PythonExecutionError::PrepareRuntime)?;
2301 return Ok(Some(PythonJavascriptSyncRpcAction::Success(
2302 python_host_stat_value(&metadata),
2303 )));
2304 }
2305
2306 let Some(path) = request.args.first().and_then(Value::as_str) else {
2307 return Ok(None);
2308 };
2309 let path_kind = python_managed_path_kind(pyodide_dist_path, path);
2310 let Some(host_path) = path_kind.host_path() else {
2311 return Ok(None);
2312 };
2313
2314 Ok(Some(match request.method.as_str() {
2315 "fs.openSync" => {
2316 let flags = request.args.get(1).unwrap_or(&Value::Null);
2317 let read_only = matches!(flags.as_str(), Some("r"))
2318 || flags.as_u64().is_some_and(|flags| flags == 0);
2319 if !read_only {
2320 PythonJavascriptSyncRpcAction::Error {
2321 code: "EROFS",
2322 message: format!(
2323 "EROFS: managed Python runtime assets are read-only, open '{path}'"
2324 ),
2325 }
2326 } else if managed_host_files.files.len() >= managed_host_files.max_files {
2327 PythonJavascriptSyncRpcAction::Error {
2328 code: "EMFILE",
2329 message: format!(
2330 "EMFILE: managed Python host descriptor limit {} reached (limits.resources.maxOpenFds); raise limits.resources.maxOpenFds",
2331 managed_host_files.max_files
2332 ),
2333 }
2334 } else {
2335 match fs::File::open(&host_path) {
2336 Ok(file) => {
2337 let fd = managed_host_files.next_fd;
2338 managed_host_files.next_fd =
2339 managed_host_files.next_fd.checked_add(1).ok_or_else(|| {
2340 PythonExecutionError::RpcResponse(String::from(
2341 "managed Python host descriptor ids exhausted",
2342 ))
2343 })?;
2344 managed_host_files.files.insert(fd, file);
2345 PythonJavascriptSyncRpcAction::Success(json!(fd))
2346 }
2347 Err(error) => {
2348 return python_sync_rpc_fs_action_error(path, "open", error).map(Some);
2349 }
2350 }
2351 }
2352 }
2353 "fs.promises.readFile" | "fs.readFileSync" => {
2354 let bytes = match fs::read(&host_path) {
2355 Ok(bytes) => bytes,
2356 Err(error) => {
2357 return python_sync_rpc_fs_action_error(path, "open", error).map(Some);
2358 }
2359 };
2360 let encoding = python_prewarm_sync_rpc_encoding(&request.args);
2361 match encoding.as_deref() {
2362 Some("utf8") | Some("utf-8") => PythonJavascriptSyncRpcAction::Success(
2363 Value::String(String::from_utf8_lossy(&bytes).into_owned()),
2364 ),
2365 _ => PythonJavascriptSyncRpcAction::Success(json!({
2366 "__agentOSType": "bytes",
2367 "base64": v8_runtime::base64_encode_pub(&bytes),
2368 })),
2369 }
2370 }
2371 "fs.statSync" | "fs.promises.stat" => match fs::metadata(&host_path) {
2372 Ok(metadata) => {
2373 PythonJavascriptSyncRpcAction::Success(python_host_stat_value(&metadata))
2374 }
2375 Err(error) => return python_sync_rpc_fs_action_error(path, "stat", error).map(Some),
2376 },
2377 "fs.lstatSync" | "fs.promises.lstat" => match fs::symlink_metadata(&host_path) {
2378 Ok(metadata) => {
2379 PythonJavascriptSyncRpcAction::Success(python_host_stat_value(&metadata))
2380 }
2381 Err(error) => return python_sync_rpc_fs_action_error(path, "lstat", error).map(Some),
2382 },
2383 "fs.existsSync" => PythonJavascriptSyncRpcAction::Success(Value::Bool(host_path.exists())),
2384 "fs.accessSync" | "fs.promises.access" => match fs::metadata(&host_path) {
2385 Ok(_) => PythonJavascriptSyncRpcAction::Success(Value::Null),
2386 Err(error) => return python_sync_rpc_fs_action_error(path, "access", error).map(Some),
2387 },
2388 "fs.readdirSync" | "fs.promises.readdir" => match fs::read_dir(&host_path) {
2389 Ok(entries) => PythonJavascriptSyncRpcAction::Success(python_readdir_value(
2390 entries
2391 .filter_map(|entry| entry.ok())
2392 .filter_map(|entry| entry.file_name().into_string().ok())
2393 .collect(),
2394 )),
2395 Err(error) => return python_sync_rpc_fs_action_error(path, "scandir", error).map(Some),
2396 },
2397 "fs.mkdirSync" | "fs.promises.mkdir" => {
2398 let recursive = python_sync_rpc_recursive_flag(&request.args);
2399 if recursive {
2400 fs::create_dir_all(&host_path).map_err(PythonExecutionError::PrepareRuntime)?;
2401 } else {
2402 match fs::create_dir(&host_path) {
2403 Ok(()) => {}
2404 Err(error) => {
2405 return python_sync_rpc_fs_action_error(path, "mkdir", error).map(Some);
2406 }
2407 }
2408 }
2409 PythonJavascriptSyncRpcAction::Success(Value::Null)
2410 }
2411 "fs.writeFileSync" | "fs.promises.writeFile" => {
2412 let contents = python_sync_rpc_bytes_arg(&request.args, 1)?;
2413 if let Some(parent) = host_path.parent() {
2414 fs::create_dir_all(parent).map_err(PythonExecutionError::PrepareRuntime)?;
2415 }
2416 fs::write(&host_path, contents).map_err(PythonExecutionError::PrepareRuntime)?;
2417 PythonJavascriptSyncRpcAction::Success(Value::Null)
2418 }
2419 "fs.realpathSync" | "fs.realpathSync.native" => match fs::canonicalize(&host_path) {
2420 Ok(canonical) => PythonJavascriptSyncRpcAction::Success(Value::String(
2421 path_kind.render_path(pyodide_dist_path, &canonical, path),
2422 )),
2423 Err(error) => {
2424 return python_sync_rpc_fs_action_error(path, "realpath", error).map(Some);
2425 }
2426 },
2427 _ => return Ok(None),
2428 }))
2429}
2430
2431fn python_sync_rpc_fs_action_error(
2432 path: &str,
2433 syscall: &str,
2434 error: std::io::Error,
2435) -> Result<PythonJavascriptSyncRpcAction, PythonExecutionError> {
2436 let action = match error.kind() {
2437 std::io::ErrorKind::NotFound => PythonJavascriptSyncRpcAction::Error {
2438 code: "ENOENT",
2439 message: format!("ENOENT: no such file or directory, {syscall} '{path}'"),
2440 },
2441 std::io::ErrorKind::AlreadyExists => PythonJavascriptSyncRpcAction::Error {
2442 code: "EEXIST",
2443 message: format!("EEXIST: file already exists, {syscall} '{path}'"),
2444 },
2445 std::io::ErrorKind::PermissionDenied => PythonJavascriptSyncRpcAction::Error {
2446 code: "EACCES",
2447 message: format!("EACCES: permission denied, {syscall} '{path}'"),
2448 },
2449 _ => {
2450 return Err(PythonExecutionError::PrepareRuntime(std::io::Error::new(
2451 error.kind(),
2452 error.to_string(),
2453 )));
2454 }
2455 };
2456 Ok(action)
2457}
2458
2459fn respond_python_javascript_sync_rpc_action(
2460 execution: &mut JavascriptExecution,
2461 id: u64,
2462 action: PythonJavascriptSyncRpcAction,
2463) -> Result<(), PythonExecutionError> {
2464 match action {
2465 PythonJavascriptSyncRpcAction::Success(value) => execution
2466 .respond_sync_rpc_success(id, value)
2467 .map_err(map_javascript_error),
2468 PythonJavascriptSyncRpcAction::RawSuccess(bytes) => execution
2469 .respond_sync_rpc_raw_success(id, bytes)
2470 .map_err(map_javascript_error),
2471 PythonJavascriptSyncRpcAction::Error { code, message } => execution
2472 .respond_sync_rpc_error(id, code, message)
2473 .map_err(map_javascript_error),
2474 }
2475}
2476
2477#[derive(Debug, Clone)]
2478enum PythonManagedPathKind {
2479 GuestPyodide,
2480 GuestCache,
2481 HostManaged,
2482 Unmanaged,
2483}
2484
2485impl PythonManagedPathKind {
2486 fn render_path(&self, pyodide_dist_path: &Path, canonical: &Path, original: &str) -> String {
2487 match self {
2488 Self::GuestPyodide | Self::GuestCache => {
2489 python_host_path_to_guest(pyodide_dist_path, canonical)
2490 .unwrap_or_else(|| original.to_owned())
2491 }
2492 Self::HostManaged => canonical.display().to_string(),
2493 Self::Unmanaged => original.to_owned(),
2494 }
2495 }
2496}
2497
2498fn python_managed_path_kind(pyodide_dist_path: &Path, path: &str) -> PythonManagedResolvedPath {
2499 let cache_path = pyodide_cache_path(pyodide_dist_path);
2500
2501 if let Some(normalized) = strip_guest_managed_root(path, PYODIDE_GUEST_ROOT) {
2502 let root = canonicalize_existing_or_self(pyodide_dist_path);
2503 let relative = normalize_relative_guest_suffix(normalized);
2504 let host_path = if relative.as_os_str().is_empty() {
2505 root.clone()
2506 } else {
2507 root.join(relative)
2508 };
2509 if confined_managed_path(&host_path, &root) {
2510 return PythonManagedResolvedPath {
2511 kind: PythonManagedPathKind::GuestPyodide,
2512 host_path: Some(host_path),
2513 };
2514 }
2515 return PythonManagedResolvedPath {
2516 kind: PythonManagedPathKind::Unmanaged,
2517 host_path: None,
2518 };
2519 }
2520
2521 if let Some(normalized) = strip_guest_managed_root(path, PYODIDE_CACHE_GUEST_ROOT) {
2522 let root = canonicalize_existing_or_self(&cache_path);
2523 let relative = normalize_relative_guest_suffix(normalized);
2524 let host_path = if relative.as_os_str().is_empty() {
2525 root.clone()
2526 } else {
2527 root.join(relative)
2528 };
2529 if confined_managed_path(&host_path, &root) {
2530 return PythonManagedResolvedPath {
2531 kind: PythonManagedPathKind::GuestCache,
2532 host_path: Some(host_path),
2533 };
2534 }
2535 return PythonManagedResolvedPath {
2536 kind: PythonManagedPathKind::Unmanaged,
2537 host_path: None,
2538 };
2539 }
2540
2541 let candidate = PathBuf::from(path);
2542 let pyodide_root = canonicalize_existing_or_self(pyodide_dist_path);
2543 let cache_root = canonicalize_existing_or_self(&cache_path);
2544 if candidate.is_absolute()
2545 && !path_has_parent_or_prefix_component(&candidate)
2546 && confined_managed_path(&candidate, &pyodide_root)
2547 {
2548 return PythonManagedResolvedPath {
2549 kind: PythonManagedPathKind::HostManaged,
2550 host_path: Some(candidate),
2551 };
2552 }
2553 if candidate.is_absolute()
2554 && !path_has_parent_or_prefix_component(&candidate)
2555 && confined_managed_path(&candidate, &cache_root)
2556 {
2557 return PythonManagedResolvedPath {
2558 kind: PythonManagedPathKind::HostManaged,
2559 host_path: Some(candidate),
2560 };
2561 }
2562
2563 PythonManagedResolvedPath {
2564 kind: PythonManagedPathKind::Unmanaged,
2565 host_path: None,
2566 }
2567}
2568
2569#[derive(Debug, Clone)]
2570struct PythonManagedResolvedPath {
2571 kind: PythonManagedPathKind,
2572 host_path: Option<PathBuf>,
2573}
2574
2575impl PythonManagedResolvedPath {
2576 fn host_path(&self) -> Option<PathBuf> {
2577 self.host_path.clone()
2578 }
2579
2580 fn render_path(&self, pyodide_dist_path: &Path, canonical: &Path, original: &str) -> String {
2581 self.kind
2582 .render_path(pyodide_dist_path, canonical, original)
2583 }
2584}
2585
2586fn strip_guest_managed_root<'a>(path: &'a str, root: &str) -> Option<&'a str> {
2587 if path == root {
2588 return Some("");
2589 }
2590 path.strip_prefix(root)?.strip_prefix('/')
2591}
2592
2593fn normalize_relative_guest_suffix(suffix: &str) -> PathBuf {
2594 let mut normalized = PathBuf::new();
2595 for segment in suffix.trim_start_matches('/').split('/') {
2596 if segment.is_empty() || segment == "." {
2597 continue;
2598 }
2599 if segment == ".." {
2600 normalized.pop();
2601 } else {
2602 normalized.push(segment);
2603 }
2604 }
2605 normalized
2606}
2607
2608fn path_has_parent_or_prefix_component(path: &Path) -> bool {
2609 path.components()
2610 .any(|component| matches!(component, Component::ParentDir | Component::Prefix(_)))
2611}
2612
2613fn canonicalize_existing_or_self(path: &Path) -> PathBuf {
2614 fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf())
2615}
2616
2617fn confined_managed_path(path: &Path, root: &Path) -> bool {
2618 let canonical_root = canonicalize_existing_or_self(root);
2619 let Some(canonical_path) = canonicalize_managed_candidate(path) else {
2620 return false;
2621 };
2622
2623 canonical_path == canonical_root || canonical_path.starts_with(canonical_root)
2624}
2625
2626fn canonicalize_managed_candidate(path: &Path) -> Option<PathBuf> {
2627 let mut missing_components = Vec::new();
2628 let mut current = path;
2629 loop {
2630 match fs::canonicalize(current) {
2631 Ok(mut canonical) => {
2632 for component in missing_components.iter().rev() {
2633 canonical.push(component);
2634 }
2635 return Some(canonical);
2636 }
2637 Err(_) => {
2638 let file_name = current.file_name()?.to_owned();
2639 if Path::new(&file_name)
2640 .components()
2641 .any(|component| !matches!(component, Component::Normal(_)))
2642 {
2643 return None;
2644 }
2645 missing_components.push(file_name);
2646 current = current.parent()?;
2647 }
2648 }
2649 }
2650}
2651
2652fn python_host_path_to_guest(pyodide_dist_path: &Path, host_path: &Path) -> Option<String> {
2653 if let Ok(relative) = host_path.strip_prefix(pyodide_dist_path) {
2654 let suffix = relative.to_string_lossy().replace('\\', "/");
2655 return Some(if suffix.is_empty() {
2656 String::from(PYODIDE_GUEST_ROOT)
2657 } else {
2658 format!("{PYODIDE_GUEST_ROOT}/{suffix}")
2659 });
2660 }
2661
2662 let cache_path = pyodide_cache_path(pyodide_dist_path);
2663 let relative = host_path.strip_prefix(cache_path).ok()?;
2664 let suffix = relative.to_string_lossy().replace('\\', "/");
2665 Some(if suffix.is_empty() {
2666 String::from(PYODIDE_CACHE_GUEST_ROOT)
2667 } else {
2668 format!("{PYODIDE_CACHE_GUEST_ROOT}/{suffix}")
2669 })
2670}
2671
2672fn python_host_stat_value(metadata: &fs::Metadata) -> Value {
2673 json!({
2674 "mode": metadata.mode(),
2675 "size": metadata.size(),
2676 "blocks": metadata.blocks(),
2677 "dev": metadata.dev(),
2678 "rdev": metadata.rdev(),
2679 "isDirectory": metadata.is_dir(),
2680 "isSymbolicLink": metadata.file_type().is_symlink(),
2681 "atimeMs": metadata.atime() * 1000 + (metadata.atime_nsec() / 1_000_000),
2682 "mtimeMs": metadata.mtime() * 1000 + (metadata.mtime_nsec() / 1_000_000),
2683 "ctimeMs": metadata.ctime() * 1000 + (metadata.ctime_nsec() / 1_000_000),
2684 "birthtimeMs": metadata.ctime() * 1000 + (metadata.ctime_nsec() / 1_000_000),
2685 "ino": metadata.ino(),
2686 "nlink": metadata.nlink(),
2687 "uid": metadata.uid(),
2688 "gid": metadata.gid(),
2689 })
2690}
2691
2692fn python_readdir_value(entries: Vec<String>) -> Value {
2693 json!(entries
2694 .into_iter()
2695 .filter(|entry| entry != "." && entry != "..")
2696 .collect::<Vec<_>>())
2697}
2698
2699fn python_sync_rpc_recursive_flag(args: &[Value]) -> bool {
2700 args.get(1)
2701 .and_then(|value| {
2702 value
2703 .as_bool()
2704 .or_else(|| value.get("recursive").and_then(Value::as_bool))
2705 })
2706 .unwrap_or(false)
2707}
2708
2709fn python_sync_rpc_bytes_arg(
2710 args: &[Value],
2711 index: usize,
2712) -> Result<Vec<u8>, PythonExecutionError> {
2713 let Some(value) = args.get(index) else {
2714 return Err(PythonExecutionError::RpcResponse(format!(
2715 "sync RPC argument {index} is required"
2716 )));
2717 };
2718
2719 if let Some(text) = value.as_str() {
2720 return Ok(text.as_bytes().to_vec());
2721 }
2722
2723 let Some(base64_value) = value
2724 .get("__agentOSType")
2725 .and_then(Value::as_str)
2726 .filter(|kind| *kind == "bytes")
2727 .and_then(|_| value.get("base64"))
2728 .and_then(Value::as_str)
2729 else {
2730 return Err(PythonExecutionError::RpcResponse(format!(
2731 "sync RPC argument {index} must be a string or encoded bytes payload"
2732 )));
2733 };
2734
2735 base64::engine::general_purpose::STANDARD
2736 .decode(base64_value)
2737 .map_err(|error| {
2738 PythonExecutionError::RpcResponse(format!(
2739 "sync RPC argument {index} contains invalid base64: {error}"
2740 ))
2741 })
2742}
2743
2744fn python_prewarm_sync_rpc_encoding(args: &[Value]) -> Option<String> {
2745 args.get(1).and_then(|value| {
2746 value.as_str().map(str::to_owned).or_else(|| {
2747 value
2748 .get("encoding")
2749 .and_then(Value::as_str)
2750 .map(str::to_owned)
2751 })
2752 })
2753}
2754
2755fn python_javascript_sync_rpc_error(
2756 request: &JavascriptSyncRpcRequest,
2757) -> Option<(&'static str, String)> {
2758 if matches!(
2759 request.method.as_str(),
2760 "net.connect"
2761 | "net.createConnection"
2762 | "dns.lookup"
2763 | "dns.resolve"
2764 | "dns.resolve4"
2765 | "dns.resolve6"
2766 | "dns.reverse"
2767 | "dgram.send"
2768 | "http.request"
2769 | "https.request"
2770 | "tls.connect"
2771 ) {
2772 return Some((
2773 "ERR_ACCESS_DENIED",
2774 String::from(
2775 "network access is not available during standalone guest Python execution",
2776 ),
2777 ));
2778 }
2779
2780 None
2781}
2782
2783fn warmup_marker_contents(
2784 import_cache: &NodeImportCache,
2785 context: &PythonContext,
2786 request: &StartPythonExecutionRequest,
2787) -> String {
2788 let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd);
2789 let compile_cache_dir = import_cache.shared_compile_cache_dir();
2790
2791 [
2792 env!("CARGO_PKG_NAME").to_string(),
2793 env!("CARGO_PKG_VERSION").to_string(),
2794 PYTHON_WARMUP_MARKER_VERSION.to_string(),
2795 String::from("agentos-v8"),
2796 python_max_old_space_mb(request).to_string(),
2797 compile_cache_dir.display().to_string(),
2798 pyodide_dist_path.display().to_string(),
2799 file_fingerprint(&pyodide_dist_path.join("pyodide.mjs")),
2800 file_fingerprint(&pyodide_dist_path.join("pyodide-lock.json")),
2801 file_fingerprint(&pyodide_dist_path.join("pyodide.asm.js")),
2802 file_fingerprint(&pyodide_dist_path.join("pyodide.asm.wasm")),
2803 file_fingerprint(&pyodide_dist_path.join("python_stdlib.zip")),
2804 ]
2805 .join("\n")
2806}
2807
2808fn python_warmup_metrics_enabled(request: &StartPythonExecutionRequest) -> bool {
2809 env_flag_enabled(&request.env, PYTHON_WARMUP_DEBUG_ENV)
2810}
2811
2812fn warmup_metrics_line(
2813 debug_enabled: bool,
2814 executed: bool,
2815 reason: &str,
2816 duration_ms: f64,
2817 import_cache: &NodeImportCache,
2818 context: &PythonContext,
2819 request: &StartPythonExecutionRequest,
2820) -> Option<Vec<u8>> {
2821 if !debug_enabled {
2822 return None;
2823 }
2824
2825 let compile_cache_dir = import_cache.shared_compile_cache_dir();
2826 let pyodide_dist_path = resolved_pyodide_dist_path(&context.pyodide_dist_path, &request.cwd);
2827
2828 Some(
2829 format!(
2830 "{PYTHON_WARMUP_METRICS_PREFIX}{{\"phase\":\"prewarm\",\"executed\":{},\"reason\":{},\"durationMs\":{duration_ms:.3},\"heapLimitMb\":{},\"compileCacheDir\":{},\"pyodideDistPath\":{}}}\n",
2831 if executed { "true" } else { "false" },
2832 encode_json_string(reason),
2833 python_max_old_space_mb(request),
2834 encode_json_string(&compile_cache_dir.display().to_string()),
2835 encode_json_string(&pyodide_dist_path.display().to_string()),
2836 )
2837 .into_bytes(),
2838 )
2839}
2840
2841#[cfg(test)]
2842mod tests {
2843 use super::{
2844 clear_pending_vfs_rpc, python_javascript_sync_rpc_action, python_managed_path_kind,
2845 python_runner_javascript_limits, python_wait_remaining, CreatePythonContextRequest,
2846 JavascriptSyncRpcRequest, PendingVfsRpc, PendingVfsRpcResolution, PendingVfsRpcState,
2847 PythonExecutionEngine, PythonExecutionLimits, PythonJavascriptSyncRpcAction,
2848 PythonManagedHostFiles, PythonManagedPathKind, PYODIDE_CACHE_GUEST_ROOT,
2849 PYODIDE_GUEST_ROOT,
2850 };
2851 use std::collections::HashMap;
2852 use std::fs;
2853 #[cfg(unix)]
2854 use std::os::unix::fs::symlink;
2855 use std::sync::{Arc, Mutex};
2856 use std::time::{Duration, Instant};
2857 use tempfile::tempdir;
2858
2859 #[test]
2860 fn python_runner_forwards_vm_reactor_limits_to_javascript() {
2861 let limits = PythonExecutionLimits {
2862 reactor_work_quantum: Some(23),
2863 bridge_call_timeout_ms: Some(54_321),
2864 ..PythonExecutionLimits::default()
2865 };
2866 let javascript = python_runner_javascript_limits(&limits, 256);
2867
2868 assert_eq!(javascript.v8_heap_limit_mb, Some(256));
2869 assert_eq!(javascript.reactor_work_quantum, Some(23));
2870 assert_eq!(javascript.bridge_call_timeout_ms, Some(54_321));
2871 }
2872
2873 #[test]
2874 fn dispose_context_reclaims_python_and_nested_javascript_metadata() {
2875 let mut engine = PythonExecutionEngine::default();
2876 let baseline = (
2877 engine.context_count_for_test(),
2878 engine.javascript_context_count_for_test(),
2879 );
2880 let temp = tempdir().expect("create Pyodide fixture root");
2881 let context = engine.create_context(CreatePythonContextRequest {
2882 vm_id: String::from("vm-python-context-dispose"),
2883 pyodide_dist_path: temp.path().to_path_buf(),
2884 });
2885 assert_eq!(engine.context_count_for_test(), baseline.0 + 1);
2886 assert_eq!(engine.javascript_context_count_for_test(), baseline.1 + 1);
2887
2888 assert!(engine.dispose_context(&context.context_id));
2889 assert_eq!(
2890 (
2891 engine.context_count_for_test(),
2892 engine.javascript_context_count_for_test(),
2893 ),
2894 baseline
2895 );
2896 }
2897
2898 #[test]
2899 fn idle_wait_uses_readiness_instead_of_turn_polling() {
2900 let started = Instant::now();
2901 assert_eq!(python_wait_remaining(None, started), None);
2902
2903 let remaining = python_wait_remaining(Some(Duration::from_secs(1)), started)
2904 .expect("finite wait keeps one deadline");
2905 assert!(remaining > Duration::from_millis(900));
2906 assert!(remaining <= Duration::from_secs(1));
2907 }
2908
2909 #[test]
2910 fn stale_python_vfs_completion_has_no_pending_waiter() {
2911 let pending = Arc::new(Mutex::new(None));
2912
2913 assert_eq!(
2914 clear_pending_vfs_rpc(&pending, 41).expect("inspect pending request"),
2915 PendingVfsRpcResolution::Missing
2916 );
2917 }
2918
2919 #[test]
2920 fn timed_out_python_vfs_completion_is_consumed_as_stale() {
2921 let pending = Arc::new(Mutex::new(Some(PendingVfsRpc {
2922 state: PendingVfsRpcState::TimedOut(42),
2923 timeout_abort: None,
2924 })));
2925
2926 assert_eq!(
2927 clear_pending_vfs_rpc(&pending, 42).expect("clear timed-out request"),
2928 PendingVfsRpcResolution::TimedOut
2929 );
2930 assert!(pending.lock().expect("pending request lock").is_none());
2931 }
2932
2933 #[test]
2934 fn python_managed_asset_descriptor_reads_use_raw_bounded_responses() {
2935 let temp = tempdir().expect("create temp dir");
2936 let pyodide = temp.path().join("pyodide");
2937 fs::create_dir_all(&pyodide).expect("create pyodide root");
2938 fs::write(pyodide.join("python_stdlib.zip"), b"stdlib-bytes").expect("write managed asset");
2939 let mut files = PythonManagedHostFiles::default();
2940 let open = JavascriptSyncRpcRequest {
2941 id: 1,
2942 method: String::from("fs.openSync"),
2943 args: vec![
2944 serde_json::Value::String(format!("{PYODIDE_GUEST_ROOT}/python_stdlib.zip")),
2945 serde_json::json!(0),
2946 serde_json::Value::Null,
2947 ],
2948 raw_bytes_args: HashMap::new(),
2949 };
2950 let fd = match python_javascript_sync_rpc_action(&pyodide, &mut files, &open)
2951 .expect("route managed open")
2952 .expect("managed open action")
2953 {
2954 PythonJavascriptSyncRpcAction::Success(value) => {
2955 value.as_u64().expect("managed descriptor")
2956 }
2957 other => panic!("unexpected managed open action: {other:?}"),
2958 };
2959
2960 let read = JavascriptSyncRpcRequest {
2961 id: 2,
2962 method: String::from("fs.readSync"),
2963 args: vec![
2964 serde_json::json!(fd),
2965 serde_json::json!(64),
2966 serde_json::Value::Null,
2967 ],
2968 raw_bytes_args: HashMap::from([(usize::MAX, Vec::new())]),
2969 };
2970 match python_javascript_sync_rpc_action(&pyodide, &mut files, &read)
2971 .expect("route managed read")
2972 .expect("managed read action")
2973 {
2974 PythonJavascriptSyncRpcAction::RawSuccess(bytes) => {
2975 assert_eq!(bytes, b"stdlib-bytes")
2976 }
2977 other => panic!("unexpected managed read action: {other:?}"),
2978 }
2979
2980 let close = JavascriptSyncRpcRequest {
2981 id: 3,
2982 method: String::from("fs.closeSync"),
2983 args: vec![serde_json::json!(fd)],
2984 raw_bytes_args: HashMap::new(),
2985 };
2986 assert!(matches!(
2987 python_javascript_sync_rpc_action(&pyodide, &mut files, &close)
2988 .expect("route managed close"),
2989 Some(PythonJavascriptSyncRpcAction::Success(
2990 serde_json::Value::Null
2991 ))
2992 ));
2993 assert!(files.files.is_empty());
2994 }
2995
2996 #[test]
2997 fn python_managed_asset_descriptors_enforce_limit_and_reuse_capacity_after_close() {
2998 let temp = tempdir().expect("create temp dir");
2999 let pyodide = temp.path().join("pyodide");
3000 fs::create_dir_all(&pyodide).expect("create pyodide root");
3001 fs::write(pyodide.join("python_stdlib.zip"), b"stdlib-bytes").expect("write managed asset");
3002 let mut files = PythonManagedHostFiles::new(2);
3003 let open = |id| JavascriptSyncRpcRequest {
3004 id,
3005 method: String::from("fs.openSync"),
3006 args: vec![
3007 serde_json::Value::String(format!("{PYODIDE_GUEST_ROOT}/python_stdlib.zip")),
3008 serde_json::json!(0),
3009 serde_json::Value::Null,
3010 ],
3011 raw_bytes_args: HashMap::new(),
3012 };
3013
3014 let opened_fd = |action: PythonJavascriptSyncRpcAction| match action {
3015 PythonJavascriptSyncRpcAction::Success(value) => {
3016 value.as_u64().expect("managed descriptor")
3017 }
3018 other => panic!("unexpected managed open action: {other:?}"),
3019 };
3020 let first = opened_fd(
3021 python_javascript_sync_rpc_action(&pyodide, &mut files, &open(1))
3022 .expect("route first open")
3023 .expect("first open action"),
3024 );
3025 let _second = opened_fd(
3026 python_javascript_sync_rpc_action(&pyodide, &mut files, &open(2))
3027 .expect("route second open")
3028 .expect("second open action"),
3029 );
3030 assert_eq!(files.files.len(), 2);
3031
3032 assert!(matches!(
3033 python_javascript_sync_rpc_action(&pyodide, &mut files, &open(3))
3034 .expect("route saturated open"),
3035 Some(PythonJavascriptSyncRpcAction::Error { code: "EMFILE", message })
3036 if message.contains("limits.resources.maxOpenFds")
3037 ));
3038 assert_eq!(files.files.len(), 2);
3039
3040 let close = JavascriptSyncRpcRequest {
3041 id: 4,
3042 method: String::from("fs.closeSync"),
3043 args: vec![serde_json::json!(first)],
3044 raw_bytes_args: HashMap::new(),
3045 };
3046 assert!(matches!(
3047 python_javascript_sync_rpc_action(&pyodide, &mut files, &close).expect("route close"),
3048 Some(PythonJavascriptSyncRpcAction::Success(
3049 serde_json::Value::Null
3050 ))
3051 ));
3052 let _replacement = opened_fd(
3053 python_javascript_sync_rpc_action(&pyodide, &mut files, &open(5))
3054 .expect("route replacement open")
3055 .expect("replacement open action"),
3056 );
3057 assert_eq!(files.files.len(), 2);
3058 }
3059
3060 #[test]
3061 fn python_managed_guest_paths_normalize_dot_dot_inside_root() {
3062 let temp = tempdir().expect("create temp dir");
3063 let pyodide = temp.path().join("pyodide");
3064 fs::create_dir_all(pyodide.join("lib")).expect("create pyodide lib");
3065
3066 let resolved = python_managed_path_kind(
3067 &pyodide,
3068 &format!("{PYODIDE_GUEST_ROOT}/lib/../pyodide.mjs"),
3069 );
3070
3071 assert!(matches!(resolved.kind, PythonManagedPathKind::GuestPyodide));
3072 assert_eq!(
3073 resolved.host_path().expect("host path"),
3074 pyodide.join("pyodide.mjs")
3075 );
3076 }
3077
3078 #[test]
3079 fn python_managed_guest_paths_clamp_dot_dot_escape_to_root() {
3080 let temp = tempdir().expect("create temp dir");
3081 let pyodide = temp.path().join("pyodide");
3082 fs::create_dir_all(&pyodide).expect("create pyodide root");
3083
3084 let resolved =
3085 python_managed_path_kind(&pyodide, &format!("{PYODIDE_GUEST_ROOT}/../../outside.txt"));
3086
3087 assert!(matches!(resolved.kind, PythonManagedPathKind::GuestPyodide));
3088 assert_eq!(
3089 resolved.host_path().expect("host path"),
3090 pyodide.join("outside.txt")
3091 );
3092 }
3093
3094 #[cfg(unix)]
3095 #[test]
3096 fn python_managed_guest_paths_reject_symlink_escape() {
3097 let temp = tempdir().expect("create temp dir");
3098 let pyodide = temp.path().join("pyodide");
3099 let outside = temp.path().join("outside");
3100 fs::create_dir_all(&pyodide).expect("create pyodide root");
3101 fs::create_dir_all(&outside).expect("create outside dir");
3102 symlink(&outside, pyodide.join("escape")).expect("create escape symlink");
3103
3104 let resolved =
3105 python_managed_path_kind(&pyodide, &format!("{PYODIDE_GUEST_ROOT}/escape/file.txt"));
3106
3107 assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged));
3108 assert!(resolved.host_path().is_none());
3109 }
3110
3111 #[cfg(unix)]
3112 #[test]
3113 fn python_managed_guest_paths_reject_symlink_escape_to_missing_descendant() {
3114 let temp = tempdir().expect("create temp dir");
3115 let pyodide = temp.path().join("pyodide");
3116 let outside = temp.path().join("outside");
3117 fs::create_dir_all(&pyodide).expect("create pyodide root");
3118 fs::create_dir_all(&outside).expect("create outside dir");
3119 symlink(&outside, pyodide.join("escape")).expect("create escape symlink");
3120
3121 let resolved = python_managed_path_kind(
3122 &pyodide,
3123 &format!("{PYODIDE_GUEST_ROOT}/escape/missing/file.txt"),
3124 );
3125
3126 assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged));
3127 assert!(resolved.host_path().is_none());
3128 }
3129
3130 #[test]
3131 fn python_managed_host_paths_accept_canonical_root_descendants() {
3132 let temp = tempdir().expect("create temp dir");
3133 let pyodide = temp.path().join("pyodide");
3134 fs::create_dir_all(pyodide.join("pkg")).expect("create pyodide package dir");
3135 let candidate = pyodide.join("pkg/module.py");
3136
3137 let resolved = python_managed_path_kind(&pyodide, &candidate.display().to_string());
3138
3139 assert!(matches!(resolved.kind, PythonManagedPathKind::HostManaged));
3140 assert_eq!(resolved.host_path().expect("host path"), candidate);
3141 }
3142
3143 #[test]
3144 fn python_managed_host_paths_reject_unresolved_dot_dot_escape() {
3145 let temp = tempdir().expect("create temp dir");
3146 let pyodide = temp.path().join("pyodide");
3147 fs::create_dir_all(&pyodide).expect("create pyodide root");
3148 let candidate = pyodide.join("missing/../../outside.txt");
3149
3150 let resolved = python_managed_path_kind(&pyodide, &candidate.display().to_string());
3151
3152 assert!(matches!(resolved.kind, PythonManagedPathKind::Unmanaged));
3153 assert!(resolved.host_path().is_none());
3154 }
3155
3156 #[test]
3157 fn python_managed_cache_guest_paths_resolve_inside_cache_root() {
3158 let temp = tempdir().expect("create temp dir");
3159 let pyodide = temp.path().join("pyodide");
3160 fs::create_dir_all(&pyodide).expect("create pyodide root");
3161
3162 let resolved = python_managed_path_kind(
3163 &pyodide,
3164 &format!("{PYODIDE_CACHE_GUEST_ROOT}/wheels/pkg.whl"),
3165 );
3166 let host_path = resolved.host_path().expect("host path");
3167
3168 assert!(matches!(resolved.kind, PythonManagedPathKind::GuestCache));
3169 assert!(host_path.ends_with("pyodide-package-cache/wheels/pkg.whl"));
3170 }
3171}