khive_changeset/
changeset.rs

1//! The change-set: an envelope plus its ordered op-list, and NDJSON-delta codec.
2
3use crate::envelope::{Envelope, CURRENT_SCHEMA_VERSION};
4use crate::op::Op;
5
6/// A staged change-set: envelope metadata plus an ordered list of operations.
7///
8/// Operation order is semantically load-bearing (a `link` may target an
9/// earlier `create`'s stage-time id) and is preserved exactly through
10/// [`to_ndjson`] / [`from_ndjson`].
11#[derive(Clone, Debug)]
12pub struct ChangeSet {
13    pub envelope: Envelope,
14    pub ops: Vec<Op>,
15}
16
17impl ChangeSet {
18    pub fn new(envelope: Envelope, ops: Vec<Op>) -> Self {
19        Self { envelope, ops }
20    }
21}
22
23/// Errors from NDJSON-delta encode/decode. No variant touches the filesystem
24/// or performs any I/O — every value here is constructed from in-memory input.
25#[derive(Debug, thiserror::Error)]
26pub enum ChangeSetError {
27    #[error("NDJSON-delta input is empty; expected an envelope header line")]
28    Empty,
29    #[error("line {line} is not valid JSON: {source}")]
30    MalformedLine {
31        line: usize,
32        #[source]
33        source: serde_json::Error,
34    },
35    #[error(
36        "envelope schema_version {found} is not supported (this crate emits and \
37         accepts schema_version {expected})"
38    )]
39    UnsupportedSchemaVersion { found: u32, expected: u32 },
40    #[error("failed to serialize change-set: {0}")]
41    Serialize(#[source] serde_json::Error),
42}
43
44/// Encode a [`ChangeSet`] as NDJSON-delta: the envelope as line 1, then one
45/// line per op in stage order. No filesystem access — returns an in-memory
46/// `String`; the caller decides what to do with it.
47pub fn to_ndjson(changeset: &ChangeSet) -> Result<String, ChangeSetError> {
48    let mut out = String::new();
49    out.push_str(&serde_json::to_string(&changeset.envelope).map_err(ChangeSetError::Serialize)?);
50    out.push('\n');
51    for op in &changeset.ops {
52        out.push_str(&serde_json::to_string(op).map_err(ChangeSetError::Serialize)?);
53        out.push('\n');
54    }
55    Ok(out)
56}
57
58/// Decode NDJSON-delta text into a [`ChangeSet`]. Line 1 must parse as the
59/// envelope; every subsequent non-empty line must parse as one [`Op`], in
60/// stage order. Rejects an envelope whose `schema_version` this crate does
61/// not recognize, and rejects any line carrying an unknown field (fail-loud,
62/// matching the `deny_unknown_fields` posture the rest of the codebase uses
63/// for configuration surfaces).
64pub fn from_ndjson(input: &str) -> Result<ChangeSet, ChangeSetError> {
65    let mut lines = input.lines().enumerate();
66    let (_, first_line) = lines.next().ok_or(ChangeSetError::Empty)?;
67    let envelope: Envelope = serde_json::from_str(first_line)
68        .map_err(|source| ChangeSetError::MalformedLine { line: 1, source })?;
69    if envelope.schema_version != CURRENT_SCHEMA_VERSION {
70        return Err(ChangeSetError::UnsupportedSchemaVersion {
71            found: envelope.schema_version,
72            expected: CURRENT_SCHEMA_VERSION,
73        });
74    }
75
76    // A blank line (including a genuinely empty one) is not skipped: it is
77    // fed to the same Op parser as every other line and rejected with the
78    // same MalformedLine shape, keeping "every line must be a valid op"
79    // exception-free rather than special-casing whitespace.
80    let mut ops = Vec::new();
81    for (idx, line) in lines {
82        let op: Op =
83            serde_json::from_str(line).map_err(|source| ChangeSetError::MalformedLine {
84                line: idx + 1,
85                source,
86            })?;
87        ops.push(op);
88    }
89    Ok(ChangeSet { envelope, ops })
90}
91
92#[cfg(test)]
93mod tests {
94    use super::*;
95    use crate::op::{CreateOp, CreateTarget, EntityCreateFields};
96    use khive_types::{EntityKind, Id128, Namespace, Timestamp};
97
98    fn sample_changeset() -> ChangeSet {
99        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
100        let create = Op::Create(CreateOp {
101            id: Id128::from_u128(1),
102            namespace: Namespace::local(),
103            target: CreateTarget::Entity(EntityCreateFields {
104                entity_kind: EntityKind::Concept,
105                entity_type: None,
106                name: "X".into(),
107                description: None,
108                properties: Default::default(),
109                tags: vec![],
110            }),
111        });
112        ChangeSet::new(envelope, vec![create])
113    }
114
115    #[test]
116    fn round_trips_envelope_and_ops() {
117        let cs = sample_changeset();
118        let text = to_ndjson(&cs).unwrap();
119        let decoded = from_ndjson(&text).unwrap();
120        assert_eq!(decoded.envelope, cs.envelope);
121        assert_eq!(decoded.ops.len(), 1);
122        let text2 = to_ndjson(&decoded).unwrap();
123        assert_eq!(text, text2, "re-serialization must be byte-identical");
124    }
125
126    #[test]
127    fn envelope_is_line_one() {
128        let cs = sample_changeset();
129        let text = to_ndjson(&cs).unwrap();
130        let first_line = text.lines().next().unwrap();
131        let parsed_env: Envelope = serde_json::from_str(first_line).unwrap();
132        assert_eq!(parsed_env, cs.envelope);
133    }
134
135    #[test]
136    fn envelope_batch_id_round_trips_through_ndjson() {
137        let mut cs = sample_changeset();
138        cs.envelope = cs.envelope.with_batch_id("batch-xyz");
139        let text = to_ndjson(&cs).unwrap();
140        let decoded = from_ndjson(&text).unwrap();
141        assert_eq!(decoded.envelope.batch_id.as_deref(), Some("batch-xyz"));
142        assert_eq!(decoded.envelope, cs.envelope);
143    }
144
145    #[test]
146    fn envelope_without_batch_id_round_trips_through_ndjson() {
147        let cs = sample_changeset();
148        assert_eq!(cs.envelope.batch_id, None);
149        let text = to_ndjson(&cs).unwrap();
150        let decoded = from_ndjson(&text).unwrap();
151        assert_eq!(decoded.envelope.batch_id, None);
152        assert_eq!(decoded.envelope, cs.envelope);
153    }
154
155    #[test]
156    fn empty_input_errors() {
157        let result = from_ndjson("");
158        assert!(matches!(result, Err(ChangeSetError::Empty)));
159    }
160
161    #[test]
162    fn malformed_op_line_reports_correct_line_number() {
163        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
164        let env_line = serde_json::to_string(&envelope).unwrap();
165        let text = format!("{env_line}\n{{\"op\": \"create\", not json}}\n");
166        let err = from_ndjson(&text).unwrap_err();
167        match err {
168            ChangeSetError::MalformedLine { line, .. } => assert_eq!(line, 2),
169            other => panic!("expected MalformedLine, got {other:?}"),
170        }
171    }
172
173    #[test]
174    fn blank_line_between_ops_is_malformed() {
175        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
176        let env_line = serde_json::to_string(&envelope).unwrap();
177        let text = format!("{env_line}\n\n");
178        let err = from_ndjson(&text).unwrap_err();
179        match err {
180            ChangeSetError::MalformedLine { line, .. } => assert_eq!(line, 2),
181            other => panic!("expected MalformedLine, got {other:?}"),
182        }
183    }
184
185    #[test]
186    fn unsupported_schema_version_is_rejected() {
187        let mut envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
188        envelope.schema_version = 99;
189        let env_line = serde_json::to_string(&envelope).unwrap();
190        let err = from_ndjson(&env_line).unwrap_err();
191        assert!(matches!(
192            err,
193            ChangeSetError::UnsupportedSchemaVersion {
194                found: 99,
195                expected: CURRENT_SCHEMA_VERSION
196            }
197        ));
198    }
199
200    #[test]
201    fn op_order_is_preserved() {
202        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
203        let mk_create = |n: u128| {
204            Op::Create(CreateOp {
205                id: Id128::from_u128(n),
206                namespace: Namespace::local(),
207                target: CreateTarget::Entity(EntityCreateFields {
208                    entity_kind: EntityKind::Concept,
209                    entity_type: None,
210                    name: format!("entity-{n}"),
211                    description: None,
212                    properties: Default::default(),
213                    tags: vec![],
214                }),
215            })
216        };
217        let ops = vec![mk_create(3), mk_create(1), mk_create(2)];
218        let cs = ChangeSet::new(envelope, ops);
219        let text = to_ndjson(&cs).unwrap();
220        let decoded = from_ndjson(&text).unwrap();
221        let ids: Vec<u128> = decoded
222            .ops
223            .iter()
224            .map(|op| match op {
225                Op::Create(c) => c.id.to_u128(),
226                _ => unreachable!(),
227            })
228            .collect();
229        assert_eq!(ids, vec![3, 1, 2]);
230    }
231}
khive_changeset/changeset.rs

khive_changeset/
changeset.rs
