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