config_disassembler/xml/handlers/

reassemble.rs

//! Reassemble XML from disassembled directory.

use crate::xml::builders::{
    merge_fragments, merge_xml_elements_owned, reorder_root_keys_owned, write_xml_file,
    ShardFragments,
};
use crate::xml::element_order::{entry_sort_key, ElementOrder};
use crate::xml::multi_level::{ensure_segment_files_structure, load_multi_level_config};
use crate::xml::parsers::parse_to_xml_object;
use crate::xml::types::{MultiLevelRule, SidecarSpec, XmlElement};
use crate::xml::utils::normalize_path_unix;
use serde_json::Value;
use std::collections::HashSet;
use std::ffi::OsString;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use tokio::fs;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

/// Upper bound on shard files being read + parsed at once across the **whole
/// process**, not per call. Callers such as sf-decomposer run dozens of
/// `reassemble` calls concurrently; a per-call bound would multiply by that
/// count and could exhaust file descriptors (Linux's default soft limit is
/// 1024). Parsing is CPU-bound, so anything well above the core count only
/// buys I/O overlap.
const SHARD_PARSE_CONCURRENCY: usize = 32;
static SHARD_PARSE_PERMITS: Semaphore = Semaphore::const_new(SHARD_PARSE_CONCURRENCY);

/// `true` for a regular file, following symlinks like the `Path::is_file` this
/// replaces. Uses the file type the directory listing already returned instead
/// of a blocking `stat` per entry on an async worker; only symlinks (and the
/// rare platform that can't report a type) pay for a real metadata call.
async fn is_regular_file(entry: &fs::DirEntry, path: &Path) -> bool {
    match entry.file_type().await {
        Ok(file_type) if !file_type.is_symlink() => file_type.is_file(),
        _ => fs::metadata(path)
            .await
            .map(|meta| meta.is_file())
            .unwrap_or(false),
    }
}

/// Read + parse every shard in `files` concurrently (bounded by
/// [`SHARD_PARSE_PERMITS`]) and return the results **in input order**, so the
/// merged output is byte-identical to parsing them one by one. `None` marks a
/// shard that failed to read or parse, exactly as the sequential path skipped it.
///
/// Each parsed shard is passed through `convert` inside its parse task, so
/// per-shard work (fragment reassembly serializes the shard and drops its tree)
/// runs concurrently too.
async fn parse_shards_in_order<T: Send + 'static>(
    files: Vec<String>,
    convert: fn(XmlElement) -> T,
) -> Result<Vec<Option<T>>, Box<dyn std::error::Error + Send + Sync>> {
    // 0 or 1 shards: nothing to overlap, skip the JoinSet setup cost.
    if files.len() <= 1 {
        let mut out = Vec::with_capacity(files.len());
        for file in files {
            out.push(parse_to_xml_object(&file).await.map(convert));
        }
        return Ok(out);
    }

    let mut results: Vec<Option<T>> = (0..files.len()).map(|_| None).collect();
    let mut join_set = JoinSet::new();
    for (index, file) in files.into_iter().enumerate() {
        join_set.spawn(async move {
            let _permit = SHARD_PARSE_PERMITS
                .acquire()
                .await
                .expect("static semaphore is never closed");
            (index, parse_to_xml_object(&file).await.map(convert))
        });
    }
    while let Some(joined) = join_set.join_next().await {
        let (index, parsed) =
            joined.map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;
        results[index] = parsed;
    }
    Ok(results)
}

/// The identity conversion for [`parse_shards_in_order`]: keep the parsed tree.
fn keep_tree(element: XmlElement) -> XmlElement {
    element
}

/// Read a `.key_order.json` file (if present) and parse it as a list of root key names.
async fn read_key_order(path: &Path) -> Option<Vec<String>> {
    let bytes = fs::read(path).await.ok()?;
    serde_json::from_slice::<Vec<String>>(&bytes).ok()
}

/// Read a `.trailing_newline.json` file (if present) and parse it as a bool. Absent --
/// e.g. a directory disassembled by a crate version predating this sidecar -- defaults
/// to `false`, preserving the pre-existing "always trim trailing whitespace" behavior.
async fn read_trailing_newline(path: &Path) -> Option<bool> {
    let bytes = fs::read(path).await.ok()?;
    String::from_utf8(bytes).ok()?.trim().parse::<bool>().ok()
}

/// Remove @xmlns from an object so the reassembled segment wrapper (e.g. programProcesses) has no xmlns.
fn strip_xmlns_from_value(v: Value) -> Value {
    match v {
        Value::Object(obj) => {
            Value::Object(obj.into_iter().filter(|(k, _)| k != "@xmlns").collect())
        }
        other => other,
    }
}

/// When recursing into a nested multi-level rule's `path_segment`, the
/// deeper-level recursion needs the *sibling* rules — every rule
/// except the one we just matched — so a sub-directory that happens
/// to share its parent's `path_segment` doesn't re-enter the same
/// rule. Returns the cloned slice with the matched segment filtered
/// out. Pure helper extracted from
/// `reassemble_multi_level_segment_inner`.
fn deeper_candidate_rules(
    all_rules: &[MultiLevelRule],
    exclude_path_segment: &str,
) -> Vec<MultiLevelRule> {
    all_rules
        .iter()
        .filter(|r| r.path_segment != exclude_path_segment)
        .cloned()
        .collect()
}

/// True when the current directory is the disassembly root for any
/// of the supplied multi-level rules. Each rule stores the base path
/// it was disassembled from; if `dir_path` matches one, the caller is
/// allowed to match that rule's child segments. Pure helper extracted
/// from `process_files_in_directory` so the `dir_path == base`
/// equality is testable without a temporary directory tree.
fn is_at_base_path(dir_path: &str, base_segments: &[(String, String, bool)]) -> bool {
    base_segments.iter().any(|(base, _, _)| dir_path == base)
}

type ProcessDirFuture<'a, T> = Pin<
    Box<dyn Future<Output = Result<Vec<T>, Box<dyn std::error::Error + Send + Sync>>> + Send + 'a>,
>;

type SegmentFuture<'a> =
    Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send + 'a>>;

pub struct ReassembleXmlFileHandler;

impl ReassembleXmlFileHandler {
    pub fn new() -> Self {
        Self
    }

    pub async fn reassemble(
        &self,
        file_path: &str,
        file_extension: Option<&str>,
        post_purge: bool,
        sidecar_specs: Option<&[SidecarSpec]>,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let file_path = normalize_path_unix(file_path);
        if !self.validate_directory(&file_path).await? {
            return Ok(());
        }

        let path = Path::new(&file_path);
        let config = load_multi_level_config(path).await;
        if let Some(ref config) = config {
            // Process each rule whose path_segment exists as a directory at the
            // disassembly root. Inner-only rules (whose segment lives nested under another
            // rule's item dir) are handled dynamically when the parent rule walks its
            // items; we hand them in as `nested_rules` candidates here.
            for (i, rule) in config.rules.iter().enumerate() {
                let segment_path = path.join(&rule.path_segment);
                if !segment_path.is_dir() {
                    continue;
                }
                let nested: Vec<MultiLevelRule> = config
                    .rules
                    .iter()
                    .enumerate()
                    .filter(|(j, _)| *j != i)
                    .map(|(_, r)| r.clone())
                    .collect();
                self.reassemble_multi_level_segment(&segment_path, rule, &nested)
                    .await?;
            }
        }

        // Build one base-segment entry per multi-level rule so the recursive walker can
        // recognize each rule's path_segment under the disassembly root.
        let base_segments: Vec<(String, String, bool)> = config
            .as_ref()
            .map(|c| {
                c.rules
                    .iter()
                    .map(|r| (file_path.clone(), r.path_segment.clone(), true))
                    .collect()
            })
            .unwrap_or_default();
        // When multi-level reassembly is done, purge the entire disassembled directory
        let post_purge_final = post_purge || config.is_some();
        self.reassemble_plain(
            &file_path,
            file_extension,
            post_purge_final,
            &base_segments,
            sidecar_specs,
        )
        .await
    }

    /// Reassemble a single multi-level segment directory.
    ///
    /// For each item directory under `segment_path` (e.g. each `<dialog>/` under
    /// `botDialogs/`):
    ///
    /// 1. **Phase 1 — nested rules first.** For every immediate sub-directory whose name
    ///    matches a `nested_rules` candidate's `path_segment`, recursively reassemble
    ///    that sub-directory as its own segment. This wraps each per-step file in
    ///    `<wrap_root_element><inner_segment>...</inner_segment></wrap_root_element>` *before*
    ///    the outer-level merge sees it, so multiple inner items survive as siblings
    ///    rather than collapsing into a single bag of leaves.
    ///
    /// 2. **Phase 2 — flat sub-directories.** Any remaining sub-directory (anything not
    ///    consumed by phase 1) is collapsed into a per-item `.xml` at the parent level
    ///    via [`Self::reassemble_plain`], the original behaviour for things like
    ///    decompose-rule outputs.
    ///
    /// 3. **Phase 3 — merge item.** Everything in the item directory (the `.xml` files
    ///    written by phases 1 and 2 plus any leaf `.xml` already there) is merged into
    ///    a single `.xml` at the parent level.
    ///
    /// Finally, [`ensure_segment_files_structure`] wraps every `.xml` in `segment_path`
    /// in `<wrap_root_element><path_segment>...</path_segment></wrap_root_element>` so
    /// the parent reassembly sees correctly-wrapped siblings.
    fn reassemble_multi_level_segment<'a>(
        &'a self,
        segment_path: &'a Path,
        rule: &'a MultiLevelRule,
        nested_rules: &'a [MultiLevelRule],
    ) -> SegmentFuture<'a> {
        let segment_path = segment_path.to_path_buf();
        let rule = rule.clone();
        let nested_rules = nested_rules.to_vec();
        Box::pin(async move {
            self.reassemble_multi_level_segment_inner(&segment_path, &rule, &nested_rules)
                .await
        })
    }

    async fn reassemble_multi_level_segment_inner(
        &self,
        segment_path: &Path,
        rule: &MultiLevelRule,
        nested_rules: &[MultiLevelRule],
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        if !segment_path.is_dir() {
            return Ok(());
        }
        let mut entries = Vec::new();
        let mut read_dir = fs::read_dir(segment_path).await?;
        while let Some(entry) = read_dir.next_entry().await? {
            entries.push(entry);
        }
        entries.sort_by_cached_key(|e| e.file_name());
        for entry in entries {
            let process_path = entry.path();
            if !process_path.is_dir() {
                continue;
            }
            let process_path_str = normalize_path_unix(&process_path.to_string_lossy());
            let mut sub_entries = Vec::new();
            let mut sub_read = fs::read_dir(&process_path).await?;
            while let Some(e) = sub_read.next_entry().await? {
                sub_entries.push(e);
            }
            sub_entries.sort_by_cached_key(|e| e.file_name());

            // Phase 1: drain any sub-directory that matches a nested rule's
            // `path_segment` so it is re-wrapped before the outer merge runs.
            let mut handled: HashSet<OsString> = HashSet::new();
            for sub_entry in &sub_entries {
                let sub_path: PathBuf = sub_entry.path();
                if !sub_path.is_dir() {
                    continue;
                }
                let sub_name = sub_path.file_name().and_then(|n| n.to_str()).unwrap_or("");
                let Some(nested_rule) = nested_rules.iter().find(|r| r.path_segment == sub_name)
                else {
                    continue;
                };
                // Pass everything *except* the rule we just matched as deeper candidates.
                // Sibling rules remain candidates further down the tree without re-entering
                // the same rule on a sub-dir that happens to share its name.
                let deeper = deeper_candidate_rules(nested_rules, &nested_rule.path_segment);
                self.reassemble_multi_level_segment(&sub_path, nested_rule, &deeper)
                    .await?;
                handled.insert(sub_entry.file_name());
            }

            // Phase 2: collapse remaining sub-directories into per-item .xml files at
            // the parent level (preserves existing behaviour for non-nested-rule subdirs).
            for sub_entry in &sub_entries {
                let sub_path = sub_entry.path();
                if !sub_path.is_dir() {
                    continue;
                }
                if handled.contains(&sub_entry.file_name()) {
                    continue;
                }
                let sub_path_str = normalize_path_unix(&sub_path.to_string_lossy());
                self.reassemble_plain(&sub_path_str, Some("xml"), true, &[], None)
                    .await?;
            }

            // Phase 3: merge everything in the item dir into a single .xml at the parent.
            self.reassemble_plain(&process_path_str, Some("xml"), true, &[], None)
                .await?;
        }
        ensure_segment_files_structure(
            segment_path,
            &rule.wrap_root_element,
            &rule.path_segment,
            &rule.wrap_xmlns,
        )
        .await?;
        Ok(())
    }

    /// Merge and write reassembled XML (no multi-level pre-step). Used internally.
    /// `base_segments` carries one tuple `(base_path, segment_name, extract_inner)` per
    /// multi-level rule. When the recursive walker reaches `base_path` and finds a subdir
    /// whose name matches one of the segment_names, that subdir's XML files are folded
    /// into a single array under the segment_name key. When extract_inner is true, each
    /// file's structure is `document_root > segment_name > content` and only the content
    /// is collected; otherwise the whole root is kept.
    async fn reassemble_plain(
        &self,
        file_path: &str,
        file_extension: Option<&str>,
        post_purge: bool,
        base_segments: &[(String, String, bool)],
        sidecar_specs: Option<&[SidecarSpec]>,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let file_path = normalize_path_unix(file_path);
        log::debug!("Parsing directory to reassemble: {}", file_path);
        // Each shard is serialized in its parse task and its parsed tree dropped
        // (see builders/fragments.rs), so the merged document is never held in
        // memory: only every shard's serialized bytes plus a small skeleton.
        // Restore the recorded order of repeated siblings (see element_order.rs);
        // absent for directories written before it existed, which keep filename order.
        let element_order = ElementOrder::load(&file_path).await.map(Arc::new);
        let mut shards = self
            .process_files_in_directory(
                file_path.to_string(),
                base_segments.to_vec(),
                ShardFragments::from_element,
                element_order,
            )
            .await?;

        if shards.is_empty() {
            log_nothing_parsed(&file_path);
            return Ok(());
        }

        let Some(mut merged) = merge_fragments(&mut shards) else {
            log_no_root(&file_path);
            return Ok(());
        };

        // A compact root can't be written from serialized children; rebuild it
        // from the parsed tree instead (rare: shards written by disassembly are
        // indented, so their roots are never compact).
        if merged.needs_tree_path() {
            drop(merged);
            drop(shards);
            return self
                .reassemble_plain_tree(
                    &file_path,
                    file_extension,
                    post_purge,
                    base_segments,
                    sidecar_specs,
                )
                .await;
        }

        let effective_specs = resolve_sidecar_specs(&file_path, sidecar_specs).await;
        // Inject sidecar element content before key reordering so the element
        // lands at its original position rather than being appended at the end.
        if let Some(specs) = effective_specs.as_deref() {
            for (element, value) in load_sidecar_elements(&file_path, specs).await {
                merged.insert(element, value);
            }
        }

        // Apply stored key order so reassembled XML matches original document order.
        let key_order_path = Path::new(&file_path).join(".key_order.json");
        if let Some(order) = read_key_order(&key_order_path).await {
            merged.reorder(&order);
        }

        // Reproduce the original file's trailing newline (or lack of one) instead of
        // always trimming it away -- see `.trailing_newline.json` in build_disassembled_files.rs.
        let trailing_newline_path = Path::new(&file_path).join(".trailing_newline.json");
        let trailing_newline = read_trailing_newline(&trailing_newline_path)
            .await
            .unwrap_or(false);

        let output_path = self.get_output_path(&file_path, file_extension);
        // Inline rather than on `spawn_blocking`; see the tree path below.
        merged.write_file(&shards, Path::new(&output_path), trailing_newline)?;
        drop(merged);
        drop(shards);

        if post_purge {
            purge_disassembled(&file_path, effective_specs.as_deref()).await;
        }

        Ok(())
    }

    /// Reassemble from the parsed shards' merged tree: the path before fragment
    /// reassembly, kept for documents [`reassemble_plain`] can't write from
    /// fragments (a compact root).
    async fn reassemble_plain_tree(
        &self,
        file_path: &str,
        file_extension: Option<&str>,
        post_purge: bool,
        base_segments: &[(String, String, bool)],
        sidecar_specs: Option<&[SidecarSpec]>,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let element_order = ElementOrder::load(file_path).await.map(Arc::new);
        let parsed_objects = self
            .process_files_in_directory(
                file_path.to_string(),
                base_segments.to_vec(),
                keep_tree,
                element_order,
            )
            .await?;

        if parsed_objects.is_empty() {
            log_nothing_parsed(file_path);
            return Ok(());
        }

        // merge_xml_elements only returns None when every parsed element is empty or
        // declaration-only (no usable root). Treat that the same as "nothing parsed"
        // rather than emitting an `<root></root>` stub.
        // Moves the parsed shards into the merged tree (no second copy of the document).
        let Some(mut merged) = merge_xml_elements_owned(parsed_objects) else {
            log_no_root(file_path);
            return Ok(());
        };

        let effective_specs = resolve_sidecar_specs(file_path, sidecar_specs).await;
        if let Some(specs) = effective_specs.as_deref() {
            inject_sidecar_elements(file_path, &mut merged, specs).await?;
        }

        let key_order_path = Path::new(file_path).join(".key_order.json");
        if let Some(order) = read_key_order(&key_order_path).await {
            merged = reorder_root_keys_owned(merged, &order);
        }

        let trailing_newline_path = Path::new(file_path).join(".trailing_newline.json");
        let trailing_newline = read_trailing_newline(&trailing_newline_path)
            .await
            .unwrap_or(false);

        let output_path = self.get_output_path(file_path, file_extension);

        // Serialize straight into the output file rather than building the whole
        // document as a String first. This runs inline, on the current worker, like
        // the String-building serializer it replaces. It deliberately does not use
        // `spawn_blocking`: with many small documents reassembled concurrently, the
        // extra blocking-pool threads (each with its own allocator arena) cost more
        // memory than the output buffer saves -- measured as a consistent ~10% max-RSS
        // regression on sf-decomposer's `manyfiles` workload.
        write_xml_file(&merged, Path::new(&output_path), trailing_newline)?;
        drop(merged);

        if post_purge {
            purge_disassembled(file_path, effective_specs.as_deref()).await;
        }

        Ok(())
    }

    /// Parse every shard under `dir_path` (recursively, collecting multi-level
    /// segments) and pass each through `convert`, in merge order.
    fn process_files_in_directory<'a, T: Send + 'static>(
        &'a self,
        dir_path: String,
        base_segments: Vec<(String, String, bool)>,
        convert: fn(XmlElement) -> T,
        element_order: Option<Arc<ElementOrder>>,
    ) -> ProcessDirFuture<'a, T> {
        Box::pin(async move {
            let mut parsed = Vec::new();
            let mut entries = Vec::new();
            let mut read_dir = fs::read_dir(&dir_path).await?;
            while let Some(entry) = read_dir.next_entry().await? {
                entries.push(entry);
            }
            // Sort by full filename for deterministic cross-platform ordering. The key
            // is computed once per entry: building it inside a comparator cost four
            // allocations per comparison (`file_name()` allocates, then the lossy
            // copy), which heap-profiling showed as ~27% of all of reassemble's
            // allocations on a directory of ~136k shards.
            // Recorded siblings first in their original order, then the rest by filename
            // (see element_order.rs). Without an order file this is plain filename order.
            entries.sort_by_cached_key(|e| {
                entry_sort_key(
                    element_order.as_deref(),
                    &dir_path,
                    e.file_name().to_string_lossy().into_owned(),
                )
            });

            // We are at the disassembly root for a given rule when our dir_path matches
            // the base_path stored on that rule. Each rule shares the same base_path in
            // the current implementation, but tracking them per-entry keeps the door open
            // for future per-rule base_paths without another signature change.
            let is_base = is_at_base_path(&dir_path, &base_segments);

            // Pass 1: classify entries in sorted order. Shard files are queued for the
            // concurrent parse below; each slot remembers its position so `parsed` ends
            // up in exactly the order the old one-file-at-a-time loop produced.
            enum Slot {
                Shard(usize),
                Dir(PathBuf, String),
            }
            let mut slots = Vec::with_capacity(entries.len());
            let mut shard_files = Vec::new();
            for entry in entries {
                let path = entry.path();
                let file_path = normalize_path_unix(&path.to_string_lossy()).to_string();
                if is_regular_file(&entry, &path).await {
                    let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
                    if !name.starts_with('.') && self.is_parsable_file(name) {
                        slots.push(Slot::Shard(shard_files.len()));
                        shard_files.push(file_path);
                    }
                } else {
                    slots.push(Slot::Dir(path, file_path));
                }
            }

            // Pass 2: read + parse every shard in this directory concurrently. This was a
            // sequential `.await` per file, which left one large component (e.g. a
            // permission set with thousands of shards) on a single core for its whole
            // reassemble.
            let mut shard_results = parse_shards_in_order(shard_files, convert).await?;

            // Pass 3: assemble in the original order; subdirectories recurse as before.
            for slot in slots {
                let (path, file_path) = match slot {
                    Slot::Shard(index) => {
                        if let Some(parsed_obj) = shard_results[index].take() {
                            parsed.push(parsed_obj);
                        }
                        continue;
                    }
                    Slot::Dir(path, file_path) => (path, file_path),
                };
                // Anything not a regular file is treated as a directory; symlinks and
                // other exotic entries simply recurse via read_dir below.
                let dir_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
                let matched_segment = if is_base {
                    base_segments
                        .iter()
                        .find(|(_, seg_name, _)| seg_name == dir_name)
                        .cloned()
                } else {
                    None
                };
                if let Some((_, segment_name, extract_inner)) = matched_segment {
                    let segment_element = self
                        .collect_segment_as_array(
                            &file_path,
                            &segment_name,
                            extract_inner,
                            element_order.as_deref(),
                        )
                        .await?;
                    if let Some(el) = segment_element {
                        parsed.push(convert(el));
                    }
                } else {
                    let sub_parsed = self
                        .process_files_in_directory(
                            file_path,
                            base_segments.clone(),
                            convert,
                            element_order.clone(),
                        )
                        .await?;
                    parsed.extend(sub_parsed);
                }
            }

            Ok(parsed)
        })
    }

    /// Collect all .xml files in a directory, parse each, and build one element with
    /// root_key and single key segment_name whose value is array of each file's content.
    /// When extract_inner is true, each file has root > segment_name > content; we push that content.
    async fn collect_segment_as_array(
        &self,
        segment_dir: &str,
        segment_name: &str,
        extract_inner: bool,
        element_order: Option<&ElementOrder>,
    ) -> Result<Option<XmlElement>, Box<dyn std::error::Error + Send + Sync>> {
        let mut xml_files = Vec::new();
        let mut read_dir = fs::read_dir(segment_dir).await?;
        while let Some(entry) = read_dir.next_entry().await? {
            let path = entry.path();
            let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
            if !name.starts_with('.')
                && self.is_parsable_file(name)
                && is_regular_file(&entry, &path).await
            {
                xml_files.push(normalize_path_unix(&path.to_string_lossy()).to_string());
            }
        }
        // Items in their recorded order when known (multi-level items are reassembled
        // under a new file name, which element_order::item_key still matches).
        let segment_dir_norm = normalize_path_unix(segment_dir);
        xml_files.sort_by_cached_key(|f| {
            let name = f.rsplit('/').next().unwrap_or(f).to_string();
            entry_sort_key(element_order, &segment_dir_norm, name)
        });

        let mut root_contents = Vec::new();
        let mut first_xml: Option<(String, Option<Value>)> = None;
        // Parsed concurrently, consumed in sorted order (see parse_shards_in_order).
        for parsed in parse_shards_in_order(xml_files, keep_tree).await? {
            // parse_to_xml_object always yields a JSON object on success; treat any other
            // shape (including parse failure) as a skip without branching explicitly.
            let Some(Value::Object(mut obj)) = parsed else {
                continue;
            };
            let Some(root_key) = obj.keys().find(|k| *k != "?xml").cloned() else {
                continue;
            };
            // Each file's content is moved out of its parsed tree: this used to clone
            // the whole document, then its root, then (with extract_inner) the inner
            // segment -- up to three copies of every segment file.
            let root_val = obj
                .get_mut(&root_key)
                .map(std::mem::take)
                .unwrap_or_else(|| Value::Object(serde_json::Map::new()));
            let mut content = if extract_inner {
                match root_val {
                    Value::Object(mut root) => root.get_mut(segment_name).map(std::mem::take),
                    _ => None,
                }
                .unwrap_or_else(|| Value::Object(serde_json::Map::new()))
            } else {
                root_val
            };
            // Inner segment element (e.g. programProcesses) should not have xmlns in output
            if extract_inner {
                content = strip_xmlns_from_value(content);
            }
            root_contents.push(content);
            if first_xml.is_none() {
                first_xml = Some((root_key, obj.get_mut("?xml").map(std::mem::take)));
            }
        }
        if root_contents.is_empty() {
            return Ok(None);
        }
        let (root_key, decl_opt) = first_xml.unwrap();
        let mut content = serde_json::Map::new();
        content.insert(segment_name.to_string(), Value::Array(root_contents));
        let mut top = serde_json::Map::new();
        if let Some(decl) = decl_opt {
            top.insert("?xml".to_string(), decl);
        } else {
            let mut d = serde_json::Map::new();
            d.insert("@version".to_string(), Value::String("1.0".to_string()));
            d.insert("@encoding".to_string(), Value::String("UTF-8".to_string()));
            top.insert("?xml".to_string(), Value::Object(d));
        }
        top.insert(root_key, Value::Object(content));
        Ok(Some(Value::Object(top)))
    }

    fn is_parsable_file(&self, file_name: &str) -> bool {
        let lower = file_name.to_lowercase();
        lower.ends_with(".xml")
            || lower.ends_with(".json")
            || lower.ends_with(".json5")
            || lower.ends_with(".yaml")
            || lower.ends_with(".yml")
    }

    async fn validate_directory(
        &self,
        path: &str,
    ) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
        let meta = fs::metadata(path).await?;
        if !meta.is_dir() {
            log::error!(
                "The provided path to reassemble is not a directory: {}",
                path
            );
            return Ok(false);
        }
        Ok(true)
    }

    fn get_output_path(&self, dir_path: &str, extension: Option<&str>) -> String {
        let path = Path::new(dir_path);
        let parent = path.parent().unwrap_or(Path::new("."));
        let base_name = path
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("output");
        let ext = extension.unwrap_or("xml");
        parent
            .join(format!("{}.{}", base_name, ext))
            .to_string_lossy()
            .to_string()
    }
}

impl Default for ReassembleXmlFileHandler {
    fn default() -> Self {
        Self::new()
    }
}

/// Convert `content` (in whatever format the sidecar file uses) to `target_format`.
///
/// - `"json"` → parse as YAML (superset of JSON) then emit as pretty JSON
/// - `"yaml"` / `"yml"` → only convert when content is strict JSON; YAML passes through
/// - anything else → pass through unchanged
///
/// Falls back to raw content with a warning when conversion fails.
fn convert_to_format(content: &str, target_format: &str) -> String {
    match target_format.to_ascii_lowercase().as_str() {
        "json" => {
            match serde_saphyr::from_str::<serde_json::Value>(content) {
                // Serializing a serde_json::Value cannot fail.
                Ok(val) => {
                    serde_json::to_string_pretty(&val).unwrap_or_else(|_| content.to_string())
                }
                Err(e) => {
                    log::warn!("sidecar reassemble: could not parse content for JSON conversion ({e}); using raw content");
                    content.to_string()
                }
            }
        }
        "yaml" | "yml" => {
            if serde_json::from_str::<serde_json::Value>(content).is_ok() {
                match serde_json::from_str::<serde_json::Value>(content)
                    .ok()
                    .and_then(|v| serde_saphyr::to_string(&v).ok())
                {
                    Some(yaml) => yaml,
                    None => {
                        log::warn!(
                            "sidecar reassemble: YAML serialization failed; using raw content"
                        );
                        content.to_string()
                    }
                }
            } else {
                content.to_string()
            }
        }
        _ => content.to_string(),
    }
}

fn log_nothing_parsed(file_path: &str) {
    log::error!(
        "No files under {} were parsed successfully. A reassembled XML file was not created.",
        file_path
    );
}

fn log_no_root(file_path: &str) {
    log::error!(
        "No usable root element found while merging files under {}. A reassembled XML file was not created.",
        file_path
    );
}

/// Resolve sidecar specs: the caller-supplied slice when non-empty, otherwise
/// auto-detected from `.sidecars.json` written by disassembly.
async fn resolve_sidecar_specs(
    dir_path: &str,
    sidecar_specs: Option<&[SidecarSpec]>,
) -> Option<Vec<SidecarSpec>> {
    if let Some(specs) = sidecar_specs.filter(|s| !s.is_empty()) {
        return Some(specs.to_vec());
    }
    let meta_path = Path::new(dir_path).join(".sidecars.json");
    let content = fs::read_to_string(&meta_path).await.ok()?;
    serde_json::from_str::<Vec<SidecarSpec>>(&content).ok()
}

/// Remove the disassembled directory (and, first, its sidecar files) after a
/// successful reassemble. They are decomposition artifacts: the content is now
/// inside the reassembled XML.
async fn purge_disassembled(dir_path: &str, specs: Option<&[SidecarSpec]>) {
    if let Some(specs) = specs {
        let path = Path::new(dir_path);
        let base = path
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("output");
        for spec in specs {
            let sidecar = path.join(format!("{}.{}", base, spec.extension));
            fs::remove_file(&sidecar).await.ok();
        }
    }
    fs::remove_dir_all(dir_path).await.ok();
}

/// Read each spec's sidecar file and return the root elements to set, in spec
/// order: `(element name, {"#raw-text": content})`.
///
/// Each sidecar is located inside the decomposed directory itself
/// (e.g. `MySvc/MySvc.yaml`). If a sidecar is absent (element was not in
/// the original XML, or was never extracted), the spec is silently skipped.
///
/// The value uses the `{"#raw-text": content}` shape that `build_xml_string`
/// writes with `partial_escape`: mandatory XML chars (`<`, `>`, `&`) are escaped
/// but `"` is left as-is so YAML content round-trips without `&quot;` inflation.
async fn load_sidecar_elements(dir_path: &str, specs: &[SidecarSpec]) -> Vec<(String, Value)> {
    let path = Path::new(dir_path);
    let base = path
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("output");
    let mut elements = Vec::new();
    for spec in specs {
        let sidecar_path = path.join(format!("{}.{}", base, spec.extension));
        let Ok(content) = fs::read_to_string(&sidecar_path).await else {
            continue;
        };
        // When the original XML embedded a different format than the sidecar
        // extension (e.g. JSON schema extracted to a .yaml sidecar), convert
        // the sidecar content back to the original format before injecting.
        let final_content = match &spec.original_format {
            Some(fmt) => convert_to_format(&content, fmt),
            None => content,
        };
        elements.push((
            spec.element.clone(),
            serde_json::json!({ "#raw-text": final_content }),
        ));
    }
    elements
}

/// Read sidecar files and inject their content back into the merged XML value
/// before serialisation (tree path; see [`load_sidecar_elements`]).
async fn inject_sidecar_elements(
    dir_path: &str,
    merged: &mut XmlElement,
    specs: &[SidecarSpec],
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let root_key = merged
        .as_object()
        .and_then(|o| o.keys().find(|k| *k != "?xml").cloned());
    let Some(root_key) = root_key else {
        return Ok(());
    };

    if let Some(root_obj) = merged
        .as_object_mut()
        .and_then(|o| o.get_mut(&root_key))
        .and_then(Value::as_object_mut)
    {
        for (element, value) in load_sidecar_elements(dir_path, specs).await {
            root_obj.insert(element, value);
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    #[allow(clippy::default_constructed_unit_structs)]
    fn reassemble_handler_default_equals_new() {
        let _ = ReassembleXmlFileHandler::default();
    }

    #[test]
    fn strip_xmlns_from_value_passes_non_object_through() {
        let s = Value::String("hello".to_string());
        assert_eq!(
            strip_xmlns_from_value(s),
            Value::String("hello".to_string())
        );
        let arr = json!([1, 2]);
        assert_eq!(strip_xmlns_from_value(arr.clone()), arr);
    }

    #[test]
    fn strip_xmlns_from_value_removes_xmlns_key() {
        let obj = json!({ "@xmlns": "ns", "child": 1 });
        let stripped = strip_xmlns_from_value(obj);
        let map = stripped.as_object().unwrap();
        assert!(map.get("@xmlns").is_none());
        assert_eq!(map.get("child").and_then(|v| v.as_i64()), Some(1));
    }

    #[test]
    fn is_parsable_file_recognises_supported_extensions() {
        let h = ReassembleXmlFileHandler::new();
        assert!(h.is_parsable_file("a.xml"));
        assert!(h.is_parsable_file("a.json"));
        assert!(h.is_parsable_file("a.json5"));
        assert!(h.is_parsable_file("a.yaml"));
        assert!(h.is_parsable_file("a.yml"));
        assert!(h.is_parsable_file("A.XML"));
        assert!(!h.is_parsable_file("a.txt"));
    }

    #[test]
    fn get_output_path_appends_extension_and_uses_parent_dir() {
        let h = ReassembleXmlFileHandler::new();
        let out = h.get_output_path("/tmp/foo", Some("xml"));
        assert!(out.ends_with("foo.xml"));
        let out_default = h.get_output_path("/tmp/bar", None);
        assert!(out_default.ends_with("bar.xml"));
        // No parent - uses "." fallback
        assert_eq!(h.get_output_path("only", Some("json")), "only.json");
    }

    #[tokio::test]
    async fn reassemble_multi_level_segment_noop_when_not_dir() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let file = tmp.path().join("not_a_dir.txt");
        tokio::fs::write(&file, "hi").await.unwrap();
        let rule = crate::xml::types::MultiLevelRule {
            file_pattern: String::new(),
            root_to_strip: String::new(),
            unique_id_elements: String::new(),
            path_segment: String::new(),
            wrap_root_element: "Root".to_string(),
            wrap_xmlns: String::new(),
        };
        h.reassemble_multi_level_segment(&file, &rule, &[])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn reassemble_multi_level_segment_skips_files_in_segment_root() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let segment = tmp.path().join("segment");
        tokio::fs::create_dir(&segment).await.unwrap();
        // A bare file inside the segment dir should be skipped (not a subdir).
        tokio::fs::write(segment.join("stray.txt"), "x")
            .await
            .unwrap();
        let rule = crate::xml::types::MultiLevelRule {
            file_pattern: String::new(),
            root_to_strip: String::new(),
            unique_id_elements: String::new(),
            path_segment: "segment".to_string(),
            wrap_root_element: "Root".to_string(),
            wrap_xmlns: "http://example.com".to_string(),
        };
        h.reassemble_multi_level_segment(&segment, &rule, &[])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn collect_segment_as_array_returns_none_for_empty_dir() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", true, None)
            .await
            .unwrap();
        assert!(out.is_none());
    }

    #[tokio::test]
    async fn collect_segment_as_array_skips_unparseable_and_empty_roots() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        // Unparseable XML
        tokio::fs::write(tmp.path().join("bad.xml"), "<<")
            .await
            .unwrap();
        // Valid XML but only declaration and no root after parse
        tokio::fs::write(tmp.path().join("only-decl.xml"), "")
            .await
            .unwrap();
        // Hidden file is skipped
        tokio::fs::write(tmp.path().join(".hidden.xml"), "<r/>")
            .await
            .unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", false, None)
            .await
            .unwrap();
        assert!(out.is_none());
    }

    #[tokio::test]
    async fn collect_segment_as_array_without_xml_decl_inserts_default_decl() {
        // XML files without a `<?xml ?>` declaration: the `else` branch in
        // collect_segment_as_array must synthesize a default declaration with
        // version="1.0" and encoding="UTF-8".
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        tokio::fs::write(
            tmp.path().join("a.xml"),
            r#"<Root><seg><x>1</x></seg></Root>"#,
        )
        .await
        .unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", true, None)
            .await
            .unwrap()
            .unwrap();
        let obj = out.as_object().unwrap();
        let decl = obj
            .get("?xml")
            .and_then(|v| v.as_object())
            .expect("default declaration must be inserted when XML has none");
        assert_eq!(decl.get("@version").and_then(|v| v.as_str()), Some("1.0"));
        assert_eq!(
            decl.get("@encoding").and_then(|v| v.as_str()),
            Some("UTF-8")
        );
    }

    #[tokio::test]
    async fn collect_segment_as_array_with_xml_decl_preserves_it() {
        // XML file WITH a `<?xml ?>` declaration: `decl_opt` is Some → line 438
        // `top.insert("?xml", decl)` is executed (the `if let Some(decl)` branch).
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        tokio::fs::write(
            tmp.path().join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><seg><x>1</x></seg></Root>"#,
        )
        .await
        .unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", true, None)
            .await
            .unwrap()
            .unwrap();
        let obj = out.as_object().unwrap();
        let decl = obj
            .get("?xml")
            .and_then(|v| v.as_object())
            .expect("?xml declaration must be preserved from source");
        assert_eq!(decl.get("@version").and_then(|v| v.as_str()), Some("1.0"));
        assert_eq!(
            decl.get("@encoding").and_then(|v| v.as_str()),
            Some("UTF-8")
        );
    }

    #[tokio::test]
    async fn collect_segment_as_array_extract_inner_moves_segment_content() {
        // Two files: one with the segment (its xmlns stripped), one whose root is
        // plain text (no segment to extract -> an empty object).
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        tokio::fs::write(
            tmp.path().join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><seg xmlns="ns"><x>1</x></seg><other>2</other></Root>"#,
        )
        .await
        .unwrap();
        tokio::fs::write(tmp.path().join("b.xml"), r#"<Root>text</Root>"#)
            .await
            .unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", true, None)
            .await
            .unwrap()
            .unwrap();
        let items = out["Root"]["seg"].as_array().unwrap();
        assert_eq!(items.len(), 2);
        assert_eq!(items[0]["x"]["#text"], "1");
        assert!(items[0].get("@xmlns").is_none(), "xmlns must be stripped");
        assert!(items[0].get("other").is_none(), "only the segment is taken");
        assert_eq!(items[1], json!({}));
        assert_eq!(out["?xml"]["@version"], "1.0");
    }

    #[tokio::test]
    async fn collect_segment_as_array_without_extract_inner_wraps_root() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        tokio::fs::write(tmp.path().join("a.xml"), r#"<Root><child>1</child></Root>"#)
            .await
            .unwrap();
        let out = h
            .collect_segment_as_array(tmp.path().to_str().unwrap(), "seg", false, None)
            .await
            .unwrap()
            .unwrap();
        let obj = out.as_object().unwrap();
        assert!(obj.contains_key("?xml"));
        let root = obj.get("Root").and_then(|r| r.as_object()).unwrap();
        assert!(root.get("seg").and_then(|v| v.as_array()).is_some());
    }

    fn rule_with_segment(segment: &str) -> MultiLevelRule {
        MultiLevelRule {
            file_pattern: String::new(),
            root_to_strip: String::new(),
            unique_id_elements: String::new(),
            path_segment: segment.to_string(),
            wrap_root_element: String::new(),
            wrap_xmlns: String::new(),
        }
    }

    #[test]
    fn deeper_candidate_rules_excludes_the_matched_segment() {
        // The matched rule must be filtered out, otherwise the
        // recursion would re-enter that rule when a child directory
        // happens to share its `path_segment`.
        let rules = vec![rule_with_segment("seg_a"), rule_with_segment("seg_b")];
        let deeper = deeper_candidate_rules(&rules, "seg_a");
        assert_eq!(deeper.len(), 1);
        assert_eq!(deeper[0].path_segment, "seg_b");
    }

    #[test]
    fn deeper_candidate_rules_keeps_all_when_no_segment_matches() {
        // When `exclude_path_segment` doesn't correspond to any rule
        // the input is forwarded unchanged. Pins the `!= -> ==` mutant
        // which would otherwise return an empty vec here.
        let rules = vec![rule_with_segment("seg_a"), rule_with_segment("seg_b")];
        let deeper = deeper_candidate_rules(&rules, "missing");
        assert_eq!(deeper.len(), 2);
    }

    #[test]
    fn deeper_candidate_rules_returns_empty_for_empty_input() {
        let deeper: Vec<MultiLevelRule> = deeper_candidate_rules(&[], "anything");
        assert!(deeper.is_empty());
    }

    #[test]
    fn is_at_base_path_true_when_dir_matches_any_segment() {
        let segs = vec![
            ("/base/other".to_string(), "seg1".to_string(), false),
            ("/base/here".to_string(), "seg2".to_string(), false),
        ];
        assert!(is_at_base_path("/base/here", &segs));
    }

    #[test]
    fn is_at_base_path_false_when_dir_matches_nothing() {
        let segs = vec![("/base/a".to_string(), "seg".to_string(), false)];
        assert!(!is_at_base_path("/base/b", &segs));
    }

    #[test]
    fn is_at_base_path_false_for_empty_segments() {
        let segs: Vec<(String, String, bool)> = Vec::new();
        assert!(!is_at_base_path("/anywhere", &segs));
    }

    /// Reassemble `files` (name, content) in a fresh `Doc/` directory with both the
    /// fragment path (`reassemble`) and the tree path, and return both outputs.
    async fn reassemble_both_ways(files: &[(&str, &str)]) -> (String, String) {
        let mut outputs = Vec::new();
        for tree in [false, true] {
            let dir = tempfile::tempdir().unwrap();
            let doc = dir.path().join("Doc");
            std::fs::create_dir(&doc).unwrap();
            for (name, content) in files {
                std::fs::write(doc.join(name), content).unwrap();
            }
            let doc = normalize_path_unix(&doc.to_string_lossy());
            let handler = ReassembleXmlFileHandler::new();
            if tree {
                handler
                    .reassemble_plain_tree(&doc, Some("xml"), false, &[], None)
                    .await
                    .unwrap();
            } else {
                handler
                    .reassemble(&doc, Some("xml"), false, None)
                    .await
                    .unwrap();
            }
            outputs.push(std::fs::read_to_string(dir.path().join("Doc.xml")).unwrap());
        }
        let tree = outputs.pop().unwrap();
        (outputs.pop().unwrap(), tree)
    }

    #[tokio::test]
    async fn compact_root_falls_back_to_tree_path() {
        // No whitespace anywhere: the parser marks <Root> compact (one object child),
        // which only the tree path can write inline.
        let shard = r#"<?xml version="1.0" encoding="UTF-8"?><Root><only><p>1</p></only></Root>"#;
        let (fragments, tree) = reassemble_both_ways(&[("a.xml", shard)]).await;
        assert_eq!(fragments, tree);
        assert!(fragments.contains("<Root><only>"), "{fragments}");
    }

    #[tokio::test]
    async fn fragment_path_matches_tree_path_with_sidecar_key_order_and_newline() {
        let files = [
            (
                "a.xml",
                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<Root xmlns=\"ns\">\n    <label>A</label>\n    <item>\n        <name>x</name>\n    </item>\n</Root>",
            ),
            (
                "b.xml",
                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<Root xmlns=\"ns\">\n    <item>\n        <name>y &amp; z</name>\n    </item>\n    <flag>true</flag>\n</Root>",
            ),
            (
                ".key_order.json",
                r#"["flag", "notes", "item", "label"]"#,
            ),
            (".trailing_newline.json", "true"),
            (
                ".sidecars.json",
                r#"[{"element": "notes", "extension": "yaml"}]"#,
            ),
            ("Doc.yaml", "key: \"value\" & more\n"),
        ];
        let (fragments, tree) = reassemble_both_ways(&files).await;
        assert_eq!(fragments, tree);
        assert!(fragments.ends_with("</Root>\n"), "{fragments}");
        let (flag, notes, item, label) = (
            fragments.find("<flag>"),
            fragments.find("<notes>"),
            fragments.find("<item>"),
            fragments.find("<label>"),
        );
        assert!(flag < notes && notes < item && item < label, "{fragments}");
    }

    #[tokio::test]
    async fn parse_shards_in_order_keeps_input_order_and_marks_failures() {
        let tmp = tempfile::tempdir().unwrap();
        let mut files = Vec::new();
        for i in 0..100 {
            // Every 10th entry points at a file that doesn't exist -> None, in place.
            let path = tmp.path().join(format!("{i:03}.xml"));
            if i % 10 != 0 {
                tokio::fs::write(&path, format!("<Root><n>{i}</n></Root>"))
                    .await
                    .unwrap();
            }
            files.push(path.to_string_lossy().to_string());
        }

        let results = parse_shards_in_order(files, keep_tree).await.unwrap();

        assert_eq!(results.len(), 100);
        for (i, result) in results.iter().enumerate() {
            if i % 10 == 0 {
                assert!(result.is_none(), "missing shard {i} should be None");
            } else {
                let n = result.as_ref().unwrap()["Root"]["n"]["#text"]
                    .as_str()
                    .unwrap();
                assert_eq!(n, i.to_string(), "shard {i} out of order");
            }
        }
    }

    // `is_regular_file` must follow symlinks like the `Path::is_file` it replaced:
    // a symlinked shard is parsed as a shard (not recursed into as a directory,
    // which would fail), and a symlinked directory is still recursed into.
    // Unix-only: creating symlinks on Windows needs elevated privileges.
    #[cfg(unix)]
    #[tokio::test]
    async fn reassemble_follows_symlinked_shards_and_directories() {
        use std::os::unix::fs::symlink;

        let tmp = tempfile::tempdir().unwrap();
        let outside = tmp.path().join("outside");
        tokio::fs::create_dir_all(outside.join("nested"))
            .await
            .unwrap();
        tokio::fs::write(outside.join("b.xml"), "<Root><n>b</n></Root>")
            .await
            .unwrap();
        tokio::fs::write(
            outside.join("nested").join("c.xml"),
            "<Root><n>c</n></Root>",
        )
        .await
        .unwrap();

        let dir = tmp.path().join("Bench");
        tokio::fs::create_dir(&dir).await.unwrap();
        tokio::fs::write(dir.join("a.xml"), "<Root><n>a</n></Root>")
            .await
            .unwrap();
        symlink(outside.join("b.xml"), dir.join("b.xml")).unwrap();
        symlink(outside.join("nested"), dir.join("c_dir")).unwrap();

        ReassembleXmlFileHandler::new()
            .reassemble(dir.to_str().unwrap(), Some("xml"), false, None)
            .await
            .unwrap();

        let output = tokio::fs::read_to_string(tmp.path().join("Bench.xml"))
            .await
            .unwrap();
        let a = output.find("<n>a</n>").expect("regular shard a");
        let b = output.find("<n>b</n>").expect("symlinked shard b");
        let c = output
            .find("<n>c</n>")
            .expect("shard c under symlinked dir");
        assert!(a < b && b < c, "sorted order a, b, c_dir/: {output}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn reassemble_many_shards_preserves_sorted_order() {
        // Well above SHARD_PARSE_CONCURRENCY, on a multi-thread runtime, so shards
        // finish parsing out of order; the merged output must still follow the
        // sorted file names.
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("Bench");
        tokio::fs::create_dir(&dir).await.unwrap();
        for i in 0..200 {
            // Uneven shard sizes make completion order differ from input order.
            let padding = "x".repeat((i * 7919) % 4096);
            tokio::fs::write(
                dir.join(format!("{i:04}.field-meta.xml")),
                format!(
                    "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<Root><fields><n>{i}</n><pad>{padding}</pad></fields></Root>"
                ),
            )
            .await
            .unwrap();
        }

        ReassembleXmlFileHandler::new()
            .reassemble(dir.to_str().unwrap(), Some("xml"), false, None)
            .await
            .unwrap();

        let output = tokio::fs::read_to_string(tmp.path().join("Bench.xml"))
            .await
            .unwrap();
        let order: Vec<usize> = output
            .match_indices("<n>")
            .map(|(at, _)| {
                let rest = &output[at + 3..];
                rest[..rest.find('<').unwrap()].parse().unwrap()
            })
            .collect();
        assert_eq!(order, (0..200).collect::<Vec<_>>());
    }

    // Pins the `delete ! in is_some_and(|s| !s.is_empty())` mutant at line 299.
    // When sidecar_specs is Some(&[]) the caller supplied nothing useful, so the
    // code must fall through to auto-detect via .sidecars.json — not treat the
    // empty slice as authoritative.
    #[tokio::test]
    async fn reassemble_plain_some_empty_sidecar_specs_falls_through_to_auto_detect() {
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("mydir");
        tokio::fs::create_dir(&dir).await.unwrap();

        tokio::fs::write(
            dir.join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><Child>hello</Child></Root>"#,
        )
        .await
        .unwrap();

        tokio::fs::write(
            dir.join(".sidecars.json"),
            r#"[{"element":"Notes","extension":"yaml"}]"#,
        )
        .await
        .unwrap();

        // Sidecar file name = directory name + extension (matches inject_sidecar_elements logic).
        tokio::fs::write(dir.join("mydir.yaml"), "key: value")
            .await
            .unwrap();

        h.reassemble_plain(dir.to_str().unwrap(), Some("xml"), false, &[], Some(&[]))
            .await
            .unwrap();

        let output = tokio::fs::read_to_string(tmp.path().join("mydir.xml"))
            .await
            .unwrap();
        assert!(
            output.contains("key: value"),
            "sidecar content missing — auto-detect did not run:\n{output}"
        );
    }

    // Pins `delete match arm "json"` mutant: without the arm, YAML input to
    // convert_to_format("json") falls through to `_ =>` and returns raw YAML.
    #[test]
    fn convert_to_format_yaml_to_json() {
        let yaml = "openapi: 3.0.1\ninfo:\n  title: \"Test API\"\n  version: 1.0.0\n";
        let out = convert_to_format(yaml, "json");
        let val: serde_json::Value = serde_json::from_str(&out).expect("output must be valid JSON");
        assert_eq!(val["openapi"], "3.0.1");
        assert_eq!(val["info"]["title"], "Test API");
        assert_eq!(val["info"]["version"], "1.0.0");
    }

    // Pins `delete match arm "yaml" | "yml"` mutant: without the arm, JSON input
    // to convert_to_format("yaml") falls through to `_ =>` and returns raw JSON.
    #[test]
    fn convert_to_format_json_to_yaml() {
        let json = r#"{"key":"value","num":42}"#;
        let out = convert_to_format(json, "yaml");
        assert!(
            serde_json::from_str::<serde_json::Value>(&out).is_err(),
            "output must be YAML format, not raw JSON: {out}"
        );
        let val: serde_json::Value =
            serde_saphyr::from_str(&out).expect("output must be valid YAML");
        assert_eq!(val["key"], "value");
        assert_eq!(val["num"], 42);
    }

    #[test]
    fn convert_to_format_yml_extension_same_as_yaml() {
        let json = r#"{"x":true}"#;
        let out = convert_to_format(json, "yml");
        let val: serde_json::Value = serde_saphyr::from_str(&out).unwrap();
        assert_eq!(val["x"], true);
    }

    #[test]
    fn convert_to_format_yaml_passes_through_unchanged() {
        // Pure YAML fails serde_json parse → returned as-is without re-serialization.
        let yaml = "title: \"@AuraEnabled\"\nversion: 1.0.0\n";
        assert_eq!(convert_to_format(yaml, "yaml"), yaml);
    }

    #[test]
    fn convert_to_format_unknown_extension_passes_through() {
        let raw = "arbitrary content";
        assert_eq!(convert_to_format(raw, "txt"), raw);
        assert_eq!(convert_to_format(raw, ""), raw);
    }

    #[test]
    fn convert_to_format_malformed_falls_back_to_raw() {
        let bad = "{{{{ not valid json or yaml at all >>>>>";
        assert_eq!(convert_to_format(bad, "json"), bad);
    }

    #[test]
    fn convert_to_format_non_string_key_falls_back_to_raw() {
        // YAML permits compound (sequence) mapping keys via explicit-key
        // syntax; JSON object keys must be strings, so parsing this into a
        // serde_json::Value fails and the raw content must come back unchanged.
        let yaml_with_sequence_key = "? [a, b]\n: value\n";
        assert_eq!(
            convert_to_format(yaml_with_sequence_key, "json"),
            yaml_with_sequence_key
        );
    }

    #[tokio::test]
    async fn reassemble_plain_uses_caller_supplied_sidecar_specs_directly() {
        // Exercises the `sidecar_specs` (non-empty, caller-supplied) arm
        // rather than falling through to `.sidecars.json` auto-detection.
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("mydir");
        tokio::fs::create_dir(&dir).await.unwrap();

        tokio::fs::write(
            dir.join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><Child>hello</Child></Root>"#,
        )
        .await
        .unwrap();
        // Deliberately no .sidecars.json — the caller-supplied spec must be used as-is.
        tokio::fs::write(dir.join("mydir.yaml"), "key: value")
            .await
            .unwrap();

        let specs = [SidecarSpec {
            element: "Notes".to_string(),
            extension: "yaml".to_string(),
            original_format: None,
        }];
        h.reassemble_plain(dir.to_str().unwrap(), Some("xml"), false, &[], Some(&specs))
            .await
            .unwrap();

        let output = tokio::fs::read_to_string(tmp.path().join("mydir.xml"))
            .await
            .unwrap();
        assert!(
            output.contains("key: value"),
            "caller-supplied sidecar spec was not used:\n{output}"
        );
    }

    #[tokio::test]
    async fn reassemble_plain_ignores_malformed_sidecars_json() {
        // Exercises the `Err(_)` arm of parsing `.sidecars.json`: malformed
        // content must not fail reassembly, just skip sidecar injection.
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("mydir");
        tokio::fs::create_dir(&dir).await.unwrap();

        tokio::fs::write(
            dir.join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><Child>hello</Child></Root>"#,
        )
        .await
        .unwrap();
        tokio::fs::write(dir.join(".sidecars.json"), "not valid json")
            .await
            .unwrap();

        h.reassemble_plain(dir.to_str().unwrap(), Some("xml"), false, &[], None)
            .await
            .unwrap();

        let output = tokio::fs::read_to_string(tmp.path().join("mydir.xml"))
            .await
            .unwrap();
        assert!(output.contains("Child"), "reassembly must still succeed");
    }

    #[tokio::test]
    async fn reassemble_plain_post_purge_removes_sidecar_files() {
        // Exercises the post_purge branch that removes sidecar files (and
        // the disassembled directory) once their content has been injected.
        let h = ReassembleXmlFileHandler::new();
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("mydir");
        tokio::fs::create_dir(&dir).await.unwrap();

        tokio::fs::write(
            dir.join("a.xml"),
            r#"<?xml version="1.0" encoding="UTF-8"?><Root><Child>hello</Child></Root>"#,
        )
        .await
        .unwrap();
        let sidecar_path = dir.join("mydir.yaml");
        tokio::fs::write(&sidecar_path, "key: value").await.unwrap();

        let specs = [SidecarSpec {
            element: "Notes".to_string(),
            extension: "yaml".to_string(),
            original_format: None,
        }];
        h.reassemble_plain(dir.to_str().unwrap(), Some("xml"), true, &[], Some(&specs))
            .await
            .unwrap();

        assert!(!dir.exists(), "post_purge must remove the disassembled dir");
        let output = tokio::fs::read_to_string(tmp.path().join("mydir.xml"))
            .await
            .unwrap();
        assert!(output.contains("key: value"));
    }

    #[tokio::test]
    async fn inject_sidecar_elements_ok_when_no_root_key() {
        // A merged value with no usable root key (only "?xml") must return
        // Ok(()) without attempting injection.
        let mut merged: XmlElement = json!({ "?xml": { "@version": "1.0" } });
        let specs = [SidecarSpec {
            element: "Notes".to_string(),
            extension: "yaml".to_string(),
            original_format: None,
        }];
        inject_sidecar_elements("does-not-matter", &mut merged, &specs)
            .await
            .unwrap();
        assert_eq!(merged, json!({ "?xml": { "@version": "1.0" } }));
    }

    #[tokio::test]
    async fn inject_sidecar_elements_skips_missing_sidecar_file() {
        // When the sidecar file referenced by the spec doesn't exist on disk,
        // the spec must be silently skipped rather than erroring.
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("mydir");
        tokio::fs::create_dir(&dir).await.unwrap();
        let mut merged: XmlElement = json!({ "Root": { "Child": { "#text": "hello" } } });
        let specs = [SidecarSpec {
            element: "Notes".to_string(),
            extension: "yaml".to_string(),
            original_format: None,
        }];
        inject_sidecar_elements(dir.to_str().unwrap(), &mut merged, &specs)
            .await
            .unwrap();
        assert_eq!(merged, json!({ "Root": { "Child": { "#text": "hello" } } }));
    }
}