feat(libxce-alloc-preason-equality): stamp every allocation's from with the deciding session start (task 3.2)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

.kiro/specs/libxce-alloc-preason-equality/tasks.md

@@ -28,14 +28,14 @@ Test-first throughout (red → green → refactor). Every sub-task leaves the bu
   - _Boundary: libxce pending-reason module; the allocation OOM test binary (shared test harness, sequenced after 2.1)_
   - _Depends: 2.1_
 
-- [ ] 3. Core: reproducible reservation start time in the scheduler
+- [x] 3. Core: reproducible reservation start time in the scheduler
 - [x] 3.1 (P) Record each session's start from the same clock reading as the session start
   - Write the test first, in its own test function (not shared with 3.2): with recording enabled, a recorded session's begin time equals the session's start time exactly.
   - Take the recorded begin time from the session start instead of a second clock reading (design.md, Components: Recording).
   - Done when the test passes and the existing recording tests still pass.
   - _Requirements: 4.2, 6.5_
   - _Boundary: scheduler normal-mode driver_
-- [ ] 3.2 (P) Stamp every allocation with its deciding session's start time, from the session core
+- [x] 3.2 (P) Stamp every allocation with its deciding session's start time, from the session core
   - Write the tests first, in test functions separate from 3.1's: every allocation of one normal-mode session, including every primary of a merged COMPOUND request, carries the session start in whole seconds; the same holds in simulation mode, where the allocators' scheduler handle is the simulator's session-less wrapper, using a realistic epoch-scale recorded session start so a stamp taken from the wrong core (0) fails.
   - Stamp the reservation start time on every per-host and shared allocation record where the allocator result is received with the session core, not in the record-building helper (design.md, Components: `from` stamping, Decision 8).
   - Confirm no allocation path bypasses that point; if one does, stop and raise it.

src/scheduler/src/scheduler.c

@@ -182,6 +182,56 @@ static void sched_pend_from_cached_reason_(xce_sched_ctx_t* ctx, int category)
     add_job_preason(ctx->reason->reasons, category, NULL, NULL);
 }
 
+/*
+ * sched_stamp_session_from_ - libxce-alloc-preason-equality Req 4.1/4.4,
+ * design.md Decision 8: stamp every allocation record the allocator just
+ * produced (each per-host allocs[i]->allocs[j] and each allocs_shared[k])
+ * with the deciding session's start, in whole seconds since the epoch
+ * (core->session_start is nanoseconds; floor division). make_alloc()'s
+ * own time(NULL) default is not reproducible under replay.
+ *
+ * Done here, with the SESSION core sched_make_primary_alloc_() receives,
+ * not in sched_make_allocs(): in simulation mode that helper's `sched` is
+ * sim_alloc.c's session-less wrapper, whose core's session_start is 0.
+ * Every allocation (PRIMARY, each primary of a COMPOUND before
+ * merge_job_alloc() folds it, each ALTER alternative) passes through
+ * sched_make_primary_alloc_(), so all of one session's allocations carry
+ * the same `from`.
+ */
+#define SCHED_NSEC_PER_SEC 1000000000ULL
+
+static void sched_stamp_session_from_(const xce_sched_core_t* core,
+                                      xce_job_alloc_t*        alloc)
+{
+    uint64_t from = core->session_start / SCHED_NSEC_PER_SEC;
+    int      i;
+    int      j;
+
+    if (alloc == NULL) {
+        return;
+    }
+
+    for (i = 0; (alloc->allocs != NULL) && (i < alloc->nhosts); i++) {
+        xce_host_alloc_t* ha = alloc->allocs[i];
+
+        if ((ha == NULL) || (ha->allocs == NULL)) {
+            continue;
+        }
+        for (j = 0; j < ha->nallocs; j++) {
+            if (ha->allocs[j] != NULL) {
+                ha->allocs[j]->from = from;
+            }
+        }
+    }
+
+    for (i = 0; (alloc->allocs_shared != NULL) && (i < alloc->nres_shared);
+         i++) {
+        if (alloc->allocs_shared[i] != NULL) {
+            alloc->allocs_shared[i]->from = from;
+        }
+    }
+}
+
 static int sched_make_primary_alloc_(xce_sched_core_t* core,
                                      xce_sched_ctx_t* ctx)
 {
@@ -214,6 +264,13 @@ static int sched_make_primary_alloc_(xce_sched_core_t* core,
         rc = sched_make_parjob_allocation(core, ctx);
     }
 
+    /* Req 4.1/4.4: the allocator's result is received here with the
+     * session core -- stamp every record's `from` before a COMPOUND's
+     * merge_job_alloc() or the dispatch decision sees it. */
+    if ((rc > 0) && (ctx->alloc != NULL)) {
+        sched_stamp_session_from_(core, ctx->alloc);
+    }
+
     /* Req 7.3: once the chosen allocator has exhausted every candidate
      * plan (the not-satisfied signal, rc == 0), populate the cache with
      * this resreq's fingerprint + identity and the exhausted attempt's own

src/scheduler/testsuite/meson.build

@@ -2307,3 +2307,29 @@ sched_sim_e2e_limit_visibility_test = executable(
 test('scheduler sim e2e limit visibility test',
      sched_sim_e2e_limit_visibility_test, suite: 'scheduler',
      env: ['ASAN_OPTIONS=detect_leaks=0'])
+
+# libxce-alloc-preason-equality spec, task 3.2 (Req 4.1, 4.4, 6.5): every
+# allocation of one session -- normal mode (real sched_init()) and
+# simulation mode (sim_alloc.c's session-less wrapper allocators) --
+# carries the session start in whole seconds as `from`, stamped by
+# sched_make_primary_alloc_() from the session core. Same real-
+# scheduler_srcs link technique as sched_parjob_e2e_multislot_test.
+sched_alloc_from_stamp_test_srcs = ['test_sched_alloc_from_stamp.c']
+foreach f : scheduler_srcs
+  if f != 'src/main.c'
+    sched_alloc_from_stamp_test_srcs += ['../' + f]
+  endif
+endforeach
+
+sched_alloc_from_stamp_test = executable(
+    'sched_alloc_from_stamp_test',
+    sched_alloc_from_stamp_test_srcs,
+    include_directories:[incdir],
+    dependencies: [scheduler_deps, unity_dep],
+    link_args: ['-Wl,--wrap=xcecli_start',
+               '-Wl,--wrap=xcecli_join'],
+    install: false)
+
+test('scheduler alloc from stamp test',
+     sched_alloc_from_stamp_test, suite: 'scheduler',
+     env: ['ASAN_OPTIONS=detect_leaks=0'])

src/scheduler/testsuite/test_sched_alloc_from_stamp.c

@@ -0,0 +1,569 @@
+/*
+ * test_sched_alloc_from_stamp.c - libxce-alloc-preason-equality spec, task
+ * 3.2 (Req 4.1, 4.4, 4.5, 6.5, 6.6; design.md "`from` stamping
+ * (`scheduler.c`)", Decision 8).
+ *
+ * Every allocation a scheduling session produces must carry, as its
+ * reservation start time `from`, the deciding session's start in whole
+ * seconds (core->session_start is nanoseconds; `from` is
+ * session_start / 10^9, floor) -- not the wall-clock second make_alloc()
+ * happened to run in. The stamp is applied where the allocator's result is
+ * received together with the SESSION core (sched_make_primary_alloc_(),
+ * scheduler.c), never in sched_make_allocs(): in simulation mode that
+ * helper's `sched` is sim_alloc.c's session-less wrapper, whose own core
+ * never begins a session (session_start == 0).
+ *
+ * Two tests, one per driver shape, both going through the real, public
+ * sched_make_allocation() chain and the real seqjob/parjob allocator
+ * plugins (real scheduler_srcs link, same technique as test_parjob_e2e_
+ * multislot.c/test_sim_alloc_limit_fanout.c):
+ *
+ *   - normal mode: a real sched_init()'d scheduler, real wire ingest, the
+ *     live allocator thread; one session dispatches a plain PRIMARY job
+ *     and a two-primary COMPOUND job (merged by merge_job_alloc()).
+ *   - simulation mode: a bare xce_sched_sim_driver_t bootstrapped with
+ *     sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_allocator()
+ *     (so the allocators' plugin->sched is the session-less wrapper);
+ *     sched_make_allocation(&sim.core, ...) called directly.
+ *
+ * Both sessions start at a realistic epoch-scale time deliberately far from
+ * "now", so neither a time(NULL) stamp (the old make_alloc() default) nor a
+ * stamp taken from the wrong core (0) can pass.
+ *
+ * Fixtures are never torn down (same convention as every other real-
+ * scheduler_srcs test in this directory).
+ */
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+
+#include <unity/unity.h>
+
+#include "client.h"
+#include "controller.capnp.h"
+#include "core.h"
+#include "decision.h"
+#include "plugin.h"
+#include "resobj.h"
+#include "resreq.h"
+#include "sched/alloc_cache.h"
+#include "sched/types.h"
+#include "scheduler.h"
+#include "sim.h"
+#include "xce/cluster.h"
+#include "xstring.h"
+
+extern int  scheduler_req_handler(xce_request_t*, void*);
+extern void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim);
+extern void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim);
+
+#define NSEC_PER_SEC 1000000000ULL
+
+/* 2023-11-14T22:13:20.123456789Z -- epoch-scale, but years away from any
+ * time(NULL) this test can run at, and not a whole second (so the floor
+ * division is exercised). */
+#define NORMAL_SESSION_EPOCH_NS ((uint64_t)1700000000123456789ULL)
+/* A recorded session start (beginRound) for the simulation test, also
+ * epoch-scale and far from now. */
+#define SIM_RECORDED_BEGIN_NS ((uint64_t)1690000000987654321ULL)
+
+#define WORKER_A_NAME "from-stamp-host-a"
+#define WORKER_B_NAME "from-stamp-host-b"
+#define WORKER_SLOTS  2.0f
+#define QUEUE_NAME    "from_stamp_queue"
+
+#define PRIMARY_RESREQ  "rusage[mem=1]"
+#define COMPOUND_RESREQ "1*{rusage[mem=1]} + 2*{rusage[mem=1]}"
+
+#define NORMAL_PRIMARY_JOBID  ((uint64_t)630201ULL)
+#define NORMAL_COMPOUND_JOBID ((uint64_t)630202ULL)
+#define SIM_PRIMARY_JOBID     ((uint64_t)630211ULL)
+#define SIM_COMPOUND_JOBID    ((uint64_t)630212ULL)
+
+int __wrap_xcecli_start(xcecli_ctx_t c)
+{
+    (void)c;
+    return 0;
+}
+
+int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
+{
+    (void)c;
+    (void)role;
+    (void)attrs;
+    return 0;
+}
+
+void setUp(void) { return; }
+void tearDown(void) { return; }
+
+/* ---------------------------------------------------------------------- */
+/* Shared helpers                                                          */
+/* ---------------------------------------------------------------------- */
+
+static xce_resource_t* register_plain_resource(var resmgr, const char* name)
+{
+    xce_resource_t res;
+
+    memset(&res, 0, sizeof(res));
+    res.name = strdup(name);
+
+    return makeResource(resmgr, &res);
+}
+
+/* Anchor a core's virtual clock so that it reads `epoch_ns` right now and
+ * advances at `speed_num` (1: normal mode's real-time pace; 0: the
+ * simulator's frozen clock between recorded timestamps). Same atomics
+ * xce_core_set_wallclock_anchor()/sim.c's sim_set_virtual_clock() store. */
+static void anchor_core_clock(xce_sched_core_t* core, uint64_t epoch_ns,
+                              uint64_t speed_num)
+{
+    struct timespec ts;
+    uint64_t        mono_ns;
+
+    clock_gettime(CLOCK_MONOTONIC, &ts);
+    mono_ns = ((uint64_t)ts.tv_sec * NSEC_PER_SEC) + (uint64_t)ts.tv_nsec;
+
+    atomic_store_explicit(&(core->virtual_clock_base), epoch_ns,
+                          memory_order_release);
+    atomic_store_explicit(&(core->real_clock_base), mono_ns,
+                          memory_order_release);
+    atomic_store_explicit(&(core->speed_num), speed_num, memory_order_release);
+    atomic_store_explicit(&(core->speed_den), 1, memory_order_release);
+}
+
+/* Assert every per-host and shared allocation record of `alloc` carries
+ * `want_from`; returns the number of records checked. */
+static int assert_every_alloc_from(const xce_job_alloc_t* alloc,
+                                   uint64_t want_from, const char* what)
+{
+    char msg[256];
+    int  checked = 0;
+    int  i, j;
+
+    TEST_ASSERT_NOT_NULL(alloc);
+
+    for (i = 0; i < alloc->nhosts; i++) {
+        const xce_host_alloc_t* ha = alloc->allocs[i];
+
+        TEST_ASSERT_NOT_NULL(ha);
+        for (j = 0; j < ha->nallocs; j++) {
+            const xce_alloc_t* a = ha->allocs[j];
+
+            if (a == NULL) {
+                continue;
+            }
+            snprintf(msg, sizeof(msg),
+                     "%s: allocs[%d](host=%s).allocs[%d](resname=%s).from "
+                     "must be the session start in whole seconds",
+                     what, i, ha->host != NULL ? ha->host : "?", j,
+                     a->resname != NULL ? a->resname : "?");
+            TEST_ASSERT_EQUAL_UINT64_MESSAGE(want_from, a->from, msg);
+            checked++;
+        }
+    }
+
+    for (i = 0; i < alloc->nres_shared; i++) {
+        const xce_alloc_t* a = alloc->allocs_shared[i];
+
+        if (a == NULL) {
+            continue;
+        }
+        snprintf(msg, sizeof(msg),
+                 "%s: allocs_shared[%d](resname=%s).from must be the "
+                 "session start in whole seconds",
+                 what, i, a->resname != NULL ? a->resname : "?");
+        TEST_ASSERT_EQUAL_UINT64_MESSAGE(want_from, a->from, msg);
+        checked++;
+    }
+
+    return checked;
+}
+
+static double alloc_total_slots(const xce_job_alloc_t* alloc)
+{
+    double total = 0.0;
+    int    i, j;
+
+    for (i = 0; i < alloc->nhosts; i++) {
+        for (j = 0; j < alloc->allocs[i]->nallocs; j++) {
+            const xce_alloc_t* a = alloc->allocs[i]->allocs[j];
+
+            if ((a->resname != NULL) && (strcmp(a->resname, "slots") == 0)) {
+                total += a->value;
+            }
+        }
+    }
+    return total;
+}
+
+/* ---------------------------------------------------------------------- */
+/* Normal-mode fixture (real sched_init(), real wire ingest)               */
+/* ---------------------------------------------------------------------- */
+
+static xce_worker_t* make_wire_worker(const char* name, float slots)
+{
+    xce_worker_t*    w       = (xce_worker_t*)calloc(1, sizeof(*w));
+    xce_val_t*       val     = (xce_val_t*)calloc(1, sizeof(*val));
+    xce_name_val_t*  nv      = (xce_name_val_t*)calloc(1, sizeof(*nv));
+    xce_name_val_t** nv_ptrs = (xce_name_val_t**)calloc(1, sizeof(*nv_ptrs));
+
+    TEST_ASSERT_NOT_NULL(w);
+    TEST_ASSERT_NOT_NULL(val);
+    TEST_ASSERT_NOT_NULL(nv);
+    TEST_ASSERT_NOT_NULL(nv_ptrs);
+
+    val->which    = Value_u_fvalue;
+    val->u.fvalue = slots;
+
+    nv->name   = strdup("slots");
+    nv->val    = val;
+    nv_ptrs[0] = nv;
+
+    w->name              = strdup(name);
+    w->status            = 0;
+    w->nresvals_to_sched = 1;
+    w->resvals_to_sched  = nv_ptrs;
+
+    return w;
+}
+
+static xce_job_t* make_push_job(uint64_t jobid, const char* resreq)
+{
+    xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
+
+    TEST_ASSERT_NOT_NULL(j);
+
+    j->jobid      = jobid;
+    j->state      = JOB_STAT_PEND;
+    j->sub_time   = 1000;
+    j->queue_time = 1000;
+
+    j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
+    TEST_ASSERT_NOT_NULL(j->submit);
+
+    j->submit->queue        = (char*)QUEUE_NAME;
+    j->submit->user         = (char*)"from_stamp_user";
+    j->submit->project_name = (char*)"from_stamp_proj";
+    j->submit->resreq       = (char*)resreq;
+    j->submit->mintasks     = 1;
+    j->submit->maxtasks     = 1;
+
+    return j;
+}
+
+static xce_scheduler_t* start_normal_sched(void)
+{
+    static xce_client_config_t  config;
+    static xce_cluster_config_t cluster_cfg;
+    xce_scheduler_t*            sched
+        = (xce_scheduler_t*)calloc(1, sizeof(*sched));
+    xce_core_event_t            ev;
+    xce_worker_t*               worker_ptrs[2];
+    xce_queue_t*                q;
+    int                         rc;
+
+    TEST_ASSERT_NOT_NULL(sched);
+
+    memset(&config, 0, sizeof(config));
+    config.nmasters        = 1;
+    config.masters[0]      = "127.0.0.1";
+    config.master_ports[0] = 19996;
+    config.from            = "sched_alloc_from_stamp_test";
+    config.key             = "keepsecret$";
+
+    memset(&cluster_cfg, 0, sizeof(cluster_cfg));
+    cluster_cfg.candidates_count = 1;
+    strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1",
+            XCE_MAXHOSTNAME_LEN - 1);
+    cluster_cfg.candidates[0].port = 19996;
+    strncpy(cluster_cfg.identity, "sched_alloc_from_stamp_test",
+            sizeof(cluster_cfg.identity) - 1);
+
+    rc = sched_init(sched, &config, &cluster_cfg, scheduler_req_handler);
+    TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: sched_init() must succeed");
+
+    TEST_ASSERT_NOT_NULL(
+        register_plain_resource(sched->core.resmgr, "slots"));
+    xce_core_set_cache_clear_cb(&(sched->core), xce_alloc_cache_clear_cb);
+
+    q = (xce_queue_t*)calloc(1, sizeof(*q));
+    TEST_ASSERT_NOT_NULL(q);
+    q->queue    = strdup(QUEUE_NAME);
+    q->status   = QUEUE_STAT_OPEN;
+    q->priority = 1;
+
+    memset(&ev, 0, sizeof(ev));
+    ev.type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
+    ev.u.queue_update.queue = q;
+    TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0);
+
+    worker_ptrs[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS);
+    worker_ptrs[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS);
+    memset(&ev, 0, sizeof(ev));
+    ev.type                    = XCE_CORE_EVENT_WORKERS_PUSH;
+    ev.u.workers_push.nworkers = 2;
+    ev.u.workers_push.workers  = worker_ptrs;
+    TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0);
+
+    return sched;
+}
+
+static void push_job(xce_scheduler_t* sched, uint64_t jobid,
+                     const char* resreq)
+{
+    xce_core_event_t ev;
+    xce_job_t*       job_ptrs[1];
+
+    job_ptrs[0] = make_push_job(jobid, resreq);
+    memset(&ev, 0, sizeof(ev));
+    ev.type              = XCE_CORE_EVENT_JOBS_PUSH;
+    ev.u.jobs_push.njobs = 1;
+    ev.u.jobs_push.jobs  = job_ptrs;
+    TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0);
+    TEST_ASSERT_NOT_NULL(sched_get_job(&(sched->core), jobid));
+}
+
+/* Feed jobid straight into the live allocator thread (the picker stays
+ * parked: xce_core_signal_go() is never called) and wait, bounded and
+ * non-blocking, for its decision -- same technique as
+ * test_parjob_e2e_multislot.c's dispatch_and_wait(). */
+static xce_decision_t* dispatch_and_wait(xce_scheduler_t* sched,
+                                         uint64_t          jobid)
+{
+    xce_decision_t* decision = NULL;
+    int             rc;
+    int             waited_ms;
+
+    rc = cbuf_tryput(&(sched->core.jbuf), &jobid);
+    TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: jbuf must have room");
+
+    for (waited_ms = 0; waited_ms < 2000; waited_ms += 5) {
+        struct timespec ts = { 0, 5 * 1000 * 1000 };
+
+        if (cbuf_tryget(&(sched->core.decisions), &decision) >= 0) {
+            return decision;
+        }
+        nanosleep(&ts, NULL);
+    }
+
+    return NULL;
+}
+
+/*
+ * Req 4.1, 4.4 (normal mode): one session dispatches a plain PRIMARY job
+ * and a two-primary COMPOUND job; every per-host and shared allocation
+ * record of both carries session_start / 10^9.
+ */
+static void test_normal_mode_every_session_alloc_from_is_session_start(void)
+{
+    xce_scheduler_t* sched = start_normal_sched();
+    xce_decision_t*  dprimary;
+    xce_decision_t*  dcompound;
+    uint64_t         want_from;
+    int              n;
+
+    push_job(sched, NORMAL_PRIMARY_JOBID, PRIMARY_RESREQ);
+    push_job(sched, NORMAL_COMPOUND_JOBID, COMPOUND_RESREQ);
+
+    /* Normal mode's clock runs at real-time pace (speed 1/1) from an
+     * epoch anchor, exactly as xce_core_set_wallclock_anchor() sets it --
+     * anchored here at a known past instant rather than "now", so a
+     * time(NULL) stamp cannot coincide with the session start. */
+    anchor_core_clock(&(sched->core), NORMAL_SESSION_EPOCH_NS, 1);
+    TEST_ASSERT_EQUAL_INT(0, xce_core_begin_session(&(sched->core)));
+
+    TEST_ASSERT_TRUE_MESSAGE(
+        sched->core.session_start >= NORMAL_SESSION_EPOCH_NS,
+        "fixture: the session must start at the anchored epoch time");
+    want_from = sched->core.session_start / NSEC_PER_SEC;
+    TEST_ASSERT_NOT_EQUAL_MESSAGE(
+        (uint64_t)time(NULL), want_from,
+        "fixture: the session start must not be the current second");
+
+    dprimary = dispatch_and_wait(sched, NORMAL_PRIMARY_JOBID);
+    TEST_ASSERT_NOT_NULL(dprimary);
+    TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, dprimary->op);
+    n = assert_every_alloc_from(dprimary->u.alloc, want_from,
+                                "normal-mode PRIMARY");
+    TEST_ASSERT_TRUE_MESSAGE(n > 0, "PRIMARY must carry allocations");
+
+    dcompound = dispatch_and_wait(sched, NORMAL_COMPOUND_JOBID);
+    TEST_ASSERT_NOT_NULL(dcompound);
+    TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, dcompound->op);
+    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
+        3.0f, (float)alloc_total_slots(dcompound->u.alloc),
+        "fixture: the merged COMPOUND must cover both primaries (1 + 2)");
+    n = assert_every_alloc_from(dcompound->u.alloc, want_from,
+                                "normal-mode COMPOUND");
+    TEST_ASSERT_TRUE_MESSAGE(n > 0, "COMPOUND must carry allocations");
+}
+
+/* ---------------------------------------------------------------------- */
+/* Simulation-mode fixture (bare sim core, sim_alloc.c wrapper allocators) */
+/* ---------------------------------------------------------------------- */
+
+static void add_sim_worker(xce_sched_core_t* core, const char* name,
+                           double slots)
+{
+    xce_sched_worker_t* worker
+        = (xce_sched_worker_t*)calloc(1, sizeof(*worker));
+    xce_resource_t*     r   = NULL;
+    xce_val_t*          val = NULL;
+    int                 rc;
+
+    TEST_ASSERT_NOT_NULL(worker);
+
+    worker->name   = strdup(name);
+    worker->status = 0;
+    worker->idx    = -1;
+    zarray_init(&(worker->resvals), 128,
+                (zarray_item_killer_t)kill_sched_resval);
+    zarray_init(&(worker->resInUse), 128,
+                (zarray_item_killer_t)kill_sched_resval);
+
+    rc = key_find(core->resmgr, (void*)"slots", (void**)&r);
+    TEST_ASSERT_TRUE(rc >= 0);
+    TEST_ASSERT_NOT_NULL(r);
+
+    val = (xce_val_t*)calloc(1, sizeof(*val));
+    TEST_ASSERT_NOT_NULL(val);
+    val->which    = Value_u_fvalue;
+    val->u.fvalue = slots;
+    zarray_put(&(worker->resvals), r->idx, val);
+
+    rc = sched_add_worker(core, worker);
+    TEST_ASSERT_TRUE(rc >= 0);
+}
+
+static xce_sched_job_t* make_sim_job(xce_scheduler_t* job_wrapper,
+                                     uint64_t jobid, const char* resreq)
+{
+    xce_job_t*         j      = (xce_job_t*)calloc(1, sizeof(*j));
+    job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit));
+    xce_sched_job_t*   job;
+
+    TEST_ASSERT_NOT_NULL(j);
+    TEST_ASSERT_NOT_NULL(submit);
+
+    submit->queue        = (char*)QUEUE_NAME;
+    submit->user         = (char*)"from_stamp_user";
+    submit->project_name = (char*)"from_stamp_proj";
+    submit->resreq       = (char*)resreq;
+    submit->mintasks     = 1;
+    submit->maxtasks     = 1;
+
+    j->jobid  = jobid;
+    j->state  = JOB_STAT_PEND;
+    j->submit = submit;
+
+    job = make_sched_job(job_wrapper, j);
+    TEST_ASSERT_NOT_NULL(job);
+    TEST_ASSERT_NOT_NULL(job->schedobj);
+
+    return job;
+}
+
+/*
+ * Req 4.1, 4.4 (simulation mode): the allocators' plugin->sched is
+ * sim_alloc.c's session-less wrapper (its core's session_start is 0); the
+ * session core (sim.core) starts at a recorded, epoch-scale beginRound.
+ * Every allocation must carry that recorded start in whole seconds.
+ */
+static void test_sim_mode_every_session_alloc_from_is_session_start(void)
+{
+    xce_sched_sim_driver_t sim;
+    xce_scheduler_t        job_wrapper; /* make_sched_job() setup only */
+    xce_sched_queue_t*     queue;
+    xce_sched_job_t*       jprimary;
+    xce_sched_job_t*       jcompound;
+    xce_sched_plugin_t*    seqp;
+    xce_sched_plugin_t*    parp;
+    uint64_t               want_from;
+    int                    rc;
+    int                    n;
+
+    memset(&sim, 0, sizeof(sim));
+    rc = xce_sched_core_init(&(sim.core));
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    TEST_ASSERT_NOT_NULL(register_plain_resource(sim.core.resmgr, "slots"));
+
+    sim_bootstrap_seqjob_allocator(&sim);
+    sim_bootstrap_parjob_allocator(&sim);
+    seqp = (xce_sched_plugin_t*)sim.core.seqjob_allocator;
+    parp = (xce_sched_plugin_t*)sim.core.parjob_allocator;
+    TEST_ASSERT_NOT_NULL(seqp);
+    TEST_ASSERT_NOT_NULL(parp);
+    TEST_ASSERT_NOT_NULL(seqp->sched);
+    TEST_ASSERT_NOT_NULL(parp->sched);
+    TEST_ASSERT_TRUE_MESSAGE(
+        &(((xce_scheduler_t*)seqp->sched)->core)
+            != &(sim.core),
+        "fixture: the allocators' scheduler handle must be the simulator's "
+        "wrapper, not the session core");
+
+    queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue));
+    TEST_ASSERT_NOT_NULL(queue);
+    queue->name = makeString(QUEUE_NAME);
+    TEST_ASSERT_TRUE(sched_add_queue(&(sim.core), queue) >= 0);
+
+    add_sim_worker(&(sim.core), WORKER_A_NAME, WORKER_SLOTS);
+    add_sim_worker(&(sim.core), WORKER_B_NAME, WORKER_SLOTS);
+
+    memset(&job_wrapper, 0, sizeof(job_wrapper));
+    TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(job_wrapper.core)));
+    job_wrapper.core.resmgr  = sim.core.resmgr;
+    job_wrapper.core.workers = sim.core.workers;
+    job_wrapper.core.queues  = sim.core.queues;
+
+    jprimary = make_sim_job(&job_wrapper, SIM_PRIMARY_JOBID, PRIMARY_RESREQ);
+    TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, jprimary->schedobj->tag);
+    jprimary->schedobj->u.primary->resreq->minTasks = 1;
+    jprimary->schedobj->u.primary->resreq->maxTasks = 1;
+    TEST_ASSERT_TRUE(sched_add_job(&(sim.core), jprimary) >= 0);
+
+    jcompound
+        = make_sim_job(&job_wrapper, SIM_COMPOUND_JOBID, COMPOUND_RESREQ);
+    TEST_ASSERT_EQUAL_INT(XCE_RESREQ_COMPOUND, jcompound->schedobj->tag);
+    TEST_ASSERT_TRUE(sched_add_job(&(sim.core), jcompound) >= 0);
+
+    /* The simulator's clock is frozen at the recorded beginRound for the
+     * session (speed 0), so the session core's session_start is exactly
+     * the recorded begin time. */
+    anchor_core_clock(&(sim.core), SIM_RECORDED_BEGIN_NS, 0);
+    TEST_ASSERT_EQUAL_INT(0, xce_core_begin_session(&(sim.core)));
+    TEST_ASSERT_EQUAL_UINT64(SIM_RECORDED_BEGIN_NS, sim.core.session_start);
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
+        0,
+        ((xce_scheduler_t*)seqp->sched)
+            ->core.session_start,
+        "fixture: the wrapper's core never begins a session");
+    want_from = SIM_RECORDED_BEGIN_NS / NSEC_PER_SEC;
+
+    rc = sched_make_allocation(&(sim.core), jprimary->schedobj);
+    TEST_ASSERT_TRUE_MESSAGE(rc > 0, "sim PRIMARY must be allocated");
+    n = assert_every_alloc_from(jprimary->schedobj->alloc, want_from,
+                                "sim-mode PRIMARY");
+    TEST_ASSERT_TRUE_MESSAGE(n > 0, "PRIMARY must carry allocations");
+
+    rc = sched_make_allocation(&(sim.core), jcompound->schedobj);
+    TEST_ASSERT_TRUE_MESSAGE(rc > 0, "sim COMPOUND must be allocated");
+    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
+        3.0f, (float)alloc_total_slots(jcompound->schedobj->alloc),
+        "fixture: the merged COMPOUND must cover both primaries (1 + 2)");
+    n = assert_every_alloc_from(jcompound->schedobj->alloc, want_from,
+                                "sim-mode COMPOUND");
+    TEST_ASSERT_TRUE_MESSAGE(n > 0, "COMPOUND must carry allocations");
+}
+
+int main(void)
+{
+    UNITY_BEGIN();
+    RUN_TEST(test_normal_mode_every_session_alloc_from_is_session_start);
+    RUN_TEST(test_sim_mode_every_session_alloc_from_is_session_start);
+    return UNITY_END();
+}