Skip to main content

kmp_adapter_embedded/adapter/
portability.rs

1//! Export/import (E6): the append-only event log is the portable form of an
2//! embedded store. Export dumps it in sequence order; import replays it into
3//! an empty store, reproducing identical revisions, idempotency outcomes and
4//! projections — temporal reads and relation proof survive the round trip by
5//! construction.
6
7use std::collections::{BTreeMap, BTreeSet};
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use kmp_domain::{ContextEventStore, ContextUpdatedEvent, PortError, ProjectionMutation};
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13
14use super::replay::ProjectionRebuildReport;
15use super::store::EmbeddedKernelStore;
16
17/// Inclusive positions in this bundle's event stream. Export is a portable
18/// replay, not a view over the store's internal sequence keys: full and
19/// filtered bundles both renumber their payload positions from one while
20/// preserving every event's aggregate revision (and therefore every ref).
21/// An empty snapshot has neither bound.
22#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
23pub struct BundleEventRange {
24    pub first: Option<u64>,
25    pub last: Option<u64>,
26}
27
28/// First line of a bundle file: identity and integrity metadata for fail-fast
29/// import. Fields added in bundle format 2 default only so format-1 bundles
30/// remain readable; every format-2 field is validated before replay.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub struct BundleHeader {
33    pub bundle_format: u32,
34    /// Format of the portable event payload, not the on-disk SQLite layout.
35    /// `store_format` is accepted from format-1 bundles because that
36    /// older name described the field ambiguously.
37    #[serde(rename = "event_format", alias = "store_format")]
38    pub event_format: u32,
39    pub event_count: u64,
40    pub kernel_version: String,
41    #[serde(default)]
42    pub snapshot_id: String,
43    #[serde(default)]
44    pub created_at_unix_ms: u64,
45    #[serde(default)]
46    pub event_range: BundleEventRange,
47    #[serde(default)]
48    pub abouts: Vec<String>,
49    #[serde(default)]
50    pub content_digest: String,
51}
52
53pub const BUNDLE_FORMAT_VERSION: u32 = 2;
54
55/// Outcome of an import: events replayed and projections rebuilt.
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub struct ImportReport {
58    pub events_imported: u64,
59    pub rebuild: ProjectionRebuildReport,
60}
61
62impl EmbeddedKernelStore {
63    /// Synchronous export for already-blocking operational paths such as
64    /// Doctor. Async application paths should use [`Self::export_bundle`].
65    pub fn export_bundle_blocking(&self) -> Result<String, PortError> {
66        encode_bundle(&self.read_event_log()?, None)
67    }
68
69    /// Serializes the full event log as a JSON-Lines bundle: one header line
70    /// followed by one event per line, in sequence order.
71    pub async fn export_bundle(&self) -> Result<String, PortError> {
72        self.run(|store| store.export_bundle_blocking()).await
73    }
74
75    /// Serializes only events rooted at one of `requested_abouts`.
76    ///
77    /// Abouts are opaque routing identifiers: matching is exact, with no
78    /// trimming, case folding, prefix expansion or other normalisation. The
79    /// filtered stream keeps the store's event order and each aggregate's
80    /// recorded revisions, then receives bundle-local positions starting at
81    /// one. A missing requested about is refused before callers can create a
82    /// destination file.
83    pub async fn export_bundle_for_abouts(
84        &self,
85        requested_abouts: &[String],
86    ) -> Result<String, PortError> {
87        let events = self.run(EmbeddedKernelStore::read_event_log).await?;
88        let events = filter_events_for_abouts(events, requested_abouts)?;
89        encode_bundle(&events, None)
90    }
91
92    /// Serializes every event *except* those rooted at one of
93    /// `excluded_abouts`.
94    ///
95    /// The mirror of [`Self::export_bundle_for_abouts`], for the caller that
96    /// knows what does not belong in a bundle rather than what does. A store
97    /// holds content it did not author — a synced guide, say — and a bundle
98    /// that is meant to carry authored memory has to be able to leave it out
99    /// without first enumerating everything else. An excluded about that is
100    /// absent is not an error: exclusion asks for a stream without something,
101    /// and a store that never had it already satisfies that.
102    ///
103    /// Abouts are opaque: matching is exact, with no trimming, case folding or
104    /// prefix expansion.
105    pub fn export_bundle_excluding_abouts_blocking(
106        &self,
107        excluded_abouts: &[String],
108    ) -> Result<String, PortError> {
109        let events = self.read_event_log()?;
110        encode_bundle(
111            &filter_events_excluding_abouts(events, excluded_abouts),
112            None,
113        )
114    }
115
116    /// The async form of [`Self::export_bundle_excluding_abouts_blocking`].
117    pub async fn export_bundle_excluding_abouts(
118        &self,
119        excluded_abouts: &[String],
120    ) -> Result<String, PortError> {
121        let events = self.run(EmbeddedKernelStore::read_event_log).await?;
122        encode_bundle(
123            &filter_events_excluding_abouts(events, excluded_abouts),
124            None,
125        )
126    }
127
128    /// Exports the same complete stream with a human-selected snapshot id.
129    /// The id is metadata, not a filename: callers may store the bundle in git,
130    /// an artifact store, or anywhere else without changing what it identifies.
131    pub async fn export_named_bundle(&self, snapshot_id: &str) -> Result<String, PortError> {
132        if snapshot_id.trim().is_empty() {
133            return Err(PortError::InvalidState(
134                "snapshot id must not be empty".to_string(),
135            ));
136        }
137        let events = self.run(EmbeddedKernelStore::read_event_log).await?;
138        encode_bundle(&events, Some(snapshot_id))
139    }
140
141    /// Replays a bundle into this store. Fail-fast rules: the store must be
142    /// empty (no merge semantics in v1 — ADR-011 rationale applies), the
143    /// header must match supported formats, and every event must reproduce
144    /// exactly the revision it was exported with.
145    pub async fn import_bundle<F>(&self, bundle: &str, derive: F) -> Result<ImportReport, PortError>
146    where
147        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
148    {
149        let (log_length, _) = self.event_log_stats().await?;
150        if log_length != 0 {
151            return Err(PortError::Conflict(format!(
152                "import requires an empty store; this store already holds {log_length} events \
153                 (merging bundles is not supported)"
154            )));
155        }
156
157        let verified = parse_bundle(bundle)?;
158        let header = verified.header;
159        let events = verified.events;
160        validate_revisions(&events)?;
161        // Projection derivation is pure. Prove every payload is rebuildable
162        // before the first append so a malformed later line cannot leave a
163        // half-restored store behind.
164        for event in &events {
165            derive(event)?;
166        }
167        let events_imported = self.replay_event_stream(events).await?;
168        debug_assert_eq!(events_imported, header.event_count);
169
170        let rebuild = self.rebuild_projections(derive).await?;
171        Ok(ImportReport {
172            events_imported,
173            rebuild,
174        })
175    }
176}
177
178/// Validates a bundle without opening or mutating a store. This is the
179/// recovery check: identity, range, about coverage and digest are all proved
180/// before an operator trusts a saved copy.
181pub fn verify_bundle(bundle: &str) -> Result<BundleHeader, PortError> {
182    parse_bundle(bundle).map(|verified| verified.header)
183}
184
185/// Re-encodes a verified bundle without events rooted at `excluded_abouts`.
186///
187/// This is the bundle-side counterpart to
188/// [`EmbeddedKernelStore::export_bundle_excluding_abouts`]. Operational
189/// comparisons must apply the same authored-memory policy to a legacy bundle
190/// and to the live store; otherwise release-owned guide events can look like a
191/// divergent project history even when both streams are identical.
192pub fn bundle_excluding_abouts(
193    bundle: &str,
194    excluded_abouts: &[String],
195) -> Result<String, PortError> {
196    let verified = parse_bundle(bundle)?;
197    encode_bundle(
198        &filter_events_excluding_abouts(verified.events, excluded_abouts),
199        None,
200    )
201}
202
203/// Merges only histories that have a deterministic answer: identical streams
204/// or one exact prefix of the other. Two branches that both appended at the
205/// same position are a semantic conflict, so KMP refuses to invent an order.
206pub fn merge_bundles(left: &str, right: &str, snapshot_id: &str) -> Result<String, PortError> {
207    if snapshot_id.trim().is_empty() {
208        return Err(PortError::InvalidState(
209            "merged snapshot id must not be empty".to_string(),
210        ));
211    }
212    let left = parse_bundle(left)?;
213    let right = parse_bundle(right)?;
214    let shared = left.events.len().min(right.events.len());
215    if let Some(position) =
216        (0..shared).find(|position| left.events[*position] != right.events[*position])
217    {
218        return Err(PortError::Conflict(format!(
219            "bundle histories diverge at event position {}; KMP only fast-forwards an exact \
220             prefix and will not invent causal order for two branches",
221            position + 1
222        )));
223    }
224    let events = if left.events.len() >= right.events.len() {
225        left.events
226    } else {
227        right.events
228    };
229    encode_bundle(&events, Some(snapshot_id))
230}
231
232struct VerifiedBundle {
233    header: BundleHeader,
234    events: Vec<ContextUpdatedEvent>,
235}
236
237fn parse_bundle(bundle: &str) -> Result<VerifiedBundle, PortError> {
238    let mut lines = bundle.lines().filter(|line| !line.trim().is_empty());
239    let header: BundleHeader = decode_line(
240        "bundle header",
241        lines.next().ok_or_else(|| {
242            PortError::InvalidState("bundle is empty: missing header line".to_string())
243        })?,
244    )?;
245    if !matches!(header.bundle_format, 1 | BUNDLE_FORMAT_VERSION) {
246        return Err(PortError::InvalidState(format!(
247            "bundle format {} is not supported (this binary reads 1 and {})",
248            header.bundle_format, BUNDLE_FORMAT_VERSION
249        )));
250    }
251    if header.event_format != super::format_version::EVENT_FORMAT_VERSION {
252        return Err(PortError::InvalidState(format!(
253            "bundle carries event format {}, this binary supports {}",
254            header.event_format,
255            super::format_version::EVENT_FORMAT_VERSION
256        )));
257    }
258
259    let mut events = Vec::new();
260    let mut event_payload = String::new();
261    for line in lines {
262        events.push(decode_line::<ContextUpdatedEvent>("bundle event", line)?);
263        event_payload.push_str(line);
264        event_payload.push('\n');
265    }
266    if events.len() as u64 != header.event_count {
267        return Err(PortError::InvalidState(format!(
268            "bundle header declares {} events but {} were present",
269            header.event_count,
270            events.len()
271        )));
272    }
273
274    if header.bundle_format == BUNDLE_FORMAT_VERSION {
275        validate_v2_header(&header, &events, &event_payload)?;
276    }
277    Ok(VerifiedBundle { header, events })
278}
279
280fn validate_v2_header(
281    header: &BundleHeader,
282    events: &[ContextUpdatedEvent],
283    event_payload: &str,
284) -> Result<(), PortError> {
285    if header.snapshot_id.trim().is_empty() {
286        return Err(PortError::InvalidState(
287            "bundle format 2 requires snapshot_id".to_string(),
288        ));
289    }
290    if header.created_at_unix_ms == 0 {
291        return Err(PortError::InvalidState(
292            "bundle format 2 requires created_at_unix_ms".to_string(),
293        ));
294    }
295    let expected_range = event_range(events.len());
296    if header.event_range != expected_range {
297        return Err(PortError::InvalidState(format!(
298            "bundle event_range {:?} does not cover its {} events (expected {:?})",
299            header.event_range,
300            events.len(),
301            expected_range
302        )));
303    }
304    let expected_abouts = abouts(events);
305    if header.abouts != expected_abouts {
306        return Err(PortError::InvalidState(format!(
307            "bundle abouts do not match its events (expected {})",
308            expected_abouts.join(", ")
309        )));
310    }
311    let expected_digest = content_digest(event_payload.as_bytes());
312    if header.content_digest != expected_digest {
313        return Err(PortError::InvalidState(format!(
314            "bundle content digest mismatch: header says {}, events produce {expected_digest}",
315            header.content_digest
316        )));
317    }
318    Ok(())
319}
320
321fn encode_bundle(
322    events: &[ContextUpdatedEvent],
323    snapshot_id: Option<&str>,
324) -> Result<String, PortError> {
325    let mut event_payload = String::new();
326    for event in events {
327        event_payload.push_str(&encode_line("bundle event", event)?);
328    }
329    let digest = content_digest(event_payload.as_bytes());
330    let named = snapshot_id.is_some();
331    let snapshot_id = snapshot_id
332        .map(str::to_string)
333        .unwrap_or_else(|| format!("content-{}", &digest[7..23]));
334    // Content-addressed head exports must be byte-identical across storage
335    // layouts and repeated exports. Their creation coordinate is
336    // therefore the newest event time. A named recovery point records when
337    // the operator created that point.
338    let created_at = if named {
339        SystemTime::now()
340    } else {
341        events
342            .iter()
343            .map(|event| event.occurred_at)
344            .max()
345            .unwrap_or(UNIX_EPOCH + Duration::from_millis(1))
346    };
347    let created_at_unix_ms = created_at
348        .duration_since(UNIX_EPOCH)
349        .unwrap_or(Duration::ZERO)
350        .as_millis() as u64;
351    let header = BundleHeader {
352        bundle_format: BUNDLE_FORMAT_VERSION,
353        event_format: super::format_version::EVENT_FORMAT_VERSION,
354        event_count: events.len() as u64,
355        kernel_version: env!("CARGO_PKG_VERSION").to_string(),
356        snapshot_id,
357        created_at_unix_ms,
358        event_range: event_range(events.len()),
359        abouts: abouts(events),
360        content_digest: digest,
361    };
362    let mut out = encode_line("bundle header", &header)?;
363    out.push_str(&event_payload);
364    Ok(out)
365}
366
367fn event_range(event_count: usize) -> BundleEventRange {
368    if event_count == 0 {
369        BundleEventRange::default()
370    } else {
371        BundleEventRange {
372            first: Some(1),
373            last: Some(event_count as u64),
374        }
375    }
376}
377
378fn abouts(events: &[ContextUpdatedEvent]) -> Vec<String> {
379    events
380        .iter()
381        .map(|event| event.root_node_id.clone())
382        .collect::<BTreeSet<_>>()
383        .into_iter()
384        .collect()
385}
386
387fn filter_events_excluding_abouts(
388    events: Vec<ContextUpdatedEvent>,
389    excluded_abouts: &[String],
390) -> Vec<ContextUpdatedEvent> {
391    if excluded_abouts.is_empty() {
392        return events;
393    }
394    let excluded = excluded_abouts.iter().cloned().collect::<BTreeSet<_>>();
395    events
396        .into_iter()
397        .filter(|event| !excluded.contains(&event.root_node_id))
398        .collect()
399}
400
401fn filter_events_for_abouts(
402    events: Vec<ContextUpdatedEvent>,
403    requested_abouts: &[String],
404) -> Result<Vec<ContextUpdatedEvent>, PortError> {
405    if requested_abouts.is_empty() {
406        return Err(PortError::InvalidState(
407            "filtered export requires at least one about".to_string(),
408        ));
409    }
410    let requested = requested_abouts.iter().cloned().collect::<BTreeSet<_>>();
411    let found = events
412        .iter()
413        .filter(|event| requested.contains(&event.root_node_id))
414        .map(|event| event.root_node_id.clone())
415        .collect::<BTreeSet<_>>();
416    let missing = requested.difference(&found).cloned().collect::<Vec<_>>();
417    if !missing.is_empty() {
418        return Err(PortError::InvalidState(format!(
419            "cannot export missing about{}: {}",
420            if missing.len() == 1 { "" } else { "s" },
421            missing
422                .iter()
423                .map(|about| format!("`{about}`"))
424                .collect::<Vec<_>>()
425                .join(", ")
426        )));
427    }
428    Ok(events
429        .into_iter()
430        .filter(|event| requested.contains(&event.root_node_id))
431        .collect())
432}
433
434fn content_digest(bytes: &[u8]) -> String {
435    format!("sha256:{:x}", Sha256::digest(bytes))
436}
437
438fn validate_revisions(events: &[ContextUpdatedEvent]) -> Result<(), PortError> {
439    let mut revisions: BTreeMap<(&str, &str), u64> = BTreeMap::new();
440    for (position, event) in events.iter().enumerate() {
441        let previous = revisions
442            .get(&(event.root_node_id.as_str(), event.role.as_str()))
443            .copied()
444            .unwrap_or(0);
445        let expected = previous + 1;
446        if event.revision != expected {
447            return Err(PortError::InvalidState(format!(
448                "bundle event position {} carries revision {} for ({}, {}), expected {}; no \
449                 events were imported",
450                position + 1,
451                event.revision,
452                event.root_node_id,
453                event.role,
454                expected
455            )));
456        }
457        revisions.insert(
458            (event.root_node_id.as_str(), event.role.as_str()),
459            event.revision,
460        );
461    }
462    Ok(())
463}
464
465impl EmbeddedKernelStore {
466    /// Replays a history into this store, in order, checking that every
467    /// event lands on the revision it was recorded with.
468    ///
469    /// That check is the whole point: a replay that silently renumbers
470    /// history would produce a store that reads plausibly and cites
471    /// revisions that never existed. Shared by import and migration, which
472    /// are the same operation seen from two different distances.
473    pub(crate) async fn replay_event_stream<I>(&self, events: I) -> Result<u64, PortError>
474    where
475        I: IntoIterator<Item = ContextUpdatedEvent>,
476    {
477        let mut replayed = 0u64;
478        for event in events {
479            let recorded_revision = event.revision;
480            let expected_previous = recorded_revision.checked_sub(1).ok_or_else(|| {
481                PortError::InvalidState("event carries revision 0; the log is corrupt".to_string())
482            })?;
483            let assigned = self.append(event, expected_previous).await?;
484            if assigned != recorded_revision {
485                return Err(PortError::Conflict(format!(
486                    "replay integrity violation: assigned revision {assigned}, \
487                     history recorded {recorded_revision}"
488                )));
489            }
490            replayed += 1;
491        }
492        Ok(replayed)
493    }
494}
495
496fn encode_line<T: Serialize>(what: &str, value: &T) -> Result<String, PortError> {
497    let mut line = serde_json::to_string(value)
498        .map_err(|error| PortError::InvalidState(format!("could not encode {what}: {error}")))?;
499    line.push('\n');
500    Ok(line)
501}
502
503fn decode_line<T: for<'de> Deserialize<'de>>(what: &str, line: &str) -> Result<T, PortError> {
504    serde_json::from_str(line)
505        .map_err(|error| PortError::InvalidState(format!("could not decode {what}: {error}")))
506}
507
508#[cfg(test)]
509mod tests {
510    use super::*;
511
512    fn event(root: &str, revision: u64, content_hash: &str) -> ContextUpdatedEvent {
513        ContextUpdatedEvent {
514            root_node_id: root.to_string(),
515            role: "agent".to_string(),
516            revision,
517            content_hash: content_hash.to_string(),
518            changes: Vec::new(),
519            idempotency_key: Some(format!("{root}:{revision}")),
520            logical_digest: None,
521            requested_by: Some("portability-test".to_string()),
522            occurred_at: UNIX_EPOCH + Duration::from_secs(revision),
523        }
524    }
525
526    #[test]
527    fn format_two_identifies_and_covers_the_snapshot() {
528        let events = vec![event("project:b", 1, "b"), event("project:a", 1, "a")];
529        let bundle = encode_bundle(&events, Some("pre-release")).expect("bundle");
530        let header = verify_bundle(&bundle).expect("verified");
531
532        assert_eq!(header.bundle_format, BUNDLE_FORMAT_VERSION);
533        assert_eq!(
534            header.event_format,
535            super::super::format_version::EVENT_FORMAT_VERSION
536        );
537        assert_eq!(header.snapshot_id, "pre-release");
538        assert!(header.created_at_unix_ms > 0);
539        assert_eq!(
540            header.event_range,
541            BundleEventRange {
542                first: Some(1),
543                last: Some(2),
544            }
545        );
546        assert_eq!(header.abouts, ["project:a", "project:b"]);
547        assert!(header.content_digest.starts_with("sha256:"));
548    }
549
550    #[test]
551    fn exclusion_keeps_everything_the_excluded_abouts_do_not_root() {
552        let events = vec![
553            event("project:a", 1, "a1"),
554            event("guide:kmp", 1, "g1"),
555            event("project:a", 2, "a2"),
556            event("guide:kmp-agent", 1, "g2"),
557        ];
558        let excluded = vec!["guide:kmp".to_string(), "guide:kmp-agent".to_string()];
559
560        let kept = filter_events_excluding_abouts(events, &excluded);
561
562        assert_eq!(
563            kept.iter()
564                .map(|event| (event.root_node_id.as_str(), event.revision))
565                .collect::<Vec<_>>(),
566            vec![("project:a", 1), ("project:a", 2)]
567        );
568    }
569
570    #[test]
571    fn exclusion_matches_exactly_because_abouts_are_opaque() {
572        let events = vec![
573            event("guide:kmp", 1, "g1"),
574            event("guide:kmp:extra", 1, "x1"),
575            event(" guide:kmp", 1, "s1"),
576        ];
577        let excluded = vec!["guide:kmp".to_string()];
578
579        let kept = filter_events_excluding_abouts(events, &excluded);
580
581        assert_eq!(
582            kept.iter()
583                .map(|event| event.root_node_id.as_str())
584                .collect::<Vec<_>>(),
585            vec!["guide:kmp:extra", " guide:kmp"]
586        );
587    }
588
589    #[test]
590    fn excluding_an_about_the_store_never_had_is_not_an_error() {
591        // Exclusion asks for a stream without something. A store that never
592        // held it already satisfies that, unlike a filtered export, which is
593        // asking for something and must refuse when it is missing.
594        let events = vec![event("project:a", 1, "a1")];
595        let excluded = vec!["guide:kmp".to_string()];
596
597        let kept = filter_events_excluding_abouts(events, &excluded);
598
599        assert_eq!(kept.len(), 1);
600        assert_eq!(kept[0].root_node_id, "project:a");
601    }
602
603    #[test]
604    fn verified_bundle_exclusion_applies_the_same_policy_to_legacy_bundles() {
605        let bundle = encode_bundle(
606            &[
607                event("project:a", 1, "a1"),
608                event("guide:kmp-agent", 1, "g1"),
609                event("project:a", 2, "a2"),
610            ],
611            None,
612        )
613        .expect("bundle");
614
615        let filtered = bundle_excluding_abouts(&bundle, &["guide:kmp-agent".to_string()])
616            .expect("filtered verified bundle");
617        let verified = parse_bundle(&filtered).expect("verified filtered bundle");
618
619        assert_eq!(verified.header.event_count, 2);
620        assert_eq!(verified.header.abouts, ["project:a"]);
621        assert_eq!(
622            verified
623                .events
624                .iter()
625                .map(|event| event.content_hash.as_str())
626                .collect::<Vec<_>>(),
627            ["a1", "a2"]
628        );
629    }
630
631    #[test]
632    fn filtered_export_matches_opaque_abouts_exactly_and_renumbers_its_range() {
633        let events = vec![
634            event("project:a", 1, "a1"),
635            event("project:ab", 1, "ab1"),
636            event("project:a", 2, "a2"),
637        ];
638        let filtered = filter_events_for_abouts(events, &["project:a".to_string()])
639            .expect("exact about exists");
640        assert_eq!(filtered.len(), 2);
641        assert!(
642            filtered
643                .iter()
644                .all(|event| event.root_node_id == "project:a")
645        );
646
647        let bundle = encode_bundle(&filtered, None).expect("filtered bundle");
648        let header = verify_bundle(&bundle).expect("filtered bundle verifies");
649        assert_eq!(header.abouts, ["project:a"]);
650        assert_eq!(header.event_count, 2);
651        assert_eq!(
652            header.event_range,
653            BundleEventRange {
654                first: Some(1),
655                last: Some(2),
656            }
657        );
658    }
659
660    #[test]
661    fn filtered_export_names_every_requested_about_that_is_missing() {
662        let error = filter_events_for_abouts(
663            vec![event("project:a", 1, "a")],
664            &["project:a".into(), "project:none".into()],
665        )
666        .expect_err("missing about must fail");
667        assert!(error.to_string().contains("`project:none`"), "{error}");
668    }
669
670    #[test]
671    fn tampering_is_rejected_before_a_bundle_can_be_replayed() {
672        let bundle =
673            encode_bundle(&[event("project:a", 1, "before")], Some("saved")).expect("bundle");
674        let tampered = bundle.replace("\"content_hash\":\"before\"", "\"content_hash\":\"after\"");
675        let error = verify_bundle(&tampered).expect_err("digest catches changed payload");
676        assert!(error.to_string().contains("content digest mismatch"));
677    }
678
679    #[test]
680    fn merge_fast_forwards_an_exact_prefix() {
681        let first = event("project:a", 1, "one");
682        let second = event("project:a", 2, "two");
683        let left = encode_bundle(std::slice::from_ref(&first), Some("left")).expect("left");
684        let right = encode_bundle(&[first, second], Some("right")).expect("right");
685
686        let merged = merge_bundles(&left, &right, "merged").expect("fast forward");
687        let header = verify_bundle(&merged).expect("verified merge");
688        assert_eq!(header.snapshot_id, "merged");
689        assert_eq!(header.event_count, 2);
690    }
691
692    #[test]
693    fn merge_refuses_two_histories_at_the_same_position() {
694        let left = encode_bundle(&[event("project:a", 1, "left")], Some("left")).expect("left");
695        let right = encode_bundle(&[event("project:a", 1, "right")], Some("right")).expect("right");
696
697        let error = merge_bundles(&left, &right, "invented").expect_err("must refuse");
698        assert!(error.to_string().contains("diverge at event position 1"));
699        assert!(error.to_string().contains("will not invent causal order"));
700    }
701
702    #[test]
703    fn legacy_format_one_remains_readable() {
704        let legacy =
705            r#"{"bundle_format":1,"store_format":1,"event_count":0,"kernel_version":"0.1.3"}"#;
706        let header = verify_bundle(legacy).expect("format one remains portable");
707        assert_eq!(header.event_format, 1);
708        assert!(header.snapshot_id.is_empty());
709    }
710
711    #[test]
712    fn invalid_later_revision_is_rejected_in_preflight() {
713        let events = [event("project:a", 1, "one"), event("project:a", 3, "three")];
714        let error = validate_revisions(&events).expect_err("revision gap");
715        assert!(error.to_string().contains("position 2"));
716        assert!(error.to_string().contains("no events were imported"));
717    }
718}