parallax_adapter_python/
lib.rs

1//! Python runtime adapter (subprocess worker).
2
3#![deny(unsafe_code)]
4#![warn(missing_docs)]
5
6use async_trait::async_trait;
7use parallax_core::{
8    ErrorCode, ExceptionInfo, ExecutionRequest, ExecutionResult, ExecutionState, ParallaxError,
9    ProgramSource, Remediation, RestoreResult, RuntimeCapabilities, RuntimeKind, RuntimeMetadata,
10    RuntimeStatus, PARALLAX_VERSION,
11};
12use parallax_ir::PirDocument;
13use parallax_protocol::{
14    validate_hello, Envelope, ExecuteRequest, ExecuteResponse, HelloRequest, HelloResponse,
15    ProtocolError, RestoreRequest, RestoreResponse, ResumeCheckpointRequest,
16    ResumeCheckpointResponse,
17};
18use parallax_runtime::{discover_python, BoxRuntimeAdapter, RuntimeAdapter, WorkerProcess};
19use parallax_security::SandboxPolicy;
20use std::path::PathBuf;
21use std::sync::Arc;
22use std::time::Duration;
23use tracing::{debug, warn};
24
25const WORKER_SOURCE: &str = include_str!("../worker.py");
26
27/// Python adapter.
28pub struct PythonAdapter {
29    /// Discovered interpreter path, if any.
30    pub python: Option<PathBuf>,
31    host_version: Option<String>,
32    worker_path: PathBuf,
33}
34
35impl PythonAdapter {
36    /// Create adapter, discovering Python and materializing the worker script.
37    pub fn new() -> parallax_core::Result<Self> {
38        let discovered = discover_python();
39        let worker_path = materialize_worker()?;
40        Ok(Self {
41            python: discovered.as_ref().map(|d| d.path.clone()),
42            host_version: discovered.and_then(|d| d.version),
43            worker_path,
44        })
45    }
46
47    /// Boxed factory helper.
48    pub fn boxed() -> parallax_core::Result<BoxRuntimeAdapter> {
49        Ok(Arc::new(Self::new()?))
50    }
51
52    async fn spawn_ready(&self) -> parallax_core::Result<WorkerProcess> {
53        let python = self.python.as_ref().ok_or_else(|| {
54            ParallaxError::new(
55                ErrorCode::RuntimeUnavailable,
56                "Python interpreter not found on this host",
57            )
58            .with_runtime(RuntimeKind::Python)
59            .with_source("parallax-adapter-python")
60            .remediate(Remediation::with_detail(
61                "Install Python 3 and ensure `python`, `python3`, or `py` works",
62                "On Windows, disable the Microsoft Store python alias or install from python.org",
63            ))
64        })?;
65
66        let args = vec![self.worker_path.to_string_lossy().to_string()];
67        let mut worker = WorkerProcess::spawn(python.clone(), &args, RuntimeKind::Python).await?;
68        let hello = Envelope::request(
69            "hello",
70            serde_json::to_value(HelloRequest {
71                protocol_version: parallax_protocol::VERSION,
72                runtime: RuntimeKind::Python,
73            })?,
74        );
75        let resp = worker.request(hello, Duration::from_secs(10)).await?;
76        if resp.ok != Some(true) {
77            let err = resp.error.unwrap_or_else(|| ProtocolError {
78                code: "RUNTIME_INITIALIZATION_FAILURE".into(),
79                message: "hello failed".into(),
80                diagnostic: None,
81            });
82            return Err(err.into_parallax(Some(RuntimeKind::Python)));
83        }
84        let hello_payload: HelloResponse = serde_json::from_value(
85            resp.payload.unwrap_or(serde_json::Value::Null),
86        )
87        .map_err(|e| {
88            ParallaxError::new(
89                ErrorCode::ProtocolViolation,
90                format!("invalid hello response: {e}"),
91            )
92            .with_runtime(RuntimeKind::Python)
93            .with_source("parallax-adapter-python")
94        })?;
95        validate_hello(&hello_payload)?;
96        Ok(worker)
97    }
98}
99
100fn materialize_worker() -> parallax_core::Result<PathBuf> {
101    let dir = std::env::temp_dir().join("parallax-workers");
102    std::fs::create_dir_all(&dir)?;
103    let path = dir.join("python_worker.py");
104    std::fs::write(&path, WORKER_SOURCE)?;
105    Ok(path)
106}
107
108fn validate_capture_name(name: &str) -> parallax_core::Result<()> {
109    let ok = name
110        .chars()
111        .next()
112        .map(|c| c.is_ascii_alphabetic() || c == '_')
113        .unwrap_or(false)
114        && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_');
115    if ok {
116        Ok(())
117    } else {
118        Err(ParallaxError::new(
119            ErrorCode::InvalidArgument,
120            format!("invalid capture binding name: {name}"),
121        )
122        .with_runtime(RuntimeKind::Python)
123        .with_operation("validate_capture_name")
124        .remediate(Remediation::new(
125            "Use ASCII identifier names (letters, digits, underscore)",
126        )))
127    }
128}
129
130fn load_source(program: &ProgramSource) -> parallax_core::Result<(String, String, Vec<String>)> {
131    match program {
132        ProgramSource::File { path } => {
133            let source = std::fs::read_to_string(path)?;
134            Ok((source, path.clone(), Vec::new()))
135        }
136        ProgramSource::Inline { source, filename } => {
137            Ok((source.clone(), filename.clone(), Vec::new()))
138        }
139        ProgramSource::CaptureBindings {
140            source,
141            names,
142            filename,
143        } => {
144            for name in names {
145                validate_capture_name(name)?;
146            }
147            Ok((source.clone(), filename.clone(), names.clone()))
148        }
149        ProgramSource::Bytes { .. } => Err(ParallaxError::new(
150            ErrorCode::UnsupportedValue,
151            "Python adapter does not execute raw bytes",
152        )
153        .with_runtime(RuntimeKind::Python)),
154    }
155}
156
157#[async_trait]
158impl RuntimeAdapter for PythonAdapter {
159    fn kind(&self) -> RuntimeKind {
160        RuntimeKind::Python
161    }
162
163    fn metadata(&self) -> RuntimeMetadata {
164        let mut meta = RuntimeMetadata::builtin(
165            RuntimeKind::Python,
166            "CPython",
167            "Python adapter via NDJSON subprocess worker",
168        );
169        meta.host_version = self.host_version.clone();
170        meta.adapter_version = PARALLAX_VERSION.to_string();
171        meta
172    }
173
174    fn capabilities(&self) -> RuntimeCapabilities {
175        RuntimeCapabilities::python()
176    }
177
178    async fn probe(&self) -> RuntimeStatus {
179        if self.python.is_some() {
180            RuntimeStatus::Ready
181        } else {
182            RuntimeStatus::Unavailable {
183                reason: "Python not found (tried python, python3, py, and common install paths)"
184                    .into(),
185            }
186        }
187    }
188
189    async fn execute(
190        &self,
191        request: ExecutionRequest,
192        policy: &SandboxPolicy,
193    ) -> parallax_core::Result<ExecutionResult> {
194        let (source, filename, mut capture) = load_source(&request.program)?;
195        if request.capture_state && capture.is_empty() {
196            capture.push("state".into());
197        }
198        for name in &capture {
199            validate_capture_name(name)?;
200        }
201        let limits = policy.limits.clone();
202        let max_message_bytes = limits.max_message_bytes;
203        let wait = limits.timeout + Duration::from_secs(5);
204        let execution_id = request.execution_id;
205        let program_label = filename.clone();
206
207        let mut worker = self.spawn_ready().await?;
208        let env = Envelope::request(
209            "execute",
210            serde_json::to_value(ExecuteRequest {
211                source,
212                filename,
213                capture,
214                limits,
215                continuum: request.continuum,
216            })?,
217        );
218        let resp = worker.request_bounded(env, wait, max_message_bytes).await;
219        worker.shutdown().await;
220        let resp = resp?;
221        if resp.ok != Some(true) {
222            let err = resp.error.unwrap_or_else(|| ProtocolError {
223                code: "EXECUTION_FAILURE".into(),
224                message: "execute failed".into(),
225                diagnostic: None,
226            });
227            return Err(err.into_parallax(Some(RuntimeKind::Python)));
228        }
229        let payload: ExecuteResponse =
230            serde_json::from_value(resp.payload.unwrap_or(serde_json::Value::Null))?;
231
232        let state = if payload
233            .bindings
234            .as_object()
235            .map(|o| !o.is_empty())
236            .unwrap_or(false)
237        {
238            let pir = PirDocument::from_bindings_json(&payload.bindings)?;
239            let mut st = ExecutionState::empty(RuntimeKind::Python, RuntimeCapabilities::python());
240            st.heap = pir.bindings_to_json();
241            st.program = Some(program_label);
242            Some(st)
243        } else {
244            None
245        };
246
247        let exception = payload.exception.as_ref().map(|v| ExceptionInfo {
248            type_name: v
249                .get("type_name")
250                .and_then(|x| x.as_str())
251                .unwrap_or("Exception")
252                .into(),
253            message: v
254                .get("message")
255                .and_then(|x| x.as_str())
256                .unwrap_or("")
257                .into(),
258            stack: v
259                .get("stack")
260                .and_then(|x| x.as_str())
261                .map(|s| s.to_string()),
262        });
263
264        Ok(ExecutionResult {
265            execution_id,
266            runtime: RuntimeKind::Python,
267            success: payload.success || payload.suspended,
268            value: None,
269            stdout: payload.stdout,
270            stderr: payload.stderr,
271            duration_us: payload.duration_us,
272            exception,
273            state,
274            suspended: payload.suspended,
275            ues: payload.ues,
276            safepoint: payload.safepoint,
277            stats: Default::default(),
278        })
279    }
280
281    async fn restore(
282        &self,
283        pir: &PirDocument,
284        policy: &SandboxPolicy,
285    ) -> parallax_core::Result<RestoreResult> {
286        let bindings = pir.bindings_to_json();
287        let limits = policy.limits.clone();
288        let max_message_bytes = limits.max_message_bytes;
289        let wait = limits.timeout + Duration::from_secs(5);
290        let mut worker = self.spawn_ready().await?;
291        let env = Envelope::request(
292            "restore",
293            serde_json::to_value(RestoreRequest { bindings, limits })?,
294        );
295        let resp = worker.request_bounded(env, wait, max_message_bytes).await;
296        worker.shutdown().await;
297        let resp = resp?;
298        if resp.ok != Some(true) {
299            let err = resp.error.unwrap_or_else(|| ProtocolError {
300                code: "RESTORE_FAILURE".into(),
301                message: "restore failed".into(),
302                diagnostic: None,
303            });
304            return Err(err.into_parallax(Some(RuntimeKind::Python)));
305        }
306        let payload: RestoreResponse =
307            serde_json::from_value(resp.payload.unwrap_or(serde_json::Value::Null))?;
308        Ok(RestoreResult {
309            success: payload.success,
310            runtime: RuntimeKind::Python,
311            warnings: payload.warnings,
312            restored_bindings: payload.restored.0.into_iter().collect(),
313            duration_us: payload.duration_us,
314        })
315    }
316}
317
318impl PythonAdapter {
319    /// Same-runtime resume from a checkpoint-produced UES (Continuum experimental).
320    pub async fn resume_checkpoint(
321        &self,
322        ues: serde_json::Value,
323        policy: &SandboxPolicy,
324    ) -> parallax_core::Result<ResumeCheckpointResponse> {
325        let limits = policy.limits.clone();
326        let max_message_bytes = limits.max_message_bytes;
327        let wait = limits.timeout + Duration::from_secs(5);
328        let mut worker = self.spawn_ready().await?;
329        let env = Envelope::request(
330            "resume_checkpoint",
331            serde_json::to_value(ResumeCheckpointRequest { ues, limits })?,
332        );
333        let resp = worker.request_bounded(env, wait, max_message_bytes).await;
334        worker.shutdown().await;
335        let resp = resp?;
336        if resp.ok != Some(true) {
337            let err = resp.error.unwrap_or_else(|| ProtocolError {
338                code: "RESTORE_FAILURE".into(),
339                message: "resume_checkpoint failed".into(),
340                diagnostic: None,
341            });
342            return Err(err.into_parallax(Some(RuntimeKind::Python)));
343        }
344        serde_json::from_value(resp.payload.unwrap_or(serde_json::Value::Null)).map_err(|e| {
345            ParallaxError::new(
346                ErrorCode::ProtocolViolation,
347                format!("invalid resume_checkpoint response: {e}"),
348            )
349            .with_runtime(RuntimeKind::Python)
350            .with_source("parallax-adapter-python")
351        })
352    }
353}
354
355/// Always register; probe reports unavailable if Python missing.
356pub fn register_lenient(manager: &parallax_runtime::RuntimeManager) {
357    match PythonAdapter::new() {
358        Ok(adapter) => {
359            debug!(python = ?adapter.python, "registered python adapter");
360            manager.register(Arc::new(adapter));
361        }
362        Err(e) => warn!("could not materialize python worker: {e}"),
363    }
364}
parallax_adapter_python/lib.rs

parallax_adapter_python/
lib.rs
