heddle_object_model/op_record/
codec.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Versioned codec for current OpRecord payloads stored in the packed oplog.
3//!
4//! Repository format v3 changed state identity from 16-byte logical ChangeIds
5//! to 32-byte content-addressed StateIds. That boundary is intentionally not
6//! migratable: repository open refuses older formats, and the oplog codec
7//! refuses record schema versions 1 through 3 without rewriting their bytes.
8//! Future incompatible record changes must allocate a new schema version.
9
10use crate::{
11    error::{HeddleError, Result},
12    object::{Attribution, ContentHash, StateId, VisibilityTier},
13};
14use serde::Deserialize;
15
16use super::{ConflictResolutionMode, OpRecord, RecordedHead, ThreadUpdateSnapshots};
17
18pub const CURRENT_OP_RECORD_SCHEMA_VERSION: u32 = 4;
19const CURRENT_OP_RECORD_SCHEMA_NAME: &str = "state-id-v4";
20const OP_RECORD_STORAGE: &str = "oplog record schema";
21
22pub fn validate_op_record_schema_version(version: u32) -> Result<()> {
23    if version < CURRENT_OP_RECORD_SCHEMA_VERSION {
24        return Err(HeddleError::StorageFormatMigrationRequired {
25            storage: OP_RECORD_STORAGE.to_string(),
26            found: version,
27            required: CURRENT_OP_RECORD_SCHEMA_VERSION,
28        });
29    }
30    if version > CURRENT_OP_RECORD_SCHEMA_VERSION {
31        return Err(HeddleError::StorageFormatTooNew {
32            storage: OP_RECORD_STORAGE.to_string(),
33            found: version,
34            supported: CURRENT_OP_RECORD_SCHEMA_VERSION,
35        });
36    }
37    Ok(())
38}
39
40pub fn decode_current_record(bytes: &[u8]) -> Result<OpRecord> {
41    let record: StrictCurrentOpRecord = decode_rmp(bytes, CURRENT_OP_RECORD_SCHEMA_NAME)?;
42    Ok(record.into_current())
43}
44
45pub fn encode_current_record(record: &OpRecord) -> Result<Vec<u8>> {
46    rmp_serde::to_vec(record).map_err(|e| HeddleError::Serialization(e.to_string()))
47}
48
49fn decode_rmp<T>(bytes: &[u8], schema_name: &str) -> Result<T>
50where
51    T: for<'de> Deserialize<'de>,
52{
53    rmp_serde::from_slice(bytes).map_err(|e| {
54        HeddleError::Serialization(format!(
55            "failed to decode OpRecord payload as {schema_name}: {e}"
56        ))
57    })
58}
59
60/// Strict snapshot of record schema version 4.
61///
62/// This mirror has no defaults for structural fields: malformed or older
63/// positional payloads fail instead of being interpreted as current records.
64#[derive(Debug, Clone, Deserialize)]
65enum StrictCurrentOpRecord {
66    Snapshot {
67        new_state: StateId,
68        prev_head: Option<StateId>,
69        head: Option<StateId>,
70        thread: Option<String>,
71    },
72    Goto {
73        target: StateId,
74        prev_head: Option<StateId>,
75        head: StateId,
76    },
77    ThreadCreate {
78        name: String,
79        state: StateId,
80        manager_snapshot: Option<Vec<u8>>,
81    },
82    ThreadDelete {
83        name: String,
84        state: StateId,
85    },
86    ThreadUpdate {
87        name: String,
88        old_state: StateId,
89        new_state: StateId,
90        #[serde(default)]
91        manager_snapshots: Option<ThreadUpdateSnapshots>,
92    },
93    Fork {
94        from: StateId,
95        new_state: StateId,
96        thread: Option<String>,
97        head: Option<StateId>,
98    },
99    Collapse {
100        sources: Vec<StateId>,
101        result: StateId,
102        thread: Option<String>,
103        pre_thread_state: Option<StateId>,
104    },
105    MarkerCreate {
106        name: String,
107        state: StateId,
108    },
109    MarkerDelete {
110        name: String,
111        state: StateId,
112    },
113    Checkpoint {
114        parent: Option<StateId>,
115        state: StateId,
116        thread: Option<String>,
117    },
118    TransactionAbort {
119        transaction_id: String,
120        reason: String,
121    },
122    EphemeralThreadCollapse {
123        thread: String,
124        final_state: StateId,
125    },
126    ConflictResolved {
127        conflict_id: String,
128        resolution: String,
129        resolver: Attribution,
130        mode: ConflictResolutionMode,
131    },
132    TransactionCommit {
133        transaction_id: String,
134        op_count: u32,
135    },
136    Redact {
137        redaction_id: ContentHash,
138        blob: ContentHash,
139        state: StateId,
140        path: String,
141    },
142    Purge {
143        redaction_id: ContentHash,
144        blob: ContentHash,
145    },
146    FastForward {
147        source_thread: String,
148        target_thread: String,
149        pre_target_id: StateId,
150        post_target_id: StateId,
151    },
152    GitCheckpoint {
153        branch: String,
154        state: StateId,
155        previous_git_oid: Option<String>,
156        new_git_oid: String,
157    },
158    RemoteThreadUpdate {
159        remote: String,
160        thread: String,
161        state: StateId,
162    },
163    RemoteThreadDelete {
164        remote: String,
165        thread: String,
166        state: StateId,
167    },
168    UndoRecoveryUpdate {
169        state: StateId,
170    },
171    StateVisibilitySet {
172        state: StateId,
173        record_id: ContentHash,
174        tier: VisibilityTier,
175        #[serde(default)]
176        prior_sidecar: Option<Vec<u8>>,
177        #[serde(default)]
178        new_sidecar: Option<Vec<u8>>,
179    },
180    StateVisibilityPromote {
181        state: StateId,
182        superseded: ContentHash,
183        record_id: ContentHash,
184        tier: VisibilityTier,
185        #[serde(default)]
186        prior_sidecar: Option<Vec<u8>>,
187        #[serde(default)]
188        new_sidecar: Option<Vec<u8>>,
189    },
190    HeadUpdate {
191        previous: RecordedHead,
192        new: RecordedHead,
193    },
194}
195
196impl StrictCurrentOpRecord {
197    fn into_current(self) -> OpRecord {
198        match self {
199            Self::Snapshot {
200                new_state,
201                prev_head,
202                head,
203                thread,
204            } => OpRecord::Snapshot {
205                new_state,
206                prev_head,
207                head,
208                thread,
209            },
210            Self::Goto {
211                target,
212                prev_head,
213                head,
214            } => OpRecord::Goto {
215                target,
216                prev_head,
217                head,
218            },
219            Self::ThreadCreate {
220                name,
221                state,
222                manager_snapshot,
223            } => OpRecord::ThreadCreate {
224                name,
225                state,
226                manager_snapshot,
227            },
228            Self::ThreadDelete { name, state } => OpRecord::ThreadDelete { name, state },
229            Self::ThreadUpdate {
230                name,
231                old_state,
232                new_state,
233                manager_snapshots,
234            } => OpRecord::ThreadUpdate {
235                name,
236                old_state,
237                new_state,
238                manager_snapshots,
239            },
240            Self::Fork {
241                from,
242                new_state,
243                thread,
244                head,
245            } => OpRecord::Fork {
246                from,
247                new_state,
248                thread,
249                head,
250            },
251            Self::Collapse {
252                sources,
253                result,
254                thread,
255                pre_thread_state,
256            } => OpRecord::Collapse {
257                sources,
258                result,
259                thread,
260                pre_thread_state,
261            },
262            Self::MarkerCreate { name, state } => OpRecord::MarkerCreate { name, state },
263            Self::MarkerDelete { name, state } => OpRecord::MarkerDelete { name, state },
264            Self::Checkpoint {
265                parent,
266                state,
267                thread,
268            } => OpRecord::Checkpoint {
269                parent,
270                state,
271                thread,
272            },
273            Self::TransactionAbort {
274                transaction_id,
275                reason,
276            } => OpRecord::TransactionAbort {
277                transaction_id,
278                reason,
279            },
280            Self::EphemeralThreadCollapse {
281                thread,
282                final_state,
283            } => OpRecord::EphemeralThreadCollapse {
284                thread,
285                final_state,
286            },
287            Self::ConflictResolved {
288                conflict_id,
289                resolution,
290                resolver,
291                mode,
292            } => OpRecord::ConflictResolved {
293                conflict_id,
294                resolution,
295                resolver,
296                mode,
297            },
298            Self::TransactionCommit {
299                transaction_id,
300                op_count,
301            } => OpRecord::TransactionCommit {
302                transaction_id,
303                op_count,
304            },
305            Self::Redact {
306                redaction_id,
307                blob,
308                state,
309                path,
310            } => OpRecord::Redact {
311                redaction_id,
312                blob,
313                state,
314                path,
315            },
316            Self::Purge { redaction_id, blob } => OpRecord::Purge { redaction_id, blob },
317            Self::FastForward {
318                source_thread,
319                target_thread,
320                pre_target_id,
321                post_target_id,
322            } => OpRecord::FastForward {
323                source_thread,
324                target_thread,
325                pre_target_id,
326                post_target_id,
327            },
328            Self::GitCheckpoint {
329                branch,
330                state,
331                previous_git_oid,
332                new_git_oid,
333            } => OpRecord::GitCheckpoint {
334                branch,
335                state,
336                previous_git_oid,
337                new_git_oid,
338            },
339            Self::RemoteThreadUpdate {
340                remote,
341                thread,
342                state,
343            } => OpRecord::RemoteThreadUpdate {
344                remote,
345                thread,
346                state,
347            },
348            Self::RemoteThreadDelete {
349                remote,
350                thread,
351                state,
352            } => OpRecord::RemoteThreadDelete {
353                remote,
354                thread,
355                state,
356            },
357            Self::UndoRecoveryUpdate { state } => OpRecord::UndoRecoveryUpdate { state },
358            Self::StateVisibilitySet {
359                state,
360                record_id,
361                tier,
362                prior_sidecar,
363                new_sidecar,
364            } => OpRecord::StateVisibilitySet {
365                state,
366                record_id,
367                tier,
368                prior_sidecar,
369                new_sidecar,
370            },
371            Self::StateVisibilityPromote {
372                state,
373                superseded,
374                record_id,
375                tier,
376                prior_sidecar,
377                new_sidecar,
378            } => OpRecord::StateVisibilityPromote {
379                state,
380                superseded,
381                record_id,
382                tier,
383                prior_sidecar,
384                new_sidecar,
385            },
386            Self::HeadUpdate { previous, new } => OpRecord::HeadUpdate { previous, new },
387        }
388    }
389}
390
391#[cfg(test)]
392mod tests {
393    use crate::object::{Agent, Principal};
394
395    use super::*;
396
397    fn state(byte: u8) -> StateId {
398        StateId::from_bytes([byte; 32])
399    }
400
401    fn hash(byte: u8) -> ContentHash {
402        ContentHash::from_bytes([byte; 32])
403    }
404
405    fn assert_round_trip(record: OpRecord) {
406        let bytes = encode_current_record(&record).unwrap();
407        let decoded = decode_current_record(&bytes).unwrap();
408        assert_eq!(format!("{decoded:?}"), format!("{record:?}"));
409    }
410
411    fn canonical_current_records() -> Vec<OpRecord> {
412        vec![
413            OpRecord::Snapshot {
414                new_state: state(1),
415                prev_head: Some(state(2)),
416                head: None,
417                thread: Some("main".into()),
418            },
419            OpRecord::Goto {
420                target: state(3),
421                prev_head: Some(state(2)),
422                head: state(3),
423            },
424            OpRecord::ThreadCreate {
425                name: "topic".into(),
426                state: state(4),
427                manager_snapshot: Some(vec![1, 2, 3]),
428            },
429            OpRecord::ThreadDelete {
430                name: "old".into(),
431                state: state(5),
432            },
433            OpRecord::ThreadUpdate {
434                name: "main".into(),
435                old_state: state(6),
436                new_state: state(7),
437                manager_snapshots: ThreadUpdateSnapshots::from_record_sets(
438                    Some(vec![6]),
439                    Some(vec![7]),
440                    vec![vec![60], vec![61]],
441                    vec![vec![70]],
442                    true,
443                ),
444            },
445            OpRecord::Fork {
446                from: state(8),
447                new_state: state(9),
448                thread: Some("topic".into()),
449                head: None,
450            },
451            OpRecord::Collapse {
452                sources: vec![state(8), state(9)],
453                result: state(10),
454                thread: Some("main".into()),
455                pre_thread_state: Some(state(7)),
456            },
457            OpRecord::MarkerCreate {
458                name: "release".into(),
459                state: state(11),
460            },
461            OpRecord::MarkerDelete {
462                name: "draft".into(),
463                state: state(12),
464            },
465            OpRecord::Checkpoint {
466                parent: Some(state(12)),
467                state: state(13),
468                thread: Some("main".into()),
469            },
470            OpRecord::TransactionAbort {
471                transaction_id: "abort".into(),
472                reason: "reason".into(),
473            },
474            OpRecord::EphemeralThreadCollapse {
475                thread: "ephemeral".into(),
476                final_state: state(14),
477            },
478            OpRecord::ConflictResolved {
479                conflict_id: "conflict".into(),
480                resolution: "ours".into(),
481                resolver: Attribution::with_agent(
482                    Principal::new("Resolver", "resolver@example.com"),
483                    Agent::new("openai", "gpt-5-codex"),
484                ),
485                mode: ConflictResolutionMode::Ours,
486            },
487            OpRecord::TransactionCommit {
488                transaction_id: "tx".into(),
489                op_count: 2,
490            },
491            OpRecord::Redact {
492                redaction_id: hash(1),
493                blob: hash(2),
494                state: state(15),
495                path: "secret.txt".into(),
496            },
497            OpRecord::Purge {
498                redaction_id: hash(3),
499                blob: hash(4),
500            },
501            OpRecord::FastForward {
502                source_thread: "feature".into(),
503                target_thread: "main".into(),
504                pre_target_id: state(17),
505                post_target_id: state(18),
506            },
507            OpRecord::GitCheckpoint {
508                branch: "main".into(),
509                state: state(20),
510                previous_git_oid: Some("abc".into()),
511                new_git_oid: "def".into(),
512            },
513            OpRecord::RemoteThreadUpdate {
514                remote: "origin".into(),
515                thread: "main".into(),
516                state: state(21),
517            },
518            OpRecord::RemoteThreadDelete {
519                remote: "origin".into(),
520                thread: "old".into(),
521                state: state(22),
522            },
523            OpRecord::UndoRecoveryUpdate { state: state(23) },
524            OpRecord::StateVisibilitySet {
525                state: state(24),
526                record_id: hash(5),
527                tier: VisibilityTier::Internal,
528                prior_sidecar: None,
529                new_sidecar: Some(vec![1, 2, 3]),
530            },
531            OpRecord::StateVisibilityPromote {
532                state: state(25),
533                superseded: hash(6),
534                record_id: hash(7),
535                tier: VisibilityTier::Restricted {
536                    scope_label: "embargo".into(),
537                },
538                prior_sidecar: Some(vec![4]),
539                new_sidecar: Some(vec![5]),
540            },
541            OpRecord::HeadUpdate {
542                previous: RecordedHead::Detached { state: state(26) },
543                new: RecordedHead::Attached {
544                    thread: "main".into(),
545                },
546            },
547        ]
548    }
549
550    fn variant_name(record: &OpRecord) -> &'static str {
551        match record {
552            OpRecord::Snapshot { .. } => "Snapshot",
553            OpRecord::Goto { .. } => "Goto",
554            OpRecord::ThreadCreate { .. } => "ThreadCreate",
555            OpRecord::ThreadDelete { .. } => "ThreadDelete",
556            OpRecord::ThreadUpdate { .. } => "ThreadUpdate",
557            OpRecord::Fork { .. } => "Fork",
558            OpRecord::Collapse { .. } => "Collapse",
559            OpRecord::MarkerCreate { .. } => "MarkerCreate",
560            OpRecord::MarkerDelete { .. } => "MarkerDelete",
561            OpRecord::Checkpoint { .. } => "Checkpoint",
562            OpRecord::TransactionAbort { .. } => "TransactionAbort",
563            OpRecord::EphemeralThreadCollapse { .. } => "EphemeralThreadCollapse",
564            OpRecord::ConflictResolved { .. } => "ConflictResolved",
565            OpRecord::TransactionCommit { .. } => "TransactionCommit",
566            OpRecord::Redact { .. } => "Redact",
567            OpRecord::Purge { .. } => "Purge",
568            OpRecord::FastForward { .. } => "FastForward",
569            OpRecord::GitCheckpoint { .. } => "GitCheckpoint",
570            OpRecord::RemoteThreadUpdate { .. } => "RemoteThreadUpdate",
571            OpRecord::RemoteThreadDelete { .. } => "RemoteThreadDelete",
572            OpRecord::UndoRecoveryUpdate { .. } => "UndoRecoveryUpdate",
573            OpRecord::StateVisibilitySet { .. } => "StateVisibilitySet",
574            OpRecord::StateVisibilityPromote { .. } => "StateVisibilityPromote",
575            OpRecord::HeadUpdate { .. } => "HeadUpdate",
576        }
577    }
578
579    #[test]
580    fn schema_four_is_current_and_legacy_versions_are_refused() {
581        assert_eq!(CURRENT_OP_RECORD_SCHEMA_VERSION, 4);
582        validate_op_record_schema_version(4).unwrap();
583        for legacy in 1..=3 {
584            let error = validate_op_record_schema_version(legacy).unwrap_err();
585            assert!(matches!(
586                error,
587                HeddleError::StorageFormatMigrationRequired {
588                    found,
589                    required: 4,
590                    ..
591                } if found == legacy
592            ));
593        }
594        assert!(matches!(
595            validate_op_record_schema_version(5).unwrap_err(),
596            HeddleError::StorageFormatTooNew {
597                found: 5,
598                supported: 4,
599                ..
600            }
601        ));
602    }
603
604    #[test]
605    fn every_current_variant_round_trips() {
606        let records = canonical_current_records();
607        assert_eq!(
608            records.iter().map(variant_name).collect::<Vec<_>>(),
609            [
610                "Snapshot",
611                "Goto",
612                "ThreadCreate",
613                "ThreadDelete",
614                "ThreadUpdate",
615                "Fork",
616                "Collapse",
617                "MarkerCreate",
618                "MarkerDelete",
619                "Checkpoint",
620                "TransactionAbort",
621                "EphemeralThreadCollapse",
622                "ConflictResolved",
623                "TransactionCommit",
624                "Redact",
625                "Purge",
626                "FastForward",
627                "GitCheckpoint",
628                "RemoteThreadUpdate",
629                "RemoteThreadDelete",
630                "UndoRecoveryUpdate",
631                "StateVisibilitySet",
632                "StateVisibilityPromote",
633                "HeadUpdate",
634            ]
635        );
636        for record in records {
637            assert_round_trip(record);
638        }
639    }
640
641    #[test]
642    fn state_id_v4_visibility_tail_bytes_are_frozen() {
643        let record = OpRecord::StateVisibilityPromote {
644            state: state(1),
645            superseded: hash(2),
646            record_id: hash(3),
647            tier: VisibilityTier::Internal,
648            prior_sidecar: Some(vec![4]),
649            new_sidecar: Some(vec![5]),
650        };
651
652        let expected = [
653            &[
654                129, 182, 83, 116, 97, 116, 101, 86, 105, 115, 105, 98, 105, 108, 105, 116, 121,
655                80, 114, 111, 109, 111, 116, 101, 150, 220, 0, 32,
656            ][..],
657            &[1; 32],
658            &[220, 0, 32],
659            &[2; 32],
660            &[220, 0, 32],
661            &[3; 32],
662            &[168, 73, 110, 116, 101, 114, 110, 97, 108, 145, 4, 145, 5],
663        ]
664        .concat();
665
666        assert_eq!(encode_current_record(&record).unwrap(), expected);
667    }
668
669    #[test]
670    fn historical_sixteen_byte_payload_is_not_a_state_id_record() {
671        let historical = [
672            129, 168, 67, 111, 108, 108, 97, 112, 115, 101, 147, 146, 220, 0, 16, 10, 10, 10, 10,
673            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 220, 0, 16, 11, 11, 11, 11, 11, 11, 11,
674            11, 11, 11, 11, 11, 11, 11, 11, 11, 220, 0, 16, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12,
675            12, 12, 12, 12, 12, 12, 164, 109, 97, 105, 110,
676        ];
677        let error = decode_current_record(&historical)
678            .expect_err("16-byte ChangeIds must not decode as StateIds");
679        assert!(error.to_string().contains("expected an array of length 32"));
680    }
681}
heddle_object_model/op_record/codec.rs

heddle_object_model/op_record/
codec.rs
