Recover owned CAC catalog and native morph metadata with explicit runtime gates

tools/asset_pipeline/customisation_catalog.py

@@ -0,0 +1,295 @@
+"""Index owned CAC authoring data; extract only explicitly selected resources.
+
+This is a lossless metadata boundary, not a character assembler or a material
+compositor. It deliberately does not equate an authored variant with a usable
+in-game option. Gender, unlock, layering, material-loop and stamp rules still
+have to be applied by the native CAC implementation.
+
+Run with ``python -m tools.asset_pipeline.customisation_catalog --help``.
+Never distribute its output: the catalog is derived from the user's disc.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import tempfile
+import xml.etree.ElementTree as ET
+
+from tools.owned_game.big import BigArchive
+
+
+CATALOG_PATH = "data/content/recipe/createacharacter/cas_db_xfd1ca62f_000.xml"
+MODEL_PREFIX = "data/content/createacharacter/model/cas_db"
+TEXTURE_PREFIX = "data/content/createacharacter/texture"
+
+# Navigation labels requested by the user. This is a scope contract, not a list
+# of implemented pages. Do not generate working controls from this list alone.
+HIERARCHY = [
+    {"name": "Body mods", "children": [
+        "Gender", "Skin", "Hair", "Body shape", "Facial presets",
+        "Face modification", "Facial hair", "Upper body tattoo", "Lower body tattoo",
+    ]},
+    {"name": "Merchandise", "children": [
+        "Hats", "Tshirts", "Button shirts", "Hoodies", "Jackets", "Sweaters",
+        "Pants", "Shoes", "Socks", "Accessories", "Skateboards",
+    ]},
+    {"name": "Edit style", "children": ["Gestures", "Stance", "Style", "Posture"]},
+    {"name": "Trucks adjustment", "children": []},
+    {"name": "Wheels adjustment", "children": []},
+]
+
+
+def asset_id(value: str) -> str:
+    if not re.fullmatch(r"(?:0x)?[0-9a-fA-F]{16}", value):
+        raise ValueError(f"Invalid retail asset ID: {value!r}")
+    return value.removeprefix("0x").lower()
+
+
+def authored(node: ET.Element) -> dict:
+    """Keep ordering, attributes, and scalar strings; never invent shader data."""
+    result = {"tag": node.tag, "attributes": dict(node.attrib)}
+    if node.text and node.text.strip():
+        result["text"] = node.text
+    if len(node):
+        result["children"] = [authored(child) for child in node]
+    return result
+
+
+def authoring_flags(node: ET.Element) -> dict[str, str]:
+    result = {}
+    for group in node.findall("ud"):
+        if group.get("n") != "cas":
+            continue
+        for field in group:
+            if field.tag in result:
+                raise ValueError(f"Duplicate CAC field {field.tag}")
+            result[field.tag] = field.get("value", "")
+    return result
+
+
+def parse_catalog(xml: bytes, archive_paths: set[str]) -> dict:
+    root = ET.fromstring(xml)
+    if (root.tag, root.get("n"), root.get("type"), root.get("schemaversion")) != (
+        "compositeasset", "cas_db", "CreateACharacter", "6.0"
+    ):
+        raise ValueError("Unsupported CAC authoring schema")
+    if any(child.tag not in {"mat", "comp"} for child in root):
+        raise ValueError("Unrecognized CAC root record")
+    available = {p.lower() for p in archive_paths}
+    materials = {}
+    for material in root.findall("mat"):
+        key = asset_id(material.attrib["id"])
+        if key in materials:
+            raise ValueError(f"Duplicate material {key}")
+        textures = []
+        for parameter in material.findall("sp"):
+            # Keep duplicate channels and their order rather than silently
+            # collapsing separate authored bindings into a channel dictionary.
+            texture = asset_id(parameter.attrib["id"])
+            path = f"{TEXTURE_PREFIX}/0x{texture}.rx2"
+            textures.append({"channel": parameter.attrib["chn"], "id": texture,
+                             "path": path, "in_archive": path.lower() in available})
+        materials[key] = {
+            "shader": material.attrib["type"], "textures": textures,
+            "flags": authoring_flags(material), "authored": authored(material),
+        }
+    components = []
+    slots = set()
+    for component in root.findall("comp"):
+        slot = component.attrib["n"]
+        if not re.fullmatch(r"[A-Za-z]+", slot) or slot in slots:
+            raise ValueError(f"Invalid or duplicate component slot {slot!r}")
+        slots.add(slot)
+        if any(child.tag != "mod" for child in component):
+            raise ValueError(f"Unrecognized record in {slot}")
+        models = []
+        ids = set()
+        for model in component.findall("mod"):
+            key = asset_id(model.attrib["id"])
+            if key in ids:
+                raise ValueError(f"Duplicate model {slot}/{key}")
+            ids.add(key)
+            lods = []
+            lod_indices = set()
+            for lod in model.findall("lod"):
+                index = int(lod.attrib["idx"])
+                if index < 0 or index in lod_indices:
+                    raise ValueError(f"Invalid or duplicate LOD {slot}/{key}/{index}")
+                lod_indices.add(index)
+                # XML arenaid names the RX2 FILE; XML id is the arena asset ID.
+                # The fallback .recipe stores them in the opposite field order.
+                model_id = asset_id(lod.attrib["arenaid"])
+                path = f"{MODEL_PREFIX}/{slot}/0x{model_id}.rx2"
+                instances = []
+                for instance in lod.findall("matinst"):
+                    variants = []
+                    variant_ids = set()
+                    for variant in instance.findall("matvar"):
+                        material_id = asset_id(variant.attrib["id"])
+                        if material_id not in materials:
+                            raise ValueError(f"Missing material definition {material_id}")
+                        if material_id in variant_ids:
+                            raise ValueError(f"Duplicate material variant {slot}/{key}/{material_id}")
+                        variant_ids.add(material_id)
+                        variants.append({"id": material_id, "name": variant.attrib["n"],
+                                         "authored": authored(variant)})
+                    if not variants:
+                        raise ValueError(f"Empty material instance {slot}/{key}")
+                    instances.append(variants)
+                if not instances:
+                    raise ValueError(f"Missing material instances {slot}/{key}")
+                lods.append({"index": index, "arena_asset_id": asset_id(lod.attrib["id"]),
+                             "model_id": model_id, "path": path,
+                             "in_archive": path.lower() in available,
+                             "material_instances": instances})
+            if not lods:
+                raise ValueError(f"Missing LODs {slot}/{key}")
+            models.append({"id": key, "name": model.attrib["n"],
+                           "flags": authoring_flags(model), "lods": lods,
+                           "authored": authored(model)})
+        components.append({"slot": slot, "models": models})
+    return {"version": 1, "source": {"entry": CATALOG_PATH,
+            "xml_sha256": hashlib.sha256(xml).hexdigest(), "attributes": dict(root.attrib)},
+            "scope": HIERARCHY, "materials": materials, "components": components}
+
+
+def selected_paths(catalog: dict, selections: list[dict]) -> list[str]:
+    """Validate extraction requests, not gender/outfit/material compatibility.
+
+    Each request supplies slot, asset_id, lod and one material_id per ordered
+    material instance. Requiring every instance prevents losing female head's
+    second material loop. Requests do not authorize arbitrary archive paths.
+    """
+    paths = set()
+    slots = set()
+    for selection in selections:
+        slot = selection["slot"]
+        if slot in slots:
+            raise ValueError(f"Repeated selected slot {slot}")
+        slots.add(slot)
+        candidates = [m for c in catalog["components"] if c["slot"] == slot
+                      for m in c["models"] if m["id"] == asset_id(selection["asset_id"])]
+        if len(candidates) != 1:
+            raise ValueError(f"Unknown selected component {slot}/{selection['asset_id']}")
+        lods = [lod for lod in candidates[0]["lods"] if lod["index"] == selection["lod"]]
+        if len(lods) != 1:
+            raise ValueError(f"Unknown selected LOD for {slot}")
+        lod = lods[0]
+        if not lod["in_archive"]:
+            raise ValueError(f"Selected model is absent from archive: {lod['path']}")
+        choices = selection["material_ids"]
+        if len(choices) != len(lod["material_instances"]):
+            raise ValueError(f"{slot} requires {len(lod['material_instances'])} material selections")
+        paths.add(lod["path"])
+        for choice, instance in zip(choices, lod["material_instances"]):
+            key = asset_id(choice)
+            if key not in {v["id"] for v in instance}:
+                raise ValueError(f"Material {key} is not a variant of this {slot} instance")
+            for texture in catalog["materials"][key]["textures"]:
+                if not texture["in_archive"]:
+                    raise ValueError(f"Selected texture is absent from archive: {texture['path']}")
+                paths.add(texture["path"])
+    return sorted(paths)
+
+
+def write_private(path: Path, data: bytes) -> str:
+    """Idempotent, no-clobber staging, including failed or interrupted runs."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.exists():
+        if path.read_bytes() != data:
+            raise ValueError(f"Refusing to replace different staged content: {path}")
+        return "reused"
+    # Publish a complete file without replacing another extractor's output.
+    # Windows rename is atomic and refuses an existing destination. POSIX
+    # rename replaces it, so use an exclusive hard link there instead.
+    descriptor, temporary = tempfile.mkstemp(prefix=".cac-", dir=path.parent)
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            stream.write(data)
+        try:
+            if os.name == "nt":
+                os.rename(temporary, path)
+            else:
+                os.link(temporary, path)
+        except FileExistsError:
+            if path.read_bytes() != data:
+                raise ValueError(f"Conflicting concurrent staged content: {path}")
+            return "reused"
+    finally:
+        Path(temporary).unlink(missing_ok=True)
+    return "written"
+
+
+def prepare(game_root: Path, output: Path, selections: list[dict] | None = None) -> dict:
+    game_root = game_root.resolve(strict=True)
+    output = output.resolve()
+    if output == game_root or output.is_relative_to(game_root) or game_root.is_relative_to(output):
+        raise ValueError("The private staging directory must be separate from the owned source")
+    archive = BigArchive(game_root / "data/content/createacharacter.big")
+    entries = {e.path.lower(): e for e in archive.entries}
+    if len(entries) != len(archive.entries):
+        raise ValueError("Ambiguous case-insensitive paths in CAC archive")
+    xml = archive.read(entries[CATALOG_PATH.lower()])
+    catalog = parse_catalog(xml, set(entries))
+    # Resolve every selected dependency before writing any selected resources.
+    paths = selected_paths(catalog, selections or [])
+    from tools.asset_pipeline.customisation_native import read_database
+    native = read_database(game_root, output / "database")
+    paths.append(CATALOG_PATH)
+    records = []
+    for name in paths:
+        entry = entries[name.lower()]
+        data = xml if name == CATALOG_PATH else archive.read(entry)
+        destination = (output / "source" / name).resolve()
+        if not destination.is_relative_to(output):
+            raise ValueError("Staged asset escaped the private directory")
+        state = write_private(destination, data)
+        records.append({"path": name, "size": len(data), "sha256": hashlib.sha256(data).hexdigest(),
+                        "state": state})
+    models = [m for c in catalog["components"] for m in c["models"]]
+    report = {"version": 1, "source_xml_sha256": catalog["source"]["xml_sha256"],
+              "component_slots": len(catalog["components"]), "model_variants": len(models),
+              "materials": len(catalog["materials"]),
+              "native_collections": len(native["collections"]),
+              "native_morphs": len(native["morphs"]),
+              "multiple_material_instance_lods": sum(len(lod["material_instances"]) > 1
+                                                       for m in models for lod in m["lods"]),
+              "missing_model_paths": sorted({lod["path"] for m in models for lod in m["lods"]
+                                              if not lod["in_archive"]}),
+              "missing_texture_paths": sorted({t["path"] for m in catalog["materials"].values()
+                                                for t in m["textures"] if not t["in_archive"]}),
+              "extracted": records,
+              "runtime_ready": False,
+              "requires": ["native CAC assembly and rig validation", "native material and tattoo composition",
+                           "native recipe colour/morph/style defaults and menu filtering"]}
+    # Catalog output is deterministic. A different disc/version needs its own
+    # staging directory rather than silently reusing another catalog's files.
+    write_private(output / "catalog.json", (json.dumps(catalog, indent=2) + "\n").encode())
+    write_private(output / "native.json", (json.dumps(native, indent=2) + "\n").encode())
+    return report
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--game-root", required=True, type=Path, help="Owned extracted Xbox game directory")
+    parser.add_argument("--output", required=True, type=Path, help="Private task staging directory, outside game source")
+    parser.add_argument("--selection", type=Path, help="JSON array of explicit component extraction requests")
+    args = parser.parse_args()
+    selections = json.loads(args.selection.read_text(encoding="utf-8")) if args.selection else None
+    if selections is not None and not isinstance(selections, list):
+        parser.error("--selection must contain a JSON array")
+    report = prepare(args.game_root, args.output, selections)
+    report_path = (args.output / "extraction-report.json").resolve()
+    if not report_path.is_relative_to(args.output.resolve()):
+        raise ValueError("Report path escaped the private staging directory")
+    report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    print(f"CAC_CATALOG_READY slots={report['component_slots']} models={report['model_variants']} "
+          f"materials={report['materials']} staged={len(report['extracted'])} runtime_ready=false")
+
+
+if __name__ == "__main__":
+    main()

tools/asset_pipeline/customisation_native.py

@@ -0,0 +1,118 @@
+"""Read native CAC settings without changing the gameplay collections schema.
+
+The XML catalog contains models/materials; BIN/VLT contains preset arrays,
+colour parameters, morph order/ranges, cameras and menu filters. Preserve both.
+Raw reflected fields remain authoritative; decoded helpers only cover layouts
+corroborated with the native code. See docs/character-customisation.md.
+"""
+from __future__ import annotations
+
+import hashlib
+import math
+from pathlib import Path
+import struct
+
+from tools.asset_pipeline.vlt import convert, hash64
+from tools.owned_game.big import BigArchive
+
+
+DATABASE_FILES = {
+    "skaterschema.bin", "skaterschema.vlt", "skatercollections.bin",
+    "skatercollections.vlt", "skaterschema_summaryreport.txt",
+}
+
+# Reflected layout offsets are independently corroborated by 8253CA60..CD18.
+MORPH_CLASS = 0x7F47CA62F4100F41
+MORPH_FIELDS = {
+    "target": 0xAAC0F30B119BD14F,  # layout +0
+    "target_order": 0x0D381DFA82C751E1,  # layout +4
+    "ui_zone": 0x3D15CA957C745188,  # layout +8; not always target_order!
+    "min_a": 0x0279ED3135F03661,  # layout +12
+    "max_a": 0xD4340AC527D228E1,  # layout +16
+    "default_a": 0x49E3727FEB2F6475,  # layout +20
+    "min_b": 0x11691A8EEF816175,  # layout +24
+    "max_b": 0x7680C10C0F43EB3E,  # layout +28
+    "default_b": 0xBE3909745E8A3935,  # layout +32
+}
+
+
+def key_hash(key: str) -> int:
+    return int(key[5:], 16) if key.startswith("Hash_") else hash64(key)
+
+
+def fields_by_hash(row: dict) -> dict[int, dict]:
+    return {key_hash(key): value for key, value in row["fields"].items()}
+
+
+def float32(field: dict) -> float:
+    data = bytes.fromhex(field["data"])
+    if field["type"] != "EA::Reflection::Float" or len(data) != 4:
+        raise ValueError("Expected one native Float field")
+    value, = struct.unpack(">f", data)
+    if not math.isfinite(value):
+        raise ValueError("Non-finite native Float field")
+    return value
+
+
+def decode_morphs(rows: list[dict]) -> list[dict]:
+    result = []
+    for row in rows:
+        if key_hash(row["class"]) != MORPH_CLASS:
+            continue
+        fields = fields_by_hash(row)
+        values = {name: fields[key] for name, key in MORPH_FIELDS.items()}
+        order, = struct.unpack(">i", bytes.fromhex(values["target_order"]["data"]))
+        zone, = struct.unpack(">I", bytes.fromhex(values["ui_zone"]["data"]))
+        if order == -1:
+            continue  # schema default, not an additional user morph
+        if not 0 <= order < 19 or not 0 <= zone < 19:
+            raise ValueError("Native morph index outside the 19-field container")
+        if values["target"]["type"] != "EA::Reflection::Text":
+            raise ValueError("Undecoded native morph target name")
+        entry = {"key": row["key"], "target": values["target"]["data"],
+                 "target_order": order, "ui_zone": zone}
+        # Retain native branch labels until the gender boolean's meaning is
+        # verified. The inspected retail bank gives identical A/B ranges.
+        for branch in ("a", "b"):
+            bounds = {name: float32(values[f"{name}_{branch}"])
+                      for name in ("min", "max", "default")}
+            if not bounds["min"] <= bounds["default"] <= bounds["max"] or bounds["min"] == bounds["max"]:
+                raise ValueError(f"Invalid native morph range: {row['key']}")
+            entry[f"branch_{branch}"] = bounds
+        result.append(entry)
+    if len(result) != 19 or {m["target_order"] for m in result} != set(range(19)):
+        raise ValueError("Native morph target order is incomplete or duplicated")
+    if len({m["target"] for m in result}) != 19:
+        raise ValueError("Duplicate native morph target name")
+    return sorted(result, key=lambda m: m["target_order"])
+
+
+def read_database(game_root: Path, staging: Path) -> dict:
+    # Import locally to avoid a module cycle with the catalog's optional stage.
+    from tools.asset_pipeline.customisation_catalog import write_private
+
+    game_root = game_root.resolve(strict=True)
+    staging = staging.resolve()
+    if staging == game_root or staging.is_relative_to(game_root) or game_root.is_relative_to(staging):
+        raise ValueError("Database staging must be separate from owned source")
+    archive = BigArchive(game_root / "data/big/db.big")
+    entries = [e for e in archive.entries if e.path.lower().startswith("data/db/")
+               and Path(e.path).name.lower() in DATABASE_FILES]
+    if len(entries) != len(DATABASE_FILES):
+        raise ValueError("Missing or ambiguous owned CAC BIN/VLT inputs")
+    sources = []
+    for entry in entries:
+        data = archive.read(entry)
+        name = Path(entry.path).name.lower()
+        write_private(staging / name, data)
+        sources.append({"entry": entry.path, "size": len(data),
+                        "sha256": hashlib.sha256(data).hexdigest()})
+    # The game ships its class/collection names in this report. No local dump
+    # or machine-specific reverse-engineering name cache is needed.
+    names = Path(__file__).with_name("names.txt").read_text(encoding="utf-8").splitlines()
+    for line in (staging / "skaterschema_summaryreport.txt").read_text(encoding="utf-8").splitlines():
+        names.extend(line.replace(".class", "").replace(".xml", "").split("/"))
+    names.extend(["Sk8::CAC::MorphPreset", "Sk8::CAC::ColourPreset", "Sk8::CAC::MORPH_ZONES"])
+    data = convert(staging / "skaterschema", staging / "skatercollections", names)
+    rows = [r for r in data["collections"] if r["class"].startswith(("cac_", "cas_"))]
+    return {"version": 1, "sources": sources, "collections": rows, "morphs": decode_morphs(rows)}

tools/asset_pipeline/test_customisation_catalog.py

@@ -0,0 +1,130 @@
+"""Synthetic CAC records only; no game content is needed or stored here."""
+from pathlib import Path
+import struct
+import tempfile
+import unittest
+
+from tools.asset_pipeline.customisation_catalog import (
+    MODEL_PREFIX, TEXTURE_PREFIX, parse_catalog, prepare,
+    selected_paths, write_private,
+)
+from tools.asset_pipeline.customisation_native import MORPH_FIELDS, decode_morphs
+
+
+M1 = "0000000000000001"
+M2 = "0000000000000002"
+MODEL = "0000000000000020"
+ASSET = "0000000000000030"
+TEX1 = "0000000000000040"
+TEX2 = "0000000000000050"
+
+
+def fixture():
+    xml = f'''<compositeasset n="cas_db" type="CreateACharacter" schemaversion="6.0">
+    <mat id="0x{M1}" type="test_shader">
+      <ud n="cas"><cas.Gender value="female"/><cas.RequiresRemovalOfComponents value="Hat,Hair"/></ud>
+      <sp chn="diffuse" id="0x{TEX1}" uv="1" sampler="authored"/>
+      <scalar value="0.375"/>
+    </mat>
+    <mat id="0x{M2}" type="test_shader_second"><sp chn="alpha" id="0x{TEX2}"/></mat>
+    <comp n="Head"><mod id="0x{ASSET}" n="synthetic_head">
+      <lod idx="0" id="0x0000000000000060" arenaid="0x{MODEL}">
+        <matinst><matvar id="0x{M1}" n="first"/></matinst>
+        <matinst><matvar id="0x{M2}" n="second"/></matinst>
+      </lod>
+    </mod></comp></compositeasset>'''.encode()
+    paths = {f"{MODEL_PREFIX}/Head/0x{MODEL}.rx2",
+             f"{TEXTURE_PREFIX}/0x{TEX1}.rx2", f"{TEXTURE_PREFIX}/0x{TEX2}.rx2"}
+    selection = [{"slot": "Head", "asset_id": ASSET, "lod": 0, "material_ids": [M1, M2]}]
+    return xml, paths, selection
+
+
+class CatalogTests(unittest.TestCase):
+    def test_preserves_material_instance_order_and_authored_shader_inputs(self):
+        xml, paths, selection = fixture()
+        catalog = parse_catalog(xml, paths)
+        lod = catalog["components"][0]["models"][0]["lods"][0]
+        self.assertEqual(lod["model_id"], MODEL)  # arenaid names the RX2, not id
+        self.assertEqual([group[0]["id"] for group in lod["material_instances"]], [M1, M2])
+        definition = catalog["materials"][M1]["authored"]["children"]
+        self.assertEqual(definition[1]["attributes"]["sampler"], "authored")
+        self.assertEqual(definition[1]["attributes"]["uv"], "1")
+        self.assertEqual(definition[2]["attributes"]["value"], "0.375")
+        self.assertEqual(catalog["materials"][M1]["flags"]["cas.RequiresRemovalOfComponents"], "Hat,Hair")
+        self.assertEqual(set(selected_paths(catalog, selection)), paths)
+
+    def test_rejects_incomplete_or_reordered_material_selection(self):
+        xml, paths, selection = fixture()
+        catalog = parse_catalog(xml, paths)
+        for choices in ([M1], [M2, M1], [M1, "000000000000ffff"]):
+            selection[0]["material_ids"] = choices
+            with self.assertRaises(ValueError):
+                selected_paths(catalog, selection)
+
+    def test_missing_archived_texture_is_reported_and_cannot_be_extracted(self):
+        xml, paths, selection = fixture()
+        paths.remove(f"{TEXTURE_PREFIX}/0x{TEX2}.rx2")
+        catalog = parse_catalog(xml, paths)
+        self.assertFalse(catalog["materials"][M2]["textures"][0]["in_archive"])
+        with self.assertRaisesRegex(ValueError, "Selected texture is absent"):
+            selected_paths(catalog, selection)
+
+    def test_rejects_broken_material_references_and_path_slots(self):
+        xml, paths, _ = fixture()
+        broken = xml.replace(f'<matvar id="0x{M1}"'.encode(), b'<matvar id="0x000000000000ffff"')
+        with self.assertRaisesRegex(ValueError, "Missing material definition"):
+            parse_catalog(broken, paths)
+        with self.assertRaisesRegex(ValueError, "Invalid or duplicate component"):
+            parse_catalog(xml.replace(b'comp n="Head"', b'comp n="../Head"'), paths)
+
+    def test_never_overwrites_staged_content_and_reuses_identical_files(self):
+        with tempfile.TemporaryDirectory() as root:
+            target = Path(root) / "private" / "resource"
+            self.assertEqual(write_private(target, b"first"), "written")
+            self.assertEqual(write_private(target, b"first"), "reused")
+            with self.assertRaisesRegex(ValueError, "Refusing to replace"):
+                write_private(target, b"second")
+            self.assertEqual(target.read_bytes(), b"first")
+            self.assertEqual([p.name for p in target.parent.iterdir()], ["resource"])
+
+    def test_rejects_staging_inside_or_above_owned_source_before_opening_archive(self):
+        with tempfile.TemporaryDirectory() as root:
+            source = Path(root) / "owned"
+            source.mkdir()
+            for destination in (source, source / "overlay", Path(root)):
+                with self.assertRaisesRegex(ValueError, "separate from the owned source"):
+                    prepare(source, destination)
+
+
+def morph_rows():
+    rows = []
+    for index in range(19):
+        values = {"target": {"type": "EA::Reflection::Text", "data": f"target_{index}"},
+                  "target_order": {"type": "EA::Reflection::Int32", "data": struct.pack(">i", index).hex()},
+                  "ui_zone": {"type": "zone", "data": struct.pack(">I", 18 - index).hex()}}
+        for branch in ("a", "b"):
+            for name, number in (("min", 0.0), ("max", 0.5), ("default", 0.25)):
+                values[f"{name}_{branch}"] = {"type": "EA::Reflection::Float", "data": struct.pack(">f", number).hex()}
+        rows.append({"class": "cac_morph_params", "key": f"field_{index}", "fields": {
+            f"Hash_{MORPH_FIELDS[name]:016X}": value for name, value in values.items()}})
+    return rows
+
+
+class NativeMetadataTests(unittest.TestCase):
+    def test_target_order_and_ui_zone_remain_independent(self):
+        decoded = decode_morphs(list(reversed(morph_rows())))
+        self.assertEqual(decoded[3]["target_order"], 3)
+        self.assertEqual(decoded[3]["ui_zone"], 15)
+        self.assertEqual(decoded[3]["branch_a"], {"min": 0.0, "max": 0.5, "default": 0.25})
+
+    def test_rejects_missing_targets_and_nonfinite_ranges(self):
+        with self.assertRaisesRegex(ValueError, "incomplete or duplicated"):
+            decode_morphs(morph_rows()[:-1])
+        rows = morph_rows()
+        rows[3]["fields"][f"Hash_{MORPH_FIELDS['max_a']:016X}"]["data"] = "7fc00000"
+        with self.assertRaisesRegex(ValueError, "Non-finite"):
+            decode_morphs(rows)
+
+
+if __name__ == "__main__":
+    unittest.main()