parallax_adapter_js/
lib.rs

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

parallax_adapter_js/
lib.rs
