wm_memory/
envelope.rs

1//! Envelope v2 — one validator, three uses (V8 S4).
2//!
3//! Every bulk record stream this store produces carries an optional
4//! single-line JSON envelope header as its first line:
5//!
6//! ```text
7//! {"wm_envelope":{"count":5,"created_at":"2026-08-31T16:00:00+00:00","format_version":2,"generator":"wm 9.0.0","kind":"session_export"}}
8//! ```
9//!
10//! Rules:
11//! - **Writers always emit the header; readers accept streams with or
12//!   without one.** Bare v1 payloads (plain record JSONL) stay importable
13//!   forever — the header is additive, never a format break.
14//! - The `wm_envelope` top-level key is the discriminator. Record types
15//!   (`Memory` et al.) never serialize that key, so a header line can never
16//!   collide with a record line.
17//! - A header with `format_version` newer than [`ENVELOPE_FORMAT_VERSION`]
18//!   is refused: a forward stream may carry records this build cannot
19//!   parse honestly, and silently skipping records is the failure mode
20//!   this module exists to prevent.
21//! - `count` is advisory but checked: a mismatch is a warning, never a
22//!   refusal (partial streams are still worth importing).
23//!
24//! Uses: `session.export` (header line in the JSONL stream),
25//! `session.import` (validation), `wm backup`/`wm restore` (envelope.json
26//! beside SHA256SUMS). All three go through this module — there is no
27//! second implementation.
28
29/// The envelope format this build writes. Readers accept this version and
30/// any older header; newer headers are refused.
31pub const ENVELOPE_FORMAT_VERSION: u32 = 2;
32
33/// Top-level discriminator key on the header line.
34pub const ENVELOPE_KEY: &str = "wm_envelope";
35
36/// Header of an enveloped record stream.
37#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
38pub struct EnvelopeHeader {
39    /// Stream format version. This build writes [`ENVELOPE_FORMAT_VERSION`].
40    pub format_version: u32,
41    /// What the stream carries: `"session_export"`, `"store_backup"`, ...
42    pub kind: String,
43    /// RFC 3339 creation timestamp of the stream.
44    pub created_at: String,
45    /// Declared record count (advisory; validated as a warning on mismatch).
46    pub count: usize,
47    /// Producing binary and version, e.g. `"wm 9.0.0"`.
48    pub generator: String,
49}
50
51impl EnvelopeHeader {
52    /// A header for a stream being written right now.
53    #[must_use]
54    pub fn new(kind: &str, count: usize) -> Self {
55        Self {
56            format_version: ENVELOPE_FORMAT_VERSION,
57            kind: kind.to_string(),
58            created_at: chrono::Utc::now().to_rfc3339(),
59            count,
60            generator: format!("wm {}", env!("CARGO_PKG_VERSION")),
61        }
62    }
63
64    /// The header as a single JSON line for stream prefixes.
65    #[must_use]
66    pub fn header_line(&self) -> String {
67        let mut v = serde_json::to_value(self).unwrap_or_else(|_| serde_json::json!({}));
68        // Re-wrap under the discriminator key so the line is
69        // distinguishable from any record line.
70        let inner = v.take();
71        serde_json::to_string(&serde_json::json!({ ENVELOPE_KEY: inner }))
72            .unwrap_or_else(|_| format!("{{\"{ENVELOPE_KEY}\":null}}"))
73    }
74
75    /// Validate `format_version` against this build. Newer formats are
76    /// refused with an actionable message; equal or older are accepted.
77    ///
78    /// # Errors
79    /// When the header declares a format newer than this build supports.
80    pub fn check_version(&self) -> Result<(), String> {
81        if self.format_version > ENVELOPE_FORMAT_VERSION {
82            return Err(format!(
83                "envelope format_version {} is newer than this build supports ({}); \
84                 upgrade `wm` to read this stream — refusing to import it partially",
85                self.format_version, ENVELOPE_FORMAT_VERSION
86            ));
87        }
88        Ok(())
89    }
90}
91
92/// Result of reading a header line.
93#[derive(Debug, Clone, PartialEq, Eq)]
94pub enum HeaderRead {
95    /// The line is not a header — the stream is bare (v1) JSONL.
96    NotAHeader,
97    /// Valid header.
98    Header(EnvelopeHeader),
99    /// Header-shaped but refused (newer format, malformed).
100    Refused(String),
101}
102
103/// Read the first non-empty line of a stream as a potential header.
104///
105/// Returns [`HeaderRead::NotAHeader`] when the line parses as JSON but has
106/// no `wm_envelope` key, or is not JSON at all (a bare v1 record).
107#[must_use]
108pub fn read_header_line(line: &str) -> HeaderRead {
109    let trimmed = line.trim();
110    let Ok(v) = serde_json::from_str::<serde_json::Value>(trimmed) else {
111        return HeaderRead::NotAHeader;
112    };
113    let Some(inner) = v.get(ENVELOPE_KEY) else {
114        return HeaderRead::NotAHeader;
115    };
116    if inner.is_null() {
117        return HeaderRead::Refused(format!(
118            "envelope header present but malformed (null {ENVELOPE_KEY})"
119        ));
120    }
121    match serde_json::from_value::<EnvelopeHeader>(inner.clone()) {
122        Ok(h) => match h.check_version() {
123            Ok(()) => HeaderRead::Header(h),
124            Err(msg) => HeaderRead::Refused(msg),
125        },
126        Err(e) => HeaderRead::Refused(format!(
127            "envelope header failed to parse: {e} (required: format_version, kind, \
128             created_at, count, generator)"
129        )),
130    }
131}
132
133/// Scan result for a whole stream.
134#[derive(Debug, Default, PartialEq, Eq)]
135pub struct StreamScan {
136    /// Header, when the stream carries one.
137    pub header: Option<EnvelopeHeader>,
138    /// Non-fatal findings (count mismatch, skipped lines).
139    pub warnings: Vec<String>,
140    /// JSON-parseable record lines seen (header excluded).
141    pub records: usize,
142    /// 1-based line numbers (in the original payload) whose JSON did not
143    /// parse as an object.
144    pub unparseable_lines: Vec<usize>,
145}
146
147/// Scan a whole stream: header, per-line JSON validity, record count.
148///
149/// Line-shape checks beyond JSON-object are the caller's business (the
150/// import tool deserializes `Memory`; backup validates files differently).
151/// The header line, when present, must be the first non-empty line.
152#[must_use]
153pub fn scan_stream(payload: &str) -> StreamScan {
154    let mut scan = StreamScan::default();
155    let mut header_checked = false;
156    for (idx, line) in payload.lines().enumerate() {
157        if line.trim().is_empty() {
158            continue;
159        }
160        if !header_checked {
161            header_checked = true;
162            match read_header_line(line) {
163                HeaderRead::NotAHeader => { /* bare v1 stream; this line is a record */ }
164                HeaderRead::Header(h) => {
165                    scan.header = Some(h);
166                    continue;
167                }
168                HeaderRead::Refused(msg) => {
169                    scan.warnings.push(msg);
170                    continue;
171                }
172            }
173        }
174        match serde_json::from_str::<serde_json::Value>(line.trim()) {
175            Ok(v) if v.is_object() => scan.records += 1,
176            _ => scan.unparseable_lines.push(idx + 1),
177        }
178    }
179    if let Some(h) = &scan.header {
180        if h.count != scan.records {
181            scan.warnings.push(format!(
182                "envelope declares count {} but stream carries {} records",
183                h.count, scan.records
184            ));
185        }
186    }
187    if !scan.unparseable_lines.is_empty() {
188        scan.warnings.push(format!(
189            "{} line(s) skipped as unparseable JSON at lines {:?}",
190            scan.unparseable_lines.len(),
191            scan.unparseable_lines
192        ));
193    }
194    scan
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200
201    #[test]
202    fn header_line_roundtrips_through_read() {
203        let h = EnvelopeHeader::new("session_export", 5);
204        let line = h.header_line();
205        assert!(line.contains(ENVELOPE_KEY));
206        match read_header_line(&line) {
207            HeaderRead::Header(parsed) => {
208                assert_eq!(parsed, h);
209                assert_eq!(parsed.format_version, ENVELOPE_FORMAT_VERSION);
210                assert_eq!(parsed.kind, "session_export");
211                assert_eq!(parsed.count, 5);
212            }
213            other => panic!("expected header, got {other:?}"),
214        }
215    }
216
217    #[test]
218    fn bare_record_line_is_not_a_header() {
219        let mem = serde_json::json!({
220            "metadata": {"id": "00000000-0000-0000-0000-000000000000"},
221            "content": "old memory",
222            "embedding": null
223        });
224        let line = serde_json::to_string(&mem).unwrap();
225        assert_eq!(read_header_line(&line), HeaderRead::NotAHeader);
226    }
227
228    #[test]
229    fn non_json_line_is_not_a_header() {
230        assert_eq!(read_header_line("not json at all"), HeaderRead::NotAHeader);
231    }
232
233    #[test]
234    fn newer_format_version_is_refused() {
235        let h = EnvelopeHeader {
236            format_version: ENVELOPE_FORMAT_VERSION + 1,
237            kind: "session_export".into(),
238            created_at: chrono::Utc::now().to_rfc3339(),
239            count: 1,
240            generator: "wm 99.0.0".into(),
241        };
242        match read_header_line(&h.header_line()) {
243            HeaderRead::Refused(msg) => {
244                assert!(msg.contains("newer than this build supports"), "{msg}");
245            }
246            other => panic!("expected refusal, got {other:?}"),
247        }
248    }
249
250    #[test]
251    fn malformed_header_is_refused_with_field_names() {
252        let line = serde_json::to_string(&serde_json::json!({
253            ENVELOPE_KEY: {"format_version": 2}
254        }))
255        .unwrap();
256        match read_header_line(&line) {
257            HeaderRead::Refused(msg) => assert!(msg.contains("required"), "{msg}"),
258            other => panic!("expected refusal, got {other:?}"),
259        }
260    }
261
262    #[test]
263    fn scan_stream_v2_payload() {
264        let rec1 = serde_json::json!({"metadata": {}, "content": "a"}).to_string();
265        let rec2 = serde_json::json!({"metadata": {}, "content": "b"}).to_string();
266        let payload = format!(
267            "{}\n{rec1}\n{rec2}\n",
268            EnvelopeHeader::new("session_export", 2).header_line()
269        );
270        let scan = scan_stream(&payload);
271        assert!(scan.header.is_some());
272        assert_eq!(scan.records, 2);
273        assert!(scan.warnings.is_empty(), "{:?}", scan.warnings);
274        assert!(scan.unparseable_lines.is_empty());
275    }
276
277    #[test]
278    fn scan_stream_bare_v1_payload_has_no_header() {
279        let rec1 = serde_json::json!({"metadata": {}, "content": "a"}).to_string();
280        let payload = format!("{rec1}\n");
281        let scan = scan_stream(&payload);
282        assert!(scan.header.is_none());
283        assert_eq!(scan.records, 1);
284        assert!(scan.warnings.is_empty());
285    }
286
287    #[test]
288    fn scan_stream_flags_count_mismatch_and_bad_lines() {
289        let good = serde_json::json!({"metadata": {}, "content": "a"}).to_string();
290        let payload = format!(
291            "{}\n{good}\n{{\"broken\":\nnot json\n",
292            EnvelopeHeader::new("session_export", 3).header_line()
293        );
294        let scan = scan_stream(&payload);
295        assert_eq!(scan.records, 1);
296        assert!(
297            scan.warnings
298                .iter()
299                .any(|w| w.contains("declares count 3 but stream carries 1"))
300        );
301        assert!(scan.warnings.iter().any(|w| w.contains("unparseable JSON")));
302        assert_eq!(scan.unparseable_lines, vec![3, 4]);
303    }
304
305    #[test]
306    fn header_survives_backup_json_roundtrip() {
307        // `wm backup` writes the header as a JSON object (envelope.json);
308        // the same struct must deserialize back identically.
309        let h = EnvelopeHeader::new("store_backup", 42);
310        let json = serde_json::to_string_pretty(&h).unwrap();
311        let parsed: EnvelopeHeader = serde_json::from_str(&json).unwrap();
312        assert_eq!(parsed, h);
313    }
314}
wm_memory/envelope.rs

wm_memory/
envelope.rs
