myxce/src/scheduler/testsuite/test_sim_equivalence.c

/*
 * test_sim_equivalence.c - deterministic replay and normal-to-simulation
 * equivalence test (scheduler-core spec, task 11.6, Req 5 AC 7.1). Per
 * design.md's own SimDriver component doc: "the highest-value test -- it
 * directly exercises the central invariant." Per requirements.md's own
 * framing, this is NOT a debugging convenience -- it is the validation
 * contract for the entire core/driver architecture: given identical initial
 * configuration and an identical event sequence, the scheduler must produce
 * identical scheduling decisions whether it runs in normal (production) mode
 * or in simulation (replay) mode.
 *
 * What this test does (AC 7.1's own recipe, driven end to end against REAL,
 * already-implemented production code -- no new production code is added by
 * this task):
 *
 *   1. Record a synthetic scheduling session through the REAL normal-mode
 *      driver (xce_sched_driver_run_normal_mode_round(), driver.c) with
 *      recording genuinely active (sched_record_start(), recording.c/
 *      recording_events.c/recording_session.c, task 9.2) -- producing a
 *      genuine `.replay` file R on disk, exactly the technique
 *      test_record_lifecycle.c's own
 *      test_full_cycle_start_event_session_stop_produces_expected_stream
 *      establishes, extended here with real jobs across two queues so the
 *      session produces real decisions instead of zero.
 *   2. Replay R through the REAL simulation driver
 *      (sched_sim_driver_run(), sim.c/sim_loop.c, tasks 11.1/11.2) with
 *      --sim-output set, producing a second `.replay` file S -- exactly the
 *      technique test_sim_record_output.c's own
 *      test_output_file_is_decodable_with_full_stream_symmetry establishes.
 *   3. Decode both files' ReplaySession.decisions lists
 *      (decode_ReplaySession_ptr()/free_ReplaySession_ptr(),
 *      controller.capnp.h) and assert R's and S's decisions match
 *      identically -- same jobid, same pend reason content, per decision,
 *      in order -- while deliberately never comparing ReplayHeader or
 *      SessionSummary fields (AC 7.1's own explicit carve-out: "only
 *      non-decision metadata may differ").
 *
 * Why every decision here is PEND, never DISPATCH (a load-bearing scope
 * decision, not an oversight -- read this before changing the fixture):
 *
 *   sim_loop.c's sim_picker_thread_fn() (task 11.2) -- the simulation
 *   driver's ONLY source of live decisions -- conservatively PENDs every
 *   scheduling candidate it is handed. Its own file header comment
 *   documents this as a "Known scope boundary": real resource-matching
 *   allocation (sched_make_allocation(), scheduler.c) requires a live
 *   xce_scheduler_t* the simulation driver deliberately does not have (Req 5
 *   AC 17), and has no pure-core equivalent yet -- that extraction belongs
 *   to the separate, not-yet-designed scheduler-allocation spec
 *   (.kiro/specs/scheduler-allocation). testsuite/meson.build's own comment
 *   on sched_sim_verify_test independently confirms the same fact from task
 *   11.4's side: "including DISPATCH, which sim_loop.c's own real picker
 *   thread can never produce live ... so [a hand-built decision pair] is
 *   the only way to prove that comparator for real."
 *
 *   In other words: as of this task, the simulation driver structurally
 *   CANNOT emit a DISPATCH decision under any replayed event sequence --
 *   this was independently re-confirmed while building this test (a
 *   synthetic normal-mode recording containing a fabricated DISPATCH
 *   decision, replayed through the unmodified production sim driver,
 *   deterministically comes back as PEND for that same job every time,
 *   exactly matching sim_picker_thread_fn()'s documented behavior -- not a
 *   flake, not a probe of this test's own bug). Given this task's explicit
 *   production-code boundary (do not touch sim.c/sim_loop.c/sim_verify.c;
 *   closing this gap is scheduler-allocation-spec territory, a materially
 *   bigger, separate problem than "wire the equivalence test"), and given
 *   the task text's own explicit instruction not to paper over a genuine
 *   divergence with a weakened test, this test's fixture is scoped to PEND
 *   decisions -- the ONE decision type BOTH pipelines can genuinely,
 *   faithfully reproduce today -- rather than silently asserting equality
 *   only on a subset of a mixed DISPATCH+PEND fixture (which would hide a
 *   real, confirmed architectural gap behind an apparently-passing test).
 *   This is reported explicitly in this task's own status report, not
 *   hidden here.
 *
 * The fixture is still a genuine, non-trivial regression guard, not a
 * single-job happy path that could pass by accident: TWO queues at
 * different priorities ("high"=5, "normal"=1) and FOUR jobs split across
 * them, forcing the real builtin_fcfs picker's actual priority-desc /
 * submission-time-asc candidate ordering to run for real, identically, on
 * both the recording side and the replay side -- a single-job fixture could
 * pass even if event translation or picker-state-rebuild silently dropped
 * every field the ordering logic itself does not read.
 *
 * Comparator design: reuses sim_verify.c's own leaf-level structural
 * comparators, sim_verify_job_alloc_eq()/sim_verify_job_preason_eq() (both
 * `static`, but visible in this TU via the white-box #include of
 * ../src/sim_verify.c below -- ordinary C translation-unit visibility, not a
 * new sharing mechanism). sim_decision_eq() itself (also in sim_verify.c)
 * is NOT reused directly: it compares a replay_decision_t against a LIVE
 * xce_decision_t (recorded-vs-live), a different pair of top-level shapes
 * than this file needs (recorded-vs-recorded, replay_decision_t vs
 * replay_decision_t, R vs S) -- but the SAME xce_job_alloc_t/
 * xce_job_preason_t leaf types and their SAME field-by-field comparators
 * apply either way, so this file's own small
 * replay_decision_content_eq() below delegates to those two leaf
 * comparators instead of re-deriving the field walk from scratch.
 *
 * White-box #include set: the union of test_record_lifecycle.c's own set
 * (core.c, core_clock.c, core_events.c, core_session.c, driver.c, plugin.c,
 * recording.c, recording_events.c, recording_session.c, snapshot.c,
 * snapshot_wire.c) and test_sim_record_output.c's own set (core.c,
 * core_clock.c, core_events.c, core_session.c, decision.c, plugin.c,
 * plugins/builtin_fcfs.c, recording.c, recording_session.c, repl.c, sim.c,
 * sim_loop.c, sim_record.c, sim_verify.c, snapshot.c, snapshot_load.c,
 * snapshot_wire.c -- repl.c added by task 12.1, required at link time by
 * sim_loop.c's own repl.h inline wrappers) -- no prior test file combines
 * both the normal-mode
 * driver stack and the full simulation stack in one TU; this task is the
 * first to genuinely need both halves of the pipeline live simultaneously.
 * Verified no static-symbol collisions exist across the combined file set
 * (mechanical scan across all 17 files' `static` function names) before
 * writing this file.
 *
 * Stub surface: the union of test_record_lifecycle.c's driver-side stubs
 * (kill_sched_*, sched_worker_release_host_ctx, sched_fill_limit_requests,
 * sched_account_jobs, sched_resolve_job_picker) and
 * test_driver_decision_cb.c's publish-side stubs
 * (sched_publish_dispatch/sched_publish_pend_reason -- never actually
 * invoked here since this test never registers a decision callback, but
 * driver.c's xce_sched_driver_on_decision() still references them at link
 * time) plus xcecli_reply_schedadmin() (test_admin_request.c's own spy
 * pattern -- never invoked here either, since this test never enqueues an
 * XCE_SCHED_OP_ADMIN_RECORD request, but handle_deferred_admin_record()
 * still references it at link time).
 */
#include <errno.h>
#include <limits.h>
#include <pthread.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 every other file in this directory establishes -- see
 * this file's own header comment for the exact combined file set and why
 * it is novel). */
#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/driver.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_events.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#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 -- 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) linker dependency -- never
 * invoked (this fixture never drives real resource-matching allocation --
 * see this file's header comment). */
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
                              xlist_t* limits)
{
    (void)sched;
    (void)vec;
    (void)limits;
    return 0;
}

/* xce_sched_driver_run_normal_mode_round()'s (driver.c) linker dependency --
 * a no-op stand-in, same rationale test_record_lifecycle.c's own copy
 * documents (real resource accounting is scheduler.c/resource.c territory,
 * not exercised here). */
int sched_account_jobs(xce_scheduler_t* sched)
{
    (void)sched;
    return 0;
}

/* xce_sched_driver_on_decision()'s (driver.c) publish-side linker
 * dependencies -- never actually invoked at runtime: this test never
 * registers a decision callback (xce_core_set_decision_cb() is never
 * called), so xce_sched_driver_run_normal_mode_round()'s
 * `sched->core.on_decision != NULL` check is always false and every
 * decision is simply WARN-logged and dropped after being recorded. Same
 * link-satisfying-only rationale as test_driver_decision_cb.c's own spies,
 * simplified to plain no-ops since this file asserts nothing about publish
 * call counts. */
int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d)
{
    (void)sched;
    (void)d;
    return 0;
}

int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
{
    (void)sched;
    (void)r;
    return 0;
}

/* handle_deferred_admin_record()'s (driver.c) linker dependency -- never
 * invoked (this test never enqueues XCE_SCHED_OP_ADMIN_RECORD). Same spy
 * signature test_admin_request.c's own stub establishes. */
int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq,
                            sched_admin_reply_t* reply)
{
    (void)c;
    (void)seq;
    (void)reply;
    return 0;
}

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * algorithm every white-box test in this directory that does not link
 * schedobj.c already establishes (test_record_lifecycle.c,
 * test_sim_record_output.c, etc.). Shared by core_events.c's internal
 * on_queue_update/on_job_push dispatch, this file's own picker/allocator
 * thread functions below, AND sim_loop.c's sim_picker_thread_fn()/
 * sim_register_default_picker() -- one real definition, every call site in
 * this combined TU resolves through it. */
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; }

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

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

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

/* Real xce_sched_core_init() + manually dataq_init'd pend_updates + the
 * REAL builtin_fcfs picker registered directly against the core (task
 * 11.2's plugin_register_core()/builtin_fcfs_new_plugin() path -- NOT the
 * xce_scheduler_t*-requiring plugin_register()) -- deliberately the REAL
 * picker, not a stub, since the whole point of this test is that the SAME
 * candidate-ordering algorithm must run, identically, on both the
 * recording side and the replay side. `record_dir` must outlive `sched`
 * (caller-owned storage), same convention test_record_lifecycle.c's own
 * fixture_init() establishes. */
static void fixture_init(xce_scheduler_t* sched, char* record_dir)
{
    xce_sched_plugin_t* fcfs;
    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.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
    dataq_init(sched->core.pend_updates[0], NULL);
    sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
    dataq_init(sched->core.pend_updates[1], NULL);

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

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

/* Builds one heap-owned, deep-freeable xce_queue_t -- release_translated_
 * payload()'s (driver.c) unconditional QUEUE_UPDATE cleanup
 * (free_QueueConf_ptr()) genuinely frees this struct and its string fields,
 * so every field it touches (queue, resreq) must be independently
 * heap-allocated, never a string literal. */
static xce_queue_t* make_queue(const char* name, int32_t priority)
{
    xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));

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

    return q;
}

/* Builds one heap-owned, deep-freeable xce_job_t -- release_translated_
 * payload()'s (driver.c) unconditional UPDATE_JOB cleanup (free_Job_ptr()
 * per element) genuinely frees this struct, its submit_data_t, and every
 * string field inside it, so every field set below is independently
 * heap-allocated (strdup() -- NOT makeString(), whose returned pointer is
 * NOT a plain malloc()-compatible buffer and crashes a later plain free(),
 * confirmed the hard way against this exact free_QueueConf()/free_Job()
 * cleanup path while building this fixture), never a string literal --
 * same rationale as make_queue() above. `resreq`/`mintasks`/`maxtasks` are
 * set to the same
 * trivially-satisfiable values test_core_full_cycle.c's own make_job()
 * uses -- core.c's sched_job_is_ready()/sched_queue_is_ready() stubs (Phase
 * 2/3's own documented readiness-check placeholder) make every pending job
 * an immediate picker candidate regardless of resreq content, so this test
 * does not depend on resreq being satisfiable against any real resource
 * data. */
static xce_job_t* make_job(uint64_t jobid, uint64_t sub_time, const char* queue,
                           const char* user)
{
    xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));

    j->jobid      = jobid;
    j->state      = JOB_STAT_PEND;
    j->sub_time   = sub_time;
    j->queue_time = sub_time;
    j->start_time = 0;

    j->submit        = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    j->submit->queue = strdup(queue);
    j->submit->user  = strdup(user);
    j->submit->project_name = strdup("equiv_proj");
    j->submit->resreq       = strdup("type == any");
    j->submit->mintasks     = 1;
    j->submit->maxtasks     = 1;

    return j;
}

/* Enqueues one XCE_SCHED_OP_UPDATE_QUEUE request into pend_updates[0] --
 * the SAME dataq mechanism handler.c populates in production, drained by
 * xce_sched_driver_translate_pending_events() (called from
 * xce_sched_driver_run_normal_mode_round()), matching
 * test_record_lifecycle.c's own JSTATE_CHANGE example, extended here to the
 * QUEUE_UPDATE/UPDATE_JOB op types this fixture actually needs. */
static void enqueue_queue_update(xce_scheduler_t* sched, const char* name,
                                 int32_t priority)
{
    xce_sched_update_request_t* req
        = (xce_sched_update_request_t*)calloc(1, sizeof(*req));

    req->op                  = XCE_SCHED_OP_UPDATE_QUEUE;
    req->u.queueupdate.queue = make_queue(name, priority);

    dataq_enq(sched->core.pend_updates[0], &(req->link));
}

static void enqueue_jobs_push(xce_scheduler_t* sched, xce_job_t** jobs,
                              int njobs)
{
    xce_sched_update_request_t* req
        = (xce_sched_update_request_t*)calloc(1, sizeof(*req));

    req->op                = XCE_SCHED_OP_UPDATE_JOB;
    req->u.jobupdate.njobs = njobs;
    req->u.jobupdate.jobs  = jobs;

    dataq_enq(sched->core.pend_updates[0], &(req->link));
}

/* ---------------------------------------------------------------------- */
/* Normal-mode picker/allocator threads -- jbuf's hardcoded capacity-1      */
/* rendezvous (Req 5 AC 7) requires a REAL, separate picker thread, same    */
/* design test_record_lifecycle.c/test_core_full_cycle.c already establish. */
/* ---------------------------------------------------------------------- */

/* Generic real-picker-driving thread -- resolves WHATEVER picker
 * core->job_pick_plugin names (here, the REAL builtin_fcfs registered by
 * fixture_init()) and drives it through one session's
 * begin_session/next_job/end_session protocol. Identical in structure to
 * test_record_lifecycle.c's own picker_thread_fn() (which drives a stub
 * picker) -- this function is picker-agnostic by construction. */
static void* 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;
}

/*
 * stand_in_allocator_thread_fn - normal-mode side's stand-in for
 * scheduler.c's real sched_allocator() thread (real resource-matching
 * allocation is out of this test's scope -- see this file's own header
 * comment). Deliberately mirrors sim_loop.c's sim_picker_thread_fn() inner
 * loop content byte-for-byte: for every candidate jobid handed off via
 * jbuf, call the resolved picker's real on_pend() hook then
 * make_pend_decision(make_pendreason(jobid)) -- the SAME production
 * constructors sim_picker_thread_fn() calls, with the SAME jobid argument.
 * This is what makes the equivalence assertion below non-tautological in a
 * useful way: the recorded decision's CONTENT is produced by an
 * independent call to the same real constructors, not copied from
 * anywhere, and must still match the simulation side's independently
 * produced decision for the same jobid.
 */
static void* stand_in_allocator_thread_fn(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;
    uint64_t            jobid;

    if (sched_resolve_job_picker(core, &picker) < 0) {
        xce_decision_t* decision = NULL;

        cbuf_get(&(core->jbuf), &jobid); /* drain the lone 0 sentinel */
        cbuf_put(&(core->decisions), &decision);
        return NULL;
    }

    for (;;) {
        xce_decision_t* decision;

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

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

    return NULL;
}

/* Runs one complete normal-mode round (real picker thread + stand-in
 * allocator thread) and blocks until it completes -- same design as
 * test_record_lifecycle.c's own run_one_real_session(). */
static void run_one_real_session(xce_scheduler_t* sched)
{
    pthread_t picker_tid;
    pthread_t allocator_tid;
    int       rc;

    rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    rc = pthread_create(&allocator_tid, NULL, stand_in_allocator_thread_fn,
                        &(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    xce_sched_driver_run_normal_mode_round(sched);

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

/* ---------------------------------------------------------------------- */
/* Decoding + comparison                                                   */
/* ---------------------------------------------------------------------- */

/* This fixture pushes exactly 3 events in one batch (2 QUEUE_UPDATE + 1
 * UPDATE_JOB carrying all 4 jobs) -- both sched_record_start()'s own
 * ReplayHeader+initial-SnapshotEntry pair (Req 5 AC 20) and those 3 events
 * come before the ReplaySession message on BOTH R (recording.c's direct
 * write order) and S (sim_loop.c's pass-through write order, Req 5 AC 19 --
 * every ReplayEvent it reads from R it re-writes to S in the same order,
 * before ever reaching the ReplaySession boundary) -- so this fixed skip
 * count is valid for decoding EITHER file. */
#define N_MSGS_BEFORE_SESSION 5 /* ReplayHeader, SnapshotEntry, 3 events */

/* Opens `path`, skips exactly `n_skip` capnp messages structurally (same
 * "capn_init_fp() + capn_free(), no decode" technique
 * test_record_lifecycle.c's own skip_header_and_initial_snapshot() already
 * establishes), then decodes the NEXT message as a ReplaySession via
 * decode_ReplaySession_ptr() -- the SAME decoder sim_loop.c's own
 * --sim-verify comparison path and test_sim_record_output.c's own decode
 * test both already use. */
static replay_session_t* decode_session_after_skip(const char* path, int n_skip)
{
    FILE*             f;
    struct capn       c;
    ReplaySession_ptr p;
    replay_session_t* dst = NULL;
    int               i;

    f = fopen(path, "r");
    TEST_ASSERT_NOT_NULL(f);

    for (i = 0; i < n_skip; i++) {
        TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
        capn_free(&c);
    }

    TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
    p.p = capn_getp(capn_root(&c), 0, 1);
    decode_ReplaySession_ptr(&dst, p);
    capn_free(&c);

    fclose(f);

    return dst;
}

/*
 * replay_decision_content_eq - structural equality between two DECODED
 * ReplayDecision entries -- R's recorded decision vs S's simulated decision
 * (a different comparison pair than sim_verify.c's own sim_decision_eq(),
 * which compares a recorded replay_decision_t against a LIVE xce_decision_t
 * -- see this file's own header comment for why a small new comparator is
 * written here rather than reusing sim_decision_eq() itself, while still
 * reusing its two LEAF comparators, sim_verify_job_alloc_eq()/
 * sim_verify_job_preason_eq() -- both `static` in sim_verify.c but visible
 * in this TU via this file's white-box #include of that file above).
 * AC 7.1 excludes ReplayHeader/SessionSummary from the equivalence claim --
 * this function, and every call site below, touches only `body_which` and
 * the DISPATCH/PEND payload, never `ts` or any summary field.
 */
static int replay_decision_content_eq(const replay_decision_t* r,
                                      const replay_decision_t* s)
{
    if ((r == NULL) || (s == NULL)) {
        return 0;
    }
    if (r->body_which != s->body_which) {
        return 0;
    }

    switch (r->body_which) {
    case ReplayDecision_body_dispatch:
        if ((r->body.dispatch == NULL) || (s->body.dispatch == NULL)) {
            return 0;
        }
        return sim_verify_job_alloc_eq(r->body.dispatch->alloc,
                                       s->body.dispatch->alloc);

    case ReplayDecision_body_pend:
        if ((r->body.pend == NULL) || (s->body.pend == NULL)) {
            return 0;
        }
        return (r->body.pend->jobid == s->body.pend->jobid)
            && sim_verify_job_preason_eq(r->body.pend->reason,
                                         s->body.pend->reason);

    default:
        return 0;
    }
}

/* ---------------------------------------------------------------------- */
/* The central regression test (task 11.6's own Observable).               */
/* ---------------------------------------------------------------------- */

static void
test_normal_mode_recording_and_simulation_replay_produce_identical_decisions(
    void)
{
    char                   record_dir[PATH_MAX];
    xce_scheduler_t        sched;
    char                   r_path[PATH_MAX];
    char                   s_path[PATH_MAX];
    int                    s_fd;
    xce_job_t**            jobs;
    sched_sim_args_t       sim_args;
    xce_sched_sim_driver_t sim;
    replay_session_t*      r_session = NULL;
    replay_session_t*      s_session = NULL;
    int                    rc;
    int                    i;

    /* ---- Half 1: record a real normal-mode session -> file R ---- */
    make_tmp_dir(record_dir, sizeof(record_dir), "_r");
    fixture_init(&sched, record_dir);

    rc = sched_record_start(&sched, "normal.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < sizeof(r_path));
    strcpy(r_path, sched.record_file_path);

    /* Two queues at different priorities, four jobs split across them --
     * see this file's header comment for why this variety matters (forces
     * the real builtin_fcfs cross-queue priority ordering to run for real,
     * identically, on both sides). */
    enqueue_queue_update(&sched, "high", 5);
    enqueue_queue_update(&sched, "normal", 1);

    jobs    = (xce_job_t**)calloc(4, sizeof(xce_job_t*));
    jobs[0] = make_job(9001ULL, 1000ULL, "high", "alice");
    jobs[1] = make_job(9002ULL, 1001ULL, "high", "bob");
    jobs[2] = make_job(9003ULL, 1002ULL, "normal", "carol");
    jobs[3] = make_job(9004ULL, 1003ULL, "normal", "dave");
    enqueue_jobs_push(&sched, jobs, 4);

    run_one_real_session(&sched);

    /* Sanity check on the recorded batch shape (2 queue updates + 1 batched
     * jobs-push = 3 events) -- pins N_MSGS_BEFORE_SESSION's own assumption. */
    TEST_ASSERT_EQUAL_UINT64(3ULL, sched.record_event_count);

    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* ---- Half 2: replay R through the REAL, unmodified simulation
     * driver, with --sim-output set -> file S ---- */
    snprintf(s_path, sizeof(s_path), "/tmp/xce_simequiv_out_XXXXXX");
    s_fd = mkstemp(s_path);
    TEST_ASSERT_TRUE(s_fd >= 0);
    close(s_fd);

    memset(&sim_args, 0, sizeof(sim_args));
    sim_args.replay_path  = r_path;
    sim_args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    sim_args.sim_speed    = 0; /* frozen -- deterministic, no wall-clock skew */
    sim_args.output_path  = s_path;

    rc = sched_sim_driver_run(&sim, &sim_args);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* ---- Decode both files' ReplaySession and compare decision content
     * only -- never ReplayHeader, never SessionSummary (AC 7.1's own
     * explicit carve-out: "only non-decision metadata may differ"). R's and
     * S's ReplayHeader genuinely DO differ (S's is fresh sim-run metadata,
     * never copied from R -- Req 5 AC 19, already proven independently by
     * test_sim_record_output.c's own header test), and this comparison
     * must -- and does -- still succeed regardless. ---- */
    r_session = decode_session_after_skip(r_path, N_MSGS_BEFORE_SESSION);
    s_session = decode_session_after_skip(s_path, N_MSGS_BEFORE_SESSION);

    TEST_ASSERT_NOT_NULL(r_session);
    TEST_ASSERT_NOT_NULL(s_session);

    /* Non-tautology guard: an empty/zero-decision comparison would
     * trivially "match" without proving anything (this task's own explicit
     * concern) -- both sides must genuinely contain all 4 decisions this
     * fixture pushed. */
    TEST_ASSERT_EQUAL_INT(4, r_session->n_decisions);
    TEST_ASSERT_EQUAL_INT(r_session->n_decisions, s_session->n_decisions);

    for (i = 0; i < r_session->n_decisions; i++) {
        /* Independently pin the expected decision TYPE (not just "R and S
         * agree with each other", which a shared bug could satisfy too --
         * see this file's header comment on why PEND is this fixture's
         * only achievable decision type today). */
        TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_pend,
                              (int)r_session->decisions[i]->body_which);

        TEST_ASSERT_TRUE(replay_decision_content_eq(r_session->decisions[i],
                                                    s_session->decisions[i]));
    }

    free_ReplaySession_ptr(&r_session);
    free_ReplaySession_ptr(&s_session);

    unlink(s_path);

    printf("test_normal_mode_recording_and_simulation_replay_produce_"
           "identical_decisions: PASS\n");
}

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