myxce/src/scheduler/testsuite/test_repl_resource_table.c

/*
 * test_repl_resource_table.c - contract test for the interactive simulation
 * REPL's Resource Value Formatter table/grouping layout, units-mode
 * rendering, and the Jobid Range Compressor's adoption in `info worker`'s
 * assigned-job listing (scheduler-sim-repl-verbosity spec, task 8.1,
 * Requirements 6.4, 6.5, 6.6, 4.5).
 *
 * Split out of test_repl_breakpoints_inspection.c -- design.md's own File
 * Structure Plan originally named that file as task 8.1's landing spot
 * ("extended with `info worker`/`info resource` table-layout and
 * units-mode-display tests"), but landing these ten tests there per that
 * plan pushed the file from 2095 to 3215 lines -- 3.2x this exact file
 * family's own established ~1000-line split trigger, the same threshold
 * that already split test_repl_round_rollup.c (1023 lines, task 2.1) and
 * test_repl_event_echo.c (705 lines, task 3.1) out of related files for
 * structurally identical reasons (see those two files' own header comments,
 * and this file pair's own meson.build comments). Following that
 * established convention, this file duplicates (rather than shares via a
 * new header) whatever pre-existing static fixture helpers it needs from
 * test_repl_breakpoints_inspection.c.
 *
 * Fixture strategy and white-box #include set: identical to
 * test_repl_breakpoints_inspection.c's own Group C (see that file's own
 * header comment) -- a REAL xce_sched_core_t populated through a genuine
 * xce_core_push_events() call, driven via repl_dispatch() against that real
 * core/sim state, and captured via stdout.
 *
 * See the section header comment below (carried over verbatim from
 * test_repl_breakpoints_inspection.c, task 8.1's original landing spot) for
 * the RED-phase rationale and the assumed output contract these tests pin.
 */
#include <errno.h>
#include <limits.h>
#include <sys/wait.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <unity/unity.h>

#include "controller.capnp.h"
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention as test_repl_breakpoints_inspection.c/
 * test_repl_round_rollup.c/test_repl_event_echo.c). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_fmt.c"
#include "../src/repl_rollup.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/repl_params.c"
/* 5.1: the REPL trace sources below call the trace model. */
#include "../src/sched_trace.c"
#include "../src/sched_trace_scan.c"
#include "../src/sched_trace_capture.c"
#include "../src/sched_trace_outcome.c"
#include "../src/repl_trace_render.c"
#include "../src/repl_trace_render_outcome.c"
#include "../src/repl_trace_store.c"
#include "../src/repl_trace.c"
#include "../src/repl_trace_emit.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_picker.c" /* bug-1567: sched_sim_driver_run() now calls sim_bootstrap_picker() (sim_loop.c-cap sibling) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_limit.c" /* bug-1825: sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin() (sim_loop.c-cap sibling, sim_picker.c's own bug-1567 precedent) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"

void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }

/* repl_info.c (scheduler-sim-diagnostics spec, task 2.1) now calls
 * limit_get_configured_limits() (plugins/limit.c, not part of this TU) to
 * resolve `info limit`'s name/all argument -- link-satisfying stub only,
 * same rationale as kill_sched_job()/kill_sched_resval() above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_configured_limits()'s own documented "plugin absent" return)
 * is also the behaviorally correct answer here, not just a link
 * placeholder. repl_info_limit() itself is not yet wired into
 * repl_cmd_info() (task 4), so no test in this file can reach it anyway. */
var limit_get_configured_limits(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 3.1: repl_info.c now also calls
 * limit_get_usage_instances() (plugins/limit.c, not part of this TU) to
 * read a limit's live usage instances -- identical link-satisfying stub,
 * same rationale as limit_get_configured_limits() immediately above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_usage_instances()'s own documented "plugin absent" return) is
 * also the behaviorally correct answer here, not just a link placeholder. */
var limit_get_usage_instances(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 6 (bug-1825): sim_loop.c's
 * sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin()
 * (sim_limit.c, pulled into this TU via the #include above), which in
 * turn calls limit_new_plugin() (plugins/limit.c, not part of this TU) --
 * identical link-satisfying stub rationale as limit_get_configured_
 * limits()/limit_get_usage_instances() immediately above. Returning NULL
 * here is also the behaviorally correct answer, not just a link
 * placeholder: sim_bootstrap_limit_plugin() already handles a NULL
 * limit_new_plugin() return gracefully (logs an ERROR, registers nothing,
 * returns) -- the exact same "out of memory" path sim_bootstrap_picker()
 * (sim_picker.c) takes for a NULL builtin_fcfs_new_plugin() -- so this
 * file's fixtures simply end up with no limit plugin registered, same
 * observable state as before this task. */
xce_sched_plugin_t* limit_new_plugin(void)
{
    return NULL;
}

/* limit_seed_config() (plugins/limit.c, not part of this TU) -- bug-1825
 * remediation round 2's new honestly-typed config-seed entry point.
 * sim_bootstrap_limit_plugin() (sim_limit.c) only calls it after
 * plugin_register_core() has already succeeded against a real, non-NULL
 * limit_new_plugin() plugin -- unreachable in this file since the
 * limit_new_plugin() stub immediately above always returns NULL, but the
 * symbol must still resolve at link time. Same link-satisfying-stub
 * rationale as limit_new_plugin() immediately above. */
int limit_seed_config(xce_sched_plugin_t* plugin, xce_sched_core_t* core,
                      xce_conf4sched_t* config)
{
    (void)plugin;
    (void)core;
    (void)config;
    return 0;
}

/* resource-accounting.md's reconciliation (resource.c, not linked here --
 * this file only #includes core_events.c). core_events.c's
 * XCE_CORE_EVENT_JOB_RUSAGE handler calls this whenever job->accounts !=
 * NULL; link-satisfying stub only, same rationale as every other stub in
 * this block (see test_job_update.c's identical stub). */
int sched_reconcile_job_resource_usage(xce_sched_core_t*  c,
                                       xce_job_account_t* a,
                                       xce_job_rusages_t* r)
{
    (void)c; (void)a; (void)r; return 0;
}

void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

int sched_fill_limit_requests(xce_sched_alloc_req_t* alloc_req,
                              xlist_t* limits, xlist_t* rusages, int nhosts,
                              int nslots)
{
    (void)alloc_req;
    (void)limits;
    (void)rusages;
    (void)nhosts;
    (void)nslots;
    return 0;
}

/* sched_sim_driver_run()'s (sim_loop.c, scheduler-sim-decision-parity task
 * 2.1) new round-boundary call to the relocated sched_account_jobs()
 * (core.h) -- a no-op stand-in here since this file does not link
 * scheduler.c/resource.c, same rationale test_sim_equivalence.c's own copy
 * documents. */
int sched_account_jobs(xce_sched_core_t* core)
{
    (void)core;
    return 0;
}

/* bug-1836 (scheduler-driver-equivalence fix, Recorded-Decision
 * Reconciliation capacity gate): link-satisfying stubs, mirroring this
 * file's existing sched_account_jobs() no-op precedent immediately above --
 * sim_loop.c's sim_reconcile_unaccounted_jobs() now also calls
 * sched_check_host_satisfy()/sched_check_shared_satisfy() (resource.c) to
 * re-validate a recorded decision against live capacity, and
 * sched_charge_account() (resource.c) to keep that check's own view of
 * capacity up to date across multiple candidates in the same round. This
 * file never exercises real resource contention (same rationale as the
 * existing sched_account_jobs()/sched_ctx_get_uncomit_host() no-ops), so
 * the satisfy checks trivially always succeed and the charge call is a
 * no-op. */
int sched_check_host_satisfy(xce_sched_core_t* core, xce_sched_worker_t* w,
                             xvec_t* uncomitted, xvec_idval_t* wanted,
                             xvec_idval_t* unmeet)
{
    (void)core;
    (void)w;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_check_shared_satisfy(xce_sched_core_t* core, xvec_t* uncomitted,
                               xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
    (void)core;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_charge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    return 0;
}

int sched_uncharge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    return 0;
}


/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * copy test_repl_execution_control.c/test_sim_alt_events.c/etc. each carry
 * (schedobj.c is not linked here). */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
    xce_sched_plugin_t* plugin = NULL;
    int                 rc     = 0;

    if ((core == NULL) || (out == NULL)) {
        return -EINVAL;
    }

    *out = NULL;

    if (core->job_pick_plugin == NULL) {
        return -EINVAL;
    }

    lock(core->plugins, RTTI_LOCK_READ);
    rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
    unlock(core->plugins);

    if (rc <= 0) {
        return (rc < 0) ? rc : -ENOENT;
    }

    if ((plugin == NULL) || (plugin->picker_iface == NULL)
        || (plugin->picker_iface->next_job == NULL)
        || (plugin->picker_iface->begin_session == NULL)
        || (plugin->picker_iface->end_session == NULL)
        || (plugin->picker_iface->on_dispatch == NULL)
        || (plugin->picker_iface->on_pend == NULL)) {
        return -EINVAL;
    }

    *out = plugin;

    return 0;
}

/* scheduler-sim-live-allocation spec, task 4.1: sim_loop.c's real
 * sched_sim_driver_run() (white-box #include'd into this TU above) now
 * spawns sched_job_picker()/sched_allocator() (sched_election.c) directly
 * instead of its own, now-removed sim_picker_thread_fn() -- an undefined
 * reference at link time in this white-box TU otherwise, since
 * sched_election.c's own sched_allocator() calls the REAL
 * sched_make_allocation()/sched_commit_decision() (scheduler.c), neither
 * part of this file's own hand-picked white-box #include set (this file's
 * own header comment lists that set). This file's own test(s) do not
 * exercise genuine resource-matching allocation, so this link-satisfying
 * pair faithfully reproduces the exact PEND-only contract the REMOVED
 * sim_picker_thread_fn() itself used to provide -- split across the SAME
 * two-thread jbuf hand-off the real production pair now uses (one
 * producer, one consumer, matching the project's jbuf capacity-1
 * rendezvous invariant) instead of the old single fused thread -- so this
 * file's own existing assertions are unaffected by this task's production
 * change. Genuine resource-based dispatch through the real, unmodified
 * sched_job_picker()/sched_allocator() pair is proved separately by
 * test_sim_live_allocation.c (task 4.1's own dedicated fixture, which
 * links the real scheduler_srcs instead of white-boxing individual files,
 * precisely so it can reach the real allocation chain this stub
 * deliberately does not). */
void* sched_job_picker(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

    for (;;) {
        int      go    = 0;
        uint64_t jobid = 0;

        cbuf_get(&(core->wbuf), &go);

        if (sched_resolve_job_picker(core, &picker) < 0) {
            jobid = 0;
            cbuf_put(&(core->jbuf), &jobid);
            continue;
        }

        picker->picker_iface->begin_session(picker);

        for (;;) {
            xce_sched_job_t* job = picker->picker_iface->next_job(picker);

            if (job == NULL) {
                break;
            }

            jobid = job->jobid;
            cbuf_put(&(core->jbuf), &jobid);
        }

        jobid = 0;
        cbuf_put(&(core->jbuf), &jobid);

        picker->picker_iface->end_session(picker);
    }

    return NULL;
}

void* sched_allocator(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

    for (;;) {
        uint64_t         jobid    = 0;
        xce_decision_t*  decision = NULL;
        xce_sched_job_t* job;

        cbuf_get(&(core->jbuf), &jobid);
        if (jobid == 0) {
            decision = NULL;
            cbuf_put(&(core->decisions), &decision);
            continue;
        }

        job = sched_get_job(core, jobid);
        if (job == NULL) {
            continue;
        }

        (void)sched_resolve_job_picker(core, &picker);
        if (picker != NULL) {
            picker->picker_iface->on_pend(picker, job->jobid);
        }

        decision = make_pend_decision(make_pendreason(job->jobid));
        cbuf_put(&(core->decisions), &decision);
    }

    return NULL;
}

/* scheduler-sim-live-allocation spec, task 4.1 (bug-1829 fix): sim_loop.c's
 * real sched_sim_driver_run() (white-box #include'd into this TU above) now
 * also calls sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_
 * allocator() (sim_alloc.c) -- undefined references at link time in this
 * white-box TU otherwise, since neither symbol is part of this file's own
 * hand-picked white-box #include set. This file's own test(s) exercise the
 * link-satisfying sched_job_picker()/sched_allocator() stub pair above
 * (which never reaches real primary allocation), so a true no-op is the
 * correct, behavior-preserving stand-in here -- the real bootstrap logic is
 * exercised separately by test_sim_live_allocation.c (task 4.1's own
 * dedicated fixture, which links the real scheduler_srcs). */
void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

void setUp(void) { return; }
void tearDown(void) { return; }

/* ---------------------------------------------------------------------- */
/* Fixture helpers -- duplicated from test_repl_breakpoints_inspection.c   */
/* (same in-memory core-state construction and stdout-capture techniques   */
/* that file's own header comment establishes for its Group C), per this   */
/* file pair's established split convention: copied, not shared via a new  */
/* header.                                                                 */
/* ---------------------------------------------------------------------- */

static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue,
                           char* user, char* project)
{
    xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));

    j->jobid      = jobid;
    j->state      = state;
    j->sub_time   = 1000;
    j->queue_time = 2000;
    j->start_time = 0;
    j->submit     = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    j->submit->queue        = queue;
    j->submit->user         = user;
    j->submit->project_name = project;
    j->submit->resreq       = "type == any";
    j->submit->mintasks     = 1;
    j->submit->maxtasks     = 1;

    return j;
}

/* Same dup()/dup2()-around-STDOUT_FILENO capture technique
 * test_repl_execution_control.c's own capture_stdout_around() establishes. */
static char* capture_stdout_around(void (*fn)(void))
{
    char  tmp[] = "/tmp/xce_replrtbl_stdoutcap_XXXXXX";
    int   fd    = mkstemp(tmp);
    int   saved_stdout;
    off_t len;
    char* buf;

    TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "mkstemp for stdout capture must "
                                      "succeed");

    fflush(stdout);
    saved_stdout = dup(STDOUT_FILENO);
    TEST_ASSERT_TRUE_MESSAGE(saved_stdout >= 0, "dup(STDOUT_FILENO) must "
                                                "succeed");
    TEST_ASSERT_TRUE_MESSAGE(dup2(fd, STDOUT_FILENO) >= 0,
                             "dup2 redirect of STDOUT_FILENO must succeed");

    fn();

    fflush(stdout);
    dup2(saved_stdout, STDOUT_FILENO);
    close(saved_stdout);

    len = lseek(fd, 0, SEEK_CUR);
    TEST_ASSERT_TRUE(len >= 0);

    buf = (char*)calloc(1, (size_t)len + 1);
    TEST_ASSERT_NOT_NULL(buf);
    TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
    if (len > 0) {
        ssize_t nread = read(fd, buf, (size_t)len);
        TEST_ASSERT_TRUE(nread == len);
    }
    close(fd);
    unlink(tmp);

    return buf;
}

/* Captures the output of dispatching one REPL command line against `sim`.
 * Never a `step`/`continue`/`quit` line (this helper does not expect
 * repl_dispatch() to return true). */
static xce_sched_sim_driver_t* g_dispatch_sim;
static const char*             g_dispatch_line;

static void run_dispatch_for_capture(void)
{
    (void)repl_dispatch(g_dispatch_sim, g_dispatch_line);
}

static char* capture_dispatch(xce_sched_sim_driver_t* sim, const char* line)
{
    g_dispatch_sim  = sim;
    g_dispatch_line = line;
    return capture_stdout_around(run_dispatch_for_capture);
}

/*
 * ------------------------------------------------------------------------
 * Resource Value Formatter + Jobid Range Compressor adoption tests
 * (scheduler-sim-repl-verbosity spec, task 8.1, Requirements 6.4, 6.5, 6.6,
 * 4.5).
 *
 * RED-PHASE ONLY: task 8.2 (a separate, later task) rewrites
 * repl_info_worker_one()/repl_info_resource_one() (repl_info.c) to actually
 * group resources into labeled table sections, read rs->units_mode for a
 * byte/megabyte-valued resource, suppress a whole numeric value's trailing
 * ".00", and adopt repl_fmt_jobid_range() for the assigned-job listing.
 * None of that exists yet -- every test below is expected to FAIL A RUNTIME
 * ASSERTION against today's unconditional flat per-resource fprintf loop
 * and today's un-compressed per-job "assigned job: jobid=..." loop -- not a
 * build/link failure, since every symbol these tests call
 * (repl_dispatch/repl_cmd_info/rs->units_mode/repl_fmt_jobid_range/
 * repl_format_bytes_human/etc.) already exists (task 7.2/1.2's own work).
 *
 * Format contract this task DEFINES for task 8.2 to satisfy (design.md's
 * Resource Value Formatter section describes the target categories but
 * deliberately "does NOT specify an exact column-width/formatting string" --
 * the same latitude task 2.1 exercised for the collapsed-range notation):
 *
 *   - Three labeled sections, in this fixed order, printed only when that
 *     section has at least one resource to show (an empty section prints no
 *     header at all -- info resource's GLOBAL/SHARED-only scope, bug-1593,
 *     may genuinely have none of a given category):
 *       "-- Consumable/Capacity --"        (numeric + XCE_RESF_CONSUMABLE)
 *       "-- Load Metrics --"                (numeric, not CONSUMABLE)
 *       "-- Static Hardware Attributes --"  (string/boolean)
 *     matching design.md's own category list and order verbatim.
 *   - Every existing per-resource "resource NAME: value=...[ used=...]" /
 *     "value=...[ used=...]" line (bug-1590/1593/1688/1727's own pinned
 *     shapes) is UNCHANGED -- only reorganized under its section header --
 *     so none of this file's existing info-worker/info-resource assertions
 *     above need to change.
 *   - AC 6.5's "no trailing .00 on a whole number" applies ONLY to a
 *     non-consumable ("Load Metrics"/"Static Hardware Attributes") numeric
 *     value -- deliberately NOT the Consumable/Capacity value=/used= pair,
 *     which keeps its existing %.2f-always formatting
 *     (test_info_worker_reports_consumable_value_and_used/
 *     test_info_resource_reports_consumable_value_and_used, above, already
 *     pin "value=5.00 used=2.00"/"value=8.00 used=3.00" for whole-number
 *     consumable pairs -- a scope decision documented here, not an
 *     oversight, to avoid contradicting those still-active assertions).
 *   - units_mode governs ONLY a repl_is_bytes_valued_resource()-true
 *     resource or `swp` (MB-special-cased, repl_internal.h's own doc
 *     comment): raw -> "value=%f" (bare wire value, C's default 6-decimal
 *     %f, mirroring xce_test's own cmd_resval raw-value convention,
 *     test.c:2540's "%f (%s)\n"); human -> "value=%s" (repl_format_bytes_
 *     human()'s own output alone); both (the default) -> "value=%f (%s)",
 *     the exact "%f (%s)\n" shape test.c's cmd_resval already establishes
 *     as this codebase's precedent for combined raw+human display.
 *   - Never mutates the underlying stored xce_val_t -- each units_mode test
 *     below re-reads the raw stored value after the display call and
 *     asserts it is unchanged (AC 6.4). Full end-to-end replay-determinism
 *     proof is task 10.2's job, not this unit-level file's -- these display
 *     functions have no return value/side effect that could feed back into
 *     a scheduling decision, so a per-value mutation check is the
 *     correct-granularity proxy here.
 *   - The assigned-job listing adopts repl_fmt_jobid_range() itself (not a
 *     hand-rolled duplicate) -- proven by computing the SAME function's
 *     output over the SAME jobid list directly in the test and asserting
 *     the identical string appears in repl_info_worker()'s own output,
 *     rather than pinning a hand-transcribed golden string.
 * ------------------------------------------------------------------------
 */

/*
 * Shared fixture helper for the units_mode tests below: a real
 * xce_sched_core_t with one worker ("hostA") carrying exactly one
 * host-based numeric resource `resname` with wire value `raw_value`, pushed
 * through a genuine XCE_CORE_EVENT_WORKERS_PUSH the same way
 * test_info_worker_reports_resource_values() (above) does by hand --
 * factored out since four separate tests need an identical
 * single-numeric-resource worker fixture, only resname/raw_value differing.
 *
 * Unlike every pre-existing test in this file (which builds its wire
 * structures as STACK locals directly in the test function body, never
 * returning before capture_dispatch() reads them), `val`/`nv`/`nv_ptrs`/
 * `wire_worker` here are heap-allocated and deliberately never freed: a
 * live ASAN stack-use-after-return failure (repl_info_worker_one() reading
 * w->resvals[idx] after this helper had already returned) proved
 * xce_core_push_events()'s WORKERS_PUSH handling retains the raw
 * nv->val pointer inside w->resvals rather than dup_val()-ing it -- that
 * data must outlive this HELPER's own return, not just its stack frame.
 * Safe to leak: this file's test runner sets ASAN_OPTIONS=detect_leaks=0.
 * `wire_resource` stays a stack local -- makeResource() (used identically,
 * stack-local, all over this file) copies what it needs synchronously,
 * before this helper returns.
 */
static xce_resource_t* build_worker_with_one_numeric_resource(
    xce_sched_sim_driver_t* sim, const char* resname, double raw_value)
{
    xce_resource_t    wire_resource;
    xce_worker_t*     wire_worker;
    xce_worker_t*     worker_ptrs[1];
    xce_val_t*        val;
    xce_name_val_t*   nv;
    xce_name_val_t**  nv_ptrs;
    xce_core_event_t  events[1];
    xce_resource_t*   r;
    int               rc;

    rc = xce_sched_core_init(&(sim->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name    = (char*)resname;
    wire_resource.valtype = XCE_VAL_NUMERIC;
    r                     = makeResource(sim->core.resmgr, &wire_resource);
    TEST_ASSERT_NOT_NULL(r);

    val = (xce_val_t*)calloc(1, sizeof(*val));
    TEST_ASSERT_NOT_NULL(val);
    val->which    = Value_u_fvalue;
    val->u.fvalue = raw_value;

    nv = (xce_name_val_t*)calloc(1, sizeof(*nv));
    TEST_ASSERT_NOT_NULL(nv);
    nv->name = (char*)resname;
    nv->val  = val;

    nv_ptrs = (xce_name_val_t**)calloc(1, sizeof(*nv_ptrs));
    TEST_ASSERT_NOT_NULL(nv_ptrs);
    nv_ptrs[0] = nv;

    wire_worker = (xce_worker_t*)calloc(1, sizeof(*wire_worker));
    TEST_ASSERT_NOT_NULL(wire_worker);
    wire_worker->name              = "hostA";
    wire_worker->status            = 0;
    wire_worker->nresvals_to_sched = 1;
    wire_worker->resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                 = wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim->core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    return r;
}

/*
 * mk_array_jobid_bpi/mk_plain_jobid_bpi -- real, correctly bit-packed
 * jobids, the SAME MAKE_JOBID/TAG_JOB_ARRAY/SEQUENCE_HIGH_SHIFT
 * construction test_repl_output_compression.c's own mk_array_jobid()/
 * mk_plain_jobid() establish (that file's own header comment has the full
 * bit-layout rationale) -- duplicated here (not shared) rather than made
 * non-static there, matching this file family's own established
 * "duplicate pre-existing static fixture helpers" convention (design.md's
 * File Structure Plan, test_repl_round_rollup.c/test_repl_event_echo.c's
 * own Correction notes). "_bpi" suffix avoids any name collision with that
 * file's own (unrelated TU) static helpers of the same short name.
 */
static uint64_t mk_array_jobid_bpi(uint32_t cluster, uint32_t apiserver,
                                   uint32_t timestamp, uint32_t array_base,
                                   uint32_t idx)
{
    uint64_t seq = (((uint64_t)array_base) << SEQUENCE_HIGH_SHIFT)
                   | (uint64_t)idx;

    return MAKE_JOBID(cluster, apiserver, timestamp, TAG_JOB_ARRAY, seq);
}

static uint64_t mk_plain_jobid_bpi(uint32_t cluster, uint32_t apiserver,
                                   uint32_t timestamp, uint32_t seq)
{
    return MAKE_JOBID(cluster, apiserver, timestamp, 0x0, (uint64_t)seq);
}

/*
 * AC 6.6: info worker's resource listing groups into three labeled
 * sections -- Consumable/Capacity, Load Metrics, Static Hardware
 * Attributes -- in that fixed order, instead of one flat undifferentiated
 * list. One resource per category, so section-header-before-resource-line
 * ordering can be asserted end to end. Uses whole-number consumable values
 * (5.0/2.0) -- deliberately NOT asserting on trailing-.00 suppression here;
 * see this section's own header comment above for why (that rule is scoped
 * to non-consumable numeric resources only).
 */
static void test_info_worker_groups_resources_into_labeled_table_sections_in_order(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_slots;
    xce_resource_t         wire_r1m;
    xce_resource_t         wire_type;
    xce_resource_t*        r_slots;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_val_t              val_slots;
    xce_val_t              val_r1m;
    xce_val_t              val_type;
    xce_name_val_t         nv_slots;
    xce_name_val_t         nv_r1m;
    xce_name_val_t         nv_type;
    xce_name_val_t*        nv_ptrs[3];
    xce_sched_worker_t*    sw;
    xce_val_t               used_val;
    char*                   captured;
    char*                   pos_consumable_header;
    char*                   pos_slots_line;
    char*                   pos_load_header;
    char*                   pos_r1m_line;
    char*                   pos_static_header;
    char*                   pos_type_line;
    int                     rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_slots, 0, sizeof(wire_slots));
    wire_slots.name    = "slots";
    wire_slots.valtype = XCE_VAL_NUMERIC;
    wire_slots.flags   = XCE_RESF_CONSUMABLE;
    r_slots            = makeResource(sim.core.resmgr, &wire_slots);
    TEST_ASSERT_NOT_NULL(r_slots);

    memset(&wire_r1m, 0, sizeof(wire_r1m));
    wire_r1m.name    = "r1m";
    wire_r1m.valtype = XCE_VAL_NUMERIC;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_r1m));

    memset(&wire_type, 0, sizeof(wire_type));
    wire_type.name    = "type";
    wire_type.valtype = XCE_VAL_STRING;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_type));

    memset(&val_slots, 0, sizeof(val_slots));
    val_slots.which    = Value_u_fvalue;
    val_slots.u.fvalue = 5.0;
    memset(&nv_slots, 0, sizeof(nv_slots));
    nv_slots.name = "slots";
    nv_slots.val  = &val_slots;

    memset(&val_r1m, 0, sizeof(val_r1m));
    val_r1m.which    = Value_u_fvalue;
    val_r1m.u.fvalue = 3.5;
    memset(&nv_r1m, 0, sizeof(nv_r1m));
    nv_r1m.name = "r1m";
    nv_r1m.val  = &val_r1m;

    memset(&val_type, 0, sizeof(val_type));
    val_type.which    = Value_u_svalue;
    val_type.u.svalue = "linux86";
    memset(&nv_type, 0, sizeof(nv_type));
    nv_type.name = "type";
    nv_type.val  = &val_type;

    nv_ptrs[0] = &nv_slots;
    nv_ptrs[1] = &nv_r1m;
    nv_ptrs[2] = &nv_type;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 3;
    wire_worker.resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = 2.0;
    zarray_put(&(sw->resInUse), r_slots->idx, dup_val(&used_val));

    captured = capture_dispatch(&sim, "info worker hostA");

    pos_consumable_header = strstr(captured, "-- Consumable/Capacity --");
    pos_slots_line
        = strstr(captured, "resource slots: value=5.00 used=2.00");
    pos_load_header = strstr(captured, "-- Load Metrics --");
    pos_r1m_line     = strstr(captured, "resource r1m: value=3.50");
    pos_static_header
        = strstr(captured, "-- Static Hardware Attributes --");
    pos_type_line = strstr(captured, "resource type: value=linux86");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_consumable_header,
        "info worker must print a 'Consumable/Capacity' section header "
        "(AC 6.6) -- absent from today's flat per-resource loop");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_slots_line,
        "the CONSUMABLE resource's existing value=/used= line must still "
        "appear (bug-1727's pinned format), now under its section header");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_load_header,
        "info worker must print a 'Load Metrics' section header (AC 6.6)");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_r1m_line,
        "the non-consumable numeric resource's value= line must still "
        "appear, now under 'Load Metrics'");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_static_header,
        "info worker must print a 'Static Hardware Attributes' section "
        "header (AC 6.6)");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_type_line,
        "the string-valued resource's value= line must still appear "
        "(bug-1590's pinned format), now under 'Static Hardware "
        "Attributes'");

    TEST_ASSERT_TRUE_MESSAGE(
        pos_consumable_header < pos_slots_line,
        "the Consumable/Capacity header must precede its own resource "
        "line");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_slots_line < pos_load_header,
        "Consumable/Capacity's content must precede the Load Metrics "
        "section (design.md's own category order)");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_load_header < pos_r1m_line,
        "the Load Metrics header must precede its own resource line");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_r1m_line < pos_static_header,
        "Load Metrics' content must precede the Static Hardware "
        "Attributes section (design.md's own category order)");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_static_header < pos_type_line,
        "the Static Hardware Attributes header must precede its own "
        "resource line");

    free(captured);

    printf("test_info_worker_groups_resources_into_labeled_table_sections_"
          "in_order: PASS\n");
}

/*
 * AC 6.5: a whole-number unitless (non-consumable) resource value renders
 * with no trailing ".00"; a fractional one keeps its two decimals.
 */
static void
test_info_worker_load_metric_whole_number_omits_trailing_zeros_fractional_keeps_decimals(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_r1m;
    xce_resource_t         wire_r15m;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_val_t              val_r1m;
    xce_val_t              val_r15m;
    xce_name_val_t         nv_r1m;
    xce_name_val_t         nv_r15m;
    xce_name_val_t*        nv_ptrs[2];
    char*                  captured;
    int                    rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_r1m, 0, sizeof(wire_r1m));
    wire_r1m.name    = "r1m";
    wire_r1m.valtype = XCE_VAL_NUMERIC;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_r1m));

    memset(&wire_r15m, 0, sizeof(wire_r15m));
    wire_r15m.name    = "r15m";
    wire_r15m.valtype = XCE_VAL_NUMERIC;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_r15m));

    memset(&val_r1m, 0, sizeof(val_r1m));
    val_r1m.which    = Value_u_fvalue;
    val_r1m.u.fvalue = 5.0;
    memset(&nv_r1m, 0, sizeof(nv_r1m));
    nv_r1m.name = "r1m";
    nv_r1m.val  = &val_r1m;

    memset(&val_r15m, 0, sizeof(val_r15m));
    val_r15m.which    = Value_u_fvalue;
    val_r15m.u.fvalue = 3.25;
    memset(&nv_r15m, 0, sizeof(nv_r15m));
    nv_r15m.name = "r15m";
    nv_r15m.val  = &val_r15m;

    nv_ptrs[0] = &nv_r1m;
    nv_ptrs[1] = &nv_r15m;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 2;
    wire_worker.resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    captured = capture_dispatch(&sim, "info worker hostA");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "resource r1m: value=5\n"),
        "AC 6.5: a whole-number unitless (non-consumable) resource value "
        "must render with no trailing '.00' -- e.g. 'value=5', not "
        "'value=5.00'");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "resource r1m: value=5.00"),
        "must not keep the old unconditional \"%.2f\" formatting for a "
        "whole-number unitless value");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "resource r15m: value=3.25"),
        "AC 6.5: a fractional unitless value must keep its decimals");

    free(captured);

    printf("test_info_worker_load_metric_whole_number_omits_trailing_"
          "zeros_fractional_keeps_decimals: PASS\n");
}

/*
 * AC 6.2 'raw': a bytes-valued resource (mem/tmp/maxmem/maxtmp/maxswap)
 * under units_mode=raw shows ONLY the bare numeric wire value, no
 * human-readable unit suffix. AC 6.4: the display call must never mutate
 * the underlying stored value.
 */
static void
test_info_worker_units_mode_raw_renders_bare_wire_value_without_mutating_stored_value(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_resource_t*        r;
    xce_sched_worker_t*    sw;
    xce_val_t*              stored;
    char*                   captured;
    char                    expected_raw[64];

    memset(&sim, 0, sizeof(sim));
    r = build_worker_with_one_numeric_resource(&sim, "mem", 8589934592.0);

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_RAW;

    captured = capture_dispatch(&sim, "info worker hostA");

    snprintf(expected_raw, sizeof(expected_raw), "value=%f", 8589934592.0);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_raw),
        "AC 6.2 'raw': a bytes-valued resource must render its bare wire "
        "numeric value under units_mode=raw");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "GB"),
        "AC 6.2 'raw': must show ONLY the numeric wire value, no "
        "human-readable unit suffix");

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);
    stored = (xce_val_t*)zarray_get(&(sw->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        8589934592.0f, (float)stored->u.fvalue,
        "AC 6.4: displaying a resource under any units_mode must never "
        "mutate the underlying stored/wire value");

    free(captured);

    printf("test_info_worker_units_mode_raw_renders_bare_wire_value_"
          "without_mutating_stored_value: PASS\n");
}

/*
 * AC 6.2 'human': the same bytes-valued resource under units_mode=human
 * shows ONLY the adaptively-scaled human-readable form, no raw wire value.
 */
static void
test_info_worker_units_mode_human_renders_scaled_value_without_mutating_stored_value(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_resource_t*        r;
    xce_sched_worker_t*    sw;
    xce_val_t*              stored;
    char*                   captured;
    char                    expected_human[64];
    char                    expected_raw[64];

    memset(&sim, 0, sizeof(sim));
    r = build_worker_with_one_numeric_resource(&sim, "mem", 8589934592.0);

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_HUMAN;

    captured = capture_dispatch(&sim, "info worker hostA");

    repl_format_bytes_human(8589934592.0, expected_human,
                            sizeof(expected_human));
    snprintf(expected_raw, sizeof(expected_raw), "%f", 8589934592.0);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_human),
        "AC 6.2 'human': a bytes-valued resource must render its "
        "adaptively-scaled human-readable value under units_mode=human");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, expected_raw),
        "AC 6.2 'human': must show ONLY the human-readable form, no raw "
        "numeric wire value");

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);
    stored = (xce_val_t*)zarray_get(&(sw->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        8589934592.0f, (float)stored->u.fvalue,
        "AC 6.4: displaying a resource under any units_mode must never "
        "mutate the underlying stored/wire value");

    free(captured);

    printf("test_info_worker_units_mode_human_renders_scaled_value_"
          "without_mutating_stored_value: PASS\n");
}

/*
 * AC 6.2 'both' (also Req 6.3's default): the same bytes-valued resource
 * renders raw AND human forms together, "<raw> (<human>)", mirroring
 * xce_test's own cmd_resval "%f (%s)\n" convention (test.c:2540).
 */
static void
test_info_worker_units_mode_both_renders_raw_and_human_without_mutating_stored_value(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_resource_t*        r;
    xce_sched_worker_t*    sw;
    xce_val_t*              stored;
    char*                   captured;
    char                    expected_raw[64];
    char                    expected_human[64];
    char                    expected_both[160];

    memset(&sim, 0, sizeof(sim));
    r = build_worker_with_one_numeric_resource(&sim, "mem", 8589934592.0);

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_BOTH;

    captured = capture_dispatch(&sim, "info worker hostA");

    snprintf(expected_raw, sizeof(expected_raw), "%f", 8589934592.0);
    repl_format_bytes_human(8589934592.0, expected_human,
                            sizeof(expected_human));
    snprintf(expected_both, sizeof(expected_both), "value=%s (%s)",
            expected_raw, expected_human);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_both),
        "AC 6.2 'both' (also Req 6.3's default): a bytes-valued resource "
        "must render raw AND human forms together, \"<raw> (<human>)\"");

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);
    stored = (xce_val_t*)zarray_get(&(sw->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        8589934592.0f, (float)stored->u.fvalue,
        "AC 6.4: displaying a resource under any units_mode must never "
        "mutate the underlying stored/wire value");

    free(captured);

    printf("test_info_worker_units_mode_both_renders_raw_and_human_"
          "without_mutating_stored_value: PASS\n");
}

/*
 * AC 6.1: `swp`'s canonical wire unit is megabytes (repl_internal.h's own
 * doc comment) -- its value must be converted MB->bytes (x1024x1024)
 * BEFORE human-scaling, so it renders on the same adaptive B/KB/MB/GB/TB/PB
 * scale a bytes-native resource of equivalent magnitude would. AC 6.4: the
 * conversion is display-only -- the stored value stays in megabytes.
 */
static void
test_info_worker_units_mode_converts_swp_megabytes_before_human_scaling(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_sched_worker_t*    sw;
    xce_resource_t*        r;
    xce_val_t*              stored;
    char*                   captured;
    char                    expected_human[64];
    double                  swp_mb;
    double                  swp_bytes;

    swp_mb    = 2048.0;                    /* 2048 MB */
    swp_bytes = swp_mb * 1024.0 * 1024.0; /* -> exactly 2 GB */

    memset(&sim, 0, sizeof(sim));
    r = build_worker_with_one_numeric_resource(&sim, "swp", swp_mb);

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_HUMAN;

    captured = capture_dispatch(&sim, "info worker hostA");

    repl_format_bytes_human(swp_bytes, expected_human,
                            sizeof(expected_human));

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_human),
        "repl_internal.h's documented swp special case: swp's megabyte "
        "wire value must be converted to bytes before human-scaling");

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);
    stored = (xce_val_t*)zarray_get(&(sw->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        (float)swp_mb, (float)stored->u.fvalue,
        "AC 6.4: the MB->bytes conversion is display-only -- the stored "
        "wire value must remain in its original megabytes unit, "
        "unconverted");

    free(captured);

    printf("test_info_worker_units_mode_converts_swp_megabytes_before_"
          "human_scaling: PASS\n");
}

/*
 * Requirement 4.5: info worker's assigned-job listing adopts the SAME
 * jobid-range-compression helper (repl_fmt_jobid_range(), task 1.2,
 * repl_fmt.c) every other multi-jobid listing in this file family already
 * uses (repl_break.c's JOB_PUSH event echo, Round Rollup's pend-jobid
 * display) -- three contiguous array-job indices assigned to the SAME
 * worker must render as ONE compressed-range line, not three separate
 * "assigned job: jobid=..." lines. A fourth, non-array job assigned to a
 * DIFFERENT worker proves the existing per-host filter is unaffected by
 * compression.
 *
 * bug-1595 jobid-reuse hazard: every jobid below (including the distractor)
 * uses a distinct (cluster, apiserver, timestamp, array-base/seq) tuple --
 * xce_core_push_events()'s jobs kvmgr is keyed by jobid, so two fixture
 * jobs sharing one jobid would silently collapse into a single stored job
 * instead of the four distinct ones this test needs.
 */
static void
test_info_worker_assigned_job_listing_compresses_contiguous_array_indices(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[2];
    xce_queue_t            wire_queue;
    xce_worker_t           wire_worker_a;
    xce_worker_t           wire_worker_b;
    xce_worker_t*          worker_ptrs[2];
    xce_job_t*             job_arr[3];
    xce_job_t*             job_plain;
    xce_job_t*             all_jobs[4];
    uint64_t               jobid_arr[3];
    uint64_t               jobid_plain;
    xce_sched_job_t*       sjob;
    xce_host_alloc_t       ha_hostA;
    xce_host_alloc_t*      ha_hostA_ptrs[1];
    xce_job_alloc_t        alloc_hostA;
    xce_host_alloc_t       ha_hostB;
    xce_host_alloc_t*      ha_hostB_ptrs[1];
    xce_job_alloc_t        alloc_hostB;
    char*                  captured;
    char                   expected_range[256];
    int                    rc;
    int                    i;
    int                    n_assigned_lines;
    char*                  scan;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "normal";
    wire_queue.priority = 5;
    wire_queue.status   = QUEUE_STAT_OPEN;

    for (i = 0; i < 3; i++) {
        jobid_arr[i]
            = mk_array_jobid_bpi(3, 2, 1700000000, 1, (uint32_t)i);
        job_arr[i] = make_job(jobid_arr[i], JOB_STAT_PEND, "normal",
                              "alice", "proj1");
    }
    jobid_plain = mk_plain_jobid_bpi(3, 2, 1700000001, 77);
    job_plain
        = make_job(jobid_plain, JOB_STAT_PEND, "normal", "alice", "proj1");

    all_jobs[0] = job_arr[0];
    all_jobs[1] = job_arr[1];
    all_jobs[2] = job_arr[2];
    all_jobs[3] = job_plain;

    memset(events, 0, sizeof(events));
    events[0].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[0].u.queue_update.queue = &wire_queue;
    events[1].type                = XCE_CORE_EVENT_JOBS_PUSH;
    events[1].u.jobs_push.njobs   = 4;
    events[1].u.jobs_push.jobs    = all_jobs;

    rc = xce_core_push_events(&(sim.core), events, 2);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_worker_a, 0, sizeof(wire_worker_a));
    wire_worker_a.name   = "hostA";
    wire_worker_a.status = 0;
    memset(&wire_worker_b, 0, sizeof(wire_worker_b));
    wire_worker_b.name   = "hostB";
    wire_worker_b.status = 0;
    worker_ptrs[0] = &wire_worker_a;
    worker_ptrs[1] = &wire_worker_b;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 2;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&ha_hostA, 0, sizeof(ha_hostA));
    ha_hostA.host    = "hostA";
    ha_hostA_ptrs[0] = &ha_hostA;
    memset(&alloc_hostA, 0, sizeof(alloc_hostA));
    alloc_hostA.nhosts = 1;
    alloc_hostA.allocs = ha_hostA_ptrs;

    memset(&ha_hostB, 0, sizeof(ha_hostB));
    ha_hostB.host    = "hostB";
    ha_hostB_ptrs[0] = &ha_hostB;
    memset(&alloc_hostB, 0, sizeof(alloc_hostB));
    alloc_hostB.nhosts = 1;
    alloc_hostB.allocs = ha_hostB_ptrs;

    for (i = 0; i < 3; i++) {
        sjob = sched_get_job(&(sim.core), jobid_arr[i]);
        TEST_ASSERT_NOT_NULL(sjob);
        sjob->allocs = &alloc_hostA;
    }
    sjob = sched_get_job(&(sim.core), jobid_plain);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc_hostB;

    rc = repl_fmt_jobid_range(expected_range, sizeof(expected_range),
                              jobid_arr, 3);
    TEST_ASSERT_TRUE_MESSAGE(
        rc >= 0, "the compression helper itself must succeed for this "
                "fixture (task 1.2, already implemented/tested)");

    captured = capture_dispatch(&sim, "info worker hostA");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_range),
        "Requirement 4.5: info worker's assigned-job listing must render "
        "the SAME compressed-range string repl_fmt_jobid_range() itself "
        "produces for these three contiguous array indices, instead of "
        "three separate uncompressed 'assigned job: jobid=...' lines");

    n_assigned_lines = 0;
    scan             = captured;
    while ((scan = strstr(scan, "assigned job")) != NULL) {
        n_assigned_lines++;
        scan++;
    }
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, n_assigned_lines,
        "the three array-job indices must collapse into exactly ONE "
        "assigned-job listing line, not three");

    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "hostB"),
        "hostA's assigned-job listing must never include a job assigned "
        "to a different worker (hostB) -- the existing per-host filter "
        "must remain unaffected by range compression");

    free(captured);

    printf("test_info_worker_assigned_job_listing_compresses_contiguous_"
          "array_indices: PASS\n");
}

/*
 * Improvement request (simulator REPL user feedback): a job consuming more
 * than one "slots" unit on a worker should be visibly distinguishable from
 * a single-slot job in the assigned-jobs listing (e.g. "jobid*2"), instead
 * of every assigned job looking identical regardless of how much of the
 * worker it actually occupies.
 */
static void
test_info_worker_assigned_job_listing_appends_slot_count_when_uniform(void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[2];
    xce_queue_t            wire_queue;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_job_t*             job;
    xce_job_t*             all_jobs[1];
    uint64_t               jobid;
    xce_sched_job_t*       sjob;
    xce_alloc_t            slots_alloc;
    xce_alloc_t*           slots_alloc_ptrs[1];
    xce_host_alloc_t       ha_host;
    xce_host_alloc_t*      ha_host_ptrs[1];
    xce_job_alloc_t        alloc;
    char*                  captured;
    char                   expected_range[256];
    char                   expected_line[300];
    int                    rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "normal";
    wire_queue.priority = 5;
    wire_queue.status   = QUEUE_STAT_OPEN;

    jobid = mk_plain_jobid_bpi(4, 2, 1700000010, 1);
    job   = make_job(jobid, JOB_STAT_RUN, "normal", "alice", "proj1");
    all_jobs[0] = job;

    memset(events, 0, sizeof(events));
    events[0].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[0].u.queue_update.queue = &wire_queue;
    events[1].type                = XCE_CORE_EVENT_JOBS_PUSH;
    events[1].u.jobs_push.njobs   = 1;
    events[1].u.jobs_push.jobs    = all_jobs;

    rc = xce_core_push_events(&(sim.core), events, 2);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name   = "hostA";
    wire_worker.status = 0;
    worker_ptrs[0] = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&slots_alloc, 0, sizeof(slots_alloc));
    slots_alloc.resname = "slots";
    slots_alloc.value   = 2.0;
    slots_alloc_ptrs[0] = &slots_alloc;

    memset(&ha_host, 0, sizeof(ha_host));
    ha_host.host    = "hostA";
    ha_host.nallocs = 1;
    ha_host.allocs  = slots_alloc_ptrs;
    ha_host_ptrs[0] = &ha_host;

    memset(&alloc, 0, sizeof(alloc));
    alloc.nhosts = 1;
    alloc.allocs = ha_host_ptrs;

    sjob = sched_get_job(&(sim.core), jobid);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc;

    rc = repl_fmt_jobid_range(expected_range, sizeof(expected_range), &jobid,
                              1);
    TEST_ASSERT_TRUE(rc >= 0);
    snprintf(expected_line, sizeof(expected_line), "assigned jobs: %s*2\n",
            expected_range);

    captured = capture_dispatch(&sim, "info worker hostA");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_line),
        "a job consuming 2 'slots' on this worker must be annotated "
        "'*2' in the assigned-jobs listing");

    free(captured);

    printf("test_info_worker_assigned_job_listing_appends_slot_count_when_"
          "uniform: PASS\n");
}

/*
 * bug-1778 (interim display-layer mitigation): a job already at a terminal
 * state (DONE/EXIT) must never appear in the assigned-jobs listing, even if
 * its allocs pointer is still (incorrectly) non-NULL -- live-observed via a
 * real captured replay where three DONE jobs kept showing as assigned
 * despite the worker's own slot-usage accounting (used=) correctly
 * excluding them. Root cause of WHY allocs stays non-NULL for a terminal
 * job is not yet pinned (suspected snapshot save/load inconsistency, see
 * buglog.json) -- this test only pins the defensive display-layer
 * exclusion, mirroring the same "terminal is permanent" guard
 * job_state_regresses_terminal()/sim_reconcile_unaccounted_jobs() already
 * apply elsewhere.
 */
static void
test_info_worker_assigned_job_listing_excludes_terminal_job_despite_stale_allocs(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[2];
    xce_queue_t            wire_queue;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_job_t*             job_done;
    xce_job_t*             job_run;
    xce_job_t*             all_jobs[2];
    uint64_t               jobid_done;
    uint64_t               jobid_run;
    xce_sched_job_t*       sjob;
    xce_host_alloc_t       ha_host;
    xce_host_alloc_t*      ha_host_ptrs[1];
    xce_job_alloc_t        alloc;
    char*                  captured;
    char                   jobidStr_done[64];
    char                   jobidStr_run[64];
    int                    rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "normal";
    wire_queue.priority = 5;
    wire_queue.status   = QUEUE_STAT_OPEN;

    jobid_done = mk_plain_jobid_bpi(6, 2, 1700000030, 1);
    jobid_run  = mk_plain_jobid_bpi(6, 2, 1700000031, 2);
    job_done = make_job(jobid_done, JOB_STAT_RUN, "normal", "alice", "proj1");
    job_run  = make_job(jobid_run, JOB_STAT_RUN, "normal", "alice", "proj1");
    all_jobs[0] = job_done;
    all_jobs[1] = job_run;

    memset(events, 0, sizeof(events));
    events[0].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[0].u.queue_update.queue = &wire_queue;
    events[1].type                = XCE_CORE_EVENT_JOBS_PUSH;
    events[1].u.jobs_push.njobs   = 2;
    events[1].u.jobs_push.jobs    = all_jobs;

    rc = xce_core_push_events(&(sim.core), events, 2);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name   = "hostA";
    wire_worker.status = 0;
    worker_ptrs[0] = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&ha_host, 0, sizeof(ha_host));
    ha_host.host    = "hostA";
    ha_host_ptrs[0] = &ha_host;
    memset(&alloc, 0, sizeof(alloc));
    alloc.nhosts = 1;
    alloc.allocs = ha_host_ptrs;

    /* Both jobs get the SAME allocs struct assigned directly (bypassing the
     * normal event-driven cleanup) -- job_done's state is then forced to
     * JOB_STAT_DONE directly, reproducing bug-1778's exact observed
     * inconsistency (state=DONE, allocs still non-NULL) regardless of which
     * code path actually produces it live. */
    sjob = sched_get_job(&(sim.core), jobid_done);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc;
    sjob->state  = JOB_STAT_DONE;

    sjob = sched_get_job(&(sim.core), jobid_run);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc;

    captured = capture_dispatch(&sim, "info worker hostA");

    fmt_jobid(jobidStr_done, sizeof(jobidStr_done), jobid_done);
    fmt_jobid(jobidStr_run, sizeof(jobidStr_run), jobid_run);

    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, jobidStr_done),
        "bug-1778: a job already at a terminal state (DONE) must never "
        "appear in the assigned-jobs listing, even if its allocs pointer "
        "is still (incorrectly) non-NULL");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, jobidStr_run),
        "the genuinely-running job on the same worker must still be "
        "listed as assigned");

    free(captured);

    printf("test_info_worker_assigned_job_listing_excludes_terminal_job_"
          "despite_stale_allocs: PASS\n");
}

/*
 * Two jobs on the SAME worker consuming DIFFERENT slot counts cannot be
 * annotated with a single trailing "*N" on a compressed range (which count
 * would it even mean?) -- must fall back to the pre-existing uncompressed
 * per-job listing, each line annotated with its own count.
 */
static void
test_info_worker_assigned_job_listing_falls_back_per_job_when_slots_differ(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[2];
    xce_queue_t            wire_queue;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_job_t*             job_a;
    xce_job_t*             job_b;
    xce_job_t*             all_jobs[2];
    uint64_t               jobid_a;
    uint64_t               jobid_b;
    xce_sched_job_t*       sjob;
    xce_alloc_t            slots_alloc_a;
    xce_alloc_t*           slots_alloc_a_ptrs[1];
    xce_alloc_t            slots_alloc_b;
    xce_alloc_t*           slots_alloc_b_ptrs[1];
    xce_host_alloc_t       ha_a;
    xce_host_alloc_t*      ha_a_ptrs[1];
    xce_job_alloc_t        alloc_a;
    xce_host_alloc_t       ha_b;
    xce_host_alloc_t*      ha_b_ptrs[1];
    xce_job_alloc_t        alloc_b;
    char*                  captured;
    char                   jobidStr_a[64];
    char                   jobidStr_b[64];
    char                   expected_a[96];
    char                   expected_b[96];
    int                    rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "normal";
    wire_queue.priority = 5;
    wire_queue.status   = QUEUE_STAT_OPEN;

    jobid_a = mk_plain_jobid_bpi(5, 2, 1700000020, 1);
    jobid_b = mk_plain_jobid_bpi(5, 2, 1700000021, 2);
    job_a   = make_job(jobid_a, JOB_STAT_RUN, "normal", "alice", "proj1");
    job_b   = make_job(jobid_b, JOB_STAT_RUN, "normal", "alice", "proj1");
    all_jobs[0] = job_a;
    all_jobs[1] = job_b;

    memset(events, 0, sizeof(events));
    events[0].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[0].u.queue_update.queue = &wire_queue;
    events[1].type                = XCE_CORE_EVENT_JOBS_PUSH;
    events[1].u.jobs_push.njobs   = 2;
    events[1].u.jobs_push.jobs    = all_jobs;

    rc = xce_core_push_events(&(sim.core), events, 2);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name   = "hostA";
    wire_worker.status = 0;
    worker_ptrs[0] = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&slots_alloc_a, 0, sizeof(slots_alloc_a));
    slots_alloc_a.resname = "slots";
    slots_alloc_a.value   = 2.0;
    slots_alloc_a_ptrs[0] = &slots_alloc_a;
    memset(&ha_a, 0, sizeof(ha_a));
    ha_a.host    = "hostA";
    ha_a.nallocs = 1;
    ha_a.allocs  = slots_alloc_a_ptrs;
    ha_a_ptrs[0] = &ha_a;
    memset(&alloc_a, 0, sizeof(alloc_a));
    alloc_a.nhosts = 1;
    alloc_a.allocs = ha_a_ptrs;

    memset(&slots_alloc_b, 0, sizeof(slots_alloc_b));
    slots_alloc_b.resname = "slots";
    slots_alloc_b.value   = 4.0;
    slots_alloc_b_ptrs[0] = &slots_alloc_b;
    memset(&ha_b, 0, sizeof(ha_b));
    ha_b.host    = "hostA";
    ha_b.nallocs = 1;
    ha_b.allocs  = slots_alloc_b_ptrs;
    ha_b_ptrs[0] = &ha_b;
    memset(&alloc_b, 0, sizeof(alloc_b));
    alloc_b.nhosts = 1;
    alloc_b.allocs = ha_b_ptrs;

    sjob = sched_get_job(&(sim.core), jobid_a);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc_a;
    sjob = sched_get_job(&(sim.core), jobid_b);
    TEST_ASSERT_NOT_NULL(sjob);
    sjob->allocs = &alloc_b;

    fmt_jobid(jobidStr_a, sizeof(jobidStr_a), jobid_a);
    fmt_jobid(jobidStr_b, sizeof(jobidStr_b), jobid_b);
    snprintf(expected_a, sizeof(expected_a), "assigned job: jobid=%s*2\n",
            jobidStr_a);
    snprintf(expected_b, sizeof(expected_b), "assigned job: jobid=%s*4\n",
            jobidStr_b);

    captured = capture_dispatch(&sim, "info worker hostA");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_a),
        "differing per-job slot counts on the same worker must fall back "
        "to the uncompressed per-job listing, annotated with this job's "
        "own slot count");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_b),
        "differing per-job slot counts on the same worker must fall back "
        "to the uncompressed per-job listing, annotated with this job's "
        "own slot count");

    free(captured);

    printf("test_info_worker_assigned_job_listing_falls_back_per_job_when_"
          "slots_differ: PASS\n");
}

/*
 * AC 6.6, `info resource all`: the same three-section grouping as `info
 * worker`'s own version above, applied to the cluster-wide GLOBAL/SHARED
 * resource listing.
 */
static void
test_info_resource_groups_resources_into_labeled_table_sections_in_order(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_lic;
    xce_resource_t         wire_load;
    xce_resource_t         wire_site;
    xce_resource_t*        r_lic;
    xce_val_t               val_lic;
    xce_val_t               val_load;
    xce_val_t               val_site;
    xce_resval_t            rv_lic;
    xce_resval_t            rv_load;
    xce_resval_t            rv_site;
    xce_resval_t*           rv_ptrs[3];
    xce_val_t               used_val;
    char*                   captured;
    char*                   pos_consumable_header;
    char*                   pos_lic_line;
    char*                   pos_load_header;
    char*                   pos_load_line;
    char*                   pos_static_header;
    char*                   pos_site_line;
    int                     rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_lic, 0, sizeof(wire_lic));
    wire_lic.name    = "gpu_licenses";
    wire_lic.valtype = XCE_VAL_NUMERIC;
    wire_lic.flags   = XCE_RESF_SHARED | XCE_RESF_CONSUMABLE;
    r_lic            = makeResource(sim.core.resmgr, &wire_lic);
    TEST_ASSERT_NOT_NULL(r_lic);

    memset(&wire_load, 0, sizeof(wire_load));
    wire_load.name    = "cluster_load1m";
    wire_load.valtype = XCE_VAL_NUMERIC;
    wire_load.flags   = XCE_RESF_SHARED;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_load));

    memset(&wire_site, 0, sizeof(wire_site));
    wire_site.name    = "site_name";
    wire_site.valtype = XCE_VAL_STRING;
    wire_site.flags   = XCE_RESF_SHARED;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_site));

    memset(&val_lic, 0, sizeof(val_lic));
    val_lic.which    = Value_u_fvalue;
    val_lic.u.fvalue = 8.0;
    memset(&rv_lic, 0, sizeof(rv_lic));
    rv_lic.name = "gpu_licenses";
    rv_lic.val  = &val_lic;

    memset(&val_load, 0, sizeof(val_load));
    val_load.which    = Value_u_fvalue;
    val_load.u.fvalue = 4.5;
    memset(&rv_load, 0, sizeof(rv_load));
    rv_load.name = "cluster_load1m";
    rv_load.val  = &val_load;

    memset(&val_site, 0, sizeof(val_site));
    val_site.which    = Value_u_svalue;
    val_site.u.svalue = "gpu-cluster";
    memset(&rv_site, 0, sizeof(rv_site));
    rv_site.name = "site_name";
    rv_site.val  = &val_site;

    rv_ptrs[0] = &rv_lic;
    rv_ptrs[1] = &rv_load;
    rv_ptrs[2] = &rv_site;

    memset(events, 0, sizeof(events));
    events[0].type                   = XCE_CORE_EVENT_RESVALS_FULL;
    events[0].u.resvals_full.n       = 3;
    events[0].u.resvals_full.resvals = rv_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = 3.0;
    zarray_put(&(sim.core.sharedInUse), r_lic->idx, dup_val(&used_val));

    captured = capture_dispatch(&sim, "info resource all");

    pos_consumable_header = strstr(captured, "-- Consumable/Capacity --");
    pos_lic_line          = strstr(captured, "value=8.00 used=3.00");
    pos_load_header       = strstr(captured, "-- Load Metrics --");
    pos_load_line         = strstr(captured, "value=4.50");
    pos_static_header
        = strstr(captured, "-- Static Hardware Attributes --");
    pos_site_line = strstr(captured, "value=gpu-cluster");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_consumable_header,
        "info resource all must print a 'Consumable/Capacity' section "
        "header (AC 6.6)");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_lic_line,
        "the CONSUMABLE global resource's existing value=/used= line "
        "(bug-1727's pinned format) must still appear");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_load_header,
        "info resource all must print a 'Load Metrics' section header "
        "(AC 6.6)");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_load_line,
        "the non-consumable numeric global resource's value= line must "
        "still appear");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_static_header,
        "info resource all must print a 'Static Hardware Attributes' "
        "section header (AC 6.6)");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        pos_site_line,
        "the string-valued global resource's value= line must still "
        "appear");

    TEST_ASSERT_TRUE_MESSAGE(
        pos_consumable_header < pos_lic_line,
        "the Consumable/Capacity header must precede its own resource "
        "line");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_lic_line < pos_load_header,
        "Consumable/Capacity content must precede the Load Metrics "
        "section (design.md's own category order)");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_load_header < pos_load_line,
        "the Load Metrics header must precede its own resource line");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_load_line < pos_static_header,
        "Load Metrics content must precede the Static Hardware "
        "Attributes section (design.md's own category order)");
    TEST_ASSERT_TRUE_MESSAGE(
        pos_static_header < pos_site_line,
        "the Static Hardware Attributes header must precede its own "
        "resource line");

    free(captured);

    printf("test_info_resource_groups_resources_into_labeled_table_"
          "sections_in_order: PASS\n");
}

/*
 * AC 6.5, `info resource <name>`: same whole-number-vs-fractional rule as
 * info worker's own version above, applied to a cluster-wide GLOBAL/SHARED
 * non-consumable numeric resource.
 */
static void
test_info_resource_load_metric_whole_number_omits_trailing_zeros(void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_load;
    xce_val_t               val_load;
    xce_resval_t            rv_load;
    xce_resval_t*           rv_ptrs[1];
    char*                   captured;
    int                     rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_load, 0, sizeof(wire_load));
    wire_load.name    = "cluster_load1m";
    wire_load.valtype = XCE_VAL_NUMERIC;
    wire_load.flags   = XCE_RESF_SHARED;
    TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_load));

    memset(&val_load, 0, sizeof(val_load));
    val_load.which    = Value_u_fvalue;
    val_load.u.fvalue = 5.0;
    memset(&rv_load, 0, sizeof(rv_load));
    rv_load.name = "cluster_load1m";
    rv_load.val  = &val_load;
    rv_ptrs[0]   = &rv_load;

    memset(events, 0, sizeof(events));
    events[0].type                   = XCE_CORE_EVENT_RESVALS_FULL;
    events[0].u.resvals_full.n       = 1;
    events[0].u.resvals_full.resvals = rv_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    captured = capture_dispatch(&sim, "info resource cluster_load1m");

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "value=5\n"),
        "AC 6.5: a whole-number unitless (non-consumable) global resource "
        "value must render with no trailing '.00'");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "value=5.00"),
        "must not keep the old unconditional \"%.2f\" formatting for a "
        "whole-number unitless global resource value");

    free(captured);

    printf("test_info_resource_load_metric_whole_number_omits_trailing_"
          "zeros: PASS\n");
}

/*
 * bug-1824: info resource bounded a shared resource's slot by
 * sharedInst.count -- the number of POPULATED slots, not the array's
 * capacity -- so a populated slot at an index >= .count (a sparsely filled
 * array) was reported as "value=(none)". Same shape as bug-1688's
 * stale-count test for info worker.
 */
static void test_info_resource_shows_value_when_slot_is_above_count(void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_load;
    xce_resource_t*        r_load;
    xce_val_t              val_load;
    xce_resval_t           rv_load;
    xce_resval_t*          rv_ptrs[1];
    char*                  captured;
    int                    rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_load, 0, sizeof(wire_load));
    wire_load.name    = "cluster_load1m";
    wire_load.valtype = XCE_VAL_NUMERIC;
    wire_load.flags   = XCE_RESF_SHARED;
    r_load            = makeResource(sim.core.resmgr, &wire_load);
    TEST_ASSERT_NOT_NULL(r_load);

    memset(&val_load, 0, sizeof(val_load));
    val_load.which    = Value_u_fvalue;
    val_load.u.fvalue = 5.0;
    memset(&rv_load, 0, sizeof(rv_load));
    rv_load.name = "cluster_load1m";
    rv_load.val  = &val_load;
    rv_ptrs[0]   = &rv_load;

    memset(events, 0, sizeof(events));
    events[0].type                   = XCE_CORE_EVENT_RESVALS_FULL;
    events[0].u.resvals_full.n       = 1;
    events[0].u.resvals_full.resvals = rv_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NOT_NULL(zarray_get(&(sim.core.sharedInst), r_load->idx));

    /* The slot stays populated; only .count under-reports it. */
    sim.core.sharedInst.count = r_load->idx;

    captured = capture_dispatch(&sim, "info resource cluster_load1m");

    TEST_ASSERT_NOT_NULL_MESSAGE(strstr(captured, "value=5\n"),
        "bug-1824: a populated shared slot at an index >= sharedInst.count "
        "must still display its value");

    free(captured);

    printf("test_info_resource_shows_value_when_slot_is_above_count: "
          "PASS\n");
}

/*
 * AC 6.2 'both' / Req 6.3 default, `info resource <name>`: same
 * units_mode-aware bytes rendering as info worker's own version above,
 * applied to a cluster-wide GLOBAL resource named "mem" -- the fixture
 * name is chosen purely to exercise repl_is_bytes_valued_resource()'s
 * classification at this second call site, independent of whether a real
 * deployment would ever flag "mem" GLOBAL rather than host-based.
 */
static void
test_info_resource_units_mode_both_renders_raw_and_human_for_bytes_resource(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_mem;
    xce_resource_t*        r;
    xce_val_t               val_mem;
    xce_resval_t            rv_mem;
    xce_resval_t*           rv_ptrs[1];
    xce_val_t*              stored;
    char*                   captured;
    char                    expected_raw[64];
    char                    expected_human[64];
    char                    expected_both[160];
    int                     rc;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_mem, 0, sizeof(wire_mem));
    wire_mem.name    = "mem";
    wire_mem.valtype = XCE_VAL_NUMERIC;
    wire_mem.flags   = XCE_RESF_SHARED;
    r                = makeResource(sim.core.resmgr, &wire_mem);
    TEST_ASSERT_NOT_NULL(r);

    memset(&val_mem, 0, sizeof(val_mem));
    val_mem.which    = Value_u_fvalue;
    val_mem.u.fvalue = 8589934592.0;
    memset(&rv_mem, 0, sizeof(rv_mem));
    rv_mem.name = "mem";
    rv_mem.val  = &val_mem;
    rv_ptrs[0]  = &rv_mem;

    memset(events, 0, sizeof(events));
    events[0].type                   = XCE_CORE_EVENT_RESVALS_FULL;
    events[0].u.resvals_full.n       = 1;
    events[0].u.resvals_full.resvals = rv_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_BOTH;

    captured = capture_dispatch(&sim, "info resource mem");

    snprintf(expected_raw, sizeof(expected_raw), "%f", 8589934592.0);
    repl_format_bytes_human(8589934592.0, expected_human,
                            sizeof(expected_human));
    snprintf(expected_both, sizeof(expected_both), "value=%s (%s)",
            expected_raw, expected_human);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_both),
        "AC 6.2 'both' (also the Req 6.3 default): a bytes-valued global "
        "resource must render its raw wire value alongside the "
        "human-readable form");

    stored = (xce_val_t*)zarray_get(&(sim.core.sharedInst), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        8589934592.0f, (float)stored->u.fvalue,
        "AC 6.4: displaying a resource under any units_mode must never "
        "mutate the underlying stored/wire value");

    free(captured);

    printf("test_info_resource_units_mode_both_renders_raw_and_human_for_"
          "bytes_resource: PASS\n");
}

/*
 * User-reported (2026-09-11, live replay session): a CONSUMABLE byte-valued
 * per-host resource (mem/tmp/swp) rendered its value=/used= pair as a raw
 * `%.2f` byte count (e.g. "used=87960930222080.00"), unlike every other
 * bytes-valued field in this same listing (maxmem/maxswap/maxtmp, and mem/
 * tmp/swp themselves when NOT flagged consumable, via the Load Metrics
 * path) which already render human-scaled. repl_info_worker_one()'s own
 * pre-fix header comment explicitly scoped this out ("no bytes-valued
 * CONSUMABLE resource is exercised anywhere in this codebase's tests
 * today") -- this test (and its "raw"/swp/info-resource siblings below)
 * closes that gap. A non-bytes-valued consumable (slots/licenses) must keep
 * its existing pinned "%.2f"/"%.2f" shape unchanged -- see
 * test_info_worker_reports_consumable_value_and_used (bug-1727,
 * test_repl_breakpoints_inspection.c), left untouched by this fix.
 */
static void
test_info_worker_units_mode_both_renders_raw_and_human_for_consumable_bytes_resource(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_resource;
    xce_resource_t*        r;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_val_t              val;
    xce_name_val_t         nv;
    xce_name_val_t*        nv_ptrs[1];
    xce_val_t              used_val;
    xce_sched_worker_t*    sw;
    char*                  captured;
    char                   expected_value[64];
    char                   expected_used[64];
    char                   expected_line[160];
    double                 avail;
    double                 used;
    int                    rc;

    avail = 8589934592.0;  /* 8 GB */
    used  = 4294967296.0; /* 4 GB */

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name    = "mem";
    wire_resource.valtype = XCE_VAL_NUMERIC;
    wire_resource.flags   = XCE_RESF_CONSUMABLE;
    r                     = makeResource(sim.core.resmgr, &wire_resource);
    TEST_ASSERT_NOT_NULL(r);

    memset(&val, 0, sizeof(val));
    val.which    = Value_u_fvalue;
    val.u.fvalue = avail;
    memset(&nv, 0, sizeof(nv));
    nv.name    = "mem";
    nv.val     = &val;
    nv_ptrs[0] = &nv;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 1;
    wire_worker.resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = used;
    zarray_put(&(sw->resInUse), r->idx, dup_val(&used_val));

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_BOTH;

    captured = capture_dispatch(&sim, "info worker hostA");

    repl_format_bytes_human(avail, expected_value, sizeof(expected_value));
    repl_format_bytes_human(used, expected_used, sizeof(expected_used));
    snprintf(expected_line, sizeof(expected_line), "value=%f (%s) used=%f (%s)",
            avail, expected_value, used, expected_used);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_line),
        "a CONSUMABLE byte-valued resource (mem/tmp/swp) must render both "
        "value= and used= human-scaled under units_mode=both, the same "
        "as every other bytes-valued field in this listing");

    free(captured);

    printf(
        "test_info_worker_units_mode_both_renders_raw_and_human_for_"
        "consumable_bytes_resource: PASS\n");
}

/*
 * AC 6.2 'raw' counterpart for a CONSUMABLE byte-valued resource: under
 * units_mode=raw, value=/used= must show ONLY the bare wire numbers, no
 * "GB"/human suffix -- mirrors
 * test_info_worker_units_mode_raw_renders_bare_wire_value_without_mutating_stored_value's
 * contract, applied to the consumable value=/used= pair instead of a
 * single value= line.
 */
static void
test_info_worker_units_mode_raw_renders_bare_values_for_consumable_bytes_resource(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_resource;
    xce_resource_t*        r;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_val_t              val;
    xce_name_val_t         nv;
    xce_name_val_t*        nv_ptrs[1];
    xce_val_t              used_val;
    xce_sched_worker_t*    sw;
    char*                  captured;
    char                   expected_line[160];
    double                 avail;
    double                 used;
    int                    rc;

    avail = 8589934592.0;
    used  = 4294967296.0;

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name    = "mem";
    wire_resource.valtype = XCE_VAL_NUMERIC;
    wire_resource.flags   = XCE_RESF_CONSUMABLE;
    r                     = makeResource(sim.core.resmgr, &wire_resource);
    TEST_ASSERT_NOT_NULL(r);

    memset(&val, 0, sizeof(val));
    val.which    = Value_u_fvalue;
    val.u.fvalue = avail;
    memset(&nv, 0, sizeof(nv));
    nv.name    = "mem";
    nv.val     = &val;
    nv_ptrs[0] = &nv;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 1;
    wire_worker.resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = used;
    zarray_put(&(sw->resInUse), r->idx, dup_val(&used_val));

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_RAW;

    captured = capture_dispatch(&sim, "info worker hostA");

    snprintf(expected_line, sizeof(expected_line), "value=%f used=%f", avail,
            used);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_line),
        "AC 6.2 'raw': a CONSUMABLE byte-valued resource must render bare "
        "wire values for both value= and used= under units_mode=raw");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "GB"),
        "AC 6.2 'raw': must show no human-readable unit suffix");

    free(captured);

    printf(
        "test_info_worker_units_mode_raw_renders_bare_values_for_"
        "consumable_bytes_resource: PASS\n");
}

/*
 * AC 6.1's swp-is-megabytes special case, applied to the CONSUMABLE
 * value=/used= pair: both fields must be converted MB->bytes before
 * human-scaling, mirroring
 * test_info_worker_units_mode_converts_swp_megabytes_before_human_scaling's
 * contract for the Load Metrics (non-consumable) path.
 */
static void
test_info_worker_units_mode_converts_swp_megabytes_for_consumable_resource(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_resource;
    xce_resource_t*        r;
    xce_worker_t           wire_worker;
    xce_worker_t*          worker_ptrs[1];
    xce_val_t              val;
    xce_name_val_t         nv;
    xce_name_val_t*        nv_ptrs[1];
    xce_val_t              used_val;
    xce_sched_worker_t*    sw;
    char*                  captured;
    char                   expected_value_human[64];
    char                   expected_used_human[64];
    double                 avail_mb;
    double                 used_mb;
    int                    rc;

    avail_mb = 16384.0; /* 16384 MB -> 16 GB */
    used_mb  = 4096.0;  /* 4096 MB -> 4 GB */

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name    = "swp";
    wire_resource.valtype = XCE_VAL_NUMERIC;
    wire_resource.flags   = XCE_RESF_CONSUMABLE;
    r                     = makeResource(sim.core.resmgr, &wire_resource);
    TEST_ASSERT_NOT_NULL(r);

    memset(&val, 0, sizeof(val));
    val.which    = Value_u_fvalue;
    val.u.fvalue = avail_mb;
    memset(&nv, 0, sizeof(nv));
    nv.name    = "swp";
    nv.val     = &val;
    nv_ptrs[0] = &nv;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 1;
    wire_worker.resvals_to_sched  = nv_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(events, 0, sizeof(events));
    events[0].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers = 1;
    events[0].u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    sw = sched_get_worker(&(sim.core), "hostA");
    TEST_ASSERT_NOT_NULL(sw);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = used_mb;
    zarray_put(&(sw->resInUse), r->idx, dup_val(&used_val));

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_HUMAN;

    captured = capture_dispatch(&sim, "info worker hostA");

    repl_format_bytes_human(avail_mb * 1024.0 * 1024.0, expected_value_human,
                            sizeof(expected_value_human));
    repl_format_bytes_human(used_mb * 1024.0 * 1024.0, expected_used_human,
                            sizeof(expected_used_human));

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_value_human),
        "swp's value= must be converted MB->bytes before human-scaling, "
        "same as the Load Metrics path");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_used_human),
        "swp's used= must be converted MB->bytes before human-scaling too");

    free(captured);

    printf(
        "test_info_worker_units_mode_converts_swp_megabytes_for_"
        "consumable_resource: PASS\n");
}

/*
 * info resource's counterpart: a GLOBAL/SHARED CONSUMABLE byte-valued
 * resource must render its value=/used= pair human-scaled too, mirroring
 * test_info_resource_units_mode_both_renders_raw_and_human_for_bytes_resource
 * (the non-consumable case) and this file's own info-worker tests above.
 */
static void
test_info_resource_units_mode_both_renders_raw_and_human_for_consumable_bytes_resource(
    void)
{
    xce_sched_sim_driver_t sim;
    xce_core_event_t       events[1];
    xce_resource_t         wire_mem;
    xce_resource_t*        r;
    xce_val_t               val_mem;
    xce_resval_t            rv_mem;
    xce_resval_t*           rv_ptrs[1];
    xce_val_t*              stored;
    xce_val_t               used_val;
    char*                   captured;
    char                    expected_value[64];
    char                    expected_used[64];
    char                    expected_line[160];
    double                  avail;
    double                  used;
    int                     rc;

    avail = 8589934592.0; /* 8 GB */
    used  = 2147483648.0; /* 2 GB */

    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&wire_mem, 0, sizeof(wire_mem));
    wire_mem.name    = "mem";
    wire_mem.valtype = XCE_VAL_NUMERIC;
    wire_mem.flags   = XCE_RESF_SHARED | XCE_RESF_CONSUMABLE;
    r                = makeResource(sim.core.resmgr, &wire_mem);
    TEST_ASSERT_NOT_NULL(r);

    memset(&val_mem, 0, sizeof(val_mem));
    val_mem.which    = Value_u_fvalue;
    val_mem.u.fvalue = avail;
    memset(&rv_mem, 0, sizeof(rv_mem));
    rv_mem.name = "mem";
    rv_mem.val  = &val_mem;
    rv_ptrs[0]  = &rv_mem;

    memset(events, 0, sizeof(events));
    events[0].type                   = XCE_CORE_EVENT_RESVALS_FULL;
    events[0].u.resvals_full.n       = 1;
    events[0].u.resvals_full.resvals = rv_ptrs;

    rc = xce_core_push_events(&(sim.core), events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    memset(&used_val, 0, sizeof(used_val));
    used_val.which    = Value_u_fvalue;
    used_val.u.fvalue = used;
    zarray_put(&(sim.core.sharedInUse), r->idx, dup_val(&used_val));

    sim.repl_state = repl_state_ensure(&sim);
    TEST_ASSERT_NOT_NULL(sim.repl_state);
    sim.repl_state->units_mode = XCE_REPL_UNITS_BOTH;

    captured = capture_dispatch(&sim, "info resource mem");

    repl_format_bytes_human(avail, expected_value, sizeof(expected_value));
    repl_format_bytes_human(used, expected_used, sizeof(expected_used));
    snprintf(expected_line, sizeof(expected_line), "value=%f (%s) used=%f (%s)",
            avail, expected_value, used, expected_used);

    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, expected_line),
        "a GLOBAL/SHARED CONSUMABLE byte-valued resource must render both "
        "value= and used= human-scaled under units_mode=both");

    stored = (xce_val_t*)zarray_get(&(sim.core.sharedInst), r->idx);
    TEST_ASSERT_NOT_NULL(stored);
    TEST_ASSERT_EQUAL_INT(Value_u_fvalue, stored->which);
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        (float)avail, (float)stored->u.fvalue,
        "AC 6.4: displaying a resource under any units_mode must never "
        "mutate the underlying stored/wire value");

    free(captured);

    printf(
        "test_info_resource_units_mode_both_renders_raw_and_human_for_"
        "consumable_bytes_resource: PASS\n");
}

int main(void)
{
    UNITY_BEGIN();

    RUN_TEST(test_info_worker_groups_resources_into_labeled_table_sections_in_order);
    RUN_TEST(test_info_worker_load_metric_whole_number_omits_trailing_zeros_fractional_keeps_decimals);
    RUN_TEST(test_info_worker_units_mode_raw_renders_bare_wire_value_without_mutating_stored_value);
    RUN_TEST(test_info_worker_units_mode_human_renders_scaled_value_without_mutating_stored_value);
    RUN_TEST(test_info_worker_units_mode_both_renders_raw_and_human_without_mutating_stored_value);
    RUN_TEST(test_info_worker_units_mode_converts_swp_megabytes_before_human_scaling);
    RUN_TEST(test_info_worker_units_mode_both_renders_raw_and_human_for_consumable_bytes_resource);
    RUN_TEST(test_info_worker_units_mode_raw_renders_bare_values_for_consumable_bytes_resource);
    RUN_TEST(test_info_worker_units_mode_converts_swp_megabytes_for_consumable_resource);
    RUN_TEST(test_info_worker_assigned_job_listing_compresses_contiguous_array_indices);
    RUN_TEST(test_info_worker_assigned_job_listing_appends_slot_count_when_uniform);
    RUN_TEST(test_info_worker_assigned_job_listing_excludes_terminal_job_despite_stale_allocs);
    RUN_TEST(test_info_worker_assigned_job_listing_falls_back_per_job_when_slots_differ);
    RUN_TEST(test_info_resource_groups_resources_into_labeled_table_sections_in_order);
    RUN_TEST(test_info_resource_load_metric_whole_number_omits_trailing_zeros);
    RUN_TEST(test_info_resource_shows_value_when_slot_is_above_count);
    RUN_TEST(test_info_resource_units_mode_both_renders_raw_and_human_for_bytes_resource);
    RUN_TEST(test_info_resource_units_mode_both_renders_raw_and_human_for_consumable_bytes_resource);

    return UNITY_END();
}