feat(xceadmin): add 'config dump' command and size-string thresholds

Adds `xceadmin config dump CLUSTER_BIN_PATH`, the reverse of `config
build` -- decodes a serialized ClusterConfig back to human-readable text
so a built config can be inspected without a throwaway pycapnp script.
Motivated directly by how bug-1556 (hosts.yaml's max_slot: -1 auto-detect
sentinel silently becoming mxj=0) was actually found: by hand-decoding a
live cluster.bin. The dump renders the same sentinel as
"max_slot=-1 (auto-detect from CPU count)", not the raw uint32.

Also adds parse_size_or_number() to xcecommon.mapping so a queue/host's
sched_stop threshold on a byte-valued resource (maxmem/mem/tmp/maxswap/
maxtmp) can be written as a human-friendly size string ("500MB", "2GB")
instead of a raw byte count; a plain number still passes through
unchanged for non-byte resources like load averages. hosts.schema.json
and queues.schema.json's sched_stop schema gains a matching SizeOrNumber
type. dump_cluster_config() humanizes the same byte-valued thresholds
back on the way out (_format_bytes_human).

New ClusterBinFileError covers every local cluster.bin read/decode
failure mode without letting a raw OSError/capnp.KjException escape the
CLI. RESOURCE_VALTYPE/RESOURCE_ORDTYPE/RESOURCE_FLAG_* made public in
mapping.py so dump_cluster_config can reuse the same int<->name tables
build_cluster_config already uses, keeping both directions in sync by
construction.

256 xceadmin tests pass (incl. new test_config_dump.py and the
parse_size_or_number/SizeOrNumber cases in test_mapping.py/test_cli.py),
verified via `xce-build.sh xceadmin`'s full pinned out-of-tree install
+ pytest run before this commit.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/configs/schema/hosts.schema.json

@@ -68,8 +68,8 @@
       "additionalProperties": false,
       "properties": {
         "resname": { "type": "string" },
-        "sched": { "type": "number" },
-        "stop": { "type": "number" }
+        "sched": { "$ref": "#/$defs/SizeOrNumber" },
+        "stop": { "$ref": "#/$defs/SizeOrNumber" }
       },
       "required": ["resname"],
       "anyOf": [
@@ -77,6 +77,13 @@
         { "required": ["stop"] }
       ]
     },
+    "SizeOrNumber": {
+      "description": "A plain number (the resource's own unit, e.g. a load average), or a size string with a unit suffix (e.g. \"500MB\", \"2GB\") for a byte-valued resource -- converted to bytes when the config is parsed.",
+      "oneOf": [
+        { "type": "number" },
+        { "type": "string", "pattern": "^[+-]?[0-9]+(\\.[0-9]+)?\\s*[A-Za-z]+$" }
+      ]
+    },
     "HostGroup": {
       "type": "object",
       "additionalProperties": false,

src/configs/schema/queues.schema.json

@@ -34,8 +34,8 @@
       "additionalProperties": false,
       "properties": {
         "resname": { "type": "string" },
-        "sched": { "type": "number" },
-        "stop": { "type": "number" }
+        "sched": { "$ref": "#/$defs/SizeOrNumber" },
+        "stop": { "$ref": "#/$defs/SizeOrNumber" }
       },
       "required": ["resname"],
       "anyOf": [
@@ -43,6 +43,13 @@
         { "required": ["stop"] }
       ]
     },
+    "SizeOrNumber": {
+      "description": "A plain number (the resource's own unit, e.g. a load average), or a size string with a unit suffix (e.g. \"500MB\", \"2GB\") for a byte-valued resource -- converted to bytes when the config is parsed.",
+      "oneOf": [
+        { "type": "number" },
+        { "type": "string", "pattern": "^[+-]?[0-9]+(\\.[0-9]+)?\\s*[A-Za-z]+$" }
+      ]
+    },
     "FairShare": {
       "type": "object",
       "additionalProperties": false,

src/xceadmin/src/xceadmin/build.py

@@ -22,7 +22,11 @@ import hashlib
 import sys
 from pathlib import Path
 
+import capnp
+
 from xcecommon import mapping, validation
+from xcecommon.errors import ClusterBinFileError
+from xcecommon.schema import load_schema
 
 #: Schema-valid YAML fields that intentionally have no ClusterConfig
 #: destination (design.md's ``xcecommon.mapping`` module docstring,
@@ -137,3 +141,235 @@ def write_output(data: bytes, output_path: str) -> str:
     else:
         Path(output_path).write_bytes(data)
     return hashlib.sha256(data).hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# ``config dump``: reverse of build_cluster_config -- decode a cluster.bin
+# back to human-readable text, so a built config can be inspected without a
+# throwaway pycapnp script (the exact pain point that motivated this: the
+# hosts.yaml `max_slot: -1` auto-detect sentinel silently becoming mxj=0 was
+# only found by hand-decoding a live cluster.bin).
+#
+# Plain text, not YAML: unlike the .replay dump (xceadmin.scheduler.dump),
+# ClusterConfig is a single well-typed top-level struct, not a stream of
+# heterogeneous messages -- no shape-discrimination is needed, only
+# `ClusterConfig.from_bytes(data)` and a field walk. Output deliberately
+# does NOT try to look like hosts.yaml/resources.yaml source: the source
+# YAML may carry comments and fields with no ClusterConfig destination
+# (module docstring, "Deliberately Unmapped Fields"), so a dump can never
+# be diffed 1:1 against it anyway -- a flat, readable listing of what's
+# actually on the wire is more honest than a fake YAML reconstruction.
+# ---------------------------------------------------------------------------
+
+#: Resource names whose values are byte counts (bytes chosen as the sole
+#: internal unit -- see resources.yaml's field descriptions) -- mirrors
+#: controller/testsuite/test.c's ``is_bytes_valued_resource()`` name list.
+#: Kept as a small explicit set here too rather than a shared cross-language
+#: constant: the C and Python copies each serve a single CLI's display
+#: logic, and duplicating five literal names costs far less than the
+#: machinery a shared source would need.
+_BYTES_VALUED_RESOURCES = frozenset({"maxmem", "mem", "tmp", "maxswap", "maxtmp"})
+
+_RESOURCE_VALTYPE_NAMES = {v: k for k, v in mapping.RESOURCE_VALTYPE.items()}
+_RESOURCE_ORDTYPE_NAMES = {v: k for k, v in mapping.RESOURCE_ORDTYPE.items()}
+
+
+def _format_bytes_human(n: float) -> str:
+    """Adaptive B/KB/MB/GB/TB/PB formatter -- mirrors test.c's
+    ``format_bytes_human`` so the same byte count reads the same way in
+    both tools.
+    """
+    value = float(n)
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if abs(value) < 1024.0:
+            return f"{value:.2f} {unit}"
+        value /= 1024.0
+    return f"{value:.2f} PB"
+
+
+def _format_resource(r) -> str:
+    valtype = _RESOURCE_VALTYPE_NAMES.get(r.valtype, r.valtype)
+    ordtype = _RESOURCE_ORDTYPE_NAMES.get(r.ordtype, r.ordtype)
+    consumable = bool(r.flags & mapping.RESOURCE_FLAG_CONSUMABLE)
+    release = bool(r.flags & mapping.RESOURCE_FLAG_RELEASE)
+    parts = [f"name={r.name}", f"type={valtype}", f"order={ordtype}"]
+    if r.desc:
+        parts.append(f"desc={r.desc!r}")
+    if r.interval:
+        parts.append(f"interval={r.interval}")
+    parts.append(f"consumable={consumable}")
+    parts.append(f"release={release}")
+    return "  " + " ".join(parts)
+
+
+def _format_group(g, label: str) -> str:
+    members = list(g.members)
+    exmembers = list(g.exmembers)
+    name = f"{label}={g.name} " if g.name else f"{label} "
+    return f"{name}members={members} exmembers={exmembers}"
+
+
+def _format_time_window_seg(seg) -> str:
+    o, c = seg.open, seg.close
+    return f"tz={seg.timezone} open={o.day}/{o.hour:02d}:{o.min:02d} close={c.day}/{c.hour:02d}:{c.min:02d}"
+
+
+def _format_time_window(tw, label: str) -> str:
+    segs = list(tw.segs)
+    if not segs:
+        return f"{label}: (none)"
+    return f"{label}: [" + "; ".join(_format_time_window_seg(s) for s in segs) + "]"
+
+
+def _format_threshold(t) -> str:
+    if t.name in _BYTES_VALUED_RESOURCES:
+        return f"name={t.name} sched={_format_bytes_human(t.sched)} stop={_format_bytes_human(t.stop)}"
+    return f"name={t.name} sched={t.sched} stop={t.stop}"
+
+
+def _format_queue(q) -> list[str]:
+    lines = [f"  === queue: {q.name} ==="]
+    if q.priority:
+        lines.append(f"    priority={q.priority}")
+    if q.resreq:
+        lines.append(f"    resreq={q.resreq!r}")
+    if q.preExec:
+        lines.append(f"    pre_exec={q.preExec!r}")
+    if q.postExec:
+        lines.append(f"    post_exec={q.postExec!r}")
+    lines.append(f"    {_format_group(q.users, 'users')}")
+    lines.append(f"    {_format_group(q.hosts, 'hosts')}")
+    for t in q.thresholds:
+        lines.append(f"    threshold: {_format_threshold(t)}")
+    lines.append(f"    {_format_time_window(q.dispatchWindow, 'dispatch_window')}")
+    lines.append(f"    {_format_time_window(q.runWindow, 'run_window')}")
+    return lines
+
+
+def _format_host_config(h) -> list[str]:
+    lines = [f"  === host: {h.name} ==="]
+    if h.model:
+        lines.append(f"    model={h.model}")
+    if h.type:
+        lines.append(f"    type={h.type}")
+    if h.mxj == mapping.HOSTCONFIG_MXJ_AUTO_DETECT:
+        lines.append("    max_slot=-1 (auto-detect from CPU count)")
+    else:
+        lines.append(f"    max_slot={h.mxj}")
+    for cv in h.resources:
+        lines.append(f"    resource: {cv.name}={cv.value}")
+    for t in h.thresholds:
+        lines.append(f"    threshold: {_format_threshold(t)}")
+    lines.append(f"    {_format_time_window(h.dispatchwindows, 'dispatch_window')}")
+    return lines
+
+
+def _format_scaler(s) -> str:
+    parts = [f"name={s.name}", f"type={s.type}", f"script_path={s.scriptPath!r}"]
+    if s.confPath:
+        parts.append(f"conf_path={s.confPath!r}")
+    if s.scriptOption:
+        parts.append(f"script_option={s.scriptOption!r}")
+    return "  " + " ".join(parts)
+
+
+def _format_limit(l) -> list[str]:  # noqa: E741 (matches mapping.py's own `l` for a limits.yaml entry)
+    val = l.val
+    value = _format_bytes_human(val.value) if val.resname in _BYTES_VALUED_RESOURCES else val.value
+    lines = [
+        f"  === limit: {l.name} ===",
+        f"    val: resname={val.resname} value={value} percentage={val.percentage}",
+        f"    per_user={l.perUser} per_queue={l.perQueue} per_project={l.perProject} per_host={l.perHost}",
+        f"    {_format_group(l.users, 'users')}",
+        f"    {_format_group(l.queues, 'queues')}",
+        f"    {_format_group(l.projects, 'projects')}",
+        f"    {_format_group(l.hosts, 'hosts')}",
+    ]
+    return lines
+
+
+def _format_cluster_config(cfg) -> str:
+    lines = [
+        "=== ClusterConfig ===",
+        f"clusterName: {cfg.clusterName}",
+        f"heartbeatInterval: {cfg.heartbeatInterval}",
+        f"inactiveHeartbeatCount: {cfg.inactiveHeartbeatCount}",
+        f"fetchJobInterval: {cfg.fetchJobInterval}",
+        f"netioTimeout: {cfg.netioTimeout}",
+        f"configGeneration: {cfg.configGeneration}",
+    ]
+
+    lines.append("")
+    lines.append(f"=== resources ({len(cfg.resources)}) ===")
+    for r in cfg.resources:
+        lines.append(_format_resource(r))
+
+    lines.append("")
+    lines.append(f"=== queues ({len(cfg.queues)}) ===")
+    for q in cfg.queues:
+        lines.extend(_format_queue(q))
+
+    lines.append("")
+    lines.append(f"=== ugroups ({len(cfg.ugroups)}) ===")
+    for g in cfg.ugroups:
+        lines.append("  " + _format_group(g, "group"))
+
+    lines.append("")
+    lines.append(f"=== hgroups ({len(cfg.hgroups)}) ===")
+    for g in cfg.hgroups:
+        lines.append("  " + _format_group(g, "group"))
+
+    lines.append("")
+    lines.append(f"=== hostConfigs ({len(cfg.hostConfigs)}) ===")
+    for h in cfg.hostConfigs:
+        lines.extend(_format_host_config(h))
+
+    lines.append("")
+    lines.append(f"=== hproviders ({len(cfg.hproviders)}) ===")
+    for s in cfg.hproviders:
+        lines.append(_format_scaler(s))
+
+    lines.append("")
+    lines.append(f"=== limits ({len(cfg.limits)}) ===")
+    for l in cfg.limits:  # noqa: E741
+        lines.extend(_format_limit(l))
+
+    return "\n".join(lines) + "\n"
+
+
+def dump_cluster_config(cluster_bin_path: str) -> str:
+    """Decode CLUSTER_BIN_PATH (a serialized ``ClusterConfig``, e.g.
+    ``config build``'s output) to human-readable text.
+
+    Returns the full text as one string (mirrors ``xceadmin.scheduler.dump``'s
+    convention) -- CLI printing is ``xceadmin.cli``'s job, not this
+    function's.
+
+    Raises :class:`~xcecommon.errors.ClusterBinFileError` if the path can't
+    be opened/read, the file is empty, or the bytes don't decode as a
+    ``ClusterConfig`` -- never lets a raw ``OSError``/``capnp.KjException``
+    escape.
+    """
+    try:
+        with open(cluster_bin_path, "rb") as f:
+            data = f.read()
+    except OSError as exc:
+        raise ClusterBinFileError(
+            f"cannot read {cluster_bin_path!r}: {exc}", path=cluster_bin_path, reason=str(exc)
+        ) from exc
+
+    if not data:
+        raise ClusterBinFileError(
+            f"{cluster_bin_path!r} is empty", path=cluster_bin_path, reason="empty file"
+        )
+
+    mod = load_schema()
+    try:
+        with mod.ClusterConfig.from_bytes(data) as cfg:
+            return _format_cluster_config(cfg)
+    except (capnp.KjException, ValueError) as exc:
+        raise ClusterBinFileError(
+            f"{cluster_bin_path!r} is not a valid ClusterConfig: {exc}",
+            path=cluster_bin_path,
+            reason=str(exc),
+        ) from exc

src/xceadmin/src/xceadmin/cli.py

@@ -33,7 +33,7 @@ from datetime import datetime, timezone
 import click
 
 from xcecommon.errors import SchedAdminError, XceAdminError
-from xceadmin.build import build_cluster_config, write_output
+from xceadmin.build import build_cluster_config, dump_cluster_config, write_output
 from xceadmin.push import probe_config, push_config
 from xceadmin.scheduler import RecordResult, dump, record_start, record_status, record_stop
 
@@ -99,6 +99,17 @@ def config_push(cluster_bin_path: str, controller: str, dry_run: bool) -> None:
         raise click.ClickException("cancelled") from None
 
 
+@config.command("dump")
+@click.argument("cluster_bin_path")
+def config_dump(cluster_bin_path: str) -> None:
+    """Decode CLUSTER_BIN_PATH (a ClusterConfig binary) to human-readable text."""
+    try:
+        text = dump_cluster_config(cluster_bin_path)
+    except XceAdminError as e:
+        raise click.ClickException(e.message) from e
+    click.echo(text, nl=False)  # dump_cluster_config() already ends its own text with "\n".
+
+
 @cli.group()
 def scheduler() -> None:
     """Control and inspect the scheduler's runtime event recording (5.10)."""

src/xceadmin/src/xcecommon/errors.py

@@ -151,6 +151,24 @@ class ReplayFileError(XceAdminError):
         self.reason = reason
 
 
+class ClusterBinFileError(XceAdminError):
+    """A ``cluster.bin`` file (a serialized ``ClusterConfig``) could not be
+    read or decoded (``xceadmin.build.dump_cluster_config``).
+
+    Covers every local-file failure mode: the path doesn't exist / isn't
+    readable, the file is empty, or the bytes don't decode as a
+    ``ClusterConfig`` at all (wrong file, truncated, or corrupt) --
+    ``ClusterConfig`` is a single well-typed top-level struct, so unlike
+    :class:`ReplayFileError` there is no message-stream framing or
+    shape-collision concern here, only "does this decode."
+    """
+
+    def __init__(self, message: str, path: str, reason: str) -> None:
+        super().__init__(message)
+        self.path = path
+        self.reason = reason
+
+
 class SchedAdminError(XceAdminError):
     """A ``SchedAdminRequest`` (scheduler recording control) round-tripped
     successfully but the reply itself reports failure --

src/xceadmin/src/xcecommon/mapping.py

@@ -69,9 +69,11 @@ reason -- never a silent drop.
     ``include/xce/types.h`` (``XCE_VAL_*``, ``xce_order_t``,
     ``XCE_RESF_CONSUMABLE``/``XCE_RESF_RELEASE``) and is actively read at
     runtime (e.g. ``scheduler.c`` copies these three fields straight out
-    of the parsed resource). See :data:`_RESOURCE_VALTYPE`,
-    :data:`_RESOURCE_ORDTYPE`, and the flag constants below for the exact
-    values. ``idx`` and ``lastReport`` remain unset: ``idx`` has no YAML source at all
+    of the parsed resource). See :data:`RESOURCE_VALTYPE`,
+    :data:`RESOURCE_ORDTYPE`, and the flag constants below for the exact
+    values (also reused by ``xceadmin.build.dump_cluster_config`` to render
+    these int-encoded fields back to their YAML names). ``idx`` and
+    ``lastReport`` remain unset: ``idx`` has no YAML source at all
     (controller/runtime-assigned), and ``lastReport`` is a runtime
     timestamp field (epoch seconds of last report) -- excluded by
     Requirement 9.2's "no timestamp" rule. ``resourcemap`` has no
@@ -131,6 +133,7 @@ reason -- never a silent drop.
 
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING, Literal, TypedDict
 
 from xcecommon.schema import load_schema
@@ -182,13 +185,13 @@ _XCE_SCALAR_FIELDS: tuple[tuple[str, str], ...] = (
 # resources.yaml ``type``/``order`` -> Resource.valtype/ordtype, mirroring
 # include/xce/types.h's XCE_VAL_*/xce_order_t (both plain capnp Int32
 # fields, not capnp enums -- see controller.capnp:203-205).
-_RESOURCE_VALTYPE: dict[str, int] = {"string": 0, "numeric": 1, "boolean": 2}
-_RESOURCE_ORDTYPE: dict[str, int] = {"increase": 0, "decrease": 1, "na": 2}
+RESOURCE_VALTYPE: dict[str, int] = {"string": 0, "numeric": 1, "boolean": 2}
+RESOURCE_ORDTYPE: dict[str, int] = {"increase": 0, "decrease": 1, "na": 2}
 
 # resources.yaml ``consumable``/``release`` -> Resource.flags bits, mirroring
 # include/xce/types.h's XCE_RESF_CONSUMABLE/XCE_RESF_RELEASE.
-_RESOURCE_FLAG_CONSUMABLE = 0x80
-_RESOURCE_FLAG_RELEASE = 0x20
+RESOURCE_FLAG_CONSUMABLE = 0x80
+RESOURCE_FLAG_RELEASE = 0x20
 
 
 def new_cluster_config_builder() -> "capnp._DynamicStructBuilder":
@@ -243,12 +246,54 @@ def _build_time_window(entries: list[dict] | None, target: "capnp._DynamicStruct
         seg.close.min = end_min
 
 
+_SIZE_UNIT_MULTIPLIERS: dict[str, int] = {
+    "b": 1,
+    "kb": 1024,
+    "mb": 1024**2,
+    "gb": 1024**3,
+    "tb": 1024**4,
+}
+_SIZE_STRING_RE = re.compile(r"^\s*([+-]?[0-9]+(?:\.[0-9]+)?)\s*([A-Za-z]+)\s*$")
+
+
+def parse_size_or_number(value: int | float | str) -> float:
+    """Parse a ``sched_stop`` threshold value that may be either a plain
+    number (the target resource's own unit -- e.g. a load average like
+    ``3.5``) or a size string with a unit suffix (``"500MB"``, ``"2GB"``)
+    for a byte-valued resource (maxmem/mem/tmp/maxswap/maxtmp -- see
+    controller/testsuite/test.c's ``is_bytes_valued_resource``), converted
+    to bytes. The caller doesn't need to know in advance which kind a
+    given ``resname`` is -- a plain number always passes through
+    unchanged, and only an explicit unit-suffixed string is converted, so
+    an operator opts in per-value by writing (or not writing) a suffix.
+    """
+    if isinstance(value, bool):
+        raise TypeError(f"threshold value must be a number or a size string, got {value!r}")
+    if isinstance(value, (int, float)):
+        return float(value)
+    if not isinstance(value, str):
+        raise TypeError(f"threshold value must be a number or a size string, got {type(value)!r}")
+
+    match = _SIZE_STRING_RE.match(value)
+    if not match:
+        raise ValueError(f'cannot parse {value!r} as a number or a size string (e.g. "500MB")')
+
+    number_part, unit_part = match.groups()
+    unit_key = unit_part.lower()
+    if unit_key not in _SIZE_UNIT_MULTIPLIERS:
+        recognized = "/".join(u.upper() for u in _SIZE_UNIT_MULTIPLIERS)
+        raise ValueError(f"unrecognized size unit {unit_part!r} in {value!r} -- expected one of {recognized}")
+
+    return float(number_part) * _SIZE_UNIT_MULTIPLIERS[unit_key]
+
+
 def _build_threshold_list(
     entries: list[dict] | None, target: "capnp._DynamicStructBuilder", field_name: str
 ) -> None:
     """Populate a ``List(Threshold)`` field from a YAML ``sched_stop`` list
     (``{resname, sched, stop}`` -> ``Threshold{name, sched, stop}``),
-    preserving order.
+    preserving order. ``sched``/``stop`` accept a plain number or a size
+    string (see :func:`parse_size_or_number`).
     """
     if not entries:
         return
@@ -256,9 +301,9 @@ def _build_threshold_list(
     for threshold, entry in zip(thresholds, entries):
         threshold.name = entry["resname"]
         if "sched" in entry:
-            threshold.sched = entry["sched"]
+            threshold.sched = parse_size_or_number(entry["sched"])
         if "stop" in entry:
-            threshold.stop = entry["stop"]
+            threshold.stop = parse_size_or_number(entry["stop"])
 
 
 def _build_config_var_list(entries: list[dict] | None, target: "capnp._DynamicStructBuilder") -> None:
@@ -315,14 +360,14 @@ def _build_resource(item: dict, target: "capnp._DynamicStructBuilder") -> None:
     if "interval" in item:
         target.interval = item["interval"]
     if "type" in item:
-        target.valtype = _RESOURCE_VALTYPE[item["type"]]
+        target.valtype = RESOURCE_VALTYPE[item["type"]]
     if "order" in item:
-        target.ordtype = _RESOURCE_ORDTYPE[item["order"]]
+        target.ordtype = RESOURCE_ORDTYPE[item["order"]]
     flags = 0
     if item.get("consumable"):
-        flags |= _RESOURCE_FLAG_CONSUMABLE
+        flags |= RESOURCE_FLAG_CONSUMABLE
     if item.get("release"):
-        flags |= _RESOURCE_FLAG_RELEASE
+        flags |= RESOURCE_FLAG_RELEASE
     target.flags = flags
 
 

src/xceadmin/tests/test_cli.py

@@ -51,6 +51,7 @@ def test_cli_config_help_shows_build_and_push() -> None:
     assert result.exit_code == 0
     assert "build" in result.output
     assert "push" in result.output
+    assert "dump" in result.output
 
 
 def test_cli_group_invoked_with_no_args_does_not_traceback() -> None:
@@ -132,6 +133,39 @@ def test_config_build_missing_configs_dir_does_not_raw_traceback(tmp_path):
     assert result.exit_code == 0, "a nonexistent dir behaves like one with zero known files present (1.5)"
 
 
+# ---------------------------------------------------------------------------
+# config dump
+# ---------------------------------------------------------------------------
+
+
+def test_config_dump_prints_decoded_text(tmp_path):
+    (tmp_path / "xce.yaml").write_text(
+        yaml.safe_dump(
+            {"cluster_name": "eda1", "heartbeat_interval": 30, "inactive_heartbeat_count": 3}
+        )
+    )
+    bin_path = tmp_path / "cluster.bin"
+    runner = CliRunner()
+    build_result = runner.invoke(cli, ["config", "build", str(tmp_path), "-o", str(bin_path)])
+    assert build_result.exit_code == 0, build_result.output
+
+    result = runner.invoke(cli, ["config", "dump", str(bin_path)])
+
+    assert result.exit_code == 0, result.output
+    assert "=== ClusterConfig ===" in result.output
+    assert "clusterName: eda1" in result.output
+
+
+def test_config_dump_missing_file_exits_nonzero_with_message(tmp_path):
+    missing = tmp_path / "does-not-exist.bin"
+    runner = CliRunner()
+    result = runner.invoke(cli, ["config", "dump", str(missing)])
+
+    assert result.exit_code != 0
+    assert not isinstance(result.exception, KeyError | AttributeError | TypeError)
+    assert result.output.strip() != ""
+
+
 # ---------------------------------------------------------------------------
 # config push / config push --dry-run (mock controller)
 # ---------------------------------------------------------------------------

src/xceadmin/tests/test_config_dump.py

@@ -0,0 +1,128 @@
+"""Tests for ``xceadmin.build.dump_cluster_config`` -- the reverse of
+``build_cluster_config`` (decode a ``cluster.bin`` back to human-readable
+text). Uses the same ``tests/fixtures/full`` real-config fixture as
+test_config_build.py rather than a second golden text fixture: the
+assertions below check for specific substrings (mirrors
+test_scheduler.py's ``dump()`` test style), not a byte-exact text match,
+so formatting can evolve without a brittle golden file.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import yaml
+
+from xcecommon.errors import ClusterBinFileError
+from xceadmin import build
+
+FIXTURES_DIR = Path(__file__).parent / "fixtures"
+
+
+@pytest.fixture
+def full_cluster_bin(tmp_path):
+    data = build.build_cluster_config(str(FIXTURES_DIR / "full" / "configs"))
+    path = tmp_path / "cluster.bin"
+    path.write_bytes(data)
+    return path
+
+
+# ---------------------------------------------------------------------------
+# Section content
+# ---------------------------------------------------------------------------
+
+
+def test_dump_lists_cluster_scalars(full_cluster_bin):
+    text = build.dump_cluster_config(str(full_cluster_bin))
+    assert "=== ClusterConfig ===" in text
+    assert "clusterName:" in text
+    assert "heartbeatInterval:" in text
+
+
+def test_dump_lists_each_resource_with_decoded_type_and_order(full_cluster_bin):
+    text = build.dump_cluster_config(str(full_cluster_bin))
+    assert "=== resources (" in text
+    # "mem" is numeric/decrease/not consumable -- valtype/ordtype/flags int
+    # codes must render back to their YAML names, not raw ints.
+    assert "name=mem type=numeric order=decrease" in text
+    assert "consumable=False" in text
+
+
+def test_dump_renders_mxj_auto_detect_sentinel_not_raw_uint32(full_cluster_bin):
+    """The exact bug that motivated this feature: hosts.yaml's `default`
+    host has `max_slot: -1`, which wire-encodes as HOSTCONFIG_MXJ_AUTO_DETECT
+    (0xFFFFFFFF). The dump must show the human meaning, not the raw UInt32.
+    """
+    text = build.dump_cluster_config(str(full_cluster_bin))
+    assert "=== host: default ===" in text
+    assert "max_slot=-1 (auto-detect from CPU count)" in text
+    # worker2 has an explicit max_slot: 256 -- shown as a plain number.
+    assert "=== host: worker2 ===" in text
+    assert "max_slot=256" in text
+
+
+def test_dump_lists_queues_groups_scalers_and_limits(full_cluster_bin):
+    text = build.dump_cluster_config(str(full_cluster_bin))
+    assert "=== queues (" in text
+    assert "=== ugroups (" in text
+    assert "=== hgroups (" in text
+    assert "=== hproviders (" in text
+    assert "=== limits (" in text
+    assert "=== limit: ug1_slots ===" in text
+
+
+def test_dump_output_ends_with_trailing_newline(full_cluster_bin):
+    text = build.dump_cluster_config(str(full_cluster_bin))
+    assert text.endswith("\n")
+    assert not text.endswith("\n\n")
+
+
+def test_dump_humanizes_byte_valued_threshold(tmp_path):
+    """A host's sched_stop threshold on a byte-valued resource (maxmem) is
+    shown as a human-readable size, not a raw float byte count -- mirrors
+    the mem/tmp/maxswap/maxtmp/maxmem unit-consistency work this session.
+    """
+    (tmp_path / "hosts.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "hosts": [
+                    {
+                        "hostname": "h1",
+                        "sched_stop": [{"resname": "maxmem", "sched": "500MB", "stop": "1GB"}],
+                    }
+                ]
+            }
+        )
+    )
+    data = build.build_cluster_config(str(tmp_path))
+    bin_path = tmp_path / "cluster.bin"
+    bin_path.write_bytes(data)
+
+    text = build.dump_cluster_config(str(bin_path))
+    assert "threshold: name=maxmem sched=500.00 MB stop=1.00 GB" in text
+
+
+# ---------------------------------------------------------------------------
+# Error handling
+# ---------------------------------------------------------------------------
+
+
+def test_dump_missing_file_raises_cluster_bin_file_error(tmp_path):
+    missing = tmp_path / "does-not-exist.bin"
+    with pytest.raises(ClusterBinFileError):
+        build.dump_cluster_config(str(missing))
+
+
+def test_dump_empty_file_raises_cluster_bin_file_error(tmp_path):
+    empty = tmp_path / "empty.bin"
+    empty.write_bytes(b"")
+    with pytest.raises(ClusterBinFileError):
+        build.dump_cluster_config(str(empty))
+
+
+def test_dump_garbage_bytes_raises_cluster_bin_file_error_not_crash(tmp_path):
+    garbage = tmp_path / "garbage.bin"
+    garbage.write_bytes(b"not a cluster config, just noise")
+    with pytest.raises(ClusterBinFileError):
+        build.dump_cluster_config(str(garbage))

src/xceadmin/tests/test_mapping.py

@@ -293,6 +293,61 @@ def test_map_list_section_host_config_sets_mxj_auto_detect_sentinel_when_max_slo
     assert mapping.HOSTCONFIG_MXJ_AUTO_DETECT != 0
 
 
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (3.5, 3.5),
+        (2000, 2000.0),
+        (0, 0.0),
+        ("500MB", 500 * 1024**2),
+        ("2GB", 2 * 1024**3),
+        ("1.5GB", 1.5 * 1024**3),
+        ("500 MB", 500 * 1024**2),  # space between number and unit
+        ("500mb", 500 * 1024**2),  # lowercase unit
+        ("1024B", 1024.0),
+        ("-1KB", -1024.0),
+    ],
+)
+def test_parse_size_or_number_accepts_plain_numbers_and_size_strings(value, expected):
+    assert mapping.parse_size_or_number(value) == expected
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "not a number",
+        "500XB",  # unrecognized unit
+        "GB",  # missing number
+        "500",  # missing unit -- ambiguous, must be a plain number instead
+        None,
+        [1, 2],
+    ],
+)
+def test_parse_size_or_number_rejects_unparseable_values(value):
+    with pytest.raises((TypeError, ValueError)):
+        mapping.parse_size_or_number(value)
+
+
+def test_map_list_section_host_config_sched_stop_accepts_size_strings(cluster_config):
+    """The production case for parse_size_or_number: an operator writes a
+    human-friendly size string for a byte-valued resource's threshold
+    (e.g. maxmem/mem/tmp/maxswap/maxtmp -- see test.c's
+    is_bytes_valued_resource) instead of computing the raw byte count by
+    hand."""
+    items = [
+        {
+            "hostname": "worker4",
+            "sched_stop": [{"resname": "tmp", "sched": "500MB", "stop": "100MB"}],
+        }
+    ]
+    mapping.map_list_section(_spec_for("hostConfigs"), items, cluster_config)
+
+    threshold = cluster_config.hostConfigs[0].thresholds[0]
+    assert threshold.name == "tmp"
+    assert threshold.sched == 500 * 1024**2
+    assert threshold.stop == 100 * 1024**2
+
+
 def test_time_window_day_prefixed_spec_parses_day_hour_minute(cluster_config):
     items = [
         {