mig-bo4e 0.14.0

//! PID field requirements — built from PID schema JSON + TOML mappings.
//!
//! Used by `validate_pid()` to check whether a populated BO4E struct
//! has all required fields for a given PID.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::definition::{FieldMapping, MappingDefinition};
use crate::error::MappingError;
use crate::pid_schema_index::PidSchemaIndex;
use crate::MappingEngine;

/// Cardinality values at or above this threshold are treated as "no practical limit"
/// and serialized as `Cardinality::max = None` (unbounded).
pub const UNCAPPED_SENTINEL: u32 = 9999;

/// Cardinality bound for a typed reference. `max == None` means unbounded.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub struct Cardinality {
    pub min: u32,
    pub max: Option<u32>,
}

impl Cardinality {
    pub const REQUIRED: Self = Self {
        min: 1,
        max: Some(1),
    };
    pub const OPTIONAL: Self = Self {
        min: 0,
        max: Some(1),
    };
    pub const LIST: Self = Self { min: 0, max: None };
    pub const NON_EMPTY: Self = Self { min: 1, max: None };

    pub fn unbounded(self) -> bool {
        self.max.is_none()
    }
    pub fn max_or(self, fallback: u32) -> u32 {
        self.max.unwrap_or(fallback)
    }
    /// True when this cardinality permits more than one element.
    pub fn is_list(self) -> bool {
        self.max.map(|m| m > 1).unwrap_or(true)
    }
}

/// Primitive BO4E field types as recognized by the catalog and PID requirements.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum Bo4ePrimitive {
    String,
    Integer,
    Decimal,
    Boolean,
    Date,
    DateTime,
    Null,
}

/// Type description of a BO4E reference, with embedded cardinality.
///
/// This is the single home of cardinality across both [`EntityRequirement`] and
/// [`FieldRequirement`]. The `Object` variant is also used at entity level —
/// the entity is conceptually a reference from the transaction root to a BO4E type.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Bo4eRefType {
    Primitive {
        primitive: Bo4ePrimitive,
        cardinality: Cardinality,
    },
    Enum {
        type_name: String,
        cardinality: Cardinality,
    },
    Object {
        type_name: String,
        cardinality: Cardinality,
    },
    Unknown,
}

impl Bo4eRefType {
    pub fn cardinality(&self) -> Option<Cardinality> {
        match self {
            Self::Primitive { cardinality, .. }
            | Self::Enum { cardinality, .. }
            | Self::Object { cardinality, .. } => Some(*cardinality),
            Self::Unknown => None,
        }
    }
    pub fn object_type_name(&self) -> Option<&str> {
        if let Self::Object { type_name, .. } = self {
            Some(type_name)
        } else {
            None
        }
    }
    pub fn enum_type_name(&self) -> Option<&str> {
        if let Self::Enum { type_name, .. } = self {
            Some(type_name)
        } else {
            None
        }
    }
}

/// Build a `Cardinality` from raw min/max where values >= [`UNCAPPED_SENTINEL`]
/// are treated as unbounded.
pub(crate) fn cardinality_from_reps(min: u32, max: u32) -> Cardinality {
    Cardinality {
        min,
        max: if max >= UNCAPPED_SENTINEL {
            None
        } else {
            Some(max)
        },
    }
}

/// Complete field requirements for a single PID.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PidRequirements {
    pub pid: String,
    pub beschreibung: String,
    pub entities: Vec<EntityRequirement>,
}

/// The scope of an entity within a PID interchange.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum EntityScope {
    /// Message-level entity (e.g., SG2, SG2.SG3 — Marktteilnehmer, Kontakt).
    /// Present once per message, outside transactions.
    Message,
    /// Transaction-level entity (e.g., SG4.*, SG4.SG8.SG10 — Prozessdaten, Marktlokation).
    /// Present within each transaction.
    Transaction,
}

/// Requirements for one BO4E entity within a PID.
///
/// The entity's BO4E type and cardinality both live in [`Bo4eRefType::Object`] on
/// `ref_type`. Cardinality is never duplicated outside `ref_type`.
///
/// `Deserialize` is implemented by hand for backwards compatibility: legacy
/// caches that pre-date the cardinality unification have `bo4e_type`/`min_reps`/
/// `max_reps` fields instead of `ref_type`. Such caches are still loadable —
/// the legacy fields are folded into a synthetic `Bo4eRefType::Object`.
#[derive(Debug, Clone, Serialize)]
pub struct EntityRequirement {
    pub entity: String,
    pub ref_type: Bo4eRefType,
    pub ahb_status: String,
    pub fields: Vec<FieldRequirement>,
    /// If this entity is eligible for map-keying (BTreeMap instead of Vec),
    /// the key field name and its allowed values.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub map_key: Option<EntityMapKeyInfo>,
    /// Whether this entity is at message level or transaction level.
    #[serde(default = "default_scope")]
    pub scope: EntityScope,
    /// Per-variant field requirements for entities fed by several AHB segment-group
    /// variants through one mapping (e.g. `Geschaeftspartner` from `sg12_z03`,
    /// `sg12_z07`, … via a single `source_path = "sg4.sg12"` TOML).
    ///
    /// `fields` holds the union over all variants. Each variant here holds the
    /// requirements of exactly one discriminator code, so a validator can pick the
    /// requirements matching an element's qualifier instead of applying every
    /// variant's required fields to every element. See
    /// [`crate::pid_validation::effective_field_requirements`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub variants: Vec<EntityVariantRequirement>,
    /// The segment groups feeding this entity, when there is more than one
    /// (e.g. `Marktlokation` from SG5 LOC+Z16 and SG5 LOC+Z22).
    ///
    /// The entity being present then says nothing about any one of its groups,
    /// so a group none of whose fields is filled is judged by its own AHB
    /// status: its fields are demanded only if the group is required. Empty for
    /// an entity fed by a single group, whose presence is the group's. See
    /// [`crate::pid_validation::absent_groups`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub groups: Vec<EntityGroupRequirement>,
}

/// One segment group feeding a multi-group entity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityGroupRequirement {
    /// The definitions' `source_path` (`sg4.sg5_z22`), or `source_group` when
    /// they have none.
    pub source_path: String,
    /// The group's own AHB status (`Soll [2003]`).
    pub ahb_status: String,
    /// Every BO4E field the group's definitions write. A field written by
    /// several groups is listed in each.
    pub fields: Vec<String>,
}

/// Field requirements of one AHB segment-group variant of an entity, selected by
/// the value of a discriminating BO4E field.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityVariantRequirement {
    /// BO4E field carrying the group's qualifier (e.g. `"partnerrolle"`).
    pub discriminator_field: String,
    /// Raw EDIFACT qualifier code selecting this variant (e.g. `"Z07"`).
    pub code: String,
    /// The BO4E value the mapping's `enum_map` translates `code` to
    /// (e.g. `"kundeMsb"`), if any.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bo4e_value: Option<String>,
    /// Schema group paths contributing to this variant (e.g. `["sg4.sg12_z07"]`).
    #[serde(default)]
    pub source_paths: Vec<String>,
    /// Requirements for the fields this variant's schema group actually contains.
    pub fields: Vec<FieldRequirement>,
}

impl<'de> serde::Deserialize<'de> for EntityRequirement {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        // Helper that accepts BOTH the new shape (ref_type) and the legacy shape
        // (bo4e_type + min_reps + max_reps). Either may be present; if both are
        // present the explicit ref_type wins.
        #[derive(Deserialize)]
        #[serde(rename_all = "snake_case")]
        struct Helper {
            entity: String,
            #[serde(default)]
            ref_type: Option<Bo4eRefType>,
            // Legacy fields (pre-cardinality-unification caches):
            #[serde(default)]
            bo4e_type: Option<String>,
            #[serde(default)]
            min_reps: Option<u32>,
            #[serde(default)]
            max_reps: Option<u32>,
            #[serde(default)]
            ahb_status: String,
            #[serde(default)]
            fields: Vec<FieldRequirement>,
            #[serde(default)]
            map_key: Option<EntityMapKeyInfo>,
            #[serde(default = "default_scope")]
            scope: EntityScope,
            #[serde(default)]
            variants: Vec<EntityVariantRequirement>,
            #[serde(default)]
            groups: Vec<EntityGroupRequirement>,
        }
        let h = Helper::deserialize(d)?;
        let ref_type = h.ref_type.unwrap_or_else(|| {
            let min = h.min_reps.unwrap_or(0);
            let max = h.max_reps.unwrap_or(1);
            Bo4eRefType::Object {
                type_name: h.bo4e_type.unwrap_or_default(),
                cardinality: cardinality_from_reps(min, max),
            }
        });
        Ok(EntityRequirement {
            entity: h.entity,
            ref_type,
            ahb_status: h.ahb_status,
            fields: h.fields,
            map_key: h.map_key,
            scope: h.scope,
            variants: h.variants,
            groups: h.groups,
        })
    }
}

impl EntityRequirement {
    /// Convenience accessor: BO4E type name for the entity (always Object variant in practice).
    pub fn bo4e_type(&self) -> &str {
        self.ref_type.object_type_name().unwrap_or("")
    }
    /// Convenience accessor: cardinality of the entity at the transaction root.
    pub fn cardinality(&self) -> Cardinality {
        self.ref_type.cardinality().unwrap_or(Cardinality::REQUIRED)
    }
}

fn default_scope() -> EntityScope {
    EntityScope::Transaction
}

/// Map-key metadata for a discriminated entity.
///
/// When an array entity has a well-known discriminator field with a fixed set
/// of values, it can be represented as a `BTreeMap<K, T>` instead of `Vec<T>`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityMapKeyInfo {
    /// The BO4E field name that acts as the key (e.g., "marktrolle").
    pub field: String,
    /// The allowed key values with their codes and descriptions.
    pub values: Vec<EntityMapKeyValue>,
}

/// A single allowed value for a map key.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EntityMapKeyValue {
    pub code: String,
    pub name: String,
}

/// Requirements for a single field within an entity.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldRequirement {
    pub bo4e_name: String,
    pub ahb_status: String,
    pub field_type: String,
    pub format: Option<String>,
    pub enum_name: Option<String>,
    pub valid_codes: Vec<CodeValue>,
    /// For fields from nested child groups (e.g., SG10 under SG8):
    /// tracks the child group name and its max_reps.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub child_group: Option<ChildGroupInfo>,
    /// BO4E-side type description with cardinality. Defaults to `Unknown` —
    /// PID-level field classification is intentionally minimal here; the
    /// authoritative typed catalog of BO4E fields lives in `bo4e_catalog`.
    #[serde(default = "default_unknown_ref")]
    pub ref_type: Bo4eRefType,
}

fn default_unknown_ref() -> Bo4eRefType {
    Bo4eRefType::Unknown
}

/// Metadata about a nested child group that a field originates from.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChildGroupInfo {
    /// Source group suffix, e.g., "sg10" from "SG4.SG8.SG10".
    pub name: String,
    /// max_reps of the child group from MIG.
    pub max_reps: i32,
}

/// A valid code value with its human-readable meaning.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CodeValue {
    pub code: String,
    pub meaning: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub enum_name: Option<String>,
    /// The BO4E value the mapping's `enum_map` translates `code` to (e.g. `Z07` ->
    /// `"kundeMsb"`). BO4E JSON carries this value rather than the raw code, and
    /// `to_edifact` accepts either, so validation accepts either as well.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bo4e_value: Option<String>,
}

// ── Loading ────────────────────────────────────────────────────────────────

/// Load all TOML mapping definitions for a PID from the standard directory layout.
///
/// Pattern:
/// - `message_dir` — message-level definitions (unfiltered, always included)
/// - `common_dir` — common/shared definitions, filtered by PID schema
/// - `pid_dir` — PID-specific definitions (override common where matching)
pub fn load_definitions_for_pid(
    common_dir: &Path,
    pid_dir: &Path,
    message_dir: &Path,
    schema: &Value,
) -> Result<Vec<MappingDefinition>, MappingError> {
    let mut all_defs = Vec::new();

    // 1. Message-level definitions (always included, no schema filtering)
    if message_dir.exists() {
        let msg_engine = MappingEngine::load(message_dir)?;
        all_defs.extend(msg_engine.definitions().to_vec());
    }

    // 2. Common definitions filtered by schema + PID overrides
    let schema_index = PidSchemaIndex::from_json(schema);
    if common_dir.exists() && pid_dir.exists() {
        let tx_engine = MappingEngine::load_with_common(common_dir, pid_dir, &schema_index)?;
        all_defs.extend(tx_engine.definitions().to_vec());
    } else if common_dir.exists() {
        let common_engine = MappingEngine::load_common_only(common_dir, &schema_index)?;
        all_defs.extend(common_engine.definitions().to_vec());
    } else if pid_dir.exists() {
        let pid_engine = MappingEngine::load(pid_dir)?;
        all_defs.extend(pid_engine.definitions().to_vec());
    }

    Ok(all_defs)
}

// ── Building PidRequirements ───────────────────────────────────────────────

/// Intermediate builder for accumulating fields per entity.
struct EntityBuilder {
    bo4e_type: String,
    ahb_status: String,
    min_reps: u32,
    max_reps: u32,
    fields: BTreeMap<String, FieldRequirement>,
    /// Accumulated map_key info (set once, from source_group + schema detection).
    map_key: Option<EntityMapKeyInfo>,
    /// Source groups that contributed to this entity (for map_key detection).
    source_groups: BTreeSet<String>,
    /// Whether this entity is message-level or transaction-level.
    scope: EntityScope,
    /// Per-variant field requirements, keyed by (discriminator field, code).
    variants: BTreeMap<(String, String), VariantBuilder>,
    /// Contributing groups, keyed by source path: (AHB status, written fields).
    groups: BTreeMap<String, (String, BTreeSet<String>)>,
}

/// Intermediate builder for one [`EntityVariantRequirement`].
#[derive(Default)]
struct VariantBuilder {
    bo4e_value: Option<String>,
    source_paths: BTreeSet<String>,
    fields: BTreeMap<String, FieldRequirement>,
}

impl PidRequirements {
    /// Build requirements by cross-referencing PID schema JSON with TOML definitions.
    ///
    /// For each definition, resolves its `source_path` against the schema to find the
    /// group's segments, then matches each TOML field path to a schema element to
    /// extract AHB status, valid codes, format, and enum name.
    pub fn from_schema_and_definitions(schema: &Value, definitions: &[MappingDefinition]) -> Self {
        Self::from_schema_definitions_and_code_lists(
            schema,
            definitions,
            &crate::code_lists::CodeLists::default(),
        )
    }

    /// As [`Self::from_schema_and_definitions`], with the shared tables a
    /// definition's `code_list` names. Validation compares a BO4E value against
    /// the raw EDIFACT code list, so without them a translated value reads as
    /// an invalid code.
    pub fn from_schema_definitions_and_code_lists(
        schema: &Value,
        definitions: &[MappingDefinition],
        code_lists: &crate::code_lists::CodeLists,
    ) -> Self {
        let pid = schema
            .get("pid")
            .and_then(|v| {
                v.as_u64()
                    .map(|n| n.to_string())
                    .or_else(|| v.as_str().map(|s| s.to_string()))
            })
            .unwrap_or_default();

        let beschreibung = schema
            .get("beschreibung")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        // Build a synthetic group JSON from root_segments for message-level defs
        let root_group_json = schema
            .get("root_segments")
            .map(|rs| serde_json::json!({ "segments": rs }));

        let mut entity_map: BTreeMap<String, EntityBuilder> = BTreeMap::new();

        for def in definitions {
            // `parent_field` definitions (e.g. SG12.SG13 → Geschaeftspartner.kontaktwege)
            // map a nested list inside each parent object. Their per-element fields
            // must not become flat requirements of the parent entity, where every
            // partner without a contact would be reported as missing them.
            if def.meta.parent_field.is_some() {
                continue;
            }
            let source_path = def.meta.source_path.as_deref().unwrap_or("");

            let group_json_owned: Value;
            // Set when the definition's group was assembled by merging several
            // schema group variants (see `resolve_schema_group_fuzzy`).
            let mut fuzzy_resolved = false;
            let group_json: &Value = if source_path.is_empty() {
                // No source_path — try deriving from source_group, else use root_segments
                if let Some(g) = resolve_schema_group_fuzzy(schema, &def.meta.source_group, None) {
                    group_json_owned = g;
                    fuzzy_resolved = true;
                    &group_json_owned
                } else {
                    match &root_group_json {
                        Some(g) => g,
                        None => continue,
                    }
                }
            } else {
                match resolve_schema_group(schema, source_path) {
                    Some(g) => g,
                    None => {
                        // Try fuzzy fallback with source_group
                        if let Some(g) =
                            resolve_schema_group_fuzzy(schema, &def.meta.source_group, None)
                        {
                            group_json_owned = g;
                            fuzzy_resolved = true;
                            &group_json_owned
                        } else {
                            continue;
                        }
                    }
                }
            };

            // Skip definitions whose discriminator doesn't match the schema group.
            // This prevents common/ TOMLs (e.g., RFF+TN → transaktionsnummer) from
            // being resolved against an incompatible group (e.g., RFF+Z13 in PID 55001).
            if let Some(ref disc) = def.meta.discriminator {
                if !discriminator_matches_schema_group(disc, group_json) {
                    continue;
                }
            }

            // Get the parent group's AHB status as fallback.
            // Root-level entities (empty source_group) map message-header segments
            // (BGM, DTM) that are always mandatory per EDIFACT/AHB rules.  The PID
            // schema's root_segments don't carry ahb_status because the AHB annotates
            // them at the segment level (e.g., S_BGM AHB_Status="Muss"), which isn't
            // propagated into the schema JSON.  Treat them as "Muss" so validation
            // catches a missing Nachricht entity.
            let parent_ahb = if def.meta.source_group.is_empty() {
                "Muss"
            } else {
                group_json
                    .get("ahb_status")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
            };

            // Derive cardinality from the DEPTH-1 group's max_reps only.
            // For "SG4.SG8.SG10": depth-1 is "SG8" (parts[1]).
            // For "SG4.SG6": depth-1 is "SG6" (parts[1]).
            // For "SG4": depth-1 is "SG4" itself (parts[0]).
            // Child groups (SG10 under SG8, SG6 under SG4) should not independently
            // affect entity cardinality — their repetitions are handled by child_group.
            let source_parts: Vec<&str> = def.meta.source_group.split('.').collect();

            let raw_max_reps = group_json
                .get("max_reps")
                .and_then(|v| v.as_u64())
                .unwrap_or(1) as u32;

            let depth = source_parts.len();
            let is_message_level = depth == 1 && source_parts[0] != "SG4";
            let contributes_cardinality = is_message_level || depth == 2;

            let builder = entity_map
                .entry(def.meta.entity.clone())
                .or_insert_with(|| {
                    // Determine scope: SG4-prefixed groups are transaction-level,
                    // everything else (SG2, SG2.SG3, root) is message-level.
                    let scope = if source_parts.first().map(|s| s.to_uppercase())
                        == Some("SG4".to_string())
                    {
                        EntityScope::Transaction
                    } else {
                        EntityScope::Message
                    };
                    EntityBuilder {
                        bo4e_type: def.meta.bo4e_type.clone(),
                        ahb_status: parent_ahb.to_string(),
                        min_reps: if parent_ahb.trim() == "X" { 1 } else { 0 },
                        max_reps: if contributes_cardinality {
                            raw_max_reps
                        } else {
                            1
                        },
                        fields: BTreeMap::new(),
                        map_key: None,
                        source_groups: BTreeSet::new(),
                        scope,
                        variants: BTreeMap::new(),
                        groups: BTreeMap::new(),
                    }
                });
            builder.source_groups.insert(def.meta.source_group.clone());

            // Allow later definitions at the same depth to increase max_reps
            // max_reps is updated across all contributing definitions (take the max)
            if contributes_cardinality {
                builder.max_reps = builder.max_reps.max(raw_max_reps);
            }
            // min_reps is set only at first insertion (in or_insert_with above).
            // Entities rarely change AHB status across source groups, so the first
            // group's parent_ahb is used as a conservative estimate.

            // Update ahb_status if we get a non-empty one
            if !parent_ahb.is_empty() && builder.ahb_status.is_empty() {
                builder.ahb_status = parent_ahb.to_string();
            }

            // Detect child group info for nested definitions (SG4.SG8.SG10 → 3 parts)
            let child_group_info = if source_parts.len() >= 3 {
                let child_name = source_parts.last().unwrap().to_lowercase();
                Some(ChildGroupInfo {
                    name: child_name,
                    max_reps: raw_max_reps.min(i32::MAX as u32) as i32,
                })
            } else {
                None
            };

            let fields_before: std::collections::BTreeSet<String> =
                builder.fields.keys().cloned().collect();

            let group_key = if source_path.is_empty() {
                def.meta.source_group.clone()
            } else {
                source_path.to_string()
            };
            let (group_status, group_fields) = builder
                .groups
                .entry(group_key)
                .or_insert_with(|| (parent_ahb.to_string(), BTreeSet::new()));
            if group_status.is_empty() {
                *group_status = parent_ahb.to_string();
            }
            group_fields.extend(
                def.fields
                    .values()
                    .filter_map(extract_target_name)
                    .filter(|name| !name.is_empty()),
            );

            // Fields resolve against the variants the discriminator names; the
            // entity and group statuses above keep the merged group's. A PID
            // with two transaction kinds (55065: `sg4_24`, `sg4_z01`) has
            // SEQ+Z22 only under one of them, and this validator does not know
            // which kind a transaction is — the narrowed variant's plain `Muss`
            // would demand the entity in both.
            let field_group_owned = if fuzzy_resolved {
                resolve_schema_group_fuzzy(
                    schema,
                    &def.meta.source_group,
                    def.meta.discriminator.as_deref(),
                )
            } else {
                None
            };
            let field_group = field_group_owned.as_ref().unwrap_or(group_json);

            // Process [fields]
            process_field_section(
                &def.fields,
                field_group,
                parent_ahb,
                &mut builder.fields,
                true,
                code_lists,
                def.meta.discriminator.as_deref(),
            );

            // Set child_group on all newly-added fields from this definition
            if let Some(ref cg) = child_group_info {
                for (key, field) in builder.fields.iter_mut() {
                    if !fields_before.contains(key) && field.child_group.is_none() {
                        field.child_group = Some(cg.clone());
                    }
                }
            }

            // A definition spanning several discriminated group variants (e.g. one
            // SG12 TOML for NAD+Z03/Z05/Z07/Z08) additionally gets per-variant
            // requirements, so each element is validated against its own variant.
            if fuzzy_resolved {
                add_variant_requirements(
                    schema,
                    def,
                    child_group_info.as_ref(),
                    &mut builder.variants,
                    code_lists,
                );
            }
        }

        // Detect map_key for entities that are candidates for BTreeMap representation.
        // This post-processes the entity builders using schema group data.
        // Note: we check ALL entities, not just those with max_reps>1, because some schemas
        // have max_reps=1 for SG2 even though it contains MS+MR (2 reps). When
        // map_key detection succeeds, we ensure max_reps reflects this.
        for (entity_name, builder) in entity_map.iter_mut() {
            if builder.map_key.is_some() {
                continue;
            }
            let detected = detect_entity_map_key(entity_name, &builder.source_groups, schema);
            if let Some(mk) = detected {
                builder.map_key = Some(mk);
                // map-keyed entities are always arrays. If MIG did not already produce
                // max_reps > 1 (e.g. merged_variant_count covers it), set a sentinel of 2
                // so callers see max_reps > 1. The real cardinality is unknown here.
                if builder.max_reps <= 1 {
                    builder.max_reps = 2; // sentinel: real count unknown, implies > 1
                }
            }
        }

        // Convert builders to EntityRequirements. Cardinality lives only inside
        // ref_type now — entity-level min_reps/max_reps are folded into the
        // Object variant's cardinality field.
        let entities: Vec<EntityRequirement> = entity_map
            .into_iter()
            .map(|(name, builder)| {
                let cardinality = cardinality_from_reps(builder.min_reps, builder.max_reps);
                EntityRequirement {
                    entity: name,
                    ref_type: Bo4eRefType::Object {
                        type_name: builder.bo4e_type,
                        cardinality,
                    },
                    ahb_status: builder.ahb_status,
                    fields: builder.fields.into_values().collect(),
                    map_key: builder.map_key,
                    scope: builder.scope,
                    variants: builder
                        .variants
                        .into_iter()
                        .map(|((field, code), vb)| EntityVariantRequirement {
                            discriminator_field: field,
                            code,
                            bo4e_value: vb.bo4e_value,
                            source_paths: vb.source_paths.into_iter().collect(),
                            fields: vb.fields.into_values().collect(),
                        })
                        .collect(),
                    groups: if builder.groups.len() > 1 {
                        builder
                            .groups
                            .into_iter()
                            .map(
                                |(source_path, (ahb_status, fields))| EntityGroupRequirement {
                                    source_path,
                                    ahb_status,
                                    fields: fields.into_iter().collect(),
                                },
                            )
                            .collect()
                    } else {
                        Vec::new()
                    },
                }
            })
            .collect();

        PidRequirements {
            pid,
            beschreibung,
            entities,
        }
    }
}

// ── Map-Key Detection ─────────────────────────────────────────────────────

/// Known entity→key field mappings for map-key detection.
///
/// These are hardcoded because the relationship between entity names and their
/// discriminator field names is a domain convention, not derivable from schema.
const MAP_KEY_RULES: &[(&str, &str, &str)] = &[
    // (entity_name, source_group_pattern, key_field)
    ("Marktteilnehmer", "SG2", "marktrolle"),
    ("Geschaeftspartner", "SG12", "nad_qualifier"),
    ("Kontakt", "SG3", "ctaFunctionCode"),
];

/// Detect map_key for an entity based on its source_groups and the PID schema.
///
/// For single-group patterns (e.g., SG2 with both MS/MR inside one node),
/// inspects the group's discriminator codes directly.
///
/// For split-child patterns (e.g., SG12 with separate sg12_z63, sg12_z65 nodes),
/// aggregates discriminator codes from all sibling child groups.
fn detect_entity_map_key(
    entity_name: &str,
    source_groups: &BTreeSet<String>,
    schema: &Value,
) -> Option<EntityMapKeyInfo> {
    // Find a matching rule
    let (_, sg_pattern, key_field) = MAP_KEY_RULES.iter().find(|(ent, sg_pat, _)| {
        *ent == entity_name
            && source_groups
                .iter()
                .any(|sg| sg == *sg_pat || sg.contains(sg_pat))
    })?;

    let fields = schema.get("fields")?.as_object()?;
    let sg_lower = sg_pattern.to_lowercase();

    // Strategy 1: Single-group detection (e.g., SG2 with both MS and MR)
    if let Some(group_json) = fields.get(&sg_lower) {
        if let Some(info) = detect_map_key_from_group(group_json, key_field) {
            return Some(info);
        }
    }

    // Strategy 2: Aggregate from split-child groups (e.g., sg12_z63, sg12_z65, ...)
    // For "SG12" in source_group "SG4.SG12", look in fields.sg4.children.sg12_*
    // For "SG2" at top level, look in fields.sg2_* siblings
    let info = aggregate_split_child_codes(fields, &sg_lower, key_field);
    if info.is_some() {
        return info;
    }

    // Strategy 3: For nested groups like SG4.SG12, look inside sg4's children
    for sg in source_groups {
        let parts: Vec<&str> = sg.split('.').collect();
        if parts.len() >= 2 {
            let parent_lower = parts[0].to_lowercase();
            let child_prefix = parts[1].to_lowercase();
            if child_prefix.starts_with(&sg_lower) || sg_lower.starts_with(&child_prefix) {
                // Look in parent group's children
                if let Some(parent) = find_group_in_fields(fields, &parent_lower) {
                    if let Some(children) = parent.get("children").and_then(|c| c.as_object()) {
                        let info = aggregate_codes_from_children(children, &sg_lower, key_field);
                        if info.is_some() {
                            return info;
                        }
                    }
                }
            }
        }
    }

    None
}

/// Find a group in the fields object, supporting both exact match and prefix variants.
fn find_group_in_fields<'a>(
    fields: &'a serde_json::Map<String, Value>,
    group_lower: &str,
) -> Option<&'a Value> {
    if let Some(v) = fields.get(group_lower) {
        return Some(v);
    }
    // Check for prefixed variants (sg4_xxx, etc.)
    let prefix = format!("{group_lower}_");
    for (k, v) in fields {
        if k.starts_with(&prefix) {
            return Some(v);
        }
    }
    None
}

/// Detect map_key from a single group's discriminator (works when >=2 codes in one node).
fn detect_map_key_from_group(group_json: &Value, key_field: &str) -> Option<EntityMapKeyInfo> {
    let disc = group_json.get("discriminator")?;
    let disc_element = disc.get("element")?.as_str()?;
    let disc_segment = disc.get("segment")?.as_str()?;

    let mut values: Vec<EntityMapKeyValue> = Vec::new();
    let mut seen_codes: BTreeSet<String> = BTreeSet::new();

    if let Some(segments) = group_json.get("segments").and_then(|s| s.as_array()) {
        for seg in segments {
            let seg_id = seg.get("id").and_then(|v| v.as_str()).unwrap_or_default();
            if seg_id != disc_segment {
                continue;
            }
            if let Some(elements) = seg.get("elements").and_then(|e| e.as_array()) {
                for el in elements {
                    let el_id = el.get("id").and_then(|v| v.as_str()).unwrap_or_default();
                    if el_id != disc_element {
                        continue;
                    }
                    collect_codes_from_element(el, &mut seen_codes, &mut values);
                }
            }
        }
    }

    // Also check discriminator.values fallback
    if let Some(disc_values) = disc.get("values").and_then(|v| v.as_array()) {
        for val in disc_values {
            if let Some(code) = val.as_str() {
                if seen_codes.insert(code.to_string()) {
                    values.push(EntityMapKeyValue {
                        code: code.to_string(),
                        name: String::new(),
                    });
                }
            }
        }
    }

    if values.len() >= 2 {
        Some(EntityMapKeyInfo {
            field: key_field.to_string(),
            values,
        })
    } else {
        None
    }
}

/// Collect code values from an element's codes array and nested components.
fn collect_codes_from_element(
    el: &Value,
    seen_codes: &mut BTreeSet<String>,
    values: &mut Vec<EntityMapKeyValue>,
) {
    if let Some(codes) = el.get("codes").and_then(|c| c.as_array()) {
        for code_obj in codes {
            if let Some(code) = code_obj.get("value").and_then(|v| v.as_str()) {
                if seen_codes.insert(code.to_string()) {
                    let name = code_obj
                        .get("name")
                        .and_then(|v| v.as_str())
                        .unwrap_or_default()
                        .to_string();
                    values.push(EntityMapKeyValue {
                        code: code.to_string(),
                        name,
                    });
                }
            }
        }
    }
    // Check components for nested codes
    if let Some(components) = el.get("components").and_then(|c| c.as_array()) {
        for comp in components {
            collect_codes_from_element(comp, seen_codes, values);
        }
    }
}

/// Aggregate discriminator codes from split child groups (e.g., sg12_z63, sg12_z65).
fn aggregate_split_child_codes(
    fields: &serde_json::Map<String, Value>,
    sg_lower: &str,
    key_field: &str,
) -> Option<EntityMapKeyInfo> {
    aggregate_codes_from_children(fields, sg_lower, key_field)
}

/// Aggregate codes from children whose key starts with `prefix_`.
fn aggregate_codes_from_children(
    children: &serde_json::Map<String, Value>,
    prefix: &str,
    key_field: &str,
) -> Option<EntityMapKeyInfo> {
    let prefix_underscore = format!("{prefix}_");
    let mut values: Vec<EntityMapKeyValue> = Vec::new();
    let mut seen_codes: BTreeSet<String> = BTreeSet::new();

    for (k, child) in children {
        if !k.starts_with(&prefix_underscore) && k != prefix {
            continue;
        }
        // Extract the qualifier suffix from the key (e.g., "z63" from "sg12_z63")
        if let Some(suffix) = k.strip_prefix(&prefix_underscore) {
            let code = suffix.to_uppercase();
            if seen_codes.insert(code.clone()) {
                // Try to get a name from the child's beschreibung or name
                let name = child
                    .get("beschreibung")
                    .or_else(|| child.get("name"))
                    .and_then(|v| v.as_str())
                    .unwrap_or_default()
                    .to_string();
                values.push(EntityMapKeyValue { code, name });
            }
        }

        // Also check discriminator codes inside the child
        if let Some(disc) = child.get("discriminator") {
            if let Some(disc_values) = disc.get("values").and_then(|v| v.as_array()) {
                for val in disc_values {
                    if let Some(code) = val.as_str() {
                        if seen_codes.insert(code.to_string()) {
                            values.push(EntityMapKeyValue {
                                code: code.to_string(),
                                name: String::new(),
                            });
                        }
                    }
                }
            }
        }
    }

    if values.len() >= 2 {
        Some(EntityMapKeyInfo {
            field: key_field.to_string(),
            values,
        })
    } else {
        None
    }
}

/// Check whether a TOML definition's discriminator is compatible with a schema group.
///
/// The schema group has a `discriminator` object with `{segment, element}` and the
/// group's segments contain `codes` arrays listing valid values. If the definition's
/// discriminator value (e.g., `TN` from `RFF.c506.d1153=TN`) is not among the schema
/// group's codes for that element, the definition doesn't belong to this PID.
///
/// Returns `true` if compatible (or if we can't determine — be permissive).
fn discriminator_matches_schema_group(disc: &str, group_json: &Value) -> bool {
    let Some((_, disc_value)) = disc.split_once('=') else {
        return true; // Can't parse — be permissive
    };
    // Strip occurrence suffix like "#0" from discriminator value
    let disc_value = disc_value.split('#').next().unwrap_or(disc_value);

    let Some(schema_disc) = group_json.get("discriminator") else {
        return true; // No schema discriminator — can't verify
    };
    let Some(disc_element) = schema_disc.get("element").and_then(|v| v.as_str()) else {
        return true;
    };

    // Find the discriminator element in the group's segments and check if disc_value
    // is among its codes
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return true;
    };

    // Collect ALL codes for the discriminator element across all segments.
    // The schema may have multiple segment variants (e.g., two RFF segments:
    // one with Z13, another with TN) and we need to check across all of them.
    let mut found_element = false;
    let mut all_codes: Vec<String> = Vec::new();

    for seg in segments {
        let elements = match seg.get("elements").and_then(|v| v.as_array()) {
            Some(e) => e,
            None => continue,
        };
        for el in elements {
            let mut collect_codes = |element: &Value| -> bool {
                let Some(id) = element.get("id").and_then(|v| v.as_str()) else {
                    return false;
                };
                if id != disc_element {
                    return false;
                }
                if let Some(codes) = element.get("codes").and_then(|v| v.as_array()) {
                    for c in codes {
                        if let Some(v) = c.get("value").and_then(|v| v.as_str()) {
                            all_codes.push(v.to_string());
                        }
                    }
                }
                true
            };

            // Check top-level element
            if collect_codes(el) {
                found_element = true;
            }
            // Check components within composite
            if let Some(components) = el.get("components").and_then(|v| v.as_array()) {
                for comp in components {
                    if collect_codes(comp) {
                        found_element = true;
                    }
                }
            }
        }
    }

    if !found_element {
        return true; // Element not found in schema — be permissive
    }

    all_codes.iter().any(|c| c == disc_value)
}

// ── Schema Navigation Helpers ──────────────────────────────────────────────

/// Fuzzy-resolve a source_group like "SG4.SG8.SG10" against the schema.
///
/// Converts source_group parts to lowercase and finds ALL matching variant paths
/// in the schema tree, merging their segments into a synthetic group. This is needed
/// because the assembler merges same-ID groups (e.g., sg8_z01 + sg8_z75 → SG8),
/// and TOML mappings reference the merged result.
fn resolve_schema_group_fuzzy(
    schema: &Value,
    source_group: &str,
    discriminator: Option<&str>,
) -> Option<Value> {
    let parts: Vec<String> = source_group.split('.').map(|p| p.to_lowercase()).collect();
    if parts.is_empty() {
        return None;
    }

    let fields = schema.get("fields")?.as_object()?;

    // Collect ALL matching groups at each level and merge their segments
    let initial = find_all_matching_values(fields, &parts[0]);
    if initial.is_empty() {
        return None;
    }

    let mut current_groups: Vec<&Value> = initial;

    for part in &parts[1..] {
        let mut next_groups = Vec::new();
        for group in &current_groups {
            if let Some(children) = group.get("children").and_then(|c| c.as_object()) {
                next_groups.extend(find_all_matching_values(children, part));
            }
        }
        if next_groups.is_empty() {
            return None;
        }
        current_groups = next_groups;
    }

    // A discriminator names one of the variants (`RFF…=TN` over `sg6_tn`,
    // `sg6_z13`): merge only those. Merging all of them folds RFF+TN into
    // RFF+Z13 — one segment with leading codes {Z13, TN} and Z13's D_1154
    // codes — so `transaktionsnummer` was checked against the PID as its only
    // valid value.
    if let Some((lhs, value)) = discriminator.and_then(|d| d.split_once('=')) {
        let tag = lhs.split('.').next().unwrap_or("");
        let value = value.split('#').next().unwrap_or(value);
        let chosen: Vec<&Value> = current_groups
            .iter()
            .copied()
            .filter(|g| {
                g.get("segments")
                    .and_then(|s| s.as_array())
                    .is_some_and(|segs| {
                        segs.iter().any(|s| {
                            s.get("id")
                                .and_then(|v| v.as_str())
                                .is_some_and(|id| id.eq_ignore_ascii_case(tag))
                                && leading_codes(s).is_some_and(|c| c.contains(&value))
                        })
                    })
            })
            .collect();
        if !chosen.is_empty() {
            current_groups = chosen;
        }
    }

    // Merge all matching groups' segments into one synthetic group
    merge_groups_segments(&current_groups)
}

/// Find ALL values in a JSON object whose key matches exactly or has the given prefix followed by '_'.
fn find_all_matching_values<'a>(
    obj: &'a serde_json::Map<String, Value>,
    prefix: &str,
) -> Vec<&'a Value> {
    let mut results = Vec::new();
    // Exact match
    if let Some(v) = obj.get(prefix) {
        results.push(v);
        return results;
    }
    // Prefix match: "sg8" matches "sg8_z01", "sg8_z75", etc.
    let prefix_underscore = format!("{prefix}_");
    for (k, v) in obj {
        if k.starts_with(&prefix_underscore) {
            results.push(v);
        }
    }
    results
}

/// Merge segments from multiple schema groups into a single synthetic group.
///
/// Unions segments by tag (keeping all elements from all variants) and merges
/// ahb_status, taking the first non-empty one.
fn merge_groups_segments(groups: &[&Value]) -> Option<Value> {
    let mut merged_segments: Vec<Value> = Vec::new();
    let mut ahb_status = String::new();

    for group in groups {
        if ahb_status.is_empty() {
            if let Some(s) = group.get("ahb_status").and_then(|v| v.as_str()) {
                ahb_status = s.to_string();
            }
        }
        if let Some(segments) = group.get("segments").and_then(|v| v.as_array()) {
            for seg in segments {
                let seg_id = seg.get("id").and_then(|v| v.as_str()).unwrap_or("");
                // Check if we already have this segment tag
                if let Some(existing) = merged_segments
                    .iter_mut()
                    .find(|s| s.get("id").and_then(|v| v.as_str()).unwrap_or("") == seg_id)
                {
                    // Merge elements: add any elements/components not already present
                    merge_segment_elements(existing, seg);
                } else {
                    merged_segments.push(seg.clone());
                }
            }
        }
    }

    if merged_segments.is_empty() {
        return None;
    }

    let mut result = serde_json::json!({ "segments": merged_segments });
    if !ahb_status.is_empty() {
        result["ahb_status"] = Value::String(ahb_status);
    }
    // Propagate max_reps from source groups (take the max)
    let max_reps = groups
        .iter()
        .filter_map(|g| g.get("max_reps").and_then(|v| v.as_i64()))
        .max();
    if let Some(max_reps) = max_reps {
        result["max_reps"] = serde_json::Value::Number(max_reps.into());
    }
    Some(result)
}

/// Merge elements from `source` segment into `target` segment.
fn merge_segment_elements(target: &mut Value, source: &Value) {
    let source_elements = match source.get("elements").and_then(|v| v.as_array()) {
        Some(e) => e,
        None => return,
    };

    let target_elements = match target.get_mut("elements").and_then(|v| v.as_array_mut()) {
        Some(e) => e,
        None => return,
    };

    for src_elem in source_elements {
        // For composites, merge components (especially codes)
        if let Some(src_composite) = src_elem.get("composite").and_then(|v| v.as_str()) {
            if let Some(tgt_elem) = target_elements
                .iter_mut()
                .find(|e| e.get("composite").and_then(|v| v.as_str()) == Some(src_composite))
            {
                merge_composite_components(tgt_elem, src_elem);
            } else {
                target_elements.push(src_elem.clone());
            }
        } else if let Some(src_id) = src_elem.get("id").and_then(|v| v.as_str()) {
            // Simple element — union its codes into the existing copy (e.g. the
            // DE3035 NAD qualifier: Z03 in one variant, Z07 in another), or add it
            // if not present.
            if let Some(tgt_elem) = target_elements.iter_mut().find(|e| {
                e.get("id").and_then(|v| v.as_str()) == Some(src_id) && e.get("composite").is_none()
            }) {
                merge_element_codes(tgt_elem, src_elem);
            } else {
                target_elements.push(src_elem.clone());
            }
        }
    }
}

/// Merge components within a composite, unioning code values.
fn merge_composite_components(target: &mut Value, source: &Value) {
    let source_comps = match source.get("components").and_then(|v| v.as_array()) {
        Some(c) => c,
        None => return,
    };
    let target_comps = match target.get_mut("components").and_then(|v| v.as_array_mut()) {
        Some(c) => c,
        None => return,
    };

    for src_comp in source_comps {
        let src_id = src_comp.get("id").and_then(|v| v.as_str()).unwrap_or("");
        if let Some(tgt_comp) = target_comps
            .iter_mut()
            .find(|c| c.get("id").and_then(|v| v.as_str()).unwrap_or("") == src_id)
        {
            merge_element_codes(tgt_comp, src_comp);
        } else {
            target_comps.push(src_comp.clone());
        }
    }
}

/// Union the `codes` of schema element `source` into schema element `target`.
fn merge_element_codes(target: &mut Value, source: &Value) {
    let Some(src_codes) = source.get("codes").and_then(|v| v.as_array()) else {
        return;
    };
    if let Some(tgt_codes) = target.get_mut("codes").and_then(|v| v.as_array_mut()) {
        for code in src_codes {
            let code_val = code.get("value").and_then(|v| v.as_str()).unwrap_or("");
            let already = tgt_codes
                .iter()
                .any(|c| c.get("value").and_then(|v| v.as_str()).unwrap_or("") == code_val);
            if !already {
                tgt_codes.push(code.clone());
            }
        }
    } else if let Some(tgt_obj) = target.as_object_mut() {
        // Target has no codes, source does — add them
        tgt_obj.insert("codes".to_string(), Value::Array(src_codes.clone()));
        // Also update type to "code" if source is code
        if source.get("type").and_then(|v| v.as_str()) == Some("code") {
            tgt_obj.insert("type".to_string(), Value::String("code".to_string()));
        }
    }
}

// ── Per-Variant Requirements ───────────────────────────────────────────────

/// A leaf schema group matched by a fuzzy `source_group`, with its discriminator.
struct DiscriminatedGroup<'a> {
    path: String,
    group: &'a Value,
    segment: String,
    element: String,
    codes: Vec<String>,
}

/// Find ALL schema groups matching `source_group` (like `resolve_schema_group_fuzzy`)
/// together with their schema paths, keeping only those that carry a
/// `discriminator` with at least one value.
fn discriminated_groups<'a>(schema: &'a Value, source_group: &str) -> Vec<DiscriminatedGroup<'a>> {
    let parts: Vec<String> = source_group.split('.').map(|p| p.to_lowercase()).collect();
    let Some(fields) = schema.get("fields").and_then(|f| f.as_object()) else {
        return Vec::new();
    };
    let Some((first, rest)) = parts.split_first() else {
        return Vec::new();
    };

    let mut current: Vec<(String, &Value)> = find_all_matching_entries(fields, first);
    for part in rest {
        let mut next = Vec::new();
        for (path, group) in &current {
            if let Some(children) = group.get("children").and_then(|c| c.as_object()) {
                for (key, child) in find_all_matching_entries(children, part) {
                    next.push((format!("{path}.{key}"), child));
                }
            }
        }
        current = next;
    }

    current
        .into_iter()
        .filter_map(|(path, group)| {
            let disc = group.get("discriminator")?;
            let segment = disc.get("segment")?.as_str()?.to_uppercase();
            let element = disc.get("element")?.as_str()?.to_uppercase();
            let codes: Vec<String> = disc
                .get("values")?
                .as_array()?
                .iter()
                .filter_map(|v| v.as_str().map(str::to_string))
                .collect();
            if codes.is_empty() {
                return None;
            }
            Some(DiscriminatedGroup {
                path,
                group,
                segment,
                element,
                codes,
            })
        })
        .collect()
}

/// Like `find_all_matching_values`, but also returns the matched keys.
fn find_all_matching_entries<'a>(
    obj: &'a serde_json::Map<String, Value>,
    prefix: &str,
) -> Vec<(String, &'a Value)> {
    if let Some(v) = obj.get(prefix) {
        return vec![(prefix.to_string(), v)];
    }
    let prefix_underscore = format!("{prefix}_");
    obj.iter()
        .filter(|(k, _)| k.starts_with(&prefix_underscore))
        .map(|(k, v)| (k.clone(), v))
        .collect()
}

/// Does TOML field `path` address the discriminator element of `dg`?
///
/// Resolves the path against the group's schema and checks that it lands on the
/// discriminator segment/element and that element lists the discriminator codes.
fn field_is_discriminator(path: &str, dg: &DiscriminatedGroup<'_>) -> bool {
    let Some((seg_tag, composite_or_element, component_id)) = parse_toml_field_path(path) else {
        return false;
    };
    if path.contains('[') || seg_tag.to_uppercase() != dg.segment {
        return false;
    }
    let Some(elem) = find_schema_element(
        dg.group,
        &seg_tag,
        None,
        &composite_or_element,
        component_id.as_deref(),
    ) else {
        return false;
    };
    if elem
        .get("id")
        .and_then(|v| v.as_str())
        .map(str::to_uppercase)
        != Some(dg.element.clone())
    {
        return false;
    }
    let elem_codes: BTreeSet<&str> = elem
        .get("codes")
        .and_then(|c| c.as_array())
        .map(|codes| {
            codes
                .iter()
                .filter_map(|c| c.get("value").and_then(|v| v.as_str()))
                .collect()
        })
        .unwrap_or_default();
    dg.codes.iter().all(|c| elem_codes.contains(c.as_str()))
}

/// Build per-variant requirements for a definition whose `source_group` matches
/// several discriminated schema group variants (e.g. `SG4.SG12` → `sg12_z03`,
/// `sg12_z07`, …), provided the definition maps the discriminator element to a
/// BO4E field. Each discriminator code becomes one variant whose field
/// requirements come only from the schema groups carrying that code.
fn add_variant_requirements(
    schema: &Value,
    def: &MappingDefinition,
    child_group_info: Option<&ChildGroupInfo>,
    variants: &mut BTreeMap<(String, String), VariantBuilder>,
    code_lists: &crate::code_lists::CodeLists,
) {
    let groups = discriminated_groups(schema, &def.meta.source_group);
    let distinct_codes: BTreeSet<&str> = groups
        .iter()
        .flat_map(|g| g.codes.iter().map(String::as_str))
        .collect();
    if groups.len() < 2 || distinct_codes.len() < 2 {
        return;
    }

    // The BO4E field mapped from the discriminator element in every variant.
    let Some((disc_target, disc_enum_map)) = def.fields.iter().find_map(|(path, fm)| {
        let target = extract_target_name(fm).filter(|t| !t.is_empty())?;
        if !groups.iter().all(|g| field_is_discriminator(path, g)) {
            return None;
        }
        let enum_map = match fm {
            FieldMapping::Structured(s) => s.enum_map.clone().or_else(|| {
                s.code_list
                    .as_deref()
                    .and_then(|n| code_lists.get(n))
                    .cloned()
            }),
            _ => None,
        };
        Some((target, enum_map))
    }) else {
        return;
    };

    for code in distinct_codes {
        let members: Vec<&DiscriminatedGroup<'_>> = groups
            .iter()
            .filter(|g| g.codes.iter().any(|c| c == code))
            .collect();
        let member_values: Vec<&Value> = members.iter().map(|g| g.group).collect();
        let Some(merged) = merge_groups_segments(&member_values) else {
            continue;
        };
        let parent_ahb = merged
            .get("ahb_status")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        let vb = variants
            .entry((disc_target.clone(), code.to_string()))
            .or_default();
        if vb.bo4e_value.is_none() {
            vb.bo4e_value = disc_enum_map.as_ref().and_then(|m| m.get(code).cloned());
        }
        vb.source_paths
            .extend(members.iter().map(|g| g.path.clone()));

        let before: BTreeSet<String> = vb.fields.keys().cloned().collect();
        process_field_section(
            &def.fields,
            &merged,
            &parent_ahb,
            &mut vb.fields,
            false,
            code_lists,
            def.meta.discriminator.as_deref(),
        );
        if let Some(cg) = child_group_info {
            for (key, field) in vb.fields.iter_mut() {
                if !before.contains(key) && field.child_group.is_none() {
                    field.child_group = Some(cg.clone());
                }
            }
        }
    }
}

/// Resolve a source_path like "sg4.sg5_z16" into the corresponding schema group JSON.
///
/// Navigates: `schema.fields.{part1}.children.{part2}.children.{part3}...`
fn resolve_schema_group<'a>(schema: &'a Value, source_path: &str) -> Option<&'a Value> {
    let parts: Vec<&str> = source_path.split('.').collect();
    if parts.is_empty() {
        return None;
    }

    let mut current = schema.get("fields")?.get(parts[0])?;

    for part in &parts[1..] {
        current = current.get("children")?.get(*part)?;
    }

    Some(current)
}

/// Process a TOML field section ([fields] or [companion_fields]) and resolve each
/// field against the schema to build FieldRequirements.
///
/// `include_unresolved` controls fields whose path is not found in `group_json`:
/// when `true` they get a minimal requirement that demands nothing (the PID's AHB
/// does not list them in this group); when `false`
/// they are skipped (used for per-variant requirements, where a field absent from
/// the variant's schema group simply does not apply to that variant).
fn process_field_section(
    section: &indexmap::IndexMap<String, FieldMapping>,
    group_json: &Value,
    parent_ahb: &str,
    output: &mut BTreeMap<String, FieldRequirement>,
    include_unresolved: bool,
    code_lists: &crate::code_lists::CodeLists,
    discriminator: Option<&str>,
) {
    for (path, field_mapping) in section {
        let bo4e_name = match extract_target_name(field_mapping) {
            Some(name) if !name.is_empty() => name,
            _ => continue, // Skip empty targets / defaults-only
        };

        // Deduplicate by bo4e_name
        if output.contains_key(&bo4e_name) {
            continue;
        }

        let parsed = match parse_toml_field_path(path) {
            Some(p) => p,
            None => continue,
        };

        let (seg_tag, composite_or_element, component_id) = parsed;
        // `rff[ACW].c506.d1154` resolves only against the RFF+ACW segment variant.
        // Without a path qualifier the definition's discriminator selects the
        // segment (`RFF…=TN` over an unqualified `rff.c506.d1154`): otherwise
        // the element resolved against the first same-tag segment — RFF+Z13 —
        // and `transaktionsnummer` was demanded with the PID as its only code.
        let qualifier = field_path_qualifier(path)
            .or_else(|| discriminator_qualifier(discriminator, &seg_tag, group_json));

        let field = if let Some(schema_elem) = find_schema_element(
            group_json,
            &seg_tag,
            qualifier,
            &composite_or_element,
            component_id.as_deref(),
        ) {
            let mut req = build_field_requirement(&schema_elem, &bo4e_name, parent_ahb);
            // Merge codes from ALL matching segments with the same element.
            // E.g., SG2 has two NAD segments (MS and MR) — merge both codes
            // into one list so the enum includes both variants.
            merge_codes_from_all_segments(
                group_json,
                &seg_tag,
                qualifier,
                &composite_or_element,
                component_id.as_deref(),
                &mut req.valid_codes,
            );
            // The mapping's enum_map translates raw codes into BO4E values; record
            // the translation so validation can accept values in either code space.
            if let FieldMapping::Structured(s) = field_mapping {
                // Either spelling of the table: inline, or the shared one the
                // rule names. Reading only the inline form makes a translated
                // value look like an invalid code, because validation compares
                // it against the raw EDIFACT list.
                let map = s
                    .enum_map
                    .as_ref()
                    .or_else(|| s.code_list.as_deref().and_then(|n| code_lists.get(n)));
                if let Some(map) = map {
                    for cv in &mut req.valid_codes {
                        if cv.bo4e_value.is_none() {
                            cv.bo4e_value = map.get(&cv.code).cloned();
                        }
                    }
                }
            }
            // An element's `X` is a statement about its *segment*: fill it if the
            // segment is there. Where the segment itself is `Kann`, a message that
            // omits the whole segment omits the element legitimately, so the
            // element cannot be unconditionally required.
            relax_status_inside_optional_segment(
                group_json,
                &seg_tag,
                qualifier,
                &mut req.ahb_status,
            );
            relax_status_of_shared_qualifier(group_json, &seg_tag, qualifier, &mut req.ahb_status);
            if qualifier.is_none() {
                scope_status_to_holding_segments(
                    group_json,
                    &seg_tag,
                    &composite_or_element,
                    component_id.as_deref(),
                    &mut req.ahb_status,
                );
            }
            req
        } else if !include_unresolved {
            continue;
        } else {
            // The PID's AHB does not list this element in this group: a rule
            // shared with other PIDs (or format versions) maps it, but nothing
            // here demands it. It used to inherit the *group's* status, so
            // FV2610 55043 demanded `geschaeftspartnerId` under SG12
            // Messlokationsadresse's `Muss [586]` — a NAD+Z03 without C082 — and
            // `produktIdentifikation` under SEQ+Z18's `Muss [2284]`, which has no
            // PIA; with no group status it became "unknown", demanded all the same.
            FieldRequirement {
                bo4e_name: bo4e_name.clone(),
                ahb_status: String::new(),
                field_type: "data".to_string(),
                format: None,
                enum_name: None,
                valid_codes: Vec::new(),
                child_group: None,
                ref_type: Bo4eRefType::Unknown,
            }
        };

        output.insert(bo4e_name, field);
    }
}

/// Extract the target field name from a FieldMapping.
fn extract_target_name(fm: &FieldMapping) -> Option<String> {
    match fm {
        FieldMapping::Simple(s) => {
            if s.is_empty() {
                None
            } else {
                Some(s.clone())
            }
        }
        FieldMapping::Structured(s) => {
            if s.target.is_empty() {
                None
            } else {
                Some(s.target.clone())
            }
        }
        FieldMapping::Nested(_) => None,
    }
}

/// Parse a TOML field path into (segment_tag, composite_or_element, component).
///
/// Examples:
/// - "loc.c517.d3225" -> ("loc", "c517", Some("d3225"))
/// - "nad.d3035" -> ("nad", "d3035", None)
/// - "cav[Z91].c889.d7111" -> ("cav", "c889", Some("d7111"))
fn parse_toml_field_path(path: &str) -> Option<(String, String, Option<String>)> {
    let parts: Vec<&str> = path.split('.').collect();
    if parts.len() < 2 || parts.len() > 3 {
        return None;
    }

    // Strip qualifier from first part: "cav[Z91]" -> "cav"
    let seg_tag = parts[0].split('[').next().unwrap_or(parts[0]).to_string();

    let second = parts[1].to_string();

    let third = if parts.len() == 3 {
        Some(parts[2].to_string())
    } else {
        None
    };

    Some((seg_tag, second, third))
}

/// The segment qualifier of a TOML field path: `Z91` for `cav[Z91].c889.d7111`
/// or `rff[Z34,1].0.1`; `None` without one or for the `[*,N]` wildcard.
/// The value of a definition's discriminator (`RFF.0.0=TN`,
/// `RFF.c506.d1153=TN`) when it is on `seg_tag` and names a segment variant of
/// the group — a leading code of one of its same-tag segments.
fn discriminator_qualifier<'a>(
    discriminator: Option<&'a str>,
    seg_tag: &str,
    group_json: &Value,
) -> Option<&'a str> {
    let (lhs, value) = discriminator?.split_once('=')?;
    let tag = lhs.split('.').next()?;
    if !tag.eq_ignore_ascii_case(seg_tag) || value.is_empty() {
        return None;
    }
    let segments = group_json.get("segments")?.as_array()?;
    segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .is_some_and(|id| id.eq_ignore_ascii_case(seg_tag))
        })
        .any(|s| leading_codes(s).is_some_and(|codes| codes.contains(&value)))
        .then_some(value)
}

fn field_path_qualifier(path: &str) -> Option<&str> {
    let tag_part = path.split('.').next()?;
    let inner = tag_part.split_once('[')?.1.trim_end_matches(']');
    let qualifier = inner.split(',').next().unwrap_or(inner);
    (!qualifier.is_empty() && qualifier != "*").then_some(qualifier)
}

/// The codes a schema segment lists at its leading position (element 0,
/// component 0) — the position a `tag[QUAL]` field path selects on. `None` if the
/// schema lists no codes there.
fn leading_codes(segment: &Value) -> Option<Vec<&str>> {
    let element0 = segment
        .get("elements")
        .and_then(|e| e.as_array())?
        .iter()
        .find(|el| el.get("index").and_then(|v| v.as_u64()) == Some(0))?;
    let leading = match element0.get("components").and_then(|c| c.as_array()) {
        Some(components) => components
            .iter()
            .find(|c| c.get("sub_index").and_then(|v| v.as_u64()) == Some(0))?,
        None => element0,
    };
    let codes: Vec<&str> = leading
        .get("codes")
        .and_then(|c| c.as_array())?
        .iter()
        .filter_map(|c| c.get("value").and_then(|v| v.as_str()))
        .collect();
    (!codes.is_empty()).then_some(codes)
}

/// The same-tag schema segments a field path with `qualifier` reads.
///
/// The segments whose leading codes include the qualifier; if there are none, the
/// segments whose leading position lists no codes (they cannot be ruled out).
/// Segments of other qualifiers are never used: their codes and formats belong
/// to a different segment variant.
/// Downgrade a field status that is required only *within* its segment, when
/// the AHB marks that segment optional.
///
/// The AHB annotates requiredness at two levels, and both apply: `Kann` on the
/// segment means the message may leave the segment out entirely, while `X` on a
/// component means "if the segment is there, fill this". Reading only the
/// component made every such element unconditionally required, so a message that
/// legitimately omits an optional segment was reported as missing its
/// components. 55042's SG12 `RFF` (segment `Kann`, `D_1154` `X [951]`) is the
/// case that surfaced it: `[951]` is a *format* condition that always evaluates
/// true, so the conditional validator reported a hard error for a reference the
/// AHB does not ask for.
///
/// The condition suffix is kept, so `X [951]` becomes `Kann [951]`: both
/// `is_unconditionally_required` and `evaluate_ahb_requirement` read a
/// `Kann`-prefixed status as optional, and the format note survives for display.
///
/// A segment the schema does not annotate (`ahb_status` absent) is left alone:
/// absence means "not stated here", not "optional" — root segments (BGM, DTM of
/// the message header) carry no status in the schema and are mandatory.
fn relax_status_inside_optional_segment(
    group_json: &Value,
    seg_tag: &str,
    qualifier: Option<&str>,
    ahb_status: &mut String,
) {
    let required = matches!(status_keyword(ahb_status), "X" | "Muss" | "Soll");
    if !required {
        return;
    }
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return;
    };
    let seg_tag_upper = seg_tag.to_uppercase();
    let matching: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .is_some_and(|id| id.to_uppercase() == seg_tag_upper)
        })
        .collect();
    let matching = select_segments_for_qualifier(matching, qualifier);
    if matching.is_empty() {
        return;
    }
    // Several segments share the tag (SG2's NAD+MS and NAD+MR, SG6's RFF
    // variants): the field is required as long as *one* of them requires it.
    let all_optional = matching.iter().all(|s| {
        s.get("ahb_status")
            .and_then(|v| v.as_str())
            .is_some_and(|status| {
                let head = status_keyword(status);
                head == "Kann" || head == "K"
            })
    });
    if !all_optional {
        return;
    }
    let condition = ahb_status
        .find(['[', '('])
        .map(|i| format!(" {}", ahb_status[i..].trim()))
        .unwrap_or_default();
    *ahb_status = format!("Kann{condition}");
}

/// Scope an unqualified field's status to the same-tag segments that hold its
/// element.
///
/// A path without a qualifier (`sts.c556.d1131`) resolves against whichever
/// same-tag segment has the element, and takes that element's status — but
/// the other segments do not have it. 55023's SG4 has STS+7 (no D_1131) and
/// STS+E01 `Muss [249]` (D_1131 `X`): with the fixture's [249] false the
/// segment is absent, yet `ebdNummer` was demanded. 55553's FTX qualifiers
/// hold one to five D_4440, so every `Freitext` was asked for `textZeile2`.
///
/// One holding segment: its condition joins the element's (`X` in
/// `Muss [249]` becomes `X [249]`). Several but not all: which one an
/// instance is cannot be told from the status, so the field is optional
/// (`Kann`, condition kept).
fn scope_status_to_holding_segments(
    group_json: &Value,
    seg_tag: &str,
    composite_or_element: &str,
    component_id: Option<&str>,
    ahb_status: &mut String,
) {
    if !matches!(status_keyword(ahb_status), "X" | "Muss" | "Soll") {
        return;
    }
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return;
    };
    let seg_tag_upper = seg_tag.to_uppercase();
    let same_tag: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .is_some_and(|id| id.to_uppercase() == seg_tag_upper)
        })
        .collect();
    if same_tag.len() < 2 {
        return;
    }
    let holders: Vec<&Value> = same_tag
        .iter()
        .copied()
        .filter(|seg| {
            let alone = serde_json::json!({ "segments": [seg] });
            find_schema_element(&alone, seg_tag, None, composite_or_element, component_id).is_some()
        })
        .collect();
    if holders.is_empty() || holders.len() == same_tag.len() {
        return;
    }
    let condition_of = |status: &str| -> Option<String> {
        status
            .find(['[', '('])
            .map(|i| status[i..].trim().to_string())
    };
    let own = condition_of(ahb_status);
    if let [holder] = holders[..] {
        let seg_status = holder
            .get("ahb_status")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        match status_keyword(seg_status) {
            "Kann" | "K" => {}
            // An unconditionally required segment adds nothing.
            _ if condition_of(seg_status).is_none() => return,
            _ => {
                let seg_cond = condition_of(seg_status).unwrap_or_default();
                let keyword = status_keyword(ahb_status).to_string();
                *ahb_status = match own {
                    Some(own) => format!("{keyword} ({seg_cond}) ∧ ({own})"),
                    None => format!("{keyword} {seg_cond}"),
                };
                return;
            }
        }
    }
    *ahb_status = match own {
        Some(own) => format!("Kann {own}"),
        None => "Kann".to_string(),
    };
}

/// The requirement keyword of an AHB status: `X`, `Muss`, `Soll`, `Kann`, `K`.
///
/// Everything up to the first `[` is not enough: a parenthesised expression
/// (`X ([442] ∧ [951]) ⊻ …`) leaves `X (`, which then matched none of the
/// keywords, so such statuses were never relaxed.
fn status_keyword(status: &str) -> &str {
    let s = status.trim_start();
    let end = s.find(|c: char| !c.is_alphabetic()).unwrap_or(s.len());
    &s[..end]
}

/// Downgrade a field read through one qualifier of a segment that allows
/// several.
///
/// The AHB writes one segment for all of them — Lokationsbündel's `RFF` with
/// `D_1153` Z34 (vorgelagerte Messlokation) or Z35 (vorgelagerte Netzlokation),
/// and one `D_1154` status for both: `X ([442] ∧ [951]) ⊻ ([440] ∧ [960])`,
/// where `[442]`/`[440]` read "in demselben RFF". A mapping splits the element
/// per qualifier (`rff[Z34].c506.d1154`, `rff[Z35].c506.d1154`), so each target
/// carried the shared status, and a message with only RFF+Z34 was reported as
/// missing the Z35 target. The element is required within *its* segment
/// instance, which a message may carry under either qualifier; the segment's
/// own presence is judged at group level.
///
/// The condition suffix is kept (`X [...]` becomes `Kann [...]`), as in
/// [`relax_status_inside_optional_segment`].
fn relax_status_of_shared_qualifier(
    group_json: &Value,
    seg_tag: &str,
    qualifier: Option<&str>,
    ahb_status: &mut String,
) {
    let Some(qualifier) = qualifier else {
        return;
    };
    let required = matches!(status_keyword(ahb_status), "X" | "Muss" | "Soll");
    if !required {
        return;
    }
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return;
    };
    let seg_tag_upper = seg_tag.to_uppercase();
    let exact: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .is_some_and(|id| id.to_uppercase() == seg_tag_upper)
        })
        .filter(|s| leading_codes(s).is_some_and(|codes| codes.contains(&qualifier)))
        .collect();
    let shared = !exact.is_empty()
        && exact
            .iter()
            .all(|s| leading_codes(s).is_some_and(|codes| codes.len() > 1));
    if !shared {
        return;
    }
    let condition = ahb_status
        .find(['[', '('])
        .map(|i| format!(" {}", ahb_status[i..].trim()))
        .unwrap_or_default();
    *ahb_status = format!("Kann{condition}");
}

fn select_segments_for_qualifier<'a>(
    segments: Vec<&'a Value>,
    qualifier: Option<&str>,
) -> Vec<&'a Value> {
    let Some(qualifier) = qualifier else {
        return segments;
    };
    let exact: Vec<&Value> = segments
        .iter()
        .copied()
        .filter(|s| leading_codes(s).is_some_and(|codes| codes.contains(&qualifier)))
        .collect();
    if !exact.is_empty() {
        return exact;
    }
    segments
        .into_iter()
        .filter(|s| leading_codes(s).is_none())
        .collect()
}

/// Merge codes from ALL matching segments into the `valid_codes` list.
///
/// When a group has multiple segments with the same tag (e.g., two NAD segments
/// for MS and MR in SG2), `find_schema_element` only returns the first match.
/// This function finds all matches and adds any codes not already in the list.
fn merge_codes_from_all_segments(
    group_json: &Value,
    seg_tag: &str,
    qualifier: Option<&str>,
    composite_or_element: &str,
    component_id: Option<&str>,
    valid_codes: &mut Vec<CodeValue>,
) {
    let Some(segments) = group_json.get("segments").and_then(|v| v.as_array()) else {
        return;
    };
    let seg_tag_upper = seg_tag.to_uppercase();

    let matching_segs: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .map(|id| id.to_uppercase() == seg_tag_upper)
                .unwrap_or(false)
        })
        .collect();
    let matching_segs = select_segments_for_qualifier(matching_segs, qualifier);

    // Collect existing code values for deduplication
    let existing: std::collections::HashSet<String> =
        valid_codes.iter().map(|c| c.code.clone()).collect();

    for seg in &matching_segs {
        // Try to find the same element in this segment
        if let Some(elem) = find_element_in_segment(seg, composite_or_element, component_id) {
            if let Some(codes) = elem.get("codes").and_then(|v| v.as_array()) {
                for code_val in codes {
                    let code = code_val.get("value").and_then(|v| v.as_str()).unwrap_or("");
                    if !code.is_empty() && !existing.contains(code) {
                        let meaning = code_val
                            .get("name")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .to_string();
                        let enum_name = code_val
                            .get("enum_name")
                            .or_else(|| code_val.get("enum"))
                            .and_then(|v| v.as_str())
                            .map(|s| s.to_string());
                        valid_codes.push(CodeValue {
                            code: code.to_string(),
                            meaning,
                            enum_name,
                            bo4e_value: None,
                        });
                    }
                }
            }
        }
    }
}

/// Find a specific element within a single segment (helper for merge).
///
/// Handles both named (d3035, c517.d3225) and numeric (0, 1.0) paths.
fn find_element_in_segment(
    seg: &Value,
    composite_or_element: &str,
    component_id: Option<&str>,
) -> Option<Value> {
    let elements = seg.get("elements")?.as_array()?;

    // Check if numeric path
    let is_numeric = composite_or_element.chars().all(|c| c.is_ascii_digit());

    if is_numeric {
        let elem_idx: usize = composite_or_element.parse().ok()?;
        // Find element at this index
        let elem = elements
            .iter()
            .find(|e| e.get("index").and_then(|v| v.as_u64()) == Some(elem_idx as u64))?;

        if let Some(comp_str) = component_id {
            if let Ok(sub_idx) = comp_str.parse::<usize>() {
                // Numeric component sub_index
                let components = elem.get("components")?.as_array()?;
                return components
                    .iter()
                    .find(|c| c.get("sub_index").and_then(|v| v.as_u64()) == Some(sub_idx as u64))
                    .cloned();
            }
        }
        // Simple element — return itself
        if elem.get("composite").is_none() {
            return Some(elem.clone());
        }
        // Composite without component — return first component
        let components = elem.get("components")?.as_array()?;
        return components.first().cloned();
    }

    let normalized_id = normalize_edifact_id(composite_or_element);
    let is_composite = normalized_id.starts_with('C');

    if is_composite {
        for elem in elements {
            let cid = elem.get("composite").and_then(|v| v.as_str()).unwrap_or("");
            if cid.to_uppercase() == normalized_id {
                if let Some(comp_id) = component_id {
                    let comp_normalized = normalize_edifact_id(comp_id);
                    let components = elem.get("components")?.as_array()?;
                    return components
                        .iter()
                        .find(|c| {
                            c.get("id")
                                .and_then(|v| v.as_str())
                                .map(|id| id.to_uppercase() == comp_normalized)
                                .unwrap_or(false)
                        })
                        .cloned();
                }
                return Some(elem.clone());
            }
        }
    } else {
        for elem in elements {
            if elem.get("composite").is_some() {
                continue;
            }
            let eid = elem.get("id").and_then(|v| v.as_str()).unwrap_or("");
            if eid.to_uppercase() == normalized_id {
                return Some(elem.clone());
            }
        }
    }
    None
}

/// Find a schema element matching a TOML field path within a group's segments.
///
/// Handles EDIFACT ID paths (`c517.d3225`), numeric paths (`2.0`),
/// and simple elements (`d3035` or `0`).
fn find_schema_element(
    group_json: &Value,
    seg_tag: &str,
    qualifier: Option<&str>,
    composite_or_element: &str,
    component_id: Option<&str>,
) -> Option<Value> {
    let segments = group_json.get("segments")?.as_array()?;
    let seg_tag_upper = seg_tag.to_uppercase();

    // Find matching segment(s) by id
    let matching_segs: Vec<&Value> = segments
        .iter()
        .filter(|s| {
            s.get("id")
                .and_then(|v| v.as_str())
                .map(|id| id.to_uppercase() == seg_tag_upper)
                .unwrap_or(false)
        })
        .collect();
    let matching_segs = select_segments_for_qualifier(matching_segs, qualifier);

    if matching_segs.is_empty() {
        return None;
    }

    // Check if this is a numeric index path (e.g., "2.0" instead of "c517.d3225")
    let is_numeric = composite_or_element.chars().all(|c| c.is_ascii_digit());

    if is_numeric {
        return find_schema_element_by_index(&matching_segs, composite_or_element, component_id);
    }

    let normalized_id = normalize_edifact_id(composite_or_element);
    let is_composite = normalized_id.starts_with('C');

    for seg in &matching_segs {
        let elements = seg.get("elements")?.as_array()?;

        if is_composite {
            for elem in elements {
                let composite_id = elem.get("composite").and_then(|v| v.as_str()).unwrap_or("");
                if composite_id.to_uppercase() == normalized_id {
                    if let Some(comp_id) = component_id {
                        // Check if component is numeric sub_index
                        let comp_stripped = strip_ordinal_suffix(comp_id);
                        if comp_stripped.chars().all(|c| c.is_ascii_digit()) {
                            let sub_idx: usize = comp_stripped.parse().ok()?;
                            if let Some(found) = elem
                                .get("components")
                                .and_then(|v| v.as_array())
                                .and_then(|components| {
                                    components.iter().find(|c| {
                                        c.get("sub_index").and_then(|v| v.as_u64())
                                            == Some(sub_idx as u64)
                                    })
                                })
                            {
                                return Some(found.clone());
                            }
                            // Not in this segment's composite: try the next one.
                            continue;
                        }
                        // Named component ID
                        let comp_normalized = normalize_edifact_id(comp_id);
                        let ordinal = extract_ordinal(comp_id);
                        if let Some(components) = elem.get("components").and_then(|v| v.as_array())
                        {
                            let mut count = 0usize;
                            for component in components {
                                let cid =
                                    component.get("id").and_then(|v| v.as_str()).unwrap_or("");
                                if cid.to_uppercase() == comp_normalized {
                                    if count == ordinal {
                                        return Some(component.clone());
                                    }
                                    count += 1;
                                }
                            }
                        }
                    } else {
                        return Some(elem.clone());
                    }
                }
            }
        } else {
            // Simple data element ID like "d3227" -> normalized "3227"
            let ordinal = extract_ordinal(composite_or_element);
            let mut count = 0usize;
            for elem in elements {
                if elem.get("composite").is_some() {
                    continue;
                }
                let eid = elem.get("id").and_then(|v| v.as_str()).unwrap_or("");
                if eid.to_uppercase() == normalized_id {
                    if count == ordinal {
                        return Some(elem.clone());
                    }
                    count += 1;
                }
            }
        }
    }

    None
}

/// Find a schema element by numeric index path (e.g., "2.0" = element at index 2, component at sub_index 0).
fn find_schema_element_by_index(
    matching_segs: &[&Value],
    element_index_str: &str,
    component_sub_index_str: Option<&str>,
) -> Option<Value> {
    let elem_idx: usize = element_index_str.parse().ok()?;

    for seg in matching_segs {
        let elements = match seg.get("elements").and_then(|v| v.as_array()) {
            Some(e) => e,
            None => continue,
        };

        let elem = match elements
            .iter()
            .find(|e| e.get("index").and_then(|v| v.as_u64()) == Some(elem_idx as u64))
        {
            Some(e) => e,
            None => continue,
        };

        if let Some(comp_str) = component_sub_index_str {
            let comp_stripped = strip_ordinal_suffix(comp_str);
            if let Ok(sub_idx) = comp_stripped.parse::<usize>() {
                if let Some(components) = elem.get("components").and_then(|v| v.as_array()) {
                    // Several segments share the tag (STS+7 and STS+E01): the
                    // component may sit in a later one — STS+7's C556 has no
                    // D_1131, STS+E01's does. Returning on the first segment
                    // left `sts.c556.d1131` unresolved.
                    match components.iter().find(|c| {
                        c.get("sub_index").and_then(|v| v.as_u64()) == Some(sub_idx as u64)
                    }) {
                        Some(c) => return Some(c.clone()),
                        None => continue,
                    }
                }
            }
        }

        return Some(elem.clone());
    }

    None
}

/// Build a FieldRequirement from a resolved schema element.
fn build_field_requirement(
    schema_elem: &Value,
    bo4e_name: &str,
    _parent_ahb: &str,
) -> FieldRequirement {
    // Only use the element's own AHB status — do NOT inherit from parent group.
    // In EDIFACT AHB, each data element has its own status. Elements without
    // explicit ahb_status are implicitly optional within the group (e.g., NAD
    // C080 sub-components for namenszusatz, additional address lines).
    let ahb_status = schema_elem
        .get("ahb_status")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();

    let field_type = schema_elem
        .get("type")
        .and_then(|v| v.as_str())
        .unwrap_or("data")
        .to_string();

    let format = schema_elem
        .get("format")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());

    let enum_name = schema_elem
        .get("codes")
        .and_then(|v| v.as_array())
        .and_then(|codes| {
            codes
                .first()
                .and_then(|c| c.get("enum").and_then(|v| v.as_str()))
                .map(to_pascal_case)
        });

    let valid_codes = extract_valid_codes(schema_elem);

    FieldRequirement {
        bo4e_name: bo4e_name.to_string(),
        ahb_status,
        field_type,
        format,
        enum_name,
        valid_codes,
        child_group: None,
        ref_type: Bo4eRefType::Unknown,
    }
}

/// Extract valid code values from a schema element's "codes" array.
fn extract_valid_codes(element: &Value) -> Vec<CodeValue> {
    element
        .get("codes")
        .and_then(|v| v.as_array())
        .map(|codes| {
            codes
                .iter()
                .filter_map(|c| {
                    let code = c.get("value").and_then(|v| v.as_str())?.to_string();
                    let meaning = c
                        .get("name")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();
                    let enum_name = c
                        .get("enum")
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string());
                    Some(CodeValue {
                        code,
                        meaning,
                        enum_name,
                        bo4e_value: None,
                    })
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Convert SCREAMING_SNAKE_CASE to PascalCase.
///
/// "HAUSHALTSKUNDE_ENWG" -> "HaushaltskundeEnwg"
fn to_pascal_case(s: &str) -> String {
    s.split('_')
        .map(|part| {
            let mut chars = part.chars();
            match chars.next() {
                Some(c) => {
                    let upper = c.to_uppercase().to_string();
                    let lower: String = chars.map(|c| c.to_ascii_lowercase()).collect();
                    format!("{upper}{lower}")
                }
                None => String::new(),
            }
        })
        .collect()
}

/// Normalize an EDIFACT ID path component to match schema IDs.
///
/// TOML paths use lowercase prefixed IDs: `c517`, `d3227`.
/// Schema JSON uses: `"C517"` for composites, `"3227"` for data elements (no D prefix).
fn normalize_edifact_id(id: &str) -> String {
    let stripped = strip_ordinal_suffix(id);
    let upper = stripped.to_uppercase();
    // Composite IDs start with C followed by digits
    if upper.starts_with('C') && upper.len() > 1 && upper[1..].chars().all(|c| c.is_ascii_digit()) {
        return upper;
    }
    // Data element IDs: strip 'D' prefix if present -> "D3227" -> "3227"
    if upper.starts_with('D') && upper.len() > 1 && upper[1..].chars().all(|c| c.is_ascii_digit()) {
        return upper[1..].to_string();
    }
    upper
}

/// Strip ordinal suffix from an EDIFACT ID: "c556_2" -> "c556", "d3036_2" -> "d3036".
fn strip_ordinal_suffix(id: &str) -> &str {
    if let Some(pos) = id.rfind('_') {
        let suffix = &id[pos + 1..];
        if !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit()) {
            return &id[..pos];
        }
    }
    id
}

/// Extract ordinal from a suffixed ID: "d3036_2" -> 1 (0-indexed), "d3036" -> 0.
fn extract_ordinal(id: &str) -> usize {
    if let Some(pos) = id.rfind('_') {
        let suffix = &id[pos + 1..];
        if let Ok(n) = suffix.parse::<usize>() {
            // Ordinal suffixes are 2-based (_2 = 2nd occurrence = index 1)
            return n.saturating_sub(1);
        }
    }
    0
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::Path;

    #[test]
    fn test_to_pascal_case() {
        assert_eq!(to_pascal_case("HAUSHALTSKUNDE"), "Haushaltskunde");
        assert_eq!(to_pascal_case("HAUSHALTSKUNDE_ENWG"), "HaushaltskundeEnwg");
        assert_eq!(to_pascal_case("EMAIL"), "Email");
    }

    #[test]
    fn test_parse_toml_field_path() {
        let (tag, comp, elem) = parse_toml_field_path("loc.c517.d3225").unwrap();
        assert_eq!(tag, "loc");
        assert_eq!(comp, "c517");
        assert_eq!(elem.as_deref(), Some("d3225"));

        let (tag, elem, none) = parse_toml_field_path("nad.d3035").unwrap();
        assert_eq!(tag, "nad");
        assert_eq!(elem, "d3035");
        assert!(none.is_none());

        let (tag, comp, elem) = parse_toml_field_path("cav[Z91].c889.d7111").unwrap();
        assert_eq!(tag, "cav");
        assert_eq!(comp, "c889");
        assert_eq!(elem.as_deref(), Some("d7111"));
    }

    #[test]
    fn test_normalize_edifact_id() {
        assert_eq!(normalize_edifact_id("c517"), "C517");
        assert_eq!(normalize_edifact_id("d3225"), "3225");
        assert_eq!(normalize_edifact_id("d3036_2"), "3036");
        assert_eq!(normalize_edifact_id("c556_2"), "C556");
    }

    #[test]
    fn test_extract_ordinal() {
        assert_eq!(extract_ordinal("d3036"), 0);
        assert_eq!(extract_ordinal("d3036_2"), 1);
        assert_eq!(extract_ordinal("c556_3"), 2);
    }

    /// A qualified field path (`rff[ACW]`) resolves against its own segment
    /// variant: RFF+ACW's reference number is free data, not RFF+Z13's PID code.
    #[test]
    fn qualified_field_path_uses_its_own_segment_variant() {
        let rff = |qual: &str, id: Value| {
            serde_json::json!({"id": "RFF", "elements": [{"index": 0, "composite": "C506", "components": [
                {"sub_index": 0, "id": "1153", "type": "code", "ahb_status": "X",
                 "codes": [{"value": qual, "name": qual, "ahb_status": "X"}]},
                id,
            ]}]})
        };
        let data = serde_json::json!({"sub_index": 1, "id": "1154", "type": "data", "ahb_status": "X", "format": "an..70"});
        let schema = serde_json::json!({"pid": "21037", "beschreibung": "", "fields": {"sg14": {
        "ahb_status": "Muss", "segments": [], "children": {"sg15": {"ahb_status": "Muss", "segments": [
            rff("Z13", serde_json::json!({"sub_index": 1, "id": "1154", "type": "code", "ahb_status": "X",
                "codes": [{"value": "21037", "name": "RD / NB-Bewertung", "ahb_status": "X"}]})),
            rff("ACW", data.clone()),
            rff("ACE", data),
        ]}}}}});
        let def = MappingDefinition::from_toml_str(
            r#"
[meta]
entity = "Status"
bo4e_type = "Status"
source_group = "SG14.SG15"
source_path = "sg14.sg15"

[fields]
"rff[Z13].c506.d1154" = "pruefidentifikator"
"rff[ACW].c506.d1153" = "acwQualifier"
"rff[ACW].c506.d1154" = "referenz"
"rff[ACE].0.1" = "gegenvorschlagReferenz"
"#,
        )
        .unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &[def]);
        let status = reqs.entities.iter().find(|e| e.entity == "Status").unwrap();
        let field = |name: &str| status.fields.iter().find(|f| f.bo4e_name == name).unwrap();
        let codes = |name: &str| {
            field(name)
                .valid_codes
                .iter()
                .map(|c| c.code.as_str())
                .collect::<Vec<_>>()
        };
        for free in ["referenz", "gegenvorschlagReferenz"] {
            assert_eq!(field(free).field_type, "data", "{free}");
            assert!(codes(free).is_empty(), "{free}: {:?}", codes(free));
        }
        assert_eq!(codes("pruefidentifikator"), ["21037"]);
        assert_eq!(codes("acwQualifier"), ["ACW"]);
    }

    #[test]
    fn test_from_schema_and_definitions_pid_55001() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        assert!(!defs.is_empty(), "should have loaded definitions");

        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Basic metadata
        assert_eq!(reqs.pid, "55001");
        assert_eq!(reqs.beschreibung, "Anmeldung verb. MaLo");
        assert!(!reqs.entities.is_empty(), "should have entities");

        // Prozessdaten entity with pruefidentifikator field
        let prozessdaten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Prozessdaten")
            .expect("Prozessdaten entity should exist");

        let pruefid = prozessdaten
            .fields
            .iter()
            .find(|f| f.bo4e_name == "pruefidentifikator")
            .expect("pruefidentifikator field should exist");
        assert_eq!(pruefid.ahb_status, "X");
        assert_eq!(pruefid.field_type, "code");
        assert!(
            pruefid.valid_codes.iter().any(|c| c.code == "55001"),
            "should have 55001 as valid code"
        );

        // Marktlokation entity should exist with marktlokationsId field
        let marktlokation = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Marktlokation")
            .expect("Marktlokation entity should exist");
        assert!(
            marktlokation
                .fields
                .iter()
                .any(|f| f.bo4e_name == "marktlokationsId"),
            "Marktlokation should have marktlokationsId field"
        );

        // SG10 CCI/CAV data is merged into the entity of its SG8 parent
        // variant — for Z01 ("Daten der Marktlokation") that is
        // MarktlokationDaten, not the SG5 LOC entity.
        let marktlokation_daten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "MarktlokationDaten")
            .expect("MarktlokationDaten entity should exist");
        let malo_code_fields: Vec<_> = marktlokation_daten
            .fields
            .iter()
            .filter(|f| f.field_type == "code" && f.valid_codes.len() >= 2)
            .collect();
        assert!(
            !malo_code_fields.is_empty(),
            "MarktlokationDaten should have code fields with >= 2 valid codes (haushaltskunde), fields: {:?}",
            marktlokation_daten
                .fields
                .iter()
                .map(|f| (&f.bo4e_name, &f.field_type, f.valid_codes.len()))
                .collect::<Vec<_>>()
        );

        // Geschaeftspartner should be detected as array (PID 55001 has SG12_Z04)
        let geschaeftspartner = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        // PID 55001 may have only 1 SG12 variant, so max_reps depends on schema
        // At minimum, the entity should exist
        assert!(
            geschaeftspartner.is_some(),
            "Geschaeftspartner entity should exist"
        );
    }

    #[test]
    fn test_extract_valid_codes_includes_enum_name() {
        let element = serde_json::json!({
            "codes": [
                {"value": "ZF9", "name": "Der Code ist anzuwenden wenn...", "enum": "ENFG_VERRINGERUNG_UMLAGE"},
                {"value": "ZG0", "name": "Der Code ist anzuwenden wenn...", "enum": "ENFG_KEINE_VERRINGERUNG_UMLAGE"},
                {"value": "ZG1", "name": "Der Code ist anzuwenden wenn..."}
            ]
        });

        let codes = extract_valid_codes(&element);
        assert_eq!(codes.len(), 3);

        assert_eq!(codes[0].code, "ZF9");
        assert_eq!(
            codes[0].enum_name.as_deref(),
            Some("ENFG_VERRINGERUNG_UMLAGE")
        );

        assert_eq!(codes[1].code, "ZG0");
        assert_eq!(
            codes[1].enum_name.as_deref(),
            Some("ENFG_KEINE_VERRINGERUNG_UMLAGE")
        );

        assert_eq!(codes[2].code, "ZG1");
        assert_eq!(codes[2].enum_name, None, "missing enum should be None");
    }

    #[test]
    fn test_enum_name_roundtrips_through_pid_requirements() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let defs = load_definitions_for_pid(
            &base.join("common"),
            &base.join("pid_55001"),
            &base.join("message"),
            &schema,
        )
        .unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Find any code field with valid_codes — its enum_name should be populated
        let code_field = reqs
            .entities
            .iter()
            .flat_map(|e| &e.fields)
            .find(|f| f.field_type == "code" && !f.valid_codes.is_empty())
            .expect("should have at least one code field");

        // At least one code should have enum_name set (schema has "enum" keys)
        let has_enum_name = code_field.valid_codes.iter().any(|c| c.enum_name.is_some());
        assert!(
            has_enum_name,
            "code field {:?} should have at least one code with enum_name set, codes: {:?}",
            code_field.bo4e_name, code_field.valid_codes
        );
    }

    #[test]
    fn test_pid_55001_max_reps_from_group_json() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Schema file not found, skipping test");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let defs = load_definitions_for_pid(
            &base.join("common"),
            &base.join("pid_55001"),
            &base.join("message"),
            &schema,
        )
        .unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // ProduktpaketDaten comes from SG8 with max_reps=99999 → should be array
        let produktpaket = reqs
            .entities
            .iter()
            .find(|e| e.entity == "ProduktpaketDaten");
        assert!(produktpaket.is_some(), "ProduktpaketDaten should exist");
        assert!(
            produktpaket.unwrap().cardinality().is_list(),
            "ProduktpaketDaten should have max_reps > 1 (SG8 max_reps=99999)"
        );

        // Geschaeftspartner should still have max_reps > 1 (SG12 max_reps from MIG)
        let gp = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        if let Some(gp) = gp {
            // SG12 max_reps depends on the MIG; verify it's set correctly
            assert!(
                gp.cardinality().is_list(),
                "Geschaeftspartner should have max_reps > 1 (SG12 repeats)"
            );
        }

        // Fields from SG10 (child group) should have child_group metadata
        let produktpaket = reqs
            .entities
            .iter()
            .find(|e| e.entity == "ProduktpaketDaten")
            .unwrap();
        let sg10_field = produktpaket.fields.iter().find(|f| {
            f.bo4e_name.contains("produktMerkmal") || f.bo4e_name.contains("produktWert")
        });
        assert!(
            sg10_field.is_some(),
            "ProduktpaketDaten should have SG10-sourced fields"
        );
        let cg = &sg10_field.unwrap().child_group;
        assert!(
            cg.is_some(),
            "SG10-sourced field should have child_group set, field: {:?}",
            sg10_field.unwrap().bo4e_name
        );
        assert_eq!(cg.as_ref().unwrap().name, "sg10");
    }

    #[test]
    fn test_entity_map_key_info_serde_roundtrip() {
        let info = EntityMapKeyInfo {
            field: "marktrolle".to_string(),
            values: vec![
                EntityMapKeyValue {
                    code: "MS".to_string(),
                    name: "Sender".to_string(),
                },
                EntityMapKeyValue {
                    code: "MR".to_string(),
                    name: "Recipient".to_string(),
                },
            ],
        };
        let json = serde_json::to_string(&info).unwrap();
        let roundtripped: EntityMapKeyInfo = serde_json::from_str(&json).unwrap();
        assert_eq!(roundtripped.field, "marktrolle");
        assert_eq!(roundtripped.values.len(), 2);
        assert_eq!(roundtripped.values[0].code, "MS");
        assert_eq!(roundtripped.values[1].code, "MR");
    }

    #[test]
    fn test_entity_requirement_map_key_skip_serializing_if_none() {
        let req = EntityRequirement {
            entity: "Test".to_string(),
            ref_type: Bo4eRefType::Object {
                type_name: "Test".to_string(),

                cardinality: Cardinality::OPTIONAL,
            },

            ahb_status: "X".to_string(),
            fields: vec![],
            variants: vec![],
            groups: vec![],
            map_key: None,
            scope: EntityScope::Transaction,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(
            !json.contains("map_key"),
            "map_key should be omitted when None: {json}"
        );
    }

    #[test]
    fn test_entity_requirement_map_key_serialized_when_present() {
        let req = EntityRequirement {
            entity: "Marktteilnehmer".to_string(),
            ref_type: Bo4eRefType::Object {
                type_name: "Marktteilnehmer".to_string(),

                cardinality: Cardinality {
                    min: 1,
                    max: Some(2),
                },
            },

            ahb_status: "X".to_string(),
            fields: vec![],
            variants: vec![],
            groups: vec![],
            map_key: Some(EntityMapKeyInfo {
                field: "marktrolle".to_string(),
                values: vec![
                    EntityMapKeyValue {
                        code: "MS".to_string(),
                        name: "Sender".to_string(),
                    },
                    EntityMapKeyValue {
                        code: "MR".to_string(),
                        name: "Recipient".to_string(),
                    },
                ],
            }),
            scope: EntityScope::Transaction,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(
            json.contains("map_key"),
            "map_key should be present: {json}"
        );
        assert!(
            json.contains("marktrolle"),
            "field name should be present: {json}"
        );
        assert!(json.contains("MS"), "MS code should be present: {json}");
        assert!(json.contains("MR"), "MR code should be present: {json}");

        // Roundtrip
        let roundtripped: EntityRequirement = serde_json::from_str(&json).unwrap();
        let mk = roundtripped.map_key.unwrap();
        assert_eq!(mk.field, "marktrolle");
        assert_eq!(mk.values.len(), 2);
    }

    #[test]
    fn test_pid_55001_marktteilnehmer_has_map_key() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Skipping: schema not found");
            return;
        }
        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Marktteilnehmer should have a map_key with marktrolle (MS + MR)
        let mt = reqs.entities.iter().find(|e| e.entity == "Marktteilnehmer");
        assert!(mt.is_some(), "should have Marktteilnehmer entity");
        let mt = mt.unwrap();
        assert!(
            mt.cardinality().is_list(),
            "Marktteilnehmer should have max_reps > 1"
        );
        assert!(
            mt.map_key.is_some(),
            "Marktteilnehmer should have map_key for marktrolle"
        );
        let mk = mt.map_key.as_ref().unwrap();
        assert_eq!(mk.field, "marktrolle");
        assert!(
            mk.values.len() >= 2,
            "Should have at least MS and MR, got {} values",
            mk.values.len()
        );
        let codes: Vec<&str> = mk.values.iter().map(|v| v.code.as_str()).collect();
        assert!(
            codes.contains(&"MS"),
            "Should contain MS code, got: {:?}",
            codes
        );
        assert!(
            codes.contains(&"MR"),
            "Should contain MR code, got: {:?}",
            codes
        );
    }

    #[test]
    fn test_detect_entity_map_key_geschaeftspartner_sg12() {
        // PID 55013 has 7 SG12 variants under SG4: sg12_z63, sg12_z65..sg12_z70
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2504/utilmd/pids/pid_55013_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("Skipping: schema not found");
            return;
        }
        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2504/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55013");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Geschaeftspartner should be an array entity with a map_key for nad_qualifier
        let gp = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Geschaeftspartner");
        assert!(gp.is_some(), "should have Geschaeftspartner entity");
        let gp = gp.unwrap();
        assert!(
            gp.cardinality().is_list(),
            "Geschaeftspartner should have max_reps > 1"
        );
        assert!(
            gp.map_key.is_some(),
            "Geschaeftspartner should have map_key for nad_qualifier"
        );
        let mk = gp.map_key.as_ref().unwrap();
        assert_eq!(mk.field, "nad_qualifier");

        // PID 55013 has 7 NAD qualifiers: Z63, Z65, Z66, Z67, Z68, Z69, Z70
        let codes: BTreeSet<&str> = mk.values.iter().map(|v| v.code.as_str()).collect();
        let expected = ["Z63", "Z65", "Z66", "Z67", "Z68", "Z69", "Z70"];
        for code in &expected {
            assert!(
                codes.contains(code),
                "Should contain {code} code, got: {:?}",
                codes
            );
        }
        assert_eq!(
            codes.len(),
            expected.len(),
            "Should have exactly {} codes, got {:?}",
            expected.len(),
            codes
        );
    }

    /// Regression test for issue #52: validate PID 55001 FV2604 requirements
    /// to catch false positives (transaktionsnummer, Kontakt, namenszusatz).
    #[test]
    fn test_issue_52_pid_55001_fv2604_false_positives() {
        let schema_path = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../mig-types/src/generated/fv2604/utilmd/pids/pid_55001_schema.json"
        ));
        if !schema_path.exists() {
            eprintln!("FV2604 schema not found, skipping");
            return;
        }

        let schema: Value =
            serde_json::from_str(&std::fs::read_to_string(schema_path).unwrap()).unwrap();

        let base = Path::new(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../mappings/FV2604/UTILMD_Strom"
        ));
        let common_dir = base.join("common");
        let pid_dir = base.join("pid_55001");
        let message_dir = base.join("message");

        let defs = load_definitions_for_pid(&common_dir, &pid_dir, &message_dir, &schema).unwrap();
        let reqs = PidRequirements::from_schema_and_definitions(&schema, &defs);

        // Issue #52 Q1: transaktionsnummer should NOT be in PID 55001 requirements.
        // PID 55001 SG6 only has RFF+Z13, not RFF+TN. The common _21_rff_tn.toml
        // discriminator (RFF.c506.d1153=TN) should be filtered out.
        let prozessdaten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Prozessdaten")
            .expect("Prozessdaten should exist");
        assert!(
            !prozessdaten
                .fields
                .iter()
                .any(|f| f.bo4e_name == "transaktionsnummer"),
            "transaktionsnummer should NOT be in PID 55001 (SG6 has only RFF+Z13, not RFF+TN)"
        );

        // Issue #52 Q2: Kontakt is a message-level entity (SG2.SG3).
        // When validate_pid receives only transaction-level data, it should not
        // require Kontakt. Verify it's scoped as Message.
        let kontakt = reqs.entities.iter().find(|e| e.entity == "Kontakt");
        if let Some(kontakt) = kontakt {
            assert_eq!(
                kontakt.scope,
                EntityScope::Message,
                "Kontakt should be Message-scoped (SG2.SG3), not Transaction"
            );
        }

        // Verify that validate_pid_json_transaction skips message-level entities
        let tx_errors =
            crate::pid_validation::validate_pid_json_transaction(&serde_json::json!({}), &reqs);
        assert!(
            !tx_errors.iter().any(|e| matches!(
                e,
                crate::pid_validation::PidValidationError::MissingEntity { entity, .. }
                if entity == "Kontakt"
            )),
            "Transaction-scoped validation should NOT report missing Kontakt"
        );

        // Issue #52 Q3: namenszusatz1/namenszusatz2 should NOT be required.
        // These C080 components (sub_index 2,3) have no explicit ahb_status
        // in the schema and should not inherit "X" from the parent group.
        let ansprechpartner = reqs
            .entities
            .iter()
            .find(|e| e.entity == "Ansprechpartner")
            .expect("Ansprechpartner should exist");
        for name in &["namenszusatz1", "namenszusatz2"] {
            if let Some(field) = ansprechpartner.fields.iter().find(|f| f.bo4e_name == *name) {
                assert!(
                    !matches!(field.ahb_status.as_str(), "X" | "Muss" | "Soll"),
                    "{name} should NOT be required (has no explicit AHB status in schema), \
                     but got ahb_status={:?}",
                    field.ahb_status
                );
            }
        }

        // Issue #52 Q4: haushaltskunde IS legitimately required. It comes
        // from the SG10 of the SG8 `SEQ+Z01` variant ("Daten der
        // Marktlokation"), which is its own entity, off the SG5 LOC.
        let marktlokation_daten = reqs
            .entities
            .iter()
            .find(|e| e.entity == "MarktlokationDaten")
            .expect("MarktlokationDaten should exist");
        let haushaltskunde = marktlokation_daten
            .fields
            .iter()
            .find(|f| f.bo4e_name == "haushaltskunde");
        assert!(
            haushaltskunde.is_some(),
            "haushaltskunde field should exist on MarktlokationDaten"
        );
    }

    #[test]
    fn deserializes_legacy_entity_requirement_shape() {
        // Pre-cardinality-unification cache files have the old field set
        // (bo4e_type / min_reps / max_reps). Make sure those still load and
        // get folded into the new Bo4eRefType::Object shape.
        let json = serde_json::json!({
            "entity": "Marktlokation",
            "bo4e_type": "Marktlokation",
            "ahb_status": "Muss",
            "min_reps": 1,
            "max_reps": 1,
            "fields": [],
            "scope": "Transaction"
        });
        let req: EntityRequirement = serde_json::from_value(json).unwrap();
        assert_eq!(req.bo4e_type(), "Marktlokation");
        assert_eq!(req.cardinality(), Cardinality::REQUIRED);
        match &req.ref_type {
            Bo4eRefType::Object {
                type_name,
                cardinality,
            } => {
                assert_eq!(type_name, "Marktlokation");
                assert_eq!(*cardinality, Cardinality::REQUIRED);
            }
            other => panic!("expected Object, got {other:?}"),
        }
    }

    #[test]
    fn legacy_entity_with_unbounded_max_reps_becomes_unbounded_cardinality() {
        let json = serde_json::json!({
            "entity": "Geschaeftspartner",
            "bo4e_type": "Geschaeftspartner",
            "ahb_status": "Kann",
            "min_reps": 0,
            "max_reps": 99999,
            "fields": []
        });
        let req: EntityRequirement = serde_json::from_value(json).unwrap();
        assert!(req.cardinality().unbounded());
        assert_eq!(req.cardinality().min, 0);
    }

    /// Simple (non-composite) elements must union their codes when group variants
    /// are merged, like composite components already do (issue #104: DE3035 of
    /// NAD+Z03 and NAD+Z07 used to keep only Z03).
    #[test]
    fn merge_segment_elements_unions_simple_element_codes() {
        let nad = |code: &str, extra: Option<&str>| {
            let mut elements = vec![serde_json::json!({
                "id": "3035", "index": 0, "type": "code",
                "codes": [{ "value": code, "name": format!("name {code}") }]
            })];
            if let Some(id) = extra {
                elements.push(serde_json::json!({ "id": id, "index": 5, "type": "data" }));
            }
            serde_json::json!({ "id": "NAD", "elements": elements })
        };
        let mut target = nad("Z03", Some("3164"));
        merge_segment_elements(&mut target, &nad("Z07", None));
        merge_segment_elements(&mut target, &nad("Z03", None));

        let elements = target["elements"].as_array().unwrap();
        assert_eq!(elements.len(), 2, "no duplicate DE3035: {elements:?}");
        let codes: Vec<&str> = elements[0]["codes"]
            .as_array()
            .unwrap()
            .iter()
            .map(|c| c["value"].as_str().unwrap())
            .collect();
        assert_eq!(codes, ["Z03", "Z07"]);
    }

    fn load_55042_requirements(fv: &str, resolve_paths: bool) -> PidRequirements {
        let schema_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join(format!(
            "../mig-types/src/generated/{}/utilmd/pids",
            fv.to_lowercase()
        ));
        let schema: Value = serde_json::from_str(
            &std::fs::read_to_string(schema_dir.join("pid_55042_schema.json")).unwrap(),
        )
        .unwrap();
        let base =
            Path::new(env!("CARGO_MANIFEST_DIR")).join(format!("../../mappings/{fv}/UTILMD_Strom"));
        let mut defs = load_definitions_for_pid(
            &base.join("common"),
            &base.join("pid_55042"),
            &base.join("message"),
            &schema,
        )
        .unwrap();
        // The definitions name shared tables; the requirements are built from
        // the same tables, or a translated value reads as an invalid code.
        let code_lists = crate::code_lists::CodeLists::discover(&base);
        if resolve_paths {
            // compile-mappings resolves EDIFACT ID paths to numeric paths first.
            let resolver = crate::path_resolver::PathResolver::from_schema_dir(&schema_dir);
            defs = MappingEngine::from_definitions_with_code_lists(
                std::sync::Arc::clone(&code_lists),
                defs,
            )
            .with_path_resolver(resolver)
            .definitions()
            .to_vec();
        }
        PidRequirements::from_schema_definitions_and_code_lists(&schema, &defs, &code_lists)
    }

    /// Issue #104: PID 55042 maps Geschaeftspartner from sg12_z03/z05/z07/z08 via a
    /// single `source_path = "sg4.sg12"` TOML. The entity-level `partnerrolle` must
    /// allow all four codes (with their enum_map names), and per-variant
    /// requirements must keep each group's own required fields apart.
    #[test]
    fn issue_104_pid_55042_geschaeftspartner_variants() {
        for fv in ["FV2604", "FV2610"] {
            for resolve_paths in [false, true] {
                let ctx = format!("{fv} resolve_paths={resolve_paths}");
                let reqs = load_55042_requirements(fv, resolve_paths);
                let gp = reqs
                    .entities
                    .iter()
                    .find(|e| e.entity == "Geschaeftspartner")
                    .unwrap_or_else(|| panic!("{ctx}: Geschaeftspartner requirement"));

                let partnerrolle = gp
                    .fields
                    .iter()
                    .find(|f| f.bo4e_name == "partnerrolle")
                    .unwrap();
                let codes: Vec<(&str, Option<&str>)> = partnerrolle
                    .valid_codes
                    .iter()
                    .map(|c| (c.code.as_str(), c.bo4e_value.as_deref()))
                    .collect();
                assert_eq!(
                    codes,
                    [
                        ("Z03", Some("messlokationsadresse")),
                        ("Z05", Some("ablesekarte")),
                        ("Z07", Some("kundeMsb")),
                        ("Z08", Some("korrespondenzKundeMsb")),
                    ],
                    "{ctx}"
                );

                let variant = |code: &str| {
                    gp.variants
                        .iter()
                        .find(|v| v.code == code)
                        .unwrap_or_else(|| panic!("{ctx}: variant {code}: {:?}", gp.variants))
                };
                assert_eq!(gp.variants.len(), 4, "{ctx}");
                let z07 = variant("Z07");
                assert_eq!(z07.discriminator_field, "partnerrolle", "{ctx}");
                assert_eq!(z07.bo4e_value.as_deref(), Some("kundeMsb"), "{ctx}");
                assert_eq!(z07.source_paths, ["sg4.sg12_z07"], "{ctx}");
                let z07_names: Vec<&str> =
                    z07.fields.iter().map(|f| f.bo4e_name.as_str()).collect();
                assert!(z07_names.contains(&"name1"), "{ctx}: {z07_names:?}");
                assert!(
                    !z07_names.contains(&"adresse.ort"),
                    "{ctx}: NAD+Z07 carries no address: {z07_names:?}"
                );
                let z07_role = z07
                    .fields
                    .iter()
                    .find(|f| f.bo4e_name == "partnerrolle")
                    .unwrap();
                assert_eq!(z07_role.valid_codes.len(), 1, "{ctx}");
                assert_eq!(z07_role.valid_codes[0].code, "Z07", "{ctx}");

                let z03 = variant("Z03");
                let z03_ort = z03
                    .fields
                    .iter()
                    .find(|f| f.bo4e_name == "adresse.ort")
                    .unwrap_or_else(|| panic!("{ctx}: Z03 has adresse.ort"));
                assert_eq!(z03_ort.ahb_status, "X", "{ctx}");
                assert!(
                    !z03.fields.iter().any(|f| f.bo4e_name == "name1"),
                    "{ctx}: NAD+Z03 has no name"
                );
            }
        }
    }

    /// An element marked `X` inside a segment the AHB marks `Kann` is required
    /// only when the segment is present, so the requirement must not be
    /// unconditional: a message that omits the optional segment omits the
    /// element legitimately.
    ///
    /// 55042's SG12 is the live case. Its `RFF` is `Kann` and `D_1154` is
    /// `X [951]`; `[951]` is a *format* condition that always evaluates true, so
    /// the condition-aware validator reported a hard "missing
    /// Geschaeftspartner.referenz" for every 55042 message.
    #[test]
    fn a_required_element_of_an_optional_segment_is_not_required() {
        for fv in ["FV2604", "FV2610"] {
            for resolve_paths in [false, true] {
                let ctx = format!("{fv} resolve_paths={resolve_paths}");
                let reqs = load_55042_requirements(fv, resolve_paths);
                let gp = reqs
                    .entities
                    .iter()
                    .find(|e| e.entity == "Geschaeftspartner")
                    .unwrap_or_else(|| panic!("{ctx}: Geschaeftspartner requirement"));

                let referenz = gp
                    .fields
                    .iter()
                    .find(|f| f.bo4e_name == "messlokationsId")
                    .unwrap_or_else(|| panic!("{ctx}: the SG12 RFF is mapped"));
                assert_eq!(
                    referenz.ahb_status, "Kann [951]",
                    "{ctx}: RFF is Kann, so its X component is conditional on the \
                     segment; the format condition is kept"
                );

                // The NAD of the same group is `Muss`, so its X components stay
                // required — the relaxation must not leak across segments.
                let name = gp.fields.iter().find(|f| f.bo4e_name == "name1").unwrap();
                assert_eq!(name.ahb_status, "X", "{ctx}");
            }
        }
    }

    #[test]
    fn the_relaxation_reads_the_segment_the_field_belongs_to() {
        let group = serde_json::json!({
            "segments": [
                { "id": "NAD", "ahb_status": "Muss", "elements": [] },
                { "id": "RFF", "ahb_status": "Kann", "elements": [] },
                { "id": "DTM", "elements": [] },
            ]
        });
        let mut status = "X [951]".to_string();
        relax_status_inside_optional_segment(&group, "rff", None, &mut status);
        assert_eq!(status, "Kann [951]");

        let mut status = "Muss".to_string();
        relax_status_inside_optional_segment(&group, "nad", None, &mut status);
        assert_eq!(status, "Muss", "a Muss segment changes nothing");

        // No `ahb_status` on the segment means "not stated here" (the message
        // header segments carry none and are mandatory), not "optional".
        let mut status = "X".to_string();
        relax_status_inside_optional_segment(&group, "dtm", None, &mut status);
        assert_eq!(status, "X");

        // Nothing to say about a status that was not required to begin with.
        let mut status = "Kann".to_string();
        relax_status_inside_optional_segment(&group, "rff", None, &mut status);
        assert_eq!(status, "Kann");
    }

    /// Serialized requirements without variants must not grow a `variants` key, and
    /// caches written before variants existed must still load.
    #[test]
    fn entity_requirement_variants_are_optional_in_serde() {
        let json = serde_json::json!({
            "entity": "Geschaeftspartner",
            "ahb_status": "Muss",
            "fields": [],
        });
        let req: EntityRequirement = serde_json::from_value(json).unwrap();
        assert!(req.variants.is_empty());
        let out = serde_json::to_string(&req).unwrap();
        assert!(!out.contains("variants"), "{out}");
        assert!(!out.contains("bo4e_value"), "{out}");
    }
}