myxce/src/scheduler/src/repl_info.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "controller.capnp.h"
#include "repl.h"
#include "repl_internal.h"
#include "resreq.h"
#include "utils.h"

/*
 * repl_info.c - state inspection (scheduler-core spec, task 12.2, Req 5 AC
 * 22 "State inspection"; `info config` added by task 12.3, Req 5 AC 22
 * "Simulation parameters"). Sibling of repl.c/repl_break.c/repl_params.c --
 * see repl_internal.h's own "Why a split at all" section for the full
 * rationale behind this four-file layout.
 *
 * Owns: `info session`/`info job`/`info jobs`/`info queue`/`info worker`/
 * `info resource`/`info summary`/`info break`/`info config`, and the
 * `repl_cmd_info()` dispatcher repl.c's repl_dispatch() hands off to
 * (repl_internal.h's own extern declaration). `info config` is the
 * read-only counterpart to repl_params.c's `set speed`/`set output`/
 * `set verify` -- grouped here with every other `info` dump rather than in
 * repl_params.c, per this file's own domain-ownership scope.
 *
 * Every dump below reads core->jobs/queues/workers/resmgr the SAME way
 * repl.c's own repl_count_jobs() and snapshot.c's
 * xce_sched_core_save_snapshot() already do -- no new traversal mechanism.
 * Field selection is deliberately cross-checked against
 * xce_sched_core_save_snapshot()'s own documented field set (snapshot.c's
 * file header comment, "Field-by-field mapping") rather than invented
 * fresh, matching AC 22's own text: inspection output should "report state
 * matching what a snapshot dump of the same moment would show." Where a
 * requested field genuinely does not exist in this codebase's data model
 * yet (e.g. xce_sched_queue_t has no RUNLIMIT/MAX_RUNNING field), that gap
 * is reported honestly ("n/a"), never fabricated -- the same discipline
 * snapshot.c's own "Scope decisions" section already establishes for
 * job-level priority.
 *
 * Boundary (Req 5 AC 17): same as repl.c/repl_break.c -- only
 * xce_sched_sim_driver_t/xce_sched_core_t (via repl.h/repl_internal.h ->
 * sim.h -> core.h) plus stdio/string.h/stdlib.h, plus resreq.h for
 * format_resreq() (`info job`'s resreq= field, see repl_info_job() below).
 * resreq.h carries no capnp/controller/client symbols of its own (its
 * #include set is kvmgr.h/xlist.h/xvec.h plus libc headers) and is already
 * part of this same isolation-boundary family -- snapshot_wire.c's own
 * build_wire_job() (task 8.1) already includes it and calls format_resreq()
 * for the identical purpose.
 */

/* dup_val/arr_append/find_resource_name_by_idx -- implemented in
 * snapshot_wire.c (task 8.1's own 800-line-cap split); redeclared here
 * rather than pulling in a private header, same "extern declared where
 * consumed" idiom snapshot.c itself already uses for these exact symbols.
 * Only arr_append/find_resource_name_by_idx are needed here (`info worker`/
 * `info resource` resolve a resource idx to its name the SAME way
 * xce_sched_core_save_snapshot() already does, per this file's own header
 * comment). */
extern int         arr_append(void***, int*, int*, void*);
extern const char* find_resource_name_by_idx(xce_resource_t**, int, int);

/* repl_job_state_name - a single readable label for `info job`'s state=
 * field (task 12.2, AC 22's "info job <jobid>: Dump one job: state, ..."
 * row). Priority order matches this codebase's own convention for a job
 * whose `state` bitmask carries more than one bit (e.g. PEND|PSUSP): the
 * most "advanced" lifecycle bit wins, mirroring repl_break.c's own
 * repl_job_state_matches() check order (used by `break ... state=`
 * filters) so a filter and this display never disagree about which single
 * word describes a given state. Non-static (repl_internal.h) since task
 * 12.5's per-event summary line (repl_break.c) shares it for
 * JOB_STATE_CHANGE. */
const char* repl_job_state_name(uint32_t state)
{
    if (JOB_IS_RUN(state)) {
        return "RUN";
    }
    if (JOB_IS_EXIT(state)) {
        return "EXIT";
    }
    if (JOB_IS_DONE(state)) {
        return "DONE";
    }
    if (JOB_IS_PEND(state)) {
        return "PEND";
    }
    return "UNKNOWN";
}

/*
 * repl_info_job - AC 22's "info job <jobid>": "Dump one job: state, queue,
 * priority, queue_time, resreq, allocations." Field set cross-checked
 * against xce_sched_core_save_snapshot()'s own documented mapping
 * (snapshot.c file header, "Field-by-field mapping", jobs row) rather than
 * invented fresh -- e.g. `resreq` is formatted via format_resreq()
 * (resreq.h), the SAME print-form snapshot_wire.c's own build_wire_job()
 * (task 8.1) already calls to populate a restored job's `submit->resreq`
 * wire field, not a fabricated format. (job-level priority remains the one
 * genuine gap snapshot.c's own "Scope decisions" section documents -- no
 * wire slot exists for it at all, unlike resreq.) sched_get_job() (core.h)
 * is the exact accessor xce_core_push_events()'s own JOB_STATE_CHANGE/
 * JOB_UPDATE handling (core_events.c) already uses.
 */
static void repl_info_job(xce_sched_sim_driver_t* sim, const char* jobid_str)
{
    uint64_t         jobid      = 0;
    int              parsed;
    xce_sched_job_t* job        = NULL;
    char*            resreq_str = NULL;
    char             idbuf[64];

    /* Accept human-readable "0-7-1786350309-0" format (parse_jobid) or
     * raw packed 64-bit integer (strtoull fallback for backward compat). */
    parsed = parse_jobid(&jobid, (char*)jobid_str);
    if (parsed <= 0) {
        jobid = strtoull(jobid_str, NULL, 10);
    }

    job = sched_get_job(&(sim->core), jobid);

    if (job == NULL) {
        fmt_jobid(idbuf, sizeof(idbuf), jobid);
        fprintf(stdout, "[sim] info job: no such job %llu (%s)\n",
               (unsigned long long)jobid, idbuf);
        return;
    }

    /* Same format_resreq() call shape/error-handling build_wire_job()
     * (snapshot_wire.c) already uses to populate a restored job's
     * `submit->resreq` wire field -- WARN + fall back to "(none)" on
     * failure, never fabricate a format. Unlike build_wire_job() (whose
     * caller owns the string via j->submit->resreq for the lifetime of the
     * wire job), this dump is transient -- resreq_str is freed below right
     * after printing. */
    if (job->resreq != NULL) {
        int rc = format_resreq(&resreq_str, job->resreq);

        if (rc < 0) {
            WARN("%s format_resreq failed for jobid %llu (rc=%d); resreq "
                 "will be reported as unavailable for this dump",
                 __func__, (unsigned long long)job->jobid, rc);
            resreq_str = NULL;
        }
    }

    fmt_jobid(idbuf, sizeof(idbuf), job->jobid);
    fprintf(stdout, "[sim] job %llu (%s): state=%s queue=%s priority=%d queue_time=%lld "
          "resreq=%s allocs=%d\n",
          (unsigned long long)job->jobid, idbuf, repl_job_state_name(job->state),
          (job->queue != NULL) ? job->queue->name : "(none)", job->priority,
          (long long)job->queue_time,
          (resreq_str != NULL) ? resreq_str : "(none)",
          (job->allocs != NULL) ? job->allocs->nhosts : 0);

    FREEUP(resreq_str);
}

/*
 * repl_info_jobs - list all job IDs with a one-line summary per job.
 * Iteration mirrors repl_count_jobs() in repl.c.  Each line prints the
 * packed uint64_t (usable directly in `info job <id>`) plus state and
 * queue — compact enough for a few dozen jobs without scrolling off
 * screen. */
static void repl_info_jobs(xce_sched_sim_driver_t* sim)
{
    xce_sched_core_t* core = &(sim->core);
    iterator_t        it;
    int               count = 0;
    int               rc;
    char              idbuf[64];

    rc = iter_make(&it, core->jobs);
    if (rc < 0) {
        fprintf(stdout, "[sim] info jobs: cannot iterate job table\n");
        return;
    }

    fprintf(stdout, "[sim] jobs:\n");
    for (;;) {
        xce_sched_job_t* job = NULL;

        rc = iter_next(&it, (void**)&job);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA) || (job == NULL)) {
            break;
        }
        count++;
        fmt_jobid(idbuf, sizeof(idbuf), job->jobid);
        fprintf(stdout, "  %-25s  %s  %s\n",
               idbuf,
               repl_job_state_name(job->state),
               (job->queue != NULL) ? job->queue->name : "-");
    }

    fprintf(stdout, "[sim] info jobs: %d job(s)\n", count);
}

/*
 * repl_info_queue_one - AC 22's "info queue <name>": "Dump one queue:
 * priority, state (open/closed), RUNLIMIT, MAX_RUNNING, job_pick_plugin,
 * list of queued job IDs with priorities." xce_sched_queue_t (sched/types.h)
 * has no RUNLIMIT/MAX_RUNNING field at all -- confirmed against both the
 * live struct and its wire xce_queue_t counterpart -- the same
 * pre-existing, documented gap snapshot.c's own "Scope decisions" section
 * already notes for other AC 21/22-adjacent fields (e.g. job-level
 * priority) that requirements.md names but no C struct in this codebase
 * carries yet. Reported honestly as "n/a" rather than a fabricated number.
 * "queued" below means JOB_IS_PEND() -- a running job is not queued.
 */
static void repl_info_queue_one(xce_sched_sim_driver_t* sim,
                                xce_sched_queue_t*      q)
{
    iterator_t       it;
    xce_sched_job_t* job = NULL;
    int              rc;

    fprintf(stdout, "[sim] queue %s: priority=%d status=%s RUNLIMIT=n/a "
          "MAX_RUNNING=n/a job_pick_plugin=%s\n",
          q->name, q->priority,
          ((q->status & QUEUE_STAT_CLOSED) != 0) ? "closed" : "open",
          (q->job_pick_plugin != NULL) ? q->job_pick_plugin : "(inherit)");

    rc = iter_make(&it, sim->core.jobs);
    if (rc < 0) {
        return;
    }
    for (;;) {
        rc = iter_next(&it, (void**)&job);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }
        if ((job->queue != NULL) && (strcmp(job->queue->name, q->name) == 0)
            && JOB_IS_PEND(job->state)) {
            char jobidStr[64];

            /* bug-1591: see repl_info_worker_one()'s own comment on
             * fmt_jobid() (below) -- same fix, same rationale. */
            fmt_jobid(jobidStr, sizeof(jobidStr), job->jobid);
            fprintf(stdout, "[sim]   queued job: jobid=%s priority=%d\n",
                  jobidStr, job->priority);
        }
    }
    iter_destroy(&it);
}

/* repl_info_queue - `info queue <name|all>` dispatch. "all" walks
 * core->queues (same iter_make()/iter_next() idiom every other walk in this
 * file uses); a specific name goes through sched_get_queue() (core.h). */
static void repl_info_queue(xce_sched_sim_driver_t* sim, const char* arg)
{
    if (strcmp(arg, "all") == 0) {
        iterator_t         it;
        xce_sched_queue_t* q  = NULL;
        int                rc = iter_make(&it, sim->core.queues);

        if (rc < 0) {
            return;
        }
        for (;;) {
            rc = iter_next(&it, (void**)&q);
            if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                break;
            }
            repl_info_queue_one(sim, q);
        }
        iter_destroy(&it);
        return;
    }

    {
        xce_sched_queue_t* q = sched_get_queue(&(sim->core), (char*)arg);

        if (q == NULL) {
            fprintf(stdout, "[sim] info queue: no such queue '%s'\n", arg);
            return;
        }
        repl_info_queue_one(sim, q);
    }
}

/*
 * repl_info_worker_one - AC 22's "info worker <name>": "Dump one worker:
 * state, resources, assigned jobs." `w->resvals` holds the resource's
 * STATIC TOTAL capacity (as self-reported by the worker daemon at join/
 * heartbeat time -- WORKERS_PUSH/RESVALS_DELTA are the only writers,
 * core_events.c -- and NEVER decremented as jobs consume it); `w->resInUse`
 * holds the currently-committed amount at the same index. Only a resource
 * flagged XCE_RESF_CONSUMABLE has a capacity/in-use split worth reporting --
 * its value= is the raw total (`w->resvals[idx]`, already inclusive of
 * whatever is in use -- see resource.c's unsatisfy_(), whose admission
 * formula `total[i] < inuse[i] + uncomitted[i] + wanted[i]` is the
 * authoritative proof this field is a total, not a live remaining-capacity
 * counter) and used= is inUse; a non-consumable numeric resource (a static
 * per-worker attribute nothing ever allocates against) gets a single value=
 * line instead, mirroring repl_info_resource_one()'s identical CONSUMABLE
 * gate for the cluster-wide case.
 *
 * bug-1727 (2026-08-30, CONFIRMED + FIXED): this comment used to
 * claim value= should be printed as available+inUse ("a conservation
 * invariant this codebase's own resource accounting already relies on") --
 * that claim was FALSE for how w->resvals is actually populated (it is
 * never decremented at charge time, so it already equals the total, not a
 * live "available" residue), and printing avail+used double-counted `used`
 * on top of a value that already included it. Live-reproduced against a
 * real 3-node cluster + worker (50-element overlapping-runtime array job):
 * a fully-packed 22-slot worker showed "value=44.00 used=22.00" instead of
 * the correct "value=22.00 used=22.00" -- exactly 22 (never-decremented
 * total) + 22 (used) = 44, matching this bug's mechanism precisely. Fixed
 * by printing avail_v as-is (no addition) at both this call site and
 * repl_info_resource_one()'s identical shared-resource case.
 *
 * Resource names/flags are resolved against a resmgr walk built the same
 * way xce_sched_core_save_snapshot() builds its own `res_arr` (snapshot.c)
 * -- not a new lookup mechanism. "assigned jobs" scans core->jobs for any
 * per-host allocation naming this worker (xce_job_alloc_t.allocs[]->host,
 * xce/types.h).
 */
static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
                                 xce_sched_worker_t*     w)
{
    xce_resource_t** res_arr = NULL;
    int              res_cap = 0;
    int              res_n   = 0;
    iterator_t       it;
    xce_resource_t*  r = NULL;
    xce_sched_job_t* job;
    int              rc;
    int              i;

    fprintf(stdout, "[sim] worker %s: status=%s\n", w->name,
          (IS_WORKER_UNAVAIL(w->status)) ? "unavail" : "online");

    rc = iter_make(&it, sim->core.resmgr);
    if (rc >= 0) {
        for (;;) {
            rc = iter_next(&it, (void**)&r);
            if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                break;
            }
            arr_append((void***)&res_arr, &res_cap, &res_n, r);
        }
        iter_destroy(&it);
    }

    /* bug-1688: `.count` (number of currently-populated slots) is NOT a
     * safe index upper bound for iterating this zarray BY INDEX -- it can
     * legitimately be less than the highest occupied index once entries
     * are sparse/non-contiguous (e.g. zarray_put_ex()'s count bookkeeping,
     * zarray.c, only tracks "how many slots read NULL at write time", not
     * "the current occupied-index high-water mark"). `.caps` (allocated
     * capacity, monotonic, never shrinks) is the correct bound for
     * index-based iteration -- mirrors resource.c:228's
     * `int num_res = (int)w->resvals.caps;` for this identical structure.
     * A `.caps`-bound loop can hit legitimately-empty slots that a
     * `.count`-bound loop never would, so avail_v/rname's existing
     * NULL-skip below (unchanged) is what keeps this safe. */
    for (i = 0; i < w->resvals.caps; i++) {
        xce_val_t*      avail_v = (xce_val_t*)zarray_get(&(w->resvals), i);
        xce_val_t*      used_v  = (xce_val_t*)zarray_get(&(w->resInUse), i);
        const char*     rname   = find_resource_name_by_idx(res_arr, res_n, i);
        xce_resource_t* rdef    = NULL;
        int              j;

        if ((avail_v == NULL) || (rname == NULL)) {
            continue;
        }

        for (j = 0; j < res_n; j++) {
            if (res_arr[j]->idx == i) {
                rdef = res_arr[j];
                break;
            }
        }

        /* bug-1590: xce_val_t.which (a Value_u_which wire tag,
         * controller.capnp.h) tells us the value's REAL type -- bug-1568
         * fixed comparing it against the wrong enum for the numeric case,
         * but left every non-numeric resource (Value_u_svalue/
         * Value_u_bvalue -- e.g. this worker's `type`/`model` string
         * resources, or any boolean-typed resource) silently coerced to
         * 0.0 by the old unconditional %.2f formatting below. A value/used
         * split is only meaningful for a CONSUMABLE numeric resource -- a
         * string/boolean resource, or a non-consumable numeric one (a
         * static per-worker attribute nothing ever allocates against), gets
         * its own single "value=" line instead. */
        switch (avail_v->which) {
        case Value_u_svalue:
            fprintf(stdout, "[sim]   resource %s: value=%s\n", rname,
                  avail_v->u.svalue ? avail_v->u.svalue : "(null)");
            break;

        case Value_u_bvalue:
            fprintf(stdout, "[sim]   resource %s: value=%s\n", rname,
                  avail_v->u.bvalue ? "true" : "false");
            break;

        case Value_u_fvalue:
            if ((rdef != NULL) && (rdef->flags & XCE_RESF_CONSUMABLE)) {
                /* bug-1727: w->resvals[idx] (avail_v here) is the resource's
                 * STATIC TOTAL capacity, not a live "remaining available"
                 * quantity that decreases as jobs consume it -- confirmed
                 * against the actual admission-control formula this exact
                 * field feeds, resource.c's unsatisfy_(): `total[i] <
                 * inuse[i] + uncomitted[i] + wanted[i]` treats
                 * get_resvals(&w->resvals, ...)'s output as `total` directly,
                 * comparing it against inuse (not total-inuse already
                 * subtracted from avail_v). Nothing else ever decrements
                 * w->resvals[idx] as jobs run (core_apply_worker_resvals()
                 * only overwrites it wholesale on a WORKERS_PUSH/RESVALS_
                 * DELTA re-announce, and built-in load indices --
                 * resource_sync.c's init_builtin_load_indices() -- never
                 * include "slots"). Printing avail+used therefore double-
                 * counts `used` on top of a value that is ALREADY the
                 * total, growing without bound as more jobs run --
                 * live-reproduced and matching the user's exact reported
                 * numbers (value=44.00 used=22.00 on a 22-slot worker at
                 * full occupancy: 22 (never-decremented total) + 22 (used)
                 * = 44). Print avail_v as-is. */
                double avail = avail_v->u.fvalue;
                double used  = ((used_v != NULL) &&
                               (used_v->which == Value_u_fvalue))
                                  ? used_v->u.fvalue
                                  : 0.0;

                fprintf(stdout,
                      "[sim]   resource %s: value=%.2f used=%.2f\n", rname,
                      avail, used);
            } else {
                fprintf(stdout, "[sim]   resource %s: value=%.2f\n", rname,
                      avail_v->u.fvalue);
            }
            break;

        default:
            /* ivalue32/uvalue32/ivalue64/uvalue64 -- not a shape this
             * codebase's resource system currently produces for a resval
             * (xce_resource_t.valtype, xce/types.h, only ever numeric/
             * string/boolean); reported honestly rather than guessed at,
             * same "silently skipped ... rather than guessed at" posture
             * core_events.c's bug-1573 fix already established for this
             * identical ambiguity. */
            break;
        }
    }
    free(res_arr);

    job = NULL;
    rc  = iter_make(&it, sim->core.jobs);
    if (rc < 0) {
        return;
    }
    for (;;) {
        rc = iter_next(&it, (void**)&job);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }
        if (job->allocs == NULL) {
            continue;
        }
        for (i = 0; i < job->allocs->nhosts; i++) {
            xce_host_alloc_t* ha = job->allocs->allocs[i];

            if ((ha != NULL) && (ha->host != NULL)
                && (strcmp(ha->host, w->name) == 0)) {
                char jobidStr[64];

                /* bug-1591: fmt_jobid() renders the raw idgen()-packed
                 * jobid in the same human-readable form already used by
                 * every other jobid display in this codebase (scheduler.c/
                 * handler.c/plugins, worker, xce_test's waldump, and
                 * job.c's log lines since bug-1588) -- this REPL command
                 * was one of the few remaining raw-integer holdouts. */
                fmt_jobid(jobidStr, sizeof(jobidStr), job->jobid);
                fprintf(stdout, "[sim]   assigned job: jobid=%s\n", jobidStr);
                break;
            }
        }
    }
    iter_destroy(&it);
}

/* repl_info_worker - `info worker <name|all>` dispatch, same "all" vs.
 * sched_get_worker() split as repl_info_queue() above. */
static void repl_info_worker(xce_sched_sim_driver_t* sim, const char* arg)
{
    if (strcmp(arg, "all") == 0) {
        iterator_t          it;
        xce_sched_worker_t* w  = NULL;
        int                 rc = iter_make(&it, sim->core.workers);

        if (rc < 0) {
            return;
        }
        for (;;) {
            rc = iter_next(&it, (void**)&w);
            if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                break;
            }
            repl_info_worker_one(sim, w);
        }
        iter_destroy(&it);
        return;
    }

    {
        xce_sched_worker_t* w = sched_get_worker(&(sim->core), (char*)arg);

        if (w == NULL) {
            fprintf(stdout, "[sim] info worker: no such worker '%s'\n", arg);
            return;
        }
        repl_info_worker_one(sim, w);
    }
}

/* XCE_RESF_FLAG_TABLE - every xce/types.h XCE_RESF_* bit, human name.
 * Shared by repl_info_resource_one()'s metadata line (task 12.5, bug-1593). */
static const struct {
    int         bit;
    const char* name;
} XCE_RESF_FLAG_TABLE[] = {
    { XCE_RESF_BUILTIN, "BUILTIN" },       { XCE_RESF_DYNAMIC, "DYNAMIC" },
    { XCE_RESF_GLOBAL, "GLOBAL" },         { XCE_RESF_SHARED, "SHARED" },
    { XCE_RESF_EXTERNAL, "EXTERNAL" },     { XCE_RESF_RELEASE, "RELEASE" },
    { XCE_RESF_DEFAULT_HOST, "DEFAULT_HOST" },
    { XCE_RESF_CONSUMABLE, "CONSUMABLE" }, { XCE_RESF_LIMIT, "LIMIT" },
};

/* repl_resource_flags_str - renders `flags` as a comma-joined list of the
 * XCE_RESF_* names it carries ("GLOBAL,SHARED"), or "(none)" if none set. */
static void repl_resource_flags_str(int flags, char* buf, size_t bufsz)
{
    size_t i;
    size_t n = 0;

    buf[0] = '\0';
    for (i = 0; i < (sizeof(XCE_RESF_FLAG_TABLE) / sizeof(XCE_RESF_FLAG_TABLE[0]));
         i++) {
        int written;

        if ((flags & XCE_RESF_FLAG_TABLE[i].bit) == 0) {
            continue;
        }
        written = snprintf(buf + n, bufsz - n, "%s%s", (n > 0) ? "," : "",
                           XCE_RESF_FLAG_TABLE[i].name);
        if ((written > 0) && ((size_t)written < (bufsz - n))) {
            n += (size_t)written;
        }
    }
    if (n == 0) {
        snprintf(buf, bufsz, "(none)");
    }
}

/*
 * repl_info_resource_one - AC 22 amendment (task 12.5, bug-1593): "info
 * resource <name>" reports the resource DEFINITION's own metadata (desc,
 * value type, order type, XCE_RESF_* flags, reporting interval) -- never a
 * per-host aggregation, which is repl_info_worker_one()'s job. A resource
 * flagged XCE_RESF_GLOBAL or XCE_RESF_SHARED additionally gets its current
 * value, read directly from the core's single defining pool
 * (`core->sharedInst`, indexed by `r->idx` -- the same pool RESVALS_FULL/
 * RESVALS_DELTA event handling and snapshot save/load already treat as
 * authoritative for these resources, core_events.c/snapshot.c). A host-based
 * resource (the common case -- cpus, mem, ...) has no single cluster-wide
 * value to report here; its per-host values remain `info worker <name>`'s
 * job. If the resource is additionally XCE_RESF_CONSUMABLE, the numeric
 * value line also reports how much is currently in use, read from
 * `core->sharedInUse` at the same index (mirrors repl_info_worker_one()'s
 * resInUse-derived split below) -- a non-consumable numeric resource has no
 * "used" counterpart, so it keeps a single value= line.
 */
static void repl_info_resource_one(xce_sched_sim_driver_t* sim,
                                   xce_resource_t*         r)
{
    char        flags_str[160];
    const char* valtype_str;
    const char* ordtype_str;

    switch (r->valtype) {
    case XCE_VAL_STRING:
        valtype_str = "string";
        break;
    case XCE_VAL_BOOLEAN:
        valtype_str = "boolean";
        break;
    case XCE_VAL_NUMERIC:
    default:
        valtype_str = "numeric";
        break;
    }

    switch (r->ordtype) {
    case XCE_ORD_INCR:
        ordtype_str = "incr";
        break;
    case XCE_ORD_DECR:
        ordtype_str = "decr";
        break;
    case XCE_ORD_NA:
    default:
        ordtype_str = "na";
        break;
    }

    repl_resource_flags_str(r->flags, flags_str, sizeof(flags_str));

    fprintf(stdout,
          "[sim] resource %s: desc=%s valtype=%s ordtype=%s flags=%s "
          "interval=%d\n",
          r->name, (r->desc != NULL) ? r->desc : "(none)", valtype_str,
          ordtype_str, flags_str, r->interval);

    if ((r->flags & (XCE_RESF_GLOBAL | XCE_RESF_SHARED)) == 0) {
        fprintf(stdout, "[sim]   host-based resource -- see 'info worker "
              "<name>' for per-host values\n");
        return;
    }

    if ((r->idx < 0) || (r->idx >= sim->core.sharedInst.count)) {
        fprintf(stdout, "[sim]   value=(none)\n");
        return;
    }

    {
        xce_val_t* v = (xce_val_t*)zarray_get(&(sim->core.sharedInst), r->idx);

        if (v == NULL) {
            fprintf(stdout, "[sim]   value=(none)\n");
            return;
        }

        switch (v->which) {
        case Value_u_svalue:
            fprintf(stdout, "[sim]   value=%s\n",
                  v->u.svalue ? v->u.svalue : "(null)");
            break;
        case Value_u_bvalue:
            fprintf(stdout, "[sim]   value=%s\n",
                  v->u.bvalue ? "true" : "false");
            break;
        case Value_u_fvalue:
            /* A consumable resource has a capacity/in-use split worth
             * reporting; a non-consumable numeric resource (e.g. a static
             * cluster-wide attribute) has no "used" counterpart, so it gets
             * a single value= line like the string/boolean cases above. */
            if (r->flags & XCE_RESF_CONSUMABLE) {
                xce_val_t* used_v = NULL;
                double     used   = 0.0;

                if ((r->idx >= 0) &&
                    (r->idx < sim->core.sharedInUse.count)) {
                    used_v = (xce_val_t*)zarray_get(&(sim->core.sharedInUse),
                                                    r->idx);
                }
                if ((used_v != NULL) && (used_v->which == Value_u_fvalue)) {
                    used = used_v->u.fvalue;
                }

                /* bug-1727: same double-count fix as
                 * repl_info_worker_one() above -- core->sharedInst[idx] (v
                 * here) is already the resource's static total capacity,
                 * never decremented as jobs consume it (confirmed against
                 * resource.c's unsatisfy_() admission formula, which reads
                 * it as `total` and compares against `inuse` separately).
                 * Print it as-is instead of adding `used` on top. */
                fprintf(stdout, "[sim]   value=%.2f used=%.2f\n",
                      v->u.fvalue, used);
            } else {
                fprintf(stdout, "[sim]   value=%.2f\n", v->u.fvalue);
            }
            break;
        default:
            fprintf(stdout, "[sim]   value=(unrecognized wire type)\n");
            break;
        }
    }
}

/* repl_info_resource - `info resource <name|all>` dispatch. resmgr has no
 * exported sched_get_*() accessor (unlike jobs/queues/workers, core.h) --
 * key_find() directly, the same accessor core_events.c's own
 * QUEUE_UPDATE/RESOURCES_PUSH handling already uses against core->resmgr.
 * bug-1593 (task 12.5): `all` is scoped to cluster-wide (GLOBAL/SHARED)
 * resources only -- a host-based resource never appears in this listing, it
 * remains `info worker`'s job. A named lookup (`info resource <name>`)
 * still resolves any resource, host-based included, so an operator can
 * check its metadata/flags even though no cluster-wide value is shown for
 * it (repl_info_resource_one()'s own host-based branch above). */
static void repl_info_resource(xce_sched_sim_driver_t* sim, const char* arg)
{
    if (strcmp(arg, "all") == 0) {
        iterator_t      it;
        xce_resource_t* r       = NULL;
        int             n_shown = 0;
        int             rc      = iter_make(&it, sim->core.resmgr);

        if (rc < 0) {
            return;
        }
        for (;;) {
            rc = iter_next(&it, (void**)&r);
            if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                break;
            }
            if ((r->flags & (XCE_RESF_GLOBAL | XCE_RESF_SHARED)) == 0) {
                continue;
            }
            repl_info_resource_one(sim, r);
            n_shown++;
        }
        iter_destroy(&it);

        /* Every resource in this session was host-based (e.g. only 'slots'/
         * 'mem'/'cpus'-style consumables, no GLOBAL/SHARED resource at
         * all) -- say so explicitly rather than silently printing nothing,
         * which reads identically to a REPL/dispatch bug from the
         * operator's side. */
        if (n_shown == 0) {
            fprintf(stdout,
                  "[sim] info resource: no global/shared resources defined "
                  "(host-based resources are 'info worker <name>''s job)\n");
        }
        return;
    }

    {
        xce_resource_t* r  = NULL;
        int             rc = key_find(sim->core.resmgr, (void*)arg, (void**)&r);

        if (rc <= 0) {
            fprintf(stdout, "[sim] info resource: no such resource '%s'\n", arg);
            return;
        }
        repl_info_resource_one(sim, r);
    }
}

/*
 * repl_info_summary - AC 22's "info summary": "Quick overview: N_jobs (M
 * pending, K running), N_workers (M online), N_queues, N_resources." Four
 * independent iter_make()/iter_next() walks, same idiom repl.c's own
 * repl_count_jobs() already establishes -- no persistent counters,
 * computed fresh each call.
 */
static void repl_info_summary(xce_sched_sim_driver_t* sim)
{
    iterator_t       it;
    int              rc;
    int              n_jobs = 0, n_pending = 0, n_running = 0;
    int              n_workers = 0, n_online = 0;
    int              n_queues = 0, n_resources = 0;
    xce_sched_job_t* job = NULL;

    rc = iter_make(&it, sim->core.jobs);
    if (rc >= 0) {
        for (;;) {
            rc = iter_next(&it, (void**)&job);
            if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                break;
            }
            n_jobs++;
            if (JOB_IS_PEND(job->state)) {
                n_pending++;
            }
            if (JOB_IS_RUN(job->state)) {
                n_running++;
            }
        }
        iter_destroy(&it);
    }

    {
        xce_sched_worker_t* w = NULL;

        rc = iter_make(&it, sim->core.workers);
        if (rc >= 0) {
            for (;;) {
                rc = iter_next(&it, (void**)&w);
                if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                    break;
                }
                n_workers++;
                if (!IS_WORKER_UNAVAIL(w->status)) {
                    n_online++;
                }
            }
            iter_destroy(&it);
        }
    }

    {
        void* tmp = NULL;

        rc = iter_make(&it, sim->core.queues);
        if (rc >= 0) {
            for (;;) {
                rc = iter_next(&it, &tmp);
                if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                    break;
                }
                n_queues++;
            }
            iter_destroy(&it);
        }

        rc = iter_make(&it, sim->core.resmgr);
        if (rc >= 0) {
            for (;;) {
                rc = iter_next(&it, &tmp);
                if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
                    break;
                }
                n_resources++;
            }
            iter_destroy(&it);
        }
    }

    fprintf(stdout, "[sim] summary: %d jobs (%d pending, %d running), %d workers "
          "(%d online), %d queues, %d resources\n",
          n_jobs, n_pending, n_running, n_workers, n_online, n_queues,
          n_resources);
}

/*
 * repl_info_session - AC 22's "info session": "Show current simulation
 * position: logical time (xce_core_now()), event index (Nth message
 * processed), session count, last SessionSummary if available." The last-
 * session dispatch/pend/preempt tally is this task's own stand-in for "last
 * SessionSummary" -- no in-memory SessionSummary wire struct is decoded on
 * the replay-read path (only written on the record path,
 * recording_session.c), so this reports the same live decision tally
 * repl_break.c's xce_repl_on_decision() already accumulates per session,
 * strictly more current than a stale recorded summary would be for a
 * replay in progress.
 */
static void repl_info_session(xce_sched_sim_driver_t* sim)
{
    xce_sched_repl_state_t* rs = sim->repl_state;
    char                    ts_human[32];

    repl_format_ts_human(xce_core_now(&(sim->core)), ts_human,
                         sizeof(ts_human));

    fprintf(stdout,
          "[sim] session: ts=%llu (%s) event_index=%llu sessions=%llu\n",
          (unsigned long long)xce_core_now(&(sim->core)), ts_human,
          (unsigned long long)((rs != NULL) ? rs->event_index : 0),
          (unsigned long long)((rs != NULL) ? rs->session_count : 0));

    if ((rs != NULL) && (rs->session_count > 0)) {
        fprintf(stdout, "[sim]   last session: round=%llu dispatch=%d pend=%d "
              "preempt=%d\n",
              (unsigned long long)rs->last_begin_round, rs->last_n_dispatch,
              rs->last_n_pend, rs->last_n_preempt);
    } else {
        fprintf(stdout, "[sim]   last session: (none yet)\n");
    }
}

/*
 * repl_info_config - AC 22's "info config": "Show current sim parameters:
 * speed=N, output=<path or \"off\">, verify=on|off, events_processed=N,
 * current_ts=<logical_time>, sessions=N." Read-only counterpart to
 * repl_params.c's `set speed`/`set output`/`set verify` (task 12.3) --
 * reads sim->sim_speed_num (the driver-level shadow copy sim.h's own doc
 * comment says exists precisely "for the REPL/diagnostics to read without
 * reaching into core", kept in sync with the real core.speed_num/speed_den
 * atomics by every `set speed` call, repl_params.c's repl_set_speed()),
 * sim->record_file/rs->output_path (repl_params.c's own output_path field
 * doc comment explains the "path unknown" fallback for a --sim-output
 * stream this REPL session never itself opened), sim->verify, and the SAME
 * rs->event_index/rs->session_count `info session` already reports
 * (repl_info_session() above).
 */
static void repl_info_config(xce_sched_sim_driver_t* sim)
{
    xce_sched_repl_state_t* rs = sim->repl_state;
    const char*             output_desc;
    char                    ts_human[32];

    if (sim->record_file == NULL) {
        output_desc = "off";
    } else if ((rs != NULL) && (rs->output_path != NULL)) {
        output_desc = rs->output_path;
    } else {
        output_desc = "(active, path unknown)";
    }

    repl_format_ts_human(xce_core_now(&(sim->core)), ts_human,
                         sizeof(ts_human));

    fprintf(stdout, "[sim] config: speed=%llu output=%s verify=%s "
          "events_processed=%llu current_ts=%llu (%s) sessions=%llu\n",
          (unsigned long long)sim->sim_speed_num, output_desc,
          sim->verify ? "on" : "off",
          (unsigned long long)((rs != NULL) ? rs->event_index : 0),
          (unsigned long long)xce_core_now(&(sim->core)), ts_human,
          (unsigned long long)((rs != NULL) ? rs->session_count : 0));
}

/* repl_info_break - AC 22's "info break": "List all breakpoints with their
 * IDs, event types, and filters." */
static void repl_info_break(xce_sched_sim_driver_t* sim)
{
    xce_sched_repl_state_t* rs = sim->repl_state;
    int                     i, j;

    if ((rs == NULL) || (rs->nbreakpoints == 0)) {
        fprintf(stdout, "[sim] no breakpoints set\n");
        return;
    }

    fprintf(stdout, "[sim] breakpoints:\n");
    for (i = 0; i < rs->nbreakpoints; i++) {
        xce_repl_breakpoint_t* bp = rs->breakpoints[i];

        fprintf(stdout, "[sim]   %d: %s", bp->id, bp->token);
        for (j = 0; j < bp->nfilters; j++) {
            fprintf(stdout, " %s=%s", bp->filters[j].key, bp->filters[j].value);
        }
        fprintf(stdout, "\n");
    }
}

/*
 * repl_cmd_info - Req 5 AC 22's "info ..." family: dispatches on the second
 * token to repl_info_break()/repl_info_session()/repl_info_config()/
 * repl_info_job()/repl_info_queue()/repl_info_worker()/repl_info_resource()/
 * repl_info_summary() (all above). No `info` subcommand ever hands control
 * back to the driver loop -- inspecting state is never a reason to advance
 * the simulation. repl.c's repl_dispatch() calls this (repl_internal.h's
 * own extern declaration).
 */
bool repl_cmd_info(xce_sched_sim_driver_t* sim, const char* line)
{
    char tokens[XCE_REPL_MAX_TOKENS][XCE_REPL_TOKEN_LEN];
    int  ntok = repl_split_tokens(line, tokens, XCE_REPL_MAX_TOKENS);

    if (ntok < 2) {
        fprintf(stdout, "[sim] info: usage: info <break|session|config|job|jobs|queue|"
              "worker|resource|summary> [arg]\n");
        return false;
    }

    if (strcmp(tokens[1], "break") == 0) {
        repl_info_break(sim);
    } else if (strcmp(tokens[1], "session") == 0) {
        repl_info_session(sim);
    } else if (strcmp(tokens[1], "config") == 0) {
        repl_info_config(sim);
    } else if (strcmp(tokens[1], "summary") == 0) {
        repl_info_summary(sim);
    } else if (strcmp(tokens[1], "job") == 0) {
        if (ntok < 3) {
            fprintf(stdout, "[sim] info job: usage: info job <jobid>\n");
        } else {
            repl_info_job(sim, tokens[2]);
        }
    } else if (strcmp(tokens[1], "jobs") == 0) {
        repl_info_jobs(sim);
    } else if (strcmp(tokens[1], "queue") == 0) {
        if (ntok < 3) {
            fprintf(stdout, "[sim] info queue: usage: info queue <name>|all\n");
        } else {
            repl_info_queue(sim, tokens[2]);
        }
    } else if (strcmp(tokens[1], "worker") == 0) {
        if (ntok < 3) {
            fprintf(stdout, "[sim] info worker: usage: info worker <name>|all\n");
        } else {
            repl_info_worker(sim, tokens[2]);
        }
    } else if (strcmp(tokens[1], "resource") == 0) {
        if (ntok < 3) {
            fprintf(stdout, "[sim] info resource: usage: info resource <name>|all\n");
        } else {
            repl_info_resource(sim, tokens[2]);
        }
    } else {
        fprintf(stdout, "[sim] info: unrecognized subcommand '%s'\n", tokens[1]);
    }

    return false;
}