/*
 * test_sim_e2e_limit_visibility.c - scheduler-sim-live-allocation spec,
 * task 5.2 (Req 2.1, 2.2, 2.3, 2.4, Depends: 4.1, 4.2, 4.3): the feature's
 * concrete, original acceptance test -- bug-1826 itself. "if the `info
 * limit` can not show the usage information of limit resources, then the
 * power of this command will get limited."
 *
 * ── What this proves that no existing test does ───────────────────────────
 * test_sim_alloc_limit_fanout.c (task 4.3) proves a HOST-scoped limit's
 * usage instance is created via a live seqjob dispatch, but builds its
 * fixture by hand (xce_sched_core_init() + the same four sim_bootstrap_*()
 * calls sched_sim_driver_run() itself makes, called directly) -- it never
 * drives an actual `.replay` FILE through the real sched_sim_driver_run()
 * entry point, and it configures ONLY a host-scoped limit, never a global
 * one. test_sim_contention_limit_driver_parity.c (task 5.1) is narrower
 * still: a bare-core contention fixture proving normal/sim parity, again
 * with only a host-scoped limit and again never touching a real `.replay`
 * file. test_repl_limit_table.c's own sim-mode-registration tests (task
 * 2.3/6) go even further from a real replay: they reproduce
 * sim->core.config's end state "via direct construction... no capnp/.replay
 * file needed" (that file's own header comment), explicitly bypassing the
 * wire/snapshot round trip.
 *
 * THIS file is the first in this feature (and, so far as a full-codebase
 * grep for `sched.core.config =` before `sched_record_start()` confirms,
 * the first anywhere) to seed a limit config into a REAL SnapshotEntry
 * (Conf4Sched, capnp-encoded, containing BOTH a global and a host-scoped
 * xce_limit_config_t), write it to a real `.replay` file, and drive that
 * file through the unmodified, real `sched_sim_driver_run()` end-to-end --
 * exactly what an operator's real `xce_scheduler -s <replay>` invocation
 * does. It is the concrete Integration Test #1 design.md's own "Testing
 * Strategy" section names: "A real xce_sched_sim_driver_t driven through
 * sched_sim_driver_run() against a replay whose original session had global
 * AND host-scoped limits configured and actively enforced -- info limit
 * shows real usage for both".
 *
 * ── Why config seeded via the snapshot, not a configPush replay event ─────
 * A tempting alternative is a `ReplayEvent.body.configPush` wire event
 * (schema-legal, `XCE_CORE_EVENT_CONFIG_PUSH`, core_events.c) -- but reading
 * xce_core_push_events()'s CONFIG_PUSH case (core_events.c) before writing
 * this fixture shows it only decomposes queues/resources/resvals and fires
 * the PICKER's on_config hook; it never calls plugin_config_update() or any
 * plugin's own config_hook. plugin_config_update() (plugin.c) has exactly
 * ONE call site in this whole codebase: scheduler.c's own one-time
 * sched_init_config() (normal mode's startup config load, never re-invoked
 * from a live CONFIG_PUSH). A configPush replay event would therefore
 * silently never reach the limit plugin in EITHER mode -- not a
 * sim-mode-specific gap, just not how limit config ever reaches a plugin
 * post-startup in this codebase today. The REAL mechanism (confirmed by
 * reading sim_loop.c's sched_sim_driver_run(), sim.c's snapshot-load path,
 * and sim_limit.c's own header comment) is: a snapshot's `config` field
 * (SnapshotData.config, a real Conf4Sched capnp field, schema-verified
 * against src/schema/controller.capnp) round-trips through
 * xce_sched_core_load_snapshot() into sim->core.config (snapshot_load.c:200,
 * a borrowed-reference assignment), which sim_bootstrap_limit_plugin()
 * (called by sched_sim_driver_run() immediately after driver_start(), per
 * sim_limit.c's own header comment) then seeds via limit_seed_config() --
 * exactly what would happen for a genuine recorded production session whose
 * limits.yaml was active when the recording snapshot was taken. This
 * fixture reproduces that shape as literally as a hand-built `.replay` file
 * can: the two limit resources ("e2e_global_res"/"e2e_host_res") are baked
 * into the snapshot's own resmgr (required -- sim_bootstrap_limit_plugin()
 * runs strictly before any replay event is processed, and
 * is_host_based_limit()/build_limit() key_find() these resources by name),
 * while "slots" and the worker/queue/job all arrive afterward as ordinary
 * wire ReplayEvents, the same real JOBS_PUSH/WORKERS_PUSH/QUEUE_UPDATE
 * ingest path test_sim_live_allocation.c's own proven fixture already
 * establishes.
 *
 * ── Link style: real scheduler_srcs, not a white-box #include ────────────
 * Same technique as test_sim_live_allocation.c/test_sim_alloc_limit_
 * fanout.c/test_sim_contention_limit_driver_parity.c: `info limit`'s own
 * backing functions (repl_info_limit(), repl_dispatch()) are `static` to
 * repl.c/repl_info.c and unreachable from a real-scheduler_srcs-linked TU
 * (only a white-box #include of those files, as test_repl_limit_table.c
 * uses, can call them directly) -- but this fixture NEEDS the real,
 * unmodified resource.c/scheduler.c/sched_election.c/plugins/
 * seqjob_allocator.c chain for a genuine dispatch, which a white-box
 * #include of repl_info.c's own narrower dependency set does not link.
 * This file therefore reads the exact same accessors repl_info_limit()
 * itself reads -- limit_get_configured_limits()/limit_get_usage_
 * instances() (plugin.h, unchanged since scheduler-sim-diagnostics) -- the
 * "existing limit-inspection accessor" task 5.2's own text calls for,
 * and additionally reads back each usage instance's actual numeric value
 * via get_resvals() (scheduler.h) against the exact array
 * update_limit_res_value() (plugins/limit.c) wrote it into (core->
 * sharedInst for the global limit, the worker's own resvals for the
 * host-scoped one), to prove "real, non-zero usage" literally, not merely
 * "an instance object exists".
 *
 * ── bug-1598/1632/1763 fixture-safety requirement ─────────────────────────
 * g_sim below is `static` (file/process lifetime), not stack-local, for the
 * identical reason test_sim_live_allocation.c's own header comment
 * documents in full: sched_sim_driver_run() spawns two real, never-joined
 * pthreads that read &(sim->core) for the rest of the process's life.
 *
 * ── RED-phase note (see this task's own status report for the full
 *    account) ───────────────────────────────────────────────────────────
 * This is a pure validation task (no production code changes expected).
 * Task 4.3's commit (f6a28621) is the specific, narrowly-scoped prior
 * change that closes the exact gap this test's host-scoped half exercises
 * (before it, plugin_prepare_allocation()'s fan-out during a live seqjob/
 * parjob dispatch attempt reached a throwaway wrapper's own separate plugin
 * registry, never sim->core.plugins, so a host-scoped limit registered
 * directly against sim->core could never be reached). The status report
 * documents a targeted revert of exactly that commit's production files,
 * confirming this test's host-scoped assertion genuinely fails without it.
 */
#include <errno.h>
#include <limits.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 "recording.h"
#include "resobj.h"
#include "scheduler.h"
#include "sim.h"
#include "xcelimit.h"
#include "xstring.h"

/* write_capnp_message (recording.c) has no public header declaration --
 * see test_sim_live_allocation.c's own identical comment for the
 * established "extern declared where consumed" rationale. */
extern int write_capnp_message(FILE* f, struct capn* c);

#define WORKER_NAME         "e2e_limit_worker"
#define QUEUE_NAME          "normal"
#define TEST_RESREQ         ((char*)"select[type==any] rusage[mem=1]")
#define TEST_JOBID          ((uint64_t)940301ULL)
#define WORKER_SLOTS        4.0f

#define GLOBAL_LIMIT_NAME   "e2e_global_limit"
#define GLOBAL_LIMIT_RESNAME "e2e_global_res"
#define GLOBAL_LIMIT_VAL    10.0f

#define HOST_LIMIT_NAME     "e2e_host_limit"
#define HOST_LIMIT_RESNAME  "e2e_host_res"
#define HOST_LIMIT_PERCENT_VAL 50.0f
#define HOST_RES_CAPACITY   1000.0f

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

/* ---------------------------------------------------------------------- */
/* Fixture helpers -- capnp event writers duplicated from test_sim_live_    */
/* allocation.c's own proven shape (this directory's established           */
/* "duplicate, don't share" testsuite convention).                         */
/* ---------------------------------------------------------------------- */

static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
    char tmpl[PATH_MAX];

    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_e2elimit%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
}

/* register_plain_resource - same shape test_sim_contention_limit_driver_
 * parity.c's/test_sim_alloc_limit_fanout.c's own identically-named helper
 * establishes: a bare resource with no LIMIT/CONSUMABLE/host flags --
 * is_host_based_limit() (xcelimit.c) resolves host-vs-global scope purely
 * from limit->val.percent (plus, for a flagged resource, XCE_RESF_GLOBAL/
 * XCE_RESF_SHARED), never from any flag on the resource itself. */
static void register_plain_resource(var resmgr, const char* name)
{
    xce_resource_t res;

    memset(&res, 0, sizeof(res));
    res.name = strdup(name);

    TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res));
}

/* make_wire_worker_two_resvals - test_sim_live_allocation.c's own
 * make_wire_worker(), extended from one resval to two: this fixture's
 * worker needs both "slots" (the job's auto-injected per-task want) and
 * the host-scoped limit's own underlying resource (a real, non-zero
 * capacity -- without it, update_limit_res_value()'s get_resvals() call
 * against the worker's own resvals would fail before ever recording a
 * usage value, per plugins/limit.c's own body, read in full before writing
 * this fixture). */
static xce_worker_t* make_wire_worker_two_resvals(const char* name,
                                                   const char* res1name,
                                                   float res1val,
                                                   const char* res2name,
                                                   float res2val)
{
    xce_worker_t*    w       = (xce_worker_t*)calloc(1, sizeof(*w));
    xce_name_val_t** nv_ptrs = (xce_name_val_t**)calloc(2, sizeof(*nv_ptrs));
    xce_val_t*       val1    = (xce_val_t*)calloc(1, sizeof(*val1));
    xce_val_t*       val2    = (xce_val_t*)calloc(1, sizeof(*val2));
    xce_name_val_t*  nv1     = (xce_name_val_t*)calloc(1, sizeof(*nv1));
    xce_name_val_t*  nv2     = (xce_name_val_t*)calloc(1, sizeof(*nv2));

    TEST_ASSERT_NOT_NULL(w);
    TEST_ASSERT_NOT_NULL(nv_ptrs);
    TEST_ASSERT_NOT_NULL(val1);
    TEST_ASSERT_NOT_NULL(val2);
    TEST_ASSERT_NOT_NULL(nv1);
    TEST_ASSERT_NOT_NULL(nv2);

    val1->which    = Value_u_fvalue;
    val1->u.fvalue = res1val;
    nv1->name      = strdup(res1name);
    nv1->val       = val1;

    val2->which    = Value_u_fvalue;
    val2->u.fvalue = res2val;
    nv2->name      = strdup(res2name);
    nv2->val       = val2;

    nv_ptrs[0] = nv1;
    nv_ptrs[1] = nv2;

    w->name              = strdup(name);
    w->status            = 0;
    w->nresvals_to_sched = 2;
    w->resvals_to_sched  = nv_ptrs;

    return w;
}

static xce_queue_t* make_wire_queue(const char* name, int32_t priority)
{
    xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));

    TEST_ASSERT_NOT_NULL(q);

    q->queue    = strdup(name);
    q->status   = QUEUE_STAT_OPEN;
    q->priority = priority;

    return q;
}

static xce_job_t* make_push_job(uint64_t jobid, char* resreq_str, char* user,
                                char* queue, char* project, uint32_t mintasks,
                                uint32_t maxtasks)
{
    xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));

    TEST_ASSERT_NOT_NULL(j);

    j->jobid      = jobid;
    j->state      = JOB_STAT_PEND;
    j->sub_time   = 1000;
    j->queue_time = 1000;

    j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    TEST_ASSERT_NOT_NULL(j->submit);

    j->submit->queue        = queue;
    j->submit->user         = user;
    j->submit->project_name = project;
    j->submit->resreq       = resreq_str;
    j->submit->mintasks     = mintasks;
    j->submit->maxtasks     = maxtasks;

    return j;
}

static void append_resourcespush_event(FILE* f, uint64_t ts,
                                       xce_resource_t** resources, int n)
{
    xce_resource_data_t wrap;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr       p;

    memset(&wrap, 0, sizeof(wrap));
    wrap.n_total   = n;
    wrap.count     = n;
    wrap.resources = resources;

    memset(&src, 0, sizeof(src));
    src.ts                 = ts;
    src.body_which         = ReplayEvent_body_resourcesPush;
    src.body.resources_push = &wrap;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_workerspush_event(FILE* f, uint64_t ts,
                                     xce_worker_t** workers, int n)
{
    xce_workers_data_t   wrap;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr       p;

    memset(&wrap, 0, sizeof(wrap));
    wrap.n_total = n;
    wrap.count   = n;
    wrap.workers = workers;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_workersPush;
    src.body.workers_push = &wrap;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_queueupdate_event(FILE* f, uint64_t ts, xce_queue_t* q)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr       p;

    memset(&src, 0, sizeof(src));
    src.ts               = ts;
    src.body_which       = ReplayEvent_body_queueUpdate;
    src.body.queue_update = q;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_jobspush_event(FILE* f, uint64_t ts, xce_job_t** jobs,
                                  int njobs)
{
    xce_jobs_data_t      wrap;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr       p;

    memset(&wrap, 0, sizeof(wrap));
    wrap.n_total = njobs;
    wrap.count   = njobs;
    wrap.jobs    = jobs;

    memset(&src, 0, sizeof(src));
    src.ts             = ts;
    src.body_which     = ReplayEvent_body_jobsPush;
    src.body.jobs_push = &wrap;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* append_replay_session - EMPTY decisions list is load-bearing here, not
 * incidental -- see test_sim_live_allocation.c's own identical rationale
 * ("Why the fixture cannot simply replay a normal-mode recording"): an
 * empty decisions list guarantees sim_reconcile_unaccounted_jobs() finds
 * nothing to reconstruct, so this fixture's job reaches the live
 * sched_job_picker()/sched_allocator() pair genuinely, not via
 * reconciliation. */
static void append_replay_session(FILE* f, uint64_t begin_round)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplaySession_ptr    sp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    sp = new_ReplaySession(cs);
    ReplaySession_set_beginRound(sp, begin_round);
    capn_setp(capn_root(&c), 0, sp.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_end_of_stream_marker(FILE* f)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr       p;

    memset(&src, 0, sizeof(src));
    src.body_which = ReplayEvent_body_endSession;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* write_header_and_snapshot - produces a valid ReplayHeader+SnapshotEntry
 * prefix at a fresh temp path via the real sched_record_start(), same
 * throwaway-xce_scheduler_t technique test_sim_live_allocation.c's own
 * write_header_and_snapshot() establishes -- extended here to additionally
 * bake BOTH limit resources ("e2e_global_res"/"e2e_host_res") into the
 * throwaway core's own resmgr and set core.config to a real, 2-entry
 * xce_conf4sched_t (one global, one host-scoped limit) BEFORE
 * sched_record_start() takes the snapshot. This is what makes the snapshot
 * genuinely carry a Conf4Sched with both limits on the wire -- see this
 * file's own header comment ("Why config seeded via the snapshot") for the
 * full mechanism trace.
 *
 * The `config`/`limit_val`/`limit_cfg`/`limits` locals below need only
 * survive through sched_record_start()'s own synchronous call (which
 * capnp-encodes their contents into the wire buffer before returning) --
 * exactly like every other throwaway construction this function already
 * makes (`sched` itself, `fcfs`), all discarded the moment this function
 * returns. */
static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len)
{
    char                record_dir[PATH_MAX];
    xce_scheduler_t     sched;
    xce_sched_plugin_t* fcfs;
    xce_limit_val_t     global_val = { .resname = (char*)GLOBAL_LIMIT_RESNAME,
                                       .percent = 0,
                                       .val     = GLOBAL_LIMIT_VAL };
    xce_limit_config_t  global_cfg = { .name = (char*)GLOBAL_LIMIT_NAME,
                                       .val  = &global_val };
    xce_limit_val_t     host_val = { .resname = (char*)HOST_LIMIT_RESNAME,
                                     .percent = 1,
                                     .val     = HOST_LIMIT_PERCENT_VAL };
    xce_limit_config_t  host_cfg = { .name = (char*)HOST_LIMIT_NAME,
                                     .val  = &host_val };
    xce_limit_config_t* limits[2] = { &global_cfg, &host_cfg };
    xce_conf4sched_t    config;
    int                 rc;

    make_tmp_dir(record_dir, sizeof(record_dir), "_r");

    memset(&sched, 0, sizeof(sched));
    mylock_init(&(sched.record_lock));

    rc = xce_sched_core_init(&(sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    fcfs = builtin_fcfs_new_plugin();
    TEST_ASSERT_NOT_NULL(fcfs);
    rc = plugin_register_core(&(sched.core), fcfs);
    TEST_ASSERT_TRUE(rc >= 0);

    /* Both limit resources must exist in resmgr BEFORE sim_bootstrap_
     * limit_plugin() runs (sched_sim_driver_run() calls it immediately
     * after driver_start()/snapshot-load, strictly before any replay event
     * -- including a RESOURCES_PUSH for "slots" -- is ever processed), so
     * they are baked into the snapshot itself rather than pushed as a
     * later wire event. */
    register_plain_resource(sched.core.resmgr, GLOBAL_LIMIT_RESNAME);
    register_plain_resource(sched.core.resmgr, HOST_LIMIT_RESNAME);

    memset(&config, 0, sizeof(config));
    config.nlimits = 2;
    config.limits  = limits;
    sched.core.config = &config;

    sched.core.job_pick_plugin = (char*)"builtin_fcfs";
    sched.record_dir           = record_dir;

    rc = sched_record_start(&sched, "e2e_limit.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len);
    strcpy(out_path, sched.record_file_path);

    return sched.record_file;
}

/* ---------------------------------------------------------------------- */
/* The sim driver under test -- static storage, see this file's own header */
/* comment ("bug-1598/1632/1763 fixture-safety requirement") for why.      */
/* ---------------------------------------------------------------------- */

static xce_sched_sim_driver_t g_sim;

/* ---------------------------------------------------------------------- */
/* The test                                                                */
/* ---------------------------------------------------------------------- */

/*
 * Observable (task 5.2's own text): "the existing limit-inspection command
 * displays real, non-zero usage for both limits against this replay, where
 * today it reports no active usage instance for either". This is bug-1826
 * itself, proven end to end: a real `.replay` file, a real
 * sched_sim_driver_run() invocation, and both a global and a host-scoped
 * limit that genuinely match the one dispatched job -- not two separate
 * fixtures, not a bare-core shortcut.
 */
static void
test_sim_replay_shows_real_global_and_host_scoped_limit_usage(void)
{
    char             path[PATH_MAX];
    FILE*            f;
    xce_resource_t*  slots_res;
    xce_resource_t*  res_ptrs[1];
    xce_worker_t*    worker_ptrs[1];
    xce_queue_t*     queue;
    xce_job_t*       job_ptrs[1];
    sched_sim_args_t args;
    xce_sched_job_t* sjob;
    xce_sched_worker_t* worker;
    var              limits_handle;
    var              instances_handle;
    iterator_t       it;
    int              n_configured = 0;
    int              n_instances  = 0;
    int              saw_global   = 0;
    int              saw_host     = 0;
    int              rc;

    f = write_header_and_snapshot(path, sizeof(path));

    /* "slots" -- the job's auto-injected per-task want -- arrives as an
     * ordinary post-snapshot wire event, the same proven shape
     * test_sim_live_allocation.c's own fixture already establishes. */
    slots_res = (xce_resource_t*)calloc(1, sizeof(*slots_res));
    TEST_ASSERT_NOT_NULL(slots_res);
    slots_res->name = strdup("slots");
    res_ptrs[0]     = slots_res;
    append_resourcespush_event(f, 1001ULL, res_ptrs, 1);

    /* The worker carries real, sufficient capacity for BOTH the job's
     * "slots" want (so the job genuinely DISPATCHES -- the strongest
     * possible proof, matching test_sim_live_allocation.c's own precedent)
     * and the host-scoped limit's own underlying resource (a generous,
     * non-zero capacity, so the limit itself never denies anything -- the
     * same "generous cap" discipline test_sim_contention_limit_driver_
     * parity.c's own fixture documents). */
    worker_ptrs[0] = make_wire_worker_two_resvals(
        WORKER_NAME, "slots", WORKER_SLOTS, HOST_LIMIT_RESNAME,
        HOST_RES_CAPACITY);
    append_workerspush_event(f, 1002ULL, worker_ptrs, 1);

    queue = make_wire_queue(QUEUE_NAME, 1);
    append_queueupdate_event(f, 1003ULL, queue);

    /* A single job is enough to exercise BOTH limits: plugin_job_election()
     * (global-limit match) fires once per job before any allocation
     * attempt, unconditionally; plugin_prepare_allocation() (host-scoped
     * match) fires once per candidate host considered during the
     * allocation search that follows -- both permissive-scoped limits
     * (no per-user/queue/project/host restriction configured) match any
     * job/host, per this codebase's own build_consumer() "all"-group
     * fallback, the same default every sibling limit fixture in this
     * feature already relies on. */
    job_ptrs[0] = make_push_job(TEST_JOBID, TEST_RESREQ, (char*)"alice",
                                (char*)QUEUE_NAME, (char*)"proj1", 1, 1);
    append_jobspush_event(f, 1004ULL, job_ptrs, 1);

    /* Empty decisions -- see this file's own header comment ("Why config
     * seeded via the snapshot") and test_sim_live_allocation.c's identical
     * rationale for why this is load-bearing, not incidental. */
    append_replay_session(f, 2000ULL);
    append_end_of_stream_marker(f);
    fclose(f);

    memset(&args, 0, sizeof(args));
    args.replay_path  = path;
    args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    args.sim_speed    = 0;

    rc = sched_sim_driver_run(&g_sim, &args);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc, "sched_sim_driver_run() must complete this fixture's single "
               "round and reach its own end-of-stream marker");

    sjob = sched_get_job(&(g_sim.core), TEST_JOBID);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        sjob, "the job pushed via the real JOBS_PUSH ingest path must be "
              "present in the simulation driver's own core state");
    TEST_ASSERT_EQUAL_UINT_MESSAGE(
        JOB_STAT_RUN, sjob->state,
        "the job must genuinely DISPATCH -- a real worker with sufficient "
        "'slots' capacity is provided specifically so both limit hooks "
        "(job election AND host-prepare-allocation) are reached along the "
        "real live dispatch path, not merely a PEND that happens to touch "
        "one of them");

    worker = sched_get_worker(&(g_sim.core), (char*)WORKER_NAME);
    TEST_ASSERT_NOT_NULL(worker);

    /* ---- `info limit all` today reports "no limits configured" for a
     * replay -- confirm both limits are genuinely configured, seeded from
     * this replay's own snapshot config (sim_bootstrap_limit_plugin(),
     * bug-1825's mechanism, extended by this feature to actually MATCH,
     * not merely display). ---- */
    limits_handle = limit_get_configured_limits(&(g_sim.core));
    TEST_ASSERT_NOT_NULL(limits_handle);

    rc = iter_make(&it, limits_handle);
    TEST_ASSERT_TRUE(rc >= 0);
    for (;;) {
        void* citem = NULL;

        rc = iter_next(&it, &citem);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }
        n_configured++;
    }
    iter_destroy(&it);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        2, n_configured,
        "both the global and host-scoped limit must be configured, seeded "
        "from the replay's own snapshot config -- Requirement 2.3's own "
        "'info limit all' resolution depends on this");

    /* ---- THE acceptance test: `info limit` must show REAL, non-zero
     * usage for BOTH limits -- where today (before this feature) it
     * reports no active usage instance for either (Requirement 2.1, 2.2,
     * 2.3). ---- */
    instances_handle = limit_get_usage_instances(&(g_sim.core));
    TEST_ASSERT_NOT_NULL_MESSAGE(
        instances_handle,
        "before this feature, plugin->sched was NULL for every "
        "sim-registered plugin and no usage instance could ever be "
        "created against a replay -- info limit reported none active for "
        "either limit");

    rc = iter_make(&it, instances_handle);
    TEST_ASSERT_TRUE(rc >= 0);
    for (;;) {
        xce_vres_t* vres  = NULL;
        float       value = -1.0f;
        int         grc;

        rc = iter_next(&it, (void**)&vres);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }
        TEST_ASSERT_NOT_NULL(vres);
        TEST_ASSERT_NOT_NULL(vres->limit);
        TEST_ASSERT_NOT_NULL(vres->vres);

        n_instances++;

        if (vres->limit->val.percent == 0) {
            saw_global = 1;
            grc = get_resvals(&(g_sim.core.sharedInst), 1,
                              &(vres->vres->idx), &value);
            TEST_ASSERT_EQUAL_INT_MESSAGE(
                1, grc,
                "the global limit's usage value must be readable back from "
                "core->sharedInst -- exactly what plugin_job_election() -> "
                "match_job_global_limit_() -> update_limit_res_value() "
                "wrote into, the same array 'info limit' reads to display "
                "global usage");
            TEST_ASSERT_TRUE_MESSAGE(
                value > 0.0f,
                "Requirement 2.1: the global limit's usage must be real "
                "and non-zero, not merely an empty/zeroed instance");
        } else {
            saw_host = 1;
            grc = get_resvals(&(worker->resvals), 1, &(vres->vres->idx),
                              &value);
            TEST_ASSERT_EQUAL_INT_MESSAGE(
                1, grc,
                "the host-scoped limit's usage value must be readable back "
                "from the dispatched-onto worker's own resvals -- exactly "
                "what plugin_prepare_allocation() -> match_job_host_limit_"
                "() -> update_limit_res_value() wrote into");
            TEST_ASSERT_TRUE_MESSAGE(
                value > 0.0f,
                "Requirement 2.2: the host-scoped limit's usage must be "
                "real and non-zero, not merely an empty/zeroed instance");
        }
    }
    iter_destroy(&it);

    TEST_ASSERT_TRUE_MESSAGE(
        saw_global,
        "Requirement 2.1/2.3: the global limit must have a real usage "
        "instance with non-zero usage, visible via the same accessor "
        "'info limit' reads -- this is bug-1826's own global-limit half");
    TEST_ASSERT_TRUE_MESSAGE(
        saw_host,
        "Requirement 2.2/2.3: the host-scoped limit must have a real "
        "usage instance with non-zero usage, visible via the same "
        "accessor 'info limit' reads -- this is bug-1826's own "
        "host-scoped-limit half");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        2, n_instances,
        "exactly one usage instance per configured limit -- no spurious "
        "extra instance, no limit silently missing one (Requirement 2.4's "
        "own 'a never-matched limit reports none' guarantee, checked here "
        "from the opposite direction: a limit that DID match reports "
        "exactly one)");

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

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_sim_replay_shows_real_global_and_host_scoped_limit_usage);
    return UNITY_END();
}