contextvm_sdk/transport/open_stream/
frame.rs

1//! CEP-41 `cvm` frame types and their `notifications/progress` envelope.
2//!
3//! A frame is the ContextVM `cvm` extension object embedded in the `params` of
4//! an MCP `notifications/progress` notification. The outer `params` also carry
5//! the `progressToken` (the stream id) and a strictly-monotonic `progress`
6//! value that orders *all* frames in one direction. See the frame shapes in
7//! `sdk/src/transport/open-stream/types.ts`.
8//!
9//! Unlike CEP-22, an open-stream `chunk` also carries a `chunkIndex` (contiguous
10//! from 0) that tracks payload completeness independently of the outer
11//! `progress` counter.
12
13use serde::{Deserialize, Serialize};
14use serde_json::{Map, Value};
15
16use crate::core::types::JsonRpcNotification;
17use crate::transport::oversized_transfer::NOTIFICATIONS_PROGRESS_METHOD;
18
19use super::constants::OPEN_STREAM_TYPE;
20
21/// A CEP-41 open-stream frame (the `cvm` object).
22///
23/// Internally tagged on `frameType`. The constant `cvm.type` (`"open-stream"`)
24/// is handled separately by [`to_cvm_value`](Self::to_cvm_value) /
25/// [`from_cvm_value`](Self::from_cvm_value) rather than encoded as an enum
26/// field, exactly mirroring CEP-22's `oversized_transfer/frame.rs`.
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(tag = "frameType", rename_all = "camelCase")]
29pub enum OpenStreamFrame {
30    /// Opens a stream. Carries only optional application-defined advisory
31    /// metadata; receivers MUST NOT depend on it for stream correctness.
32    #[serde(rename_all = "camelCase")]
33    Start {
34        /// Optional advisory content type (writer-settable, receiver-ignored).
35        #[serde(default, skip_serializing_if = "Option::is_none")]
36        content_type: Option<String>,
37    },
38    /// Reader handshake acknowledgement, sent only in reply to an inbound `start`.
39    Accept,
40    /// One ordered payload fragment of the stream.
41    #[serde(rename_all = "camelCase")]
42    Chunk {
43        /// Contiguous-from-0 chunk index used for completeness/ordering.
44        chunk_index: u64,
45        /// The chunk payload (an arbitrary UTF-8 string).
46        data: String,
47    },
48    /// Keepalive probe; the peer must echo the `nonce` in a `pong`.
49    Ping {
50        /// Opaque probe nonce echoed back by the peer.
51        nonce: String,
52    },
53    /// Keepalive response echoing a prior `ping` nonce.
54    Pong {
55        /// The nonce of the `ping` being acknowledged.
56        nonce: String,
57    },
58    /// Closes the stream gracefully (terminal). Declares the final chunk index
59    /// when any chunks were emitted, so the reader can verify completeness.
60    #[serde(rename_all = "camelCase")]
61    Close {
62        /// The highest `chunkIndex` emitted, or `None` for a zero-chunk stream.
63        #[serde(default, skip_serializing_if = "Option::is_none")]
64        last_chunk_index: Option<u64>,
65    },
66    /// Terminates the stream early (terminal).
67    Abort {
68        /// Optional advisory reason.
69        #[serde(default, skip_serializing_if = "Option::is_none")]
70        reason: Option<String>,
71    },
72}
73
74impl OpenStreamFrame {
75    /// Returns the `frameType` discriminator string for this frame.
76    pub fn frame_type(&self) -> &'static str {
77        match self {
78            Self::Start { .. } => "start",
79            Self::Accept => "accept",
80            Self::Chunk { .. } => "chunk",
81            Self::Ping { .. } => "ping",
82            Self::Pong { .. } => "pong",
83            Self::Close { .. } => "close",
84            Self::Abort { .. } => "abort",
85        }
86    }
87
88    /// Serialize this frame into a `cvm` object [`Value`], injecting the constant
89    /// `type` discriminator (`"open-stream"`).
90    pub fn to_cvm_value(&self) -> Result<Value, serde_json::Error> {
91        let mut value = serde_json::to_value(self)?;
92        if let Value::Object(map) = &mut value {
93            map.insert(
94                "type".to_string(),
95                Value::String(OPEN_STREAM_TYPE.to_string()),
96            );
97        }
98        Ok(value)
99    }
100
101    /// Parse a `cvm` object [`Value`] into a typed frame, verifying the `type`
102    /// discriminator. Returns `None` when `value` is not an open-stream frame or
103    /// fails to parse.
104    pub fn from_cvm_value(value: &Value) -> Option<Self> {
105        if value.get("type").and_then(Value::as_str) != Some(OPEN_STREAM_TYPE) {
106            return None;
107        }
108        serde_json::from_value(value.clone()).ok()
109    }
110
111    /// Returns `true` when `value` structurally looks like an open-stream frame
112    /// (`type == "open-stream"` and a string `frameType`).
113    ///
114    /// Mirrors the TS `isOpenStreamFrame` structural narrowing.
115    pub fn is_frame_value(value: &Value) -> bool {
116        value.get("type").and_then(Value::as_str) == Some(OPEN_STREAM_TYPE)
117            && value.get("frameType").and_then(Value::as_str).is_some()
118    }
119
120    /// Wrap this frame in a `notifications/progress` [`JsonRpcNotification`].
121    ///
122    /// Builds the outer `params` with `progressToken`, `progress`, an optional
123    /// human-readable `message` (non-normative UX), and the `cvm` frame. The
124    /// token is always emitted as a JSON **string**, even when the originating
125    /// request carried a numeric one (matching the TS SDK's
126    /// `String(progressToken)`); see
127    /// [`progress_token_string`](crate::transport::oversized_transfer::progress_token_string).
128    pub fn into_progress_notification(
129        &self,
130        progress_token: &str,
131        progress: u64,
132        message: Option<&str>,
133    ) -> Result<JsonRpcNotification, serde_json::Error> {
134        let mut params = Map::new();
135        params.insert(
136            "progressToken".to_string(),
137            Value::String(progress_token.to_string()),
138        );
139        params.insert("progress".to_string(), Value::Number(progress.into()));
140        if let Some(message) = message {
141            params.insert("message".to_string(), Value::String(message.to_string()));
142        }
143        params.insert("cvm".to_string(), self.to_cvm_value()?);
144
145        Ok(JsonRpcNotification {
146            jsonrpc: "2.0".to_string(),
147            method: NOTIFICATIONS_PROGRESS_METHOD.to_string(),
148            params: Some(Value::Object(params)),
149        })
150    }
151}
152
153/// Extract a typed [`OpenStreamFrame`] from a `notifications/progress` payload's
154/// `params.cvm`, returning `None` for any non-open-stream notification.
155pub fn open_stream_frame_from_notification(
156    notification: &JsonRpcNotification,
157) -> Option<OpenStreamFrame> {
158    notification
159        .params
160        .as_ref()
161        .and_then(|params| params.get("cvm"))
162        .and_then(OpenStreamFrame::from_cvm_value)
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168    use serde_json::json;
169
170    fn assert_cvm_roundtrip(frame: OpenStreamFrame) {
171        let value = frame.to_cvm_value().unwrap();
172        assert_eq!(value["type"], json!("open-stream"));
173        assert_eq!(value["frameType"], json!(frame.frame_type()));
174        assert_eq!(OpenStreamFrame::from_cvm_value(&value), Some(frame));
175    }
176
177    #[test]
178    fn all_frame_variants_roundtrip() {
179        assert_cvm_roundtrip(OpenStreamFrame::Start {
180            content_type: Some("text/plain".to_string()),
181        });
182        assert_cvm_roundtrip(OpenStreamFrame::Start { content_type: None });
183        assert_cvm_roundtrip(OpenStreamFrame::Accept);
184        assert_cvm_roundtrip(OpenStreamFrame::Chunk {
185            chunk_index: 7,
186            data: "payload".to_string(),
187        });
188        assert_cvm_roundtrip(OpenStreamFrame::Ping {
189            nonce: "tok:1".to_string(),
190        });
191        assert_cvm_roundtrip(OpenStreamFrame::Pong {
192            nonce: "tok:1".to_string(),
193        });
194        assert_cvm_roundtrip(OpenStreamFrame::Close {
195            last_chunk_index: Some(4),
196        });
197        assert_cvm_roundtrip(OpenStreamFrame::Close {
198            last_chunk_index: None,
199        });
200        assert_cvm_roundtrip(OpenStreamFrame::Abort {
201            reason: Some("boom".to_string()),
202        });
203        assert_cvm_roundtrip(OpenStreamFrame::Abort { reason: None });
204    }
205
206    #[test]
207    fn chunk_frame_serializes_with_camel_case_fields() {
208        let value = OpenStreamFrame::Chunk {
209            chunk_index: 3,
210            data: "abc".to_string(),
211        }
212        .to_cvm_value()
213        .unwrap();
214        assert_eq!(value["type"], json!("open-stream"));
215        assert_eq!(value["frameType"], json!("chunk"));
216        assert_eq!(value["chunkIndex"], json!(3));
217        assert_eq!(value["data"], json!("abc"));
218        // snake_case field name must not leak onto the wire.
219        assert!(!value.as_object().unwrap().contains_key("chunk_index"));
220    }
221
222    #[test]
223    fn start_content_type_serializes_camel_case_and_omits_when_none() {
224        let with = OpenStreamFrame::Start {
225            content_type: Some("application/json".to_string()),
226        }
227        .to_cvm_value()
228        .unwrap();
229        assert_eq!(with["contentType"], json!("application/json"));
230
231        let without = OpenStreamFrame::Start { content_type: None }
232            .to_cvm_value()
233            .unwrap();
234        assert!(!without.as_object().unwrap().contains_key("contentType"));
235    }
236
237    #[test]
238    fn close_last_chunk_index_serializes_camel_case_and_omits_when_none() {
239        let with = OpenStreamFrame::Close {
240            last_chunk_index: Some(9),
241        }
242        .to_cvm_value()
243        .unwrap();
244        assert_eq!(with["lastChunkIndex"], json!(9));
245
246        let without = OpenStreamFrame::Close {
247            last_chunk_index: None,
248        }
249        .to_cvm_value()
250        .unwrap();
251        assert!(!without.as_object().unwrap().contains_key("lastChunkIndex"));
252    }
253
254    #[test]
255    fn abort_reason_omitted_when_none() {
256        let value = OpenStreamFrame::Abort { reason: None }
257            .to_cvm_value()
258            .unwrap();
259        assert!(!value.as_object().unwrap().contains_key("reason"));
260    }
261
262    #[test]
263    fn from_cvm_value_rejects_wrong_type() {
264        let value = json!({ "type": "oversized-transfer", "frameType": "start" });
265        assert_eq!(OpenStreamFrame::from_cvm_value(&value), None);
266        assert!(!OpenStreamFrame::is_frame_value(&value));
267    }
268
269    #[test]
270    fn from_cvm_value_rejects_unknown_frame_type() {
271        let value = json!({ "type": "open-stream", "frameType": "bogus" });
272        assert_eq!(OpenStreamFrame::from_cvm_value(&value), None);
273    }
274
275    #[test]
276    fn is_frame_value_requires_type_and_frame_type() {
277        assert!(OpenStreamFrame::is_frame_value(
278            &json!({ "type": "open-stream", "frameType": "chunk", "chunkIndex": 0, "data": "x" })
279        ));
280        assert!(!OpenStreamFrame::is_frame_value(
281            &json!({ "type": "open-stream" })
282        ));
283        assert!(!OpenStreamFrame::is_frame_value(
284            &json!({ "frameType": "chunk" })
285        ));
286        assert!(!OpenStreamFrame::is_frame_value(&json!("not an object")));
287    }
288
289    #[test]
290    fn into_progress_notification_builds_progress_envelope() {
291        let notification = OpenStreamFrame::Chunk {
292            chunk_index: 2,
293            data: "frag".to_string(),
294        }
295        .into_progress_notification("tok-1", 7, Some("hi"))
296        .unwrap();
297
298        assert_eq!(notification.method, "notifications/progress");
299        let params = notification.params.as_ref().unwrap();
300        // progressToken is always a string on the wire.
301        assert_eq!(params["progressToken"], json!("tok-1"));
302        // progress is numeric.
303        assert_eq!(params["progress"], json!(7));
304        assert_eq!(params["message"], json!("hi"));
305        assert_eq!(params["cvm"]["frameType"], json!("chunk"));
306        assert_eq!(params["cvm"]["chunkIndex"], json!(2));
307        assert_eq!(params["cvm"]["data"], json!("frag"));
308        assert_eq!(params["cvm"]["type"], json!("open-stream"));
309    }
310
311    #[test]
312    fn into_progress_notification_omits_absent_message() {
313        let notification = OpenStreamFrame::Accept
314            .into_progress_notification("tok-1", 9, None)
315            .unwrap();
316        let params = notification.params.as_ref().unwrap();
317        assert!(!params.as_object().unwrap().contains_key("message"));
318    }
319
320    #[test]
321    fn open_stream_frame_from_notification_extracts_typed_frame() {
322        let notification = OpenStreamFrame::Ping {
323            nonce: "n".to_string(),
324        }
325        .into_progress_notification("tok", 1, None)
326        .unwrap();
327        assert_eq!(
328            open_stream_frame_from_notification(&notification),
329            Some(OpenStreamFrame::Ping {
330                nonce: "n".to_string()
331            })
332        );
333
334        // A plain (non-cvm) progress notification yields None.
335        let plain = JsonRpcNotification {
336            jsonrpc: "2.0".to_string(),
337            method: "notifications/progress".to_string(),
338            params: Some(json!({ "progressToken": "t", "progress": 3 })),
339        };
340        assert_eq!(open_stream_frame_from_notification(&plain), None);
341    }
342}
contextvm_sdk/transport/open_stream/frame.rs

contextvm_sdk/transport/open_stream/
frame.rs
