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