feat(scheduler-core): implement builtin_fcfs session and post-allocation hooks (task 2.3)

Adds begin_session/next_job/end_session/on_dispatch/on_pend, driving a
dual-cursor walk (queue then job) over builtin_fcfs's private ordq_t
state. next_job() returns candidates in (queue priority desc, job
priority desc, queue_time asc) order and NULL once exhausted;
on_dispatch removes permanently, on_pend is a true no-op.

Review flagged an unsynchronized cross-thread access risk between
next_job() (picker thread) and on_dispatch/on_pend (allocator thread)
once wired live in task 2.4 — recorded in tasks.md Implementation Notes
for that task to close before registering builtin_fcfs as the default.

Requirements: 3.2, 3.4, 3.6

.kiro/specs/scheduler-core/tasks.md

@@ -41,7 +41,7 @@
   - _Requirements: 3.2_
   - _Depends: 2.1_
 
-- [ ] 2.3 Implement builtin_fcfs session and post-allocation hooks
+- [x] 2.3 Implement builtin_fcfs session and post-allocation hooks
   - `begin_session`: reset the iteration cursor to the head of the ordered queue list
   - `next_job`: walk queues then jobs in ordering, skipping not-ready queues/jobs, returning the first ready job and advancing the cursor; returns NULL when exhausted
   - `end_session`: no-op (iteration state already reset in `begin_session`)
@@ -356,3 +356,4 @@
 - Background IDE diagnostics ("file not found: rtti.h", "unknown type name" for various xce_* types) have appeared consistently on every scheduler.h/plugin.h/types.h/schedobj.c change so far (tasks 1.1-1.3) and are confirmed stale clangd artifacts each time via independent real `meson compile`/`meson test` runs — not real build failures. Don't let these block a task; verify with the real build instead.
 - Meson build dir is `src/xce.build`, not repo-root `xce.build` — build/test commands must run from `src/` (e.g. `cd src && nix develop . --command bash -c "meson test -C xce.build --suite scheduler"`).
 - Scheduler testsuite convention: run test binaries directly with `ASAN_OPTIONS=detect_leaks=0` (fixtures aren't torn down, by design — matches existing `meson.build` test registration).
+- Task 2.3's review found a real concurrency gap for task 2.4 to close: once builtin_fcfs is wired live, `next_job()` (picker thread) and `on_dispatch`/`on_pend` (allocator thread) can race on the same unsynchronized `fcfs_plugin_data_t` private state — `cbuf_get()` unblocks the picker's next `cbuf_put()` before the allocator finishes processing the dequeued job (jbuf capacity is 1), so the picker can re-enter `next_job()` while the allocator is still mutating `priv->queues` via `on_dispatch`/`on_pend`. Today's production path never hits this because the allocator thread calls a different plugin category (`limit.c`'s resource-limit hook), never `fcfs.c`'s own private state. Task 2.4 must add a mutex guarding `fcfs_plugin_data_t` across all five picker hooks (or an equivalent handoff redesign) before registering builtin_fcfs as the live default.

src/scheduler/src/plugins/builtin_fcfs.c

@@ -49,6 +49,7 @@
 #include "ordq.h"
 #include "plugin.h"
 #include "sched/types.h"
+#include "scheduler.h"
 #include "utils.h"
 
 /* Private per-queue candidate state (Req 3 AC 2 "init" row: "one
@@ -72,12 +73,22 @@ typedef struct {
  * next_job()"). Allocated by builtin_fcfs_init(), freed by
  * builtin_fcfs_fini() -- see below. */
 typedef struct {
-    ordq_t queues;       /* ordered by queue->priority desc, see
+    ordq_t queues;           /* ordered by queue->priority desc, see
-                          * fcfs_queue_cmp */
+                              * fcfs_queue_cmp */
-    ordq_item_t* cursor; /* iteration cursor over queues; NULL until
+    ordq_item_t* cursor;     /* iteration cursor over queues; NULL until
-                          * begin_session (task 2.3) sets it -- init() only
+                              * begin_session (task 2.3) sets it -- init() only
-                          * zeroes it here, next_job() (task 2.3) is what
+                              * zeroes it here, next_job() (task 2.3) is what
-                          * walks it */
+                              * walks it */
+    ordq_item_t* job_cursor; /* iteration cursor over cursor's queue's own
+                              * pendJobs (task 2.3). Kept in lockstep with
+                              * `cursor` by fcfs_cursor_enter_queue() below:
+                              * whenever `cursor` is repointed at a queue,
+                              * `job_cursor` is repointed at that queue's
+                              * pendJobs.head in the same step. That
+                              * invariant is what makes job_cursor == NULL
+                              * unambiguous inside next_job() -- it always
+                              * means "current queue's job list exhausted",
+                              * never "haven't started this queue yet". */
 } fcfs_plugin_data_t;
 
 /* ordq_data_killer_t for a queue's pendJobs: frees the fcfs_job_t wrapper.
@@ -203,7 +214,8 @@ int builtin_fcfs_init(void* picker, xce_sched_core_t* core)
         return rc;
     }
 
-    priv->cursor = NULL;
+    priv->cursor     = NULL;
+    priv->job_cursor = NULL;
 
     plugin->data = (plugin_data_t)priv;
 
@@ -548,3 +560,207 @@ int builtin_fcfs_on_job_update(void* picker, xce_sched_job_t* job)
 
     return ordq_put(&(fq->pendJobs), (void*)fj);
 }
+
+/* ---------------------------------------------------------------------
+ * Task 2.3: session hooks (Req 3 AC 2's begin_session/next_job/end_session
+ * rows; Req 1 AC 2's session-hook signatures; Req 1 AC 3 "Session hooks ...
+ * run in the picker thread") and post-allocation hooks (Req 3 AC 2's
+ * on_dispatch/on_pend rows; Req 1 AC 2's post-allocation-hook signatures;
+ * Req 1 AC 3 "Post-allocation hooks ... run in the allocator thread").
+ * These are the only functions that ever read priv->cursor/job_cursor --
+ * the state-change hooks above never touch either field.
+ * ------------------------------------------------------------------- */
+
+/* Repoints priv->cursor at `queue_item` and resyncs priv->job_cursor to
+ * that queue's own pendJobs.head in the same step, so the two fields are
+ * always in lockstep. This is what makes job_cursor == NULL unambiguous
+ * inside next_job() below: it always means "cursor's queue has no more
+ * candidates", never "haven't started walking this queue's pendJobs yet".
+ * queue_item == NULL (all queues exhausted) correctly leaves job_cursor
+ * NULL too. */
+static void fcfs_cursor_enter_queue(fcfs_plugin_data_t* priv,
+                                    ordq_item_t*        queue_item)
+{
+    priv->cursor     = queue_item;
+    priv->job_cursor = (queue_item != NULL)
+                         ? ((fcfs_queue_t*)queue_item->data)->pendJobs.head
+                         : NULL;
+}
+
+/*
+ * builtin_fcfs_begin_session - xce_job_picker_iface_t.begin_session (Req 3
+ * AC 2 "begin_session" row, Req 3 AC 6 "iteration state reset in
+ * begin_session()").
+ *
+ * Resets the iteration cursor to the head of priv->queues, and -- in
+ * lockstep, via fcfs_cursor_enter_queue -- the job cursor to that first
+ * queue's pendJobs.head. Called once at the start of every scheduling round
+ * (picker thread).
+ */
+int builtin_fcfs_begin_session(void* picker)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
+    fcfs_plugin_data_t* priv   = NULL;
+
+    if (plugin == NULL) {
+        return -EINVAL;
+    }
+
+    priv = (fcfs_plugin_data_t*)plugin->data;
+    if (priv == NULL) {
+        return -EINVAL;
+    }
+
+    fcfs_cursor_enter_queue(priv, priv->queues.head);
+
+    return 0;
+}
+
+/*
+ * builtin_fcfs_next_job - xce_job_picker_iface_t.next_job (Req 3 AC 2
+ * "next_job" row, Req 3 AC 4 ordering equivalence with today's
+ * fcfs_elect_job).
+ *
+ * Walks priv->queues from priv->cursor, skipping any queue for which
+ * sched_queue_is_ready() is false -- the same gate fcfs_elect_job applies
+ * (design.md: "sched_queue_is_ready()/sched_job_is_ready() ... called as-is
+ * by both today's fcfs_elect_job and the new builtin_fcfs"). Within a ready
+ * queue, walks that queue's pendJobs from priv->job_cursor and returns the
+ * first job for which sched_job_is_ready() is true, leaving both cursors
+ * positioned just past it so the very next call resumes there. This is
+ * what makes next_job() a true single-job pull interface -- unlike
+ * fcfs_elect_job, which pushes every ready job of a session in one pass,
+ * next_job() returns exactly one job per call and remembers its place.
+ *
+ * When a queue's job list is exhausted (or the queue itself is not ready),
+ * advances to the next queue via fcfs_cursor_enter_queue and keeps
+ * searching. Returns NULL once priv->cursor runs off the end of
+ * priv->queues -- "all candidates exhausted" (task 2.3 Observable).
+ *
+ * Cancel-point plumbing (pthread_testcancel() calls for the live picker
+ * thread) is task 2.4 scope -- this task only implements the walking
+ * contract itself.
+ */
+xce_sched_job_t* builtin_fcfs_next_job(void* picker)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
+    fcfs_plugin_data_t* priv   = NULL;
+
+    if (plugin == NULL) {
+        return NULL;
+    }
+
+    priv = (fcfs_plugin_data_t*)plugin->data;
+    if (priv == NULL) {
+        return NULL;
+    }
+
+    while (priv->cursor != NULL) {
+        fcfs_queue_t* fq = (fcfs_queue_t*)priv->cursor->data;
+
+        if (!sched_queue_is_ready(fq->queue)) {
+            fcfs_cursor_enter_queue(priv, priv->cursor->next);
+            continue;
+        }
+
+        while (priv->job_cursor != NULL) {
+            fcfs_job_t* fj = (fcfs_job_t*)priv->job_cursor->data;
+
+            priv->job_cursor = priv->job_cursor->next;
+
+            if (sched_job_is_ready(fj->job)) {
+                return fj->job;
+            }
+        }
+
+        fcfs_cursor_enter_queue(priv, priv->cursor->next);
+    }
+
+    return NULL;
+}
+
+/*
+ * builtin_fcfs_end_session - xce_job_picker_iface_t.end_session (Req 3 AC 2
+ * "end_session" row: "No-op for FCFS (iteration state reset in
+ * begin_session)"; Req 3 AC 6).
+ *
+ * Validates its argument (same defensive "picker must be a real, init()'d
+ * instance" contract every hook in this file follows -- plugin non-NULL AND
+ * already init()'d) but otherwise touches nothing: priv->cursor/job_cursor
+ * are left exactly where next_job() last positioned them (both NULL, since
+ * end_session() is only ever called after next_job() has returned NULL);
+ * begin_session() is what resets them for the next round.
+ */
+int builtin_fcfs_end_session(void* picker)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
+
+    if (plugin == NULL) {
+        return -EINVAL;
+    }
+
+    if (plugin->data == NULL) {
+        return -EINVAL;
+    }
+
+    return 0;
+}
+
+/*
+ * builtin_fcfs_on_dispatch - xce_job_picker_iface_t.on_dispatch (Req 3 AC 2
+ * "on_dispatch" row).
+ *
+ * The job was allocated -- remove it from its queue's pendJobs permanently
+ * so it never becomes a candidate again. Only `jobid` is available here (no
+ * job or queue pointer), so this reuses fcfs_remove_job_everywhere (task
+ * 2.2's helper for on_job_update's queue-reassignment case), which already
+ * searches every registered queue for the jobid -- a job is never a
+ * pending candidate in more than one queue at once, so at most one queue's
+ * ordq_del ever actually removes anything.
+ */
+int builtin_fcfs_on_dispatch(void* picker, uint64_t jobid)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
+    fcfs_plugin_data_t* priv   = NULL;
+
+    if (plugin == NULL) {
+        return -EINVAL;
+    }
+
+    priv = (fcfs_plugin_data_t*)plugin->data;
+    if (priv == NULL) {
+        return -EINVAL;
+    }
+
+    fcfs_remove_job_everywhere(priv, jobid);
+
+    return 0;
+}
+
+/*
+ * builtin_fcfs_on_pend - xce_job_picker_iface_t.on_pend (Req 3 AC 2
+ * "on_pend" row: "No-op for FCFS -- job remains in pendJobs, will be
+ * retried next round in same position").
+ *
+ * Validates its argument (same defensive "picker must be a real, init()'d
+ * instance" contract every hook in this file follows) but otherwise does
+ * nothing: the job's fcfs_job_t wrapper was never removed from its queue's
+ * pendJobs (only on_dispatch removes it), so it is already exactly where it
+ * needs to be for the next round.
+ */
+int builtin_fcfs_on_pend(void* picker, uint64_t jobid)
+{
+    xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)picker;
+
+    (void)jobid;
+
+    if (plugin == NULL) {
+        return -EINVAL;
+    }
+
+    if (plugin->data == NULL) {
+        return -EINVAL;
+    }
+
+    return 0;
+}

src/scheduler/testsuite/meson.build

@@ -104,6 +104,20 @@ sched_picker_fcfs_test = executable('sched_picker_fcfs_test',
                                     dependencies: [scheduler_test_deps],
                                     install: false)
 
+# White-box test for builtin_fcfs's session hooks (begin_session/next_job/
+# end_session) and post-allocation hooks (on_dispatch/on_pend) (scheduler-core
+# spec, task 2.3). Split out of test_picker_fcfs.c rather than appended to it
+# -- that file was already ~765 lines after tasks 2.1/2.2, against this
+# project's 800-line ceiling. Also #includes ../src/plugins/builtin_fcfs.c
+# directly (same white-box pattern as sched_picker_fcfs_test), and supplies
+# its own sched_queue_is_ready()/sched_job_is_ready() test doubles since this
+# translation unit does not link scheduler.c (see the file header comment).
+sched_picker_fcfs_session_test = executable('sched_picker_fcfs_session_test',
+                                            ['test_picker_fcfs_session.c'],
+                                            include_directories:[incdir],
+                                            dependencies: [scheduler_test_deps],
+                                            install: false)
+
 # Register as meson tests so `meson test --suite scheduler` runs them all.
 # detect_leaks=0 matches the suite convention (fixtures are not torn down).
 test('scheduler test', scheduler_test, suite: 'scheduler',
@@ -124,3 +138,5 @@ test('scheduler picker config fields test', sched_picker_config_fields_test, sui
      env: ['ASAN_OPTIONS=detect_leaks=0'])
 test('scheduler picker fcfs test', sched_picker_fcfs_test, suite: 'scheduler',
      env: ['ASAN_OPTIONS=detect_leaks=0'])
+test('scheduler picker fcfs session test', sched_picker_fcfs_session_test, suite: 'scheduler',
+     env: ['ASAN_OPTIONS=detect_leaks=0'])

src/scheduler/testsuite/test_picker_fcfs.c

@@ -32,6 +32,27 @@
 #include "plugin.h"
 #include "sched/types.h"
 
+/* builtin_fcfs.c's next_job() (task 2.3) calls sched_queue_is_ready()/
+ * sched_job_is_ready() (declared in scheduler.h, defined in scheduler.c),
+ * which this translation unit does not link -- same as every other
+ * white-box picker test in this directory that #includes builtin_fcfs.c
+ * directly. This file's own tests (tasks 2.1/2.2) never call next_job(), so
+ * a simple always-ready stub (mirroring scheduler.c's own current stub,
+ * scheduler.c:1267/1270) is enough to satisfy the linker;
+ * test_picker_fcfs_session.c is where the real skip-when-not-ready behavior
+ * is exercised, via controllable test doubles. */
+int sched_queue_is_ready(xce_sched_queue_t* q)
+{
+    (void)q;
+    return 1;
+}
+
+int sched_job_is_ready(xce_sched_job_t* j)
+{
+    (void)j;
+    return 1;
+}
+
 #include "../src/plugins/builtin_fcfs.c"
 
 void setUp(void) { return; }

src/scheduler/testsuite/test_picker_fcfs_session.c

@@ -0,0 +1,396 @@
+/*
+ * test_picker_fcfs_session.c - white-box tests for builtin_fcfs's session
+ * hooks (begin_session/next_job/end_session) and post-allocation hooks
+ * (on_dispatch/on_pend) -- scheduler-core spec, Req 3 AC 2's corresponding
+ * hook-table rows, Req 3 AC 4/AC 6.
+ *
+ * Split out of test_picker_fcfs.c (task 2.3, tasks.md): that file already
+ * sits at ~765 lines after tasks 2.1/2.2 (lifecycle + state-change hooks),
+ * against this project's 800-line ceiling (typical 200-400) -- adding this
+ * task's session/post-allocation coverage on top would push meaningfully
+ * past 800. This file is therefore a separate, cohesive translation unit
+ * covering exactly task 2.3's hooks, registered as its own meson test
+ * executable (testsuite/meson.build) alongside (not instead of)
+ * test_picker_fcfs.c.
+ *
+ *   RED:   before task 2.3, plugins/builtin_fcfs.c did not define
+ *          builtin_fcfs_begin_session/builtin_fcfs_next_job/
+ *          builtin_fcfs_end_session/builtin_fcfs_on_dispatch/
+ *          builtin_fcfs_on_pend -- this file failed to compile (implicit
+ *          declaration of every symbol under test).
+ *   GREEN: the hooks implement the walking/cursor/no-op contract Req 3 AC 2
+ *          specifies, verified here by driving them directly (no live
+ *          core) and asserting on next_job()'s actual return sequence.
+ *
+ * #include's the real plugins/builtin_fcfs.c (white-box, same convention as
+ * test_picker_fcfs.c) so this test exercises the real cursor logic and the
+ * real private struct layout, not a re-implementation of either.
+ *
+ * builtin_fcfs.c's next_job() calls sched_queue_is_ready()/
+ * sched_job_is_ready() (declared in scheduler.h, defined in scheduler.c --
+ * design.md: "called as-is by both today's fcfs_elect_job and the new
+ * builtin_fcfs"). This translation unit does not link scheduler.c (same as
+ * test_picker_fcfs.c, which never needed these symbols since tasks 2.1/2.2
+ * never call them), so this file supplies its own test-double definitions
+ * below -- which doubles as the only way to unit-test the "skip a not-ready
+ * queue/job" branch, since scheduler.c's real definitions are permanently
+ * `return 1;` stubs (scheduler.c:1267/1270, FIXME) that could never
+ * exercise the skip path.
+ */
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "plugin.h"
+#include "sched/types.h"
+
+/* ---------------------------------------------------------------------- */
+/* Test doubles for sched_queue_is_ready()/sched_job_is_ready() -- see file
+ * header. Each test that needs to force a skip points the relevant global
+ * at the queue/job under test before calling next_job(), and resets it
+ * (via fixture_init, below) before the next test runs. */
+static xce_sched_queue_t* g_not_ready_queue = NULL;
+static xce_sched_job_t*   g_not_ready_job   = NULL;
+
+int sched_queue_is_ready(xce_sched_queue_t* q)
+{
+    return (q != g_not_ready_queue);
+}
+
+int sched_job_is_ready(xce_sched_job_t* j) { return (j != g_not_ready_job); }
+
+#include "../src/plugins/builtin_fcfs.c"
+
+void setUp(void) { return; }
+void tearDown(void) { return; }
+
+/* ---------------------------------------------------------------------- */
+/* Fixture -- same convention as test_picker_fcfs.c's fixture_init: "picker"
+ * is passed to every hook as this struct's address. Also resets the
+ * sched_queue_is_ready()/sched_job_is_ready() test doubles to "everything
+ * ready", so a skip forced by one test never leaks into the next. Nothing
+ * is torn down between tests (suite convention, see meson.build:
+ * detect_leaks=0). */
+static void fixture_init(xce_sched_plugin_t* plugin)
+{
+    memset(plugin, 0, sizeof(*plugin));
+    g_not_ready_queue = NULL;
+    g_not_ready_job   = NULL;
+}
+
+/* ---------------------------------------------------------------------- */
+/* Task 2.3 Observable: "a full begin_session -> repeated next_job ->
+ * end_session cycle against a fixed candidate set returns jobs in (queue
+ * priority desc, job priority desc, queue_time asc) order and returns NULL
+ * exactly once all candidates are exhausted."
+ *
+ * Fixture: three queues (mid=5, high=9, low=1, pushed out of priority
+ * order) each with two-or-three jobs, including a same-queue priority tie
+ * broken by queue_time asc. Expected next_job() sequence, fully spelled
+ * out against the documented ordering rule:
+ *   high queue (prio 9): hA (prio 9), hB (prio 3)
+ *   mid  queue (prio 5): mTieEarly (prio 4, t=100), mTieLate (prio 4,
+ *                        t=200), mLow (prio 1)
+ *   low  queue (prio 1): lOnly (prio 1)
+ */
+static void test_full_session_cycle_returns_documented_order_then_null(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  qmid  = { .name = "mid", .priority = 5 };
+    xce_sched_queue_t  qhigh = { .name = "high", .priority = 9 };
+    xce_sched_queue_t  qlow  = { .name = "low", .priority = 1 };
+
+    xce_sched_job_t hA = { .jobid      = 101,
+                           .priority   = 9,
+                           .queue_time = 10,
+                           .state      = JOB_STAT_PEND,
+                           .queue      = &qhigh };
+    xce_sched_job_t hB = { .jobid      = 102,
+                           .priority   = 3,
+                           .queue_time = 20,
+                           .state      = JOB_STAT_PEND,
+                           .queue      = &qhigh };
+
+    xce_sched_job_t mTieLate  = { .jobid      = 201,
+                                  .priority   = 4,
+                                  .queue_time = 200,
+                                  .state      = JOB_STAT_PEND,
+                                  .queue      = &qmid };
+    xce_sched_job_t mLow      = { .jobid      = 202,
+                                  .priority   = 1,
+                                  .queue_time = 5,
+                                  .state      = JOB_STAT_PEND,
+                                  .queue      = &qmid };
+    xce_sched_job_t mTieEarly = { .jobid      = 203,
+                                  .priority   = 4,
+                                  .queue_time = 100,
+                                  .state      = JOB_STAT_PEND,
+                                  .queue      = &qmid };
+
+    xce_sched_job_t lOnly = { .jobid      = 301,
+                              .priority   = 1,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &qlow };
+
+    uint64_t         expected[6] = { 101, 102, 203, 201, 202, 301 };
+    size_t           i;
+    xce_sched_job_t* got;
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+
+    /* pushed out of priority order on purpose */
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qmid));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qhigh));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qlow));
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &hA));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &hB));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &mTieLate));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &mLow));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &mTieEarly));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &lOnly));
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+
+    for (i = 0; i < 6; i++) {
+        got = builtin_fcfs_next_job(&plugin);
+        TEST_ASSERT_NOT_NULL(got);
+        TEST_ASSERT_EQUAL_UINT64(expected[i], got->jobid);
+    }
+
+    /* NULL-on-exhaustion, and stays NULL on a repeated call (task
+     * Observable: "returns NULL exactly once all candidates are
+     * exhausted"). */
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_end_session(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* begin_session() with no queues registered at all: next_job() must return
+ * NULL immediately, not crash walking a NULL priv->queues.head. */
+static void test_next_job_returns_null_when_no_queues_registered(void)
+{
+    xce_sched_plugin_t plugin;
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* A registered queue with an empty pendJobs must be walked past (not
+ * returned as a job, not crashed on) straight to NULL. */
+static void test_next_job_returns_null_when_queue_has_no_jobs(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  q1 = { .name = "q1", .priority = 1 };
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1));
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* Req 3 AC 2 "next_job" row: "skip if sched_queue_is_ready() returns
+ * false". Two queues, one forced not-ready via the test double -- its job
+ * must never be returned; the other queue's job still is. */
+static void test_next_job_skips_not_ready_queue(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  qready    = { .name = "ready", .priority = 1 };
+    xce_sched_queue_t  qnotready = { .name = "notready", .priority = 9 };
+    xce_sched_job_t    jready    = { .jobid      = 1,
+                                     .priority   = 1,
+                                     .queue_time = 1,
+                                     .state      = JOB_STAT_PEND,
+                                     .queue      = &qready };
+    xce_sched_job_t    jnotready = { .jobid      = 2,
+                                     .priority   = 1,
+                                     .queue_time = 1,
+                                     .state      = JOB_STAT_PEND,
+                                     .queue      = &qnotready };
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qready));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &qnotready));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &jready));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &jnotready));
+
+    g_not_ready_queue = &qnotready; /* higher priority, would sort first */
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+
+    TEST_ASSERT_EQUAL_UINT64(1, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* Req 3 AC 2 "next_job" row: "Return first job where sched_job_is_ready()
+ * returns true" -- a single not-ready job within an otherwise-ready queue
+ * must be skipped, not stop iteration of the rest of that queue's jobs. */
+static void test_next_job_skips_not_ready_job(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  q1 = { .name = "q1", .priority = 1 };
+    /* jskip outranks jok on priority, so it would be returned FIRST if it
+     * weren't forced not-ready. */
+    xce_sched_job_t jskip = { .jobid      = 1,
+                              .priority   = 9,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+    xce_sched_job_t jok   = { .jobid      = 2,
+                              .priority   = 1,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &jskip));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &jok));
+
+    g_not_ready_job = &jskip;
+
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+
+    TEST_ASSERT_EQUAL_UINT64(2, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* Defensive-argument contract, matching every other hook in this file: NULL
+ * or never-init()'d pickers report -EINVAL/NULL rather than crash. */
+static void test_session_hooks_reject_invalid_picker(void)
+{
+    xce_sched_plugin_t plugin;
+
+    fixture_init(&plugin); /* plugin.data == NULL, never init()'d */
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_begin_session(NULL));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(NULL));
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_end_session(NULL));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_end_session(&plugin));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_dispatch(NULL, 1));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_dispatch(&plugin, 1));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_pend(NULL, 1));
+    TEST_ASSERT_EQUAL_INT(-EINVAL, builtin_fcfs_on_pend(&plugin, 1));
+}
+
+/* Req 3 AC 2 "on_dispatch" row: a dispatched job is permanently gone -- a
+ * later session (begin_session -> next_job) never returns it again, while
+ * the queue's remaining candidate still comes back. */
+static void test_on_dispatch_removes_job_permanently(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  q1 = { .name = "q1", .priority = 1 };
+    xce_sched_job_t    j1 = { .jobid      = 1,
+                              .priority   = 5,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+    xce_sched_job_t    j2 = { .jobid      = 2,
+                              .priority   = 1,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j2));
+
+    /* session 1: j1 elected (higher priority), allocator dispatches it */
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_EQUAL_UINT64(1, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_dispatch(&plugin, 1));
+    TEST_ASSERT_EQUAL_UINT64(2, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_end_session(&plugin));
+
+    /* session 2: j1 must not reappear; j2 (never dispatched) still does */
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_EQUAL_UINT64(2, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_end_session(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+/* Req 3 AC 2 "on_pend" row: a true no-op -- the job reappears in the SAME
+ * position on the very next session (not moved, not dropped, not
+ * duplicated). */
+static void test_on_pend_is_true_noop_job_reappears_same_position(void)
+{
+    xce_sched_plugin_t plugin;
+    xce_sched_queue_t  q1 = { .name = "q1", .priority = 1 };
+    xce_sched_job_t    j1 = { .jobid      = 1,
+                              .priority   = 5,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+    xce_sched_job_t    j2 = { .jobid      = 2,
+                              .priority   = 1,
+                              .queue_time = 1,
+                              .state      = JOB_STAT_PEND,
+                              .queue      = &q1 };
+
+    fixture_init(&plugin);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_init(&plugin, NULL));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_queue_update(&plugin, &q1));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j1));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_job_push(&plugin, &j2));
+
+    /* session 1: j1 elected first, allocator can't place it, pends it */
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_EQUAL_UINT64(1, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_on_pend(&plugin, 1));
+    TEST_ASSERT_EQUAL_UINT64(2, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_end_session(&plugin));
+
+    /* session 2: j1 still comes back first, in the same position */
+    TEST_ASSERT_EQUAL_INT(0, builtin_fcfs_begin_session(&plugin));
+    TEST_ASSERT_EQUAL_UINT64(1, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_EQUAL_UINT64(2, builtin_fcfs_next_job(&plugin)->jobid);
+    TEST_ASSERT_NULL(builtin_fcfs_next_job(&plugin));
+
+    builtin_fcfs_fini(&plugin);
+}
+
+int main(void)
+{
+    UNITY_BEGIN();
+    RUN_TEST(test_full_session_cycle_returns_documented_order_then_null);
+    RUN_TEST(test_next_job_returns_null_when_no_queues_registered);
+    RUN_TEST(test_next_job_returns_null_when_queue_has_no_jobs);
+    RUN_TEST(test_next_job_skips_not_ready_queue);
+    RUN_TEST(test_next_job_skips_not_ready_job);
+    RUN_TEST(test_session_hooks_reject_invalid_picker);
+    RUN_TEST(test_on_dispatch_removes_job_permanently);
+    RUN_TEST(test_on_pend_is_true_noop_job_reappears_same_position);
+
+    return UNITY_END();
+}