myxce/src/scheduler/testsuite/test_sim_snapshot_reconcile_gap.c

/*
 * test_sim_snapshot_reconcile_gap.c - Regression coverage for scheduler-
 * sim-decision-parity task 4.2 (Req 4.1, 4.2): a replay starting from a
 * snapshot that already has a job in RUN state, with NO recorded decision
 * anywhere in this replay's own window, must degrade gracefully -- report
 * that job's usage as unavailable (job->accounts stays NULL, nothing
 * fabricated), never crash, and continue processing the rest of the replay
 * normally for every other job.
 *
 * This is a DIFFERENT trigger shape from test_sim_verify.c's own
 * test_verify_with_already_resolved_job_reports_no_mismatch (task 3, Req
 * 5.2): that test's already-resolved job (A) transitions PEND->RUN via a
 * JOB_STATE_CHANGE event WITHIN this replay's own recorded window, with a
 * matching recorded DISPATCH decision in the SAME round -- reconciliation
 * finds a match and rebuilds job->allocs/job->accounts from it. This file's
 * job is already RUN *in the initial snapshot itself* (the state a job is
 * in when it was dispatched before this replay's own recording ever started
 * capturing decisions) -- design.md's Implementation Notes calls this out
 * explicitly: "xce_sched_core_load_snapshot() restores job->state from
 * snapshot data but never touches job->accounts/job->allocs ... this is
 * unambiguously Requirement 4's graceful-degradation case." No recorded
 * DISPATCH for this job exists ANYWHERE in the replay stream (there is
 * nothing to reconstruct from), so sim_reconcile_unaccounted_jobs()'s
 * per-round search must run to exhaustion and take the "not found" branch
 * (task 2.2) every round, forever -- not treated as an error, not
 * fabricating an allocation, and not blocking a second, ordinary job in the
 * same replay from being decided normally.
 *
 * task 2.2's own unit-level test (test_sim_reconcile_unaccounted.c) already
 * proves the underlying "not found" branch directly, calling
 * sim_reconcile_unaccounted_jobs() by hand against a synthetic core/session
 * pair. This file adds the integration-level proof that same branch was
 * missing until now: a job restored through a REAL
 * xce_sched_core_load_snapshot() call (via the real `.replay` header+
 * snapshot prefix sched_record_start() writes, the same production path
 * every operator-facing replay uses), reaching the round boundary through
 * the real, unmodified sched_sim_driver_run(), with a genuinely empty
 * search (the only recorded decision in the whole file belongs to a
 * different, ordinary job) -- not a hand-built xce_sched_core_t the unit
 * test constructs directly. It also confirms (via builtin_fcfs.c's own
 * JOB_IS_PEND(job->state) guard in on_job_push, unchanged by this feature)
 * that the RUN-state job restored from the snapshot never becomes a live
 * picker candidate either, so the only path that can ever give it visible
 * usage is reconciliation -- exactly the gap this feature closes.
 *
 * Fixture construction: build_fixture_scheduler_with_run_job() pushes a
 * QUEUE_UPDATE + JOBS_PUSH (job 901, JOB_STAT_RUN -- core_build_job()
 * (core_events.c) copies job->state verbatim from the pushed wire job; this
 * is NOT a JOB_STATE_CHANGE transition) directly onto the fixture
 * xce_scheduler_t's own core, via the real xce_core_push_events(), BEFORE
 * calling sched_record_start() -- so the snapshot sched_record_start()
 * writes (recording.c's own write_snapshot_entry(), unchanged) captures job
 * 901 already in RUN state, with no job->accounts/allocs (core_build_job()
 * only sets job->accounts when the pushed wire job's own `allocs` field is
 * non-NULL, which it never is here; snapshot.c's own header comment
 * separately confirms save/load never captures job->allocs/accounts either
 * way). No picker plugin is registered on this throwaway fixture core (same
 * "no picker needs to be key_add'ed" convention test_sim_verify.c's own
 * build_fixture_scheduler() documents) -- xce_core_push_events()'s
 * sched_resolve_job_picker() failure just WARNs and continues with a NULL
 * picker (core_events.c), so job 901 is still built and added to core->jobs
 * correctly.
 *
 * White-box #include set: identical to test_sim_verify.c's own (see that
 * file's header comment for the full rationale of each entry).
 */
#include <errno.h>
#include <limits.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <unity/unity.h>

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

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention/set as test_sim_verify.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() 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;
}

/* core_reset_workers()'s (core_session.c) linker dependency -- same stub
 * test_sim_verify.c already establishes; 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's own scenario keeps job->accounts NULL by design
 * (see this file's own header comment), 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, the "limit" plugin category)
 * linker dependency -- never invoked (this file never drives real
 * allocation). */
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) 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_verify.c's own copy
 * documents. This file's Observable does not depend on resInUse/
 * sharedInUse totals at all -- only on job->accounts staying NULL and the
 * replay completing cleanly -- so a no-op is sufficient here. */
int sched_account_jobs(xce_sched_core_t* core)
{
    (void)core;
    return 0;
}

/* bug-1836 (scheduler-driver-equivalence fix, Recorded-Decision
 * Reconciliation capacity gate): link-satisfying stubs, mirroring this
 * file's existing sched_account_jobs() no-op precedent immediately above --
 * sim_loop.c's sim_reconcile_unaccounted_jobs() now also calls
 * sched_check_host_satisfy()/sched_check_shared_satisfy() (resource.c) to
 * re-validate a recorded decision against live capacity, and
 * sched_charge_account() (resource.c) to keep that check's own view of
 * capacity up to date across multiple candidates in the same round. This
 * file never exercises real resource contention (same rationale as the
 * existing sched_account_jobs()/sched_ctx_get_uncomit_host() no-ops), so
 * the satisfy checks trivially always succeed and the charge call is a
 * no-op. */
int sched_check_host_satisfy(xce_sched_core_t* core, xce_sched_worker_t* w,
                             xvec_t* uncomitted, xvec_idval_t* wanted,
                             xvec_idval_t* unmeet)
{
    (void)core;
    (void)w;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_check_shared_satisfy(xce_sched_core_t* core, xvec_t* uncomitted,
                               xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
    (void)core;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_charge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    return 0;
}

int sched_uncharge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    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. */
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;
}

/* scheduler-sim-live-allocation spec, task 4.1: sim_loop.c's real
 * sched_sim_driver_run() (white-box #include'd into this TU above) now
 * spawns sched_job_picker()/sched_allocator() (sched_election.c) directly
 * instead of its own, now-removed sim_picker_thread_fn() -- an undefined
 * reference at link time in this white-box TU otherwise, since
 * sched_election.c's own sched_allocator() calls the REAL
 * sched_make_allocation()/sched_commit_decision() (scheduler.c), neither
 * part of this file's own hand-picked white-box #include set (this file's
 * own header comment lists that set). This file's own test(s) do not
 * exercise genuine resource-matching allocation, so this link-satisfying
 * pair faithfully reproduces the exact PEND-only contract the REMOVED
 * sim_picker_thread_fn() itself used to provide -- split across the SAME
 * two-thread jbuf hand-off the real production pair now uses (one
 * producer, one consumer, matching the project's jbuf capacity-1
 * rendezvous invariant) instead of the old single fused thread -- so this
 * file's own existing assertions are unaffected by this task's production
 * change. Genuine resource-based dispatch through the real, unmodified
 * sched_job_picker()/sched_allocator() pair is proved separately by
 * test_sim_live_allocation.c (task 4.1's own dedicated fixture, which
 * links the real scheduler_srcs instead of white-boxing individual files,
 * precisely so it can reach the real allocation chain this stub
 * deliberately does not). */
void* sched_job_picker(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

    for (;;) {
        int      go    = 0;
        uint64_t jobid = 0;

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

        if (sched_resolve_job_picker(core, &picker) < 0) {
            jobid = 0;
            cbuf_put(&(core->jbuf), &jobid);
            continue;
        }

        picker->picker_iface->begin_session(picker);

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

            if (job == NULL) {
                break;
            }

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

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

        picker->picker_iface->end_session(picker);
    }

    return NULL;
}

void* sched_allocator(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

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

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

        job = sched_get_job(core, jobid);
        if (job == NULL) {
            continue;
        }

        (void)sched_resolve_job_picker(core, &picker);
        if (picker != NULL) {
            picker->picker_iface->on_pend(picker, job->jobid);
        }

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

    return NULL;
}

/* scheduler-sim-live-allocation spec, task 4.1 (bug-1829 fix): sim_loop.c's
 * real sched_sim_driver_run() (white-box #include'd into this TU above) now
 * also calls sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_
 * allocator() (sim_alloc.c) -- undefined references at link time in this
 * white-box TU otherwise, since neither symbol is part of this file's own
 * hand-picked white-box #include set. This file's own test(s) exercise the
 * link-satisfying sched_job_picker()/sched_allocator() stub pair above
 * (which never reaches real primary allocation), so a true no-op is the
 * correct, behavior-preserving stand-in here -- the real bootstrap logic is
 * exercised separately by test_sim_live_allocation.c (task 4.1's own
 * dedicated fixture, which links the real scheduler_srcs). */
void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

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_simsnapgap%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
}

/* Same xce_job_t wire-job builder test_snapshot_save_load.c's own
 * make_job() establishes -- reused here (this file's own no-shared-header
 * convention, matching every other file in this directory) but with an
 * EXPLICIT state parameter, not the JOB_STAT_PEND every other fixture in
 * this directory hardcodes. */
static xce_job_t* make_job_with_state(uint64_t jobid, uint32_t state,
                                      const char* queue)
{
    xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));

    TEST_ASSERT_NOT_NULL(j);
    j->jobid      = jobid;
    j->state      = state;
    j->sub_time   = 1000;
    j->queue_time = 1000;
    j->start_time = (state == JOB_STAT_RUN) ? 1000 : 0;
    j->submit     = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    TEST_ASSERT_NOT_NULL(j->submit);
    j->submit->queue    = (char*)queue;
    j->submit->user     = (char*)"alice";
    j->submit->mintasks = 1;
    j->submit->maxtasks = 1;

    return j;
}

/*
 * build_fixture_scheduler_with_run_job - test_sim_verify.c's own
 * build_fixture_scheduler() (throwaway xce_scheduler_t: xce_sched_core_init(),
 * job_pick_plugin set to the NAME "builtin_fcfs" but no plugin object ever
 * key_add'ed -- Req 5 AC 21, xce_sched_core_save_snapshot() does not
 * require a resolvable picker), PLUS one queue and one job already in RUN
 * state pushed directly onto `sched->core` via the real
 * xce_core_push_events() -- BEFORE the caller's own sched_record_start()
 * call snapshots it. This is what makes the snapshot sched_record_start()
 * writes contain a RUN-state job with job->accounts == NULL from the very
 * first byte of the replay stream, the exact shape design.md's
 * Implementation Notes describes for a job "already running at the
 * snapshot the replay starts from."
 */
static void build_fixture_scheduler_with_run_job(xce_scheduler_t* sched,
                                                 char* record_dir,
                                                 uint64_t    run_jobid,
                                                 const char* queue_name)
{
    xce_queue_t       wire_queue;
    xce_job_t*        run_job;
    xce_job_t*        jobs[1];
    xce_core_event_t  events[2];
    xce_sched_job_t*  job;
    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;

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

    run_job = make_job_with_state(run_jobid, JOB_STAT_RUN, queue_name);
    jobs[0] = run_job;

    memset(events, 0, sizeof(events));
    events[0].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[0].u.queue_update.queue = &wire_queue;
    events[1].type                 = XCE_CORE_EVENT_JOBS_PUSH;
    events[1].u.jobs_push.njobs    = 1;
    events[1].u.jobs_push.jobs     = jobs;

    rc = xce_core_push_events(&(sched->core), events, 2);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Sanity: job is really RUN-state with no accounts yet, in THIS
     * fixture core, before it's ever snapshotted -- non-tautology guard,
     * proving the pre-snapshot setup itself is correct. */
    job = sched_get_job(&(sched->core), run_jobid);
    TEST_ASSERT_NOT_NULL(job);
    TEST_ASSERT_TRUE(JOB_IS_RUN(job->state));
    TEST_ASSERT_NULL(job->accounts);
}

/* Same "leave the file open, skip sched_record_stop()'s own endSession
 * marker" technique test_sim_verify.c's own write_header_and_snapshot()
 * establishes -- this copy snapshots a core carrying one already-RUN job
 * (via build_fixture_scheduler_with_run_job() above), not an empty one. */
static FILE* write_header_and_snapshot_with_run_job(char*       out_path,
                                                    size_t      out_path_len,
                                                    const char* suffix,
                                                    uint64_t    run_jobid,
                                                    const char* queue_name)
{
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;

    make_tmp_dir(record_dir, sizeof(record_dir), suffix);
    build_fixture_scheduler_with_run_job(&sched, record_dir, run_jobid,
                                         queue_name);

    rc = sched_record_start(&sched, "fixture.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len);
    strcpy(out_path, sched.record_file_path);

    return sched.record_file;
}

/* Appends one ReplayEvent{jobsPush} message with exactly one JOB_STAT_PEND
 * job -- same technique test_sim_verify.c's own append_jobs_push_event()
 * establishes; this file's own ordinary/control job. */
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*)"bob";
    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_end_of_stream_marker(FILE* f)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

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

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

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

/* Appends one ReplaySession carrying exactly one recorded PendDecision for
 * `jobid` -- same technique test_sim_verify.c's own
 * append_replay_session_with_pend() establishes. Deliberately carries NO
 * decision at all for the snapshot's own already-RUN job: this file's
 * entire point is that no such recorded decision exists anywhere in this
 * replay's own window. */
static void append_replay_session_with_pend(FILE* f, uint64_t begin_round,
                                            uint64_t jobid, int main_reason)
{
    xce_job_preason_t     reason;
    sched_pend_decision_t pend;
    replay_decision_t     decision;
    replay_decision_t*    decisions[1];
    replay_session_t      session;
    struct capn           c;
    struct capn_segment*  cs;
    ReplaySession_ptr     sp;

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

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

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

    decisions[0] = &decision;

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

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

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

/* Same dup()/dup2() STDOUT_FILENO capture technique test_sim_verify.c's own
 * run_sim_and_capture_stdout() establishes -- also hands back the driver
 * struct itself (`out_sim`) so the caller can inspect post-replay core
 * state (job->accounts/job->state) directly, which test_sim_verify.c's own
 * copy never needed to do. */
static char* run_sim_and_capture_stdout(const char*             path,
                                        bool                    verify,
                                        xce_sched_sim_driver_t* out_sim,
                                        int*                    out_rc)
{
    sched_sim_args_t args;
    char             tmp[] = "/tmp/xce_simsnapgap_run_XXXXXX";
    int              fd    = mkstemp(tmp);
    int              saved_stdout;
    off_t            len;
    char*            buf;

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

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

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

    *out_rc = sched_sim_driver_run(out_sim, &args);

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

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

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

    return buf;
}

/* ---------------------------------------------------------------------- */
/* The Observable: Requirement 4's graceful degradation, from a snapshot.  */
/* ---------------------------------------------------------------------- */

/*
 * test_replay_from_snapshot_with_already_running_job_degrades_gracefully --
 * scheduler-sim-decision-parity spec, task 4.2 (Req 4.1, 4.2): a replay
 * starting from a snapshot with a job (901) already RUN, with NO recorded
 * decision for it anywhere in this replay's own window, must: (1) never
 * crash or abort the run (rc == 0), (2) leave that job's usage genuinely
 * unavailable rather than fabricate an allocation (job->accounts/allocs
 * stay NULL, job->state stays RUN, exactly as the snapshot restored it),
 * and (3) continue processing the rest of the replay normally -- a second,
 * ordinary job (902) pushed fresh mid-replay must still be decided (PEND)
 * exactly as it would be with no already-running job in the mix at all.
 */
static void
test_replay_from_snapshot_with_already_running_job_degrades_gracefully(void)
{
    char                   path[PATH_MAX];
    FILE*                  f;
    int                    rc;
    char*                  captured;
    xce_sched_sim_driver_t sim;
    xce_sched_job_t*       run_job;
    xce_sched_job_t*       fresh_job;

    /* Snapshot carries job 901, already RUN, no accounts -- see
     * build_fixture_scheduler_with_run_job()'s own doc comment. */
    f = write_header_and_snapshot_with_run_job(path, sizeof(path), "_gap",
                                               901ULL, "normal");
    TEST_ASSERT_NOT_NULL(f);

    /* Job 902: an ordinary, freshly-arriving job -- the real picker
     * independently PENDs it, exactly like every other integration fixture
     * in this directory. Its own recorded PEND decision is the ONLY
     * decision anywhere in this replay's window -- job 901 has none. */
    append_jobs_push_event(f, 1500ULL, 902ULL, "normal");
    append_replay_session_with_pend(f, 2000ULL, 902ULL, 0);
    append_end_of_stream_marker(f);
    fclose(f);

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

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "a replay starting from a snapshot with an already-running job and "
        "no in-window recorded decision for it must complete cleanly, not "
        "abort (Requirement 4.2)");
    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "MISMATCH"),
        "the genuinely unaccounted-for snapshot job must not produce a "
        "spurious --sim-verify mismatch either");

    /* Requirement 4.1: usage reported unavailable, nothing fabricated. */
    run_job = sched_get_job(&(sim.core), 901ULL);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        run_job, "the snapshot-restored job must still exist after replay");
    TEST_ASSERT_TRUE_MESSAGE(
        JOB_IS_RUN(run_job->state),
        "the snapshot-restored job's RUN state must be preserved");
    TEST_ASSERT_NULL_MESSAGE(
        run_job->accounts,
        "no recorded decision exists in-window for this job -- "
        "job->accounts must stay NULL (unavailable), never fabricated");
    TEST_ASSERT_NULL_MESSAGE(
        run_job->allocs,
        "job->allocs must likewise stay NULL -- nothing to reconstruct "
        "from and nothing must be fabricated in its place");

    /* Requirement 4.2: the rest of the replay is unaffected -- job 902
     * still gets its ordinary, real PEND decision. */
    fresh_job = sched_get_job(&(sim.core), 902ULL);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        fresh_job, "the ordinary, freshly-arriving job must still exist");
    TEST_ASSERT_FALSE_MESSAGE(
        JOB_IS_RUN(fresh_job->state),
        "the ordinary job was only ever PEND -- must not have been "
        "affected by the unrelated unaccounted job's own processing");

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

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