feat(admin-config-tool): task 2.1 - xcecommon.validation

Validate operator-authored YAML against the 7 bundled JSON schemas,
enforce reserved-name and duplicate-name rejection, and flag undefined
group/resource cross-references as non-blocking warnings.

.kiro/specs/admin-config-tool/tasks.md

@@ -22,7 +22,7 @@
   - _Requirements: 5.5, 8.1, 8.2, 8.6_
 
 - [ ] 2. Core: xcecommon shared library
-- [ ] 2.1 (P) Validate operator-authored YAML against the bundled schemas
+- [x] 2.1 (P) Validate operator-authored YAML against the bundled schemas
   - Each of the seven config file types can be checked against its own bundled validation schema, independent of anything the operator supplies
   - Reserved queue/host names are rejected even when otherwise schema-valid
   - Duplicate names within a single file/section are rejected
@@ -146,3 +146,4 @@
 - **1.3**: nix's bare `python3` does not put `pip` on PATH — `python3Packages.pip` had to be added to `nativeBuildInputs` alongside the packages design.md's Technology Stack table lists; not a spec gap, just an operational dependency the table didn't need to spell out.
 - **1.3**: Claude Code `PostToolUse` hooks receive tool input via stdin JSON (`tool_input.file_path`), not a `$FILE_PATH` env var — brief.md's ruff-hook sketch used the env-var form, which would have silently no-op'd. The actual hooks added to `.claude/settings.json` follow this repo's existing `.wolf/hooks/post-write.js` stdin-JSON convention instead. Any future hook work in this repo should follow the same pattern, not brief.md's original sketch.
 - **1.3**: `src/xceadmin/src/xcecommon/schema_files/` (synced capnp + JSON schema copies) is a build artifact, regenerated every `xce-build.sh xceadmin` run — never hand-edit it, never commit it (`src/xceadmin/.gitignore` excludes it).
+- **2.1**: `check_undefined_refs` deliberately does not parse resource names out of `res_req`/`gpu_req` free-form LSF-style expression strings (e.g. `"mem>1000 rusage[mem=1000]"`) — design.md's `xcecommon.validation` component scopes this check to queue→host-group and limit→group/`resname` only. `limits.yaml`'s `projects` field and `scaler.yaml` name collisions are likewise out of scope — no `projects.yaml` catalog exists among the 7 config types, and Requirement 4.3 only enumerates queue/host/resource/group/limit names. `src/configs/hosts.yaml` has 2 pre-existing undefined refs (resource `mg`, host `h3`) that now surface as warnings — expected, not a regression; fixing the example fixture is outside this task's boundary.

src/xceadmin/src/xcecommon/validation.py

@@ -0,0 +1,332 @@
+"""JSON Schema validation plus business-rule checks for operator-authored
+YAML config files (design.md, ``xcecommon.validation``; Requirements 4.1,
+4.2, 4.3, 4.4, 4.5, 5.5).
+
+This module has no dependency on ``xcecommon.schema``/``mapping``/``wire``
+(design.md, Dependency Direction) — it only touches YAML dicts, the bundled
+JSON schemas, and the shared exception hierarchy in ``xcecommon.errors``.
+
+Four checks, run in this order by the (later) build orchestration:
+
+1. ``validate_yaml`` — JSON Schema (Draft 2020-12) structural validation
+   against a bundled schema (Requirement 4.1). Hard failure.
+2. ``check_reserved_names`` — ``"lost_and_found"`` is reserved and cannot
+   be used as a queue or host name (Requirement 4.2). Hard failure.
+3. ``check_duplicates`` — duplicate queue/host/resource/group/limit names,
+   scoped to their own file/section (Requirement 4.3). Hard failure.
+4. ``check_undefined_refs`` — cross-file references (queue -> host group,
+   limit -> user/queue/host group or resname) that do not resolve to a
+   declared entry (Requirement 4.4, 4.5). Warnings only, never raises.
+"""
+
+from __future__ import annotations
+
+import importlib.resources
+import json
+from pathlib import Path
+from typing import Any
+
+import jsonschema
+import yaml
+
+from xcecommon.errors import DuplicateEntryError, ReservedNameError, ValidationError
+
+RESERVED_NAME = "lost_and_found"
+
+# The membership-reference "all" keyword and the fields consulted for each
+# reference check below are documented per-check, not centralized here —
+# see check_undefined_refs.
+
+
+def default_schema_dir() -> str:
+    """Resolve the directory containing the 7 bundled JSON validation
+    schemas (Requirement 5.5).
+
+    Uses ``importlib.resources`` so this resolves correctly under both
+    ``pip install -e .`` (points into the checked-out ``schema_files/json``
+    copy) and ``pip install .`` (points into the installed package's copy)
+    — the same mechanism ``xcecommon.schema.default_schema_path()`` uses
+    for ``controller.capnp``. ``configs_dir`` is never consulted for a
+    ``schema/`` subdirectory (superseded Requirement 1.6).
+    """
+    return str(importlib.resources.files("xcecommon") / "schema_files" / "json")
+
+
+def _schema_path_for(schema_name: str) -> Path:
+    return Path(default_schema_dir()) / f"{schema_name}.schema.json"
+
+
+def _json_pointer(path_parts: Any) -> str:
+    if not path_parts:
+        return "/"
+    return "/" + "/".join(str(part) for part in path_parts)
+
+
+def validate_yaml(file_path: str, schema_name: str) -> dict:
+    """Parse ``file_path`` as YAML and validate it against the bundled
+    schema named ``schema_name`` (the YAML file's base name with ``.yaml``
+    stripped, e.g. ``"queues"`` for ``queues.yaml`` -> ``queues.schema.json``).
+
+    Returns the parsed (and now schema-valid) top-level dict on success.
+    Raises :class:`ValidationError` — carrying ``file``, ``json_pointer``,
+    and ``constraint`` (Requirement 4.1) — on any failure: missing file,
+    malformed YAML syntax, missing/unreadable bundled schema, or a genuine
+    schema violation.
+    """
+    schema_path = _schema_path_for(schema_name)
+    if not schema_path.is_file():
+        raise ValidationError(
+            f"{file_path}: no bundled schema named {schema_name!r} "
+            f"(expected {schema_path})",
+            file=file_path,
+            json_pointer="/",
+            constraint="schema-not-found",
+        )
+
+    try:
+        with open(file_path, encoding="utf-8") as fh:
+            raw_text = fh.read()
+    except OSError as exc:
+        raise ValidationError(
+            f"{file_path}: cannot read file: {exc}",
+            file=file_path,
+            json_pointer="/",
+            constraint="file-not-found",
+        ) from exc
+
+    try:
+        data = yaml.safe_load(raw_text)
+    except yaml.YAMLError as exc:
+        raise ValidationError(
+            f"{file_path}: invalid YAML syntax: {exc}",
+            file=file_path,
+            json_pointer="/",
+            constraint="yaml-syntax",
+        ) from exc
+
+    if data is None:
+        data = {}
+
+    schema = json.loads(schema_path.read_text(encoding="utf-8"))
+    validator = jsonschema.Draft202012Validator(schema)
+    errors = sorted(validator.iter_errors(data), key=lambda e: list(e.path))
+    if errors:
+        first = errors[0]
+        pointer = _json_pointer(list(first.path))
+        raise ValidationError(
+            f"{file_path}: {first.message} (at {pointer})",
+            file=file_path,
+            json_pointer=pointer,
+            constraint=str(first.validator),
+        )
+
+    return data
+
+
+def check_reserved_names(sections: dict) -> None:
+    """Reject the reserved name ``"lost_and_found"`` when used as a queue
+    or host name, even if the file otherwise passed schema validation
+    (Requirement 4.2).
+
+    ``sections`` maps schema_name -> the dict returned by ``validate_yaml``
+    for that file (missing/absent files simply have no key and are
+    skipped).
+    """
+    hosts_data = sections.get("hosts") or {}
+    for host in hosts_data.get("hosts") or []:
+        if host.get("hostname") == RESERVED_NAME:
+            raise ReservedNameError(
+                f'hosts.yaml: "{RESERVED_NAME}" is reserved and cannot be '
+                "used as a host name"
+            )
+
+    queues_data = sections.get("queues") or {}
+    for queue in queues_data.get("queues") or []:
+        if queue.get("name") == RESERVED_NAME:
+            raise ReservedNameError(
+                f'queues.yaml: "{RESERVED_NAME}" is reserved and cannot be '
+                "used as a queue name"
+            )
+
+
+def _raise_on_duplicates(
+    items: list[dict] | None, key: str, file_label: str, entity_label: str
+) -> None:
+    seen: set[str] = set()
+    duplicates: set[str] = set()
+    for item in items or []:
+        name = item.get(key)
+        if name is None:
+            continue
+        if name in seen:
+            duplicates.add(name)
+        seen.add(name)
+
+    if duplicates:
+        names = ", ".join(sorted(duplicates))
+        raise DuplicateEntryError(
+            f"{file_label}: duplicate {entity_label} name(s): {names}"
+        )
+
+
+def check_duplicates(sections: dict) -> None:
+    """Reject duplicate names within their own file/section (Requirement
+    4.3): queue names, host names, resource names, host-group names,
+    user-group names, and limit names. Each category is checked
+    independently — a name reused across categories (e.g. a limit named
+    the same as a queue) is intentionally allowed.
+    """
+    hosts_data = sections.get("hosts") or {}
+    _raise_on_duplicates(hosts_data.get("hosts"), "hostname", "hosts.yaml", "host")
+    _raise_on_duplicates(
+        hosts_data.get("hostgroups"), "groupname", "hosts.yaml", "host group"
+    )
+
+    queues_data = sections.get("queues") or {}
+    _raise_on_duplicates(queues_data.get("queues"), "name", "queues.yaml", "queue")
+
+    resources_data = sections.get("resources") or {}
+    _raise_on_duplicates(
+        resources_data.get("resources"), "name", "resources.yaml", "resource"
+    )
+
+    users_data = sections.get("users") or {}
+    _raise_on_duplicates(
+        users_data.get("usergroups"), "groupname", "users.yaml", "user group"
+    )
+
+    limits_data = sections.get("limits") or {}
+    _raise_on_duplicates(limits_data.get("limits"), "name", "limits.yaml", "limit")
+
+
+def _membership_names(membership: dict | None) -> list[str]:
+    if not membership:
+        return []
+    members = list(membership.get("members") or [])
+    excludes = list(membership.get("exclude_members") or [])
+    return members + excludes
+
+
+def _warn_undefined_membership(
+    membership: dict | None,
+    known: set[str],
+    file_label: str,
+    context_label: str,
+    warnings: list[str],
+) -> None:
+    for name in _membership_names(membership):
+        if name not in known:
+            warnings.append(
+                f"{file_label}: {context_label} references undefined entry {name!r}"
+            )
+
+
+def check_undefined_refs(sections: dict) -> list[str]:  # noqa: C901 - single cohesive pass
+    """Warn (never raise, Requirement 4.4) about cross-file references
+    that do not resolve to a declared entry:
+
+    - a queue's ``hosts``/``users`` membership referencing an undefined
+      host/host-group or user-group
+    - a queue's or host's ``sched_stop`` ``resname`` referencing an
+      undefined resource
+    - a host's per-host ``resource`` entries referencing an undefined
+      resource
+    - a limit's ``users``/``queues``/``hosts`` membership or ``resname``
+      referencing an undefined group/queue/host or resource
+
+    Per Requirement 4.5, ``resources.yaml`` is the exhaustive resource
+    catalog: built-in/dynamic resource names (``slots``, ``mem``, and the
+    8 ``add_hostattrs_resvals()`` host attributes) are NOT special-cased
+    as implicitly valid — they must actually be declared in
+    ``resources.yaml`` like any other resource, or a reference to them
+    warns just like a reference to any other undefined name.
+
+    ``limits.yaml``'s ``projects`` membership field has no corresponding
+    catalog file in this xceadmin dataset (no ``projects.yaml`` input, no
+    ``ClusterConfig`` field maps to it) — there is nothing to check it
+    against, so it is intentionally not validated here.
+
+    Free-form resource-requirement expression strings (``queues.yaml``'s
+    ``res_req``/``gpu_req``) are not parsed for embedded resource names —
+    that would require reimplementing LSF-style expression syntax
+    (``select[]``, ``rusage[]``, etc.); only the structured ``resname``
+    fields (``sched_stop``, ``limits.resname``, per-host ``resource``
+    entries) are checked.
+    """
+    warnings: list[str] = []
+
+    hosts_data = sections.get("hosts") or {}
+    queues_data = sections.get("queues") or {}
+    users_data = sections.get("users") or {}
+    limits_data = sections.get("limits") or {}
+    resources_data = sections.get("resources") or {}
+
+    host_names = {h.get("hostname") for h in hosts_data.get("hosts") or [] if h.get("hostname")}
+    hostgroup_names = {
+        g.get("groupname") for g in hosts_data.get("hostgroups") or [] if g.get("groupname")
+    }
+    known_hosts = host_names | hostgroup_names | {"all"}
+
+    usergroup_names = {
+        g.get("groupname") for g in users_data.get("usergroups") or [] if g.get("groupname")
+    }
+    known_users = usergroup_names | {"all"}
+
+    queue_names = {q.get("name") for q in queues_data.get("queues") or [] if q.get("name")}
+    known_queues = queue_names | {"all"}
+
+    known_resources = {
+        r.get("name") for r in resources_data.get("resources") or [] if r.get("name")
+    }
+
+    for queue in queues_data.get("queues") or []:
+        qname = queue.get("name", "?")
+        _warn_undefined_membership(
+            queue.get("hosts"), known_hosts, "queues.yaml", f'queue "{qname}" hosts', warnings
+        )
+        _warn_undefined_membership(
+            queue.get("users"), known_users, "queues.yaml", f'queue "{qname}" users', warnings
+        )
+        for sched_stop in queue.get("sched_stop") or []:
+            resname = sched_stop.get("resname")
+            if resname and resname not in known_resources:
+                warnings.append(
+                    f'queues.yaml: queue "{qname}" sched_stop references '
+                    f"undefined resname {resname!r}"
+                )
+
+    for host in hosts_data.get("hosts") or []:
+        hname = host.get("hostname", "?")
+        for resource in host.get("resource") or []:
+            rname = resource.get("name")
+            if rname and rname not in known_resources:
+                warnings.append(
+                    f'hosts.yaml: host "{hname}" resource references '
+                    f"undefined resname {rname!r}"
+                )
+        for sched_stop in host.get("sched_stop") or []:
+            resname = sched_stop.get("resname")
+            if resname and resname not in known_resources:
+                warnings.append(
+                    f'hosts.yaml: host "{hname}" sched_stop references '
+                    f"undefined resname {resname!r}"
+                )
+
+    for limit in limits_data.get("limits") or []:
+        lname = limit.get("name", "?")
+        _warn_undefined_membership(
+            limit.get("users"), known_users, "limits.yaml", f'limit "{lname}" users', warnings
+        )
+        _warn_undefined_membership(
+            limit.get("queues"), known_queues, "limits.yaml", f'limit "{lname}" queues', warnings
+        )
+        _warn_undefined_membership(
+            limit.get("hosts"), known_hosts, "limits.yaml", f'limit "{lname}" hosts', warnings
+        )
+        resname = limit.get("resname")
+        if resname and resname not in known_resources:
+            warnings.append(
+                f'limits.yaml: limit "{lname}" references undefined resname {resname!r}'
+            )
+
+    return warnings

src/xceadmin/tests/test_validation.py

@@ -0,0 +1,528 @@
+"""Tests for xcecommon.validation (task 2.1).
+
+Covers Requirements 4.1, 4.2, 4.3, 4.4, 4.5, 5.5: JSON Schema validation of
+operator-authored YAML against the 7 bundled schemas, plus business-rule
+checks (reserved names, duplicates, undefined cross-references) that JSON
+Schema alone cannot express.
+
+Two fixture strategies are used deliberately:
+  - `validate_yaml()` is exercised against the *real* bundled schemas
+    (`default_schema_dir()`) and the *real* example configs under
+    `src/configs/*.yaml` — proving the actual shipped schemas accept actual
+    shipped example data, not just hand-rolled inline schema dicts.
+  - `check_reserved_names()` / `check_duplicates()` / `check_undefined_refs()`
+    are pure functions over already-validated section dicts, so they are
+    exercised with small in-memory fixtures built directly in this file —
+    no file I/O needed for business-rule logic that never touches a schema.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from xcecommon.errors import DuplicateEntryError, ReservedNameError, ValidationError
+from xcecommon.validation import (
+    check_duplicates,
+    check_reserved_names,
+    check_undefined_refs,
+    default_schema_dir,
+    validate_yaml,
+)
+
+# The real, operator-facing example configs this repo ships under
+# src/configs/ — used as realistic valid-case fixtures per design.md's
+# Testing Strategy ("full fixture... covers all 7 YAML files"). Failing
+# loud (not skipping) if this repo layout assumption is wrong is
+# intentional: a silently-skipped test would hide a real fixture gap.
+REPO_CONFIGS_DIR = Path(__file__).resolve().parent.parent.parent / "configs"
+
+ALL_SCHEMA_NAMES = [
+    "hosts",
+    "queues",
+    "resources",
+    "users",
+    "limits",
+    "xce",
+    "scaler",
+]
+
+
+def _real_config_path(schema_name: str) -> Path:
+    path = REPO_CONFIGS_DIR / f"{schema_name}.yaml"
+    assert path.is_file(), f"expected real example config at {path}"
+    return path
+
+
+# ---------------------------------------------------------------------------
+# default_schema_dir()
+# ---------------------------------------------------------------------------
+
+
+def test_default_schema_dir_resolves_to_real_bundled_schemas() -> None:
+    """default_schema_dir() must resolve to the packaged schema_files/json/
+    directory, containing all 7 bundled schema files (Requirement 5.5)."""
+    schema_dir = Path(default_schema_dir())
+
+    assert schema_dir.is_dir()
+    for schema_name in ALL_SCHEMA_NAMES:
+        schema_file = schema_dir / f"{schema_name}.schema.json"
+        assert schema_file.is_file(), f"missing bundled schema: {schema_file}"
+
+
+# ---------------------------------------------------------------------------
+# validate_yaml() — real bundled schemas against real example YAML
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("schema_name", ALL_SCHEMA_NAMES)
+def test_validate_yaml_accepts_real_example_config(schema_name: str) -> None:
+    """Every one of the 7 real src/configs/*.yaml example files validates
+    successfully against its corresponding bundled schema — proving the
+    actual shipped schema accepts actual shipped example data, not just an
+    inline hand-rolled schema dict."""
+    data = validate_yaml(str(_real_config_path(schema_name)), schema_name)
+
+    assert isinstance(data, dict)
+
+
+def test_validate_yaml_no_schema_subdirectory_needed_in_configs_dir(tmp_path: Path) -> None:
+    """Observable from tasks.md: validating a config directory with no
+    schema files present anywhere in it still succeeds — schemas are
+    bundled with xceadmin (Requirement 5.5), never read from configs_dir."""
+    configs_dir = tmp_path / "configs"
+    configs_dir.mkdir()
+    hosts_yaml = configs_dir / "hosts.yaml"
+    hosts_yaml.write_text(_real_config_path("hosts").read_text())
+
+    # No "schema/" subdirectory anywhere under configs_dir.
+    assert not any(configs_dir.rglob("*schema*"))
+
+    data = validate_yaml(str(hosts_yaml), "hosts")
+
+    assert "hosts" in data
+
+
+def test_validate_yaml_reports_file_pointer_and_constraint_on_schema_violation(
+    tmp_path: Path,
+) -> None:
+    """Requirement 4.1: a schema violation reports the file name, the
+    JSON-pointer field path, and the violated constraint, then the caller
+    can treat it as a hard failure."""
+    bad_queues = tmp_path / "queues.yaml"
+    bad_queues.write_text(
+        "queues:\n"
+        "  - name: badqueue\n"
+        "    priority: 30\n"
+        "    preemption: not-a-real-value\n"
+    )
+
+    with pytest.raises(ValidationError) as exc_info:
+        validate_yaml(str(bad_queues), "queues")
+
+    caught = exc_info.value
+    assert caught.file == str(bad_queues)
+    assert "preemption" in caught.json_pointer
+    assert caught.constraint == "enum"
+
+
+def test_validate_yaml_reports_missing_required_field(tmp_path: Path) -> None:
+    """Requirement 4.1: a missing required field (queue with no
+    "priority") is a schema violation, not silently accepted."""
+    bad_queues = tmp_path / "queues.yaml"
+    bad_queues.write_text("queues:\n  - name: no-priority\n")
+
+    with pytest.raises(ValidationError) as exc_info:
+        validate_yaml(str(bad_queues), "queues")
+
+    assert exc_info.value.constraint == "required"
+
+
+def test_validate_yaml_missing_yaml_file_is_not_this_functions_concern(tmp_path: Path) -> None:
+    """A missing individual YAML file is handled by the caller (build
+    orchestration treats it as an empty section, Requirement 1.5) — but if
+    validate_yaml is asked to open a nonexistent file directly, it must
+    fail loud with a ValidationError, not an unhandled OSError leaking out
+    of the validation boundary."""
+    missing = tmp_path / "does-not-exist.yaml"
+
+    with pytest.raises(ValidationError):
+        validate_yaml(str(missing), "queues")
+
+
+def test_validate_yaml_reports_malformed_yaml_syntax(tmp_path: Path) -> None:
+    """Invalid YAML syntax is a validation failure too, not an unhandled
+    parser exception escaping the boundary."""
+    broken = tmp_path / "queues.yaml"
+    broken.write_text("queues: [this is not: valid yaml")
+
+    with pytest.raises(ValidationError):
+        validate_yaml(str(broken), "queues")
+
+
+# ---------------------------------------------------------------------------
+# check_reserved_names() — Requirement 4.2
+# ---------------------------------------------------------------------------
+
+
+def test_check_reserved_names_rejects_reserved_queue_name() -> None:
+    sections = {
+        "queues": {"queues": [{"name": "lost_and_found", "priority": 10}]},
+    }
+
+    with pytest.raises(ReservedNameError):
+        check_reserved_names(sections)
+
+
+def test_check_reserved_names_rejects_reserved_host_name() -> None:
+    sections = {
+        "hosts": {"hosts": [{"hostname": "lost_and_found"}]},
+    }
+
+    with pytest.raises(ReservedNameError):
+        check_reserved_names(sections)
+
+
+def test_check_reserved_names_allows_ordinary_names() -> None:
+    sections = {
+        "hosts": {"hosts": [{"hostname": "worker1"}]},
+        "queues": {"queues": [{"name": "normal", "priority": 10}]},
+    }
+
+    check_reserved_names(sections)  # must not raise
+
+
+def test_check_reserved_names_ignores_missing_sections() -> None:
+    """Sections not present in the dict (file was absent, Requirement 1.5)
+    must not crash the check."""
+    check_reserved_names({})  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# check_duplicates() — Requirement 4.3
+# ---------------------------------------------------------------------------
+
+
+def test_check_duplicates_rejects_duplicate_queue_names() -> None:
+    sections = {
+        "queues": {
+            "queues": [
+                {"name": "batch", "priority": 10},
+                {"name": "batch", "priority": 20},
+            ]
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_rejects_duplicate_host_names() -> None:
+    sections = {
+        "hosts": {
+            "hosts": [{"hostname": "worker1"}, {"hostname": "worker1"}],
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_rejects_duplicate_resource_names() -> None:
+    sections = {
+        "resources": {
+            "resources": [
+                {"name": "gpu", "type": "numeric"},
+                {"name": "gpu", "type": "numeric"},
+            ]
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_rejects_duplicate_host_group_names() -> None:
+    sections = {
+        "hosts": {
+            "hosts": [],
+            "hostgroups": [
+                {"groupname": "hg1", "members": ["all"]},
+                {"groupname": "hg1", "members": ["all"]},
+            ],
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_rejects_duplicate_user_group_names() -> None:
+    sections = {
+        "users": {
+            "usergroups": [
+                {"groupname": "ug1", "members": ["u1"]},
+                {"groupname": "ug1", "members": ["u2"]},
+            ],
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_rejects_duplicate_limit_names() -> None:
+    sections = {
+        "limits": {
+            "limits": [
+                {"name": "lim1", "resname": "slots", "value": 1},
+                {"name": "lim1", "resname": "mem", "value": 2},
+            ]
+        },
+    }
+
+    with pytest.raises(DuplicateEntryError):
+        check_duplicates(sections)
+
+
+def test_check_duplicates_allows_limit_name_to_collide_with_queue_name() -> None:
+    """Requirement 4.3, scoped intentionally: a limit name is not required
+    to differ from a queue, host, or group name — duplicates are checked
+    within their own file/section only."""
+    sections = {
+        "queues": {"queues": [{"name": "shared_name", "priority": 10}]},
+        "limits": {
+            "limits": [{"name": "shared_name", "resname": "slots", "value": 1}]
+        },
+    }
+
+    check_duplicates(sections)  # must not raise
+
+
+def test_check_duplicates_allows_host_group_and_user_group_same_name() -> None:
+    """Host groups and user groups are separate files/sections — a name
+    used in both is not a duplicate."""
+    sections = {
+        "hosts": {"hostgroups": [{"groupname": "shared", "members": ["all"]}]},
+        "users": {"usergroups": [{"groupname": "shared", "members": ["u1"]}]},
+    }
+
+    check_duplicates(sections)  # must not raise
+
+
+def test_check_duplicates_ignores_missing_sections() -> None:
+    check_duplicates({})  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# check_undefined_refs() — Requirement 4.4, 4.5
+# ---------------------------------------------------------------------------
+
+
+def test_check_undefined_refs_never_raises_on_missing_sections() -> None:
+    """This is the one intentionally non-fatal check (4.4) — it must never
+    raise, even with an empty/partial sections dict."""
+    warnings = check_undefined_refs({})
+
+    assert warnings == []
+
+
+def test_check_undefined_refs_warns_on_undefined_host_group_in_queue() -> None:
+    """Requirement 4.4 example: a queue references a host group that is not
+    defined."""
+    sections = {
+        "hosts": {"hosts": [], "hostgroups": [{"groupname": "hg1", "members": ["all"]}]},
+        "queues": {
+            "queues": [
+                {
+                    "name": "batch",
+                    "priority": 10,
+                    "hosts": {"members": ["hg_not_defined"]},
+                }
+            ]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert any("hg_not_defined" in w for w in warnings)
+
+
+def test_check_undefined_refs_does_not_warn_on_defined_host_group_in_queue() -> None:
+    sections = {
+        "hosts": {"hosts": [], "hostgroups": [{"groupname": "hg1", "members": ["all"]}]},
+        "queues": {
+            "queues": [{"name": "batch", "priority": 10, "hosts": {"members": ["hg1"]}}]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert warnings == []
+
+
+def test_check_undefined_refs_does_not_warn_on_literal_all() -> None:
+    sections = {
+        "hosts": {"hosts": [], "hostgroups": []},
+        "queues": {
+            "queues": [{"name": "batch", "priority": 10, "hosts": {"members": ["all"]}}]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert warnings == []
+
+
+def test_check_undefined_refs_warns_on_undefined_user_group_in_queue() -> None:
+    sections = {
+        "users": {"usergroups": [{"groupname": "ug1", "members": ["u1"]}]},
+        "queues": {
+            "queues": [
+                {
+                    "name": "batch",
+                    "priority": 10,
+                    "users": {"members": ["ug_not_defined"]},
+                }
+            ]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert any("ug_not_defined" in w for w in warnings)
+
+
+def test_check_undefined_refs_warns_on_undefined_resname_in_limit() -> None:
+    """Requirement 4.4 example: a limit references a resname not defined in
+    resources.yaml."""
+    sections = {
+        "resources": {"resources": [{"name": "slots", "type": "numeric"}]},
+        "limits": {
+            "limits": [{"name": "lim1", "resname": "not_a_real_resource", "value": 1}]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert any("not_a_real_resource" in w for w in warnings)
+
+
+def test_check_undefined_refs_warns_on_undefined_queue_in_limit() -> None:
+    sections = {
+        "queues": {"queues": [{"name": "batch", "priority": 10}]},
+        "limits": {
+            "limits": [
+                {
+                    "name": "lim1",
+                    "resname": "slots",
+                    "value": 1,
+                    "queues": {"members": ["not_a_real_queue"]},
+                }
+            ]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert any("not_a_real_queue" in w for w in warnings)
+
+
+def test_check_undefined_refs_built_in_resource_name_not_implicitly_exempt() -> None:
+    """Requirement 4.5: resources.yaml is the exhaustive resource catalog —
+    built-in/dynamic resource names (slots, mem, and the 8
+    add_hostattrs_resvals() host attributes) are NOT treated as implicit
+    built-ins exempt from the catalog. A limit referencing "slots" when
+    resources.yaml does not declare it must still warn."""
+    sections = {
+        "resources": {"resources": [{"name": "mem", "type": "numeric"}]},  # no "slots"
+        "limits": {
+            "limits": [{"name": "lim1", "resname": "slots", "value": 1}]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert any("slots" in w for w in warnings)
+
+
+def test_check_undefined_refs_all_ten_built_in_names_pass_when_declared() -> None:
+    """The converse of the above: once resources.yaml actually declares
+    all the built-in/dynamic names (as the real example resources.yaml
+    does), referencing them from a limit produces no warning."""
+    built_in_names = [
+        "slots",
+        "mem",
+        "type",
+        "model",
+        "cpuf",
+        "maxcpus",
+        "maxmem",
+        "maxswap",
+        "maxtmp",
+        "ndisks",
+    ]
+    sections = {
+        "resources": {
+            "resources": [{"name": name, "type": "numeric"} for name in built_in_names]
+        },
+        "limits": {
+            "limits": [
+                {"name": f"lim_{name}", "resname": name, "value": 1}
+                for name in built_in_names
+            ]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)
+
+    assert warnings == []
+
+
+def test_check_undefined_refs_does_not_block_on_multiple_violations() -> None:
+    """4.4: undefined references warn but never raise, regardless of how
+    many violations are found."""
+    sections = {
+        "queues": {
+            "queues": [
+                {
+                    "name": "batch",
+                    "priority": 10,
+                    "hosts": {"members": ["ghost_group"]},
+                    "users": {"members": ["ghost_users"]},
+                }
+            ]
+        },
+        "limits": {
+            "limits": [{"name": "lim1", "resname": "ghost_resource", "value": 1}]
+        },
+    }
+
+    warnings = check_undefined_refs(sections)  # must not raise
+
+    assert len(warnings) >= 3
+
+
+# ---------------------------------------------------------------------------
+# Integration: real example configs, parsed via validate_yaml, produce no
+# reserved-name / duplicate / undefined-ref surprises against each other.
+# ---------------------------------------------------------------------------
+
+
+def test_business_rule_checks_run_clean_against_all_real_example_sections() -> None:
+    """Load every real src/configs/*.yaml section through validate_yaml,
+    then run all three business-rule checks against the combined sections
+    dict — check_reserved_names/check_duplicates must not raise (the real
+    fixtures use no reserved or duplicate names), and check_undefined_refs
+    must return a list without raising."""
+    sections = {
+        schema_name: validate_yaml(str(_real_config_path(schema_name)), schema_name)
+        for schema_name in ALL_SCHEMA_NAMES
+    }
+
+    check_reserved_names(sections)
+    check_duplicates(sections)
+    warnings = check_undefined_refs(sections)
+
+    assert isinstance(warnings, list)