feat(scheduler-core): implement scheduler-side admin request handler (task 9.4)

Implements the scheduler's first request-reply opcode (OPCode_schedAdmin).
sched_handle_admin_request() (handler.c) decodes SchedAdminRequest and
dispatches by action: recordStatus answers synchronously (pure read);
recordStart/recordStop are deferred onto the existing pend_updates queue
(XCE_SCHED_OP_ADMIN_RECORD) since sched_record_start()'s snapshot call
requires the session_active==0 precondition only guaranteed on the driver
thread, not the async request-handling thread. driver.c's
xce_sched_driver_translate_pending_events() performs the deferred action
and sends the reply via a new xcecli_reply_schedadmin() (client.c/client.h),
correlated by seq copied before the original request is freed.

Adversarial review found a genuine cross-thread race: sched_record_status()
read driver-owned recording state (including a pointer the driver thread
frees) from the vtask thread with no lock -- the codebase's first such
race, missed by ASan (doesn't detect races). Fixed with a dedicated
record_lock scoped strictly to in-memory field access, never spanning file
I/O; sched_record_status() now strdup()s under the lock instead of
returning a borrowed live pointer. Verified with a real two-thread
lock-contention test (independently confirmed non-vacuous by removing the
lock and observing the test fail) plus a deterministic use-after-free
reproduction.

26/26 scheduler suite, clang-format clean.

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

@@ -242,11 +242,12 @@
   - _Depends: 9.2_
   - _Verified: 2026-08-05 — the write-failure error-handling half was already built in task 9.2 (`sched_record_abort_on_error()`, matching AC 10's clause verbatim: log, fflush+fclose, free+NULL path, recording=false); this task added the missing end-to-end test (`test_write_failure_disables_recording_and_driver_keeps_running`) using a genuine closed-fd write failure (fflush + setvbuf(_IONBF) + close(fd), confirmed by review running the binary directly and observing a real `rc=-5` abort log line, then a real subsequent session completing) rather than a mocked return code. Periodic checkpointing (the new work) hooks into `sched_record_write_events()`: once per call, `record_snapshot_interval > 0 && record_event_count % record_snapshot_interval == 0` triggers a fresh `SnapshotEntry` via the same `write_snapshot_entry()` helper `sched_record_start()` already uses (made non-static, zero duplication) — safe from task 8.2's `session_active` precondition because this call site is only ever reached from `xce_sched_driver_translate_pending_events()`, confirmed the sole production caller and confirmed to always run strictly before `xce_core_begin_session()`. `record_snapshot_interval == 0` genuinely produces zero checkpoints. The "once per call, not once per crossed boundary" semantics matches requirements.md's own embedded pseudocode verbatim, not an implementer shortcut. 13/13 new test binary, 25/25 scheduler suite, clang-format clean — independently re-verified by me via a fresh `meson compile`/`meson test --suite scheduler` run before marking complete. Two follow-ups flagged by review, logged below rather than folded in: `--record-snapshot-interval` has no CLI wiring anywhere (AC 11 cannot be configured non-zero in production as currently scoped — this task's own Observable is satisfied via the struct field directly, which is sufficient for this task, but the operator-facing gap remains); `test_record_lifecycle.c` grew to 1232 lines (826→1232), past this suite's prior precedent, with a natural AC10/AC11 split point identified but not acted on._
 
-- [ ] 9.4 Implement the scheduler-side admin request handler
+- [x] 9.4 Implement the scheduler-side admin request handler
   - Decode a forwarded `SchedAdminRequest` arriving on the scheduler's controller-facing endpoint, dispatch by action to the record-start/stop/status functions built in 9.2/9.3, and encode+send the `SchedAdminReply` back to the controller on the same endpoint — this is the scheduler-side counterpart to the controller's forwarding dispatch (task 10.1), without which 10.1 has nothing to forward to
   - Observable: sending each of the three admin actions to the scheduler's endpoint produces the matching reply (recording state, file, event count, started-at populated correctly for each action)
   - _Requirements: 5.10_
   - _Depends: 9.3_
+  - _Verified: 2026-08-05 — this is the scheduler's first-ever request-reply opcode (`OPCode_schedAdmin`, no prior precedent). `sched_handle_admin_request()` (new, handler.c) decodes `SchedAdminRequest` and dispatches by action: `recordStatus` is answered synchronously (new `sched_record_status()`, recording.c, pure read); `recordStart`/`recordStop` are deferred — enqueued as a new `XCE_SCHED_OP_ADMIN_RECORD` op onto the existing `pend_updates` queue (same mechanism every controller-sourced update already uses) rather than called directly from the request-handling thread, because `sched_record_start()`'s `xce_sched_core_save_snapshot()` call requires task 8.2's `session_active==0` precondition, only guaranteed on the driver thread — confirmed by review as a genuine (not theoretical) concern: `vtaskmgr.c` spawns real pthreads for request handling, distinct from `main.c`'s single driver-loop thread. `driver.c`'s `xce_sched_driver_translate_pending_events()` intercepts the deferred op, performs the real start/stop, and sends the reply via a new `xcecli_reply_schedadmin()` (src/lib/client.c/client.h) correlated by a plain `seq` copied by value before the original request is freed — confirmed safe since `xcecli_ctx_t` is a single unmultiplexed connection. Every reply-field branch (ok/err for each action) independently verified against requirements.md's semantics table, including the subtlety that `recordStop`'s `file` value must be captured before `sched_record_stop()` frees `record_file_path`. Round 1 review found one real bug beyond either implementer session's own testing: `sched_record_status()`, reachable from the vtask thread, read `recording`/`record_file_path`/`record_event_count`/`record_start_ts` with no lock while the driver thread's `sched_record_stop()` mutates and frees those same fields unsynchronized — a genuine cross-thread use-after-free, the codebase's first such race, missed by the ASan-only test config (ASan doesn't catch races). Fixed with a new dedicated `mylock_t record_lock` (deliberately separate from `core.lock`), scoped strictly to the in-memory field reads/writes only (never across `fopen()`/`fwrite()`/capnp I/O — confirmed no vtask-blocks-on-disk-I/O hazard reintroduced), with `sched_record_status()` now `strdup()`ing `record_file_path` while holding the lock rather than returning a borrowed live pointer. Round 2 review independently confirmed the fix: exhaustive grep found every access site now correctly locked (with `recording_session.c`'s driver-thread-only functions and `handle_deferred_admin_record()`'s post-start/stop reads correctly reasoned as safe without the lock, both same-thread-only by construction); confirmed non-recursive `mylock_t` and no nested acquisition (no new deadlock); independently reproduced the fix's necessity by temporarily stripping the lock and confirming the new lock-contention test fails exactly as predicted, then restoring and reconfirming green. 26/26 scheduler suite (14/14 in the new `sched_admin_request_test`), clang-format clean — independently re-verified by me via a fresh `meson compile`/`meson test --suite scheduler` run before marking complete. New `xcecli_reply_schedadmin()` addition to shared `src/lib/client.c`/`src/include/client.h` scoped purely to this one function, confirmed not touching `src/controller/` (task 10.1's exclusive boundary)._
 
 - [ ] 9.5 Recording lifecycle integration tests
   - Test the full start/stop/status round trip against a real filesystem (through the request handler from 9.4, not by calling the internal functions directly), including the path-validation rejection cases and a simulated disk-full condition

src/include/client.h

@@ -133,4 +133,24 @@ extern int             xcecli_jobinfo(xcecli_ctx_t, uint64_t, xce_job_t**);
 extern int xcecli_reply_fetchfile(xcecli_ctx_t, xce_request_t*, xce_fetchfile_reply_t*);
 extern int xcecli_fetchfile(xcecli_ctx_t, xce_fetchfile_t*, xce_fetchfile_reply_t**);
 
+/*
+ * xcecli_reply_schedadmin: send a SchedAdminReply back over `c`'s
+ * connection in response to a previously-received OPCode_schedAdmin
+ * request (scheduler-core spec, task 9.4, Req 5 AC 10). Mirrors
+ * xcecli_reply_fetchfile() above exactly, with one difference: it takes
+ * the original request's `seq` directly rather than an `xce_request_t*`,
+ * since `recordStart`/`recordStop` defer this reply to the driver thread
+ * (see driver.c's xce_sched_driver_translate_pending_events()), by which
+ * point the original xce_request_t has already been freed by the
+ * request-handling thread (xcectx.c's vtask_entry(), immediately after the
+ * registered handler returns) -- seq is the only field of it the reply
+ * needs (opcode is always OPCode_schedAdmin). `recordStatus` (answered
+ * synchronously from the handler thread, real request still in scope)
+ * uses this same function, simply passing `request->seq` directly.
+ * Returns xce_reply()'s own return value; -EINVAL if `c` or `areply` is
+ * NULL.
+ */
+extern int xcecli_reply_schedadmin(xcecli_ctx_t, uint64_t seq,
+                                   sched_admin_reply_t* areply);
+
 #endif

src/lib/client.c

@@ -969,6 +969,40 @@ int xcecli_reply_fetchfile(xcecli_ctx_t c, xce_request_t* request,
     return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply);
 }
 
+/*
+ * xcecli_reply_schedadmin - see client.h's declaration comment for the full
+ * contract (scheduler-core spec, task 9.4, Req 5 AC 10). Builds the
+ * RequestReply envelope directly (no init_reply_by_req(): unlike
+ * xcecli_reply_fetchfile() above, the caller may not have a live
+ * xce_request_t any more -- recordStart/recordStop replies are sent from
+ * the driver thread, after the original request has already been freed).
+ * `reply.status` is always 0: a successfully-built SchedAdminReply is
+ * always a valid RPC response, whether its own `.ok` field is true or
+ * false (business-level success/failure lives inside the SchedAdminReply
+ * payload itself, per its own `ok`/`error` fields -- same convention
+ * handle_config_request() (controller/config_repl.c) uses for
+ * ReplConfigReply's "up to date" case).
+ */
+int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq,
+                            sched_admin_reply_t* areply)
+{
+    api_ctx_t*  ctx = (api_ctx_t*)c;
+    xce_reply_t reply;
+
+    if ((ctx == NULL) || (areply == NULL)) {
+        return -EINVAL;
+    }
+
+    memset(&reply, 0, sizeof(reply));
+    reply.seq                     = seq;
+    reply.opcode                  = OPCode_schedAdmin;
+    reply.status                  = 0;
+    reply.data_which              = RequestReply_data_schedadminreplymsg;
+    reply.data.schedadminreplymsg = areply;
+
+    return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply);
+}
+
 int xcecli_fetchfile(xcecli_ctx_t c, xce_fetchfile_t* msg,
                      xce_fetchfile_reply_t** reply)
 {

src/scheduler/include/recording.h

@@ -170,6 +170,12 @@ extern int sched_validate_record_filename(const char* record_dir,
  *
  * `record_file` may be NULL or empty to auto-generate a timestamped name
  * (`scheduler-replay-<UTC timestamp>.replay`) beneath `sched->record_dir`.
+ *
+ * Concurrency (task 9.4 fix round): step 4's field commit takes
+ * `sched->record_lock` (scheduler.h's field comments) -- see
+ * sched_record_status()'s doc comment below for the cross-thread reader
+ * this protects against. The lock is held only around the five assignments
+ * themselves, never across steps 1-3's file I/O.
  */
 extern int sched_record_start(xce_scheduler_t* sched, const char* record_file);
 
@@ -192,6 +198,13 @@ extern int sched_record_start(xce_scheduler_t* sched, const char* record_file);
  * marker failed to write (state is still fully cleared either way, per the
  * error-handling clause above -- the negative return is diagnostic only,
  * not a signal that recording is still active).
+ *
+ * Concurrency (task 9.4 fix round): the fflush()+fclose()+state-clear
+ * sequence above takes `sched->record_lock` (scheduler.h's field
+ * comments) around the state-clear portion only -- see
+ * sched_record_status()'s doc comment below for the cross-thread reader
+ * this protects against. In particular, freeing `record_file_path` and
+ * setting it NULL happens atomically from that reader's point of view.
  */
 extern int sched_record_stop(xce_scheduler_t* sched);
 
@@ -225,6 +238,13 @@ extern int sched_record_stop(xce_scheduler_t* sched);
  * rationale. A checkpoint write failure disables recording via
  * sched_record_abort_on_error(), the exact same path as any other
  * recording write failure -- never special-cased.
+ *
+ * Concurrency (task 9.4 fix round): the per-event increment of
+ * `sched->record_event_count` takes `sched->record_lock` (scheduler.h's
+ * field comments), scoped to just the increment -- see
+ * sched_record_status()'s doc comment below for the cross-thread reader
+ * this protects against. `record_snapshot_interval` is driver-thread-only
+ * and not protected by this lock (scheduler.h's field comment).
  */
 extern int sched_record_write_events(xce_scheduler_t*  sched,
                                      xce_core_event_t* events, int n);
@@ -283,4 +303,48 @@ extern int sched_record_session_end(xce_scheduler_t*            sched,
                                     sched_record_session_ctx_t* ctx,
                                     uint64_t                    end_ts);
 
+/*
+ * sched_record_status - Req 5 AC 10 (scheduler-core spec, task 9.4,
+ * "Reply field semantics by action" -- the `recordStatus` column). A pure
+ * read of `sched`'s CURRENT recording-state fields into a
+ * `sched_admin_reply_t` the caller (handler.c's scheduler_req_handler())
+ * sends back over the wire immediately -- no file I/O, no state mutation,
+ * and (unlike sched_record_start()/sched_record_stop()) no session_active
+ * dependency, so it is safe to answer directly from the request-handling
+ * thread rather than deferring through the driver-thread pend_updates
+ * queue the way recordStart/recordStop must.
+ *
+ * Concurrency (task 9.4 fix round): this function is called from a vtask
+ * worker pthread (handler.c's scheduler_req_handler(), invoked with no
+ * lock held -- lib/xcectx.c), which runs concurrently with the
+ * single-threaded driver loop's sched_record_start()/sched_record_stop()/
+ * sched_record_write_events()/sched_record_abort_on_error() -- all of
+ * which mutate the same fields this function reads, with no other
+ * synchronization. This function therefore takes `sched->record_lock`
+ * (scheduler.h's field comments) around every read of those fields,
+ * INCLUDING the copy of `record_file_path` described next.
+ *
+ * `reply->file` is a fresh, caller-owned heap copy of
+ * `sched->record_file_path` (via strdup(), taken while still holding
+ * `sched->record_lock` -- copying the bytes before the lock is released is
+ * what makes this safe), NOT a borrowed alias of the live pointer. Unlike
+ * every other capnp encode call in this codebase, this reply crosses a
+ * function-return boundary before it is ever encoded, so a borrowed
+ * pointer into `sched`'s live, mutable state cannot be handed back safely
+ * -- the driver thread could free() it the instant `record_lock` is
+ * released, well before the caller's eventual xcecli_reply_schedadmin()
+ * call. The caller MUST free(reply->file) once it is done (NULL when not
+ * recording -- nothing to free).
+ *
+ * Returns 0 on success, populating every field of `*reply`
+ * (`reply->ok = true` unconditionally). Returns -EINVAL if `sched` or
+ * `reply` is NULL; `*reply` is left untouched in that case. Returns
+ * -ENOMEM if the internal strdup() of `record_file_path` fails; `*reply`
+ * is left zeroed (via the initial memset()) in that case -- this is the
+ * one way reading status can now itself fail, a direct consequence of no
+ * longer handing back a borrowed pointer.
+ */
+extern int sched_record_status(xce_scheduler_t*     sched,
+                               sched_admin_reply_t* reply);
+
 #endif

src/scheduler/include/scheduler.h

@@ -39,7 +39,22 @@ typedef enum {
     /* Notify-only single-job in-place field update. Distinct from
      * XCE_SCHED_OP_UPDATE_JOB (the dataset bulk-create path) so a notify
      * update to an existing job mutates it in place and never creates a job. */
-    XCE_SCHED_OP_UPDATE_JOB_FIELDS
+    XCE_SCHED_OP_UPDATE_JOB_FIELDS,
+    /* Scheduler admin recording control (scheduler-core spec, task 9.4, Req
+     * 5 AC 10): a `recordStart`/`recordStop` SchedAdminRequest, deferred
+     * from the request-handling thread (handler.c's scheduler_req_handler())
+     * to the driver thread. sched_record_start()/sched_record_stop() do
+     * real file I/O and sched_record_start() calls
+     * xce_sched_core_save_snapshot(), which requires the core's
+     * session_active precondition (task 8.2) -- a guarantee only held on
+     * the single-threaded driver loop (main.c), not the async request-
+     * handling thread. See xce_sched_update_request_t's `adminrecord`
+     * member below and driver.c's xce_sched_driver_translate_pending_events()
+     * for the full deferred-dispatch + deferred-reply design.
+     * `recordStatus` is a pure read of already-current struct fields (no
+     * file I/O, no session_active dependency) and is answered synchronously
+     * from the handler thread instead -- it never reaches this queue. */
+    XCE_SCHED_OP_ADMIN_RECORD
 } xce_sched_op_t;
 
 typedef struct {
@@ -79,6 +94,27 @@ typedef struct {
         struct {
             xce_job_rusages_t* rusages;
         } job_rusage;
+        struct {
+            /* SchedAdminAction_recordStart or SchedAdminAction_recordStop
+             * (controller.capnp.h) -- recordStatus never reaches this
+             * queue (answered synchronously, see xce_sched_op_t's own
+             * comment on XCE_SCHED_OP_ADMIN_RECORD above). */
+            int action;
+            /* strdup'd copy of SchedAdminRequest.recordFile for
+             * recordStart (NULL for auto-generate, and always NULL for
+             * recordStop, which ignores it) -- the original xce_request_t
+             * is freed by the handler-thread caller (xcectx.c's
+             * vtask_entry()) immediately after scheduler_req_handler()
+             * returns, so this must be an independent, owned copy for the
+             * driver thread to read later. Freed by the deferred
+             * dispatcher (driver.c) once the action completes. */
+            char* record_file;
+            /* The original request's seq -- the only field of the
+             * now-freed xce_request_t the deferred reply needs
+             * (init_reply_by_req()'s other field, opcode, is always
+             * OPCode_schedAdmin here, so it is not worth storing). */
+            uint64_t seq;
+        } adminrecord;
     } u;
 } xce_sched_update_request_t;
 
@@ -132,27 +168,63 @@ typedef struct {
      * writes -- see xce_sched_driver_translate_pending_events()'s and
      * xce_sched_driver_run_normal_mode_round()'s own comments.
      *
-     * Invariant: recording == true iff record_file != NULL. */
+     * Invariant: recording == true iff record_file != NULL.
+     *
+     * Concurrency (scheduler-core spec, task 9.4 fix round): `recording`,
+     * `record_file_path`, `record_file`, `record_event_count`, and
+     * `record_start_ts` are written on the single-threaded driver loop
+     * (main.c) but ALSO read by sched_record_status() (recording.c), which
+     * runs on whichever vtask worker pthread handled the synchronous
+     * `recordStatus` SchedAdminRequest (handler.c's
+     * scheduler_req_handler(), invoked with no lock held --
+     * lib/xcectx.c). That makes this a genuine cross-thread reader/writer
+     * pair with no other synchronization, so every read or write of these
+     * five fields MUST hold `record_lock` below -- see its own comment. */
     bool recording;
     /* strdup'd copy of the resolved record file path; owned by the record
      * file lifecycle (strdup on start, free+NULL on stop or start/write
-     * failure) -- lets a future status query (task 9.4, out of this task's
+     * failure) -- lets a status query (task 9.4) answer with the current
-     * scope) answer with the current path even after record_file closes. */
+     * path even after record_file closes. Protected by `record_lock` (see
+     * `recording`'s comment above and `record_lock`'s own comment below) --
+     * sched_record_status() must copy this string's bytes (e.g. strdup())
+     * while still holding the lock rather than handing back the live
+     * pointer, since the driver thread may free() it the instant the lock
+     * is released. */
     char* record_file_path;
-    /* replay file handle; NULL when not recording. */
+    /* replay file handle; NULL when not recording. Protected by
+     * `record_lock` for the same reason as `record_file_path` above. */
     FILE* record_file;
     /* count of ReplayEvent messages written since recording started (used
      * by task 9.3's periodic-checkpoint trigger; incremented by
-     * recording.c's event-batch writer, task 9.2). */
+     * recording.c's event-batch writer, task 9.2). Protected by
+     * `record_lock`. */
     uint64_t record_event_count;
     /* N events between periodic snapshot checkpoints (0 = disabled). Read
      * by recording_events.c's sched_record_write_events() (task 9.3):
      * every time cumulative `record_event_count` lands exactly on a
      * multiple of this value, a fresh SnapshotEntry is appended to the
-     * record file (Req 5 AC 11). */
+     * record file (Req 5 AC 11). Driver-thread-only (never read by
+     * sched_record_status()) -- not protected by `record_lock`. */
     uint64_t record_snapshot_interval;
-    /* logical time (xce_core_now()) recording started; 0 = not recording. */
+    /* logical time (xce_core_now()) recording started; 0 = not recording.
+     * Protected by `record_lock`. */
     uint64_t record_start_ts;
+    /* Dedicated mutex (scheduler-core spec, task 9.4 fix round) guarding
+     * `recording`/`record_file_path`/`record_file`/`record_event_count`/
+     * `record_start_ts` -- see `recording`'s comment above for the
+     * cross-thread reader/writer hazard this closes. Deliberately a NEW,
+     * separate lock rather than reusing `core.lock`: `core`
+     * (xce_sched_core_t) is the conceptually distinct core-vs-driver piece
+     * of state this struct's own header comment describes, and these
+     * recording fields are driver-level (like `record_dir`/`apictx`/
+     * `cluster_cfg` above), so widening core.lock's scope to also cover
+     * them would blur that boundary for no benefit. Held only around the
+     * struct-field reads/writes themselves -- never across the actual file
+     * I/O (fopen()/fwrite()/capn_write_mem() etc.) -- matching this
+     * codebase's low-contention-lock-scope convention. Initialized by
+     * sched_init(), destroyed by sched_fini() (scheduler.c) -- same
+     * lifecycle pairing as `core.lock` (core.c). */
+    mylock_t record_lock;
 } xce_scheduler_t;
 
 extern int  sched_init(xce_scheduler_t*, xce_client_config_t*,

src/scheduler/src/driver.c

@@ -1,4 +1,6 @@
 #include <errno.h>
+#include <limits.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -281,6 +283,163 @@ static int translate_one_pending_update(xce_sched_core_t*           core,
     }
 }
 
+/*
+ * validation_error_text - classifies a sched_record_start() failure rc into
+ * the AC 10 error-codes-table bucket it belongs to (scheduler-core spec,
+ * task 9.4). sched_record_start() calls sched_validate_record_filename()
+ * (recording.c, task 9.1) FIRST, before ever touching the filesystem for
+ * real -- that function's own doc comment documents its complete return
+ * set as exactly these six codes, so a match here means the failure was a
+ * REJECTED record file NAME, never an I/O failure. Returns NULL for any
+ * other rc (the caller then falls back to the disk/write-failure message).
+ * `-EALREADY` (already recording) is handled by the caller before this is
+ * ever consulted -- not one of this function's cases.
+ */
+static char* validation_error_text(int rc)
+{
+    switch (rc) {
+    case -EINVAL:
+    case -ENAMETOOLONG:
+    case -ENOENT:
+    case -ENOTDIR:
+    case -EACCES:
+    case -EEXIST:
+        return "invalid record file name";
+    default:
+        return NULL;
+    }
+}
+
+/*
+ * handle_deferred_admin_record - performs the ACTUAL sched_record_start()/
+ * sched_record_stop() call the request-handling thread (handler.c's
+ * sched_handle_admin_request()) deferred onto this queue, then sends the
+ * SchedAdminReply the original caller is still waiting on (scheduler-core
+ * spec, task 9.4, Req 5 AC 10's "Reply field semantics by action" table).
+ *
+ * `record_ctx` is `xce_sched_driver_run_normal_mode_round()`'s own `sched`
+ * parameter, threaded through unchanged (the SAME opaque void* task 9.2's
+ * per-event-batch recording-write hook already uses -- see this file's
+ * translate-loop comment). A NULL `record_ctx` (a caller that does not
+ * want recording, e.g. some unit tests) means there is no `xcecli_ctx_t`
+ * to reply on either -- this is logged loudly and the (still fully
+ * owned/allocated) request is simply dropped: the enqueuing caller was
+ * always guaranteed a NULL-record_ctx round would never legitimately
+ * enqueue this op in the first place (only handler.c's real production
+ * wiring does, and main.c always passes the real `sched`).
+ *
+ * Field-population strategy: every reply field except `ok`/`error` is
+ * populated by reading `sched`'s CURRENT state (recording/record_file_path/
+ * record_event_count/record_start_ts) strictly AFTER the
+ * sched_record_start()/sched_record_stop() call returns. This single
+ * "always snapshot post-call" rule is deliberately uniform across every
+ * outcome (success, precondition failure, validation failure, I/O
+ * failure): sched_record_start()/sched_record_stop() never partially
+ * mutate state on any failure path (recording.c's own doc comments on
+ * each), so reading post-call state naturally yields the table's "current"
+ * values on success and its "(unchanged)" values on failure -- no
+ * per-outcome branching is needed for those four fields, only for
+ * `ok`/`error`.
+ *
+ * One exception each side must still capture BEFORE the call, since the
+ * function itself clears it as part of a successful call:
+ *   - recordStop(ok): `file` must report the JUST-ENDED recording's path
+ *     ("last path (informational)") -- but sched_record_stop() frees +
+ *     NULLs sched->record_file_path unconditionally as part of its own
+ *     state-clearing (recording.c), so it is snapshotted into a local
+ *     buffer BEFORE the call.
+ *
+ * recordStop's only documented failure in the AC 10 error-codes table is
+ * "not recording" (rc == -EALREADY). A negative rc from a genuine
+ * end-of-recording-marker WRITE failure is treated as a successful stop
+ * for reply purposes: recording.h's own doc comment on sched_record_stop()
+ * says state "is still fully cleared either way ... the negative return is
+ * diagnostic only, not a signal that recording is still active" -- the
+ * admin caller asked for a stop, and a stop genuinely happened.
+ */
+static void handle_deferred_admin_record(void*                       record_ctx,
+                                         xce_sched_update_request_t* req)
+{
+    xce_scheduler_t*    sched = (xce_scheduler_t*)record_ctx;
+    sched_admin_reply_t reply;
+    char                iobuf[256];
+    int                 rc;
+
+    memset(&reply, 0, sizeof(reply));
+
+    if (sched == NULL) {
+        ERROR("%s admin record request (action=%d) enqueued with no "
+              "record_ctx -- cannot execute or reply",
+              __func__, req->u.adminrecord.action);
+        free(req->u.adminrecord.record_file);
+        return;
+    }
+
+    if (req->u.adminrecord.action == SchedAdminAction_recordStart) {
+        rc = sched_record_start(sched, req->u.adminrecord.record_file);
+
+        if (rc == 0) {
+            reply.ok = 1;
+        } else if (rc == -EALREADY) {
+            reply.ok    = 0;
+            reply.error = "already recording";
+        } else {
+            char* verr = validation_error_text(rc);
+
+            reply.ok = 0;
+            if (verr != NULL) {
+                reply.error = verr;
+            } else {
+                snprintf(iobuf, sizeof(iobuf), "cannot write to %s: %s",
+                         (req->u.adminrecord.record_file != NULL)
+                             ? req->u.adminrecord.record_file
+                             : "(auto-generated)",
+                         strerror(-rc));
+                reply.error = iobuf;
+            }
+        }
+
+        reply.recording  = sched->recording ? 1 : 0;
+        reply.file       = sched->record_file_path;
+        reply.eventCount = sched->record_event_count;
+        reply.startedAt  = sched->record_start_ts;
+
+        xcecli_reply_schedadmin(sched->apictx, req->u.adminrecord.seq, &reply);
+    } else if (req->u.adminrecord.action == SchedAdminAction_recordStop) {
+        char last_path[PATH_MAX];
+
+        last_path[0] = '\0';
+        if (sched->record_file_path != NULL) {
+            strncpy(last_path, sched->record_file_path, sizeof(last_path) - 1);
+            last_path[sizeof(last_path) - 1] = '\0';
+        }
+
+        rc = sched_record_stop(sched);
+
+        if (rc == -EALREADY) {
+            reply.ok         = 0;
+            reply.error      = "not recording";
+            reply.recording  = 0;
+            reply.file       = NULL;
+            reply.eventCount = 0;
+            reply.startedAt  = 0;
+        } else {
+            reply.ok         = 1;
+            reply.recording  = sched->recording ? 1 : 0;
+            reply.file       = (last_path[0] != '\0') ? last_path : NULL;
+            reply.eventCount = sched->record_event_count;
+            reply.startedAt  = sched->record_start_ts;
+        }
+
+        xcecli_reply_schedadmin(sched->apictx, req->u.adminrecord.seq, &reply);
+    } else {
+        ERROR("%s unrecognized deferred admin action %d", __func__,
+              req->u.adminrecord.action);
+    }
+
+    free(req->u.adminrecord.record_file);
+}
+
 int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
                                               void*             record_ctx)
 {
@@ -321,6 +480,20 @@ int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
         xce_sched_update_request_t* req = ITPTR_TO_UPDATE_REQ(it);
         xce_core_event_t            ev;
 
+        /* Req 5 AC 10 (scheduler-core spec, task 9.4): recording admin
+         * control is a driver-LEVEL concern, not a core scheduling event
+         * -- "the core has zero knowledge of whether recording is active"
+         * (requirements.md). Intercepted here, before
+         * translate_one_pending_update()/xce_core_push_events() ever see
+         * it, so it is never mistaken for (or batched alongside) a real
+         * xce_core_event_t. See handle_deferred_admin_record()'s own
+         * comment for the full deferred-dispatch-and-reply contract. */
+        if (req->op == XCE_SCHED_OP_ADMIN_RECORD) {
+            handle_deferred_admin_record(record_ctx, req);
+            free(req);
+            continue;
+        }
+
         if (!translate_one_pending_update(core, req, now, &ev)) {
             release_translated_payload(core, req, 0);
             free(req);

src/scheduler/src/handler.c

@@ -10,6 +10,7 @@
 #include "helper.h"
 #include "xce/types.h"
 
+#include "recording.h"
 #include "scheduler.h"
 #include "utils.h"
 
@@ -68,14 +69,14 @@ static job_startinfo_t* clone_job_startinfo(const job_startinfo_t* src)
         return NULL;
     }
 
-#define CLONE_STR_FIELD(f)                                                    \
+#define CLONE_STR_FIELD(f)                                                     \
-    do {                                                                      \
+    do {                                                                       \
-        if (src->f != NULL) {                                                 \
+        if (src->f != NULL) {                                                  \
-            d->f = strdup(src->f);                                           \
+            d->f = strdup(src->f);                                             \
-            if (d->f == NULL) {                                              \
+            if (d->f == NULL) {                                                \
-                goto fail;                                                    \
+                goto fail;                                                     \
-            }                                                                 \
+            }                                                                  \
-        }                                                                     \
+        }                                                                      \
     } while (0)
 
     CLONE_STR_FIELD(exechome);
@@ -114,14 +115,14 @@ static job_submit_data_t* clone_job_submit_data(const job_submit_data_t* src)
     d->mintasks = src->mintasks;
     d->maxtasks = src->maxtasks;
 
-#define CLONE_STR_FIELD(f)                                                    \
+#define CLONE_STR_FIELD(f)                                                     \
-    do {                                                                      \
+    do {                                                                       \
-        if (src->f != NULL) {                                                 \
+        if (src->f != NULL) {                                                  \
-            d->f = strdup(src->f);                                           \
+            d->f = strdup(src->f);                                             \
-            if (d->f == NULL) {                                              \
+            if (d->f == NULL) {                                                \
-                goto fail;                                                    \
+                goto fail;                                                     \
-            }                                                                 \
+            }                                                                  \
-        }                                                                     \
+        }                                                                      \
     } while (0)
 
     CLONE_STR_FIELD(queue);
@@ -175,7 +176,7 @@ static xce_job_t* make_array_element(const xce_job_submit_t* job_array,
     return j;
 }
 
-static int sched_on_job_array(xce_scheduler_t* sched,
+static int sched_on_job_array(xce_scheduler_t*  sched,
                               xce_job_submit_t* job_array)
 {
     xce_job_t** jobs;
@@ -236,8 +237,8 @@ static int sched_on_job_array(xce_scheduler_t* sched,
 
     fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid);
     INFO("job array push received and expanded: base_jobid=%s "
-        "array_size=%llu new_tasks=%d",
+         "array_size=%llu new_tasks=%d",
-        jobidString, (unsigned long long)array_size, n_new);
+         jobidString, (unsigned long long)array_size, n_new);
 
     rc = sched_on_jobs(sched, n_new, jobs);
     if (rc < 0) {
@@ -286,9 +287,9 @@ static int sched_handle_resources(xce_scheduler_t* sched, int count,
         return -ENOMEM;
     }
 
-    req->op                           = XCE_SCHED_OP_UPDATE_RESOURCES;
+    req->op                          = XCE_SCHED_OP_UPDATE_RESOURCES;
-    req->u.resourceupdate.nresources  = count;
+    req->u.resourceupdate.nresources = count;
-    req->u.resourceupdate.resources   = resources;
+    req->u.resourceupdate.resources  = resources;
 
     enq_pend_update(sched, req);
 
@@ -309,9 +310,9 @@ static int sched_handle_resvals(xce_scheduler_t* sched, int count,
         return -ENOMEM;
     }
 
-    req->op                        = XCE_SCHED_OP_UPDATE_RESVALS_FULL;
+    req->op                       = XCE_SCHED_OP_UPDATE_RESVALS_FULL;
-    req->u.resvalfullupdate.nvals  = count;
+    req->u.resvalfullupdate.nvals = count;
-    req->u.resvalfullupdate.vals   = vals;
+    req->u.resvalfullupdate.vals  = vals;
 
     enq_pend_update(sched, req);
 
@@ -400,8 +401,7 @@ static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
     return 0;
 }
 
-static int sched_handle_sched_update_job(xce_scheduler_t* sched,
+static int sched_handle_sched_update_job(xce_scheduler_t* sched, xce_job_t* job)
-                                         xce_job_t*        job)
 {
     xce_sched_update_request_t* req;
     char                        jobidString[128];
@@ -422,8 +422,8 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched,
      * (XCE_SCHED_OP_UPDATE_JOB_FIELDS), NOT the dataset bulk-create path
      * (XCE_SCHED_OP_UPDATE_JOB). An unknown jobid must yield -ENOENT, never a
      * new job. The dataset bulk-create path (sched_on_jobs) is unchanged. */
-    req->op                     = XCE_SCHED_OP_UPDATE_JOB_FIELDS;
+    req->op                    = XCE_SCHED_OP_UPDATE_JOB_FIELDS;
-    req->u.jobfieldsupdate.job  = job;
+    req->u.jobfieldsupdate.job = job;
 
     enq_pend_update(sched, req);
 
@@ -431,7 +431,7 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched,
 }
 
 static int sched_handle_sched_update_queue(xce_scheduler_t* sched,
-                                           xce_queue_t*      queue)
+                                           xce_queue_t*     queue)
 {
     xce_sched_update_request_t* req;
 
@@ -444,8 +444,8 @@ static int sched_handle_sched_update_queue(xce_scheduler_t* sched,
         return -ENOMEM;
     }
 
-    req->op                    = XCE_SCHED_OP_UPDATE_QUEUE;
+    req->op                  = XCE_SCHED_OP_UPDATE_QUEUE;
-    req->u.queueupdate.queue   = queue;
+    req->u.queueupdate.queue = queue;
 
     enq_pend_update(sched, req);
 
@@ -466,8 +466,8 @@ static int sched_handle_sched_update_resvals(xce_scheduler_t*   sched,
         return -ENOMEM;
     }
 
-    req->op                      = XCE_SCHED_OP_UPDATE_RESVALS;
+    req->op                     = XCE_SCHED_OP_UPDATE_RESVALS;
-    req->u.resvalupdate.resvals  = resvals;
+    req->u.resvalupdate.resvals = resvals;
 
     enq_pend_update(sched, req);
 
@@ -475,7 +475,7 @@ static int sched_handle_sched_update_resvals(xce_scheduler_t*   sched,
 }
 
 static int sched_handle_sched_job_rusage(xce_scheduler_t*   sched,
-                                          xce_job_rusages_t* rusages)
+                                         xce_job_rusages_t* rusages)
 {
     xce_sched_update_request_t* req;
 
@@ -488,8 +488,8 @@ static int sched_handle_sched_job_rusage(xce_scheduler_t*   sched,
         return -ENOMEM;
     }
 
-    req->op                    = XCE_SCHED_OP_JOB_RUSAGE;
+    req->op                   = XCE_SCHED_OP_JOB_RUSAGE;
-    req->u.job_rusage.rusages  = rusages;
+    req->u.job_rusage.rusages = rusages;
 
     enq_pend_update(sched, req);
 
@@ -547,6 +547,111 @@ int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
     return rc;
 }
 
+/*
+ * sched_handle_admin_request - decodes and dispatches an OPCode_schedAdmin
+ * request (scheduler-core spec, task 9.4, Req 5 AC 10). This is the
+ * scheduler-side counterpart to task 10.1's controller-side forwarding
+ * dispatch: the controller relays a SchedAdminRequest it received from
+ * xceadmin, verbatim, to this endpoint; this function is where it is
+ * finally interpreted.
+ *
+ * `recordStatus` is a pure read (recording.h's sched_record_status()) with
+ * no file I/O and no session_active dependency -- it is safe to answer
+ * directly, synchronously, on this (request-handling) thread, and does so
+ * via xcecli_reply_schedadmin() while `request` is still fully valid.
+ *
+ * `recordStart`/`recordStop` do real file I/O and (recordStart) call
+ * xce_sched_core_save_snapshot(), which requires the core's
+ * session_active == 0 precondition (task 8.2) -- a guarantee only held on
+ * the single-threaded driver loop (main.c), never on this async
+ * request-handling thread (tasks.md's Implementation Notes, "Cross-task
+ * interaction" entry preceding this task). So neither the action nor its
+ * reply can happen here: this function enqueues an
+ * XCE_SCHED_OP_ADMIN_RECORD request onto the SAME pend_updates queue every
+ * other controller-sourced update already flows through (enq_pend_update()
+ * above), captured with just enough context (action, a strdup'd copy of
+ * recordFile, and the request's seq) to survive past this function
+ * returning and `request` being freed by the caller (xcectx.c's
+ * vtask_entry(), immediately after this handler returns). Actually
+ * performing the deferred action and sending its reply is driver.c's
+ * xce_sched_driver_translate_pending_events()'s job -- see that file's own
+ * comment on its XCE_SCHED_OP_ADMIN_RECORD interception.
+ *
+ * Returns 0 on successful dispatch (synchronous reply sent, or the
+ * deferred request successfully enqueued); a negative errno otherwise
+ * (-EINVAL for a NULL argument/missing body/unrecognized action, -ENOMEM
+ * on an allocation failure). Mirrors every other case in this function:
+ * the return value is for internal logging only (vtask_entry(), xcectx.c,
+ * discards it) -- it is NOT itself what causes a reply to be sent.
+ */
+static int sched_handle_admin_request(xce_scheduler_t* sched,
+                                      xce_request_t*   request)
+{
+    sched_admin_req_t* areq;
+
+    if ((sched == NULL) || (request == NULL)
+        || (request->body.schedadminmsg == NULL)) {
+        return -EINVAL;
+    }
+
+    areq = request->body.schedadminmsg;
+
+    switch (areq->action) {
+    case SchedAdminAction_recordStatus: {
+        sched_admin_reply_t reply;
+        int                 rc = sched_record_status(sched, &reply);
+        int                 send_rc;
+
+        if (rc < 0) {
+            return rc;
+        }
+
+        /* recording.h's sched_record_status() doc comment (task 9.4 fix
+         * round): reply.file is now a fresh strdup()'d copy (or NULL), not
+         * a borrowed alias of sched->record_file_path -- caller-owned, so
+         * it must be freed here once the encode call above is done with
+         * it. */
+        send_rc = xcecli_reply_schedadmin(sched->apictx, request->seq, &reply);
+        free(reply.file);
+
+        return send_rc;
+    }
+
+    case SchedAdminAction_recordStart:
+    case SchedAdminAction_recordStop: {
+        xce_sched_update_request_t* req;
+
+        req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
+        if (req == NULL) {
+            return -ENOMEM;
+        }
+
+        req->op                   = XCE_SCHED_OP_ADMIN_RECORD;
+        req->u.adminrecord.action = areq->action;
+        req->u.adminrecord.seq    = request->seq;
+
+        if ((areq->action == SchedAdminAction_recordStart)
+            && (areq->recordFile != NULL) && (areq->recordFile[0] != '\0')) {
+            req->u.adminrecord.record_file = strdup(areq->recordFile);
+            if (req->u.adminrecord.record_file == NULL) {
+                free(req);
+                return -ENOMEM;
+            }
+        } else {
+            req->u.adminrecord.record_file = NULL;
+        }
+
+        enq_pend_update(sched, req);
+
+        return 0;
+    }
+
+    default:
+        WARN("%s unrecognized SchedAdminAction %d", __func__, areq->action);
+        return -EINVAL;
+    }
+}
+
 int scheduler_req_handler(xce_request_t* request, void* args)
 {
     int              rc    = 0;
@@ -579,6 +684,9 @@ int scheduler_req_handler(xce_request_t* request, void* args)
     case OPCode_notify:
         rc = sched_handle_notify(sched, request->body.notifymsg);
         break;
+    case OPCode_schedAdmin:
+        rc = sched_handle_admin_request(sched, request);
+        break;
     default:
         rc = -EINVAL;
         break;

src/scheduler/src/recording.c

@@ -484,11 +484,19 @@ int sched_record_start(xce_scheduler_t* sched, const char* record_file)
         return rc;
     }
 
+    /* record_lock (scheduler.h field comments, task 9.4 fix round): guards
+     * this commit against sched_record_status(), which can run
+     * concurrently on a vtask worker thread. Scope is deliberately just
+     * these five assignments -- every fopen()/write_replay_header()/
+     * write_snapshot_entry() I/O call above has already completed by this
+     * point, so the lock never spans file I/O. */
+    mylock_lock(&(sched->record_lock));
     sched->record_file        = f;
     sched->record_file_path   = strdup(resolved_path);
     sched->recording          = true;
     sched->record_event_count = 0;
     sched->record_start_ts    = start_ts;
+    mylock_unlock(&(sched->record_lock));
 
     return 0;
 }
@@ -529,13 +537,28 @@ int sched_record_stop(xce_scheduler_t* sched)
 
     /* Clear recording state unconditionally -- Req 5 AC 10's "Error
      * handling" clause: a failed write disables recording rather than
-     * leaving it half-active. */
+     * leaving it half-active. fflush()/fclose() run BEFORE record_lock is
+     * taken (task 9.4 fix round) -- they are file I/O on `record_file`,
+     * never contended by sched_record_status() (which never touches
+     * `record_file` itself, only reads the OTHER four fields under the
+     * lock), so there is nothing to protect here and no reason to hold the
+     * lock across them. */
     fflush(sched->record_file);
     fclose(sched->record_file);
+
+    /* record_lock (scheduler.h field comments): guards the four-field
+     * commit below against sched_record_status(), which can run
+     * concurrently on a vtask worker thread. In particular, `free()`ing
+     * record_file_path and then setting it NULL must happen atomically
+     * from that reader's point of view -- otherwise a concurrent
+     * sched_record_status() could read the freed pointer between the two
+     * statements (the exact use-after-free this fix closes). */
+    mylock_lock(&(sched->record_lock));
     sched->record_file = NULL;
     sched->recording   = false;
     free(sched->record_file_path);
     sched->record_file_path = NULL;
+    mylock_unlock(&(sched->record_lock));
 
     return rc;
 }
@@ -550,20 +573,98 @@ int sched_record_stop(xce_scheduler_t* sched)
  * already broken -- disk full, permission revoked mid-write, etc. -- so
  * attempting one more write is not productive).
  */
+/*
+ * sched_record_status - see recording.h's doc comment for the full
+ * contract (scheduler-core spec, task 9.4, Req 5 AC 10). A pure read of
+ * `sched`'s current recording-state fields -- no file I/O, no mutation.
+ * `reply->file` deliberately aliases `sched->record_file_path` rather than
+ * copying it (recording.h documents the borrowed-pointer lifetime the
+ * caller must respect).
+ */
+int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
+{
+    char* file_copy;
+    int   had_path;
+
+    if ((sched == NULL) || (reply == NULL)) {
+        return -EINVAL;
+    }
+
+    memset(reply, 0, sizeof(*reply));
+
+    /* record_lock (scheduler.h field comments, task 9.4 fix round): this
+     * function runs on a vtask worker pthread (handler.c's
+     * scheduler_req_handler(), invoked with no lock held -- lib/xcectx.c),
+     * concurrently with the driver thread's sched_record_start()/
+     * sched_record_stop()/sched_record_abort_on_error(), which mutate
+     * these same five fields with no other synchronization. The lock must
+     * stay held through the strdup() below, not just the four field reads
+     * -- copying record_file_path's BYTES while the lock is held is what
+     * makes the copy safe; handing back sched->record_file_path itself
+     * (the pre-fix behavior) would let the driver thread free() it out
+     * from under the caller the instant this function returns. */
+    mylock_lock(&(sched->record_lock));
+
+    reply->recording  = sched->recording ? 1 : 0;
+    reply->eventCount = sched->record_event_count;
+    reply->startedAt  = sched->record_start_ts;
+
+    had_path  = (sched->record_file_path != NULL);
+    file_copy = had_path ? strdup(sched->record_file_path) : NULL;
+
+    mylock_unlock(&(sched->record_lock));
+
+    /* strdup() failure (ENOMEM) is the one way this now-copying function
+     * can fail where the old borrowed-pointer version could not -- surface
+     * it rather than silently returning a status reply with a missing
+     * `file` field for a recording that is genuinely active. */
+    if (had_path && (file_copy == NULL)) {
+        return -ENOMEM;
+    }
+
+    reply->ok    = 1;
+    reply->error = NULL;
+    /* Ownership: `reply->file` is now a caller-owned heap allocation (NULL
+     * when not recording) -- unlike every other capnp-encode call in this
+     * codebase, this reply crosses a function-return boundary (back to
+     * handler.c's sched_handle_admin_request()) before it is ever encoded,
+     * so a borrowed pointer into `sched`'s live state cannot be made safe
+     * here the way handle_deferred_admin_record()'s recordStop path (a
+     * same-stack-frame local buffer) manages it. The caller MUST
+     * free(reply->file). */
+    reply->file = file_copy;
+
+    return 0;
+}
+
 void sched_record_abort_on_error(xce_scheduler_t* sched, int rc)
 {
+    FILE* f;
+
     ERROR("%s scheduler recording write to %s failed (rc=%d) -- disabling "
           "recording",
           __func__,
           (sched->record_file_path != NULL) ? sched->record_file_path : "?",
           rc);
 
-    if (sched->record_file != NULL) {
+    /* fflush()/fclose() happen BEFORE record_lock is taken, same rationale
-        fflush(sched->record_file);
+     * as sched_record_stop()'s identical ordering above -- `record_file`
-        fclose(sched->record_file);
+     * itself is never read by sched_record_status(), so there is nothing
-        sched->record_file = NULL;
+     * to protect, and the lock must not span file I/O. */
+    f = sched->record_file;
+    if (f != NULL) {
+        fflush(f);
+        fclose(f);
     }
+
+    /* record_lock (scheduler.h field comments): same atomicity requirement
+     * as sched_record_stop()'s identical commit -- free()ing
+     * record_file_path and setting it NULL must be indivisible from
+     * sched_record_status()'s point of view. */
+    mylock_lock(&(sched->record_lock));
+    sched->record_file = NULL;
     free(sched->record_file_path);
     sched->record_file_path = NULL;
     sched->recording        = false;
+    mylock_unlock(&(sched->record_lock));
 }

src/scheduler/src/recording_events.c

@@ -39,9 +39,16 @@ extern int  write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts);
  * unconditionally (cheap stack space) so its address stays valid across
  * the single encode+write call after the switch, regardless of which case
  * populated it.
+ *
+ * Takes `sched` (rather than a bare `uint64_t*` counter, task 9.4 fix
+ * round) so the post-write increment of `sched->record_event_count` can be
+ * taken under `sched->record_lock` -- that field is read concurrently by
+ * sched_record_status() on a vtask worker thread (scheduler.h's field
+ * comment). The lock is held ONLY around the increment itself, after the
+ * encode+write I/O above has already completed.
  */
 static int write_one_replay_event(FILE* f, xce_core_event_t* ev,
-                                  uint64_t* event_count)
+                                  xce_scheduler_t* sched)
 {
     struct capn          c;
     struct capn_segment* cs;
@@ -151,7 +158,9 @@ static int write_one_replay_event(FILE* f, xce_core_event_t* ev,
     capn_free(&c);
 
     if (rc == 0) {
-        (*event_count)++;
+        mylock_lock(&(sched->record_lock));
+        sched->record_event_count++;
+        mylock_unlock(&(sched->record_lock));
     }
 
     return rc;
@@ -202,8 +211,7 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
     }
 
     for (i = 0; i < n; i++) {
-        int rc = write_one_replay_event(sched->record_file, &events[i],
+        int rc = write_one_replay_event(sched->record_file, &events[i], sched);
-                                        &(sched->record_event_count));
         if (rc < 0) {
             sched_record_abort_on_error(sched, rc);
             return rc;

src/scheduler/src/scheduler.c

@@ -477,7 +477,7 @@ static int sched_register_plugins(xce_scheduler_t* sched)
 
 int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
                const xce_cluster_config_t* cluster_cfg,
-               xce_request_handler_t handler)
+               xce_request_handler_t       handler)
 {
     int rc = 0;
     /* Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, task 2.4): resolved
@@ -493,6 +493,17 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
 
     memset(sched, 0x0, sizeof(*sched));
 
+    /* scheduler-core spec, task 9.4 fix round: record_lock guards the
+     * recording-state fields below record_dir (scheduler.h's field
+     * comments) against the vtask-thread sched_record_status() reader /
+     * driver-thread writer race. Initialized here (paired with
+     * mylock_fini() in sched_fini()) the same way core.lock is initialized
+     * by xce_sched_core_init() just below -- must happen before any thread
+     * that could call sched_record_status() (the vtask workers, created far
+     * later by lib/xcectx.c once this scheduler registers its handler) can
+     * possibly run. */
+    mylock_init(&(sched->record_lock));
+
     if (cluster_cfg != NULL) {
         sched->cluster_cfg = *cluster_cfg;
     }
@@ -651,6 +662,10 @@ void sched_fini(xce_scheduler_t* sched)
      * same order as before this task, just relocated. */
     xce_sched_core_fini(&(sched->core));
 
+    /* scheduler-core spec, task 9.4 fix round: pairs with sched_init()'s
+     * mylock_init(&(sched->record_lock)) above. */
+    mylock_fini(&(sched->record_lock));
+
     return;
 }
 
@@ -692,8 +707,8 @@ static int sched_init_sharedInst(xce_scheduler_t* sched)
 
         if (key_find(sched->core.resmgr, (void*)nv->name, (void**)&r) < 0
             || r == NULL) {
-            WARN("%s unknown resource '%s' in resvals, skipping",
+            WARN("%s unknown resource '%s' in resvals, skipping", __func__,
-                 __func__, nv->name);
+                 nv->name);
             continue;
         }
 
@@ -793,18 +808,30 @@ static char* sched_op_name(xce_sched_op_t op)
     static char buf[32];
 
     switch (op) {
-    case XCE_SCHED_OP_NONE:             return "NONE";
+    case XCE_SCHED_OP_NONE:
-    case XCE_SCHED_OP_UPDATE_CONFIG:    return "UPDATE_CONFIG";
+        return "NONE";
-    case XCE_SCHED_OP_UPDATE_JOB:       return "UPDATE_JOB";
+    case XCE_SCHED_OP_UPDATE_CONFIG:
-    case XCE_SCHED_OP_JSTATE_CHANGE:    return "JSTATE_CHANGE";
+        return "UPDATE_CONFIG";
-    case XCE_SCHED_OP_UPDATE_WORKER:    return "UPDATE_WORKER";
+    case XCE_SCHED_OP_UPDATE_JOB:
-    case XCE_SCHED_OP_UPDATE_QUEUE:     return "UPDATE_QUEUE";
+        return "UPDATE_JOB";
-    case XCE_SCHED_OP_UPDATE_RESOURCES: return "UPDATE_RESOURCES";
+    case XCE_SCHED_OP_JSTATE_CHANGE:
-    case XCE_SCHED_OP_UPDATE_RESVALS_FULL: return "UPDATE_RESVALS_FULL";
+        return "JSTATE_CHANGE";
-    case XCE_SCHED_OP_UPDATE_RESVALS:   return "UPDATE_RESVALS";
+    case XCE_SCHED_OP_UPDATE_WORKER:
-    case XCE_SCHED_OP_JOB_RUSAGE:       return "JOB_RUSAGE";
+        return "UPDATE_WORKER";
-    case XCE_SCHED_OP_UPDATE_JOB_FIELDS: return "UPDATE_JOB_FIELDS";
+    case XCE_SCHED_OP_UPDATE_QUEUE:
-    default: break;
+        return "UPDATE_QUEUE";
+    case XCE_SCHED_OP_UPDATE_RESOURCES:
+        return "UPDATE_RESOURCES";
+    case XCE_SCHED_OP_UPDATE_RESVALS_FULL:
+        return "UPDATE_RESVALS_FULL";
+    case XCE_SCHED_OP_UPDATE_RESVALS:
+        return "UPDATE_RESVALS";
+    case XCE_SCHED_OP_JOB_RUSAGE:
+        return "JOB_RUSAGE";
+    case XCE_SCHED_OP_UPDATE_JOB_FIELDS:
+        return "UPDATE_JOB_FIELDS";
+    default:
+        break;
     }
 
     snprintf(buf, sizeof(buf), "UNKNOWN(%d)", (int)op);
@@ -858,13 +885,15 @@ int sched_update(xce_scheduler_t* sched)
             rc = sched_update_existing_queue(sched, req->u.queueupdate.queue);
             break;
         case XCE_SCHED_OP_UPDATE_RESOURCES: {
-            int              i;
+            int i;
             for (i = 0; i < req->u.resourceupdate.nresources; i++) {
                 xce_resource_t* r = req->u.resourceupdate.resources[i];
-                if (r == NULL) continue;
+                if (r == NULL)
+                    continue;
                 xce_resource_t* existing = makeResource(sched->core.resmgr, r);
                 if (existing == NULL) {
-                    /* EEXIST: update-in-place on existing resource (add-or-update) */
+                    /* EEXIST: update-in-place on existing resource
+                     * (add-or-update) */
                     xce_resource_t* found = NULL;
                     if (key_find(sched->core.resmgr, r->name, (void**)&found)
                             == 0
@@ -875,9 +904,11 @@ int sched_update(xce_scheduler_t* sched)
                         found->ordtype  = r->ordtype;
                         found->flags    = r->flags;
                         found->interval = r->interval;
-                        WARN("%s updated existing resource %s", __func__, r->name);
+                        WARN("%s updated existing resource %s", __func__,
+                             r->name);
                     } else {
-                        WARN("%s makeResource failed for %s and lookup not found",
+                        WARN("%s makeResource failed for %s and lookup not "
+                             "found",
                              __func__, r->name);
                     }
                 }
@@ -886,7 +917,7 @@ int sched_update(xce_scheduler_t* sched)
             break;
         }
         case XCE_SCHED_OP_UPDATE_RESVALS_FULL: {
-            int               i;
+            int i;
             for (i = 0; i < req->u.resvalfullupdate.nvals; i++) {
                 xce_resval_t*   rv = req->u.resvalfullupdate.vals[i];
                 xce_resource_t* r  = NULL;
@@ -905,15 +936,17 @@ int sched_update(xce_scheduler_t* sched)
             break;
         }
         case XCE_SCHED_OP_UPDATE_RESVALS: {
-            int              i;
+            int                i;
             xce_resval_list_t* resvals = req->u.resvalupdate.resvals;
             for (i = 0; i < resvals->count; i++) {
                 xce_resval_t*       rv = resvals->vals[i];
                 xce_sched_worker_t* w  = NULL;
                 xce_resource_t*     r  = NULL;
-                if (rv == NULL) continue;
+                if (rv == NULL)
+                    continue;
                 w = sched_get_worker(&(sched->core), rv->host);
-                if (w == NULL) continue;
+                if (w == NULL)
+                    continue;
                 if (key_find(sched->core.resmgr, rv->name, (void**)&r) < 0
                     || r == NULL) {
                     WARN("%s unknown resource %s in per-worker resvals update",
@@ -943,8 +976,8 @@ int sched_update(xce_scheduler_t* sched)
         sched_pend_update_killer(it);
 
         if (rc < 0) {
-            WARN("%s: op %s failed: %s (%d)", __func__,
+            WARN("%s: op %s failed: %s (%d)", __func__, sched_op_name(req->op),
-                 sched_op_name(req->op), strerror(-rc), rc);
+                 strerror(-rc), rc);
         }
     }
 

src/scheduler/testsuite/meson.build

@@ -434,6 +434,30 @@ sched_record_lifecycle_test = executable(
     dependencies: [scheduler_test_deps],
     install: false)
 
+# Contract test for the scheduler-side OPCode_schedAdmin request handler
+# (scheduler-core spec, task 9.4, Req 5 AC 10): decodes a forwarded
+# SchedAdminRequest, dispatches by action to sched_record_start()/
+# sched_record_stop()/sched_record_status(), and sends back a
+# SchedAdminReply matching requirements.md's field-semantics table.
+# recordStatus is proven synchronous (answered directly on the handler
+# thread, both idle and while genuinely recording); recordStart/recordStop
+# are proven deferred (enqueued, not executed, until a subsequent
+# xce_sched_driver_translate_pending_events() drain actually performs the
+# real sched_record_start()/sched_record_stop() call and sends the reply).
+# Same white-box #include pattern as sched_driver_translate_test
+# (core.c/core_clock.c/core_events.c/driver.c) PLUS ../src/recording.c/
+# recording_events.c/snapshot.c/snapshot_wire.c (real record start/stop/
+# status/write, task 9.1/9.2/9.3) PLUS ../src/handler.c (the code under
+# test). xcecli_reply_schedadmin() is a SPY test double, same pattern
+# sched_driver_decision_cb_test already establishes for
+# sched_publish_dispatch()/sched_publish_pend_reason().
+sched_admin_request_test = executable(
+    'sched_admin_request_test',
+    ['test_admin_request.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',
@@ -500,3 +524,6 @@ test('scheduler recording path validation test',
 test('scheduler record lifecycle test',
      sched_record_lifecycle_test, suite: 'scheduler',
      env: ['ASAN_OPTIONS=detect_leaks=0'])
+test('scheduler admin request test',
+     sched_admin_request_test, suite: 'scheduler',
+     env: ['ASAN_OPTIONS=detect_leaks=0'])

src/scheduler/testsuite/test_admin_request.c

@@ -0,0 +1,903 @@
+/*
+ * test_admin_request.c - contract test for the scheduler-side
+ * OPCode_schedAdmin request handler (scheduler-core spec, task 9.4, Req 5
+ * AC 10). This is the scheduler-side counterpart to task 10.1's
+ * controller-side forwarding dispatch (out of THIS spec's boundary): the
+ * controller relays a SchedAdminRequest verbatim to this endpoint, and this
+ * file proves the scheduler decodes it, dispatches by action to
+ * sched_record_start()/sched_record_stop()/sched_record_status() (tasks
+ * 9.2/9.3/9.4), and sends back a SchedAdminReply whose fields match
+ * requirements.md's "Reply field semantics by action" table.
+ *
+ * Observable under test:
+ *   - recordStatus is answered SYNCHRONOUSLY, on the calling (handler)
+ *     thread, directly from handler.c's sched_handle_admin_request() --
+ *     never enqueued -- both when not recording (all-zero/empty reply) and
+ *     while genuinely recording (live event count/file/started-at).
+ *   - recordStart/recordStop are DEFERRED: enqueued onto the same
+ *     pend_updates dataq every other controller-sourced update flows
+ *     through (XCE_SCHED_OP_ADMIN_RECORD), NOT executed synchronously --
+ *     `sched->recording` is provably unchanged immediately after
+ *     scheduler_req_handler() returns -- and only actually performed (real
+ *     sched_record_start()/sched_record_stop() call, real reply send) by a
+ *     subsequent xce_sched_driver_translate_pending_events() drain
+ *     (driver.c's handle_deferred_admin_record()).
+ *   - recordStart while already recording / recordStop while not recording
+ *     reply {ok=false, error="already recording"/"not recording"} with the
+ *     table's documented (unchanged) field values.
+ *   - An invalid recordFile (path separator) surfaces all the way through
+ *     the deferred path as {ok=false, error="invalid record file name"},
+ *     never just an internal log line, and never actually opens a file.
+ *   - An unrecognized SchedAdminAction returns -EINVAL, logs, and neither
+ *     enqueues anything nor sends any reply -- no crash.
+ *
+ * White-box #include of ../src/core.c, ../src/core_clock.c,
+ * ../src/core_events.c, ../src/driver.c (same set test_driver_translate.c
+ * uses to exercise xce_sched_driver_translate_pending_events() for real)
+ * PLUS ../src/recording.c/recording_events.c (task 9.1/9.2/9.3's REAL
+ * sched_record_start()/sched_record_stop()/sched_record_status()/
+ * sched_record_write_events() -- this file's whole point is proving the
+ * deferred path performs REAL file I/O, so these must be genuine, not
+ * stubbed) PLUS ../src/snapshot.c/snapshot_wire.c (task 8.1's real
+ * xce_sched_core_save_snapshot(), needed by sched_record_start()) PLUS
+ * ../src/handler.c (task 9.4's sched_handle_admin_request()/
+ * scheduler_req_handler(), the code under test).
+ *
+ * recording_session.c is NOT linked: its sched_record_session_begin()/
+ * sched_record_session_add_decision()/sched_record_session_end() trio is
+ * only reachable from xce_sched_driver_run_normal_mode_round(), which this
+ * file never calls (it drives xce_sched_driver_translate_pending_events()
+ * directly, same as test_driver_translate.c) -- stubbed below purely to
+ * satisfy the linker, same rationale as that file's identical stubs for the
+ * same reason. Likewise core_session.c/plugin.c are not linked (their
+ * exported functions are stubbed below); no real scheduling session is ever
+ * run by any test in this file.
+ *
+ * xcecli_reply_schedadmin() is a SPY test double (client.c's real
+ * implementation needs a live capnp connection this file never sets up) --
+ * same pattern test_driver_decision_cb.c already establishes for
+ * sched_publish_dispatch()/sched_publish_pend_reason(). It deep-copies
+ * every field of the passed sched_admin_reply_t (including string fields)
+ * into fixed-size static storage rather than retaining the passed pointers:
+ * both handle_deferred_admin_record() (recordStop's `file`) and
+ * sched_record_status() (`file`, a BORROWED alias of
+ * sched->record_file_path) document that their reply pointers are only
+ * valid for the duration of the synchronous call -- recordStop's in
+ * particular aliases a stack-local buffer that is gone the instant
+ * handle_deferred_admin_record() returns, so retaining the raw pointer
+ * would be a stack-use-after-return read the very next line.
+ */
+#include <errno.h>
+#include <limits.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include <unity/unity.h>
+
+#include "controller.capnp.h"
+#include "dataq.h"
+#include "kvmgr.h"
+#include "mylock.h"
+#include "scheduler.h"
+#include "xstring.h"
+
+/* Pull in the real functions under test into this TU (white-box #include,
+ * same convention as test_driver_translate.c/test_record_lifecycle.c). */
+#include "../src/core.c"
+#include "../src/core_clock.c"
+#include "../src/core_events.c"
+#include "../src/driver.c"
+#include "../src/recording.c"
+#include "../src/recording_events.c"
+#include "../src/snapshot.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; }
+void plugin_unregister(void* arg) { (void)arg; }
+
+/* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/
+ * sched_publish_pend_reason() -- driver.c's (real, #included above)
+ * xce_sched_driver_on_decision() references both; never actually invoked by
+ * any test in this file (no session is ever run, so no decision is ever
+ * drained). Same rationale as test_driver_translate.c's identical stubs. */
+int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d)
+{
+    (void)sched;
+    (void)d;
+    return 0;
+}
+
+int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
+{
+    (void)sched;
+    (void)r;
+    return 0;
+}
+
+/* Link-satisfying stubs: xce_sched_driver_run_normal_mode_round() (driver.c,
+ * real, #included above, but never CALLED by any test here -- this file
+ * only drives xce_sched_driver_translate_pending_events() directly, same as
+ * test_driver_translate.c) references xce_core_begin_session()/
+ * xce_core_signal_go()/xce_core_end_session() (core_session.c) and
+ * plugin_begin_session()/plugin_end_session() (plugin.c) /
+ * sched_account_jobs() (scheduler.c). Never actually invoked at runtime. */
+int xce_core_begin_session(xce_sched_core_t* core)
+{
+    (void)core;
+    return 0;
+}
+
+int xce_core_signal_go(xce_sched_core_t* core)
+{
+    (void)core;
+    return 0;
+}
+
+int xce_core_end_session(xce_sched_core_t* core)
+{
+    (void)core;
+    return 0;
+}
+
+int plugin_begin_session(void* arg)
+{
+    (void)arg;
+    return 0;
+}
+
+int plugin_end_session(void* arg)
+{
+    (void)arg;
+    return 0;
+}
+
+int sched_account_jobs(xce_scheduler_t* sched)
+{
+    (void)sched;
+    return 0;
+}
+
+/* Link-satisfying stubs for recording_session.c's real functions -- see
+ * this file's header comment for why recording_session.c itself is not
+ * linked. Never actually invoked (no session is ever run). */
+sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
+                                                       uint64_t begin_round_ts)
+{
+    (void)sched;
+    (void)begin_round_ts;
+    return NULL;
+}
+
+void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
+                                       uint64_t ts, xce_decision_t* decision)
+{
+    (void)ctx;
+    (void)ts;
+    (void)decision;
+}
+
+int sched_record_session_end(xce_scheduler_t*            sched,
+                             sched_record_session_ctx_t* ctx, uint64_t end_ts)
+{
+    (void)sched;
+    (void)ctx;
+    (void)end_ts;
+    return 0;
+}
+
+/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
+ * rationale as test_driver_translate.c's own copy (schedobj.c is not
+ * linked). Never actually exercised (no test here pushes core events, so
+ * xce_core_push_events() -- the only real caller -- is never reached: every
+ * enqueued request in this file is XCE_SCHED_OP_ADMIN_RECORD, intercepted
+ * by driver.c BEFORE the translated batch/push path). Present only to
+ * satisfy the linker for core_events.c. */
+int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
+{
+    (void)core;
+    (void)out;
+    return -EINVAL;
+}
+
+/* Link-satisfying stub for scheduler.c's sched_init_config() -- handler.c's
+ * (real, #included below) OPCode_push/configpushmsg case references it;
+ * this file never exercises that path (out of scope -- this file's whole
+ * surface is OPCode_schedAdmin). Never actually invoked. */
+int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
+{
+    (void)s;
+    (void)c;
+    return 0;
+}
+
+/* Pull in the real sched_handle_admin_request()/scheduler_req_handler()
+ * under test into this TU. */
+#include "../src/handler.c"
+
+/* ---------------------------------------------------------------------- */
+/* Spy test double for client.c's xcecli_reply_schedadmin() -- see file     */
+/* header comment for why every field is deep-copied rather than retaining  */
+/* the passed sched_admin_reply_t's reply->file/reply->error pointers.      */
+/* ---------------------------------------------------------------------- */
+
+static int          g_reply_calls;
+static uint64_t     g_last_seq;
+static xcecli_ctx_t g_last_ctx;
+static int          g_last_ok;
+static int          g_last_error_null;
+static char         g_last_error[256];
+static int          g_last_recording;
+static int          g_last_file_null;
+static char         g_last_file[PATH_MAX];
+static uint64_t     g_last_event_count;
+static uint64_t     g_last_started_at;
+
+static void reset_reply_spy(void)
+{
+    g_reply_calls      = 0;
+    g_last_seq         = 0;
+    g_last_ctx         = NULL;
+    g_last_ok          = 0;
+    g_last_error_null  = 1;
+    g_last_error[0]    = '\0';
+    g_last_recording   = 0;
+    g_last_file_null   = 1;
+    g_last_file[0]     = '\0';
+    g_last_event_count = 0;
+    g_last_started_at  = 0;
+}
+
+int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq,
+                            sched_admin_reply_t* areply)
+{
+    g_reply_calls++;
+    g_last_ctx = c;
+    g_last_seq = seq;
+
+    if (areply == NULL) {
+        return 0;
+    }
+
+    g_last_ok          = areply->ok;
+    g_last_recording   = areply->recording;
+    g_last_event_count = areply->eventCount;
+    g_last_started_at  = areply->startedAt;
+
+    g_last_error_null = (areply->error == NULL);
+    g_last_error[0]   = '\0';
+    if (areply->error != NULL) {
+        strncpy(g_last_error, areply->error, sizeof(g_last_error) - 1);
+    }
+
+    g_last_file_null = (areply->file == NULL);
+    g_last_file[0]   = '\0';
+    if (areply->file != NULL) {
+        strncpy(g_last_file, areply->file, sizeof(g_last_file) - 1);
+    }
+
+    return 0;
+}
+
+void setUp(void) { reset_reply_spy(); }
+void tearDown(void) { return; }
+
+/* ---------------------------------------------------------------------- */
+/* Fixture helpers -- same pattern as test_record_lifecycle.c's own.        */
+/* ---------------------------------------------------------------------- */
+
+static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
+{
+    char tmpl[PATH_MAX];
+
+    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_adminreq%s_XXXXXX", suffix);
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
+    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
+    strcpy(out, tmpl);
+}
+
+/* `record_dir` must outlive `sched` (caller-owned storage), same convention
+ * as test_record_lifecycle.c's fixture_init(). No picker plugin is
+ * registered -- unlike test_record_lifecycle.c's fixture, no test in this
+ * file ever runs a scheduling session, so core->job_pick_plugin staying
+ * NULL is fine (sched_resolve_job_picker() above is never actually
+ * reached). */
+static void fixture_init(xce_scheduler_t* sched, char* record_dir)
+{
+    int rc;
+
+    memset(sched, 0, sizeof(*sched));
+
+    /* This fixture hand-builds xce_scheduler_t rather than calling the full
+     * sched_init() (which starts real threads/network I/O this white-box
+     * suite never wants) -- record_lock (scheduler.h field comments, task
+     * 9.4 fix round) is normally initialized by sched_init(), so it must be
+     * initialized here too, the same way core.lock is initialized by the
+     * xce_sched_core_init() call just below. Every test in this file
+     * exercises sched_record_status()/sched_record_start()/
+     * sched_record_stop()/sched_record_write_events() for real, all of
+     * which now take this lock. */
+    mylock_init(&(sched->record_lock));
+
+    rc = xce_sched_core_init(&(sched->core));
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
+    dataq_init(sched->core.pend_updates[0], NULL);
+    sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
+    dataq_init(sched->core.pend_updates[1], NULL);
+
+    sched->record_dir = record_dir;
+    /* Dummy non-NULL xcecli_ctx_t (an opaque void*) -- xcecli_reply_schedadmin
+     * is spied above, so nothing ever dereferences this; a real production
+     * ctx would come from xcecli_ctx_init(). */
+    sched->apictx = (xcecli_ctx_t)sched;
+}
+
+/* Builds a request/reply pair on the stack and drives it through the REAL
+ * production entry point, scheduler_req_handler() (handler.c) -- exactly
+ * the function main.c registers as the scheduler's request handler and the
+ * function that dispatches OPCode_schedAdmin to sched_handle_admin_request()
+ * (handler.c's own new `case` this task added). */
+static int send_admin_request(xce_scheduler_t* sched, int action,
+                              const char* record_file, uint64_t seq)
+{
+    sched_admin_req_t areq;
+    xce_request_t     request;
+
+    memset(&areq, 0, sizeof(areq));
+    areq.action     = action;
+    areq.recordFile = (char*)record_file;
+
+    memset(&request, 0, sizeof(request));
+    request.opcode             = OPCode_schedAdmin;
+    request.seq                = seq;
+    request.body_which         = Request_body_schedadminmsg;
+    request.body.schedadminmsg = &areq;
+
+    return scheduler_req_handler(&request, sched);
+}
+
+/* ---------------------------------------------------------------------- */
+/* recordStatus -- synchronous, answered directly from the handler thread. */
+/* ---------------------------------------------------------------------- */
+
+static void test_recordstatus_not_recording_returns_empty_state(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_status_idle");
+    fixture_init(&sched, record_dir);
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStatus, NULL, 111);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_EQUAL_UINT64(111, g_last_seq);
+    TEST_ASSERT_EQUAL_PTR(sched.apictx, g_last_ctx);
+    TEST_ASSERT_TRUE(g_last_ok);
+    TEST_ASSERT_FALSE(g_last_recording);
+    TEST_ASSERT_TRUE(g_last_file_null);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at);
+
+    /* recordStatus never touches the deferred queue. */
+    TEST_ASSERT_NULL(dataq_deq(sched.core.pend_updates[0], NULL));
+}
+
+static void test_recordstatus_while_recording_reflects_live_state(void)
+{
+    char             record_dir[PATH_MAX];
+    xce_scheduler_t  sched;
+    xce_core_event_t ev;
+    int              rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_status_live");
+    fixture_init(&sched, record_dir);
+
+    rc = sched_record_start(&sched, "live-status.replay");
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_TRUE(sched.recording);
+
+    /* Real event writes via recording_events.c's genuine
+     * sched_record_write_events() -- record_event_count genuinely advances,
+     * not a fabricated field. */
+    memset(&ev, 0, sizeof(ev));
+    ev.type                     = XCE_CORE_EVENT_JOB_STATE_CHANGE;
+    ev.u.job_state_change.jobid = 7ULL;
+    TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1));
+    ev.u.job_state_change.jobid = 8ULL;
+    TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1));
+    TEST_ASSERT_EQUAL_UINT64(2ULL, sched.record_event_count);
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStatus, NULL, 222);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_TRUE(g_last_ok);
+    TEST_ASSERT_TRUE(g_last_recording);
+    TEST_ASSERT_FALSE(g_last_file_null);
+    TEST_ASSERT_TRUE(strstr(g_last_file, "live-status.replay") != NULL);
+    TEST_ASSERT_EQUAL_UINT64(2ULL, g_last_event_count);
+    TEST_ASSERT_TRUE(g_last_started_at > 0);
+    TEST_ASSERT_EQUAL_UINT64(sched.record_start_ts, g_last_started_at);
+}
+
+/* ---------------------------------------------------------------------- */
+/* recordStatus concurrency -- record_lock (scheduler.h field comments,     */
+/* task 9.4 fix round) genuinely protects the recording-state fields       */
+/* against the vtask-thread sched_record_status() reader / driver-thread   */
+/* writer race the code review flagged. Two DETERMINISTIC proofs rather    */
+/* than a single flaky, purely timing-dependent race reproduction --       */
+/* task 9.2's TOCTOU test already established this codebase's precedent    */
+/* of preferring a deterministic proof over a genuinely racy test:         */
+/*                                                                          */
+/*   1. record_lock is a REAL mutex a second, genuine pthread actually     */
+/*      blocks on -- not just "the code calls mylock_lock()" verified by   */
+/*      inspection. Proven by holding the lock on the main thread          */
+/*      (simulating "driver thread is mid-mutation"), launching a real     */
+/*      thread that calls sched_record_status(), and asserting it has NOT  */
+/*      completed after a generous wait -- then asserting it DOES          */
+/*      complete once the lock is released. The only timing dependency is  */
+/*      in the safe direction: a slow CI runner can only make the "still   */
+/*      blocked" assertion harder to observe, never produce a false        */
+/*      positive (record_lock is genuinely held for the whole wait).       */
+/*   2. reply->file is a genuinely independent allocation, not an alias of */
+/*      the live sched->record_file_path -- proven by taking a status      */
+/*      snapshot, then calling the real sched_record_stop() (which frees   */
+/*      sched->record_file_path for real) and confirming the snapshot's    */
+/*      reply->file is untouched. This reproduces the exact               */
+/*      use-after-free the finding identified, deterministically, with no  */
+/*      reliance on OS thread-scheduling timing at all.                    */
+/* ---------------------------------------------------------------------- */
+
+typedef struct {
+    xce_scheduler_t*    sched;
+    sched_admin_reply_t reply;
+    int                 rc;
+    atomic_int          done;
+} status_thread_arg_t;
+
+static void* status_thread_fn(void* arg)
+{
+    status_thread_arg_t* a = (status_thread_arg_t*)arg;
+
+    a->rc = sched_record_status(a->sched, &(a->reply));
+    atomic_store(&(a->done), 1);
+
+    return NULL;
+}
+
+static void test_recordstatus_genuinely_blocks_on_record_lock(void)
+{
+    char                record_dir[PATH_MAX];
+    xce_scheduler_t     sched;
+    status_thread_arg_t arg;
+    pthread_t           th;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_status_lockblock");
+    fixture_init(&sched, record_dir);
+
+    TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "lockblock.replay"));
+
+    memset(&arg, 0, sizeof(arg));
+    arg.sched = &sched;
+    atomic_store(&(arg.done), 0);
+
+    /* Simulate "driver thread is mid-mutation": hold record_lock on THIS
+     * (main test) thread, exactly the window sched_record_start()/
+     * sched_record_stop()/sched_record_abort_on_error() hold it for in
+     * production. */
+    mylock_lock(&(sched.record_lock));
+
+    TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, status_thread_fn, &arg));
+
+    /* Generous wait -- if record_lock were NOT actually taken by
+     * sched_record_status() (i.e. the fix regressed to a no-op), the
+     * spawned thread would complete almost instantly and `done` would
+     * already be 1 here. */
+    usleep(150000); /* 150ms */
+
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&(arg.done)));
+
+    mylock_unlock(&(sched.record_lock));
+
+    TEST_ASSERT_EQUAL_INT(0, pthread_join(th, NULL));
+
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&(arg.done)));
+    TEST_ASSERT_EQUAL_INT(0, arg.rc);
+    TEST_ASSERT_TRUE(arg.reply.recording);
+    TEST_ASSERT_NOT_NULL(arg.reply.file);
+    TEST_ASSERT_TRUE(strstr(arg.reply.file, "lockblock.replay") != NULL);
+
+    free(arg.reply.file);
+    TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched));
+}
+
+static void test_recordstatus_file_survives_concurrent_record_stop_free(void)
+{
+    char                record_dir[PATH_MAX];
+    xce_scheduler_t     sched;
+    sched_admin_reply_t snapshot;
+    int                 rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_status_uaf");
+    fixture_init(&sched, record_dir);
+
+    rc = sched_record_start(&sched, "uaftest.replay");
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    rc = sched_record_status(&sched, &snapshot);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_NOT_NULL(snapshot.file);
+    /* Genuinely a DIFFERENT allocation from the live pointer -- not an
+     * alias (the exact bug this fix closes: the pre-fix version set
+     * reply->file = sched->record_file_path directly). */
+    TEST_ASSERT_TRUE(sched.record_file_path != snapshot.file);
+    TEST_ASSERT_EQUAL_STRING(sched.record_file_path, snapshot.file);
+
+    /* sched_record_stop() frees sched->record_file_path for real -- if
+     * snapshot.file aliased it, this would be a genuine use-after-free
+     * (ASan would abort) the moment the next assertion reads it. */
+    rc = sched_record_stop(&sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_NULL(sched.record_file_path);
+
+    TEST_ASSERT_TRUE(strstr(snapshot.file, "uaftest.replay") != NULL);
+
+    free(snapshot.file);
+}
+
+/* ---------------------------------------------------------------------- */
+/* recordStart -- deferred, not synchronous.                               */
+/* ---------------------------------------------------------------------- */
+
+static void test_recordstart_is_deferred_then_performed_by_driver_round(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    char            expect_path[PATH_MAX];
+    int             rc;
+    struct stat     st;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_start_deferred");
+    fixture_init(&sched, record_dir);
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStart,
+                            "admin-start.replay", 333);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* Deferred, not synchronous: no start yet, no reply yet. */
+    TEST_ASSERT_FALSE(sched.recording);
+    TEST_ASSERT_NULL(sched.record_file_path);
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+
+    /* A subsequent driver round actually performs the deferred start. */
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc); /* n==0: the admin op never enters the
+                                   * translated event batch. */
+
+    TEST_ASSERT_TRUE(sched.recording);
+    TEST_ASSERT_NOT_NULL(sched.record_file_path);
+    TEST_ASSERT_TRUE(strstr(sched.record_file_path, "admin-start.replay")
+                     != NULL);
+
+    /* The file is genuinely created on disk. */
+    snprintf(expect_path, sizeof(expect_path), "%s/admin-start.replay",
+             record_dir);
+    TEST_ASSERT_EQUAL_INT(0, stat(expect_path, &st));
+    TEST_ASSERT_TRUE(S_ISREG(st.st_mode));
+
+    /* And the deferred reply was genuinely sent, matching the "recordStart
+     * (ok)" column of requirements.md's field-semantics table. */
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_EQUAL_UINT64(333, g_last_seq);
+    TEST_ASSERT_TRUE(g_last_ok);
+    TEST_ASSERT_TRUE(g_last_error_null);
+    TEST_ASSERT_TRUE(g_last_recording);
+    TEST_ASSERT_FALSE(g_last_file_null);
+    TEST_ASSERT_TRUE(strstr(g_last_file, "admin-start.replay") != NULL);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count);
+    TEST_ASSERT_TRUE(g_last_started_at > 0);
+}
+
+static void test_recordstart_empty_file_auto_generates_name(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_start_auto");
+    fixture_init(&sched, record_dir);
+
+    /* NULL recordFile -- "empty = auto-generate" (requirements.md's
+     * SchedAdminRequest.recordFile field comment). */
+    rc = send_admin_request(&sched, SchedAdminAction_recordStart, NULL, 444);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    TEST_ASSERT_TRUE(sched.recording);
+    TEST_ASSERT_NOT_NULL(sched.record_file_path);
+    TEST_ASSERT_TRUE(strstr(sched.record_file_path, record_dir) != NULL);
+    TEST_ASSERT_TRUE(strstr(sched.record_file_path, ".replay") != NULL);
+
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_TRUE(g_last_ok);
+    TEST_ASSERT_FALSE(g_last_file_null);
+}
+
+static void test_recordstart_while_already_recording_replies_error(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    char            path_before[PATH_MAX];
+    uint64_t        started_before;
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_start_already");
+    fixture_init(&sched, record_dir);
+
+    rc = sched_record_start(&sched, "first.replay");
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    strcpy(path_before, sched.record_file_path);
+    started_before = sched.record_start_ts;
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStart,
+                            "second.replay", 555);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* Original recording untouched by the rejected second start. */
+    TEST_ASSERT_TRUE(sched.recording);
+    TEST_ASSERT_EQUAL_STRING(path_before, sched.record_file_path);
+    TEST_ASSERT_EQUAL_UINT64(0ULL, sched.record_event_count);
+
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_EQUAL_UINT64(555, g_last_seq);
+    TEST_ASSERT_FALSE(g_last_ok);
+    TEST_ASSERT_FALSE(g_last_error_null);
+    TEST_ASSERT_EQUAL_STRING("already recording", g_last_error);
+    /* "(unchanged)" -- requirements.md's recordStart(err) column. */
+    TEST_ASSERT_TRUE(g_last_recording);
+    TEST_ASSERT_FALSE(g_last_file_null);
+    TEST_ASSERT_EQUAL_STRING(path_before, g_last_file);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count);
+    TEST_ASSERT_EQUAL_UINT64(started_before, g_last_started_at);
+
+    TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched));
+}
+
+static void test_recordstart_invalid_filename_replies_error(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    char            escape_path[PATH_MAX];
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_start_invalid");
+    fixture_init(&sched, record_dir);
+
+    /* Contains a path separator -- rejected before any filesystem access
+     * (recording.c's sched_validate_record_filename(), task 9.1). */
+    rc = send_admin_request(&sched, SchedAdminAction_recordStart,
+                            "sub/dir.replay", 666);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* No file opened, no directory created, recording never started. */
+    TEST_ASSERT_FALSE(sched.recording);
+    TEST_ASSERT_NULL(sched.record_file_path);
+    TEST_ASSERT_NULL(sched.record_file);
+    snprintf(escape_path, sizeof(escape_path), "%s/sub", record_dir);
+    TEST_ASSERT_FALSE(
+        (access(escape_path, F_OK) == 0)); /* "sub" was never created */
+
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_FALSE(g_last_ok);
+    TEST_ASSERT_FALSE(g_last_error_null);
+    TEST_ASSERT_EQUAL_STRING("invalid record file name", g_last_error);
+    TEST_ASSERT_FALSE(g_last_recording);
+    TEST_ASSERT_TRUE(g_last_file_null);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at);
+}
+
+/* ---------------------------------------------------------------------- */
+/* recordStop -- deferred, not synchronous.                                */
+/* ---------------------------------------------------------------------- */
+
+static void test_recordstop_is_deferred_then_performed_by_driver_round(void)
+{
+    char             record_dir[PATH_MAX];
+    xce_scheduler_t  sched;
+    char             path_before[PATH_MAX];
+    uint64_t         started_before;
+    xce_core_event_t ev;
+    int              rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_stop_deferred");
+    fixture_init(&sched, record_dir);
+
+    rc = sched_record_start(&sched, "stopme.replay");
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    strcpy(path_before, sched.record_file_path);
+    started_before = sched.record_start_ts;
+
+    memset(&ev, 0, sizeof(ev));
+    ev.type                     = XCE_CORE_EVENT_JOB_STATE_CHANGE;
+    ev.u.job_state_change.jobid = 9ULL;
+    TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1));
+    TEST_ASSERT_EQUAL_UINT64(1ULL, sched.record_event_count);
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStop, NULL, 777);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* Deferred, not synchronous: still recording, no reply yet. */
+    TEST_ASSERT_TRUE(sched.recording);
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    TEST_ASSERT_FALSE(sched.recording);
+    TEST_ASSERT_NULL(sched.record_file_path);
+    TEST_ASSERT_NULL(sched.record_file);
+
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_EQUAL_UINT64(777, g_last_seq);
+    TEST_ASSERT_TRUE(g_last_ok);
+    TEST_ASSERT_TRUE(g_last_error_null);
+    TEST_ASSERT_FALSE(g_last_recording);
+    /* "last path (informational)". */
+    TEST_ASSERT_FALSE(g_last_file_null);
+    TEST_ASSERT_EQUAL_STRING(path_before, g_last_file);
+    /* "final count". */
+    TEST_ASSERT_EQUAL_UINT64(1ULL, g_last_event_count);
+    /* "original start (historical)". */
+    TEST_ASSERT_EQUAL_UINT64(started_before, g_last_started_at);
+}
+
+static void test_recordstop_while_not_recording_replies_error(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_stop_notrecording");
+    fixture_init(&sched, record_dir);
+
+    rc = send_admin_request(&sched, SchedAdminAction_recordStop, NULL, 888);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls); /* deferred */
+
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    TEST_ASSERT_FALSE(sched.recording);
+
+    TEST_ASSERT_EQUAL_INT(1, g_reply_calls);
+    TEST_ASSERT_EQUAL_UINT64(888, g_last_seq);
+    TEST_ASSERT_FALSE(g_last_ok);
+    TEST_ASSERT_FALSE(g_last_error_null);
+    TEST_ASSERT_EQUAL_STRING("not recording", g_last_error);
+    TEST_ASSERT_FALSE(g_last_recording);
+    TEST_ASSERT_TRUE(g_last_file_null);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_event_count);
+    TEST_ASSERT_EQUAL_UINT64(0, g_last_started_at);
+}
+
+/* ---------------------------------------------------------------------- */
+/* Unrecognized action / NULL argument guards.                             */
+/* ---------------------------------------------------------------------- */
+
+static void test_unrecognized_action_returns_einval_no_crash_no_reply(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    int             rc;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_bad_action");
+    fixture_init(&sched, record_dir);
+
+    rc = send_admin_request(&sched, 999 /* not a valid SchedAdminAction */,
+                            NULL, 1010);
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);          /* no synchronous reply */
+    TEST_ASSERT_NULL(
+        dataq_deq(sched.core.pend_updates[0], NULL)); /* nothing enqueued */
+
+    /* A subsequent driver round is a genuine no-op -- nothing to drain. */
+    rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+}
+
+static void test_null_sched_returns_einval(void)
+{
+    sched_admin_req_t areq;
+    xce_request_t     request;
+
+    memset(&areq, 0, sizeof(areq));
+    areq.action = SchedAdminAction_recordStatus;
+
+    memset(&request, 0, sizeof(request));
+    request.opcode             = OPCode_schedAdmin;
+    request.body_which         = Request_body_schedadminmsg;
+    request.body.schedadminmsg = &areq;
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_admin_request(NULL, &request));
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+}
+
+static void test_null_request_returns_einval(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_null_req");
+    fixture_init(&sched, record_dir);
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_admin_request(&sched, NULL));
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+}
+
+static void test_null_schedadminmsg_body_returns_einval(void)
+{
+    char            record_dir[PATH_MAX];
+    xce_scheduler_t sched;
+    xce_request_t   request;
+
+    make_tmp_dir(record_dir, sizeof(record_dir), "_null_body");
+    fixture_init(&sched, record_dir);
+
+    memset(&request, 0, sizeof(request));
+    request.opcode             = OPCode_schedAdmin;
+    request.body_which         = Request_body_schedadminmsg;
+    request.body.schedadminmsg = NULL;
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL,
+                          sched_handle_admin_request(&sched, &request));
+    TEST_ASSERT_EQUAL_INT(0, g_reply_calls);
+}
+
+int main(void)
+{
+    UNITY_BEGIN();
+
+    RUN_TEST(test_recordstatus_not_recording_returns_empty_state);
+    RUN_TEST(test_recordstatus_while_recording_reflects_live_state);
+    RUN_TEST(test_recordstatus_genuinely_blocks_on_record_lock);
+    RUN_TEST(test_recordstatus_file_survives_concurrent_record_stop_free);
+
+    RUN_TEST(test_recordstart_is_deferred_then_performed_by_driver_round);
+    RUN_TEST(test_recordstart_empty_file_auto_generates_name);
+    RUN_TEST(test_recordstart_while_already_recording_replies_error);
+    RUN_TEST(test_recordstart_invalid_filename_replies_error);
+
+    RUN_TEST(test_recordstop_is_deferred_then_performed_by_driver_round);
+    RUN_TEST(test_recordstop_while_not_recording_replies_error);
+
+    RUN_TEST(test_unrecognized_action_returns_einval_no_crash_no_reply);
+    RUN_TEST(test_null_sched_returns_einval);
+    RUN_TEST(test_null_request_returns_einval);
+    RUN_TEST(test_null_schedadminmsg_body_returns_einval);
+
+    return UNITY_END();
+}

src/scheduler/testsuite/test_driver_decision_cb.c

@@ -134,6 +134,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
     return 0;
 }
 
+/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) --
+ * driver.c's handle_deferred_admin_record() (new this task) calls both;
+ * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so
+ * neither is actually invoked at runtime, same rationale as
+ * sched_record_write_events above. */
+int sched_record_start(xce_scheduler_t* sched, const char* record_file)
+{
+    (void)sched;
+    (void)record_file;
+    return 0;
+}
+
+int sched_record_stop(xce_scheduler_t* sched)
+{
+    (void)sched;
+    return 0;
+}
+
 sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
                                                        uint64_t begin_round_ts)
 {

src/scheduler/testsuite/test_driver_session_ordering.c

@@ -229,6 +229,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
     return 0;
 }
 
+/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) --
+ * driver.c's handle_deferred_admin_record() (new this task) calls both;
+ * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so
+ * neither is actually invoked at runtime, same rationale as
+ * sched_record_write_events above. */
+int sched_record_start(xce_scheduler_t* sched, const char* record_file)
+{
+    (void)sched;
+    (void)record_file;
+    return 0;
+}
+
+int sched_record_stop(xce_scheduler_t* sched)
+{
+    (void)sched;
+    return 0;
+}
+
 sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
                                                        uint64_t begin_round_ts)
 {

src/scheduler/testsuite/test_driver_translate.c

@@ -112,6 +112,24 @@ int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
     return 0;
 }
 
+/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) --
+ * driver.c's handle_deferred_admin_record() (new this task) calls both;
+ * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so
+ * neither is actually invoked at runtime, same rationale as
+ * sched_record_write_events above. */
+int sched_record_start(xce_scheduler_t* sched, const char* record_file)
+{
+    (void)sched;
+    (void)record_file;
+    return 0;
+}
+
+int sched_record_stop(xce_scheduler_t* sched)
+{
+    (void)sched;
+    return 0;
+}
+
 sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
                                                        uint64_t begin_round_ts)
 {

src/scheduler/testsuite/test_handlers.c

@@ -27,14 +27,30 @@
  * test_job_array_expand.c for the real kvmgr-backed sched_get_job). */
 int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
 {
-    (void)s; (void)c; return 0;
+    (void)s;
+    (void)c;
+    return 0;
 }
 
 /* sched_get_job relocated to core.h (scheduler-core spec, task 4.1) -- now
  * operates on xce_sched_core_t*, not xce_scheduler_t*. */
 xce_sched_job_t* sched_get_job(xce_sched_core_t* sched, uint64_t jobid)
 {
-    (void)sched; (void)jobid; return NULL;
+    (void)sched;
+    (void)jobid;
+    return NULL;
+}
+
+/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
+ * handler.c's OPCode_schedAdmin/recordStatus path references it; this file
+ * never exercises OPCode_schedAdmin (its own scope is notify/dataset
+ * enqueue), so a link-satisfying stub is enough, same rationale as
+ * sched_init_config above. */
+int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
+{
+    (void)sched;
+    (void)reply;
+    return -EINVAL;
 }
 
 #include "../src/handler.c"
@@ -118,8 +134,7 @@ static void test_handle_job_rusage_enqueues_rusage(void)
     xce_sched_op_t  op = (xce_sched_op_t)-1;
     fixture_init(&sched);
 
-    TEST_ASSERT_EQUAL_INT(-EINVAL,
+    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_sched_job_rusage(&sched, NULL));
-                          sched_handle_sched_job_rusage(&sched, NULL));
 
     xce_job_rusages_t* ru = (xce_job_rusages_t*)calloc(1, sizeof(*ru));
     ru->jobid             = 7;
@@ -132,20 +147,18 @@ static void test_handle_job_rusage_enqueues_rusage(void)
 static void test_handle_update_resvals_skips_empty_enqueues_nonempty(void)
 {
     xce_scheduler_t   sched;
-    xce_sched_op_t    op = (xce_sched_op_t)-1;
+    xce_sched_op_t    op    = (xce_sched_op_t)-1;
     xce_resval_list_t empty = { .count = 0, .vals = NULL };
     xce_resval_list_t full  = { .count = 1, .vals = NULL };
     fixture_init(&sched);
 
     /* NULL or count==0 -> skip (return 0, nothing enqueued) */
     TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, NULL));
-    TEST_ASSERT_EQUAL_INT(0,
+    TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &empty));
-                          sched_handle_sched_update_resvals(&sched, &empty));
     TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL));
 
     /* count>0 -> enqueue UPDATE_RESVALS */
-    TEST_ASSERT_EQUAL_INT(0,
+    TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &full));
-                          sched_handle_sched_update_resvals(&sched, &full));
     TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
     TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_RESVALS, op);
 }
@@ -154,9 +167,9 @@ static void test_handle_update_resvals_skips_empty_enqueues_nonempty(void)
 
 static void test_handle_resources_skips_empty_enqueues_nonempty(void)
 {
-    xce_scheduler_t  sched;
+    xce_scheduler_t sched;
-    xce_sched_op_t   op = (xce_sched_op_t)-1;
+    xce_sched_op_t  op = (xce_sched_op_t)-1;
-    xce_resource_t*  one[1];
+    xce_resource_t* one[1];
     fixture_init(&sched);
 
     TEST_ASSERT_EQUAL_INT(0, sched_handle_resources(&sched, 0, NULL));
@@ -188,7 +201,8 @@ static void test_handle_resvals_full_skips_empty_enqueues_nonempty(void)
 
 int main(int argc, char* argv[])
 {
-    (void)argc; (void)argv;
+    (void)argc;
+    (void)argv;
     UNITY_BEGIN();
     RUN_TEST(test_handle_update_job_enqueues_job_fields, 0);
     RUN_TEST(test_handle_update_queue_enqueues_queue, 0);

src/scheduler/testsuite/test_job_array_expand.c

@@ -29,7 +29,20 @@
 
 int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
 {
-    (void)s; (void)c; return 0;
+    (void)s;
+    (void)c;
+    return 0;
+}
+
+/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
+ * handler.c's OPCode_schedAdmin/recordStatus path references it; this file
+ * never exercises OPCode_schedAdmin, so a link-satisfying stub is enough,
+ * same rationale as sched_init_config above. */
+int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
+{
+    (void)sched;
+    (void)reply;
+    return -EINVAL;
 }
 
 /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors
@@ -84,7 +97,7 @@ static void fixture_init(xce_scheduler_t* sched)
 static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid)
 {
     xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
-    job->jobid            = jobid;
+    job->jobid           = jobid;
     sched_add_job(&(sched->core), job);
 }
 
@@ -99,9 +112,8 @@ static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid)
 static xce_job_submit_t* init_job_array(uint64_t base_jobid,
                                         uint64_t array_size)
 {
-    job_submit_data_t* submit
+    job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit));
-        = (job_submit_data_t*)calloc(1, sizeof(*submit));
+    xce_job_submit_t*  ja     = (xce_job_submit_t*)calloc(1, sizeof(*ja));
-    xce_job_submit_t*  ja = (xce_job_submit_t*)calloc(1, sizeof(*ja));
 
     submit->queue    = strdup("normal");
     submit->user     = strdup("tester");
@@ -145,12 +157,12 @@ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out)
 
 static void test_job_array_push_expands_into_n_tasks(void)
 {
-    xce_scheduler_t    sched;
+    xce_scheduler_t   sched;
-    xce_job_submit_t*  ja;
+    xce_job_submit_t* ja;
-    dataset_t          dataset;
+    dataset_t         dataset;
-    xce_job_t**        jobs = NULL;
+    xce_job_t**       jobs = NULL;
-    int                njobs;
+    int               njobs;
-    int                i;
+    int               i;
 
     fixture_init(&sched);
     ja = init_job_array(5000, 3);
@@ -188,11 +200,11 @@ static void test_job_array_push_expands_into_n_tasks(void)
 
 static void test_job_array_redelivery_does_not_duplicate(void)
 {
-    xce_scheduler_t   sched;
+    xce_scheduler_t sched;
-    dataset_t         dataset;
+    dataset_t       dataset;
-    xce_job_t**       jobs = NULL;
+    xce_job_t**     jobs = NULL;
-    int               njobs;
+    int             njobs;
-    int               i;
+    int             i;
 
     fixture_init(&sched);
 
@@ -289,9 +301,9 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
     xce_data_t        data;
     xce_request_t     request;
     xce_request_t     decoded;
-    io_packet_t      *pkt;
+    io_packet_t*      pkt;
     dataset_t         dataset;
-    xce_job_t       **jobs = NULL;
+    xce_job_t**       jobs = NULL;
     int               njobs;
     int               i;
 
@@ -319,12 +331,13 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
     request.body_which   = Request_body_pushmsg;
     request.body.pushmsg = &data;
 
-    pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
+    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
     TEST_ASSERT_NOT_NULL(pkt);
     TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt));
 
     memset(&decoded, 0, sizeof(decoded));
-    TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded)); /* consumes/frees pkt */
+    TEST_ASSERT_EQUAL_INT(0,
+                          xce_decode(pkt, &decoded)); /* consumes/frees pkt */
 
     /* Act 1: the REAL transport dispatcher (task 5.3b's fix) -- exactly what
      * vtask_entry() (xcectx.c) calls on every inbound push. */
@@ -364,7 +377,8 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
 
 int main(int argc, char* argv[])
 {
-    (void)argc; (void)argv;
+    (void)argc;
+    (void)argv;
     UNITY_BEGIN();
     RUN_TEST(test_job_array_push_expands_into_n_tasks, 0);
     RUN_TEST(test_job_array_redelivery_does_not_duplicate, 0);

src/scheduler/testsuite/test_job_array_wire_e2e.c

@@ -79,6 +79,17 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
     return 0;
 }
 
+/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
+ * handler.c's OPCode_schedAdmin/recordStatus path references it; this file
+ * never exercises OPCode_schedAdmin, so a link-satisfying stub is enough,
+ * same rationale as sched_init_config above. */
+int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
+{
+    (void)sched;
+    (void)reply;
+    return -EINVAL;
+}
+
 /* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors
  * test_job_array_expand.c's precedent), needed so sched_on_job_array()'s
  * per-element dedup check (sched_get_job) is exercised for real. Relocated

src/scheduler/testsuite/test_record_lifecycle.c

@@ -342,6 +342,15 @@ static void fixture_init(xce_scheduler_t* sched, char* record_dir)
 
     memset(sched, 0, sizeof(*sched));
 
+    /* record_lock (scheduler.h field comments, task 9.4 fix round) is
+     * normally initialized by sched_init(), which this white-box fixture
+     * deliberately bypasses (same rationale as this function's own header
+     * comment) -- initialize it here too, same pairing as the
+     * xce_sched_core_init() call just below initializing core.lock. Every
+     * sched_record_start()/sched_record_stop() call in this file now takes
+     * this lock. */
+    mylock_init(&(sched->record_lock));
+
     rc = xce_sched_core_init(&(sched->core));
     TEST_ASSERT_EQUAL_INT(0, rc);