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