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/// Envelope metadata plus an ordered list of staged operations.
7///
8/// Operation order is semantically load-bearing (a `link` may target an
9/// earlier `create` ID) and is preserved exactly by the NDJSON codec.
10/// See `crates/khive-changeset/docs/api/ndjson-codec.md` for the wire contract.
11#[derive(Clone, Debug)]
12pub struct ChangeSet {
13    /// Change-set-wide stage-time provenance.
14    pub envelope: Envelope,
15    /// Operations in semantic application order.
16    pub ops: Vec<Op>,
17}
18
19impl ChangeSet {
20    /// Stores `envelope` and `ops` without reordering or validation.
21    pub fn new(envelope: Envelope, ops: Vec<Op>) -> Self {
22        Self { envelope, ops }
23    }
24}
25
26/// In-memory NDJSON encode/decode errors; no variant represents I/O.
27///
28/// See `crates/khive-changeset/docs/api/ndjson-codec.md` for variant semantics.
29#[derive(Debug, thiserror::Error)]
30pub enum ChangeSetError {
31    #[error("NDJSON-delta input is empty; expected an envelope header line")]
32    Empty,
33    #[error("line {line} is not valid JSON: {source}")]
34    MalformedLine {
35        line: usize,
36        #[source]
37        source: serde_json::Error,
38    },
39    #[error(
40        "envelope schema_version {found} is not supported (this crate emits and \
41         accepts schema_version {expected})"
42    )]
43    UnsupportedSchemaVersion { found: u32, expected: u32 },
44    #[error("failed to serialize change-set: {0}")]
45    Serialize(#[source] serde_json::Error),
46}
47
48/// Encodes the envelope as line one and each op on a following line, in order.
49///
50/// The returned in-memory string ends every line, including the last, with `\n`.
51///
52/// # Errors
53///
54/// Returns [`ChangeSetError::Serialize`] if any envelope or op cannot serialize.
55pub fn to_ndjson(changeset: &ChangeSet) -> Result<String, ChangeSetError> {
56    let mut out = String::new();
57    out.push_str(&serde_json::to_string(&changeset.envelope).map_err(ChangeSetError::Serialize)?);
58    out.push('\n');
59    for op in &changeset.ops {
60        out.push_str(&serde_json::to_string(op).map_err(ChangeSetError::Serialize)?);
61        out.push('\n');
62    }
63    Ok(out)
64}
65
66/// Decodes an envelope header and every following line as one ordered [`Op`].
67///
68/// Blank lines and unknown fields are errors; schema version must equal
69/// [`CURRENT_SCHEMA_VERSION`].
70///
71/// # Errors
72///
73/// Returns [`ChangeSetError::Empty`], [`ChangeSetError::MalformedLine`], or
74/// [`ChangeSetError::UnsupportedSchemaVersion`] according to the failed stage.
75/// See `crates/khive-changeset/docs/api/ndjson-codec.md` for line semantics.
76pub fn from_ndjson(input: &str) -> Result<ChangeSet, ChangeSetError> {
77    let mut lines = input.lines().enumerate();
78    let (_, first_line) = lines.next().ok_or(ChangeSetError::Empty)?;
79    let envelope: Envelope = serde_json::from_str(first_line)
80        .map_err(|source| ChangeSetError::MalformedLine { line: 1, source })?;
81    if envelope.schema_version != CURRENT_SCHEMA_VERSION {
82        return Err(ChangeSetError::UnsupportedSchemaVersion {
83            found: envelope.schema_version,
84            expected: CURRENT_SCHEMA_VERSION,
85        });
86    }
87
88    // Parse blank lines too, preserving “every post-header line is an op.”
89    let mut ops = Vec::new();
90    for (idx, line) in lines {
91        let op: Op =
92            serde_json::from_str(line).map_err(|source| ChangeSetError::MalformedLine {
93                line: idx + 1,
94                source,
95            })?;
96        ops.push(op);
97    }
98    Ok(ChangeSet { envelope, ops })
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use crate::op::{CreateOp, CreateTarget, EntityCreateFields};
105    use khive_types::{EntityKind, Id128, Namespace, Timestamp};
106
107    fn sample_changeset() -> ChangeSet {
108        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
109        let create = Op::Create(CreateOp {
110            id: Id128::from_u128(1),
111            namespace: Namespace::local(),
112            target: CreateTarget::Entity(EntityCreateFields {
113                entity_kind: EntityKind::Concept,
114                entity_type: None,
115                name: "X".into(),
116                description: None,
117                properties: Default::default(),
118                tags: vec![],
119            }),
120        });
121        ChangeSet::new(envelope, vec![create])
122    }
123
124    #[test]
125    fn round_trips_envelope_and_ops() {
126        let cs = sample_changeset();
127        let text = to_ndjson(&cs).unwrap();
128        let decoded = from_ndjson(&text).unwrap();
129        assert_eq!(decoded.envelope, cs.envelope);
130        assert_eq!(decoded.ops.len(), 1);
131        let text2 = to_ndjson(&decoded).unwrap();
132        assert_eq!(text, text2, "re-serialization must be byte-identical");
133    }
134
135    #[test]
136    fn envelope_is_line_one() {
137        let cs = sample_changeset();
138        let text = to_ndjson(&cs).unwrap();
139        let first_line = text.lines().next().unwrap();
140        let parsed_env: Envelope = serde_json::from_str(first_line).unwrap();
141        assert_eq!(parsed_env, cs.envelope);
142    }
143
144    #[test]
145    fn envelope_batch_id_round_trips_through_ndjson() {
146        let mut cs = sample_changeset();
147        cs.envelope = cs.envelope.with_batch_id("batch-xyz");
148        let text = to_ndjson(&cs).unwrap();
149        let decoded = from_ndjson(&text).unwrap();
150        assert_eq!(decoded.envelope.batch_id.as_deref(), Some("batch-xyz"));
151        assert_eq!(decoded.envelope, cs.envelope);
152    }
153
154    #[test]
155    fn envelope_without_batch_id_round_trips_through_ndjson() {
156        let cs = sample_changeset();
157        assert_eq!(cs.envelope.batch_id, None);
158        let text = to_ndjson(&cs).unwrap();
159        let decoded = from_ndjson(&text).unwrap();
160        assert_eq!(decoded.envelope.batch_id, None);
161        assert_eq!(decoded.envelope, cs.envelope);
162    }
163
164    #[test]
165    fn empty_input_errors() {
166        let result = from_ndjson("");
167        assert!(matches!(result, Err(ChangeSetError::Empty)));
168    }
169
170    #[test]
171    fn malformed_op_line_reports_correct_line_number() {
172        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
173        let env_line = serde_json::to_string(&envelope).unwrap();
174        let text = format!("{env_line}\n{{\"op\": \"create\", not json}}\n");
175        let err = from_ndjson(&text).unwrap_err();
176        match err {
177            ChangeSetError::MalformedLine { line, .. } => assert_eq!(line, 2),
178            other => panic!("expected MalformedLine, got {other:?}"),
179        }
180    }
181
182    #[test]
183    fn blank_line_between_ops_is_malformed() {
184        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
185        let env_line = serde_json::to_string(&envelope).unwrap();
186        let text = format!("{env_line}\n\n");
187        let err = from_ndjson(&text).unwrap_err();
188        match err {
189            ChangeSetError::MalformedLine { line, .. } => assert_eq!(line, 2),
190            other => panic!("expected MalformedLine, got {other:?}"),
191        }
192    }
193
194    #[test]
195    fn unsupported_schema_version_is_rejected() {
196        let mut envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
197        envelope.schema_version = 99;
198        let env_line = serde_json::to_string(&envelope).unwrap();
199        let err = from_ndjson(&env_line).unwrap_err();
200        assert!(matches!(
201            err,
202            ChangeSetError::UnsupportedSchemaVersion {
203                found: 99,
204                expected: CURRENT_SCHEMA_VERSION
205            }
206        ));
207    }
208
209    #[test]
210    fn op_order_is_preserved() {
211        let envelope = Envelope::new("agent:test", "family:sonnet", Timestamp::from_secs(1));
212        let mk_create = |n: u128| {
213            Op::Create(CreateOp {
214                id: Id128::from_u128(n),
215                namespace: Namespace::local(),
216                target: CreateTarget::Entity(EntityCreateFields {
217                    entity_kind: EntityKind::Concept,
218                    entity_type: None,
219                    name: format!("entity-{n}"),
220                    description: None,
221                    properties: Default::default(),
222                    tags: vec![],
223                }),
224            })
225        };
226        let ops = vec![mk_create(3), mk_create(1), mk_create(2)];
227        let cs = ChangeSet::new(envelope, ops);
228        let text = to_ndjson(&cs).unwrap();
229        let decoded = from_ndjson(&text).unwrap();
230        let ids: Vec<u128> = decoded
231            .ops
232            .iter()
233            .map(|op| match op {
234                Op::Create(c) => c.id.to_u128(),
235                _ => unreachable!(),
236            })
237            .collect();
238        assert_eq!(ids, vec![3, 1, 2]);
239    }
240}
khive_changeset/changeset.rs

khive_changeset/
changeset.rs
