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::{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>,
}

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,
        }
    }

    /// 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 =
                    toml::from_str(&content).map_err(|e| MappingError::TomlParse {
                        file: path.display().to_string(),
                        message: e.to_string(),
                    })?;
                definitions.push(def);
            }
        }

        Ok(Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
        })
    }

    /// 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,
        })
    }

    /// 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,
        })
    }

    /// 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,
        })
    }

    /// 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.
    pub fn from_definitions(definitions: Vec<MappingDefinition>) -> Self {
        Self {
            definitions,
            segment_structure: None,
            code_lookup: None,
            transaction_group: None,
            current_pid: None,
        }
    }

    /// 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,
        })
    }

    /// 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 companion field values.
    ///
    /// When set, companion 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
    }

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

    /// Find a definition by entity name.
    pub fn definition_for_entity(&self, entity: &str) -> Option<&MappingDefinition> {
        self.definitions.iter().find(|d| d.meta.entity == entity)
    }

    // ── 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.
    ///
    /// If the definition has `companion_fields`, those are extracted into a nested
    /// object keyed by `companion_type` — or flattened directly into the result
    /// when `companion_type` is not specified.
    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);
            self.extract_companion_fields(&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,
                        );
                        self.extract_companion_fields(
                            &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);
            self.extract_companion_fields(instance, def, &mut result, enrich_codes);
        }

        serde_json::Value::Object(result)
    }

    /// Extract companion_fields into the result map.
    ///
    /// When `companion_type` is set in `[meta]`, fields are nested under a camelCased
    /// key (e.g. `"betragEdifact": { ... }`). When it is omitted, companion fields are
    /// flattened directly into the parent `result` — the previous behavior of wrapping
    /// them under an internal `_companion` sentinel leaked an underscore-prefixed key
    /// into the BO4E JSON output.
    ///
    /// When a `code_lookup` is configured, code-type fields are emitted as
    /// `{"code": "Z15", "meaning": "Ja"}` objects. Data-type fields remain plain strings.
    fn extract_companion_fields(
        &self,
        instance: &AssembledGroupInstance,
        def: &MappingDefinition,
        result: &mut serde_json::Map<String, serde_json::Value>,
        enrich_codes: bool,
    ) {
        if let Some(ref companion_fields) = def.companion_fields {
            let companion_key = def.meta.companion_type.as_deref().map(to_camel_case);
            let mut companion_result = serde_json::Map::new();

            for (path, field_mapping) in companion_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(),
                        s.enum_map.as_ref(),
                        s.also_target.as_deref(),
                        s.also_enum_map.as_ref(),
                    ),
                    FieldMapping::Nested(_) => continue,
                };
                if target.is_empty() {
                    continue;
                }

                // Wildcard collect: rff[Z34,*].0.1 → JSON array of all matches
                if is_collect_all_path(path) {
                    let all = Self::extract_all_from_instance(instance, path);
                    if !all.is_empty() {
                        let arr: Vec<serde_json::Value> = all
                            .into_iter()
                            .map(|v| {
                                let mapped = if let Some(map) = enum_map {
                                    map.get(&v).cloned().unwrap_or_else(|| v.clone())
                                } else {
                                    v
                                };
                                serde_json::Value::String(mapped)
                            })
                            .collect();
                        set_nested_value_json(
                            &mut companion_result,
                            target,
                            serde_json::Value::Array(arr),
                        );
                    }
                    continue;
                }

                if let Some(val) = Self::extract_from_instance(instance, path) {
                    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, _qualifier, _occ) = parse_tag_qualifier(parts[0]);
                            let (element_idx, component_idx) =
                                Self::parse_element_component(&parts[1..]);
                            let disc_qualifier = Self::discriminator_qualifier(def);
                            let q = disc_qualifier.as_deref();

                            if code_lookup.is_code_field_q(
                                source_path,
                                &seg_tag,
                                q,
                                element_idx,
                                component_idx,
                            ) {
                                // Class C: PID self-reference — emit a plain string,
                                // skipping decoration. The schema's only allowed code
                                // here is the PID itself, so {code, meaning, enum}
                                // wrappers add no signal and leak dedup-suffixed
                                // enum names like PRUEFIDENTIFIKATOR_3.
                                if let Some(ref pid) = self.current_pid {
                                    if code_lookup.is_pid_self_reference(
                                        source_path,
                                        &seg_tag,
                                        q,
                                        element_idx,
                                        component_idx,
                                        pid,
                                    ) {
                                        set_nested_value(&mut companion_result, target, mapped_val);
                                        if let (Some(at), Some(am)) = (also_target, also_enum_map) {
                                            if let Some(also_mapped) = am.get(&val) {
                                                set_nested_value(
                                                    &mut companion_result,
                                                    at,
                                                    also_mapped.clone(),
                                                );
                                            }
                                        }
                                        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 = code_lookup.enrichment_for_q(
                                    source_path,
                                    &seg_tag,
                                    q,
                                    element_idx,
                                    component_idx,
                                    &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(&mut companion_result, target, enriched);
                                continue;
                            }
                        }
                    }

                    set_nested_value(&mut companion_result, target, mapped_val);

                    // Dual decomposition: also extract a second field from the same value.
                    // Only set also_target when the code IS in also_enum_map (mixed codes
                    // without a quality dimension simply don't get the second field).
                    if let (Some(at), Some(am)) = (also_target, also_enum_map) {
                        if let Some(also_mapped) = am.get(&val) {
                            set_nested_value(&mut companion_result, at, also_mapped.clone());
                        }
                    }
                }
            }

            if !companion_result.is_empty() {
                match companion_key {
                    Some(key) => {
                        result.insert(key, serde_json::Value::Object(companion_result));
                    }
                    None => {
                        // No companion_type set: flatten companion fields into the
                        // parent result directly. Existing keys are preserved (core
                        // fields take precedence — companion fields don't overwrite
                        // them).
                        for (k, v) in companion_result {
                            result.entry(k).or_insert(v);
                        }
                    }
                }
            }
        }
    }

    /// 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(), s.enum_map.as_ref()),
                FieldMapping::Nested(_) => continue,
            };
            if target.is_empty() {
                continue;
            }
            if let Some(val) = Self::extract_from_instance(instance, path) {
                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, _qualifier, _occ) = parse_tag_qualifier(parts[0]);
                        let (element_idx, component_idx) =
                            Self::parse_element_component(&parts[1..]);
                        let disc_qualifier = Self::discriminator_qualifier(def);
                        let q = disc_qualifier.as_deref();

                        if code_lookup.is_code_field_q(
                            source_path,
                            &seg_tag,
                            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 code_lookup.is_pid_self_reference(
                                    source_path,
                                    &seg_tag,
                                    q,
                                    element_idx,
                                    component_idx,
                                    pid,
                                ) {
                                    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 = code_lookup.enrichment_for_q(
                                source_path,
                                &seg_tag,
                                q,
                                element_idx,
                                component_idx,
                                &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);
            }
        }
    }

    /// 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.
    fn discriminator_qualifier(def: &MappingDefinition) -> Option<String> {
        def.meta
            .discriminator
            .as_deref()
            .and_then(|d| d.split_once('=').map(|(_, v)| v.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(),
                };
            }
        }
        self.map_reverse_single(bo4e_value, def)
    }

    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) = match field_mapping {
                FieldMapping::Simple(t) => (t.as_str(), None, None, None),
                FieldMapping::Structured(s) => (
                    s.target.as_str(),
                    s.default.as_ref(),
                    s.enum_map.as_ref(),
                    s.when_filled.as_ref(),
                ),
                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
                    // Check both core and companion objects (ref field may be in either)
                    (Some(d), Some(fields)) => {
                        let companion_key_for_check =
                            def.meta.companion_type.as_deref().map(to_camel_case);
                        let companion_for_check = companion_key_for_check
                            .as_ref()
                            .and_then(|k| bo4e_value.get(k))
                            .unwrap_or(&serde_json::Value::Null);
                        let any_filled = fields.iter().any(|f| {
                            self.populate_field(bo4e_value, f).is_some()
                                || self.populate_field(companion_for_check, f).is_some()
                        });
                        if any_filled {
                            // A successful when_filled check confirms real data
                            // exists — prevent phantom suppression even when
                            // companion data fields are absent.
                            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)) => {
                        // 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()));
                }
            }
        }

        // Process companion_fields — values are nested under the companion type key.
        // Fallback: when no *Edifact wrapper exists (typed PID format) or when
        // `companion_type` is unset (companion fields are flattened into the
        // entity root on forward), look directly in the entity root object.
        if let Some(ref companion_fields) = def.companion_fields {
            let companion_value = match def.meta.companion_type.as_deref() {
                Some(raw_key) => {
                    let key = to_camel_case(raw_key);
                    bo4e_value.get(&key).unwrap_or(bo4e_value)
                }
                None => bo4e_value,
            };

            for (path, field_mapping) in companion_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(),
                            s.enum_map.as_ref(),
                            s.when_filled.as_ref(),
                            s.also_target.as_deref(),
                            s.also_enum_map.as_ref(),
                        ),
                        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]);
                let seg_key = parts[0].to_uppercase();
                let sub_path = &parts[1..];

                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,
                    }
                };

                // Wildcard collect reverse: read JSON array, expand to N segments
                if is_collect_all_path(path) && !target.is_empty() {
                    if let Some(arr) = self
                        .populate_field_json(companion_value, target)
                        .and_then(|v| v.as_array().cloned())
                    {
                        has_data_fields = true;
                        if !arr.is_empty() {
                            has_real_data = true;
                        }
                        for (idx, item) in arr.iter().enumerate() {
                            if let Some(val_str) = item.as_str() {
                                let mapped = if let Some(map) = enum_map {
                                    map.iter()
                                        .find(|(_, bo4e_v)| *bo4e_v == val_str)
                                        .map(|(edifact_k, _)| edifact_k.clone())
                                        .unwrap_or_else(|| val_str.to_string())
                                } else {
                                    val_str.to_string()
                                };
                                let occ_key = if let Some(q) = qualifier {
                                    format!("{}[{},{}]", seg_tag, q, idx)
                                } else {
                                    format!("{}[*,{}]", seg_tag, idx)
                                };
                                field_values.push((
                                    occ_key.clone(),
                                    seg_tag.clone(),
                                    element_idx,
                                    component_idx,
                                    mapped,
                                ));
                                // Inject qualifier for each occurrence
                                if let Some(q) = qualifier {
                                    if injected_qualifiers.insert(occ_key.clone()) {
                                        field_values.push((
                                            occ_key,
                                            seg_tag.clone(),
                                            0,
                                            0,
                                            q.to_string(),
                                        ));
                                    }
                                }
                            }
                        }
                    }
                    continue;
                }

                let val = if target.is_empty() {
                    match (default, when_filled) {
                        (Some(d), Some(fields)) => {
                            let any_filled = fields.iter().any(|f| {
                                self.populate_field(bo4e_value, f).is_some()
                                    || self.populate_field(companion_value, f).is_some()
                            });
                            if any_filled {
                                has_real_data = true;
                                Some(d.clone())
                            } else {
                                None
                            }
                        }
                        (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(companion_value, target);
                    if bo4e_val.is_some() {
                        has_real_data = true;
                        seg_has_real_data.insert(seg_key.clone());
                    }
                    let mapped_val = match (bo4e_val, enum_map) {
                        (Some(v), Some(map)) => {
                            if let (Some(at), Some(am)) = (also_target, also_enum_map) {
                                let also_val = self.populate_field(companion_value, at);
                                if let Some(av) = also_val.as_deref() {
                                    // Joint lookup: find code where BOTH maps match
                                    map.iter()
                                        .find(|(edifact_k, bo4e_v)| {
                                            *bo4e_v == &v
                                                && am.get(*edifact_k).is_some_and(|am_v| am_v == av)
                                        })
                                        .map(|(edifact_k, _)| edifact_k.clone())
                                        .or(Some(v))
                                } else {
                                    // also_target absent: find code matching enum_map
                                    // that is NOT in also_enum_map (unpaired code)
                                    map.iter()
                                        .find(|(edifact_k, bo4e_v)| {
                                            *bo4e_v == &v && !am.contains_key(*edifact_k)
                                        })
                                        .or_else(|| {
                                            // Fallback: any matching code
                                            map.iter().find(|(_, bo4e_v)| *bo4e_v == &v)
                                        })
                                        .map(|(edifact_k, _)| edifact_k.clone())
                                        .or(Some(v))
                                }
                            } else {
                                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 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)
    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())
    }

    /// Extract a raw JSON value from a BO4E JSON object by target field name.
    /// Like `populate_field` but returns the `serde_json::Value` instead of coercing to String.
    fn populate_field_json<'a>(
        &self,
        bo4e_value: &'a serde_json::Value,
        target_field: &str,
    ) -> Option<&'a serde_json::Value> {
        let mut current = bo4e_value;
        for part in target_field.split('.') {
            current = current.get(part)?;
        }
        Some(current)
    }

    /// 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, dp_routing)` 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).
    /// - `dp_routing` carries side-channel metadata for any NAD+DP segments
    ///   that were routed out of marktteilnehmer[] into a lokation entity
    ///   (Phase 8). The reverse mapper consumes it to rebuild the segment.
    fn map_all_forward_inner(
        &self,
        tree: &AssembledTree,
        enrich_codes: bool,
    ) -> (
        serde_json::Value,
        std::collections::HashMap<String, Vec<usize>>,
        DpRouting,
    ) {
        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>>,
        DpRouting,
    ) {
        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 {
            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);
                        self.extract_companion_fields(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);
                    self.extract_companion_fields(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 bo4e = inject_bo4e_metadata(bo4e, &def.meta.bo4e_type);
                let key = to_camel_case(entity);
                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,
        );

        // Post-process: route NAD+DP entries out of marktteilnehmer[] into a
        // standalone Marktlokation or Messlokation entity. Phase 8 of the
        // 2026-04-28 audit: NAD+DP carries a delivery-point reference + address,
        // not a market role. Routing metadata travels in the side-channel so
        // the public BO4E JSON stays free of internal `_dpSource` markers.
        let dp_routing = route_nad_dp_to_lokation(&mut result);

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

    /// 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 {
        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 {
            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(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 companion 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(ref cf) = def.companion_fields {
                            let disc_path = def
                                .meta
                                .discriminator
                                .as_deref()
                                .unwrap()
                                .split_once('=')
                                .unwrap()
                                .0
                                .to_lowercase();
                            for (path, mapping) in cf {
                                // Compare paths, handling 2-part vs 3-part format mismatch.
                                // resolve_path produces "nad.0" (2-part for simple elements),
                                // resolve_discriminator produces "NAD.0.0" (always 3-part).
                                let cf_path = path.to_lowercase();
                                let matches =
                                    cf_path == disc_path || format!("{}.0", cf_path) == disc_path;
                                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() {
                                        if let Some(obj) = injected.as_object_mut() {
                                            let entry = obj
                                                .entry(target.to_string())
                                                .or_insert(serde_json::Value::Null);
                                            if entry.is_null() {
                                                *entry = serde_json::Value::String(
                                                    disc_value.to_string(),
                                                );
                                            }
                                        }
                                    }
                                    break;
                                }
                            }
                        }
                        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_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();
                        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))
                });
                for (i, child_rep) in child_group.repetitions.into_iter().enumerate() {
                    let target_idx = distribution
                        .and_then(|dist| dist.get(i))
                        .copied()
                        .unwrap_or(0);

                    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
    }

    /// 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 {
        // Map message-level entities (also captures nesting + DP routing info)
        let (stammdaten, nesting_info, dp_routing) =
            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_dp_routing) = tx_engine
                            .map_all_forward_inner_with_tx(
                                &wrapped_tree,
                                enrich_codes,
                                Some(transaction_group),
                            );

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

        crate::model::MappedMessage {
            stammdaten,
            transaktionen,
            nesting_info,
            dp_routing,
            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. If forward mapping routed
        // any NAD+DP entries out of marktteilnehmer[] into a lokation entity,
        // unroute them first using the side-channel metadata so the per-def
        // loop sees the original marktteilnehmer shape. Clone only when
        // routing actually happened — keeps the common path zero-copy.
        let _owned_msg: Option<serde_json::Value>;
        let msg_stammdaten = if !mapped.dp_routing.is_empty() {
            if let serde_json::Value::Object(map) = &mapped.stammdaten {
                let mut cloned = map.clone();
                unroute_lokation_to_nad_dp(&mut cloned, &mapped.dp_routing);
                _owned_msg = Some(serde_json::Value::Object(cloned));
                _owned_msg.as_ref().unwrap()
            } else {
                _owned_msg = None;
                &mapped.stammdaten
            }
        } else {
            _owned_msg = None;
            &mapped.stammdaten
        };

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

        // 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()
            .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();

            // Apply per-transaction NAD+DP unroute when forward mapping
            // captured tx-level DP routing. Without this, fixtures whose AHB
            // places NAD+DP under a transaction-level group (vs message-level
            // SG2) would lose the segment on reverse — see issue #61.
            let _owned_tx: Option<serde_json::Value>;
            let tx_stammdaten: &serde_json::Value = if !tx.dp_routing.is_empty() {
                if let serde_json::Value::Object(map) = &tx.stammdaten {
                    let mut cloned = map.clone();
                    unroute_lokation_to_nad_dp(&mut cloned, &tx.dp_routing);
                    _owned_tx = Some(serde_json::Value::Object(cloned));
                    _owned_tx.as_ref().unwrap()
                } else {
                    _owned_tx = None;
                    &tx.stammdaten
                }
            } 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(ref cf) = dm.def.companion_fields {
                                let disc_path = dm
                                    .def
                                    .meta
                                    .discriminator
                                    .as_deref()
                                    .unwrap()
                                    .split_once('=')
                                    .unwrap()
                                    .0
                                    .to_lowercase();
                                for (path, mapping) in cf {
                                    let cf_path = path.to_lowercase();
                                    let matches = cf_path == disc_path
                                        || format!("{}.0", cf_path) == disc_path;
                                    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() {
                                            if let Some(obj) = injected.as_object_mut() {
                                                obj.entry(target.to_string()).or_insert_with(
                                                    || {
                                                        serde_json::Value::String(
                                                            disc_value.to_string(),
                                                        )
                                                    },
                                                );
                                            }
                                        }
                                        break;
                                    }
                                }
                            }
                            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() {
                        root_segs.extend(instance.segments);
                    } 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);
                            }
                        }
                    }
                }
            }

            // Sort variant reps within each child group to match MIG order.
            // The reverse mapper appends reps in definition-filename order, but
            // the assembler captures them in MIG variant order. Use the filtered
            // MIG's nested_groups as the canonical ordering.
            if let Some(mig) = filtered_mig {
                sort_variant_reps_by_mig(&mut child_groups, mig, transaction_group);
            }

            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());
            }
        }

        AssembledTree {
            segments: root_segments,
            groups: all_groups,
            post_group_start: pre_group_count,
            inter_group_segments: inter_group,
        }
    }

    /// 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,
    {
        let stammdaten = serde_json::to_value(&nachricht.stammdaten)?;
        let transaktionen: Vec<crate::model::MappedTransaktion> = nachricht
            .transaktionen
            .iter()
            .map(|t| {
                Ok(crate::model::MappedTransaktion {
                    stammdaten: serde_json::to_value(t)?,
                    nesting_info: Default::default(),
                    dp_routing: Default::default(),
                })
            })
            .collect::<Result<Vec<_>, serde_json::Error>>()?;
        let mapped = crate::model::MappedMessage {
            stammdaten,
            transaktionen,
            nesting_info: Default::default(),
            dp_routing: 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.
fn strip_all_rep_indices(relative: &str) -> String {
    relative
        .split('.')
        .map(|part| {
            let (id, _) = parse_group_spec(part);
            id
        })
        .collect::<Vec<_>>()
        .join(".")
}

/// Check whether a path uses the `*` occurrence wildcard (e.g., `rff[Z34,*].0.1`).
///
/// When `*` appears in the occurrence position, `extract_all_from_instance` should
/// be used to collect ALL matching segments instead of selecting a single one.
fn is_collect_all_path(path: &str) -> bool {
    let tag_part = path.split('.').next().unwrap_or("");
    if let Some(bracket_start) = tag_part.find('[') {
        let inner = tag_part[bracket_start + 1..].trim_end_matches(']');
        if let Some(comma_pos) = inner.find(',') {
            let qualifier = &inner[..comma_pos];
            let occ = &inner[comma_pos + 1..];
            // Collect-all: qualifier is NOT *, but occurrence IS *
            qualifier != "*" && occ == "*"
        } else {
            false
        }
    } else {
        false
    }
}

/// 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; use `is_collect_all_path` to detect
/// - `"rff"`         → `("RFF", None, 0)`
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)
    }
}

/// Inject `boTyp` and `versionStruktur` metadata into a BO4E JSON value.
///
/// For objects, inserts both fields (without overwriting existing ones).
/// For arrays, injects into each element object.
fn inject_bo4e_metadata(mut value: serde_json::Value, bo4e_type: &str) -> serde_json::Value {
    match &mut value {
        serde_json::Value::Object(map) => {
            map.entry("boTyp")
                .or_insert_with(|| serde_json::Value::String(bo4e_type.to_uppercase()));
            map.entry("versionStruktur")
                .or_insert_with(|| serde_json::Value::String("1".to_string()));
        }
        serde_json::Value::Array(items) => {
            for item in items {
                if let serde_json::Value::Object(map) = item {
                    map.entry("boTyp")
                        .or_insert_with(|| serde_json::Value::String(bo4e_type.to_uppercase()));
                    map.entry("versionStruktur")
                        .or_insert_with(|| serde_json::Value::String("1".to_string()));
                }
            }
        }
        _ => {}
    }
    value
}

/// 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);
}

/// 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 {
            continue;
        }
        let disc = def.meta.discriminator.as_deref()?;
        let (disc_path, _) = disc.split_once('=')?;
        let disc_path_lower = disc_path.to_lowercase();

        // Search both [companion_fields] and [fields] — the qualifier field may
        // be in either section (e.g., Marktteilnehmer has "marktrolle" in [fields]).
        let sections: Vec<&indexmap::IndexMap<String, FieldMapping>> =
            [def.companion_fields.as_ref(), Some(&def.fields)]
                .into_iter()
                .flatten()
                .collect();

        for section in sections {
            for (path, mapping) in section {
                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>,
) {
    // 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 {
            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(),
            ));
        }
    }

    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);
    }
}

// NAD+DP routing post-processor lives in `crate::dp_routing`.
pub use crate::dp_routing::{route_nad_dp_to_lokation, unroute_lokation_to_nad_dp, DpRouting};

/// 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())
        })
}

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: HashMap<String, Vec<MappingDefinition>>,
    /// Per-PID combined definitions (key: "pid_55001").
    pub combined_defs: HashMap<String, Vec<MappingDefinition>>,
    /// Per-PID code lookups (key: "pid_55001"). Cached to avoid reading schema JSONs at load time.
    #[serde(default)]
    pub code_lookups: HashMap<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>,
    /// 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: HashMap<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: HashMap<String, crate::pid_requirements::PidRequirements>,
    /// 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: HashMap<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)?;
        serde_json::from_slice(&bytes).map_err(|e| MappingError::CacheRead {
            path: path.display().to_string(),
            message: e.to_string(),
        })
    }

    /// 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(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(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: HashMap<String, VariantCache>,
}

impl DataBundle {
    pub const CURRENT_VERSION: u32 = 2;

    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 bundle = Self::read_from(reader)?;
        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)
    }
}

/// Sort variant reps within child groups to match MIG-defined variant order.
///
/// The reverse mapper appends reps in definition-filename order, but the
/// assembler captures them in the order MIG variants are defined (which is
/// the canonical EDIFACT order). This function reorders reps within same-ID
/// groups to match the MIG's nested_groups ordering.
///
/// Uses position-aware qualifier matching: each MIG variant has a
/// `variant_code` and `variant_qualifier_position` that specifies WHERE
/// the qualifier lives in the entry segment (e.g., SEQ qualifier at [0][0],
/// CCI qualifier at [2][0]). This correctly handles groups where different
/// variants have qualifiers at different positions.
fn sort_variant_reps_by_mig(
    child_groups: &mut [AssembledGroup],
    mig: &MigSchema,
    transaction_group: &str,
) {
    let tx_def = match mig
        .segment_groups
        .iter()
        .find(|sg| sg.id == transaction_group)
    {
        Some(d) => d,
        None => return,
    };

    for cg in child_groups.iter_mut() {
        if cg.repetitions.len() <= 1 {
            continue;
        }

        // Collect all MIG variant definitions for this group_id, in MIG order.
        let variant_defs: Vec<(usize, &mig_types::schema::mig::MigSegmentGroup)> = tx_def
            .nested_groups
            .iter()
            .enumerate()
            .filter(|(_, ng)| ng.id == cg.group_id && ng.variant_code.is_some())
            .collect();

        if variant_defs.is_empty() {
            continue;
        }

        // Sort reps: for each rep, find which MIG variant it matches by
        // checking the entry segment's qualifier at each variant's specific position.
        cg.repetitions.sort_by_key(|rep| {
            let entry_seg = rep.segments.first();
            for &(mig_pos, variant_def) in &variant_defs {
                let (ei, ci) = variant_def.variant_qualifier_position.unwrap_or((0, 0));
                let actual_qual = entry_seg
                    .and_then(|s| s.elements.get(ei))
                    .and_then(|e| e.get(ci))
                    .map(|s| s.as_str())
                    .unwrap_or("");
                let matches = if !variant_def.variant_codes.is_empty() {
                    variant_def
                        .variant_codes
                        .iter()
                        .any(|c| actual_qual.eq_ignore_ascii_case(c))
                } else if let Some(ref expected_code) = variant_def.variant_code {
                    actual_qual.eq_ignore_ascii_case(expected_code)
                } else {
                    false
                };
                if matches {
                    return mig_pos;
                }
            }
            usize::MAX // unmatched reps go to the end
        });
    }
}

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

    fn make_test_cache() -> VariantCache {
        let mut tx_groups = HashMap::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 = HashMap::new();
        transaction_defs.insert("pid_55001".to_string(), vec![]);
        transaction_defs.insert("pid_21007".to_string(), vec![]);

        VariantCache {
            message_defs: vec![],
            transaction_defs,
            combined_defs: HashMap::new(),
            code_lookups: HashMap::new(),
            mig_schema: None,
            segment_structure: None,
            pid_segment_numbers: HashMap::new(),
            pid_requirements: HashMap::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());
    }
}

#[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(),
                companion_type: None,
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields,
            companion_fields: None,
            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(),
                    companion_type: None,
                    source_group: "SG4".to_string(),
                    source_path: None,
                    discriminator: None,
                    repeat_on_tag: None,
                },
                fields: tx_fields,
                companion_fields: None,
                complex_handlers: None,
            },
            MappingDefinition {
                meta: MappingMeta {
                    entity: "Marktlokation".to_string(),
                    bo4e_type: "Marktlokation".to_string(),
                    companion_type: None,
                    source_group: "SG4.SG5".to_string(),
                    source_path: None,
                    discriminator: None,
                    repeat_on_tag: None,
                },
                fields: malo_fields,
                companion_fields: None,
                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].stammdaten["prozessdaten"]["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,
                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,
                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,
                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(),
                companion_type: None,
                source_group: "SG2".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: msg_fields,
            companion_fields: None,
            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(),
                companion_type: None,
                source_group: "".to_string(), // Root-level within transaction sub-tree
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: proz_fields,
            companion_fields: None,
            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(),
                companion_type: None,
                source_group: "SG5".to_string(), // Relative to SG4, not "SG4.SG5"
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: malo_fields,
            companion_fields: None,
            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(),
                companion_type: None,
                source_group: "SG2".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: msg_fields,
            companion_fields: None,
            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(),
                companion_type: None,
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: tx_fields,
            companion_fields: None,
            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].stammdaten["prozessdaten"]["vorgangId"]
                .as_str()
                .unwrap(),
            "TX001"
        );
        assert_eq!(
            result.transaktionen[1].stammdaten["prozessdaten"]["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,
                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 HashMap
        let mut counts = std::collections::HashMap::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_extract_companion_fields_with_code_enrichment() {
        use crate::code_lookup::CodeLookup;
        use mig_assembly::assembler::*;

        let schema = serde_json::json!({
            "fields": {
                "sg4": {
                    "children": {
                        "sg8_z01": {
                            "children": {
                                "sg10": {
                                    "segments": [{
                                        "id": "CCI",
                                        "elements": [{
                                            "index": 2,
                                            "components": [{
                                                "sub_index": 0,
                                                "type": "code",
                                                "codes": [
                                                    {"value": "Z15", "name": "Haushaltskunde"},
                                                    {"value": "Z18", "name": "Kein Haushaltskunde"}
                                                ]
                                            }]
                                        }]
                                    }],
                                    "source_group": "SG10"
                                }
                            },
                            "segments": [],
                            "source_group": "SG8"
                        }
                    },
                    "segments": [],
                    "source_group": "SG4"
                }
            }
        });

        let code_lookup = CodeLookup::from_schema_value(&schema);

        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![],
                            child_groups: vec![AssembledGroup {
                                group_id: "SG10".to_string(),
                                repetitions: vec![AssembledGroupInstance {
                                    segments: vec![AssembledSegment {
                                        tag: "CCI".to_string(),
                                        elements: vec![vec![], vec![], vec!["Z15".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(),
        };

        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.2".to_string(),
            FieldMapping::Simple("haushaltskunde".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Marktlokation".to_string(),
                bo4e_type: "Marktlokation".to_string(),
                companion_type: Some("MarktlokationEdifact".to_string()),
                source_group: "SG4.SG8.SG10".to_string(),
                source_path: Some("sg4.sg8_z01.sg10".to_string()),
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

        // Without code lookup — plain string
        let engine_plain = MappingEngine::from_definitions(vec![]);
        let bo4e_plain = engine_plain.map_forward(&tree, &def, 0);
        assert_eq!(
            bo4e_plain["marktlokationEdifact"]["haushaltskunde"].as_str(),
            Some("Z15"),
            "Without code lookup, should be plain string"
        );

        // With code lookup — enriched object
        let engine_enriched = MappingEngine::from_definitions(vec![]).with_code_lookup(code_lookup);
        let bo4e_enriched = engine_enriched.map_forward(&tree, &def, 0);
        let hk = &bo4e_enriched["marktlokationEdifact"]["haushaltskunde"];
        assert_eq!(hk["code"].as_str(), Some("Z15"));
        assert_eq!(hk["meaning"].as_str(), Some("Haushaltskunde"));
        // Without "enum" in schema codes, no "enum" in output
        assert!(hk.get("enum").is_none());
    }

    #[test]
    fn test_extract_companion_fields_with_enum_enrichment() {
        use crate::code_lookup::CodeLookup;
        use mig_assembly::assembler::*;

        // Schema with "enum" field on codes
        let schema = serde_json::json!({
            "fields": {
                "sg4": {
                    "children": {
                        "sg8_z01": {
                            "children": {
                                "sg10": {
                                    "segments": [{
                                        "id": "CCI",
                                        "elements": [{
                                            "index": 2,
                                            "components": [{
                                                "sub_index": 0,
                                                "type": "code",
                                                "codes": [
                                                    {"value": "Z15", "name": "Haushaltskunde", "enum": "HAUSHALTSKUNDE"},
                                                    {"value": "Z18", "name": "Kein Haushaltskunde", "enum": "KEIN_HAUSHALTSKUNDE"}
                                                ]
                                            }]
                                        }]
                                    }],
                                    "source_group": "SG10"
                                }
                            },
                            "segments": [],
                            "source_group": "SG8"
                        }
                    },
                    "segments": [],
                    "source_group": "SG4"
                }
            }
        });

        let code_lookup = CodeLookup::from_schema_value(&schema);

        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![],
                            child_groups: vec![AssembledGroup {
                                group_id: "SG10".to_string(),
                                repetitions: vec![AssembledGroupInstance {
                                    segments: vec![AssembledSegment {
                                        tag: "CCI".to_string(),
                                        elements: vec![vec![], vec![], vec!["Z15".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(),
        };

        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.2".to_string(),
            FieldMapping::Simple("haushaltskunde".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Marktlokation".to_string(),
                bo4e_type: "Marktlokation".to_string(),
                companion_type: Some("MarktlokationEdifact".to_string()),
                source_group: "SG4.SG8.SG10".to_string(),
                source_path: Some("sg4.sg8_z01.sg10".to_string()),
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

        let engine = MappingEngine::from_definitions(vec![]).with_code_lookup(code_lookup);
        let bo4e = engine.map_forward(&tree, &def, 0);
        let hk = &bo4e["marktlokationEdifact"]["haushaltskunde"];
        assert_eq!(hk["code"].as_str(), Some("Z15"));
        assert_eq!(hk["meaning"].as_str(), Some("Haushaltskunde"));
        assert_eq!(
            hk["enum"].as_str(),
            Some("HAUSHALTSKUNDE"),
            "enum field should be present"
        );
    }

    #[test]
    fn test_reverse_mapping_accepts_enriched_with_enum() {
        // Reverse mapping should ignore "enum" field — only reads "code"
        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.2".to_string(),
            FieldMapping::Simple("haushaltskunde".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Test".to_string(),
                bo4e_type: "Test".to_string(),
                companion_type: Some("TestEdifact".to_string()),
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

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

        let bo4e = serde_json::json!({
            "testEdifact": {
                "haushaltskunde": {
                    "code": "Z15",
                    "meaning": "Haushaltskunde",
                    "enum": "HAUSHALTSKUNDE"
                }
            }
        });
        let instance = engine.map_reverse(&bo4e, &def);
        assert_eq!(instance.segments[0].elements[2], vec!["Z15"]);
    }

    #[test]
    fn test_reverse_mapping_accepts_enriched_companion() {
        // Reverse mapping should accept both plain string and enriched object format
        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.2".to_string(),
            FieldMapping::Simple("haushaltskunde".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Test".to_string(),
                bo4e_type: "Test".to_string(),
                companion_type: Some("TestEdifact".to_string()),
                source_group: "SG4".to_string(),
                source_path: None,
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

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

        // Test 1: Plain string format (backward compat)
        let bo4e_plain = serde_json::json!({
            "testEdifact": {
                "haushaltskunde": "Z15"
            }
        });
        let instance_plain = engine.map_reverse(&bo4e_plain, &def);
        assert_eq!(instance_plain.segments[0].elements[2], vec!["Z15"]);

        // Test 2: Enriched object format
        let bo4e_enriched = serde_json::json!({
            "testEdifact": {
                "haushaltskunde": {
                    "code": "Z15",
                    "meaning": "Haushaltskunde gem. EnWG"
                }
            }
        });
        let instance_enriched = engine.map_reverse(&bo4e_enriched, &def);
        assert_eq!(instance_enriched.segments[0].elements[2], vec!["Z15"]);
    }

    #[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_companion_dotted_path_forward() {
        use mig_assembly::assembler::*;

        // Build an assembled tree with a CCI segment inside SG4.SG8.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![],
                            child_groups: vec![AssembledGroup {
                                group_id: "SG10".to_string(),
                                repetitions: vec![AssembledGroupInstance {
                                    segments: vec![AssembledSegment {
                                        tag: "CCI".to_string(),
                                        elements: vec![
                                            vec!["11XAB-1234".to_string()],
                                            vec!["305".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(),
        };

        // Companion fields with dotted targets
        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.0".to_string(),
            FieldMapping::Simple("bilanzkreis.id".to_string()),
        );
        companion_fields.insert(
            "cci.1".to_string(),
            FieldMapping::Simple("bilanzkreis.codelist".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Test".to_string(),
                bo4e_type: "Test".to_string(),
                companion_type: Some("TestEdifact".to_string()),
                source_group: "SG4.SG8.SG10".to_string(),
                source_path: Some("sg4.sg8_z01.sg10".to_string()),
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

        let engine = MappingEngine::from_definitions(vec![]);
        let bo4e = engine.map_forward(&tree, &def, 0);

        // Verify nested structure under companion type key
        let companion = &bo4e["testEdifact"];
        assert!(
            companion.is_object(),
            "testEdifact should be an object, got: {companion}"
        );
        let bilanzkreis = &companion["bilanzkreis"];
        assert!(
            bilanzkreis.is_object(),
            "bilanzkreis should be a nested object, got: {bilanzkreis}"
        );
        assert_eq!(
            bilanzkreis["id"].as_str(),
            Some("11XAB-1234"),
            "bilanzkreis.id should be 11XAB-1234"
        );
        assert_eq!(
            bilanzkreis["codelist"].as_str(),
            Some("305"),
            "bilanzkreis.codelist should be 305"
        );
    }

    #[test]
    fn test_companion_dotted_path_reverse() {
        // Test that populate_field resolves dotted paths in nested JSON
        let engine = MappingEngine::from_definitions(vec![]);

        let companion_value = serde_json::json!({
            "bilanzkreis": {
                "id": "11XAB-1234",
                "codelist": "305"
            }
        });

        assert_eq!(
            engine.populate_field(&companion_value, "bilanzkreis.id"),
            Some("11XAB-1234".to_string()),
            "dotted path bilanzkreis.id should resolve"
        );
        assert_eq!(
            engine.populate_field(&companion_value, "bilanzkreis.codelist"),
            Some("305".to_string()),
            "dotted path bilanzkreis.codelist should resolve"
        );

        // Also test full reverse mapping roundtrip through map_reverse
        let mut companion_fields: IndexMap<String, FieldMapping> = IndexMap::new();
        companion_fields.insert(
            "cci.0".to_string(),
            FieldMapping::Simple("bilanzkreis.id".to_string()),
        );
        companion_fields.insert(
            "cci.1".to_string(),
            FieldMapping::Simple("bilanzkreis.codelist".to_string()),
        );

        let def = MappingDefinition {
            meta: MappingMeta {
                entity: "Test".to_string(),
                bo4e_type: "Test".to_string(),
                companion_type: Some("TestEdifact".to_string()),
                source_group: "SG4.SG8.SG10".to_string(),
                source_path: Some("sg4.sg8_z01.sg10".to_string()),
                discriminator: None,
                repeat_on_tag: None,
            },
            fields: IndexMap::new(),
            companion_fields: Some(companion_fields),
            complex_handlers: None,
        };

        let bo4e = serde_json::json!({
            "testEdifact": {
                "bilanzkreis": {
                    "id": "11XAB-1234",
                    "codelist": "305"
                }
            }
        });

        let instance = engine.map_reverse(&bo4e, &def);
        assert_eq!(instance.segments.len(), 1, "should produce one CCI segment");
        let cci = &instance.segments[0];
        assert_eq!(cci.tag, "CCI");
        assert_eq!(
            cci.elements[0],
            vec!["11XAB-1234"],
            "element 0 should contain bilanzkreis.id"
        );
        assert_eq!(
            cci.elements[1],
            vec!["305"],
            "element 1 should contain bilanzkreis.codelist"
        );
    }

    #[test]
    fn test_when_filled_injects_when_field_present() {
        let toml_str = r#"
[meta]
entity = "Test"
bo4e_type = "Test"
companion_type = "TestEdifact"
source_group = "SG4.SG8.SG10"

[fields]

[companion_fields]
"cci.0.0" = { target = "", default = "Z83", when_filled = ["merkmalCode"] }
"cav.0.0" = "merkmalCode"
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();

        // BO4E with merkmalCode present → should inject Z83
        let bo4e_with = serde_json::json!({
            "testEdifact": { "merkmalCode": "ZA7" }
        });
        let engine = MappingEngine::new_empty();
        let instance = engine.map_reverse(&bo4e_with, &def);
        let cci = instance
            .segments
            .iter()
            .find(|s| s.tag == "CCI")
            .expect("CCI should exist");
        assert_eq!(cci.elements[0][0], "Z83");

        // BO4E without merkmalCode → should NOT inject CCI
        let bo4e_without = serde_json::json!({
            "testEdifact": {}
        });
        let instance2 = engine.map_reverse(&bo4e_without, &def);
        let cci2 = instance2.segments.iter().find(|s| s.tag == "CCI");
        assert!(
            cci2.is_none(),
            "CCI should not be emitted when merkmalCode is absent"
        );
    }

    #[test]
    fn test_when_filled_checks_core_and_companion() {
        let toml_str = r#"
[meta]
entity = "Test"
bo4e_type = "Test"
companion_type = "TestEdifact"
source_group = "SG4.SG5"

[fields]
"loc.1.0" = "marktlokationsId"

[companion_fields]
"loc.0.0" = { target = "", default = "Z16", when_filled = ["marktlokationsId"] }
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();

        // Core field present → inject
        let bo4e_with = serde_json::json!({
            "marktlokationsId": "51234567890"
        });
        let engine = MappingEngine::new_empty();
        let instance = engine.map_reverse(&bo4e_with, &def);
        let loc = instance
            .segments
            .iter()
            .find(|s| s.tag == "LOC")
            .expect("LOC should exist");
        assert_eq!(loc.elements[0][0], "Z16");
        assert_eq!(loc.elements[1][0], "51234567890");

        // Core field absent → no injection
        let bo4e_without = serde_json::json!({});
        let instance2 = engine.map_reverse(&bo4e_without, &def);
        let loc2 = instance2.segments.iter().find(|s| s.tag == "LOC");
        assert!(loc2.is_none());
    }

    #[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()));
    }

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

        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,
                },
            ],
            child_groups: vec![],
            entry_mig_number: None,
            variant_mig_numbers: vec![],
            skipped_segments: vec![],
            skipped_positions: Vec::new(),
        };

        let toml_str = r#"
[meta]
entity = "Test"
bo4e_type = "Test"
companion_type = "TestEdifact"
source_group = "SG4.SG8"

[fields]

[companion_fields]
"rff[Z34,*].0.1" = "messlokationsIdRefs"
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();
        let engine = MappingEngine::new_empty();

        let mut result = serde_json::Map::new();
        engine.extract_companion_fields(&instance, &def, &mut result, false);

        let companion = result.get("testEdifact").unwrap().as_object().unwrap();
        let refs = companion
            .get("messlokationsIdRefs")
            .unwrap()
            .as_array()
            .unwrap();
        assert_eq!(refs.len(), 2);
        assert_eq!(refs[0].as_str().unwrap(), "REF_A");
        assert_eq!(refs[1].as_str().unwrap(), "REF_B");
    }

    #[test]
    fn test_reverse_json_array_produces_multiple_segments() {
        let toml_str = r#"
[meta]
entity = "Test"
bo4e_type = "Test"
companion_type = "TestEdifact"
source_group = "SG4.SG8"

[fields]

[companion_fields]
"seq.0.0" = { target = "", default = "ZD6" }
"rff[Z34,*].0.1" = "messlokationsIdRefs"
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();
        let engine = MappingEngine::new_empty();

        let bo4e = serde_json::json!({
            "testEdifact": {
                "messlokationsIdRefs": ["REF_A", "REF_B", "REF_C"]
            }
        });

        let instance = engine.map_reverse(&bo4e, &def);

        // Should have SEQ + 3 RFF segments
        let rff_segs: Vec<_> = instance
            .segments
            .iter()
            .filter(|s| s.tag == "RFF")
            .collect();
        assert_eq!(rff_segs.len(), 3);
        assert_eq!(rff_segs[0].elements[0][0], "Z34");
        assert_eq!(rff_segs[0].elements[0][1], "REF_A");
        assert_eq!(rff_segs[1].elements[0][0], "Z34");
        assert_eq!(rff_segs[1].elements[0][1], "REF_B");
        assert_eq!(rff_segs[2].elements[0][0], "Z34");
        assert_eq!(rff_segs[2].elements[0][1], "REF_C");
    }

    #[test]
    fn test_when_filled_dotted_path() {
        let toml_str = r#"
[meta]
entity = "Test"
bo4e_type = "Test"
companion_type = "TestEdifact"
source_group = "SG4.SG8.SG10"

[fields]

[companion_fields]
"cci.0.0" = { target = "", default = "Z83", when_filled = ["merkmal.code"] }
"cav.0.0" = "merkmal.code"
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();

        let bo4e = serde_json::json!({
            "testEdifact": { "merkmal": { "code": "ZA7" } }
        });
        let engine = MappingEngine::new_empty();
        let instance = engine.map_reverse(&bo4e, &def);
        let cci = instance
            .segments
            .iter()
            .find(|s| s.tag == "CCI")
            .expect("CCI should exist");
        assert_eq!(cci.elements[0][0], "Z83");
    }

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

        let instance = AssembledGroupInstance {
            segments: vec![AssembledSegment {
                tag: "NAD".to_string(),
                elements: vec![vec!["Z47".to_string()], vec!["12345".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(),
        };

        let toml_str = r#"
[meta]
entity = "Geschaeftspartner"
bo4e_type = "Geschaeftspartner"
companion_type = "GeschaeftspartnerEdifact"
source_group = "SG4.SG12"

[fields]
"nad.1.0" = "identifikation"

[companion_fields."nad.0.0"]
target = "partnerrolle"
enum_map = { "Z47" = "kundeDesLf", "Z48" = "kundeDesLf", "Z51" = "kundeDesNb", "Z52" = "kundeDesNb" }
also_target = "datenqualitaet"
also_enum_map = { "Z47" = "erwartet", "Z48" = "imSystemVorhanden", "Z51" = "erwartet", "Z52" = "imSystemVorhanden" }
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();
        let engine = MappingEngine::new_empty();

        let mut result = serde_json::Map::new();
        engine.extract_companion_fields(&instance, &def, &mut result, false);

        let companion = result
            .get("geschaeftspartnerEdifact")
            .unwrap()
            .as_object()
            .unwrap();
        assert_eq!(
            companion.get("partnerrolle").unwrap().as_str().unwrap(),
            "kundeDesLf"
        );
        assert_eq!(
            companion.get("datenqualitaet").unwrap().as_str().unwrap(),
            "erwartet"
        );
    }

    #[test]
    fn test_also_target_reverse_joint_lookup() {
        let toml_str = r#"
[meta]
entity = "Geschaeftspartner"
bo4e_type = "Geschaeftspartner"
companion_type = "GeschaeftspartnerEdifact"
source_group = "SG4.SG12"

[fields]

[companion_fields."nad.0.0"]
target = "partnerrolle"
enum_map = { "Z47" = "kundeDesLf", "Z48" = "kundeDesLf", "Z51" = "kundeDesNb", "Z52" = "kundeDesNb" }
also_target = "datenqualitaet"
also_enum_map = { "Z47" = "erwartet", "Z48" = "imSystemVorhanden", "Z51" = "erwartet", "Z52" = "imSystemVorhanden" }
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();
        let engine = MappingEngine::new_empty();

        // kundeDesLf + erwartet → Z47
        let bo4e = serde_json::json!({
            "geschaeftspartnerEdifact": {
                "partnerrolle": "kundeDesLf",
                "datenqualitaet": "erwartet"
            }
        });
        let instance = engine.map_reverse(&bo4e, &def);
        let nad = instance
            .segments
            .iter()
            .find(|s| s.tag == "NAD")
            .expect("NAD");
        assert_eq!(nad.elements[0][0], "Z47");

        // kundeDesNb + imSystemVorhanden → Z52
        let bo4e2 = serde_json::json!({
            "geschaeftspartnerEdifact": {
                "partnerrolle": "kundeDesNb",
                "datenqualitaet": "imSystemVorhanden"
            }
        });
        let instance2 = engine.map_reverse(&bo4e2, &def);
        let nad2 = instance2
            .segments
            .iter()
            .find(|s| s.tag == "NAD")
            .expect("NAD");
        assert_eq!(nad2.elements[0][0], "Z52");
    }

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

        // Mixed: Z09 (unpaired) + Z47/Z48 (paired)
        let toml_str = r#"
[meta]
entity = "Geschaeftspartner"
bo4e_type = "Geschaeftspartner"
companion_type = "GeschaeftspartnerEdifact"
source_group = "SG4.SG12"

[fields]

[companion_fields."nad.0.0"]
target = "partnerrolle"
enum_map = { "Z09" = "kundeDesLf", "Z47" = "kundeDesLf", "Z48" = "kundeDesLf" }
also_target = "datenqualitaet"
also_enum_map = { "Z47" = "erwartet", "Z48" = "imSystemVorhanden" }
"#;
        let def: MappingDefinition = toml::from_str(toml_str).unwrap();
        let engine = MappingEngine::new_empty();

        // Forward: Z09 (unpaired) → partnerrolle set, datenqualitaet NOT set
        let instance_z09 = AssembledGroupInstance {
            segments: vec![AssembledSegment {
                tag: "NAD".to_string(),
                elements: vec![vec!["Z09".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(),
        };
        let mut result = serde_json::Map::new();
        engine.extract_companion_fields(&instance_z09, &def, &mut result, false);
        let comp = result
            .get("geschaeftspartnerEdifact")
            .unwrap()
            .as_object()
            .unwrap();
        assert_eq!(
            comp.get("partnerrolle").unwrap().as_str().unwrap(),
            "kundeDesLf"
        );
        assert!(
            comp.get("datenqualitaet").is_none(),
            "Z09 should not set datenqualitaet"
        );

        // Reverse: kundeDesLf WITHOUT datenqualitaet → Z09 (not Z47/Z48)
        let bo4e = serde_json::json!({
            "geschaeftspartnerEdifact": { "partnerrolle": "kundeDesLf" }
        });
        let instance = engine.map_reverse(&bo4e, &def);
        let nad = instance
            .segments
            .iter()
            .find(|s| s.tag == "NAD")
            .expect("NAD");
        assert_eq!(nad.elements[0][0], "Z09");

        // Reverse: kundeDesLf WITH datenqualitaet=erwartet → Z47
        let bo4e2 = serde_json::json!({
            "geschaeftspartnerEdifact": {
                "partnerrolle": "kundeDesLf",
                "datenqualitaet": "erwartet"
            }
        });
        let instance2 = engine.map_reverse(&bo4e2, &def);
        let nad2 = instance2
            .segments
            .iter()
            .find(|s| s.tag == "NAD")
            .expect("NAD");
        assert_eq!(nad2.elements[0][0], "Z47");
    }

    /// Issue #61 — `unroute_lokation_to_nad_dp` rebuilds the marktteilnehmer
    /// DP entry from a side-channel metadata blob without needing a `_dpSource`
    /// stamp on the lokation entity. This is what enables tx-level NAD+DP
    /// roundtrip: the per-tx reverse path calls this helper using
    /// `MappedTransaktion::dp_routing` rather than scanning JSON for markers.
    #[test]
    fn test_unroute_uses_side_channel_metadata() {
        let mut entities = serde_json::json!({
            "marktlokation": {
                "marktlokationsId": "12345678901",
                "boTyp": "MARKTLOKATION",
                "adresse": {
                    "ort": "Testhausen",
                    "postleitzahl": "99011",
                    "land": "DE"
                }
            }
        })
        .as_object()
        .unwrap()
        .clone();

        let mut metadata = serde_json::Map::new();
        metadata.insert(
            "originalIdx".to_string(),
            serde_json::Value::Number(serde_json::Number::from(2usize)),
        );
        metadata.insert(
            "marktrolle".to_string(),
            serde_json::json!({"code": "DP"}),
        );
        metadata.insert(
            "ortQualifier".to_string(),
            serde_json::json!({"code": "172"}),
        );

        let mut dp_routing: DpRouting = std::collections::HashMap::new();
        dp_routing.insert("marktlokation".to_string(), vec![metadata]);

        unroute_lokation_to_nad_dp(&mut entities, &dp_routing);

        // marktlokation key removed; marktteilnehmer rebuilt with DP entry.
        assert!(
            !entities.contains_key("marktlokation"),
            "marktlokation should be consumed by unroute"
        );
        let mts = entities
            .get("marktteilnehmer")
            .and_then(|v| v.as_array())
            .expect("marktteilnehmer array reconstructed");
        assert_eq!(mts.len(), 1);
        let dp = &mts[0];
        assert_eq!(
            dp.get("meldepunktId").and_then(|v| v.as_str()),
            Some("12345678901")
        );
        assert_eq!(
            dp.get("ort").and_then(|v| v.as_str()),
            Some("Testhausen")
        );
        // No `_dpSource` marker should appear on the reconstructed entry.
        assert!(
            dp.as_object().is_some_and(|m| !m.contains_key("_dpSource")),
            "rebuilt DP entry must not carry _dpSource: {dp}"
        );
        // Side-channel-only fields restored from metadata.
        assert!(dp.get("marktrolle").is_some());
        assert!(dp.get("ortQualifier").is_some());
    }
}