bynk_check/

schema_registry.rs

//! Events track, slice 3c (#980): the cross-build schema registry's
//! reconciliation logic — verifying a project's live event shapes against a
//! committed `bynk.schema.lock`, keyed by qualified event name, storing each
//! event's field shape (name, type, whether the field has a default) and its
//! current schema version. Every build reconciles the live shape against the
//! stored one and either leaves the version alone, bumps it (a purely
//! additive change), or fails the build (a change that isn't safely
//! additive).
//!
//! P4.0 (#1113, [DECISION A]): the document shape itself — `SchemaRegistry`,
//! `EventEntry`, `FieldShape`, and their `parse`/`serialize` — moved to
//! `bynk_project::schema_registry`; `bynk-emit/src/project/schema_registry.rs`
//! re-exports it there, unchanged, for every existing `schema_registry::X`
//! call site in `project.rs`.
//!
//! P5.3 (`design/tracks/semantics-in-the-checker.md` §6): [`reconcile`] and
//! its helpers moved here from that same `bynk-emit` module — category 1 of
//! `bynk-check/src/analysis.rs`'s own seven-category residual-gap accounting
//! ("gap in name only": `reconcile` only ever ran under `SchemaLock::On`, and
//! the analyse path always passed `SchemaLock::Off`, so relocating it changes
//! nothing observable — see that module's doc for why R3.5 still requires the
//! move). `bynk-emit`'s `run_checks` is now this function's caller, not its
//! owner, the same way P4.0/P4.1 turned `project.rs` into a caller of
//! `bynk-project`/`bynk-check` throughout.
//!
//! P5.5 (§6, §3.2's "eighth site"): [`parse_or_diagnose`] moved here too —
//! `bynk.project.schema_registry_corrupt`, a real `CompileError::new`
//! construction the seven-category accounting above didn't cover (it isn't a
//! whole-project *check*, it's the read step immediately before
//! [`reconcile`]), found and scoped by name in §3.2 rather than left for a
//! future sweep to rediscover. `bynk_project::schema_registry::parse` itself
//! — the document-shape parse — stays in `bynk-project` (P4.0); only the
//! diagnostic construction on its error path moves, the same split
//! `reconcile`'s own move drew between document shape and checking.

use std::collections::BTreeMap;

use bynk_syntax::ast::{EventDecl, TypeRef};
use bynk_syntax::error::CompileError;
use bynk_syntax::span::Span;

use bynk_project::schema_registry::{EventEntry, FieldShape, SchemaRegistry};

use crate::symbols::UnitTable;

/// Parse `bynk.schema.lock`'s content, diagnosing a corrupt file as
/// `bynk.project.schema_registry_corrupt` rather than handing the caller a
/// bare `String` to wrap itself. `existing`/`project_root` are exactly
/// `bynk_project::schema_registry::parse`'s own parameters — `None` (no lock
/// file yet) always succeeds with a fresh, empty registry.
pub fn parse_or_diagnose(
    existing: Option<&str>,
    project_root: &std::path::Path,
) -> Result<SchemaRegistry, CompileError> {
    bynk_project::schema_registry::parse(existing, project_root).map_err(|msg| {
        CompileError::new("bynk.project.schema_registry_corrupt", Span::default(), msg)
    })
}

/// A shallow, per-field snapshot of an event's current shape — deliberately
/// **not** `bynk-check/src/contract.rs`'s `canon_named_in`: that renders a
/// field as `name: type` with no signal for default-presence, so an additive
/// change (new field, has a default) and a breaking one (new field, no
/// default) perturb it identically. This snapshot exists to tell those two
/// apart, one field at a time.
fn snapshot(event: &EventDecl) -> Vec<FieldShape> {
    let mut fields: Vec<FieldShape> = event
        .body
        .fields
        .iter()
        .map(|f| FieldShape {
            name: f.name.name.clone(),
            ty: canon_type(&f.type_ref),
            default: f.init.is_some(),
        })
        .collect();
    fields.sort_by(|a, b| a.name.cmp(&b.name));
    fields
}

/// A stable surface-form rendering of a `TypeRef`, for the registry's own
/// stored shape only. Deliberately **not** `bynk-fmt`'s `type_ref_to_string`
/// (private, and neither `bynk-emit` nor `bynk-check` depends on `bynk-fmt`)
/// — and deliberately not required to match either the formatter's or the
/// checker's own rendering byte-for-byte: this string is compared only
/// against itself, across builds, never displayed to a user. Exhaustive, no
/// wildcard arm, so a new `TypeRef` variant is a compile error here rather
/// than a silent "no signal" gap in the registry's diffing.
fn canon_type(t: &TypeRef) -> String {
    match t {
        TypeRef::Base(b, _) => b.name().to_string(),
        TypeRef::Named(id) => id.name.clone(),
        TypeRef::Result(a, b, _) => format!("Result[{}, {}]", canon_type(a), canon_type(b)),
        TypeRef::Option(t, _) => format!("Option[{}]", canon_type(t)),
        TypeRef::Effect(t, _) => format!("Effect[{}]", canon_type(t)),
        TypeRef::HttpResult(t, _) => format!("HttpResult[{}]", canon_type(t)),
        TypeRef::QueueResult(_) => "QueueResult".to_string(),
        TypeRef::List(t, _) => format!("List[{}]", canon_type(t)),
        TypeRef::Map(k, v, _) => format!("Map[{}, {}]", canon_type(k), canon_type(v)),
        TypeRef::Query(t, _) => format!("Query[{}]", canon_type(t)),
        TypeRef::Stream(t, _) => format!("Stream[{}]", canon_type(t)),
        TypeRef::Connection(t, _) => format!("Connection[{}]", canon_type(t)),
        TypeRef::History(t, _) => format!("History[{}]", canon_type(t)),
        TypeRef::ValidationError(_) => "ValidationError".to_string(),
        TypeRef::JsonError(_) => "JsonError".to_string(),
        TypeRef::Unit(_) => "()".to_string(),
        TypeRef::Fn(params, ret, _) => format!(
            "({}) -> {}",
            params.iter().map(canon_type).collect::<Vec<_>>().join(", "),
            canon_type(ret)
        ),
        TypeRef::App { name, args, .. } => format!(
            "{}[{}]",
            name.name,
            args.iter().map(canon_type).collect::<Vec<_>>().join(", ")
        ),
    }
}

/// The outcome of reconciling one event against its (possibly absent)
/// registry entry.
enum Reconciled {
    /// No prior entry: baseline silently at `declared` (the event's current
    /// `@schema(N)`, or `1`).
    Baseline { fields: Vec<FieldShape> },
    /// Shape unchanged since the stored entry.
    Unchanged {
        fields: Vec<FieldShape>,
        stored: i64,
    },
    /// Every added field carries a default; nothing removed or retyped.
    Additive {
        fields: Vec<FieldShape>,
        bumped: i64,
    },
    /// A field was removed, retyped, added without a default, or lost a
    /// default it previously had.
    NonAdditive {
        removed: Vec<String>,
        retyped: Vec<String>,
        added_without_default: Vec<String>,
        lost_default: Vec<String>,
    },
}

fn reconcile_one(current: &[FieldShape], stored: Option<&EventEntry>) -> Reconciled {
    let Some(stored) = stored else {
        return Reconciled::Baseline {
            fields: current.to_vec(),
        };
    };
    if current == stored.fields.as_slice() {
        return Reconciled::Unchanged {
            fields: current.to_vec(),
            stored: stored.schema,
        };
    }

    fn by_name(fields: &[FieldShape]) -> BTreeMap<&str, &FieldShape> {
        fields.iter().map(|f| (f.name.as_str(), f)).collect()
    }
    let old = by_name(&stored.fields);
    let new = by_name(current);

    let removed: Vec<String> = old
        .keys()
        .filter(|n| !new.contains_key(*n))
        .map(|n| n.to_string())
        .collect();
    let retyped: Vec<String> = old
        .iter()
        .filter_map(|(n, old_field)| {
            new.get(n)
                .filter(|new_field| new_field.ty != old_field.ty)
                .map(|_| n.to_string())
        })
        .collect();
    let added_without_default: Vec<String> = new
        .iter()
        .filter(|(n, f)| !old.contains_key(*n) && !f.default)
        .map(|(n, _)| n.to_string())
        .collect();
    // A field present in both, same type, that had a default and no longer
    // does: an older wire event that omitted this key (relying on the
    // default) can no longer decode — the same forward-compatibility break
    // as a field added without one, just approached from the other side.
    let lost_default: Vec<String> = old
        .iter()
        .filter_map(|(n, old_field)| {
            new.get(n).filter(|new_field| {
                new_field.ty == old_field.ty && old_field.default && !new_field.default
            })
        })
        .map(|f| f.name.clone())
        .collect();

    if removed.is_empty()
        && retyped.is_empty()
        && added_without_default.is_empty()
        && lost_default.is_empty()
    {
        Reconciled::Additive {
            fields: current.to_vec(),
            bumped: stored.schema + 1,
        }
    } else {
        Reconciled::NonAdditive {
            removed,
            retyped,
            lost_default,
            added_without_default,
        }
    }
}

/// Reconcile every event across every unit against the given registry.
/// Returns the updated registry document and the effective schema version
/// for each event, keyed the same way as the document itself
/// (`<unit>.<EventName>`). Diagnostics for a mismatched `@schema(N)` or a
/// non-additive change are pushed onto `errors`.
pub fn reconcile(
    existing: &SchemaRegistry,
    unit_tables: &std::collections::HashMap<String, UnitTable>,
    errors: &mut Vec<CompileError>,
) -> (SchemaRegistry, std::collections::HashMap<String, i64>) {
    let mut updated = SchemaRegistry::new();
    let mut effective = std::collections::HashMap::new();

    let mut units: Vec<_> = unit_tables.iter().collect();
    units.sort_by_key(|(name, _)| *name);

    for (unit_name, table) in units {
        let mut events: Vec<_> = table.events.iter().collect();
        events.sort_by_key(|(name, _)| *name);
        for (event_name, event) in events {
            let key = format!("{unit_name}.{event_name}");
            let fields = snapshot(event);
            let declared = event.schema_version();
            let annotation_span = event
                .annotations
                .iter()
                .find(|a| a.name.name == "schema")
                .map(|a| a.span);
            let has_annotation = annotation_span.is_some();

            let (effective_version, entry) = match reconcile_one(&fields, existing.get(&key)) {
                Reconciled::Baseline { fields } => (
                    declared,
                    EventEntry {
                        schema: declared,
                        fields,
                    },
                ),
                Reconciled::Unchanged { fields, stored } => {
                    if has_annotation && declared != stored {
                        errors.push(mismatch_error(
                            event_name,
                            annotation_span.unwrap(),
                            declared,
                            stored,
                        ));
                    }
                    (
                        stored,
                        EventEntry {
                            schema: stored,
                            fields,
                        },
                    )
                }
                Reconciled::Additive { fields, bumped } => {
                    if has_annotation && declared != bumped {
                        errors.push(mismatch_error(
                            event_name,
                            annotation_span.unwrap(),
                            declared,
                            bumped,
                        ));
                    }
                    (
                        bumped,
                        EventEntry {
                            schema: bumped,
                            fields,
                        },
                    )
                }
                Reconciled::NonAdditive {
                    removed,
                    retyped,
                    added_without_default,
                    lost_default,
                } => {
                    errors.push(non_additive_error(
                        event,
                        event_name,
                        &removed,
                        &retyped,
                        &added_without_default,
                        &lost_default,
                    ));
                    // Keep the old entry as-is: a build that fails here
                    // must not perturb the registry (`compile_project`
                    // skips the write on any error too — this just keeps
                    // the in-memory document consistent with that).
                    let old = existing
                        .get(&key)
                        .expect("a NonAdditive verdict only fires against a stored entry")
                        .clone();
                    (old.schema, old)
                }
            };

            effective.insert(key.clone(), effective_version);
            updated.insert(key, entry);
        }
    }

    (updated, effective)
}

fn mismatch_error(
    event_name: &str,
    span: bynk_syntax::span::Span,
    declared: i64,
    computed: i64,
) -> CompileError {
    CompileError::new(
        "bynk.event.schema_version_mismatch",
        span,
        format!(
            "`{event_name}`'s `@schema({declared})` disagrees with the schema \
             registry, which computes version {computed} from the event's \
             build history"
        ),
    )
    .with_note(format!(
        "update the annotation to `@schema({computed})`, or remove it to let \
         the compiler track the version automatically"
    ))
}

fn non_additive_error(
    event: &EventDecl,
    event_name: &str,
    removed: &[String],
    retyped: &[String],
    added_without_default: &[String],
    lost_default: &[String],
) -> CompileError {
    let mut parts = Vec::new();
    if !removed.is_empty() {
        parts.push(format!("field(s) removed: {}", removed.join(", ")));
    }
    if !retyped.is_empty() {
        parts.push(format!("field(s) retyped: {}", retyped.join(", ")));
    }
    if !added_without_default.is_empty() {
        parts.push(format!(
            "field(s) added without a default: {}",
            added_without_default.join(", ")
        ));
    }
    if !lost_default.is_empty() {
        parts.push(format!(
            "field(s) lost their default: {}",
            lost_default.join(", ")
        ));
    }
    CompileError::new(
        "bynk.event.non_additive_schema_change",
        event.span,
        format!(
            "`{event_name}` changed in a way the schema registry cannot \
             evolve additively — {}",
            parts.join("; ")
        ),
    )
    .with_note(
        "an additive change adds only fields that carry a default; give a \
         breaking change a new event type name instead",
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use bynk_syntax::ast::{
        Annotation, AnnotationArg, BaseType, Expr, ExprId, ExprKind, Ident, RecordBody, Trivia,
    };
    use bynk_syntax::span::Span;
    use std::collections::HashMap as StdHashMap;

    fn ident(name: &str) -> Ident {
        Ident {
            name: name.to_string(),
            span: Span::default(),
        }
    }

    fn int_lit(value: i64) -> Expr {
        Expr {
            id: ExprId::SYNTHETIC,
            kind: ExprKind::IntLit {
                value,
                lexeme: value.to_string(),
            },
            span: Span::default(),
        }
    }

    fn field(name: &str, ty: TypeRef, has_default: bool) -> bynk_syntax::ast::RecordField {
        bynk_syntax::ast::RecordField {
            trivia: Default::default(),
            name: ident(name),
            type_ref: ty,
            refinement: None,
            init: has_default.then(|| int_lit(0)),
            span: Span::default(),
        }
    }

    fn schema_annotation(n: i64) -> Annotation {
        Annotation {
            name: ident("schema"),
            args: vec![AnnotationArg {
                label: None,
                value: int_lit(n),
                span: Span::default(),
            }],
            span: Span::default(),
        }
    }

    fn event(
        name: &str,
        annotations: Vec<Annotation>,
        fields: Vec<bynk_syntax::ast::RecordField>,
    ) -> EventDecl {
        EventDecl {
            name: ident(name),
            annotations,
            body: RecordBody {
                trailing_comments: Default::default(),
                fields,
                span: Span::default(),
            },
            documentation: None,
            span: Span::default(),
            trivia: Trivia::default(),
        }
    }

    fn int_ty() -> TypeRef {
        TypeRef::Base(BaseType::Int, Span::default())
    }
    fn string_ty() -> TypeRef {
        TypeRef::Base(BaseType::String, Span::default())
    }

    fn shape(name: &str, ty: &str, default: bool) -> FieldShape {
        FieldShape {
            name: name.to_string(),
            ty: ty.to_string(),
            default,
        }
    }

    // -- canon_type -----------------------------------------------------

    #[test]
    fn canon_type_renders_base_and_generic_shapes() {
        assert_eq!(canon_type(&int_ty()), "Int");
        assert_eq!(
            canon_type(&TypeRef::Option(Box::new(string_ty()), Span::default())),
            "Option[String]"
        );
        assert_eq!(
            canon_type(&TypeRef::List(Box::new(int_ty()), Span::default())),
            "List[Int]"
        );
    }

    // -- reconcile_one ----------------------------------------------------

    #[test]
    fn no_entry_baselines_silently() {
        let current = vec![shape("orderId", "String", false)];
        match reconcile_one(&current, None) {
            Reconciled::Baseline { fields } => assert_eq!(fields, current),
            _ => panic!("expected Baseline"),
        }
    }

    #[test]
    fn unchanged_shape_keeps_stored_version() {
        let current = vec![shape("orderId", "String", false)];
        let stored = EventEntry {
            schema: 2,
            fields: current.clone(),
        };
        match reconcile_one(&current, Some(&stored)) {
            Reconciled::Unchanged { stored: v, .. } => assert_eq!(v, 2),
            _ => panic!("expected Unchanged"),
        }
    }

    #[test]
    fn additive_field_with_default_bumps_version() {
        let old = vec![shape("orderId", "String", false)];
        let new = vec![
            shape("orderId", "String", false),
            shape("region", "Region", true),
        ];
        let stored = EventEntry {
            schema: 1,
            fields: old,
        };
        match reconcile_one(&new, Some(&stored)) {
            Reconciled::Additive { bumped, .. } => assert_eq!(bumped, 2),
            _ => panic!("expected Additive"),
        }
    }

    #[test]
    fn field_removed_is_non_additive() {
        let old = vec![
            shape("orderId", "String", false),
            shape("region", "Region", false),
        ];
        let new = vec![shape("orderId", "String", false)];
        let stored = EventEntry {
            schema: 1,
            fields: old,
        };
        match reconcile_one(&new, Some(&stored)) {
            Reconciled::NonAdditive { removed, .. } => {
                assert_eq!(removed, vec!["region".to_string()])
            }
            _ => panic!("expected NonAdditive"),
        }
    }

    #[test]
    fn field_retyped_is_non_additive() {
        let old = vec![shape("orderId", "String", false)];
        let new = vec![shape("orderId", "Int", false)];
        let stored = EventEntry {
            schema: 1,
            fields: old,
        };
        match reconcile_one(&new, Some(&stored)) {
            Reconciled::NonAdditive { retyped, .. } => {
                assert_eq!(retyped, vec!["orderId".to_string()])
            }
            _ => panic!("expected NonAdditive"),
        }
    }

    #[test]
    fn field_added_without_default_is_non_additive() {
        let old = vec![shape("orderId", "String", false)];
        let new = vec![
            shape("orderId", "String", false),
            shape("region", "Region", false),
        ];
        let stored = EventEntry {
            schema: 1,
            fields: old,
        };
        match reconcile_one(&new, Some(&stored)) {
            Reconciled::NonAdditive {
                added_without_default,
                ..
            } => assert_eq!(added_without_default, vec!["region".to_string()]),
            _ => panic!("expected NonAdditive"),
        }
    }

    #[test]
    fn a_field_losing_its_default_is_non_additive() {
        // An older wire event that omitted `region` (relying on the default)
        // could no longer decode once the default is gone — the same
        // forward-compatibility break as adding a field without one, from
        // the other direction.
        let old = vec![
            shape("orderId", "String", false),
            shape("region", "String", true),
        ];
        let new = vec![
            shape("orderId", "String", false),
            shape("region", "String", false),
        ];
        let stored = EventEntry {
            schema: 1,
            fields: old,
        };
        match reconcile_one(&new, Some(&stored)) {
            Reconciled::NonAdditive { lost_default, .. } => {
                assert_eq!(lost_default, vec!["region".to_string()])
            }
            _ => panic!("expected NonAdditive"),
        }
    }

    // -- reconcile (full pipeline, with real EventDecls) -------------------

    fn table_with(events: Vec<(&str, EventDecl)>) -> UnitTable {
        UnitTable {
            kind: None,
            types: StdHashMap::new(),
            fns: StdHashMap::new(),
            methods: StdHashMap::new(),
            capabilities: StdHashMap::new(),
            providers: StdHashMap::new(),
            services: StdHashMap::new(),
            agents: StdHashMap::new(),
            actors: StdHashMap::new(),
            exported_capabilities: Default::default(),
            events: events
                .into_iter()
                .map(|(n, e)| (n.to_string(), e))
                .collect(),
            flattened_caps: StdHashMap::new(),
        }
    }

    #[test]
    fn reconcile_baselines_a_brand_new_event_at_its_declared_annotation() {
        let e = event(
            "PaymentConfirmed",
            vec![schema_annotation(3)],
            vec![field("orderId", string_ty(), false)],
        );
        let mut units = StdHashMap::new();
        units.insert(
            "commerce.order".to_string(),
            table_with(vec![("PaymentConfirmed", e)]),
        );
        let existing = SchemaRegistry::new();
        let mut errors = Vec::new();
        let (updated, effective) = reconcile(&existing, &units, &mut errors);
        assert!(errors.is_empty(), "a first-ever compile must not error");
        assert_eq!(effective.get("commerce.order.PaymentConfirmed"), Some(&3));
        assert_eq!(
            updated
                .get("commerce.order.PaymentConfirmed")
                .unwrap()
                .schema,
            3
        );
    }

    #[test]
    fn reconcile_rejects_a_mismatched_schema_annotation_on_an_unchanged_event() {
        let e = event(
            "PaymentConfirmed",
            vec![schema_annotation(5)],
            vec![field("orderId", string_ty(), false)],
        );
        let mut units = StdHashMap::new();
        units.insert(
            "commerce.order".to_string(),
            table_with(vec![("PaymentConfirmed", e)]),
        );
        let mut existing = SchemaRegistry::new();
        existing.insert(
            "commerce.order.PaymentConfirmed".to_string(),
            EventEntry {
                schema: 2,
                fields: vec![shape("orderId", "String", false)],
            },
        );
        let mut errors = Vec::new();
        let (_, effective) = reconcile(&existing, &units, &mut errors);
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0].category, "bynk.event.schema_version_mismatch");
        // The registry's computed version wins as the effective one, not the
        // author's stale annotation.
        assert_eq!(effective.get("commerce.order.PaymentConfirmed"), Some(&2));
    }

    #[test]
    fn reconcile_auto_bumps_an_unannotated_additive_change() {
        let e = event(
            "OrderCancelled",
            vec![],
            vec![
                field("orderId", string_ty(), false),
                field("reason", string_ty(), true),
            ],
        );
        let mut units = StdHashMap::new();
        units.insert(
            "commerce.order".to_string(),
            table_with(vec![("OrderCancelled", e)]),
        );
        let mut existing = SchemaRegistry::new();
        existing.insert(
            "commerce.order.OrderCancelled".to_string(),
            EventEntry {
                schema: 1,
                fields: vec![shape("orderId", "String", false)],
            },
        );
        let mut errors = Vec::new();
        let (updated, effective) = reconcile(&existing, &units, &mut errors);
        assert!(errors.is_empty());
        assert_eq!(effective.get("commerce.order.OrderCancelled"), Some(&2));
        assert_eq!(
            updated.get("commerce.order.OrderCancelled").unwrap().schema,
            2
        );
    }

    #[test]
    fn reconcile_rejects_a_non_additive_change_and_keeps_the_old_entry() {
        let e = event(
            "OrderCancelled",
            vec![],
            vec![field("orderId", string_ty(), false)],
        );
        let mut units = StdHashMap::new();
        units.insert(
            "commerce.order".to_string(),
            table_with(vec![("OrderCancelled", e)]),
        );
        let mut existing = SchemaRegistry::new();
        existing.insert(
            "commerce.order.OrderCancelled".to_string(),
            EventEntry {
                schema: 4,
                fields: vec![
                    shape("orderId", "String", false),
                    shape("reason", "String", false),
                ],
            },
        );
        let mut errors = Vec::new();
        let (updated, effective) = reconcile(&existing, &units, &mut errors);
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0].category, "bynk.event.non_additive_schema_change");
        assert_eq!(effective.get("commerce.order.OrderCancelled"), Some(&4));
        assert_eq!(
            updated.get("commerce.order.OrderCancelled").unwrap().schema,
            4
        );
    }

    #[test]
    fn a_stale_key_for_a_renamed_event_is_dropped_silently() {
        // The old table had `PaymentConfirmed`; the new one only has
        // `PaymentConfirmedV2` (the track's prescribed rename-for-breaking-
        // change path). No diagnostic, and the old key does not survive into
        // the updated document.
        let e = event(
            "PaymentConfirmedV2",
            vec![],
            vec![field("orderId", string_ty(), false)],
        );
        let mut units = StdHashMap::new();
        units.insert(
            "commerce.order".to_string(),
            table_with(vec![("PaymentConfirmedV2", e)]),
        );
        let mut existing = SchemaRegistry::new();
        existing.insert(
            "commerce.order.PaymentConfirmed".to_string(),
            EventEntry {
                schema: 3,
                fields: vec![shape("orderId", "String", false)],
            },
        );
        let mut errors = Vec::new();
        let (updated, _) = reconcile(&existing, &units, &mut errors);
        assert!(errors.is_empty());
        assert!(updated.get("commerce.order.PaymentConfirmed").is_none());
        assert!(updated.get("commerce.order.PaymentConfirmedV2").is_some());
    }

    // `serialize`, and `parse`'s own parsing rules, are
    // `bynk_project::schema_registry`'s (P4.0, #1113) — covered by that
    // crate's own tests; nothing left to duplicate here.

    // -- parse_or_diagnose (P5.5) ------------------------------------------

    #[test]
    fn parse_or_diagnose_passes_through_a_valid_registry() {
        let mut reg = SchemaRegistry::new();
        reg.insert(
            "commerce.order.PaymentConfirmed".to_string(),
            EventEntry {
                schema: 1,
                fields: vec![shape("orderId", "String", false)],
            },
        );
        let text = bynk_project::schema_registry::serialize(&reg);
        let parsed = parse_or_diagnose(Some(&text), std::path::Path::new("/tmp"))
            .expect("a freshly serialized registry must parse");
        assert_eq!(
            parsed
                .get("commerce.order.PaymentConfirmed")
                .map(|e| e.schema),
            Some(1)
        );
    }

    #[test]
    fn parse_or_diagnose_reports_a_corrupt_registry_under_its_own_code() {
        let err = parse_or_diagnose(Some("not valid toml {{{"), std::path::Path::new("/tmp"))
            .expect_err("garbage content must not parse");
        assert_eq!(err.category, "bynk.project.schema_registry_corrupt");
    }

    #[test]
    fn parse_or_diagnose_with_no_existing_content_baselines_empty() {
        let parsed = parse_or_diagnose(None, std::path::Path::new("/tmp"))
            .expect("no lock file yet is not corruption");
        assert!(parsed.get("anything.at_all").is_none());
    }
}