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