parallax_protocol/
lib.rs

1//! Versioned NDJSON worker protocol for Parallax adapters.
2
3#![deny(unsafe_code)]
4#![warn(missing_docs)]
5
6use parallax_core::{
7    check_protocol, ErrorCode, ExecutionLimits, ParallaxError, RequestId, RuntimeKind,
8    PROTOCOL_VERSION,
9};
10use serde::{Deserialize, Serialize};
11use serde_json::Value;
12
13/// Protocol version re-export.
14pub const VERSION: u32 = PROTOCOL_VERSION;
15
16/// Envelope wrapping every worker message.
17#[derive(Clone, Debug, Serialize, Deserialize)]
18pub struct Envelope {
19    /// Protocol version.
20    pub v: u32,
21    /// Correlation id.
22    pub id: RequestId,
23    /// Operation name.
24    pub op: String,
25    /// Success flag (responses only; omitted on requests).
26    #[serde(default, skip_serializing_if = "Option::is_none")]
27    pub ok: Option<bool>,
28    /// Payload.
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub payload: Option<Value>,
31    /// Structured error (responses only).
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    pub error: Option<ProtocolError>,
34}
35
36impl Envelope {
37    /// Build a request envelope.
38    pub fn request(op: impl Into<String>, payload: Value) -> Self {
39        Self {
40            v: VERSION,
41            id: RequestId::new(),
42            op: op.into(),
43            ok: None,
44            payload: Some(payload),
45            error: None,
46        }
47    }
48
49    /// Build a successful response.
50    pub fn ok_response(id: RequestId, op: impl Into<String>, payload: Value) -> Self {
51        Self {
52            v: VERSION,
53            id,
54            op: op.into(),
55            ok: Some(true),
56            payload: Some(payload),
57            error: None,
58        }
59    }
60
61    /// Build an error response.
62    pub fn err_response(id: RequestId, op: impl Into<String>, error: ProtocolError) -> Self {
63        Self {
64            v: VERSION,
65            id,
66            op: op.into(),
67            ok: Some(false),
68            payload: None,
69            error: Some(error),
70        }
71    }
72
73    /// Encode as a single NDJSON line (with trailing newline).
74    pub fn to_ndjson_line(&self) -> parallax_core::Result<String> {
75        let mut s = serde_json::to_string(self).map_err(ParallaxError::from)?;
76        s.push('\n');
77        Ok(s)
78    }
79
80    /// Parse a single NDJSON line.
81    pub fn from_ndjson_line(line: &str) -> parallax_core::Result<Self> {
82        let env: Self = serde_json::from_str(line.trim()).map_err(|e| {
83            ParallaxError::new(ErrorCode::ProtocolViolation, e.to_string())
84                .with_source("parallax-protocol")
85                .with_operation("from_ndjson_line")
86        })?;
87        check_protocol(env.v).map_err(|e| {
88            e.with_source("parallax-protocol")
89                .with_operation("from_ndjson_line")
90        })?;
91        Ok(env)
92    }
93}
94
95/// Protocol-level error payload.
96#[derive(Clone, Debug, Serialize, Deserialize)]
97pub struct ProtocolError {
98    /// Error code string.
99    pub code: String,
100    /// Message.
101    pub message: String,
102    /// Optional diagnostic.
103    #[serde(default, skip_serializing_if = "Option::is_none")]
104    pub diagnostic: Option<String>,
105}
106
107impl ProtocolError {
108    /// From a ParallaxError.
109    pub fn from_parallax(err: &ParallaxError) -> Self {
110        Self {
111            code: err.code.to_string(),
112            message: err.message.clone(),
113            diagnostic: err.diagnostic.clone(),
114        }
115    }
116
117    /// Convert to ParallaxError.
118    pub fn into_parallax(self, runtime: Option<RuntimeKind>) -> ParallaxError {
119        let code = match self.code.as_str() {
120            "RUNTIME_UNAVAILABLE" => ErrorCode::RuntimeUnavailable,
121            "EXECUTION_FAILURE" => ErrorCode::ExecutionFailure,
122            "EXECUTION_TIMEOUT" => ErrorCode::ExecutionTimeout,
123            "EXECUTION_CANCELLED" => ErrorCode::ExecutionCancelled,
124            "CAPTURE_FAILURE" => ErrorCode::CaptureFailure,
125            "RESTORE_FAILURE" => ErrorCode::RestoreFailure,
126            "UNSUPPORTED_VALUE" => ErrorCode::UnsupportedValue,
127            "PROTOCOL_VIOLATION" => ErrorCode::ProtocolViolation,
128            "RESOURCE_LIMIT_EXCEEDED" => ErrorCode::ResourceLimitExceeded,
129            "ADAPTER_CRASHED" => ErrorCode::AdapterCrashed,
130            _ => ErrorCode::Internal,
131        };
132        let mut err = ParallaxError::new(code, self.message)
133            .with_source("worker")
134            .with_operation("protocol");
135        if let Some(rt) = runtime {
136            err = err.with_runtime(rt);
137        }
138        if let Some(d) = self.diagnostic {
139            err = err.with_diagnostic(d);
140        }
141        err
142    }
143}
144
145/// Hello request payload.
146#[derive(Clone, Debug, Serialize, Deserialize)]
147pub struct HelloRequest {
148    /// Core protocol version.
149    pub protocol_version: u32,
150    /// Expected runtime kind.
151    pub runtime: RuntimeKind,
152}
153
154/// Hello response payload.
155#[derive(Clone, Debug, Serialize, Deserialize)]
156pub struct HelloResponse {
157    /// Worker protocol version.
158    pub protocol_version: u32,
159    /// Runtime kind.
160    pub runtime: RuntimeKind,
161    /// Host language/runtime version string.
162    pub host_version: String,
163    /// Adapter version.
164    pub adapter_version: String,
165}
166
167/// Validate a hello response against the host protocol version.
168pub fn validate_hello(resp: &HelloResponse) -> parallax_core::Result<()> {
169    check_protocol(resp.protocol_version).map_err(|e| {
170        e.with_source("parallax-protocol")
171            .with_operation("validate_hello")
172    })
173}
174
175/// Execute request payload.
176#[derive(Clone, Debug, Serialize, Deserialize)]
177pub struct ExecuteRequest {
178    /// Source code.
179    pub source: String,
180    /// Filename hint.
181    pub filename: String,
182    /// Binding names to capture after execution.
183    #[serde(default)]
184    pub capture: Vec<String>,
185    /// Resource limits.
186    pub limits: ExecutionLimits,
187    /// Enable Continuum safepoint / checkpoint capture (explicit `parallax.checkpoint`).
188    #[serde(default)]
189    pub continuum: bool,
190}
191
192/// Execute response payload.
193#[derive(Clone, Debug, Serialize, Deserialize)]
194pub struct ExecuteResponse {
195    /// Captured stdout.
196    pub stdout: String,
197    /// Captured stderr.
198    pub stderr: String,
199    /// Wall duration microseconds.
200    pub duration_us: u64,
201    /// Captured bindings as PIR-tagged JSON object.
202    #[serde(default)]
203    pub bindings: Value,
204    /// Optional exception.
205    #[serde(default, skip_serializing_if = "Option::is_none")]
206    pub exception: Option<Value>,
207    /// Success flag for guest code.
208    pub success: bool,
209    /// True when paused at an explicit Continuum safepoint (not a failure).
210    #[serde(default)]
211    pub suspended: bool,
212    /// Captured Universal Execution State JSON when suspended at a checkpoint.
213    #[serde(default, skip_serializing_if = "Option::is_none")]
214    pub ues: Option<Value>,
215    /// Safepoint report JSON when suspended.
216    #[serde(default, skip_serializing_if = "Option::is_none")]
217    pub safepoint: Option<Value>,
218}
219
220/// Resume a checkpoint-produced UES on the same runtime (Continuum).
221#[derive(Clone, Debug, Serialize, Deserialize)]
222pub struct ResumeCheckpointRequest {
223    /// UES JSON document.
224    pub ues: Value,
225    /// Resource limits.
226    pub limits: ExecutionLimits,
227}
228
229/// Response from same-runtime checkpoint resume.
230#[derive(Clone, Debug, Serialize, Deserialize)]
231pub struct ResumeCheckpointResponse {
232    /// Whether resume succeeded.
233    pub success: bool,
234    /// Captured stdout from post-checkpoint region.
235    pub stdout: String,
236    /// Captured stderr.
237    pub stderr: String,
238    /// Duration microseconds.
239    pub duration_us: u64,
240    /// Bindings after resume (PIR-tagged).
241    #[serde(default)]
242    pub bindings: Value,
243    /// Optional exception.
244    #[serde(default, skip_serializing_if = "Option::is_none")]
245    pub exception: Option<Value>,
246    /// Warnings (e.g. experimental).
247    #[serde(default)]
248    pub warnings: Vec<String>,
249}
250
251/// Restore request payload.
252#[derive(Clone, Debug, Serialize, Deserialize)]
253pub struct RestoreRequest {
254    /// Bindings as PIR-tagged JSON object.
255    pub bindings: Value,
256    /// Resource limits.
257    pub limits: ExecutionLimits,
258}
259
260/// Restore response payload.
261#[derive(Clone, Debug, Serialize, Deserialize)]
262pub struct RestoreResponse {
263    /// Whether restore succeeded.
264    pub success: bool,
265    /// Warnings.
266    #[serde(default)]
267    pub warnings: Vec<String>,
268    /// Restored binding summaries.
269    #[serde(default)]
270    pub restored: IndexMapCompat,
271    /// Duration microseconds.
272    pub duration_us: u64,
273    /// Optional verification dump of restored values as PIR JSON.
274    #[serde(default, skip_serializing_if = "Option::is_none")]
275    pub bindings: Option<Value>,
276}
277
278/// Thin serde-friendly string map.
279#[derive(Clone, Debug, Default, Serialize, Deserialize)]
280#[serde(transparent)]
281pub struct IndexMapCompat(pub std::collections::BTreeMap<String, String>);
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286
287    #[test]
288    fn ndjson_round_trip() {
289        let env = Envelope::request("ping", serde_json::json!({}));
290        let line = env.to_ndjson_line().unwrap();
291        let parsed = Envelope::from_ndjson_line(&line).unwrap();
292        assert_eq!(parsed.op, "ping");
293    }
294
295    #[test]
296    fn rejects_future_protocol_major() {
297        let line = format!(
298            r#"{{"v":{},"id":"00000000-0000-4000-8000-000000000001","op":"ping"}}"#,
299            VERSION + 1
300        );
301        let err = Envelope::from_ndjson_line(&line).unwrap_err();
302        assert_eq!(err.code, ErrorCode::ProtocolViolation);
303    }
304
305    #[test]
306    fn hello_rejects_mismatched_protocol() {
307        let resp = HelloResponse {
308            protocol_version: VERSION + 7,
309            runtime: RuntimeKind::Python,
310            host_version: "3.12".into(),
311            adapter_version: "0.1.0".into(),
312        };
313        let err = validate_hello(&resp).unwrap_err();
314        assert_eq!(err.code, ErrorCode::ProtocolViolation);
315    }
316}
parallax_protocol/lib.rs

parallax_protocol/
lib.rs
