myxce/src/scheduler/testsuite/test_sim_verify.c

/*
 * test_sim_verify.c - contract test for decision verification
 * (scheduler-core spec, task 11.4, Req 5 AC 15/18).
 *
 * Observable under test (task 11.4's own text):
 *   - replaying a fixture file with verification enabled against its own
 *     previously-recorded output reports zero mismatches.
 *   - replaying against a deliberately-altered copy reports the exact
 *     mismatch.
 *
 * Two granularities, same split test_sim_event_loop.c's own header comment
 * already establishes for task 11.2:
 *
 *   Unit-level (white-box): calls sim_decision_eq()/sim_log_mismatch()
 *   (sim_verify.c, task 11.4's own sibling file to sim_loop.c) directly
 *   against hand-built replay_decision_t/xce_decision_t pairs -- DISPATCH
 *   match/mismatch (jobid, host, resource value), PEND match/mismatch
 *   (jobid, main_reason), a body_which/op type mismatch, and PREEMPT's
 *   documented always-false path. This is the only way to exercise DISPATCH
 *   comparison at all: sim_loop.c's real picker thread
 *   (sim_picker_thread_fn) only ever emits PEND decisions (its own doc
 *   comment: "conservatively PENDs every candidate ... rather than
 *   fabricate a DISPATCH decision with no real worker/resource assignment
 *   behind it") -- real resource-matching allocation belongs to the
 *   not-yet-designed scheduler-allocation spec. sim_decision_eq() must
 *   still compare DISPATCH decisions correctly (Req 5 AC 15 does not
 *   exempt them), so that comparator is proven directly here rather than
 *   left unreachable.
 *
 *   Integration-level: drives the full sched_sim_driver_run() (this task's
 *   other Observable) against a fixture built from a REAL
 *   QUEUE_UPDATE+JOBS_PUSH event pair (one queue, one JOB_STAT_PEND job)
 *   feeding the REAL builtin_fcfs picker -- since core.c's own
 *   sched_job_is_ready()/sched_queue_is_ready() are unconditional `return
 *   1` stubs (task 2.4's own documented, still-live scope note), any queue/
 *   job pair is immediately "ready", so this always produces exactly one
 *   live PEND decision via make_pendreason() (preason.c), whose
 *   xce_job_preason_t is always freshly calloc()'d (main_reason=0, no host/
 *   res reasons) -- a fully deterministic, reproducible target to compare
 *   against. `--sim-verify`'s ERROR()-logged mismatch report is observed by
 *   redirecting STDOUT_FILENO around the sched_sim_driver_run() call, the
 *   same dup()/dup2() capture technique
 *   test_apply_correctness_verification.c's own
 *   test_apply_divergent_loud_fence_log_omits_false_peer_hwm_clause
 *   establishes (that file's own doc comment explains WHY this works: with
 *   tlog never tlog_init()'d in this test process -- true here too, nothing
 *   in this TU calls it -- every ERROR()/WARN() falls through tlog's
 *   early-print path, a raw write() to STDOUT_FILENO).
 *
 * Fixture strategy: same production-code-reuse technique test_sim_startup.c/
 * test_sim_event_loop.c already establish for the header+snapshot prefix
 * (sched_record_start() against a small real xce_scheduler_t fixture) PLUS
 * hand-constructed ReplayEvent/ReplaySession messages appended afterward via
 * the real encode_ReplayEvent_ptr()/encode_ReplaySession_ptr()/
 * write_capnp_message() codec, same technique test_sim_event_loop.c's own
 * append_jobstatechange_event()/append_replay_session() and
 * test_e2e_capnp.c's test_dispatch_decision_round_trip()/
 * test_pend_decision_round_trip() already prove works.
 *
 * White-box #include of the same TU set test_sim_event_loop.c uses (see
 * that file's own header comment, including task 11.5's
 * recording_session.c/sim_record.c AND task 12.1's repl.c -- all required at
 * link time by sched_sim_driver_run()'s/repl.h's unconditional externs
 * regardless of whether --sim-output/--interactive is ever set in a given
 * test) PLUS ../src/sim_verify.c (the code under test).
 */
#include <errno.h>
#include <limits.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <unity/unity.h>

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

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention as test_sim_event_loop.c/test_sim_startup.c). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/repl_params.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_picker.c" /* bug-1567: sched_sim_driver_run() now calls sim_bootstrap_picker() (sim_loop.c-cap sibling) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"

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

/* core_reset_workers()'s (core_session.c) linker dependency -- same stub
 * test_sim_event_loop.c/test_core_full_cycle.c already establish; never
 * invoked (no worker is ever added here). */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

/* plugin_prepare_allocation()'s (plugin.c, the "limit" plugin category)
 * linker dependency -- never invoked (this file never drives real
 * allocation). */
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
                              xlist_t* limits)
{
    (void)sched;
    (void)vec;
    (void)limits;
    return 0;
}

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * copy test_sim_event_loop.c/test_sim_startup.c/test_core_full_cycle.c each
 * carry (schedobj.c is not linked here: it would need plugin_register(),
 * which requires a genuine xce_scheduler_t* neither this file's fixtures
 * nor sim_loop.c's own core-scoped registration path have, Req 5 AC 17). */
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; }

/* ---------------------------------------------------------------------- */
/* Unit-level fixture helpers: hand-built xce_job_alloc_t trees.           */
/* ---------------------------------------------------------------------- */

static xce_alloc_t* make_test_alloc(const char* resname, const char* reserve,
                                    uint64_t instid, double value,
                                    uint64_t from, int64_t duration)
{
    xce_alloc_t* a = (xce_alloc_t*)calloc(1, sizeof(*a));

    TEST_ASSERT_NOT_NULL(a);
    a->resname  = strdup(resname);
    a->reserve  = strdup(reserve);
    a->instid   = instid;
    a->value    = value;
    a->from     = from;
    a->duration = duration;

    return a;
}

static xce_host_alloc_t* make_test_host_alloc(const char*   host,
                                              xce_alloc_t** allocs, int nallocs)
{
    xce_host_alloc_t* h = (xce_host_alloc_t*)calloc(1, sizeof(*h));

    TEST_ASSERT_NOT_NULL(h);
    h->host    = strdup(host);
    h->nallocs = nallocs;
    h->allocs  = allocs;

    return h;
}

static xce_job_alloc_t*
make_test_job_alloc(uint64_t jobid, xce_host_alloc_t** hosts, int nhosts)
{
    xce_job_alloc_t* j = (xce_job_alloc_t*)calloc(1, sizeof(*j));

    TEST_ASSERT_NOT_NULL(j);
    j->jobid         = jobid;
    j->nhosts        = nhosts;
    j->allocs        = hosts;
    j->nres_shared   = 0;
    j->allocs_shared = NULL;

    return j;
}

static replay_decision_t make_replay_dispatch_decision(xce_job_alloc_t* alloc)
{
    replay_decision_t          d;
    sched_dispatch_decision_t* dd
        = (sched_dispatch_decision_t*)calloc(1, sizeof(*dd));

    TEST_ASSERT_NOT_NULL(dd);
    dd->alloc = alloc;

    memset(&d, 0, sizeof(d));
    d.body_which    = ReplayDecision_body_dispatch;
    d.body.dispatch = dd;

    return d;
}

static replay_decision_t make_replay_pend_decision(uint64_t jobid,
                                                   int      main_reason)
{
    replay_decision_t      d;
    sched_pend_decision_t* pd = (sched_pend_decision_t*)calloc(1, sizeof(*pd));
    xce_job_preason_t* reason = (xce_job_preason_t*)calloc(1, sizeof(*reason));

    TEST_ASSERT_NOT_NULL(pd);
    TEST_ASSERT_NOT_NULL(reason);
    reason->main_reason = main_reason;
    pd->jobid           = jobid;
    pd->reason          = reason;

    memset(&d, 0, sizeof(d));
    d.body_which = ReplayDecision_body_pend;
    d.body.pend  = pd;

    return d;
}

/* ---------------------------------------------------------------------- */
/* Unit-level tests: sim_decision_eq() directly.                          */
/* ---------------------------------------------------------------------- */

static void test_dispatch_decisions_with_identical_content_match(void)
{
    xce_alloc_t* res_a[1]
        = { make_test_alloc("gpu", "resv1", 1, 2.0, 100, 1800) };
    xce_host_alloc_t* host_a[1] = { make_test_host_alloc("hostA", res_a, 1) };
    xce_job_alloc_t*  expected_alloc = make_test_job_alloc(42, host_a, 1);

    xce_alloc_t* res_b[1]
        = { make_test_alloc("gpu", "resv1", 1, 2.0, 100, 1800) };
    xce_host_alloc_t* host_b[1] = { make_test_host_alloc("hostA", res_b, 1) };
    xce_job_alloc_t*  actual_alloc = make_test_job_alloc(42, host_b, 1);

    replay_decision_t expected = make_replay_dispatch_decision(expected_alloc);
    xce_decision_t*   actual   = make_dispatch_decision(actual_alloc);

    /* Two structurally-identical, but never pointer-equal, allocation
     * trees: proves the comparison is real structural equality, not a
     * pointer/memcmp shortcut (this file's own header comment). */
    TEST_ASSERT_NOT_EQUAL(expected_alloc, actual_alloc);
    TEST_ASSERT_TRUE(sim_decision_eq(&expected, actual));

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

static void test_dispatch_decisions_differing_jobid_mismatch(void)
{
    xce_alloc_t*      res[1]  = { make_test_alloc("gpu", "r", 1, 1.0, 0, 0) };
    xce_host_alloc_t* host[1] = { make_test_host_alloc("hostA", res, 1) };
    xce_job_alloc_t*  expected_alloc = make_test_job_alloc(42, host, 1);

    xce_alloc_t*      res2[1]  = { make_test_alloc("gpu", "r", 1, 1.0, 0, 0) };
    xce_host_alloc_t* host2[1] = { make_test_host_alloc("hostA", res2, 1) };
    xce_job_alloc_t*  actual_alloc = make_test_job_alloc(43, host2, 1);

    replay_decision_t expected = make_replay_dispatch_decision(expected_alloc);
    xce_decision_t*   actual   = make_dispatch_decision(actual_alloc);

    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_dispatch_decisions_differing_host_mismatch(void)
{
    xce_alloc_t*      res[1]  = { make_test_alloc("gpu", "r", 1, 1.0, 0, 0) };
    xce_host_alloc_t* host[1] = { make_test_host_alloc("hostA", res, 1) };
    xce_job_alloc_t*  expected_alloc = make_test_job_alloc(42, host, 1);

    xce_alloc_t*      res2[1]  = { make_test_alloc("gpu", "r", 1, 1.0, 0, 0) };
    xce_host_alloc_t* host2[1] = { make_test_host_alloc("hostB", res2, 1) };
    xce_job_alloc_t*  actual_alloc = make_test_job_alloc(42, host2, 1);

    replay_decision_t expected = make_replay_dispatch_decision(expected_alloc);
    xce_decision_t*   actual   = make_dispatch_decision(actual_alloc);

    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_dispatch_decisions_differing_resource_value_mismatch(void)
{
    xce_alloc_t*      res[1]  = { make_test_alloc("gpu", "r", 1, 1.0, 0, 0) };
    xce_host_alloc_t* host[1] = { make_test_host_alloc("hostA", res, 1) };
    xce_job_alloc_t*  expected_alloc = make_test_job_alloc(42, host, 1);

    xce_alloc_t*      res2[1]  = { make_test_alloc("gpu", "r", 1, 2.0, 0, 0) };
    xce_host_alloc_t* host2[1] = { make_test_host_alloc("hostA", res2, 1) };
    xce_job_alloc_t*  actual_alloc = make_test_job_alloc(42, host2, 1);

    replay_decision_t expected = make_replay_dispatch_decision(expected_alloc);
    xce_decision_t*   actual   = make_dispatch_decision(actual_alloc);

    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_pend_decisions_with_identical_content_match(void)
{
    replay_decision_t expected = make_replay_pend_decision(9001ULL, 3);
    xce_pendreason_t* reason   = make_pendreason(9001ULL);
    xce_decision_t*   actual;

    reason->reasons->main_reason = 3;
    actual                       = make_pend_decision(reason);

    TEST_ASSERT_TRUE(sim_decision_eq(&expected, actual));

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

static void test_pend_decisions_differing_jobid_mismatch(void)
{
    replay_decision_t expected = make_replay_pend_decision(9001ULL, 0);
    xce_decision_t*   actual   = make_pend_decision(make_pendreason(9002ULL));

    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_pend_decisions_differing_main_reason_mismatch(void)
{
    replay_decision_t expected = make_replay_pend_decision(9001ULL, 3);
    xce_decision_t*   actual   = make_pend_decision(make_pendreason(9001ULL));

    /* actual's main_reason stays 0 (make_pendreason()'s own fresh-calloc
     * default) -- differs from expected's 3. */
    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_decision_kind_mismatch_reported(void)
{
    replay_decision_t expected = make_replay_pend_decision(1ULL, 0);
    xce_job_alloc_t*  alloc    = make_test_job_alloc(1, NULL, 0);
    xce_decision_t*   actual   = make_dispatch_decision(alloc);

    /* Recorded PEND vs. live DISPATCH for the SAME jobid: still a
     * mismatch -- op/body_which disagreement is reported, not ignored. */
    TEST_ASSERT_FALSE(sim_decision_eq(&expected, actual));

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

static void test_preempt_decision_always_reports_mismatch(void)
{
    xce_preempt_t     preempt;
    xce_decision_t    actual;
    replay_decision_t expected;

    memset(&preempt, 0, sizeof(preempt));
    preempt.jobid = 55ULL;
    memset(&actual, 0, sizeof(actual));
    actual.op        = XCE_DECISION_PREEMPT;
    actual.u.preempt = &preempt;

    /* Even a recorded entry that ALSO claims PREEMPT for the same jobid
     * still reports a mismatch -- recording_session.c's own documented
     * "no live path produces this decision type today" makes any PREEMPT
     * comparison unreachable-in-practice and therefore untrustworthy. */
    memset(&expected, 0, sizeof(expected));
    expected.body_which = ReplayDecision_body_preempt;

    TEST_ASSERT_FALSE(sim_decision_eq(&expected, &actual));

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

static void test_decision_eq_defends_against_null_inputs(void)
{
    xce_decision_t*   actual   = make_pend_decision(make_pendreason(1ULL));
    replay_decision_t expected = make_replay_pend_decision(1ULL, 0);

    TEST_ASSERT_FALSE(sim_decision_eq(NULL, actual));
    TEST_ASSERT_FALSE(sim_decision_eq(&expected, NULL));
    TEST_ASSERT_FALSE(sim_decision_eq(NULL, NULL));

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

/* ---------------------------------------------------------------------- */
/* Unit-level test: sim_log_mismatch()'s captured output content.         */
/* ---------------------------------------------------------------------- */

static char* capture_stdout_around(void (*fn)(void))
{
    char  tmp[] = "/tmp/xce_simverify_logcap_XXXXXX";
    int   fd    = mkstemp(tmp);
    int   saved_stdout;
    off_t len;
    char* buf;

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

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

    fn();

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

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

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

    return buf;
}

static void log_one_mismatch_for_capture(void)
{
    replay_decision_t expected = make_replay_pend_decision(9001ULL, 3);
    xce_decision_t*   actual   = make_pend_decision(make_pendreason(9001ULL));

    sim_log_mismatch(12345ULL, 2, &expected, actual);
}

static void test_log_mismatch_reports_logical_time_and_decision_detail(void)
{
    char* buf = capture_stdout_around(log_one_mismatch_for_capture);

    TEST_ASSERT_NOT_NULL_MESSAGE(strstr(buf, "MISMATCH"),
                                 "log_mismatch must report a MISMATCH");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(buf, "begin_round=12345"),
        "log_mismatch must report the session's exact logical time");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(buf, "decision_index=2"),
        "log_mismatch must report the decision's position within the "
        "session");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(buf, "jobid=9001"),
        "log_mismatch must report actionable decision detail (jobid)");

    free(buf);
    printf("test_log_mismatch_reports_logical_time_and_decision_detail: "
           "PASS\n");
}

/* ---------------------------------------------------------------------- */
/* Integration-level fixture helpers: a real .replay stream driving the   */
/* REAL builtin_fcfs picker to a genuine live PEND decision.              */
/* ---------------------------------------------------------------------- */

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

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

/* Builds a small, real xce_scheduler_t fixture purely to drive
 * sched_record_start() and produce a REAL, valid `.replay` header+snapshot
 * prefix (same technique test_sim_startup.c/test_sim_event_loop.c's own
 * build_fixture_scheduler() establish) -- but, unlike
 * test_sim_event_loop.c's copy, deliberately records schedPolicy=
 * "builtin_fcfs" (a string literal, never freed -- this throwaway fixture
 * scheduler is never torn down, matching this whole suite's documented
 * accepted-leak convention) so sim_open_and_validate_header() carries
 * "builtin_fcfs" into sim->core.job_pick_plugin and
 * sim_register_default_picker() (sim_loop.c) registers the REAL
 * builtin_fcfs plugin -- this file's own Observable needs the real picker,
 * not a stub, to prove sim_decision_eq() against a genuinely-produced live
 * decision. No picker needs to be key_add'ed into THIS throwaway
 * scheduler's own core.plugins: xce_sched_core_save_snapshot() does not
 * require a resolvable picker (Req 5 AC 21: "Picker state is NOT stored in
 * snapshot").
 */
static void build_fixture_scheduler(xce_scheduler_t* sched, char* record_dir)
{
    int rc;

    memset(sched, 0, sizeof(*sched));
    mylock_init(&(sched->record_lock));

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

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

/* Same "leave the file open, skip sched_record_stop()'s own endSession
 * marker" technique test_sim_event_loop.c's own write_header_and_snapshot()
 * establishes -- see that function's doc comment for the full rationale. */
static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len,
                                       const char* suffix)
{
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;

    make_tmp_dir(record_dir, sizeof(record_dir), suffix);
    build_fixture_scheduler(&sched, record_dir);

    rc = sched_record_start(&sched, "fixture.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;
}

/* Appends one ReplayEvent{queueUpdate} message -- creates `queue_name` in
 * the core (core_events.c's core_upsert_queue(), fires
 * on_queue_update(picker, q) against the now-registered real builtin_fcfs).
 * status/priority are set but never gate readiness: core.c's own
 * sched_queue_is_ready() is an unconditional `return 1` stub (task 2.4's
 * still-live documented scope note) -- any value here works. */
static void append_queue_update_event(FILE* f, uint64_t ts,
                                      const char* queue_name)
{
    xce_queue_t          queue;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&queue, 0, sizeof(queue));
    queue.queue    = (char*)queue_name;
    queue.status   = QUEUE_STAT_OPEN;
    queue.priority = 1;

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

    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);
}

/* Appends one ReplayEvent{jobsPush} message with exactly one
 * JOB_STAT_PEND job assigned to `queue_name` -- MUST be appended AFTER
 * append_queue_update_event() for the same queue name within the same
 * session's accumulated batch: core_build_job() (core_events.c) looks the
 * queue up via sched_get_queue() with no get-or-create fallback (unlike
 * user/project), so the queue must already exist. Drives
 * core_events.c's real JOBS_PUSH handling: sched_add_job() +
 * on_job_push(picker, job) against the real builtin_fcfs -- which (since
 * sched_job_is_ready()/sched_queue_is_ready() are unconditional `return 1`
 * stubs) makes this job an immediate next_job() candidate, so the picker
 * thread PENDs it for real via make_pend_decision(make_pendreason(jobid))
 * (sim_loop.c's own sim_picker_thread_fn).
 */
static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
                                   const char* queue_name)
{
    job_submit_data_t    submit;
    xce_job_t            job;
    xce_job_t*           jobs[1];
    xce_jobs_data_t      jobs_data;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&submit, 0, sizeof(submit));
    submit.queue    = (char*)queue_name;
    submit.user     = (char*)"alice";
    submit.mintasks = 1;
    submit.maxtasks = 1;

    memset(&job, 0, sizeof(job));
    job.jobid  = jobid;
    job.state  = JOB_STAT_PEND;
    job.submit = &submit;

    jobs[0] = &job;

    memset(&jobs_data, 0, sizeof(jobs_data));
    jobs_data.n_total = 1;
    jobs_data.count   = 1;
    jobs_data.jobs    = jobs;

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

    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);
}

/* Appends the .replay stream's whole-stream termination marker
 * (ReplayEvent{endSession}, Req 5 AC 20) -- same technique
 * test_sim_event_loop.c's own append_end_of_stream_marker() establishes. */
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);
}

/* Appends one ReplaySession carrying exactly one recorded PendDecision
 * (jobid, main_reason) -- the "recorded output" side of this file's own
 * Observable, built via the real encode_ReplaySession_ptr() codec
 * (recording_session.c's own real encode call site, not a hand-rolled
 * byte layout).
 */
static void append_replay_session_with_pend(FILE* f, uint64_t begin_round,
                                            uint64_t jobid, int main_reason)
{
    xce_job_preason_t     reason;
    sched_pend_decision_t pend;
    replay_decision_t     decision;
    replay_decision_t*    decisions[1];
    replay_session_t      session;
    struct capn           c;
    struct capn_segment*  cs;
    ReplaySession_ptr     sp;

    memset(&reason, 0, sizeof(reason));
    reason.main_reason = main_reason;

    memset(&pend, 0, sizeof(pend));
    pend.jobid  = jobid;
    pend.reason = &reason;

    memset(&decision, 0, sizeof(decision));
    decision.ts         = begin_round;
    decision.body_which = ReplayDecision_body_pend;
    decision.body.pend  = &pend;

    decisions[0] = &decision;

    memset(&session, 0, sizeof(session));
    session.begin_round = begin_round;
    session.decisions   = decisions;
    session.n_decisions = 1;
    session.summary     = NULL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplaySession_ptr(cs, &sp, &session);
    capn_setp(capn_root(&c), 0, sp.p);

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

/* Runs sched_sim_driver_run() against `path` with --sim-verify set to
 * `verify`, capturing everything written to stdout during the call (the
 * same dup()/dup2() technique capture_stdout_around() above uses, inlined
 * here since sched_sim_driver_run() needs live arguments, not a bare
 * void(void) callback). */
static char* run_sim_and_capture_stdout(const char* path, bool verify,
                                        int* out_rc)
{
    sched_sim_args_t       args;
    xce_sched_sim_driver_t sim;
    char                   tmp[] = "/tmp/xce_simverify_run_XXXXXX";
    int                    fd    = mkstemp(tmp);
    int                    saved_stdout;
    off_t                  len;
    char*                  buf;

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

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

    memset(&args, 0, sizeof(args));
    args.replay_path  = path;
    args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    args.sim_speed    = 0; /* frozen -- deterministic, no wall-clock wait */
    args.verify       = verify;

    *out_rc = sched_sim_driver_run(&sim, &args);

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

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

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

    return buf;
}

/* ---------------------------------------------------------------------- */
/* Integration-level tests: full sched_sim_driver_run(), --sim-verify.    */
/* ---------------------------------------------------------------------- */

/*
 * Observable (a): replaying a fixture against its own previously-recorded
 * output (here: the exact PEND decision the real picker deterministically
 * produces -- main_reason=0, make_pendreason()'s own fresh-calloc default)
 * with verification enabled reports ZERO mismatches.
 */
static void test_verify_against_self_reports_zero_mismatches(void)
{
    char  path[PATH_MAX];
    FILE* f;
    int   rc;
    char* captured;

    f = write_header_and_snapshot(path, sizeof(path), "_selfmatch");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, 777ULL, "normal");
    append_replay_session_with_pend(f, 2000ULL, 777ULL, 0);
    append_end_of_stream_marker(f);
    fclose(f);

    captured = run_sim_and_capture_stdout(path, true, &rc);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "MISMATCH"),
        "verifying a fixture against its own previously-recorded decision "
        "must report zero mismatches");

    free(captured);
    printf("test_verify_against_self_reports_zero_mismatches: PASS\n");
}

/*
 * Observable (b): replaying against a deliberately-altered copy (recorded
 * main_reason flipped to 99, which the real picker thread can never
 * produce -- make_pendreason() always calloc()s a fresh, all-zero
 * xce_job_preason_t) reports the EXACT mismatch: MISMATCH is logged,
 * naming the specific job.
 */
static void test_verify_against_altered_copy_reports_exact_mismatch(void)
{
    char  path[PATH_MAX];
    FILE* f;
    int   rc;
    char* captured;

    f = write_header_and_snapshot(path, sizeof(path), "_altered");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, 777ULL, "normal");
    append_replay_session_with_pend(f, 2000ULL, 777ULL, 99);
    append_end_of_stream_marker(f);
    fclose(f);

    captured = run_sim_and_capture_stdout(path, true, &rc);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "MISMATCH"),
        "verifying against a deliberately-altered recorded decision must "
        "report a mismatch");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "jobid=777"),
        "the mismatch report must name the specific job");

    free(captured);
    printf("test_verify_against_altered_copy_reports_exact_mismatch: "
           "PASS\n");
}

/*
 * sim->verify == false must skip verification entirely, even against the
 * SAME deliberately-altered fixture the previous test proves DOES mismatch
 * when verification is on -- proves the --sim-verify gate itself, not just
 * the comparator.
 */
static void test_verify_disabled_reports_no_mismatches_even_when_altered(void)
{
    char  path[PATH_MAX];
    FILE* f;
    int   rc;
    char* captured;

    f = write_header_and_snapshot(path, sizeof(path), "_verifyoff");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, 777ULL, "normal");
    append_replay_session_with_pend(f, 2000ULL, 777ULL, 99);
    append_end_of_stream_marker(f);
    fclose(f);

    captured = run_sim_and_capture_stdout(path, false, &rc);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "MISMATCH"),
        "sim->verify == false must never report a mismatch, even against a "
        "fixture the verify-enabled test proves mismatches");

    free(captured);
    printf("test_verify_disabled_reports_no_mismatches_even_when_altered: "
           "PASS\n");
}

int main(void)
{
    UNITY_BEGIN();

    RUN_TEST(test_dispatch_decisions_with_identical_content_match);
    RUN_TEST(test_dispatch_decisions_differing_jobid_mismatch);
    RUN_TEST(test_dispatch_decisions_differing_host_mismatch);
    RUN_TEST(test_dispatch_decisions_differing_resource_value_mismatch);
    RUN_TEST(test_pend_decisions_with_identical_content_match);
    RUN_TEST(test_pend_decisions_differing_jobid_mismatch);
    RUN_TEST(test_pend_decisions_differing_main_reason_mismatch);
    RUN_TEST(test_decision_kind_mismatch_reported);
    RUN_TEST(test_preempt_decision_always_reports_mismatch);
    RUN_TEST(test_decision_eq_defends_against_null_inputs);
    RUN_TEST(test_log_mismatch_reports_logical_time_and_decision_detail);
    RUN_TEST(test_verify_against_self_reports_zero_mismatches);
    RUN_TEST(test_verify_against_altered_copy_reports_exact_mismatch);
    RUN_TEST(test_verify_disabled_reports_no_mismatches_even_when_altered);

    return UNITY_END();
}