bmad 6.11

_bmad/scripts/config_utils.py

@@ -0,0 +1,119 @@
+"""Shared strict TOML loading and structural merge support."""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+from typing import Any, Iterable
+
+
+class ConfigError(ValueError):
+    """Raised when a present configuration layer cannot be used safely."""
+
+
+_KEYED_MERGE_FIELDS = ("code", "id")
+
+
+def load_toml(path: Path, *, required: bool = False) -> dict[str, Any]:
+    """Load a TOML table, allowing absence only for optional layers."""
+    if not path.exists():
+        if required:
+            raise ConfigError(f"required TOML file not found: {path}")
+        return {}
+    if not path.is_file():
+        raise ConfigError(f"TOML layer is not a file: {path}")
+    try:
+        with path.open("rb") as stream:
+            parsed = tomllib.load(stream)
+    except tomllib.TOMLDecodeError as error:
+        raise ConfigError(f"failed to parse {path}: {error}") from error
+    except OSError as error:
+        raise ConfigError(f"failed to read {path}: {error}") from error
+    if not isinstance(parsed, dict):
+        raise ConfigError(f"TOML layer did not parse to a table: {path}")
+    return parsed
+
+
+def _detect_keyed_merge_field(items: list[Any]) -> str | None:
+    if not items or not all(isinstance(item, dict) for item in items):
+        return None
+    for candidate in _KEYED_MERGE_FIELDS:
+        if all(candidate in item for item in items):
+            for item in items:
+                value = item[candidate]
+                if not isinstance(value, str):
+                    raise ConfigError(
+                        f"keyed array identifier `{candidate}` must be a string, "
+                        f"got {type(value).__name__}"
+                    )
+                if not value:
+                    raise ConfigError(
+                        f"keyed array identifier `{candidate}` must not be empty"
+                    )
+            return candidate
+    return None
+
+
+def _merge_arrays(base: list[Any], override: list[Any]) -> list[Any]:
+    keyed_field = _detect_keyed_merge_field(base + override)
+    if keyed_field is None:
+        return list(base) + list(override)
+
+    result: list[Any] = []
+    index_by_key: dict[str, int] = {}
+    for item in base:
+        copied = dict(item)
+        index_by_key[copied[keyed_field]] = len(result)
+        result.append(copied)
+    for item in override:
+        copied = dict(item)
+        key = copied[keyed_field]
+        if key in index_by_key:
+            result[index_by_key[key]] = copied
+        else:
+            index_by_key[key] = len(result)
+            result.append(copied)
+    return result
+
+
+def structural_merge(base: Any, override: Any) -> Any:
+    """Merge tables recursively, keyed table arrays by identity, and append other arrays."""
+    if isinstance(base, dict) and isinstance(override, dict):
+        result = dict(base)
+        for key, value in override.items():
+            result[key] = structural_merge(result[key], value) if key in result else value
+        return result
+    if isinstance(base, list) and isinstance(override, list):
+        return _merge_arrays(base, override)
+    return override
+
+
+def merge_layers(layers: Iterable[dict[str, Any]]) -> dict[str, Any]:
+    merged: dict[str, Any] = {}
+    for layer in layers:
+        merged = structural_merge(merged, layer)
+    return merged
+
+
+def load_central_config(project_root: Path) -> dict[str, Any]:
+    bmad_dir = project_root / "_bmad"
+    return merge_layers(
+        (
+            load_toml(bmad_dir / "config.toml", required=True),
+            load_toml(bmad_dir / "config.user.toml"),
+            load_toml(bmad_dir / "custom" / "config.toml"),
+            load_toml(bmad_dir / "custom" / "config.user.toml"),
+        )
+    )
+
+
+def load_customization(project_root: Path | None, skill_dir: Path) -> dict[str, Any]:
+    skill_name = skill_dir.name
+    custom_dir = project_root / "_bmad" / "custom" if project_root else None
+    return merge_layers(
+        (
+            load_toml(skill_dir / "customize.toml", required=True),
+            load_toml(custom_dir / f"{skill_name}.toml") if custom_dir else {},
+            load_toml(custom_dir / f"{skill_name}.user.toml") if custom_dir else {},
+        )
+    )