mig_bo4e/

engine.rs

//! Mapping engine — loads TOML definitions and provides bidirectional conversion.
//!
//! Supports nested group paths (e.g., "SG4.SG5") for navigating the assembled tree
//! and provides `map_forward` / `map_reverse` for full entity conversion.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::Path;

use mig_assembly::assembler::{
    AssembledGroup, AssembledGroupInstance, AssembledSegment, AssembledTree,
};
use mig_types::schema::mig::MigSchema;
use mig_types::segment::OwnedSegment;

use crate::definition::{FieldMapping, MappingDefinition};
use crate::error::MappingError;
use crate::segment_structure::SegmentStructure;

/// The mapping engine holds all loaded mapping definitions
/// and provides methods for bidirectional conversion.
pub struct MappingEngine {
    definitions: Vec<MappingDefinition>,
    segment_structure: Option<SegmentStructure>,
    code_lookup: Option<crate::code_lookup::CodeLookup>,
    /// Transaction-root SG id (e.g. "SG4" for UTILMD), when the engine is
    /// operating at transaction scope. Child entities whose parent group
    /// equals this id are left at the top level of the forward-mapped JSON
    /// rather than being nested — SG4 is the transaction envelope, so
    /// entities inside it (Marktlokation, Geschaeftspartner, …) are peers of
    /// the transaction metadata, not sub-objects of it.
    ///
    /// Nesting still applies to other parent groups: e.g. Kontakt (SG2.SG3)
    /// remains nested under Marktteilnehmer (SG2) because SG2 is a
    /// message-level group, not the transaction root.
    transaction_group: Option<String>,
    /// PID currently being processed (e.g., "55002"). Used to suppress codelist
    /// decoration of self-referential PID-identifier fields (e.g., RFF+Z13's
    /// d1154 in PID 55002 has only "55002" as an allowed value).
    current_pid: Option<String>,
    /// The shared code-list tables a definition's `code_list` names. One `Arc`
    /// per mappings tree, shared by every engine built from it -- a format
    /// version builds a couple of thousand engines and the tables are the same
    /// for all of them.
    code_lists: std::sync::Arc<crate::code_lists::CodeLists>,
}

impl MappingEngine {
    /// Create an empty engine with no definitions (for unit testing).
    pub fn new_empty() -> Self {
        Self {
            definitions: Vec::new(),
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: std::sync::Arc::new(crate::code_lists::CodeLists::default()),
        }
    }

    /// Load all TOML mapping files from a directory.
    pub fn load(dir: &Path) -> Result<Self, MappingError> {
        let mut definitions = Vec::new();

        let mut entries: Vec<_> = std::fs::read_dir(dir)?.filter_map(|e| e.ok()).collect();
        entries.sort_by_key(|e| e.file_name());

        for entry in entries {
            let path = entry.path();
            if path.extension().map(|e| e == "toml").unwrap_or(false) {
                let content = std::fs::read_to_string(&path)?;
                let def = MappingDefinition::from_toml_str(&content).map_err(|message| {
                    MappingError::TomlParse {
                        file: path.display().to_string(),
                        message,
                    }
                })?;
                definitions.push(def);
            }
        }

        // Emission order comes from `meta.order` where a definition states it,
        // and from the filename otherwise — which is what every file relies on
        // today, via the `_30_12_` prefix convention. `sort_by_key` is stable,
        // so definitions without the key keep their filename order exactly, and
        // `u32::MAX` puts them after any that opt in.
        definitions.sort_by_key(|d| d.meta.order.unwrap_or(u32::MAX));

        Ok(Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: crate::code_lists::CodeLists::discover(dir),
        })
    }

    /// Load message-level and transaction-level TOML mappings from separate directories.
    ///
    /// Returns `(message_engine, transaction_engine)` where:
    /// - `message_engine` maps SG2/SG3/root-level definitions (shared across PIDs)
    /// - `transaction_engine` maps SG4+ definitions (PID-specific)
    pub fn load_split(
        message_dir: &Path,
        transaction_dir: &Path,
    ) -> Result<(Self, Self), MappingError> {
        let msg_engine = Self::load(message_dir)?;
        let tx_engine = Self::load(transaction_dir)?;
        Ok((msg_engine, tx_engine))
    }

    /// Load TOML mapping files from multiple directories into a single engine.
    ///
    /// Useful for combining message-level and transaction-level mappings
    /// when a single engine with all definitions is needed.
    pub fn load_merged(dirs: &[&Path]) -> Result<Self, MappingError> {
        let mut definitions = Vec::new();
        for dir in dirs {
            let engine = Self::load(dir)?;
            definitions.extend(engine.definitions);
        }
        Ok(Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: crate::code_lists::CodeLists::discover(
                dirs.first().copied().unwrap_or(Path::new("")),
            ),
        })
    }

    /// Load transaction-level mappings with common template inheritance.
    ///
    /// 1. Loads all `.toml` from `common_dir`
    /// 2. Filters: keeps only definitions whose `source_path` exists in the PID schema
    /// 3. Loads all `.toml` from `pid_dir`
    /// 4. For each PID definition, if a common definition has matching
    ///    `(source_group, discriminator)`, replaces the common one (file-level replacement)
    /// 5. Merges both sets: common first, then PID additions
    pub fn load_with_common(
        common_dir: &Path,
        pid_dir: &Path,
        schema_index: &crate::pid_schema_index::PidSchemaIndex,
    ) -> Result<Self, MappingError> {
        let mut common_defs = Self::load(common_dir)?.definitions;

        // Filter common defs by schema — keep only groups that exist in this PID
        common_defs.retain(|d| {
            d.meta
                .source_path
                .as_deref()
                .map(|sp| schema_index.has_group(sp))
                .unwrap_or(true)
        });

        let pid_defs = Self::load(pid_dir)?.definitions;

        // Build set of PID override keys: (source_group_normalized, discriminator)
        // Normalizations applied:
        // 1. Strip positional indices from source_group: "SG4.SG5:1" → "SG4.SG5"
        // 2. Strip occurrence indices from discriminator: "RFF.c506.d1153=TN#0" → "RFF.c506.d1153=TN"
        let normalize_sg = |sg: &str| -> String {
            sg.split('.')
                .map(|part| part.split(':').next().unwrap_or(part))
                .collect::<Vec<_>>()
                .join(".")
        };
        let pid_keys: HashSet<(String, Option<String>)> = pid_defs
            .iter()
            .flat_map(|d| {
                let sg = normalize_sg(&d.meta.source_group);
                let disc = d.meta.discriminator.clone();
                let mut keys = vec![(sg.clone(), disc.clone())];
                // If discriminator has occurrence index (#N), also add base form
                if let Some(ref disc_str) = disc {
                    if let Some(base) = disc_str.rsplit_once('#') {
                        if base.1.chars().all(|c| c.is_ascii_digit()) {
                            keys.push((sg, Some(base.0.to_string())));
                        }
                    }
                }
                keys
            })
            .collect();

        // Remove common defs that are overridden by PID defs
        common_defs.retain(|d| {
            let key = (
                normalize_sg(&d.meta.source_group),
                d.meta.discriminator.clone(),
            );
            !pid_keys.contains(&key)
        });

        // Combine: common first, then PID
        let mut definitions = common_defs;
        definitions.extend(pid_defs);

        Ok(Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: crate::code_lists::CodeLists::discover(pid_dir),
        })
    }

    /// Load common definitions only (no per-PID dir), filtered by schema index.
    ///
    /// Used for PIDs that have no per-PID directory but can use shared common/ definitions.
    pub fn load_common_only(
        common_dir: &Path,
        schema_index: &crate::pid_schema_index::PidSchemaIndex,
    ) -> Result<Self, MappingError> {
        let mut common_defs = Self::load(common_dir)?.definitions;

        // Filter common defs by schema — keep only groups that exist in this PID
        common_defs.retain(|d| {
            d.meta
                .source_path
                .as_deref()
                .map(|sp| schema_index.has_group(sp))
                .unwrap_or(true)
        });

        Ok(Self {
            definitions: common_defs,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: crate::code_lists::CodeLists::discover(common_dir),
        })
    }

    /// Load message + transaction engines with common template inheritance.
    ///
    /// Returns `(message_engine, transaction_engine)` where the transaction engine
    /// inherits shared templates from `common_dir`, filtered by the PID schema.
    pub fn load_split_with_common(
        message_dir: &Path,
        common_dir: &Path,
        transaction_dir: &Path,
        schema_index: &crate::pid_schema_index::PidSchemaIndex,
    ) -> Result<(Self, Self), MappingError> {
        let msg_engine = Self::load(message_dir)?;
        let tx_engine = Self::load_with_common(common_dir, transaction_dir, schema_index)?;
        Ok((msg_engine, tx_engine))
    }

    /// Create an engine from an already-parsed list of definitions.
    /// Whether any definition names a shared code list.
    fn names_a_code_list(definitions: &[MappingDefinition]) -> bool {
        definitions.iter().any(|d| {
            d.fields.values().any(|f| {
                matches!(f, FieldMapping::Structured(s)
                    if s.code_list.is_some() || s.also_code_list.is_some())
            })
        })
    }

    /// The table a structured mapping translates through: its own inline
    /// `enum_map`, or the shared list its `code_list` names.
    ///
    /// Returning a reference rather than resolving at load time is what keeps
    /// the tables out of the compiled cache: a definition serialises the name,
    /// not 94 entries, in each of the files that use it.
    fn table<'a>(
        &'a self,
        inline: Option<&'a BTreeMap<String, String>>,
        named: Option<&str>,
    ) -> Option<&'a BTreeMap<String, String>> {
        inline.or_else(|| named.and_then(|n| self.code_lists.get(n)))
    }

    /// Attach shared code lists to an engine built from cached definitions.
    pub fn with_code_lists(
        mut self,
        code_lists: std::sync::Arc<crate::code_lists::CodeLists>,
    ) -> Self {
        self.code_lists = code_lists;
        self
    }

    /// The shared tables this engine resolves `code_list` names against.
    pub fn code_lists(&self) -> &std::sync::Arc<crate::code_lists::CodeLists> {
        &self.code_lists
    }

    /// Build from cached definitions, with the shared tables their `code_list`
    /// names resolve against.
    pub fn from_definitions_with_code_lists(
        code_lists: std::sync::Arc<crate::code_lists::CodeLists>,
        definitions: Vec<MappingDefinition>,
    ) -> Self {
        // The assertion belongs here too, not only in `from_definitions`:
        // `DataBundle::load` called *this* constructor with an empty `Arc` for
        // months of work, so checking only the other one meant the check could
        // not see the one path that actually shipped.
        debug_assert!(
            !(code_lists.is_empty() && Self::names_a_code_list(&definitions)),
            "definitions name a shared code list but the supplied tables are \
             empty — whatever produced them (a bundle, a cache) is not carrying \
             them, and every code they translate will reach the output raw"
        );
        // Built directly rather than through `from_definitions`, whose debug
        // assertion is precisely "nobody supplied the tables" -- routing the
        // correct call through it would fire on every translated definition.
        Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists,
        }
    }

    pub fn from_definitions(definitions: Vec<MappingDefinition>) -> Self {
        // A definition that names a code list is useless without the tables:
        // the name resolves to nothing and the EDIFACT code reaches the output
        // raw, which reads as "the guide lists no codes here" rather than as
        // the wiring mistake it is. It has happened twice -- once in the API,
        // once in the test harness -- so say so where it happens instead of
        // letting a wrong value travel.
        debug_assert!(
            !Self::names_a_code_list(&definitions),
            "definitions name a shared code list but none were supplied — build \
             this engine with `from_definitions_with_code_lists`, or the codes \
             they translate will reach the output untranslated"
        );
        Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: std::sync::Arc::new(crate::code_lists::CodeLists::default()),
        }
    }

    /// Save definitions to a cache file.
    ///
    /// Only the `definitions` are serialized — `segment_structure` and `code_lookup`
    /// must be re-attached after loading from cache. Paths in the definitions are
    /// already resolved to numeric indices, so no `PathResolver` is needed at load time.
    pub fn save_cached(&self, path: &Path) -> Result<(), MappingError> {
        let encoded =
            serde_json::to_vec(&self.definitions).map_err(|e| MappingError::CacheWrite {
                path: path.display().to_string(),
                message: e.to_string(),
            })?;
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(path, encoded)?;
        Ok(())
    }

    /// Load from cache if available, otherwise fall back to TOML directory.
    ///
    /// When loading from cache, PathResolver is NOT needed (paths pre-resolved).
    /// When falling back to TOML, the caller should chain `.with_path_resolver()`.
    pub fn load_cached_or_toml(cache_path: &Path, toml_dir: &Path) -> Result<Self, MappingError> {
        if cache_path.exists() {
            Self::load_cached(cache_path)
        } else {
            Self::load(toml_dir)
        }
    }

    /// Load definitions from a cache file.
    ///
    /// Returns an engine with only `definitions` populated. Attach `segment_structure`
    /// and `code_lookup` via the builder methods if needed.
    pub fn load_cached(path: &Path) -> Result<Self, MappingError> {
        let bytes = std::fs::read(path)?;
        let definitions: Vec<MappingDefinition> =
            serde_json::from_slice(&bytes).map_err(|e| MappingError::CacheRead {
                path: path.display().to_string(),
                message: e.to_string(),
            })?;
        Ok(Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
            code_lists: crate::code_lists::CodeLists::discover(path),
        })
    }

    /// Attach a MIG-derived segment structure for trailing element padding.
    ///
    /// When set, `map_reverse` pads each segment's elements up to the
    /// MIG-defined count, ensuring trailing empty elements are preserved.
    pub fn with_segment_structure(mut self, ss: SegmentStructure) -> Self {
        self.segment_structure = Some(ss);
        self
    }

    /// Attach a code lookup for enriching code-type field values.
    ///
    /// When set, fields that map to code-type elements in the PID schema
    /// are emitted as `{"code": "Z15", "meaning": "Ja"}` objects instead of plain strings.
    pub fn with_code_lookup(mut self, cl: crate::code_lookup::CodeLookup) -> Self {
        self.code_lookup = Some(cl);
        self
    }

    /// Declare which PID this engine is currently processing.
    ///
    /// When combined with [`with_code_lookup`](Self::with_code_lookup), code
    /// fields whose only allowed value equals the PID itself (Class C in the
    /// 2026-04-28 audit — e.g., RFF+Z13's d1154 in PID 55002 enumerates only
    /// `55002`) are emitted as plain strings instead of being decorated with
    /// `{code, meaning, enum}` and a dedup-suffixed enum name.
    pub fn with_pid(mut self, pid: impl Into<String>) -> Self {
        self.current_pid = Some(pid.into());
        self
    }

    /// Attach a path resolver to normalize EDIFACT ID paths to numeric indices.
    ///
    /// This allows TOML mapping files to use named paths like `loc.c517.d3225`
    /// instead of numeric indices like `loc.1.0`. Resolution happens once at
    /// load time — the engine hot path is completely unchanged.
    pub fn with_path_resolver(mut self, resolver: crate::path_resolver::PathResolver) -> Self {
        for def in &mut self.definitions {
            def.normalize_paths(&resolver);
        }
        self
    }

    /// Declare the transaction-root SG id (e.g. `"SG4"` for UTILMD).
    ///
    /// When set, entities whose parent group equals this id are not nested
    /// into their parent in the forward-mapped JSON. See the
    /// [`transaction_group`](Self#structfield.transaction_group-1) field doc
    /// on `MappingEngine` for the full rationale.
    pub fn with_transaction_group(mut self, tx: impl Into<String>) -> Self {
        self.transaction_group = Some(tx.into());
        self
    }

    /// Add definitions to an already-built engine, keeping everything else it
    /// carries (code lookup, segment structure, PID, transaction group).
    ///
    /// Used to widen a message-level engine into one flat engine over a whole
    /// variant — what APERAK and CONTRL are converted with, since they have no
    /// message/transaction split in the v2 `convert` route. A definition whose
    /// `(entity, source_group, source_path, discriminator, parent_field)` the
    /// engine already has is skipped, so the same rule reached through two
    /// PIDs is added once.
    ///
    /// The definitions are taken as they are; run them through
    /// [`with_path_resolver`](Self::with_path_resolver) first if their paths
    /// are still named.
    pub fn extend_definitions(mut self, defs: impl IntoIterator<Item = MappingDefinition>) -> Self {
        fn key(d: &MappingDefinition) -> (String, String, String, String, String) {
            (
                d.meta.entity.clone(),
                d.meta.source_group.clone(),
                d.meta.source_path.clone().unwrap_or_default(),
                d.meta.discriminator.clone().unwrap_or_default(),
                d.meta.parent_field.clone().unwrap_or_default(),
            )
        }
        let mut seen: std::collections::HashSet<_> = self.definitions.iter().map(key).collect();
        for def in defs {
            if seen.insert(key(&def)) {
                self.definitions.push(def);
            }
        }
        self
    }

    /// Get all loaded definitions.
    pub fn definitions(&self) -> &[MappingDefinition] {
        &self.definitions
    }

    /// Find a definition by entity name.
    /// The entity's own definition. `parent_field` children (which carry their
    /// parent's entity name and are mapped inside the parent's instance) are
    /// skipped — they are not a definition *of* the entity.
    pub fn definition_for_entity(&self, entity: &str) -> Option<&MappingDefinition> {
        self.definitions
            .iter()
            .find(|d| d.meta.entity == entity && d.meta.parent_field.is_none())
    }

    // ── Forward mapping: tree → BO4E ──

    /// Extract a field value from an assembled tree using a mapping path.
    ///
    /// `group_path` supports dotted notation for nested groups (e.g., "SG4.SG5").
    /// Parent groups default to repetition 0; `repetition` applies to the leaf group.
    ///
    /// Path format: "segment.composite.data_element" e.g., "loc.c517.d3225"
    pub fn extract_field(
        &self,
        tree: &AssembledTree,
        group_path: &str,
        path: &str,
        repetition: usize,
    ) -> Option<String> {
        let instance = Self::resolve_group_instance(tree, group_path, repetition)?;
        Self::extract_from_instance(instance, path)
    }

    /// Navigate a potentially nested group path to find a group instance.
    ///
    /// For "SG4.SG5", finds SG4\[0\] then SG5 at the given repetition within it.
    /// For "SG8", finds SG8 at the given repetition in the top-level groups.
    ///
    /// Supports intermediate repetition with colon syntax: "SG4.SG8:1.SG10"
    /// means SG4\[0\] → SG8\[1\] → SG10\[repetition\]. Without a colon suffix,
    /// intermediate groups default to repetition 0.
    pub fn resolve_group_instance<'a>(
        tree: &'a AssembledTree,
        group_path: &str,
        repetition: usize,
    ) -> Option<&'a AssembledGroupInstance> {
        let parts: Vec<&str> = group_path.split('.').collect();

        let (first_id, first_rep) = parse_group_spec(parts[0]);
        let first_group = tree.groups.iter().find(|g| g.group_id == first_id)?;

        if parts.len() == 1 {
            // Single part — use the explicit rep from spec or the `repetition` param
            let rep = first_rep.unwrap_or(repetition);
            return first_group.repetitions.get(rep);
        }

        // Navigate through groups; intermediate parts default to rep 0
        // unless explicitly specified via `:N` suffix
        let mut current_instance = first_group.repetitions.get(first_rep.unwrap_or(0))?;

        for (i, part) in parts[1..].iter().enumerate() {
            let (group_id, explicit_rep) = parse_group_spec(part);
            let child_group = current_instance
                .child_groups
                .iter()
                .find(|g| g.group_id == group_id)?;

            if i == parts.len() - 2 {
                // Last part — use explicit rep, or fall back to `repetition`
                let rep = explicit_rep.unwrap_or(repetition);
                return child_group.repetitions.get(rep);
            }
            // Intermediate — use explicit rep or 0
            current_instance = child_group.repetitions.get(explicit_rep.unwrap_or(0))?;
        }

        None
    }

    /// Navigate the assembled tree using a source_path with qualifier suffixes.
    ///
    /// Source paths like `"sg4.sg8_z98.sg10"` encode qualifiers inline:
    /// `sg8_z98` means "find the SG8 repetition whose entry segment has qualifier Z98".
    /// Parts without underscores (e.g., `sg4`, `sg10`) use the first repetition.
    ///
    /// Returns `None` if any part of the path can't be resolved.
    pub fn resolve_by_source_path<'a>(
        tree: &'a AssembledTree,
        source_path: &str,
    ) -> Option<&'a AssembledGroupInstance> {
        let parts: Vec<&str> = source_path.split('.').collect();
        if parts.is_empty() {
            return None;
        }

        let (first_id, first_qualifier) = parse_source_path_part(parts[0]);
        let first_group = tree
            .groups
            .iter()
            .find(|g| g.group_id.eq_ignore_ascii_case(first_id))?;

        let mut current_instance = if let Some(q) = first_qualifier {
            find_rep_by_entry_qualifier(&first_group.repetitions, q)?
        } else {
            first_group.repetitions.first()?
        };

        if parts.len() == 1 {
            return Some(current_instance);
        }

        for part in &parts[1..] {
            let (group_id, qualifier) = parse_source_path_part(part);
            let child_group = current_instance
                .child_groups
                .iter()
                .find(|g| g.group_id.eq_ignore_ascii_case(group_id))?;

            current_instance = if let Some(q) = qualifier {
                find_rep_by_entry_qualifier(&child_group.repetitions, q)?
            } else {
                child_group.repetitions.first()?
            };
        }

        Some(current_instance)
    }

    /// Resolve ALL matching instances for a source_path, returning a Vec.
    ///
    /// Like `resolve_by_source_path` but returns all repetitions matching
    /// at any level, not just the first.  For example, if there are two SG5
    /// reps with LOC+Z17, `resolve_all_by_source_path(tree, "sg4.sg5_z17")`
    /// returns both.  For deeper paths like "sg4.sg8_zf3.sg10", if there are
    /// two SG8 reps with ZF3, it returns SG10 children from both.
    pub fn resolve_all_by_source_path<'a>(
        tree: &'a AssembledTree,
        source_path: &str,
    ) -> Vec<&'a AssembledGroupInstance> {
        let parts: Vec<&str> = source_path.split('.').collect();
        if parts.is_empty() {
            return vec![];
        }

        // First part: match against top-level groups
        let (first_id, first_qualifier) = parse_source_path_part(parts[0]);
        let first_group = match tree
            .groups
            .iter()
            .find(|g| g.group_id.eq_ignore_ascii_case(first_id))
        {
            Some(g) => g,
            None => return vec![],
        };

        let mut current_instances: Vec<&AssembledGroupInstance> = if let Some(q) = first_qualifier {
            find_all_reps_by_entry_qualifier(&first_group.repetitions, q)
        } else {
            first_group.repetitions.iter().collect()
        };

        // Navigate remaining parts, branching at each level when multiple
        // instances match a qualifier (e.g., two SG8 reps with ZF3).
        for part in &parts[1..] {
            let (group_id, qualifier) = parse_source_path_part(part);
            let mut next_instances = Vec::new();

            for instance in &current_instances {
                if let Some(child_group) = instance
                    .child_groups
                    .iter()
                    .find(|g| g.group_id.eq_ignore_ascii_case(group_id))
                {
                    if let Some(q) = qualifier {
                        next_instances.extend(find_all_reps_by_entry_qualifier(
                            &child_group.repetitions,
                            q,
                        ));
                    } else {
                        next_instances.extend(child_group.repetitions.iter());
                    }
                }
            }

            current_instances = next_instances;
        }

        current_instances
    }

    /// Like `resolve_all_by_source_path` but also returns the direct parent
    /// rep index that each leaf instance came from. The "direct parent" is the
    /// group one level above the leaf in the path.
    ///
    /// For `"sg2.sg3"`: parent is the SG2 rep index.
    /// For `"sg17.sg36.sg40"`: parent is the SG36 rep index (not SG17).
    ///
    /// For single-level paths, all indices are 0.
    ///
    /// Compute child rep indices for the leaf group in a source_path.
    /// E.g., for "sg29.sg30", returns the position of each matched SG30 rep
    /// within its parent SG29's SG30 child group.
    fn compute_child_indices(
        tree: &AssembledTree,
        source_path: &str,
        indexed: &[(usize, &AssembledGroupInstance)],
    ) -> Vec<usize> {
        let parts: Vec<&str> = source_path.split('.').collect();
        if parts.len() < 2 {
            return vec![];
        }
        // Navigate to the parent level and find the child group
        let (first_id, first_qualifier) = parse_source_path_part(parts[0]);
        let first_group = match tree
            .groups
            .iter()
            .find(|g| g.group_id.eq_ignore_ascii_case(first_id))
        {
            Some(g) => g,
            None => return vec![],
        };
        let parent_reps: Vec<&AssembledGroupInstance> = if let Some(q) = first_qualifier {
            find_all_reps_by_entry_qualifier(&first_group.repetitions, q)
        } else {
            first_group.repetitions.iter().collect()
        };
        // For 2-level paths (sg29.sg30), find the child group in the parent
        let (child_id, _child_qualifier) = parse_source_path_part(parts[parts.len() - 1]);
        let mut result = Vec::new();
        for (_, inst) in indexed {
            // Find which rep index this instance is at in the child group
            let mut found = false;
            for parent in &parent_reps {
                if let Some(child_group) = parent
                    .child_groups
                    .iter()
                    .find(|g| g.group_id.eq_ignore_ascii_case(child_id))
                {
                    if let Some(pos) = child_group
                        .repetitions
                        .iter()
                        .position(|r| std::ptr::eq(r, *inst))
                    {
                        result.push(pos);
                        found = true;
                        break;
                    }
                }
            }
            if !found {
                result.push(usize::MAX); // fallback
            }
        }
        result
    }

    /// Returns `Vec<(parent_rep_index, &AssembledGroupInstance)>`.
    pub fn resolve_all_with_parent_indices<'a>(
        tree: &'a AssembledTree,
        source_path: &str,
    ) -> Vec<(usize, &'a AssembledGroupInstance)> {
        let parts: Vec<&str> = source_path.split('.').collect();
        if parts.is_empty() {
            return vec![];
        }

        // First part: match against top-level groups
        let (first_id, first_qualifier) = parse_source_path_part(parts[0]);
        let first_group = match tree
            .groups
            .iter()
            .find(|g| g.group_id.eq_ignore_ascii_case(first_id))
        {
            Some(g) => g,
            None => return vec![],
        };

        // If single-level path, just return instances with index 0
        if parts.len() == 1 {
            let instances: Vec<&AssembledGroupInstance> = if let Some(q) = first_qualifier {
                find_all_reps_by_entry_qualifier(&first_group.repetitions, q)
            } else {
                first_group.repetitions.iter().collect()
            };
            return instances.into_iter().map(|i| (0, i)).collect();
        }

        // Multi-level: navigate tracking (parent_rep_idx, instance) at each level.
        // At intermediate levels, parent_rep_idx is updated to the current rep's
        // position within its group. At the leaf level, the parent_rep_idx from
        // the previous level is preserved — giving us the DIRECT parent index.
        let first_reps: Vec<(usize, &AssembledGroupInstance)> = if let Some(q) = first_qualifier {
            let matching = find_all_reps_by_entry_qualifier(&first_group.repetitions, q);
            let mut result = Vec::new();
            for m in matching {
                let idx = first_group
                    .repetitions
                    .iter()
                    .position(|r| std::ptr::eq(r, m))
                    .unwrap_or(0);
                result.push((idx, m));
            }
            result
        } else {
            first_group.repetitions.iter().enumerate().collect()
        };

        let mut current: Vec<(usize, &AssembledGroupInstance)> = first_reps;
        let remaining = &parts[1..];

        for (level, part) in remaining.iter().enumerate() {
            let is_leaf = level == remaining.len() - 1;
            let (group_id, qualifier) = parse_source_path_part(part);
            let mut next: Vec<(usize, &AssembledGroupInstance)> = Vec::new();

            for (prev_parent_idx, instance) in &current {
                if let Some(child_group) = instance
                    .child_groups
                    .iter()
                    .find(|g| g.group_id.eq_ignore_ascii_case(group_id))
                {
                    let matching: Vec<(usize, &AssembledGroupInstance)> = if let Some(q) = qualifier
                    {
                        let filtered =
                            find_all_reps_by_entry_qualifier(&child_group.repetitions, q);
                        filtered
                            .into_iter()
                            .map(|m| {
                                let idx = child_group
                                    .repetitions
                                    .iter()
                                    .position(|r| std::ptr::eq(r, m))
                                    .unwrap_or(0);
                                (idx, m)
                            })
                            .collect()
                    } else {
                        child_group.repetitions.iter().enumerate().collect()
                    };

                    for (rep_idx, child_rep) in matching {
                        if is_leaf {
                            // At the leaf: keep the parent index from the previous level
                            next.push((*prev_parent_idx, child_rep));
                        } else {
                            // At intermediate: pass down the current rep index
                            next.push((rep_idx, child_rep));
                        }
                    }
                }
            }

            current = next;
        }

        current
    }

    /// Extract a field from a group instance by path.
    ///
    /// Supports qualifier-based segment selection with `tag[qualifier]` syntax:
    /// - `"dtm.0.1"` → first DTM segment, elements\[0\]\[1\]
    /// - `"dtm[92].0.1"` → DTM where elements\[0\]\[0\] == "92", then elements\[0\]\[1\]
    pub fn extract_from_instance(instance: &AssembledGroupInstance, path: &str) -> Option<String> {
        let parts: Vec<&str> = path.split('.').collect();
        if parts.is_empty() {
            return None;
        }

        // Parse segment tag, optional qualifier, and occurrence index:
        // "dtm[92]" → ("DTM", Some("92"), 0), "rff[Z34,1]" → ("RFF", Some("Z34"), 1)
        let (segment_tag, qualifier, occurrence) = parse_tag_qualifier(parts[0]);

        let segment = if let Some(q) = qualifier {
            instance
                .segments
                .iter()
                .filter(|s| {
                    s.tag.eq_ignore_ascii_case(&segment_tag)
                        && s.elements
                            .first()
                            .and_then(|e| e.first())
                            .map(|v| v.as_str())
                            == Some(q)
                })
                .nth(occurrence)?
        } else {
            instance
                .segments
                .iter()
                .filter(|s| s.tag.eq_ignore_ascii_case(&segment_tag))
                .nth(occurrence)?
        };

        Self::resolve_field_path(segment, &parts[1..])
    }

    /// Extract ALL matching values from a group instance for a collect-all path.
    ///
    /// Used with wildcard occurrence syntax `tag[qualifier,*]` to collect values
    /// from every segment matching the qualifier, not just the Nth one.
    /// Returns a `Vec<String>` of all extracted values in segment order.
    pub fn extract_all_from_instance(instance: &AssembledGroupInstance, path: &str) -> Vec<String> {
        let parts: Vec<&str> = path.split('.').collect();
        if parts.is_empty() {
            return vec![];
        }

        let (segment_tag, qualifier, _) = parse_tag_qualifier(parts[0]);

        let matching_segments: Vec<&AssembledSegment> = if let Some(q) = qualifier {
            instance
                .segments
                .iter()
                .filter(|s| {
                    s.tag.eq_ignore_ascii_case(&segment_tag)
                        && s.elements
                            .first()
                            .and_then(|e| e.first())
                            .map(|v| v.as_str())
                            == Some(q)
                })
                .collect()
        } else {
            instance
                .segments
                .iter()
                .filter(|s| s.tag.eq_ignore_ascii_case(&segment_tag))
                .collect()
        };

        matching_segments
            .into_iter()
            .filter_map(|seg| Self::resolve_field_path(seg, &parts[1..]))
            .collect()
    }

    /// Map all fields in a definition from the assembled tree to a BO4E JSON object.
    ///
    /// `group_path` is the definition's `source_group` (may be dotted, e.g., "SG4.SG5").
    /// An empty `source_group` maps root-level segments (BGM, DTM, etc.).
    /// Returns a flat JSON object with target field names as keys.
    pub fn map_forward(
        &self,
        tree: &AssembledTree,
        def: &MappingDefinition,
        repetition: usize,
    ) -> serde_json::Value {
        self.map_forward_inner(tree, def, repetition, true)
    }

    /// Inner implementation with enrichment control.
    fn map_forward_inner(
        &self,
        tree: &AssembledTree,
        def: &MappingDefinition,
        repetition: usize,
        enrich_codes: bool,
    ) -> serde_json::Value {
        let mut result = serde_json::Map::new();

        // Root-level mapping: source_group is empty → use tree's own segments.
        // Include all root segments (both pre-group and post-group, e.g., summary
        // MOA after UNS+S in REMADV) plus any inter_group_segments (e.g., UNS+S
        // consumed between groups by the assembler).
        if def.meta.source_group.is_empty() {
            let mut all_root_segs = tree.segments.clone();
            for segs in tree.inter_group_segments.values() {
                all_root_segs.extend(segs.iter().cloned());
            }
            let root_instance = AssembledGroupInstance {
                segments: all_root_segs,
                child_groups: vec![],
                entry_mig_number: None,
                variant_mig_numbers: vec![],
                skipped_segments: Vec::new(),
                skipped_positions: Vec::new(),
            };
            self.extract_fields_from_instance(&root_instance, def, &mut result, enrich_codes);
            return serde_json::Value::Object(result);
        }

        // Try source_path-based resolution when:
        //   1. source_path has qualifier suffixes (e.g., "sg4.sg8_z98.sg10")
        //   2. source_group has no explicit :N indices (those take priority)
        // This allows definitions without positional indices to navigate via
        // entry-segment qualifiers (e.g., SEQ qualifier Z98).
        let instance = if let Some(ref sp) = def.meta.source_path {
            if has_source_path_qualifiers(sp) && !def.meta.source_group.contains(':') {
                Self::resolve_by_source_path(tree, sp).or_else(|| {
                    Self::resolve_group_instance(tree, &def.meta.source_group, repetition)
                })
            } else {
                Self::resolve_group_instance(tree, &def.meta.source_group, repetition)
            }
        } else {
            Self::resolve_group_instance(tree, &def.meta.source_group, repetition)
        };

        if let Some(instance) = instance {
            // repeat_on_tag: iterate over all segments of that tag, producing an array
            if let Some(ref tag) = def.meta.repeat_on_tag {
                let matching: Vec<_> = instance
                    .segments
                    .iter()
                    .filter(|s| s.tag.eq_ignore_ascii_case(tag))
                    .collect();

                if matching.len() > 1 {
                    let mut arr = Vec::new();
                    for seg in &matching {
                        let sub_instance = AssembledGroupInstance {
                            segments: vec![(*seg).clone()],
                            child_groups: vec![],
                            entry_mig_number: None,
                            variant_mig_numbers: vec![],
                            skipped_segments: Vec::new(),
                            skipped_positions: Vec::new(),
                        };
                        let mut elem_result = serde_json::Map::new();
                        self.extract_fields_from_instance(
                            &sub_instance,
                            def,
                            &mut elem_result,
                            enrich_codes,
                        );
                        if !elem_result.is_empty() {
                            arr.push(serde_json::Value::Object(elem_result));
                        }
                    }
                    if !arr.is_empty() {
                        return serde_json::Value::Array(arr);
                    }
                }
            }

            self.extract_fields_from_instance(instance, def, &mut result, enrich_codes);
        }

        serde_json::Value::Object(result)
    }

    /// Extract all fields from an instance into a result map.
    ///
    /// When a `code_lookup` is configured, code-type fields are emitted as
    /// `{"code": "E01", "meaning": "..."}` objects. Data-type fields remain plain strings.
    fn extract_fields_from_instance(
        &self,
        instance: &AssembledGroupInstance,
        def: &MappingDefinition,
        result: &mut serde_json::Map<String, serde_json::Value>,
        enrich_codes: bool,
    ) {
        for (path, field_mapping) in &def.fields {
            let (target, enum_map) = match field_mapping {
                FieldMapping::Simple(t) => (t.as_str(), None),
                FieldMapping::Structured(s) => (
                    s.target.as_str(),
                    self.table(s.enum_map.as_ref(), s.code_list.as_deref()),
                ),
                FieldMapping::Nested(_) => continue,
            };
            if target.is_empty() {
                continue;
            }
            if let Some(val) = Self::extract_from_instance(instance, path) {
                // Dual decomposition: one EDIFACT code also feeds a second BO4E
                // field (e.g. the NAD qualifier carries both partnerrolle and
                // datenqualitaet). Without this the code cannot be recovered in
                // reverse, because several codes share the primary value.
                if let FieldMapping::Structured(s) = field_mapping {
                    if let (Some(also), Some(also_map)) = (
                        s.also_target.as_deref(),
                        self.table(s.also_enum_map.as_ref(), s.also_code_list.as_deref()),
                    ) {
                        if let Some(also_val) = also_map.get(&val) {
                            set_nested_value(result, also, also_val.clone());
                        }
                    }
                }

                let mapped_val = if let Some(map) = enum_map {
                    map.get(&val).cloned().unwrap_or_else(|| val.clone())
                } else {
                    val.clone()
                };

                // Enrich code fields with meaning from PID schema
                if enrich_codes {
                    if let (Some(ref code_lookup), Some(ref source_path)) =
                        (&self.code_lookup, &def.meta.source_path)
                    {
                        let parts: Vec<&str> = path.split('.').collect();
                        let (seg_tag, path_qualifier, _occ) = parse_tag_qualifier(parts[0]);
                        let (element_idx, component_idx) =
                            Self::parse_element_component(&parts[1..]);
                        let disc_qualifier = Self::discriminator_qualifier_for_tag(def, &seg_tag);
                        let q = disc_qualifier.as_deref();

                        if let Some(codes) = code_lookup.enrichment_codes(
                            source_path,
                            &seg_tag,
                            path_qualifier,
                            q,
                            element_idx,
                            component_idx,
                        ) {
                            // Class C: PID self-reference — emit a plain string,
                            // skipping {code, meaning, enum} decoration when the
                            // schema's only allowed value at this position is the
                            // PID itself.
                            if let Some(ref pid) = self.current_pid {
                                if codes.len() == 1 && codes.contains_key(pid.as_str()) {
                                    set_nested_value(result, target, mapped_val);
                                    continue;
                                }
                            }

                            // Look up the original EDIFACT value for enrichment,
                            // since schema codes use raw values (e.g., "293")
                            // not enum_map targets (e.g., "BDEW").
                            let enrichment = codes.get(&val);
                            let meaning = enrichment
                                .map(|e| serde_json::Value::String(e.meaning.clone()))
                                .unwrap_or(serde_json::Value::Null);

                            let mut obj = serde_json::Map::new();
                            obj.insert("code".into(), serde_json::json!(mapped_val));
                            obj.insert("meaning".into(), meaning);
                            if let Some(enum_key) = enrichment.and_then(|e| e.enum_key.as_ref()) {
                                obj.insert("enum".into(), serde_json::json!(enum_key));
                            }
                            let enriched = serde_json::Value::Object(obj);
                            set_nested_value_json(result, target, enriched);
                            continue;
                        }
                    }
                }

                set_nested_value(result, target, mapped_val);
            }
        }

        // Also for `parent_field` children: they can be parents of deeper
        // `parent_field` definitions (SG15 → SG17 → SG18).
        if !instance.child_groups.is_empty() {
            self.extract_nested_children(instance, def, result, enrich_codes);
        }
    }

    /// Forward half of `[meta] parent_field`: map the child groups of `instance`
    /// (the parent group repetition `def` was just extracted from) into array
    /// fields of the same object. Placement is instance-local — a child can only
    /// land in the object produced from the group repetition that contains it.
    fn extract_nested_children(
        &self,
        instance: &AssembledGroupInstance,
        def: &MappingDefinition,
        result: &mut serde_json::Map<String, serde_json::Value>,
        enrich_codes: bool,
    ) {
        for child in self
            .definitions
            .iter()
            .filter(|c| is_nested_child_of(c, def))
        {
            if nested_parent_qualifier(child).is_some_and(|q| !entry_qualifier_matches(instance, q))
            {
                continue;
            }
            let (leaf_id, leaf_qualifier) = nested_child_leaf(child);
            let Some(group) = instance
                .child_groups
                .iter()
                .find(|g| g.group_id.eq_ignore_ascii_case(&leaf_id))
            else {
                continue;
            };
            let reps: Vec<&AssembledGroupInstance> = match leaf_qualifier {
                Some(q) => find_all_reps_by_entry_qualifier(&group.repetitions, q),
                None => group.repetitions.iter().collect(),
            };

            let mut items: Vec<serde_json::Value> = Vec::new();
            let mut push_item = |sub: &AssembledGroupInstance| {
                let mut obj = serde_json::Map::new();
                self.extract_fields_from_instance(sub, child, &mut obj, enrich_codes);
                if !obj.is_empty() {
                    items.push(serde_json::Value::Object(obj));
                }
            };
            for rep in reps {
                let repeat_tag = child
                    .meta
                    .repeat_on_tag
                    .as_deref()
                    .filter(|tag| rep.segments.iter().any(|s| s.tag.eq_ignore_ascii_case(tag)));
                let Some(tag) = repeat_tag else {
                    push_item(rep);
                    continue;
                };
                // One element per repeating segment; the group's other segments
                // (e.g. the CTA entry segment) are visible to every element.
                let shared: Vec<AssembledSegment> = rep
                    .segments
                    .iter()
                    .filter(|s| !s.tag.eq_ignore_ascii_case(tag))
                    .cloned()
                    .collect();
                for seg in rep
                    .segments
                    .iter()
                    .filter(|s| s.tag.eq_ignore_ascii_case(tag))
                {
                    let mut segments = shared.clone();
                    segments.push(seg.clone());
                    push_item(&AssembledGroupInstance {
                        segments,
                        child_groups: vec![],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: Vec::new(),
                        skipped_positions: Vec::new(),
                    });
                }
            }
            if items.is_empty() {
                continue;
            }
            let field = child.meta.parent_field.as_deref().unwrap_or_default();
            match result.get_mut(field) {
                Some(serde_json::Value::Array(existing)) => existing.extend(items),
                _ => {
                    result.insert(field.to_string(), serde_json::Value::Array(items));
                }
            }
        }
    }

    /// Reverse half of `[meta] parent_field`: emit the elements of
    /// `bo4e_value[parent_field]` as child group(s) of `instance`, the parent
    /// group repetition just rebuilt from that same object.
    fn reverse_nested_children(
        &self,
        bo4e_value: &serde_json::Value,
        def: &MappingDefinition,
        instance: &mut AssembledGroupInstance,
    ) {
        let mut handled_fields: Vec<&str> = Vec::new();
        for child in self
            .definitions
            .iter()
            .filter(|c| is_nested_child_of(c, def))
        {
            let field = child.meta.parent_field.as_deref().unwrap_or_default();
            if handled_fields.contains(&field) {
                continue;
            }
            if nested_parent_qualifier(child)
                .is_some_and(|q| !rebuilt_entry_qualifier_matches(instance, def, q))
            {
                continue;
            }
            let elements: Vec<&serde_json::Value> = match bo4e_value.get(field) {
                Some(serde_json::Value::Array(arr)) => arr.iter().collect(),
                Some(serde_json::Value::Null) | None => continue,
                Some(other) => vec![other],
            };

            let mut reps: Vec<AssembledGroupInstance> = Vec::new();
            if let Some(tag) = child.meta.repeat_on_tag.as_deref() {
                // All elements share one group repetition: the non-repeating
                // segments (entry segment) once, then one repeating segment each.
                let mut merged: Option<AssembledGroupInstance> = None;
                for element in elements {
                    let sub = self.map_reverse_single(element, child);
                    if sub.segments.is_empty() {
                        continue;
                    }
                    match merged.as_mut() {
                        None => merged = Some(sub),
                        Some(m) => m.segments.extend(
                            sub.segments
                                .into_iter()
                                .filter(|s| s.tag.eq_ignore_ascii_case(tag)),
                        ),
                    }
                }
                reps.extend(merged);
            } else {
                for element in elements {
                    let mut sub = self.map_reverse_single(element, child);
                    if sub.segments.is_empty() {
                        continue;
                    }
                    // Grandchildren nested in this element (multi-level nesting).
                    self.reverse_nested_children(element, child, &mut sub);
                    reps.push(sub);
                }
            }
            if reps.is_empty() {
                continue;
            }
            handled_fields.push(field);

            let (leaf_id, _) = nested_child_leaf(child);
            match instance
                .child_groups
                .iter_mut()
                .find(|g| g.group_id.eq_ignore_ascii_case(&leaf_id))
            {
                Some(group) => group.repetitions.extend(reps),
                None => instance.child_groups.push(AssembledGroup {
                    group_id: leaf_id,
                    repetitions: reps,
                }),
            }
        }
    }

    /// Extract the discriminator's qualifier value from a definition's `[meta]`.
    ///
    /// `discriminator` strings look like `"RFF.0.0=Z13"` (numeric, post path-resolution)
    /// or `"RFF.c506.d1153=TN"` (named, pre-resolution). The qualifier is the
    /// substring after the first `=`. Returns `None` when no discriminator is set
    /// or the format is unexpected.
    pub(crate) fn discriminator_qualifier(def: &MappingDefinition) -> Option<String> {
        def.meta
            .discriminator
            .as_deref()
            .and_then(|d| d.split_once('=').map(|(_, v)| v.to_string()))
    }

    /// The discriminator value when the discriminator selects on `segment_tag`
    /// itself (`RFF.0.0=Z13` for an RFF field). A discriminator on another segment
    /// (`SEQ.0.0=Z98` for a CCI field) says nothing about which variant of
    /// `segment_tag` a field reads.
    pub(crate) fn discriminator_qualifier_for_tag(
        def: &MappingDefinition,
        segment_tag: &str,
    ) -> Option<String> {
        let (lhs, value) = def.meta.discriminator.as_deref()?.split_once('=')?;
        let disc_tag = lhs.split('.').next().unwrap_or(lhs);
        disc_tag
            .eq_ignore_ascii_case(segment_tag)
            .then(|| value.to_string())
    }

    /// Map a PID struct field's segments to BO4E JSON.
    ///
    /// `segments` are the `OwnedSegment`s from a PID wrapper field.
    /// Converts to `AssembledSegment` format for compatibility with existing
    /// field extraction logic, then applies the definition's field mappings.
    pub fn map_forward_from_segments(
        &self,
        segments: &[OwnedSegment],
        def: &MappingDefinition,
    ) -> serde_json::Value {
        let assembled_segments: Vec<AssembledSegment> = segments
            .iter()
            .map(|s| AssembledSegment {
                tag: s.id.clone(),
                elements: s.elements.clone(),
                mig_number: None,
                segment_number: Some(s.segment_number),
            })
            .collect();

        let instance = AssembledGroupInstance {
            segments: assembled_segments,
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: Vec::new(),
            skipped_positions: Vec::new(),
        };

        let mut result = serde_json::Map::new();
        self.extract_fields_from_instance(&instance, def, &mut result, true);
        serde_json::Value::Object(result)
    }

    // ── Reverse mapping: BO4E → tree ──

    /// Map a BO4E JSON object back to an assembled group instance.
    ///
    /// Uses the definition's field mappings to populate segment elements.
    /// Fields with `default` values are used when no BO4E value is present
    /// (useful for fixed qualifiers like LOC qualifier "Z16").
    ///
    /// Supports:
    /// - Named paths: `"d3227"` → element\[0\]\[0\], `"c517.d3225"` → element\[1\]\[0\]
    /// - Numeric index: `"0"` → element\[0\]\[0\], `"1.2"` → element\[1\]\[2\]
    /// - Qualifier selection: `"dtm[92].0.1"` → DTM segment with qualifier "92"
    pub fn map_reverse(
        &self,
        bo4e_value: &serde_json::Value,
        def: &MappingDefinition,
    ) -> AssembledGroupInstance {
        // repeat_on_tag + array input: reverse each element independently, merge segments
        if def.meta.repeat_on_tag.is_some() {
            if let Some(arr) = bo4e_value.as_array() {
                let mut all_segments = Vec::new();
                for elem in arr {
                    let sub = self.map_reverse_single(elem, def);
                    all_segments.extend(sub.segments);
                }
                return AssembledGroupInstance {
                    segments: all_segments,
                    child_groups: vec![],
                    entry_mig_number: None,
                    variant_mig_numbers: vec![],
                    skipped_segments: Vec::new(),
                    skipped_positions: Vec::new(),
                };
            }
        }
        let mut instance = self.map_reverse_single(bo4e_value, def);
        if def.meta.parent_field.is_none() && !instance.segments.is_empty() {
            self.reverse_nested_children(bo4e_value, def, &mut instance);
        }
        instance
    }

    fn map_reverse_single(
        &self,
        bo4e_value: &serde_json::Value,
        def: &MappingDefinition,
    ) -> AssembledGroupInstance {
        // Collect (segment_key, element_index, component_index, value) tuples.
        // segment_key includes qualifier for disambiguation: "DTM" or "DTM[92]".
        let mut field_values: Vec<(String, String, usize, usize, String)> =
            Vec::with_capacity(def.fields.len());

        // Track whether any field with a non-empty target resolved to an actual
        // BO4E value.  When a definition has data fields but none resolved to
        // values, only defaults (qualifiers) would be emitted — producing phantom
        // segments for groups not present in the original EDIFACT message.
        // Definitions with ONLY qualifier/default fields (no data targets) are
        // "container" definitions (e.g., SEQ entry segments) and are always kept.
        let mut has_real_data = false;
        let mut has_data_fields = false;
        // Per-segment phantom tracking: segments with data fields but no resolved
        // data are phantoms — their entries should be removed from field_values.
        let mut seg_has_data_field: HashSet<String> = HashSet::new();
        let mut seg_has_real_data: HashSet<String> = HashSet::new();
        let mut injected_qualifiers: HashSet<String> = HashSet::new();

        for (path, field_mapping) in &def.fields {
            let (target, default, enum_map, when_filled, also_target, also_enum_map) =
                match field_mapping {
                    FieldMapping::Simple(t) => (t.as_str(), None, None, None, None, None),
                    FieldMapping::Structured(s) => (
                        s.target.as_str(),
                        s.default.as_ref(),
                        self.table(s.enum_map.as_ref(), s.code_list.as_deref()),
                        s.when_filled.as_ref(),
                        s.also_target.as_deref(),
                        self.table(s.also_enum_map.as_ref(), s.also_code_list.as_deref()),
                    ),
                    FieldMapping::Nested(_) => continue,
                };

            let parts: Vec<&str> = path.split('.').collect();
            if parts.len() < 2 {
                continue;
            }

            let (seg_tag, qualifier, _occ) = parse_tag_qualifier(parts[0]);
            // Use the raw first part as segment key to group fields by segment instance.
            // Indexed qualifiers like "RFF[Z34,1]" produce a distinct key from "RFF[Z34]".
            let seg_key = parts[0].to_uppercase();
            let sub_path = &parts[1..];

            // Determine (element_idx, component_idx) from path
            let (element_idx, component_idx) = if let Ok(ei) = sub_path[0].parse::<usize>() {
                let ci = if sub_path.len() > 1 {
                    sub_path[1].parse::<usize>().unwrap_or(0)
                } else {
                    0
                };
                (ei, ci)
            } else {
                match sub_path.len() {
                    1 => (0, 0),
                    2 => (1, 0),
                    _ => continue,
                }
            };

            // Try BO4E value first, fall back to default
            let val = if target.is_empty() {
                match (default, when_filled) {
                    // has when_filled → conditional injection
                    (Some(d), Some(fields)) => {
                        let any_filled = fields
                            .iter()
                            .any(|f| self.populate_field(bo4e_value, f).is_some());
                        if any_filled {
                            // A successful when_filled check confirms real data
                            // exists — prevent phantom suppression.
                            has_real_data = true;
                            Some(d.clone())
                        } else {
                            None
                        }
                    }
                    // no when_filled → unconditional (backward compat)
                    (Some(d), None) => Some(d.clone()),
                    (None, _) => None,
                }
            } else {
                has_data_fields = true;
                seg_has_data_field.insert(seg_key.clone());
                let bo4e_val = self.populate_field(bo4e_value, target);
                if bo4e_val.is_some() {
                    has_real_data = true;
                    seg_has_real_data.insert(seg_key.clone());
                }
                // Apply reverse enum_map: BO4E value → EDIFACT value
                let mapped_val = match (bo4e_val, enum_map) {
                    (Some(v), Some(map)) => {
                        // Dual decomposition (`also_target`): one EDIFACT code was
                        // split across two BO4E fields, so neither alone identifies
                        // it. Find the code both maps agree on; several codes share
                        // a `partnerrolle` and are told apart only by the second
                        // field. Falls back to the single-map lookup when the
                        // second field is absent or no code matches both.
                        let joint = match (also_target, also_enum_map) {
                            (Some(also), Some(also_map)) => {
                                self.populate_field(bo4e_value, also).and_then(|also_v| {
                                    map.iter()
                                        .find(|(code, bo4e_v)| {
                                            *bo4e_v == &v && also_map.get(*code) == Some(&also_v)
                                        })
                                        .map(|(code, _)| code.clone())
                                })
                            }
                            _ => None,
                        };
                        joint
                            .or_else(|| {
                                // Reverse lookup: find EDIFACT key for BO4E value
                                map.iter()
                                    .find(|(_, bo4e_v)| *bo4e_v == &v)
                                    .map(|(edifact_k, _)| edifact_k.clone())
                            })
                            .or(Some(v))
                    }
                    (v, _) => v,
                };
                mapped_val.or_else(|| default.cloned())
            };

            if let Some(val) = val {
                field_values.push((
                    seg_key.clone(),
                    seg_tag.clone(),
                    element_idx,
                    component_idx,
                    val,
                ));
            }

            // If there's a qualifier, also inject it at elements[0][0]
            if let Some(q) = qualifier {
                if injected_qualifiers.insert(seg_key.clone()) {
                    field_values.push((seg_key, seg_tag, 0, 0, q.to_string()));
                }
            }
        }

        // Per-segment phantom prevention for qualified segments: remove entries
        // for segments using tag[qualifier] syntax (e.g., FTX[ACB], DTM[Z07])
        // that have data fields but none resolved to actual BO4E values.  This
        // prevents phantom segments when a definition maps multiple segment types
        // and optional qualified segments are not in the original message.
        // Unqualified segments (plain tags like SEQ, IDE) are always kept — they
        // are typically entry/mandatory segments of their group.
        field_values.retain(|(seg_key, _, _, _, _)| {
            if !seg_key.contains('[') {
                return true; // unqualified segments always kept
            }
            !seg_has_data_field.contains(seg_key) || seg_has_real_data.contains(seg_key)
        });

        // If the definition has data fields but none resolved to actual BO4E values,
        // return an empty instance to prevent phantom segments for groups not
        // present in the original EDIFACT message.  Definitions with only
        // qualifier/default fields (has_data_fields=false) are always kept.
        if has_data_fields && !has_real_data {
            return AssembledGroupInstance {
                segments: vec![],
                child_groups: vec![],
                entry_mig_number: None,
                variant_mig_numbers: vec![],
                skipped_segments: Vec::new(),
                skipped_positions: Vec::new(),
            };
        }

        // Build segments with elements/components in correct positions.
        // Group by segment_key to create separate segments for "DTM[92]" vs "DTM[93]".
        let mut segments: Vec<AssembledSegment> = Vec::with_capacity(field_values.len());
        let mut seen_keys: HashMap<String, usize> = HashMap::new();

        for (seg_key, seg_tag, element_idx, component_idx, val) in &field_values {
            let seg = if let Some(&pos) = seen_keys.get(seg_key) {
                &mut segments[pos]
            } else {
                let pos = segments.len();
                seen_keys.insert(seg_key.clone(), pos);
                segments.push(AssembledSegment {
                    tag: seg_tag.clone(),
                    elements: vec![],
                    mig_number: None,
                    segment_number: None,
                });
                &mut segments[pos]
            };

            while seg.elements.len() <= *element_idx {
                seg.elements.push(vec![]);
            }
            while seg.elements[*element_idx].len() <= *component_idx {
                seg.elements[*element_idx].push(String::new());
            }
            seg.elements[*element_idx][*component_idx] = val.clone();
        }

        // Pad intermediate empty elements: any [] between position 0 and the last
        // populated position becomes [""] so the EDIFACT renderer emits the `+` separator.
        for seg in &mut segments {
            let last_populated = seg.elements.iter().rposition(|e| !e.is_empty());
            if let Some(last_idx) = last_populated {
                for i in 0..last_idx {
                    if seg.elements[i].is_empty() {
                        seg.elements[i] = vec![String::new()];
                    }
                }
            }
        }

        // MIG-aware trailing padding: extend each segment to the MIG-defined element count.
        if let Some(ref ss) = self.segment_structure {
            for seg in &mut segments {
                if let Some(expected) = ss.element_count(&seg.tag) {
                    while seg.elements.len() < expected {
                        seg.elements.push(vec![String::new()]);
                    }
                }
            }
        }

        AssembledGroupInstance {
            segments,
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: Vec::new(),
            skipped_positions: Vec::new(),
        }
    }

    /// Resolve a field path within a segment to extract a value.
    ///
    /// Two path conventions are supported:
    ///
    /// **Named paths** (backward compatible):
    /// - 1-part `"d3227"` → elements\[0\]\[0\]
    /// - 2-part `"c517.d3225"` → elements\[1\]\[0\]
    ///
    /// **Numeric index paths** (for multi-component access):
    /// - `"0"` → elements\[0\]\[0\]
    /// - `"1.0"` → elements\[1\]\[0\]
    /// - `"1.2"` → elements\[1\]\[2\]
    fn resolve_field_path(segment: &AssembledSegment, path: &[&str]) -> Option<String> {
        if path.is_empty() {
            return None;
        }

        // Numeric paths only: index-based resolution.
        if let Ok(element_idx) = path[0].parse::<usize>() {
            let component_idx = if path.len() > 1 {
                path[1].parse::<usize>().unwrap_or(0)
            } else {
                0
            };
            return segment
                .elements
                .get(element_idx)?
                .get(component_idx)
                .filter(|v| !v.is_empty())
                .cloned();
        }

        // Non-numeric path[0] indicates an EDIFACT ID path that the PathResolver
        // failed to normalize (e.g. composite/element absent from any loaded PID
        // schema). Returning None lets the field be omitted from output instead
        // of silently guessing element index 1, which previously surfaced
        // unrelated data (e.g. NAD c819.d3229 read as c082.d3039 / rollencodenummer).
        None
    }

    /// Parse element and component indices from path parts after the segment tag.
    /// E.g., ["2"] -> (2, 0), ["0", "3"] -> (0, 3), ["1", "0"] -> (1, 0)
    pub(crate) fn parse_element_component(parts: &[&str]) -> (usize, usize) {
        if parts.is_empty() {
            return (0, 0);
        }
        let element_idx = parts[0].parse::<usize>().unwrap_or(0);
        let component_idx = if parts.len() > 1 {
            parts[1].parse::<usize>().unwrap_or(0)
        } else {
            0
        };
        (element_idx, component_idx)
    }

    /// Extract a value from a BO4E JSON object by target field name.
    /// Supports dotted paths like "nested.field_name".
    pub fn populate_field(
        &self,
        bo4e_value: &serde_json::Value,
        target_field: &str,
    ) -> Option<String> {
        let mut current = bo4e_value;
        for part in target_field.split('.') {
            current = current.get(part)?;
        }
        // Handle enriched code objects: {"code": "Z15", "meaning": "..."}
        if let Some(code) = current.get("code").and_then(|v| v.as_str()) {
            return Some(code.to_string());
        }
        current.as_str().map(|s| s.to_string())
    }

    /// Build a segment from BO4E values using the reverse mapping.
    pub fn build_segment_from_bo4e(
        &self,
        bo4e_value: &serde_json::Value,
        segment_tag: &str,
        target_field: &str,
    ) -> AssembledSegment {
        let value = self.populate_field(bo4e_value, target_field);
        let elements = if let Some(val) = value {
            vec![vec![val]]
        } else {
            vec![]
        };
        AssembledSegment {
            tag: segment_tag.to_uppercase(),
            elements,
            mig_number: None,
            segment_number: None,
        }
    }

    // ── Multi-entity forward mapping ──

    /// Parse a discriminator string (e.g., "SEQ.0.0=Z79") and find the matching
    /// repetition index within the given group path.
    ///
    /// Discriminator format: `"TAG.element_idx.component_idx=expected_value"`
    /// Scans all repetitions of the leaf group and returns the first rep index
    /// where the entry segment matches.
    pub fn resolve_repetition(
        tree: &AssembledTree,
        group_path: &str,
        discriminator: &str,
    ) -> Option<usize> {
        let (spec, expected) = discriminator.split_once('=')?;
        let parts: Vec<&str> = spec.split('.').collect();
        if parts.len() != 3 {
            return None;
        }
        let tag = parts[0];
        let element_idx: usize = parts[1].parse().ok()?;
        let component_idx: usize = parts[2].parse().ok()?;

        // Navigate to the parent and get the leaf group with all its repetitions
        let path_parts: Vec<&str> = group_path.split('.').collect();

        let leaf_group = if path_parts.len() == 1 {
            let (group_id, _) = parse_group_spec(path_parts[0]);
            tree.groups.iter().find(|g| g.group_id == group_id)?
        } else {
            // Navigate to the parent instance, then find the leaf group
            let parent_parts = &path_parts[..path_parts.len() - 1];
            let mut current_instance = {
                let (first_id, first_rep) = parse_group_spec(parent_parts[0]);
                let first_group = tree.groups.iter().find(|g| g.group_id == first_id)?;
                first_group.repetitions.get(first_rep.unwrap_or(0))?
            };
            for part in &parent_parts[1..] {
                let (group_id, explicit_rep) = parse_group_spec(part);
                let child_group = current_instance
                    .child_groups
                    .iter()
                    .find(|g| g.group_id == group_id)?;
                current_instance = child_group.repetitions.get(explicit_rep.unwrap_or(0))?;
            }
            let (leaf_id, _) = parse_group_spec(path_parts.last()?);
            current_instance
                .child_groups
                .iter()
                .find(|g| g.group_id == leaf_id)?
        };

        // Scan all repetitions for the matching discriminator
        let expected_values: Vec<&str> = expected.split('|').collect();
        for (rep_idx, instance) in leaf_group.repetitions.iter().enumerate() {
            let matches = instance.segments.iter().any(|s| {
                s.tag.eq_ignore_ascii_case(tag)
                    && s.elements
                        .get(element_idx)
                        .and_then(|e| e.get(component_idx))
                        .map(|v| expected_values.iter().any(|ev| v == ev))
                        .unwrap_or(false)
            });
            if matches {
                return Some(rep_idx);
            }
        }

        None
    }

    /// Like `resolve_repetition`, but returns ALL matching rep indices instead of just the first.
    ///
    /// This is used for multi-Zeitscheibe support where multiple SG6 reps may match
    /// the same discriminator (e.g., multiple RFF+Z49 time slices).
    pub fn resolve_all_repetitions(
        tree: &AssembledTree,
        group_path: &str,
        discriminator: &str,
    ) -> Vec<usize> {
        let Some((spec, expected)) = discriminator.split_once('=') else {
            return Vec::new();
        };
        let parts: Vec<&str> = spec.split('.').collect();
        if parts.len() != 3 {
            return Vec::new();
        }
        let tag = parts[0];
        let element_idx: usize = match parts[1].parse() {
            Ok(v) => v,
            Err(_) => return Vec::new(),
        };
        let component_idx: usize = match parts[2].parse() {
            Ok(v) => v,
            Err(_) => return Vec::new(),
        };

        // Navigate to the parent and get the leaf group with all its repetitions
        let path_parts: Vec<&str> = group_path.split('.').collect();

        let leaf_group = if path_parts.len() == 1 {
            let (group_id, _) = parse_group_spec(path_parts[0]);
            match tree.groups.iter().find(|g| g.group_id == group_id) {
                Some(g) => g,
                None => return Vec::new(),
            }
        } else {
            let parent_parts = &path_parts[..path_parts.len() - 1];
            let mut current_instance = {
                let (first_id, first_rep) = parse_group_spec(parent_parts[0]);
                let first_group = match tree.groups.iter().find(|g| g.group_id == first_id) {
                    Some(g) => g,
                    None => return Vec::new(),
                };
                match first_group.repetitions.get(first_rep.unwrap_or(0)) {
                    Some(i) => i,
                    None => return Vec::new(),
                }
            };
            for part in &parent_parts[1..] {
                let (group_id, explicit_rep) = parse_group_spec(part);
                let child_group = match current_instance
                    .child_groups
                    .iter()
                    .find(|g| g.group_id == group_id)
                {
                    Some(g) => g,
                    None => return Vec::new(),
                };
                current_instance = match child_group.repetitions.get(explicit_rep.unwrap_or(0)) {
                    Some(i) => i,
                    None => return Vec::new(),
                };
            }
            let (leaf_id, _) = match path_parts.last() {
                Some(p) => parse_group_spec(p),
                None => return Vec::new(),
            };
            match current_instance
                .child_groups
                .iter()
                .find(|g| g.group_id == leaf_id)
            {
                Some(g) => g,
                None => return Vec::new(),
            }
        };

        // Parse optional occurrence index from expected value: "TN#1" → ("TN", Some(1))
        let (expected_raw, occurrence) = parse_discriminator_occurrence(expected);

        // Collect ALL matching rep indices
        let expected_values: Vec<&str> = expected_raw.split('|').collect();
        let mut result = Vec::new();
        for (rep_idx, instance) in leaf_group.repetitions.iter().enumerate() {
            let matches = instance.segments.iter().any(|s| {
                s.tag.eq_ignore_ascii_case(tag)
                    && s.elements
                        .get(element_idx)
                        .and_then(|e| e.get(component_idx))
                        .map(|v| expected_values.iter().any(|ev| v == ev))
                        .unwrap_or(false)
            });
            if matches {
                result.push(rep_idx);
            }
        }

        // If occurrence index specified, return only that match
        if let Some(occ) = occurrence {
            result.into_iter().nth(occ).into_iter().collect()
        } else {
            result
        }
    }

    /// Resolve a discriminated instance using source_path for parent navigation.
    ///
    /// Like `resolve_repetition` + `resolve_group_instance`, but navigates to the
    /// parent group via source_path qualifier suffixes. Returns the matching instance
    /// directly (not just a rep index) to avoid re-navigation in `map_forward_inner`.
    ///
    /// For example, `source_path = "sg4.sg8_z98.sg10"` with `discriminator = "CCI.2.0=ZB3"`
    /// navigates to the SG8 instance with SEQ qualifier Z98, then finds the SG10 rep
    /// where CCI element 2 component 0 equals "ZB3".
    /// Map all definitions against a tree, returning a JSON object with entity names as keys.
    ///
    /// For each definition:
    /// - Has discriminator → find matching rep via `resolve_repetition`, map single instance
    /// - Root-level (empty source_group) → map rep 0 as single object
    /// - No discriminator, 1 rep in tree → map as single object
    /// - No discriminator, multiple reps in tree → map ALL reps into a JSON array
    ///
    /// When multiple definitions share the same `entity` name, their fields are
    /// deep-merged into a single JSON object. This allows related TOML files
    /// (e.g., LOC location + SEQ info + SG10 characteristics) to contribute
    /// fields to the same BO4E entity.
    pub fn map_all_forward(&self, tree: &AssembledTree) -> serde_json::Value {
        self.map_all_forward_inner(tree, true).0
    }

    /// Like [`map_all_forward`](Self::map_all_forward) but with explicit
    /// `enrich_codes` control (when `false`, code fields are plain strings
    /// instead of `{"code": …, "meaning": …}` objects).
    pub fn map_all_forward_enriched(
        &self,
        tree: &AssembledTree,
        enrich_codes: bool,
    ) -> serde_json::Value {
        self.map_all_forward_inner(tree, enrich_codes).0
    }

    /// Inner implementation with enrichment control.
    ///
    /// Returns `(json_value, nesting_info)`, where `nesting_info` maps entity
    /// keys to the parent rep index for each child element (used by the reverse
    /// mapper to distribute nested group children among their parent reps).
    fn map_all_forward_inner(
        &self,
        tree: &AssembledTree,
        enrich_codes: bool,
    ) -> (
        serde_json::Value,
        std::collections::HashMap<String, Vec<usize>>,
    ) {
        self.map_all_forward_inner_with_tx(tree, enrich_codes, self.transaction_group.as_deref())
    }

    /// Like `map_all_forward_inner` but with an explicit transaction-group
    /// override. Used by `map_interchange`, which knows the tx group even when
    /// the caller-supplied tx_engine wasn't built with `with_transaction_group`.
    fn map_all_forward_inner_with_tx(
        &self,
        tree: &AssembledTree,
        enrich_codes: bool,
        tx_group_override: Option<&str>,
    ) -> (
        serde_json::Value,
        std::collections::HashMap<String, Vec<usize>>,
    ) {
        let mut result = serde_json::Map::new();
        let mut nesting_info: std::collections::HashMap<String, Vec<usize>> =
            std::collections::HashMap::new();

        for def in &self.definitions {
            // `parent_field` children are mapped inside their parent's instance
            // (see `extract_nested_children`), never as top-level entities.
            if def.meta.parent_field.is_some() {
                continue;
            }
            let entity = &def.meta.entity;

            let bo4e = if let Some(ref disc) = def.meta.discriminator {
                // Has discriminator — resolve to matching rep(s).
                // Use source_path navigation when qualifiers are present
                // (e.g., "sg4.sg8_z98.sg10" navigates to Z98's SG10 reps,
                //  "sg4.sg5_z17" finds all LOC+Z17 when there are multiple).
                let use_source_path = def
                    .meta
                    .source_path
                    .as_ref()
                    .is_some_and(|sp| has_source_path_qualifiers(sp));
                if use_source_path {
                    // Navigate via source_path, then filter by discriminator.
                    let sp = def.meta.source_path.as_deref().unwrap();
                    let all_instances = Self::resolve_all_by_source_path(tree, sp);
                    // Apply discriminator filter to resolved instances (respects #N occurrence)
                    let instances: Vec<_> = if let Some(matcher) = DiscriminatorMatcher::parse(disc)
                    {
                        matcher.filter_instances(all_instances)
                    } else {
                        all_instances
                    };
                    let extract = |instance: &AssembledGroupInstance| {
                        let mut r = serde_json::Map::new();
                        self.extract_fields_from_instance(instance, def, &mut r, enrich_codes);
                        serde_json::Value::Object(r)
                    };
                    match instances.len() {
                        0 => None,
                        1 => Some(extract(instances[0])),
                        _ => Some(serde_json::Value::Array(
                            instances.iter().map(|i| extract(i)).collect(),
                        )),
                    }
                } else {
                    let reps = Self::resolve_all_repetitions(tree, &def.meta.source_group, disc);
                    match reps.len() {
                        0 => None,
                        1 => Some(self.map_forward_inner(tree, def, reps[0], enrich_codes)),
                        _ => Some(serde_json::Value::Array(
                            reps.iter()
                                .map(|&rep| self.map_forward_inner(tree, def, rep, enrich_codes))
                                .collect(),
                        )),
                    }
                }
            } else if def.meta.source_group.is_empty() {
                // Root-level mapping — always single object
                Some(self.map_forward_inner(tree, def, 0, enrich_codes))
            } else if def.meta.source_path.as_ref().is_some_and(|sp| {
                has_source_path_qualifiers(sp) || def.meta.source_group.contains('.')
            }) {
                // Multi-level source path — navigate via source_path to collect all
                // instances across all parent repetitions. Handles both qualified
                // paths (e.g., "sg4.sg8_zd7.sg10") and unqualified paths (e.g.,
                // "sg17.sg36.sg40") where multiple parent reps each have children.
                let sp = def.meta.source_path.as_deref().unwrap();
                let mut indexed = Self::resolve_all_with_parent_indices(tree, sp);

                // When the LAST part of source_path has no qualifier (e.g., "sg29.sg30"),
                // exclude reps that match a qualified sibling definition's qualifier
                // (e.g., "sg29.sg30_z35"). This prevents double-extraction when both
                // qualified and unqualified definitions target the same group.
                if let Some(last_part) = sp.rsplit('.').next() {
                    if !last_part.contains('_') {
                        // Collect qualifiers from sibling definitions that share the
                        // same base group name. E.g., for "sg29.sg30", only match
                        // "sg29.sg30_z35" (same base "sg30"), NOT "sg29.sg31_z35".
                        let base_prefix = if let Some(parent) = sp.rsplit_once('.') {
                            format!("{}.", parent.0)
                        } else {
                            String::new()
                        };
                        let sibling_qualifiers: Vec<String> = self
                            .definitions
                            .iter()
                            .filter_map(|d| d.meta.source_path.as_deref())
                            .filter(|other_sp| {
                                *other_sp != sp
                                    && other_sp.starts_with(&base_prefix)
                                    && other_sp.split('.').count() == sp.split('.').count()
                            })
                            .filter_map(|other_sp| {
                                let other_last = other_sp.rsplit('.').next()?;
                                // Only match siblings with the same base group name
                                // e.g., "sg30_z35" has base "sg30", must match "sg30"
                                let (base, q) = other_last.split_once('_')?;
                                if base == last_part {
                                    Some(q.to_string())
                                } else {
                                    None
                                }
                            })
                            .collect();

                        if !sibling_qualifiers.is_empty() {
                            indexed.retain(|(_, inst)| {
                                let entry_qual = inst
                                    .segments
                                    .first()
                                    .and_then(|seg| seg.elements.first())
                                    .and_then(|el| el.first())
                                    .map(|v| v.to_lowercase());
                                // Keep reps whose entry qualifier does NOT match
                                // any sibling's qualifier
                                !entry_qual.is_some_and(|q| {
                                    sibling_qualifiers.iter().any(|sq| {
                                        sq.split('_').any(|part| part.eq_ignore_ascii_case(&q))
                                    })
                                })
                            });
                        }
                    }
                }
                let extract = |instance: &AssembledGroupInstance| {
                    let mut r = serde_json::Map::new();
                    self.extract_fields_from_instance(instance, def, &mut r, enrich_codes);
                    serde_json::Value::Object(r)
                };
                // Track parent rep indices for nesting reconstruction.
                // Key by source_path (not entity or source_group) so that definitions
                // at different depths or with different qualifiers don't collide.
                // e.g., "sg5.sg8_z41.sg9" vs "sg5.sg8_z42.sg9" are distinct keys.
                if def.meta.source_group.contains('.') && !indexed.is_empty() {
                    if let Some(sp) = &def.meta.source_path {
                        let parent_indices: Vec<usize> =
                            indexed.iter().map(|(idx, _)| *idx).collect();
                        nesting_info.entry(sp.clone()).or_insert(parent_indices);

                        // Also store child rep indices (position within the leaf group)
                        // for depth-1 reverse placement. Key: "{sp}#child".
                        let child_key = format!("{sp}#child");
                        if let std::collections::hash_map::Entry::Vacant(e) =
                            nesting_info.entry(child_key)
                        {
                            let child_indices: Vec<usize> =
                                Self::compute_child_indices(tree, sp, &indexed);
                            if !child_indices.is_empty() {
                                e.insert(child_indices);
                            }
                        }
                    }
                }
                match indexed.len() {
                    0 => None,
                    1 => Some(extract(indexed[0].1)),
                    _ => Some(serde_json::Value::Array(
                        indexed.iter().map(|(_, i)| extract(i)).collect(),
                    )),
                }
            } else {
                let num_reps = Self::count_repetitions(tree, &def.meta.source_group);
                if num_reps <= 1 {
                    Some(self.map_forward_inner(tree, def, 0, enrich_codes))
                } else {
                    // Multiple reps, no discriminator — map all into array
                    let mut items = Vec::with_capacity(num_reps);
                    for rep in 0..num_reps {
                        items.push(self.map_forward_inner(tree, def, rep, enrich_codes));
                    }
                    Some(serde_json::Value::Array(items))
                }
            };

            if let Some(bo4e) = bo4e {
                let key = to_camel_case(entity);
                match def.meta.target_list.as_deref() {
                    Some(list_field) => append_to_list_field(&mut result, &key, list_field, bo4e),
                    None => deep_merge_insert(&mut result, &key, bo4e),
                }
            }
        }

        // Post-process: nest child entities under their parent entities.
        // E.g., Kontakt (source_group="SG2.SG3") moves under Marktteilnehmer (source_group="SG2").
        // Children whose parent group is the transaction root (e.g. SG4 for UTILMD) are
        // left at the top level — see MappingEngine::transaction_group.
        nest_child_entities_in_result(
            &mut result,
            &self.definitions,
            &nesting_info,
            tx_group_override,
        );

        (serde_json::Value::Object(result), nesting_info)
    }

    /// Reverse-map a BO4E entity map back to an AssembledTree.
    ///
    /// For each definition:
    /// 1. Look up entity in input by `meta.entity` name
    /// 2. If entity value is an array, map each element as a separate group repetition
    /// 3. Place results by `source_group`: `""` → root segments, `"SGn"` → groups
    ///
    /// This is the inverse of `map_all_forward()`.
    pub fn map_all_reverse(
        &self,
        entities: &serde_json::Value,
        nesting_info: Option<&std::collections::HashMap<String, Vec<usize>>>,
    ) -> AssembledTree {
        self.map_all_reverse_with_mig(entities, nesting_info, None)
    }

    /// [`map_all_reverse`](Self::map_all_reverse) with the PID-filtered MIG,
    /// which decides the parent of a nested child entity the BO4E JSON does
    /// not link to a parent (see the nesting step below).
    pub fn map_all_reverse_with_mig(
        &self,
        entities: &serde_json::Value,
        nesting_info: Option<&std::collections::HashMap<String, Vec<usize>>>,
        mig: Option<&MigSchema>,
    ) -> AssembledTree {
        let mut root_segments: Vec<AssembledSegment> = Vec::new();
        let mut groups: Vec<AssembledGroup> = Vec::new();
        // Track parent rep indices for child entities extracted from map-keyed
        // or array parents.  Used as fallback when nesting_info is empty.
        let mut inferred_nesting: std::collections::HashMap<String, Vec<usize>> =
            std::collections::HashMap::new();

        for def in &self.definitions {
            // `parent_field` children are reversed with their parent object
            // (see `reverse_nested_children`).
            if def.meta.parent_field.is_some() {
                continue;
            }
            let entity_key = to_camel_case(&def.meta.entity);

            // Look up entity value — first at top level, then nested under parent.
            // `_extracted` keeps the owned value alive for the borrow below.
            let _extracted: Option<serde_json::Value>;
            let entity_value = if let Some(list_field) = def.meta.target_list.as_deref() {
                // `target_list`: this definition's data is not the entity object,
                // it is the elements of a list field on it. Handing the array
                // straight to the array branch below turns each element back into
                // one group repetition, which is the exact inverse of the forward
                // "one repetition -> one element" rule.
                match entities.get(&entity_key).and_then(|e| e.get(list_field)) {
                    Some(v) if v.is_array() => {
                        _extracted = None;
                        v
                    }
                    _ => continue,
                }
            } else if let Some(v) = entities.get(&entity_key) {
                _extracted = None;
                v
            } else if def.meta.source_group.contains('.') {
                // Child entity not at top level — try extracting from parent entity
                match extract_child_from_parent_with_indices(entities, &self.definitions, def) {
                    Some((v, parent_indices)) => {
                        // Record inferred parent rep indices for nesting distribution
                        if let Some(sp) = def.meta.source_path.as_deref() {
                            inferred_nesting
                                .entry(sp.to_string())
                                .or_insert(parent_indices);
                        }
                        _extracted = Some(v);
                        _extracted.as_ref().unwrap()
                    }
                    None => continue,
                }
            } else {
                continue;
            };

            // Support map-keyed entities from typed PID format.
            // E.g., geschaeftspartner: {"Z04": {name1: "..."}} with discriminator NAD.0.0=Z04.
            // Extract inner value using discriminator's qualifier value as key,
            // and inject the qualifier into the inner object so companion fields find it.
            //
            // Also handles non-discriminated maps (e.g., marktteilnehmer: {"MS": {...}, "MR": {...}})
            // by converting them to arrays of inner values.
            let unwrapped: Option<serde_json::Value>;
            let entity_value = if entity_value.is_object() && !entity_value.is_array() {
                if let Some(disc_value) = def
                    .meta
                    .discriminator
                    .as_deref()
                    .and_then(|d| d.split_once('='))
                    .map(|(_, v)| v)
                {
                    // Discriminated definition: try to extract map key matching qualifier
                    if let Some(inner) = entity_value.get(disc_value) {
                        let mut injected = inner.clone();
                        // Find the field that maps to the discriminator's EDIFACT path
                        // and inject the map key as that field's value (e.g., nadQualifier = "Z04")
                        if let Some(qualifier_field) =
                            find_qualifier_companion_field(&self.definitions, &def.meta.entity)
                        {
                            if let Some(obj) = injected.as_object_mut() {
                                let entry = obj
                                    .entry(qualifier_field)
                                    .or_insert(serde_json::Value::Null);
                                if entry.is_null() {
                                    *entry = serde_json::Value::String(disc_value.to_string());
                                }
                            }
                        }
                        unwrapped = Some(injected);
                        unwrapped.as_ref().unwrap()
                    } else {
                        entity_value
                    }
                } else if is_map_keyed_object(entity_value) {
                    // Non-discriminated definition: convert map to array
                    // e.g., marktteilnehmer: {"MS": {...}, "MR": {...}} → [{...}, {...}]
                    // Inject each map key into its inner object using the companion field
                    // that maps to the discriminator path (if identifiable from other defs).
                    let map = entity_value.as_object().unwrap();
                    let arr: Vec<serde_json::Value> = map
                        .iter()
                        .map(|(key, val)| {
                            let mut item = val.clone();
                            // Try to find a qualifier companion field from peer definitions
                            // that share this entity name and have a discriminator
                            if let Some(obj) = item.as_object_mut() {
                                if let Some(qualifier_field) = find_qualifier_companion_field(
                                    &self.definitions,
                                    &def.meta.entity,
                                ) {
                                    let entry = obj
                                        .entry(qualifier_field)
                                        .or_insert(serde_json::Value::Null);
                                    if entry.is_null() {
                                        *entry = serde_json::Value::String(key.clone());
                                    }
                                }
                            }
                            item
                        })
                        .collect();
                    unwrapped = Some(serde_json::Value::Array(arr));
                    unwrapped.as_ref().unwrap()
                } else {
                    entity_value
                }
            } else {
                entity_value
            };

            // Determine target group from source_group (use leaf part after last dot)
            let leaf_group = def
                .meta
                .source_group
                .rsplit('.')
                .next()
                .unwrap_or(&def.meta.source_group);

            if def.meta.source_group.is_empty() {
                // Root-level: reverse into root segments
                let instance = self.map_reverse(entity_value, def);
                root_segments.extend(instance.segments);
            } else if entity_value.is_array() {
                // Array entity: each element becomes a group repetition
                let arr = entity_value.as_array().unwrap();
                let reps: Vec<_> = arr.iter().map(|item| self.map_reverse(item, def)).collect();

                // Merge into existing group or create new one
                if let Some(existing) = groups.iter_mut().find(|g| g.group_id == leaf_group) {
                    existing.repetitions.extend(reps);
                } else {
                    groups.push(AssembledGroup {
                        group_id: leaf_group.to_string(),
                        repetitions: reps,
                    });
                }
            } else {
                // Single object: one repetition
                let instance = self.map_reverse(entity_value, def);

                if let Some(existing) = groups.iter_mut().find(|g| g.group_id == leaf_group) {
                    existing.repetitions.push(instance);
                } else {
                    groups.push(AssembledGroup {
                        group_id: leaf_group.to_string(),
                        repetitions: vec![instance],
                    });
                }
            }
        }

        // Post-process: move nested groups under their parent repetitions.
        // Definitions with multi-level source_group (e.g., "SG2.SG3") produce
        // top-level groups that must be nested inside their parent group.
        // Children are distributed sequentially among parent reps (child[i] → parent[i])
        // matching the forward mapper's extraction order.
        let nested_specs: Vec<(String, String)> = self
            .definitions
            .iter()
            .filter(|def| def.meta.parent_field.is_none())
            .filter_map(|def| {
                let parts: Vec<&str> = def.meta.source_group.split('.').collect();
                if parts.len() > 1 {
                    Some((parts[0].to_string(), parts[parts.len() - 1].to_string()))
                } else {
                    None
                }
            })
            .collect();
        for (parent_id, child_id) in &nested_specs {
            // Only nest if both parent and child exist at the top level
            let has_parent = groups.iter().any(|g| g.group_id == *parent_id);
            let has_child = groups.iter().any(|g| g.group_id == *child_id);
            if has_parent && has_child {
                let child_idx = groups.iter().position(|g| g.group_id == *child_id).unwrap();
                let child_group = groups.remove(child_idx);
                let parent = groups
                    .iter_mut()
                    .find(|g| g.group_id == *parent_id)
                    .unwrap();
                // Distribute child reps among parent reps using nesting info
                // if available, falling back to all-under-first when not.
                // Nesting info is keyed by source_path (e.g., "sg2.sg3").
                let child_source_path = self
                    .definitions
                    .iter()
                    .find(|d| {
                        let parts: Vec<&str> = d.meta.source_group.split('.').collect();
                        d.meta.parent_field.is_none()
                            && parts.len() > 1
                            && parts[parts.len() - 1] == *child_id
                    })
                    .and_then(|d| d.meta.source_path.as_deref());
                let distribution = child_source_path.and_then(|key| {
                    nesting_info
                        .and_then(|ni| ni.get(key))
                        .or_else(|| inferred_nesting.get(key))
                });
                // Without a link from the JSON, the parent follows from the MIG:
                // the first repetition (in MIG variant order) whose variant
                // defines this child group — e.g. the SG2 NAD+MS repetition for
                // the sender's SG3 contact. Not "the first array element": BO4E
                // carries no ordering information.
                let unlinked_target = mig
                    .and_then(|m| {
                        mig_assembly::repetition_order::preferred_parent_repetition(
                            parent,
                            &m.segment_groups,
                            child_id,
                        )
                    })
                    .unwrap_or(0);
                for (i, child_rep) in child_group.repetitions.into_iter().enumerate() {
                    let target_idx = distribution
                        .and_then(|dist| dist.get(i))
                        .copied()
                        .unwrap_or(unlinked_target);

                    if let Some(target_rep) = parent.repetitions.get_mut(target_idx) {
                        if let Some(existing) = target_rep
                            .child_groups
                            .iter_mut()
                            .find(|g| g.group_id == *child_id)
                        {
                            existing.repetitions.push(child_rep);
                        } else {
                            target_rep.child_groups.push(AssembledGroup {
                                group_id: child_id.clone(),
                                repetitions: vec![child_rep],
                            });
                        }
                    }
                }
            }
        }

        let post_group_start = root_segments.len();
        AssembledTree {
            segments: root_segments,
            groups,
            post_group_start,
            inter_group_segments: std::collections::BTreeMap::new(),
        }
    }

    /// Count the number of repetitions available for a group path in the tree.
    fn count_repetitions(tree: &AssembledTree, group_path: &str) -> usize {
        let parts: Vec<&str> = group_path.split('.').collect();

        let (first_id, first_rep) = parse_group_spec(parts[0]);
        let first_group = match tree.groups.iter().find(|g| g.group_id == first_id) {
            Some(g) => g,
            None => return 0,
        };

        if parts.len() == 1 {
            return first_group.repetitions.len();
        }

        // Navigate to parent, then count leaf group reps
        let mut current_instance = match first_group.repetitions.get(first_rep.unwrap_or(0)) {
            Some(i) => i,
            None => return 0,
        };

        for (i, part) in parts[1..].iter().enumerate() {
            let (group_id, explicit_rep) = parse_group_spec(part);
            let child_group = match current_instance
                .child_groups
                .iter()
                .find(|g| g.group_id == group_id)
            {
                Some(g) => g,
                None => return 0,
            };

            if i == parts.len() - 2 {
                // Last part — return rep count
                return child_group.repetitions.len();
            }
            current_instance = match child_group.repetitions.get(explicit_rep.unwrap_or(0)) {
                Some(i) => i,
                None => return 0,
            };
        }

        0
    }

    /// Translate an assembled tree into BO4E, without code enrichment.
    ///
    /// This is the translation proper: every code field is a plain string, as it
    /// appears in the EDIFACT message. Enrichment (`{code, meaning, enum}`) is a
    /// display concern and is applied separately by [`Self::enrich_bo4e_types`],
    /// so a caller that does not need it never pays for it and never has to
    /// strip it back out.
    pub fn translate_edifact_to_bo4e(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        tree: &AssembledTree,
        transaction_group: &str,
    ) -> crate::model::MappedMessage {
        Self::map_interchange_inner(msg_engine, tx_engine, tree, transaction_group, false)
    }

    /// Decorate code fields of an already-translated message in place.
    ///
    /// Replaces the plain string at each code position with
    /// `{"code": …, "meaning": …, "enum": …}`. Needs a [`CodeLookup`] on the
    /// engines; without one this is a no-op, which is why CI — which never
    /// attaches a lookup — sees the unenriched shape.
    ///
    /// Works from the mapping definitions rather than from the EDIFACT tree: a
    /// definition knows both where a value came from (`source_path` plus the
    /// segment/element coordinates of the field) and where it went (`target`),
    /// which is all the lookup needs. The original EDIFACT value is recovered by
    /// inverting `enum_map` the same way the reverse mapper does, including the
    /// `also_target` disambiguation for codes that share a primary value.
    pub fn enrich_bo4e_types(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        mapped: &mut crate::model::MappedMessage,
    ) {
        msg_engine.enrich_entities(&mut mapped.stammdaten);
        for tx in &mut mapped.transaktionen {
            tx_engine.enrich_entities(&mut tx.stammdaten);
        }

        // The metadata slots hold mapped entities too. The forward pass splits
        // them out of `stammdaten`, so walking `stammdaten` alone no longer
        // reaches them — and their code fields would silently stay plain.
        msg_engine.enrich_named_entity(
            &mut mapped.nachricht_meta,
            crate::model::MSG_METADATA_ENTITY,
        );
        for tx in &mut mapped.transaktionen {
            tx_engine
                .enrich_named_entity(&mut tx.transaktionsdaten, crate::model::TX_METADATA_ENTITY);
        }
    }

    /// Apply the code sites of one named entity to a value holding that entity.
    ///
    /// The entity-map walk keys on the enclosing object's field name; a metadata
    /// slot has no such name, so the entity is named explicitly here.
    fn enrich_named_entity(&self, value: &mut serde_json::Value, entity_key: &str) {
        if self.code_lookup.is_none() || value.is_null() {
            return;
        }
        let sites = self.code_sites();
        if let Some(entity_sites) = sites.get(entity_key) {
            Self::apply_sites(self, value, entity_sites);
        }
    }

    /// Apply this engine's code enrichment to one entity map.
    ///
    /// Entities are located by key at any depth, because the forward pass moves
    /// them after extraction: `nest_child_entities_in_result` puts children
    /// under their parents.
    fn enrich_entities(&self, value: &mut serde_json::Value) {
        if self.code_lookup.is_none() {
            return;
        }
        let sites: HashMap<String, Vec<CodeSite<'_>>> = self.code_sites();
        if sites.is_empty() {
            return;
        }
        Self::walk_and_enrich(self, value, &sites);
    }

    /// Every code-field position this engine's definitions write to, grouped by
    /// the entity key the value ends up under.
    fn code_sites(&self) -> HashMap<String, Vec<CodeSite<'_>>> {
        let Some(ref code_lookup) = self.code_lookup else {
            return HashMap::new();
        };
        let mut sites: HashMap<String, Vec<CodeSite<'_>>> = HashMap::new();

        for def in &self.definitions {
            let Some(ref source_path) = def.meta.source_path else {
                continue;
            };
            let entity_key = to_camel_case(&def.meta.entity);

            for (path, field_mapping) in &def.fields {
                let (target, enum_map, also_target, also_enum_map) = match field_mapping {
                    FieldMapping::Simple(t) => (t.as_str(), None, None, None),
                    FieldMapping::Structured(s) => (
                        s.target.as_str(),
                        self.table(s.enum_map.as_ref(), s.code_list.as_deref()),
                        s.also_target.as_deref(),
                        self.table(s.also_enum_map.as_ref(), s.also_code_list.as_deref()),
                    ),
                    FieldMapping::Nested(_) => continue,
                };
                if target.is_empty() {
                    continue;
                }

                let parts: Vec<&str> = path.split('.').collect();
                let (seg_tag, path_qualifier, _occ) = parse_tag_qualifier(parts[0]);
                let (element_idx, component_idx) = Self::parse_element_component(&parts[1..]);
                // Same predicate, and the same two qualifiers, as the pre-split
                // path in `extract_fields_from_instance`: the field key's own
                // qualifier selects the schema variant, the discriminator's only
                // where the key has none. Asking with one merged qualifier — as
                // this did — calls `cav[Z30]`'s device number a code field and
                // decorates it, which the pre-split path never did.
                let disc_qualifier = Self::discriminator_qualifier_for_tag(def, &seg_tag);
                if code_lookup
                    .enrichment_codes(
                        source_path,
                        &seg_tag,
                        path_qualifier,
                        disc_qualifier.as_deref(),
                        element_idx,
                        component_idx,
                    )
                    .is_none()
                {
                    continue;
                }

                sites.entry(entity_key.clone()).or_default().push(CodeSite {
                    target,
                    parent_field: def.meta.parent_field.as_deref(),
                    source_path,
                    seg_tag,
                    path_qualifier: path_qualifier.map(str::to_string),
                    disc_qualifier,
                    element_idx,
                    component_idx,
                    enum_map,
                    also_target,
                    also_enum_map,
                });
            }
        }
        sites
    }

    /// Descend through the result, enriching every object that sits under a key
    /// naming an entity this engine maps.
    fn walk_and_enrich(
        engine: &MappingEngine,
        value: &mut serde_json::Value,
        sites: &HashMap<String, Vec<CodeSite<'_>>>,
    ) {
        match value {
            serde_json::Value::Object(map) => {
                for (key, child) in map.iter_mut() {
                    if let Some(entity_sites) = sites.get(key.as_str()) {
                        Self::apply_sites(engine, child, entity_sites);
                    }
                    Self::walk_and_enrich(engine, child, sites);
                }
            }
            serde_json::Value::Array(items) => {
                for item in items.iter_mut() {
                    Self::walk_and_enrich(engine, item, sites);
                }
            }
            _ => {}
        }
    }

    /// Apply one entity's code sites to an entity value (an object, or an array
    /// of them when the group repeats).
    fn apply_sites(engine: &MappingEngine, value: &mut serde_json::Value, sites: &[CodeSite<'_>]) {
        match value {
            serde_json::Value::Array(items) => {
                for item in items.iter_mut() {
                    Self::apply_sites(engine, item, sites);
                }
            }
            serde_json::Value::Object(_) => {
                for site in sites {
                    match site.parent_field {
                        None => engine.enrich_one(value, site),
                        Some(field) => {
                            if let Some(nested) = value.get_mut(field) {
                                Self::apply_nested_site(engine, nested, site);
                            }
                        }
                    }
                }
            }
            _ => {}
        }
    }

    /// Apply one nested site to every element of the `parent_field` array.
    fn apply_nested_site(
        engine: &MappingEngine,
        value: &mut serde_json::Value,
        site: &CodeSite<'_>,
    ) {
        match value {
            serde_json::Value::Array(items) => {
                for item in items.iter_mut() {
                    Self::apply_nested_site(engine, item, site);
                }
            }
            serde_json::Value::Object(_) => engine.enrich_one(value, site),
            _ => {}
        }
    }

    /// Enrich a single position, if it currently holds a plain string.
    fn enrich_one(&self, entity: &mut serde_json::Value, site: &CodeSite<'_>) {
        let Some(ref code_lookup) = self.code_lookup else {
            return;
        };
        // Already an object means another definition enriched this position.
        let Some(mapped_val) = Self::read_plain_string(entity, site.target) else {
            return;
        };

        // Recover the EDIFACT value: the schema's codes are raw ("293"), while
        // the JSON holds the enum_map target ("BDEW").
        let raw = match site.enum_map {
            None => mapped_val.clone(),
            Some(map) => {
                let joint = match (site.also_target, site.also_enum_map) {
                    (Some(also), Some(also_map)) => {
                        Self::read_plain_string(entity, also).and_then(|also_v| {
                            map.iter()
                                .find(|(code, bo4e_v)| {
                                    *bo4e_v == &mapped_val && also_map.get(*code) == Some(&also_v)
                                })
                                .map(|(code, _)| code.clone())
                        })
                    }
                    _ => None,
                };
                joint
                    .or_else(|| {
                        map.iter()
                            .find(|(_, bo4e_v)| *bo4e_v == &mapped_val)
                            .map(|(code, _)| code.clone())
                    })
                    .unwrap_or_else(|| mapped_val.clone())
            }
        };

        let Some(codes) = code_lookup.enrichment_codes(
            site.source_path,
            &site.seg_tag,
            site.path_qualifier.as_deref(),
            site.disc_qualifier.as_deref(),
            site.element_idx,
            site.component_idx,
        ) else {
            return;
        };

        // Class C: PID self-reference stays a plain string.
        if let Some(ref pid) = self.current_pid {
            if codes.len() == 1 && codes.contains_key(pid.as_str()) {
                return;
            }
        }

        let enrichment = codes.get(&raw);
        let meaning = enrichment
            .map(|e| serde_json::Value::String(e.meaning.clone()))
            .unwrap_or(serde_json::Value::Null);

        let mut obj = serde_json::Map::new();
        obj.insert("code".into(), serde_json::json!(mapped_val));
        obj.insert("meaning".into(), meaning);
        if let Some(enum_key) = enrichment.and_then(|e| e.enum_key.as_ref()) {
            obj.insert("enum".into(), serde_json::json!(enum_key));
        }

        if let serde_json::Value::Object(map) = entity {
            set_nested_value_json(map, site.target, serde_json::Value::Object(obj));
        }
    }

    /// The string at a dotted target path, or `None` when it is absent or has
    /// already been replaced by an enrichment object.
    fn read_plain_string(entity: &serde_json::Value, target: &str) -> Option<String> {
        let mut current = entity;
        for part in target.split('.') {
            current = current.get(part)?;
        }
        current.as_str().map(str::to_string)
    }

    /// Map an assembled tree into message-level and transaction-level results.
    ///
    /// - `msg_engine`: MappingEngine loaded with message-level definitions (SG2, SG3, root segments)
    /// - `tx_engine`: MappingEngine loaded with transaction-level definitions (relative to SG4)
    /// - `tree`: The assembled tree for one message
    /// - `transaction_group`: The group ID that represents transactions (e.g., "SG4")
    ///
    /// Returns a `MappedMessage` with message stammdaten and per-transaction results.
    pub fn map_interchange(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        tree: &AssembledTree,
        transaction_group: &str,
        enrich_codes: bool,
    ) -> crate::model::MappedMessage {
        let mut mapped =
            Self::translate_edifact_to_bo4e(msg_engine, tx_engine, tree, transaction_group);
        if enrich_codes {
            Self::enrich_bo4e_types(msg_engine, tx_engine, &mut mapped);
        }
        mapped
    }

    /// The translation itself, with enrichment still inlined in the extraction.
    ///
    /// Retained so the split can be proven equivalent: `map_interchange_inner`
    /// with `enrich_codes = true` must produce exactly what
    /// `translate_edifact_to_bo4e` followed by `enrich_bo4e_types` produces.
    /// See `enrich_split_parity_test`.
    /// Test-only door onto the pre-split path, so the parity gate can compare
    /// the two. Not part of the public pipeline.
    #[doc(hidden)]
    pub fn map_interchange_inner_for_test(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        tree: &AssembledTree,
        transaction_group: &str,
        enrich_codes: bool,
    ) -> crate::model::MappedMessage {
        Self::map_interchange_inner(msg_engine, tx_engine, tree, transaction_group, enrich_codes)
    }

    pub(crate) fn map_interchange_inner(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        tree: &AssembledTree,
        transaction_group: &str,
        enrich_codes: bool,
    ) -> crate::model::MappedMessage {
        // Map message-level entities (also captures nesting info)
        let (stammdaten, nesting_info) = msg_engine.map_all_forward_inner(tree, enrich_codes);

        // Find the transaction group and map each repetition
        let transaktionen = tree
            .groups
            .iter()
            .find(|g| g.group_id == transaction_group)
            .map(|sg| {
                sg.repetitions
                    .iter()
                    .map(|instance| {
                        // Wrap the instance in its group so that definitions with
                        // source_group paths like "SG4.SG5" can resolve correctly.
                        let wrapped_tree = AssembledTree {
                            segments: vec![],
                            groups: vec![AssembledGroup {
                                group_id: transaction_group.to_string(),
                                repetitions: vec![instance.clone()],
                            }],
                            post_group_start: 0,
                            inter_group_segments: std::collections::BTreeMap::new(),
                        };

                        // Pass the transaction_group into the tx_engine so its direct
                        // children (Marktlokation etc.) stay top-level peers of
                        // Prozessdaten rather than nested under it.
                        let (tx_result, tx_nesting) = tx_engine.map_all_forward_inner_with_tx(
                            &wrapped_tree,
                            enrich_codes,
                            Some(transaction_group),
                        );

                        // Split the transaction's own metadata out of its
                        // business objects. The engine maps `Prozessdaten` like
                        // any other entity; it just does not belong among the
                        // BOs once mapped.
                        let mut tx_result = tx_result;
                        let transaktionsdaten = crate::model::take_entity(
                            &mut tx_result,
                            crate::model::TX_METADATA_ENTITY,
                        );

                        crate::model::MappedTransaktion {
                            stammdaten: tx_result,
                            transaktionsdaten,
                            nesting_info: tx_nesting,
                        }
                    })
                    .collect()
            })
            .unwrap_or_default();

        // Same split one level up: `Nachricht` is metadata about the message.
        let mut stammdaten = stammdaten;
        let nachricht_meta =
            crate::model::take_entity(&mut stammdaten, crate::model::MSG_METADATA_ENTITY);

        crate::model::MappedMessage {
            stammdaten,
            nachricht_meta,
            transaktionen,
            nesting_info,
            inter_group_segments: tree.inter_group_segments.clone(),
        }
    }

    /// Reverse-map a `MappedMessage` back to an `AssembledTree`.
    ///
    /// Two-engine approach mirroring `map_interchange()`:
    /// - `msg_engine` handles message-level stammdaten → SG2/SG3 groups
    /// - `tx_engine` handles per-transaction stammdaten → SG4 instances
    ///
    /// All entities (including prozessdaten/nachricht) are in `tx.stammdaten`.
    /// Results are merged into one `AssembledGroupInstance` per transaction,
    /// collected into an SG4 `AssembledGroup`, then combined with message-level groups.
    pub fn map_interchange_reverse(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        mapped: &crate::model::MappedMessage,
        transaction_group: &str,
        filtered_mig: Option<&MigSchema>,
    ) -> AssembledTree {
        // Step 1: Reverse message-level stammdaten.
        //
        // The message's metadata entity goes back in here first: the forward
        // pass split `Nachricht` out into its own slot, but the definitions
        // resolve against one flat entity map, so without this the BGM/DTM
        // segments it feeds cannot be rebuilt. Clone only when there is
        // metadata to restore — keeps the common path zero-copy.
        let _owned_msg: Option<serde_json::Value>;
        let msg_stammdaten = if !mapped.nachricht_meta.is_null() {
            let mut merged = mapped.stammdaten.clone();
            crate::model::restore_entity(
                &mut merged,
                crate::model::MSG_METADATA_ENTITY,
                &mapped.nachricht_meta,
            );
            _owned_msg = Some(merged);
            _owned_msg.as_ref().unwrap()
        } else {
            _owned_msg = None;
            &mapped.stammdaten
        };

        let msg_tree = msg_engine.map_all_reverse_with_mig(
            msg_stammdaten,
            if mapped.nesting_info.is_empty() {
                None
            } else {
                Some(&mapped.nesting_info)
            },
            filtered_mig,
        );

        // Step 2: Build transaction instances from each Transaktion
        let mut sg4_reps: Vec<AssembledGroupInstance> = Vec::new();

        // Collect all definitions with their relative paths and sort by depth.
        // Shallower paths (SG8) must be processed before deeper ones (SG8:0.SG10)
        // so that parent group repetitions exist before children are added.
        struct DefWithMeta<'a> {
            def: &'a MappingDefinition,
            relative: String,
            depth: usize,
        }

        let mut sorted_defs: Vec<DefWithMeta> = tx_engine
            .definitions
            .iter()
            // `parent_field` children are reversed with their parent object
            // (see `reverse_nested_children`).
            .filter(|def| def.meta.parent_field.is_none())
            .map(|def| {
                let relative = strip_tx_group_prefix(&def.meta.source_group, transaction_group);
                let depth = if relative.is_empty() {
                    0
                } else {
                    relative.chars().filter(|c| *c == '.').count() + 1
                };
                DefWithMeta {
                    def,
                    relative,
                    depth,
                }
            })
            .collect();

        // Build parent source_path → rep_index map from deeper definitions.
        // SG10 defs like "SG4.SG8:0.SG10" with source_path "sg4.sg8_z79.sg10"
        // tell us that the SG8 def with source_path "sg4.sg8_z79" should be rep 0.
        let mut parent_rep_map: std::collections::HashMap<String, usize> =
            std::collections::HashMap::new();
        for dm in &sorted_defs {
            if dm.depth >= 2 {
                let parts: Vec<&str> = dm.relative.split('.').collect();
                let (_, parent_rep) = parse_group_spec(parts[0]);
                if let Some(rep_idx) = parent_rep {
                    if let Some(sp) = &dm.def.meta.source_path {
                        if let Some((parent_path, _)) = sp.rsplit_once('.') {
                            parent_rep_map
                                .entry(parent_path.to_string())
                                .or_insert(rep_idx);
                        }
                    }
                }
            }
        }

        // Augment shallow definitions with explicit rep indices from the map,
        // but only for single-rep cases (no multi-rep — those use dynamic tracking).
        for dm in &mut sorted_defs {
            if dm.depth == 1 && !dm.relative.contains(':') {
                if let Some(sp) = &dm.def.meta.source_path {
                    if let Some(rep_idx) = parent_rep_map.get(sp.as_str()) {
                        dm.relative = format!("{}:{}", dm.relative, rep_idx);
                    }
                }
            }
        }

        // Sort: shallower depth first, so SG8 defs create reps before SG8:N.SG10 defs.
        // Within same depth, sort by MIG group position (if available) for correct emission order,
        // falling back to alphabetical relative path for deterministic ordering.
        //
        // For variant groups (SG8 with Z01/Z03/Z07 etc.), use per-variant MIG positions
        // extracted from each definition's source_path qualifier suffix (e.g., "sg4.sg8_z01" → "Z01").
        if let Some(mig) = filtered_mig {
            let mig_order = build_reverse_mig_group_order(mig, transaction_group);
            sorted_defs.sort_by(|a, b| {
                a.depth.cmp(&b.depth).then_with(|| {
                    let a_id = a.relative.split(':').next().unwrap_or(&a.relative);
                    let b_id = b.relative.split(':').next().unwrap_or(&b.relative);
                    // Try per-variant lookup from source_path (e.g., "sg4.sg8_z01" → "SG8_Z01")
                    let a_pos = variant_mig_position(a.def, a_id, &mig_order);
                    let b_pos = variant_mig_position(b.def, b_id, &mig_order);
                    a_pos.cmp(&b_pos).then(a.relative.cmp(&b.relative))
                })
            });
        } else {
            sorted_defs.sort_by(|a, b| a.depth.cmp(&b.depth).then(a.relative.cmp(&b.relative)));
        }

        for tx in &mapped.transaktionen {
            let mut root_segs: Vec<AssembledSegment> = Vec::new();
            let mut child_groups: Vec<AssembledGroup> = Vec::new();

            // `transaktionsdaten` is merged back for the same reason as the
            // message's metadata above — the definitions expect one flat map.
            let _owned_tx: Option<serde_json::Value>;
            let tx_stammdaten: &serde_json::Value = if !tx.transaktionsdaten.is_null() {
                let mut merged = tx.stammdaten.clone();
                crate::model::restore_entity(
                    &mut merged,
                    crate::model::TX_METADATA_ENTITY,
                    &tx.transaktionsdaten,
                );
                _owned_tx = Some(merged);
                _owned_tx.as_ref().unwrap()
            } else {
                _owned_tx = None;
                &tx.stammdaten
            };

            // Track source_path → repetition indices for parent groups (top-down).
            // Built during depth-1 processing, used by depth-2+ defs without
            // explicit rep indices to find their correct parent via source_path.
            // Vec<usize> supports multi-rep parents (e.g., two SG8+ZF3 reps).
            let mut source_path_to_rep: std::collections::HashMap<String, Vec<usize>> =
                std::collections::HashMap::new();

            for dm in &sorted_defs {
                // Determine the BO4E value to reverse-map from.
                // Check top level first, then nested under parent entity.
                let entity_key = to_camel_case(&dm.def.meta.entity);
                let _tx_extracted: Option<serde_json::Value>;
                let bo4e_value = if let Some(v) = tx_stammdaten.get(&entity_key) {
                    _tx_extracted = None;
                    v
                } else if dm.def.meta.source_group.contains('.') {
                    match extract_child_from_parent(tx_stammdaten, &tx_engine.definitions, dm.def) {
                        Some(v) => {
                            _tx_extracted = Some(v);
                            _tx_extracted.as_ref().unwrap()
                        }
                        None => continue,
                    }
                } else {
                    continue;
                };

                // Support map-keyed entities from typed PID format (same logic as map_all_reverse).
                let unwrapped_value: Option<serde_json::Value>;
                let bo4e_value = if bo4e_value.is_object() && !bo4e_value.is_array() {
                    if let Some(disc_value) = dm
                        .def
                        .meta
                        .discriminator
                        .as_deref()
                        .and_then(|d| d.split_once('='))
                        .map(|(_, v)| v)
                    {
                        if let Some(inner) = bo4e_value.get(disc_value) {
                            let mut injected = inner.clone();
                            if let Some(qualifier_field) = find_qualifier_companion_field(
                                &tx_engine.definitions,
                                &dm.def.meta.entity,
                            ) {
                                if let Some(obj) = injected.as_object_mut() {
                                    obj.entry(qualifier_field).or_insert_with(|| {
                                        serde_json::Value::String(disc_value.to_string())
                                    });
                                }
                            }
                            unwrapped_value = Some(injected);
                            unwrapped_value.as_ref().unwrap()
                        } else {
                            bo4e_value
                        }
                    } else if is_map_keyed_object(bo4e_value) {
                        let map = bo4e_value.as_object().unwrap();
                        let arr: Vec<serde_json::Value> = map
                            .iter()
                            .map(|(key, val)| {
                                let mut item = val.clone();
                                if let Some(obj) = item.as_object_mut() {
                                    if let Some(qualifier_field) = find_qualifier_companion_field(
                                        &tx_engine.definitions,
                                        &dm.def.meta.entity,
                                    ) {
                                        let entry = obj
                                            .entry(qualifier_field)
                                            .or_insert(serde_json::Value::Null);
                                        if entry.is_null() {
                                            *entry = serde_json::Value::String(key.clone());
                                        }
                                    }
                                }
                                item
                            })
                            .collect();
                        unwrapped_value = Some(serde_json::Value::Array(arr));
                        unwrapped_value.as_ref().unwrap()
                    } else {
                        bo4e_value
                    }
                } else {
                    bo4e_value
                };

                // Handle array entities: each element becomes a separate group rep.
                // This supports both the NAD/SG12 pattern (multiple qualifiers) and
                // the multi-rep pattern (e.g., two LOC+Z17 Messlokationen).
                let items: Vec<&serde_json::Value> = if bo4e_value.is_array() {
                    bo4e_value.as_array().unwrap().iter().collect()
                } else {
                    vec![bo4e_value]
                };

                for (item_idx, item) in items.iter().enumerate() {
                    let instance = tx_engine.map_reverse(item, dm.def);

                    // Skip empty instances (definition had no real BO4E data)
                    if instance.segments.is_empty() && instance.child_groups.is_empty() {
                        continue;
                    }

                    if dm.relative.is_empty() {
                        // The definition maps the transaction group itself
                        // (CONTRL's SG1, UTILMD's SG4): its segments are the
                        // instance's own root segments. Children it nested with
                        // `parent_field` (CONTRL SG1.SG2, the UCS/UCD errors of
                        // this checked message) are already built as child
                        // groups of that instance and must travel with it.
                        root_segs.extend(instance.segments);
                        for child in instance.child_groups {
                            match child_groups
                                .iter_mut()
                                .find(|g| g.group_id == child.group_id)
                            {
                                Some(existing) => existing.repetitions.extend(child.repetitions),
                                None => child_groups.push(child),
                            }
                        }
                    } else {
                        // For depth-2+ defs without explicit rep index, resolve
                        // parent rep from source_path matching (qualifier-based).
                        // item_idx selects the correct parent rep for multi-rep entities.
                        let effective_relative = if dm.depth >= 2 {
                            // Multi-rep: strip hardcoded parent :N indices so
                            // resolve_child_relative uses source_path lookup instead.
                            let rel = if items.len() > 1 {
                                strip_all_rep_indices(&dm.relative)
                            } else {
                                dm.relative.clone()
                            };
                            // Use tx nesting info for multi-rep arrays, BUT skip it
                            // when source_path is present and resolves to a single
                            // parent rep. In that case, nesting_info indices (from the
                            // original tree) may not match the reverse tree's rep layout.
                            // resolve_child_relative uses reverse-tree source_path_to_rep
                            // which is always correct.
                            let skip_nesting = dm
                                .def
                                .meta
                                .source_path
                                .as_ref()
                                .and_then(|sp| sp.rsplit_once('.'))
                                .and_then(|(parent_path, _)| source_path_to_rep.get(parent_path))
                                .is_some_and(|reps| reps.len() == 1);
                            let nesting_idx = if items.len() > 1 && !skip_nesting {
                                dm.def
                                    .meta
                                    .source_path
                                    .as_ref()
                                    .and_then(|sp| tx.nesting_info.get(sp))
                                    .and_then(|dist| dist.get(item_idx))
                                    .copied()
                            } else {
                                None
                            };
                            if let Some(parent_rep) = nesting_idx {
                                // Direct placement using known nesting distribution
                                let parts: Vec<&str> = rel.split('.').collect();
                                let parent_id = parts[0].split(':').next().unwrap_or(parts[0]);
                                let rest = parts[1..].join(".");
                                format!("{}:{}.{}", parent_id, parent_rep, rest)
                            } else {
                                resolve_child_relative(
                                    &rel,
                                    dm.def.meta.source_path.as_deref(),
                                    &source_path_to_rep,
                                    item_idx,
                                )
                            }
                        } else if dm.depth == 1 {
                            // Depth-1: use nesting_info child indices for correct
                            // rep placement (preserves original interleaving order).
                            let child_key = dm
                                .def
                                .meta
                                .source_path
                                .as_ref()
                                .map(|sp| format!("{sp}#child"));
                            if let Some(child_indices) =
                                child_key.as_ref().and_then(|ck| tx.nesting_info.get(ck))
                            {
                                if let Some(&target) = child_indices.get(item_idx) {
                                    if target != usize::MAX {
                                        let base =
                                            dm.relative.split(':').next().unwrap_or(&dm.relative);
                                        format!("{}:{}", base, target)
                                    } else {
                                        dm.relative.clone()
                                    }
                                } else if items.len() > 1 && item_idx > 0 {
                                    strip_rep_index(&dm.relative)
                                } else {
                                    dm.relative.clone()
                                }
                            } else if items.len() > 1 && item_idx > 0 {
                                strip_rep_index(&dm.relative)
                            } else {
                                dm.relative.clone()
                            }
                        } else if items.len() > 1 && item_idx > 0 {
                            // Multi-rep entity with hardcoded :N index: first item uses
                            // the original index, subsequent items append (strip :N).
                            strip_rep_index(&dm.relative)
                        } else {
                            dm.relative.clone()
                        };

                        let rep_used =
                            place_in_groups(&mut child_groups, &effective_relative, instance);

                        // Track source_path → rep_index for depth-1 (parent) defs
                        if dm.depth == 1 {
                            if let Some(sp) = &dm.def.meta.source_path {
                                source_path_to_rep
                                    .entry(sp.clone())
                                    .or_default()
                                    .push(rep_used);
                            }
                        }
                    }
                }
            }

            sg4_reps.push(AssembledGroupInstance {
                segments: root_segs,
                child_groups,
                entry_mig_number: None,
                variant_mig_numbers: vec![],
                skipped_segments: Vec::new(),
                skipped_positions: Vec::new(),
            });
        }

        // Step 3: Combine message tree with transaction group.
        // Move UNS section separator from root segments to inter_group_segments.
        // UNS+D (detail) goes BEFORE the tx group (MSCONS: header/detail boundary).
        // UNS+S (summary) goes AFTER the tx group (ORDERS: detail/summary boundary).
        // Any segments that follow UNS in the sequence (e.g., summary MOA in REMADV)
        // are also placed in inter_group_segments alongside UNS.
        let mut root_segments = Vec::new();
        let mut uns_segments = Vec::new();
        let mut uns_is_summary = false;
        let mut found_uns = false;
        for seg in msg_tree.segments {
            if seg.tag == "UNS" {
                // Check if this is UNS+S (summary separator) vs UNS+D (detail separator)
                uns_is_summary = seg
                    .elements
                    .first()
                    .and_then(|el| el.first())
                    .map(|v| v == "S")
                    .unwrap_or(false);
                uns_segments.push(seg);
                found_uns = true;
            } else if found_uns {
                // Segments after UNS belong in the same inter_group position
                uns_segments.push(seg);
            } else {
                root_segments.push(seg);
            }
        }

        let pre_group_count = root_segments.len();
        let mut all_groups = msg_tree.groups;
        let mut inter_group = msg_tree.inter_group_segments;

        // Helper: parse SG number from group_id (e.g., "SG26" → 26).
        let sg_num = |id: &str| -> usize {
            id.strip_prefix("SG")
                .and_then(|n| n.parse::<usize>().ok())
                .unwrap_or(0)
        };

        if !sg4_reps.is_empty() {
            if uns_is_summary {
                // UNS+S: place AFTER the transaction group (detail/summary boundary)
                all_groups.push(AssembledGroup {
                    group_id: transaction_group.to_string(),
                    repetitions: sg4_reps,
                });
                if !uns_segments.is_empty() {
                    // Sort groups by SG number so the disassembler emits them
                    // in MIG order.  Insert UNS right after the tx_group —
                    // any groups with higher SG numbers (e.g., SG50/SG52 in
                    // INVOIC) are post-UNS summary groups.
                    all_groups.sort_by_key(|g| sg_num(&g.group_id));
                    let tx_num = sg_num(transaction_group);
                    let uns_pos = all_groups
                        .iter()
                        .rposition(|g| sg_num(&g.group_id) <= tx_num)
                        .map(|i| i + 1)
                        .unwrap_or(all_groups.len());
                    inter_group.insert(uns_pos, uns_segments);
                }
            } else {
                // UNS+D: place BEFORE the transaction group (header/detail boundary)
                if !uns_segments.is_empty() {
                    inter_group.insert(all_groups.len(), uns_segments);
                }
                all_groups.push(AssembledGroup {
                    group_id: transaction_group.to_string(),
                    repetitions: sg4_reps,
                });
            }
        } else if !uns_segments.is_empty() {
            if transaction_group.is_empty() {
                // Truly message-only (tx_group=""): UNS is a section separator.
                // UNS+S (summary) goes AFTER all groups — e.g., ORDCHG UNS+S
                // follows SG1 (NAD+CTA+COM) groups.
                // UNS+D (detail) goes BEFORE groups.
                all_groups.sort_by_key(|g| sg_num(&g.group_id));
                if uns_is_summary {
                    inter_group.insert(all_groups.len(), uns_segments);
                } else {
                    inter_group.insert(0, uns_segments);
                }
            } else {
                // Has a tx_group but no tx reps (e.g., INVOIC PID 31004
                // Storno — no SG26 data).  Sort groups and insert UNS after
                // the last group with SG number ≤ tx_group number.
                all_groups.sort_by_key(|g| sg_num(&g.group_id));
                let tx_num = sg_num(transaction_group);
                let uns_pos = all_groups
                    .iter()
                    .rposition(|g| sg_num(&g.group_id) <= tx_num)
                    .map(|i| i + 1)
                    .unwrap_or(all_groups.len());
                inter_group.insert(uns_pos, uns_segments);
            }
        }

        // Restore inter_group_segments captured during forward mapping
        // (e.g. PID-foreign top-level segments preserved by the assembler's
        // skip-unknown mode — see `Assembler::assemble_generic`). Without
        // this, BO4E forward + reverse drops anything not represented in a
        // TOML mapping definition. We append rather than overwrite so the
        // UNS placement computed above survives — same-key collisions are
        // rare in practice (UNS goes at well-known positions).
        for (k, segs) in &mapped.inter_group_segments {
            if segs.is_empty() {
                continue;
            }
            let existing_tags: std::collections::HashSet<String> = inter_group
                .get(k)
                .map(|v| v.iter().map(|s| s.tag.clone()).collect())
                .unwrap_or_default();
            for seg in segs {
                if existing_tags.contains(&seg.tag) {
                    continue;
                }
                inter_group.entry(*k).or_default().push(seg.clone());
            }
        }

        let mut tree = AssembledTree {
            segments: root_segments,
            groups: all_groups,
            post_group_start: pre_group_count,
            inter_group_segments: inter_group,
        };

        // Order repetitions of same-ID group variants (SG2 NAD+MS / NAD+MR,
        // SG12 NAD+Z07 / NAD+Z08, SG10 CCI variants, …) by MIG variant order.
        // The reps above were appended in definition order and, within one
        // definition, in the order of the BO4E JSON array — which carries no
        // ordering information. The transaction group itself keeps its order:
        // the order of transactions is data.
        if let Some(mig) = filtered_mig {
            mig_assembly::repetition_order::sort_repetitions_by_mig_variant(
                &mut tree,
                mig,
                (!transaction_group.is_empty()).then_some(transaction_group),
            );
        }
        tree
    }

    /// Build an assembled group from BO4E values and a definition.
    pub fn build_group_from_bo4e(
        &self,
        bo4e_value: &serde_json::Value,
        def: &MappingDefinition,
    ) -> AssembledGroup {
        let instance = self.map_reverse(bo4e_value, def);
        let leaf_group = def
            .meta
            .source_group
            .rsplit('.')
            .next()
            .unwrap_or(&def.meta.source_group);

        AssembledGroup {
            group_id: leaf_group.to_string(),
            repetitions: vec![instance],
        }
    }

    /// Forward-map an assembled tree to a typed interchange.
    ///
    /// Runs the dynamic mapping pipeline, wraps the result with metadata,
    /// then converts via JSON serialization into the caller's typed structs.
    ///
    /// - `M`: message-level stammdaten type (e.g., `Pid55001MsgStammdaten`)
    /// - `T`: transaction-level stammdaten type (e.g., `Pid55001TxStammdaten`)
    pub fn map_interchange_typed<M, T>(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        tree: &AssembledTree,
        tx_group: &str,
        enrich_codes: bool,
        nachrichtendaten: crate::model::Nachrichtendaten,
        interchangedaten: crate::model::Interchangedaten,
    ) -> Result<crate::model::Interchange<M, T>, serde_json::Error>
    where
        M: serde::de::DeserializeOwned,
        T: serde::de::DeserializeOwned,
    {
        let mapped = Self::map_interchange(msg_engine, tx_engine, tree, tx_group, enrich_codes);
        let nachricht = mapped.into_dynamic_nachricht(nachrichtendaten);
        let dynamic = crate::model::DynamicInterchange {
            interchangedaten,
            nachrichten: vec![nachricht],
        };
        let value = serde_json::to_value(&dynamic)?;
        serde_json::from_value(value)
    }

    /// Reverse-map a typed interchange nachricht back to an assembled tree.
    ///
    /// Serializes the typed struct to JSON, then runs the dynamic reverse pipeline.
    ///
    /// - `M`: message-level stammdaten type
    /// - `T`: transaction-level stammdaten type
    pub fn map_interchange_reverse_typed<M, T>(
        msg_engine: &MappingEngine,
        tx_engine: &MappingEngine,
        nachricht: &crate::model::Nachricht<M, T>,
        tx_group: &str,
    ) -> Result<AssembledTree, serde_json::Error>
    where
        M: serde::Serialize,
        T: serde::Serialize,
    {
        // The reverse resolves definitions against one flat entity map, so both
        // metadata slots go back where the mappings expect to find them.
        let mut stammdaten = serde_json::to_value(&nachricht.stammdaten)?;
        crate::model::restore_message_metadata(&mut stammdaten, &nachricht.nachrichtendaten);
        let transaktionen: Vec<crate::model::MappedTransaktion> = nachricht
            .transaktionen
            .iter()
            .map(|t| {
                Ok(crate::model::MappedTransaktion {
                    stammdaten: serde_json::to_value(t)?,
                    transaktionsdaten: serde_json::Value::Null,
                    nesting_info: Default::default(),
                })
            })
            .collect::<Result<Vec<_>, serde_json::Error>>()?;
        let mapped = crate::model::MappedMessage {
            stammdaten,
            nachricht_meta: serde_json::Value::Null,
            transaktionen,
            nesting_info: Default::default(),
            inter_group_segments: Default::default(),
        };
        Ok(Self::map_interchange_reverse(
            msg_engine, tx_engine, &mapped, tx_group, None,
        ))
    }
}

/// Parse a group path part with optional repetition: "SG8:1" → ("SG8", Some(1)).
/// Parse a source_path part into (group_id, optional_qualifier).
///
/// `"sg8_z98"` → `("sg8", Some("z98"))`
/// `"sg4"` → `("sg4", None)`
/// `"sg10"` → `("sg10", None)`
fn parse_source_path_part(part: &str) -> (&str, Option<&str>) {
    // Find the first underscore that separates group from qualifier.
    // Source path parts look like "sg8_z98", "sg4", "sg10", "sg12_z04".
    // The group ID is always "sgN", so the underscore after the digits is the separator.
    if let Some(pos) = part.find('_') {
        let group = &part[..pos];
        let qualifier = &part[pos + 1..];
        if !qualifier.is_empty() {
            return (group, Some(qualifier));
        }
    }
    (part, None)
}

/// Build a map from group ID (e.g., "SG5", "SG8") to its position index
/// within the transaction group's nested_groups Vec.
/// Used by `map_interchange_reverse` to sort definitions in MIG order.
///
/// For variant groups (same ID with variant_code set, e.g., SG8 with Z01, Z03, Z07),
/// stores per-variant positions (e.g., "SG8_Z01" → 0, "SG8_Z03" → 1) so that
/// definitions are sorted in MIG XML order rather than alphabetical qualifier order.
fn build_reverse_mig_group_order(mig: &MigSchema, tx_group_id: &str) -> HashMap<String, usize> {
    let mut order = HashMap::new();
    if let Some(tg) = mig.segment_groups.iter().find(|g| g.id == tx_group_id) {
        for (i, nested) in tg.nested_groups.iter().enumerate() {
            // For variant groups, store per-variant key (e.g., "SG8_Z01" → i)
            if let Some(ref vc) = nested.variant_code {
                let variant_key = format!("{}_{}", nested.id, vc.to_uppercase());
                order.insert(variant_key, i);
            }
            // Always store base group ID for fallback
            order.entry(nested.id.clone()).or_insert(i);
        }
    }
    order
}

/// Extract the MIG position for a definition, using per-variant lookup when possible.
///
/// For a definition with source_path "sg4.sg8_z01", extracts the variant qualifier "Z01"
/// and looks up "SG8_Z01" in the MIG order map. Falls back to the base group ID (e.g., "SG8")
/// if no variant qualifier is found or if the per-variant key isn't in the map.
fn variant_mig_position(
    def: &MappingDefinition,
    base_group_id: &str,
    mig_order: &HashMap<String, usize>,
) -> usize {
    // Try to extract variant qualifier from source_path.
    // source_path like "sg4.sg8_z01" or "sg4.sg8_z01.sg10" — we want the part matching base_group_id.
    if let Some(ref sp) = def.meta.source_path {
        // Find the path segment matching the base group (e.g., "sg8_z01" for base "SG8")
        let base_lower = base_group_id.to_lowercase();
        for part in sp.split('.') {
            if part.starts_with(&base_lower)
                || part.starts_with(base_group_id.to_lowercase().as_str())
            {
                // Extract qualifier suffix: "sg8_z01" → "z01"
                if let Some(underscore_pos) = part.find('_') {
                    let qualifier = &part[underscore_pos + 1..];
                    let variant_key = format!("{}_{}", base_group_id, qualifier.to_uppercase());
                    if let Some(&pos) = mig_order.get(&variant_key) {
                        return pos;
                    }
                }
            }
        }
    }
    // Fallback to base group position
    mig_order.get(base_group_id).copied().unwrap_or(usize::MAX)
}

/// Find a group repetition whose entry segment has a matching qualifier.
///
/// The entry segment is the first segment in the instance (e.g., SEQ for SG8).
/// The qualifier is matched against `elements[0][0]` (case-insensitive).
fn find_rep_by_entry_qualifier<'a>(
    reps: &'a [AssembledGroupInstance],
    qualifier: &str,
) -> Option<&'a AssembledGroupInstance> {
    // Support compound qualifiers like "za1_za2" — match any part.
    let parts: Vec<&str> = qualifier.split('_').collect();
    reps.iter().find(|inst| {
        inst.segments.first().is_some_and(|seg| {
            seg.elements
                .first()
                .and_then(|e| e.first())
                .is_some_and(|v| parts.iter().any(|part| v.eq_ignore_ascii_case(part)))
        })
    })
}

/// Find ALL repetitions whose entry segment qualifier matches (case-insensitive).
fn find_all_reps_by_entry_qualifier<'a>(
    reps: &'a [AssembledGroupInstance],
    qualifier: &str,
) -> Vec<&'a AssembledGroupInstance> {
    // Support compound qualifiers like "za1_za2" — match any part.
    let parts: Vec<&str> = qualifier.split('_').collect();
    reps.iter()
        .filter(|inst| {
            inst.segments.first().is_some_and(|seg| {
                seg.elements
                    .first()
                    .and_then(|e| e.first())
                    .is_some_and(|v| parts.iter().any(|part| v.eq_ignore_ascii_case(part)))
            })
        })
        .collect()
}

/// Check if a source_path contains qualifier suffixes (e.g., "sg8_z98").
fn has_source_path_qualifiers(source_path: &str) -> bool {
    source_path.split('.').any(|part| {
        if let Some(pos) = part.find('_') {
            pos < part.len() - 1
        } else {
            false
        }
    })
}

fn parse_group_spec(part: &str) -> (&str, Option<usize>) {
    if let Some(colon_pos) = part.find(':') {
        let id = &part[..colon_pos];
        let rep = part[colon_pos + 1..].parse::<usize>().ok();
        (id, rep)
    } else {
        (part, None)
    }
}

/// Strip the transaction group prefix from a source_group path.
///
/// Given `source_group = "SG4.SG8:0.SG10"` and `tx_group = "SG4"`,
/// returns `"SG8:0.SG10"`.
/// Given `source_group = "SG4"` and `tx_group = "SG4"`, returns `""`.
fn strip_tx_group_prefix(source_group: &str, tx_group: &str) -> String {
    if source_group == tx_group || source_group.is_empty() {
        String::new()
    } else if let Some(rest) = source_group.strip_prefix(tx_group) {
        rest.strip_prefix('.').unwrap_or(rest).to_string()
    } else {
        source_group.to_string()
    }
}

/// Place a reverse-mapped group instance into the correct nesting position.
///
/// `relative_path` is the group path relative to the transaction group:
/// - `"SG5"` → top-level child group
/// - `"SG8:0.SG10"` → SG10 inside SG8 repetition 0
///
/// Returns the repetition index used at the first nesting level.
fn place_in_groups(
    groups: &mut Vec<AssembledGroup>,
    relative_path: &str,
    instance: AssembledGroupInstance,
) -> usize {
    let parts: Vec<&str> = relative_path.split('.').collect();

    if parts.len() == 1 {
        // Leaf group: "SG5", "SG8", "SG12", or with explicit index "SG8:0"
        let (id, rep) = parse_group_spec(parts[0]);

        // Find or create the group
        let group = if let Some(g) = groups.iter_mut().find(|g| g.group_id == id) {
            g
        } else {
            groups.push(AssembledGroup {
                group_id: id.to_string(),
                repetitions: vec![],
            });
            groups.last_mut().unwrap()
        };

        if let Some(rep_idx) = rep {
            // Explicit index: place at specific position, merging into existing
            while group.repetitions.len() <= rep_idx {
                group.repetitions.push(AssembledGroupInstance {
                    segments: vec![],
                    child_groups: vec![],
                    entry_mig_number: None,
                    variant_mig_numbers: vec![],
                    skipped_segments: Vec::new(),
                    skipped_positions: Vec::new(),
                });
            }
            group.repetitions[rep_idx]
                .segments
                .extend(instance.segments);
            group.repetitions[rep_idx]
                .child_groups
                .extend(instance.child_groups);
            rep_idx
        } else {
            // No index: append new repetition
            let pos = group.repetitions.len();
            group.repetitions.push(instance);
            pos
        }
    } else {
        // Nested path: e.g., "SG8:0.SG10" → place SG10 inside SG8 rep 0
        let (parent_id, parent_rep) = parse_group_spec(parts[0]);
        let rep_idx = parent_rep.unwrap_or(0);

        // Find or create the parent group
        let parent_group = if let Some(g) = groups.iter_mut().find(|g| g.group_id == parent_id) {
            g
        } else {
            groups.push(AssembledGroup {
                group_id: parent_id.to_string(),
                repetitions: vec![],
            });
            groups.last_mut().unwrap()
        };

        // Ensure the target repetition exists (extend with empty instances if needed)
        while parent_group.repetitions.len() <= rep_idx {
            parent_group.repetitions.push(AssembledGroupInstance {
                segments: vec![],
                child_groups: vec![],
                entry_mig_number: None,
                variant_mig_numbers: vec![],
                skipped_segments: Vec::new(),
                skipped_positions: Vec::new(),
            });
        }

        let remaining = parts[1..].join(".");
        place_in_groups(
            &mut parent_group.repetitions[rep_idx].child_groups,
            &remaining,
            instance,
        );
        rep_idx
    }
}

/// Resolve the effective relative path for a child definition (depth >= 2).
///
/// If the child's relative already has an explicit parent rep index (e.g., "SG8:5.SG10"),
/// use it as-is. Otherwise, use the `source_path` to look up the parent's actual
/// repetition index from `source_path_to_rep`.
///
/// `item_idx` selects which parent rep to use when the parent created multiple reps
/// (e.g., two SG8 reps with ZF3 → item_idx 0 picks the first, 1 picks the second).
///
/// Example: relative = "SG8.SG10", source_path = "sg4.sg8_zf3.sg10"
/// → looks up "sg4.sg8_zf3" in map → finds reps [3, 4] → item_idx=1 → returns "SG8:4.SG10"
fn resolve_child_relative(
    relative: &str,
    source_path: Option<&str>,
    source_path_to_rep: &std::collections::HashMap<String, Vec<usize>>,
    item_idx: usize,
) -> String {
    let parts: Vec<&str> = relative.split('.').collect();
    if parts.is_empty() {
        return relative.to_string();
    }

    // If first part already has explicit index, keep as-is
    let (parent_id, parent_rep) = parse_group_spec(parts[0]);
    if parent_rep.is_some() {
        return relative.to_string();
    }

    // Try to resolve from source_path: extract parent path and look up its rep
    if let Some(sp) = source_path {
        if let Some((parent_path, _child)) = sp.rsplit_once('.') {
            // Exact match first.
            if let Some(rep_indices) = source_path_to_rep.get(parent_path) {
                let rep_idx = rep_indices
                    .get(item_idx)
                    .or_else(|| rep_indices.last())
                    .copied()
                    .unwrap_or(0);
                let rest = parts[1..].join(".");
                return format!("{}:{}.{}", parent_id, rep_idx, rest);
            }
            // Fallback: variant wildcard. When TOMLs use a flat parent path
            // like "sg4" but the schema splits it into variants (e.g. sg4_su,
            // sg4_z10..z21), union the reps from every matching variant so a
            // per-item iteration can place each child under its own parent.
            // `PidSchemaIndex::has_group` already accepts this style for
            // forward mapping — reverse mapping needs the same or children
            // from all-but-one variant get dropped (PARTIN 12 SG4 reps).
            let prefix = format!("{}_", parent_path);
            let mut unioned: Vec<usize> = source_path_to_rep
                .iter()
                .filter(|(k, _)| k.starts_with(&prefix))
                .flat_map(|(_, v)| v.iter().copied())
                .collect();
            if !unioned.is_empty() {
                unioned.sort_unstable();
                unioned.dedup();
                let rep_idx = unioned
                    .get(item_idx)
                    .or_else(|| unioned.last())
                    .copied()
                    .unwrap_or(0);
                let rest = parts[1..].join(".");
                return format!("{}:{}.{}", parent_id, rep_idx, rest);
            }
        }
    }

    // No resolution possible, keep original
    relative.to_string()
}

/// Parsed discriminator for filtering assembled group instances.
///
/// Discriminator format: "TAG.element_idx.component_idx=VALUE" or
/// "TAG.element_idx.component_idx=VAL1|VAL2" (pipe-separated multi-value).
/// E.g., "LOC.0.0=Z17" → match LOC segments where elements[0][0] == "Z17"
/// E.g., "RFF.0.0=Z49|Z53" → match RFF where elements[0][0] is Z49 OR Z53
struct DiscriminatorMatcher<'a> {
    tag: &'a str,
    element_idx: usize,
    component_idx: usize,
    expected_values: Vec<&'a str>,
    /// Optional occurrence index: `#N` selects the Nth match among instances.
    occurrence: Option<usize>,
}

impl<'a> DiscriminatorMatcher<'a> {
    fn parse(disc: &'a str) -> Option<Self> {
        let (spec, expected) = disc.split_once('=')?;
        let parts: Vec<&str> = spec.split('.').collect();
        if parts.len() != 3 {
            return None;
        }
        let (expected_raw, occurrence) = parse_discriminator_occurrence(expected);
        Some(Self {
            tag: parts[0],
            element_idx: parts[1].parse().ok()?,
            component_idx: parts[2].parse().ok()?,
            expected_values: expected_raw.split('|').collect(),
            occurrence,
        })
    }

    fn matches(&self, instance: &AssembledGroupInstance) -> bool {
        instance.segments.iter().any(|s| {
            s.tag.eq_ignore_ascii_case(self.tag)
                && s.elements
                    .get(self.element_idx)
                    .and_then(|e| e.get(self.component_idx))
                    .map(|v| self.expected_values.iter().any(|ev| v == ev))
                    .unwrap_or(false)
        })
    }

    /// Filter instances, respecting the occurrence index if present.
    fn filter_instances<'b>(
        &self,
        instances: Vec<&'b AssembledGroupInstance>,
    ) -> Vec<&'b AssembledGroupInstance> {
        let matching: Vec<_> = instances
            .into_iter()
            .filter(|inst| self.matches(inst))
            .collect();
        if let Some(occ) = self.occurrence {
            matching.into_iter().nth(occ).into_iter().collect()
        } else {
            matching
        }
    }
}

/// Parse an optional occurrence index from a discriminator expected value.
///
/// `"TN#1"` → `("TN", Some(1))` — select the 2nd matching rep
/// `"TN"`   → `("TN", None)` — select all matching reps
/// `"Z13|Z14#0"` → `("Z13|Z14", Some(0))` — first match among Z13 or Z14
fn parse_discriminator_occurrence(expected: &str) -> (&str, Option<usize>) {
    if let Some(hash_pos) = expected.rfind('#') {
        if let Ok(occ) = expected[hash_pos + 1..].parse::<usize>() {
            return (&expected[..hash_pos], Some(occ));
        }
    }
    (expected, None)
}

/// Strip explicit rep index from a relative path: "SG5:4" → "SG5", "SG8:3" → "SG8".
/// Used for multi-rep entities where subsequent items should append rather than
/// merge into the same rep position.
fn strip_rep_index(relative: &str) -> String {
    let (id, _) = parse_group_spec(relative);
    id.to_string()
}

/// Strip all explicit rep indices from a multi-part relative path:
/// "SG8:3.SG10" → "SG8.SG10", "SG8:3.SG10:0" → "SG8.SG10".
/// Used for multi-rep depth-2+ entities so resolve_child_relative uses
/// source_path lookup instead of hardcoded indices.
pub(crate) fn strip_all_rep_indices(relative: &str) -> String {
    relative
        .split('.')
        .map(|part| {
            let (id, _) = parse_group_spec(part);
            id
        })
        .collect::<Vec<_>>()
        .join(".")
}

// ── Nested child groups (`[meta] parent_field`) ──

/// Whether `child` is a `parent_field` definition nested directly below `parent`
/// (which may itself be a `parent_field` definition — nesting can span several
/// group levels, one `parent_field` per level):
/// same entity, `source_group` exactly one level deeper, and (when both carry a
/// `source_path`) a structurally compatible parent path. Qualifiers on the parent
/// part are compared only when both sides specify one; the instance-level check
/// is [`nested_parent_qualifier`] + [`entry_qualifier_matches`].
pub fn is_nested_child_of(child: &MappingDefinition, parent: &MappingDefinition) -> bool {
    if child.meta.parent_field.is_none() || child.meta.entity != parent.meta.entity {
        return false;
    }
    let child_sg = strip_all_rep_indices(&child.meta.source_group);
    let parent_sg = strip_all_rep_indices(&parent.meta.source_group);
    match child_sg.rsplit_once('.') {
        Some((head, _)) if head.eq_ignore_ascii_case(&parent_sg) => {}
        _ => return false,
    }
    let (Some(child_sp), Some(parent_sp)) = (
        child.meta.source_path.as_deref(),
        parent.meta.source_path.as_deref(),
    ) else {
        return true;
    };
    let Some((child_parent_sp, _)) = child_sp.rsplit_once('.') else {
        return false;
    };
    let child_parts: Vec<&str> = child_parent_sp.split('.').collect();
    let parent_parts: Vec<&str> = parent_sp.split('.').collect();
    child_parts.len() == parent_parts.len()
        && child_parts.iter().zip(&parent_parts).all(|(c, p)| {
            let (c_id, c_q) = parse_source_path_part(c);
            let (p_id, p_q) = parse_source_path_part(p);
            c_id.eq_ignore_ascii_case(p_id)
                && match (c_q, p_q) {
                    (Some(cq), Some(pq)) => cq.eq_ignore_ascii_case(pq),
                    _ => true,
                }
        })
}

/// Entry qualifier the parent group instance must carry for a nested child
/// definition to apply (e.g. `"z08"` for `source_path = "sg4.sg12_z08.sg13"`).
fn nested_parent_qualifier(child: &MappingDefinition) -> Option<&str> {
    let (parent_path, _) = child.meta.source_path.as_deref()?.rsplit_once('.')?;
    let last = parent_path.rsplit('.').next()?;
    parse_source_path_part(last).1
}

/// Leaf group id and optional entry qualifier of a nested child definition
/// (e.g. `("SG13", None)` for `source_group = "SG4.SG12.SG13"`).
fn nested_child_leaf(child: &MappingDefinition) -> (String, Option<&str>) {
    let leaf_group = strip_all_rep_indices(
        child
            .meta
            .source_group
            .rsplit('.')
            .next()
            .unwrap_or(&child.meta.source_group),
    );
    let leaf_qualifier = child
        .meta
        .source_path
        .as_deref()
        .and_then(|sp| sp.rsplit('.').next())
        .and_then(|part| parse_source_path_part(part).1);
    (leaf_group, leaf_qualifier)
}

/// Whether the instance's entry segment (its first segment) carries `qualifier`
/// at `elements[0][0]`. Compound qualifiers (`"z53_z54"`) match any part.
fn entry_qualifier_matches(instance: &AssembledGroupInstance, qualifier: &str) -> bool {
    segment_qualifier_matches(instance.segments.first(), qualifier)
}

/// [`entry_qualifier_matches`] for a group repetition rebuilt by the reverse
/// mapping from `def`. Its segments follow the order of `def`'s fields, so the
/// entry segment need not come first (e.g. `PIA` listed before `SEQ`). When
/// `def` has a discriminator, its segment tag names the entry segment.
fn rebuilt_entry_qualifier_matches(
    instance: &AssembledGroupInstance,
    def: &MappingDefinition,
    qualifier: &str,
) -> bool {
    let entry_tag = def
        .meta
        .discriminator
        .as_deref()
        .and_then(|d| d.split('.').next())
        .filter(|tag| !tag.is_empty());
    let entry = match entry_tag {
        Some(tag) => instance
            .segments
            .iter()
            .find(|s| s.tag.eq_ignore_ascii_case(tag)),
        None => instance.segments.first(),
    };
    segment_qualifier_matches(entry, qualifier)
}

fn segment_qualifier_matches(segment: Option<&AssembledSegment>, qualifier: &str) -> bool {
    segment
        .and_then(|seg| seg.elements.first())
        .and_then(|e| e.first())
        .is_some_and(|v| qualifier.split('_').any(|q| v.eq_ignore_ascii_case(q)))
}

/// Parse a segment tag with optional qualifier and occurrence index.
///
/// - `"dtm[92]"`    → `("DTM", Some("92"), 0)` — first (default) occurrence
/// - `"rff[Z34,1]"` → `("RFF", Some("Z34"), 1)` — second occurrence (0-indexed)
/// - `"rff[Z34,*]"` → `("RFF", Some("Z34"), 0)` — wildcard occurrence
/// - `"rff"`         → `("RFF", None, 0)`
pub(crate) fn parse_tag_qualifier(tag_part: &str) -> (String, Option<&str>, usize) {
    if let Some(bracket_start) = tag_part.find('[') {
        let tag = tag_part[..bracket_start].to_uppercase();
        let inner = tag_part[bracket_start + 1..].trim_end_matches(']');
        if let Some(comma_pos) = inner.find(',') {
            let qualifier = &inner[..comma_pos];
            let index = inner[comma_pos + 1..].parse::<usize>().unwrap_or(0);
            // "*" wildcard means no qualifier filter — positional access only
            if qualifier == "*" {
                (tag, None, index)
            } else {
                (tag, Some(qualifier), index)
            }
        } else {
            (tag, Some(inner), 0)
        }
    } else {
        (tag_part.to_uppercase(), None, 0)
    }
}

/// Deep-merge a BO4E value into the result map.
///
/// If the entity already exists as an object, new fields are merged in
/// (existing fields are NOT overwritten). This allows multiple TOML
/// definitions with the same `entity` name to contribute fields to one object.
pub fn deep_merge_insert(
    result: &mut serde_json::Map<String, serde_json::Value>,
    entity: &str,
    bo4e: serde_json::Value,
) {
    if let Some(existing) = result.get_mut(entity) {
        // Array + Array: element-wise merge (same entity from multiple TOML defs,
        // each producing an array for multi-rep groups like two LOC+Z17).
        if let (Some(existing_arr), Some(new_arr)) =
            (existing.as_array().map(|a| a.len()), bo4e.as_array())
        {
            if existing_arr == new_arr.len() {
                let existing_arr = existing.as_array_mut().unwrap();
                for (existing_elem, new_elem) in existing_arr.iter_mut().zip(new_arr) {
                    if let (Some(existing_map), Some(new_map)) =
                        (existing_elem.as_object_mut(), new_elem.as_object())
                    {
                        for (k, v) in new_map {
                            if let Some(existing_v) = existing_map.get_mut(k) {
                                if let (Some(existing_inner), Some(new_inner)) =
                                    (existing_v.as_object_mut(), v.as_object())
                                {
                                    for (ik, iv) in new_inner {
                                        existing_inner
                                            .entry(ik.clone())
                                            .or_insert_with(|| iv.clone());
                                    }
                                }
                            } else {
                                existing_map.insert(k.clone(), v.clone());
                            }
                        }
                    }
                }
                return;
            }
        }
        // Object + Object: field-level merge
        if let (Some(existing_map), serde_json::Value::Object(new_map)) =
            (existing.as_object_mut(), &bo4e)
        {
            for (k, v) in new_map {
                if let Some(existing_v) = existing_map.get_mut(k) {
                    // Recursively merge nested objects (e.g., companion types)
                    if let (Some(existing_inner), Some(new_inner)) =
                        (existing_v.as_object_mut(), v.as_object())
                    {
                        for (ik, iv) in new_inner {
                            existing_inner
                                .entry(ik.clone())
                                .or_insert_with(|| iv.clone());
                        }
                    }
                    // Don't overwrite existing scalar/array values
                } else {
                    existing_map.insert(k.clone(), v.clone());
                }
            }
            return;
        }
    }
    result.insert(entity.to_string(), bo4e);
}

/// Append a definition's per-repetition output to a **list-valued field** on an
/// entity — the write half of `MappingMeta::target_list`.
///
/// This cannot go through `deep_merge_insert`, which documents that it does
/// "not overwrite existing scalar/array values". That rule is right for ordinary
/// fields and wrong here: a list field is the one place where several
/// definitions are *expected* to contribute to the same key (four separate
/// `Obis*` definitions all feed `zaehlwerke`), and under `deep_merge_insert`
/// every contribution after the first would be dropped without a trace.
///
/// Empty elements are skipped so an absent optional group does not leave a
/// `[{}]` behind, which would reverse into a phantom segment.
fn append_to_list_field(
    result: &mut serde_json::Map<String, serde_json::Value>,
    entity: &str,
    list_field: &str,
    bo4e: serde_json::Value,
) {
    let mut items = match bo4e {
        serde_json::Value::Array(a) => a,
        other => vec![other],
    };
    items.retain(|v| !v.as_object().is_some_and(|o| o.is_empty()));
    if items.is_empty() {
        return;
    }
    let entry = result
        .entry(entity.to_string())
        .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
    // An entity carrying a list field is a single object. If it is already an
    // array, some other definition made it multi-rep and the two shapes are
    // incompatible — leave it alone rather than corrupt it silently.
    let Some(obj) = entry.as_object_mut() else {
        return;
    };
    match obj.get_mut(list_field).and_then(|v| v.as_array_mut()) {
        Some(existing) => existing.extend(items),
        None => {
            obj.insert(list_field.to_string(), serde_json::Value::Array(items));
        }
    }
}

/// Convert a PascalCase name to camelCase by lowering the first character.
///
/// E.g., `"Ansprechpartner"` → `"ansprechpartner"`,
/// `"AnsprechpartnerEdifact"` → `"ansprechpartnerEdifact"`,
/// `"ProduktpaketPriorisierung"` → `"produktpaketPriorisierung"`.
/// Detect whether a JSON object looks like a map-keyed entity (typed PID format).
///
/// Map-keyed objects have short uppercase/alphanumeric keys that look like qualifier
/// codes (e.g., `{"Z04": {...}, "Z09": {...}}` or `{"MS": {...}, "MR": {...}}`),
/// as opposed to normal field-name objects (e.g., `{"name1": "...", "adresse": {...}}`).
fn is_map_keyed_object(value: &serde_json::Value) -> bool {
    let Some(obj) = value.as_object() else {
        return false;
    };
    if obj.is_empty() {
        return false;
    }
    // All keys must be short (≤5 chars), uppercase/digit only, and all values must be objects
    obj.iter().all(|(k, v)| {
        k.len() <= 5
            && k.chars()
                .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit())
            && v.is_object()
    })
}

/// Find the BO4E companion field name used for the qualifier/discriminator
/// across definitions that share the same entity name.
///
/// For example, if `Geschaeftspartner` has a definition with discriminator
/// `NAD.0.0=Z04` and companion field `nad.0.0 → nadQualifier`, this returns
/// `Some("nadQualifier")`.
///
/// Used to inject map keys into inner objects when converting map-keyed entities.
fn find_qualifier_companion_field(
    definitions: &[crate::definition::MappingDefinition],
    entity: &str,
) -> Option<String> {
    for def in definitions {
        if def.meta.entity != *entity || def.meta.parent_field.is_some() {
            continue;
        }
        let disc = def.meta.discriminator.as_deref()?;
        let (disc_path, _) = disc.split_once('=')?;
        let disc_path_lower = disc_path.to_lowercase();

        // Search [fields] for the qualifier field (e.g., Marktteilnehmer has
        // "marktrolle" in [fields]).
        for (path, mapping) in &def.fields {
            let cf_path = path.to_lowercase();
            let matches = cf_path == disc_path_lower || format!("{}.0", cf_path) == disc_path_lower;
            if matches {
                let target = match mapping {
                    FieldMapping::Simple(t) => t.as_str(),
                    FieldMapping::Structured(s) => s.target.as_str(),
                    FieldMapping::Nested(_) => continue,
                };
                if !target.is_empty() {
                    return Some(target.to_string());
                }
            }
        }
    }
    None
}

/// Extract a child entity from its parent entity in the reverse mapping input.
///
/// When a child entity (e.g., Kontakt with source_group="SG2.SG3") isn't found
/// at the top level, look inside the parent entity (e.g., Marktteilnehmer with
/// source_group="SG2") for a nested field matching the child's camelCase name.
///
/// For map-keyed parents ({"MS": {...}, "MR": {...}}), collects child values
/// from all inner objects that have the field, returning them as an array.
fn extract_child_from_parent(
    entities: &serde_json::Value,
    definitions: &[MappingDefinition],
    child_def: &MappingDefinition,
) -> Option<serde_json::Value> {
    extract_child_from_parent_with_indices(entities, definitions, child_def).map(|(v, _)| v)
}

/// Like `extract_child_from_parent`, but also returns the parent rep indices
/// from which each child was extracted.  This allows the nesting distribution
/// to place child groups under the correct parent rep even when `nesting_info`
/// is unavailable (e.g., typed struct / manual JSON construction).
fn extract_child_from_parent_with_indices(
    entities: &serde_json::Value,
    definitions: &[MappingDefinition],
    child_def: &MappingDefinition,
) -> Option<(serde_json::Value, Vec<usize>)> {
    let parts: Vec<&str> = child_def.meta.source_group.split('.').collect();
    if parts.len() < 2 {
        return None;
    }
    let parent_group = parts[0];
    let parent_def = definitions
        .iter()
        .find(|d| d.meta.source_group == parent_group && d.meta.entity != child_def.meta.entity)?;
    let parent_key = to_camel_case(&parent_def.meta.entity);
    let child_key = to_camel_case(&child_def.meta.entity);
    let parent_value = entities.get(&parent_key)?;

    // Map-keyed parent: collect child from each inner object
    if let Some(parent_map) = parent_value.as_object() {
        if is_map_keyed_value(parent_map) {
            let mut children: Vec<serde_json::Value> = Vec::new();
            let mut indices: Vec<usize> = Vec::new();
            for (i, (_key, inner)) in parent_map.iter().enumerate() {
                if let Some(child) = inner.get(&child_key) {
                    if !child.is_null() {
                        children.push(child.clone());
                        indices.push(i);
                    }
                }
            }
            return match children.len() {
                0 => None,
                1 => Some((children.into_iter().next().unwrap(), indices)),
                _ => Some((serde_json::Value::Array(children), indices)),
            };
        }
    }

    // Array parent: collect child from each element
    if let Some(parent_arr) = parent_value.as_array() {
        let mut children: Vec<serde_json::Value> = Vec::new();
        let mut indices: Vec<usize> = Vec::new();
        for (i, item) in parent_arr.iter().enumerate() {
            if let Some(child) = item.get(&child_key) {
                if !child.is_null() {
                    children.push(child.clone());
                    indices.push(i);
                }
            }
        }
        return match children.len() {
            0 => None,
            1 => Some((children.into_iter().next().unwrap(), indices)),
            _ => Some((serde_json::Value::Array(children), indices)),
        };
    }

    // Single parent object — always index 0
    let child = parent_value.get(&child_key)?;
    if child.is_null() {
        return None;
    }
    Some((child.clone(), vec![0]))
}

/// Move child entities under their parent entities in the forward-mapped result.
///
/// For each definition with a dotted `source_group` (e.g., "SG2.SG3"), finds the
/// parent definition (e.g., "SG2") and moves the child entity from the top-level
/// result into the parent entity as a nested field.
fn nest_child_entities_in_result(
    result: &mut serde_json::Map<String, serde_json::Value>,
    definitions: &[MappingDefinition],
    nesting_info: &std::collections::HashMap<String, Vec<usize>>,
    transaction_group: Option<&str>,
) {
    let nesting_pairs = child_entity_nesting_pairs(definitions, transaction_group);

    for (_parent_group, parent_entity, child_entity, child_source_path) in nesting_pairs {
        let parent_key = to_camel_case(&parent_entity);
        let child_key = to_camel_case(&child_entity);

        // Remove child from top level (if present)
        let child_value = match result.remove(&child_key) {
            Some(v) => v,
            None => continue,
        };

        // Get parent value.
        // If the parent is a plain array (not map-keyed), nesting would silently
        // place the child into arbitrary array elements. Skip and leave the child
        // at the top level where the reverse mapper can find it.
        let Some(parent_value) = result.get_mut(&parent_key) else {
            // Parent doesn't exist — put child back
            result.insert(child_key, child_value);
            continue;
        };
        if parent_value.is_array() {
            result.insert(child_key, child_value);
            continue;
        }

        // Get the nesting distribution (which parent rep each child rep belongs to)
        let distribution = child_source_path
            .as_deref()
            .and_then(|sp| nesting_info.get(sp));

        // Normalize child to a list of (index, value) pairs
        let child_items: Vec<(usize, &serde_json::Value)> = match &child_value {
            serde_json::Value::Array(arr) => arr.iter().enumerate().collect(),
            other => vec![(0, other)],
        };

        // Helper: insert or append child value into a parent object field.
        // First call inserts the value; subsequent calls convert to array and append.
        let insert_or_append = |obj: &mut serde_json::Map<String, serde_json::Value>,
                                key: &str,
                                val: &serde_json::Value| {
            match obj.get_mut(key) {
                Some(existing) => {
                    // Convert single value to array, then push
                    if !existing.is_array() {
                        let prev = existing.take();
                        *existing = serde_json::Value::Array(vec![prev]);
                    }
                    if let Some(arr) = existing.as_array_mut() {
                        arr.push(val.clone());
                    }
                }
                None => {
                    obj.insert(key.to_string(), val.clone());
                }
            }
        };

        // Handle parent as map-keyed object: {"MS": {...}, "MR": {...}}
        if let Some(parent_map) = parent_value.as_object_mut() {
            if is_map_keyed_value(parent_map) {
                // Map keys in insertion order correspond to rep indices
                let keys: Vec<String> = parent_map.keys().cloned().collect();
                for (i, child_item) in &child_items {
                    let target_idx = distribution
                        .and_then(|dist| dist.get(*i))
                        .copied()
                        .unwrap_or(0);
                    if let Some(key) = keys.get(target_idx) {
                        if let Some(inner) = parent_map.get_mut(key).and_then(|v| v.as_object_mut())
                        {
                            insert_or_append(inner, &child_key, child_item);
                        }
                    }
                }
                continue;
            }
        }

        // Handle parent as array
        if let Some(parent_arr) = parent_value.as_array_mut() {
            for (i, child_item) in &child_items {
                let target_idx = distribution
                    .and_then(|dist| dist.get(*i))
                    .copied()
                    .unwrap_or(0);
                if let Some(parent_obj) = parent_arr
                    .get_mut(target_idx)
                    .and_then(|v| v.as_object_mut())
                {
                    insert_or_append(parent_obj, &child_key, child_item);
                }
            }
            continue;
        }

        // Handle parent as single object
        if let Some(parent_obj) = parent_value.as_object_mut() {
            for (_i, child_item) in &child_items {
                insert_or_append(parent_obj, &child_key, child_item);
            }
            continue;
        }

        // Fallback: put child back at top level
        result.insert(child_key, child_value);
    }
}

/// Parent/child entity pairs the forward mapping nests (see
/// [`nest_child_entities_in_result`]): `(parent_group, parent_entity,
/// child_entity, child_source_path)`.
///
/// A child entity (dotted `source_group`, e.g. `SG2.SG3` Kontakt) is moved into
/// the object of the entity mapped from its parent group (e.g. `SG2`
/// Marktteilnehmer) — unless the parent group is the transaction root, the
/// child also has a definition at the parent level (same-entity enrichment), or
/// the parent maps a dotted field of the child's name.
pub(crate) fn child_entity_nesting_pairs(
    definitions: &[MappingDefinition],
    transaction_group: Option<&str>,
) -> Vec<(String, String, String, Option<String>)> {
    // Collect parent→child relationships from definitions.
    // parent_group → (parent_entity, child_entity, child_source_path)
    let mut nesting_pairs: Vec<(String, String, String, Option<String>)> = Vec::new();
    for def in definitions {
        let parts: Vec<&str> = def.meta.source_group.split('.').collect();
        if parts.len() < 2 || def.meta.parent_field.is_some() {
            continue;
        }
        let parent_group = parts[0];
        // Skip nesting when the parent group is the transaction root. SG4 in UTILMD
        // IS the transaction — its direct children (Marktlokation, Geschaeftspartner,
        // ProduktpaketDaten, …) are peers of the transaction metadata (Prozessdaten),
        // not sub-objects of it. Nesting still applies to other parents (e.g. SG2.SG3
        // Kontakt stays nested under SG2 Marktteilnehmer).
        if transaction_group.is_some_and(|tx| tx == parent_group) {
            continue;
        }
        let child_entity = def.meta.entity.clone();
        // Skip if the child entity also has a definition at the parent group level.
        // E.g., Prozessdaten at SG4.SG6 enriches Prozessdaten at SG4 via deep_merge —
        // this is same-entity enrichment, not a parent-child nesting relationship.
        let child_has_parent_level_def = definitions
            .iter()
            .any(|d| d.meta.source_group == parent_group && d.meta.entity == child_entity);
        if child_has_parent_level_def {
            continue;
        }
        // Find the parent definition (a different entity at the parent group level)
        let parent_entity = definitions
            .iter()
            .find(|d| d.meta.source_group == parent_group && d.meta.entity != child_entity)
            .map(|d| d.meta.entity.clone());
        if let Some(ref parent_entity) = parent_entity {
            // Skip nesting if the parent definition has a dotted field target
            // that creates a sub-object with the same name as the child entity.
            // E.g., Prozessdaten has "zeitscheibe.referenz" which creates
            // prozessdaten.zeitscheibe — collides with nesting Zeitscheibe entity.
            let child_key_lc = to_camel_case(&child_entity);
            let parent_defs: Vec<_> = definitions
                .iter()
                .filter(|d| d.meta.entity == *parent_entity)
                .collect();
            let has_conflicting_field = parent_defs.iter().any(|pd| {
                pd.fields.values().any(|fm| {
                    let target = match fm {
                        crate::definition::FieldMapping::Simple(t) => t.as_str(),
                        crate::definition::FieldMapping::Structured(s) => s.target.as_str(),
                        crate::definition::FieldMapping::Nested(_) => "",
                    };
                    target.starts_with(&child_key_lc)
                        && target.get(child_key_lc.len()..child_key_lc.len() + 1) == Some(".")
                })
            });
            if has_conflicting_field {
                continue;
            }
            // Avoid duplicates
            if nesting_pairs
                .iter()
                .any(|(_, pe, ce, _)| *pe == *parent_entity && *ce == child_entity)
            {
                continue;
            }
            nesting_pairs.push((
                parent_group.to_string(),
                parent_entity.clone(),
                child_entity,
                def.meta.source_path.clone(),
            ));
        }
    }

    nesting_pairs
}

/// Check if a JSON map looks like a map-keyed entity (short uppercase/code keys → objects).
fn is_map_keyed_value(map: &serde_json::Map<String, serde_json::Value>) -> bool {
    if map.is_empty() {
        return false;
    }
    map.values().all(|v| v.is_object())
        && map.keys().all(|k| {
            k.len() <= 5
                || k.chars()
                    .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit())
        })
}

/// One code-field position recovered from a mapping definition: where the value
/// came from in EDIFACT, and where it landed in BO4E.
struct CodeSite<'a> {
    target: &'a str,
    /// `[meta] parent_field`: the site is a key of an element of that array on
    /// the entity, not a key of the entity. Without this the split enrichment
    /// looks for the target directly on the carrier and finds nothing, while
    /// the pre-split path enriched it at write time — the two would disagree on
    /// every nested rule.
    parent_field: Option<&'a str>,
    source_path: &'a str,
    seg_tag: String,
    /// The qualifier on the field key itself (`cav[Z30]...`).
    path_qualifier: Option<String>,
    /// The qualifier the definition's discriminator pins for this tag.
    disc_qualifier: Option<String>,
    element_idx: usize,
    component_idx: usize,
    enum_map: Option<&'a std::collections::BTreeMap<String, String>>,
    also_target: Option<&'a str>,
    also_enum_map: Option<&'a std::collections::BTreeMap<String, String>>,
}

pub(crate) fn to_camel_case(name: &str) -> String {
    let mut chars = name.chars();
    match chars.next() {
        Some(c) => c.to_lowercase().to_string() + chars.as_str(),
        None => String::new(),
    }
}

/// Set a value in a nested JSON map using a dotted path.
/// E.g., "address.city" sets `{"address": {"city": "value"}}`.
fn set_nested_value(map: &mut serde_json::Map<String, serde_json::Value>, path: &str, val: String) {
    set_nested_value_json(map, path, serde_json::Value::String(val));
}

/// Like `set_nested_value` but accepts a `serde_json::Value` instead of a `String`.
fn set_nested_value_json(
    map: &mut serde_json::Map<String, serde_json::Value>,
    path: &str,
    val: serde_json::Value,
) {
    if let Some((prefix, leaf)) = path.rsplit_once('.') {
        let mut current = map;
        for part in prefix.split('.') {
            let entry = current
                .entry(part.to_string())
                .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
            current = entry.as_object_mut().expect("expected object in path");
        }
        current.insert(leaf.to_string(), val);
    } else {
        map.insert(path.to_string(), val);
    }
}

/// Precompiled cache for a single format-version/variant (e.g., FV2504/UTILMD_Strom).
///
/// Contains all engines with paths pre-resolved, ready for immediate use.
/// Loading one `VariantCache` file replaces thousands of individual `.bin` reads.
#[derive(serde::Serialize, serde::Deserialize)]
pub struct VariantCache {
    /// Message-level definitions (shared across PIDs).
    pub message_defs: Vec<MappingDefinition>,
    /// Per-PID transaction definitions (key: "pid_55001").
    pub transaction_defs: BTreeMap<String, Vec<MappingDefinition>>,
    /// Per-PID combined definitions (key: "pid_55001").
    pub combined_defs: BTreeMap<String, Vec<MappingDefinition>>,
    /// Per-PID code lookups (key: "pid_55001"). Cached to avoid reading schema JSONs at load time.
    #[serde(default)]
    pub code_lookups: BTreeMap<String, crate::code_lookup::CodeLookup>,
    /// Parsed MIG schema — cached to avoid re-parsing MIG XML at startup.
    #[serde(default)]
    pub mig_schema: Option<mig_types::schema::mig::MigSchema>,
    /// Segment element counts derived from MIG — cached for reverse mapping padding.
    #[serde(default)]
    pub segment_structure: Option<crate::segment_structure::SegmentStructure>,
    /// The shared code-list tables the definitions' `code_list` names resolve
    /// against. Not part of the cache file: the tables live once beside it, so
    /// `load` finds them and every engine this cache builds inherits them.
    /// Without that a translated code reaches the output raw.
    #[serde(skip)]
    pub code_lists: std::sync::Arc<crate::code_lists::CodeLists>,
    /// Per-PID AHB segment numbers (key: "pid_55001"). Used for MIG filtering at runtime.
    /// Eliminates the need to parse AHB XML files at startup.
    #[serde(default)]
    pub pid_segment_numbers: BTreeMap<String, Vec<String>>,
    /// Per-PID field requirements (key: "pid_55001"). Built from PID schema + TOML definitions.
    /// Used by `validate_pid()` to check field completeness.
    #[serde(default)]
    pub pid_requirements: BTreeMap<String, crate::pid_requirements::PidRequirements>,
    /// Per-PID pre-built AHB workflow (key: "pid_55001"). The EDIFACT-side rulebook
    /// (segment-path keyed), twin of `pid_requirements` (BO4E-entity keyed). Built at
    /// compile-mappings from the PID schema JSON so downstream consumers can run full
    /// raw-EDIFACT validation (`Mapper::validate_edifact`) without the schema files.
    #[serde(default)]
    pub pid_ahb_workflows: BTreeMap<String, ahb_types::AhbWorkflow>,
    /// Per-PID transaction group ID (key: "pid_55001", value: "SG4").
    /// Derived from the common `source_group` prefix of transaction definitions.
    /// Empty string for message-only variants (e.g., ORDCHG).
    #[serde(default)]
    pub tx_groups: BTreeMap<String, String>,
}

impl VariantCache {
    /// Save this variant cache to a single JSON file.
    pub fn save(&self, path: &Path) -> Result<(), MappingError> {
        let encoded = serde_json::to_vec(self).map_err(|e| MappingError::CacheWrite {
            path: path.display().to_string(),
            message: e.to_string(),
        })?;
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(path, encoded)?;
        Ok(())
    }

    /// Load a variant cache from a single JSON file.
    pub fn load(path: &Path) -> Result<Self, MappingError> {
        let bytes = std::fs::read(path)?;
        let mut cache: Self =
            serde_json::from_slice(&bytes).map_err(|e| MappingError::CacheRead {
                path: path.display().to_string(),
                message: e.to_string(),
            })?;
        cache.code_lists = crate::code_lists::CodeLists::discover(path);
        Ok(cache)
    }

    /// Get the transaction group for a PID (e.g., "SG4" for UTILMD PIDs).
    /// Returns `None` if the PID is not in this variant.
    /// Returns `Some("")` for message-only variants (no transaction group).
    pub fn tx_group(&self, pid: &str) -> Option<&str> {
        self.tx_groups
            .get(&format!("pid_{pid}"))
            .map(|s| s.as_str())
    }

    /// Build a `MappingEngine` from the message-level definitions, attaching
    /// the per-PID code lookup so forward mapping enriches code fields with
    /// `{ code, meaning, enum }` objects.
    pub fn msg_engine(&self, pid: &str) -> MappingEngine {
        let mut eng = MappingEngine::from_definitions_with_code_lists(
            std::sync::Arc::clone(&self.code_lists),
            self.message_defs.clone(),
        )
        .with_pid(pid);
        if let Some(cl) = self.code_lookups.get(&format!("pid_{pid}")) {
            eng = eng.with_code_lookup(cl.clone());
        }
        eng
    }

    /// Build a `MappingEngine` from the transaction-level definitions for a PID,
    /// attaching the per-PID code lookup. Returns `None` if the PID is not in
    /// this variant.
    pub fn tx_engine(&self, pid: &str) -> Option<MappingEngine> {
        self.transaction_defs
            .get(&format!("pid_{pid}"))
            .map(|defs| {
                let mut eng = MappingEngine::from_definitions_with_code_lists(
                    std::sync::Arc::clone(&self.code_lists),
                    defs.clone(),
                )
                .with_pid(pid);
                if let Some(cl) = self.code_lookups.get(&format!("pid_{pid}")) {
                    eng = eng.with_code_lookup(cl.clone());
                }
                eng
            })
    }

    /// Get a PID-filtered MIG schema.
    /// Returns `None` if no MIG schema or no segment numbers for this PID.
    ///
    /// Falls back to the empty-PID workflow's segment numbers when the AHB
    /// has no Pruefidentifikator attribute (e.g., APERAK — one workflow for
    /// all BGM doc codes). This lets `from_edifact` work for variants whose
    /// AHB doesn't enumerate per-PID segment numbers.
    pub fn filtered_mig(&self, pid: &str) -> Option<mig_types::schema::mig::MigSchema> {
        let mig = self.mig_schema.as_ref()?;
        let numbers = self
            .pid_segment_numbers
            .get(&format!("pid_{pid}"))
            .or_else(|| self.pid_segment_numbers.get("pid_"))?;
        let number_set: std::collections::HashSet<String> = numbers.iter().cloned().collect();
        Some(mig_assembly::pid_filter::filter_mig_for_pid(
            mig,
            &number_set,
        ))
    }
}

/// Bundled data for a single format version (e.g., FV2504).
///
/// Contains all VariantCaches for every message type in that FV,
/// serialized as one bincode file for distribution via GitHub releases.
#[derive(serde::Serialize, serde::Deserialize)]
pub struct DataBundle {
    pub format_version: String,
    pub bundle_version: u32,
    pub variants: BTreeMap<String, VariantCache>,
    /// PID-agnostic BO4E type catalog (parsed from `bo4e-german` source).
    ///
    /// Populated by the bundle generator at compile-mappings time. Older bundles
    /// without this field deserialize to an empty catalog.
    #[serde(default)]
    pub bo4e_catalog: crate::bo4e_catalog::Bo4eCatalog,

    /// The crate version that produced this bundle.
    ///
    /// Distinct from [`bundle_version`](Self::bundle_version), which guards the
    /// serialisation *format* and has been unchanged for many releases — a
    /// bundle can satisfy it while its mappings, schemas and code lists come
    /// from another era. That is not a hypothetical: a bundle five months old
    /// passed the format check, loaded cleanly, and rendered 12% of a message
    /// with no error (issue #158).
    ///
    /// `None` for bundles produced before this field existed, which is itself
    /// evidence of age.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub built_by: Option<String>,
    /// The shared code-list tables the definitions' `code_list` names resolve
    /// against.
    ///
    /// Carried IN the bundle, unlike `VariantCache`, which is written beside a
    /// copy of `code_lists.toml` and repairs itself from it on load. A bundle
    /// is a bare `.bin` fetched into `~/.edifact/data` with nothing beside it,
    /// so a bundle that does not carry its tables cannot resolve a single
    /// name: forward, the EDIFACT code reaches the output untranslated;
    /// reverse, the BO4E name is written into the EDIFACT slot verbatim. The
    /// deduplication that made naming worth doing does not argue against this
    /// -- there is one bundle per format version, so the tables appear once.
    #[serde(default)]
    pub code_lists: crate::code_lists::CodeLists,
}

impl DataBundle {
    pub const CURRENT_VERSION: u32 = 2;

    /// The release a bundle built now belongs to, and the one a bundle must
    /// have been built by to be read.
    ///
    /// Taken from `mig-bo4e` rather than from whichever crate produces or
    /// consumes a bundle: `mig-bo4e` carries the workspace version, which is
    /// what the release process stamps, while `automapper-generator` versions
    /// itself separately. Reading it from the producer gave `0.1.0` against a
    /// consumer expecting `0.1.1` — a mismatch that is an artefact of where the
    /// constant was read, not of the data.
    pub const PRODUCING_VERSION: &'static str = env!("CARGO_PKG_VERSION");

    pub fn variant(&self, name: &str) -> Option<&VariantCache> {
        self.variants.get(name)
    }

    pub fn write_to<W: std::io::Write>(&self, writer: &mut W) -> Result<(), MappingError> {
        let encoded = serde_json::to_vec(self).map_err(|e| MappingError::CacheWrite {
            path: "<stream>".to_string(),
            message: e.to_string(),
        })?;
        writer.write_all(&encoded).map_err(MappingError::Io)
    }

    pub fn read_from<R: std::io::Read>(reader: &mut R) -> Result<Self, MappingError> {
        let mut bytes = Vec::new();
        reader.read_to_end(&mut bytes).map_err(MappingError::Io)?;
        serde_json::from_slice(&bytes).map_err(|e| MappingError::CacheRead {
            path: "<stream>".to_string(),
            message: e.to_string(),
        })
    }

    pub fn read_from_checked<R: std::io::Read>(reader: &mut R) -> Result<Self, MappingError> {
        let mut bundle = Self::read_from(reader)?;
        // Every engine this bundle builds resolves names through its variant's
        // `Arc`, which `#[serde(skip)]` left empty on the way in.
        let shared = std::sync::Arc::new(std::mem::take(&mut bundle.code_lists));
        for variant in bundle.variants.values_mut() {
            variant.code_lists = std::sync::Arc::clone(&shared);
        }
        bundle.code_lists = (*shared).clone();
        if bundle.bundle_version != Self::CURRENT_VERSION {
            return Err(MappingError::CacheRead {
                path: "<stream>".to_string(),
                message: format!(
                    "Incompatible bundle version {}, expected version {}. \
                     Run `edifact-data update` to fetch compatible bundles.",
                    bundle.bundle_version,
                    Self::CURRENT_VERSION
                ),
            });
        }
        Ok(bundle)
    }

    pub fn save(&self, path: &Path) -> Result<(), MappingError> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let mut file = std::fs::File::create(path).map_err(MappingError::Io)?;
        self.write_to(&mut file)
    }

    pub fn load(path: &Path) -> Result<Self, MappingError> {
        let mut file = std::fs::File::open(path).map_err(MappingError::Io)?;
        Self::read_from_checked(&mut file)
    }
}

#[cfg(test)]
mod variant_cache_helper_tests {
    use super::*;

    fn make_test_cache() -> VariantCache {
        let mut tx_groups = BTreeMap::new();
        tx_groups.insert("pid_55001".to_string(), "SG4".to_string());
        tx_groups.insert("pid_21007".to_string(), "SG14".to_string());

        let mut transaction_defs = BTreeMap::new();
        transaction_defs.insert("pid_55001".to_string(), vec![]);
        transaction_defs.insert("pid_21007".to_string(), vec![]);

        VariantCache {
            code_lists: Default::default(),
            message_defs: vec![],
            transaction_defs,
            combined_defs: BTreeMap::new(),
            code_lookups: BTreeMap::new(),
            mig_schema: None,
            segment_structure: None,
            pid_segment_numbers: BTreeMap::new(),
            pid_requirements: BTreeMap::new(),
            pid_ahb_workflows: BTreeMap::new(),
            tx_groups,
        }
    }

    #[test]
    fn test_tx_group_returns_correct_group() {
        let vc = make_test_cache();
        assert_eq!(vc.tx_group("55001").unwrap(), "SG4");
        assert_eq!(vc.tx_group("21007").unwrap(), "SG14");
    }

    #[test]
    fn test_tx_group_unknown_pid_returns_none() {
        let vc = make_test_cache();
        assert!(vc.tx_group("99999").is_none());
    }

    #[test]
    fn test_msg_engine_returns_engine() {
        let vc = make_test_cache();
        let engine = vc.msg_engine("55001");
        assert_eq!(engine.definitions().len(), 0);
    }

    #[test]
    fn test_tx_engine_returns_engine_for_known_pid() {
        let vc = make_test_cache();
        assert!(vc.tx_engine("55001").is_some());
    }

    #[test]
    fn test_tx_engine_returns_none_for_unknown_pid() {
        let vc = make_test_cache();
        assert!(vc.tx_engine("99999").is_none());
    }

    /// Build a cache whose every map holds many keys. Each call creates fresh
    /// `HashMap`s (fresh random hash seeds), so an order-dependent serializer
    /// produces different bytes on different calls.
    fn make_populated_cache() -> VariantCache {
        let pids: Vec<String> = (0..40).map(|i| format!("pid_{}", 55000 + i * 7)).collect();
        let schema: serde_json::Value = serde_json::from_str(include_str!(
            "../../mig-types/src/generated/fv2504/utilmd/pids/pid_55001_schema.json"
        ))
        .unwrap();
        let code_lookup = crate::code_lookup::CodeLookup::from_schema_value(&schema);
        let element_counts: serde_json::Map<String, serde_json::Value> = (0..40)
            .map(|i| (format!("T{i:02}"), serde_json::json!(i)))
            .collect();
        let segment_structure: SegmentStructure =
            serde_json::from_value(serde_json::json!({ "element_counts": element_counts }))
                .unwrap();
        let ubs: serde_json::Map<String, serde_json::Value> = (0..40)
            .map(|i| (format!("UB{i}"), serde_json::json!({ "Ref": i })))
            .collect();
        let workflow: ahb_types::AhbWorkflow = serde_json::from_value(serde_json::json!({
            "pruefidentifikator": "55001",
            "description": "",
            "communication_direction": null,
            "fields": [],
            "ub_definitions": ubs,
        }))
        .unwrap();

        let mut vc = make_test_cache();
        vc.segment_structure = Some(segment_structure);
        for pid in &pids {
            vc.transaction_defs.insert(pid.clone(), vec![]);
            vc.combined_defs.insert(pid.clone(), vec![]);
            vc.code_lookups.insert(pid.clone(), code_lookup.clone());
            vc.pid_segment_numbers
                .insert(pid.clone(), vec!["00001".to_string()]);
            vc.pid_ahb_workflows.insert(pid.clone(), workflow.clone());
            vc.tx_groups.insert(pid.clone(), "SG4".to_string());
        }
        vc
    }

    /// Enrichment of qualified field paths uses the codes of the segment variant
    /// the field reads, never those of a sibling variant of the same tag.
    #[test]
    fn test_enrichment_uses_codes_of_the_path_qualifier_variant() {
        let comp = |sub: u64, id: &str, codes: Option<serde_json::Value>| match codes {
            Some(c) => serde_json::json!({"sub_index": sub, "id": id, "type": "code", "codes": c}),
            None => serde_json::json!({"sub_index": sub, "id": id, "type": "data"}),
        };
        let code = |v: &str, n: &str| serde_json::json!([{"value": v, "name": n}]);
        let seg = |tag: &str, composite: &str, comps: Vec<serde_json::Value>| serde_json::json!({"id": tag, "elements": [{"index": 0, "composite": composite, "components": comps}]});
        let schema = serde_json::json!({"fields": {"sg15": {"segments": [
            seg("RFF", "C506", vec![comp(0, "1153", Some(code("Z13", "PID"))), comp(1, "1154", Some(code("21037", "RD / NB-Bewertung")))]),
            seg("RFF", "C506", vec![comp(0, "1153", Some(code("ACW", "Referenz"))), comp(1, "1154", None)]),
            seg("CAV", "C889", vec![comp(0, "7111", Some(code("Z91", "Z91"))), comp(1, "7110", Some(code("A", "Alpha")))]),
            seg("CAV", "C889", vec![comp(0, "7111", Some(code("ZF0", "ZF0"))), comp(1, "7110", Some(code("C", "Gamma")))]),
        ]}}});
        let engine = MappingEngine::new_empty()
            .with_code_lookup(crate::code_lookup::CodeLookup::from_schema_value(&schema));
        let def = MappingDefinition::from_toml_str(
            r#"
[meta]
entity = "Status"
bo4e_type = "Status"
source_group = "SG15"
source_path = "sg15"
discriminator = "RFF.0.0=Z13"

[fields]
"rff.0.1" = "pruefidentifikator"
"rff[ACW].0.1" = "referenz"
"cav[Z91].0.1" = "z91Wert"
"cav[ZF0].0.1" = "zf0Wert"
"#,
        )
        .unwrap();
        let segment = |tag: &str, elements: &[&[&str]]| OwnedSegment {
            id: tag.to_string(),
            elements: elements
                .iter()
                .map(|e| e.iter().map(|c| c.to_string()).collect())
                .collect(),
            segment_number: 1,
        };
        let json = engine.map_forward_from_segments(
            &[
                segment("RFF", &[&["Z13", "21037"]]),
                segment("RFF", &[&["ACW", "REF-1"]]),
                segment("CAV", &[&["Z91", "C"]]),
                segment("CAV", &[&["ZF0", "C"]]),
            ],
            &def,
        );
        assert_eq!(
            json["referenz"],
            serde_json::json!("REF-1"),
            "RFF+ACW d1154 is data; RFF+Z13's codes must not apply: {json}"
        );
        assert_eq!(json["pruefidentifikator"]["meaning"], "RD / NB-Bewertung");
        assert_eq!(
            json["z91Wert"]["meaning"],
            serde_json::Value::Null,
            "'C' is a CAV+ZF0 code, unknown to CAV+Z91: {json}"
        );
        assert_eq!(json["zf0Wert"]["meaning"], "Gamma");
    }

    #[test]
    fn test_variant_cache_serialization_is_deterministic() {
        let reference = serde_json::to_vec(&make_populated_cache()).unwrap();
        for _ in 0..5 {
            let again = serde_json::to_vec(&make_populated_cache()).unwrap();
            assert!(
                reference == again,
                "VariantCache serialization must not depend on HashMap iteration order"
            );
        }
    }

    #[test]
    fn test_variant_cache_serializes_map_keys_sorted() {
        use indexmap::IndexMap;
        use serde::de::IgnoredAny;

        #[derive(serde::Deserialize)]
        struct ProbeWorkflow {
            ub_definitions: IndexMap<String, IgnoredAny>,
        }
        #[derive(serde::Deserialize)]
        struct ProbeStructure {
            element_counts: IndexMap<String, usize>,
        }
        #[derive(serde::Deserialize)]
        struct Probe {
            transaction_defs: IndexMap<String, IgnoredAny>,
            combined_defs: IndexMap<String, IgnoredAny>,
            code_lookups: IndexMap<String, IndexMap<String, IgnoredAny>>,
            segment_structure: ProbeStructure,
            pid_segment_numbers: IndexMap<String, IgnoredAny>,
            pid_requirements: IndexMap<String, IgnoredAny>,
            pid_ahb_workflows: IndexMap<String, ProbeWorkflow>,
            tx_groups: IndexMap<String, String>,
        }
        fn assert_sorted<'a>(what: &str, keys: impl Iterator<Item = &'a String>) {
            let keys: Vec<&String> = keys.collect();
            let mut sorted = keys.clone();
            sorted.sort();
            assert_eq!(keys, sorted, "{what} keys must serialize in sorted order");
        }

        let json = serde_json::to_string(&make_populated_cache()).unwrap();
        let probe: Probe = serde_json::from_str(&json).unwrap();
        assert_sorted("transaction_defs", probe.transaction_defs.keys());
        assert_sorted("combined_defs", probe.combined_defs.keys());
        assert_sorted("code_lookups", probe.code_lookups.keys());
        let lookup = probe.code_lookups.values().next().unwrap();
        assert!(lookup.len() > 10, "fixture lookup should have many entries");
        assert_sorted("code_lookup entries", lookup.keys());
        assert_sorted(
            "segment_structure",
            probe.segment_structure.element_counts.keys(),
        );
        assert_sorted("pid_segment_numbers", probe.pid_segment_numbers.keys());
        assert_sorted("pid_requirements", probe.pid_requirements.keys());
        assert_sorted("pid_ahb_workflows", probe.pid_ahb_workflows.keys());
        let wf = probe.pid_ahb_workflows.values().next().unwrap();
        assert_sorted("ub_definitions", wf.ub_definitions.keys());
        assert_sorted("tx_groups", probe.tx_groups.keys());
    }

    #[test]
    fn test_data_bundle_serializes_variants_sorted() {
        use indexmap::IndexMap;
        use serde::de::IgnoredAny;

        #[derive(serde::Deserialize)]
        struct Probe {
            variants: IndexMap<String, IgnoredAny>,
        }
        let variants: BTreeMap<String, VariantCache> = (0..20)
            .map(|i| (format!("VARIANT_{i:02}"), make_test_cache()))
            .collect();
        let bundle = DataBundle {
            format_version: "FV2504".to_string(),
            bundle_version: DataBundle::CURRENT_VERSION,
            built_by: Some(DataBundle::PRODUCING_VERSION.to_string()),
            variants,
            bo4e_catalog: Default::default(),
            code_lists: Default::default(),
        };
        let mut bytes = Vec::new();
        bundle.write_to(&mut bytes).unwrap();
        let probe: Probe = serde_json::from_slice(&bytes).unwrap();
        let keys: Vec<&String> = probe.variants.keys().collect();
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(keys, sorted);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::definition::{MappingDefinition, MappingMeta, StructuredFieldMapping};
    use indexmap::IndexMap;

    fn make_def(fields: IndexMap<String, FieldMapping>) -> MappingDefinition {
        MappingDefinition {
            meta: MappingMeta {
                entity: "Test".to_string(),
                bo4e_type: "Test".to_string(),
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields,
            complex_handlers: None,
        }
    }

    #[test]
    fn test_map_interchange_single_transaction_backward_compat() {
        use mig_assembly::assembler::*;

        // Single SG4 with SG5 — the common case for current PID 55001 fixtures
        let tree = AssembledTree {
            segments: vec![
                AssembledSegment {
                    tag: "UNH".to_string(),
                    elements: vec![vec!["001".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "BGM".to_string(),
                    elements: vec![vec!["E01".to_string()], vec!["DOC001".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
            ],
            groups: vec![
                AssembledGroup {
                    group_id: "SG2".to_string(),
                    repetitions: vec![AssembledGroupInstance {
                        segments: vec![AssembledSegment {
                            tag: "NAD".to_string(),
                            elements: vec![vec!["MS".to_string()], vec!["9900123".to_string()]],
                            mig_number: None,
                            segment_number: None,
                        }],
                        child_groups: vec![],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: vec![],
                        skipped_positions: Vec::new(),
                    }],
                },
                AssembledGroup {
                    group_id: "SG4".to_string(),
                    repetitions: vec![AssembledGroupInstance {
                        segments: vec![AssembledSegment {
                            tag: "IDE".to_string(),
                            elements: vec![vec!["24".to_string()], vec!["TX001".to_string()]],
                            mig_number: None,
                            segment_number: None,
                        }],
                        child_groups: vec![AssembledGroup {
                            group_id: "SG5".to_string(),
                            repetitions: vec![AssembledGroupInstance {
                                segments: vec![AssembledSegment {
                                    tag: "LOC".to_string(),
                                    elements: vec![
                                        vec!["Z16".to_string()],
                                        vec!["DE000111222333".to_string()],
                                    ],
                                    mig_number: None,
                                    segment_number: None,
                                }],
                                child_groups: vec![],
                                entry_mig_number: None,
                                variant_mig_numbers: vec![],
                                skipped_segments: vec![],
                                skipped_positions: Vec::new(),
                            }],
                        }],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: vec![],
                        skipped_positions: Vec::new(),
                    }],
                },
            ],
            post_group_start: 2,
            inter_group_segments: std::collections::BTreeMap::new(),
        };

        // Empty message engine (no message-level defs for this test)
        let msg_engine = MappingEngine::from_definitions(vec![]);

        // Transaction defs
        let mut tx_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        tx_fields.insert(
            "ide.1".to_string(),
            FieldMapping::Simple("vorgangId".to_string()),
        );
        let mut malo_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        malo_fields.insert(
            "loc.1".to_string(),
            FieldMapping::Simple("marktlokationsId".to_string()),
        );

        let tx_engine = MappingEngine::from_definitions(vec![
            MappingDefinition {
                meta: MappingMeta {
                    entity: "Prozessdaten".to_string(),
                    bo4e_type: "Prozessdaten".to_string(),
                    source_group: "SG4".to_string(),
                    source_path: None,
                    discriminator: None,
                    repeat_on_tag: None,
                    parent_field: None,
                    target_list: None,
                    order: None,
                },
                fields: tx_fields,
                complex_handlers: None,
            },
            MappingDefinition {
                meta: MappingMeta {
                    entity: "Marktlokation".to_string(),
                    bo4e_type: "Marktlokation".to_string(),
                    source_group: "SG4.SG5".to_string(),
                    source_path: None,
                    discriminator: None,
                    repeat_on_tag: None,
                    parent_field: None,
                    target_list: None,
                    order: None,
                },
                fields: malo_fields,
                complex_handlers: None,
            },
        ]);

        let result = MappingEngine::map_interchange(&msg_engine, &tx_engine, &tree, "SG4", true);

        assert_eq!(result.transaktionen.len(), 1);
        assert_eq!(
            result.transaktionen[0].transaktionsdaten["vorgangId"]
                .as_str()
                .unwrap(),
            "TX001"
        );
        // Marktlokation (SG4.SG5) stays top-level — SG4 IS the transaction root,
        // so Marktlokation is a peer of Prozessdaten, not a child of it.
        assert_eq!(
            result.transaktionen[0].stammdaten["marktlokation"]["marktlokationsId"]
                .as_str()
                .unwrap(),
            "DE000111222333"
        );
    }

    #[test]
    fn test_map_reverse_pads_intermediate_empty_elements() {
        // NAD+Z09+++Muster:Max — positions 0 and 3 populated, 1 and 2 should become [""]
        let mut fields = IndexMap::new();
        fields.insert(
            "nad.0".to_string(),
            FieldMapping::Structured(StructuredFieldMapping {
                target: String::new(),
                transform: None,
                when: None,
                default: Some("Z09".to_string()),
                enum_map: None,
                code_list: None,
                also_code_list: None,
                when_filled: None,
                also_target: None,
                also_enum_map: None,
            }),
        );
        fields.insert(
            "nad.3.0".to_string(),
            FieldMapping::Simple("name".to_string()),
        );
        fields.insert(
            "nad.3.1".to_string(),
            FieldMapping::Simple("vorname".to_string()),
        );

        let def = make_def(fields);
        let engine = MappingEngine::from_definitions(vec![]);

        let bo4e = serde_json::json!({
            "name": "Muster",
            "vorname": "Max"
        });

        let instance = engine.map_reverse(&bo4e, &def);
        assert_eq!(instance.segments.len(), 1);

        let nad = &instance.segments[0];
        assert_eq!(nad.tag, "NAD");
        assert_eq!(nad.elements.len(), 4);
        assert_eq!(nad.elements[0], vec!["Z09"]);
        // Intermediate positions 1 and 2 should be padded to [""]
        assert_eq!(nad.elements[1], vec![""]);
        assert_eq!(nad.elements[2], vec![""]);
        assert_eq!(nad.elements[3][0], "Muster");
        assert_eq!(nad.elements[3][1], "Max");
    }

    #[test]
    fn test_map_reverse_no_padding_when_contiguous() {
        // DTM+92:20250531:303 — all three components in element 0, no gaps
        let mut fields = IndexMap::new();
        fields.insert(
            "dtm.0.0".to_string(),
            FieldMapping::Structured(StructuredFieldMapping {
                target: String::new(),
                transform: None,
                when: None,
                default: Some("92".to_string()),
                enum_map: None,
                code_list: None,
                also_code_list: None,
                when_filled: None,
                also_target: None,
                also_enum_map: None,
            }),
        );
        fields.insert(
            "dtm.0.1".to_string(),
            FieldMapping::Simple("value".to_string()),
        );
        fields.insert(
            "dtm.0.2".to_string(),
            FieldMapping::Structured(StructuredFieldMapping {
                target: String::new(),
                transform: None,
                when: None,
                default: Some("303".to_string()),
                enum_map: None,
                code_list: None,
                also_code_list: None,
                when_filled: None,
                also_target: None,
                also_enum_map: None,
            }),
        );

        let def = make_def(fields);
        let engine = MappingEngine::from_definitions(vec![]);

        let bo4e = serde_json::json!({ "value": "20250531" });

        let instance = engine.map_reverse(&bo4e, &def);
        let dtm = &instance.segments[0];
        // Single element with 3 components — no intermediate padding needed
        assert_eq!(dtm.elements.len(), 1);
        assert_eq!(dtm.elements[0], vec!["92", "20250531", "303"]);
    }

    #[test]
    fn test_map_message_level_extracts_sg2_only() {
        use mig_assembly::assembler::*;

        // Build a tree with SG2 (message-level) and SG4 (transaction-level)
        let tree = AssembledTree {
            segments: vec![
                AssembledSegment {
                    tag: "UNH".to_string(),
                    elements: vec![vec!["001".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "BGM".to_string(),
                    elements: vec![vec!["E01".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
            ],
            groups: vec![
                AssembledGroup {
                    group_id: "SG2".to_string(),
                    repetitions: vec![AssembledGroupInstance {
                        segments: vec![AssembledSegment {
                            tag: "NAD".to_string(),
                            elements: vec![vec!["MS".to_string()], vec!["9900123".to_string()]],
                            mig_number: None,
                            segment_number: None,
                        }],
                        child_groups: vec![],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: vec![],
                        skipped_positions: Vec::new(),
                    }],
                },
                AssembledGroup {
                    group_id: "SG4".to_string(),
                    repetitions: vec![AssembledGroupInstance {
                        segments: vec![AssembledSegment {
                            tag: "IDE".to_string(),
                            elements: vec![vec!["24".to_string()], vec!["TX001".to_string()]],
                            mig_number: None,
                            segment_number: None,
                        }],
                        child_groups: vec![],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: vec![],
                        skipped_positions: Vec::new(),
                    }],
                },
            ],
            post_group_start: 2,
            inter_group_segments: std::collections::BTreeMap::new(),
        };

        // Message-level definition maps SG2
        let mut msg_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        msg_fields.insert(
            "nad.0".to_string(),
            FieldMapping::Simple("marktrolle".to_string()),
        );
        msg_fields.insert(
            "nad.1".to_string(),
            FieldMapping::Simple("rollencodenummer".to_string()),
        );
        let msg_def = MappingDefinition {
            meta: MappingMeta {
                entity: "Marktteilnehmer".to_string(),
                bo4e_type: "Marktteilnehmer".to_string(),
                source_group: "SG2".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields: msg_fields,
            complex_handlers: None,
        };

        let engine = MappingEngine::from_definitions(vec![msg_def.clone()]);
        let result = engine.map_all_forward(&tree);

        // Should contain Marktteilnehmer from SG2
        assert!(result.get("marktteilnehmer").is_some());
        let mt = &result["marktteilnehmer"];
        assert_eq!(mt["marktrolle"].as_str().unwrap(), "MS");
        assert_eq!(mt["rollencodenummer"].as_str().unwrap(), "9900123");
    }

    #[test]
    fn test_map_transaction_scoped_to_sg4_instance() {
        use mig_assembly::assembler::*;

        // Build a tree with SG4 containing SG5 (LOC+Z16)
        let tree = AssembledTree {
            segments: vec![
                AssembledSegment {
                    tag: "UNH".to_string(),
                    elements: vec![vec!["001".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "BGM".to_string(),
                    elements: vec![vec!["E01".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
            ],
            groups: vec![AssembledGroup {
                group_id: "SG4".to_string(),
                repetitions: vec![AssembledGroupInstance {
                    segments: vec![AssembledSegment {
                        tag: "IDE".to_string(),
                        elements: vec![vec!["24".to_string()], vec!["TX001".to_string()]],
                        mig_number: None,
                        segment_number: None,
                    }],
                    child_groups: vec![AssembledGroup {
                        group_id: "SG5".to_string(),
                        repetitions: vec![AssembledGroupInstance {
                            segments: vec![AssembledSegment {
                                tag: "LOC".to_string(),
                                elements: vec![
                                    vec!["Z16".to_string()],
                                    vec!["DE000111222333".to_string()],
                                ],
                                mig_number: None,
                                segment_number: None,
                            }],
                            child_groups: vec![],
                            entry_mig_number: None,
                            variant_mig_numbers: vec![],
                            skipped_segments: vec![],
                            skipped_positions: Vec::new(),
                        }],
                    }],
                    entry_mig_number: None,
                    variant_mig_numbers: vec![],
                    skipped_segments: vec![],
                    skipped_positions: Vec::new(),
                }],
            }],
            post_group_start: 2,
            inter_group_segments: std::collections::BTreeMap::new(),
        };

        // Transaction-level definitions: prozessdaten (root of SG4) + marktlokation (SG5)
        let mut proz_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        proz_fields.insert(
            "ide.1".to_string(),
            FieldMapping::Simple("vorgangId".to_string()),
        );
        let proz_def = MappingDefinition {
            meta: MappingMeta {
                entity: "Prozessdaten".to_string(),
                bo4e_type: "Prozessdaten".to_string(),
                source_group: "".to_string(), // Root-level within transaction sub-tree
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields: proz_fields,
            complex_handlers: None,
        };

        let mut malo_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        malo_fields.insert(
            "loc.1".to_string(),
            FieldMapping::Simple("marktlokationsId".to_string()),
        );
        let malo_def = MappingDefinition {
            meta: MappingMeta {
                entity: "Marktlokation".to_string(),
                bo4e_type: "Marktlokation".to_string(),
                source_group: "SG5".to_string(), // Relative to SG4, not "SG4.SG5"
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields: malo_fields,
            complex_handlers: None,
        };

        let tx_engine = MappingEngine::from_definitions(vec![proz_def, malo_def]);

        // Scope to the SG4 instance and map
        let sg4 = &tree.groups[0]; // SG4 group
        let sg4_instance = &sg4.repetitions[0];
        let sub_tree = sg4_instance.as_assembled_tree();

        let result = tx_engine.map_all_forward(&sub_tree);

        // Should contain Prozessdaten from SG4 root segments
        assert_eq!(
            result["prozessdaten"]["vorgangId"].as_str().unwrap(),
            "TX001"
        );

        // Should contain Marktlokation from SG5 within SG4
        assert_eq!(
            result["marktlokation"]["marktlokationsId"]
                .as_str()
                .unwrap(),
            "DE000111222333"
        );
    }

    #[test]
    fn test_map_interchange_produces_full_hierarchy() {
        use mig_assembly::assembler::*;

        // Build a tree with SG2 (message-level) and SG4 with two repetitions (two transactions)
        let tree = AssembledTree {
            segments: vec![
                AssembledSegment {
                    tag: "UNH".to_string(),
                    elements: vec![vec!["001".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "BGM".to_string(),
                    elements: vec![vec!["E01".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
            ],
            groups: vec![
                AssembledGroup {
                    group_id: "SG2".to_string(),
                    repetitions: vec![AssembledGroupInstance {
                        segments: vec![AssembledSegment {
                            tag: "NAD".to_string(),
                            elements: vec![vec!["MS".to_string()], vec!["9900123".to_string()]],
                            mig_number: None,
                            segment_number: None,
                        }],
                        child_groups: vec![],
                        entry_mig_number: None,
                        variant_mig_numbers: vec![],
                        skipped_segments: vec![],
                        skipped_positions: Vec::new(),
                    }],
                },
                AssembledGroup {
                    group_id: "SG4".to_string(),
                    repetitions: vec![
                        AssembledGroupInstance {
                            segments: vec![AssembledSegment {
                                tag: "IDE".to_string(),
                                elements: vec![vec!["24".to_string()], vec!["TX001".to_string()]],
                                mig_number: None,
                                segment_number: None,
                            }],
                            child_groups: vec![],
                            entry_mig_number: None,
                            variant_mig_numbers: vec![],
                            skipped_segments: vec![],
                            skipped_positions: Vec::new(),
                        },
                        AssembledGroupInstance {
                            segments: vec![AssembledSegment {
                                tag: "IDE".to_string(),
                                elements: vec![vec!["24".to_string()], vec!["TX002".to_string()]],
                                mig_number: None,
                                segment_number: None,
                            }],
                            child_groups: vec![],
                            entry_mig_number: None,
                            variant_mig_numbers: vec![],
                            skipped_segments: vec![],
                            skipped_positions: Vec::new(),
                        },
                    ],
                },
            ],
            post_group_start: 2,
            inter_group_segments: std::collections::BTreeMap::new(),
        };

        // Message-level definitions
        let mut msg_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        msg_fields.insert(
            "nad.0".to_string(),
            FieldMapping::Simple("marktrolle".to_string()),
        );
        let msg_defs = vec![MappingDefinition {
            meta: MappingMeta {
                entity: "Marktteilnehmer".to_string(),
                bo4e_type: "Marktteilnehmer".to_string(),
                source_group: "SG2".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields: msg_fields,
            complex_handlers: None,
        }];

        // Transaction-level definitions (source_group includes SG4 prefix)
        let mut tx_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        tx_fields.insert(
            "ide.1".to_string(),
            FieldMapping::Simple("vorgangId".to_string()),
        );
        let tx_defs = vec![MappingDefinition {
            meta: MappingMeta {
                entity: "Prozessdaten".to_string(),
                bo4e_type: "Prozessdaten".to_string(),
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
                parent_field: None,
                target_list: None,
                order: None,
            },
            fields: tx_fields,
            complex_handlers: None,
        }];

        let msg_engine = MappingEngine::from_definitions(msg_defs);
        let tx_engine = MappingEngine::from_definitions(tx_defs);

        let result = MappingEngine::map_interchange(&msg_engine, &tx_engine, &tree, "SG4", true);

        // Message-level stammdaten
        assert!(result.stammdaten["marktteilnehmer"].is_object());
        assert_eq!(
            result.stammdaten["marktteilnehmer"]["marktrolle"]
                .as_str()
                .unwrap(),
            "MS"
        );

        // Two transactions
        assert_eq!(result.transaktionen.len(), 2);
        assert_eq!(
            result.transaktionen[0].transaktionsdaten["vorgangId"]
                .as_str()
                .unwrap(),
            "TX001"
        );
        assert_eq!(
            result.transaktionen[1].transaktionsdaten["vorgangId"]
                .as_str()
                .unwrap(),
            "TX002"
        );
    }

    #[test]
    fn test_map_reverse_with_segment_structure_pads_trailing() {
        // STS+7++E01 — position 0 and 2 populated, MIG says 5 elements
        let mut fields = IndexMap::new();
        fields.insert(
            "sts.0".to_string(),
            FieldMapping::Structured(StructuredFieldMapping {
                target: String::new(),
                transform: None,
                when: None,
                default: Some("7".to_string()),
                enum_map: None,
                code_list: None,
                also_code_list: None,
                when_filled: None,
                also_target: None,
                also_enum_map: None,
            }),
        );
        fields.insert(
            "sts.2".to_string(),
            FieldMapping::Simple("grund".to_string()),
        );

        let def = make_def(fields);

        // Build a SegmentStructure manually via BTreeMap
        let mut counts = std::collections::BTreeMap::new();
        counts.insert("STS".to_string(), 5usize);
        let ss = SegmentStructure {
            element_counts: counts,
        };

        let engine = MappingEngine::from_definitions(vec![]).with_segment_structure(ss);

        let bo4e = serde_json::json!({ "grund": "E01" });

        let instance = engine.map_reverse(&bo4e, &def);
        let sts = &instance.segments[0];
        // Should have 5 elements: pos 0 = ["7"], pos 1 = [""] (intermediate pad),
        // pos 2 = ["E01"], pos 3 = [""] (trailing pad), pos 4 = [""] (trailing pad)
        assert_eq!(sts.elements.len(), 5);
        assert_eq!(sts.elements[0], vec!["7"]);
        assert_eq!(sts.elements[1], vec![""]);
        assert_eq!(sts.elements[2], vec!["E01"]);
        assert_eq!(sts.elements[3], vec![""]);
        assert_eq!(sts.elements[4], vec![""]);
    }

    #[test]
    fn test_resolve_child_relative_with_source_path() {
        let mut map: std::collections::HashMap<String, Vec<usize>> =
            std::collections::HashMap::new();
        map.insert("sg4.sg8_ze1".to_string(), vec![6]);
        map.insert("sg4.sg8_z98".to_string(), vec![0]);

        // Child without explicit index → resolved from source_path
        assert_eq!(
            resolve_child_relative("SG8.SG10", Some("sg4.sg8_ze1.sg10"), &map, 0),
            "SG8:6.SG10"
        );

        // Child with explicit index → kept as-is
        assert_eq!(
            resolve_child_relative("SG8:3.SG10", Some("sg4.sg8_ze1.sg10"), &map, 0),
            "SG8:3.SG10"
        );

        // Source path not in map → kept as-is
        assert_eq!(
            resolve_child_relative("SG8.SG10", Some("sg4.sg8_unknown.sg10"), &map, 0),
            "SG8.SG10"
        );

        // No source_path → kept as-is
        assert_eq!(
            resolve_child_relative("SG8.SG10", None, &map, 0),
            "SG8.SG10"
        );

        // SG9 also works
        assert_eq!(
            resolve_child_relative("SG8.SG9", Some("sg4.sg8_z98.sg9"), &map, 0),
            "SG8:0.SG9"
        );

        // Multi-rep parent: item_idx selects the correct parent rep
        map.insert("sg4.sg8_zf3".to_string(), vec![3, 4]);
        assert_eq!(
            resolve_child_relative("SG8.SG10", Some("sg4.sg8_zf3.sg10"), &map, 0),
            "SG8:3.SG10"
        );
        assert_eq!(
            resolve_child_relative("SG8.SG10", Some("sg4.sg8_zf3.sg10"), &map, 1),
            "SG8:4.SG10"
        );
    }

    #[test]
    fn test_place_in_groups_returns_rep_index() {
        let mut groups: Vec<AssembledGroup> = Vec::new();

        // Append (no index) → returns position 0
        let instance = AssembledGroupInstance {
            segments: vec![],
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: Vec::new(),
        };
        assert_eq!(place_in_groups(&mut groups, "SG8", instance), 0);

        // Append again → returns position 1
        let instance = AssembledGroupInstance {
            segments: vec![],
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: Vec::new(),
        };
        assert_eq!(place_in_groups(&mut groups, "SG8", instance), 1);

        // Explicit index → returns that index
        let instance = AssembledGroupInstance {
            segments: vec![],
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: Vec::new(),
        };
        assert_eq!(place_in_groups(&mut groups, "SG8:5", instance), 5);
    }

    #[test]
    fn test_resolve_by_source_path() {
        use mig_assembly::assembler::*;

        // Build a tree: SG4[0] → SG8 with two reps (Z98 and ZD7) → each has SG10
        let tree = AssembledTree {
            segments: vec![],
            groups: vec![AssembledGroup {
                group_id: "SG4".to_string(),
                repetitions: vec![AssembledGroupInstance {
                    segments: vec![],
                    child_groups: vec![AssembledGroup {
                        group_id: "SG8".to_string(),
                        repetitions: vec![
                            AssembledGroupInstance {
                                segments: vec![AssembledSegment {
                                    tag: "SEQ".to_string(),
                                    elements: vec![vec!["Z98".to_string()]],
                                    mig_number: None,
                                    segment_number: None,
                                }],
                                child_groups: vec![AssembledGroup {
                                    group_id: "SG10".to_string(),
                                    repetitions: vec![AssembledGroupInstance {
                                        segments: vec![AssembledSegment {
                                            tag: "CCI".to_string(),
                                            elements: vec![vec![], vec![], vec!["ZB3".to_string()]],
                                            mig_number: None,
                                            segment_number: None,
                                        }],
                                        child_groups: vec![],
                                        entry_mig_number: None,
                                        variant_mig_numbers: vec![],
                                        skipped_segments: vec![],
                                        skipped_positions: Vec::new(),
                                    }],
                                }],
                                entry_mig_number: None,
                                variant_mig_numbers: vec![],
                                skipped_segments: vec![],
                                skipped_positions: Vec::new(),
                            },
                            AssembledGroupInstance {
                                segments: vec![AssembledSegment {
                                    tag: "SEQ".to_string(),
                                    elements: vec![vec!["ZD7".to_string()]],
                                    mig_number: None,
                                    segment_number: None,
                                }],
                                child_groups: vec![AssembledGroup {
                                    group_id: "SG10".to_string(),
                                    repetitions: vec![AssembledGroupInstance {
                                        segments: vec![AssembledSegment {
                                            tag: "CCI".to_string(),
                                            elements: vec![vec![], vec![], vec!["ZE6".to_string()]],
                                            mig_number: None,
                                            segment_number: None,
                                        }],
                                        child_groups: vec![],
                                        entry_mig_number: None,
                                        variant_mig_numbers: vec![],
                                        skipped_segments: vec![],
                                        skipped_positions: Vec::new(),
                                    }],
                                }],
                                entry_mig_number: None,
                                variant_mig_numbers: vec![],
                                skipped_segments: vec![],
                                skipped_positions: Vec::new(),
                            },
                        ],
                    }],
                    entry_mig_number: None,
                    variant_mig_numbers: vec![],
                    skipped_segments: vec![],
                    skipped_positions: Vec::new(),
                }],
            }],
            post_group_start: 0,
            inter_group_segments: std::collections::BTreeMap::new(),
        };

        // Resolve SG10 under Z98
        let inst = MappingEngine::resolve_by_source_path(&tree, "sg4.sg8_z98.sg10");
        assert!(inst.is_some());
        assert_eq!(inst.unwrap().segments[0].elements[2][0], "ZB3");

        // Resolve SG10 under ZD7
        let inst = MappingEngine::resolve_by_source_path(&tree, "sg4.sg8_zd7.sg10");
        assert!(inst.is_some());
        assert_eq!(inst.unwrap().segments[0].elements[2][0], "ZE6");

        // Unknown qualifier → None
        let inst = MappingEngine::resolve_by_source_path(&tree, "sg4.sg8_zzz.sg10");
        assert!(inst.is_none());

        // Without qualifier → first rep (Z98)
        let inst = MappingEngine::resolve_by_source_path(&tree, "sg4.sg8.sg10");
        assert!(inst.is_some());
        assert_eq!(inst.unwrap().segments[0].elements[2][0], "ZB3");
    }

    #[test]
    fn test_parse_source_path_part() {
        assert_eq!(parse_source_path_part("sg4"), ("sg4", None));
        assert_eq!(parse_source_path_part("sg8_z98"), ("sg8", Some("z98")));
        assert_eq!(parse_source_path_part("sg10"), ("sg10", None));
        assert_eq!(parse_source_path_part("sg12_z04"), ("sg12", Some("z04")));
    }

    #[test]
    fn test_has_source_path_qualifiers() {
        assert!(has_source_path_qualifiers("sg4.sg8_z98.sg10"));
        assert!(has_source_path_qualifiers("sg4.sg8_ze1.sg9"));
        assert!(!has_source_path_qualifiers("sg4.sg6"));
        assert!(!has_source_path_qualifiers("sg4.sg8.sg10"));
    }

    #[test]
    fn test_extract_all_from_instance_collects_all_qualifier_matches() {
        use mig_assembly::assembler::*;

        // Instance with 3 RFF+Z34 segments
        let instance = AssembledGroupInstance {
            segments: vec![
                AssembledSegment {
                    tag: "SEQ".to_string(),
                    elements: vec![vec!["ZD6".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "RFF".to_string(),
                    elements: vec![vec!["Z34".to_string(), "REF_A".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "RFF".to_string(),
                    elements: vec![vec!["Z34".to_string(), "REF_B".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "RFF".to_string(),
                    elements: vec![vec!["Z34".to_string(), "REF_C".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
                AssembledSegment {
                    tag: "RFF".to_string(),
                    elements: vec![vec!["Z35".to_string(), "OTHER".to_string()]],
                    mig_number: None,
                    segment_number: None,
                },
            ],
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: Vec::new(),
        };

        // Wildcard collect: rff[Z34,*] should collect all 3 RFF+Z34 values
        let all = MappingEngine::extract_all_from_instance(&instance, "rff[Z34,*].0.1");
        assert_eq!(all, vec!["REF_A", "REF_B", "REF_C"]);

        // Non-wildcard still returns single value via extract_from_instance
        let single = MappingEngine::extract_from_instance(&instance, "rff[Z34].0.1");
        assert_eq!(single, Some("REF_A".to_string()));

        let second = MappingEngine::extract_from_instance(&instance, "rff[Z34,1].0.1");
        assert_eq!(second, Some("REF_B".to_string()));
    }
}