myxce/src/scheduler/testsuite/test_repl_equivalence.c

/*
 * test_repl_equivalence.c - REPL step/continue-to-completion vs batch
 * simulation equivalence test (scheduler-core spec, task 12.4, Req 5 AC 22).
 *
 * This is NOT task 11.6's own regression test (test_sim_equivalence.c),
 * which compares the NORMAL-mode driver against the SIMULATION driver. This
 * task compares the SAME simulation driver (sched_sim_driver_run(),
 * sim_loop.c) against itself, run twice against the SAME `.replay` fixture
 * file, differing ONLY in `sim->interactive` -- once driven by a scripted
 * `step`/`continue` REPL session (stdin freopen()'d from a script file, the
 * SAME mechanism test_repl_execution_control.c's own Test 2/Test 3
 * establish), once run straight through in batch mode. This is the exact
 * empirical claim design.md's SimDriver "Invariants" section and
 * requirements.md's Req 5 AC 22 "Deterministic replay is preserved"
 * paragraph make: "The REPL never alters core state or event stream. It
 * only controls *when* events are processed (by pausing), not *what* events
 * are processed ... Interactive use with `step` and `continue` that runs to
 * completion produces the same output as `--sim` without `--interactive`."
 *
 * Fixture (a genuine, non-trivial, PEND-only sequence -- see
 * test_sim_equivalence.c's own header comment for why every decision the
 * simulation driver can produce today is PEND, never DISPATCH:
 * sim_picker_thread_fn()'s documented, out-of-this-task's-scope gap): TWO
 * queues at different priorities ("high"=5, "normal"=1, forcing the real
 * builtin_fcfs cross-queue priority ordering to run for real) and FOUR jobs
 * split across THREE separate ReplaySession boundaries (2 jobs before
 * session 1, one more before session 2, one more before session 3) -- the
 * multi-session shape matters here specifically (unlike task 11.6's
 * single-session fixture) because a `step 1` / `continue` script needs more
 * than one session boundary to exercise both commands meaningfully.
 *
 * Capture mechanism: both runs set `sched_sim_args_t.output_path`
 * (--sim-output, task 11.5) directly -- unconditional on `interactive`
 * (confirmed by reading sim.c's sched_sim_driver_start(): the
 * `args->output_path != NULL` -> fopen() gate has no interactive check) --
 * so no `set output` REPL command is needed; both runs simply write their
 * own full `.replay`-format decision stream to their own file, exactly like
 * test_sim_equivalence.c's own file S. Since this fixture has THREE
 * ReplaySession boundaries (not task 11.6's one), decoding cannot reuse that
 * file's fixed-skip-count technique (N_MSGS_BEFORE_SESSION) -- this file's
 * own decode_all_sessions() instead scans the whole output file structurally
 * via sim.c's shared msg_root_has_shape() (ReplaySession's own distinct
 * (1,2) word/pointer-count shape, per that function's own header comment),
 * decoding every ReplaySession-shaped message in file order and skipping
 * every other shape -- robust regardless of exactly how many ReplayEvent
 * pass-through messages land between sessions on either side.
 *
 * Comparator: 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 below), the same technique
 * test_sim_equivalence.c's own replay_decision_content_eq() already
 * establishes -- duplicated here (this file family's own established
 * per-file-duplication convention; no shared test-utility header exists
 * across this directory) rather than imported.
 *
 * Script design ("step 1\ncontinue\n", exactly 2 lines, regardless of how
 * many sessions the fixture has): the pre-replay pause consumes "step 1"
 * (mode=STEP, budget=1) -> session 1 runs -> post-session pause (budget
 * reaches 0) consumes "continue" (mode=CONTINUE) -> every remaining session
 * (2 and 3 here) runs to EOF with no further stdin read at all
 * (repl_should_pause() returns false unconditionally in CONTINUE mode,
 * repl.c). This exercises BOTH `step` and `continue` (task 12.4's own
 * literal text: "using step/continue to completion") while staying safe
 * against ever exhausting the script file and hitting readline()'s NULL/EOF
 * path (which calls repl_cmd_quit() -> exit(0) and would kill this whole
 * in-process test binary, not just the one test) -- CONTINUE mode, once
 * entered, never reads stdin again no matter how many sessions remain.
 *
 * Why the RED-phase probe targets xce_repl_on_event(), not
 * repl_should_pause(): task 12.1's own already-verified RED-phase probe
 * (tasks.md) defeated repl_should_pause() (`return false;` instead of the
 * real budget check) to prove `step N` pauses at the right session. That
 * probe changes ONLY *when* stdin is read/control is regained -- it does
 * NOT change *what* gets processed or in what order, so it would NOT turn
 * THIS test's final decision-sequence comparison red (a genuinely
 * reassuring, independently-confirmed fact about this architecture, not a
 * gap in this test): every session still eventually runs to EOF regardless
 * of pause timing, and this file's script already tolerates any pause
 * schedule via CONTINUE's sticky non-pausing behavior. The correct
 * synthetic probe for THIS specific invariant ("never alters ... what
 * events are processed") has to touch a hook that can affect *content*, and
 * only under `sim->interactive`: xce_repl_on_event(xce_sched_sim_driver_t*,
 * xce_core_event_t* ev) (repl_break.c) takes a non-const `ev` and is called
 * (sim_loop.c, via repl.h's xce_repl_check_event() wrapper, gated on
 * `sim->interactive`) strictly BEFORE the event is appended to the
 * accumulated batch that xce_core_push_events() later applies -- so a
 * one-line mutation there (corrupting a JOBS_PUSH job's jobid, batch mode
 * never calls this hook at all) is a real, load-bearing content divergence
 * between the REPL run and the batch run, exactly the failure mode task
 * 12.4 exists to catch. See this task's own tasks.md `_Verified:` entry for
 * the actual probe text and observed failing test output.
 *
 * White-box #include set: identical to test_repl_execution_control.c's own
 * (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, repl_break.c, repl_info.c, repl_params.c, sim.c, sim_loop.c,
 * sim_record.c, sim_verify.c, snapshot.c, snapshot_load.c, snapshot_wire.c)
 * -- no driver.c/recording_events.c: unlike test_sim_equivalence.c, this
 * file never drives a real normal-mode recording session; both runs are
 * pure simulation-driver replays of one hand-built fixture file, so it only
 * needs the sim+repl stack, not the normal-mode driver stack.
 */
#include <errno.h>
#include <limits.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>

#include <unity/unity.h>

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

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention as test_repl_execution_control.c/test_sim_equivalence.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_fmt.c"
#include "../src/repl_rollup.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_limit.c" /* bug-1825: sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin() (sim_loop.c-cap sibling, sim_picker.c's own bug-1567 precedent) 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; }

/* repl_info.c (scheduler-sim-diagnostics spec, task 2.1) now calls
 * limit_get_configured_limits() (plugins/limit.c, not part of this TU) to
 * resolve `info limit`'s name/all argument -- link-satisfying stub only,
 * same rationale as kill_sched_job()/kill_sched_resval() above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_configured_limits()'s own documented "plugin absent" return)
 * is also the behaviorally correct answer here, not just a link
 * placeholder. repl_info_limit() itself is not yet wired into
 * repl_cmd_info() (task 4), so no test in this file can reach it anyway. */
var limit_get_configured_limits(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 3.1: repl_info.c now also calls
 * limit_get_usage_instances() (plugins/limit.c, not part of this TU) to
 * read a limit's live usage instances -- identical link-satisfying stub,
 * same rationale as limit_get_configured_limits() immediately above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_usage_instances()'s own documented "plugin absent" return) is
 * also the behaviorally correct answer here, not just a link placeholder. */
var limit_get_usage_instances(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 6 (bug-1825): sim_loop.c's
 * sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin()
 * (sim_limit.c, pulled into this TU via the #include above), which in
 * turn calls limit_new_plugin() (plugins/limit.c, not part of this TU) --
 * identical link-satisfying stub rationale as limit_get_configured_
 * limits()/limit_get_usage_instances() immediately above. Returning NULL
 * here is also the behaviorally correct answer, not just a link
 * placeholder: sim_bootstrap_limit_plugin() already handles a NULL
 * limit_new_plugin() return gracefully (logs an ERROR, registers nothing,
 * returns) -- the exact same "out of memory" path sim_bootstrap_picker()
 * (sim_picker.c) takes for a NULL builtin_fcfs_new_plugin() -- so this
 * file's fixtures simply end up with no limit plugin registered, same
 * observable state as before this task. */
xce_sched_plugin_t* limit_new_plugin(void)
{
    return NULL;
}

/* limit_seed_config() (plugins/limit.c, not part of this TU) -- bug-1825
 * remediation round 2's new honestly-typed config-seed entry point.
 * sim_bootstrap_limit_plugin() (sim_limit.c) only calls it after
 * plugin_register_core() has already succeeded against a real, non-NULL
 * limit_new_plugin() plugin -- unreachable in this file since the
 * limit_new_plugin() stub immediately above always returns NULL, but the
 * symbol must still resolve at link time. Same link-satisfying-stub
 * rationale as limit_new_plugin() immediately above. */
int limit_seed_config(xce_sched_plugin_t* plugin, xce_sched_core_t* core,
                      xce_conf4sched_t* config)
{
    (void)plugin;
    (void)core;
    (void)config;
    return 0;
}

/* resource-accounting.md's reconciliation (resource.c, not linked here --
 * this file only #includes core_events.c). core_events.c's
 * XCE_CORE_EVENT_JOB_RUSAGE handler calls this whenever job->accounts !=
 * NULL; link-satisfying stub only, same rationale as every other stub in
 * this block (see test_job_update.c's identical stub). */
int sched_reconcile_job_resource_usage(xce_sched_core_t*  c,
                                       xce_job_account_t* a,
                                       xce_job_rusages_t* r)
{
    (void)c; (void)a; (void)r; return 0;
}

void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

int sched_fill_limit_requests(xce_sched_alloc_req_t* alloc_req,
                              xlist_t* limits, xlist_t* rusages, int nhosts,
                              int nslots)
{
    (void)alloc_req;
    (void)limits;
    (void)rusages;
    (void)nhosts;
    (void)nslots;
    return 0;
}

/* sched_sim_driver_run()'s (sim_loop.c, scheduler-sim-decision-parity task
 * 2.1) new round-boundary call to the relocated sched_account_jobs()
 * (core.h) -- a no-op stand-in here since this file does not link
 * scheduler.c/resource.c, same rationale test_sim_equivalence.c's own copy
 * documents. */
int sched_account_jobs(xce_sched_core_t* core)
{
    (void)core;
    return 0;
}

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * copy every white-box test in this directory that does not link
 * schedobj.c already establishes (test_repl_execution_control.c,
 * test_sim_equivalence.c, etc.). */
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 -- same techniques test_repl_execution_control.c/       */
/* test_sim_equivalence.c already establish.                               */
/* ---------------------------------------------------------------------- */

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

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

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

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

/* Same as test_repl_execution_control.c's own helper, extended with a
 * `priority` parameter (this file's own fixture needs two queues at
 * different priorities, matching test_sim_equivalence.c's cross-queue
 * ordering fixture, not the single fixed priority=1 that file's own
 * single-queue fixture needed). */
static void append_queue_update_event(FILE* f, uint64_t ts,
                                      const char* queue_name,
                                      int32_t     priority)
{
    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 = priority;

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

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

static void append_replay_session(FILE* f, uint64_t begin_round)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplaySession_ptr    sp;

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

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

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

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

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

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

static void write_text_file(const char* path, const char* content)
{
    FILE* f = fopen(path, "w");

    TEST_ASSERT_NOT_NULL(f);
    TEST_ASSERT_TRUE(fputs(content, f) >= 0);
    fclose(f);
}

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

/*
 * decode_all_sessions - scans `path` end to end, decoding every
 * ReplaySession-shaped message (sim.c's own shared msg_root_has_shape(),
 * ReplaySession's distinct (1,2) word/pointer-count shape) into a growable
 * array, in file order, skipping every other message shape (ReplayHeader/
 * SnapshotEntry/ReplayEvent). Unlike test_sim_equivalence.c's
 * decode_session_after_skip() (a fixed skip count valid for exactly ONE
 * ReplaySession), this file's fixture has THREE session boundaries, so a
 * structural scan is used instead of a hardcoded message-index -- also
 * robust against neither side needing to hand-count how many ReplayEvent
 * pass-through messages precede each session (an implementation detail this
 * test correctly does not otherwise care about).
 */
static void decode_all_sessions(const char* path, replay_session_t*** out,
                                int* out_count)
{
    FILE*              f = fopen(path, "r");
    replay_session_t** sessions = NULL;
    int                count    = 0;

    TEST_ASSERT_NOT_NULL(f);

    for (;;) {
        struct capn c;
        capn_ptr    root;

        if (capn_init_fp(&c, f, 0) != 0) {
            break; /* EOF */
        }

        if (msg_root_has_shape(&c, &root, ReplaySession_word_count,
                               ReplaySession_pointer_count)) {
            ReplaySession_ptr  sp;
            replay_session_t*  decoded = NULL;
            replay_session_t** grown;

            sp.p = root;
            decode_ReplaySession_ptr(&decoded, sp);
            capn_free(&c);

            TEST_ASSERT_NOT_NULL(decoded);

            grown = (replay_session_t**)realloc(
                sessions, (size_t)(count + 1) * sizeof(*sessions));
            TEST_ASSERT_NOT_NULL(grown);
            sessions        = grown;
            sessions[count] = decoded;
            count++;
            continue;
        }

        capn_free(&c);
    }

    fclose(f);

    *out       = sessions;
    *out_count = count;
}

static void free_all_sessions(replay_session_t** sessions, int count)
{
    int i;

    for (i = 0; i < count; i++) {
        free_ReplaySession_ptr(&(sessions[i]));
    }
    free(sessions);
}

/*
 * replay_decision_content_eq - structural equality between two DECODED
 * ReplayDecision entries (batch's vs the REPL's own, at the same session
 * and position). Same technique test_sim_equivalence.c's own function of
 * the same name establishes (delegating to sim_verify.c's leaf comparators,
 * duplicated here per this file family's own no-shared-header convention --
 * see this file's header comment).
 */
static int replay_decision_content_eq(const replay_decision_t* a,
                                      const replay_decision_t* b)
{
    if ((a == NULL) || (b == NULL)) {
        return 0;
    }
    if (a->body_which != b->body_which) {
        return 0;
    }

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

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

    default:
        return 0;
    }
}

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

static void
test_repl_step_continue_to_completion_matches_batch_decision_sequence(void)
{
    char                   fixture_path[PATH_MAX];
    char                   script_dir[PATH_MAX];
    char                   script_path[PATH_MAX];
    char                   batch_out_path[PATH_MAX];
    char                   repl_out_path[PATH_MAX];
    int                    batch_fd;
    int                    repl_fd;
    FILE*                  f;
    xce_sched_sim_driver_t sim_batch;
    xce_sched_sim_driver_t sim_repl;
    sched_sim_args_t       args_batch;
    sched_sim_args_t       args_repl;
    replay_session_t**     batch_sessions = NULL;
    replay_session_t**     repl_sessions  = NULL;
    int                    n_batch        = 0;
    int                    n_repl         = 0;
    int                    total_batch_decisions;
    int                    saw_jobid_501;
    int                    rc;
    int                    i;
    int                    j;

    /* ---- Build ONE `.replay` fixture: two queues at different
     * priorities, four jobs split across three ReplaySession boundaries --
     * this file's own header comment explains why three sessions matter
     * here (task 11.6's own single-session fixture would not exercise both
     * `step` and `continue`). ---- */
    f = write_header_and_snapshot(fixture_path, sizeof(fixture_path),
                                  "_fixture");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "high", 5);
    append_queue_update_event(f, 1001ULL, "normal", 1);

    append_jobs_push_event(f, 1500ULL, 501ULL, "high");
    append_jobs_push_event(f, 1501ULL, 502ULL, "normal");
    append_replay_session(f, 2000ULL); /* session 1 */

    append_jobs_push_event(f, 2500ULL, 503ULL, "high");
    append_replay_session(f, 3000ULL); /* session 2 */

    append_jobs_push_event(f, 3500ULL, 504ULL, "normal");
    append_replay_session(f, 4000ULL); /* session 3 */

    append_end_of_stream_marker(f);
    fclose(f);

    /* ---- Run 1: batch mode (interactive=false), straight through, no
     * stdin involvement at all. ---- */
    snprintf(batch_out_path, sizeof(batch_out_path),
             "/tmp/xce_replequiv_batch_XXXXXX");
    batch_fd = mkstemp(batch_out_path);
    TEST_ASSERT_TRUE(batch_fd >= 0);
    close(batch_fd);

    memset(&sim_batch, 0, sizeof(sim_batch));
    memset(&args_batch, 0, sizeof(args_batch));
    args_batch.replay_path  = fixture_path;
    args_batch.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    args_batch.sim_speed    = 0; /* frozen -- deterministic, no wall-clock
                                  * skew, matching every other equivalence
                                  * fixture in this directory */
    args_batch.output_path  = batch_out_path;
    args_batch.interactive  = false;

    rc = sched_sim_driver_run(&sim_batch, &args_batch);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* ---- Run 2: the SAME fixture, driven via the REPL's step/continue
     * command loop to completion (see this file's header comment for why
     * exactly "step 1\ncontinue\n" is both sufficient and safe against ever
     * exhausting the script). ---- */
    snprintf(repl_out_path, sizeof(repl_out_path),
             "/tmp/xce_replequiv_repl_XXXXXX");
    repl_fd = mkstemp(repl_out_path);
    TEST_ASSERT_TRUE(repl_fd >= 0);
    close(repl_fd);

    make_tmp_dir(script_dir, sizeof(script_dir), "_script");
    snprintf(script_path, sizeof(script_path), "%s/script.txt", script_dir);
    /* task 12.5 (bug-1594): trailing `quit` -- `continue` now re-enters
     * the REPL at replay completion instead of returning, so this side
     * must run in a FORKED child (matching test_repl_execution_control.c's
     * own Test 3 pattern); otherwise quit's exit(0) would tear down this
     * whole test binary before the batch-vs-REPL comparison below ever
     * runs. */
    write_text_file(script_path, "step 1\ncontinue\nquit\n");

    memset(&args_repl, 0, sizeof(args_repl));
    args_repl.replay_path  = fixture_path;
    args_repl.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    args_repl.sim_speed    = 0;
    args_repl.output_path  = repl_out_path;
    args_repl.interactive  = true;

    {
        pid_t pid = fork();
        int   status;

        TEST_ASSERT_TRUE_MESSAGE(pid >= 0, "fork() must succeed");

        if (pid == 0) {
            if (freopen(script_path, "r", stdin) == NULL) {
                _exit(120);
            }
            memset(&sim_repl, 0, sizeof(sim_repl));
            (void)sched_sim_driver_run(&sim_repl, &args_repl);
            _exit(111); /* unreachable -- quit must already have exited */
        }

        TEST_ASSERT_EQUAL_INT_MESSAGE(pid, waitpid(pid, &status, 0),
                                      "waitpid must reap the REPL child");
        TEST_ASSERT_TRUE_MESSAGE(WIFEXITED(status),
                                 "the REPL child must exit normally via "
                                 "quit's exit(0), not crash/signal");
        TEST_ASSERT_EQUAL_INT_MESSAGE(0, WEXITSTATUS(status),
                                      "quit's exit(0) is the only correct "
                                      "exit path");
    }

    /* ---- Decode every ReplaySession from both output files and compare
     * the decision sequence, session by session, decision by decision. ---- */
    decode_all_sessions(batch_out_path, &batch_sessions, &n_batch);
    decode_all_sessions(repl_out_path, &repl_sessions, &n_repl);

    TEST_ASSERT_EQUAL_INT_MESSAGE(3, n_batch,
                                  "the fixture pushes exactly 3 "
                                  "ReplaySession boundaries");
    TEST_ASSERT_EQUAL_INT_MESSAGE(n_batch, n_repl,
                                  "REPL step/continue-to-completion must "
                                  "produce the SAME number of sessions as "
                                  "batch mode");

    /* Non-tautology guard #1: session 1 has EXACTLY 2 pending candidates
     * (jobs 501/502) and nothing else could possibly be pending yet -- an
     * unambiguous floor independent of whether a PEND'd job persists as a
     * candidate into later sessions. */
    TEST_ASSERT_EQUAL_INT(2, batch_sessions[0]->n_decisions);

    /* Non-tautology guard #2: independently pin real decision CONTENT (not
     * merely "both sides agree", which a shared bug in both runs could
     * satisfy too) -- job 501 must appear as a PEND decision somewhere in
     * session 1 on the batch side. */
    saw_jobid_501 = 0;
    for (j = 0; j < batch_sessions[0]->n_decisions; j++) {
        replay_decision_t* d = batch_sessions[0]->decisions[j];

        TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_pend,
                              (int)d->body_which);
        if ((d->body.pend != NULL) && (d->body.pend->jobid == 501ULL)) {
            saw_jobid_501 = 1;
        }
    }
    TEST_ASSERT_TRUE_MESSAGE(saw_jobid_501,
                             "session 1's batch-mode decisions must "
                             "genuinely include job 501");

    /* Non-tautology guard #3: at least 4 total decisions across the whole
     * batch run (2 from session 1's own guaranteed floor above, plus at
     * least 1 new candidate pushed before each of sessions 2 and 3) --
     * proves this is not a degenerate empty-everything comparison. */
    total_batch_decisions = 0;
    for (i = 0; i < n_batch; i++) {
        total_batch_decisions += batch_sessions[i]->n_decisions;
    }
    TEST_ASSERT_TRUE_MESSAGE(total_batch_decisions >= 4,
                             "the fixture must produce a genuinely "
                             "non-trivial decision sequence");

    /* ---- The central equivalence assertion: per session, per decision,
     * REPL step/continue-to-completion and batch mode must match exactly. ---- */
    for (i = 0; i < n_batch; i++) {
        TEST_ASSERT_EQUAL_INT_MESSAGE(
            batch_sessions[i]->n_decisions, repl_sessions[i]->n_decisions,
            "session decision COUNT must match between REPL and batch");

        for (j = 0; j < batch_sessions[i]->n_decisions; j++) {
            replay_decision_t* bd = batch_sessions[i]->decisions[j];
            replay_decision_t* rd = repl_sessions[i]->decisions[j];

            /* Independently pin the decision TYPE on each side (matches
             * test_sim_equivalence.c's own "pin the type, don't just diff
             * the two sides against each other" guard) -- every decision
             * this fixture can produce is PEND (sim_picker_thread_fn()'s
             * documented, out-of-scope DISPATCH gap). */
            TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_pend,
                                  (int)bd->body_which);
            TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_pend,
                                  (int)rd->body_which);

            TEST_ASSERT_TRUE_MESSAGE(
                replay_decision_content_eq(bd, rd),
                "REPL step/continue-to-completion must produce a "
                "byte-identical decision to batch mode at this session "
                "and position");
        }
    }

    free_all_sessions(batch_sessions, n_batch);
    free_all_sessions(repl_sessions, n_repl);

    unlink(batch_out_path);
    unlink(repl_out_path);

    printf("test_repl_step_continue_to_completion_matches_batch_decision_"
          "sequence: PASS\n");
}

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