memstead-schema 0.11.0

//! Schema loader — reads on-disk schema directories (or in-memory YAML) into
//! validated `Schema` values with `edge_weights` resolved.
//!
//! Three validation layers coordinate here:
//! 1. Structural (serde + `deny_unknown_fields`) — handled by the deserialize.
//! 2. Semantic (this module) — cross-field rules listed in `SchemaLoadError`.
//! 3. Editor (JSON Schemas) — generated by `emit_json_schemas`, consumed by
//!    schema authors via `# yaml-language-server: $schema=...`.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use indexmap::IndexMap;
use thiserror::Error;

use crate::base_metadata;
use crate::manifest::SchemaManifest;
use crate::schema::Schema;
use crate::types::TypeDefinition;

#[derive(Debug, Error)]
pub enum SchemaLoadError {
    #[error("i/o error reading {}: {source}", .path.display())]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },

    #[error("failed to parse manifest {}: {source}", .path.display())]
    ParseManifest {
        path: PathBuf,
        #[source]
        source: serde_yaml_ng::Error,
    },

    #[error("failed to parse type file {}: {source}", .path.display())]
    ParseType {
        path: PathBuf,
        #[source]
        source: serde_yaml_ng::Error,
    },

    #[error("invalid version '{value}': must be semver (e.g. 1.0.0)")]
    InvalidVersion { value: String },

    #[error("invalid schema name '{value}': {reason}")]
    InvalidName { value: String, reason: &'static str },

    #[error(
        "schema type file mismatch — declared in manifest: [{}], found in types/: [{}]",
        declared.join(", "),
        found.join(", ")
    )]
    TypeFileMismatch {
        declared: Vec<String>,
        found: Vec<String>,
    },

    #[error(
        "type file '{file}.yaml' has `name: {declared}` — filename and `name` field must match"
    )]
    TypeNameMismatch { file: String, declared: String },
    /// The retired `propagating_relationships:` key was used in an
    /// authoring context (a schema package loaded from a directory —
    /// workspace `.memstead/schemas/`, `schema install`, `schema
    /// validate`). The key's name promised propagation the engine
    /// never performed; its only effect — refusing self-loops on the
    /// listed rel-types — now lives under `no_self_loop_relationships`.
    /// Sealed content (built-ins, installed refs) keeps loading with
    /// the old key translated; only authoring refuses, so the fix is
    /// one mechanical key rename per schema.
    #[error(
        "type '{type_name}': `propagating_relationships` was renamed — its only effect is \
         refusing self-loops on the listed rel-types, so the key is now \
         `no_self_loop_relationships` (optional; empty lists can simply be deleted). \
         Rename the key and retry."
    )]
    PropagatingRelationshipsRenamed { type_name: String },

    #[error(
        "type '{type_name}' declares the retired `examples:` list — it was never \
         validated nor served and is replaced by the engine-validated `exemplar:` \
         (one canonical entity: title, metadata, sections, relations with \
         placeholder targets). Move the material into `exemplar:` and retry."
    )]
    ExamplesRetired { type_name: String },

    /// The retired `optional:` metadata-field key was used in an
    /// authoring context. The polarity flipped (first-author-path
    /// plan 07): a field is optional unless it declares
    /// `required: true` — the same rule sections follow. Sealed
    /// content keeps loading with the old key inverted; only
    /// authoring refuses, so the fix is one mechanical edit.
    #[error(
        "type '{type_name}' metadata field '{field}' declares the retired `optional:` key — \
         fields are optional unless they declare `required: true`. Fix: delete `optional: true`; \
         replace `optional: false` with `required: true`. Then retry."
    )]
    OptionalRetired { type_name: String, field: String },

    /// A `due:` declaration referencing fields the type does not have
    /// in the required shapes. `offender` names the bad reference,
    /// `reason` states the shape rule it violates — the due axis is
    /// spec (an external declaration), so its references validate at
    /// load with the loader's usual recovery quality.
    #[error("type '{type_name}' due axis is invalid: {reason} — offending name: '{offender}'")]
    InvalidDueAxis {
        type_name: String,
        offender: String,
        reason: String,
    },

    #[error("schema relationship vocabulary must include a '_default' definition")]
    MissingDefaultWeight,

    #[error("duplicate relationship definition: '{name}'")]
    DuplicateRelationship { name: String },

    #[error(
        "type '{type_name}' references relationship '{relationship}' in field '{field}' — not declared in schema. Available: [{}]. {}",
        available.join(", "),
        format_suggestion(relationship, available)
    )]
    UndeclaredRelationship {
        type_name: String,
        field: &'static str,
        relationship: String,
        available: Vec<String>,
    },

    #[error(
        "type '{type_name}' must have exactly one section with `catch_all: true` (found {count})"
    )]
    CatchAllViolation { type_name: String, count: usize },

    #[error(
        "type '{type_name}' field '{field}' references unknown key '{reference}' — not a section or metadata field"
    )]
    UnknownFieldReference {
        type_name: String,
        field: &'static str,
        reference: String,
    },

    #[error(
        "type '{type_name}' constraint ({kind}) is invalid: {reason} — offending name: '{offender}'"
    )]
    InvalidConstraint {
        type_name: String,
        kind: &'static str,
        offender: String,
        reason: String,
    },

    #[error("relationships.acyclic_sets is invalid: {reason} — offending entry: '{offender}'")]
    InvalidAcyclicSet { offender: String, reason: String },

    #[error("relationships.labelling is invalid: {reason} — offending name: '{offender}'")]
    InvalidLabelling { offender: String, reason: String },

    #[error(
        "type '{type_name}' section '{section}' format declaration is invalid: {}",
        problems.join("; ")
    )]
    InvalidSectionFormat {
        type_name: String,
        section: String,
        /// EVERY problem of the section's declaration — the loader
        /// names all offenders, never the first only, so one repair
        /// pass fixes the schema.
        problems: Vec<String>,
    },

    #[error(
        "type '{type_name}' metadata field '{field}' default '{default}' is not listed in enum_values: [{}]",
        allowed.join(", ")
    )]
    DefaultValueNotInEnum {
        type_name: String,
        field: String,
        default: String,
        allowed: Vec<String>,
    },

    #[error(
        "type '{type_name}' redeclares engine-implicit metadata key '{field}' — remove it from the YAML; the loader injects it automatically"
    )]
    RedeclaredBaseField { type_name: String, field: String },

    #[error(
        "relationship '{relationship}' field '{field}' references unknown type '{reference}'. Declared types: [{}]. {}",
        declared.join(", "),
        format_suggestion(reference, declared)
    )]
    UndeclaredRelationshipType {
        relationship: String,
        field: &'static str,
        reference: String,
        declared: Vec<String>,
    },

    /// Schema declares a regular section or metadata field whose key
    /// collides with an engine-invariant key. The reserved set covers
    /// section key `relationships` (the parser's auto-managed
    /// `## Relationships` section) and the metadata identity/
    /// discriminator triple `type` / `mem` / `id`
    /// ([`reserved_metadata_field_keys`]). Sections refuse at load;
    /// metadata keys refuse on the install/strict-validation path
    /// ([`check_reserved_metadata_keys`]) so sealed schemas keep
    /// booting.
    #[error(
        "type '{type_name}' declares {kind} with reserved key '{offending_key}' — reserved keys: [{}]",
        reserved_keys.join(", ")
    )]
    ReservedSchemaKey {
        type_name: String,
        kind: &'static str,
        offending_key: String,
        reserved_keys: Vec<String>,
    },

    /// A `cross_mem_relationships:` entry's `to_schema:` field is
    /// not a bare schema name. Cross-mem eligibility is name-based —
    /// versioned (`software@1.0.0`) and range (`software@^1.0`) forms
    /// are refused so a version component can never silently re-enter
    /// the eligibility path.
    #[error(
        "cross_mem_relationships[].to_schema '{value}' {reason} — expected a bare schema name (e.g. 'software', not 'software@1.0.0')"
    )]
    InvalidCrossMemToSchema { value: String, reason: String },

    /// Two `cross_mem_relationships:` entries declare the same
    /// `to_schema:`. A schema declares each target-schema at most once
    /// — the second entry would otherwise silently shadow or split
    /// the vocabulary.
    #[error("cross_mem_relationships declares duplicate to_schema '{to_schema}'")]
    DuplicateCrossMemToSchema { to_schema: String },

    /// A sealed package's file list carries no `schema.yaml`. Only
    /// reachable through [`load_sealed_package`] — the directory
    /// loader surfaces a missing manifest as [`Self::Io`] instead.
    #[error("sealed schema package carries no schema.yaml")]
    SealedPackageMissingManifest,

    /// A `to_schema: "*"` wildcard entry in a schema that declares no
    /// `alias_target_rel_type`. The wildcard is BOUND to the
    /// alias-synthesised rel-type — a schema that has not opted into
    /// alias synthesis has made no decision the wildcard could extend.
    #[error(
        "cross_mem_relationships declares to_schema '*' but the schema declares no \
         alias_target_rel_type — the wildcard is bound to the alias-synthesised rel-type; \
         declare alias_target_rel_type, or name each destination schema explicitly"
    )]
    CrossMemWildcardWithoutAliasTarget,

    /// A `to_schema: "*"` wildcard entry declaring a rel-type other
    /// than the schema's `alias_target_rel_type`. Hand-authored
    /// structural edges keep requiring a per-destination-schema
    /// declaration — the wildcard only extends the soft, auto-emitted
    /// alias references the author already permitted.
    #[error(
        "cross_mem_relationships[to_schema='*'] declares rel-type '{rel_type}', but the \
         wildcard is bound to the schema's alias_target_rel_type '{alias_target}' — \
         hand-authored structural edges need a per-destination-schema declaration"
    )]
    CrossMemWildcardNonAliasRelType {
        rel_type: String,
        alias_target: String,
    },

    /// A `cross_mem_relationships[].definitions[*].source_types` entry
    /// references a type name not declared in the source schema's
    /// `types` list. Source types belong to the source schema's
    /// namespace; unknown names raise this error at load time.
    /// (Target types are accepted as opaque strings — they belong to
    /// the target schema's namespace, which is not in scope here.)
    #[error(
        "cross_mem_relationships[to_schema='{to_schema}'].definitions[name='{relationship}'].source_types references unknown type '{reference}'. Declared types: [{}]. {}",
        declared.join(", "),
        format_suggestion(reference, declared)
    )]
    UndeclaredCrossMemSourceType {
        to_schema: String,
        relationship: String,
        reference: String,
        declared: Vec<String>,
    },

    /// The schema's `alias_target_rel_type:` pointer names a rel-type
    /// not declared in `relationships.definitions`. Surfaces at
    /// schema-load time so the alias-synthesis pass can trust the
    /// pointer is resolvable at every later mutation call.
    #[error(
        "schema '{schema}' alias_target_rel_type '{target}' is not declared in relationships. Declared: [{}]. {}",
        declared.join(", "),
        format_suggestion(target, declared)
    )]
    AliasTargetRelTypeNotDeclared {
        schema: String,
        target: String,
        declared: Vec<String>,
    },

    /// One or more declared section headings do not derive back to their
    /// declared keys (`derive_section_key(heading) != key`), so content
    /// written under the heading could never be parsed back into the
    /// section — it would silently fork into a second heading or fall
    /// through to the catch-all. Raised by
    /// [`check_section_heading_roundtrip`] on the authoring/installation
    /// path only; a schema already sealed into a mem-repo keeps loading
    /// and surfaces the condition through health instead.
    #[error(
        "schema declares section heading(s) that cannot round-trip to their key(s): {}. \
         Fix: make each heading derive to its key — lowercasing the heading and replacing \
         spaces with underscores must yield the key exactly (key `current_state` ⇒ heading \
         `Current State`)",
        format_heading_violations(violations)
    )]
    SectionHeadingMismatch {
        violations: Vec<HeadingKeyViolation>,
    },

    /// Two or more independent semantic violations found in one load
    /// pass. The loader accumulates every violation it can prove on
    /// successfully parsed structure and refuses once, so the author
    /// fixes the whole set in one edit instead of one violation per
    /// validate round. Structural failures (a manifest that does not
    /// parse, a declared-vs-found type-file mismatch) still
    /// short-circuit — everything downstream of them would be noise
    /// derived from a value that does not exist. A single violation is
    /// returned bare, never as a one-element list.
    #[error(
        "schema has {} violations:\n{}",
        errors.len(),
        format_multiple(errors)
    )]
    Multiple { errors: Vec<SchemaLoadError> },
}

fn format_multiple(errors: &[SchemaLoadError]) -> String {
    errors
        .iter()
        .enumerate()
        .map(|(i, e)| format!("  {}. {e}", i + 1))
        .collect::<Vec<_>>()
        .join("\n")
}

/// Fold an accumulated violation list into one error: a single
/// violation stays bare (the common case keeps today's message shape),
/// several wrap in [`SchemaLoadError::Multiple`]. Callers guarantee
/// the list is non-empty.
fn collapse(mut errors: Vec<SchemaLoadError>) -> SchemaLoadError {
    debug_assert!(!errors.is_empty());
    if errors.len() == 1 {
        errors.remove(0)
    } else {
        SchemaLoadError::Multiple { errors }
    }
}

/// One `(type, key, heading, derived_key)` tuple in a
/// [`SchemaLoadError::SectionHeadingMismatch`] refusal.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeadingKeyViolation {
    pub type_name: String,
    pub key: String,
    pub heading: String,
    pub derived_key: String,
}

fn format_heading_violations(violations: &[HeadingKeyViolation]) -> String {
    violations
        .iter()
        .map(|v| {
            format!(
                "type '{}' section key '{}' has heading '{}' (derives to '{}')",
                v.type_name, v.key, v.heading, v.derived_key
            )
        })
        .collect::<Vec<_>>()
        .join("; ")
}

/// Refuse a schema in which any declared section's heading does not
/// derive back to that section's declared key. Collects **every**
/// offending `(type, key, heading, derived_key)` tuple — a schema with
/// one good and one bad section is refused whole, and the author sees
/// the complete list in one round.
///
/// Installation-path gate only: callers are the schema-authoring
/// surfaces (CLI `schema validate` / `schema install`, the engine's
/// `install_schema` primitive). Boot and sealed-schema loads must NOT
/// call this — a schema already sealed on `__MEMSTEAD` that violates
/// the rule keeps loading, and the violation surfaces as a health
/// finding, never as a boot failure.
pub fn check_section_heading_roundtrip(schema: &Schema) -> Result<(), SchemaLoadError> {
    let mut violations = Vec::new();
    // Deterministic report order: sort type names (Schema.types is a
    // HashMap); sections keep declaration order within a type.
    let mut type_names: Vec<&String> = schema.types.keys().collect();
    type_names.sort();
    for type_name in type_names {
        let t = &schema.types[type_name];
        for s in &t.sections {
            let derived_key = crate::types::derive_section_key(&s.heading);
            if derived_key != s.key {
                violations.push(HeadingKeyViolation {
                    type_name: type_name.clone(),
                    key: s.key.clone(),
                    heading: s.heading.clone(),
                    derived_key,
                });
            }
        }
    }
    if violations.is_empty() {
        Ok(())
    } else {
        Err(SchemaLoadError::SectionHeadingMismatch { violations })
    }
}

/// Engine-invariant section keys reserved against schema use. The
/// parser's auto-managed `## Relationships` section is the only entry
/// today.
pub fn reserved_section_keys() -> &'static [&'static str] {
    &["relationships"]
}

/// Engine-invariant metadata-field keys reserved against schema use —
/// the entity's identity/discriminator triple. `type` is the engine-set
/// frontmatter discriminator; `mem` and `id` are the entity's
/// structural identity, owned by the engine's id grammar and mount
/// routing. One reservation, one behaviour: no installable schema may
/// declare any of them (see [`check_reserved_metadata_keys`]), and the
/// engine write paths refuse them as caller-supplied metadata.
pub fn reserved_metadata_field_keys() -> &'static [&'static str] {
    &["type", "mem", "id"]
}

/// Author/install-path check: refuse a schema whose types **declare**
/// an engine-reserved metadata-field key (`type` / `mem` / `id`).
/// Reads the raw pre-merge declaration list the loader records
/// (`TypeDefinition::declared_metadata_keys`) so the engine-injected
/// base fields never false-positive.
///
/// Same posture as [`check_section_heading_roundtrip`]: install and
/// strict validation call this and refuse; boot and sealed-schema
/// loads must NOT — a schema already sealed that violates the rule
/// keeps loading (refusing at boot would brick the workspace), and the
/// engine's write-path refusals keep the reserved keys unwritable
/// regardless.
pub fn check_reserved_metadata_keys(schema: &crate::Schema) -> Result<(), SchemaLoadError> {
    for td in schema.types.values() {
        for key in &td.declared_metadata_keys {
            if reserved_metadata_field_keys().contains(&key.as_str()) {
                return Err(SchemaLoadError::ReservedSchemaKey {
                    type_name: td.name.clone(),
                    kind: "metadata_field",
                    offending_key: key.clone(),
                    reserved_keys: reserved_metadata_field_keys()
                        .iter()
                        .map(|s| s.to_string())
                        .collect(),
                });
            }
        }
    }
    Ok(())
}

fn format_suggestion(needle: &str, candidates: &[String]) -> String {
    let mut best: Option<(usize, &String)> = None;
    for cand in candidates {
        let d = strsim::levenshtein(needle, cand);
        match best {
            Some((bd, _)) if bd <= d => {}
            _ => best = Some((d, cand)),
        }
    }
    match best {
        Some((d, cand)) if d > 0 && d <= needle.len().saturating_add(3) => {
            format!("Did you mean '{cand}'?")
        }
        _ => String::new(),
    }
}

/// Load a schema from a directory containing `schema.yaml` and `types/*.yaml`.
pub fn load_schema_from_dir(path: &Path) -> Result<Schema, SchemaLoadError> {
    let manifest_path = path.join("schema.yaml");
    let manifest_text =
        std::fs::read_to_string(&manifest_path).map_err(|e| SchemaLoadError::Io {
            path: manifest_path.clone(),
            source: e,
        })?;

    let types_dir = path.join("types");
    let mut type_files: Vec<(String, String)> = Vec::new();
    if types_dir.is_dir() {
        let entries = std::fs::read_dir(&types_dir).map_err(|e| SchemaLoadError::Io {
            path: types_dir.clone(),
            source: e,
        })?;
        for entry in entries {
            let entry = entry.map_err(|e| SchemaLoadError::Io {
                path: types_dir.clone(),
                source: e,
            })?;
            let p = entry.path();
            if p.extension().and_then(|s| s.to_str()) != Some("yaml") {
                continue;
            }
            let Some(stem) = p.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
                continue;
            };
            let contents = std::fs::read_to_string(&p).map_err(|e| SchemaLoadError::Io {
                path: p.clone(),
                source: e,
            })?;
            type_files.push((stem, contents));
        }
    }
    // read_dir order is filesystem-dependent; accumulated violations
    // must report in a stable order across runs on the same input.
    type_files.sort_by(|a, b| a.0.cmp(&b.0));

    load_with_context(
        &manifest_text,
        &type_files,
        Some(&manifest_path),
        Some(&types_dir),
        // Authoring context — the author writes the current language.
        MetadataPolarityFormat::RequiredOptIn,
    )
}

/// The metadata-field polarity generation of a sealed package —
/// decided by the presence of the package's format marker
/// (`schema-format.json`), never by heuristics over the document body.
///
/// Under the pre-flip language an absent `required`/`optional` key
/// meant **required**; under the current language absence means
/// **optional**. The two are syntactically indistinguishable, so an
/// unmarked sealed package reads with [`Self::Legacy`] semantics —
/// its effective behaviour conserved — while packages sealed from
/// this change on carry the marker and read as
/// [`Self::RequiredOptIn`]. Directory (authoring) loads are always
/// `RequiredOptIn`: the author writes against the current language.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MetadataPolarityFormat {
    /// Pre-flip sealed content: an absent key means required.
    Legacy,
    /// Current language: an absent key means optional.
    RequiredOptIn,
}

/// The format-marker file name sealed alongside a schema package on
/// the genuinely sealed surfaces — the `__MEMSTEAD` ref, published
/// `.mem` archives (the export carries it, the archive validator
/// admits it, the archive loader honors it), and new builtin version
/// directories. Presence ⇒ [`MetadataPolarityFormat::RequiredOptIn`];
/// absence ⇒ legacy. Content is informative JSON; presence is the
/// contract.
///
/// Directory loads (`load_schema_from_dir`: workspace
/// `.memstead/schemas/`, cache extractions, `schema validate` /
/// `install`) are the AUTHORING tier and never consult the marker —
/// they always read the current language, with the retired
/// `optional:` key refusing loudly. A pre-flip directory package
/// relying on absent-key-means-required flips soft (fields become
/// optional — admits more, refuses nothing); that is the fail-soft
/// direction by design, with `health_required_fields` and
/// constraints as the data-quality backstop.
pub const SCHEMA_FORMAT_MARKER_FILE: &str = "schema-format.json";

/// The marker file's canonical content.
pub const SCHEMA_FORMAT_MARKER_CONTENT: &str = "{\"metadata_polarity\":\"required-opt-in\"}\n";

/// Append the format marker to a package's file list if absent.
///
/// **For content verified current-language only.** The marker is a
/// generation claim, and this helper cannot verify one — a legacy
/// package stamped here would flip every bare field from required to
/// optional the moment the sealed copy is read. The one place that
/// KNOWS a package is current is the authoring/install resolver (a
/// directory source just validated under the current language, where
/// the retired `optional:` key refuses loudly); sealed and builtin
/// sources carry their generation as-found and are never stamped —
/// absence IS the legacy claim, and inventing a marker over it is the
/// silent-flip defect this contract exists to prevent.
pub fn with_format_marker(mut files: Vec<(String, Vec<u8>)>) -> Vec<(String, Vec<u8>)> {
    if !files
        .iter()
        .any(|(rel, _)| rel == SCHEMA_FORMAT_MARKER_FILE)
    {
        files.push((
            SCHEMA_FORMAT_MARKER_FILE.to_string(),
            SCHEMA_FORMAT_MARKER_CONTENT.as_bytes().to_vec(),
        ));
    }
    files
}

/// Load a schema from in-memory YAML strings — **legacy sealed
/// semantics** (an absent required/optional key means required).
/// Use [`load_schema_from_memory_with_format`] when the caller knows
/// the package's format generation from its marker.
///
/// `types_yamls` is a slice of `(filename_stem, contents)` tuples — the stems
/// must match `manifest.types` exactly.
pub fn load_schema_from_memory(
    manifest_yaml: &str,
    types_yamls: &[(String, String)],
) -> Result<Schema, SchemaLoadError> {
    load_with_context(
        manifest_yaml,
        types_yamls,
        None,
        None,
        MetadataPolarityFormat::Legacy,
    )
}

/// Load a **sealed** schema package from its `(relative-path, bytes)`
/// file list — the shape every sealed surface carries: a published
/// `.mem` archive's `.memstead/schema/` tree, the git-branch
/// `__MEMSTEAD:schemas/<name>@<version>/` tree, a built-in version
/// directory. Paths are package-relative (`schema.yaml`,
/// `types/<stem>.yaml`, `schema-format.json`); anything else is
/// ignored.
///
/// One function so the surfaces that *admit* a sealed package (the
/// archive validator) and the surfaces that later *read it back* (the
/// local schema source the install stages into) can never disagree —
/// a package valid to publish stays valid to install. The metadata
/// polarity comes from the package's own format marker, and keys
/// retired after the package was sealed read with their written
/// meaning instead of refusing.
///
/// Authoring content takes [`load_schema_from_dir`] instead, which
/// refuses retired keys loudly so the author can act.
pub fn load_sealed_package(files: &[(String, Vec<u8>)]) -> Result<Schema, SchemaLoadError> {
    let mut manifest: Option<String> = None;
    let mut types: Vec<(String, String)> = Vec::new();
    let mut marked = false;
    for (rel, bytes) in files {
        if rel == "schema.yaml" {
            manifest = Some(String::from_utf8_lossy(bytes).into_owned());
        } else if rel == SCHEMA_FORMAT_MARKER_FILE {
            marked = true;
        } else if let Some(stem) = rel
            .strip_prefix("types/")
            .and_then(|f| f.strip_suffix(".yaml"))
        {
            types.push((
                stem.to_string(),
                String::from_utf8_lossy(bytes).into_owned(),
            ));
        }
    }
    let manifest = manifest.ok_or(SchemaLoadError::SealedPackageMissingManifest)?;
    // Deterministic order regardless of how the caller enumerated the
    // package — accumulated violations must report the same way twice.
    types.sort_by(|a, b| a.0.cmp(&b.0));
    let format = if marked {
        MetadataPolarityFormat::RequiredOptIn
    } else {
        MetadataPolarityFormat::Legacy
    };
    load_with_context(&manifest, &types, None, None, format)
}

/// Would this package content pass the AUTHORING tier under the
/// current language? Same strictness as [`load_schema_from_dir`] —
/// retired keys (`propagating_relationships`, `examples:`, `optional:`)
/// refuse loudly — but over in-memory content, so a caller holding a
/// sealed package's files (git-branch `__MEMSTEAD:schemas/` tree,
/// folder seal, archive) can ask whether the content is still
/// re-authorable/installable without materialising a directory. The
/// tolerant sealed read ([`load_sealed_package`]) is unaffected; this
/// is a pure diagnosis — the health rot axis is the consumer.
pub fn check_package_reauthorable(
    manifest_yaml: &str,
    types_yamls: &[(String, String)],
) -> Result<(), SchemaLoadError> {
    // The `types_dir: Some(..)` context is what selects the authoring-
    // strict legacy-key gates; the path itself only labels error
    // messages.
    let strict_context = Path::new("<sealed package>");
    load_with_context(
        manifest_yaml,
        types_yamls,
        Some(strict_context),
        Some(strict_context),
        MetadataPolarityFormat::RequiredOptIn,
    )
    .map(|_| ())
}

/// Load a schema from in-memory YAML strings with an explicit
/// metadata-polarity format generation (from the sealed package's
/// format marker).
pub fn load_schema_from_memory_with_format(
    manifest_yaml: &str,
    types_yamls: &[(String, String)],
    format: MetadataPolarityFormat,
) -> Result<Schema, SchemaLoadError> {
    load_with_context(manifest_yaml, types_yamls, None, None, format)
}

fn load_with_context(
    manifest_yaml: &str,
    types_yamls: &[(String, String)],
    manifest_path: Option<&Path>,
    types_dir: Option<&Path>,
    format: MetadataPolarityFormat,
) -> Result<Schema, SchemaLoadError> {
    // Semantic-violation accumulator. Every check operating on
    // successfully parsed structure pushes here instead of returning,
    // so the author sees the complete violation set in one refusal;
    // only structural failures short-circuit (see
    // [`SchemaLoadError::Multiple`]). Order is deterministic:
    // manifest checks in declaration order, then type files in
    // `types_yamls` order (sorted by stem when loaded from a
    // directory).
    let mut errors: Vec<SchemaLoadError> = Vec::new();

    let mut manifest: SchemaManifest =
        serde_yaml_ng::from_str(manifest_yaml).map_err(|e| SchemaLoadError::ParseManifest {
            path: manifest_path
                .map(Path::to_path_buf)
                .unwrap_or_else(|| PathBuf::from("<memory>")),
            source: e,
        })?;

    if let Err(e) = validate_name(&manifest.name) {
        errors.push(e);
    }

    // `None` only ever coexists with a non-empty accumulator, so the
    // `Schema` construction at the bottom (reached only when the
    // accumulator is empty) can unwrap.
    let version = match semver::Version::parse(&manifest.version) {
        Ok(v) => Some(v),
        Err(_) => {
            errors.push(SchemaLoadError::InvalidVersion {
                value: manifest.version.clone(),
            });
            None
        }
    };

    // Relationship vocabulary: unique names + _default present
    let mut rel_names: HashSet<String> = HashSet::new();
    for def in &manifest.relationships.definitions {
        if !rel_names.insert(def.name.clone()) {
            errors.push(SchemaLoadError::DuplicateRelationship {
                name: def.name.clone(),
            });
        }
    }
    if !rel_names.contains("_default") {
        errors.push(SchemaLoadError::MissingDefaultWeight);
    }
    let available_rels: Vec<String> = manifest
        .relationships
        .definitions
        .iter()
        .map(|d| d.name.clone())
        .collect();

    // Validate that the schema-level alias_target_rel_type pointer (if
    // set) names a declared rel-type. The synthesis pass later relies
    // on this invariant — running it as a load-time check keeps the
    // mutation path's hot loop free of resolution failures.
    if let Some(target) = &manifest.alias_target_rel_type
        && !rel_names.contains(target)
    {
        let mut declared = available_rels.clone();
        declared.sort();
        errors.push(SchemaLoadError::AliasTargetRelTypeNotDeclared {
            schema: manifest.name.clone(),
            target: target.clone(),
            declared,
        });
    }

    // Acyclicity sets: each set names two or more DECLARED rel-types,
    // and a rel-type appears in at most one set across all sets
    // (overlapping sets have no coherent refusal message; a duplicate
    // inside one set is the same defect). A single-member set is the
    // per-definition `acyclic` flag's job and refuses here.
    let mut acyclic_set_member_seen: HashSet<&str> = HashSet::new();
    for set in &manifest.relationships.acyclic_sets {
        if set.len() < 2 {
            errors.push(SchemaLoadError::InvalidAcyclicSet {
                offender: set.join(", "),
                reason: "a set needs at least two rel-types (a single member is the \
                         per-definition `acyclic` flag)"
                    .to_string(),
            });
        }
        for name in set {
            if !rel_names.contains(name.as_str()) {
                errors.push(SchemaLoadError::InvalidAcyclicSet {
                    offender: name.clone(),
                    reason: "names no declared relationship".to_string(),
                });
            }
            if !acyclic_set_member_seen.insert(name.as_str()) {
                errors.push(SchemaLoadError::InvalidAcyclicSet {
                    offender: name.clone(),
                    reason: "a rel-type may appear in at most one acyclicity set".to_string(),
                });
            }
        }
    }

    // Labelling declaration: `attack` names at least one declared
    // rel-type; a `support` block names declared rel-types too (its
    // `terminal_types` need every type loaded and are checked in the
    // schema-level pass; its `direction` is a closed enum).
    if let Some(lab) = &manifest.relationships.labelling {
        if lab.attack.is_empty() {
            errors.push(SchemaLoadError::InvalidLabelling {
                offender: "(empty)".to_string(),
                reason: "`labelling.attack` must name at least one rel-type".to_string(),
            });
        }
        for name in &lab.attack {
            if !rel_names.contains(name.as_str()) {
                errors.push(SchemaLoadError::InvalidLabelling {
                    offender: name.clone(),
                    reason: "`labelling.attack` entry names no declared relationship".to_string(),
                });
            }
        }
        if let Some(sup) = &lab.support {
            if sup.relationships.is_empty() {
                errors.push(SchemaLoadError::InvalidLabelling {
                    offender: "(empty)".to_string(),
                    reason: "`labelling.support.relationships` must name at least one rel-type"
                        .to_string(),
                });
            }
            for name in &sup.relationships {
                if !rel_names.contains(name.as_str()) {
                    errors.push(SchemaLoadError::InvalidLabelling {
                        offender: name.clone(),
                        reason: "`labelling.support.relationships` entry names no declared \
                                 relationship"
                            .to_string(),
                    });
                }
            }
        }
    }

    // Option C coupling — auto-force `manual_authoring: forbidden` on
    // the rel-type named by `alias_target_rel_type`. Schemas setting
    // the pointer opt the named rel-type out of explicit authoring;
    // the only path to a relation of that rel-type is via the
    // alias-synthesis pass that emits one per body wiki-link. This
    // closes the explicit/synthesised coexistence question: with the
    // coupling in place, edges of the pointer rel-type are always
    // engine-emitted, so `EdgeSource::BodyLink` is unambiguous and
    // GC can drop pointer-rel-type relations without risking
    // explicit-author data.
    //
    // The coupling is silent — a schema that writes
    // `manual_authoring: allow` (or `warn`) on the named rel-type
    // gets overridden to `forbidden` at load. The override is the
    // schema-strictness contract; explicit `allow`/`warn` on the
    // pointer rel-type is meaningless under the design and would
    // surprise the validator at runtime, so the loader corrects it
    // here.
    if let Some(pointer) = manifest.alias_target_rel_type.clone() {
        for def in &mut manifest.relationships.definitions {
            if def.name == pointer {
                def.manual_authoring = crate::manifest::ManualAuthoring::Forbidden;
            }
        }
    }

    // Cross-check `source_types` / `target_types` on each relationship
    // definition against the manifest's declared type list. Unknown
    // names raise `UndeclaredRelationshipType` with a "did you mean"
    // suggestion — the schema-author equivalent of `INVALID_REL_SHAPE`.
    for def in &manifest.relationships.definitions {
        for t in &def.source_types {
            if !manifest.types.iter().any(|d| d == t) {
                errors.push(SchemaLoadError::UndeclaredRelationshipType {
                    relationship: def.name.clone(),
                    field: "source_types",
                    reference: t.clone(),
                    declared: manifest.types.clone(),
                });
            }
        }
        for t in &def.target_types {
            if !manifest.types.iter().any(|d| d == t) {
                errors.push(SchemaLoadError::UndeclaredRelationshipType {
                    relationship: def.name.clone(),
                    field: "target_types",
                    reference: t.clone(),
                    declared: manifest.types.clone(),
                });
            }
        }
    }

    // Cross-mem relationships: validate `to_schema` is a bare schema
    // name (cross-mem eligibility is name-based — a version suffix or
    // range refuses), refuse duplicate target schemas, and cross-check
    // `source_types` against the source schema's types. `target_types`
    // are accepted as opaque strings — they belong to the target
    // schema's namespace, which is out of scope at source-schema load
    // time. The target schema may not even be present in the workspace
    // when the source schema loads (and cross-mem declarations
    // targeting absent schemas are legitimate for portable library
    // schemas).
    let mut seen_to_schemas: HashSet<String> = HashSet::new();
    for entry in &manifest.cross_mem_relationships {
        if entry.to_schema == "*" {
            // Wildcard destination — bound to the alias-synthesised
            // rel-type. The binding IS the safety argument: the author
            // already permitted soft auto-emitted references of that
            // type; the wildcard extends that decision across the mem
            // boundary and introduces no new permission. Structural
            // rel-types stay per-destination-schema.
            match manifest.alias_target_rel_type.as_deref() {
                None => errors.push(SchemaLoadError::CrossMemWildcardWithoutAliasTarget),
                Some(alias) => {
                    for def in &entry.definitions {
                        if def.name != alias {
                            errors.push(SchemaLoadError::CrossMemWildcardNonAliasRelType {
                                rel_type: def.name.clone(),
                                alias_target: alias.to_string(),
                            });
                        }
                    }
                }
            }
        } else if entry.to_schema.contains('@') {
            errors.push(SchemaLoadError::InvalidCrossMemToSchema {
                value: entry.to_schema.clone(),
                reason: "must not carry a version or range".into(),
            });
        } else if let Err(reason) = name_shape(&entry.to_schema) {
            errors.push(SchemaLoadError::InvalidCrossMemToSchema {
                value: entry.to_schema.clone(),
                reason: reason.into(),
            });
        }
        if !seen_to_schemas.insert(entry.to_schema.clone()) {
            errors.push(SchemaLoadError::DuplicateCrossMemToSchema {
                to_schema: entry.to_schema.clone(),
            });
        }
        for def in &entry.definitions {
            for t in &def.source_types {
                if !manifest.types.iter().any(|d| d == t) {
                    errors.push(SchemaLoadError::UndeclaredCrossMemSourceType {
                        to_schema: entry.to_schema.clone(),
                        relationship: def.name.clone(),
                        reference: t.clone(),
                        declared: manifest.types.clone(),
                    });
                }
            }
        }
    }

    // Type file cross-check: declared vs found, set equality (order-insensitive)
    let mut found_stems: Vec<String> = types_yamls.iter().map(|(s, _)| s.clone()).collect();
    found_stems.sort();
    let mut declared = manifest.types.clone();
    declared.sort();
    if found_stems != declared {
        // Structural: the per-type pass below would run against a file
        // set the manifest never described. Refuse now, together with
        // every manifest-level violation already proven.
        errors.push(SchemaLoadError::TypeFileMismatch {
            declared,
            found: found_stems,
        });
        return Err(collapse(errors));
    }

    // Per-type defaults map for edge_weights resolution
    let defaults: IndexMap<String, f32> = manifest
        .relationships
        .definitions
        .iter()
        .map(|d| (d.name.clone(), d.default_weight))
        .collect();

    let mut types_map: HashMap<String, Arc<TypeDefinition>> = HashMap::new();
    let mut had_type_parse_failure = false;

    for (stem, text) in types_yamls {
        let type_path = types_dir
            .map(|d| d.join(format!("{stem}.yaml")))
            .unwrap_or_else(|| PathBuf::from(format!("<memory>/{stem}.yaml")));

        let mut td: TypeDefinition = match serde_yaml_ng::from_str(text) {
            Ok(td) => td,
            Err(e) => {
                // A type file that does not parse cannot be checked
                // semantically — record the parse failure, keep
                // checking the other files, and skip the cross-type
                // pass below (the missing type would make it noise).
                errors.push(SchemaLoadError::ParseType {
                    path: type_path.clone(),
                    source: e,
                });
                had_type_parse_failure = true;
                continue;
            }
        };

        if td.name != *stem {
            errors.push(SchemaLoadError::TypeNameMismatch {
                file: stem.clone(),
                declared: td.name.clone(),
            });
        }

        // Legacy-key gate: authoring contexts (loaded from a
        // directory — `types_dir` is `Some`) refuse the retired
        // `propagating_relationships` key with the rename error;
        // sealed contexts (in-memory: built-ins, installed refs)
        // translate it so shipped content keeps loading (install-time
        // strict, sealed-tolerant — the section-heading round-trip
        // doctrine).
        if let Some(legacy) = td.legacy_propagating_relationships.take() {
            if types_dir.is_some() {
                errors.push(SchemaLoadError::PropagatingRelationshipsRenamed {
                    type_name: td.name.clone(),
                });
            } else if td.no_self_loop_relationships.is_empty() {
                td.no_self_loop_relationships = legacy;
            }
        }

        // Retired `examples:` list (agent-trust plan 09): dead
        // vocabulary — never validated, never served. Authoring
        // contexts refuse with the pointer at `exemplar:`; sealed
        // contexts tolerate and drop (nothing consumed it, so
        // dropping is lossless).
        if td.legacy_examples.take().is_some() && types_dir.is_some() {
            errors.push(SchemaLoadError::ExamplesRetired {
                type_name: td.name.clone(),
            });
        }

        // Metadata-required polarity (first-author-path plan 07):
        // authoring refuses the retired `optional:` key naming the
        // inversion; sealed content inverts it. An absent key resolves
        // by the package's format generation — legacy sealed content
        // reads absence as required (its written meaning), everything
        // else as optional.
        for field in &mut td.metadata_fields {
            // Current-language contexts (directory/authoring loads and
            // install validation, both RequiredOptIn) refuse the
            // retired key; only Legacy sealed loads invert silently.
            if matches!(format, MetadataPolarityFormat::RequiredOptIn)
                && field.legacy_optional.is_some()
            {
                errors.push(SchemaLoadError::OptionalRetired {
                    type_name: td.name.clone(),
                    field: field.key.clone(),
                });
            }
            field.required_resolved = match (field.required, field.legacy_optional.take()) {
                (Some(required), _) => required,
                (None, Some(optional)) => !optional,
                (None, None) => matches!(format, MetadataPolarityFormat::Legacy),
            };
        }

        // Record the raw author-declared metadata keys BEFORE the
        // base-metadata merge, so the install-path reserved-key check
        // ([`check_reserved_metadata_keys`]) can tell a declared
        // `type`/`mem`/`id` from the engine-injected base fields. The
        // check itself deliberately does NOT run here: this loader
        // serves boot and sealed-schema reads too, and a schema sealed
        // before the reservation widened must keep loading (heading-
        // round-trip posture — refusal fires on the authoring/install
        // path, never at boot).
        td.declared_metadata_keys = td.metadata_fields.iter().map(|f| f.key.clone()).collect();

        // Reject redeclarations of remaining engine-implicit base metadata
        // (`created_date`, `last_modified`, `tags`). The reserved `type`
        // case is excluded here — it refuses with the typed reserved-key
        // error on the install path instead.
        for field in &td.metadata_fields {
            if base_metadata::is_base_key(&field.key)
                && !reserved_metadata_field_keys().contains(&field.key.as_str())
            {
                errors.push(SchemaLoadError::RedeclaredBaseField {
                    type_name: td.name.clone(),
                    field: field.key.clone(),
                });
            }
        }

        // Merge base metadata around the type-declared fields. Canonical
        // order: type, created_date, last_modified, <declared>, tags.
        let mut merged = base_metadata::prefix_fields();
        merged.append(&mut td.metadata_fields);
        merged.extend(base_metadata::suffix_fields());
        td.metadata_fields = merged;

        compile_section_formats(&mut td);
        validate_type(&td, &rel_names, &available_rels, &mut errors);

        // Resolve edge_weights: start with schema defaults, apply overrides.
        let mut weights = defaults.clone();
        for (k, v) in &td.edge_weight_overrides {
            weights.insert(k.clone(), *v);
        }
        td.edge_weights = weights;

        types_map.insert(stem.clone(), Arc::new(td));
    }

    // Schema-level constraint pass — checks that need every type
    // loaded. `enum_from_neighbour.section` names a section on the
    // *reached* entity, whose type this schema cannot pin statically;
    // requiring the key to exist on at least one declared type catches
    // the typo class without over-constraining the endpoint. Skipped
    // when a type file failed to parse — the missing type's sections
    // would make the existence check report noise.
    if !had_type_parse_failure {
        let all_section_keys: HashSet<&str> = types_map
            .values()
            .flat_map(|t| t.sections.iter().map(|s| s.key.as_str()))
            .collect();
        // Deterministic report order: sort type names (types_map is a
        // HashMap); constraints keep declaration order within a type.
        let mut type_names: Vec<&String> = types_map.keys().collect();
        type_names.sort();
        for type_name in type_names {
            let td = &types_map[type_name];
            for c in &td.constraints {
                if let crate::types::ConstraintDef::EnumFromNeighbour { section, .. } = c
                    && !all_section_keys.contains(section.as_str())
                {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "enum_from_neighbour",
                        offender: section.clone(),
                        reason: "`section` names a section key no type of this schema declares"
                            .to_string(),
                    });
                }
            }
            // `must_reach.terminal_types` name types of this schema —
            // checkable only once every type is loaded.
            for ob in &td.must_reach {
                for t in &ob.terminal_types {
                    if !types_map.contains_key(t.as_str()) {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "must_reach",
                            offender: t.clone(),
                            reason: "`terminal_types` entry names no type of this schema"
                                .to_string(),
                        });
                    }
                }
            }
            // A signal's neighbour pair reads the COUNTERPART entity,
            // whose type this schema cannot pin statically: the field
            // must be declared with `enum_values` on at least one type
            // of the schema, and the value must be a member on at
            // least one such declaring type.
            for sig in &td.signals {
                if let (Some(field), Some(value)) = (&sig.neighbour_field, &sig.neighbour_value) {
                    let declaring: Vec<&crate::types::MetadataFieldDef> = types_map
                        .values()
                        .flat_map(|t| t.metadata_fields.iter())
                        .filter(|f| f.key == *field && f.enum_values.is_some())
                        .collect();
                    if declaring.is_empty() {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "signal",
                            offender: field.clone(),
                            reason: "`neighbour_field` is declared with `enum_values` on no \
                                     type of this schema"
                                .to_string(),
                        });
                    } else if !declaring.iter().any(|f| {
                        f.enum_values
                            .as_ref()
                            .is_some_and(|allowed| allowed.contains(value))
                    }) {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "signal",
                            offender: value.clone(),
                            reason: format!(
                                "`neighbour_value` is outside `{field}`'s enum_values on every \
                                 declaring type"
                            ),
                        });
                    }
                }
            }
        }
    }

    // `labelling.support.terminal_types` name types of this schema —
    // checkable only once every type is loaded (mirrors the
    // `must_reach` pass; skipped on type-parse failure like the rest
    // of the schema-level pass).
    if !had_type_parse_failure
        && let Some(lab) = &manifest.relationships.labelling
        && let Some(sup) = &lab.support
    {
        for t in &sup.terminal_types {
            if !types_map.contains_key(t.as_str()) {
                errors.push(SchemaLoadError::InvalidLabelling {
                    offender: t.clone(),
                    reason: "`labelling.support.terminal_types` entry names no type of this \
                             schema"
                        .to_string(),
                });
            }
        }
    }

    if !errors.is_empty() {
        return Err(collapse(errors));
    }

    Ok(Schema {
        manifest,
        version: version.expect("version parse failure would have accumulated an error"),
        types: types_map,
    })
}

fn validate_name(name: &str) -> Result<(), SchemaLoadError> {
    name_shape(name).map_err(|reason| SchemaLoadError::InvalidName {
        value: name.into(),
        reason,
    })
}

/// Author-time access to the schema-name shape rule — the same check
/// the loader runs on a manifest's `name:`. Exposed so scaffolding
/// tooling (`memstead schema new`) can refuse a bad name up front with
/// the loader's own reason string instead of a drifting copy of the
/// grammar.
pub fn validate_schema_name(name: &str) -> Result<(), &'static str> {
    name_shape(name)
}

/// Shared shape rule for schema names — the manifest's own `name:` and
/// every `cross_mem_relationships[].to_schema` follow the same
/// grammar; the two callers wrap violations in their field-specific
/// error variants.
fn name_shape(name: &str) -> Result<(), &'static str> {
    if name.is_empty() {
        return Err("must not be empty");
    }
    let mut chars = name.chars();
    let first = chars.next().unwrap();
    if !first.is_ascii_lowercase() {
        return Err("must start with a lowercase letter");
    }
    for c in chars {
        if !(c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') {
            return Err("must contain only lowercase letters, digits, and hyphens");
        }
    }
    Ok(())
}

/// Validate and compile the section-format declarations of a type
/// (plan 08). LENIENT at load: problems are recorded on the section
/// (`format_problems`) instead of refusing, so a sealed schema
/// carrying a bad declaration keeps loading — install and strict
/// validation refuse via [`check_section_formats`], and a defective
/// declaration is never enforced (`compiled_content` stays `None`). A
/// valid `content` expression is compiled once and cached.
fn compile_section_formats(td: &mut TypeDefinition) {
    use crate::content_expr::ContentExpr;
    for section in &mut td.sections {
        // A non-default severity only deserializes from an explicit
        // declaration, so a lone `format_severity: warn` without
        // `content` is detectable — and would otherwise load and be
        // silently ignored (`format_severity: block` alone equals the
        // default and is inherently a no-op).
        let declares_any = section.content.is_some()
            || section.item_pattern.is_some()
            || section.table.is_some()
            || section.example.is_some()
            || section.format_severity != crate::types::ConstraintSeverity::Block;
        if !declares_any {
            continue;
        }
        let mut problems: Vec<String> = Vec::new();

        let compiled = match &section.content {
            None => {
                problems.push(
                    "`item_pattern` / `table` / `example` require a `content` declaration"
                        .to_string(),
                );
                None
            }
            Some(expr_src) => match ContentExpr::parse(expr_src) {
                Ok(expr) => Some(expr),
                Err(e) => {
                    problems.push(format!("`content` is invalid: {e}"));
                    None
                }
            },
        };

        if let Some(pattern) = &section.item_pattern {
            if let Err(e) = regex::Regex::new(pattern) {
                problems.push(format!("`item_pattern` is not a valid regex: {e}"));
            }
            if let Some(expr) = &compiled {
                let names = expr.mentioned_names();
                let has_list = names.contains(&"list");
                let has_paragraph = names.contains(&"paragraph");
                if has_list == has_paragraph {
                    problems.push(
                        "`item_pattern` requires a `content` expression containing exactly                          one of `list` / `paragraph` (tables use `column_patterns`)"
                            .to_string(),
                    );
                }
            }
        }

        if let Some(table) = &section.table {
            if let Some(expr) = &compiled
                && !expr.mentioned_names().contains(&"table")
            {
                problems.push(
                    "`table` block is only legal when `content` contains `table`".to_string(),
                );
            }
            if table.columns.is_empty() {
                problems.push("`table.columns` must name at least one column".to_string());
            }
            for (column, pattern) in &table.column_patterns {
                if !table.columns.contains(column) {
                    problems.push(format!(
                        "`column_patterns` names '{column}', which is not in `columns`"
                    ));
                }
                if let Err(e) = regex::Regex::new(pattern) {
                    problems.push(format!(
                        "`column_patterns.{column}` is not a valid regex: {e}"
                    ));
                }
            }
        }

        if problems.is_empty() {
            section.compiled_content = compiled;
        } else {
            section.format_problems = problems;
        }
    }
}

/// Refuse a schema whose section-format declarations are defective —
/// the install / strict-validation half of the loader-honesty rule.
/// Same posture as [`check_reserved_metadata_keys`]: install and
/// strict validation call this and refuse (naming EVERY problem of
/// the first defective section); boot and sealed-schema loads must
/// NOT — the recorded `format_problems` surface as health findings
/// instead, and the defective declaration is never enforced.
pub fn check_section_formats(schema: &crate::Schema) -> Result<(), SchemaLoadError> {
    // Aggregate EVERY defective section across every type — the
    // refusal names all offenders, never the first only. `type_name`
    // / `section` carry the first offender; entries beyond it are
    // prefixed with their own type/section inside `problems`.
    let mut first: Option<(String, String)> = None;
    let mut problems: Vec<String> = Vec::new();
    for td in schema.types.values() {
        for section in &td.sections {
            if section.format_problems.is_empty() {
                continue;
            }
            if first.is_none() {
                first = Some((td.name.clone(), section.key.clone()));
                problems.extend(section.format_problems.iter().cloned());
            } else {
                problems.extend(
                    section
                        .format_problems
                        .iter()
                        .map(|p| format!("[{}.{}] {p}", td.name, section.key)),
                );
            }
        }
    }
    match first {
        Some((type_name, section)) => Err(SchemaLoadError::InvalidSectionFormat {
            type_name,
            section,
            problems,
        }),
        None => Ok(()),
    }
}

fn validate_type(
    td: &TypeDefinition,
    rel_names: &HashSet<String>,
    available_rels: &[String],
    errors: &mut Vec<SchemaLoadError>,
) {
    // Reserved-section check. Section key `relationships` collides
    // with the parser's auto-managed `## Relationships` section.
    // Domain conventions (`identity`, `purpose`, ...) are NOT reserved.
    // The reserved-metadata-key check runs earlier in
    // `load_with_context` against the *raw* author-declared field list
    // so the base-metadata merge doesn't false-positive against
    // engine-injected keys.
    for section in &td.sections {
        if reserved_section_keys().contains(&section.key.as_str()) {
            errors.push(SchemaLoadError::ReservedSchemaKey {
                type_name: td.name.clone(),
                kind: "section",
                offending_key: section.key.clone(),
                reserved_keys: reserved_section_keys()
                    .iter()
                    .map(|s| s.to_string())
                    .collect(),
            });
        }
    }

    if let Err(e) = check_rel(
        &td.name,
        "hierarchy_relationship",
        &td.hierarchy_relationship,
        rel_names,
        available_rels,
    ) {
        errors.push(e);
    }
    for r in &td.no_self_loop_relationships {
        if let Err(e) = check_rel(
            &td.name,
            "no_self_loop_relationships",
            r,
            rel_names,
            available_rels,
        ) {
            errors.push(e);
        }
    }
    for r in td.edge_weight_overrides.keys() {
        if let Err(e) = check_rel(
            &td.name,
            "edge_weight_overrides",
            r,
            rel_names,
            available_rels,
        ) {
            errors.push(e);
        }
    }
    for block in &td.required_outgoing {
        for r in &block.relationships {
            if let Err(e) = check_rel(&td.name, "required_outgoing", r, rel_names, available_rels) {
                errors.push(e);
            }
        }
        // Conditional blocks: `when_field` / `when_value` travel as a
        // pair, the field must be a declared metadata field carrying
        // `enum_values`, and the value must be a member. Stricter than
        // `requires_when` (which tolerates non-enum trigger fields):
        // an edge obligation armed by a free-text value would never
        // fire predictably, so the loader refuses instead of loading a
        // dead condition.
        match (&block.when_field, &block.when_value) {
            (None, None) => {}
            (Some(f), None) => {
                errors.push(SchemaLoadError::InvalidConstraint {
                    type_name: td.name.clone(),
                    kind: "required_outgoing",
                    offender: f.clone(),
                    reason: "`when_field` requires `when_value` alongside it".to_string(),
                });
            }
            (None, Some(v)) => {
                errors.push(SchemaLoadError::InvalidConstraint {
                    type_name: td.name.clone(),
                    kind: "required_outgoing",
                    offender: v.clone(),
                    reason: "`when_value` requires `when_field` alongside it".to_string(),
                });
            }
            (Some(f), Some(v)) => match td.metadata_fields.iter().find(|mf| mf.key == *f) {
                None => {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "required_outgoing",
                        offender: f.clone(),
                        reason: "`when_field` names no metadata field of this type".to_string(),
                    });
                }
                Some(when_def) => match &when_def.enum_values {
                    None => {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "required_outgoing",
                            offender: f.clone(),
                            reason: format!(
                                "`when_field` must name a metadata field with `enum_values`; `{f}` declares none"
                            ),
                        });
                    }
                    Some(allowed) if !allowed.contains(v) => {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "required_outgoing",
                            offender: v.clone(),
                            reason: format!(
                                "`when_value` is not in `{f}`'s enum_values [{}]",
                                allowed.join(", ")
                            ),
                        });
                    }
                    Some(_) => {}
                },
            },
        }
    }

    // Reachability obligations. Per-type checks here; `terminal_types`
    // needs every type loaded and is checked in the schema-level pass.
    for ob in &td.must_reach {
        if ob.relationships.is_empty() {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "must_reach",
                offender: "(empty)".to_string(),
                reason: "`relationships` must name at least one relationship".to_string(),
            });
        }
        for r in &ob.relationships {
            if let Err(e) = check_rel(&td.name, "must_reach", r, rel_names, available_rels) {
                errors.push(e);
            }
        }
        if ob.terminal_types.is_empty() {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "must_reach",
                offender: "(empty)".to_string(),
                reason: "`terminal_types` must name at least one type".to_string(),
            });
        }
        if ob.max_depth == Some(0) {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "must_reach",
                offender: "0".to_string(),
                reason: "`max_depth` must be at least 1 — nothing is reachable in zero hops"
                    .to_string(),
            });
        }
        if ob.severity == crate::types::ConstraintSeverity::Block {
            // A transitive property is established by writes on OTHER
            // entities, so a write-time refusal would punish the wrong
            // mutation — refuse the promise rather than load-and-
            // downgrade (same posture as status_propagation).
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "must_reach",
                offender: "block".to_string(),
                reason: "must_reach is always warn-tier — a reachability gap is created by \
                         writes on other entities, so no single write can be refused for it"
                    .to_string(),
            });
        }
    }

    // Aggregate signals. Per-type checks here; the neighbour pair's
    // cross-type validation needs every type loaded and runs in the
    // schema-level pass.
    let mut signal_names_seen: HashSet<&str> = HashSet::new();
    for sig in &td.signals {
        let name_ok = sig
            .name
            .chars()
            .next()
            .is_some_and(|c| c.is_ascii_lowercase())
            && sig
                .name
                .chars()
                .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
        if !name_ok {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "signal",
                offender: sig.name.clone(),
                reason: "`name` must match [a-z][a-z0-9_]*".to_string(),
            });
        }
        if !signal_names_seen.insert(sig.name.as_str()) {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "signal",
                offender: sig.name.clone(),
                reason: "duplicate signal name on this type".to_string(),
            });
        }
        if sig.relationships.is_empty() {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "signal",
                offender: sig.name.clone(),
                reason: "`relationships` must name at least one relationship".to_string(),
            });
        }
        for r in &sig.relationships {
            if let Err(e) = check_rel(&td.name, "signals", r, rel_names, available_rels) {
                errors.push(e);
            }
        }
        if sig.thresholds.is_empty() {
            errors.push(SchemaLoadError::InvalidConstraint {
                type_name: td.name.clone(),
                kind: "signal",
                offender: sig.name.clone(),
                reason: "`thresholds` must declare at least one step".to_string(),
            });
        }
        for pair in sig.thresholds.windows(2) {
            if pair[1].at_least <= pair[0].at_least {
                errors.push(SchemaLoadError::InvalidConstraint {
                    type_name: td.name.clone(),
                    kind: "signal",
                    offender: pair[1].at_least.to_string(),
                    reason: "`thresholds` must have strictly increasing `at_least` values"
                        .to_string(),
                });
            }
        }
        match (&sig.neighbour_field, &sig.neighbour_value) {
            (None, None) | (Some(_), Some(_)) => {}
            (Some(f), None) => {
                errors.push(SchemaLoadError::InvalidConstraint {
                    type_name: td.name.clone(),
                    kind: "signal",
                    offender: f.clone(),
                    reason: "`neighbour_field` requires `neighbour_value` alongside it".to_string(),
                });
            }
            (None, Some(v)) => {
                errors.push(SchemaLoadError::InvalidConstraint {
                    type_name: td.name.clone(),
                    kind: "signal",
                    offender: v.clone(),
                    reason: "`neighbour_value` requires `neighbour_field` alongside it".to_string(),
                });
            }
        }
    }

    // Constraint vocabulary (loader honesty: a malformed declaration
    // refuses with a typed error naming the offender — never
    // load-and-ignore).
    let field_keys: std::collections::HashSet<&str> =
        td.metadata_fields.iter().map(|f| f.key.as_str()).collect();
    let section_keys: std::collections::HashSet<&str> =
        td.sections.iter().map(|sec| sec.key.as_str()).collect();
    for c in &td.constraints {
        match c {
            crate::types::ConstraintDef::RequiresWhen {
                field,
                when_field,
                when_value,
                ..
            } => {
                if !field_keys.contains(field.as_str()) && !section_keys.contains(field.as_str()) {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "requires_when",
                        offender: field.clone(),
                        reason: "`field` names neither a metadata field nor a section of this type"
                            .to_string(),
                    });
                }
                let Some(when_def) = td.metadata_fields.iter().find(|f| f.key == *when_field)
                else {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "requires_when",
                        offender: when_field.clone(),
                        reason: "`when_field` names no metadata field of this type".to_string(),
                    });
                    continue;
                };
                if let Some(allowed) = &when_def.enum_values
                    && !allowed.contains(when_value)
                {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "requires_when",
                        offender: when_value.clone(),
                        reason: format!(
                            "`when_value` is not in `{when_field}`'s enum_values [{}]",
                            allowed.join(", ")
                        ),
                    });
                }
            }
            crate::types::ConstraintDef::Unique { fields, .. } => {
                if fields.is_empty() {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "unique",
                        offender: "(empty)".to_string(),
                        reason: "`fields` must name at least one metadata field".to_string(),
                    });
                }
                for f in fields {
                    if !field_keys.contains(f.as_str()) {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "unique",
                            offender: f.clone(),
                            reason: "`fields` entry names no metadata field of this type"
                                .to_string(),
                        });
                    }
                }
            }
            crate::types::ConstraintDef::EnumFromNeighbour {
                field, rel_type, ..
            } => {
                if !field_keys.contains(field.as_str()) {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "enum_from_neighbour",
                        offender: field.clone(),
                        reason: "`field` names no metadata field of this type".to_string(),
                    });
                }
                if !rel_names.contains(rel_type) {
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "enum_from_neighbour",
                        offender: rel_type.clone(),
                        reason: "`rel_type` is not in the schema's relationship vocabulary"
                            .to_string(),
                    });
                }
                // `section` names a key on the *reached* type, which
                // this per-type pass cannot see — the schema-level
                // pass after all types load checks it.
            }
            crate::types::ConstraintDef::StatusPropagation {
                field,
                value,
                rel_type,
                rel_types,
                severity,
                ..
            } => {
                match td.metadata_fields.iter().find(|f| f.key == *field) {
                    None => {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "status_propagation",
                            offender: field.clone(),
                            reason: "`field` names no metadata field of this type".to_string(),
                        });
                    }
                    Some(field_def) => {
                        if let Some(allowed) = &field_def.enum_values
                            && !allowed.contains(value)
                        {
                            errors.push(SchemaLoadError::InvalidConstraint {
                                type_name: td.name.clone(),
                                kind: "status_propagation",
                                offender: value.clone(),
                                reason: format!(
                                    "`value` is not in `{field}`'s enum_values [{}]",
                                    allowed.join(", ")
                                ),
                            });
                        }
                    }
                }
                // Exactly one of `rel_type` / `rel_types`; every named
                // member must be declared; an empty set is a defect.
                match (rel_type, rel_types) {
                    (Some(_), Some(_)) => {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "status_propagation",
                            offender: "rel_type".to_string(),
                            reason: "declare `rel_type` or `rel_types`, not both".to_string(),
                        });
                    }
                    (None, None) => {
                        errors.push(SchemaLoadError::InvalidConstraint {
                            type_name: td.name.clone(),
                            kind: "status_propagation",
                            offender: "(missing)".to_string(),
                            reason: "one of `rel_type` / `rel_types` is required".to_string(),
                        });
                    }
                    (Some(single), None) => {
                        if !rel_names.contains(single) {
                            errors.push(SchemaLoadError::InvalidConstraint {
                                type_name: td.name.clone(),
                                kind: "status_propagation",
                                offender: single.clone(),
                                reason: "`rel_type` is not in the schema's relationship vocabulary"
                                    .to_string(),
                            });
                        }
                    }
                    (None, Some(set)) => {
                        if set.is_empty() {
                            errors.push(SchemaLoadError::InvalidConstraint {
                                type_name: td.name.clone(),
                                kind: "status_propagation",
                                offender: "(empty)".to_string(),
                                reason: "`rel_types` must name at least one relationship"
                                    .to_string(),
                            });
                        }
                        for name in set {
                            if !rel_names.contains(name) {
                                errors.push(SchemaLoadError::InvalidConstraint {
                                    type_name: td.name.clone(),
                                    kind: "status_propagation",
                                    offender: name.clone(),
                                    reason: "`rel_types` entry is not in the schema's \
                                             relationship vocabulary"
                                        .to_string(),
                                });
                            }
                        }
                    }
                }
                if *severity == crate::types::ConstraintSeverity::Block {
                    // Propagation can never refuse a write (the taint
                    // arises from the ancestor's later change), so a
                    // `block` declaration would be a promise the
                    // engine will not keep — refuse it rather than
                    // load-and-downgrade.
                    errors.push(SchemaLoadError::InvalidConstraint {
                        type_name: td.name.clone(),
                        kind: "status_propagation",
                        offender: "block".to_string(),
                        reason: "status_propagation is always warn-tier — a parent falling after \
                                 the child was written cannot retroactively make the child's \
                                 write illegal"
                            .to_string(),
                    });
                }
            }
        }
    }

    // Due axis (first-author-path plan 08): the declaration's
    // references must exist on this type in the declared shapes.
    if let Some(due) = &td.due {
        match td.metadata_fields.iter().find(|f| f.key == due.date_field) {
            None => errors.push(SchemaLoadError::InvalidDueAxis {
                type_name: td.name.clone(),
                offender: due.date_field.clone(),
                reason: "`date_field` names no metadata field of this type".to_string(),
            }),
            Some(f) if f.field_type != crate::types::FieldType::Date => {
                errors.push(SchemaLoadError::InvalidDueAxis {
                    type_name: td.name.clone(),
                    offender: due.date_field.clone(),
                    reason: "`date_field` must name a date-typed metadata field".to_string(),
                })
            }
            Some(_) => {}
        }
        match td
            .metadata_fields
            .iter()
            .find(|f| f.key == due.status_field)
        {
            None => errors.push(SchemaLoadError::InvalidDueAxis {
                type_name: td.name.clone(),
                offender: due.status_field.clone(),
                reason: "`status_field` names no metadata field of this type".to_string(),
            }),
            Some(f) => match &f.enum_values {
                None => errors.push(SchemaLoadError::InvalidDueAxis {
                    type_name: td.name.clone(),
                    offender: due.status_field.clone(),
                    reason: "`status_field` must name an enum-typed metadata field \
                             (declare enum_values)"
                        .to_string(),
                }),
                Some(allowed) => {
                    for v in &due.open_values {
                        if !allowed.contains(v) {
                            errors.push(SchemaLoadError::InvalidDueAxis {
                                type_name: td.name.clone(),
                                offender: v.clone(),
                                reason: format!(
                                    "`open_values` entry is not in `{}`'s enum_values [{}]",
                                    due.status_field,
                                    allowed.join(", ")
                                ),
                            });
                        }
                    }
                }
            },
        }
        if due.open_values.is_empty() {
            errors.push(SchemaLoadError::InvalidDueAxis {
                type_name: td.name.clone(),
                offender: "(empty)".to_string(),
                reason: "`open_values` must name at least one open status value".to_string(),
            });
        }
        if let Some(lead) = &due.lead_section
            && !td.sections.iter().any(|s| s.key == *lead)
        {
            errors.push(SchemaLoadError::InvalidDueAxis {
                type_name: td.name.clone(),
                offender: lead.clone(),
                reason: "`lead_section` names no section of this type".to_string(),
            });
        }
    }

    // Exactly one catch_all section
    let catch_all_count = td.sections.iter().filter(|s| s.catch_all).count();
    if catch_all_count != 1 {
        errors.push(SchemaLoadError::CatchAllViolation {
            type_name: td.name.clone(),
            count: catch_all_count,
        });
    }

    // Field-reference integrity
    let section_keys: HashSet<&str> = td.sections.iter().map(|s| s.key.as_str()).collect();
    let meta_keys: HashSet<&str> = td.metadata_fields.iter().map(|m| m.key.as_str()).collect();

    for f in &td.text_fields {
        // text_fields point at section content — not metadata.
        if !section_keys.contains(f.as_str()) {
            errors.push(SchemaLoadError::UnknownFieldReference {
                type_name: td.name.clone(),
                field: "text_fields",
                reference: f.clone(),
            });
        }
    }
    for f in &td.health_required_fields {
        if !section_keys.contains(f.as_str()) && !meta_keys.contains(f.as_str()) {
            errors.push(SchemaLoadError::UnknownFieldReference {
                type_name: td.name.clone(),
                field: "health_required_fields",
                reference: f.clone(),
            });
        }
    }
    for f in &td.updatable_fields {
        // `title` is the entity's filename-derived title — always updatable.
        if f == "title" {
            continue;
        }
        if !section_keys.contains(f.as_str()) && !meta_keys.contains(f.as_str()) {
            errors.push(SchemaLoadError::UnknownFieldReference {
                type_name: td.name.clone(),
                field: "updatable_fields",
                reference: f.clone(),
            });
        }
    }

    // Metadata default_value must be a member of enum_values when both present.
    for m in &td.metadata_fields {
        if let (Some(default), Some(allowed)) = (m.default_value.as_ref(), m.enum_values.as_ref())
            && !allowed.contains(default)
        {
            errors.push(SchemaLoadError::DefaultValueNotInEnum {
                type_name: td.name.clone(),
                field: m.key.clone(),
                default: default.clone(),
                allowed: allowed.clone(),
            });
        }
    }
}

fn check_rel(
    type_name: &str,
    field: &'static str,
    relationship: &str,
    rel_names: &HashSet<String>,
    available: &[String],
) -> Result<(), SchemaLoadError> {
    if rel_names.contains(relationship) {
        return Ok(());
    }
    Err(SchemaLoadError::UndeclaredRelationship {
        type_name: type_name.into(),
        field,
        relationship: relationship.into(),
        available: available.to_vec(),
    })
}