myxce/src/scheduler/testsuite/test_snapshot_save_load.c

/*
 * test_snapshot_save_load.c - the cross-core behavioral-equivalence proof
 * for xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot()
 * (scheduler-core spec, task 8.1, Req 5 AC 21 / Req 6 AC 5).
 *
 * Observable under test (task 8.1's own text): "save immediately followed
 * by load into a freshly-initialized core produces a core whose observable
 * scheduling behavior (next session's decisions, given the same events) is
 * identical to the pre-save core." This is a genuinely CROSS-CORE proof,
 * not a field-level round-trip check alone: two fully independent
 * xce_sched_core_t instances (core_a, the original; core_b, freshly
 * xce_sched_core_init()'d and populated purely via
 * xce_sched_core_load_snapshot()) are each driven through the SAME
 * push-events -> go -> drain-decisions -> end-session cycle
 * test_core_full_cycle.c (task 4.6) already established as the correct,
 * real way to drive a session -- a real picker pthread plus a stand-in
 * allocator (this file's own run_identical_session(), the test function's
 * own thread) reproducing sched_job_picker()/sched_allocator()'s real
 * per-session bodies, NOT a hand-rolled shortcut. The two cores' resulting
 * decision sequences (op + jobid, in order) are asserted byte-for-byte
 * identical. Field-level assertions (queues/resources/resvals/workers/
 * users/projects/prng_seed/jobs restored onto core_b) are ALSO included --
 * cheaper to debug from, per this task's own text, but not a substitute
 * for the cross-core proof above.
 *
 * White-box #include of core.c, core_clock.c, core_events.c, core_session.c,
 * decision.c (same convention/file set as test_core_full_cycle.c) PLUS the
 * three files under test: snapshot.c, snapshot_wire.c, snapshot_load.c
 * (task 8.1's own three-file split -- see snapshot.c's file header
 * comment). Stubs below are the exact same set test_core_full_cycle.c
 * already establishes needing for this file combination (kill_sched_*
 * family, plugin_unregister, sched_worker_release_host_ctx,
 * sched_resolve_job_picker)
 * -- UNLIKE that file, sched_worker_release_host_ctx is genuinely invoked
 * here (core_reset_workers(), core_session.c) since this test adds real
 * workers to both cores; it remains a safe no-op regardless.
 * makeResource()/parse_resreq()/format_resreq()/key_find()/zarray_*() are
 * all REAL, already-linked libxce/librtti symbols (src/lib/resobj.c,
 * src/lib/resreq.c) -- no stub needed for any of them.
 *
 * Two independent stub picker plugin instances (g_stub_plugin_a/
 * g_stub_plugin_b), each with its OWN private stub_picker_state_t reached
 * via xce_sched_plugin_t's existing `data` field (the same mechanism
 * builtin_fcfs.c's real fcfs_plugin_data_t uses) -- this is what lets
 * core_a and core_b each build and walk their OWN independent candidate
 * list without cross-contaminating a shared global, while still reusing
 * the exact same core-parameterized picker-thread function
 * (test_picker_thread_fn(), copied from test_core_full_cycle.c verbatim)
 * for both: sched_resolve_job_picker(core, ...) naturally resolves to
 * whichever plugin instance is registered in THAT core's own
 * core->plugins kvmgr.
 */
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "kvmgr.h"
#include "preason.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real xce_sched_core_init()/xce_core_push_events()/
 * xce_core_go()/xce_core_end_session()/xce_core_set_decision_cb()/
 * xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() under test
 * into this TU (white-box #include, same convention as
 * test_core_full_cycle.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/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; }
void plugin_unregister(void* arg) { (void)arg; }

/* core_reset_workers()'s (core_session.c) linker dependency. UNLIKE
 * test_core_full_cycle.c's own identical stub, this one IS actually
 * invoked here (both core_a and core_b get real workers below) -- still a
 * safe no-op, see file header comment. */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
 * file header comment (same rationale/algorithm as
 * test_core_full_cycle.c's own copy). Parameterized purely on `core`, so it
 * naturally resolves core_a's own registered plugin for core_a and
 * core_b's own registered plugin for core_b. */
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;
}

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

/* ---------------------------------------------------------------------- */
/* Stub picker: per-instance private state via xce_sched_plugin_t->data    */
/* (see file header comment) -- each of g_stub_state_a/g_stub_state_b      */
/* independently accumulates ITS OWN core's candidate list.                */
/* ---------------------------------------------------------------------- */

#define STUB_MAX_CANDIDATES 8

typedef struct {
    xce_sched_job_t* candidates[STUB_MAX_CANDIDATES];
    int              ncandidates;
    int              cursor;
} stub_picker_state_t;

static stub_picker_state_t g_stub_state_a;
static stub_picker_state_t g_stub_state_b;

static int stub_on_job_push(void* picker, xce_sched_job_t* job)
{
    stub_picker_state_t* st
        = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;

    if (st->ncandidates < STUB_MAX_CANDIDATES) {
        st->candidates[st->ncandidates++] = job;
    }

    return 0;
}

static int stub_on_queue_update(void* picker, xce_sched_queue_t* queue)
{
    (void)picker;
    (void)queue;

    return 0;
}

static int stub_on_job_state_change(void* picker, xce_sched_job_t* job,
                                    uint32_t state)
{
    (void)picker;
    (void)job;
    (void)state;

    return 0; /* not exercised by this test's event batch */
}

static int stub_on_job_update(void* picker, xce_sched_job_t* job)
{
    (void)picker;
    (void)job;

    return 0; /* not exercised by this test's event batch */
}

/* begin_session()/next_job() run in the real picker thread ONLY -- same
 * "no lock needed against on_dispatch()/on_pend()" reasoning
 * test_core_full_cycle.c's own header comment gives (this stub's
 * on_dispatch/on_pend below are pure no-ops, touching nothing). */
static int stub_begin_session(void* picker)
{
    stub_picker_state_t* st
        = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;

    st->cursor = 0; /* session-aware requirement (plugin.h): never carry the
                     * iteration cursor over from a previous session. */

    return 0;
}

static xce_sched_job_t* stub_next_job(void* picker)
{
    stub_picker_state_t* st
        = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;

    if (st->cursor >= st->ncandidates) {
        return NULL;
    }

    return st->candidates[st->cursor++];
}

static int stub_end_session(void* picker)
{
    (void)picker;

    return 0;
}

/* Post-allocation hooks (plugin.h: allocator thread, in production; the
 * main thread's stand-in allocator here, run_identical_session() below) --
 * pure no-ops: this test's decision-sequence comparison is driven entirely
 * off driver_ctx->seen[] (the real xce_decision_t stream), not off these
 * hooks. */
static int stub_on_dispatch(void* picker, uint64_t jobid)
{
    (void)picker;
    (void)jobid;

    return 0;
}

static int stub_on_pend(void* picker, uint64_t jobid)
{
    (void)picker;
    (void)jobid;

    return 0;
}

static xce_job_picker_iface_t g_stub_picker_iface = {
    .init                = NULL,
    .fini                = NULL,
    .begin_session       = stub_begin_session,
    .next_job            = stub_next_job,
    .end_session         = stub_end_session,
    .on_dispatch         = stub_on_dispatch,
    .on_pend             = stub_on_pend,
    .on_queue_update     = stub_on_queue_update,
    .on_job_push         = stub_on_job_push,
    .on_job_state_change = stub_on_job_state_change,
    .on_job_update       = stub_on_job_update,
    .on_config           = NULL, /* MAY be NULL (plugin.h) -- not pushed */
    .on_job_rusage       = NULL, /* MAY be NULL (plugin.h) -- not pushed */
};

/* Same picker-plugin NAME for both -- safe: each lives in its own core's
 * OWN core->plugins kvmgr (see file header comment), so there is no name
 * collision. */
static xce_sched_plugin_t g_stub_plugin_a
    = { .name         = "test_stub_picker",
        .picker_iface = &g_stub_picker_iface,
        .data         = &g_stub_state_a };
static xce_sched_plugin_t g_stub_plugin_b
    = { .name         = "test_stub_picker",
        .picker_iface = &g_stub_picker_iface,
        .data         = &g_stub_state_b };

/* ---------------------------------------------------------------------- */
/* Picker thread: reproduces scheduler.c's sched_job_picker() real         */
/* per-session body verbatim (copied from test_core_full_cycle.c's own     */
/* test_picker_thread_fn -- identical, core-parameterized, single-shot).   */
/* ---------------------------------------------------------------------- */

static void* test_picker_thread_fn(void* arg)
{
    xce_sched_core_t*   core       = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* job_picker = NULL;
    int                 rc;
    int                 go = 0;
    uint64_t            jobid;

    rc = sched_resolve_job_picker(core, &job_picker);
    if (rc < 0) {
        return NULL;
    }

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

    job_picker->picker_iface->begin_session(job_picker);

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

        if (job == NULL) {
            break;
        }

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

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

    job_picker->picker_iface->end_session(job_picker);

    return NULL;
}

/* ---------------------------------------------------------------------- */
/* Fixture helpers                                                        */
/* ---------------------------------------------------------------------- */

static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
    int rc;

    rc = xce_sched_core_init(core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    lock(core->plugins, RTTI_LOCK_WRITE);
    rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin);
    unlock(core->plugins);
    TEST_ASSERT_GREATER_OR_EQUAL(0, rc);

    core->job_pick_plugin = plugin->name;
}

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    = 1000;
    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;
}

#define MAX_RECORDED_DECISIONS 8

typedef struct {
    int             calls;
    xce_decision_t* seen[MAX_RECORDED_DECISIONS];
} test_driver_ctx_t;

static int test_on_decision(void* driver_ctx, xce_decision_t* decision)
{
    test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx;

    TEST_ASSERT_NOT_NULL(decision);

    if (ctx->calls < MAX_RECORDED_DECISIONS) {
        ctx->seen[ctx->calls] = decision;
    }
    ctx->calls++;

    return 0;
}

static int drain_and_dispatch(xce_sched_core_t* core)
{
    int delivered = 0;

    for (;;) {
        xce_decision_t* decision = NULL;

        cbuf_get(&(core->decisions), &decision);
        if (decision == NULL) {
            break;
        }

        core->on_decision(core->driver_ctx, decision);
        delivered++;
    }

    return delivered;
}

/* run_identical_session - drives ONE core through a full session: a real
 * picker thread + a stand-in allocator (this function itself, mirroring
 * test_core_full_cycle.c's own Phase 3/4/5) that decides DISPATCH (even
 * jobid) vs PEND (odd jobid) -- a deterministic, resource-matching-free
 * rule, same scope boundary test_core_full_cycle.c's own header comment
 * documents for why real sched_allocator()/resource.c internals are out of
 * a pure core-contract test's reach. This is what makes "identical events
 * in, identical decisions out" a meaningful, reproducible comparison
 * between core_a and core_b: given the SAME candidate jobids in the SAME
 * order (which the picker rebuild on load is what this whole task proves),
 * this stand-in allocator's own logic is a pure function of jobid, so it
 * necessarily produces the SAME op+jobid sequence for both cores. */
static void run_identical_session(xce_sched_core_t* core)
{
    pthread_t picker_tid;
    int       rc;
    uint64_t  jobid;

    rc = pthread_create(&picker_tid, NULL, test_picker_thread_fn, core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    rc = xce_core_go(core);
    TEST_ASSERT_EQUAL_INT(0, rc);

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

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

        job = sched_get_job(core, jobid);
        TEST_ASSERT_NOT_NULL(job);

        if ((jobid % 2) == 0) {
            xce_job_alloc_t* alloc
                = (xce_job_alloc_t*)calloc(1, sizeof(*alloc));
            alloc->jobid = jobid;
            decision     = make_dispatch_decision(alloc);
        } else {
            xce_pendreason_t* reason = make_pendreason(jobid);
            decision                 = make_pend_decision(reason);
        }

        TEST_ASSERT_NOT_NULL(decision);
        cbuf_put(&(core->decisions), &decision);
    }

    rc = pthread_join(picker_tid, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);

    (void)drain_and_dispatch(core);

    rc = xce_core_end_session(core);
    TEST_ASSERT_EQUAL_INT(0, rc);
}

/* ---------------------------------------------------------------------- */
/* Guard tests (Part 3's own "-EBUSY/-EINVAL, directly and cheaply, unit-  */
/* level" request) -- no session, no picker plugin needed: both guards     */
/* fire before either function touches state or resolves a picker (core.h */
/* 's own doc comment).                                                    */
/* ---------------------------------------------------------------------- */

static void
test_snapshot_guards_reject_null_args_and_busy_session_without_mutating_state(
    void)
{
    xce_sched_core_t      core;
    sched_snapshot_data_t snap;
    sched_snapshot_data_t empty_in;
    int                   rc;

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

    memset(&empty_in, 0, sizeof(empty_in));

    /* -EINVAL: NULL core, or NULL out/in, for both directions. */
    rc = xce_sched_core_save_snapshot(NULL, &snap);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);

    rc = xce_sched_core_save_snapshot(&core, NULL);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);

    rc = xce_sched_core_load_snapshot(NULL, &empty_in);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);

    rc = xce_sched_core_load_snapshot(&core, NULL);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);

    /* -EBUSY: session_active != 0, checked BEFORE any mutation -- a
     * rejected call is a true no-op (core.h's own doc comment for both
     * functions, "ZERO state touched"). Directly flip the atomic (same
     * precondition-rejection technique test_core_full_cycle.c's own Phase 2
     * uses for xce_core_go()/xce_core_push_events()) rather than driving a
     * real session, since only the guard itself is under test here. */
    atomic_store(&(core.session_active), 1);

    memset(&snap, 0xAA, sizeof(snap));
    rc = xce_sched_core_save_snapshot(&core, &snap);
    TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
    /* sentinel bytes untouched -- save_snapshot() never reached its own
     * memset(out, 0, ...), proving the guard runs before any mutation. */
    TEST_ASSERT_EQUAL_UINT8(0xAA, ((uint8_t*)&snap)[0]);

    rc = xce_sched_core_load_snapshot(&core, &empty_in);
    TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
    /* no queue was created -- proves ZERO state was touched, not just that
     * the return code was right. */
    TEST_ASSERT_NULL(sched_get_queue(&core, "normal"));

    atomic_store(&(core.session_active), 0);
}

/* ---------------------------------------------------------------------- */
/* The Observable, end to end.                                             */
/* ---------------------------------------------------------------------- */

/*
 * test_snapshot_save_then_load_produces_behaviorally_equivalent_core - see
 * file header comment for the full design.
 */
static void
test_snapshot_save_then_load_produces_behaviorally_equivalent_core(void)
{
    xce_sched_core_t      core_a;
    xce_sched_core_t      core_b;
    sched_snapshot_data_t snap;
    xce_core_event_t      events[3];
    xce_conf4sched_t      config;
    xce_resource_t        wire_resource;
    xce_resource_t*       resources[1];
    xce_val_t             resval_value;
    xce_name_val_t        wire_resval;
    xce_name_val_t*       resvals[1];
    xce_queue_t           wire_queue_normal;
    xce_queue_t           wire_queue_batch;
    xce_queue_t*          queues[2];
    xce_worker_t          wire_worker_a;
    xce_worker_t          wire_worker_b;
    xce_worker_t*         workers[2];
    xce_val_t             worker_a_cpu_val;
    xce_val_t             worker_b_cpu_val;
    xce_name_val_t        worker_a_resval;
    xce_name_val_t        worker_b_resval;
    xce_name_val_t*       worker_a_resvals[1];
    xce_name_val_t*       worker_b_resvals[1];
    xce_job_t*            jobs[3];
    xce_sched_project_t*  proj1_a;
    xce_resource_t*       found_resource_a;
    xce_resource_t*       found_resource_b;
    xce_val_t*            shared_val_b;
    xce_sched_queue_t*    q_normal_b;
    xce_sched_queue_t*    q_batch_b;
    xce_sched_worker_t*   worker_a_b;
    xce_sched_worker_t*   worker_b_b;
    xce_val_t*            worker_a_resval_b;
    xce_val_t*            worker_b_resval_b;
    xce_sched_project_t*  proj1_b;
    xce_sched_project_t*  proj2_b;
    xce_sched_job_t*      job201_b;
    xce_sched_job_t*      job202_b;
    xce_sched_job_t*      job203_b;
    test_driver_ctx_t     driver_ctx_a;
    test_driver_ctx_t     driver_ctx_b;
    int                   rc;
    int                   i;

    memset(&driver_ctx_a, 0, sizeof(driver_ctx_a));
    memset(&driver_ctx_b, 0, sizeof(driver_ctx_b));
    memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
    memset(&g_stub_state_b, 0, sizeof(g_stub_state_b));

    /* ===================================================================
     * Phase 1: build core_a with REAL state -- jobs, queues, workers,
     * resources, resvals, a registered picker plugin.
     * =================================================================== */
    fixture_init(&core_a, &g_stub_plugin_a);
    core_a.prng_seed = 0xDEADBEEFULL; /* direct field poke -- prng_seed has
                                       * no wire event of its own (task
                                       * 8.1's own scope note, core.h);
                                       * seeded here to a non-zero value so
                                       * the round-trip assertion below is
                                       * meaningful, not a trivial 0==0. */

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name = "cpu";
    resources[0]       = &wire_resource;

    memset(&resval_value, 0, sizeof(resval_value));
    /* .which MUST be XCE_VAL_NUMERIC, not left at the memset-zero default
     * -- XCE_VAL_STRING is literally 0 (xce/types.h), and dup_val()
     * (snapshot_wire.c, genuinely exercised by save_snapshot() below)
     * branches on .which to decide whether .u.svalue is a string pointer
     * to copy: leaving this at 0 makes it misinterpret .u.ivalue32's raw
     * integer bits as a char* and crash inside makeString(). */
    resval_value.which      = XCE_VAL_NUMERIC;
    resval_value.u.ivalue32 = 8;
    memset(&wire_resval, 0, sizeof(wire_resval));
    wire_resval.name = "cpu";
    wire_resval.val  = &resval_value;
    resvals[0]       = &wire_resval;

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

    memset(&wire_queue_batch, 0, sizeof(wire_queue_batch));
    wire_queue_batch.queue    = "batch";
    wire_queue_batch.priority = 1;
    wire_queue_batch.status   = QUEUE_STAT_OPEN;

    queues[0] = &wire_queue_normal;
    queues[1] = &wire_queue_batch;

    memset(&config, 0, sizeof(config));
    config.n_resources = 1;
    config.resources   = resources;
    config.nresvals    = 1;
    config.resvals     = resvals;
    config.n_queues    = 2;
    config.queues      = queues;

    memset(&worker_a_cpu_val, 0, sizeof(worker_a_cpu_val));
    worker_a_cpu_val.which = XCE_VAL_NUMERIC; /* see resval_value's own
                                               * comment above */
    worker_a_cpu_val.u.ivalue32 = 4;
    memset(&worker_a_resval, 0, sizeof(worker_a_resval));
    worker_a_resval.name = "cpu";
    worker_a_resval.val  = &worker_a_cpu_val;
    worker_a_resvals[0]  = &worker_a_resval;

    memset(&worker_b_cpu_val, 0, sizeof(worker_b_cpu_val));
    worker_b_cpu_val.which      = XCE_VAL_NUMERIC;
    worker_b_cpu_val.u.ivalue32 = 6;
    memset(&worker_b_resval, 0, sizeof(worker_b_resval));
    worker_b_resval.name = "cpu";
    worker_b_resval.val  = &worker_b_cpu_val;
    worker_b_resvals[0]  = &worker_b_resval;

    memset(&wire_worker_a, 0, sizeof(wire_worker_a));
    wire_worker_a.name   = "hostA";
    wire_worker_a.status = 0; /* available: IS_WORKER_UNAVAIL
                               * checks a specific unset bit */
    wire_worker_a.nresvals_to_sched = 1;
    wire_worker_a.resvals_to_sched  = worker_a_resvals;

    memset(&wire_worker_b, 0, sizeof(wire_worker_b));
    wire_worker_b.name              = "hostB";
    wire_worker_b.status            = 0;
    wire_worker_b.nresvals_to_sched = 1;
    wire_worker_b.resvals_to_sched  = worker_b_resvals;

    workers[0] = &wire_worker_a;
    workers[1] = &wire_worker_b;

    jobs[0] = make_job(201, JOB_STAT_PEND, "normal", "alice", "proj1");
    jobs[1] = make_job(202, JOB_STAT_PEND, "normal", "bob", "proj1");
    jobs[2] = make_job(203, JOB_STAT_PEND, "batch", "carol", "proj2");

    memset(events, 0, sizeof(events));
    /* CONFIG_PUSH first: registers "cpu" into core_a->resmgr, which
     * WORKERS_PUSH (next) needs already present to resolve each worker's
     * resvals_to_sched by name. */
    events[0].type                    = XCE_CORE_EVENT_CONFIG_PUSH;
    events[0].u.config_push.config    = &config;
    events[1].type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    events[1].u.workers_push.nworkers = 2;
    events[1].u.workers_push.workers  = workers;
    events[2].type                    = XCE_CORE_EVENT_JOBS_PUSH;
    events[2].u.jobs_push.njobs       = 3;
    events[2].u.jobs_push.jobs        = jobs;

    rc = xce_core_push_events(&core_a, events, 3);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(3, g_stub_state_a.ncandidates);

    /* project priority: no wire event of its own -- direct field poke on
     * the auto-created project object (get-or-create side effect of
     * JOBS_PUSH), same "genuine, but directly-constructed" state
     * test_core_events.c's own fixtures already use for config/queue/
     * resource wire payloads. proj2 deliberately left at its default (0)
     * -- proves the round-trip is field-accurate, not just "non-zero
     * survives". */
    proj1_a = sched_get_project(&core_a, "proj1");
    TEST_ASSERT_NOT_NULL(proj1_a);
    proj1_a->priority = 5;

    /* ===================================================================
     * Phase 2: save core_a's snapshot; field-level sanity checks on the
     * snapshot struct itself (cheap, direct evidence of the field-by-field
     * mapping snapshot.c's own header comment documents).
     * =================================================================== */
    memset(&snap, 0, sizeof(snap));
    rc = xce_sched_core_save_snapshot(&core_a, &snap);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_NOT_NULL(snap.jobs);
    TEST_ASSERT_EQUAL_INT(3, snap.jobs->count);
    TEST_ASSERT_EQUAL_INT(2, snap.n_queues);
    TEST_ASSERT_NOT_NULL(snap.workers);
    TEST_ASSERT_EQUAL_INT(2, snap.workers->count);
    TEST_ASSERT_NOT_NULL(snap.resources);
    TEST_ASSERT_EQUAL_INT(1, snap.resources->count);
    TEST_ASSERT_NOT_NULL(snap.resvals);
    TEST_ASSERT_EQUAL_INT(1, snap.resvals->count);
    TEST_ASSERT_EQUAL_INT(3, snap.n_users);
    TEST_ASSERT_EQUAL_INT(2, snap.n_projects);
    TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, snap.prng_seed);

    /* ===================================================================
     * Phase 3: build core_b -- a completely separate, freshly-initialized
     * instance -- and load the snapshot into it.
     * =================================================================== */
    fixture_init(&core_b, &g_stub_plugin_b);

    rc = xce_sched_core_load_snapshot(&core_b, &snap);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* picker rebuild fired on_job_push for every restored job, exactly as
     * the original JOBS_PUSH did for core_a above. */
    TEST_ASSERT_EQUAL_INT(3, g_stub_state_b.ncandidates);

    /* ===================================================================
     * Phase 4: field-level assertions -- core_b's restored state matches
     * core_a's original values exactly.
     * =================================================================== */
    q_normal_b = sched_get_queue(&core_b, "normal");
    q_batch_b  = sched_get_queue(&core_b, "batch");
    TEST_ASSERT_NOT_NULL(q_normal_b);
    TEST_ASSERT_NOT_NULL(q_batch_b);
    TEST_ASSERT_EQUAL_INT(3, q_normal_b->priority);
    TEST_ASSERT_EQUAL_INT(1, q_batch_b->priority);

    rc = key_find(core_a.resmgr, (void*)"cpu", (void**)&found_resource_a);
    TEST_ASSERT_GREATER_THAN_INT(0, rc);
    rc = key_find(core_b.resmgr, (void*)"cpu", (void**)&found_resource_b);
    TEST_ASSERT_GREATER_THAN_INT(0, rc);
    TEST_ASSERT_NOT_NULL(found_resource_a);
    TEST_ASSERT_NOT_NULL(found_resource_b);

    shared_val_b
        = (xce_val_t*)zarray_get(&(core_b.sharedInst), found_resource_b->idx);
    TEST_ASSERT_NOT_NULL(shared_val_b);
    TEST_ASSERT_EQUAL_INT(8, shared_val_b->u.ivalue32);

    worker_a_b = sched_get_worker(&core_b, "hostA");
    worker_b_b = sched_get_worker(&core_b, "hostB");
    TEST_ASSERT_NOT_NULL(worker_a_b);
    TEST_ASSERT_NOT_NULL(worker_b_b);
    worker_a_resval_b
        = (xce_val_t*)zarray_get(&(worker_a_b->resvals), found_resource_b->idx);
    worker_b_resval_b
        = (xce_val_t*)zarray_get(&(worker_b_b->resvals), found_resource_b->idx);
    TEST_ASSERT_NOT_NULL(worker_a_resval_b);
    TEST_ASSERT_NOT_NULL(worker_b_resval_b);
    TEST_ASSERT_EQUAL_INT(4, worker_a_resval_b->u.ivalue32);
    TEST_ASSERT_EQUAL_INT(6, worker_b_resval_b->u.ivalue32);

    TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "alice"));
    TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "bob"));
    TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "carol"));

    proj1_b = sched_get_project(&core_b, "proj1");
    proj2_b = sched_get_project(&core_b, "proj2");
    TEST_ASSERT_NOT_NULL(proj1_b);
    TEST_ASSERT_NOT_NULL(proj2_b);
    TEST_ASSERT_EQUAL_INT(5, proj1_b->priority);
    TEST_ASSERT_EQUAL_INT(0, proj2_b->priority);

    TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, core_b.prng_seed);

    job201_b = sched_get_job(&core_b, 201);
    job202_b = sched_get_job(&core_b, 202);
    job203_b = sched_get_job(&core_b, 203);
    TEST_ASSERT_NOT_NULL(job201_b);
    TEST_ASSERT_NOT_NULL(job202_b);
    TEST_ASSERT_NOT_NULL(job203_b);
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, job201_b->state);
    TEST_ASSERT_EQUAL_STRING("normal", job201_b->queue->name);
    TEST_ASSERT_EQUAL_STRING("alice", job201_b->user->name);
    TEST_ASSERT_EQUAL_STRING("proj1", job201_b->project->name);
    TEST_ASSERT_EQUAL_STRING("batch", job203_b->queue->name);
    TEST_ASSERT_EQUAL_STRING("carol", job203_b->user->name);
    TEST_ASSERT_EQUAL_STRING("proj2", job203_b->project->name);

    /* ===================================================================
     * Phase 5: THE cross-core behavioral-equivalence proof. Drive core_a
     * and core_b through an IDENTICAL session (same picker-thread protocol,
     * same deterministic stand-in-allocator rule) and compare the
     * resulting decision sequences.
     * =================================================================== */
    rc = xce_core_set_decision_cb(&core_a, test_on_decision, &driver_ctx_a);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = xce_core_set_decision_cb(&core_b, test_on_decision, &driver_ctx_b);
    TEST_ASSERT_EQUAL_INT(0, rc);

    run_identical_session(&core_a);
    run_identical_session(&core_b);

    TEST_ASSERT_EQUAL_INT(3, driver_ctx_a.calls);
    TEST_ASSERT_EQUAL_INT(driver_ctx_a.calls, driver_ctx_b.calls);

    for (i = 0; i < driver_ctx_a.calls; i++) {
        xce_decision_t* da = driver_ctx_a.seen[i];
        xce_decision_t* db = driver_ctx_b.seen[i];

        TEST_ASSERT_NOT_NULL(da);
        TEST_ASSERT_NOT_NULL(db);
        TEST_ASSERT_EQUAL_INT(da->op, db->op);

        if (da->op == XCE_DECISION_DISPATCH) {
            TEST_ASSERT_EQUAL_UINT64(da->u.alloc->jobid, db->u.alloc->jobid);
        } else {
            TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, da->op);
            TEST_ASSERT_EQUAL_UINT64(da->u.reason->jobid, db->u.reason->jobid);
        }
    }

    /* Spot-check the actual expected sequence (jobid parity -- see
     * run_identical_session()'s own comment): 201 odd -> PEND, 202 even ->
     * DISPATCH, 203 odd -> PEND, in push/candidate order. Proven identical
     * for BOTH cores by the loop above; this just pins the concrete values
     * so a regression here fails loudly with a readable expected/actual
     * pair instead of only an A==B equality. */
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[0]->op);
    TEST_ASSERT_EQUAL_UINT64(201, driver_ctx_a.seen[0]->u.reason->jobid);
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx_a.seen[1]->op);
    TEST_ASSERT_EQUAL_UINT64(202, driver_ctx_a.seen[1]->u.alloc->jobid);
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[2]->op);
    TEST_ASSERT_EQUAL_UINT64(203, driver_ctx_a.seen[2]->u.reason->jobid);
}

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