myxce/src/scheduler/testsuite/test_sim_session_ordering.c

/*
 * test_sim_session_ordering.c - proves the simulation driver's round
 * boundary now upholds the same ordering discipline
 * test_driver_session_ordering.c already proves for the normal-mode driver
 * (scheduler-sim-decision-parity spec, task 2.1, Req 1.3/1.4, 5.1, 5.3;
 * "Sim Driver Round-Boundary Accounting" component, design.md).
 *
 * The property under test: sim_loop.c's sched_sim_driver_run() must call
 * the relocated sched_account_jobs() (core.h, task 1.1) once per replayed
 * round, strictly inside the round's session-active window -- i.e. after
 * xce_core_begin_session()'s resval reset, before xce_core_end_session()
 * closes the round -- mirroring driver.c's
 * xce_sched_driver_run_normal_mode_round() exactly. Before task 2.1,
 * sim_loop.c called the older monolithic xce_core_go() with zero
 * interleaving slot and never called sched_account_jobs() at all
 * (bug-1613, documented in .kiro/steering/scheduler-driver-equivalence.md)
 * -- every worker's/shared resource's resInUse stayed permanently zero
 * across a whole `--sim`/`--interactive` run regardless of what jobs were
 * actually running. This file's own central assertion is 0 spy calls
 * against the pre-fix code, 2 (one per replayed round) against the fixed
 * code -- captured as this task's RED/GREEN evidence.
 *
 * Why a spy, not the real sched_account_jobs() (resource.c/scheduler.c):
 * same rationale test_driver_session_ordering.c's own header comment gives
 * for its identical choice -- this file's target is the DRIVER's ordering
 * discipline (does it call sched_account_jobs() at all, and in the right
 * window), not whether sched_account_jobs() itself charges resources
 * correctly (task 1.1's own scope, proven by test_resource_accounting.c).
 *
 * Why no cross-thread wbuf-rendezvous proof (unlike
 * test_driver_session_ordering.c's own elaborate three-thread design): that
 * proof technique requires substituting a test-controlled picker plugin for
 * the one the driver under test resolves. sim_loop.c's
 * sched_sim_driver_run() -- unlike driver.c's decoupled, single-round
 * xce_sched_driver_run_normal_mode_round() -- is one monolithic entry point
 * that spawns its own picker thread (sim_picker_thread_fn(), sim_loop.c)
 * internally and unconditionally bootstraps the real builtin_fcfs plugin
 * (sim_bootstrap_picker(), sim_picker.c) as part of the same call; neither
 * is independently substitutable from a test without modifying production
 * code outside this task's boundary (sim_picker.c is not part of the "Sim
 * Driver Round-Boundary Accounting" component this task owns). This is not
 * a gap in rigor: begin_session()/[reconciliation slot]/account_jobs()/
 * signal_go() are four single-threaded, sequential, unconditional calls in
 * sim_loop.c's own source (no branch reorders them), so "does
 * account_jobs() run before signal_go()" is a plain, deterministic
 * same-thread program-order fact, not a race to be proven at runtime --
 * only "does it actually run, and does it see session_active==1 (i.e.
 * strictly after begin_session, strictly before end_session)" is the
 * genuinely runtime-observable part, which this file's spy does prove.
 * cbuf's own capacity-1 blocking rendezvous (core_session.c, already
 * proven by test_core_session.c) is what guarantees the picker thread
 * cannot observe xce_core_signal_go()'s wbuf signal any earlier than that
 * call actually executes.
 *
 * Error-path preservation (Requirement: "preserve the existing
 * error-handling behavior for a failed session start exactly as it works
 * today"): NOT exercised by a runtime reproduction in this file. Analysis
 * (see task's own status report): xce_core_begin_session() only fails when
 * a session is already active, and sched_sim_driver_run()'s single
 * replay-reading thread always completes a round's xce_core_end_session()
 * (via the picker thread's guaranteed end-of-session NULL decision, drained
 * before end_session runs) before its own next loop iteration could call
 * begin_session() again -- this was equally true, and equally unreachable
 * through the legitimate single-consumer replay loop, before this task's
 * change (xce_core_go() had the identical failure condition). This task's
 * diff preserves the exact same cleanup sequence (free_ReplaySession_ptr(),
 * free(record_ctx), continue) the prior xce_core_go() branch used, just
 * keyed off xce_core_begin_session()'s own rc instead -- verified by code
 * review, not a synthetic runtime trigger disproportionate to a
 * structurally unreachable branch.
 *
 * Fixture-construction technique and white-box #include/stub set: IDENTICAL
 * to test_sim_equivalence.c's own (real normal-mode recording via
 * sched_record_start()/xce_sched_driver_run_normal_mode_round() driving the
 * REAL builtin_fcfs picker + a stand-in allocator that unconditionally
 * PENDs, then replayed via the REAL, unmodified sched_sim_driver_run()) --
 * see that file's own header comment for the full rationale on why every
 * decision here is PEND (sim_loop.c's live picker thread cannot emit a
 * DISPATCH). This file records TWO separate rounds (not test_sim_equivalence.c's
 * one) specifically to prove sched_account_jobs() is called ONCE PER ROUND,
 * not merely once ever -- pinning Requirement 1.3 ("When a scheduling round
 * completes ... refresh ... not only jobs whose resource consumption was
 * already reflected in an earlier round").
 */
#include <errno.h>
#include <limits.h>
#include <pthread.h>
#include <stdatomic.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 combined set test_sim_equivalence.c already establishes). NOTE:
 * ../src/plugin.c is deliberately NOT included here (unlike
 * test_driver_session_ordering.c, which needs the real plugin_begin_session()/
 * plugin_end_session() driven through a genuinely-registered observer
 * plugin) -- this file spies plugin_begin_session()/plugin_end_session()
 * instead (see "PLUGIN SESSION-HOOK SPY" section below), same rationale as
 * its pre-existing sched_account_jobs() spy: this file's target is sim_loop.c's
 * round-boundary CALL-SITE discipline (bug-1681), already independently
 * proven to drive real per-plugin hook firing by test_driver_session_ordering.c
 * for the normal-mode side; re-proving hook-firing here would be redundant
 * AND is not practically constructible here anyway -- sim.core.plugins does
 * not exist until sched_sim_driver_run()'s own internal
 * sched_sim_driver_start()/sim_bootstrap_picker() calls run, with no hook
 * point exposed for a test to register an observer plugin in between.
 * plugin_register_core() (needed for real, both by this file's own
 * fixture_init() and transitively by sim_picker.c's sim_bootstrap_picker())
 * is reimplemented verbatim below -- same "real small reimplementation
 * instead of a whole extra file" convention this file's own
 * sched_resolve_job_picker() copy already establishes. plugin_register()/
 * plugin_unregister() are trivial link-satisfying no-op stubs: plugin_register()
 * is referenced only by plugins/builtin_fcfs.c's register_builtin_fcfs()
 * (this fixture never calls that entry point, using plugin_register_core()
 * directly instead, exactly as sim_bootstrap_picker() does in production);
 * plugin_unregister() is referenced only as core.c's plugins-kvmgr
 * datakiller (never invoked -- this fixture is never torn down, same
 * rationale test_driver_session_ordering.c's own file header documents for
 * not needing plugin_unregister() to do anything real either). */
#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/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_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() 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; }

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

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

/* 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; this fixture never populates job->accounts, so the real body is
 * never reached -- 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;
}

/* plugin_prepare_allocation()'s (plugin.c) linker dependency -- never
 * invoked (this fixture never drives real resource-matching allocation --
 * same rationale test_sim_equivalence.c's own copy documents). */
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;
}

/* xce_sched_driver_on_decision()'s (driver.c) publish-side linker
 * dependencies -- never actually invoked (this test never registers a
 * decision callback). Same rationale test_sim_equivalence.c's own copy
 * documents. */
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). */
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
 * copy every white-box test in this directory that does not link
 * schedobj.c already establishes. Shared by driver.c's normal-mode round
 * (recording phase, below) AND sim_loop.c's sim_picker_thread_fn() (replay
 * phase, below) -- one real definition, both call sites in this combined TU
 * resolve 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;
}

/* ---------------------------------------------------------------------- */
/* sched_account_jobs() SPY -- this file's own central instrument. Records  */
/* one call count PLUS, per call (up to MAX_SPY_CALLS), whether             */
/* core->session_active was 1 at the moment of the call -- proving each     */
/* invocation happened strictly inside a round's begin_session/end_session  */
/* window (see this file's own header comment for why this, rather than a   */
/* cross-thread rendezvous, is the right-strength proof here).              */
/* ---------------------------------------------------------------------- */

#define MAX_SPY_CALLS 8

static _Atomic int g_spy_calls;
static int         g_spy_session_active_ok[MAX_SPY_CALLS];

int sched_account_jobs(xce_sched_core_t* core)
{
    int idx = atomic_fetch_add(&g_spy_calls, 1);

    if ((core != NULL) && (idx < MAX_SPY_CALLS)) {
        g_spy_session_active_ok[idx]
            = (atomic_load_explicit(&(core->session_active),
                                    memory_order_acquire)
               == 1)
                ? 1
                : 0;
    }

    return 0;
}

static void reset_spy(void)
{
    atomic_store(&g_spy_calls, 0);
    memset(g_spy_session_active_ok, 0, sizeof(g_spy_session_active_ok));
}

/* ---------------------------------------------------------------------- */
/* plugin_register_core() -- REAL reimplementation (verbatim logic from    */
/* plugin.c), needed because this file does not #include ../src/plugin.c   */
/* (see the #include block's own comment above for why). Used for real by  */
/* this file's own fixture_init() (recording-side registration) AND        */
/* transitively by sim_picker.c's sim_bootstrap_picker() (replay-side).    */
/* plugin_register()/plugin_unregister() are trivial link-satisfying no-op */
/* stubs -- neither is ever actually invoked by this file's fixtures (see  */
/* #include block comment).                                                */
/* ---------------------------------------------------------------------- */

int plugin_register_core(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
    int rc = 0;

    if ((core == NULL) || (plugin == NULL)) {
        return -EINVAL;
    }

    if (plugin->init_fn != NULL) {
        rc = plugin->init_fn(plugin);
        if (rc < 0) {
            return rc;
        }
    }

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

    if (rc >= 0) {
        plugin->plugin_id = rc;
    }

    return rc;
}

int plugin_register(void* arg1, void* arg2)
{
    (void)arg1;
    (void)arg2;
    return 0;
}

void plugin_unregister(void* arg) { (void)arg; }

/* ---------------------------------------------------------------------- */
/* plugin_begin_session()/plugin_end_session() SPY (bug-1681) -- same      */
/* call-count + session_active-window proof technique as the               */
/* sched_account_jobs() spy above; this file's target is sim_loop.c's      */
/* CALL-SITE discipline, not per-plugin hook-firing semantics (already     */
/* proven for the normal-mode side by test_driver_session_ordering.c's own */
/* real-observer-plugin design -- see #include block comment for why that  */
/* design is not practically constructible against sim_loop.c).            */
/* ---------------------------------------------------------------------- */

static _Atomic int g_begin_session_spy_calls;
static _Atomic int g_end_session_spy_calls;
static int         g_begin_session_spy_active_ok[MAX_SPY_CALLS];
static int         g_end_session_spy_active_ok[MAX_SPY_CALLS];

int plugin_begin_session(xce_sched_core_t* core)
{
    int idx = atomic_fetch_add(&g_begin_session_spy_calls, 1);

    if ((core != NULL) && (idx < MAX_SPY_CALLS)) {
        g_begin_session_spy_active_ok[idx]
            = (atomic_load_explicit(&(core->session_active),
                                    memory_order_acquire)
               == 1)
                ? 1
                : 0;
    }

    return 0;
}

int plugin_end_session(xce_sched_core_t* core)
{
    int idx = atomic_fetch_add(&g_end_session_spy_calls, 1);

    if ((core != NULL) && (idx < MAX_SPY_CALLS)) {
        g_end_session_spy_active_ok[idx]
            = (atomic_load_explicit(&(core->session_active),
                                    memory_order_acquire)
               == 1)
                ? 1
                : 0;
    }

    return 0;
}

static void reset_plugin_session_spies(void)
{
    atomic_store(&g_begin_session_spy_calls, 0);
    atomic_store(&g_end_session_spy_calls, 0);
    memset(g_begin_session_spy_active_ok, 0,
          sizeof(g_begin_session_spy_active_ok));
    memset(g_end_session_spy_active_ok, 0,
          sizeof(g_end_session_spy_active_ok));
}

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

/* ---------------------------------------------------------------------- */
/* Fixture helpers -- identical to test_sim_equivalence.c's own.           */
/* ---------------------------------------------------------------------- */

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

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

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

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

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("ordering_proj");
    j->submit->resreq       = strdup("type == any");
    j->submit->mintasks     = 1;
    j->submit->maxtasks     = 1;

    return j;
}

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 (recording phase) -- identical to   */
/* test_sim_equivalence.c's own.                                            */
/* ---------------------------------------------------------------------- */

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

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

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

/* ---------------------------------------------------------------------- */
/* The contract test.                                                       */
/* ---------------------------------------------------------------------- */

static void
test_sim_replay_calls_account_jobs_once_per_round_within_session_window(
    void)
{
    char                   record_dir[PATH_MAX];
    xce_scheduler_t        sched;
    char                   r_path[PATH_MAX];
    xce_job_t**            jobs1;
    xce_job_t**            jobs2;
    sched_sim_args_t       sim_args;
    xce_sched_sim_driver_t sim;
    int                    rc;
    int                    i;

    /* ---- Record TWO real normal-mode rounds into one .replay file ---- */
    make_tmp_dir(record_dir, sizeof(record_dir), "_r");
    fixture_init(&sched, record_dir);

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

    enqueue_queue_update(&sched, "normal", 1);

    jobs1    = (xce_job_t**)calloc(1, sizeof(xce_job_t*));
    jobs1[0] = make_job(9201ULL, 1000ULL, "normal", "alice");
    enqueue_jobs_push(&sched, jobs1, 1);

    run_one_real_session(&sched); /* round 1 */

    jobs2    = (xce_job_t**)calloc(1, sizeof(xce_job_t*));
    jobs2[0] = make_job(9202ULL, 2000ULL, "normal", "bob");
    enqueue_jobs_push(&sched, jobs2, 1);

    run_one_real_session(&sched); /* round 2 */

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

    /* Reset the spy AFTER recording -- driver.c's own normal-mode round
     * boundary (task 1.1) already calls the real sched_account_jobs() once
     * per round during the recording above (already proven correct by
     * test_driver_session_ordering.c); this test's own target is
     * sim_loop.c's call site (task 2.1), isolated by resetting here. */
    reset_spy();
    reset_plugin_session_spies();

    /* ---- Replay the SAME file through the REAL, unmodified simulation
     * driver ---- */
    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 */

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

    /* THE central regression proof (task 2.1's own scope, Req 1.3/1.4):
     * before this task, this count is 0 (sim_loop.c never called
     * sched_account_jobs() at all, bug-1613) -- after, it is 2 per-round
     * calls plus ONE load-time call (task 4.6:
     * xce_sched_core_load_snapshot(), called from sched_sim_driver_start()
     * at the very top of sched_sim_driver_run(), now also charges
     * sched_account_jobs() once for whatever RUN-state jobs the loaded
     * snapshot already has -- see snapshot_load.c's own doc comment on
     * that call). Total 3, not merely once ever and not merely once per
     * round. */
    TEST_ASSERT_EQUAL_INT(3, atomic_load(&g_spy_calls));

    /* Index 0 is task 4.6's load-time call -- by design it runs BEFORE any
     * session begins (core->session_active == 0 at that point is correct,
     * not a violation of the round-boundary contract below, which only
     * governs calls made ONCE A ROUND IS UNDERWAY). */
    TEST_ASSERT_FALSE_MESSAGE(
        g_spy_session_active_ok[0],
        "the load-time sched_account_jobs() call (task 4.6) runs before "
        "any session begins -- core->session_active must be 0 here, not "
        "the round-boundary contract's session_active==1");

    for (i = 1; i < 3; i++) {
        TEST_ASSERT_TRUE_MESSAGE(
            g_spy_session_active_ok[i],
            "sched_account_jobs() must run with core->session_active==1 -- "
            "i.e. strictly after xce_core_begin_session(), strictly before "
            "xce_core_end_session(), matching driver.c's own ordering "
            "contract");
    }

    /* bug-1681's own central regression proof: before the fix, both counts
     * are 0 (sim_loop.c never called plugin_begin_session()/
     * plugin_end_session() at all) -- after, each is 2, one per replayed
     * round. */
    TEST_ASSERT_EQUAL_INT(2, atomic_load(&g_begin_session_spy_calls));
    TEST_ASSERT_EQUAL_INT(2, atomic_load(&g_end_session_spy_calls));

    for (i = 0; i < 2; i++) {
        TEST_ASSERT_TRUE_MESSAGE(
            g_begin_session_spy_active_ok[i],
            "plugin_begin_session() must run with core->session_active==1, "
            "matching driver.c's own ordering contract (bug-1681)");
        TEST_ASSERT_TRUE_MESSAGE(
            g_end_session_spy_active_ok[i],
            "plugin_end_session() must run with core->session_active==1 "
            "(strictly before xce_core_end_session() clears it), matching "
            "driver.c's own ordering contract (bug-1681)");
    }

    printf("test_sim_replay_calls_account_jobs_once_per_round_within_"
          "session_window: PASS\n");
}

/*
 * test_sim_replay_still_completes_interactive_and_batch_identically -
 * Requirement 5.1: interactive and batch simulation runs must go through
 * the SAME round-boundary call site with no divergence. sched_sim_driver_run()
 * is the one entry point BOTH `xce_scheduler -s` (batch) and `-i -s`
 * (interactive) call -- gated only by sim->interactive inside REPL hook
 * calls this file's own fixture never triggers (scheduler/CLAUDE.md: "drives
 * the exact same sched_sim_driver_run() as batch --sim, gated only by
 * sim->interactive"). This test simply confirms the round-boundary spy
 * fires identically whether sim->interactive is left at its default 0
 * (batch) -- there is no separate call site to diverge from; this is a
 * structural guarantee proven by the single call site itself, re-asserted
 * here as an explicit regression pin rather than left implicit.
 */
static void test_sim_replay_batch_mode_uses_the_same_round_boundary(void)
{
    char                   record_dir[PATH_MAX];
    xce_scheduler_t        sched;
    char                   r_path[PATH_MAX];
    xce_job_t**            jobs1;
    sched_sim_args_t       sim_args;
    xce_sched_sim_driver_t sim;
    int                    rc;

    make_tmp_dir(record_dir, sizeof(record_dir), "_b");
    fixture_init(&sched, record_dir);

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

    enqueue_queue_update(&sched, "normal", 1);

    jobs1    = (xce_job_t**)calloc(1, sizeof(xce_job_t*));
    jobs1[0] = make_job(9301ULL, 1000ULL, "normal", "carol");
    enqueue_jobs_push(&sched, jobs1, 1);

    run_one_real_session(&sched);

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

    reset_spy();
    reset_plugin_session_spies();

    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;

    memset(&sim, 0, sizeof(sim));
    TEST_ASSERT_EQUAL_INT(0, sim.interactive); /* batch: default-zero, never set */

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

    /* 2, not 1: task 4.6's load-time call (index 0, session_active==0 by
     * design -- see the two-round test's own comment above) plus this
     * batch run's one replayed round (index 1, session_active==1). */
    TEST_ASSERT_EQUAL_INT(2, atomic_load(&g_spy_calls));
    TEST_ASSERT_FALSE(g_spy_session_active_ok[0]);
    TEST_ASSERT_TRUE(g_spy_session_active_ok[1]);

    TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_begin_session_spy_calls));
    TEST_ASSERT_TRUE(g_begin_session_spy_active_ok[0]);
    TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_end_session_spy_calls));
    TEST_ASSERT_TRUE(g_end_session_spy_active_ok[0]);

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

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